import { Time } from '@n8n/constants';
import { ensureError } from '@n8n/utils/errors/ensure-error';

import {
	DEFAULT_CLOCK_SKEW_OPTIONS,
	isClockSkewSignificant,
	measureClockSkew,
	type ClockSkewOptions,
} from './clock-skew';
import {
	InvalidLifecycleOptionsError,
	InvalidReconciliationOptionsError,
	InvalidSchedulerDepsError,
} from './errors';
import {
	DEFAULT_EXECUTOR_OPTIONS,
	Executor,
	PrecisionTimer,
	TaskHandlerRegistry,
} from './executor';
import type { ExecutorOptions, ExecutorTaskStore } from './executor';
import { DEFAULT_LIFECYCLE_OPTIONS, pollLookaheadSeconds, Loop, PASS_TIMED_OUT } from './lifecycle';
import type { LifecycleOptions } from './lifecycle';
import { DEFAULT_MATERIALIZER_OPTIONS, materialize, totalDiscarded } from './materializer';
import type { MaterializerOptions, RunInTransaction } from './materializer';
import { DEFAULT_REAPER_OPTIONS, reap } from './reaper';
import type { ReaperOptions, ReaperTaskStore } from './reaper';
import { DEFAULT_RECONCILIATION_OPTIONS, reconcile } from './reconciliation';
import type {
	ReconciliationCursor,
	ReconciliationJobStore,
	ReconciliationOptions,
	ReconciliationSummary,
	ScheduledJobOwnerRegistry,
} from './reconciliation';
import { DEFAULT_RETENTION_OPTIONS, prune } from './retention';
import type { RetentionOptions, RetentionStore } from './retention';
import type { Scheduler, SchedulerPasses } from './scheduler';
import { SCHEDULER_ATTRIBUTES } from '../observability/attributes';
import { createExecutorTracing, withHandoffTracing } from '../observability/executor-tracing';
import { traceCreatedTasks } from '../observability/materializer-tracing';
import { noopMetrics, type SchedulerMetrics } from '../observability/metrics';
import { tracePass } from '../observability/pass-tracing';
import { noopTracer, type Tracer } from '../observability/tracer';

export type SchedulerEventLevel = 'debug' | 'info' | 'warn' | 'error';

/**
 * A task firing this far past its scheduled time is reported once, per fire. Kept
 * well above normal sub-second dispatch jitter so the warning flags a genuinely
 * late fire (a blocked event loop, a skewed clock) rather than routine load.
 */
export const DEFAULT_DISPATCH_LAG_WARN_THRESHOLD_SECONDS = 30;

/**
 * A lifecycle milestone, incident or notable outcome the scheduler reports,
 * already handled where it fired. The host only decides how to report it
 * (typically: route to its logger).
 */
export interface SchedulerEvent {
	level: SchedulerEventLevel;
	message: string;
	context: Record<string, unknown>;
}

/**
 * Task storage for the executor, reaper and retention passes. The `@n8n/db`
 * `ScheduledTaskRepository` satisfies the intersection structurally, so the
 * host passes it as-is.
 */
export type SchedulerTaskStore = ExecutorTaskStore & ReaperTaskStore & RetentionStore;

/** What a host binds to compose a {@link Scheduler}. */
export interface SchedulerDeps {
	/** Stable identity of this instance, stamped on every claim. */
	hostId: string;

	/** Runs one materializer pass inside one storage transaction. */
	materializerTransaction: RunInTransaction;

	taskStore: SchedulerTaskStore;

	/** Per-pass tuning; anything omitted falls back to the pass's defaults. */
	materializer?: Partial<MaterializerOptions>;
	executor?: Partial<ExecutorOptions>;
	reaper?: Partial<ReaperOptions>;
	retention?: Partial<RetentionOptions>;

	/**
	 * Enables the owner reconciliation pass. Absent, no loop is composed and
	 * `reconcile()` returns a no-op summary. Requires {@link now}: the pass's
	 * settle and grace windows are judged on the shared clock.
	 */
	reconciliation?: {
		jobStore: ReconciliationJobStore;
		/** The claimed liveness resolvers, populated by the host before `start`. */
		owners: ScheduledJobOwnerRegistry;
		/** The default timezone is shared with the materializer's. */
		options?: Partial<Omit<ReconciliationOptions, 'defaultTimezone'>>;
	};

	/** Tuning for the start-time clock-skew check (only used when {@link now} is set). */
	clockSkew?: Partial<ClockSkewOptions>;

	/**
	 * Warn through {@link onEvent} when a task fires at least this many seconds after
	 * its scheduled time (a blocked event loop or a skewed clock). Defaults to
	 * {@link DEFAULT_DISPATCH_LAG_WARN_THRESHOLD_SECONDS}.
	 */
	dispatchLagWarnThresholdSeconds?: number;

	/** Cadences of the loops `start` runs, one per pass. */
	lifecycle?: Partial<LifecycleOptions>;

	/**
	 * Reads the current time from the clock the scheduler coordinates on, the same
	 * clock due-ness and leases are judged against. Optional: when given, `start`
	 * samples it once and reports a warning through {@link onEvent} if it differs
	 * from this instance's own clock enough to fire tasks early or late. How the
	 * time is read is the host's concern; left out, the check is skipped.
	 */
	now?: () => Promise<Date>;

	onEvent?: (event: SchedulerEvent) => void;

	/** Host tracer; defaults to a no-op. */
	tracer?: Tracer;

	/** Host metrics; defaults to a no-op. */
	metrics?: SchedulerMetrics;
}

/**
 * A plain `{ ...defaults, ...overrides }` would let an explicitly-undefined
 * override clobber a default (turning e.g. `leaseMs` into `NaN` downstream),
 * so undefined entries are treated as absent.
 */
function withDefaults<T extends object>(defaults: T, overrides: Partial<T> = {}): T {
	const merged = { ...defaults };
	for (const key of Object.keys(overrides) as Array<keyof T>) {
		const value = overrides[key];
		if (value !== undefined) {
			merged[key] = value;
		}
	}
	return merged;
}

/**
 * Compose the core algorithms into a {@link Scheduler}: a registry and an
 * {@link Executor} over `taskStore`, plus the materializer, reaper and
 * retention passes bound to their options, with every incident routed to
 * `onEvent` as a described {@link SchedulerEvent}. `start` runs each pass on
 * its own jittered {@link Loop}; `stop` drains them.
 */
export function createScheduler(deps: SchedulerDeps): Scheduler & SchedulerPasses {
	const { hostId, materializerTransaction, taskStore, onEvent } = deps;
	const tracer = deps.tracer ?? noopTracer;
	const metrics = deps.metrics ?? noopMetrics;
	const materializerOptions = withDefaults(DEFAULT_MATERIALIZER_OPTIONS, deps.materializer);
	const executorOptions = withDefaults(DEFAULT_EXECUTOR_OPTIONS, deps.executor);
	const reaperOptions = withDefaults(DEFAULT_REAPER_OPTIONS, deps.reaper);
	const retentionOptions = withDefaults(DEFAULT_RETENTION_OPTIONS, deps.retention);
	const reconciliationOptions = withDefaults<ReconciliationOptions>(
		DEFAULT_RECONCILIATION_OPTIONS,
		deps.reconciliation?.options,
	);
	const lifecycleOptions = withDefaults(DEFAULT_LIFECYCLE_OPTIONS, deps.lifecycle);
	const clockSkewOptions = withDefaults(DEFAULT_CLOCK_SKEW_OPTIONS, deps.clockSkew);
	const dispatchLagWarnThresholdSeconds =
		deps.dispatchLagWarnThresholdSeconds ?? DEFAULT_DISPATCH_LAG_WARN_THRESHOLD_SECONDS;

	if (!(lifecycleOptions.jitterRatio >= 0 && lifecycleOptions.jitterRatio < 1)) {
		throw new InvalidLifecycleOptionsError(
			`jitterRatio must be at least 0 and below 1, got ${lifecycleOptions.jitterRatio}`,
		);
	}

	// A non-positive or NaN interval collapses to setTimeout's 1ms floor,
	// turning the pass into a hot loop against task storage.
	// A non-positive or NaN timeout would abandon every pass the moment it starts.
	const durationKeys = [
		'materializerIntervalSeconds',
		'executorIntervalSeconds',
		'reaperIntervalSeconds',
		'retentionIntervalSeconds',
		'reconciliationIntervalSeconds',
		'materializerTimeoutSeconds',
		'executorTimeoutSeconds',
		'reaperTimeoutSeconds',
		'retentionTimeoutSeconds',
		'reconciliationTimeoutSeconds',
	] as const;
	for (const key of durationKeys) {
		const value = lifecycleOptions[key];
		if (!(Number.isFinite(value) && value > 0)) {
			throw new InvalidLifecycleOptionsError(
				`${key} must be a positive number of seconds, got ${value}`,
			);
		}
	}

	if (
		lifecycleOptions.concurrencyMode !== 'sequential' &&
		lifecycleOptions.concurrencyMode !== 'concurrent'
	) {
		throw new InvalidLifecycleOptionsError(
			`concurrencyMode must be 'sequential' or 'concurrent', got ${String(lifecycleOptions.concurrencyMode)}`,
		);
	}

	const { maxConcurrentPasses } = lifecycleOptions;
	if (!(Number.isInteger(maxConcurrentPasses) && maxConcurrentPasses >= 1)) {
		throw new InvalidLifecycleOptionsError(
			`maxConcurrentPasses must be a positive integer, got ${maxConcurrentPasses}`,
		);
	}

	for (const key of ['settleSeconds', 'quarantineGraceSeconds'] as const) {
		const value = reconciliationOptions[key];
		if (!(Number.isFinite(value) && value > 0)) {
			throw new InvalidReconciliationOptionsError(
				'Reconciliation windows must be a positive number of seconds',
				key,
				value,
			);
		}
	}

	for (const key of ['batchSize', 'maxPagesPerPass'] as const) {
		const value = reconciliationOptions[key];
		if (!(Number.isInteger(value) && value > 0)) {
			throw new InvalidReconciliationOptionsError(
				'Reconciliation paging options must be a positive integer',
				key,
				value,
			);
		}
	}

	const emit = (level: SchedulerEventLevel, message: string, context: Record<string, unknown>) => {
		// The sink is the reporting channel itself: if it throws (a broken logger),
		// there is nothing left to report through, and the pass that emitted must
		// not be broken by its own observability.
		try {
			onEvent?.({ level, message, context });
		} catch {
			// Deliberately swallowed; see above.
		}
	};
	const described = (error: unknown) => ensureError(error).message;

	// Derived, not caller-chosen: the materializer must record a job's occurrences
	// early enough that the executor still has them in hand when it needs to fire.
	// That spans two gaps: the materializer loop's own worst-case tick gap (so a job
	// due between ticks is claimed, not noticed a tick late), plus the executor's
	// lookahead (so the recorded task row exists before the executor pre-arms its
	// timer). Drop the executor term and a boundary occurrence can land as late as
	// its `nextRunAt`, leaving the executor no slack and firing it up to one executor
	// tick late.
	materializerOptions.lookaheadSeconds =
		pollLookaheadSeconds(
			lifecycleOptions.materializerIntervalSeconds,
			lifecycleOptions.jitterRatio,
		) + executorOptions.lookaheadSeconds;
	if (materializerOptions.lookaheadSeconds > materializerOptions.windowSeconds) {
		// The lookahead is meant to sit inside the window: claim a job a little before
		// its window lapses so the next occurrences are planned ahead. A lookahead past
		// the whole window means a job is re-claimed every poll with nothing new to
		// record, degrading to no-lookahead materialization.
		emit(
			'warn',
			'Scheduler materializer lookahead exceeds the window; jobs may be reclaimed with nothing to plan',
			{
				lookaheadSeconds: materializerOptions.lookaheadSeconds,
				windowSeconds: materializerOptions.windowSeconds,
			},
		);
	}

	// Metrics are best-effort observability: a throwing sink (e.g. a broken
	// exporter) must never break the pass that emitted, so every record is
	// wrapped and its failure swallowed.
	const recordMetric = (record: () => void) => {
		try {
			record();
		} catch {
			// Deliberately swallowed; see above.
		}
	};

	const registry = new TaskHandlerRegistry();
	const executor = new Executor(
		taskStore,
		registry,
		new PrecisionTimer(),
		executorOptions,
		{
			onLeaseShorterThanLookahead: (context) => {
				emit(
					'warn',
					'Scheduler executor lookahead reaches or exceeds the lease; claimed tasks may lose their lease before firing',
					{ ...context },
				);
			},
			onMissingHandler: (task) => {
				emit('warn', 'Scheduler claimed a task with no registered handler; claim released', {
					taskId: task.id,
					taskType: task.taskType,
				});
			},
			onFireError: (task, error) => {
				emit('error', 'Scheduler could not record a task outcome; left for the reaper', {
					taskId: task.id,
					error: described(error),
				});
			},
			onReleaseError: (taskId, error) => {
				emit('error', 'Scheduler failed to release a claimed task; left for the reaper', {
					taskId,
					error: described(error),
				});
			},
			onDispatch: (taskType, lagSeconds) => {
				recordMetric(() => {
					metrics.recordDispatch(taskType);
					metrics.observeDispatchLagSeconds(taskType, lagSeconds);
				});
				if (lagSeconds >= dispatchLagWarnThresholdSeconds) {
					emit('warn', 'Scheduler fired a task later than its scheduled time', {
						taskType,
						lagSeconds: Math.round(lagSeconds),
					});
				}
			},
			onFire: (taskType, result) =>
				recordMetric(() => {
					metrics.recordFireOutcome(taskType, result);
					// An executor terminal failure (attempts exhausted) is a permanent failure = a dead-letter.
					if (result === 'failure') metrics.recordDeadLettered();
				}),
			onRetry: (taskType) => recordMetric(() => metrics.recordRetry(taskType)),
			onLeaseLost: (taskType) => {
				recordMetric(() => metrics.recordLeaseLost(taskType));
				emit(
					'warn',
					'Scheduler task finished after losing its lease; another instance may have run the same occurrence concurrently',
					{ taskType },
				);
			},
		},
		createExecutorTracing(tracer),
	);

	if (retentionOptions.failedRetentionSeconds < retentionOptions.retentionSeconds) {
		emit(
			'warn',
			'Scheduler retention keeps failed runs shorter than succeeded ones; failure evidence will be deleted first',
			{
				retentionSeconds: retentionOptions.retentionSeconds,
				failedRetentionSeconds: retentionOptions.failedRetentionSeconds,
			},
		);
	}

	// Unlike the other passes, the materializer must throw on an observed abort
	// rather than returning early: it is one open transaction, and only a throw
	// out of `runInTransaction`'s callback rolls it back. A timeout is a real
	// fault (the span stays errored, same as any other pass timing out); a plain
	// shutdown abort is expected and rolled back cleanly, so it is reported the
	// same way the other passes report one: a no-op summary, span ok.
	const runMaterialize = tracePass(
		tracer,
		{ name: 'Scheduler materialize', op: 'scheduler.materialize' },
		async (signal) => {
			try {
				const summary = await materialize(
					materializerTransaction,
					materializerOptions,
					{
						onPlanError: (job, error) => {
							emit('error', 'Scheduler could not plan a job schedule; deferred for retry', {
								jobId: job.id,
								error: described(error),
							});
						},
						onSkippedDuplicates: (context) => {
							emit(
								'debug',
								'Scheduler materializer skipped occurrences that were already recorded',
								{ ...context },
							);
						},
					},
					signal,
				);
				await traceCreatedTasks(tracer, summary.created);
				recordMetric(() => {
					metrics.recordMaterialized(summary.occurrences, summary.deferredJobs);
					metrics.recordMisfired(summary.misfires);
					metrics.recordRetired(summary.retiredOccurrences);
				});
				return summary;
			} catch (error) {
				// `throwIfAborted` always throws `signal.reason` itself, so this only
				// swallows the abort: a real storage failure that happens to race a
				// shutdown throws its own error, not `signal.reason`, and still propagates.
				if (
					signal?.aborted === true &&
					signal.reason !== PASS_TIMED_OUT &&
					error === signal.reason
				) {
					return {
						claimedJobs: 0,
						occurrences: 0,
						created: [],
						deferredJobs: 0,
						misfires: [],
						retiredOccurrences: 0,
					};
				}
				throw error;
			}
		},
		(summary) => ({
			[SCHEDULER_ATTRIBUTES.claimedJobs]: summary.claimedJobs,
			[SCHEDULER_ATTRIBUTES.occurrences]: summary.occurrences,
			[SCHEDULER_ATTRIBUTES.deferredJobs]: summary.deferredJobs,
			[SCHEDULER_ATTRIBUTES.skippedOccurrences]: totalDiscarded(summary.misfires),
			[SCHEDULER_ATTRIBUTES.retiredOccurrences]: summary.retiredOccurrences,
		}),
	);

	const runExecute = tracePass(
		tracer,
		{
			name: 'Scheduler claim',
			op: 'scheduler.claim',
			attributes: { [SCHEDULER_ATTRIBUTES.host]: hostId },
		},
		async (signal) => await executor.claimAndSchedule(hostId, signal),
		(tasks) => ({ [SCHEDULER_ATTRIBUTES.claimedCount]: tasks.length }),
	);

	const runReap = tracePass(
		tracer,
		{ name: 'Scheduler reap', op: 'scheduler.reap' },
		async (signal) => {
			const result = await reap(
				taskStore,
				reaperOptions,
				{
					onRowError: (taskId, error) => {
						emit(
							'error',
							'Scheduler could not recover an expired task; skipped until the next sweep',
							{
								taskId,
								error: described(error),
							},
						);
					},
					onRetireError: (error) => {
						emit('error', 'Scheduler could not retire stale pending occurrences', {
							error: described(error),
						});
					},
					onDeadLetter: (task) => {
						emit('warn', 'Scheduler dead-lettered a task; its last attempt lost its lease', {
							...task,
						});
					},
					onCompletedAfterDispatch: ({ taskId }) => {
						emit(
							'warn',
							'Scheduler completed a task whose lease lapsed after dispatch; its effect had already happened',
							{ taskId },
						);
					},
				},
				signal,
			);
			recordMetric(() =>
				metrics.recordReaped(result.reclaimed, result.deadLettered, result.missed),
			);
			return result;
		},
		(result) => ({
			[SCHEDULER_ATTRIBUTES.reclaimed]: result.reclaimed,
			[SCHEDULER_ATTRIBUTES.deadLettered]: result.deadLettered,
			[SCHEDULER_ATTRIBUTES.missed]: result.missed,
		}),
	);

	const runPrune = tracePass(
		tracer,
		{ name: 'Scheduler retention', op: 'scheduler.retention' },
		async (signal) => {
			const summary = await prune(taskStore, retentionOptions, signal);
			if (!summary.drained && signal?.aborted !== true) {
				emit('warn', 'Scheduler retention pass hit its batch budget; backlog remains', {
					...summary,
				});
			} else if (summary.drained && summary.deleted > 0) {
				emit('debug', 'Scheduler retention deleted finished tasks', { ...summary });
			}
			recordMetric(() => metrics.recordPruned(summary.deleted));
			return summary;
		},
		(summary) => ({
			[SCHEDULER_ATTRIBUTES.retentionDeleted]: summary.deleted,
			[SCHEDULER_ATTRIBUTES.retentionDrained]: summary.drained,
		}),
	);

	reconciliationOptions.defaultTimezone = materializerOptions.defaultTimezone;
	const noReconciliation = async (): Promise<ReconciliationSummary> =>
		await Promise.resolve({
			ownersChecked: 0,
			quarantined: 0,
			deleted: 0,
			revived: 0,
			skippedOwnerTypes: [],
			drained: true,
		});
	let runReconcile: (signal?: AbortSignal) => Promise<ReconciliationSummary> = noReconciliation;
	if (deps.reconciliation !== undefined) {
		const { jobStore, owners } = deps.reconciliation;
		let resumeFrom: ReconciliationCursor | undefined;
		// Passes overlap, and each walks from the cursor it read: only the first
		// one back may publish, or a pass that started behind rewinds the walk.
		let published = 0;
		const readNow = deps.now;
		if (readNow === undefined) {
			throw new InvalidSchedulerDepsError(
				'Owner reconciliation requires `now`: its settle and grace windows are judged on the clock every instance shares',
			);
		}
		runReconcile = tracePass(
			tracer,
			{ name: 'Scheduler reconcile owners', op: 'scheduler.reconcile_owners' },
			async (signal) => {
				const startedFrom = resumeFrom;
				const startedAfter = published;
				const summary = await reconcile(
					jobStore,
					owners,
					readNow,
					reconciliationOptions,
					{
						onUnclaimedOwnerType: (ownerType) => {
							emit(
								'warn',
								'Scheduled jobs exist for an owner type with no registered liveness resolver; leaving them alone',
								{ ownerType },
							);
						},
						onResolverFailed: (ownerType, error) => {
							emit(
								'error',
								'A scheduled job owner resolver failed; abandoning that owner type for this pass',
								{ ownerType, error: described(error) },
							);
						},
						onQuarantined: (context) => {
							emit('warn', 'Disabled scheduled jobs whose owner no longer exists', {
								...context,
								deletedAfterSeconds: reconciliationOptions.quarantineGraceSeconds,
							});
						},
						onDeleted: (context) => {
							emit(
								'info',
								'Deleted scheduled jobs whose owner stayed gone past the quarantine grace',
								{ ...context },
							);
						},
						onRevived: (context) => {
							emit('warn', 'Re-enabled scheduled jobs whose owner turned out to still exist', {
								...context,
							});
						},
						onReviveClockFailed: ({ jobId, error }) => {
							emit('warn', 'Could not recompute the clock of a revived scheduled job', {
								jobId,
								error: described(error),
							});
						},
					},
					signal,
					startedFrom,
				);
				if (published === startedAfter) {
					published += 1;
					resumeFrom = summary.resumeFrom;
				}
				if (summary.quarantined > 0 || summary.deleted > 0 || summary.revived > 0) {
					emit('info', 'Scheduled job owner reconciliation changed jobs', { ...summary });
				}
				if (summary.resumeFrom !== undefined && signal?.aborted !== true) {
					emit(
						'warn',
						'Scheduler owner reconciliation pass stopped before draining; backlog remains',
						{ ...summary },
					);
				}
				recordMetric(() =>
					metrics.recordReconciled(summary.quarantined, summary.deleted, summary.revived),
				);
				return summary;
			},
			(summary) => ({
				[SCHEDULER_ATTRIBUTES.ownersChecked]: summary.ownersChecked,
				[SCHEDULER_ATTRIBUTES.quarantinedJobs]: summary.quarantined,
				[SCHEDULER_ATTRIBUTES.deletedOrphanedJobs]: summary.deleted,
				[SCHEDULER_ATTRIBUTES.revivedJobs]: summary.revived,
				[SCHEDULER_ATTRIBUTES.reconciliationDrained]: summary.drained,
			}),
		);
	}

	const loopOver = (
		pass: string,
		run: (signal: AbortSignal) => Promise<unknown>,
		intervalSeconds: number,
		timeoutSeconds: number,
	) =>
		new Loop(
			run,
			{
				intervalMs: intervalSeconds * Time.seconds.toMilliseconds,
				jitterRatio: lifecycleOptions.jitterRatio,
				timeoutMs: timeoutSeconds * Time.seconds.toMilliseconds,
				concurrency: lifecycleOptions.concurrencyMode,
				maxConcurrent: lifecycleOptions.maxConcurrentPasses,
			},
			{
				onError: (error) => {
					emit('error', 'Scheduler pass failed; retrying on its next tick', {
						pass,
						error: described(error),
					});
				},
				onTimeout: ({ timeoutMs }) => {
					emit('error', 'Scheduler pass timed out and was abandoned; retrying on its next tick', {
						pass,
						timeoutMs,
					});
				},
				onSkippedTick: ({ inFlight, limit }) => {
					emit('warn', 'Scheduler pass tick dropped; in-flight passes at the concurrency limit', {
						pass,
						inFlight,
						limit,
					});
				},
			},
		);

	const loops = [
		loopOver(
			'materializer',
			runMaterialize,
			lifecycleOptions.materializerIntervalSeconds,
			lifecycleOptions.materializerTimeoutSeconds,
		),
		loopOver(
			'executor',
			runExecute,
			lifecycleOptions.executorIntervalSeconds,
			lifecycleOptions.executorTimeoutSeconds,
		),
		loopOver(
			'reaper',
			runReap,
			lifecycleOptions.reaperIntervalSeconds,
			lifecycleOptions.reaperTimeoutSeconds,
		),
		loopOver(
			'retention',
			runPrune,
			lifecycleOptions.retentionIntervalSeconds,
			lifecycleOptions.retentionTimeoutSeconds,
		),
	];
	if (deps.reconciliation !== undefined) {
		loops.push(
			loopOver(
				'owner reconciliation',
				runReconcile,
				lifecycleOptions.reconciliationIntervalSeconds,
				lifecycleOptions.reconciliationTimeoutSeconds,
			),
		);
	}

	const checkClockSkew = async () => {
		if (deps.now === undefined) return;
		try {
			const before = Date.now();
			const referenceNow = await deps.now();
			const after = Date.now();
			const skew = measureClockSkew({ before, referenceNow: referenceNow.getTime(), after });
			if (isClockSkewSignificant(skew, clockSkewOptions.warnThresholdMs)) {
				emit(
					'warn',
					'Scheduler detected a clock difference between this instance and the clock it coordinates on; scheduled tasks may fire slightly early or late. Synchronise this instance clock (e.g. via NTP).',
					{ offsetMs: Math.round(skew.offsetMs), roundTripMs: Math.round(skew.roundTripMs) },
				);
			}
		} catch (error) {
			emit('debug', 'Scheduler could not check the clock difference', {
				error: described(error),
			});
		}
	};

	let started = false;
	let stopping: Promise<void> | undefined;

	return {
		registerTaskHandler(taskType, handler) {
			// Wrapped here, at registration, so the executor calls handlers without
			// knowing a tracing span surrounds each run.
			registry.register(taskType, withHandoffTracing(tracer, handler));
		},

		materialize: runMaterialize,

		execute: runExecute,

		reap: runReap,

		prune: runPrune,

		reconcile: runReconcile,

		start() {
			if (!started && stopping === undefined) {
				started = true;
				for (const loop of loops) {
					loop.start();
				}
				emit('info', 'Scheduler started', { hostId });
				// Detached: the clock check reports through the event sink and must
				// never delay the loops that are already running.
				void checkClockSkew();
			}
		},

		async stop() {
			// Cache the teardown so overlapping `stop()` calls await the one run
			// instead of the second returning before the first has finished.
			stopping ??= (async () => {
				// Loops first, each draining its in-flight passes (bounded by their
				// timeouts), so no executor tick overlaps the executor teardown.
				await Promise.all(loops.map(async (loop) => await loop.stop()));
				await executor.stop();
				if (started) {
					emit('info', 'Scheduler stopped', { hostId });
				}
			})();
			await stopping;
		},
	};
}