import { spawn, type StdioNull, type StdioPipe } from "node:child_process";
import { sql } from "drizzle-orm";
import { auditReport, db, pool } from "../db";
import { assertEnv } from "../lib/env";
import { parseReport } from "../lib/report";

/**
 * The audit worker.
 *
 * Postgres is the broker. `FOR UPDATE SKIP LOCKED` is what makes that honest:
 * two workers claiming at the same instant each get a disjoint set of rows,
 * because the second one steps over what the first has locked instead of
 * blocking on it or, worse, reading it as available.
 */

/** Ten audits at once. The binary caps its own concurrency at 6 per run. */
const MAX_IN_FLIGHT = 10;

/** Pages per audit — the tool's own default, stated here so it is visible. */
const SAMPLE = 10;

/**
 * A hard ceiling per audit, below the five minutes after which the sweeper
 * declares a running job lost. The order matters: a job we kill is failed by us
 * with a reason, never swept out from under a live process.
 */
const AUDIT_TIMEOUT_MS = 4 * 60 * 1000;

/** Idle poll. Short enough to feel immediate, long enough to be free. */
const POLL_MS = 2000;

const SWEEP_MS = 60 * 1000;

/**
 * When a `running` job is declared lost.
 *
 * Derived from the timeout rather than written beside it: the two were a
 * literal `4 minutes` and a literal `5 minutes` that only a comment kept in
 * step, and raising the first without the second would have the sweeper killing
 * audits that are still working.
 */
const SWEEP_AFTER_S = AUDIT_TIMEOUT_MS / 1000 + 60;

/** A report is a few hundred kilobytes; past this, something is wrong. */
const MAX_STDOUT_BYTES = 16 * 1024 * 1024;

const BIN = process.env.HSC_BIN ?? "nextjs-seo-prerender";

let running = true;
const inFlight = new Set<Promise<void>>();

type ClaimedJob = { id: string; origin: string };

/**
 * Claim up to `limit` queued jobs and mark them running, in one statement.
 *
 * One statement because two — a SELECT then an UPDATE — would leave a window
 * in which a crash loses the rows or a second worker takes them.
 */
async function claim(limit: number): Promise<ClaimedJob[]> {
  const result = await db.execute<ClaimedJob>(sql`
    UPDATE audit_job
       SET status = 'running', started_at = now()
     WHERE id IN (
       SELECT id
         FROM audit_job
        WHERE status = 'queued'
        ORDER BY created_at
          FOR UPDATE SKIP LOCKED
        LIMIT ${limit}
     )
    RETURNING id, origin
  `);
  return result.rows;
}

type RunOutcome = { code: number; stdout: string; stderr: string; timedOut: boolean };

/** Run the binary once and collect its stdout, which it writes exactly once. */
function runAudit(origin: string): Promise<RunOutcome> {
  return new Promise((resolve) => {
    // A tuple, not an array: it is what picks the spawn overload whose stdout
    // and stderr are streams rather than possibly-null.
    const stdio: [StdioNull, StdioPipe, StdioPipe] = ["ignore", "pipe", "pipe"];

    const child = spawn(BIN, [origin, "--json", "--sample", String(SAMPLE)], {
      stdio,
      // Nothing else from our environment is any of the binary's business, and
      // a DATABASE_URL leaked into a subprocess is a leak we chose.
      env: {
        PATH: process.env.PATH ?? "/usr/local/bin:/usr/bin:/bin",
        NODE_ENV: process.env.NODE_ENV,
      },
    });

    // Buffers, not a growing string: a multi-byte character split across two
    // chunks decodes to two replacement characters if each chunk is decoded on
    // its own, and the damage lands inside a JSON string — so the document still
    // parses, and a mangled URL is stored for good.
    const chunks: Buffer[] = [];
    let bytes = 0;
    let stderr = "";
    let timedOut = false;

    const timer = setTimeout(() => {
      timedOut = true;
      child.kill("SIGKILL");
    }, AUDIT_TIMEOUT_MS);

    child.stdout.on("data", (chunk: Buffer) => {
      if (bytes >= MAX_STDOUT_BYTES) return;
      chunks.push(chunk);
      bytes += chunk.length;
    });
    // Kept for the failure message: the binary explains a fatal on stderr.
    child.stderr.on("data", (chunk: Buffer) => {
      if (stderr.length < 8192) stderr += chunk.toString("utf8");
    });

    child.on("error", (error) => {
      clearTimeout(timer);
      resolve({ code: -1, stdout: "", stderr: error.message, timedOut });
    });
    child.on("close", (code) => {
      clearTimeout(timer);
      resolve({ code: code ?? -1, stdout: Buffer.concat(chunks).toString("utf8"), stderr, timedOut });
    });
  });
}

/** Thrown to roll a transaction back, never to signal a failure. */
class Swept extends Error {}

/**
 * Settle a job we are still holding.
 *
 * `AND status = 'running'` is the whole point: a job the sweeper has already
 * declared lost belongs to nobody, and writing to it would let a process that
 * lost the race publish an outcome the visitor was already told about.
 */
async function markFailed(id: string, reason: string): Promise<void> {
  await db.execute(sql`
    UPDATE audit_job
       SET status = 'failed', error = ${reason.slice(0, 500)}, finished_at = now()
     WHERE id = ${id} AND status = 'running'
  `);
}

async function runJob(job: ClaimedJob): Promise<void> {
  let outcome: RunOutcome;
  try {
    outcome = await runAudit(job.origin);
  } catch (error) {
    await markFailed(job.id, error instanceof Error ? error.message : "the audit could not start");
    return;
  }

  if (outcome.timedOut) {
    await markFailed(job.id, `l'audit a dépassé ${AUDIT_TIMEOUT_MS / 1000} secondes`);
    return;
  }
  // 130 is SIGINT: the run was cut short, so its verdict is not a verdict.
  if (outcome.code === 130) {
    await markFailed(job.id, "l'audit a été interrompu");
    return;
  }
  // 0 and 1 are a complete run; 2 is a run that could not happen and still
  // writes a JSON document, which is a result worth keeping and showing.
  if (![0, 1, 2].includes(outcome.code)) {
    const detail = outcome.stderr.trim().split("\n").pop() ?? `code ${outcome.code}`;
    await markFailed(job.id, detail);
    return;
  }

  const report = parseReport(outcome.stdout);
  if (!report) {
    await markFailed(job.id, "l'audit n'a pas produit de rapport lisible");
    return;
  }

  const fatal = report.fatal !== null;
  const counters = fatal
    ? { blocking: 0, warnings: 0, checked: 0, total: 0 }
    : {
        blocking: report.summary.blocking,
        warnings: report.warnings,
        // Pages the tool could read, not URLs it fetched: a run answered
        // entirely in redirects verified nothing, and the column has to say so
        // for the list to render the right verdict without opening the JSON.
        checked: report.summary.checked,
        total: report.discovery.total,
      };

  try {
    await db.transaction(async (tx) => {
      // Claim the outcome before writing it. If the sweeper got here first the
      // job is `failed`, the visitor has been told so, and the honest thing is
      // to drop this result rather than overwrite a verdict they have seen.
      const settled = await tx.execute(sql`
        UPDATE audit_job SET status = 'done', finished_at = now()
         WHERE id = ${job.id} AND status = 'running'
        RETURNING id
      `);
      if (settled.rows.length === 0) throw new Swept();

      await tx.insert(auditReport).values({
        jobId: job.id,
        exitCode: report.exit_code,
        ...counters,
        report,
      });
    });
  } catch (error) {
    if (!(error instanceof Swept)) throw error;
    console.log(`worker: job ${job.id} was swept before it finished, its result is dropped`);
  }
}

/**
 * Jobs whose worker died mid-run.
 *
 * Nothing else can free them: the row says `running` and the process that would
 * have finished it is gone, so without this a crash strands a visitor on a
 * spinner for ever.
 */
async function sweep(): Promise<void> {
  const result = await db.execute(sql`
    UPDATE audit_job
       SET status = 'failed',
           error = 'l''audit ne répond plus',
           finished_at = now()
     WHERE status = 'running'
       AND started_at < now() - make_interval(secs => ${SWEEP_AFTER_S})
    RETURNING id
  `);
  if (result.rows.length > 0) {
    console.log(`worker: swept ${result.rows.length} stalled job(s)`);
  }
}

async function main(): Promise<void> {
  // Before the first claim, so a bad environment is a process that did not
  // start rather than a loop that fails every two seconds for ever.
  assertEnv("worker");
  console.log(`worker: up, ${MAX_IN_FLIGHT} audits at a time, binary ${BIN}`);

  const sweeper = setInterval(() => {
    sweep().catch((error) => console.error("worker: sweep failed", error));
  }, SWEEP_MS);

  const stop = (signal: string) => {
    // Stop claiming, let what is in flight finish and be written. A job killed
    // mid-run would be swept back out five minutes later anyway; finishing is
    // both kinder and cheaper.
    console.log(`worker: ${signal}, draining ${inFlight.size} audit(s)`);
    running = false;
  };
  process.on("SIGTERM", () => stop("SIGTERM"));
  process.on("SIGINT", () => stop("SIGINT"));

  while (running) {
    const room = MAX_IN_FLIGHT - inFlight.size;
    let jobs: ClaimedJob[] = [];
    if (room > 0) {
      try {
        jobs = await claim(room);
      } catch (error) {
        console.error("worker: claim failed", error);
      }
    }

    for (const job of jobs) {
      const task = runJob(job)
        .catch((error) => {
          console.error(`worker: job ${job.id} failed`, error);
          return markFailed(job.id, "erreur interne").catch(() => {});
        })
        .finally(() => {
          inFlight.delete(task);
        });
      inFlight.add(task);
    }

    if (jobs.length === 0) {
      // Nothing to take: wait either for the poll interval or for a slot to
      // free up, whichever comes first.
      await Promise.race([
        new Promise((resolve) => setTimeout(resolve, POLL_MS)),
        ...(inFlight.size > 0 ? [Promise.race(inFlight)] : []),
      ]);
    }
  }

  clearInterval(sweeper);
  await Promise.allSettled(inFlight);
  await pool.end();
  console.log("worker: down");
}

main().catch((error) => {
  console.error("worker: fatal", error);
  process.exit(1);
});