stack/packages/discord/src/journal.mjs

// 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
//   notices.jsonl  once-per-day fixed lines already attempted (ceiling)
//   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
// Directories are 0700, files 0600. Lines are appended, never rewritten.

import {
  appendFileSync, closeSync, existsSync, mkdirSync, openSync, readdirSync, readFileSync, renameSync, rmSync,
  unlinkSync, writeFileSync, writeSync,
} from "node:fs";
import { join } from "node:path";
import { DiscordError } from "./errors.mjs";

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);
  appendFileSync(path, line + "\n", { mode: 0o600 });
}

function readLines(path) {
  if (!existsSync(path)) return [];
  const out = [];
  const text = readFileSync(path, "utf8");
  for (const [i, line] of text.split("\n").entries()) {
    if (line.length === 0) continue;
    try {
      out.push(JSON.parse(line));
    } catch (err) {
      throw new DiscordError(`${path}:${i + 1}: not valid JSON (${err.message})`);
    }
  }
  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: ${JSON.stringify(id)}`, 1);
  return join(dir, "turns", `${id}.json`);
}

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: ${path}`, 1);
    throw err;
  }
  try {
    writeSync(fd, JSON.stringify({ ...record, id }, null, 2) + "\n");
  } finally {
    closeSync(fd);
  }
  return path;
}

export function readTurn(dir, id) {
  return JSON.parse(readFileSync(turnPath(dir, id), "utf8"));
}

export function listTurns(dir) {
  const turns = join(dir, "turns");
  if (!existsSync(turns)) return [];
  return readdirSync(turns)
    .filter((f) => f.endsWith(".json"))
    .map((f) => JSON.parse(readFileSync(join(turns, f), "utf8")));
}

// 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") {
  const path = stopPath(dir);
  const fd = openSync(path, "a", 0o600);
  try {
    writeSync(fd, JSON.stringify({ at: new Date().toISOString(), reason }) + "\n");
  } finally {
    closeSync(fd);
  }
  return path;
}

export function clearStop(dir) {
  const path = stopPath(dir);
  if (existsSync(path)) unlinkSync(path);
}

// --- 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";
}

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}`, 1);
  if (state === "invalid") return new DiscordError(`run.lock has an owner record that cannot be read; refusing. Inspect ${ownerPath(dir)} by hand`, 1);
  if (state === "live") return new DiscordError(`another connector is running for this binding (pid ${existing.pid})`, 1);
  if (state === "unknown") return new DiscordError(`run.lock belongs to pid ${existing.pid}, which is alive but whose identity cannot be verified; refusing`, 1);
  return new DiscordError(`run.lock belongs to pid ${existing.pid}, which is gone or is a different process now; ${UNLOCK_HINT}`, 1);
}

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 err;
    throw lockRefusal(dir, { identity });
  }
  const tmp = join(lock, "owner.json.tmp");
  writeFileSync(tmp, JSON.stringify({ pid, start, boot, at: new Date(now).toISOString() }) + "\n", { mode: 0o600 });
  renameSync(tmp, ownerPath(dir));
  // 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 ${dir}; remove it to run`, 1);
  }
}

// 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.
export function unlock(dir, { beforeRemove = null, identity = identityOf } = {}) {
  requestStop(dir, "unlock");
  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 ${ownerPath(dir)} by hand`, 1);
  if (beforeRemove) beforeRemove();
  rmSync(lock, { recursive: true, force: true, maxRetries: 10, retryDelay: 20 });
  return rec;
}

// 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) rmSync(lockPath(dir), { recursive: true, force: true });
}

// --- 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);
}