import 'server-only';
import { PgBoss } from 'pg-boss';
import { logger } from '@rareindex/shared';

/**
 * Thin pg-boss producer used by the admin console to enqueue pipeline jobs. Consumers live in
 * workers/ (agent D). Queue names and payloads are the contract:
 *   crawl.run      { connectorId, mode: 'incremental'|'backfill'|'probe', trigger: 'manual'|'retry'|'schedule', limit? }
 *   pipeline.normalize { connectorId? }
 *   valuation.run  { assetId? }
 */
export const QUEUES = {
  crawlRun: 'crawl.run',
  normalize: 'pipeline.normalize',
  valuation: 'valuation.run',
  indices: 'indices.run',
} as const;

let boss: PgBoss | null = null;
let starting: Promise<PgBoss> | null = null;

export async function getBoss(): Promise<PgBoss> {
  if (boss) return boss;
  if (!starting) {
    starting = (async () => {
      const b = new PgBoss({ connectionString: process.env.DATABASE_URL ?? 'postgres://localhost:5432/rareindex', application_name: 'rareindex-web-admin', max: 2, supervise: false, schedule: false });
      b.on('error', (err: Error) => logger.warn({ err: err.message }, 'pg-boss error'));
      await b.start();
      boss = b;
      return b;
    })();
  }
  return starting;
}

export async function enqueue(queue: string, data: Record<string, unknown>, opts: { singletonKey?: string; priority?: number } = {}): Promise<string | null> {
  const b = await getBoss();
  await b.createQueue(queue).catch(() => {});
  return b.send(queue, data, { ...(opts.singletonKey ? { singletonKey: opts.singletonKey } : {}), ...(opts.priority !== undefined ? { priority: opts.priority } : {}), retryLimit: 2, expireInSeconds: 3600 });
}

export async function queueDepths(): Promise<Array<{ name: string; queued: number; active: number; failed: number }>> {
  try {
    const b = await getBoss();
    const out: Array<{ name: string; queued: number; active: number; failed: number }> = [];
    for (const name of Object.values(QUEUES)) {
      const q = await b.getQueue(name).catch(() => null);
      if (!q) continue;
      const stats = (await (b as unknown as { getQueueStats?: (n: string) => Promise<{ queuedCount?: number; activeCount?: number; failedCount?: number }> }).getQueueStats?.(name).catch(() => null)) ?? null;
      out.push({ name, queued: stats?.queuedCount ?? 0, active: stats?.activeCount ?? 0, failed: stats?.failedCount ?? 0 });
    }
    return out;
  } catch (err) {
    logger.warn({ err: err instanceof Error ? err.message : String(err) }, 'queue depth unavailable');
    return [];
  }
}