/* eslint-disable @typescript-eslint/unbound-method */
import type { Logger } from '@n8n/backend-common';
import { mockLogger } from '@n8n/backend-test-utils';
import type { GlobalConfig, WorkflowsConfig } from '@n8n/config';
import type { Project, WorkflowEntity, WorkflowHistory, WorkflowRepository } from '@n8n/db';
import type { UpdateResult } from '@n8n/typeorm';
import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
import { sleep } from '@n8n/utils/sleep';
import type { ErrorReporter, InstanceSettings } from 'n8n-core';
import {
	ActiveWorkflowTriggers,
	PollTriggerExecutor,
	ScheduledTaskManager,
	Tracing,
} from 'n8n-core';
import type {
	CronExpression,
	ExecutionError,
	INodeExecutionData,
	INode,
	INodeType,
	INodeTypes,
	IPollFunctions,
	IRun,
	IWorkflowBase,
	IWorkflowExecuteAdditionalData,
	WorkflowActivateMode,
	WorkflowExecuteMode,
} from 'n8n-workflow';
import { Workflow, WorkflowActivationError } from 'n8n-workflow';
import { mock, type MockProxy } from 'vitest-mock-extended';

import type { ActivationErrorsService } from '@/activation-errors.service';
import { ActiveWorkflowManager } from '@/active-workflow-manager';
import { DuplicateExecutionError } from '@/errors/duplicate-execution.error';
import type { EventService } from '@/events/event.service';
import type { ExecutionService } from '@/executions/execution.service';
import type { NodeTypes } from '@/node-types';
import type { Push } from '@/push';
import type { Publisher } from '@/scaling/pubsub/publisher.service';
import type { PollTriggerJobRegistrar } from '@/scheduling/poll-trigger-node/poll-trigger-job-registrar';
import type {
	ScheduleTriggerCollectionSession,
	ScheduleTriggerJobRegistrar,
} from '@/scheduling/schedule-trigger-node/schedule-trigger-job-registrar';
import type { PolicyEnforcementService } from '@/policy/policy-enforcement.service';
import { PolicyViolationError } from '@/policy/policy-violation.error';
import type { OwnershipService } from '@/services/ownership.service';
import type { PollCursorService } from '@/workflows/triggers/poll-cursor.service';
import { TriggerExecutionContextFactory } from '@/workflows/triggers/trigger-execution-context.factory';
import type { WorkflowExecutionService } from '@/workflows/workflow-execution.service';
import { WorkflowPushNotifier } from '@/workflows/workflow-push-notifier.service';
import type { WorkflowSharingService } from '@/workflows/workflow-sharing.service';
import type { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';

describe('ActiveWorkflowManager', () => {
	const WORKFLOW_SCHEDULE_GROUP_TYPE = 'workflow';
	const workflowGroup = (id: string) => ({ type: WORKFLOW_SCHEDULE_GROUP_TYPE, id });
	let activeWorkflowManager: ActiveWorkflowManager;
	const instanceSettings = mock<InstanceSettings>({ isMultiMain: false });
	const nodeTypes = mock<NodeTypes>();
	const workflowRepository = mock<WorkflowRepository>();
	const workflowsConfig = mock<WorkflowsConfig>({ useWorkflowPublicationService: false });

	// Shared by every construction below; clears by default, like an instance with
	// no policy backend.
	const policyEnforcementService = mock<PolicyEnforcementService>();
	const ownershipService = mock<OwnershipService>();

	beforeEach(() => {
		vi.clearAllMocks();
		// `clearAllMocks` keeps implementations, so restore the clearing defaults.
		ownershipService.getWorkflowProjectCached.mockResolvedValue(mock<Project>({ id: 'project-1' }));
		policyEnforcementService.hasChecksFor.mockReturnValue(true);
		policyEnforcementService.enforceWorkflowPublish.mockResolvedValue(mock());
		activeWorkflowManager = new ActiveWorkflowManager(
			mockLogger(),
			mock(),
			mock(),
			mock(),
			nodeTypes,
			mock(),
			workflowRepository,
			mock(),
			mock(),
			mock(),
			instanceSettings,
			mock(),
			workflowsConfig,
			mock<TriggerExecutionContextFactory>(),
			mock(),
			mock(), // scheduleTriggerJobRegistrar
			mock(), // pollTriggerJobRegistrar
			mock(), // workflowPushNotifier
			policyEnforcementService,
			ownershipService,
		);
	});

	describe('shouldAddWebhooks', () => {
		describe('if leader', () => {
			beforeAll(() => {
				Object.assign(instanceSettings, { isLeader: true, isFollower: false });
			});

			test('should return `true` for `init`', () => {
				// ensure webhooks are populated on init: https://github.com/n8n-io/n8n/pull/8830
				const result = activeWorkflowManager.shouldAddWebhooks('init');
				expect(result).toBe(true);
			});

			test('should return `true` for `leadershipChange`', () => {
				const result = activeWorkflowManager.shouldAddWebhooks('leadershipChange');
				expect(result).toBe(true);
			});

			test('should return `true` for `update` or `activate`', () => {
				const modes = ['update', 'activate'] as WorkflowActivateMode[];
				for (const mode of modes) {
					const result = activeWorkflowManager.shouldAddWebhooks(mode);
					expect(result).toBe(true);
				}
			});
		});

		describe('if follower', () => {
			beforeAll(() => {
				Object.assign(instanceSettings, { isLeader: false, isFollower: true });
			});

			test('should return `false` for `update` or `activate`', () => {
				const modes = ['update', 'activate'] as WorkflowActivateMode[];
				for (const mode of modes) {
					const result = activeWorkflowManager.shouldAddWebhooks(mode);
					expect(result).toBe(false);
				}
			});
		});

		describe('add', () => {
			test.each<[WorkflowActivateMode]>([['init'], ['leadershipChange']])(
				'should skip inactive workflow in `%s` activation mode',
				async (mode) => {
					const addWebhooksSpy = vi.spyOn(activeWorkflowManager, 'addWebhooks');
					const addNonWebhookTriggersSpy = vi.spyOn(activeWorkflowManager, 'addNonWebhookTriggers');
					workflowRepository.findById.mockResolvedValue(
						mock<WorkflowEntity>({ active: false, activeVersionId: null, activeVersion: null }),
					);

					const added = await activeWorkflowManager.add('some-id', mode);

					expect(addWebhooksSpy).not.toHaveBeenCalled();
					expect(addNonWebhookTriggersSpy).not.toHaveBeenCalled();
					expect(added).toEqual({ triggersAndPollers: false, webhooks: false });
				},
			);

			test.each<[WorkflowActivateMode]>([['init'], ['leadershipChange'], ['activate']])(
				'should skip archived workflow in `%s` activation mode',
				async (mode) => {
					const addWebhooksSpy = vi.spyOn(activeWorkflowManager, 'addWebhooks');
					const addNonWebhookTriggersSpy = vi.spyOn(activeWorkflowManager, 'addNonWebhookTriggers');
					workflowRepository.findById.mockResolvedValue(
						mock<WorkflowEntity>({
							id: 'archived-id',
							active: true,
							activeVersionId: 'v1',
							isArchived: true,
						}),
					);

					const added = await activeWorkflowManager.add('archived-id', mode);

					expect(addWebhooksSpy).not.toHaveBeenCalled();
					expect(addNonWebhookTriggersSpy).not.toHaveBeenCalled();
					expect(added).toEqual({ triggersAndPollers: false, webhooks: false });
				},
			);
		});
	});

	describe('addActiveWorkflows', () => {
		test('should prevent concurrent activations', async () => {
			const getAllActiveIds = workflowRepository.getAllActiveIds;

			workflowRepository.getAllActiveIds.mockImplementation(
				async () => await new Promise((resolve) => setTimeout(() => resolve(['workflow-1']), 50)),
			);

			await Promise.all([
				activeWorkflowManager.addActiveWorkflows('init'),
				activeWorkflowManager.addActiveWorkflows('leadershipChange'),
			]);

			expect(getAllActiveIds).toHaveBeenCalledTimes(1);
		});
	});

	describe('policy enforcement', () => {
		// `chunk` coerces an unset `activationBatchSize` to 0 and yields no batches, so
		// the startup loop would silently never run.
		const policyWorkflowsConfig = mock<WorkflowsConfig>({
			useWorkflowPublicationService: false,
			activationBatchSize: 1,
		});
		const activationErrorsService = mock<ActivationErrorsService>();
		const activeWorkflowTriggers = mock<ActiveWorkflowTriggers>();
		const errorReporter = mock<ErrorReporter>();

		const VERSION_NODES = [{ id: 'node-1', name: 'Active trigger' } as INode];
		const DRAFT_NODES = [{ id: 'node-2', name: 'Draft node' } as INode];

		const makeManager = () =>
			new ActiveWorkflowManager(
				mockLogger(),
				errorReporter,
				activeWorkflowTriggers,
				mock(),
				nodeTypes,
				mock(),
				workflowRepository,
				activationErrorsService,
				mock(),
				mock(),
				instanceSettings,
				mock(),
				policyWorkflowsConfig,
				mock<TriggerExecutionContextFactory>(),
				mock(),
				mock(),
				mock(), // scheduleTriggerJobRegistrar
				mock(), // pollTriggerJobRegistrar
				policyEnforcementService,
				ownershipService,
			);

		const makeWorkflow = (overrides: Partial<WorkflowEntity> = {}) =>
			mock<WorkflowEntity>({
				id: 'wf-1',
				name: 'My workflow',
				active: true,
				isArchived: false,
				activeVersionId: 'v1',
				// Differs from the published version on purpose: only the latter runs.
				nodes: DRAFT_NODES,
				activeVersion: mock<WorkflowHistory>({
					versionId: 'v1',
					nodes: VERSION_NODES,
					connections: {},
				}),
				...overrides,
			});

		beforeEach(() => {
			Object.assign(instanceSettings, { isLeader: true, isFollower: false });
			activeWorkflowManager = makeManager();
		});

		test('enforces with the published version nodes, not the draft', async () => {
			workflowRepository.findById.mockResolvedValue(makeWorkflow());

			// Registration fails here (no real node types); the check runs before it.
			await activeWorkflowManager.add('wf-1', 'activate').catch(() => {});

			expect(policyEnforcementService.enforceWorkflowPublish).toHaveBeenCalledExactlyOnceWith({
				workflow: { id: 'wf-1', name: 'My workflow', nodes: VERSION_NODES },
				projectId: 'project-1',
			});
		});

		test('registers nothing and records an activation error when policy blocks', async () => {
			workflowRepository.findById.mockResolvedValue(makeWorkflow());
			const addWebhooksSpy = vi.spyOn(activeWorkflowManager, 'addWebhooks');
			const addNonWebhookTriggersSpy = vi.spyOn(activeWorkflowManager, 'addNonWebhookTriggers');
			policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(
				new PolicyViolationError([
					{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
				]),
			);

			await expect(activeWorkflowManager.add('wf-1', 'activate')).rejects.toBeInstanceOf(
				PolicyViolationError,
			);

			expect(addWebhooksSpy).not.toHaveBeenCalled();
			expect(addNonWebhookTriggersSpy).not.toHaveBeenCalled();
			expect(activationErrorsService.register).toHaveBeenCalledWith('wf-1', 'Blocked by policy');
		});

		test('does not enforce for a workflow that is no longer active', async () => {
			workflowRepository.findById.mockResolvedValue(
				makeWorkflow({ active: false, activeVersionId: null, activeVersion: null }),
			);

			await activeWorkflowManager.add('wf-1', 'init');

			expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
		});

		// An unevaluated project rule is not a passed one, so the lookup is unguarded.
		test('propagates a failed ownership lookup instead of policing a null scope', async () => {
			workflowRepository.findById.mockResolvedValue(makeWorkflow());
			ownershipService.getWorkflowProjectCached.mockRejectedValue(new Error('no owner row'));

			await expect(activeWorkflowManager.add('wf-1', 'activate')).rejects.toThrow('no owner row');

			expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
		});

		// A feature that is merely absent must not cost a lookup on every activation.
		test('does not resolve ownership when no check is registered', async () => {
			policyEnforcementService.hasChecksFor.mockReturnValue(false);
			workflowRepository.findById.mockResolvedValue(makeWorkflow());

			await activeWorkflowManager.add('wf-1', 'activate').catch(() => {});

			expect(ownershipService.getWorkflowProjectCached).not.toHaveBeenCalled();
			expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
		});

		test('does not enforce for an archived workflow', async () => {
			workflowRepository.findById.mockResolvedValue(makeWorkflow({ isArchived: true }));

			await activeWorkflowManager.add('wf-1', 'activate');

			expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
		});

		// The forward is unconditional on multi-main, not leader-specific; the leader
		// enforces when it handles the pubsub command with `shouldPublish: false`.
		test('does not enforce when a multi-main instance forwards the activation', async () => {
			Object.assign(instanceSettings, { isMultiMain: true });
			workflowRepository.findById.mockResolvedValue(makeWorkflow());

			try {
				await activeWorkflowManager.add('wf-1', 'activate');

				expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
			} finally {
				Object.assign(instanceSettings, { isMultiMain: false });
			}
		});

		// Already queued for an unrelated transient failure: if policy then blocks it,
		// the retry must drop it rather than reschedule forever.
		test('drops a queued retry when policy blocks on the retry attempt', async () => {
			vi.useFakeTimers();
			try {
				const dbWorkflow = makeWorkflow();
				workflowRepository.findById.mockResolvedValue(dbWorkflow);
				const manager = activeWorkflowManager as unknown as {
					addQueuedWorkflowActivation: (
						mode: WorkflowActivateMode,
						workflow: WorkflowEntity,
					) => void;
					queuedActivations: Record<string, unknown>;
				};

				manager.addQueuedWorkflowActivation('update', dbWorkflow);
				expect(manager.queuedActivations['wf-1']).toBeDefined();

				policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(
					new PolicyViolationError([
						{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
					]),
				);

				await vi.runOnlyPendingTimersAsync();

				expect(manager.queuedActivations['wf-1']).toBeUndefined();
				// Nor a fault report: the refusal is expected and permanent.
				expect(errorReporter.error).not.toHaveBeenCalled();
			} finally {
				vi.useRealTimers();
			}
		});

		// A refusal, not a failure: the owner's error automation must not fire.
		test('does not run the error workflow when policy blocks at startup', async () => {
			workflowRepository.getAllActiveIds.mockResolvedValue(['wf-1']);
			workflowRepository.findById.mockResolvedValue(makeWorkflow());
			policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(
				new PolicyViolationError([
					{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
				]),
			);
			const errorWorkflowSpy = vi.spyOn(activeWorkflowManager, 'executeErrorWorkflow');

			await activeWorkflowManager.addActiveWorkflows('init');

			expect(errorWorkflowSpy).not.toHaveBeenCalled();
			// Nor a fault report, or every restart alerts on an expected refusal.
			expect(errorReporter.error).not.toHaveBeenCalled();
		});

		// A block is permanent, so the indefinite retry must not pick it up.
		test('does not queue a startup retry when policy blocks', async () => {
			workflowRepository.getAllActiveIds.mockResolvedValue(['wf-1']);
			workflowRepository.findById.mockResolvedValue(makeWorkflow());
			policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(
				new PolicyViolationError([
					{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
				]),
			);
			const queueSpy = vi.spyOn(
				activeWorkflowManager as unknown as { addQueuedWorkflowActivation: () => void },
				'addQueuedWorkflowActivation',
			);

			await activeWorkflowManager.addActiveWorkflows('init');

			expect(queueSpy).not.toHaveBeenCalled();
		});
	});

	describe('handleAddWebhooksAndNonWebhookTriggers', () => {
		const push = mock<Push>();
		const publisher = mock<Publisher>();
		const workflowSharingService = mock<WorkflowSharingService>();
		const workflowPushNotifier = new WorkflowPushNotifier(push, workflowSharingService);
		const sharedUserIds = ['user-1', 'user-2'];

		beforeEach(() => {
			workflowSharingService.getUserIdsWithAccessToWorkflowSafe.mockResolvedValue(sharedUserIds);
			publisher.publishCommand.mockResolvedValue(undefined);
			activeWorkflowManager = new ActiveWorkflowManager(
				mockLogger(),
				mock(),
				mock(),
				mock(),
				nodeTypes,
				mock(),
				workflowRepository,
				mock(),
				mock(),
				mock(),
				instanceSettings,
				publisher,
				mock(),
				mock<TriggerExecutionContextFactory>(),
				mock(),
				mock(), // scheduleTriggerJobRegistrar
				mock(), // pollTriggerJobRegistrar
				workflowPushNotifier,
				policyEnforcementService,
				ownershipService,
			);
		});

		// The leader's generic failure path clears `activeVersionId`. A refusal must not
		// unpublish a workflow — that decision is not this path's to make.
		test('does not unpublish the workflow when policy blocks the forwarded activation', async () => {
			const violation = new PolicyViolationError([
				{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
			]);
			vi.spyOn(activeWorkflowManager, 'add').mockRejectedValue(violation);
			const clearWebhooksSpy = vi.spyOn(activeWorkflowManager, 'clearWebhooks');
			const removeTriggersSpy = vi.spyOn(activeWorkflowManager, 'removeNonWebhookTriggers');

			await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
				workflowId: 'wf-1',
				activeVersionId: 'v1',
				activationMode: 'init',
			});

			expect(workflowRepository.update).not.toHaveBeenCalled();
			expect(clearWebhooksSpy).not.toHaveBeenCalled();
			expect(removeTriggersSpy).not.toHaveBeenCalled();
			// The user is still told why it did not start.
			expect(push.sendToUsers.mock.calls.map(([msg]) => msg.type)).toContain(
				'workflowFailedToActivate',
			);
		});

		test('should include nodeId in broadcast when error has node', async () => {
			const triggerNode = mock<INode>({ id: 'node-123', name: 'Linear Trigger' });
			const activationError = new WorkflowActivationError('Invalid role: admin required', {
				node: triggerNode,
			});

			vi.spyOn(activeWorkflowManager, 'add').mockRejectedValue(activationError);

			await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
				workflowId: 'wf-1',
				activeVersionId: 'v1',
				activationMode: 'activate',
			});

			expect(workflowSharingService.getUserIdsWithAccessToWorkflowSafe).toHaveBeenCalledWith(
				'wf-1',
			);
			expect(push.sendToUsers).toHaveBeenCalledWith(
				{
					type: 'workflowFailedToActivate',
					data: {
						workflowId: 'wf-1',
						errorMessage: 'Invalid role: admin required',
						nodeId: 'node-123',
					},
				},
				sharedUserIds,
			);

			expect(publisher.publishCommand).toHaveBeenCalledWith({
				command: 'display-workflow-activation-error',
				payload: {
					workflowId: 'wf-1',
					errorMessage: 'Invalid role: admin required',
					nodeId: 'node-123',
				},
			});
		});

		test('should not include nodeId in broadcast when error has no node', async () => {
			vi.spyOn(activeWorkflowManager, 'add').mockRejectedValue(new Error('Some error'));

			await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
				workflowId: 'wf-1',
				activeVersionId: 'v1',
				activationMode: 'activate',
			});

			expect(push.sendToUsers).toHaveBeenCalledWith(
				{
					type: 'workflowFailedToActivate',
					data: { workflowId: 'wf-1', errorMessage: 'Some error' },
				},
				sharedUserIds,
			);
		});

		test('should tear down partial registrations before deactivating when activation fails', async () => {
			vi.spyOn(activeWorkflowManager, 'add').mockRejectedValue(new Error('Some error'));
			const callOrder: string[] = [];
			vi.spyOn(activeWorkflowManager, 'clearWebhooks').mockImplementation(async () => {
				callOrder.push('clearWebhooks');
			});
			vi.spyOn(activeWorkflowManager, 'removeNonWebhookTriggers').mockImplementation(async () => {
				callOrder.push('removeNonWebhookTriggers');
			});
			workflowRepository.update.mockImplementation(async () => {
				callOrder.push('update');
				return {} as UpdateResult;
			});

			await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
				workflowId: 'wf-1',
				activeVersionId: 'v1',
				activationMode: 'activate',
			});

			expect(callOrder).toEqual(['clearWebhooks', 'removeNonWebhookTriggers', 'update']);
			expect(workflowRepository.update).toHaveBeenCalledWith('wf-1', {
				active: false,
				activeVersionId: null,
			});
		});

		test('should still deactivate the workflow when the teardown fails', async () => {
			vi.spyOn(activeWorkflowManager, 'add').mockRejectedValue(new Error('Some error'));
			vi.spyOn(activeWorkflowManager, 'clearWebhooks').mockResolvedValue();
			vi.spyOn(activeWorkflowManager, 'removeNonWebhookTriggers').mockRejectedValue(
				new Error('teardown failed'),
			);

			await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
				workflowId: 'wf-1',
				activeVersionId: 'v1',
				activationMode: 'activate',
			});

			expect(workflowRepository.update).toHaveBeenCalledWith('wf-1', {
				active: false,
				activeVersionId: null,
			});
		});

		test('should still remove triggers when webhook cleanup fails', async () => {
			vi.spyOn(activeWorkflowManager, 'add').mockRejectedValue(new Error('Some error'));
			vi.spyOn(activeWorkflowManager, 'clearWebhooks').mockRejectedValue(
				new Error('webhook cleanup failed'),
			);
			const removeNonWebhookTriggers = vi
				.spyOn(activeWorkflowManager, 'removeNonWebhookTriggers')
				.mockResolvedValue();

			await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
				workflowId: 'wf-1',
				activeVersionId: 'v1',
				activationMode: 'activate',
			});

			expect(removeNonWebhookTriggers).toHaveBeenCalledWith('wf-1');
			expect(workflowRepository.update).toHaveBeenCalledWith('wf-1', {
				active: false,
				activeVersionId: null,
			});
		});

		test('should push workflowActivated to accessible users and relay it on successful activation', async () => {
			vi.spyOn(activeWorkflowManager, 'add').mockResolvedValue({
				webhooks: true,
				triggersAndPollers: true,
			});

			await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
				workflowId: 'wf-1',
				activeVersionId: 'v1',
				activationMode: 'activate',
			});

			expect(workflowSharingService.getUserIdsWithAccessToWorkflowSafe).toHaveBeenCalledWith(
				'wf-1',
			);
			expect(push.sendToUsers).toHaveBeenCalledWith(
				{ type: 'workflowActivated', data: { workflowId: 'wf-1', activeVersionId: 'v1' } },
				sharedUserIds,
			);
			expect(publisher.publishCommand).toHaveBeenCalledWith({
				command: 'display-workflow-activation',
				payload: { workflowId: 'wf-1', activeVersionId: 'v1' },
			});
			expect(workflowRepository.update).not.toHaveBeenCalled();
		});

		test('should not roll back a successfully activated workflow when nobody can be resolved to notify', async () => {
			vi.spyOn(activeWorkflowManager, 'add').mockResolvedValue({
				webhooks: true,
				triggersAndPollers: true,
			});
			workflowSharingService.getUserIdsWithAccessToWorkflowSafe.mockResolvedValueOnce([]);

			await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
				workflowId: 'wf-1',
				activeVersionId: 'v1',
				activationMode: 'activate',
			});

			expect(push.sendToUsers).toHaveBeenCalledWith(
				{ type: 'workflowActivated', data: { workflowId: 'wf-1', activeVersionId: 'v1' } },
				[],
			);
			expect(publisher.publishCommand).toHaveBeenCalledWith({
				command: 'display-workflow-activation',
				payload: { workflowId: 'wf-1', activeVersionId: 'v1' },
			});
			expect(workflowRepository.update).not.toHaveBeenCalled();
		});

		test('should not roll back a successfully activated workflow even if the recipient lookup itself throws', async () => {
			vi.spyOn(activeWorkflowManager, 'add').mockResolvedValue({
				webhooks: true,
				triggersAndPollers: true,
			});
			workflowSharingService.getUserIdsWithAccessToWorkflowSafe.mockRejectedValueOnce(
				new Error('db unavailable'),
			);

			await expect(
				activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
					workflowId: 'wf-1',
					activeVersionId: 'v1',
					activationMode: 'activate',
				}),
			).rejects.toThrow('db unavailable');

			expect(workflowRepository.update).not.toHaveBeenCalled();
			expect(push.sendToUsers).not.toHaveBeenCalledWith(
				expect.objectContaining({ type: 'workflowFailedToActivate' }),
				expect.anything(),
			);
		});
	});

	describe('display and removal pubsub handlers', () => {
		const push = mock<Push>();
		const publisher = mock<Publisher>();
		const workflowSharingService = mock<WorkflowSharingService>();
		const workflowPushNotifier = new WorkflowPushNotifier(push, workflowSharingService);
		const sharedUserIds = ['user-1', 'user-2'];

		beforeEach(() => {
			vi.clearAllMocks();
			workflowSharingService.getUserIdsWithAccessToWorkflowSafe.mockResolvedValue(sharedUserIds);
			activeWorkflowManager = new ActiveWorkflowManager(
				mockLogger(),
				mock(),
				mock(),
				mock(),
				nodeTypes,
				mock(),
				workflowRepository,
				mock(),
				mock(),
				mock(),
				instanceSettings,
				publisher,
				mock(),
				mock<TriggerExecutionContextFactory>(),
				mock(),
				mock(), // scheduleTriggerJobRegistrar
				mock(), // pollTriggerJobRegistrar
				workflowPushNotifier,
				policyEnforcementService,
				ownershipService,
			);
		});

		test.each([
			{
				name: 'handleDisplayWorkflowActivation',
				call: async () =>
					await activeWorkflowManager.handleDisplayWorkflowActivation({
						workflowId: 'wf-1',
						activeVersionId: 'v1',
					}),
				expectedMsg: {
					type: 'workflowActivated',
					data: { workflowId: 'wf-1', activeVersionId: 'v1' },
				},
			},
			{
				name: 'handleDisplayWorkflowDeactivation',
				call: async () =>
					await activeWorkflowManager.handleDisplayWorkflowDeactivation({ workflowId: 'wf-1' }),
				expectedMsg: { type: 'workflowDeactivated', data: { workflowId: 'wf-1' } },
			},
			{
				name: 'handleDisplayWorkflowActivationError',
				call: async () =>
					await activeWorkflowManager.handleDisplayWorkflowActivationError({
						workflowId: 'wf-1',
						errorMessage: 'boom',
						nodeId: 'node-1',
					}),
				expectedMsg: {
					type: 'workflowFailedToActivate',
					data: {
						workflowId: 'wf-1',
						errorMessage: 'boom',
						errorDescription: undefined,
						nodeId: 'node-1',
					},
				},
			},
		])('$name sends its push message only to accessible users', async ({ call, expectedMsg }) => {
			await call();

			expect(workflowSharingService.getUserIdsWithAccessToWorkflowSafe).toHaveBeenCalledWith(
				'wf-1',
			);
			expect(push.sendToUsers).toHaveBeenCalledWith(expectedMsg, sharedUserIds);
		});

		test('handleRemoveNonWebhookTriggers sends workflowDeactivated only to accessible users and relays it', async () => {
			vi.spyOn(activeWorkflowManager, 'removeActivationError').mockResolvedValue();
			vi.spyOn(activeWorkflowManager, 'removeNonWebhookTriggers').mockResolvedValue();

			await activeWorkflowManager.handleRemoveNonWebhookTriggers({ workflowId: 'wf-1' });

			expect(workflowSharingService.getUserIdsWithAccessToWorkflowSafe).toHaveBeenCalledWith(
				'wf-1',
			);
			expect(push.sendToUsers).toHaveBeenCalledWith(
				{ type: 'workflowDeactivated', data: { workflowId: 'wf-1' } },
				sharedUserIds,
			);
			expect(publisher.publishCommand).toHaveBeenCalledWith({
				command: 'display-workflow-deactivation',
				payload: { workflowId: 'wf-1' },
			});
		});
	});

	describe('activateWorkflow', () => {
		beforeEach(() => {
			// Set up as leader to allow workflow activation
			Object.assign(instanceSettings, { isLeader: true });
		});

		test('should use active version when calling executeErrorWorkflow on activation failure', async () => {
			// Create different nodes for draft vs active version
			const draftNodes = [
				{
					id: 'draft-node-1',
					name: 'Draft Webhook',
					type: 'n8n-nodes-base.webhook',
					typeVersion: 1,
					position: [0, 0] as [number, number],
					parameters: {},
				},
			];

			const activeNodes = [
				{
					id: 'active-node-1',
					name: 'Active Webhook',
					type: 'n8n-nodes-base.webhook',
					typeVersion: 1,
					position: [0, 0] as [number, number],
					parameters: {},
				},
			];

			const activeVersion = mock<WorkflowHistory>({
				versionId: 'v1',
				workflowId: 'workflow-1',
				nodes: activeNodes,
				connections: {},
				authors: 'test-user',
				createdAt: new Date(),
				updatedAt: new Date(),
			});

			const workflowEntity = mock<WorkflowEntity>({
				id: 'workflow-1',
				name: 'Test Workflow',
				active: true,
				activeVersionId: activeVersion.versionId,
				nodes: draftNodes,
				connections: {},
				activeVersion,
			});

			workflowRepository.findById.mockResolvedValue(workflowEntity);

			// Mock the add method to throw an error (simulating activation failure)
			vi.spyOn(activeWorkflowManager, 'add').mockRejectedValue(new Error('Authorization failed'));

			const executeErrorWorkflowSpy = vi
				.spyOn(activeWorkflowManager, 'executeErrorWorkflow')
				.mockImplementation(() => {});

			await activeWorkflowManager['activateWorkflow']('workflow-1', 'init');

			expect(executeErrorWorkflowSpy).toHaveBeenCalled();

			// Get the workflow data that was passed to executeErrorWorkflow
			const callArgs = executeErrorWorkflowSpy.mock.calls[0];
			const workflowData = callArgs[1];

			expect(workflowData.nodes).toEqual(activeNodes);
			expect(workflowData.nodes[0].name).toBe('Active Webhook');
		});
	});

	describe('getExecuteTriggerFunctions / getExecutePollFunctions', () => {
		const workflowStaticDataService = mock<WorkflowStaticDataService>();
		const workflowExecutionService = mock<WorkflowExecutionService>();
		const eventService = mock<EventService>();
		const activeWorkflowTriggers = mock<ActiveWorkflowTriggers>();
		const activationErrorsService = mock<ActivationErrorsService>();
		const executionService = mock<ExecutionService>();
		let scopedLogger: Logger;

		let factory: TriggerExecutionContextFactory;
		let pollCursorService: MockProxy<PollCursorService>;

		beforeEach(() => {
			vi.clearAllMocks();
			pollCursorService = mock<PollCursorService>({ enabled: false });
			pollCursorService.resolveCursor.mockResolvedValue({ migrated: false });
			workflowStaticDataService.saveStaticData.mockResolvedValue(undefined);
			workflowExecutionService.runWorkflow.mockResolvedValue('exec-123');
			activeWorkflowTriggers.remove.mockResolvedValue(true);
			activationErrorsService.register.mockResolvedValue(undefined);
			executionService.createErrorExecution.mockResolvedValue(undefined);

			scopedLogger = mock<Logger>();
			const rootLogger = mock<Logger>({ scoped: vi.fn().mockReturnValue(scopedLogger) });

			const ownershipService = mock<OwnershipService>();
			ownershipService.getWorkflowProjectCached.mockResolvedValue(
				mock<Project>({ id: 'project-1', name: 'Test Project' }),
			);

			factory = new TriggerExecutionContextFactory(
				rootLogger,
				mock(), // errorReporter
				mock(), // activeExecutions
				eventService,
				executionService,
				workflowStaticDataService,
				workflowExecutionService,
				mock(), // storageConfig
				mock(), // workflowPublishedDataService
				mock(), // scheduleTriggerJobRegistrar
				ownershipService,
				mock(), // nodeTypes
				pollCursorService,
				mock<GlobalConfig>({ scheduler: { pollTimeoutSeconds: 45, leaseDurationSeconds: 60 } }),
			);

			activeWorkflowManager = new ActiveWorkflowManager(
				rootLogger,
				mock(), // errorReporter
				activeWorkflowTriggers,
				mock(), // externalHooks
				nodeTypes,
				mock(), // webhookService
				workflowRepository,
				activationErrorsService,
				workflowStaticDataService,
				mock(), // activeWorkflowsService
				instanceSettings,
				mock(), // publisher
				workflowsConfig,
				factory,
				mock(), // eventBus
				mock(), // scheduleTriggerJobRegistrar
				mock(), // pollTriggerJobRegistrar
				mock(), // workflowPushNotifier
				policyEnforcementService,
				ownershipService,
			);
		});

		describe('emit', () => {
			test('calls workflowStaticDataService.saveStaticData, workflowExecutionService.runWorkflow, and eventService.emit', async () => {
				const workflowData = mock<WorkflowEntity>({
					id: 'wf-1',
					name: 'Test Workflow',
				});
				const additionalData = mock<IWorkflowExecuteAdditionalData>();
				const mode: WorkflowExecuteMode = 'trigger';
				const activation: WorkflowActivateMode = 'activate';
				const workflow = mock<Workflow>({ name: 'Test Workflow' });
				const node = mock<INode>({ name: 'Trigger Node' });
				const triggerData: INodeExecutionData[][] = [[]];

				const getTriggerFunctions = activeWorkflowManager.getExecuteTriggerFunctions(
					workflowData,
					additionalData,
					mode,
					activation,
					async () => workflowData,
					mock<ScheduleTriggerCollectionSession>(),
				);
				const context = getTriggerFunctions(workflow, node, additionalData, mode, activation);

				context.emit(triggerData);

				await sleep(0);

				expect(workflowStaticDataService.saveStaticData).toHaveBeenCalledWith(workflow);
				expect(workflowExecutionService.runWorkflow).toHaveBeenCalledWith(
					workflowData,
					node,
					triggerData,
					additionalData,
					mode,
					undefined,
					undefined,
				);

				expect(eventService.emit).toHaveBeenCalledWith('workflow-executed', {
					workflowId: workflowData.id,
					workflowName: workflowData.name,
					executionId: 'exec-123',
					projectId: 'project-1',
					projectName: 'Test Project',
					source: 'trigger',
				});
			});

			test('forwards deduplicationKey to workflowExecutionService.runWorkflow', async () => {
				const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
				const additionalData = mock<IWorkflowExecuteAdditionalData>();
				const mode: WorkflowExecuteMode = 'trigger';
				const activation: WorkflowActivateMode = 'activate';
				const workflow = mock<Workflow>({ name: 'Test Workflow' });
				const node = mock<INode>({ name: 'Trigger Node', id: 'node-1' });
				const triggerData: INodeExecutionData[][] = [[]];

				const getTriggerFunctions = activeWorkflowManager.getExecuteTriggerFunctions(
					workflowData,
					additionalData,
					mode,
					activation,
					async () => workflowData,
					mock<ScheduleTriggerCollectionSession>(),
				);
				const context = getTriggerFunctions(workflow, node, additionalData, mode, activation);

				context.emit(triggerData, undefined, undefined, 'wf-1:node-1:1700000000000');

				await sleep(0);

				expect(workflowExecutionService.runWorkflow).toHaveBeenCalledWith(
					workflowData,
					node,
					triggerData,
					additionalData,
					mode,
					undefined,
					'wf-1:node-1:1700000000000',
				);
			});

			test('skips event emission when runWorkflow rejects with DuplicateExecutionError', async () => {
				const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
				const additionalData = mock<IWorkflowExecuteAdditionalData>();
				const mode: WorkflowExecuteMode = 'trigger';
				const activation: WorkflowActivateMode = 'activate';
				const workflow = mock<Workflow>({ name: 'Test Workflow' });
				const node = mock<INode>({ name: 'Trigger Node', id: 'node-1' });
				const triggerData: INodeExecutionData[][] = [[]];

				workflowExecutionService.runWorkflow.mockRejectedValueOnce(
					new DuplicateExecutionError('wf-1:node-1:1700000000000'),
				);

				const getTriggerFunctions = activeWorkflowManager.getExecuteTriggerFunctions(
					workflowData,
					additionalData,
					mode,
					activation,
					async () => workflowData,
					mock<ScheduleTriggerCollectionSession>(),
				);
				const context = getTriggerFunctions(workflow, node, additionalData, mode, activation);

				context.emit(triggerData, undefined, undefined, 'wf-1:node-1:1700000000000');

				await sleep(0);

				expect(eventService.emit).not.toHaveBeenCalled();
			});

			test('resolves donePromise with undefined when runWorkflow rejects with DuplicateExecutionError', async () => {
				const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
				const additionalData = mock<IWorkflowExecuteAdditionalData>();
				const mode: WorkflowExecuteMode = 'trigger';
				const activation: WorkflowActivateMode = 'activate';
				const workflow = mock<Workflow>({ name: 'Test Workflow' });
				const node = mock<INode>({ name: 'Trigger Node', id: 'node-1' });
				const triggerData: INodeExecutionData[][] = [[]];

				workflowExecutionService.runWorkflow.mockRejectedValueOnce(
					new DuplicateExecutionError('wf-1:node-1:1700000000000'),
				);

				const getTriggerFunctions = activeWorkflowManager.getExecuteTriggerFunctions(
					workflowData,
					additionalData,
					mode,
					activation,
					async () => workflowData,
					mock<ScheduleTriggerCollectionSession>(),
				);
				const context = getTriggerFunctions(workflow, node, additionalData, mode, activation);

				const donePromise = createDeferredPromise<IRun>();
				context.emit(triggerData, undefined, donePromise, 'wf-1:node-1:1700000000000');

				await expect(donePromise.promise).resolves.toBeUndefined();
			});
		});

		describe('emitError', () => {
			test('removes workflow from activeWorkflowTriggers, registers error, calls executeErrorWorkflow and addQueuedWorkflowActivation', () => {
				const workflowData = mock<WorkflowEntity>({
					id: 'wf-1',
					name: 'Test Workflow',
				});
				const additionalData = mock<IWorkflowExecuteAdditionalData>();
				const mode: WorkflowExecuteMode = 'trigger';
				const activation: WorkflowActivateMode = 'activate';
				const workflow = mock<Workflow>({ name: 'Test Workflow' });
				const node = mock<INode>({ name: 'Trigger Node' });
				const triggerError = new Error('Trigger connection failed');

				const executeErrorWorkflowSpy = vi
					.spyOn(activeWorkflowManager, 'executeErrorWorkflow')
					.mockImplementation(() => {});
				const addQueuedWorkflowActivationSpy = vi.spyOn(
					activeWorkflowManager as unknown as Record<
						'addQueuedWorkflowActivation',
						(a: WorkflowActivateMode, w: WorkflowEntity) => void
					>,
					'addQueuedWorkflowActivation',
				);

				const getTriggerFunctions = activeWorkflowManager.getExecuteTriggerFunctions(
					workflowData,
					additionalData,
					mode,
					activation,
					async () => workflowData,
					mock<ScheduleTriggerCollectionSession>(),
				);
				const context = getTriggerFunctions(workflow, node, additionalData, mode, activation);

				context.emitError(triggerError);

				expect(activeWorkflowTriggers.remove).toHaveBeenCalledWith(workflowData.id);
				expect(activationErrorsService.register).toHaveBeenCalledWith(
					workflowData.id,
					triggerError.message,
				);
				expect(executeErrorWorkflowSpy).toHaveBeenCalled();
				expect(addQueuedWorkflowActivationSpy).toHaveBeenCalledWith(activation, workflowData);
			});

			test('wraps the cause in a WorkflowActivationError that surfaces the cause message in `description`', () => {
				const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
				const additionalData = mock<IWorkflowExecuteAdditionalData>();
				const mode: WorkflowExecuteMode = 'trigger';
				const activation: WorkflowActivateMode = 'activate';
				const workflow = mock<Workflow>({ name: 'Test Workflow' });
				const node = mock<INode>({ name: 'Trigger Node' });
				const triggerError = new Error('IMAP connection closed unexpectedly');

				const executeErrorWorkflowSpy = vi
					.spyOn(activeWorkflowManager, 'executeErrorWorkflow')
					.mockImplementation(() => {});

				const getTriggerFunctions = activeWorkflowManager.getExecuteTriggerFunctions(
					workflowData,
					additionalData,
					mode,
					activation,
					async () => workflowData,
					mock<ScheduleTriggerCollectionSession>(),
				);
				const context = getTriggerFunctions(workflow, node, additionalData, mode, activation);

				context.emitError(triggerError);

				const wrappedError = executeErrorWorkflowSpy.mock.calls[0][0] as WorkflowActivationError;
				expect(wrappedError).toBeInstanceOf(WorkflowActivationError);
				expect(wrappedError.message).not.toBe(triggerError.message);
				expect(wrappedError.description).toBe('IMAP connection closed unexpectedly');
			});
		});

		describe('saveFailedExecution', () => {
			test('calls executionService.createErrorExecution and executeErrorWorkflow', async () => {
				const workflowData = mock<WorkflowEntity>({
					id: 'wf-1',
					name: 'Test Workflow',
				});
				const additionalData = mock<IWorkflowExecuteAdditionalData>();
				const mode: WorkflowExecuteMode = 'trigger';
				const activation: WorkflowActivateMode = 'activate';
				const workflow = mock<Workflow>({ name: 'Test Workflow' });
				const node = mock<INode>({ name: 'Trigger Node' });
				const executionError = mock<ExecutionError>();

				const executeErrorWorkflowSpy = vi
					.spyOn(factory, 'executeErrorWorkflow')
					.mockImplementation(() => {});

				const getTriggerFunctions = activeWorkflowManager.getExecuteTriggerFunctions(
					workflowData,
					additionalData,
					mode,
					activation,
					async () => workflowData,
					mock<ScheduleTriggerCollectionSession>(),
				);
				const context = getTriggerFunctions(workflow, node, additionalData, mode, activation);

				context.saveFailedExecution(executionError);

				await sleep(0);

				expect(executionService.createErrorExecution).toHaveBeenCalledWith(
					executionError,
					node,
					workflowData,
					workflow,
					mode,
				);
				expect(executeErrorWorkflowSpy).toHaveBeenCalledWith(executionError, workflowData, mode);
			});
		});

		describe('getExecutePollFunctions', () => {
			const scheduledTaskManager = mock<ScheduledTaskManager>();

			const createPollContext = () => {
				const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
				const additionalData = mock<IWorkflowExecuteAdditionalData>();
				const mode: WorkflowExecuteMode = 'trigger';
				const activation: WorkflowActivateMode = 'activate';
				const workflow = mock<Workflow>({ id: 'wf-1', name: 'Test Workflow' });
				const node = mock<INode>({ name: 'Poll Node' });

				const getPollFunctions = activeWorkflowManager.getExecutePollFunctions(
					workflowData,
					additionalData,
					mode,
					activation,
					async () => workflowData,
				);
				return {
					workflow,
					node,
					context: getPollFunctions(workflow, node, additionalData, mode, activation),
				};
			};

			test('__emit persists static data and starts a workflow execution', async () => {
				const { workflow, context } = createPollContext();

				context.__emit([[{ json: {} }]]);

				await sleep(0);

				expect(workflowStaticDataService.saveStaticData).toHaveBeenCalledWith(workflow);
				expect(workflowExecutionService.runWorkflow).toHaveBeenCalled();
			});

			test('does not persist the state of an in-flight poll dropped by workflow removal', async () => {
				// ActiveWorkflowTriggers drops in-flight polls of superseded registrations before
				// `__emit`; see active-workflow-triggers.ts. That is only safe because persistence
				// happens exclusively in `__emit`, and by skipping it, a poller advancing
				// its state via getWorkflowStaticData() inside poll() only mutates in-memory,
				// so the next registration's poller re-fetches the same events from the stored state.
				let pollCount = 0;
				let resolveInFlightPoll!: (value: INodeExecutionData[][] | null) => void;
				const pollNodeType = {
					description: { properties: [] },
					async poll(this: IPollFunctions) {
						const cursor = this.getWorkflowStaticData('node');
						pollCount++;
						cursor.lastId = pollCount;
						if (pollCount === 1) return null; // activation test poll: no new events
						if (pollCount === 2) return [[{ json: { id: 2 } }]]; // first cron tick: emits
						// second cron tick: hangs in flight
						return await new Promise<INodeExecutionData[][] | null>((resolve) => {
							resolveInFlightPoll = resolve;
						});
					},
				} as unknown as INodeType;
				const pollNodeTypes = {
					getByNameAndVersion: () => pollNodeType,
				} as unknown as INodeTypes;

				const workflow = new Workflow({
					id: 'wf-1',
					name: 'Test Workflow',
					nodes: [
						{
							id: 'node-1',
							name: 'Poll Node',
							type: 'test.poll',
							typeVersion: 1,
							position: [0, 0],
							parameters: {},
						},
					],
					connections: {},
					active: true,
					nodeTypes: pollNodeTypes,
					staticData: {},
				});
				workflow.nodes['Poll Node'].parameters = {
					pollTimes: { item: [{ mode: 'everyMinute' }] },
				};

				const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
				const additionalData = mock<IWorkflowExecuteAdditionalData>();
				const logger = mock<Logger>({ scoped: vi.fn().mockReturnValue(mock<Logger>()) });
				const triggersAndPollers = {
					runPollFunction: async (wf: Workflow, node: INode, pollFunctions: IPollFunctions) =>
						await wf.nodeTypes
							.getByNameAndVersion(node.type, node.typeVersion)
							.poll!.call(pollFunctions),
				} as ConstructorParameters<typeof ActiveWorkflowTriggers>[2];
				const realActiveWorkflowTriggers = new ActiveWorkflowTriggers(
					logger,
					scheduledTaskManager,
					triggersAndPollers,
					mock(),
					new PollTriggerExecutor(logger, triggersAndPollers, new Tracing(), mock()),
				);

				await realActiveWorkflowTriggers.addAllTriggers(
					'wf-1',
					workflow,
					additionalData,
					'trigger',
					'activate',
					mock(),
					activeWorkflowManager.getExecutePollFunctions(
						workflowData,
						additionalData,
						'trigger',
						'activate',
						async () => workflowData,
					),
				);

				const executeScheduledPoll = scheduledTaskManager.register.mock.calls[0][1] as () => void;

				// First cron tick completes and emits: advanced pool state is persisted.
				executeScheduledPoll();
				await sleep(0);
				expect(workflowStaticDataService.saveStaticData).toHaveBeenCalledTimes(1);
				expect(workflowExecutionService.runWorkflow).toHaveBeenCalledTimes(1);

				// Second cron tick hangs in flight; the workflow is removed during that time
				executeScheduledPoll();
				await sleep(0);
				await realActiveWorkflowTriggers.remove('wf-1');
				resolveInFlightPoll([[{ json: { id: 3 } }]]);
				await sleep(0);

				// poll() did advance the state in memory, but the dropped poll neither
				// persisted it nor started an execution; the next registration
				// fetches the same events from upstream
				expect(workflow.getStaticData('node', workflow.nodes['Poll Node']).lastId).toBe(3);
				expect(workflowStaticDataService.saveStaticData).toHaveBeenCalledTimes(1);
				expect(workflowExecutionService.runWorkflow).toHaveBeenCalledTimes(1);
			});
		});
	});

	describe('removeNonWebhookTriggers', () => {
		// Wire the real ActiveWorkflowTriggers + real ScheduledTaskManager through the manager
		// so the test asserts the cron is actually stopped, not just that a method was
		// called.
		const hourly = '0 * * * *' as CronExpression;
		const scheduleTriggerJobRegistrar = mock<ScheduleTriggerJobRegistrar>();
		const pollTriggerJobRegistrar = mock<PollTriggerJobRegistrar>();
		const errorReporter = mock<ErrorReporter>();
		let realScheduledTaskManager: ScheduledTaskManager;
		let realActiveWorkflowTriggers: ActiveWorkflowTriggers;

		beforeEach(() => {
			vi.clearAllMocks();
			vi.useFakeTimers();
			realScheduledTaskManager = new ScheduledTaskManager(
				mock<InstanceSettings>({ isLeader: true }),
				mock<Logger>({ scoped: vi.fn().mockReturnValue(mock<Logger>()) }),
				mock(),
			);
			realActiveWorkflowTriggers = new ActiveWorkflowTriggers(
				mock<Logger>({ scoped: vi.fn().mockReturnValue(mock<Logger>()) }),
				realScheduledTaskManager,
				mock(),
				mock(),
				mock<PollTriggerExecutor>(),
			);
			activeWorkflowManager = new ActiveWorkflowManager(
				mockLogger(),
				errorReporter,
				realActiveWorkflowTriggers,
				mock(),
				nodeTypes,
				mock(),
				workflowRepository,
				mock(),
				mock(),
				mock(),
				instanceSettings,
				mock(),
				workflowsConfig,
				mock<TriggerExecutionContextFactory>(),
				mock(),
				scheduleTriggerJobRegistrar,
				pollTriggerJobRegistrar,
				mock(), // workflowPushNotifier
				policyEnforcementService,
				ownershipService,
			);
		});

		afterEach(() => {
			realScheduledTaskManager.deregisterGroups(WORKFLOW_SCHEDULE_GROUP_TYPE);
			vi.useRealTimers();
		});

		it('should stop a cron left registered for an inactive workflow', async () => {
			realScheduledTaskManager.register(
				{
					group: workflowGroup('wf-desynced'),
					targetId: 'schedule-node',
					timezone: 'GMT',
					expression: hourly,
				},
				vi.fn(),
			);
			expect(realScheduledTaskManager.hasGroup(workflowGroup('wf-desynced'))).toBe(true);
			expect(realActiveWorkflowTriggers.isActive('wf-desynced')).toBe(false);

			await activeWorkflowManager.removeNonWebhookTriggers('wf-desynced');

			expect(realScheduledTaskManager.hasGroup(workflowGroup('wf-desynced'))).toBe(false);
		});

		it('should deprovision the durable jobs of the workflow on removal', async () => {
			// A deactivation through the legacy path (e.g. a workflow transfer with the
			// publication flag on) must delete the durable rows its activation committed,
			// or the durable scheduler keeps firing a workflow now marked inactive.
			realScheduledTaskManager.register(
				{
					group: workflowGroup('wf-durable'),
					targetId: 'schedule-node',
					timezone: 'GMT',
					expression: hourly,
				},
				vi.fn(),
			);

			await activeWorkflowManager.removeNonWebhookTriggers('wf-durable');

			expect(scheduleTriggerJobRegistrar.removeWorkflow).toHaveBeenCalledWith('wf-durable');
			expect(pollTriggerJobRegistrar.removeWorkflow).toHaveBeenCalledWith('wf-durable');
		});

		it('should deprovision durable jobs even with no in-memory registration', async () => {
			// An earlier removal may have cleared the in-memory registration and then
			// failed on the durable delete; the retry must still find and drop the rows.
			expect(realActiveWorkflowTriggers.isActive('wf-retry')).toBe(false);

			await activeWorkflowManager.removeNonWebhookTriggers('wf-retry');

			expect(scheduleTriggerJobRegistrar.removeWorkflow).toHaveBeenCalledWith('wf-retry');
			expect(pollTriggerJobRegistrar.removeWorkflow).toHaveBeenCalledWith('wf-retry');
		});

		it('should not let a durable deprovision failure abort deactivation', async () => {
			// The in-memory triggers are torn down first and the deprovision is
			// idempotent, so a failure here must be reported and swallowed rather than
			// propagated: in the pubsub path an unhandled throw would skip the
			// workflowDeactivated broadcast, stranding the workflow as active in the UI.
			realScheduledTaskManager.register(
				{
					group: workflowGroup('wf-durable-fail'),
					targetId: 'schedule-node',
					timezone: 'GMT',
					expression: hourly,
				},
				vi.fn(),
			);
			const deprovisionError = new Error('durable delete failed');
			scheduleTriggerJobRegistrar.removeWorkflow.mockRejectedValueOnce(deprovisionError);

			await expect(
				activeWorkflowManager.removeNonWebhookTriggers('wf-durable-fail'),
			).resolves.not.toThrow();

			// In-memory cron was still stopped, and the failure was reported.
			expect(realScheduledTaskManager.hasGroup(workflowGroup('wf-durable-fail'))).toBe(false);
			expect(errorReporter.error).toHaveBeenCalledWith(deprovisionError);
			// A schedule-registrar failure must not skip the poll-registrar call.
			expect(pollTriggerJobRegistrar.removeWorkflow).toHaveBeenCalledWith('wf-durable-fail');
		});

		it('should not let a poll-registrar deprovision failure abort deactivation', async () => {
			const deprovisionError = new Error('durable poll delete failed');
			pollTriggerJobRegistrar.removeWorkflow.mockRejectedValueOnce(deprovisionError);

			await expect(
				activeWorkflowManager.removeNonWebhookTriggers('wf-durable-poll-fail'),
			).resolves.not.toThrow();

			expect(errorReporter.error).toHaveBeenCalledWith(deprovisionError);
		});

		it('should stop a stranded cron on leader stepdown / shutdown', async () => {
			// removeAllNonWebhookTriggerWorkflows is the @OnLeaderStepdown / @OnShutdown
			// handler. On stepdown the process keeps running as a follower, so a stranded
			// cron left behind would survive the demotion and resurface on the next takeover.
			realScheduledTaskManager.register(
				{
					group: workflowGroup('wf-orphan'),
					targetId: 'schedule-node',
					timezone: 'GMT',
					expression: hourly,
				},
				vi.fn(),
			);
			expect(realScheduledTaskManager.hasGroup(workflowGroup('wf-orphan'))).toBe(true);
			expect(realActiveWorkflowTriggers.isActive('wf-orphan')).toBe(false);

			await activeWorkflowManager.removeAllNonWebhookTriggerWorkflows();

			expect(realScheduledTaskManager.hasGroup(workflowGroup('wf-orphan'))).toBe(false);
			// Durable jobs track the published state of a workflow, not this instance's
			// leadership, so stepdown/shutdown must never deprovision them.
			expect(scheduleTriggerJobRegistrar.removeWorkflow).not.toHaveBeenCalled();
			expect(pollTriggerJobRegistrar.removeWorkflow).not.toHaveBeenCalled();
		});

		it('does not tear down triggers under the publication service flag', async () => {
			// Under the publication service, teardown runs in PublishedWorkflowTriggerDeactivator
			// under the lifecycle lock, so this handler must be a no-op.
			workflowsConfig.useWorkflowPublicationService = true;
			try {
				realScheduledTaskManager.register(
					{
						group: workflowGroup('wf-pub'),
						targetId: 'schedule-node',
						timezone: 'GMT',
						expression: hourly,
					},
					vi.fn(),
				);

				await activeWorkflowManager.removeAllNonWebhookTriggerWorkflows();

				expect(realScheduledTaskManager.hasGroup(workflowGroup('wf-pub'))).toBe(true);
			} finally {
				workflowsConfig.useWorkflowPublicationService = false;
			}
		});

		it('does not re-activate workflows on takeover under the publication service flag', async () => {
			// Re-activation goes through the outbox consumer instead.
			workflowsConfig.useWorkflowPublicationService = true;
			const addActiveWorkflows = vi
				.spyOn(activeWorkflowManager, 'addActiveWorkflows')
				.mockResolvedValue(undefined);
			try {
				await activeWorkflowManager.addAllNonWebhookTriggerWorkflows();

				expect(addActiveWorkflows).not.toHaveBeenCalled();
			} finally {
				workflowsConfig.useWorkflowPublicationService = false;
				addActiveWorkflows.mockRestore();
			}
		});
	});

	describe('remove (multi-main)', () => {
		const publisher = mock<Publisher>();
		const scheduleTriggerJobRegistrar = mock<ScheduleTriggerJobRegistrar>();
		const pollTriggerJobRegistrar = mock<PollTriggerJobRegistrar>();

		const makeManager = () =>
			new ActiveWorkflowManager(
				mockLogger(),
				mock(),
				mock(),
				mock(),
				nodeTypes,
				mock(),
				workflowRepository,
				mock(),
				mock(),
				mock(),
				mock<InstanceSettings>({ isMultiMain: true }),
				publisher,
				workflowsConfig,
				mock<TriggerExecutionContextFactory>(),
				mock(),
				scheduleTriggerJobRegistrar,
				pollTriggerJobRegistrar,
				mock(), // workflowPushNotifier
				policyEnforcementService,
				ownershipService,
			);

		beforeEach(() => vi.clearAllMocks());

		it('deprovisions durable jobs on the handling main, not only via the forwarded command', async () => {
			// The command to tear down the in-memory triggers is fire-and-forget to the
			// leader; durable rows are DB state, so their removal runs here and does not
			// depend on that command landing.
			await makeManager().remove('wf-mm');

			expect(scheduleTriggerJobRegistrar.removeWorkflow).toHaveBeenCalledWith('wf-mm');
			expect(pollTriggerJobRegistrar.removeWorkflow).toHaveBeenCalledWith('wf-mm');
			expect(publisher.publishCommand).toHaveBeenCalledWith({
				command: 'remove-triggers-and-pollers',
				payload: { workflowId: 'wf-mm' },
			});
		});

		it('still forwards the teardown command when the durable deprovision fails', async () => {
			scheduleTriggerJobRegistrar.removeWorkflow.mockRejectedValueOnce(
				new Error('durable delete failed'),
			);

			await expect(makeManager().remove('wf-mm-fail')).resolves.not.toThrow();

			expect(publisher.publishCommand).toHaveBeenCalledWith({
				command: 'remove-triggers-and-pollers',
				payload: { workflowId: 'wf-mm-fail' },
			});
			// A schedule-registrar failure must not skip the poll-registrar call.
			expect(pollTriggerJobRegistrar.removeWorkflow).toHaveBeenCalledWith('wf-mm-fail');
		});

		it('still forwards the teardown command when the poll-registrar deprovision fails', async () => {
			pollTriggerJobRegistrar.removeWorkflow.mockRejectedValueOnce(
				new Error('durable poll delete failed'),
			);

			await expect(makeManager().remove('wf-mm-poll-fail')).resolves.not.toThrow();

			expect(publisher.publishCommand).toHaveBeenCalledWith({
				command: 'remove-triggers-and-pollers',
				payload: { workflowId: 'wf-mm-poll-fail' },
			});
		});
	});

	describe('addNonWebhookTriggers', () => {
		// This legacy path can run with the durable scheduler intercepting schedule
		// nodes (e.g. a workflow transfer with the publication flag on), so the
		// rules its trigger contexts collect must be committed or discarded here
		// just like on the publication path.
		const activeWorkflowTriggers = mock<ActiveWorkflowTriggers>();
		const scheduleTriggerJobRegistrar = mock<ScheduleTriggerJobRegistrar>();
		const scheduleCollectionSession = mock<ScheduleTriggerCollectionSession>();
		const triggerNode = mock<INode>({ id: 'trigger-a' });

		const makeManager = () =>
			new ActiveWorkflowManager(
				mockLogger(),
				mock(),
				activeWorkflowTriggers,
				mock(),
				nodeTypes,
				mock(),
				workflowRepository,
				mock(),
				mock(),
				mock(),
				instanceSettings,
				mock(),
				workflowsConfig,
				mock<TriggerExecutionContextFactory>(),
				mock(),
				scheduleTriggerJobRegistrar,
				mock(), // pollTriggerJobRegistrar
				mock(), // workflowPushNotifier
				policyEnforcementService,
				ownershipService,
			);

		const makeWorkflow = () => {
			const workflow = mock<Workflow>({ id: 'wf-1' });
			workflow.getTriggerNodes.mockReturnValue([triggerNode]);
			workflow.getPollNodes.mockReturnValue([]);
			return workflow;
		};

		const registrationContext = () => ({
			activationMode: 'update' as WorkflowActivateMode,
			executionMode: 'trigger' as WorkflowExecuteMode,
			additionalData: mock<IWorkflowExecuteAdditionalData>(),
			resolveWorkflowData: async () => mock<IWorkflowBase>(),
		});

		beforeEach(() => {
			vi.clearAllMocks();
			scheduleTriggerJobRegistrar.createSession.mockReturnValue(scheduleCollectionSession);
		});

		test('commits collected durable schedules after in-memory registration, then discards', async () => {
			const workflow = makeWorkflow();

			await makeManager().addNonWebhookTriggers(
				mock<WorkflowEntity>({ id: 'wf-1' }),
				workflow,
				registrationContext(),
			);

			expect(scheduleCollectionSession.commit).toHaveBeenCalledWith('wf-1', 'trigger-a');
			expect(scheduleCollectionSession.discard).toHaveBeenCalledWith('wf-1', 'trigger-a');
		});

		test('discards without committing when in-memory registration fails', async () => {
			const workflow = makeWorkflow();
			activeWorkflowTriggers.addTriggers.mockRejectedValue(new Error('activation failed'));

			await expect(
				makeManager().addNonWebhookTriggers(
					mock<WorkflowEntity>({ id: 'wf-1' }),
					workflow,
					registrationContext(),
				),
			).rejects.toThrow('activation failed');

			expect(scheduleCollectionSession.commit).not.toHaveBeenCalled();
			expect(scheduleCollectionSession.discard).toHaveBeenCalledWith('wf-1', 'trigger-a');
		});
	});
});