/**
 * Enqueue worker jobs by hand (the worker must be running to execute them).
 *
 *   tsx workers/cli.ts run <connector> [--mode full|incremental|dry_run] [--max-minutes N] [--max-records N] [--reset-cursor]
 *   tsx workers/cli.ts counters [--then-rank]
 *   tsx workers/cli.ts rank
 *   tsx workers/cli.ts health
 *   tsx workers/cli.ts schedules            list registered schedules
 *   tsx workers/cli.ts queues               queue depths
 */
import { DATABASE_URL, JOBS, PGBOSS_SCHEMA } from './lib/env.js';
import { PgBoss } from 'pg-boss';

const [, , cmd, ...rest] = process.argv;
const flags: Record<string, string | boolean> = {};
const positional: string[] = [];
for (let i = 0; i < rest.length; i++) {
  const a = rest[i]!;
  if (a.startsWith('--')) {
    const next = rest[i + 1];
    if (next && !next.startsWith('--')) {
      flags[a.slice(2)] = next;
      i++;
    } else flags[a.slice(2)] = true;
  } else positional.push(a);
}

const boss = new PgBoss({ connectionString: DATABASE_URL, schema: PGBOSS_SCHEMA, application_name: 'cancerindex-worker-cli', supervise: false, schedule: false, max: 2 });
boss.on('error', (e) => console.error(e));
await boss.start();

async function ensure(name: string) {
  if (!(await boss.getQueue(name))) await boss.createQueue(name, { policy: 'stately', retryLimit: 0, expireInSeconds: 3 * 3600 });
}

try {
  switch (cmd) {
    case 'run': {
      const id = positional[0];
      if (!id) throw new Error('usage: run <connector>');
      // Omit undefined keys — pg-boss rejects undefined option values.
      const data: Record<string, unknown> = { id, requestedBy: 'cli' };
      if (flags.mode) data.mode = flags.mode;
      if (flags['max-minutes']) data.maxMinutes = Number(flags['max-minutes']);
      if (flags['max-records']) data.maxRecords = Number(flags['max-records']);
      if (flags['reset-cursor']) data.resetCursor = true;
      await ensure(JOBS.connectorRun);
      const jobId = await boss.send(JOBS.connectorRun, data, { singletonKey: id });
      console.log(jobId ? `queued ${JOBS.connectorRun} ${id} → job ${jobId}` : `not queued: a run for ${id} is already queued/active (singleton)`);
      break;
    }
    case 'counters': {
      await ensure(JOBS.counters);
      const jobId = await boss.send(JOBS.counters, { thenRank: !!flags['then-rank'], requestedBy: 'cli' }, { singletonKey: 'counters' });
      console.log(jobId ? `queued ${JOBS.counters} → ${jobId}` : 'not queued (already pending)');
      break;
    }
    case 'rank': {
      await ensure(JOBS.rank);
      const jobId = await boss.send(JOBS.rank, { requestedBy: 'cli' }, { singletonKey: 'rank' });
      console.log(jobId ? `queued ${JOBS.rank} → ${jobId}` : 'not queued (already pending)');
      break;
    }
    case 'health': {
      await ensure(JOBS.healthProbe);
      const jobId = await boss.send(JOBS.healthProbe, { requestedBy: 'cli' }, { singletonKey: 'health' });
      console.log(jobId ? `queued ${JOBS.healthProbe} → ${jobId}` : 'not queued (already pending)');
      break;
    }
    case 'schedules': {
      for (const s of await boss.getSchedules()) console.log(`${s.name.padEnd(22)} key=${s.key.padEnd(16)} cron=${s.cron.padEnd(14)} tz=${s.timezone} data=${JSON.stringify(s.data ?? {})}`);
      break;
    }
    case 'queues': {
      for (const q of await boss.getQueues()) console.log(`${q.name.padEnd(22)} policy=${q.policy} queued=${q.queuedCount} active=${q.activeCount} failed=${q.failedCount} total=${q.totalCount}`);
      break;
    }
    default:
      console.log('usage: tsx workers/cli.ts <run <connector>|counters|rank|health|schedules|queues>');
  }
} finally {
  await boss.stop({ graceful: false, close: true, timeout: 5000 });
}