mesh-catalog/modules/gitea/webhook.ts

// **The forge says every event on the bus** (novox/hq issue 729; to-be 10: an event makes it fast, nothing makes it
// necessary). The module keeps one system webhook on its forge for every event, pointing at its own receiver, signed
// with a secret it mints and keeps beside its token; the receiver refuses whatever is unsigned or does not verify, and
// publishes each event, trimmed, on the module's events as forge.<kind>.<action> (forge.pull-request.opened, forge.push, …),
// on the shared EVENTS path its other events take; its own stream (ADR 0363) is a follow-up once issue 725 ships.
// The polls stay as the safety net, slowed; the announced-heads and announced-merges records keep anything from
// being announced twice, whichever path saw it first.
//
// The secret is never logged: lines say it was minted or reused and where it is kept, never what it is.

import { createHmac, randomBytes, timingSafeEqual } from "node:crypto";
import { chmodSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
import { createServer, type Server } from "node:http";
import { dirname } from "node:path";

/** The name the receiver answers on, and the port it listens on unless the manifest says otherwise. */
export const HOOK_PATH = "/forge-event";
export const HOOK_PORT = 3047;

/** The secret the forge signs with: read from its 0600 file, or minted and kept there. */
export function keptSecret(file: string, log: (line: string) => void = (l) => console.log(`[gitea] ${l}`)): string {
  try {
    const kept = readFileSync(file, "utf8").trim();
    if (kept.length >= 32) {
      log(`reusing the webhook secret kept at ${file}`);
      return kept;
    }
  } catch (err) {
    if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err;
  }
  const secret = randomBytes(32).toString("hex");
  mkdirSync(dirname(file), { recursive: true, mode: 0o700 });
  const tmp = `${file}.tmp`;
  writeFileSync(tmp, secret + "\n", { mode: 0o600 });
  chmodSync(tmp, 0o600);
  renameSync(tmp, file);
  log(`minted a webhook secret, kept at ${file}`);
  return secret;
}

/** Whether a body carries the forge's HMAC-SHA256 signature (X-Gitea-Signature, hex) under the secret. */
export function verified(secret: string, body: Buffer | string, signature: string | undefined): boolean {
  if (!secret || !signature || !/^[0-9a-f]{64}$/i.test(signature.trim())) return false;
  const want = createHmac("sha256", secret).update(body).digest();
  const got = Buffer.from(signature.trim(), "hex");
  return got.length === want.length && timingSafeEqual(got, want);
}

/** One forge event as the bus carries it: its name inside the stream, and what is needed of it — no secrets. */
export interface ForgeEvent {
  name: string;
  body: Record<string, unknown>;
}

/** Every event name the module emits for the forge, without its forge. prefix (module.json `emits`). */
export const EVENTS: readonly string[] = ["push", "create", "delete", "fork", "status", "pull-request.opened", "pull-request.reopened", "pull-request.synchronized", "pull-request.edited", "pull-request.closed", "pull-request.merged", "pull-request.labeled", "pull-request.unlabeled", "pull-request.assigned", "pull-request.unassigned", "pull-request.reviewed", "pull-request.review-requested", "pull-request.commented", "pull-request.changed", "issue-comment.created", "issue-comment.edited", "issue-comment.deleted", "issues.opened", "issues.reopened", "issues.closed", "issues.edited", "issues.labeled", "issues.unlabeled", "issues.assigned", "issues.unassigned", "issues.changed", "release.published", "release.updated", "release.deleted", "release.changed", "repository.created", "repository.deleted", "repository.changed", "wiki.changed", "other"];

/** The families the module's stream carries (module.json `streams`); any other kind is said as other.<kind>; underscores become hyphens. */
export const FAMILIES = ["push", "create", "delete", "fork", "status", "pull_request", "issue_comment", "issues",
  "release", "repository", "wiki"] as const;

const ACTIONS: Record<string, string> = { label_updated: "labeled", label_cleared: "unlabeled", synchronized: "synchronized" };

function str(v: unknown): string | undefined {
  return typeof v === "string" && v ? v : undefined;
}
function num(v: unknown): number | undefined {
  return typeof v === "number" ? v : undefined;
}
function login(u: any): string | undefined {
  return str(u?.login) ?? str(u?.username);
}

/**
 * The event a forge delivery is: its kind (X-Gitea-Event) and the payload's action, trimmed to ids, repository, refs,
 * shas, author, labels and a comment's body (a ready mark is one, ADR 0364). A merge is pull-request.merged, a comment on
 * a pull request pull-request.commented.
 */
export function eventOf(kind: string, payload: any): ForgeEvent {
  // The bus's event names take lower-case letters, digits and hyphens: the forge's underscores become hyphens.
  const event = forgeEventOf(kind, payload);
  const name = event.name.toLowerCase().replace(/_/g, "-").replace(/[^a-z0-9.-]/g, "-");
  // Published on the module's events as it publishes its others, each name declared in its `emits`: one the manifest
  // does not name is forge.other, saying what the forge called it. Its own stream (ADR 0363) follows issue 725.
  if (EVENTS.includes(name)) return { name: `forge.${name}`, body: event.body };
  return { name: "forge.other", body: { ...event.body, kind, said: name } };
}

function forgeEventOf(said: string, payload: any): ForgeEvent {
  // The forge names some kinds finer than their family (pull_request_label, pull_request_sync): read by the payload.
  let kind = said;
  if (said.startsWith("pull_request") && payload?.pull_request) kind = "pull_request";
  if ((said.startsWith("issue") || said === "pull_request_comment") && payload?.comment) kind = "issue_comment";
  if (said.startsWith("issue") && !payload?.comment && payload?.issue) kind = "issues";
  const repo = payload?.repository ?? {};
  const base: Record<string, unknown> = {
    owner: login(repo.owner), repo: str(repo.name), full_name: str(repo.full_name), clone_url: str(repo.clone_url),
    sender: login(payload?.sender),
  };
  const family = (FAMILIES as readonly string[]).includes(kind) ? kind : `other`;
  let action = str(payload?.action);
  const pull = payload?.pull_request;
  if (kind === "push") {
    return { name: "push", body: { ...base, ref: str(payload.ref), before: str(payload.before), after: str(payload.after),
      commits: Array.isArray(payload.commits) ? payload.commits.length : undefined, pusher: login(payload.pusher) } };
  }
  if (kind === "pull_request" && pull) {
    if (action === "closed" && pull.merged) action = "merged";
    action = ACTIONS[action ?? ""] ?? action ?? "changed";
    return { name: `pull_request.${action}`, body: { ...base, number: num(pull.number ?? payload.number),
      title: str(pull.title), author: login(pull.user), head: str(pull.head?.ref), head_sha: str(pull.head?.sha),
      base: str(pull.base?.ref), merged: !!pull.merged, merge_commit_sha: str(pull.merge_commit_sha),
      merged_at: str(pull.merged_at), html_url: str(pull.html_url),
      labels: Array.isArray(pull.labels) ? pull.labels.map((l: any) => str(l?.name)).filter(Boolean) : undefined,
      label: str(payload.label?.name) } };
  }
  if (kind === "issue_comment") {
    const issue = payload.issue ?? {};
    const onPull = !!payload.is_pull || !!issue.pull_request;
    return { name: onPull ? "pull_request.commented" : `issue_comment.${action ?? "created"}`, body: { ...base,
      number: num(issue.number), comment_id: num(payload.comment?.id), author: login(payload.comment?.user),
      comment: str(payload.comment?.body), action } };
  }
  if (kind === "create" || kind === "delete") {
    return { name: kind, body: { ...base, ref: str(payload.ref), ref_type: str(payload.ref_type), sha: str(payload.sha) } };
  }
  if (kind === "status") {
    return { name: "status", body: { ...base, sha: str(payload.sha), context: str(payload.context), state: str(payload.state) } };
  }
  const name = family === "other" ? `other.${kind.replace(/[^a-z0-9_-]/gi, "_") || "unknown"}` :
    ["fork"].includes(family) ? family : `${family}.${action ?? "changed"}`;
  return { name, body: { ...base, number: num(payload?.number ?? payload?.issue?.number), action } };
}

export interface ReceiverOptions {
  secret: string;
  port: number;
  host?: string;
  /** Publishes one event on the bus. */
  publish: (event: ForgeEvent) => Promise<void>;
  /** Acts on it at once, beside publishing it (the announcer). */
  act?: (event: ForgeEvent) => Promise<void>;
  /** Told of every verified delivery, so the polls know events are heard. */
  heard?: () => void;
  log?: (line: string) => void;
}

/** The most a delivery may carry: the forge's push of many commits stays well under it. */
export const MAX_BODY = 5 * 1024 * 1024;
/** How many deliveries' ids are remembered, so a delivery sent again is published once. */
export const REMEMBERED = 2000;

/**
 * The receiver. A request without a signature is refused before its body is read, a body over MAX_BODY is refused, and
 * one whose signature does not verify is refused, each with 401 or 413 and a line saying so. A verified delivery is
 * answered 200 at once and queued: it is published and acted on after the reply, one at a time, so the forge's
 * delivery timeout is never reached. A delivery whose X-Gitea-Delivery id was already heard is answered and dropped.
 */
export function receiver(opts: ReceiverOptions): Server & { idle(): Promise<void> } {
  const log = opts.log ?? ((l) => console.log(`[gitea] ${l}`));
  const seen = new Set<string>();
  let queue: Promise<void> = Promise.resolve();
  const server = createServer((req, res) => {
    if (req.method !== "POST" || (req.url ?? "").split("?")[0] !== HOOK_PATH) {
      res.statusCode = 404;
      return res.end();
    }
    const from = req.socket.remoteAddress ?? "?";
    const signature = String(req.headers["x-gitea-signature"] ?? "").trim();
    if (!signature) {
      log(`refused a forge event from ${from}: it is unsigned`);
      res.writeHead(401, { Connection: "close" });
      res.end("unsigned\n");
      req.destroy();
      return;
    }
    if (Number(req.headers["content-length"] ?? 0) > MAX_BODY) {
      log(`refused a forge event from ${from}: its body is over ${MAX_BODY} bytes`);
      res.writeHead(413, { Connection: "close" });
      res.end("too large\n");
      req.destroy();
      return;
    }
    const chunks: Buffer[] = [];
    let size = 0;
    let over = false;
    req.on("data", (c: Buffer) => {
      size += c.length;
      if (size > MAX_BODY) {
        if (!over) {
          over = true;
          log(`refused a forge event from ${from}: its body is over ${MAX_BODY} bytes`);
          res.writeHead(413, { Connection: "close" });
          res.end("too large\n");
          req.destroy();
        }
      } else chunks.push(c);
    });
    req.on("end", () => {
      if (over) return;
      const body = Buffer.concat(chunks);
      if (!verified(opts.secret, body, signature)) {
        log(`refused a forge event from ${from}: its signature does not verify`);
        res.statusCode = 401;
        return res.end("signature does not verify\n");
      }
      let payload: unknown;
      try {
        payload = JSON.parse(body.toString("utf8"));
      } catch {
        res.statusCode = 400;
        return res.end("not JSON\n");
      }
      opts.heard?.();
      const delivery = String(req.headers["x-gitea-delivery"] ?? "");
      if (delivery && seen.has(delivery)) {
        res.statusCode = 200;
        return res.end("already heard\n");
      }
      if (delivery) {
        seen.add(delivery);
        if (seen.size > REMEMBERED) seen.delete(seen.values().next().value as string);
      }
      const event = eventOf(String(req.headers["x-gitea-event"] ?? "other"), payload);
      res.statusCode = 200;
      res.end("queued\n");
      queue = queue.then(async () => {
        try {
          await opts.publish(event);
          if (opts.act) await opts.act(event);
        } catch (err) {
          log(`a forge event ${event.name} could not be published or acted on; the polls catch it — ${err instanceof Error ? err.message : err}`);
        }
      });
    });
  });
  return Object.assign(server, { idle: () => queue });
}

/** What ensuring the system webhook needs of the forge's administration (GiteaAdmin's login). */
export interface HookAdmin {
  call(method: string, path: string, body?: unknown): Promise<{ status: number; body: any }>;
}

/**
 * **How fast the polls look** (issue 729): every five minutes only while the forge's events are heard — a verified
 * delivery within the last ten minutes, and none missed since. Until the first delivery, after ten minutes without one,
 * or once the poll found something no event said, every thirty seconds.
 */
export class Pace {
  private last = 0;
  private missed = 0;
  private readonly now: () => number;
  private readonly log: (l: string) => void;
  constructor(now: () => number = Date.now, log: (l: string) => void = (l) => console.error(`[gitea] ${l}`)) {
    this.now = now;
    this.log = log;
  }
  heard(): void {
    this.last = this.now();
  }
  /** The poll found a head or merge no forge event announced. */
  miss(what: string): void {
    if (this.hearing()) this.log(`THE FORGE'S EVENTS MISSED ${what}: the poll found it, so the polls look every 30 s again until the next event`);
    this.missed = this.now();
  }
  hearing(): boolean {
    const t = this.now();
    return this.last > 0 && t - this.last < 10 * 60_000 && this.last > this.missed;
  }
  every(): number {
    return this.hearing() ? 5 * 60_000 : 30_000;
  }
}

/** Every event family the forge (1.27) can send a webhook. */
export const HOOK_EVENTS = ["create", "delete", "fork", "push", "issues", "issue_comment", "pull_request",
  "pull_request_review_request", "wiki", "repository", "release", "package", "status", "workflow_run", "workflow_job"];

/**
 * **One system webhook, for every event, at the receiver** (issue 729): created when none points there, corrected
 * when one does but differs (events, active, content type), and its secret set again each start — the forge never
 * hands a secret back, so a corrected secret cannot be told from a kept one. Answers what it did.
 */
export async function ensureSystemHook(admin: HookAdmin, url: string, secret: string): Promise<"created" | "corrected"> {
  // POST /admin/hooks makes a system webhook, and its options take no flag for it (the forge's 1.27.3 API schema). The
  // forge has no "every event" word: each family is named, and a family name (issues, pull_request) takes all of it.
  const want = { type: "gitea", active: true, events: HOOK_EVENTS, branch_filter: "*",
    config: { url, content_type: "json", secret } };
  const listed = await admin.call("GET", "/admin/hooks?type=system&limit=50");
  if (listed.status !== 200) throw new Error(`the forge's system webhooks could not be listed: ${listed.status}`);
  const hooks: any[] = Array.isArray(listed.body) ? listed.body : [];
  const ours = hooks.filter((h) => h?.config?.url === url);
  if (ours.length === 0) {
    const made = await admin.call("POST", "/admin/hooks", want);
    if (made.status !== 201 && made.status !== 200) throw new Error(`the forge's system webhook could not be made: ${made.status}`);
    return "created";
  }
  for (const extra of ours.slice(1)) await admin.call("DELETE", `/admin/hooks/${extra.id}`);
  const fixed = await admin.call("PATCH", `/admin/hooks/${ours[0].id}`, { active: true, events: HOOK_EVENTS, branch_filter: "*",
    config: { url, content_type: "json", secret } });
  if (fixed.status !== 200) throw new Error(`the forge's system webhook could not be corrected: ${fixed.status}`);
  return "corrected";
}

/**
 * **Claim a head for announcing, once** (issue 729): true when `sha` is not already the head recorded for `key`, which
 * it then is. The poll and the forge's event both claim before they announce, so whichever comes second finds it taken.
 */
export function claimHead(heads: Record<string, string>, key: string, sha: string): boolean {
  if (!sha || heads[key] === sha) return false;
  heads[key] = sha;
  return true;
}