520 lines
28 KiB
TypeScript
520 lines
28 KiB
TypeScript
// **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 — its kind (X-Gitea-Event) and its delivery id — so the polls know events are
|
|
* heard, and the module can say when it last heard one. */
|
|
heard?: (kind: string, delivery: string) => void;
|
|
/** Told of every refused delivery, with the reason and where it came from. */
|
|
refused?: (reason: string, from: string) => 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 refuse = (from: string, reason: string): void => {
|
|
log(`refused a forge event from ${from}: ${reason}`);
|
|
opts.refused?.(reason, from);
|
|
};
|
|
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) {
|
|
refuse(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) {
|
|
refuse(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;
|
|
refuse(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)) {
|
|
refuse(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 {
|
|
refuse(from, "it is signed but not JSON");
|
|
res.statusCode = 400;
|
|
return res.end("not JSON\n");
|
|
}
|
|
const delivery = String(req.headers["x-gitea-delivery"] ?? "");
|
|
const kind = String(req.headers["x-gitea-event"] ?? "other");
|
|
opts.heard?.(kind, 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(kind, 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 }>;
|
|
}
|
|
|
|
/** The forge's one system webhook as ensured at start: read back from the forge after it was made. */
|
|
export interface HookState {
|
|
/** Always "system": a hook the forge's listing of system webhooks names, so it fires for every repository. */
|
|
kind: "system";
|
|
id: number;
|
|
url: string;
|
|
events: string[];
|
|
active: boolean;
|
|
content_type?: string;
|
|
/** Made where none was, or made in place of the ones there (system, default, or a repository's) at the receiver's url. */
|
|
did: "created" | "replaced";
|
|
/** The hooks at the receiver's url it took the place of, by kind: a repository's is a copy of a default webhook, which
|
|
* would have sent that repository's events a second time. */
|
|
removed: { system: number; default: number; repository: number; repositories: string[] };
|
|
}
|
|
|
|
/** How long without a verified delivery — since the start, or since the last one — a forge is said silent once the poll
|
|
* finds what no event said. */
|
|
export const SILENT_AFTER = 10 * 60_000;
|
|
|
|
/**
|
|
* **How fast the polls look, and whether the forge's events are heard** (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.
|
|
*
|
|
* It also keeps what `gitea_events` says: where the receiver listens, the hook as ensured, the verified and refused
|
|
* deliveries, and the polls' finds no event said. **A forge whose events stop is said** (issue 729): once the receiver
|
|
* listens and the hook is ensured, when nothing verified for ten minutes — since the start, or since the last delivery —
|
|
* and the poll has found at least two heads or merges no event said in that quiet (one could be the poll beating an
|
|
* event by a second), it is logged loudly once per silent spell, and the module's events are said unhealthy until a
|
|
* delivery verifies. Deliveries refused since the last verified one are said unhealthy too.
|
|
*/
|
|
export class Pace {
|
|
private last = 0;
|
|
private missed = 0;
|
|
private readonly started: number;
|
|
private readonly now: () => number;
|
|
private readonly log: (l: string) => void;
|
|
private where: { host: string; port: number; url: string } | null = null;
|
|
private hook: HookState | null = null;
|
|
private notHeard: string | null = null;
|
|
private notHeardFails = false;
|
|
private verifiedCount = 0;
|
|
private lastVerified: { at: number; event: string; delivery: string } | null = null;
|
|
private refusedCount = 0;
|
|
private refusedSpell = 0;
|
|
private lastRefused: { at: number; reason: string; from: string } | null = null;
|
|
private missCount = 0;
|
|
private lastMiss: { at: number; what: string } | null = null;
|
|
private silentSince = 0;
|
|
private quietMisses = 0;
|
|
constructor(now: () => number = Date.now, log: (l: string) => void = (l) => console.error(`[gitea] ${l}`)) {
|
|
this.now = now;
|
|
this.log = log;
|
|
this.started = now();
|
|
}
|
|
/** The receiver listens, and the hook at it was ensured: from now on, deliveries are expected. */
|
|
listening(host: string, port: number, url: string, hook: HookState): void {
|
|
this.where = { host, port, url };
|
|
this.hook = hook;
|
|
this.notHeard = null;
|
|
}
|
|
/** No events are listened for, and why: by configuration (`fails` false), or because the receiver could not listen
|
|
* or the hook could not be ensured (`fails` true, said unhealthy). The polls alone watch. */
|
|
unheard(why: string, fails = true): void {
|
|
this.notHeard = why;
|
|
this.notHeardFails = fails;
|
|
}
|
|
heard(event = "", delivery = ""): void {
|
|
this.last = this.now();
|
|
this.verifiedCount++;
|
|
this.lastVerified = { at: this.last, event, delivery };
|
|
this.refusedSpell = 0;
|
|
this.quietMisses = 0;
|
|
if (this.silentSince) this.log(`the forge's events are heard again: ${event || "a delivery"} verified`);
|
|
this.silentSince = 0;
|
|
}
|
|
refused(reason: string, from: string): void {
|
|
this.refusedCount++;
|
|
this.refusedSpell++;
|
|
this.lastRefused = { at: this.now(), reason, from };
|
|
}
|
|
/** The poll found a head or merge no forge event announced. */
|
|
miss(what: string): void {
|
|
const t = this.now();
|
|
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 = t;
|
|
this.missCount++;
|
|
this.lastMiss = { at: t, what };
|
|
// Silent: nothing verified for SILENT_AFTER — never since the start, or not since the last delivery (a hook deleted by
|
|
// hand, every delivery refused) — while the poll finds what no event said. Said once per silent spell.
|
|
const quietSince = this.last || this.started;
|
|
this.quietMisses++;
|
|
if (this.expecting() && t - quietSince >= SILENT_AFTER && this.quietMisses >= 2 && !this.silentSince) {
|
|
this.silentSince = t;
|
|
const minutes = Math.round((t - quietSince) / 60_000);
|
|
const refusedSince = this.refusedSinceHeard();
|
|
this.log(`THE FORGE'S EVENTS ARE NOT HEARD: ${this.last ? `no delivery verified since the last at ` +
|
|
`${new Date(this.last).toISOString()}` : "no delivery ever verified since the start"}, ${minutes} min, from the ` +
|
|
`system webhook at ${this.hook?.url}${refusedSince ? `; ${refusedSince} refused since, the last because ` +
|
|
`${this.lastRefused?.reason}` : ""}; the poll found ${what} — the polls alone watch the forge, every 30 s ` +
|
|
`(novox/hq issue 729)`);
|
|
}
|
|
}
|
|
/** Whether a delivery was refused after the last verified one (or with none verified). */
|
|
private refusedSinceHeard(): number {
|
|
return this.lastRefused && this.lastRefused.at >= this.last ? this.refusedSpell : 0;
|
|
}
|
|
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;
|
|
}
|
|
private expecting(): boolean {
|
|
return this.where !== null && this.hook !== null;
|
|
}
|
|
/** What `gitea_events` answers: healthy, unhealthy or waits (ADR 0283's shape), with every fact behind it. */
|
|
report(): Record<string, unknown> {
|
|
const t = this.now();
|
|
const iso = (ms: number): string => new Date(ms).toISOString();
|
|
let state: "healthy" | "unhealthy" | "waits" = "healthy";
|
|
let why: string;
|
|
const refusedSince = this.refusedSinceHeard();
|
|
if (this.silentSince) {
|
|
state = "unhealthy";
|
|
why = `${this.last ? `no delivery verified since the last at ${iso(this.last)}` : `the forge has never delivered an ` +
|
|
`event since the start at ${iso(this.started)}`}, while the poll found ${this.missCount} head(s) or merge(s) no ` +
|
|
`event said, the last ${this.lastMiss?.what} at ${iso(this.lastMiss?.at ?? t)}`;
|
|
} else if (refusedSince) {
|
|
state = "unhealthy";
|
|
why = `${refusedSince} deliveries refused ${this.last ? `since the last verified one at ${iso(this.last)}` :
|
|
"and none verified"}, the last at ${iso(this.lastRefused?.at ?? t)} because ${this.lastRefused?.reason}`;
|
|
} else if (!this.expecting()) {
|
|
why = this.notHeard ? `no events are listened for, so the polls alone watch the forge: ${this.notHeard}` :
|
|
"the receiver is not listening yet";
|
|
if (this.notHeard && this.notHeardFails) state = "unhealthy";
|
|
} else if (this.hearing()) {
|
|
why = `the forge's events are heard: ${this.verifiedCount} verified, the last at ${iso(this.last)}`;
|
|
} else if (this.last === 0) {
|
|
why = `no delivery heard yet; the forge is said silent if the poll finds what no event said ${SILENT_AFTER / 60_000} min after the start`;
|
|
} else {
|
|
why = `the last delivery was at ${iso(this.last)}; the polls look every 30 s until the next`;
|
|
}
|
|
return {
|
|
healthy: state !== "unhealthy", state, why,
|
|
started: iso(this.started),
|
|
receiver: this.where ? { listening: true, ...this.where } : { listening: false, why: this.notHeard ?? "not started" },
|
|
hook: this.hook,
|
|
verified: { count: this.verifiedCount, last: this.lastVerified && { ...this.lastVerified, at: iso(this.lastVerified.at) } },
|
|
refused: { count: this.refusedCount, since_verified: refusedSince, last: this.lastRefused && { ...this.lastRefused, at: iso(this.lastRefused.at) } },
|
|
polls: { every_seconds: this.every() / 1000, hearing: this.hearing(), found_unsaid: this.missCount,
|
|
last_found: this.lastMiss && { ...this.lastMiss, at: iso(this.lastMiss.at) } },
|
|
silent: this.silentSince ? { since: iso(this.silentSince) } : null,
|
|
};
|
|
}
|
|
}
|
|
|
|
/** 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"];
|
|
|
|
async function hooksAt(admin: HookAdmin, type: "system" | "default", url: string): Promise<any[]> {
|
|
const listed = await admin.call("GET", `/admin/hooks?type=${type}&limit=50`);
|
|
if (listed.status !== 200) throw new Error(`the forge's ${type} webhooks could not be listed: ${listed.status}`);
|
|
return (Array.isArray(listed.body) ? listed.body : []).filter((h) => h?.config?.url === url);
|
|
}
|
|
|
|
/** Every page of a listing that answers an array (or `{ data: [] }`, the repository search), PAGE at a time. */
|
|
const PAGE = 50;
|
|
async function everyPage(admin: HookAdmin, path: string, what: string): Promise<any[]> {
|
|
const all: any[] = [];
|
|
for (let page = 1; ; page++) {
|
|
const r = await admin.call("GET", `${path}${path.includes("?") ? "&" : "?"}page=${page}&limit=${PAGE}`);
|
|
if (r.status !== 200) throw new Error(`${what} could not be listed: ${r.status}`);
|
|
const items: any[] = Array.isArray(r.body) ? r.body : Array.isArray(r.body?.data) ? r.body.data : [];
|
|
all.push(...items);
|
|
if (items.length < PAGE) return all;
|
|
}
|
|
}
|
|
|
|
/** Every repository's own webhooks at the receiver's url: copies of a default webhook made at it. */
|
|
async function repoHooksAt(admin: HookAdmin, url: string): Promise<{ repo: string; id: number }[]> {
|
|
const found: { repo: string; id: number }[] = [];
|
|
for (const repo of await everyPage(admin, "/repos/search?sort=id", "the forge's repositories")) {
|
|
const full = String(repo?.full_name ?? "");
|
|
if (!full.includes("/")) continue;
|
|
const hooks = await everyPage(admin, `/repos/${full}/hooks`, `${full}'s webhooks`);
|
|
for (const h of hooks) if (h?.config?.url === url) found.push({ repo: full, id: Number(h.id) });
|
|
}
|
|
return found;
|
|
}
|
|
|
|
/**
|
|
* **One system webhook, for every event, at the receiver** (issue 729), made each start and read back.
|
|
*
|
|
* The forge (1.27.3, routers/api/v1/utils/hook.go addHook) makes a *default* webhook from POST /admin/hooks unless
|
|
* `config.is_system_webhook` is "true": a default webhook fires for no repository and is only copied into repositories
|
|
* made after it, and the listing of system webhooks does not name it — so without the flag every start made another
|
|
* one and no event came. Its edit (editHook) never changes a webhook's secret nor its kind, so a hook already there is
|
|
* not corrected but replaced: the new system webhook is made first, with the secret kept now, so no event falls in a
|
|
* gap; then every other hook at the receiver's url is deleted — system, default, and each repository's copy of a
|
|
* default one, which would send that repository's events twice. The forge's listing of system webhooks must then name
|
|
* exactly one at the url, or this fails, saying so.
|
|
*/
|
|
export async function ensureSystemHook(admin: HookAdmin, url: string, secret: string): Promise<HookState> {
|
|
// 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, is_system_webhook: "true" } };
|
|
const before = new Set((await hooksAt(admin, "system", url)).map((h) => Number(h.id)));
|
|
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}`);
|
|
const fresh = (await hooksAt(admin, "system", url)).filter((h) => !before.has(Number(h.id)));
|
|
if (fresh.length !== 1) {
|
|
throw new Error(`the forge names ${fresh.length} new system webhooks at ${url} after one was made, not one: its events ` +
|
|
`are not known to be sent`);
|
|
}
|
|
const ours = fresh[0];
|
|
const defaults = await hooksAt(admin, "default", url);
|
|
const gone = async (path: string): Promise<void> => {
|
|
const r = await admin.call("DELETE", path);
|
|
if (r.status !== 204 && r.status !== 200 && r.status !== 404) throw new Error(`the forge's webhook ${path} could not be deleted: ${r.status}`);
|
|
};
|
|
for (const id of before) await gone(`/admin/hooks/${id}`);
|
|
for (const h of defaults) await gone(`/admin/hooks/${h.id}`);
|
|
const copies = await repoHooksAt(admin, url);
|
|
for (const c of copies) await gone(`/repos/${c.repo}/hooks/${c.id}`);
|
|
const kept = await hooksAt(admin, "system", url);
|
|
if (kept.length !== 1 || Number(kept[0].id) !== Number(ours.id)) {
|
|
throw new Error(`the forge names ${kept.length} system webhooks at ${url} once the others were deleted, not the one ` +
|
|
`made (${ours.id})`);
|
|
}
|
|
const removed = { system: before.size, default: defaults.length, repository: copies.length,
|
|
repositories: [...new Set(copies.map((c) => c.repo))] };
|
|
return { kind: "system", id: Number(ours.id), url, events: Array.isArray(ours.events) ? ours.events : [],
|
|
active: !!ours.active, content_type: ours.config?.content_type,
|
|
did: removed.system + removed.default + removed.repository > 0 ? "replaced" : "created", removed };
|
|
}
|
|
|
|
/** The `gitea_events` tool: what the module knows of the forge's events, from its Pace. */
|
|
export function eventsTool(pace: Pace): { name: string; description: string; input: Record<string, unknown>;
|
|
run: (args: Readonly<Record<string, unknown>>) => Promise<unknown> } {
|
|
return {
|
|
name: "gitea_events",
|
|
description: "Whether the forge's events reach the mesh (novox/hq issue 729): where the receiver listens, the " +
|
|
"system webhook as ensured at the start (its id, url, events, kind, and what it replaced), how many deliveries " +
|
|
"verified and the last one's time and event, how many were refused and the last reason, how often the polls look " +
|
|
"and whether events are heard, and how many heads or merges the poll found that no event said. Unhealthy when the " +
|
|
"forge has delivered nothing that verified for ten minutes (since the start or since the last delivery) while the " +
|
|
"poll found what no event said, when deliveries were refused since the last verified one, or when the receiver " +
|
|
"could not listen or the webhook could not be made.",
|
|
input: {},
|
|
run: async () => pace.report(),
|
|
};
|
|
}
|
|
|
|
/**
|
|
* **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;
|
|
}
|