import type { Logger } from '@n8n/backend-common';
import type { WorkflowsConfig } from '@n8n/config';
import type { WorkflowPublicationOutbox, WorkflowPublicationOutboxRepository } from '@n8n/db';
import { mock } from 'vitest-mock-extended';
import type { ErrorReporter, InstanceSettings, Span, Tracing } from 'n8n-core';
import { UserError } from 'n8n-workflow';

import type { EventService } from '@/events/event.service';
import type { PublicationResult } from '@/workflows/publication/publication-result';
import type { PublicationStatusReporter } from '@/workflows/publication/publication-status-reporter';
import { WorkflowPublicationLifecycleLock } from '@/workflows/publication/workflow-publication-lifecycle-lock';
import type { WorkflowPublicationApplier } from '@/workflows/publication/workflow-publication-applier';
import { WorkflowPublicationOutboxConsumer } from '@/workflows/publication/workflow-publication-outbox-consumer';

describe('WorkflowPublicationOutboxConsumer', () => {
	const logger = mock<Logger>();
	logger.scoped.mockReturnValue(logger);

	const errorReporter = mock<ErrorReporter>();
	const outboxRepository = mock<WorkflowPublicationOutboxRepository>();
	const applier = mock<WorkflowPublicationApplier>();
	const reporter = mock<PublicationStatusReporter>();
	const tracing = mock<Tracing>();
	const eventService = mock<EventService>();
	const abortSignal = new AbortController().signal;

	let consumer: WorkflowPublicationOutboxConsumer;

	const POLL_INTERVAL_MS = 15_000;
	const LEASE_SECONDS = 120;
	// Abort fires at 70% of the lease; abandonment follows after the grace period.
	const ABORT_AFTER_MS = LEASE_SECONDS * 0.7 * 1000;
	const ABANDON_GRACE_MS = 10_000;

	let lifecycleLock: WorkflowPublicationLifecycleLock;

	function createConsumer(
		useWorkflowPublicationService = true,
		isLeader = true,
		concurrency = 1,
		leaseSeconds = LEASE_SECONDS,
	) {
		const workflowsConfig = mock<WorkflowsConfig>({
			useWorkflowPublicationService,
			publicationOutboxPollIntervalMs: POLL_INTERVAL_MS,
			workflowPublicationConcurrency: concurrency,
			publicationOutboxLeaseSeconds: leaseSeconds,
		});
		lifecycleLock = new WorkflowPublicationLifecycleLock();
		return new WorkflowPublicationOutboxConsumer(
			logger,
			workflowsConfig,
			errorReporter,
			outboxRepository,
			applier,
			reporter,
			mock<InstanceSettings>({ isLeader }),
			lifecycleLock,
			tracing,
			eventService,
		);
	}

	function makeRecord(
		overrides: Partial<WorkflowPublicationOutbox> = {},
	): WorkflowPublicationOutbox {
		return {
			id: 1,
			workflowId: 'wf-1',
			publishedVersionId: 'v-2',
			status: 'in_progress',
			errorMessage: null,
			createdAt: new Date(),
			updatedAt: new Date(),
			...overrides,
		} as WorkflowPublicationOutbox;
	}

	beforeEach(() => {
		vi.clearAllMocks();
		vi.useFakeTimers();
		tracing.startSpan.mockImplementation(async (_opts, spanCb) => await spanCb(mock<Span>()));
		outboxRepository.claimNextPendingRecord.mockResolvedValue(null);
		applier.apply.mockResolvedValue({ type: 'completed', triggerStatuses: [] });
		reporter.report.mockResolvedValue(undefined);
		consumer = createConsumer();
	});

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

	describe('init', () => {
		test('on the leader, drains pending records immediately and starts polling', async () => {
			const record = makeRecord({ id: 1 });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(record).mockResolvedValue(null);
			consumer = createConsumer(true, true);

			await consumer.init();

			expect(applier.apply).toHaveBeenCalledTimes(1);
			expect(reporter.report).toHaveBeenCalledTimes(1);
			expect(vi.getTimerCount()).toBe(1);
		});

		test('on a follower, does nothing', async () => {
			consumer = createConsumer(true, false);

			await consumer.init();

			expect(outboxRepository.claimNextPendingRecord).not.toHaveBeenCalled();
			expect(vi.getTimerCount()).toBe(0);
		});
	});

	describe('lifecycle', () => {
		test('startPolling starts interval when feature flag is on', () => {
			consumer.startPolling();

			expect(vi.getTimerCount()).toBe(1);
		});

		test('polling schedules the next cycle after a timer fires', async () => {
			consumer.startPolling();

			await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);

			expect(outboxRepository.claimNextPendingRecord).toHaveBeenCalledTimes(1);
			expect(vi.getTimerCount()).toBe(1);
		});

		test('polling stops scheduling when leadership is lost during a cycle', async () => {
			outboxRepository.claimNextPendingRecord.mockImplementationOnce(async () => {
				consumer.stopPolling();
				return null;
			});
			consumer.startPolling();

			await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);

			expect(outboxRepository.claimNextPendingRecord).toHaveBeenCalledTimes(1);
			expect(vi.getTimerCount()).toBe(0);
		});

		test('polling reports claim errors and schedules the next cycle', async () => {
			const error = new Error('claim failed');
			outboxRepository.claimNextPendingRecord.mockRejectedValueOnce(error);
			consumer.startPolling();

			await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);

			expect(errorReporter.error).toHaveBeenCalledWith(error, { shouldBeLogged: true });
			expect(vi.getTimerCount()).toBe(1);

			await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);
			expect(outboxRepository.claimNextPendingRecord).toHaveBeenCalledTimes(2);
		});

		test('startPolling does nothing when feature flag is off', () => {
			consumer = createConsumer(false);
			consumer.startPolling();

			expect(vi.getTimerCount()).toBe(0);
		});

		test('stopPolling clears the interval', () => {
			consumer.startPolling();
			expect(vi.getTimerCount()).toBe(1);

			consumer.stopPolling();
			expect(vi.getTimerCount()).toBe(0);
		});

		test('shutdown stops polling', async () => {
			consumer.startPolling();
			await consumer.shutdown();

			expect(vi.getTimerCount()).toBe(0);
		});
	});

	describe('concurrent drains', () => {
		test('overlapping drainPending calls never exceed the concurrency cap', async () => {
			const record = makeRecord({ id: 1 });
			let releaseApply!: () => void;
			let signalApplyStarted!: () => void;
			const applyStarted = new Promise<void>((resolve) => {
				signalApplyStarted = resolve;
			});
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(record).mockResolvedValue(null);
			applier.apply.mockImplementationOnce(async () => {
				signalApplyStarted();
				await new Promise<void>((resolve) => {
					releaseApply = resolve;
				});
				return { type: 'completed', triggerStatuses: [] };
			});
			consumer.startPolling();

			const first = consumer.drainPending();
			await applyStarted;
			const second = consumer.drainPending();
			await vi.advanceTimersByTimeAsync(0);

			// The cap is one worker, so the second call must not start another claim
			// while the first worker is busy.
			expect(outboxRepository.claimNextPendingRecord).toHaveBeenCalledTimes(1);

			releaseApply();
			await Promise.all([first, second]);

			expect(applier.apply).toHaveBeenCalledTimes(1);
			expect(reporter.report).toHaveBeenCalledTimes(1);
		});
	});

	describe('parallel drains', () => {
		test('processes up to the configured concurrency in parallel', async () => {
			consumer = createConsumer(true, true, 2);

			// Distinct workflowIds so the per-workflow lifecycle lock never serializes them.
			const r1 = makeRecord({ id: 1, workflowId: 'wf-1' });
			const r2 = makeRecord({ id: 2, workflowId: 'wf-2' });
			const r3 = makeRecord({ id: 3, workflowId: 'wf-3' });
			outboxRepository.claimNextPendingRecord
				.mockResolvedValueOnce(r1)
				.mockResolvedValueOnce(r2)
				.mockResolvedValueOnce(r3)
				.mockResolvedValue(null);

			const started: number[] = [];
			const releases = new Map<number, () => void>();
			const startedSignals = new Map<number, () => void>();
			const startedPromises = new Map<number, Promise<void>>();
			for (const id of [1, 2, 3]) {
				startedPromises.set(id, new Promise<void>((resolve) => startedSignals.set(id, resolve)));
			}
			applier.apply.mockImplementation(async (record) => {
				started.push(record.id);
				startedSignals.get(record.id)!();
				await new Promise<void>((resolve) => releases.set(record.id, resolve));
				return { type: 'completed', triggerStatuses: [] };
			});

			consumer.startPolling();
			const drain = consumer.drainPending();

			// Two workers enter apply() in parallel before either completes; the third waits.
			await Promise.all([startedPromises.get(1), startedPromises.get(2)]);
			expect(started.length).toBe(2);
			expect(started).toEqual(expect.arrayContaining([1, 2]));
			expect(started).not.toContain(3);

			// Freeing one worker lets it claim and start the third record.
			releases.get(1)!();
			await startedPromises.get(3);
			expect(started).toContain(3);

			releases.get(2)!();
			releases.get(3)!();
			await drain;
			expect(reporter.report).toHaveBeenCalledTimes(3);
		});
	});

	describe('worker pool', () => {
		test('a wake-up processes new records while another worker is stuck on a hung record', async () => {
			consumer = createConsumer(true, true, 2);
			const stuck = makeRecord({ id: 1, workflowId: 'wf-stuck' });
			let signalStuckStarted!: () => void;
			const stuckStarted = new Promise<void>((resolve) => {
				signalStuckStarted = resolve;
			});
			applier.apply.mockImplementation(async (record) => {
				if (record.workflowId === 'wf-stuck') {
					signalStuckStarted();
					return await new Promise(() => {});
				}
				return { type: 'completed', triggerStatuses: [] };
			});
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(stuck).mockResolvedValue(null);
			consumer.startPolling();

			void consumer.drainPending();
			await stuckStarted;

			// A new record commits and its wake-up arrives while the first record hangs.
			const fresh = makeRecord({ id: 2, workflowId: 'wf-2' });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(fresh).mockResolvedValue(null);
			void consumer.drainPending();
			await vi.advanceTimersByTimeAsync(0);

			expect(applier.apply).toHaveBeenCalledWith(fresh, expect.anything());
			expect(reporter.report).toHaveBeenCalledWith(
				fresh,
				expect.objectContaining({ type: 'completed' }),
			);
		});

		test('a wake-up arriving while all workers are busy triggers a follow-up pass', async () => {
			const r1 = makeRecord({ id: 1, workflowId: 'wf-1' });
			const r2 = makeRecord({ id: 2, workflowId: 'wf-2' });
			outboxRepository.claimNextPendingRecord
				.mockResolvedValueOnce(r1)
				// The pass runs dry before r2 commits, mirroring a record that lands
				// after the final claim's visibility horizon.
				.mockResolvedValueOnce(null)
				.mockResolvedValueOnce(r2)
				.mockResolvedValue(null);
			let releaseApply!: () => void;
			let signalApplyStarted!: () => void;
			const applyStarted = new Promise<void>((resolve) => {
				signalApplyStarted = resolve;
			});
			applier.apply.mockImplementationOnce(async () => {
				signalApplyStarted();
				await new Promise<void>((resolve) => {
					releaseApply = resolve;
				});
				return { type: 'completed', triggerStatuses: [] };
			});
			consumer.startPolling();

			const first = consumer.drainPending();
			await applyStarted;

			// r2's wake-up arrives while the only worker is still busy.
			const second = consumer.drainPending();
			releaseApply();
			await Promise.all([first, second]);

			expect(applier.apply).toHaveBeenCalledTimes(2);
			expect(applier.apply).toHaveBeenCalledWith(r2, expect.anything());
		});

		test('drainPending rejects with a wrapping error after the pool reported the failure', async () => {
			const error = new Error('claim failed');
			outboxRepository.claimNextPendingRecord.mockRejectedValueOnce(error);
			consumer.startPolling();

			// Awaiting callers (e.g. the reconciler) must still observe drain
			// failures, or they would report success for records nobody claimed.
			// The wrapper lets them fail their own operation; the underlying error
			// is reported by the pool alone, exactly once. `shouldReport: false`
			// pins that a caller's generic catch reporting the wrapper creates no
			// second Sentry event (the ErrorReporter drops non-reportable errors).
			await expect(consumer.drainPending()).rejects.toMatchObject({
				message: expect.stringContaining('drain failed'),
				cause: error,
				shouldReport: false,
			});
			expect(errorReporter.error).toHaveBeenCalledTimes(1);
			expect(errorReporter.error).toHaveBeenCalledWith(error, { shouldBeLogged: true });
		});

		test('an idle pass starts no tracing span', async () => {
			consumer.startPolling();

			await consumer.drainPending();

			expect(outboxRepository.claimNextPendingRecord).toHaveBeenCalledTimes(1);
			expect(tracing.startSpan).not.toHaveBeenCalled();
		});

		test('the poll fallback keeps running while a worker is stuck on a hung record', async () => {
			consumer = createConsumer(true, true, 2);
			const stuck = makeRecord({ id: 1, workflowId: 'wf-stuck' });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(stuck).mockResolvedValue(null);
			applier.apply.mockImplementationOnce(async () => await new Promise(() => {}));
			consumer.startPolling();

			// First cycle claims the record that hangs.
			await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);

			const fresh = makeRecord({ id: 2, workflowId: 'wf-2' });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(fresh).mockResolvedValue(null);
			await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);

			expect(applier.apply).toHaveBeenCalledWith(fresh, expect.anything());
			expect(reporter.report).toHaveBeenCalledWith(
				fresh,
				expect.objectContaining({ type: 'completed' }),
			);
		});
	});

	describe('shutdown', () => {
		test('waits for an in-flight record to finish before resolving', async () => {
			const record = makeRecord({ id: 1 });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(record).mockResolvedValue(null);

			let releaseApply!: () => void;
			let signalApplyStarted!: () => void;
			const applyStarted = new Promise<void>((resolve) => {
				signalApplyStarted = resolve;
			});
			applier.apply.mockImplementationOnce(async () => {
				signalApplyStarted();
				await new Promise<void>((resolve) => {
					releaseApply = resolve;
				});
				return { type: 'completed', triggerStatuses: [] };
			});

			consumer.startPolling();
			const drain = consumer.drainPending();
			// Wait until the record is actually being applied, regardless of how many
			// internal async hops (e.g. lifecycle-lock acquisition) precede the call.
			await applyStarted;

			let shutdownResolved = false;
			const shutdown = consumer.shutdown().then(() => {
				shutdownResolved = true;
			});

			// Shutdown must not resolve while the record is still being applied.
			await Promise.resolve();
			expect(shutdownResolved).toBe(false);

			releaseApply();
			await shutdown;
			await drain;

			expect(shutdownResolved).toBe(true);
			expect(reporter.report).toHaveBeenCalledTimes(1);
		});
	});

	describe('processRecord', () => {
		test('applies the record then reports the result', async () => {
			const record = makeRecord();
			const result: PublicationResult = { type: 'completed', triggerStatuses: [] };
			applier.apply.mockResolvedValue(result);

			await consumer.processRecord(record, abortSignal);

			expect(applier.apply).toHaveBeenCalledWith(
				record,
				expect.objectContaining({ signal: abortSignal }),
			);
			expect(reporter.report).toHaveBeenCalledWith(record, result);
		});

		test('reports a failed result when the applier throws unexpectedly', async () => {
			applier.apply.mockRejectedValue(new Error('teardown failed'));

			await consumer.processRecord(makeRecord(), abortSignal);

			expect(reporter.report).toHaveBeenCalledWith(
				expect.anything(),
				expect.objectContaining({
					type: 'failed',
					error: expect.objectContaining({ message: 'Unexpected: teardown failed' }),
				}),
			);
		});

		test('reports a UserError from the applier as-is, without the Unexpected wrapper', async () => {
			const userError = new UserError('Credential with ID "c-1" does not exist');
			applier.apply.mockRejectedValue(userError);

			await consumer.processRecord(makeRecord(), abortSignal);

			expect(reporter.report).toHaveBeenCalledWith(
				expect.anything(),
				expect.objectContaining({ type: 'failed', error: userError }),
			);
		});

		test('logs but swallows a reporter failure, leaving the record for retry', async () => {
			const reportError = new Error('db write failed');
			reporter.report.mockRejectedValue(reportError);

			await expect(consumer.processRecord(makeRecord(), abortSignal)).resolves.toBeUndefined();

			expect(errorReporter.error).toHaveBeenCalledWith(reportError, { shouldBeLogged: true });
		});

		test('returns the record to the queue without applying it when no longer leader', async () => {
			consumer = createConsumer(true, false);
			const record = makeRecord({ id: 7, workflowId: 'wf-7' });

			await consumer.processRecord(record, abortSignal);

			expect(outboxRepository.returnToPending).toHaveBeenCalledWith(7);
			expect(applier.apply).not.toHaveBeenCalled();
			expect(reporter.report).not.toHaveBeenCalled();
		});
	});

	describe('abort and abandon', () => {
		test('abandons a record that ignores the abort signal and keeps the drain alive', async () => {
			const stuck = makeRecord({ id: 1, workflowId: 'wf-stuck' });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(stuck).mockResolvedValue(null);
			applier.apply.mockImplementationOnce(async () => await new Promise(() => {}));
			consumer.startPolling();

			const drain = consumer.drainPending();
			await vi.advanceTimersByTimeAsync(ABORT_AFTER_MS + ABANDON_GRACE_MS);

			// The drain settles without the stuck record; no terminal status is
			// written for it (it stays in_progress for lease reclaim).
			await drain;
			expect(reporter.report).not.toHaveBeenCalled();
			expect(errorReporter.error).toHaveBeenCalledWith(
				expect.objectContaining({
					message: expect.stringContaining('Abandoned workflow publication outbox record'),
				}),
				{ shouldBeLogged: true },
			);

			// A fresh drain still processes new records.
			const next = makeRecord({ id: 2, workflowId: 'wf-2' });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(next).mockResolvedValue(null);
			await consumer.drainPending();
			expect(reporter.report).toHaveBeenCalledTimes(1);
			expect(reporter.report).toHaveBeenCalledWith(
				next,
				expect.objectContaining({ type: 'completed' }),
			);
		});

		test('a record that settles during the grace period still gets its terminal status', async () => {
			const record = makeRecord({ id: 1 });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(record).mockResolvedValue(null);
			// Settles after the abort deadline but within the grace period.
			applier.apply.mockImplementationOnce(
				async () =>
					await new Promise((resolve) =>
						setTimeout(
							() => resolve({ type: 'completed', triggerStatuses: [] }),
							ABORT_AFTER_MS + 1000,
						),
					),
			);
			consumer.startPolling();

			const drain = consumer.drainPending();
			await vi.advanceTimersByTimeAsync(ABORT_AFTER_MS + 2000);

			await drain;
			expect(reporter.report).toHaveBeenCalledWith(
				record,
				expect.objectContaining({ type: 'completed' }),
			);
		});

		test('an abort honored by the applier still writes a terminal failed status', async () => {
			const record = makeRecord({ id: 1 });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(record).mockResolvedValue(null);
			applier.apply.mockImplementationOnce(
				async (_record, abort) =>
					await new Promise((_resolve, reject) => {
						abort?.signal.addEventListener('abort', () => reject(abort.signal.reason));
					}),
			);
			consumer.startPolling();

			const drain = consumer.drainPending();
			await vi.advanceTimersByTimeAsync(ABORT_AFTER_MS);

			await drain;
			expect(reporter.report).toHaveBeenCalledWith(
				record,
				expect.objectContaining({
					type: 'failed',
					error: expect.objectContaining({
						message: expect.stringContaining('exceeded its deadline'),
					}),
				}),
			);
		});

		test('holds the workflow lock until abandoned trigger operations settle', async () => {
			const record = makeRecord({ id: 1, workflowId: 'wf-1' });
			let releaseDetached!: () => void;
			const detachedWork = new Promise<void>((resolve) => {
				releaseDetached = resolve;
			});
			applier.apply.mockImplementationOnce(async (_record, abort) => {
				abort?.onDetached(detachedWork);
				return { type: 'failed', error: new Error('deadline') };
			});
			const controller = new AbortController();

			const processing = consumer.processRecord(record, controller.signal);
			await vi.advanceTimersByTimeAsync(0);

			// The terminal status is written, but the lock is still held for the orphan.
			expect(reporter.report).toHaveBeenCalledWith(
				record,
				expect.objectContaining({ type: 'failed' }),
			);
			expect(lifecycleLock.isLocked('wf-1')).toBe(true);

			releaseDetached();
			await processing;
			expect(lifecycleLock.isLocked('wf-1')).toBe(false);
		});

		test('leaves an aborted record in progress for lease reclaim instead of applying it', async () => {
			const record = makeRecord({ id: 7, workflowId: 'wf-7' });
			const controller = new AbortController();
			controller.abort();

			await consumer.processRecord(record, controller.signal);

			// Not returned to pending: the wait for the lock may have outlived the
			// lease, and flipping the row would release a newer claimant's claim.
			expect(outboxRepository.returnToPending).not.toHaveBeenCalled();
			expect(applier.apply).not.toHaveBeenCalled();
			expect(reporter.report).not.toHaveBeenCalled();
		});

		test('scales the abandon grace down for short leases so abandonment stays within the lease', async () => {
			// Lease 16s: abort fires at 11.2s, grace is capped at 4s (a quarter of
			// the lease) instead of the fixed 10s, so abandonment lands at 15.2s —
			// still inside the lease.
			consumer = createConsumer(true, true, 1, 16);
			const stuck = makeRecord({ id: 1 });
			outboxRepository.claimNextPendingRecord.mockResolvedValueOnce(stuck).mockResolvedValue(null);
			applier.apply.mockImplementationOnce(async () => await new Promise(() => {}));
			consumer.startPolling();

			const drain = consumer.drainPending();
			await vi.advanceTimersByTimeAsync(15_200);

			await drain;
			expect(errorReporter.error).toHaveBeenCalledWith(
				expect.objectContaining({
					message: expect.stringContaining('Abandoned workflow publication outbox record'),
				}),
				{ shouldBeLogged: true },
			);
		});
	});

	describe('metrics events', () => {
		function lastOutcome() {
			const calls = eventService.emit.mock.calls.filter(
				(call) => call[0] === 'workflow-publication-outbox-record-processed',
			);
			return calls.at(-1)?.[1] as
				| { result: string; reason: string; durationMs: number }
				| undefined;
		}

		test.each([
			[{ type: 'completed', triggerStatuses: [] }, 'published', 'none'],
			[{ type: 'unpublished' }, 'unpublished', 'none'],
			[{ type: 'skipped', reason: 'workflow-not-found' }, 'skipped', 'workflow_not_found'],
			[{ type: 'version-missing' }, 'failed', 'version_missing'],
			[{ type: 'partial', triggerStatuses: [] }, 'partial_success', 'none'],
			[{ type: 'failed', error: new Error('boom') }, 'failed', 'none'],
		] as Array<[PublicationResult, string, string]>)(
			'emits result=%o as result=%s reason=%s',
			async (result, expectedResult, expectedReason) => {
				applier.apply.mockResolvedValue(result);

				await consumer.processRecord(makeRecord(), abortSignal);

				expect(lastOutcome()).toEqual(
					expect.objectContaining({ result: expectedResult, reason: expectedReason }),
				);
			},
		);

		test('emits a failed outcome when the reporter throws', async () => {
			applier.apply.mockResolvedValue({ type: 'completed', triggerStatuses: [] });
			reporter.report.mockRejectedValue(new Error('db write failed'));

			await consumer.processRecord(makeRecord(), abortSignal);

			expect(lastOutcome()).toEqual(expect.objectContaining({ result: 'failed', reason: 'none' }));
		});

		test('does not emit when the record is returned to the queue (no longer leader)', async () => {
			consumer = createConsumer(true, false);

			await consumer.processRecord(makeRecord(), abortSignal);

			expect(eventService.emit).not.toHaveBeenCalledWith(
				'workflow-publication-outbox-record-processed',
				expect.anything(),
			);
		});
	});

	describe('wakeUp', () => {
		test('starts polling and drains all pending records', async () => {
			const first = makeRecord({ id: 1 });
			const second = makeRecord({ id: 2 });
			outboxRepository.claimNextPendingRecord
				.mockResolvedValueOnce(first)
				.mockResolvedValueOnce(second)
				.mockResolvedValue(null);

			await consumer.wakeUp();

			expect(applier.apply).toHaveBeenCalledTimes(2);
			expect(reporter.report).toHaveBeenCalledTimes(2);
			expect(vi.getTimerCount()).toBe(1);
		});

		test('does nothing when the feature flag is off', async () => {
			consumer = createConsumer(false);

			await consumer.wakeUp();

			expect(outboxRepository.claimNextPendingRecord).not.toHaveBeenCalled();
			expect(vi.getTimerCount()).toBe(0);
		});

		test('on a follower, neither polls nor claims a record', async () => {
			outboxRepository.claimNextPendingRecord.mockResolvedValue(makeRecord({ id: 1 }));
			consumer = createConsumer(true, false);

			await consumer.wakeUp();

			expect(outboxRepository.claimNextPendingRecord).not.toHaveBeenCalled();
			expect(vi.getTimerCount()).toBe(0);
		});

		test('never rejects on a drain failure, which is reported exactly once', async () => {
			const error = new Error('claim failed');
			outboxRepository.claimNextPendingRecord.mockRejectedValueOnce(error);

			// Pubsub dispatch drops handler rejections, so a rejecting wakeUp would
			// surface as an unhandled promise rejection.
			await expect(consumer.wakeUp()).resolves.toBeUndefined();

			expect(errorReporter.error).toHaveBeenCalledTimes(1);
			expect(errorReporter.error).toHaveBeenCalledWith(error, { shouldBeLogged: true });
		});

		test('overlapping wake-ups report a shared drain failure exactly once', async () => {
			const error = new Error('claim failed');
			let rejectClaim!: (reason: Error) => void;
			const claimGate = new Promise<WorkflowPublicationOutbox | null>((_resolve, reject) => {
				rejectClaim = reject;
			});
			outboxRepository.claimNextPendingRecord.mockImplementationOnce(async () => await claimGate);

			const first = consumer.wakeUp();
			const second = consumer.wakeUp();
			rejectClaim(error);
			await Promise.all([first, second]);

			expect(errorReporter.error).toHaveBeenCalledTimes(1);
			expect(errorReporter.error).toHaveBeenCalledWith(error, { shouldBeLogged: true });
		});
	});

	describe('poll cycle', () => {
		test('drains all pending records through apply + report', async () => {
			const first = makeRecord({ id: 1 });
			const second = makeRecord({ id: 2 });
			outboxRepository.claimNextPendingRecord
				.mockResolvedValueOnce(first)
				.mockResolvedValueOnce(second)
				.mockResolvedValue(null);
			consumer.startPolling();

			await vi.advanceTimersByTimeAsync(POLL_INTERVAL_MS);

			expect(applier.apply).toHaveBeenCalledTimes(2);
			expect(reporter.report).toHaveBeenCalledTimes(2);
		});
	});
});