import { PgBoss } from 'pg-boss';
import { env, logger } from '@rareindex/shared';

/**
 * Queue abstraction (§140). Implemented with pg-boss (PostgreSQL-backed, schema `pgboss`).
 * Only this file knows pg-boss; handlers receive plain payloads so Redis Streams / Kafka can
 * replace it later.
 */
export const JOBS = {
  crawlRun: 'crawl.run',
  normalizeBatch: 'normalize.batch',
  resolveBatch: 'resolve.batch',
  valuationAsset: 'valuation.asset',
  valuationRebuild: 'valuation.rebuild',
  indicesDaily: 'indices.daily',
  snapshotsDaily: 'snapshots.daily',
  radarScan: 'radar.scan',
  healthCompute: 'health.compute',
  fxSync: 'fx.sync',
  benchmarksSync: 'benchmarks.sync',
  listingsExpire: 'listings.expire',
  accountJobs: 'account.jobs',
  imagesProcess: 'images.process',
  certsVerify: 'certs.verify',
  auctionsAssess: 'auctions.assess',
} as const;
export type JobName = (typeof JOBS)[keyof typeof JOBS];

export interface SendOptions {
  /** de-duplicate: only one queued/active job with this key */
  singletonKey?: string;
  /** collapse sends within this many seconds */
  singletonSeconds?: number;
  priority?: number;
  startAfterSeconds?: number;
  retryLimit?: number;
  expireInSeconds?: number;
}

export interface WorkOptions {
  batchSize?: number;
  concurrency?: number;
  pollingIntervalSeconds?: number;
}

export interface Queue {
  start(): Promise<void>;
  stop(): Promise<void>;
  send<T extends object>(name: JobName, data: T, opts?: SendOptions): Promise<string | null>;
  schedule(name: JobName, cron: string, data?: object, opts?: { tz?: string }): Promise<void>;
  work<T extends object>(name: JobName, opts: WorkOptions, handler: (data: T, job: { id: string; signal: AbortSignal }) => Promise<unknown>): Promise<void>;
  stats(): Promise<Array<{ name: string; queued: number; active: number; failed: number }>>;
}

export function createQueue(): Queue {
  const boss = new PgBoss({
    connectionString: env().DATABASE_URL,
    schema: 'pgboss',
    max: 6,
    monitorIntervalSeconds: 60,
  });
  boss.on('error', (err) => logger.error({ err }, 'pg-boss error'));
  const log = logger.child({ component: 'queue' });
  const created = new Set<string>();

  async function ensure(name: string, opts: { retryLimit?: number; expireInSeconds?: number } = {}) {
    if (created.has(name)) return;
    try {
      await boss.createQueue(name, { retryLimit: opts.retryLimit ?? 3, retryDelay: 30, retryBackoff: true, expireInSeconds: opts.expireInSeconds ?? 3600, deleteAfterSeconds: 3 * 86_400 });
    } catch (err) {
      const msg = err instanceof Error ? err.message : String(err);
      if (!/already exists|duplicate key/i.test(msg)) throw err;
    }
    created.add(name);
  }

  return {
    async start() {
      await boss.start();
      for (const name of Object.values(JOBS)) await ensure(name);
      log.info('queue started');
    },
    async stop() {
      await boss.stop({ graceful: true, timeout: 20_000 });
    },
    async send(name, data, opts = {}) {
      await ensure(name);
      // pg-boss asserts on explicitly-undefined option keys (e.g. "priority must be an integer"): only pass defined ones.
      const sendOpts: Record<string, unknown> = {};
      if (opts.singletonKey !== undefined) sendOpts.singletonKey = opts.singletonKey;
      if (opts.singletonSeconds !== undefined) sendOpts.singletonSeconds = opts.singletonSeconds;
      if (opts.priority !== undefined) sendOpts.priority = Math.trunc(opts.priority);
      if (opts.startAfterSeconds !== undefined) sendOpts.startAfter = opts.startAfterSeconds;
      if (opts.retryLimit !== undefined) sendOpts.retryLimit = opts.retryLimit;
      if (opts.expireInSeconds !== undefined) sendOpts.expireInSeconds = opts.expireInSeconds;
      return boss.send(name, data, sendOpts);
    },
    async schedule(name, cron, data = {}, opts = {}) {
      await ensure(name);
      await boss.schedule(name, cron, data, { tz: opts.tz ?? 'UTC' });
    },
    async work(name, opts, handler) {
      await ensure(name);
      await boss.work(
        name,
        { batchSize: opts.batchSize ?? 1, localConcurrency: opts.concurrency ?? 1, pollingIntervalSeconds: opts.pollingIntervalSeconds ?? 2 },
        async (jobs) => {
          for (const job of jobs) {
            try {
              await handler(job.data as never, { id: job.id, signal: job.signal });
            } catch (err) {
              log.error({ err, job: name, id: job.id }, 'job failed');
              throw err;
            }
          }
        },
      );
    },
    async stats() {
      const out: Array<{ name: string; queued: number; active: number; failed: number }> = [];
      for (const name of Object.values(JOBS)) {
        const q = await boss.getQueue(name);
        if (q) out.push({ name, queued: Number((q as { queuedCount?: number }).queuedCount ?? 0), active: Number((q as { activeCount?: number }).activeCount ?? 0), failed: Number((q as { failedCount?: number }).failedCount ?? 0) });
      }
      return out;
    },
  };
}