import { loadDotenv } from './lib/env.ts';
loadDotenv();
process.env.RI_SERVICE ??= 'rareindex-worker';

import { env, logger } from '@rareindex/shared';
import { createQueue, JOBS } from './lib/queue.ts';
import { closeDb, db } from './lib/db.ts';
import { flushCosts } from './lib/costs.ts';
import { runCrawl } from './crawler/run.ts';
import { scheduleDueCrawls } from './crawler/scheduler.ts';
import { normalizeBatch } from './normalizer/index.ts';
import { sql } from 'drizzle-orm';
import { pendingCount, resolveBatch } from './entity-resolution/index.ts';
import { assetsNeedingValuation, computePremiums, valueMany } from './valuation/run.ts';
import { runCategorySnapshots, runIndices, runRadar } from './indices/run.ts';
import { syncFx } from './fx.ts';
import { syncBenchmarks } from './benchmarks.ts';
import { computeHealth } from './health.ts';
import { expireListings } from './listings-expire.ts';
import { processImages } from './image-processing/index.ts';
import { verifyCertificates } from './certs-verify.ts';
import { assessLots } from './auctions/assess.ts';

/**
 * RareIndex worker process (PM2: rareindex-worker). One process runs the queue handlers and the
 * scheduler; scale out by starting more processes against the same database.
 */
export async function startWorker(): Promise<() => Promise<void>> {
  const log = logger.child({ component: 'worker' });
  const queue = createQueue();
  await queue.start();
  // Single-worker deployment: jobs left 'active' by a previous process (restart/crash) would wait for
  // their expiry (hours). Re-queue them so crawls resume immediately.
  await db().execute(sql`update pgboss.job set state = 'retry', started_on = null where state = 'active' and started_on < now() - interval '90 seconds'`);
  // Runs left 'running' by a dead process can never finish: close them so the per-connector guard below works.
  await db().execute(sql`update connector_runs set status = 'failed', finished_at = now(), error = 'worker restarted' where status = 'running'`);
  // Collapse duplicate queued crawls per connector (keep the oldest).
  await db().execute(sql`delete from pgboss.job j using pgboss.job k where j.name = 'crawl.run' and k.name = 'crawl.run' and j.state in ('created','retry') and k.state in ('created','retry') and j.data->>'connectorId' = k.data->>'connectorId' and j.created_on > k.created_on`);
  const concurrency = env().WORKER_CONCURRENCY;

  // ---- handlers ----
  await queue.work<{ connectorId: string; mode?: 'incremental' | 'backfill' | 'probe'; limit?: number; trigger?: string }>(JOBS.crawlRun, { concurrency: Math.max(1, Math.floor(concurrency / 2)), pollingIntervalSeconds: 5 }, async (data) => {
    // One crawl per connector at a time (duplicates waste engine credits and crawl slots).
    const [running] = (await db().execute(sql`select 1 from connector_runs where connector_id = ${data.connectorId} and status = 'running' and started_at > now() - interval '6 hours' limit 1`)) as unknown as unknown[];
    if (running) {
      log.info({ connector: data.connectorId }, 'crawl already running; skipping duplicate job');
      return;
    }
    const res = await runCrawl(data.connectorId, { mode: data.mode ?? 'incremental', limit: data.limit, trigger: data.trigger ?? 'schedule' });
    if (res.recordsRaw > 0) await queue.send(JOBS.normalizeBatch, { connectorId: data.connectorId }, { singletonKey: `normalize:${data.connectorId}`, singletonSeconds: 30 });
  });
  await queue.work<{ connectorId?: string }>(JOBS.normalizeBatch, { concurrency: 2, pollingIntervalSeconds: 5 }, async (data) => {
    let total = 0;
    let more = false;
    for (let i = 0; i < 40; i++) {
      const r = await normalizeBatch({ connectorId: data.connectorId, limit: 500 });
      total += r.processed;
      more = r.processed >= 500;
      if (!more) break;
    }
    // Large crawls (100k+ raw rows) exceed one job's budget: chain another job instead of leaving rows unprocessed.
    if (more) await queue.send(JOBS.normalizeBatch, { connectorId: data.connectorId }, { singletonKey: `normalize:${data.connectorId ?? 'all'}:${Date.now()}`, startAfterSeconds: 1 });
    if (total > 0) for (let k = 0; k < Number(process.env.RESOLVE_CONCURRENCY ?? 3); k++) await queue.send(JOBS.resolveBatch, {}, { singletonKey: `resolve:${k}`, singletonSeconds: 30 });
  });
  const RESOLVE_WORKERS = Number(process.env.RESOLVE_CONCURRENCY ?? 3);
  // Release rows claimed by a previous worker process that died mid-batch.
  await db().execute(sql`update normalized_records set status = 'pending' where status = 'processing'`);
  await queue.work<Record<string, never>>(JOBS.resolveBatch, { concurrency: RESOLVE_WORKERS, pollingIntervalSeconds: 5 }, async () => {
    const touched = new Set<string>();
    for (let i = 0; i < 40; i++) {
      const r = await resolveBatch({ limit: 500 });
      for (const a of r.touchedAssets) touched.add(a);
      if (r.processed < 500) break;
    }
    if ((await pendingCount()) > 0) for (let k = 0; k < RESOLVE_WORKERS; k++) await queue.send(JOBS.resolveBatch, {}, { singletonKey: `resolve:${k}`, singletonSeconds: 30, startAfterSeconds: 5 });
    if (touched.size) await queue.send(JOBS.valuationAsset, { assetIds: [...touched].slice(0, 5000) }, { singletonKey: `value:${Date.now()}` });
    if (touched.size) await queue.send(JOBS.imagesProcess, { assetIds: [...touched].slice(0, 5000), limit: 5000 }, { singletonKey: `images:${Date.now()}` });
  });
  await queue.work<{ assetIds: string[] }>(JOBS.valuationAsset, { concurrency: 1, pollingIntervalSeconds: 5 }, async (data) => {
    await valueMany(data.assetIds, { concurrency: 4 });
  });
  await queue.work<{ all?: boolean; rebuildHistory?: boolean }>(JOBS.valuationRebuild, { concurrency: 1, pollingIntervalSeconds: 30 }, async (data) => {
    await computePremiums();
    const ids = await assetsNeedingValuation({ all: data.all, limit: 200_000 });
    await valueMany(ids, { rebuildHistory: data.rebuildHistory, concurrency: 4 });
  });
  await queue.work(JOBS.indicesDaily, { concurrency: 1, pollingIntervalSeconds: 30 }, async () => {
    await runIndices();
  });
  await queue.work(JOBS.snapshotsDaily, { concurrency: 1, pollingIntervalSeconds: 30 }, async () => {
    await runCategorySnapshots();
  });
  await queue.work(JOBS.radarScan, { concurrency: 1, pollingIntervalSeconds: 30 }, async () => {
    await runRadar();
  });
  await queue.work(JOBS.healthCompute, { concurrency: 1, pollingIntervalSeconds: 30 }, async () => {
    await computeHealth();
  });
  await queue.work<{ backfill?: boolean }>(JOBS.fxSync, { concurrency: 1, pollingIntervalSeconds: 30 }, async (data) => {
    await syncFx({ backfill: data.backfill });
  });
  await queue.work(JOBS.benchmarksSync, { concurrency: 1, pollingIntervalSeconds: 60 }, async () => {
    await syncBenchmarks();
  });
  await queue.work<{ assetIds?: string[]; limit?: number; recheck?: boolean }>(JOBS.imagesProcess, { concurrency: 1, pollingIntervalSeconds: 30 }, async (data) => {
    await processImages({ assetIds: data.assetIds, limit: data.limit ?? 2000, recheck: data.recheck ?? false });
  });
  await queue.work(JOBS.listingsExpire, { concurrency: 1, pollingIntervalSeconds: 60 }, async () => {
    await expireListings();
  });
  await queue.work<{ limit?: number; grader?: string }>(JOBS.certsVerify, { concurrency: 1, pollingIntervalSeconds: 60 }, async (data) => {
    await verifyCertificates({ limit: data.limit ?? 100, grader: data.grader });
  });
  await queue.work<{ limit?: number; all?: boolean }>(JOBS.auctionsAssess, { concurrency: 1, pollingIntervalSeconds: 60 }, async (data) => {
    await assessLots({ limit: data.limit ?? 20_000, all: data.all ?? false });
  });
  await queue.work(JOBS.accountJobs, { concurrency: 1, pollingIntervalSeconds: 60 }, async () => {
    try {
      const accountModule = './account/index.ts'; // provided by the account module when present
      const mod = (await import(accountModule)) as { runAccountJobs?: (d: ReturnType<typeof db>) => Promise<unknown> };
      if (mod.runAccountJobs) await mod.runAccountJobs(db());
    } catch (err) {
      if (!/Cannot find module|ERR_MODULE_NOT_FOUND/.test(String(err))) log.error({ err }, 'account jobs failed');
    }
  });

  // ---- schedules (UTC) ----
  await queue.schedule(JOBS.fxSync, '15 16 * * 1-5', {}); // after ECB 16:00 CET publication
  await queue.schedule(JOBS.benchmarksSync, '30 22 * * *', {});
  await queue.schedule(JOBS.valuationRebuild, '0 3 * * *', { all: false });
  await queue.schedule(JOBS.snapshotsDaily, '20 4 * * *', {});
  await queue.schedule(JOBS.indicesDaily, '40 4 * * *', {});
  await queue.schedule(JOBS.radarScan, '0 5 * * *', {});
  await queue.schedule(JOBS.healthCompute, '*/30 * * * *', {});
  await queue.schedule(JOBS.listingsExpire, '10 * * * *', {});
  await queue.schedule(JOBS.certsVerify, '25 * * * *', { limit: 100 }); // SPEC §22: verify cert numbers seen on marketplaces against public grader lookups
  await queue.schedule(JOBS.imagesProcess, '50 5 * * *', { limit: 20000, recheck: true });
  await queue.schedule(JOBS.imagesProcess, '*/15 * * * *', { limit: 3000 });
  await queue.schedule(JOBS.accountJobs, '*/10 * * * *', {});
  await queue.schedule(JOBS.auctionsAssess, '*/20 * * * *', { limit: 20000 }); // §33–§35: all-in bid vs RIV on live lots

  // ---- crawl scheduler loop ----
  const tick = async () => {
    try {
      await scheduleDueCrawls(queue);
      // Normalize whatever raw rows exist even while long crawls are still running (sales show up progressively).
      await queue.send(JOBS.normalizeBatch, {}, { singletonKey: 'normalize:periodic', singletonSeconds: 120 });
    } catch (err) {
      log.error({ err }, 'scheduler tick failed');
    }
  };
  await tick();
  if ((await pendingCount()) > 0) for (let k = 0; k < RESOLVE_WORKERS; k++) await queue.send(JOBS.resolveBatch, {}, { singletonKey: `resolve:${k}`, singletonSeconds: 30 });
  const timer = setInterval(tick, 60_000);
  // hourly incremental valuation for assets with fresh evidence
  const hourly = setInterval(() => void queue.send(JOBS.valuationRebuild, { all: false }, { singletonKey: 'valuation:hourly', singletonSeconds: 3000 }), 3600_000);
  log.info({ concurrency }, 'worker started');

  return async () => {
    clearInterval(timer);
    clearInterval(hourly);
    await queue.stop();
    await flushCosts();
    await closeDb();
    log.info('worker stopped');
  };
}

const isMain = process.argv[1] && /workers\/main\.ts$/.test(process.argv[1]);
if (isMain) {
  startWorker()
    .then((stop) => {
      const shutdown = () => void stop().then(() => process.exit(0));
      process.on('SIGINT', shutdown);
      process.on('SIGTERM', shutdown);
    })
    .catch((err) => {
      logger.error({ err }, 'worker failed to start');
      process.exit(1);
    });
}