mesh-catalog/modules/gitea/check.ts

// **mesh/check is written from the checker's owed state, read by a watch** (ADR 0521; R4e round 2 finding R2-1; finding K1
// before it; novox/hq issue 1301; ADR 0447 §1, ADR 0503 §3: main requires mesh/check alone).
//
// No tool writes mesh/check: a tool is told no caller, and the console every agent session speaks through (node-tools)
// invokes every tool, so a tool that wrote it could be called by any session to merge past the checker (the fault issue
// 293 closed for the two statuses before it). Nor does an event: nothing in the delivery plan orders this module before the
// forge bridge, and a durable consumer whose filters change keeps its deliver policy, so an event said before this module
// consumed it was lost. The checker keeps what it owes the forge in its state `owed` (ADR 0484 §2), which it announces
// readable (ADR 0495 §2-§3), and only the checker writes it. This module watches it: the watch hands over the latest value of
// every key when it starts, whenever that is, and then every change. Each key whose act is the status mesh/check is
// written on its head, once per revision; a revision older than one already written on the key is dropped.
//
// A read refused (the grant not yet composed) is said once and asked again on a bound until it holds. A forge that refuses
// a status is asked again on a bound, until it writes or a newer revision of the key comes.

import type { CommitStatus } from "./client.js";
import { bindUntilHeld, bindWait } from "./bind.js";
import { checkName } from "./guard.js";
import { CHECK_STATUS } from "./pulls.js";

/** The checker's state this module reads, and the act of its keys this module writes. */
export const CHECKER_OWED = "mesh-delivery-checker.owed";
export const CHECK_ACT = `status=${CHECK_STATUS}`;

const states = new Set(["pending", "success", "error", "failure", "warning"]);

/** What writing the check needs of the forge. */
export interface CheckForge {
  setCommitStatus(owner: string, repo: string, sha: string, status: CommitStatus): Promise<void>;
}

/** One change of the checker's owed, as the SDK's state watch hands it. */
export interface OwedChange {
  key: string;
  op: "put" | "delete";
  value?: unknown;
  revision: number;
  current?: boolean;
}

/** A key part as mesh-sdk go/delivery EscapeName writes it, read back: `=` and two lowercase hex digits per byte. */
function unescapeName(part: string): string {
  if (!/^([-/_a-zA-Z0-9]|=[0-9a-f]{2})+$/.test(part)) throw new Error(`${JSON.stringify(part)} is not an escaped name`);
  return Buffer.from(part.replace(/=([0-9a-f]{2})/g, (_, h) => String.fromCharCode(parseInt(h, 16))), "latin1").toString("utf8");
}

/** The pull request and head of the checker's owed key (mesh-sdk go/delivery CheckerOwedKey):
 *  `<owner>/<name>.<number>.<head>.status=mesh/check`, owner and name escaped. */
export function owedKeyOf(key: string): { owner: string; repo: string; number: number; sha: string } {
  const parts = String(key ?? "").split(".");
  if (parts.length !== 4) throw new Error(`${JSON.stringify(key)} is not an owed key, <repository>.<number>.<head>.<act>`);
  const [repository, number, sha, act] = parts;
  if (act !== CHECK_ACT) throw new Error(`${JSON.stringify(key)} owes ${JSON.stringify(act)}, not ${CHECK_ACT}`);
  const m = /^([^/]+)\/([^/]+)$/.exec(repository);
  if (!m) throw new Error(`${JSON.stringify(key)} names no <owner>/<name>`);
  if (!/^(0|[1-9][0-9]*)$/.test(number)) throw new Error(`${JSON.stringify(key)} names no pull request number`);
  if (!/^([0-9a-f]{40}|[0-9a-f]{64})$/i.test(sha)) throw new Error(`${JSON.stringify(key)} names no head's whole sha`);
  return { owner: checkName("owner", unescapeName(m[1])), repo: checkName("name", unescapeName(m[2])), number: Number(number), sha };
}

/** The status an owed change asks for, checked as mesh-sdk go/delivery DecodeCheckerOwed checks it; or why it is not one. */
export function checkStatusOf(change: OwedChange): { owner: string; repo: string; sha: string; revision: number; status: CommitStatus } {
  if (change.op !== "put") throw new Error(`a ${change.op} owes nothing`);
  const { owner, repo, sha } = owedKeyOf(change.key);
  const revision = change.revision;
  if (typeof revision !== "number" || !Number.isSafeInteger(revision) || revision < 0) {
    throw new Error(`${JSON.stringify(revision)} is not a revision`);
  }
  const v = change.value;
  if (!v || typeof v !== "object") throw new Error("the value is no owed act");
  const o = v as Record<string, unknown>;
  if (o.v !== 1) throw new Error(`the owed act's version ${JSON.stringify(o.v)} is not one this module reads`);
  if (o.act !== "status" || o.context !== CHECK_STATUS) {
    throw new Error(`the owed act is ${JSON.stringify(o.act)} ${JSON.stringify(o.context)}, not the status ${CHECK_STATUS}`);
  }
  const state = String(o.state ?? "");
  if (!states.has(state)) throw new Error(`${JSON.stringify(o.state)} is not a status the forge keeps`);
  let description = String(o.text ?? "").replace(/\s+/g, " ").trim();
  if (description.length > 140) description = description.slice(0, 139) + "…";
  const target = typeof o.target === "string" && /^https?:\/\//.test(o.target) ? o.target : undefined;
  const status: CommitStatus = { state: state as CommitStatus["state"], context: CHECK_STATUS, description,
    ...(target ? { target_url: target } : {}) };
  return { owner, repo, sha, revision, status };
}

/** The one writer of mesh/check: each owed revision of the checker's, in revision order per key. */
export class CheckStatuses {
  /** The newest revision written of each key, and the newest seen; the oldest keys go past what any forge holds open. */
  private written = new Map<string, number>();
  private seen = new Map<string, number>();
  /** The turn each key's writes wait on, and the asks again still running. */
  private turns = new Map<string, Promise<unknown>>();
  private again = new Set<Promise<void>>();
  private forge: CheckForge;
  private say: (line: string) => void;
  private sleep: (ms: number) => Promise<void>;

  constructor(forge: CheckForge, say: (line: string) => void = (l) => console.error(l),
    sleep: (ms: number) => Promise<void> = (ms) => new Promise((r) => setTimeout(r, ms))) {
    this.forge = forge;
    this.say = say;
    this.sleep = sleep;
  }

  /** Write the change's status: "written", or "dropped" for one that owes no mesh/check or is older than one written. A
   *  forge that refuses throws. */
  async apply(change: OwedChange): Promise<"written" | "dropped"> {
    let asked;
    try {
      asked = checkStatusOf(change);
    } catch (err) {
      this.say(`[gitea] ${CHECKER_OWED} ${change?.key} at revision ${change?.revision} writes nothing: ${err instanceof Error ? err.message : err}`);
      return "dropped";
    }
    const key = change.key;
    const seen = this.seen.get(key);
    if (seen === undefined || asked.revision > seen) this.remember(this.seen, key, asked.revision);
    // One key's writes take turns, so two at once are weighed one after the other.
    const before = this.turns.get(key) ?? Promise.resolve();
    const turn = before.then(async (): Promise<"written" | "dropped"> => {
      const newest = this.written.get(key);
      if (newest !== undefined && asked.revision < newest) {
        this.say(`[gitea] ${asked.owner}/${asked.repo}@${asked.sha.slice(0, 12)}: mesh/check at revision ${asked.revision} is older than ${newest}, already written, and is dropped`);
        return "dropped";
      }
      await this.forge.setCommitStatus(asked.owner, asked.repo, asked.sha, asked.status);
      this.remember(this.written, key, asked.revision);
      return "written";
    });
    const settled = turn.catch(() => {});
    this.turns.set(key, settled);
    void settled.then(() => { if (this.turns.get(key) === settled) this.turns.delete(key); });
    return turn;
  }

  /** What the watch hands over: written, or when the forge refuses, asked again on a bound in the background until it is
   *  written or a newer revision of the key was seen. Never throws: a watch's handler that throws loses the change. */
  async take(change: OwedChange): Promise<void> {
    try {
      await this.apply(change);
    } catch (err) {
      const run = this.askAgain(change, err);
      this.again.add(run);
      void run.finally(() => this.again.delete(run));
    }
  }

  /** Resolves when no ask again is running. */
  async settled(): Promise<void> {
    while (this.again.size) await Promise.all([...this.again]);
  }

  private async askAgain(change: OwedChange, first: unknown): Promise<void> {
    let err = first;
    for (let attempt = 1; ; attempt++) {
      const wait = bindWait(attempt);
      this.say(`[gitea] ${change.key}: mesh/check at revision ${change.revision} could not be written (${err instanceof Error ? err.message : err}); asked again in ${wait / 1000}s`);
      await this.sleep(wait);
      const newest = this.seen.get(change.key);
      if (newest !== undefined && newest > change.revision) return;
      try {
        await this.apply(change);
        return;
      } catch (e) {
        err = e;
      }
    }
  }

  private remember(map: Map<string, number>, key: string, revision: number): void {
    map.delete(key);
    map.set(key, revision);
    if (map.size > 4000) map.delete(map.keys().next().value as string);
  }
}

/**
 * Watch the checker's owed with `watch` (the SDK's `state(CHECKER_OWED).watch`), handing each change to `checks`. A watch
 * refused — the read not granted yet — is said once and asked again on a bound until it holds (bind.ts). Never rejects.
 */
export async function watchOwed(watch: (handler: (change: OwedChange) => Promise<void>) => Promise<void>, checks: CheckStatuses,
  sleep?: (ms: number) => Promise<void>, say: (line: string) => void = (l) => console.error(l)): Promise<void> {
  let refusals = 0;
  await bindUntilHeld(CHECKER_OWED, () => watch((change) => checks.take(change)), sleep, (line) => {
    if (/could not be bound/.test(line) && refusals++ > 0) return;
    say(line);
  });
}