/**
 * BullMQ Queue Infrastructure
 *
 * Enterprise-grade async job processing with:
 * - Automatic retries with exponential backoff
 * - Job prioritization (high/normal/low)
 * - Dead Letter Queue (DLQ) for failed jobs
 * - Job metrics and monitoring
 * - Rate limiting per queue
 * - Graceful shutdown
 *
 * Queues:
 * - email: Email sending (OTP, notifications, marketing)
 * - webhook: External webhook calls (n8n, Stripe, etc.)
 * - analytics: Event tracking, metrics aggregation
 *
 * @requires bullmq, ioredis
 */

import 'server-only';
import { Queue, Worker, QueueEvents, Job } from 'bullmq';
import IORedis from 'ioredis';
import { logger } from '@/lib/logger';
import { recordSystemAlert, buildQueueAlertMessage } from '@/lib/system-alerts';

// ============================
// Redis Connection
// ============================

let redisConnection: IORedis | null = null;

function getRedisConnection(): IORedis {
  if (redisConnection) return redisConnection;

  const redisUrl = process.env.REDIS_URL || 'redis://localhost:6379';

  try {
    const url = new URL(redisUrl);
    redisConnection = new IORedis({
      host: url.hostname || 'localhost',
      port: parseInt(url.port) || 6379,
      password: url.password || undefined,
      maxRetriesPerRequest: null, // BullMQ requirement
      enableReadyCheck: false, // BullMQ recommendation
      retryStrategy(times) {
        const delay = Math.min(times * 50, 2000);
        return delay;
      },
    });

    redisConnection.on('error', (err) => {
      logger.error('[BullMQ] Redis connection error', err);
    });

    logger.info('[BullMQ] Redis connection initialized', {
      host: url.hostname,
      port: url.port,
    });

    return redisConnection;
  } catch (error) {
    logger.error('[BullMQ] Failed to parse REDIS_URL', { error });
    throw new Error('Invalid REDIS_URL configuration');
  }
}

// ============================
// Queue Configuration
// ============================

export const QUEUE_NAMES = {
  EMAIL: 'email',
  WEBHOOK: 'webhook',
  ANALYTICS: 'analytics',
  SYSTEM: 'system', // For system maintenance jobs like drift detection
} as const;

export type QueueName = (typeof QUEUE_NAMES)[keyof typeof QUEUE_NAMES];

const DEFAULT_JOB_OPTIONS = {
  attempts: 4,
  backoff: {
    type: 'exponential' as const,
    delay: 5000, // Start with 5s, then 10s, 20s, 40s
  },
  removeOnComplete: {
    age: 3600, // Keep completed jobs for 1 hour
    count: 100, // Keep last 100 completed jobs
  },
  removeOnFail: {
    age: 86400, // Keep failed jobs for 24 hours
    count: 500, // Keep last 500 failed jobs
  },
};

const queueAlertCooldowns = new Map<QueueName, number>();

function queueOnCooldown(queue: QueueName, windowMinutes = 5): boolean {
  const last = queueAlertCooldowns.get(queue);
  if (!last) return false;
  return Date.now() - last < windowMinutes * 60 * 1000;
}

async function alertQueue(queue: QueueName, reason: string, severity: 'warning' | 'error' = 'warning') {
  if (queueOnCooldown(queue)) return;
  await recordSystemAlert({
    title: 'Queue alert',
    message: buildQueueAlertMessage(queue, reason),
    severity,
    source: `queue:${queue}`,
    environment: process.env.APP_ENV || process.env.VERCEL_ENV || process.env.NODE_ENV,
    cooldownKey: `queue:${queue}:${severity}`,
    cooldownMinutes: 5,
  });
  queueAlertCooldowns.set(queue, Date.now());
}

// ============================
// Queue Instances
// ============================

const queues = new Map<QueueName, Queue>();
const workers = new Map<QueueName, Worker>();
const queueEvents = new Map<QueueName, QueueEvents>();

export function getQueue(name: QueueName): Queue {
  if (queues.has(name)) {
    return queues.get(name)!;
  }

  const connection = getRedisConnection();
  const queue = new Queue(name, {
    connection: connection as any,
    defaultJobOptions: DEFAULT_JOB_OPTIONS,
  });

  queues.set(name, queue);

  logger.info('[BullMQ] Queue created', { queue: name });

  return queue;
}

// ============================
// Job Payloads (Type-Safe)
// ============================

export interface EmailJobPayload {
  to: string;
  subject: string;
  html: string;
  text: string;
  category?: string;
  metadata?: Record<string, unknown>;
}

export interface WebhookJobPayload {
  url: string;
  method: 'GET' | 'POST' | 'PUT' | 'DELETE' | 'PATCH';
  headers?: Record<string, string>;
  body?: Record<string, unknown> | string;
  timeout?: number;
  metadata?: Record<string, unknown>;
}

export interface AnalyticsJobPayload {
  event: string;
  userId?: string;
  sessionId?: string;
  properties?: Record<string, unknown>;
  timestamp?: string;
}

export interface SystemJobPayload {
  type: 'drift_check' | 'health_check' | 'cleanup';
  countThreshold?: number;
  percentageThreshold?: number;
  sendAlertEmail?: boolean;
  metadata?: Record<string, unknown>;
}

// Union type for all job payloads
export type JobPayload = EmailJobPayload | WebhookJobPayload | AnalyticsJobPayload | SystemJobPayload;

// ============================
// Job Enqueue Functions
// ============================

export interface EnqueueOptions {
  /** Job priority (1-10, higher = more important) */
  priority?: number;
  /** Delay before processing (ms) */
  delay?: number;
  /** Max retry attempts (overrides default) */
  attempts?: number;
  /** Job ID (for deduplication) */
  jobId?: string;
  /** Remove job on completion */
  removeOnComplete?: boolean | number;
}

/**
 * Enqueue an email job
 */
export async function enqueueEmail(
  payload: EmailJobPayload,
  options?: EnqueueOptions
): Promise<string> {
  const queue = getQueue(QUEUE_NAMES.EMAIL);

  const job = await queue.add('send-email', payload, {
    priority: options?.priority ?? (payload.category === 'transactional' ? 5 : 3),
    delay: options?.delay,
    attempts: options?.attempts ?? DEFAULT_JOB_OPTIONS.attempts,
    jobId: options?.jobId,
    removeOnComplete: options?.removeOnComplete ?? DEFAULT_JOB_OPTIONS.removeOnComplete,
  });

  logger.debug('[BullMQ] Email job enqueued', {
    jobId: job.id,
    to: payload.to,
    category: payload.category,
  });

  return job.id!;
}

/**
 * Enqueue a webhook job
 */
export async function enqueueWebhook(
  payload: WebhookJobPayload,
  options?: EnqueueOptions
): Promise<string> {
  const queue = getQueue(QUEUE_NAMES.WEBHOOK);

  const job = await queue.add('call-webhook', payload, {
    priority: options?.priority ?? 3,
    delay: options?.delay,
    attempts: options?.attempts ?? DEFAULT_JOB_OPTIONS.attempts,
    jobId: options?.jobId,
  });

  logger.debug('[BullMQ] Webhook job enqueued', {
    jobId: job.id,
    url: payload.url,
    method: payload.method,
  });

  return job.id!;
}

/**
 * Enqueue an analytics job
 */
export async function enqueueAnalytics(
  payload: AnalyticsJobPayload,
  options?: EnqueueOptions
): Promise<string> {
  const queue = getQueue(QUEUE_NAMES.ANALYTICS);

  const job = await queue.add('track-event', payload, {
    priority: options?.priority ?? 1, // Analytics is low priority
    delay: options?.delay,
    attempts: options?.attempts ?? 3, // Fewer retries for analytics
    jobId: options?.jobId,
  });

  logger.debug('[BullMQ] Analytics job enqueued', {
    jobId: job.id,
    event: payload.event,
    userId: payload.userId,
  });

  return job.id!;
}

/**
 * Enqueue a system job (drift check, health check, cleanup, etc.)
 */
export async function enqueueSystemJob(
  payload: SystemJobPayload,
  options?: EnqueueOptions
): Promise<string> {
  const queue = getQueue(QUEUE_NAMES.SYSTEM);

  const job = await queue.add(`system-${payload.type}`, payload, {
    priority: options?.priority ?? 2, // System jobs are low-medium priority
    delay: options?.delay,
    attempts: options?.attempts ?? 2, // Fewer retries for system jobs
    jobId: options?.jobId,
  });

  logger.debug('[BullMQ] System job enqueued', {
    jobId: job.id,
    type: payload.type,
  });

  return job.id!;
}

/**
 * Schedule a daily drift check job
 * Should be called on application startup
 */
export async function scheduleDailyDriftCheck(
  options?: {
    countThreshold?: number;
    percentageThreshold?: number;
    sendAlertEmail?: boolean;
    cronTime?: string; // Default: '0 2 * * *' (2 AM daily)
  }
): Promise<void> {
  const queue = getQueue(QUEUE_NAMES.SYSTEM);

  // Use BullMQ's repeat feature for scheduled jobs
  // Default: Run at 2 AM daily
  const cronExpression = options?.cronTime || '0 2 * * *';

  try {
    // Add a repeatable job
    await queue.add(
      'system-drift_check',
      {
        type: 'drift_check',
        countThreshold: options?.countThreshold,
        percentageThreshold: options?.percentageThreshold,
        sendAlertEmail: options?.sendAlertEmail ?? true,
      } as SystemJobPayload,
      {
        repeat: {
          pattern: cronExpression,
        },
        jobId: 'daily-drift-check', // Deduplicate
        removeOnComplete: 10, // Keep last 10 completed
        removeOnFail: 50, // Keep last 50 failed for debugging
      }
    );

    logger.info('[BullMQ] Daily drift check scheduled', {
      cronExpression,
      countThreshold: options?.countThreshold,
      percentageThreshold: options?.percentageThreshold,
    });
  } catch (error) {
    logger.error('[BullMQ] Failed to schedule daily drift check', error);
    throw error;
  }
}

// ============================
// Worker Registration
// ============================

export type JobHandler<T = unknown> = (job: Job<T>) => Promise<void>;

export interface WorkerOptions {
  concurrency?: number;
  limiter?: {
    max: number; // Max jobs per duration
    duration: number; // Duration in ms
  };
}

/**
 * Register a worker to process jobs from a queue
 */
export function registerWorker<T = unknown>(
  queueName: QueueName,
  handler: JobHandler<T>,
  options?: WorkerOptions
): Worker<T> {
  if (workers.has(queueName)) {
    logger.warn('[BullMQ] Worker already registered', { queue: queueName });
    return workers.get(queueName)! as Worker<T>;
  }

  const connection = getRedisConnection();

  const worker = new Worker<T>(
    queueName,
    async (job) => {
      const startTime = Date.now();

      try {
        logger.debug('[BullMQ] Processing job', {
          jobId: job.id,
          queue: queueName,
          attempt: job.attemptsMade + 1,
        });

        await handler(job);

        const duration = Date.now() - startTime;
        logger.info('[BullMQ] Job completed', {
          jobId: job.id,
          queue: queueName,
          duration,
        });
      } catch (error) {
        const duration = Date.now() - startTime;
        logger.error('[BullMQ] Job failed', {
          jobId: job.id,
          queue: queueName,
          attempt: job.attemptsMade + 1,
          duration,
          error: error instanceof Error ? error.message : String(error),
        });
        throw error; // Re-throw to trigger retry
      }
    },
    {
      connection: connection as any,
      concurrency: options?.concurrency ?? 5,
      limiter: options?.limiter,
    }
  );

  // Event listeners
  worker.on('completed', (job) => {
    logger.debug('[BullMQ] Worker completed job', {
      jobId: job.id,
      queue: queueName,
    });
  });

  worker.on('failed', (job, err) => {
    logger.error('[BullMQ] Worker job failed', {
      jobId: job?.id,
      queue: queueName,
      error: err.message,
      willRetry: (job?.attemptsMade ?? 0) < (job?.opts.attempts ?? DEFAULT_JOB_OPTIONS.attempts),
    });
    void alertQueue(
      queueName,
      `Job ${job?.id || 'unknown'} failed: ${err.message}`,
      'warning'
    );
  });

  worker.on('error', (err) => {
    logger.error('[BullMQ] Worker error', { queue: queueName, error: err.message });
    void alertQueue(queueName, `Worker error: ${err.message}`, 'error');
  });

  workers.set(queueName, worker as Worker);

  logger.info('[BullMQ] Worker registered', {
    queue: queueName,
    concurrency: options?.concurrency ?? 5,
  });

  return worker;
}

// ============================
// Queue Events & Monitoring
// ============================

/**
 * Subscribe to queue events for monitoring
 */
export function subscribeToQueueEvents(queueName: QueueName): QueueEvents {
  if (queueEvents.has(queueName)) {
    return queueEvents.get(queueName)!;
  }

  const connection = getRedisConnection();
  const events = new QueueEvents(queueName, { connection: connection as any });

  events.on('completed', ({ jobId }) => {
    logger.debug('[BullMQ] Job event: completed', { queue: queueName, jobId });
  });

  events.on('failed', ({ jobId, failedReason }) => {
    logger.warn('[BullMQ] Job event: failed', {
      queue: queueName,
      jobId,
      reason: failedReason,
    });
    void alertQueue(queueName, `Job ${jobId} failed: ${failedReason || 'unknown reason'}`, 'warning');
  });

  events.on('retries-exhausted', ({ jobId }) => {
    logger.error('[BullMQ] Job event: retries exhausted', { queue: queueName, jobId });
    void alertQueue(queueName, `Retries exhausted for job ${jobId}`, 'error');
  });

  queueEvents.set(queueName, events);

  logger.info('[BullMQ] Queue events subscribed', { queue: queueName });

  return events;
}

// ============================
// Queue Statistics
// ============================

export interface QueueStats {
  waiting: number;
  active: number;
  completed: number;
  failed: number;
  delayed: number;
  paused: boolean;
}

export async function getQueueStats(queueName: QueueName): Promise<QueueStats> {
  const queue = getQueue(queueName);

  const [waiting, active, completed, failed, delayed, isPaused] = await Promise.all([
    queue.getWaitingCount(),
    queue.getActiveCount(),
    queue.getCompletedCount(),
    queue.getFailedCount(),
    queue.getDelayedCount(),
    queue.isPaused(),
  ]);

  return {
    waiting,
    active,
    completed,
    failed,
    delayed,
    paused: isPaused,
  };
}

/**
 * Get stats for all queues
 */
export async function getAllQueueStats(): Promise<Record<QueueName, QueueStats>> {
  const stats: Partial<Record<QueueName, QueueStats>> = {};

  for (const queueName of Object.values(QUEUE_NAMES)) {
    stats[queueName] = await getQueueStats(queueName);
  }

  return stats as Record<QueueName, QueueStats>;
}

// ============================
// Job Management
// ============================

/**
 * Get failed jobs from a queue
 */
export async function getFailedJobs(queueName: QueueName, limit = 50): Promise<Job[]> {
  const queue = getQueue(queueName);
  return queue.getFailed(0, limit - 1);
}

/**
 * Retry a failed job
 */
export async function retryJob(queueName: QueueName, jobId: string): Promise<void> {
  const queue = getQueue(queueName);
  const job = await queue.getJob(jobId);

  if (!job) {
    throw new Error(`Job ${jobId} not found in queue ${queueName}`);
  }

  await job.retry();
  logger.info('[BullMQ] Job retried', { queue: queueName, jobId });
}

/**
 * Remove a job from a queue
 */
export async function removeJob(queueName: QueueName, jobId: string): Promise<void> {
  const queue = getQueue(queueName);
  const job = await queue.getJob(jobId);

  if (!job) {
    throw new Error(`Job ${jobId} not found in queue ${queueName}`);
  }

  await job.remove();
  logger.info('[BullMQ] Job removed', { queue: queueName, jobId });
}

/**
 * Pause a queue (stop processing new jobs)
 */
export async function pauseQueue(queueName: QueueName): Promise<void> {
  const queue = getQueue(queueName);
  await queue.pause();
  logger.warn('[BullMQ] Queue paused', { queue: queueName });
}

/**
 * Resume a paused queue
 */
export async function resumeQueue(queueName: QueueName): Promise<void> {
  const queue = getQueue(queueName);
  await queue.resume();
  logger.info('[BullMQ] Queue resumed', { queue: queueName });
}

/**
 * Drain a queue (remove all waiting/delayed jobs)
 */
export async function drainQueue(queueName: QueueName): Promise<void> {
  const queue = getQueue(queueName);
  await queue.drain();
  logger.warn('[BullMQ] Queue drained', { queue: queueName });
}

// ============================
// Graceful Shutdown
// ============================

export async function shutdownQueues(): Promise<void> {
  logger.info('[BullMQ] Shutting down queues and workers...');

  // Close all workers
  for (const [name, worker] of workers.entries()) {
    logger.info('[BullMQ] Closing worker', { queue: name });
    await worker.close();
  }

  // Close all queue events
  for (const [name, events] of queueEvents.entries()) {
    logger.info('[BullMQ] Closing queue events', { queue: name });
    await events.close();
  }

  // Close all queues
  for (const [name, queue] of queues.entries()) {
    logger.info('[BullMQ] Closing queue', { queue: name });
    await queue.close();
  }

  // Close Redis connection
  if (redisConnection) {
    logger.info('[BullMQ] Closing Redis connection');
    await redisConnection.quit();
    redisConnection = null;
  }

  logger.info('[BullMQ] Shutdown complete');
}

// Handle process termination
if (process.env.NODE_ENV === 'production') {
  process.on('SIGTERM', async () => {
    logger.info('[BullMQ] SIGTERM received, shutting down gracefully...');
    await shutdownQueues();
    process.exit(0);
  });

  process.on('SIGINT', async () => {
    logger.info('[BullMQ] SIGINT received, shutting down gracefully...');
    await shutdownQueues();
    process.exit(0);
  });
}
