import { isObjectLiteral, Logger } from '@n8n/backend-common';
import { GlobalConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
import { ExecutionRepository } from '@n8n/db';
import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
import { Container, Service } from '@n8n/di';
import { ErrorReporter, InstanceSettings } from 'n8n-core';
import { BINARY_ENCODING, sleep, jsonStringify, ensureError, UnexpectedError } from 'n8n-workflow';
import type { IExecuteResponsePromiseData, IRun } from 'n8n-workflow';
import assert, { strict } from 'node:assert';

import { ActiveExecutions } from '@/active-executions';
import { HIGHEST_SHUTDOWN_PRIORITY } from '@/constants';
import { EventService } from '@/events/event.service';
import { assertNever } from '@/utils';

import { JOB_TYPE_NAME, QUEUE_NAME } from './constants';
import { JobProcessor } from './job-processor';
import type {
	JobQueue,
	Job,
	JobData,
	JobOptions,
	JobStatus,
	JobId,
	JobFinishedProps,
	QueueRecoveryContext,
	JobMessage,
	JobFailedMessage,
} from './scaling.types';

@Service()
export class ScalingService {
	private queue: JobQueue;

	private jobResults = new Map<string, JobFinishedProps>();

	constructor(
		private readonly logger: Logger,
		private readonly errorReporter: ErrorReporter,
		private readonly activeExecutions: ActiveExecutions,
		private readonly jobProcessor: JobProcessor,
		private readonly globalConfig: GlobalConfig,
		private readonly executionRepository: ExecutionRepository,
		private readonly instanceSettings: InstanceSettings,
		private readonly eventService: EventService,
	) {
		this.logger = this.logger.scoped('scaling');
	}

	// #region Lifecycle

	async setupQueue() {
		const { default: BullQueue } = await import('bull');
		const { RedisClientService } = await import('@/services/redis-client.service');

		if (this.queue) return;

		const service = Container.get(RedisClientService);

		const bullPrefix = this.globalConfig.queue.bull.prefix;
		const prefix = service.toValidPrefix(bullPrefix);

		this.queue = new BullQueue(QUEUE_NAME, {
			prefix,
			settings: { ...this.globalConfig.queue.bull.settings, maxStalledCount: 0 },
			createClient: (type) => service.createClient({ type: `${type}(bull)` }),
		});

		this.registerListeners();

		if (this.instanceSettings.isLeader) this.scheduleQueueRecovery(0);

		this.scheduleQueueMetrics();

		const { McpServer, QueuedExecutionStrategy, RedisSessionStore } = await import(
			'@n8n/n8n-nodes-langchain/mcp/core'
		);
		const { Publisher } = await import('@/scaling/pubsub/publisher.service');

		const publisher = Container.get(Publisher);

		const MCP_SESSION_TTL = 86400;
		const getMcpSessionKey = (sessionId: string) =>
			`${this.globalConfig.redis.prefix}:mcp-session:${sessionId}`;

		const mcpServer = McpServer.instance(this.logger);
		const redisStore = new RedisSessionStore(
			{
				set: async (key, value, ttl) => await publisher.set(key, value, ttl),
				get: async (key) => await publisher.get(key),
				clear: async (key) => await publisher.clear(key),
			},
			getMcpSessionKey,
			MCP_SESSION_TTL,
		);

		mcpServer.setSessionStore(redisStore);
		mcpServer.setExecutionStrategy(new QueuedExecutionStrategy(mcpServer.getPendingCallsManager()));

		this.logger.debug('Queue setup completed');
	}

	setupWorker(concurrency: number) {
		this.assertWorker();
		this.assertQueue();

		void this.queue.process(JOB_TYPE_NAME, concurrency, async (job: Job) => {
			try {
				this.eventService.emit('job-dequeued', {
					executionId: job.data.executionId,
					workflowId: job.data.workflowId,
					hostId: this.instanceSettings.hostId,
					jobId: job.id.toString(),
				});

				if (!this.hasValidJobData(job)) {
					throw new UnexpectedError('Worker received invalid job', {
						extra: { jobData: jsonStringify(job, { replaceCircularRefs: true }) },
					});
				}

				await this.jobProcessor.processJob(job);
			} catch (error) {
				await this.reportJobProcessingError(ensureError(error), job);
			}
		});

		this.logger.debug('Worker setup completed');
	}

	private async reportJobProcessingError(error: Error, job: Job) {
		const { executionId } = job.data;

		this.logger.error(`Worker errored while running execution ${executionId} (job ${job.id})`, {
			error,
			executionId,
			jobId: job.id,
		});

		const msg: JobFailedMessage = {
			kind: 'job-failed',
			executionId,
			workerId: this.instanceSettings.hostId,
			errorMsg: error.message,
			errorStack: error.stack ?? '',
		};

		await job.progress(msg);

		this.errorReporter.error(error, { executionId });

		throw error;
	}

	@OnShutdown(HIGHEST_SHUTDOWN_PRIORITY)
	async stop() {
		const { instanceType } = this.instanceSettings;

		if (instanceType === 'main') await this.stopMain();
		else if (instanceType === 'worker') await this.stopWorker();
	}

	private async pauseQueue() {
		await this.queue.pause(true, true); // no more jobs will be enqueued or picked up
		this.logger.debug('Paused queue');
	}

	private async stopMain() {
		if (this.instanceSettings.isSingleMain) await this.pauseQueue();

		if (this.queueRecoveryContext.timeout) this.stopQueueRecovery();
		if (this.isQueueMetricsEnabled) this.stopQueueMetrics();
	}

	private async stopWorker() {
		await this.pauseQueue();

		let count = 0;

		while (this.getRunningJobsCount() !== 0) {
			if (count++ % 4 === 0) {
				this.logger.info(
					`Waiting for ${this.getRunningJobsCount()} active executions to finish...`,
				);
			}

			await sleep(500);
		}
	}

	async pingQueue() {
		await this.queue.client.ping();
	}

	// #endregion

	// #region Jobs

	/** Get and remove the result for a completed job. */
	popJobResult(executionId: string): JobFinishedProps | undefined {
		const result = this.jobResults.get(executionId);
		this.jobResults.delete(executionId);
		return result;
	}

	async getPendingJobCounts() {
		const { active, waiting } = await this.queue.getJobCounts();

		return { active, waiting };
	}

	/**
	 * Add a job to the queue.
	 *
	 * @param jobData Data of the job to add to the queue.
	 * @param priority Priority of the job, from `1` (highest) to `MAX_SAFE_INTEGER` (lowest).
	 */
	async addJob(jobData: JobData, { priority }: { priority: number }) {
		strict(priority > 0 && priority <= Number.MAX_SAFE_INTEGER);

		const { keepLastCompleted, keepLastFailed } = this.globalConfig.executions.queueRetention;

		const jobOptions: JobOptions = {
			priority,
			removeOnComplete: keepLastCompleted,
			removeOnFail: keepLastFailed,
		};

		const job = await this.queue.add(JOB_TYPE_NAME, jobData, jobOptions);

		const { executionId } = jobData;
		const jobId = job.id;

		this.logger.info(`Enqueued execution ${executionId} (job ${jobId})`, { executionId, jobId });
		this.eventService.emit('job-enqueued', {
			executionId,
			workflowId: jobData.workflowId,
			hostId: this.instanceSettings.hostId,
			jobId: jobId.toString(),
		});

		return job;
	}

	async getJob(jobId: JobId) {
		return await this.queue.getJob(jobId);
	}

	async findJobsByStatus(statuses: JobStatus[]) {
		const jobs = await this.queue.getJobs(statuses);

		return jobs.filter((job) => job !== null);
	}

	async stopJob(job: Job) {
		const props = { jobId: job.id, executionId: job.data.executionId };

		try {
			if (await job.isActive()) {
				await job.progress({ kind: 'abort-job' }); // being processed by worker
				this.logger.debug('Sent abort signal to worker', props);
				return true;
			}

			await job.remove(); // not yet picked up, or waiting for next pickup (stalled)
			this.logger.debug('Stopped inactive job', props);
			return true;
		} catch (error: unknown) {
			assert(error instanceof Error);
			this.logger.error('Failed to stop job', {
				...props,
				error: {
					message: error.message,
					name: error.name,
					stack: error.stack,
				},
			});
			return false;
		}
	}

	getRunningJobsCount() {
		return this.jobProcessor.getRunningJobIds().length;
	}

	// #endregion

	// #region Listeners

	private registerListeners() {
		const { instanceType } = this.instanceSettings;
		if (instanceType === 'main' || instanceType === 'webhook') {
			this.registerMainOrWebhookListeners();
		} else if (instanceType === 'worker') {
			this.registerWorkerListeners();
		}
	}

	/**
	 * Register listeners on a `worker` process for Bull queue events.
	 */
	private registerWorkerListeners() {
		this.queue.on('global:progress', (jobId: JobId, msg: unknown) => {
			if (!this.isJobMessage(msg)) return;

			if (msg.kind === 'abort-job') this.jobProcessor.stopJob(jobId);
		});

		this.queue.on('error', (error: Error) => {
			if ('code' in error && error.code === 'ECONNREFUSED') return; // handled by RedisClientService.retryStrategy

			/**
			 * Non-recoverable error on worker start with Redis unavailable.
			 * Even if Redis recovers, worker will remain unable to process jobs.
			 */
			if (error.message.includes('Error initializing Lua scripts')) {
				this.logger.error('Fatal error initializing worker', { error });
				this.logger.error('Exiting process...');
				process.exit(1);
			}

			this.logger.error('Queue errored', { error });

			throw error;
		});
	}

	/**
	 * Register listeners on a `main` or `webhook` process for Bull queue events.
	 */
	private registerMainOrWebhookListeners() {
		this.queue.on('error', (error: Error) => {
			if ('code' in error && error.code === 'ECONNREFUSED') return; // handled by RedisClientService.retryStrategy

			this.logger.error('Queue errored', { error });

			throw error;
		});

		this.queue.on('global:progress', (jobId: JobId, msg: unknown) => {
			if (!this.isJobMessage(msg)) return;

			// completion and failure are reported via `global:progress` to convey more details
			// than natively provided by Bull in `global:completed` and `global:failed` events

			switch (msg.kind) {
				case 'send-chunk':
					this.activeExecutions.sendChunk(msg.executionId, msg.chunkText);
					break;
				case 'respond-to-webhook':
					const decodedResponse = this.decodeWebhookResponse(msg.response);
					this.activeExecutions.resolveResponsePromise(msg.executionId, decodedResponse);
					break;
				case 'job-finished':
					if (msg.success) {
						this.activeExecutions.resolveResponsePromise(msg.executionId, {});
					} else {
						this.activeExecutions.resolveResponsePromise(msg.executionId, {
							body: {
								message: 'Workflow execution failed',
							},
							statusCode: 500,
						});
					}

					/**
					 * We track the result received via `job-finished` message,
					 * because `removeOnComplete: true` prevents `job.finished()`
					 * from returning a value that is no longer in Redis.
					 */
					if (msg.version === 2) {
						this.jobResults.set(msg.executionId, {
							success: msg.success,
							error: msg.error,
							status: msg.status,
							lastNodeExecuted: msg.lastNodeExecuted,
							usedDynamicCredentials: msg.usedDynamicCredentials,
							metadata: msg.metadata,
							startedAt: new Date(msg.startedAt),
							stoppedAt: new Date(msg.stoppedAt),
						});
					}

					this.logger.info(`Execution ${msg.executionId} (job ${jobId}) finished`, {
						workerId: msg.workerId,
						executionId: msg.executionId,
						jobId,
						success: msg.success,
					});
					break;
				case 'job-failed':
					this.logger.error(
						[
							`Execution ${msg.executionId} (job ${jobId}) failed`,
							msg.errorStack ? `\n${msg.errorStack}\n` : '',
						].join(''),
						{
							workerId: msg.workerId,
							errorMsg: msg.errorMsg,
							executionId: msg.executionId,
							jobId,
						},
					);
					break;
				case 'abort-job':
					break; // only for worker
				case 'mcp-response':
					// Route to appropriate MCP handler based on type
					// All mains receive this; only the one with the session/pending response will handle it
					void this.handleMcpResponse(
						msg.executionId,
						msg.mcpType,
						msg.sessionId,
						msg.messageId,
						msg.response,
					);
					break;
				default:
					assertNever(msg);
			}
		});

		if (this.isQueueMetricsEnabled) {
			this.queue.on('global:completed', () => this.jobCounters.completed++);
			this.queue.on('global:failed', () => this.jobCounters.failed++);
		}
	}

	/** Whether the argument is a message sent via Bull's internal pubsub setup. */
	private isJobMessage(candidate: unknown): candidate is JobMessage {
		return typeof candidate === 'object' && candidate !== null && 'kind' in candidate;
	}

	/**
	 * Handle MCP response from worker - forward to appropriate MCP handler.
	 * For MCP Service: fetches execution data from DB and forwards IRun.
	 * For MCP Trigger: forwards the response directly (tool result from sendResponse hook).
	 */
	private async handleMcpResponse(
		executionId: string,
		mcpType: 'service' | 'trigger',
		sessionId: string,
		messageId: string,
		response: unknown,
	): Promise<void> {
		try {
			if (mcpType === 'service') {
				// For MCP Service, fetch execution data from DB
				const executionData = await this.executionRepository.findSingleExecution(executionId, {
					includeData: true,
					unflattenData: true,
				});

				if (!executionData) {
					this.logger.error('Execution not found in DB for MCP response', { executionId });
					return;
				}

				// Convert to IRun format
				const runData: IRun = {
					finished: executionData.finished,
					mode: executionData.mode,
					startedAt: executionData.startedAt,
					stoppedAt: executionData.stoppedAt,
					status: executionData.status,
					data: executionData.data,
					storedAt: executionData.storedAt,
				};

				const { McpService } = await import('@/modules/mcp/mcp.service');
				const mcpService = Container.get(McpService);
				mcpService.handleWorkerResponse(executionId, runData);
			} else {
				const { McpServer } = await import('@n8n/n8n-nodes-langchain/mcp/core');
				const mcpServer = McpServer.instance(this.logger);
				mcpServer.handleWorkerResponse(sessionId, messageId, response);
			}
		} catch (error) {
			this.logger.error('Failed to handle MCP response', {
				executionId,
				mcpType,
				error: ensureError(error).message,
			});
		}
	}

	// #endregion

	private decodeWebhookResponse(
		response: IExecuteResponsePromiseData,
	): IExecuteResponsePromiseData {
		if (
			typeof response === 'object' &&
			typeof response.body === 'object' &&
			response.body !== null &&
			'__@N8nEncodedBuffer@__' in response.body &&
			typeof response.body['__@N8nEncodedBuffer@__'] === 'string'
		) {
			response.body = Buffer.from(response.body['__@N8nEncodedBuffer@__'], BINARY_ENCODING);
		}

		return response;
	}

	private assertQueue() {
		if (this.queue) return;

		throw new UnexpectedError('This method must be called after `setupQueue`');
	}

	private assertWorker() {
		if (this.instanceSettings.instanceType === 'worker') return;

		throw new UnexpectedError('This method must be called on a `worker` instance');
	}

	// #region Queue metrics

	/** Counters for completed and failed jobs, reset on each interval tick. */
	private readonly jobCounters = { completed: 0, failed: 0 };

	/** Interval for collecting queue metrics to expose via Prometheus. */
	private queueMetricsInterval: NodeJS.Timeout | undefined;

	get isQueueMetricsEnabled() {
		return (
			this.globalConfig.endpoints.metrics.includeQueueMetrics &&
			this.instanceSettings.instanceType === 'main'
		);
	}

	/** Set up an interval to collect queue metrics and emit them in an event. */
	private scheduleQueueMetrics() {
		if (!this.isQueueMetricsEnabled || this.queueMetricsInterval) return;

		this.queueMetricsInterval = setInterval(async () => {
			const pendingJobCounts = await this.getPendingJobCounts();

			this.eventService.emit('job-counts-updated', {
				...pendingJobCounts, // active, waiting
				...this.jobCounters, // completed, failed
			});

			this.jobCounters.completed = 0;
			this.jobCounters.failed = 0;
		}, this.globalConfig.endpoints.metrics.queueMetricsInterval * Time.seconds.toMilliseconds);
	}

	/** Stop collecting queue metrics. */
	private stopQueueMetrics() {
		if (this.queueMetricsInterval) {
			clearInterval(this.queueMetricsInterval);
			this.queueMetricsInterval = undefined;

			this.logger.debug('Queue metrics collection stopped');
		}
	}

	// #endregion

	// #region Queue recovery

	private readonly queueRecoveryContext: QueueRecoveryContext = {
		batchSize: this.globalConfig.executions.queueRecovery.batchSize,
		waitMs: this.globalConfig.executions.queueRecovery.interval * 60 * 1000,
	};

	@OnLeaderTakeover()
	private scheduleQueueRecovery(waitMs = this.queueRecoveryContext.waitMs) {
		this.queueRecoveryContext.timeout = setTimeout(async () => {
			try {
				const nextWaitMs = await this.recoverFromQueue();
				this.scheduleQueueRecovery(nextWaitMs);
			} catch (error) {
				this.logger.error('Failed to recover dangling executions from queue', {
					msg: this.toErrorMsg(error),
				});
				this.logger.error('Retrying...');

				this.scheduleQueueRecovery();
			}
		}, waitMs);

		if (waitMs === 0) return;

		const wait = [this.queueRecoveryContext.waitMs / Time.minutes.toMilliseconds, 'min'].join(' ');

		this.logger.debug(`Scheduled queue recovery check for next ${wait}`);
	}

	@OnLeaderStepdown()
	private stopQueueRecovery() {
		if (!this.queueRecoveryContext.timeout) return;

		clearTimeout(this.queueRecoveryContext.timeout);

		this.logger.debug('Queue recovery stopped');
	}

	/**
	 * Mark in-progress executions as `crashed` if stored in DB as `new` or `running`
	 * but absent from the queue. Return time until next recovery cycle.
	 */
	async recoverFromQueue() {
		const { waitMs, batchSize } = this.queueRecoveryContext;

		const storedIds = await this.executionRepository.getInProgressExecutionIds(batchSize);

		if (storedIds.length === 0) {
			this.logger.debug('Completed queue recovery check, no dangling executions');
			return waitMs;
		}

		const runningJobs = await this.findJobsByStatus(['active', 'waiting']);
		const queuedIds = new Set(runningJobs.map((job) => job.data.executionId));

		const danglingIds = storedIds.filter((id) => !queuedIds.has(id));
		if (danglingIds.length === 0) {
			this.logger.debug('Completed queue recovery check, no dangling executions');
			return waitMs;
		}

		await this.executionRepository.markAsCrashed(danglingIds);

		this.logger.info('Completed queue recovery check, recovered dangling executions', {
			danglingIds,
		});

		// if this cycle used up the whole batch size, it is possible for there to be
		// dangling executions outside this check, so speed up next cycle

		return storedIds.length >= this.queueRecoveryContext.batchSize ? waitMs / 2 : waitMs;
	}

	private toErrorMsg(error: unknown) {
		return error instanceof Error
			? error.message
			: jsonStringify(error, { replaceCircularRefs: true });
	}

	private hasValidJobData(job: Job) {
		return isObjectLiteral(job.data) && 'executionId' in job.data && 'loadStaticData' in job.data;
	}

	// #endregion
}