import { test } from "node:test";
import assert from "node:assert/strict";
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, utimesSync, writeFileSync } from "node:fs";
import os from "node:os";
import path from "node:path";
import type { Paths } from "../lib/paths";
import {
  LockWaitCancelled,
  acquirePublishLock,
  START_SLACK_MS,
  holderIsGone,
  lockHostId,
  pidStartOf,
  processStartedAtMs,
  publishLockPath,
  readLockHolder,
  withPublishLock,
  type LockHolder,
} from "./stageLock";

// Run with:
//   pnpm --filter yt-dlp-transcript-common test
//
// The publish lock (release 18): a stale holder on this host is taken over, a
// live one is waited for (cancellably), and a holder on another host is never
// stolen. The process probes are injected; nothing here signals a real pid.

function tmp(): { root: string; paths: Paths } {
  const root = mkdtempSync(path.join(os.tmpdir(), "stage-lock-"));
  return { root, paths: { exportBuildsDir: path.join(root, ".export-builds") } as Paths };
}

function plant(paths: Paths, holder: Partial<LockHolder>): void {
  mkdirSync(paths.exportBuildsDir, { recursive: true });
  writeFileSync(
    publishLockPath(paths),
    JSON.stringify({ pid: 4242, host: "here", kind: "build-site", target: "jer", since: 1, ...holder }),
  );
}

const here = { host: "here", pid: 100, startOf: () => null, startedAtMs: () => null };

test("holderIsGone: same host and a dead pid, or a pid reused by a later process", () => {
  const h: LockHolder = { pid: 7, host: "here", kind: "k", target: "t", since: 1, pidStart: "500" };
  assert.equal(holderIsGone(h, { host: "here", isAlive: () => false }), true);
  assert.equal(holderIsGone(h, { host: "here", isAlive: () => true, startOf: () => "500", startedAtMs: () => null }), false);
  assert.equal(holderIsGone(h, { host: "here", isAlive: () => true, startOf: () => "900", startedAtMs: () => null }), true, "pid reused");
  assert.equal(holderIsGone(h, { host: "here", isAlive: () => true, startOf: () => null, startedAtMs: () => null }), false);
  assert.equal(holderIsGone(h, { host: "there", isAlive: () => false }), false, "another host: never");
  assert.equal(holderIsGone({ ...h, pidStart: null }, { host: "here", isAlive: () => true, startOf: () => "1", startedAtMs: () => null }), false);
});

test("holderIsGone: a live pid whose process started AFTER the lock was taken is not the holder (a recreated container's pid 1)", () => {
  const since = 1_000_000;
  const h: LockHolder = { pid: 1, host: "archilyzer-editor", kind: "k", target: "t", since };
  const probe = (startedAt: number | null) =>
    holderIsGone(h, { host: "archilyzer-editor", isAlive: () => true, startOf: () => null, startedAtMs: () => startedAt });
  assert.equal(probe(since + 60_000), true, "started a minute after the lock: stale");
  assert.equal(probe(since - 60_000), false, "started before the lock: may be the holder");
  assert.equal(probe(since + 1_000), false, "within btime's slack: not judged");
  assert.equal(probe(null), false, "no /proc: pid-alive alone decides");
});

test("the host identity is ARCHILYZER_HOST_ID when set, else the hostname", () => {
  assert.equal(lockHostId({ ARCHILYZER_HOST_ID: " archilyzer-editor " }), "archilyzer-editor");
  assert.equal(lockHostId({ ARCHILYZER_HOST_ID: "" }), os.hostname());
  assert.equal(lockHostId({}), os.hostname());
});

test("processStartedAtMs: this process started before now, and no such pid is null", () => {
  if (process.platform === "linux") {
    const t = processStartedAtMs(process.pid);
    assert.ok(t !== null && t <= Date.now() + START_SLACK_MS && t > Date.now() - 86_400_000 * 365, String(t));
  }
  assert.equal(processStartedAtMs(2 ** 30), null);
});

test("pidStartOf reads this process's start time on Linux and is null for no such pid", () => {
  if (process.platform === "linux") {
    assert.match(pidStartOf(process.pid) ?? "", /^\d+$/);
    assert.equal(pidStartOf(process.pid), pidStartOf(process.pid));
  }
  assert.equal(pidStartOf(2 ** 30), null);
});

test("the lock is taken, names its holder, and is released", async () => {
  const { root, paths } = tmp();
  try {
    const lock = await acquirePublishLock(paths, { kind: "update-index", target: "_index" }, here);
    assert.deepEqual(await readLockHolder(paths), lock.holder);
    assert.equal(lock.holder.pid, 100);
    assert.equal(lock.holder.kind, "update-index");
    await lock.release();
    assert.equal(existsSync(publishLockPath(paths)), false);
    // withPublishLock releases on a throw too.
    await assert.rejects(
      withPublishLock(paths, { kind: "k", target: "t" }, async () => {
        throw new Error("boom");
      }, here),
      /boom/,
    );
    assert.equal(existsSync(publishLockPath(paths)), false);
  } finally {
    rmSync(root, { recursive: true, force: true });
  }
});

test("a stale holder on this host is taken over, and said so", async () => {
  const { root, paths } = tmp();
  try {
    plant(paths, { pid: 4242, host: "here" });
    const logs: string[] = [];
    const lock = await acquirePublishLock(paths, { kind: "build-site", target: "ani" }, {
      ...here,
      isAlive: (pid) => pid !== 4242,
      onLog: (l) => logs.push(l),
    });
    assert.equal(lock.holder.target, "ani");
    assert.match(logs.join(""), /taking over a stale publish lock: build-site jer \(pid 4242 on here/);
    await lock.release();
  } finally {
    rmSync(root, { recursive: true, force: true });
  }
});

test("a live holder is waited for, with one log line, until it releases", async () => {
  const { root, paths } = tmp();
  try {
    plant(paths, { pid: 4242, host: "here" });
    const logs: string[] = [];
    let polls = 0;
    const taking = acquirePublishLock(paths, { kind: "build-site", target: "ani" }, {
      ...here,
      isAlive: () => true,
      pollMs: 5,
      onLog: (l) => {
        logs.push(l);
      },
      now: () => {
        polls++;
        // The holder releases after a few polls.
        if (polls === 6) rmSync(publishLockPath(paths));
        return polls;
      },
    });
    const lock = await taking;
    assert.equal(lock.holder.target, "ani");
    assert.equal(logs.length, 1, logs.join(""));
    assert.match(logs[0], /waiting for the publish lock — held by build-site jer \(pid 4242 on here/);
    await lock.release();
  } finally {
    rmSync(root, { recursive: true, force: true });
  }
});

test("a holder on another host is never stolen; the wait is cancellable", async () => {
  const { root, paths } = tmp();
  try {
    plant(paths, { pid: 4242, host: "elsewhere" });
    const ac = new AbortController();
    const logs: string[] = [];
    const taking = acquirePublishLock(paths, { kind: "k", target: "t" }, {
      ...here,
      isAlive: () => false, // dead HERE means nothing for a pid over there
      pollMs: 5,
      signal: ac.signal,
      onLog: (l) => logs.push(l),
    });
    setTimeout(() => ac.abort(), 40);
    await assert.rejects(taking, LockWaitCancelled);
    assert.equal(logs.length, 1);
    assert.match(logs[0], /on ANOTHER host \("elsewhere"; this one is "here"\)/);
    assert.match(logs[0], /If no publish stage is running there, remove .*\.publish\.lock/);
    assert.equal(JSON.parse(readFileSync(publishLockPath(paths), "utf8")).host, "elsewhere", "untouched");
  } finally {
    rmSync(root, { recursive: true, force: true });
  }
});

test("a torn lock file is waited for, then taken over after the grace", async () => {
  const { root, paths } = tmp();
  try {
    mkdirSync(paths.exportBuildsDir, { recursive: true });
    writeFileSync(publishLockPath(paths), "");
    const old = new Date(Date.now() - 120_000);
    utimesSync(publishLockPath(paths), old, old);
    const lock = await acquirePublishLock(paths, { kind: "k", target: "t" }, { ...here, pollMs: 5 });
    assert.equal(lock.holder.kind, "k");
    await lock.release();
  } finally {
    rmSync(root, { recursive: true, force: true });
  }
});

test("release never removes a lock somebody else holds now", async () => {
  const { root, paths } = tmp();
  try {
    const lock = await acquirePublishLock(paths, { kind: "k", target: "t" }, here);
    plant(paths, { pid: 9, host: "here", since: 99 });
    await lock.release();
    assert.equal(JSON.parse(readFileSync(publishLockPath(paths), "utf8")).pid, 9);
  } finally {
    rmSync(root, { recursive: true, force: true });
  }
});