import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import { afterEach, describe, it } from "node:test";
import { createResultDeliveryOwnership, type ResultTakeoverDeps } from "../../src/runs/background/result-delivery-ownership.ts";
import { ensureOwnerRecord, ownerFileStem, ownerRecordPath, readOwnerRecord, type OwnerRecord } from "../../src/shared/owner-record.ts";
import { processLiveness, processStartKey } from "../../src/shared/process-identity.ts";

const roots: string[] = [];
afterEach(() => {
	for (const root of roots.splice(0)) fs.rmSync(root, { recursive: true, force: true });
});
function tempDir(): string {
	const dir = fs.mkdtempSync(path.join(os.tmpdir(), "pi-takeover-test-"));
	roots.push(dir);
	return dir;
}

const HOST = "host-a";
const SESSION = "session-1";

function writeRecord(ownersDir: string, id: string, overrides: Partial<OwnerRecord> = {}): OwnerRecord {
	const record: OwnerRecord = { version: 1, completionOwnerId: id, pid: 4242, hostname: HOST, createdAt: 1_000, ...overrides };
	fs.mkdirSync(ownersDir, { recursive: true });
	fs.writeFileSync(ownerRecordPath(ownersDir, id)!, JSON.stringify(record));
	return record;
}

type Liveness = "alive" | "dead" | "unknown";

/** pid -> liveness table so each fake owner can live or die independently. */
function fakes(ownersDir: string, table: Record<number, Liveness>, overrides: Partial<ResultTakeoverDeps> = {}): ResultTakeoverDeps {
	return {
		ownersDir,
		hostname: () => HOST,
		processLiveness: (pid) => table[pid] ?? "unknown",
		processStartKey: () => undefined,
		...overrides,
	};
}

function ownershipFor(ownerId: string, ownersDir: string, table: Record<number, Liveness>, sessionId: string | null = SESSION, overrides: Partial<ResultTakeoverDeps> = {}) {
	const state = { currentSessionId: sessionId, completionOwnerId: ownerId };
	return { state, ownership: createResultDeliveryOwnership(state, fakes(ownersDir, table, overrides)) };
}

function claimFiles(ownersDir: string): string[] {
	try {
		return fs.readdirSync(path.join(ownersDir, "claims")).sort();
	} catch {
		return [];
	}
}

describe("cross-process result takeover", () => {
	it("takes over a dead same-host owner's result for the current session, exactly once", () => {
		const dir = tempDir();
		writeRecord(dir, "dead-owner", { pid: 100 });
		const { ownership } = ownershipFor("me", dir, { 100: "dead" }, SESSION, { now: () => 5_000 });
		assert.equal(ownership.owns(SESSION, "dead-owner", "run-1"), true);
		assert.equal(ownership.owns(SESSION, "dead-owner", "run-1"), true, "idempotent for the same claimant");
		assert.deepEqual(claimFiles(dir), ["run-1.json"]);
		assert.deepEqual(JSON.parse(fs.readFileSync(path.join(dir, "claims", "run-1.json"), "utf-8")), { claimedBy: "me", previousOwner: "dead-owner", claimedAt: 5_000 });
	});

	it("never takes over from a live owner", () => {
		const dir = tempDir();
		writeRecord(dir, "live-owner", { pid: 100 });
		const { ownership } = ownershipFor("me", dir, { 100: "alive" });
		assert.equal(ownership.owns(SESSION, "live-owner", "run-1"), false);
		assert.equal(ownership.canTakeOver(SESSION, "live-owner", "run-1"), false);
		assert.deepEqual(claimFiles(dir), []);
	});

	it("does not take over without an owner record", () => {
		const dir = tempDir();
		fs.mkdirSync(dir, { recursive: true });
		const { ownership } = ownershipFor("me", dir, { 100: "dead" });
		assert.equal(ownership.owns(SESSION, "no-record", "run-1"), false);
		fs.writeFileSync(ownerRecordPath(dir, "garbage")!, "{not json");
		assert.equal(ownership.owns(SESSION, "garbage", "run-1"), false);
		writeRecord(dir, "wrong-id", { pid: 100 });
		fs.renameSync(ownerRecordPath(dir, "wrong-id")!, ownerRecordPath(dir, "impostor")!);
		assert.equal(ownership.owns(SESSION, "impostor", "run-1"), false, "a record must describe the id it is filed under");
		assert.deepEqual(claimFiles(dir), []);
	});

	it("does not take over when liveness is unknown", () => {
		const dir = tempDir();
		writeRecord(dir, "mystery", { pid: 100 });
		const { ownership } = ownershipFor("me", dir, { 100: "unknown" });
		assert.equal(ownership.owns(SESSION, "mystery", "run-1"), false);
		assert.deepEqual(claimFiles(dir), []);
	});

	it("does not take over from another host", () => {
		const dir = tempDir();
		writeRecord(dir, "elsewhere", { pid: 100, hostname: "host-b" });
		const { ownership } = ownershipFor("me", dir, { 100: "dead" });
		assert.equal(ownership.owns(SESSION, "elsewhere", "run-1"), false);
		assert.deepEqual(claimFiles(dir), []);
	});

	it("treats a reused pid (different start key) as a dead owner, but only with comparable evidence", () => {
		const dir = tempDir();
		const cases: Array<{ name: string; recorded?: string; current?: string; takeable: boolean }> = [
			{ name: "reused", recorded: "ps:Mon Oct  4 10:00:00 2026", current: "ps:Mon Oct  4 12:30:00 2026", takeable: true },
			{ name: "same-process", recorded: "ps:Mon Oct  4 10:00:00 2026", current: "ps:Mon Oct  4 10:00:00 2026", takeable: false },
			{ name: "no-recorded-key", current: "ps:Mon Oct  4 12:30:00 2026", takeable: false },
			{ name: "no-current-key", recorded: "ps:Mon Oct  4 10:00:00 2026", takeable: false },
			{ name: "scheme-changed", recorded: "linux:12345", current: "ps:Mon Oct  4 12:30:00 2026", takeable: false },
		];
		for (const entry of cases) {
			writeRecord(dir, entry.name, { pid: 100, ...(entry.recorded ? { startKey: entry.recorded } : {}) });
			const { ownership } = ownershipFor("me", dir, { 100: "alive" }, SESSION, { processStartKey: () => entry.current });
			assert.equal(ownership.owns(SESSION, entry.name, `run-${entry.name}`), entry.takeable, entry.name);
		}
	});

	it("never takes another session's result, even from a dead owner", () => {
		const dir = tempDir();
		writeRecord(dir, "dead-owner", { pid: 100 });
		const { state, ownership } = ownershipFor("me", dir, { 100: "dead" });
		assert.equal(ownership.owns("other-session", "dead-owner", "run-1"), false);
		assert.equal(ownership.canTakeOver("other-session", "dead-owner", "run-1"), false);
		assert.deepEqual(claimFiles(dir), []);
		// An already-claimed predecessor session is part of this process's scope, exactly as for in-process ownership.
		state.currentSessionId = "old";
		assert.equal(ownership.claimPredecessor("old", "old"), true);
		state.currentSessionId = "new";
		assert.equal(ownership.owns("old", "dead-owner", "run-old"), true);
		assert.equal(ownership.owns("new", "dead-owner", "run-new"), true);
		assert.equal(ownership.owns("other-session", "dead-owner", "run-2"), false);
	});

	it("requires a run key and a non-empty foreign owner id; the in-process rule is unchanged", () => {
		const dir = tempDir();
		writeRecord(dir, "dead-owner", { pid: 100 });
		const { ownership } = ownershipFor("me", dir, { 100: "dead" });
		assert.equal(ownership.owns(SESSION, "dead-owner"), false);
		assert.equal(ownership.owns(SESSION, undefined, "run-1"), false);
		assert.equal(ownership.owns(SESSION, "", "run-1"), false);
		assert.equal(ownership.owns(SESSION, "me", "run-1"), true);
		assert.equal(ownership.owns("other-session", "me", "run-1"), false);
		assert.deepEqual(claimFiles(dir), []);
	});

	it("lets exactly one of two racing processes win a result", () => {
		for (const order of [["a", "b"], ["b", "a"]] as const) {
			const dir = tempDir();
			writeRecord(dir, "dead-owner", { pid: 100 });
			const table: Record<number, Liveness> = { 100: "dead", 201: "alive", 202: "alive" };
			writeRecord(dir, "proc-a", { pid: 201 });
			writeRecord(dir, "proc-b", { pid: 202 });
			const a = ownershipFor("proc-a", dir, table);
			const b = ownershipFor("proc-b", dir, table);
			const by = { a: a.ownership, b: b.ownership };
			const results = order.map((name) => by[name].owns(SESSION, "dead-owner", "run-1"));
			assert.deepEqual(results, [true, false], `first caller (${order[0]}) wins`);
			assert.equal(by[order[0]].owns(SESSION, "dead-owner", "run-1"), true);
			assert.equal(by[order[1]].owns(SESSION, "dead-owner", "run-1"), false);
			assert.equal(by[order[1]].canTakeOver(SESSION, "dead-owner", "run-1"), false);
			assert.deepEqual(claimFiles(dir), ["run-1.json"]);
		}
	});

	it("re-claims a result whose claimer died, race-safely and with a bounded chain", () => {
		const dir = tempDir();
		writeRecord(dir, "dead-owner", { pid: 100 });
		writeRecord(dir, "claimer-1", { pid: 201 });
		writeRecord(dir, "claimer-2", { pid: 202 });
		writeRecord(dir, "claimer-3", { pid: 203 });
		const table: Record<number, Liveness> = { 100: "dead", 201: "alive", 202: "alive", 203: "alive", 301: "alive", 302: "alive" };
		writeRecord(dir, "taker-a", { pid: 301 });
		writeRecord(dir, "taker-b", { pid: 302 });

		const first = ownershipFor("claimer-1", dir, table);
		assert.equal(first.ownership.owns(SESSION, "dead-owner", "run-1"), true);
		// While the first claimer lives nobody else may take the result.
		assert.equal(ownershipFor("taker-a", dir, table).ownership.owns(SESSION, "dead-owner", "run-1"), false);

		table[201] = "dead";
		const a = ownershipFor("taker-a", dir, table, SESSION, { livenessTtlMs: 0 });
		const b = ownershipFor("taker-b", dir, table, SESSION, { livenessTtlMs: 0 });
		assert.equal(a.ownership.canTakeOver(SESSION, "dead-owner", "run-1"), true, "peek never claims");
		assert.deepEqual(claimFiles(dir), ["run-1.json"]);
		assert.equal(a.ownership.owns(SESSION, "dead-owner", "run-1"), true);
		assert.equal(b.ownership.owns(SESSION, "dead-owner", "run-1"), false, "the second re-claimer loses");
		assert.deepEqual(claimFiles(dir), ["run-1%g2.json", "run-1.json"].sort());
		assert.equal(JSON.parse(fs.readFileSync(path.join(dir, "claims", "run-1%g2.json"), "utf-8")).claimedBy, "taker-a");

		// A claimer whose record is gone cannot be proven dead.
		fs.rmSync(ownerRecordPath(dir, "taker-a")!);
		assert.equal(ownershipFor("claimer-2", dir, table, SESSION, { livenessTtlMs: 0 }).ownership.owns(SESSION, "dead-owner", "run-1"), false);
	});

	it("gives up once the claim chain is exhausted", () => {
		const dir = tempDir();
		writeRecord(dir, "dead-owner", { pid: 100 });
		const table: Record<number, Liveness> = { 100: "dead" };
		const claimers = ["c1", "c2", "c3", "c4"];
		claimers.forEach((id, index) => {
			writeRecord(dir, id, { pid: 200 + index });
			table[200 + index] = "alive";
		});
		for (const id of claimers) {
			assert.equal(ownershipFor(id, dir, table, SESSION, { livenessTtlMs: 0 }).ownership.owns(SESSION, "dead-owner", "run-1"), true, id);
			table[200 + claimers.indexOf(id)] = "dead";
		}
		writeRecord(dir, "late", { pid: 300 });
		table[300] = "alive";
		assert.equal(ownershipFor("late", dir, table, SESSION, { livenessTtlMs: 0 }).ownership.owns(SESSION, "dead-owner", "run-1"), false);
	});

	it("fails safe on an unreadable claim marker", () => {
		const dir = tempDir();
		writeRecord(dir, "dead-owner", { pid: 100 });
		fs.mkdirSync(path.join(dir, "claims"), { recursive: true });
		fs.writeFileSync(path.join(dir, "claims", "run-1.json"), "{half");
		const { ownership } = ownershipFor("me", dir, { 100: "dead" });
		assert.equal(ownership.owns(SESSION, "dead-owner", "run-1"), false);
		assert.equal(ownership.canTakeOver(SESSION, "dead-owner", "run-1"), false);
	});

	it("keeps run keys from escaping the claims directory", () => {
		const dir = tempDir();
		writeRecord(dir, "dead-owner", { pid: 100 });
		const { ownership } = ownershipFor("me", dir, { 100: "dead" });
		assert.equal(ownership.owns(SESSION, "dead-owner", "../../escape"), true);
		assert.deepEqual(claimFiles(dir), [`${encodeURIComponent("../../escape")}.json`]);
		assert.equal(fs.existsSync(path.join(dir, "..", "escape.json")), false);
		assert.equal(ownerFileStem("../x/y"), "..%2Fx%2Fy");
	});

	it("probes each owner at most once per TTL window", () => {
		const dir = tempDir();
		writeRecord(dir, "live-owner", { pid: 100 });
		let probes = 0;
		let clock = 10_000;
		const { ownership } = ownershipFor("me", dir, {}, SESSION, {
			now: () => clock,
			processLiveness: () => { probes += 1; return "alive"; },
		});
		for (let index = 0; index < 50; index += 1) ownership.owns(SESSION, "live-owner", `run-${index}`);
		assert.equal(probes, 1);
		clock += 1_999;
		ownership.owns(SESSION, "live-owner", "run-x");
		assert.equal(probes, 1);
		clock += 2;
		ownership.owns(SESSION, "live-owner", "run-y");
		assert.equal(probes, 2);
	});

	it("garbage-collects only provably dead, week-old owner records and stale claims", () => {
		const dir = tempDir();
		const week = 8 * 24 * 60 * 60 * 1000;
		const old = new Date(Date.now() - week);
		const table: Record<number, Liveness> = { 1: "dead", 2: "alive", 3: "dead", 4: "unknown" };
		writeRecord(dir, "old-dead", { pid: 1 });
		writeRecord(dir, "old-alive", { pid: 2 });
		writeRecord(dir, "young-dead", { pid: 3 });
		writeRecord(dir, "old-unknown", { pid: 4 });
		writeRecord(dir, "me", { pid: 5 });
		table[5] = "dead";
		for (const id of ["old-dead", "old-alive", "old-unknown", "me"]) fs.utimesSync(ownerRecordPath(dir, id)!, old, old);
		fs.mkdirSync(path.join(dir, "claims"));
		fs.writeFileSync(path.join(dir, "claims", "gone.json"), JSON.stringify({ claimedBy: "young-dead", previousOwner: "x", claimedAt: 1 }));
		fs.utimesSync(path.join(dir, "claims", "gone.json"), old, old);
		fs.writeFileSync(path.join(dir, "claims", "kept.json"), JSON.stringify({ claimedBy: "old-alive", previousOwner: "x", claimedAt: 1 }));
		fs.utimesSync(path.join(dir, "claims", "kept.json"), old, old);
		fs.writeFileSync(path.join(dir, "claims", "stale.json.1.abc.tmp"), "x");
		fs.utimesSync(path.join(dir, "claims", "stale.json.1.abc.tmp"), old, old);
		writeRecord(dir, "dead-owner", { pid: 9 });
		table[9] = "dead";
		const { ownership } = ownershipFor("me", dir, table);
		ownership.owns(SESSION, "dead-owner", "run-1");
		const left = fs.readdirSync(dir).sort();
		assert.deepEqual(left, ["claims", "dead-owner.json", "me.json", "old-alive.json", "old-unknown.json", "young-dead.json"].sort());
		assert.deepEqual(claimFiles(dir), ["kept.json", "run-1.json"]);
	});

	it("takes over from a real exited process, identified by its real pid", () => {
		const dir = tempDir();
		const child = spawnSync(process.execPath, ["-e", "process.stdout.write(String(process.pid))"], { encoding: "utf-8" });
		const pid = Number(child.stdout);
		assert.ok(Number.isSafeInteger(pid) && pid > 0);
		writeRecord(dir, "exited", { pid, hostname: os.hostname() });
		writeRecord(dir, "running", { pid: process.pid, hostname: os.hostname() });
		const state = { currentSessionId: SESSION as string | null, completionOwnerId: "me" };
		const ownership = createResultDeliveryOwnership(state, { ownersDir: dir });
		assert.equal(ownership.owns(SESSION, "exited", "run-1"), true);
		assert.equal(ownership.owns(SESSION, "running", "run-2"), false, "this very process is alive");
	});
});

describe("owner record", () => {
	it("is written once per process, shared across reloads, and survives shutdown", () => {
		const dir = tempDir();
		const ids = { completionOwnerId: "owner-xyz", pid: 777, hostname: HOST, startKey: () => "ps:start" };
		assert.equal(ensureOwnerRecord({ ownersDir: dir, ...ids, now: () => 111 }), true);
		const file = ownerRecordPath(dir, "owner-xyz")!;
		const first = fs.readFileSync(file, "utf-8");
		assert.deepEqual(JSON.parse(first), { version: 1, completionOwnerId: "owner-xyz", pid: 777, startKey: "ps:start", hostname: HOST, createdAt: 111 });
		// Simulated extension reload: same owner id, later clock.
		let startKeyCalls = 0;
		assert.equal(ensureOwnerRecord({ ownersDir: dir, ...ids, startKey: () => { startKeyCalls += 1; return "x"; }, now: () => 999 }), false);
		assert.equal(fs.readFileSync(file, "utf-8"), first);
		assert.equal(startKeyCalls, 0, "a reload must not spawn ps again");
		assert.deepEqual(fs.readdirSync(dir), ["owner-xyz.json"], "no temp files left behind");
		if (process.platform !== "win32") assert.equal(fs.statSync(file).mode & 0o077, 0, "record is private to the user");
		// Not removed by anything this process does on shutdown: detached runners outlive it and need the record.
		assert.deepEqual(readOwnerRecord(dir, "owner-xyz")?.pid, 777);
	});

	it("replaces a record left by a different process that reused the owner id slot", () => {
		const dir = tempDir();
		ensureOwnerRecord({ ownersDir: dir, completionOwnerId: "o", pid: 1, hostname: HOST, startKey: () => undefined, now: () => 1 });
		assert.equal(ensureOwnerRecord({ ownersDir: dir, completionOwnerId: "o", pid: 2, hostname: HOST, startKey: () => undefined, now: () => 2 }), true);
		assert.equal(readOwnerRecord(dir, "o")?.pid, 2);
		assert.equal("startKey" in (readOwnerRecord(dir, "o") ?? {}), false);
	});

	it("resolves this process's real liveness and start key", () => {
		assert.equal(processLiveness(process.pid), "alive");
		assert.equal(processLiveness(0), "unknown");
		assert.equal(processLiveness(-5), "unknown");
		const key = processStartKey(process.pid);
		assert.equal(processStartKey(process.pid), key, "own key is cached");
	});
});