stack/packages/cli/src/notifier.mjs

// The notifier (lead decision 70): every poll it reads the human inbox
// through a reader capability, DMs each open blocking decision once, and
// sends one digest a day at 08:00 America/Chicago. It never writes to the
// bus; its memory is the journal `<dataRoot>/notify/<business>/sent.jsonl`
// (0600, in a 0700 directory), one line per send attempt:
//
//   {at, kind: "dm"|"digest", decision, day?, outcome: "confirmed"|"refused"|"unknown"|"gave-up", messageId, status?}
//
// A decision counts as sent once it has a confirmed line; a day's digest
// likewise. A refused or unknown send is retried with backoff (30 s
// doubling to 30 min): a duplicate costs less than a miss, and Discord's
// nonce folds a retry inside its dedupe window into the first message.
// The exception is a definite refusal, an HTTP 4xx other than 429 (lead
// decision 72): the next try waits the full 30 min, counted from the
// refusal's line so a restart cannot shorten it (lead decision 73), and
// after five for one decision, counted from the journal, the notifier
// appends one `gave-up` line, logs once and stops sending that DM. Five
// tries span two hours, long enough to fix a binding or token typo. The
// digest marks it "DM refused, not retried"; recovery is the operator
// deciding it. Unknown outcomes (network, 5xx, 429) retry without a limit.
// No Discord channel or user id goes in the journal, a log line or an
// error; the Discord side (packages/discord/src/notify.mjs) keeps them.

import { createHash } from "node:crypto";
import { accessSync, closeSync, constants, fstatSync, fsyncSync, ftruncateSync, lstatSync, mkdirSync, openSync, readFileSync, writeSync } from "node:fs";
import { dirname, join } from "node:path";
import { CliError } from "./errors.mjs";
import { authorizationLines, optionsLine, shortId } from "./format.mjs";

export const ZONE = "America/Chicago";
export const DIGEST_HOUR = 8;
export const POLL_MS = 30000;
const BACKOFF_MS = 30000;
const BACKOFF_MAX_MS = 30 * 60 * 1000;
const LIMIT = 2000;
export const REFUSAL_LIMIT = 5;

export const journalPath = (dataRoot, business) => join(dataRoot, "notify", business, "sent.jsonl");

// Local date and hour in the IANA zone. An unknown zone throws RangeError,
// so a broken time-zone database refuses rather than guessing.
export function zoned(date, zone = ZONE) {
  const parts = Object.fromEntries(
    new Intl.DateTimeFormat("en-US", { timeZone: zone, year: "numeric", month: "2-digit", day: "2-digit", hour: "2-digit", hourCycle: "h23" })
      .formatToParts(date)
      .map((p) => [p.type, p.value]),
  );
  return { day: `${parts.year}-${parts.month}-${parts.day}`, hour: Number(parts.hour) };
}

// Discord nonces are at most 25 characters. The digest nonce carries the
// business, so two businesses sharing a bot and a recipient never send the
// same nonce on the same day.
const hash23 = (text) => createHash("sha256").update(text).digest("hex").slice(0, 23);
export const dmNonce = (id) => `dm${hash23(String(id))}`;
export const digestNonce = (business, day) => `dg${hash23(`${business}\n${day}`)}`;

function clip(text, max) {
  return text.length > max ? `${text.slice(0, max - 1)}…` : text;
}

export function dmContent(business, d) {
  const lines = [
    `Mosaic (${business}): a blocking decision needs you.`,
    clip(d.question, 1000),
    `options: ${optionsLine(d)} (recommended: ${d.recommendation})`,
    ...authorizationLines(d),
    `raised by ${d.raised_by_role}${d.task_ref ? `, task ${d.task_ref}` : ""}`,
    `decide: mosaic decide ${shortId(d.id)} <option>`,
  ];
  return clip(lines.join("\n"), LIMIT);
}

export function digestContent(business, day, inbox, dmSent, dmGaveUp = () => false) {
  if (inbox.length === 0) return `Mosaic digest (${business}, ${day}): your inbox is empty.`;
  const head = `Mosaic digest (${business}, ${day}): ${inbox.length} open decision(s).`;
  const tail = "Run mosaic inbox for the full list.";
  const lines = [head];
  let shown = 0;
  for (const d of inbox) {
    const mark = d.blocking ? (dmSent(d.id) ? "[blocking, DM sent] " : dmGaveUp(d.id) ? "[blocking, DM refused, not retried] " : "[blocking, DM pending] ") : "";
    const line = `- ${mark}${shortId(d.id)} ${d.action}: ${clip(d.question.replace(/\s+/g, " "), 160)}`;
    const more = inbox.length - shown - 1;
    const reserve = more > 0 ? `\n… and ${more} more.`.length : 0;
    if ([...lines, line].join("\n").length + reserve + tail.length + 1 > LIMIT) break;
    lines.push(line);
    shown++;
  }
  if (shown < inbox.length) lines.push(`… and ${inbox.length - shown} more.`);
  lines.push(tail);
  return lines.join("\n");
}

const { O_APPEND, O_CREAT, O_NOFOLLOW, O_RDWR, O_WRONLY } = constants;

// UTC stamp for a torn-tail copy, e.g. 20261008T235212345Z.
const stamp = (date) => date.toISOString().replace(/[-:.]/g, "");

// Step 1 of a torn-tail repair (lead decision 71): the torn bytes go to a
// new file, torn-<UTC stamp>.bin (0600), fsynced with its directory. A name
// that exists already gets a counter, so a repeat never overwrites a copy.
function copyTorn(dir, bytes, date) {
  for (let n = 0; ; n++) {
    const name = `torn-${stamp(date)}${n ? `-${n}` : ""}.bin`;
    let fd;
    try {
      fd = openSync(join(dir, name), O_WRONLY | O_CREAT | constants.O_EXCL | O_NOFOLLOW, 0o600);
    } catch (e) {
      if (e.code === "EEXIST") continue;
      throw e;
    }
    try {
      writeSync(fd, bytes);
      fsyncSync(fd);
    } finally {
      closeSync(fd);
    }
    const dfd = openSync(dir, constants.O_RDONLY);
    try {
      fsyncSync(dfd);
    } finally {
      closeSync(dfd);
    }
    return name;
  }
}

const OUTCOMES = ["confirmed", "refused", "unknown", "gave-up"];
const DAY = /^\d{4}-\d{2}-\d{2}$/;
const isDay = (d) => typeof d === "string" && DAY.test(d) && !Number.isNaN(Date.parse(d)) && new Date(d).toISOString().startsWith(d);
const UNWRITABLE = ["EACCES", "EPERM", "EROFS"];

// The first field of a journal record that has the wrong type, or null
// (lead decision 72, J4). A confirmed line carries the message id.
function badField(r) {
  if (!r || typeof r !== "object" || Array.isArray(r)) return "record";
  if (typeof r.at !== "string" || Number.isNaN(Date.parse(r.at)) || new Date(r.at).toISOString() !== r.at) return "at";
  if (!["dm", "digest"].includes(r.kind)) return "kind";
  if (r.kind === "dm" ? typeof r.decision !== "string" || r.decision === "" : r.decision !== null && r.decision !== undefined) return "decision";
  if (r.kind === "digest" ? !isDay(r.day) : r.day !== undefined && !isDay(r.day)) return "day";
  if (!OUTCOMES.includes(r.outcome) || (r.outcome === "gave-up" && r.kind !== "dm")) return "outcome";
  if (r.outcome === "confirmed" ? typeof r.messageId !== "string" : r.messageId !== null && typeof r.messageId !== "string") return "messageId";
  if (r.status !== undefined && !Number.isInteger(r.status)) return "status";
  return null;
}

// A DM refused with an HTTP 4xx other than 429.
const definite = (r) => r.kind === "dm" && r.outcome === "refused" && Number.isInteger(r.status) && r.status >= 400 && r.status < 500 && r.status !== 429;

// Opens (creating if needed) the journal. The directory must be 0700 or
// tighter, writable and not a symlink, and the file 0600, both owned by
// this user; a symlinked journal refuses. Every complete line must parse
// and type-check, or the open refuses with exit 3.
// A final line without its newline is a write that never finished (lead
// decision 71): its bytes are copied to torn-<stamp>.bin, then the journal
// is truncated to its last newline and fsynced, and both steps are logged.
// A crash between the two leaves the tail torn, and the next open repeats
// both; the second copy is harmless.
export function openJournal(file, { log = () => {}, now = () => new Date() } = {}) {
  const dir = dirname(file);
  try {
    mkdirSync(dir, { recursive: true, mode: 0o700 });
  } catch (e) {
    // A dangling link in place of the directory fails mkdir with ENOENT.
    if (e.code === "ENOENT" && lstatSync(dir, { throwIfNoEntry: false })?.isSymbolicLink()) throw new CliError(`notify journal directory must not be a symlink: ${dir}`, 3);
    if ([...UNWRITABLE, "ENOTDIR"].includes(e.code)) throw new CliError(`notify journal directory cannot be created (${e.code}): ${dir}`, 3);
    if (e.code !== "EEXIST") throw e;
  }
  const ds = lstatSync(dir);
  if (ds.isSymbolicLink()) throw new CliError(`notify journal directory must not be a symlink: ${dir}`, 3);
  if (!ds.isDirectory() || ds.uid !== process.getuid() || (ds.mode & 0o077) !== 0) {
    throw new CliError(`notify journal directory must be mode 0700 and owned by this user: ${dir}`, 3);
  }
  try {
    accessSync(dir, constants.W_OK | constants.X_OK);
  } catch (e) {
    if (UNWRITABLE.includes(e.code)) throw new CliError(`notify journal directory is not writable (${e.code}): ${dir}`, 3);
    throw e;
  }
  let fd;
  try {
    fd = openSync(file, O_RDWR | O_APPEND | O_CREAT | O_NOFOLLOW, 0o600);
  } catch (e) {
    if (e.code === "ELOOP") throw new CliError(`notify journal must not be a symlink: ${file}`, 3);
    if (UNWRITABLE.includes(e.code)) throw new CliError(`notify journal is not writable (${e.code}): ${file}`, 3);
    if (e.code === "EISDIR") throw new CliError(`notify journal must be a regular file, mode 0600, owned by this user: ${file}`, 3);
    throw e;
  }
  const sent = new Set();
  const days = new Set();
  const refusals = new Map();
  const gaveUp = new Set();
  const refusedAt = new Map();
  const count = (r) => {
    if (r.kind === "dm" && r.outcome === "gave-up") gaveUp.add(r.decision);
    if (definite(r)) {
      refusals.set(r.decision, (refusals.get(r.decision) ?? 0) + 1);
      refusedAt.set(r.decision, Date.parse(r.at));
    }
    if (r.outcome !== "confirmed") return;
    if (r.kind === "dm") sent.add(r.decision);
    else days.add(r.day);
  };
  try {
    const st = fstatSync(fd);
    if (!st.isFile() || st.uid !== process.getuid() || (st.mode & 0o777) !== 0o600) {
      throw new CliError(`notify journal must be a regular file, mode 0600, owned by this user: ${file}`, 3);
    }
    const bytes = readFileSync(fd);
    const end = bytes.lastIndexOf(0x0a) + 1;
    const lines = bytes.subarray(0, end).toString("utf8").split("\n");
    lines.pop();
    lines.forEach((line, i) => {
      let r;
      try {
        r = JSON.parse(line);
      } catch {
        r = null;
      }
      const bad = r === null ? "json" : badField(r);
      if (bad) throw new CliError(`notify journal line ${i + 1} is malformed (${bad}): ${file}`, 3);
      count(r);
    });
    if (end < bytes.length) {
      const name = copyTorn(dir, bytes.subarray(end), now());
      log(`notify journal: copied a torn final line (${bytes.length - end} bytes) to ${name}`);
      ftruncateSync(fd, end);
      fsyncSync(fd);
      log(`notify journal: truncated ${file} to its last newline (${end} bytes)`);
    }
  } finally {
    closeSync(fd);
  }
  return {
    sent,
    days,
    refusals,
    gaveUp,
    refusedAt,
    append(record) {
      const afd = openSync(file, O_WRONLY | O_APPEND | O_NOFOLLOW);
      try {
        writeSync(afd, `${JSON.stringify(record)}\n`);
      } finally {
        closeSync(afd);
      }
      count(record);
    },
  };
}

// inbox(): the broker's inbox through the reader capability.
// direct.send({content, nonce}) resolves {messageId} or throws a RestOutcome.
export function createNotifier({ business, dataRoot, inbox, direct, now = () => new Date(), zone = ZONE, log = () => {} }) {
  zoned(now(), zone);
  const journal = openJournal(journalPath(dataRoot, business), { log, now });
  const backoff = new Map();

  function waiting(key, t) {
    const b = backoff.get(key);
    return b !== undefined && t < b.next;
  }

  // One `gave-up` line and one log line; the decision is never sent again.
  function giveUp(decision, t) {
    journal.append({ at: new Date(t).toISOString(), kind: "dm", decision, outcome: "gave-up", messageId: null });
    backoff.delete(`dm:${decision}`);
    log(`notify: dm ${shortId(decision)} refused ${journal.refusals.get(decision)} times; gave up, not retried`);
  }

  async function attempt(key, record, message) {
    const t = now().getTime();
    try {
      const { messageId } = await direct.send(message);
      backoff.delete(key);
      journal.append({ at: new Date(t).toISOString(), ...record, outcome: "confirmed", messageId });
      return true;
    } catch (err) {
      const kind = err?.kind === "refused" ? "refused" : "unknown";
      const status = Number.isInteger(err?.details?.status) ? err.details.status : null;
      const line = { at: new Date(t).toISOString(), ...record, outcome: kind, messageId: null, ...(status !== null ? { status } : {}) };
      const n = (backoff.get(key)?.n ?? -1) + 1;
      const wait = definite(line) ? BACKOFF_MAX_MS : Math.min(BACKOFF_MS * 2 ** n, BACKOFF_MAX_MS);
      backoff.set(key, { n, next: t + wait });
      journal.append(line);
      if (record.kind === "dm" && (journal.refusals.get(record.decision) ?? 0) >= REFUSAL_LIMIT) {
        giveUp(record.decision, t);
        return false;
      }
      log(`notify: ${record.kind} ${kind}${status !== null ? ` (HTTP ${status})` : ""}; retry in ${Math.round(wait / 1000)} s`);
      return false;
    }
  }

  // One poll. Returns what it did, for the tests and the log.
  async function tick() {
    const done = { dms: 0, digest: false, failed: 0 };
    let list;
    try {
      list = await inbox();
    } catch (err) {
      log(`notify: inbox read failed (${err?.code ?? err?.message ?? "error"}); next poll retries`);
      return { ...done, inboxError: true };
    }
    const t = now();
    for (const d of list) {
      if (!d.blocking || journal.sent.has(d.id) || journal.gaveUp.has(d.id)) continue;
      // A crash between the fifth refusal and its gave-up line.
      if ((journal.refusals.get(d.id) ?? 0) >= REFUSAL_LIMIT) {
        giveUp(d.id, t.getTime());
        continue;
      }
      // A definite refusal waits the full cap from its journal line, which
      // the in-memory backoff loses on a restart.
      if (t.getTime() < (journal.refusedAt.get(d.id) ?? -Infinity) + BACKOFF_MAX_MS) continue;
      const key = `dm:${d.id}`;
      if (waiting(key, t.getTime())) continue;
      if (await attempt(key, { kind: "dm", decision: d.id }, { content: dmContent(business, d), nonce: dmNonce(d.id) })) done.dms++;
      else done.failed++;
    }
    const { day, hour } = zoned(t, zone);
    const key = `digest:${day}`;
    if (hour >= DIGEST_HOUR && !journal.days.has(day) && !waiting(key, t.getTime())) {
      const content = digestContent(business, day, list, (id) => journal.sent.has(id), (id) => journal.gaveUp.has(id));
      if (await attempt(key, { kind: "digest", decision: null, day }, { content, nonce: digestNonce(business, day) })) done.digest = true;
      else done.failed++;
    }
    return done;
  }

  return Object.freeze({ tick, journal: journalPath(dataRoot, business) });
}

// Runs tick() every pollMs until stop(). Ticks never overlap.
export function runLoop(notifier, { pollMs = POLL_MS, log = () => {} } = {}) {
  let timer = null;
  let stopped = false;
  let running = Promise.resolve();
  const loop = () => {
    if (stopped) return;
    running = notifier
      .tick()
      .catch((err) => log(`notify: poll failed: ${err?.message ?? err}`))
      .finally(() => {
        if (!stopped) timer = setTimeout(loop, pollMs);
      });
  };
  loop();
  return {
    async stop() {
      stopped = true;
      clearTimeout(timer);
      await running;
    },
  };
}