// Durable records for one binding under <dataRoot>/discord/<name>/:
//   inbox.jsonl    every accepted Discord message id, appended before any
//                  other action; the restart guard reads it back
//   outbox.jsonl   one line per delivery state change: intent, confirmed,
//                  refused, unknown; keyed by nonce
//   drops.jsonl    one counter line per dropped or refused inbound message
//   admissions.jsonl one line per turn admitted, before the engine is asked
//   turns/<id>.json  one write-once record per turn
//   STOP           presence refuses new turns; one JSON line per writer
//                  ({at, reason}), appended, so the last line names who
//                  braked; `recover` removes only a STOP it wrote itself
//   notices.jsonl  once-per-day fixed lines already attempted (ceiling)
//   reloads.jsonl  one line per binding reload attempt: applied (with the
//                  differences) or refused (with the reason); the binding
//                  in memory only changes on an applied line
//   run.lock/      ownership directory (mkdir is atomic) holding owner.json
//                  {pid, start, boot}; `stop` signals only a live pid whose
//                  start time and boot id match; a stale lock refuses `run`
//                  until `unlock`, which is gated by STOP; `recover` is the
//                  supervised form for a service unit's pre-start
// Directories are 0700, files 0600. Lines are appended, never rewritten.

import { parseJsonFileText } from "../../organization/src/util.mjs";
import {
  appendFileSync, closeSync, constants, existsSync, fstatSync, fsyncSync, mkdirSync, openSync, readdirSync, readFileSync, readSync, renameSync, rmdirSync, rmSync,
  unlinkSync, writeFileSync, writeSync,
} from "node:fs";
import { basename, join } from "node:path";
import { BINDING_NAME, DiscordError, quietFs, quietRead } from "./errors.mjs";

// Where a refusal says a binding's journal is: its place under the data root
// when the directory carries a valid binding name (bindingDataDir() makes it
// so), never the host path (#1873).
export function journalLabel(dir) {
  const name = basename(String(dir));
  return BINDING_NAME.test(name) ? `discord/${name} under the data root` : "the binding's journal directory";
}

export const OUTBOX_STATUS = Object.freeze(["intent", "confirmed", "refused", "unknown"]);

export function ensureJournal(dir) {
  mkdirSync(join(dir, "turns"), { recursive: true, mode: 0o700 });
  return dir;
}

function appendLine(path, record) {
  const line = JSON.stringify(record);
  if (line.includes("\n")) throw new DiscordError("journal line must not contain a newline", 1);
  try {
    appendFileSync(path, line + "\n", { mode: 0o600 });
  } catch (e) {
    throw quietFs(e, basename(path), "append to");
  }
}

function readLines(path) {
  if (!existsSync(path)) return [];
  const out = [];
  let text;
  try {
    text = readFileSync(path, "utf8");
  } catch (e) {
    throw quietRead(e, basename(path));
  }
  for (const [i, line] of text.split("\n").entries()) {
    if (line.length === 0) continue;
    try {
      out.push(parseJsonFileText(line, `${path}:${i + 1}`));
    } catch {
      throw new DiscordError(`${basename(path)}:${i + 1}: not valid JSON`);
    }
  }
  return out;
}

// --- inbox ---

export function appendInbox(dir, entry) {
  if (typeof entry.id !== "string" || entry.id.length === 0) throw new DiscordError("inbox entry needs a message id", 1);
  appendLine(join(dir, "inbox.jsonl"), entry);
}

export function readInboxIds(dir) {
  return new Set(readLines(join(dir, "inbox.jsonl")).map((e) => e.id).filter((id) => typeof id === "string"));
}

// --- outbox ---

export function appendOutbox(dir, entry) {
  if (!OUTBOX_STATUS.includes(entry.status)) throw new DiscordError(`outbox status must be one of ${OUTBOX_STATUS.join(", ")}`, 1);
  if (typeof entry.nonce !== "string" || entry.nonce.length === 0) throw new DiscordError("outbox entry needs a nonce", 1);
  appendLine(join(dir, "outbox.jsonl"), entry);
}

// Latest state per nonce, in first-seen order. An intent with no later line
// is an "unknown": the process died between the POST and its receipt.
// `intentAt` is the first line's timestamp for that nonce and never moves;
// reconcile measures the dedupe window from it, not from the latest retry.
export function readOutbox(dir) {
  const byNonce = new Map();
  for (const e of readLines(join(dir, "outbox.jsonl"))) {
    if (typeof e.nonce !== "string") continue;
    const prev = byNonce.get(e.nonce);
    const intentAt = prev ? prev.intentAt : e.at;
    byNonce.set(e.nonce, { ...prev, ...e, intentAt });
  }
  return byNonce;
}

export function unresolvedOutbox(dir) {
  return [...readOutbox(dir).values()].filter((e) => e.status === "intent" || e.status === "unknown");
}

// --- drops ---

export function appendDrop(dir, entry) {
  appendLine(join(dir, "drops.jsonl"), entry);
}

export function readDrops(dir) {
  return readLines(join(dir, "drops.jsonl"));
}

// --- turns (write-once) ---

const TURN_ID = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/;

export function turnPath(dir, id) {
  if (!TURN_ID.test(String(id))) throw new DiscordError("invalid turn id (letters, digits, ., _ or -, at most 128)", 1);
  return join(dir, "turns", `${id}.json`);
}

// A parse refusal or a failed read or write names the turn by its id, never by
// host path (PR #1916, #1873); a listed file's id is named only when it passes TURN_ID.
const turnLabel = (id) => (TURN_ID.test(id) ? `turn record ${id}` : "turn record");

export function writeTurn(dir, id, record) {
  const path = turnPath(dir, id);
  let fd;
  try {
    fd = openSync(path, "wx", 0o600);
  } catch (err) {
    if (err.code === "EEXIST") throw new DiscordError(`turn record already exists: ${id}`, 1);
    throw quietFs(err, turnLabel(String(id)), "write");
  }
  try {
    try {
      writeSync(fd, JSON.stringify({ ...record, id }, null, 2) + "\n");
    } finally {
      closeSync(fd);
    }
  } catch (err) {
    throw quietFs(err, turnLabel(String(id)), "write");
  }
  return path;
}

function readTurnFile(path, id) {
  let text;
  try {
    text = readFileSync(path, "utf8");
  } catch (e) {
    throw quietRead(e, turnLabel(id));
  }
  return parseJsonFileText(text, path, turnLabel(id));
}

export function readTurn(dir, id) {
  return readTurnFile(turnPath(dir, id), String(id));
}

export function listTurns(dir) {
  const turns = join(dir, "turns");
  if (!existsSync(turns)) return [];
  let names;
  try {
    names = readdirSync(turns);
  } catch (e) {
    throw quietRead(e, `the turns directory in ${journalLabel(dir)}`);
  }
  return names
    .filter((f) => f.endsWith(".json"))
    .map((f) => readTurnFile(join(turns, f), f.slice(0, -".json".length)));
}

// The daily ceiling counts admissions on the current UTC date. An admission
// is appended before the engine is asked, so a turn interrupted by a crash
// still counts after restart. Refusals are drop lines, not admissions.
export function utcDate(now) {
  return new Date(now).toISOString().slice(0, 10);
}

export function appendAdmission(dir, entry) {
  if (typeof entry.id !== "string" || typeof entry.at !== "string") throw new DiscordError("admission needs id and at", 1);
  appendLine(join(dir, "admissions.jsonl"), entry);
}

export function countAdmissionsOn(dir, date) {
  const ids = new Set();
  for (const e of readLines(join(dir, "admissions.jsonl"))) {
    if (typeof e.id === "string" && typeof e.at === "string" && e.at.slice(0, 10) === date) ids.add(e.id);
  }
  return ids.size;
}

export function countTurnsOn(dir, date) {
  return listTurns(dir).filter((t) => typeof t.startedAt === "string" && t.startedAt.slice(0, 10) === date).length;
}

// --- stop switch and pid ---

export function stopPath(dir) {
  return join(dir, "STOP");
}

export function stopRequested(dir) {
  return existsSync(stopPath(dir));
}

export function requestStop(dir, reason = "stop") {
  return stopLock(dir, () => appendStop(dir, reason));
}

function stopLock(dir, action) {
  const lock = join(dir, '.stop.lock');
  try { mkdirSync(lock, { mode: 0o700 }); }
  catch { throw new DiscordError('STOP update is locked; refusing', 1); }
  try { return action(); } finally {
    try { rmdirSync(lock); } catch (e) { throw quietFs(e, `.stop.lock in ${journalLabel(dir)}`, "remove"); }
  }
}

// A STOP write or removal that fails names STOP in the journal, never its path (#1873).
const stopLabel = (dir) => `STOP in ${journalLabel(dir)}`;

function appendStop(dir, reason) {
  const path = stopPath(dir);
  try {
    const fd = openSync(path, constants.O_WRONLY | constants.O_APPEND | constants.O_CREAT | constants.O_NOFOLLOW, 0o600);
    try {
      writeSync(fd, JSON.stringify({ at: new Date().toISOString(), reason }) + "\n");
      fsyncSync(fd);
    } finally {
      closeSync(fd);
    }
  } catch (e) {
    throw quietFs(e, stopLabel(dir), "write");
  }
  return path;
}

function removeStop(dir) {
  try {
    unlinkSync(stopPath(dir));
  } catch (e) {
    throw quietFs(e, stopLabel(dir), "remove");
  }
}

const PROCESSING_GENERATION = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;
function processingStops(dir) {
  let fd;
  try {
    fd = openSync(stopPath(dir), constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
    const st = fstatSync(fd);
    if (!st.isFile() || st.uid !== process.getuid() || (st.mode & 0o7777) !== 0o600 || st.size > 65536) return null;
    const buffer = Buffer.alloc(65537), size = readSync(fd, buffer, 0, buffer.length, 0);
    if (size > 65536) return null;
    const bytes = buffer.subarray(0, size);
    return new TextDecoder('utf8', { fatal: true }).decode(bytes).split('\n').filter(Boolean).map(JSON.parse);
  } catch (e) { return e.code === 'ENOENT' ? [] : null; }
  finally { if (fd !== undefined) closeSync(fd); }
}
export function requestProcessingStop(dir, generation) {
  if (!PROCESSING_GENERATION.test(generation)) throw new DiscordError('invalid processing generation', 1);
  return stopLock(dir, () => {
    const rows = processingStops(dir);
    if (rows === null) throw new DiscordError('STOP cannot be verified', 1);
    const reason = `processing:${generation}`;
    if (!rows.some(r => r?.reason === reason)) appendStop(dir, reason);
    return stopPath(dir);
  });
}
export function clearProcessingStop(dir, generation) {
  if (!PROCESSING_GENERATION.test(generation)) return false;
  return stopLock(dir, () => {
    const rows = processingStops(dir);
    if (!rows?.length || rows.some(r => !r || Object.keys(r).sort().join(',') !== 'at,reason' ||
        r.reason !== `processing:${generation}` || !Number.isFinite(Date.parse(r.at)) ||
        new Date(r.at).toISOString() !== r.at)) return false;
    removeStop(dir);
    return true;
  });
}

export function clearStop(dir) {
  if (existsSync(stopPath(dir))) removeStop(dir);
}

// The STOP lines as written, oldest first. A line that does not parse is
// kept as {reason: null}: it was not written by this code, so it is never
// treated as ours.
export function readStop(dir) {
  if (!stopRequested(dir)) return [];
  let text;
  try {
    text = readFileSync(stopPath(dir), "utf8");
  } catch (e) {
    throw quietRead(e, stopLabel(dir));
  }
  return text.split("\n").filter((l) => l.trim() !== "").map((l) => {
    try {
      const v = JSON.parse(l);
      return v && typeof v === "object" && typeof v.reason === "string" ? v : { reason: null };
    } catch {
      return { reason: null };
    }
  });
}

// --- run lock ---
// One directory, <dir>/run.lock, is the ownership primitive: mkdir is atomic,
// so two starts cannot both create it. The owner record is published inside
// it by write-then-rename. Nothing reclaims a lock on its own: a lock whose
// record is missing (a start in progress, or one that crashed between mkdir
// and rename), or whose owner is dead or a reused pid, refuses `run` until
// an operator runs `unlock`.
//
// STOP is the quiescence gate that serializes `unlock` with every claim.
// `unlock` writes STOP before it inspects or touches the lock, and a claim
// re-checks STOP after it has published its record; a claim that finds STOP
// releases itself and refuses. So no process that claims during an unlock
// can ever hold the binding, and `unlock` only ever removes a lock whose
// owner is verified dead or that can no longer be held. Automatic reclaim
// and compare-then-restore were both rejected in review (#1509 comments
// 26123 and 26132): a rename proves nothing about which directory it moved.

export function lockPath(dir) {
  return join(dir, "run.lock");
}

export function ownerPath(dir) {
  return join(lockPath(dir), "owner.json");
}

// Process identity beyond the pid number: the /proc start time (ticks since
// boot, which a reused pid cannot reproduce within one boot) and the boot id
// (so the same pid and ticks after a reboot do not match either). Each is
// null where it cannot be read.
// Identity values have a fixed syntax: a start time is the tick count from
// /proc/<pid>/stat exactly as the kernel prints it (canonical unsigned
// decimal: no leading zeros, at most 2^64-1, and never zero for a process
// this connector could own), a boot id is the UUID from
// /proc/sys/kernel/random/boot_id. Anything else is not an identity and
// never compares: it reads as absent.
const START_RE = /^[1-9][0-9]{0,19}$/;
const START_MAX = 18446744073709551615n;
const BOOT_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
export function validStart(v) { return typeof v === "string" && START_RE.test(v) && BigInt(v) <= START_MAX; }
export function validBoot(v) { return typeof v === "string" && BOOT_RE.test(v); }

export function processStart(pid) {
  try {
    const stat = readFileSync(`/proc/${pid}/stat`, "utf8");
    const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" ");
    const start = fields[19] ?? null; // starttime is field 22 of stat; 20th after the comm field
    return validStart(start) ? start : null;
  } catch {
    return null;
  }
}

export function bootId() {
  try {
    const id = readFileSync("/proc/sys/kernel/random/boot_id", "utf8").trim();
    return validBoot(id) ? id : null;
  } catch {
    return null;
  }
}

export function identityOf(pid) {
  return { start: processStart(pid), boot: bootId() };
}

// {pid, start, boot} from the published owner record; null when there is no
// record file; {invalid: true} when the file exists but cannot be read or
// parsed or has no usable pid. An unreadable record is not the same as an
// absent one: absence is the STOP-gated publication interval, an unreadable
// record is an owner whose identity cannot be established, and that fails closed.
export function readPid(dir) {
  const path = ownerPath(dir);
  if (!existsSync(path)) return null;
  try {
    const rec = JSON.parse(readFileSync(path, "utf8"));
    if (rec === null || typeof rec !== "object" || !Number.isInteger(rec.pid) || rec.pid <= 0) return { invalid: true };
    return {
      pid: rec.pid,
      start: validStart(rec.start) ? rec.start : null,
      boot: validBoot(rec.boot) ? rec.boot : null,
    };
  } catch {
    return { invalid: true };
  }
}

export function pidAlive(pid) {
  try {
    process.kill(pid, 0);
    return true;
  } catch (err) {
    return err.code === "EPERM";
  }
}

// Identity check for a record:
//   "absent"   no record file
//   "invalid"  a record file that cannot be read, parsed, or has no pid
//   "dead"     the pid is not alive
//   "unknown"  the pid is alive but identity cannot be established: the
//              record lacks start or boot (an older record), carries a value
//              that is not a start tick or a boot id (corrupt metadata), or
//              the current /proc values cannot be read right now
//   "mismatch" the pid is alive and its identity positively differs
//   "live"     the pid is alive and start time and boot id both match
// Only "live" is ever signaled. "unknown" and "invalid" refuse everything:
// never signaled, never removed, never claimed over. Once the pid is
// positively dead, "dead" applies and unlock may clear it. `identity` is a
// test seam.
export function ownerState(rec, { identity = identityOf } = {}) {
  if (rec === null) return "absent";
  if (rec.invalid) return "invalid";
  if (!pidAlive(rec.pid)) return "dead";
  if (rec.start === null || rec.boot === null) return "unknown";
  const now = identity(rec.pid);
  if (now.start === null || now.boot === null) return "unknown";
  return now.start === rec.start && now.boot === rec.boot ? "live" : "mismatch";
}

export function ownerAlive(rec, opts) {
  return ownerState(rec, opts) === "live";
}

// Exit code for a refusal that a supervisor must not retry: STOP is present
// or the binding is held. Distinct from 1 (operation failed) so a service
// unit can restart after a crash and stay down after a brake.
export const BRAKE_EXIT = 3;
export const RECOVER_REASON = "recover";

export const UNLOCK_HINT = "if no connector is running for this binding, run `scripts/discord.sh unlock <binding>`";

// Explains why an existing lock refuses a new claim. Always a DiscordError.
function lockRefusal(dir, opts) {
  const existing = readPid(dir);
  const state = ownerState(existing, opts);
  if (state === "absent") return new DiscordError(`run.lock exists without an owner record: a start is in progress or was interrupted; ${UNLOCK_HINT}`, BRAKE_EXIT);
  if (state === "invalid") return new DiscordError(`run.lock has an owner record that cannot be read; refusing. Inspect run.lock/owner.json in ${journalLabel(dir)} by hand`, BRAKE_EXIT);
  if (state === "live") return new DiscordError(`another connector is running for this binding (pid ${existing.pid})`, BRAKE_EXIT);
  if (state === "unknown") return new DiscordError(`run.lock belongs to pid ${existing.pid}, which is alive but whose identity cannot be verified; refusing`, BRAKE_EXIT);
  return new DiscordError(`run.lock belongs to pid ${existing.pid}, which is gone or is a different process now; ${UNLOCK_HINT}`, BRAKE_EXIT);
}

export function writePid(dir, pid, { now = Date.now(), identity = identityOf } = {}) {
  const { start, boot } = identity(pid);
  if (start === null || boot === null) throw new DiscordError("cannot read this process's start time or the boot id from /proc; refusing to claim the binding", 1);
  const lock = lockPath(dir);
  try {
    mkdirSync(lock, { mode: 0o700 });
  } catch (err) {
    if (err.code !== "EEXIST") throw quietFs(err, `run.lock in ${journalLabel(dir)}`, "create");
    throw lockRefusal(dir, { identity });
  }
  const tmp = join(lock, "owner.json.tmp");
  try {
    writeFileSync(tmp, JSON.stringify({ pid, start, boot, at: new Date(now).toISOString() }) + "\n", { mode: 0o600 });
    renameSync(tmp, ownerPath(dir));
  } catch (e) {
    throw quietFs(e, `run.lock/owner.json in ${journalLabel(dir)}`, "write");
  }
  // The gate: STOP written before this point (by `stop` or `unlock`) means
  // this claim must not stand, however it interleaved with an unlock.
  if (stopRequested(dir)) {
    clearPid(dir, pid);
    throw new DiscordError(`STOP is present in ${journalLabel(dir)}; remove it to run`, BRAKE_EXIT);
  }
}

// Operator cleanup, gated by STOP. Writes STOP first, so every claim that
// publishes from now on releases itself. Refuses while the recorded owner is
// live (use `stop`) or alive with unverifiable identity (never removed).
// Refuses an owner record it cannot read. Otherwise removes the lock.
// Returns the record that was cleared (null for a lock without one), or
// false when there was no lock. STOP stays in place;
// remove it to run again. `beforeRemove` and `identity` are test seams;
// `reason` is what the STOP line says (`recover` uses its own).
export function unlock(dir, { beforeRemove = null, identity = identityOf, reason = "unlock" } = {}) {
  requestStop(dir, reason);
  const lock = lockPath(dir);
  if (!existsSync(lock)) return false;
  const rec = readPid(dir);
  const state = ownerState(rec, { identity });
  if (state === "live") throw new DiscordError(`refusing to unlock: the connector is running (pid ${rec.pid}); use stop, and unlock only a lock whose owner is gone`, 1);
  if (state === "unknown") throw new DiscordError(`refusing to unlock: pid ${rec.pid} is alive and its identity cannot be verified; nothing removed. Stop that process first`, 1);
  if (state === "invalid") throw new DiscordError(`refusing to unlock: the owner record cannot be read; nothing removed. Inspect run.lock/owner.json in ${journalLabel(dir)} by hand`, 1);
  if (beforeRemove) beforeRemove();
  removeLock(dir, { maxRetries: 10, retryDelay: 20 });
  return rec;
}

// Supervised pre-start, for the service unit's ExecStartPre. Lets a run
// that crashed (or a reboot) start again without a hand, while every
// operator brake still holds. In order:
//   STOP present         refuse with BRAKE_EXIT and touch nothing. `stop`
//                        and `unlock` wrote it; nothing automatic removes it.
//                        The one exception is a STOP made of exactly one
//                        line that an earlier `recover` wrote and then died
//                        before removing: that one is ours and goes.
//   no run.lock          nothing to do.
//   owner live           refuse: another connector holds the binding.
//   owner unknown, or record invalid   refuse, as `unlock` would.
//   owner dead, or no record           `unlock` (STOP written first, so a
//                        claim that publishes meanwhile releases itself),
//                        then that STOP is removed again, but only if it
//                        still consists of the single line `unlock` wrote
//                        for us. Any other line means an operator braked in
//                        the meantime; STOP stays and the start is refused.
// Returns "clean" or "cleared". Every refusal is a DiscordError with
// BRAKE_EXIT so a supervisor does not retry it. `beforeRemove` and
// `identity` are test seams.
export function recover(dir, { identity = identityOf, beforeRemove = null } = {}) {
  const ours = (lines) => lines.length === 1 && lines[0].reason === RECOVER_REASON;
  if (stopRequested(dir)) {
    if (!ours(readStop(dir))) throw new DiscordError(`STOP is present in ${journalLabel(dir)}; the brake is on. Remove it to run`, BRAKE_EXIT);
    clearStop(dir);
  }
  if (!existsSync(lockPath(dir))) return "clean";
  const state = ownerState(readPid(dir), { identity });
  if (state === "live" || state === "unknown" || state === "invalid") throw lockRefusal(dir, { identity });
  unlock(dir, { identity, reason: RECOVER_REASON, beforeRemove });
  if (!ours(readStop(dir))) throw new DiscordError(`STOP was written by an operator while recovering ${journalLabel(dir)}; the brake stays. Remove it to run`, BRAKE_EXIT);
  clearStop(dir);
  return "cleared";
}

// The verified live owner to signal, or null. Never returns a pid whose
// identity cannot be proven.
export function stopTarget(dir, opts) {
  const rec = readPid(dir);
  return ownerAlive(rec, opts) ? rec.pid : null;
}

export function clearPid(dir, pid) {
  const rec = readPid(dir);
  if (rec !== null && !rec.invalid && rec.pid === pid) removeLock(dir);
}

function removeLock(dir, retry = {}) {
  try {
    rmSync(lockPath(dir), { recursive: true, force: true, ...retry });
  } catch (e) {
    throw quietFs(e, `run.lock in ${journalLabel(dir)}`, "remove");
  }
}

// --- reloads ---

export function appendReload(dir, entry) {
  if (!["applied", "refused"].includes(entry.outcome)) throw new DiscordError("reload entry needs an outcome", 1);
  appendLine(join(dir, "reloads.jsonl"), entry);
}

export function readReloads(dir) {
  return readLines(join(dir, "reloads.jsonl"));
}

// --- notices ---
// Fixed lines that must go out at most once per UTC day (the ceiling
// notice). The line is appended before the delivery attempt, so a crash
// mid-delivery does not produce a second attempt after restart.
export function appendNotice(dir, entry) {
  if (typeof entry.kind !== "string" || typeof entry.date !== "string") throw new DiscordError("notice needs kind and date", 1);
  appendLine(join(dir, "notices.jsonl"), entry);
}

export function noticeOn(dir, kind, date) {
  return readLines(join(dir, "notices.jsonl")).some((e) => e.kind === kind && e.date === date);
}