import {
	AgentCredentialIntegrationConfig,
	isAgentCredentialIntegration,
	type AgentIntegrationSettings,
	type AgentIntegrationStatusResponse,
} from '@n8n/api-types';
import { Logger } from '@n8n/backend-common';
import { GlobalConfig } from '@n8n/config';
import type { User } from '@n8n/db';
import { ProjectRelationRepository, UserRepository } from '@n8n/db';
import { OnLeaderStepdown, OnLeaderTakeover, OnPubSubEvent } from '@n8n/decorators';
import { Container, Service } from '@n8n/di';
import type { Channel, Thread, UserInfo } from 'chat';
import { InstanceSettings } from 'n8n-core';

import { CredentialsFinderService } from '@/credentials/credentials-finder.service';
import { CredentialsService } from '@/credentials/credentials.service';
import { Publisher } from '@/scaling/pubsub/publisher.service';
import type { PubSubCommandMap } from '@/scaling/pubsub/pubsub.event-map';
import { UrlService } from '@/services/url.service';

import { AgentChatBridge } from './agent-chat-bridge';
import {
	ChatIntegrationRegistry,
	type AgentChatIntegrationContext,
} from './agent-chat-integration';
import { ComponentMapper } from './component-mapper';
import { loadChatSdk, loadMemoryState } from './esm-loader';
import { buildIntegrationConnectionId } from './integration-tools';
import type { Agent } from '../entities/agent.entity';
import { AgentRepository } from '../repositories/agent.repository';

// ---------------------------------------------------------------------------
// Chat SDK local interfaces
//
// The `chat` package is ESM-only, so we cannot import types at module level.
// These interfaces mirror the subset of the Chat SDK API we consume.
// ---------------------------------------------------------------------------

type WebhookHandler = (
	request: Request,
	options?: { waitUntil?: (task: Promise<unknown>) => void },
) => Promise<Response>;

export interface ChatInstance {
	initialize(): Promise<void>;
	shutdown(): Promise<void>;
	webhooks: Record<string, WebhookHandler>;
	onNewMention: (handler: unknown) => void;
	onSubscribedMessage: (handler: unknown) => void;
	onAction: (handler: unknown) => void;
	getAdapter(name: string): unknown;
	openDM(user: string): Promise<Thread>;
	thread(threadId: string): Thread;
	channel(channelId: string): Channel;
	getUser(user: string): Promise<UserInfo | null>;
}

interface ChatAgentConnection {
	chat: ChatInstance;
	bridge: AgentChatBridge;
	/**
	 * Context captured at connect time. Used by `disconnectOne` to invoke
	 * `onBeforeDisconnect` hooks with the same decrypted credential the connect
	 * ran with — re-decrypting at disconnect time would fail if the credential
	 * was rotated or deleted in between.
	 */
	context: AgentChatIntegrationContext;
}

interface ConnectOptions {
	skipExternalHooks?: boolean;
	settings?: AgentIntegrationSettings;
}

interface DisconnectOptions {
	/**
	 * Skip integration-defined external teardown.
	 * Mirror of `ConnectOptions.skipExternalHooks` — set true on peer mains
	 * reacting to a PubSub broadcast, in graceful-shutdown paths, and during
	 * leader-stepdown so the cluster-wide remote release happens exactly once.
	 */
	skipExternalHooks?: boolean;
}

/**
 * Manages per-agent Chat SDK instances and their lifecycle.
 *
 * Each integration (e.g. Slack workspace) gets its own `Chat` instance keyed
 * by `agentId:type:credentialId`. This supports multiple integrations per agent
 * (two Slack workspaces, or Slack + Discord in the future).
 */
@Service()
export class ChatIntegrationService {
	private readonly connections = new Map<string, ChatAgentConnection>();

	constructor(
		private readonly logger: Logger,
		private readonly agentRepository: AgentRepository,
		private readonly credentialsService: CredentialsService,
		private readonly credentialsFinderService: CredentialsFinderService,
		private readonly urlService: UrlService,
		private readonly integrationRegistry: ChatIntegrationRegistry,
		private readonly instanceSettings: InstanceSettings,
		private readonly publisher: Publisher,
		private readonly globalConfig: GlobalConfig,
	) {}

	/**
	 * In multi-main mode, broadcast a connect/disconnect change so every other
	 * main reconciles its in-memory `connections` map. Single-instance setups
	 * skip the round-trip.
	 *
	 * The originating main has already applied the change locally before
	 * calling this — only peer mains need to act on the published command.
	 */
	async broadcastIntegrationChange(
		agentId: string,
		integration: AgentCredentialIntegrationConfig,
		action: 'connect' | 'disconnect',
	): Promise<void> {
		if (!this.globalConfig.multiMainSetup.enabled) return;
		try {
			const payload = { agentId, integration, action };
			await this.publisher.publishCommand({
				command: 'agent-chat-integration-changed',
				payload,
			});
		} catch (error) {
			this.logger.warn(
				`[ChatIntegrationService] Failed to publish ${action} for ${integration.type} on agent ${agentId}: ${error instanceof Error ? error.message : String(error)}`,
			);
		}
	}

	private connectionKey(agentId: string, type: string, credentialId: string): string {
		return `${agentId}:${type}:${credentialId}`;
	}

	/** Extract the integration type segment from a `connectionKey()` value. */
	private connectionTypeFromKey(key: string): string | undefined {
		const parts = key.split(':');
		return parts.length >= 3 ? parts[1] : undefined;
	}

	/**
	 * Connect an agent to a chat platform via the Chat SDK.
	 *
	 * Creates a Chat instance with the appropriate adapter, initializes it,
	 * and wires up the AgentChatBridge for event handling.
	 *
	 * `options.skipExternalHooks` skips `onBeforeConnect` and `onAfterConnect`.
	 * These hooks can touch external services and must run exactly once per
	 * cluster — by the originator on a user-initiated connect, or by the leader
	 * on startup. Peer mains reacting to a PubSub broadcast pass `true` so they
	 * only build local runtime state.
	 */
	async connect(
		agentId: string,
		integration: AgentCredentialIntegrationConfig,
		userId: string,
		projectId: string,
		options: ConnectOptions = {},
	): Promise<void> {
		const key = this.connectionKey(agentId, integration.type, integration.credentialId);

		// Tear down existing connection if reconnecting
		if (this.connections.has(key)) {
			await this.disconnectOne(key);
		}

		const integrationImpl = this.integrationRegistry.require(integration.type);

		const user = await this.resolveUser(userId);

		// Decrypt the integration credential to get platform tokens
		const decryptedData = await this.decryptCredential(integration.credentialId, user);

		const ctx: AgentChatIntegrationContext = {
			agentId,
			projectId,
			credentialId: integration.credentialId,
			credential: decryptedData,
			webhookUrlFor: (platform) => this.buildWebhookUrl(agentId, projectId, platform),
		};

		// Pre-connect hook — webhook-based platforms use this to detect
		// credential conflicts (e.g. a Telegram bot token already in use) and
		// abort the connect before we touch any external API.
		if (integrationImpl.onBeforeConnect && !options.skipExternalHooks) {
			await integrationImpl.onBeforeConnect(ctx);
		}

		// Delegate adapter construction to the platform implementation.
		const adapter = await integrationImpl.createAdapter(ctx);

		// Dynamic imports — chat packages are ESM-only, use loader to bypass CJS transform
		const { Chat } = await loadChatSdk();
		const { createMemoryState } = await loadMemoryState();

		// Use the platform type as the adapter key (e.g. 'slack') so that
		// bot.webhooks.slack maps correctly to the handler.
		const chat = new Chat({
			userName: `n8n-agent-${agentId}`,
			adapters: { [integration.type]: adapter } as Record<string, never>,
			state: createMemoryState(),
		});

		// Create supporting infrastructure
		const componentMapper = new ComponentMapper();

		// Lazy-import AgentsService to avoid circular DI dependency
		// eslint-disable-next-line import-x/no-cycle
		const { AgentsService } = await import('../agents.service');
		const agentService = Container.get(AgentsService);

		const bridge = AgentChatBridge.create(
			chat,
			agentId,
			agentService,
			componentMapper,
			this.logger,
			projectId,
			integration,
		);

		// Initialize the Chat instance (connects adapters, state adapter, etc.)
		await chat.initialize();

		// Post-initialize hooks (e.g. Telegram setWebhook) run AFTER chat is live.
		// If one throws we must shut the chat down, otherwise adapters/timers leak.
		if (integrationImpl.onAfterConnect && !options.skipExternalHooks) {
			try {
				await integrationImpl.onAfterConnect(ctx);
			} catch (error) {
				await chat.shutdown().catch((shutdownError: unknown) => {
					this.logger.warn(
						`[ChatIntegrationService] Shutdown after failed onAfterConnect threw: ${shutdownError instanceof Error ? shutdownError.message : String(shutdownError)}`,
					);
				});
				bridge.dispose();
				throw error;
			}
		}

		// The `chat` variable is returned by `new Chat(...)` from the ESM-only
		// package. Its runtime shape matches our local `ChatInstance` interface.
		// We validate the required methods exist before storing.
		const chatInstance = chat as ChatInstance;

		this.connections.set(key, {
			chat: chatInstance,
			bridge,
			context: ctx,
		});
		this.logger.info(`[ChatIntegrationService] Connected: ${key}`);
	}

	/**
	 * Disconnect one or all integrations for an agent.
	 * If `type` and `credentialId` are provided, disconnects only that integration.
	 * Otherwise disconnects all integrations for the agent.
	 *
	 * `options.skipExternalHooks` skips `onBeforeDisconnect` — set this on peer
	 * mains reacting to a PubSub broadcast so the cluster-wide remote release
	 * happens exactly once.
	 */
	async disconnect(
		agentId: string,
		integration?: { credentialId: string; type: string },
		options: DisconnectOptions = {},
	): Promise<void> {
		if (integration) {
			await this.disconnectOne(
				this.connectionKey(agentId, integration.type, integration.credentialId),
				options,
			);
		} else {
			const keysToRemove = [...this.connections.keys()].filter((k) => k.startsWith(`${agentId}:`));
			for (const k of keysToRemove) {
				await this.disconnectOne(k, options);
			}
		}
	}

	/**
	 * Disconnect every active integration regardless of type. Used by tests and
	 * for explicit shutdown paths; the leader-stepdown lifecycle uses
	 * {@link disconnectLeaderOnlyIntegrations} so webhook integrations keep
	 * answering on the demoted main (now a follower).
	 */
	async disconnectAll(): Promise<void> {
		const keys = [...this.connections.keys()];
		for (const key of keys) {
			// Graceful shutdown should only clear local runtime state. Cluster-wide
			// remote state must survive so another main can keep receiving events.
			await this.disconnectOne(key, { skipExternalHooks: true });
		}
	}

	/**
	 * On leader stepdown, release only the integrations that require leader
	 * exclusivity (e.g. Telegram polling). Webhook-driven integrations stay
	 * connected on the demoted main so it can keep handling inbound webhooks
	 * routed to it by the load balancer.
	 */
	@OnLeaderStepdown()
	async disconnectLeaderOnlyIntegrations(): Promise<void> {
		for (const key of [...this.connections.keys()]) {
			const type = this.connectionTypeFromKey(key);
			if (!type) continue;
			const integration = this.integrationRegistry.get(type);
			if (integration?.requiresLeader()) {
				await this.disconnectOne(key, { skipExternalHooks: true });
			}
		}
	}

	/**
	 * Diff the previous and next chat integrations of an agent and reconcile
	 * runtime connections accordingly. Used by `AgentsService.updateConfig`
	 * after the builder writes a new integrations array, and by
	 * `AgentsService.publishAgent` to wake up integrations that were persisted
	 * while the agent was still a draft.
	 *
	 * Disconnects of removed integrations always run (so unpublishing-then-
	 * editing works). Connects of newly-added integrations are gated on
	 * `agent.activeVersionId` — matching the controller's connect endpoint,
	 * which rejects unpublished agents, and `reconnectAll`, which only restores
	 * published agents. The integration entry stays persisted on the entity so
	 * it can be picked up later by `publishAgent` calling this method again.
	 *
	 * Connection failures are logged at the call site — this method propagates
	 * errors from disconnect but swallows connect errors per integration so a
	 * single bad credential doesn't block the others.
	 */
	async syncToConfig(
		agent: Agent,
		previous: AgentCredentialIntegrationConfig[],
		next: AgentCredentialIntegrationConfig[],
	): Promise<void> {
		const previousKeys = new Set(previous.map(buildIntegrationConnectionId));
		const nextKeys = new Set(next.map(buildIntegrationConnectionId));

		for (const integration of previous) {
			if (!nextKeys.has(buildIntegrationConnectionId(integration))) {
				try {
					await this.disconnect(agent.id, integration);
					await this.broadcastIntegrationChange(agent.id, integration, 'disconnect');
				} catch (error) {
					this.logger.warn(
						`[ChatIntegrationService] Disconnect during sync failed for ${integration.type} on agent ${agent.id}: ${error instanceof Error ? error.message : String(error)}`,
					);
				}
			}
		}

		const additions = next.filter((i) => !previousKeys.has(buildIntegrationConnectionId(i)));

		if (additions.length > 0 && !agent.activeVersionId) {
			this.logger.debug(
				'[ChatIntegrationService] Skipping connect for unpublished agent — entry persisted, will connect on publish',
				{ agentId: agent.id, pendingTypes: additions.map((i) => i.type) },
			);
			return;
		}

		// TODO: AgentCredentialIntegration has no record of *who* connected the
		// integration, so we have no anchor user identity to decrypt credentials
		// with on reconnect / sync. We fall back to probing project members until
		// one has `credential:read` on the integration credential.
		// Replace with a proper solution (fetching credentials should not depend on any specific user)
		const userIds = additions.length
			? await Container.get(ProjectRelationRepository).findUserIdsByProjectId(agent.projectId)
			: [];

		for (const integration of additions) {
			let connected = false;
			for (const userId of userIds) {
				try {
					await this.connect(agent.id, integration, userId, agent.projectId);

					connected = true;
					break;
				} catch (error) {
					this.logger.debug('[ChatIntegrationService] Connect attempt failed during sync', {
						agentId: agent.id,
						userId,
						type: integration.type,
						error,
					});
				}
			}
			if (connected) {
				await this.broadcastIntegrationChange(agent.id, integration, 'connect');
			} else {
				this.logger.warn(
					'[ChatIntegrationService] Could not connect integration during sync — no project member had credential access',
					{ agentId: agent.id, type: integration.type, credentialId: integration.credentialId },
				);
			}
		}
	}

	/**
	 * Return connection status and count for an agent.
	 */
	getStatus(agentId: string): AgentIntegrationStatusResponse & { connections: number } {
		const integrations: AgentIntegrationStatusResponse['integrations'] = [];
		for (const k of this.connections.keys()) {
			if (k.startsWith(`${agentId}:`)) {
				// Key format: agentId:type:credentialId
				const parts = k.split(':');
				if (parts.length >= 3) {
					integrations.push({ type: parts[1], credentialId: parts.slice(2).join(':') });
				}
			}
		}
		return {
			status: integrations.length > 0 ? 'connected' : 'disconnected',
			connections: integrations.length,
			integrations,
		};
	}

	/**
	 * Return the first Chat instance for an agent, or undefined if not connected.
	 */
	getChatInstance(
		agentId: string,
		integration?: { type: string; credentialId: string },
	): ChatInstance | undefined {
		if (integration) {
			return this.connections.get(
				this.connectionKey(agentId, integration.type, integration.credentialId),
			)?.chat;
		}
		for (const [k, conn] of this.connections) {
			if (k.startsWith(`${agentId}:`)) return conn.chat;
		}
		return undefined;
	}

	/**
	 * Return the webhook handler for a specific platform on an agent.
	 * This is the pre-built handler from `bot.webhooks[platform]` that
	 * accepts a Web API Request and returns a Web API Response.
	 *
	 * Looks up the connection by platform so that the correct Chat instance
	 * is used when an agent has multiple integrations (e.g. Slack + Discord).
	 */
	getWebhookHandler(agentId: string, platform: string): WebhookHandler | undefined {
		for (const [key, conn] of this.connections) {
			if (key.startsWith(`${agentId}:${platform}:`)) {
				return conn.chat.webhooks[platform];
			}
		}
		return undefined;
	}

	/**
	 * Reconnect all agents that have integrations configured. Called on startup
	 * (every main) and on `leader-takeover` in multi-main mode.
	 *
	 * Webhook-driven integrations connect on every main so that inbound webhooks
	 * load-balanced across mains always find a live handler. Integrations that
	 * declare `requiresLeader()` (e.g. Telegram polling) only connect on the
	 * leader so a single instance owns the long-poll loop.
	 *
	 * Already-connected keys are skipped so this is a safe idempotent operation
	 * — important for leader takeover, where a former follower already holds
	 * webhook integrations and only needs to add the leader-only ones.
	 */
	@OnLeaderTakeover()
	async reconnectAll(): Promise<void> {
		// Only reconnect integrations for published agents — an unpublished agent must not
		// receive events, so we don't even load it.
		const agents = await this.agentRepository.findPublished();
		for (const agent of agents) {
			if (!agent.integrations || agent.integrations.length === 0) continue;
			for (const integration of agent.integrations) {
				if (!isAgentCredentialIntegration(integration)) {
					continue;
				}

				const definition = this.integrationRegistry.get(integration.type);
				if (definition?.requiresLeader() && !this.instanceSettings.isLeader) {
					this.logger.debug(
						`[ChatIntegrationService] Skipping ${integration.type} for agent ${agent.id} — leader-only and this main is a follower`,
					);
					continue;
				}

				const key = this.connectionKey(agent.id, integration.type, integration.credentialId);
				if (this.connections.has(key)) continue;

				const userIds = await Container.get(ProjectRelationRepository).findUserIdsByProjectId(
					agent.projectId,
				);
				if (userIds.length === 0) {
					this.logger.warn(
						`[ChatIntegrationService] No users found for project ${agent.projectId} — skipping reconnect for agent ${agent.id}`,
					);
					continue;
				}

				// External setup runs once per cluster — the leader claims that role
				// on startup; followers only build local runtime state.
				const skipExternalHooks = !this.instanceSettings.isLeader;
				const options = this.connectOptionsFor(integration, skipExternalHooks);

				// Try each project member until one succeeds — the first user may not
				// have access to the integration credential.
				let connected = false;
				for (const userId of userIds) {
					try {
						await this.connect(agent.id, integration, userId, agent.projectId, options);
						connected = true;
						break;
					} catch (error) {
						this.logger.debug(
							`[ChatIntegrationService] User ${userId} could not reconnect ${integration.type} for agent ${agent.id}: ${error instanceof Error ? error.message : String(error)}`,
						);
					}
				}
				if (!connected) {
					this.logger.error(
						`[ChatIntegrationService] Failed to reconnect ${integration.type} for agent ${agent.id} — no project member could access the credential`,
					);
				}
			}
		}
	}

	/**
	 * Reconcile a single integration when notified via PubSub by another main.
	 *
	 * The originating main (the one that handled the user's connect/disconnect
	 * request) updated its own state synchronously before publishing — this
	 * handler runs on every other main so all `connections` maps stay aligned.
	 *
	 * On `connect`, integrations that require the leader are skipped on
	 * followers. On `disconnect`, the integration is always torn down regardless
	 * of role.
	 */
	@OnPubSubEvent('agent-chat-integration-changed', { instanceType: 'main' })
	async handleIntegrationChanged(
		payload: PubSubCommandMap['agent-chat-integration-changed'],
	): Promise<void> {
		const { agentId, integration, action } = payload;
		const { type, credentialId } = integration;

		if (action === 'disconnect') {
			// The originating main already ran integration-defined external teardown.
			// Peers only clear local runtime state to avoid duplicate external side effects.
			await this.disconnect(agentId, integration, { skipExternalHooks: true });
			return;
		}

		const definition = this.integrationRegistry.get(type);
		if (definition?.requiresLeader() && !this.instanceSettings.isLeader) {
			this.logger.debug(
				`[ChatIntegrationService] Ignoring connect for ${type} on agent ${agentId} — leader-only integration on follower`,
			);
			return;
		}

		const key = this.connectionKey(agentId, type, credentialId);
		if (this.connections.has(key)) return;

		const agent = await this.agentRepository.findOne({ where: { id: agentId } });
		if (!agent) {
			this.logger.warn(
				`[ChatIntegrationService] Cannot connect ${type} — agent ${agentId} not found`,
			);
			return;
		}

		const userIds = await Container.get(ProjectRelationRepository).findUserIdsByProjectId(
			agent.projectId,
		);
		for (const userId of userIds) {
			try {
				// The originating main already ran integration-defined external setup.
				// Peers only build local runtime state to avoid duplicate external
				// side effects.
				const options: ConnectOptions = { skipExternalHooks: true };
				await this.connect(agentId, integration, userId, agent.projectId, options);
				return;
			} catch (error) {
				this.logger.debug(
					`[ChatIntegrationService] User ${userId} could not connect ${type} for agent ${agentId}: ${error instanceof Error ? error.message : String(error)}`,
				);
			}
		}
		this.logger.error(
			`[ChatIntegrationService] Failed to connect ${type} for agent ${agentId} — no project member could access the credential`,
		);
	}

	// ---------------------------------------------------------------------------
	// Private helpers
	// ---------------------------------------------------------------------------

	private async disconnectOne(key: string, options: DisconnectOptions = {}): Promise<void> {
		const conn = this.connections.get(key);
		if (!conn) return;

		// External teardown runs while the chat is still live — symmetric with
		// `onAfterConnect`, which runs after `chat.initialize()`. Errors are
		// logged but never re-thrown: local teardown must always complete so a
		// transient remote failure can't leak in-process resources.
		if (!options.skipExternalHooks) {
			const type = this.connectionTypeFromKey(key);
			const integration = type ? this.integrationRegistry.get(type) : undefined;
			if (integration?.onBeforeDisconnect) {
				try {
					await integration.onBeforeDisconnect(conn.context);
				} catch (error) {
					this.logger.warn(
						`[ChatIntegrationService] onBeforeDisconnect failed for ${key}: ${error instanceof Error ? error.message : String(error)}`,
					);
				}
			}
		}

		try {
			await conn.chat.shutdown();
		} catch (error) {
			this.logger.warn(
				`[ChatIntegrationService] Error during shutdown for ${key}: ${error instanceof Error ? error.message : String(error)}`,
			);
		}

		conn.bridge.dispose();

		this.connections.delete(key);
		this.logger.info(`[ChatIntegrationService] Disconnected: ${key}`);
	}

	private async resolveUser(userId: string): Promise<User> {
		const user = await Container.get(UserRepository).findOne({
			where: { id: userId },
			relations: ['role'],
		});
		if (!user) {
			throw new Error(`User ${userId} not found`);
		}
		return user;
	}

	private async decryptCredential(
		credentialId: string,
		user: User,
	): Promise<Record<string, unknown>> {
		const credential = await this.credentialsFinderService.findCredentialForUser(
			credentialId,
			user,
			['credential:read'],
		);
		if (!credential) {
			throw new Error(`Credential ${credentialId} not found or not accessible`);
		}
		const decrypted = await this.credentialsService.decrypt(credential, true);
		return decrypted as Record<string, unknown>;
	}

	private buildWebhookUrl(agentId: string, projectId: string, platform: string): string {
		// getWebhookBaseUrl returns the URL with a trailing slash, honours the
		// WEBHOOK_URL env var used by n8n's other webhook triggers.
		const base = this.urlService.getWebhookBaseUrl();
		return `${base}rest/projects/${projectId}/agents/v2/${agentId}/webhooks/${platform}`;
	}

	private connectOptionsFor(
		integration: AgentCredentialIntegrationConfig,
		skipExternalHooks: boolean,
	): ConnectOptions {
		return 'settings' in integration
			? { skipExternalHooks, settings: integration.settings }
			: { skipExternalHooks };
	}
}