/**
 * Typed BullMQ queue/worker factory.
 *
 * Usage:
 *   const thumbQueue = createQueue<ThumbnailJob>('thumbnail-processing');
 *   const thumbWorker = createWorker<ThumbnailJob>('thumbnail-processing', async (job) => {
 *     // ...
 *   });
 *
 * Workers and queues share a single ioredis connection (see ./connection.ts).
 * Graceful shutdown is registered once; on SIGINT/SIGTERM all queues + workers
 * close before the Redis connection is torn down.
 */

import { Queue, Worker, type Job, type Processor, type QueueOptions, type WorkerOptions } from 'bullmq';

import { loadConfig } from '@/app/config';
import { logger } from '@/shared/logger';

import { closeRedis, getRedis } from './connection';

export { getRedis, pingRedis, closeRedis } from './connection';
export type { Job } from 'bullmq';

const queues = new Set<Queue<unknown, unknown, string>>();
const workers = new Set<Worker<unknown, unknown, string>>();

const DEFAULT_JOB_OPTIONS: QueueOptions['defaultJobOptions'] = {
  attempts: 3,
  backoff: { type: 'exponential', delay: 1_000 },
  removeOnComplete: { age: 24 * 60 * 60, count: 1_000 },
  removeOnFail: { age: 7 * 24 * 60 * 60 },
};

export function createQueue<TJob>(name: string, options: Partial<QueueOptions> = {}): Queue<TJob> {
  const queue = new Queue<TJob>(name, {
    connection: getRedis(),
    defaultJobOptions: { ...DEFAULT_JOB_OPTIONS, ...(options.defaultJobOptions ?? {}) },
    ...options,
  });
  queues.add(queue as Queue<unknown, unknown, string>);
  logger.info('queue: registered', { name });
  return queue;
}

export function createWorker<TJob, TResult = void>(
  name: string,
  processor: Processor<TJob, TResult>,
  options: Partial<WorkerOptions> = {},
): Worker<TJob, TResult> {
  const { WORKER_CONCURRENCY } = loadConfig();
  const worker = new Worker<TJob, TResult>(name, processor, {
    connection: getRedis(),
    concurrency: WORKER_CONCURRENCY,
    ...options,
  });

  worker.on('failed', (job: Job<TJob> | undefined, err: Error) => {
    logger.warn('worker: job failed', {
      queue: name,
      jobId: job?.id,
      attempt: job?.attemptsMade,
      err: err.message,
    });
  });
  worker.on('error', (err: Error) => {
    logger.error('worker: error', { queue: name, err: err.message });
  });
  worker.on('ready', () => {
    logger.info('worker: ready', { queue: name, concurrency: WORKER_CONCURRENCY });
  });

  workers.add(worker as Worker<unknown, unknown, string>);
  return worker;
}

let shutdownRegistered = false;

export function registerGracefulShutdown(): void {
  if (shutdownRegistered) return;
  shutdownRegistered = true;

  const shutdown = async (signal: string): Promise<void> => {
    logger.info('queue: shutdown initiated', { signal });
    try {
      await Promise.all([...workers].map((w) => w.close()));
      await Promise.all([...queues].map((q) => q.close()));
      await closeRedis();
    } catch (err) {
      logger.error('queue: shutdown error', { err: err instanceof Error ? err.message : String(err) });
    }
    process.exit(0);
  };

  process.once('SIGINT', () => void shutdown('SIGINT'));
  process.once('SIGTERM', () => void shutdown('SIGTERM'));
}
