/**
 * # session/live-lock — the per-host live singleton lock (kestrel-7o2, Wave A / A6)
 *
 * Two live pumps against ONE broker account = doubled clips. Every other safety layer in this repo
 * is per-process — the kill-switch latches in one process, the L0 clamp bounds one process's orders,
 * WALL 5 ceilings one process's preflight — so NONE of them can see a second process arming the same
 * account. This module is the one guard whose subject is the HOST, not the process: an advisory,
 * owner-only lock file under the existing `~/.kestrel` home, keyed on (venue, account).
 *
 * ## Where the floor ends and the lease begins (kestrel-7o2.2)
 * This is the SPIKE FLOOR, deliberately less than the control-plane lease, and it must not be
 * mistaken for it. What is HERE:
 *   - one host, one filesystem, one supervised operator;
 *   - mutual exclusion keyed on (venue, account);
 *   - fail-closed acquisition and a pid-death-proven recovery.
 * What is NOT here, and stays in **kestrel-7o2.2** (control-plane lease, ADR-0007 / ADR-0034 §7):
 *   - **cross-host** exclusion — two machines cannot see each other's lock files, so this module
 *     REFUSES rather than reasons about a foreign host ({@link LiveLockFailure} `foreign-host`);
 *   - **pod-lineage** scope — the lease's subject is the pod lineage, not (venue, account);
 *   - **TTL / auto-expiry at market close**, and the arm-survives-restart question — both need a
 *     trusted clock this module deliberately does not have (see "Time" below);
 *   - **fencing tokens** — the lease hands the broker a monotonic token so a resumed zombie's writes
 *     are rejected AT THE VENUE. That is the only thing that truly closes the recovery race below;
 *     a file lock cannot, because the broker does not participate in it.
 * A file on a local disk is not a distributed lock. Do not let this module's existence retire 7o2.2.
 *
 * ## Fail closed
 * Every path that cannot PROVE the account is free refuses, with a typed {@link LiveLockFailure} and
 * a logged, secret-free reason. There is no "proceed unlocked" branch and no force/override flag: an
 * unreadable lock, a foreign host, a live holder, and an I/O error all refuse identically. The cost
 * is that a corrupt lock file needs a human `rm` — the refusal names the path so the operator can.
 * That is the intended trade: an outage is recoverable, a double clip is not.
 *
 * ## Recovery: pid-death, never age
 * A dead holder's lock is recoverable. The recovery predicate is exactly one thing — **the holder's
 * pid is provably gone on THIS host** — and it is deliberately NOT a TTL. A TTL steals from a slow
 * holder (a GC pause, an SIGSTOP, a long reconcile) and stealing from a LIVE holder is the precise
 * failure this module exists to prevent, so age never grants a takeover. The direction of the error
 * matters: a recycled pid reads as ALIVE and we refuse (annoying, safe), never as dead (silent,
 * catastrophic). Recovery is OBSERVABLE two ways: it announces itself through {@link LiveLockDeps.notify}
 * and the new record carries {@link LiveLockRecord.recovered_from}, so a takeover is durable evidence
 * rather than a line that scrolled past. A silent takeover would be indistinguishable from a bug.
 *
 * **The residual race, stated honestly:** between the unlink of a dead holder's file and our
 * exclusive re-create, a second recoverer can interleave. `O_EXCL` decides the winner (not
 * last-writer-wins) and {@link LiveLockHandle.held} re-reads the disk, so the loser learns it lost —
 * but only when it next asks. A driver MUST therefore call `held()` before each submit rather than
 * trusting the handle it acquired at open. Closing this window for real needs the venue-side fencing
 * token in 7o2.2.
 *
 * ## Time
 * No wall clock is read here. `now` is injected ({@link LiveLockDeps.now}, the repo's `NowFn` seam —
 * same shape as `IbkrTransportDeps.now` / `PaperHooks.now`) and is used ONLY to STAMP the record for
 * observability; no decision branches on it. That is what lets the recovery rule stay pid-death:
 * with no time in the predicate there is no clock to skew, and nothing on the deterministic Session
 * path is load-bearing on this control-plane record (OSS-ADR-0010).
 *
 * The lock is ADVISORY: it binds processes that ask. It is the floor for a supervised, single-operator
 * spike — not a mechanism that can stop a determined or unaware writer.
 */

import { homedir, hostname } from "node:os";
import { join } from "node:path";
import { createHash } from "node:crypto";
import { mkdirSync, openSync, writeSync, closeSync, readFileSync, unlinkSync, chmodSync } from "node:fs";

import { redactAccount } from "../adapters/broker/ibkr/config.ts";

/** A monotone-enough wall clock in epoch ms — injected, never ambient (mirrors `IbkrTransportDeps.now`). */
export type NowFn = () => number;

/** The singleton's SUBJECT: one live pump per (venue, account) per host. Two different accounts are
 * two different singletons; the same account at two venues likewise (a venue is a distinct book). */
export interface LiveLockScope {
  /** The venue key — the registry's `IBKR_PAPER_VENUE`-style id (e.g. `"ibkr"`). */
  readonly venue: string;
  /** The broker account id. NEVER written to disk in the clear — hashed into the path, redacted in
   * the record (a lock file that names a live account is a credential-adjacent leak, cf. 7o2.23). */
  readonly account: string;
}

/** Why an acquisition refused. Closed vocabulary — an unknown state is not representable. */
export type LiveLockFailure =
  /** A live process on this host holds the lock. The ordinary, correct refusal. */
  | "held"
  /** The lock is held by a record from ANOTHER host: we cannot prove its liveness, so we refuse. */
  | "foreign-host"
  /** The lock file exists but is not a lock file we understand (corrupt / wrong schema / newer
   * version). We refuse rather than interpret — and never delete what we cannot read. */
  | "unreadable"
  /** The filesystem refused (perms, missing home, disk). Fail closed: no lock, no trading. */
  | "io";

/** A refusal to grant the live singleton. Carries the typed {@link LiveLockFailure} and a secret-free
 * reason (the account is redacted) — mirrors `PaperSessionRefused`'s shape so a driver can log both
 * the same way. */
export class LiveLockRefused extends Error {
  override readonly name = "LiveLockRefused";
  readonly failure: LiveLockFailure;
  readonly reason: string;

  constructor(failure: LiveLockFailure, reason: string, options: { cause?: unknown } = {}) {
    super(`live singleton lock REFUSED (${failure}): ${reason}`, options);
    this.failure = failure;
    this.reason = reason;
  }
}

/** The on-disk lock record (v1) — a control-plane authority record (OSS-ADR-0010), not a projection:
 * it derives from no text truth and nothing on the graded fold reads it. */
export interface LiveLockRecord {
  readonly version: 1;
  /** The venue key (plaintext — not a secret). */
  readonly venue: string;
  /** The REDACTED account (`U***67`) — enough to identify the lock in a log, never enough to leak it. */
  readonly account: string;
  /** The host that minted this record. A record from another host is never recovered. */
  readonly host: string;
  /** The holder's pid ON `host`. The liveness subject — meaningless off that host. */
  readonly pid: number;
  /** `sha256(host|pid|acquired_at)` — the holder's identity, so `held()` distinguishes US from a
   * later holder that happens to share our pid. Derived, NOT random: no unseeded RNG (RUNTIME §0). */
  readonly token: string;
  /** The injected `now` at acquisition (epoch ms). OBSERVABILITY ONLY — nothing branches on it. */
  readonly acquired_at: number;
  /** Present iff this lock was RECOVERED from a dead holder — the durable evidence of a takeover. */
  readonly recovered_from?: {
    readonly pid: number;
    readonly acquired_at: number;
  };
}

/** The injected environment. Every ambient read (clock, pid, hostname, process liveness, home) is a
 * seam so the fixtures state facts instead of racing them. */
export interface LiveLockDeps {
  /** Base dir for `.kestrel/`. Defaults to `KESTREL_HOME` then `homedir()` — same rule as
   * `cli/credentials.ts`, deliberately reusing that home rather than inventing a second one. */
  readonly home?: string;
  /** Wall clock (epoch ms) for the record stamp. Defaults to `Date.now`. Never a decision input. */
  readonly now?: NowFn;
  /** This host's identity. Defaults to `os.hostname()`. */
  readonly hostId?: string;
  /** This process's pid. Defaults to `process.pid`. */
  readonly pid?: number;
  /**
   * The pid-liveness oracle — THE recovery predicate. Defaults to `process.kill(pid, 0)`, which
   * throws `ESRCH` iff no such process exists. Note `EPERM` means the process EXISTS but is owned by
   * another user ⇒ ALIVE ⇒ refuse. Anything we cannot disprove is treated as alive (fail closed).
   */
  readonly isAlive?: (pid: number) => boolean;
  /** Announce a takeover. Defaults to `process.stderr` — a recovery must never be silent. */
  readonly notify?: (line: string) => void;
}

/** A held lock. `held()` is the still-holder re-check a driver MUST call before each submit. */
export interface LiveLockHandle {
  /** The lock file this handle owns. */
  readonly path: string;
  /** The record we wrote when we won it. */
  readonly record: LiveLockRecord;
  /**
   * Re-READ the disk and answer whether WE still hold the lock. Never cached — the whole point is to
   * catch the case where a peer legitimately recovered us after a false death. Fail-closed: a
   * missing/corrupt/foreign file answers `false`, and it never throws (a driver calls this on the
   * hot path; a throw here would be an outage, and "I don't know" must read as "I don't hold it").
   */
  held(): boolean;
  /** Release the lock, idempotently. NEVER deletes a file whose token is not ours — a stale handle
   * must not evict the process that legitimately took over from it. */
  release(): void;
}

const LIVE_LOCK_DIR = "live";

/** Resolve the base home the same way `cli/credentials.ts` does — `KESTREL_HOME` then `homedir()`. */
function resolveHome(home: string | undefined): string {
  return home ?? process.env.KESTREL_HOME ?? homedir();
}

/**
 * The lock file for a scope: `$KESTREL_HOME/.kestrel/live/<sha256(venue,account)>.lock`.
 *
 * The basename is a HASH, not `ibkr-U1234567.lock`: a live account id must not sit in a path that
 * ends up in a shell history, a backup, or a stack trace. `JSON.stringify([venue, account])` is the
 * pre-image, which is unambiguous without needing the `\x00` delimiter trick used elsewhere in the
 * engine (and so keeps this file out of the grep-blind NUL set — see AGENTS.md).
 */
export function liveLockPath(scope: LiveLockScope, home?: string): string {
  const digest = createHash("sha256").update(JSON.stringify([scope.venue, scope.account])).digest("hex");
  return join(resolveHome(home), ".kestrel", LIVE_LOCK_DIR, `${digest}.lock`);
}

/** Mint the holder token — a pure function of the holder's identity, so it is stable under replay. */
function tokenOf(host: string, pid: number, acquiredAt: number): string {
  return createHash("sha256").update(JSON.stringify([host, pid, acquiredAt])).digest("hex");
}

/** The default liveness oracle. `kill(pid, 0)` sends no signal — it only probes existence.
 * `ESRCH` = no such process = DEAD (the only case we may recover). `EPERM` = exists, other user =
 * ALIVE. Any other error is unknown ⇒ ALIVE, because "unknown" must never authorize a takeover. */
function defaultIsAlive(pid: number): boolean {
  try {
    process.kill(pid, 0);
    return true;
  } catch (e) {
    const code = (e as NodeJS.ErrnoException).code;
    if (code === "ESRCH") return false;
    return true; // EPERM (alive, foreign owner) or anything we cannot interpret — fail closed.
  }
}

/** Parse + VALIDATE a lock record. Returns `undefined` for anything we do not positively recognize —
 * the caller turns that into an `unreadable` refusal. Validation is whitelist-shaped on purpose: an
 * unknown `version` is refused, so a future format is never half-interpreted by an old binary. */
function parseRecord(bytes: string): LiveLockRecord | undefined {
  try {
    const p = JSON.parse(bytes) as Partial<LiveLockRecord>;
    if (p?.version !== 1) return undefined;
    if (typeof p.host !== "string" || p.host.length === 0) return undefined;
    if (typeof p.pid !== "number" || !Number.isInteger(p.pid) || p.pid <= 0) return undefined;
    if (typeof p.token !== "string" || p.token.length === 0) return undefined;
    if (typeof p.acquired_at !== "number" || !Number.isFinite(p.acquired_at)) return undefined;
    if (typeof p.venue !== "string" || p.venue.length === 0) return undefined;
    return p as LiveLockRecord;
  } catch {
    return undefined;
  }
}

/** Read the record at `path`, distinguishing ABSENT (`null`) from PRESENT-BUT-UNREADABLE
 * (`undefined`) — the two lead to opposite outcomes (acquire vs refuse), so they must not collapse. */
function readRecord(path: string): LiveLockRecord | undefined | null {
  let bytes: string;
  try {
    bytes = readFileSync(path, "utf8");
  } catch (e) {
    if ((e as NodeJS.ErrnoException).code === "ENOENT") return null;
    throw e;
  }
  return parseRecord(bytes);
}

/** Exclusively CREATE the lock file (`O_CREAT|O_EXCL`), 0600. Returns false iff it already exists.
 * `wx` is what makes the winner atomic — never "check then write", which two processes both pass. */
function createExclusive(path: string, record: LiveLockRecord): boolean {
  let fd: number;
  try {
    fd = openSync(path, "wx", 0o600);
  } catch (e) {
    if ((e as NodeJS.ErrnoException).code === "EEXIST") return false;
    throw e;
  }
  try {
    writeSync(fd, JSON.stringify(record, null, 2) + "\n");
  } finally {
    closeSync(fd);
  }
  chmodSync(path, 0o600); // belt-and-braces against a permissive umask
  return true;
}

/**
 * Acquire the per-host live singleton for `scope`, or THROW {@link LiveLockRefused}.
 *
 * The ladder, fail-closed at every rung:
 *   1. exclusive create wins ⇒ held;
 *   2. it exists but is unreadable ⇒ `unreadable` (never interpreted, never deleted);
 *   3. it is from another host ⇒ `foreign-host` (we cannot prove a foreign pid's liveness);
 *   4. its pid is ALIVE (or not provably dead) ⇒ `held`;
 *   5. its pid is provably DEAD ⇒ recover, announced + recorded, then re-verify we won the race.
 *
 * There is no branch that returns without a lock.
 */
export function acquireLiveLock(scope: LiveLockScope, deps: LiveLockDeps = {}): LiveLockHandle {
  const path = liveLockPath(scope, deps.home);
  const host = deps.hostId ?? hostname();
  const pid = deps.pid ?? process.pid;
  const now = deps.now ?? Date.now;
  const isAlive = deps.isAlive ?? defaultIsAlive;
  const notify = deps.notify ?? ((line: string) => void process.stderr.write(line + "\n"));

  const acquiredAt = now();
  const mint = (recoveredFrom?: LiveLockRecord): LiveLockRecord => ({
    version: 1,
    venue: scope.venue,
    account: redactAccount(scope.account),
    host,
    pid,
    token: tokenOf(host, pid, acquiredAt),
    acquired_at: acquiredAt,
    ...(recoveredFrom === undefined
      ? {}
      : { recovered_from: { pid: recoveredFrom.pid, acquired_at: recoveredFrom.acquired_at } }),
  });

  try {
    mkdirSync(join(resolveHome(deps.home), ".kestrel", LIVE_LOCK_DIR), { recursive: true, mode: 0o700 });
  } catch (e) {
    throw new LiveLockRefused("io", `cannot create the lock dir for ${path}: ${String(e)}`, { cause: e });
  }

  // Rung 1 — the uncontended win.
  const fresh = mint();
  try {
    if (createExclusive(path, fresh)) return handleFor(path, fresh);
  } catch (e) {
    throw new LiveLockRefused("io", `cannot write the lock file ${path}: ${String(e)}`, { cause: e });
  }

  // Contended: someone's record is on disk.
  let incumbent: LiveLockRecord | undefined | null;
  try {
    incumbent = readRecord(path);
  } catch (e) {
    throw new LiveLockRefused("io", `cannot read the lock file ${path}: ${String(e)}`, { cause: e });
  }

  // It vanished between our create and our read — the holder released. One retry, then refuse: we
  // never loop, because an unbounded retry against a flapping peer is a busy-wait, not a guard.
  if (incumbent === null) {
    try {
      if (createExclusive(path, fresh)) return handleFor(path, fresh);
    } catch (e) {
      throw new LiveLockRefused("io", `cannot write the lock file ${path}: ${String(e)}`, { cause: e });
    }
    throw new LiveLockRefused("held", `the lock at ${path} was taken by a peer while acquiring it`);
  }

  // Rung 2 — present but not ours to interpret. Refuse; name the path so a human can clear it.
  if (incumbent === undefined) {
    throw new LiveLockRefused(
      "unreadable",
      `the lock file ${path} is not a readable v1 live-lock record — REFUSING to interpret or ` +
        `remove it. If no live session is running on this host, delete the file by hand and retry.`,
    );
  }

  // Rung 3 — a foreign host's record. Our process table says nothing about their pid.
  if (incumbent.host !== host) {
    throw new LiveLockRefused(
      "foreign-host",
      `the lock at ${path} is held by pid ${incumbent.pid} on host ${incumbent.host} (this host is ` +
        `${host}) — a foreign pid's liveness is not provable from here, so this REFUSES rather than ` +
        `recovers. Cross-host exclusion is the control-plane lease (kestrel-7o2.2), not this lock.`,
    );
  }

  // Rung 4 — the holder is alive (or we cannot prove otherwise). The ordinary refusal.
  if (isAlive(incumbent.pid)) {
    throw new LiveLockRefused(
      "held",
      `a live session (pid ${incumbent.pid} on ${incumbent.host}, acquired_at ${incumbent.acquired_at}) ` +
        `already holds ${incumbent.venue}/${incumbent.account} — a second live pump on one account ` +
        `would double every clip. REFUSING.`,
    );
  }

  // Rung 5 — the holder is PROVABLY dead. Recover, loudly.
  const recovered = mint(incumbent);
  notify(
    `live-lock: RECOVERED a stale lock at ${path} — holder pid ${incumbent.pid} on ${incumbent.host} ` +
      `(acquired_at ${incumbent.acquired_at}) is gone from this host's process table; pid ${pid} is ` +
      `taking ${incumbent.venue}/${incumbent.account}. If that process is in fact trading, KILL IT NOW.`,
  );
  try {
    unlinkSync(path);
  } catch (e) {
    if ((e as NodeJS.ErrnoException).code !== "ENOENT") {
      throw new LiveLockRefused("io", `cannot clear the stale lock ${path}: ${String(e)}`, { cause: e });
    }
    // ENOENT: a peer recoverer got there first. Fall through — our exclusive create decides it.
  }
  try {
    if (!createExclusive(path, recovered)) {
      // A peer recoverer won the exclusive create. THEY hold it; we refuse. This is why recovery
      // uses unlink+O_EXCL rather than an overwriting rename: last-writer-wins would let BOTH
      // recoverers believe they hold the account.
      throw new LiveLockRefused(
        "held",
        `a peer recovered the stale lock at ${path} first — REFUSING rather than racing it.`,
      );
    }
  } catch (e) {
    if (e instanceof LiveLockRefused) throw e;
    throw new LiveLockRefused("io", `cannot write the recovered lock ${path}: ${String(e)}`, { cause: e });
  }
  return handleFor(path, recovered);
}

/** Build the handle. `held()` re-reads the disk every call — the check exists precisely for the case
 * where our cached belief is the thing that is wrong. */
function handleFor(path: string, record: LiveLockRecord): LiveLockHandle {
  const stillOurs = (): boolean => {
    try {
      const cur = readRecord(path);
      return cur !== null && cur !== undefined && cur.token === record.token;
    } catch {
      return false; // unreadable ⇒ we do not hold it. "I don't know" is never "I hold it".
    }
  };
  return {
    path,
    record,
    held: stillOurs,
    release: (): void => {
      if (!stillOurs()) return; // idempotent, and never evicts a legitimate successor
      try {
        unlinkSync(path);
      } catch {
        // Best-effort: a release that cannot unlink leaves a lock whose pid is about to die, and the
        // next acquirer recovers it observably through rung 5. Never throw on the way out.
      }
    },
  };
}