/**
 * main.ts — entrypoint for the investigator runtime.
 *
 * Lifecycle:
 *  1. Connect to Postgres pool + a dedicated LISTEN client.
 *  2. On every NOTIFY (or on a slow-poll fallback) attempt to claim a job.
 *  3. Claim uses `UPDATE … WHERE status='queued' RETURNING *` so two workers
 *     can race safely.
 *  4. Dispatch via the orchestrator and update status.
 *
 * The healthcheck file at /tmp/healthy is touched once per loop so docker
 * declares us unhealthy if the LISTEN connection dies.
 */
import { utimes, writeFile } from "node:fs/promises";
import { audit } from "./lib/audit";
import { env } from "./lib/env";
import { close, listen, pool, query } from "./lib/pg";
import { dispatch, type InvestigationJob } from "./orchestrator";

const HEALTH_FILE = "/tmp/healthy";
const POLL_INTERVAL_MS = 15_000;
const CHANNEL = "investigation_jobs";

let shuttingDown = false;
let activeWorkers = 0;

async function touchHealth(): Promise<void> {
  try {
    const now = new Date();
    await writeFile(HEALTH_FILE, `${now.toISOString()} ${env.WORKER_ID}\n`);
    await utimes(HEALTH_FILE, now, now);
  } catch {
    /* docker healthcheck will trip on its own */
  }
}

async function claimNextJob(): Promise<InvestigationJob | null> {
  const rows = await query<InvestigationJob>(
    `UPDATE public.investigation_jobs
        SET status = 'running', worker_id = $1, started_at = NOW()
      WHERE job_id = (
        SELECT job_id FROM public.investigation_jobs
         WHERE status = 'queued'
         ORDER BY created_at ASC
         LIMIT 1
         FOR UPDATE SKIP LOCKED
      )
      RETURNING job_id, kind, payload, triggered_by`,
    [env.WORKER_ID],
  );
  return rows[0] ?? null;
}

async function drainOnce(): Promise<void> {
  if (shuttingDown) return;
  if (activeWorkers >= env.MAX_PARALLEL_WORKERS) return;
  const job = await claimNextJob();
  if (!job) return;
  activeWorkers++;
  // Don't await — let the dispatcher run in parallel and tick the loop.
  void dispatch(job, env.WORKER_ID).finally(() => {
    activeWorkers--;
    // Immediately check the queue again — bursty notifications shouldn't
    // sit idle while we wait for the next NOTIFY.
    void drainOnce();
  });
  // Try to fill the rest of the worker slots.
  if (activeWorkers < env.MAX_PARALLEL_WORKERS) void drainOnce();
}

async function main(): Promise<void> {
  await audit({ event: "runtime_starting", worker_id: env.WORKER_ID, max_parallel: env.MAX_PARALLEL_WORKERS });

  // Mark healthy NOW so the docker healthcheck doesn't trip during cold start.
  await touchHealth();

  // 1. LISTEN on investigation_jobs.
  const listenClient = await listen(CHANNEL, () => { void drainOnce(); });
  await audit({ event: "listening", channel: CHANNEL });

  // 2. Slow poll as a safety net (handles dropped NOTIFYs / restart races).
  const poller = setInterval(() => { void drainOnce(); void touchHealth(); }, POLL_INTERVAL_MS);

  // 3. Initial drain — any jobs queued during downtime get processed now.
  await drainOnce();

  // Graceful shutdown
  const shutdown = async (sig: string): Promise<void> => {
    if (shuttingDown) return;
    shuttingDown = true;
    console.error(`[runtime] received ${sig}, draining...`);
    clearInterval(poller);
    try {
      // Wait for in-flight workers to finish, but no longer than JOB_TIMEOUT.
      const waitUntil = Date.now() + env.JOB_TIMEOUT_SECONDS * 1000;
      while (activeWorkers > 0 && Date.now() < waitUntil) {
        await new Promise((r) => setTimeout(r, 250));
      }
      await listenClient.end().catch(() => undefined);
      await close();
    } finally {
      await audit({ event: "runtime_stopped", signal: sig });
      process.exit(0);
    }
  };
  process.on("SIGTERM", () => { void shutdown("SIGTERM"); });
  process.on("SIGINT",  () => { void shutdown("SIGINT");  });
}

main().catch(async (e) => {
  console.error(`[runtime] fatal: ${(e as Error).stack ?? e}`);
  await audit({ event: "runtime_fatal", error: String(e) });
  await pool.end().catch(() => undefined);
  process.exit(1);
});
