import { closeDb, db, migrate, sql } from "@websensor/db";
import { closeDispatcher } from "@websensor/connectors";
import { config, factoryConfig, log } from "./config";
import { claimSeeds, evaluateShadows, factoryStats, processSeedSafe, seedFromCoverage, seedFromFiles, type ProcessOutcome } from "./factory";
import { closeRedis, getRedis } from "./redis";

/**
 * Source Factory process (`websensor-factory`): a slow, polite background worker separate from the poller.
 *   N workers   — each claims one queued seed at a time (systemic organizations first) and runs deep discovery +
 *                 shadow creation; a slow organization never stalls the others (per-organization deadline).
 *   every 15 min — evaluate shadow sensors (accept / reject / defer)
 *   every 24 h   — re-seed from the coverage universes (new members become seeds automatically)
 * Heartbeat: Redis `ws:factory:status` (processed, queued, in-flight, last outcomes).
 */
async function heartbeat(extra: Record<string, unknown>): Promise<void> {
  try {
    await getRedis().set("ws:factory:status", JSON.stringify({ ...extra, at: new Date().toISOString(), pid: process.pid, version: config.version }), "EX", 120);
  } catch {
    // best effort
  }
}

async function main(): Promise<void> {
  log.info({ env: config.env, version: config.version, concurrency: factoryConfig.concurrency, budget: factoryConfig.budget, deadlineMs: factoryConfig.deadlineMs }, "websensor factory starting");
  const applied = await migrate(config.databaseUrl);
  if (applied.length) log.info({ applied }, "migrations applied");
  if (!factoryConfig.enabled) {
    log.warn("WS_FACTORY=0 — factory idle");
    setInterval(() => void heartbeat({ idle: true }), 60_000);
    return;
  }
  let stopping = false;
  let processed = 0;
  let inflight = 0;
  let lastSeed = 0;
  let lastEval = 0;
  let maintenanceBusy = false;
  const recent: { seed: string; status: string; candidates: number; shadow: number; requests: number; ms: number }[] = [];
  const note = (o: ProcessOutcome): void => {
    processed++;
    recent.unshift({ seed: o.seedId, status: o.status, candidates: o.candidates, shadow: o.shadow, requests: o.requests, ms: o.durationMs });
    if (recent.length > 12) recent.pop();
  };

  const seed = async (): Promise<void> => {
    try {
      const a = await seedFromFiles();
      const b = await seedFromCoverage({ mode: "hinted" });
      log.info({ files: a, coverage: { members: b.members, seeds: b.seeds, inserted: b.inserted, issues: b.issues.length } }, "factory: seeds refreshed");
      for (const i of b.issues) log.warn({ issue: i }, "coverage file issue");
    } catch (e) {
      log.error({ err: (e as Error).message }, "factory: seeding failed");
    }
  };

  // Continuous worker pool: each worker claims one seed, processes it, and immediately claims the next.
  const worker = async (n: number): Promise<void> => {
    while (!stopping) {
      let seeds: Awaited<ReturnType<typeof claimSeeds>> = [];
      try {
        seeds = await claimSeeds(1);
      } catch (e) {
        log.warn({ worker: n, err: (e as Error).message }, "factory: claim failed");
      }
      if (!seeds.length) {
        await new Promise((r) => setTimeout(r, factoryConfig.tickSeconds * 1000 + Math.random() * 2000));
        continue;
      }
      inflight++;
      try {
        note(await processSeedSafe(seeds[0]!));
      } finally {
        inflight--;
      }
    }
  };

  const maintenance = async (): Promise<void> => {
    if (stopping || maintenanceBusy) return;
    maintenanceBusy = true;
    try {
      if (Date.now() - lastSeed > 24 * 3600e3) {
        lastSeed = Date.now();
        await seed();
      }
      if (Date.now() - lastEval > 15 * 60e3) {
        lastEval = Date.now();
        await evaluateShadows().catch((e) => log.warn({ err: (e as Error).message }, "factory: shadow evaluation failed"));
      }
      const queued = Number((await db.execute<{ n: string }>(sql`select count(*)::text as n from factory_seeds where status = 'queued'`)).rows[0]?.n ?? 0);
      await heartbeat({ processed, queued, inflight, lastBatch: recent, concurrency: factoryConfig.concurrency });
      if (!inflight && !queued && processed % 50 === 0) {
        const st = await factoryStats();
        log.info({ seeds: st.seeds, candidates: st.candidates, shadow: st.shadow }, "factory: idle");
      }
    } catch (e) {
      log.error({ err: (e as Error).stack ?? (e as Error).message }, "factory maintenance failed");
    } finally {
      maintenanceBusy = false;
    }
  };

  await heartbeat({ starting: true });
  await maintenance();
  const timer = setInterval(() => void maintenance(), 15_000);
  const workers = Array.from({ length: factoryConfig.concurrency }, (_, i) => worker(i));

  const shutdown = async (signal: string): Promise<void> => {
    if (stopping) return;
    stopping = true;
    log.info({ signal, inflight }, "factory shutting down");
    clearInterval(timer);
    const deadline = Date.now() + 90_000;
    while (inflight > 0 && Date.now() < deadline) await new Promise((r) => setTimeout(r, 250));
    // Seeds still marked `discovering` are reclaimed automatically after 15 min (claim query; per-seed deadline is 2 min).
    await closeDispatcher();
    await closeRedis();
    await closeDb();
    process.exit(0);
  };
  process.on("SIGINT", () => void shutdown("SIGINT"));
  process.on("SIGTERM", () => void shutdown("SIGTERM"));
  process.on("unhandledRejection", (e) => log.error({ err: e instanceof Error ? e.stack : String(e) }, "unhandled rejection"));
  await Promise.all(workers);
}

main().catch((e) => {
  log.fatal({ err: (e as Error).stack ?? String(e) }, "factory failed to start");
  process.exit(1);
});