import { SystemAuditLog } from '#models/SystemAuditLog';
import { EntityChangeLog } from '#models/EntityChangeLog';
import { AuditOutbox } from '#models/AuditOutbox';
import { AuditDeadLetter } from '#models/AuditDeadLetter';
import { Op, QueryTypes } from 'sequelize';
import workerSequelize from '#services/database/sequelize/workerSequelizeService';
import Logger from '#utils/logger';
import { randomUUID } from 'crypto';
import Config from '#config';
import auditMetrics, { initAuditMetrics } from '#utils/auditMetrics';

/**
 * Outbox Poller with LISTEN/NOTIFY pattern for real-time processing
 * - LISTEN on audit_channel for immediate wake-up when new events arrive
 * - Fallback polling (10s) in case NOTIFY is missed
 * - Atomic claim with FOR UPDATE SKIP LOCKED to eliminate race conditions
 */
export class OutboxPoller {
	private static MIN_POLL_INTERVAL = 500; // 500ms when queue has events (for faster processing)
	private static MAX_POLL_INTERVAL = 30000; // 30s when queue is empty (reduced DB load)
	private static FALLBACK_POLL_INTERVAL = 15000; // 15s fallback for LISTEN/NOTIFY (increased from 10s)
	private static currentPollInterval = OutboxPoller.MIN_POLL_INTERVAL;
	private static BATCH_SIZE = 100;
	private static isRunning = false;
	private static pgClient: any = null;
	private static isListening = false;

	static async start() {
		Logger.info('OutboxPoller.start() called');
		if (this.isRunning) {
			Logger.info('OutboxPoller already running, skipping');
			return;
		}
		this.isRunning = true;
		Logger.info('Outbox poller started');

		// Initialize Prometheus metrics
		initAuditMetrics();

		// Initialize LISTEN/NOTIFY
		await this.setupListenNotify();

		// Start polling loop
		Logger.info('About to call poll()');
		this.poll();
		Logger.info('poll() called (fire-and-forget)');
	}

	static stop() {
		this.isRunning = false;
		this.isListening = false;

		// Close PostgreSQL client
		if (this.pgClient) {
			this.pgClient.end();
			this.pgClient = null;
		}

		Logger.info('Outbox poller stopped');
	}

	private static async setupListenNotify() {
		try {
			// Get a dedicated PostgreSQL client for LISTEN/NOTIFY
			const pg = await import('pg');
			const Pool = pg.Pool as any;
			this.pgClient = new Pool({
				host: Config.database.host,
				port: Config.database.port,
				database: Config.database.name,
				user: Config.database.user,
				password: Config.database.password,
			});

			// Listen for notifications
			await this.pgClient.query('LISTEN audit_channel');
			this.isListening = true;
			Logger.info('Listening on audit_channel for real-time notifications');

			// Handle notifications
			this.pgClient.on('notification', async () => {
				if (this.isRunning && this.isListening) {
					Logger.debug('Received NOTIFY on audit_channel, waking up poller');
					// Reset to minimum interval for immediate processing
					this.currentPollInterval = this.MIN_POLL_INTERVAL;
				}
			});

			// Handle connection errors
			this.pgClient.on('error', (error: any) => {
				Logger.error('PostgreSQL LISTEN/NOTIFY client error:', error);
				this.isListening = false;
				// Fallback to polling only
			});
		} catch (error) {
			Logger.error('Failed to setup LISTEN/NOTIFY, falling back to polling only:', error);
			this.isListening = false;
		}
	}

	private static async poll() {
		Logger.info(`Poll loop started, isRunning: ${this.isRunning}`);
		while (this.isRunning) {
			try {
				Logger.debug(`Poll iteration, interval: ${this.currentPollInterval}`);
				const eventCount = await this.processBatch();

				// Adjust poll interval based on queue activity
				if (eventCount > 0) {
					// Queue has events, poll faster
					this.currentPollInterval = this.MIN_POLL_INTERVAL;
				} else {
					// Queue empty, use fallback interval
					if (this.isListening) {
						// LISTEN/NOTIFY is active, use longer fallback
						this.currentPollInterval = this.FALLBACK_POLL_INTERVAL;
					} else {
						// No LISTEN/NOTIFY, use exponential backoff
						this.currentPollInterval = Math.min(this.currentPollInterval * 1.5, this.MAX_POLL_INTERVAL);
					}
				}
			} catch (error) {
				Logger.error('Outbox poller error:', error);
				// Wait longer on error
				this.currentPollInterval = 5000;
			}

			await this.sleep(this.currentPollInterval);
		}
	}

	private static async processBatch(): Promise<number> {
		Logger.debug('processBatch called');
		// Recovery: Reset stuck PROCESSING events (older than 2 minutes) to PENDING
		await this.recoverStuckEvents();

		// Monitor queue depth for backpressure alerts
		await this.monitorQueueDepth();

		// ATOMIC CLAIM: UPDATE with RETURNING eliminates race condition
		// If app crashes after claim but before COMPLETED, recovery will reset PROCESSING
		const query = `
            UPDATE audit.audit_outbox
            SET status = 'PROCESSING',
                updated_at = CURRENT_TIMESTAMP
            WHERE id IN (
                SELECT id
                FROM audit.audit_outbox
                WHERE status = 'PENDING'
                ORDER BY created_at ASC
                LIMIT $1
                FOR UPDATE SKIP LOCKED
            )
            RETURNING *;
        `;

		const events = await workerSequelize.query(query, {
			bind: [this.BATCH_SIZE],
		});

		if (!events || (events as any[]).length === 0) return 0;

		Logger.debug(`Processing ${events.length} audit events`);

		const startTime = Date.now();

		try {
			// Process events (optimistic batching)
			const systemAuditEvents = events.filter((e: any) => e.event_type === 'SYSTEM_AUDIT');
			const entityChangeEvents = events.filter((e: any) => e.event_type === 'ENTITY_CHANGE');

			Logger.debug(
				`Event breakdown: ${systemAuditEvents.length} SYSTEM_AUDIT, ${entityChangeEvents.length} ENTITY_CHANGE`,
			);

			// Batch insert system audit logs
			if (systemAuditEvents.length > 0) {
				await this.batchInsertSystemAudit(systemAuditEvents);
			}

			// Batch insert entity change logs
			if (entityChangeEvents.length > 0) {
				await this.batchInsertEntityChange(entityChangeEvents);
			}

			// Mark as COMPLETED
			const eventIds = events.map((e: any) => e.id);
			await AuditOutbox.update(
				{ status: 'COMPLETED', processed_at: new Date() },
				{ where: { id: { [Op.in]: eventIds } } },
			);

			// Record metrics
			const processingTime = (Date.now() - startTime) / 1000;
			events.forEach((event: any) => {
				auditMetrics.auditEventsProcessed.inc({ event_type: event.event_type, status: 'COMPLETED' });
				auditMetrics.auditProcessingDuration.observe({ event_type: event.event_type }, processingTime);
			});

			return events.length;
		} catch (error) {
			// Handle batch failure - mark all events as failed
			Logger.error(`Batch processing failed for ${events.length} events:`, error);
			for (const event of events) {
				await this.handleFailedEvent(event, error);
			}
			return events.length;
		}
	}

	private static async recoverStuckEvents() {
		try {
			Logger.debug('Running recovery check for stuck PROCESSING events...');
			const result = await workerSequelize.query(
				`
                UPDATE audit.audit_outbox
                SET status = 'PENDING', updated_at = CURRENT_TIMESTAMP, retry_count = 0
                WHERE status = 'PROCESSING'
                AND updated_at < NOW() - INTERVAL '2 minutes'
            `,
			);
			const recoveredCount = (result as any)[1];
			if (recoveredCount > 0) {
				Logger.warn(`Recovered ${recoveredCount} stuck PROCESSING events`);
			}
		} catch (error) {
			Logger.error('Failed to recover stuck events:', error);
		}
	}

	private static async monitorQueueDepth() {
		try {
			const [result] = await workerSequelize.query<{
				queue_depth: number;
				pending_count: number;
				processing_count: number;
				failed_count: number;
			}>(
				`
                SELECT 
                    COUNT(*) as queue_depth,
                    COUNT(*) FILTER (WHERE status = 'PENDING') as pending_count,
                    COUNT(*) FILTER (WHERE status = 'PROCESSING') as processing_count,
                    COUNT(*) FILTER (WHERE status IN ('FAILED', 'FAILED_PERMANENT')) as failed_count
                FROM audit.audit_outbox
                WHERE created_at > NOW() - INTERVAL '1 hour'
            `,
				{ type: QueryTypes.SELECT },
			);

			const queueData = result;
			if (queueData) {
				const { queue_depth, pending_count, processing_count, failed_count } = queueData;

				// Log queue depth metrics
				Logger.debug(
					`Queue depth: ${queue_depth} (pending: ${pending_count}, processing: ${processing_count}, failed: ${failed_count})`,
				);

				// Alert if queue depth is high
				const HIGH_QUEUE_THRESHOLD = 1000;
				const CRITICAL_QUEUE_THRESHOLD = 5000;

				if (queue_depth >= CRITICAL_QUEUE_THRESHOLD) {
					Logger.error(
						`CRITICAL: Audit outbox queue depth ${queue_depth} exceeds critical threshold ${CRITICAL_QUEUE_THRESHOLD}`,
					);
				} else if (queue_depth >= HIGH_QUEUE_THRESHOLD) {
					Logger.warn(
						`WARNING: Audit outbox queue depth ${queue_depth} exceeds high threshold ${HIGH_QUEUE_THRESHOLD}`,
					);
				}

				// Record metrics
				try {
					auditMetrics.auditOutboxQueueDepth.set(queue_depth);
					auditMetrics.auditOutboxPending.set(pending_count);
					auditMetrics.auditOutboxProcessing.set(processing_count);
					auditMetrics.auditOutboxFailed.set(failed_count);
				} catch {
					// Metrics optional, don't fail if not available
				}
			}
		} catch (error) {
			Logger.error('Failed to monitor queue depth:', error);
		}
	}

	private static async handleFailedEvent(event: any, error: any) {
		Logger.error(`Failed to process audit event ${event.id}:`, error);

		// Increment retry count
		const newRetryCount = (event.retry_count || 0) + 1;

		// Record failed metric
		auditMetrics.auditEventsFailed.inc({ event_type: event.event_type, error_type: error.name || 'unknown' });

		if (newRetryCount >= (event.max_retries || 3)) {
			// Move to dead letter queue
			try {
				await AuditDeadLetter.create({
					id: randomUUID(),
					original_event_id: event.id,
					event_type: event.event_type,
					original_payload: event.payload,
					error_message: error.message || 'Unknown error',
					error_stack: error.stack,
					failed_at: new Date(),
					replayed: false,
					replay_count: 0,
					investigation_status: 'PENDING',
					created_at: new Date(),
				});

				// Mark as FAILED_PERMANENT in outbox
				await AuditOutbox.update({ status: 'FAILED_PERMANENT', failed_at: new Date() }, { where: { id: event.id } });

				// Record dead letter metric
				auditMetrics.auditDeadLetterEvents.inc({ event_type: event.event_type });

				Logger.warn(`Moved audit event ${event.id} to dead letter queue`);
			} catch (dlqError) {
				Logger.error(`Failed to move event ${event.id} to dead letter queue:`, dlqError);
			}
		} else {
			// Update retry count and mark as FAILED for retry
			await AuditOutbox.update(
				{
					status: 'FAILED',
					retry_count: newRetryCount,
					error_message: error.message || 'Unknown error',
				},
				{ where: { id: event.id } },
			);
		}
	}

	private static async batchInsertSystemAudit(events: any[]) {
		const records = events.map((e) => {
			const payload = e.payload;
			return {
				id: randomUUID(),
				outbox_event_id: e.id,
				request_id: e.request_id,
				trace_id: e.trace_id,
				session_id: payload.session_id,
				tenant_id: payload.tenant_id,
				actor_id: payload.actorId,
				actor_name: payload.actorName,
				actor_email: payload.actorEmail,
				actor_role: payload.actorRole,
				action: payload.action,
				module: payload.module,
				entity_id: payload.entityId,
				entity_type: payload.entityType,
				batch_id: payload.batchId,
				description: payload.description,
				severity: payload.severity,
				ip_address: payload.ipAddress,
				user_agent: payload.userAgent,
				payload: payload.payload,
				payload_size: payload.payload_size,
				success: payload.success,
				failure_reason: payload.failureReason,
				event_version: payload.event_version,
				created_at: new Date(),
			};
		});

		// Use ON CONFLICT DO NOTHING for idempotency
		await SystemAuditLog.bulkCreate(records as any, {
			ignoreDuplicates: true,
		});
	}

	private static async batchInsertEntityChange(events: any[]) {
		const records = events.map((e) => {
			const payload = e.payload;
			return {
				id: randomUUID(),
				outbox_event_id: e.id,
				request_id: e.request_id,
				trace_id: e.trace_id,
				entity_type: payload.entityType,
				entity_id: payload.entityId,
				actor_id: payload.actorId,
				actor_name: payload.actorName,
				actor_email: payload.actorEmail,
				change_type: payload.changeType,
				payload: payload.payload,
				payload_size: payload.payload_size,
				reason: payload.reason,
				event_version: payload.event_version,
				created_at: new Date(),
			};
		});

		Logger.debug(`Preparing to insert ${records.length} entity change records`);

		try {
			const result = await EntityChangeLog.bulkCreate(records as any, {
				ignoreDuplicates: true,
			});
			Logger.debug(`bulkCreate result: ${result.length} records created`);
		} catch (error) {
			Logger.error(`bulkCreate failed:`, error);
			throw error;
		}
	}

	private static sleep(ms: number) {
		return new Promise((resolve) => setTimeout(resolve, ms));
	}
}

export default OutboxPoller;
