import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
import { GenericContainer, Wait, type StartedTestContainer } from "testcontainers";
import { Client } from "pg";
import { build } from "esbuild";
import { createRequire } from "node:module";
import { fileURLToPath } from "node:url";
import { execFileSync, spawn, type ChildProcess } from "node:child_process";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import * as http from "node:http";
import type { AddressInfo, Socket } from "node:net";

// A real process is killed with a real signal while a request to the endpoint is in flight, then the next processes must
// recover with no human step. What this does NOT cover: a Kubernetes pod (no Lease, no kubelet, no 60 s grace period).
const ROWS = Number(process.env.WEBHOOK_KILL_ROWS ?? 3000);
const hangAt = Math.floor(ROWS / 2);

let container: StartedTestContainer;
let admin: Client;
let childFile: string;
let objectDir: string;
let work: string;

beforeAll(async () => {
  work = await mkdtemp(join(tmpdir(), "retl-kill-"));
  objectDir = join(work, "objects");
  childFile = join(work, "child.cjs");
  await build({
    entryPoints: [fileURLToPath(new URL("./webhook-kill-child.ts", import.meta.url))],
    bundle: true,
    platform: "node",
    format: "cjs",
    target: "node22",
    external: ["pg-native"],
    outfile: childFile,
    logLevel: "silent",
  });
  container = await new GenericContainer("postgres:18-alpine")
    .withEnvironment({ POSTGRES_PASSWORD: "test", POSTGRES_DB: "runner_test" })
    .withExposedPorts(5432)
    .withWaitStrategy(Wait.forLogMessage(/database system is ready to accept connections/, 2))
    .start();
  const config = {
    host: container.getHost(),
    port: container.getMappedPort(5432),
    database: "runner_test",
    user: "postgres",
    password: "test",
  };
  admin = new Client(config);
  await admin.connect();
  execFileSync(
    process.execPath,
    [
      createRequire(import.meta.url).resolve("prisma/build/index.js"),
      "db",
      "push",
      `--schema=${fileURLToPath(new URL("../../../webapps/console/prisma/schema.prisma", import.meta.url))}`,
      "--skip-generate",
    ],
    {
      // eslint-disable-next-line no-restricted-properties -- disposable test database only.
      env: {
        ...process.env,
        DATABASE_URL: `postgresql://postgres:test@${config.host}:${config.port}/${config.database}?schema=newjitsu`,
      },
      stdio: "ignore",
    }
  );
  await admin.query(
    "CREATE ROLE runner_runtime LOGIN PASSWORD 'runtime'; GRANT USAGE ON SCHEMA newjitsu TO runner_runtime"
  );
  for (const table of ["control", "target_owner"])
    await admin.query(`GRANT SELECT,INSERT,UPDATE,DELETE ON newjitsu.reverse_sync_${table} TO runner_runtime`);
  await admin.query(
    "GRANT SELECT,INSERT,UPDATE ON newjitsu.source_state,newjitsu.source_task TO runner_runtime; GRANT INSERT ON newjitsu.task_log TO runner_runtime"
  );
}, 120_000);
afterAll(async () => {
  await admin?.end();
  await container?.stop();
  if (work) await rm(work, { recursive: true, force: true });
});

// The endpoint: records every delivered record, and holds ONE request open forever after processing it (the answer never
// arrives), which is the moment the process is killed.
const receiver = {
  server: undefined as unknown as http.Server,
  sockets: new Set<Socket>(),
  ids: new Map<number, number>(),
  keys: new Map<string, number>(),
  hung: false,
  onHung: () => {},
};
const startReceiver = async () => {
  receiver.ids.clear();
  receiver.keys.clear();
  receiver.hung = false;
  const hungPromise = new Promise<void>(resolve => (receiver.onHung = resolve));
  receiver.server = http.createServer((req, res) => {
    const chunks: Buffer[] = [];
    req.on("data", chunk => chunks.push(chunk));
    req.on("end", () => {
      const records = JSON.parse(Buffer.concat(chunks).toString()).records as {
        idempotencyKey: string;
        data: { id: number };
      }[];
      for (const record of records) {
        receiver.ids.set(record.data.id, (receiver.ids.get(record.data.id) ?? 0) + 1);
        receiver.keys.set(record.idempotencyKey, (receiver.keys.get(record.idempotencyKey) ?? 0) + 1);
      }
      if (!receiver.hung && records.some(record => record.data.id === hangAt)) {
        receiver.hung = true;
        receiver.onHung();
        return; // never answer
      }
      res.statusCode = 200;
      res.end();
    });
  });
  receiver.server.on("connection", socket => {
    receiver.sockets.add(socket);
    socket.on("close", () => receiver.sockets.delete(socket));
  });
  await new Promise<void>(resolve => receiver.server.listen(0, "127.0.0.1", resolve));
  return { hung: hungPromise };
};
afterEach(async () => {
  for (const socket of receiver.sockets) socket.destroy();
  await new Promise<void>(resolve => receiver.server?.close(() => resolve()));
});
beforeEach(async () => {
  await admin.query(
    "TRUNCATE newjitsu.reverse_sync_control,newjitsu.reverse_sync_target_owner,newjitsu.source_state,newjitsu.source_task,newjitsu.task_log"
  );
  await rm(objectDir, { recursive: true, force: true });
});

/** Reject if the endpoint is not hit in time, with what the child printed so far, instead of hanging until the test timeout. */
const stall = (child: { output: () => string }, ms = 90_000) =>
  new Promise<never>((_, reject) =>
    setTimeout(
      () =>
        reject(
          new Error(
            `no request reached the hang point after ${ms / 1000}s; endpoint saw ${
              receiver.ids.size
            } rows; child output: ${child.output().slice(0, 1500)}`
          )
        ),
      ms
    ).unref()
  );

function startChild(taskId: string): {
  child: ChildProcess;
  output: () => string;
  done: Promise<{ code: number | null; signal: string | null; out: string }>;
} {
  const child = spawn(process.execPath, ["--no-warnings", childFile], {
    env: {
      // eslint-disable-next-line no-restricted-properties -- test child needs PATH and node options only.
      ...process.env,
      KILL_ROWS: String(ROWS),
      KILL_RECEIVER_URL: `http://127.0.0.1:${(receiver.server.address() as AddressInfo).port}/hook`,
      KILL_DB_HOST: container.getHost(),
      KILL_DB_PORT: String(container.getMappedPort(5432)),
      KILL_DB_NAME: "runner_test",
      KILL_OBJECT_DIR: objectDir,
      KILL_TASK_ID: taskId,
    },
    stdio: ["ignore", "pipe", "pipe"],
  });
  let out = "";
  child.stdout!.on("data", chunk => (out += chunk));
  child.stderr!.on("data", chunk => (out += chunk));
  const done = new Promise<{ code: number | null; signal: string | null; out: string }>(resolve =>
    child.on("exit", (code, signal) => resolve({ code, signal, out }))
  );
  return { child, done, output: () => out };
}
const taskError = async (taskId: string) =>
  String((await admin.query("SELECT error FROM newjitsu.source_task WHERE task_id=$1", [taskId])).rows[0]?.error ?? "");
const missingIds = () => {
  const missing: number[] = [];
  for (let id = 1; id <= ROWS && missing.length < 5; id++) if (!receiver.ids.has(id)) missing.push(id);
  return missing;
};

/** After an interruption: plain scheduled runs, as production would start them, until one completes. */
async function runUntilComplete(prefix: string) {
  const runs: { result: string; error: string }[] = [];
  for (let i = 1; i <= 4; i++) {
    const taskId = `${prefix}-after-${i}`;
    const { done } = startChild(taskId);
    const { code, out } = await done;
    const result = /RESULT (\w+)/.exec(out)?.[1] ?? `NO_RESULT(exit ${code}): ${out.slice(0, 200)}`;
    runs.push({ result, error: result === "FAILED" ? await taskError(taskId) : "" });
    if (result === "COMPLETE") break;
  }
  return runs;
}

describe("a process killed while a request to the endpoint is in flight", () => {
  it("SIGKILL: the next run fails once with a readable message and the one after delivers every row", async () => {
    const { hung } = await startReceiver();
    const { child, done, output } = startChild("kill-1");
    await Promise.race([
      hung,
      done.then(r => Promise.reject(new Error(`child exited before the endpoint was hit: exit ${r.code} ${r.out}`))),
    ]); // the endpoint has processed the batch and will never answer
    child.kill("SIGKILL");
    const killed = await done;
    expect(killed.signal).toBe("SIGKILL");
    expect(receiver.ids.has(hangAt)).toBe(true);
    expect(receiver.ids.size).toBeLessThan(ROWS);

    const runs = await runUntilComplete("kill");
    process.stdout.write(`KILL-RESULT ${JSON.stringify(runs)}\n`);
    expect(runs.at(-1)!.result).toBe("COMPLETE");
    expect(runs.filter(run => run.result === "FAILED").length).toBeLessThanOrEqual(1);
    for (const run of runs.filter(run => run.result === "FAILED"))
      expect(run.error).toContain("Recovery completed cleanup; next run will restart extraction");
    expect(missingIds()).toEqual([]);
    // The repeats carry the same idempotency keys: exactly one key per row, however many times it was sent.
    expect(receiver.keys.size).toBe(ROWS);
  }, 300_000);

  it("SIGTERM: the process stops itself promptly, and the same recovery follows", async () => {
    const { hung } = await startReceiver();
    const { child, done, output } = startChild("term-1");
    await Promise.race([
      hung,
      done.then(r => Promise.reject(new Error(`child exited before the endpoint was hit: exit ${r.code} ${r.out}`))),
    ]);
    const started = Date.now();
    child.kill("SIGTERM");
    const stopped = await done;
    const seconds = (Date.now() - started) / 1000;
    process.stdout.write(
      `TERM-STOP exit=${stopped.code} signal=${stopped.signal} after=${seconds}s out=${stopped.out.trim()}\n`
    );
    expect(stopped.signal).toBeNull(); // it exited by itself, was not killed by the signal
    expect(seconds).toBeLessThan(45); // within the watchdog
    expect(stopped.code).not.toBe(0);

    const runs = await runUntilComplete("term");
    process.stdout.write(`TERM-RESULT ${JSON.stringify(runs)}\n`);
    expect(runs.at(-1)!.result).toBe("COMPLETE");
    expect(runs.filter(run => run.result === "FAILED").length).toBeLessThanOrEqual(1);
    expect(missingIds()).toEqual([]);
    expect(receiver.keys.size).toBe(ROWS);
  }, 300_000);
});