import type { InstanceAiEvent } from '@n8n/api-types';
import type { Logger } from '@n8n/backend-common';
import { Time } from '@n8n/constants';
import {
	orchestratorAgentId,
	type InstanceAiLivenessPolicy,
	type InstanceAiLivenessTimeoutReason,
} from '@n8n/instance-ai';
import { getErrorMessage } from '@n8n/utils/errors/get-error-message';

import type { InstanceAiRunTimeoutDetails } from '../run-timeout-details';

export const INSTANCE_AI_RUN_TIMEOUT_REASON = 'timeout';

const RUN_TIMEOUT_MESSAGE =
	'The run stopped making progress, so I cancelled it. You can retry or adjust the request.';

export type InstanceAiLivenessTimedOutActiveRun = {
	runId: string;
	abortController: AbortController;
};

export type InstanceAiLivenessTimedOutSuspendedRun = {
	runId: string;
	threadId: string;
	abortController: AbortController;
};

export type InstanceAiLivenessTimedOutTask = {
	threadId: string;
	taskId: string;
	role: string;
	timeoutReason?: InstanceAiLivenessTimeoutReason;
};

export type InstanceAiLivenessPendingConfirmation = {
	threadId: string;
};

export type InstanceAiLivenessSweepResult = {
	activeThreadIds: string[];
	suspendedThreadIds: string[];
	confirmationRequestIds: string[];
	activeTimeouts?: Record<string, InstanceAiRunTimeoutDetails>;
	suspendedTimeouts?: Record<string, InstanceAiRunTimeoutDetails>;
	confirmationTimeouts?: Record<string, InstanceAiRunTimeoutDetails>;
};

export type InstanceAiConsumedRunTimeout = {
	timedOut: boolean;
	details?: InstanceAiRunTimeoutDetails;
};

export type InstanceAiLivenessRunState<
	TSuspendedRun extends InstanceAiLivenessTimedOutSuspendedRun,
> = {
	sweepTimedOut: (policy: InstanceAiLivenessPolicy, now?: number) => InstanceAiLivenessSweepResult;
	cancelActiveRun: (threadId: string) => InstanceAiLivenessTimedOutActiveRun | undefined;
	cancelSuspendedRun: (threadId: string) => TSuspendedRun | undefined;
	getActiveRunId: (threadId: string) => string | undefined;
	getPendingConfirmation: (requestId: string) => InstanceAiLivenessPendingConfirmation | undefined;
	hasPendingConfirmationForThread: (threadId: string) => boolean;
	rejectPendingConfirmation: (requestId: string) => boolean;
};

export type InstanceAiLivenessBackgroundTasks = {
	timeoutTimedOutTasks: (
		policy: InstanceAiLivenessPolicy,
		now?: number,
		options?: {
			shouldSkipTask?: (task: InstanceAiLivenessTimedOutTask) => boolean;
		},
	) => Promise<InstanceAiLivenessTimedOutTask[]>;
};

export type InstanceAiLivenessEventBus = {
	publish: (threadId: string, event: InstanceAiEvent) => void;
};

export type InstanceAiLivenessServiceOptions<
	TSuspendedRun extends InstanceAiLivenessTimedOutSuspendedRun,
> = {
	policy: InstanceAiLivenessPolicy;
	backgroundTaskIdleTimeoutMs: number;
	runState: InstanceAiLivenessRunState<TSuspendedRun>;
	backgroundTasks: InstanceAiLivenessBackgroundTasks;
	eventBus: InstanceAiLivenessEventBus;
	finalizeCancelledSuspendedRun: (suspended: TSuspendedRun, reason: string) => void;
	onPendingConfirmationRejected?: (requestId: string) => void;
	logger: Logger;
};

export class InstanceAiLivenessService<
	TSuspendedRun extends InstanceAiLivenessTimedOutSuspendedRun,
> {
	private timeoutInterval?: NodeJS.Timeout;

	private readonly timedOutRunIds = new Map<string, InstanceAiRunTimeoutDetails | undefined>();

	private readonly timedOutActiveRunThreads = new Set<string>();

	/**
	 * Runs whose timeout notice has already been published, so a run cancelled
	 * by the sweep and then finalised by its own abort handler only gets the
	 * notice once. Capped FIFO (see {@link NOTICE_DEDUPE_CACHE_SIZE}): dedupe
	 * only has to hold across a single run's cancellation window, and a run
	 * lives on one main for one process lifetime.
	 */
	private readonly noticedRunIds = new Set<string>();

	/** Max retained run ids in {@link noticedRunIds}; oldest evicted first. */
	static readonly NOTICE_DEDUPE_CACHE_SIZE = 1000;

	constructor(private readonly options: InstanceAiLivenessServiceOptions<TSuspendedRun>) {}

	get backgroundTaskIdleTimeoutMs(): number {
		return this.options.backgroundTaskIdleTimeoutMs;
	}

	start(): void {
		if (!this.options.policy.hasEnabledTimeouts()) return;

		this.timeoutInterval = setInterval(() => {
			void this.sweepTimedOutWork();
		}, Time.minutes.toMilliseconds);
	}

	shutdown(): void {
		if (this.timeoutInterval) {
			clearInterval(this.timeoutInterval);
			this.timeoutInterval = undefined;
		}
		this.timedOutRunIds.clear();
		this.timedOutActiveRunThreads.clear();
		this.noticedRunIds.clear();
	}

	clearThreadState(threadId: string): void {
		this.timedOutActiveRunThreads.delete(threadId);
	}

	markRunTimedOut(runId: string, details?: InstanceAiRunTimeoutDetails): void {
		this.timedOutRunIds.set(runId, details);
	}

	consumeRunTimedOut(runId: string): boolean {
		const timedOut = this.timedOutRunIds.has(runId);
		if (timedOut) this.timedOutRunIds.delete(runId);
		return timedOut;
	}

	consumeRunTimeout(runId: string): InstanceAiConsumedRunTimeout {
		const timedOut = this.timedOutRunIds.has(runId);
		if (!timedOut) return { timedOut: false };

		const details = this.timedOutRunIds.get(runId);
		this.timedOutRunIds.delete(runId);
		return details ? { timedOut: true, details } : { timedOut: true };
	}

	hasTimedOutActiveRunThread(threadId: string): boolean {
		return this.timedOutActiveRunThreads.has(threadId);
	}

	async sweepTimedOutWork(now = Date.now()): Promise<void> {
		const {
			activeThreadIds,
			suspendedThreadIds,
			confirmationRequestIds,
			activeTimeouts,
			suspendedTimeouts,
			confirmationTimeouts,
		} = this.options.runState.sweepTimedOut(this.options.policy, now);

		for (const threadId of activeThreadIds) {
			this.options.logger.debug('Cancelling timed-out active run', { threadId });
			this.cancelTimedOutActiveRun(threadId, activeTimeouts?.[threadId]);
		}

		for (const threadId of suspendedThreadIds) {
			this.options.logger.debug('Auto-rejecting timed-out suspended run', { threadId });
			this.cancelTimedOutSuspendedRun(threadId, suspendedTimeouts?.[threadId]);
		}

		for (const reqId of confirmationRequestIds) {
			this.options.logger.debug('Auto-rejecting timed-out sub-agent confirmation', {
				requestId: reqId,
			});
			const pending = this.options.runState.getPendingConfirmation(reqId);
			if (pending) {
				const runId = this.options.runState.getActiveRunId(pending.threadId);
				if (runId) {
					this.markRunTimedOut(runId, confirmationTimeouts?.[reqId]);
					this.publishRunTimeoutNotice(pending.threadId, runId);
				}
			}
			this.options.runState.rejectPendingConfirmation(reqId);
			this.options.onPendingConfirmationRejected?.(reqId);
		}

		try {
			const timedOutTasks = await this.options.backgroundTasks.timeoutTimedOutTasks(
				this.options.policy,
				now,
				{
					shouldSkipTask: (task) =>
						this.options.runState.hasPendingConfirmationForThread(task.threadId),
				},
			);
			for (const task of timedOutTasks) {
				this.options.logger.debug('Timed out background task', {
					threadId: task.threadId,
					taskId: task.taskId,
					role: task.role,
					timeoutReason: task.timeoutReason,
				});
			}
		} catch (error) {
			this.options.logger.warn('Failed to sweep timed-out background tasks', {
				error: getErrorMessage(error),
			});
		}
	}

	cancelTimedOutActiveRun(threadId: string, details?: InstanceAiRunTimeoutDetails): void {
		const active = this.options.runState.cancelActiveRun(threadId);
		if (!active) return;

		this.markRunTimedOut(active.runId, details);
		this.timedOutActiveRunThreads.add(threadId);
		this.publishRunTimeoutNotice(threadId, active.runId);
		active.abortController.abort();
	}

	cancelTimedOutSuspendedRun(threadId: string, details?: InstanceAiRunTimeoutDetails): void {
		const suspended = this.options.runState.cancelSuspendedRun(threadId);
		if (!suspended) return;

		this.markRunTimedOut(suspended.runId, details);
		suspended.abortController.abort();
		this.options.finalizeCancelledSuspendedRun(suspended, INSTANCE_AI_RUN_TIMEOUT_REASON);
	}

	publishRunTimeoutNotice(threadId: string, runId: string): void {
		// Deduped locally rather than by scanning the run's events: the notice is
		// a delta, and under the durable log deltas are ephemeral (never stored,
		// persisted only once their segment coalesces), so a read-back cannot see
		// a notice published moments earlier.
		if (this.noticedRunIds.has(runId)) return;

		// Recorded only after the publish lands: marking first would suppress the
		// notice permanently if publish threw.
		this.options.eventBus.publish(threadId, {
			type: 'text-delta',
			runId,
			agentId: orchestratorAgentId(runId),
			responseId: `run-timeout:${runId}`,
			payload: { text: RUN_TIMEOUT_MESSAGE },
		});
		this.rememberNoticedRunId(runId);
	}

	/**
	 * Record a notified run id, evicting the oldest once the cap is reached. A
	 * Set keeps insertion order, so the first value is the oldest entry.
	 */
	private rememberNoticedRunId(runId: string): void {
		this.noticedRunIds.add(runId);
		if (this.noticedRunIds.size > InstanceAiLivenessService.NOTICE_DEDUPE_CACHE_SIZE) {
			const oldest = this.noticedRunIds.values().next().value;
			if (oldest !== undefined) this.noticedRunIds.delete(oldest);
		}
	}
}