/* eslint-disable @typescript-eslint/unbound-method */
import type { Logger } from '@n8n/backend-common';
import { ScheduledJobMisfirePolicy } from '@n8n/constants';
import { mockLogger } from '@n8n/backend-test-utils';
import type { GlobalConfig, WorkflowsConfig } from '@n8n/config';
import type { EntityManager } from '@n8n/db';
import type { CronDefinition } from '@n8n/scheduler';
import { ScheduleTrigger } from 'n8n-nodes-base/nodes/Schedule/ScheduleTrigger.node';
import type { Cron, CronExpression, INode, INodeParameters, INodeTypes } from 'n8n-workflow';
import { SCHEDULE_TRIGGER_NODE_TYPE, Workflow } from 'n8n-workflow';
import { mock } from 'vitest-mock-extended';

import type { DurableJobProvisioner } from '../../durable-job-provisioner';
import type { WorkflowScheduledJobOwner } from '../../workflow-scheduled-job-owner';
import { ScheduleTriggerJobRegistrar } from '../schedule-trigger-job-registrar';
import { SCHEDULE_TRIGGER_TASK_TYPE } from '../schedule-trigger-task';

// Monday. Deterministic clock so first-fire assertions are exact.
const NOW = new Date('2026-01-05T00:00:00.000Z');
const NEXT_NINE = new Date('2026-01-05T09:00:00.000Z');

const WORKFLOW_ID = 'wf-1';
const NODE_ID = 'node-1';

/** `<workflowId>:<nodeId>:<definition fingerprint>:<occurrence>` */
const jobNamePattern = new RegExp(`^${WORKFLOW_ID}:${NODE_ID}:[0-9a-f]{16}:\\d+$`);

const workflow = { id: WORKFLOW_ID, settings: {} } as unknown as Workflow;
const scheduleNode = mock<INode>({ id: NODE_ID, type: SCHEDULE_TRIGGER_NODE_TYPE });

const makeNode = ({
	id = NODE_ID,
	typeVersion = 1.4,
	parameters = {},
}: {
	id?: string;
	typeVersion?: number;
	parameters?: INodeParameters;
} = {}): INode =>
	mock<INode>({
		id,
		type: SCHEDULE_TRIGGER_NODE_TYPE,
		typeVersion,
		parameters,
	});

const dailyAtNine: Cron = {
	expression: '0 0 9 * * *' as CronExpression,
	recurrence: { activated: false },
};
const everyThreeWeeksMonday: Cron = {
	expression: '0 0 9 * * 1' as CronExpression,
	recurrence: { activated: true, index: 1, intervalSize: 3, typeInterval: 'weeks' },
};

const ownerOf = (workflowId: string, nodeId: string) => ({
	ownerType: 'workflow',
	ownerId: workflowId,
	ownerMemberId: nodeId,
});
const OWNER = ownerOf(WORKFLOW_ID, NODE_ID);
const OWNER_REF = { ownerType: 'workflow', ownerId: WORKFLOW_ID };

describe('ScheduleTriggerJobRegistrar', () => {
	const jobProvisioner = mock<DurableJobProvisioner>();
	const owner = mock<WorkflowScheduledJobOwner>();

	const lastRequest = () => {
		const lastCall = jobProvisioner.provision.mock.calls.at(-1);
		if (lastCall === undefined) {
			throw new Error('expected provision to have been called');
		}
		return lastCall[0];
	};

	const lastDesired = () => lastRequest().desired;

	const lastProvisionedGrace = () => lastRequest().misfireGraceSeconds;

	const makeRegistrar = ({
		schedulerEnabled = true,
		publicationEnabled = true,
		allowSkipDurableScheduler = false,
	} = {}) =>
		new ScheduleTriggerJobRegistrar(
			mockLogger(),
			mock<GlobalConfig>({
				scheduler: { enabled: schedulerEnabled, allowSkipDurableScheduler },
				generic: { timezone: 'UTC' },
			}),
			mock<WorkflowsConfig>({ useWorkflowPublicationService: publicationEnabled }),
			jobProvisioner,
			owner,
		);

	beforeEach(() => {
		vi.clearAllMocks();
		vi.useFakeTimers();
		vi.setSystemTime(NOW);
		owner.member.mockImplementation(ownerOf);
		owner.ref.mockReturnValue(OWNER_REF);
		jobProvisioner.provision.mockResolvedValue({
			inserted: [],
			redefined: [],
			unchanged: [],
			removed: [],
		});
	});

	afterEach(() => {
		vi.useRealTimers();
	});

	describe('interceptsNode', () => {
		it('intercepts a schedule trigger node when the durable scheduler and publication path are on', () => {
			expect(makeRegistrar().interceptsNode(scheduleNode)).toBe(true);
		});

		it('does not intercept when the durable scheduler is off', () => {
			expect(makeRegistrar({ schedulerEnabled: false }).interceptsNode(scheduleNode)).toBe(false);
		});

		it('does not intercept on the legacy activation path', () => {
			expect(makeRegistrar({ publicationEnabled: false }).interceptsNode(scheduleNode)).toBe(false);
		});

		it('does not intercept other node types', () => {
			const other = mock<INode>({ id: NODE_ID, type: 'n8n-nodes-base.gmailTrigger' });
			expect(makeRegistrar().interceptsNode(other)).toBe(false);
		});

		it('does not intercept a node opted out via skip when the escape hatch is enabled', () => {
			const skippingNode = mock<INode>({
				id: NODE_ID,
				type: SCHEDULE_TRIGGER_NODE_TYPE,
				parameters: { skipDurableScheduler: true },
			});
			expect(makeRegistrar({ allowSkipDurableScheduler: true }).interceptsNode(skippingNode)).toBe(
				false,
			);
		});

		it('still intercepts a skipping node when the escape hatch is disabled', () => {
			const skippingNode = mock<INode>({
				id: NODE_ID,
				type: SCHEDULE_TRIGGER_NODE_TYPE,
				parameters: { skipDurableScheduler: true },
			});
			expect(makeRegistrar({ allowSkipDurableScheduler: false }).interceptsNode(skippingNode)).toBe(
				true,
			);
		});

		it('intercepts a node that has not opted out even when the escape hatch is enabled', () => {
			const keepingNode = mock<INode>({
				id: NODE_ID,
				type: SCHEDULE_TRIGGER_NODE_TYPE,
				parameters: { skipDurableScheduler: false },
			});
			expect(makeRegistrar({ allowSkipDurableScheduler: true }).interceptsNode(keepingNode)).toBe(
				true,
			);
		});
	});

	describe('collect and commit', () => {
		it('provisions one desired job per rule, named by its definition, with the first fire planned', async () => {
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);
			collector.registerCron(dailyAtNine, vi.fn());
			collector.registerCron(everyThreeWeeksMonday, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledWith({
				owner: OWNER,
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				desired: [
					{
						name: expect.stringMatching(jobNamePattern),
						schedule: { kind: 'cron', cronExpression: '0 0 9 * * *', timezone: null },
						firstRunAt: NEXT_NINE,
					},
					{
						name: expect.stringMatching(jobNamePattern),
						schedule: {
							kind: 'recurring_cron',
							cronExpression: '0 0 9 * * 1',
							timezone: null,
							recurrenceUnit: 'weeks',
							recurrenceSize: 3,
						},
						// The first fire is ungated: the very next Monday.
						firstRunAt: NEXT_NINE,
					},
				],
				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
				misfireGraceSeconds: undefined,
			});
		});

		it('provisions a multi-rule node in one call, under the skip policy', async () => {
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);
			collector.registerCron(dailyAtNine, vi.fn());
			collector.registerCron(everyThreeWeeksMonday, vi.fn());
			collector.registerCron(
				{ expression: '0 30 9 * * *' as CronExpression, recurrence: { activated: false } },
				vi.fn(),
			);

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledTimes(1);
			const { desired, misfirePolicy } = lastRequest();
			expect(desired).toHaveLength(3);
			expect(misfirePolicy).toBe(ScheduledJobMisfirePolicy.Skip);
		});

		it('provisions a 5-field custom cron (no seconds) and plans its first fire', async () => {
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);
			collector.registerCron(
				{ expression: '0 9 * * 1-5' as CronExpression, recurrence: { activated: false } },
				vi.fn(),
			);

			await session.commit(WORKFLOW_ID, NODE_ID);

			const desired = lastDesired();
			expect(desired).toHaveLength(1);
			expect(desired[0].schedule).toEqual({
				kind: 'cron',
				cronExpression: '0 9 * * 1-5',
				timezone: null,
			});
			// NOW is Monday 00:00 UTC; a weekday 09:00 cron fires the same day.
			expect(desired[0].firstRunAt).toEqual(NEXT_NINE);
		});

		it('a rule keeps its name when rules are inserted before it or reordered', async () => {
			// A positional name would shift here and needlessly redefine the job,
			// restarting its clock; the definition-derived name must not move.
			const session = makeRegistrar().createSession();
			session.createCollector(workflow, scheduleNode).registerCron(dailyAtNine, vi.fn());
			await session.commit(WORKFLOW_ID, NODE_ID);
			const firstNames = lastDesired().map((job) => job.name);

			const reordered = session.createCollector(workflow, scheduleNode);
			reordered.registerCron(everyThreeWeeksMonday, vi.fn());
			reordered.registerCron(dailyAtNine, vi.fn());
			await session.commit(WORKFLOW_ID, NODE_ID);
			const secondNames = lastDesired().map((job) => job.name);

			// dailyAtNine moved from index 0 to index 1, but its name is unchanged.
			expect(secondNames[1]).toBe(firstNames[0]);
			expect(secondNames[0]).not.toBe(firstNames[0]);
		});

		it('identical duplicate rules get distinct names, stable by occurrence', async () => {
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);
			collector.registerCron(dailyAtNine, vi.fn());
			collector.registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			const desired = lastDesired();
			const [first, second] = desired.map((job) => job.name);
			expect(first).not.toBe(second);
			// Same definition, so same fingerprint: only the occurrence ordinal differs.
			expect(first.replace(/:\d+$/, '')).toBe(second.replace(/:\d+$/, ''));
			expect(first.endsWith(':0')).toBe(true);
			expect(second.endsWith(':1')).toBe(true);
		});

		it("passes the workflow's own timezone, and null for the instance default", async () => {
			const session = makeRegistrar().createSession();
			const zoned = {
				id: WORKFLOW_ID,
				settings: { timezone: 'Europe/Berlin' },
			} as unknown as Workflow;
			const collector = session.createCollector(zoned, scheduleNode);
			collector.registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			const desired = lastDesired();
			expect((desired[0].schedule as CronDefinition).timezone).toBe('Europe/Berlin');
			// 09:00 Berlin (UTC+1 in January).
			expect(desired[0].firstRunAt).toEqual(new Date('2026-01-05T08:00:00.000Z'));
		});

		it("treats the 'DEFAULT' timezone sentinel as the instance default", async () => {
			const session = makeRegistrar().createSession();
			const defaulted = {
				id: WORKFLOW_ID,
				settings: { timezone: 'DEFAULT' },
			} as unknown as Workflow;
			const collector = session.createCollector(defaulted, scheduleNode);
			collector.registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			const desired = lastDesired();
			expect((desired[0].schedule as CronDefinition).timezone).toBeNull();
		});

		it('resolves a null (instance-default) timezone for the first-run math, but stores it as null', async () => {
			// Instance default is not UTC, so a wrong resolution (e.g. falling back
			// to UTC) would show up in firstRunAt rather than being masked by it.
			const session = new ScheduleTriggerJobRegistrar(
				mockLogger(),
				mock<GlobalConfig>({
					scheduler: { enabled: true },
					generic: { timezone: 'Europe/Berlin' },
				}),
				mock<WorkflowsConfig>({ useWorkflowPublicationService: true }),
				jobProvisioner,
				owner,
			).createSession();
			const defaulted = { id: WORKFLOW_ID, settings: {} } as unknown as Workflow;
			const collector = session.createCollector(defaulted, scheduleNode);
			collector.registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			const desired = lastDesired();
			expect((desired[0].schedule as CronDefinition).timezone).toBeNull();
			// 09:00 Berlin (UTC+1 in January) — would be 09:00 UTC if the default
			// weren't resolved.
			expect(desired[0].firstRunAt).toEqual(new Date('2026-01-05T08:00:00.000Z'));
		});

		it('throws synchronously from registerCron on an invalid cron expression', () => {
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);

			expect(() =>
				collector.registerCron({ expression: '99 99 99 * * *' as CronExpression }, vi.fn()),
			).toThrow();
		});

		it('rejects an invalid cron expression even when the recurrence gate makes the rule never fire', () => {
			// The legacy engine parses the expression at registration, before its
			// recurrence check ever runs, so a malformed rule fails activation
			// regardless of the gate; the clock-dead shortcut must not skip that.
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);

			expect(() =>
				collector.registerCron(
					{
						expression: '99 99 99 * * *' as CronExpression,
						recurrence: { activated: true, index: 0, intervalSize: 0, typeInterval: 'weeks' },
					},
					vi.fn(),
				),
			).toThrow();
		});

		it('mirrors a never-firing legacy rule (zero interval) as a job with no next run', async () => {
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);
			collector.registerCron(
				{
					expression: '0 0 9 * * 1' as CronExpression,
					recurrence: { activated: true, index: 0, intervalSize: 0, typeInterval: 'weeks' },
				},
				vi.fn(),
			);

			await session.commit(WORKFLOW_ID, NODE_ID);

			const desired = lastDesired();
			expect(desired[0].firstRunAt).toBeNull();
			expect((desired[0].schedule as CronDefinition).kind).toBe('cron');
		});

		it('provisions a negative-interval rule as a live plain-cron job, matching legacy fire-every-tick', async () => {
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);
			collector.registerCron(
				{
					expression: '0 0 9 * * 1' as CronExpression,
					recurrence: { activated: true, index: 0, intervalSize: -1, typeInterval: 'weeks' },
				},
				vi.fn(),
			);

			await session.commit(WORKFLOW_ID, NODE_ID);

			const desired = lastDesired();
			// Legacy fires a negative-stride rule on every candidate tick, so it must
			// stay a live plain-cron job with a real first run, not a clock-dead row.
			expect((desired[0].schedule as CronDefinition).kind).toBe('cron');
			expect(desired[0].firstRunAt).toEqual(NEXT_NINE);
		});

		it('is a no-op for a node that collected nothing', async () => {
			await makeRegistrar().createSession().commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).not.toHaveBeenCalled();
		});

		it('provisions a node with no rules left as an empty desired set', async () => {
			const session = makeRegistrar().createSession();
			session.createCollector(workflow, scheduleNode);

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledWith({
				owner: OWNER,
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				desired: [],
				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
				misfireGraceSeconds: undefined,
			});
		});

		it.each<[string, INodeParameters | undefined, ScheduledJobMisfirePolicy]>([
			['1.4', undefined, ScheduledJobMisfirePolicy.Skip],
			['1.4', { misfirePolicy: 'coalesce' }, ScheduledJobMisfirePolicy.Coalesce],
			['1.4', { misfirePolicy: 'coalesce_owner' }, ScheduledJobMisfirePolicy.CoalesceOwner],
			['1.4', { misfirePolicy: 'skip' }, ScheduledJobMisfirePolicy.Skip],
			['1.3', undefined, ScheduledJobMisfirePolicy.Skip],
			['1.4', { misfirePolicy: 'nonsense' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
			['1.4', { misfirePolicy: 'wrong' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
			['1.4', { misfirePolicy: '' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
			['1.4', { misfirePolicy: 'Coalesce' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
			['1.4', { misfirePolicy: ' coalesce' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
			['1.4', { misfirePolicy: 'coalesce ' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
			[
				'1.4',
				{ misfirePolicy: 'CoalesceOwner' } as INodeParameters,
				ScheduledJobMisfirePolicy.Skip,
			],
			[
				'1.4',
				{ misfirePolicy: 'coalesce_owner ' } as INodeParameters,
				ScheduledJobMisfirePolicy.Skip,
			],
			['1.4', { misfirePolicy: '__proto__' } as INodeParameters, ScheduledJobMisfirePolicy.Skip],
		])(
			'resolves misfirePolicy %s with parameters %s to %s',
			async (typeVersionLabel, parameters, expected) => {
				const node = makeNode({
					typeVersion: Number(typeVersionLabel),
					parameters: parameters ?? {},
				});
				const session = makeRegistrar().createSession();
				session.createCollector(workflow, node).registerCron(dailyAtNine, vi.fn());

				await session.commit(WORKFLOW_ID, NODE_ID);

				expect(jobProvisioner.provision).toHaveBeenCalledWith({
					owner: OWNER,
					taskType: SCHEDULE_TRIGGER_TASK_TYPE,
					payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
					desired: expect.anything(),
					misfirePolicy: expected,
					misfireGraceSeconds: undefined,
				});
			},
		);

		const buildRealNormalizedNode = (typeVersion: number, parameters: INodeParameters) => {
			const scheduleTriggerNodeTypes = mock<INodeTypes>({
				getByNameAndVersion: () => new ScheduleTrigger(),
			});
			const realWorkflow = new Workflow({
				id: WORKFLOW_ID,
				nodes: [
					{
						id: NODE_ID,
						name: 'Schedule Trigger',
						type: SCHEDULE_TRIGGER_NODE_TYPE,
						typeVersion,
						position: [0, 0],
						parameters,
					},
				],
				connections: {},
				active: false,
				nodeTypes: scheduleTriggerNodeTypes,
			});
			return { workflow: realWorkflow, node: realWorkflow.nodes['Schedule Trigger'] };
		};

		it('resolves a typeVersion 1.3 node to Skip even when its raw JSON carries misfirePolicy, because Workflow normalisation strips a parameter gated to 1.4+', async () => {
			const { workflow: realWorkflow, node: normalizedNode } = buildRealNormalizedNode(1.3, {
				misfirePolicy: 'coalesce',
			});
			expect(normalizedNode.parameters.misfirePolicy).toBeUndefined();

			const session = makeRegistrar().createSession();
			session.createCollector(realWorkflow, normalizedNode).registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledWith({
				owner: OWNER,
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				desired: expect.anything(),
				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
				misfireGraceSeconds: undefined,
			});
		});

		it("keeps a typeVersion 1.4 node's coalesce parameter through Workflow normalisation, since the version gate is satisfied", async () => {
			const { workflow: realWorkflow, node: normalizedNode } = buildRealNormalizedNode(1.4, {
				misfirePolicy: 'coalesce',
			});
			expect(normalizedNode.parameters.misfirePolicy).toBe('coalesce');

			const session = makeRegistrar().createSession();
			session.createCollector(realWorkflow, normalizedNode).registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledWith({
				owner: OWNER,
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				desired: expect.anything(),
				misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
				misfireGraceSeconds: undefined,
			});
		});

		it('resolves misfirePolicy to skip without throwing when the node has no parameters object at all', async () => {
			const node = mock<INode>({
				id: NODE_ID,
				type: SCHEDULE_TRIGGER_NODE_TYPE,
				typeVersion: 1.4,
				parameters: undefined,
			});
			const session = makeRegistrar().createSession();
			session.createCollector(workflow, node).registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledWith({
				owner: OWNER,
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				desired: expect.anything(),
				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
				misfireGraceSeconds: undefined,
			});
		});

		it('provisions a node with no rules left as an empty desired set alongside a coalesce policy', async () => {
			const node = makeNode({ parameters: { misfirePolicy: 'coalesce' } });
			const session = makeRegistrar().createSession();
			session.createCollector(workflow, node);

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledWith({
				owner: OWNER,
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				desired: [],
				misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
				misfireGraceSeconds: undefined,
			});
		});

		it("a discarded session's policy does not leak into the other session's later commit", async () => {
			const registrar = makeRegistrar();
			const attemptA = registrar.createSession();
			const attemptB = registrar.createSession();
			const coalescingNode = makeNode({ parameters: { misfirePolicy: 'coalesce' } });
			const skippingNode = makeNode({ parameters: { misfirePolicy: 'skip' } });

			attemptA.createCollector(workflow, coalescingNode).registerCron(dailyAtNine, vi.fn());
			attemptB.createCollector(workflow, skippingNode).registerCron(dailyAtNine, vi.fn());

			attemptA.discard(WORKFLOW_ID, NODE_ID);
			await attemptA.commit(WORKFLOW_ID, NODE_ID);
			await attemptB.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledTimes(1);
			expect(jobProvisioner.provision).toHaveBeenCalledWith({
				owner: OWNER,
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				desired: expect.anything(),
				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
				misfireGraceSeconds: undefined,
			});
		});

		it('two sessions collecting different policies for the same workflow and node each provision their own resolved policy', async () => {
			const registrar = makeRegistrar();
			const attemptA = registrar.createSession();
			const attemptB = registrar.createSession();
			const coalescingNode = makeNode({ parameters: { misfirePolicy: 'coalesce' } });
			const skippingNode = makeNode({ parameters: { misfirePolicy: 'skip' } });

			attemptA.createCollector(workflow, coalescingNode).registerCron(dailyAtNine, vi.fn());
			attemptB.createCollector(workflow, skippingNode).registerCron(dailyAtNine, vi.fn());

			await attemptA.commit(WORKFLOW_ID, NODE_ID);
			await attemptB.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledTimes(2);
			expect(jobProvisioner.provision).toHaveBeenNthCalledWith(1, {
				owner: ownerOf(WORKFLOW_ID, NODE_ID),
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				desired: expect.anything(),
				misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
				misfireGraceSeconds: undefined,
			});
			expect(jobProvisioner.provision).toHaveBeenNthCalledWith(2, {
				owner: ownerOf(WORKFLOW_ID, NODE_ID),
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				desired: expect.anything(),
				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
				misfireGraceSeconds: undefined,
			});
		});

		it('commits each of two nodes collected in one session with its own distinct misfire policy', async () => {
			const session = makeRegistrar().createSession();
			const coalescingNode = makeNode({
				id: 'node-coalesce',
				parameters: { misfirePolicy: 'coalesce' },
			});
			const skippingNode = makeNode({ id: 'node-skip', parameters: { misfirePolicy: 'skip' } });

			session.createCollector(workflow, coalescingNode).registerCron(dailyAtNine, vi.fn());
			session.createCollector(workflow, skippingNode).registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, 'node-coalesce');
			await session.commit(WORKFLOW_ID, 'node-skip');

			expect(jobProvisioner.provision).toHaveBeenNthCalledWith(1, {
				owner: ownerOf(WORKFLOW_ID, 'node-coalesce'),
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: 'node-coalesce' },
				desired: expect.anything(),
				misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
				misfireGraceSeconds: undefined,
			});
			expect(jobProvisioner.provision).toHaveBeenNthCalledWith(2, {
				owner: ownerOf(WORKFLOW_ID, 'node-skip'),
				taskType: SCHEDULE_TRIGGER_TASK_TYPE,
				payload: { workflowId: WORKFLOW_ID, nodeId: 'node-skip' },
				desired: expect.anything(),
				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
				misfireGraceSeconds: undefined,
			});
		});

		it('provisions a positive misfire grace parameter as the grace of the node', async () => {
			const node = makeNode({ parameters: { misfireGraceSeconds: 90 } });
			const session = makeRegistrar().createSession();
			session.createCollector(workflow, node).registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(lastProvisionedGrace()).toBe(90);
		});

		it('provisions no misfire grace when the node carries no grace parameter, leaving the instance value to apply', async () => {
			const node = makeNode();
			const session = makeRegistrar().createSession();
			session.createCollector(workflow, node).registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(lastProvisionedGrace()).toBeUndefined();
		});

		it.each<[string, INodeParameters]>([
			['a stored 0', { misfireGraceSeconds: 0 }],
			['a stored "0" string', { misfireGraceSeconds: '0' }],
			[
				'a stored null, as empty as an absent parameter',
				{ misfireGraceSeconds: null } as unknown as INodeParameters,
			],
			['a stored false, as empty as an absent parameter', { misfireGraceSeconds: false }],
		])(
			'provisions no misfire grace for %s, which stands for the instance value',
			async (_label, parameters) => {
				const node = makeNode({ parameters });
				const session = makeRegistrar().createSession();
				session.createCollector(workflow, node).registerCron(dailyAtNine, vi.fn());

				await session.commit(WORKFLOW_ID, NODE_ID);

				expect(lastProvisionedGrace()).toBeUndefined();
			},
		);

		it.each<[string, INodeParameters]>([
			['a negative number', { misfireGraceSeconds: -30 }],
			['NaN', { misfireGraceSeconds: Number.NaN }],
			['Infinity', { misfireGraceSeconds: Number.POSITIVE_INFINITY }],
			['a non-numeric string', { misfireGraceSeconds: 'nonsense' }],
			['an empty string', { misfireGraceSeconds: '' }],
			[
				'a boolean true, which is not a number the author could have typed',
				{
					misfireGraceSeconds: true,
				},
			],
			['a value below the second the provisioner would read', { misfireGraceSeconds: 0.5 }],
		])(
			'provisions no misfire grace for %s, rather than failing the activation',
			async (_label, parameters) => {
				const node = makeNode({ parameters });
				const session = makeRegistrar().createSession();
				session.createCollector(workflow, node).registerCron(dailyAtNine, vi.fn());

				await session.commit(WORKFLOW_ID, NODE_ID);

				expect(lastProvisionedGrace()).toBeUndefined();
			},
		);

		it.each<[string, INodeParameters, number]>([
			['a numeric string', { misfireGraceSeconds: '90' }, 90],
			['a fractional number', { misfireGraceSeconds: 90.5 }, 90.5],
			['exactly the one second the provisioner reads as stated', { misfireGraceSeconds: 1 }, 1],
		])(
			'provisions %s as a stated grace of %s, leaving truncation and clamping to the provisioner',
			async (_label, parameters, expected) => {
				const node = makeNode({ parameters });
				const session = makeRegistrar().createSession();
				session.createCollector(workflow, node).registerCron(dailyAtNine, vi.fn());

				await session.commit(WORKFLOW_ID, NODE_ID);

				expect(lastProvisionedGrace()).toBe(expected);
			},
		);

		describe('unusable misfire grace warning', () => {
			const makeRegistrarWatchingWarnings = () => {
				const scopedLogger = mockLogger();
				const registrar = new ScheduleTriggerJobRegistrar(
					mock<Logger>({ scoped: vi.fn().mockReturnValue(scopedLogger) }),
					mock<GlobalConfig>({
						scheduler: { enabled: true, allowSkipDurableScheduler: false },
						generic: { timezone: 'UTC' },
					}),
					mock<WorkflowsConfig>({ useWorkflowPublicationService: true }),
					jobProvisioner,
					owner,
				);
				return { registrar, scopedLogger };
			};

			const plainNode = (parameters: INodeParameters): INode => ({
				id: NODE_ID,
				name: 'Schedule Trigger',
				type: SCHEDULE_TRIGGER_NODE_TYPE,
				typeVersion: 1.4,
				position: [0, 0],
				parameters,
			});

			it.each<[string, INodeParameters]>([
				['a non-numeric string', { misfireGraceSeconds: 'nonsense' }],
				['a blank string, which coerces to zero but states nothing', { misfireGraceSeconds: ' ' }],
				['a boolean', { misfireGraceSeconds: true }],
				['a value the provisioner would drop as below a second', { misfireGraceSeconds: 0.5 }],
			])('warns naming the workflow and node for %s', (_label, parameters) => {
				const { registrar, scopedLogger } = makeRegistrarWatchingWarnings();
				const node = plainNode(parameters);

				registrar.createSession().createCollector(workflow, node);

				expect(scopedLogger.warn).toHaveBeenCalledWith(
					'Schedule trigger node has an unusable misfire grace period; the instance setting applies',
					{ workflowId: WORKFLOW_ID, nodeId: NODE_ID },
				);
			});

			it.each<[string, INodeParameters]>([
				['no misfire grace parameter', {}],
				['a stored 0, which stands for the instance value', { misfireGraceSeconds: 0 }],
				['a stored "0" string, which stands for it just as well', { misfireGraceSeconds: '0' }],
				[
					'a stored null, as empty as an absent parameter',
					{ misfireGraceSeconds: null } as unknown as INodeParameters,
				],
				['a stored false, as empty as an absent parameter', { misfireGraceSeconds: false }],
				['a usable misfire grace', { misfireGraceSeconds: 90 }],
			])('does not warn for a node with %s', (_label, parameters) => {
				const { registrar, scopedLogger } = makeRegistrarWatchingWarnings();

				registrar.createSession().createCollector(workflow, plainNode(parameters));

				expect(scopedLogger.warn).not.toHaveBeenCalled();
			});
		});

		it('provisions no misfire grace for a typeVersion 1.3 node whose raw JSON carries one, because Workflow normalisation strips a parameter gated to 1.4+', async () => {
			const { workflow: realWorkflow, node: normalizedNode } = buildRealNormalizedNode(1.3, {
				misfireGraceSeconds: 90,
			});
			expect(normalizedNode.parameters.misfireGraceSeconds).toBeUndefined();

			const session = makeRegistrar().createSession();
			session.createCollector(realWorkflow, normalizedNode).registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(lastProvisionedGrace()).toBeUndefined();
		});

		it("keeps a typeVersion 1.4 node's misfire grace parameter through Workflow normalisation, since the version gate is satisfied", async () => {
			const { workflow: realWorkflow, node: normalizedNode } = buildRealNormalizedNode(1.4, {
				misfireGraceSeconds: 90,
			});
			expect(normalizedNode.parameters.misfireGraceSeconds).toBe(90);

			const session = makeRegistrar().createSession();
			session.createCollector(realWorkflow, normalizedNode).registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(lastProvisionedGrace()).toBe(90);
		});

		it('consumes the collected rules: a second commit is a no-op', async () => {
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);
			collector.registerCron(dailyAtNine, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);
			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).toHaveBeenCalledTimes(1);
		});

		it('discard drops collected rules so a failed activation persists nothing', async () => {
			const session = makeRegistrar().createSession();
			const collector = session.createCollector(workflow, scheduleNode);
			collector.registerCron(dailyAtNine, vi.fn());

			session.discard(WORKFLOW_ID, NODE_ID);
			await session.commit(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.provision).not.toHaveBeenCalled();
		});

		it('a fresh collector replaces the rules of a previous failed attempt', async () => {
			const session = makeRegistrar().createSession();
			session.createCollector(workflow, scheduleNode).registerCron(dailyAtNine, vi.fn());
			// The retry collects anew; only its rules must persist.
			const retry = session.createCollector(workflow, scheduleNode);
			retry.registerCron(everyThreeWeeksMonday, vi.fn());

			await session.commit(WORKFLOW_ID, NODE_ID);

			const desired = lastDesired();
			expect(desired).toHaveLength(1);
			expect((desired[0].schedule as CronDefinition).cronExpression).toBe('0 0 9 * * 1');
		});

		it('interleaved activation attempts each commit only their own rules', async () => {
			// Two attempts for the same workflow and node can interleave on the
			// legacy path, which has no lifecycle lock. Each attempt's session owns
			// its rules, so neither commit can consume the other's collection.
			const registrar = makeRegistrar();
			const attemptA = registrar.createSession();
			const attemptB = registrar.createSession();
			attemptA.createCollector(workflow, scheduleNode).registerCron(dailyAtNine, vi.fn());
			attemptB.createCollector(workflow, scheduleNode).registerCron(everyThreeWeeksMonday, vi.fn());

			await attemptA.commit(WORKFLOW_ID, NODE_ID);
			const fromA = lastDesired();
			expect(fromA).toHaveLength(1);
			expect((fromA[0].schedule as CronDefinition).cronExpression).toBe('0 0 9 * * *');

			await attemptB.commit(WORKFLOW_ID, NODE_ID);
			const fromB = lastDesired();
			expect(fromB).toHaveLength(1);
			expect((fromB[0].schedule as CronDefinition).cronExpression).toBe('0 0 9 * * 1');
		});
	});

	describe('remove', () => {
		it('removes the durable jobs of a deactivated node', async () => {
			await makeRegistrar().remove(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.deprovisionOwnerMember).toHaveBeenCalledWith(OWNER);
		});

		it('removes durable jobs left by an earlier activation even while the scheduler is off', async () => {
			// A prior activation may have persisted rows while the durable engine was
			// on; deactivation must clear them so they cannot re-fire on re-enable.
			await makeRegistrar({ schedulerEnabled: false }).remove(WORKFLOW_ID, NODE_ID);

			expect(jobProvisioner.deprovisionOwnerMember).toHaveBeenCalledWith(OWNER);
		});
	});

	describe('removeWorkflow', () => {
		it('removes the durable jobs of all schedule nodes of a deactivated workflow', async () => {
			await makeRegistrar().removeWorkflow(WORKFLOW_ID);

			expect(jobProvisioner.deprovisionOwnerTaskType).toHaveBeenCalledWith(
				OWNER_REF,
				SCHEDULE_TRIGGER_TASK_TYPE,
			);
		});

		it('removes durable jobs left by an earlier activation even while the scheduler is off', async () => {
			await makeRegistrar({ schedulerEnabled: false }).removeWorkflow(WORKFLOW_ID);

			expect(jobProvisioner.deprovisionOwnerTaskType).toHaveBeenCalledWith(
				OWNER_REF,
				SCHEDULE_TRIGGER_TASK_TYPE,
			);
		});
	});

	describe('removeWorkflowInTransaction', () => {
		it("removes the workflow's durable jobs through the caller's transaction", async () => {
			const manager = mock<EntityManager>();

			await makeRegistrar().removeWorkflowInTransaction(manager, WORKFLOW_ID);

			expect(jobProvisioner.deprovisionOwnerTaskTypeInTransaction).toHaveBeenCalledWith(
				manager,
				OWNER_REF,
				SCHEDULE_TRIGGER_TASK_TYPE,
			);
		});
	});

	describe('configuration warning', () => {
		// The registrar scopes its logger, so warnings land on the scoped instance.
		const construct = ({ schedulerEnabled = true, publicationEnabled = true } = {}) => {
			const scopedLogger = mockLogger();
			const logger = mock<Logger>({ scoped: vi.fn().mockReturnValue(scopedLogger) });
			new ScheduleTriggerJobRegistrar(
				logger,
				mock<GlobalConfig>({
					scheduler: { enabled: schedulerEnabled },
					generic: { timezone: 'UTC' },
				}),
				mock<WorkflowsConfig>({ useWorkflowPublicationService: publicationEnabled }),
				jobProvisioner,
				owner,
			);
			return scopedLogger;
		};

		it('warns when the durable scheduler is enabled but the publication service is off', () => {
			const logger = construct({ schedulerEnabled: true, publicationEnabled: false });

			expect(logger.warn).toHaveBeenCalledWith(
				expect.stringContaining('workflow publication service is disabled'),
			);
		});

		it('does not warn when both the durable scheduler and the publication service are on', () => {
			const logger = construct({ schedulerEnabled: true, publicationEnabled: true });

			expect(logger.warn).not.toHaveBeenCalled();
		});

		it('does not warn when the durable scheduler is off', () => {
			const logger = construct({ schedulerEnabled: false, publicationEnabled: false });

			expect(logger.warn).not.toHaveBeenCalled();
		});
	});
});