mesh-catalog/modules/gitea/gate.ts

// **One writer of a head's mesh/merge-gate, and the verdict lands last** (novox/hq issue 446).
//
// A pending check says where it is: "waiting in the build queue since …" when the head is announced, "building on
// <machine> since …" when a build machine takes the check — the build seat's `started` (novox/hq ADR 0157) — and the
// verdict when the controller says `checked`. The forge keeps only the newest status of a context, so a pending word
// written after the verdict hides it: a pass reads pending and the merge waits for ever; a failure hides under
// "building". Two writers cannot order their writes, and a writer that reads the forge before writing races the other
// between its read and its write. So this module is the one writer of mesh/merge-gate, and every write of one head's
// gate goes through one chain: a verdict is marked the moment it is heard, before it waits its turn, and a "building"
// still waiting behind or taking its turn after that mark is dropped, never written.
//
// **A recheck's pending is written here too** (novox/hq issue 456). mesh-delivery's recheck puts the gate back to
// pending through the gitea_commit_status tool, which this module serves from the watcher's own bundle so that it
// reaches this writer: the pending clears the verdict mark, so the new run's "building" is said, and queues behind an
// old verdict in flight, so that verdict lands first. A verdict is weighed by the ask it answers — the controller's
// checked.id, `build-<unix nanoseconds>` of when the check was asked — against when the check now pending was opened:
// an older run's verdict, redelivered after its write failed, is dropped, never written over the pending. After a
// restart, when the check was opened is read from the forge's own history of the gate.

import { CHECK_CONTEXT, pendingWords } from "./pulls.js";
import type { CommitStatus, HeadCheck } from "./client.js";

/** The build seat's `started`, as mesh-controller's link.BuildStart publishes it. */
export interface BuildStarted {
  id?: string;
  repository?: string;
  path?: string;
  ref?: string;
  on?: string;
  at?: string;
}

/** What the gate writer needs of the forge: a status set, and a head's gate as the forge holds it. */
export interface GateForge {
  setCommitStatus(owner: string, repo: string, sha: string, status: CommitStatus): Promise<void>;
  /** The head's mesh/merge-gate, newest: its state and when it was set; undefined when it has none. */
  gateOf(owner: string, repo: string, sha: string): Promise<{ state: string; updated_at?: string } | undefined>;
  /** The head's mesh/merge-gate statuses, every one the forge keeps, newest first (novox/hq issue 456). */
  gateHistory(owner: string, repo: string, sha: string): Promise<GateStatus[]>;
  /** The head's statuses by context, the newest of each, with their words (novox/hq ADR 0323). */
  headChecks(owner: string, repo: string, sha: string): Promise<Record<string, HeadCheck>>;
}

/** One status of a head's gate as the forge keeps it: its state, its words, and when it was set. */
export interface GateStatus {
  state: string;
  description?: string;
  updated_at?: string;
}

/**
 * When a check was asked, from its id as mesh-controller's link.NewBuildID writes it: `build-<unix nanoseconds>`.
 * Undefined for an id of any other shape, whose ask time is not known (novox/hq issue 456).
 */
export function askedAt(id: string | undefined): number | undefined {
  const m = /^build-(\d{1,20})$/.exec(String(id ?? ""));
  if (!m) return undefined;
  const ms = Number(BigInt(m[1]) / 1_000_000n);
  return ms > 0 ? ms : undefined;
}

/**
 * When the check now pending on a head was opened, from the gate's history, newest first: the newest status, past the
 * "building" words this writer says only inside an open check, when it is pending; undefined when the gate holds a
 * verdict, holds nothing, or its time cannot be read (novox/hq issue 456).
 */
export function openedAt(history: GateStatus[]): number | undefined {
  for (const s of history) {
    if (s.state !== "pending") return undefined;
    if ((s.description ?? "").startsWith(BUILDING)) continue;
    const t = Date.parse(s.updated_at ?? "");
    return Number.isNaN(t) ? undefined : t;
  }
  return undefined;
}

/** How far a build machine's clock may stand behind this one's when a start is weighed against the pending words. */
const CLOCK_SLACK_MS = 60_000;

/** How the words of a check a machine is running begin. */
const BUILDING = "building on ";

/** The words of a check a machine is running, in the time the operator reads. */
export function buildingWords(on: string, at: Date): string {
  const two = (n: number) => String(n).padStart(2, "0");
  return `${BUILDING}${on} since ${two(at.getHours())}:${two(at.getMinutes())}`;
}

/** Whether a repository as the mesh clones it (`http://forge/novox/hq.git`) is owner/repo. */
export function clonedAs(url: string | undefined, owner: string, repo: string): boolean {
  const u = (url ?? "").trim().toLowerCase().replace(/\.git$/, "");
  const full = `${owner}/${repo}`.toLowerCase();
  return u === full || u.endsWith("/" + full);
}

interface Head {
  /** When the pending words were last set by this writer; undefined for a head this process did not announce. */
  pendingAt?: number;
  /** A verdict was heard for the head: nothing pending is written after it. */
  verdict: boolean;
  /** Which verdict set the mark: one dropped in its turn takes the mark back only while it is still its own. */
  mark?: object;
  /** The chain the head's writes take their turns on. */
  chain: Promise<void>;
}

export class GateWriter {
  private heads = new Map<string, Head>();
  private forge: GateForge;
  private now: () => Date;

  constructor(forge: GateForge, now: () => Date = () => new Date()) {
    this.forge = forge;
    this.now = now;
  }

  private head(owner: string, repo: string, sha: string): Head {
    const key = `${owner}/${repo}@${sha}`.toLowerCase();
    let h = this.heads.get(key);
    if (!h) {
      h = { verdict: false, chain: Promise.resolve() };
      this.heads.set(key, h);
      // Kept for the heads still in play; the oldest go first, past what any forge holds open at once.
      if (this.heads.size > 2000) this.heads.delete(this.heads.keys().next().value as string);
    }
    return h;
  }

  /** One turn on the head's chain; a turn that fails never stops the next. Its failure is the caller's. */
  private turn(h: Head, fn: () => Promise<void>): Promise<void> {
    const run = h.chain.then(fn);
    h.chain = run.catch(() => {});
    return run;
  }

  /** A turn whose failure is said here: a pending word not written is not worth the event again. */
  private said(h: Head, what: string, fn: () => Promise<void>): Promise<void> {
    return this.turn(h, fn).catch((err) => {
      console.error(`[gitea] ${what} — ${err instanceof Error ? err.message : err}`);
    });
  }

  /** The head announced: it waits in the build queue from now. A new check of the head, so no verdict yet. */
  pending(owner: string, repo: string, sha: string): Promise<void> {
    const h = this.head(owner, repo, sha);
    const at = this.now();
    this.opened(h, at);
    return this.said(h, `${owner}/${repo}@${sha.slice(0, 8)}: could not say a check is pending`, () =>
      this.forge.setCommitStatus(owner, repo, sha, { state: "pending", context: CHECK_CONTEXT, description: pendingWords(at) }));
  }

  /** A new check of the head opened now: no verdict heard for it yet, and nothing older than it is its own. */
  private opened(h: Head, at: Date): void {
    h.pendingAt = at.getTime();
    h.verdict = false;
    h.mark = undefined;
  }

  /**
   * A recheck's pending, asked through the gitea_commit_status tool (novox/hq issue 456): on mesh/merge-gate it opens
   * a new check of the head — the verdict mark cleared, so the new run's "building" is said — and takes its turn
   * after every write already asked for, an old verdict in flight included, so it lands over that verdict. Any
   * other status of the merge check (mesh/repo-check, which the verdict also writes) takes its turn on the same
   * chain. A write that fails fails the caller, and the head is as it was before it.
   */
  rechecked(owner: string, repo: string, sha: string, status: CommitStatus): Promise<void> {
    const h = this.head(owner, repo, sha);
    const write = () => this.forge.setCommitStatus(owner, repo, sha, status);
    if (status.context !== CHECK_CONTEXT || status.state !== "pending") return this.turn(h, write);
    const before = { pendingAt: h.pendingAt, verdict: h.verdict, mark: h.mark };
    const at = this.now();
    this.opened(h, at);
    return this.turn(h, write).catch((err) => {
      if (h.pendingAt === at.getTime() && !h.verdict) Object.assign(h, before);
      throw err;
    });
  }

  /**
   * **A trunk's move puts the head's verdict out of date** (novox/hq ADR 0323 §4): its merge check statuses are put back
   * to pending, with the reason, in their turn after every write of the head already asked for — a verdict in flight
   * lands first. What is written is decided in the turn, from the statuses the forge holds then (`decide`, trunk.ts's
   * outOfDateStatuses): a check on its way, or a verdict already judged against the trunk's head, is left as it is. It
   * opens no check — nothing is asked — so a verdict of a check asked before it keeps its weight. Answers what it wrote;
   * a write that fails fails the caller, and the move is weighed again on the next pass.
   */
  outOfDate(owner: string, repo: string, sha: string,
    decide: (checks: Record<string, HeadCheck>) => CommitStatus[] | Promise<CommitStatus[]>, target?: string): Promise<CommitStatus[]> {
    const h = this.head(owner, repo, sha);
    const written: CommitStatus[] = [];
    return this.turn(h, async () => {
      for (const s of await decide(await this.forge.headChecks(owner, repo, sha))) {
        await this.forge.setCommitStatus(owner, repo, sha, target ? { ...s, target_url: target } : s);
        written.push(s);
      }
    }).then(() => written);
  }

  /**
   * A build machine took an ask of this head: said while no verdict was heard, and only for the check now pending —
   * a start from before the pending words were set (a run before a recheck, the stream's history) is not this check's.
   * A head this process did not announce (it restarted since) is weighed against the forge's own gate: said only while
   * that is pending, and not for a start older than it.
   */
  started(owner: string, repo: string, sha: string, b: BuildStarted): Promise<boolean> {
    const at = b.at ? new Date(b.at) : new Date(NaN);
    if (!b.on || Number.isNaN(at.getTime())) return Promise.resolve(false);
    const h = this.head(owner, repo, sha);
    let said = false;
    return this.said(h, `${owner}/${repo}@${sha.slice(0, 8)}: could not say where its check is`, async () => {
      if (h.verdict) return;
      let since = h.pendingAt;
      if (since === undefined) {
        const gate = await this.forge.gateOf(owner, repo, sha);
        if (gate && gate.state !== "pending") return;
        since = gate?.updated_at ? Date.parse(gate.updated_at) : undefined;
      }
      if (since !== undefined && at.getTime() < since - CLOCK_SLACK_MS) return;
      if (h.verdict) return; // heard while the forge was read
      await this.forge.setCommitStatus(owner, repo, sha, { state: "pending", context: CHECK_CONTEXT,
        description: buildingWords(b.on as string, at) });
      said = true;
    }).then(() => said);
  }

  /**
   * The verdict: marked at once, so no pending word is written after it, then written in its turn — last. A write
   * that fails fails the caller, so the event is offered again.
   *
   * `id` is the ask it answers (checked.id): a verdict of a check asked before the check now pending was opened is
   * an older run's — redelivered after its write failed, or heard late — and is dropped, never written over the
   * pending (novox/hq issue 456). When this process opened no check of the head (it restarted since), the opening
   * is read from the forge's history of the gate in the verdict's turn; a history that cannot be read writes the
   * verdict, as one with no readable id is: a verdict is never lost to what cannot be weighed.
   */
  verdict(owner: string, repo: string, sha: string, write: () => Promise<void>, id?: string): Promise<void> {
    const h = this.head(owner, repo, sha);
    const asked = askedAt(id);
    const head = `${owner}/${repo}@${sha.slice(0, 8)}`;
    if (asked !== undefined && h.pendingAt !== undefined && older(asked, h.pendingAt)) {
      console.log(`[gitea] ${head}: the verdict of ${id} is an older run's than the check now pending; not written`);
      return Promise.resolve();
    }
    const before = h.verdict;
    const mark = {};
    h.verdict = true;
    h.mark = mark;
    return this.turn(h, async () => {
      if (asked !== undefined && h.pendingAt === undefined) {
        let opened: number | undefined;
        try {
          opened = openedAt(await this.forge.gateHistory(owner, repo, sha));
        } catch (err) {
          console.error(`[gitea] ${head}: the gate's history could not be read, so the verdict of ${id} is written ` +
            `unweighed — ${err instanceof Error ? err.message : err}`);
        }
        if (opened !== undefined && older(asked, opened)) {
          if (h.mark === mark) h.verdict = before;
          console.log(`[gitea] ${head}: the verdict of ${id} is an older run's than the check the forge holds pending; not written`);
          return;
        }
      }
      await write();
    });
  }
}

/** Whether a check asked at `asked` was asked before a check opened at `opened`, past what the clocks may differ by. */
function older(asked: number, opened: number): boolean {
  return asked < opened - CLOCK_SLACK_MS;
}