a10db84 · 2195926 · 5bc95e5 · de14d7f · ee70458 · fdcfa46
@@ -0,0 +1,12 @@
# A failing webhook is said once
When a webhook's third delivery in a row ends `failed`, one `webhook.failing`
event goes on the bus (`src/core/webhook_streak.ts`, called from the delivery
worker in `src/core/webhooks.ts`). Once per streak: a success ends it, and a
fourth failure or a redelivery that fails again adds nothing.
Writers of the repository see one line in their feed — "Webhook to
example.com failed 3 times: answered HTTP 500" — linking to the repository, and naming Settings ›
Webhooks, where redeliver and pause are (`webhook-redeliver.md`,
`webhook-pause.md`). Readers do not: the event's audience is `writers`
(`src/core/event_audience.ts`). It names the host, never the address.
@@ -181,6 +181,9 @@
const READER_KINDS: ReadonlySet<string> = new Set([
 * `test/event-audience-coverage.test.ts` fails on an emitted kind that is.
 */
export function classifiedAudience(kind: string): EventAudience | undefined {
  // A hook that keeps failing is for those who can act on it: the host and
  // the reason, never the address (`webhook_streak.ts`).
  if (kind === "webhook.failing") return "writers";
  if (
    kind.startsWith("webhook.") ||
    kind.startsWith("app.env_") ||
@@ -298,7 +301,11 @@
export function audienceFilter(
    const known = writers.get(repoId);
    if (known !== undefined) return known;
    const repo = getRepoById(db, repoId);
    const yes = repo !== undefined && canWrite(db, repo, viewer);
    const yes =
      repo !== undefined &&
      viewer !== undefined &&
      isActive(viewer) &&
      canWrite(db, repo, viewer);
    writers.set(repoId, yes);
    return yes;
  };
@@ -0,0 +1,77 @@
/**
 * A webhook that keeps failing, said once.
 *
 * Redeliver and pause exist, but nobody is told to use them. When a hook's
 * third delivery in a row ends `failed`, one `webhook.failing` event goes on
 * the bus for the repository's writers: once per streak, not per failure. A
 * delivery that ends `ok` ends the streak; the next three failures are a new
 * one. The streak is named by its oldest failed delivery (`since`), so a
 * redelivery that fails again, or a fourth failure, finds the event already
 * written and adds nothing.
 */
import type { Database } from "bun:sqlite";
import type { EventBus } from "./events.ts";
/** Failed deliveries in a row that make a streak worth saying. */
export const STREAK = 3;
/** The longest reason the event carries. */
const REASON_MAX = 200;
/** The host a hook posts to; the whole address (often a secret) is not for writers. */
export function hostOf(url: string): string {
  try {
    return new URL(url).hostname;
  } catch {
    return "an unreadable address";
  }
}
/** Call after a delivery of this hook has ended `failed`. */
export function noteFailure(
  db: Database,
  bus: EventBus,
  webhookId: string,
  reason: string,
): void {
  const hook = db
    .query<{ repo_id: string; url: string }, [string]>(
      "SELECT repo_id, url FROM webhook WHERE id = ?1",
    )
    .get(webhookId);
  if (hook === null) return;
  // The streak is every failure since the newest success.
  const failed = db
    .query<{ id: string }, [string]>(
      `SELECT id FROM webhook_delivery
        WHERE webhook_id = ?1 AND state = 'failed'
          AND id > COALESCE((SELECT MAX(id) FROM webhook_delivery
                              WHERE webhook_id = ?1 AND state = 'ok'), '')
        ORDER BY id`,
    )
    .all(webhookId);
  if (failed.length < STREAK) return;
  // Its oldest failure names it, however long it grows.
  const since = failed[0]!.id;
  const said = db
    .query(
      `SELECT 1 FROM event
        WHERE kind = 'webhook.failing' AND repo_id = ?1
          AND json_extract(payload, '$.webhookId') = ?2
          AND json_extract(payload, '$.since') = ?3 LIMIT 1`,
    )
    .get(hook.repo_id, webhookId, since);
  if (said !== null) return;
  bus.append({
    kind: "webhook.failing",
    repoId: hook.repo_id,
    payload: {
      webhookId,
      since,
      host: hostOf(hook.url),
      reason: reason.slice(0, REASON_MAX),
      failures: STREAK,
    },
  });
}
@@ -19,7 +19,8 @@
import {
  type OutboundOptions,
} from "../kernel/outbound.ts";
import { err, ok, type Result } from "../kernel/result.ts";
import type { ForgeEvent } from "./events.ts";
import { EventBus, type ForgeEvent } from "./events.ts";
import { noteFailure } from "./webhook_streak.ts";
const MAX_ATTEMPTS = 6;
const DELIVER_TIMEOUT_MS = 10_000;
@@ -167,6 +168,9 @@
export function kindMatches(spec: string, kind: string): boolean {
 */
export function enqueueForEvent(db: Database, event: ForgeEvent): number {
  if (event.repoId === null) return 0;
  // The news that a hook is failing is for writers, not for hooks: sent to the
  // failing one it would be one more doomed delivery in the streak it reports.
  if (event.kind === "webhook.failing") return 0;
  // A hook's URL is somewhere else: nothing about a sealed report goes
  // there, not even that it moved (`sealed.ts`).
  if (eventSealed(db, event)) return 0;
@@ -300,7 +304,6 @@
interface DueRow {
  created_at: string;
}
/** Deliver all due pending deliveries. Returns how many were attempted. */
/**
 * How long a claimed delivery is its sender's before another may try it:
 * well past the longest one send may take, so a claim never lapses while
@@ -309,11 +312,18 @@
interface DueRow {
export const CLAIM_MS = 60_000;
export { DELIVER_TIMEOUT_MS };
/**
 * Deliver all due pending deliveries. Returns how many were attempted.
 * `bus` is optional only for one-shot callers (tests, scripts) with no live
 * listeners: it then writes a bus of its own, which stores the event and wakes
 * no one. The worker (`runDelivery`) always passes the forge's bus.
 */
export async function deliverDue(
  db: Database,
  log: Log,
  now = new Date(),
  outbound: OutboundOptions = {},
  bus: EventBus = new EventBus(db),
): Promise<number> {
  const nowIso = now.toISOString();
  const events = db
@@ -365,7 +375,7 @@
export async function deliverDue(
      );
    if (claimed.changes !== 1) continue;
    try {
      await deliverOne(db, log, row, now, outbound);
      await deliverOne(db, log, row, now, outbound, bus);
    } catch (error) {
      // One malformed delivery must not take the batch with it. Without
      // this, a single unsendable row stops every delivery behind it — and
@@ -377,7 +387,7 @@
export async function deliverDue(
        delivery: row.id,
        error: String(error),
      });
      backOff(db, log, row, 0, now);
      backOff(db, log, row, 0, now, bus);
    }
  }
  return due.length;
@@ -394,6 +404,7 @@
async function deliverOne(
  row: DueRow,
  now: Date,
  outbound: OutboundOptions,
  bus: EventBus,
): Promise<void> {
  // A ping's body is its payload as it stands: { kind, repo, hook, at }.
  const body =
@@ -440,6 +451,7 @@
async function deliverOne(
      db.query(
        `UPDATE ${tableOf(row)} SET state = 'failed', attempts = ?1, last_status = NULL, last_error = ?2 WHERE id = ?3`,
      ).run(attempt, error.message, row.id);
      if (!row.ping) noteFailure(db, bus, row.webhook_id, error.message);
      return;
    }
    log.warn("webhook delivery error", {
@@ -455,7 +467,7 @@
async function deliverOne(
    ).run(attempt, status, row.id);
    return;
  }
  backOff(db, log, row, status, now);
  backOff(db, log, row, status, now, bus);
}
/** Record an attempt and schedule the next one — or give up. */
@@ -465,6 +477,7 @@
function backOff(
  row: DueRow,
  status: number,
  now: Date,
  bus: EventBus,
): void {
  const attempt = row.attempts + 1;
  if (attempt >= MAX_ATTEMPTS) {
@@ -476,6 +489,13 @@
function backOff(
      attempts: attempt,
      status,
    });
    if (!row.ping)
      noteFailure(
        db,
        bus,
        row.webhook_id,
        status === 0 ? "no response" : `answered HTTP ${status}`,
      );
    return;
  }
  // Exponential backoff: ~10s, 40s, 90s, … capped implicitly by attempt count.
@@ -591,11 +611,15 @@
const delivering = new WeakSet<Database>();
/** For tests and for anyone who wants to know why nothing is being sent. */
export const deliveryBusy = (db: Database): boolean => delivering.has(db);
/** A never-throwing wrapper for the worker loop's fire-and-forget calls. */
export function runDelivery(db: Database, log: Log): void {
/**
 * A never-throwing wrapper for the worker loop's fire-and-forget calls. The
 * bus is required: it is the forge's own, so the `webhook.failing` event
 * wakes live streams and not only the outbox.
 */
export function runDelivery(db: Database, log: Log, bus: EventBus): void {
  if (delivering.has(db)) return;
  delivering.add(db);
  deliverDue(db, log)
  deliverDue(db, log, new Date(), {}, bus)
    .catch((error) => {
      // The worker must survive any delivery batch failing — a thrown rejection
      // here would take the process down (a lesson paid for elsewhere).
@@ -189,6 +189,7 @@
const WORDS: Record<string, string> = {
  "webhook.deleted": "removed a webhook",
  "webhook.paused": "paused a webhook",
  "webhook.resumed": "resumed a webhook",
  "webhook.failing": "had a webhook fail three deliveries in a row",
  "origin.connected": "connected an origin",
  "origin.disconnected": "disconnected an origin",
  "origin.needs_reauth": "needs to reconnect an origin",
@@ -731,6 +731,25 @@
function fromEvent(
            ? `#${String(number)} now waits for #${String(p["blockedBy"])}`
            : `#${String(number)} no longer waits for #${String(p["blockedBy"])}`,
      };
    // A hook that failed three deliveries in a row: writers only (`webhook_streak.ts`).
    case "webhook.failing": {
      
        sentinelcrew
        2d ago
        
      
      The link target /settings/hooks is admin-only, so a writer who is not an admin gets a refusal. Link somewhere the writer can open, or say who must act. Also add a test that renders this line to HTML and shows the reason is escaped.
    
      const host = typeof p["host"] === "string" ? p["host"] : "a webhook";
      const reason = typeof p["reason"] === "string" ? p["reason"] : "";
      const failures = typeof p["failures"] === "number" ? p["failures"] : 3;
      const what = clip(
        `Webhook to ${host} failed ${failures} times${reason === "" ? "" : `: ${reason}`}`,
        300,
      );
      // Settings › Webhooks is for admins, and a writer need not be one: the
      // line links to the repository and says who can redeliver or pause.
      return {
        ...common,
        kind: "map",
        link: base(repo),
        what: `${what} (a repository admin can redeliver or pause it in Settings › Webhooks)`,
        urgency: 3,
      };
    }
    case "crew.governed":
      return {
        ...common,
@@ -583,11 +583,11 @@
const notifier = startNotifier(app, { forgeUrl });
// Expired sessions and run tokens ride the same tick: cheap DELETEs, never
// worth a worker.
const deliveryTimer = setInterval(() => {
  runDelivery(app.db, app.log);
  runDelivery(app.db, app.log, app.bus);
  pruneSessions(app.db);
  pruneTokens(app.db);
}, 5_000);
runDelivery(app.db, app.log); // sweep anything queued while we were down
runDelivery(app.db, app.log, app.bus); // sweep anything queued while we were down
pruneSessions(app.db);
pruneTokens(app.db);
@@ -734,7 +734,7 @@
describe("backpressure", () => {
      }
      expect(deliveryBusy(app.db)).toBe(false);
      runDelivery(app.db, app.log);
      runDelivery(app.db, app.log, app.bus);
      expect(deliveryBusy(app.db)).toBe(true);
      // Every tick that lands while a batch is in flight is a no-op, which is
      // what keeps a slow endpoint from being attempted twice per delivery.
@@ -744,7 +744,7 @@
describe("backpressure", () => {
          []
        >("SELECT COALESCE(SUM(attempts), 0) AS n FROM webhook_delivery")
        .get()!.n;
      for (let i = 0; i < 10; i++) runDelivery(app.db, app.log);
      for (let i = 0; i < 10; i++) runDelivery(app.db, app.log, app.bus);
      expect(attemptsBefore).toBe(0);
      const deadline = Date.now() + 15_000;
@@ -0,0 +1,186 @@
/**
 * A hook that fails three deliveries in a row is said once, to the people who
 * can write the repository: one `webhook.failing` per streak, ended by a
 * success, never shown to readers, and never trusted as markup.
 */
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { createRepo } from "../src/actions/repos.ts";
import { audienceFilter } from "../src/core/event_audience.ts";
import { listEvents } from "../src/core/events.ts";
import type { Repo } from "../src/core/repos.ts";
import { createUser, type User } from "../src/core/users.ts";
import { id } from "../src/kernel/id.ts";
import { createWebhook, deliverDue, enqueueForEvent } from "../src/core/webhooks.ts";
import { noteFailure } from "../src/core/webhook_streak.ts";
import { readFeed } from "../src/feed/feed.ts";
import { feedCard } from "../src/web/studio/feed.ts";
import { getUserById } from "../src/core/users.ts";
import { canWrite, getRepoById } from "../src/core/repos.ts";
import { candidatesFor } from "../src/feed/items.ts";
import { noChips } from "../src/core/feed_chips.ts";
import { createLog } from "../src/kernel/log.ts";
import { promote } from "./helpers/members.ts";
describe("a failing webhook", () => {
  let dir: string;
  let app: App;
  let alice: User;
  let bob: User;
  let repo: Repo;
  let hookId: string;
  const mk = async (username: string, first = false) => {
    const r = await createUser(app.db, {
      origin: "seed", trusted: true, username, email: `${username}@forge.local`,
      password: "a-good-passphrase", adminIfFirst: first,
    });
    if (!r.ok) throw new Error(r.error);
    promote(app.db, r.value);
    return r.value;
  };
  const hook = (url: string): string => {
    const r = createWebhook(app.db, { repoId: repo.id, url, kinds: "*", secret: "s3cret-signing-key-0123456789" });
    if (!r.ok) throw new Error(r.error);
    return r.value.webhook.id;
  };
  /** A delivery that has ended in `state`, ids in the order made. */
  const ended = (webhookId: string, state: "ok" | "failed"): string => {
    const ev = app.bus.append({ kind: "push", repoId: repo.id, actorId: alice.id, payload: {} });
    // The bus has already queued it for every matching hook; settle that row.
    app.db.run(
      `INSERT INTO webhook_delivery (id, webhook_id, event_id, state, attempts) VALUES (?1, ?2, ?3, ?4, 6)
       ON CONFLICT (webhook_id, event_id) DO UPDATE SET state = ?4, attempts = 6`,
      [id(), webhookId, ev.id, state],
    );
    const did = app.db
      .query<{ id: string }, [string, string]>("SELECT id FROM webhook_delivery WHERE webhook_id = ?1 AND event_id = ?2")
      .get(webhookId, ev.id)!.id;
    return did;
  };
  const failing = () => listEvents(app.db, { kinds: ["webhook.failing"] }).filter((e) => e.repoId === repo.id);
  const fail = (webhookId: string, reason = "answered HTTP 500") => {
    ended(webhookId, "failed");
    noteFailure(app.db, app.bus, webhookId, reason);
  };
  beforeAll(async () => {
    dir = await mkdtemp(join(tmpdir(), "forge-hookfail-"));
    app = createApp({ dataDir: dir, port: 0 });
    alice = await mk("alice", true);
    bob = await mk("bob");
    const made = await createRepo(app, alice, { name: "open", visibility: "public" });
    if (!made.ok) throw new Error(made.error);
    repo = made.value;
    hookId = hook("https://hooks.example.test/services/s3cr3t-in-the-path");
  });
  afterAll(async () => {
    app.close();
    await rm(dir, { recursive: true, force: true });
  });
  test("says nothing before the third failure, then once, and not again on a fourth", () => {
    fail(hookId);
    fail(hookId);
    expect(failing()).toHaveLength(0);
    fail(hookId, "answered HTTP 502");
    expect(failing()).toHaveLength(1);
    const [event] = failing();
    expect(event!.payload.host).toBe("hooks.example.test");
    expect(event!.payload.reason).toBe("answered HTTP 502");
    expect(JSON.stringify(event!.payload)).not.toContain("s3cr3t");
    fail(hookId);
    fail(hookId);
    expect(failing()).toHaveLength(1);
  });
  test("the news is not queued as a delivery to any hook", () => {
    const [event] = failing();
    const queued = app.db
      .query<{ n: number }, [string]>("SELECT count(*) n FROM webhook_delivery WHERE event_id = ?1")
      .get(event!.id)!.n;
    expect(queued).toBe(0);
  });
  test("a success ends the streak, and three more failures are a new one", () => {
    ended(hookId, "ok");
    fail(hookId);
    fail(hookId);
    expect(failing()).toHaveLength(1);
    fail(hookId);
    expect(failing()).toHaveLength(2);
  });
  test("the worker reports a refused delivery's reason on the third", async () => {
    const other = hook("https://hooks.example.test/other");
    // A private address is refused at once, with its reason, and never retried.
    app.db.run("UPDATE webhook SET url = 'http://127.0.0.1/x' WHERE id = ?1", [other]);
    const before = failing().length;
    for (let i = 0; i < 3; i++) {
      const ev = app.bus.append({ kind: "push", repoId: repo.id, actorId: alice.id, payload: {} });
      enqueueForEvent(app.db, ev);
    }
    await deliverDue(app.db, createLog("test"), new Date(), {}, app.bus);
    const now = failing();
    expect(now.length).toBe(before + 1);
    expect(now[0]!.payload.host).toBe("127.0.0.1");
    expect(String(now[0]!.payload.reason)).toContain("refused");
  });
  test("writers hear it, readers do not", () => {
    const event = failing()[0]!;
    expect(audienceFilter(app.db, alice)(event)).toBe(true);
    expect(audienceFilter(app.db, bob)(event)).toBe(false);
    expect(audienceFilter(app.db, undefined)(event)).toBe(false);
  });
  test("the line links to the repository and says an admin acts; readers get none", () => {
    const evil = hook("https://hooks.example.test/evil");
    for (let i = 0; i < 3; i++) fail(evil, '<img src=x onerror="alert(1)">');
    const lines = candidatesFor(app.db, alice, Date.now(), noChips).filter((c) => c.what.startsWith("Webhook to "));
    const line = lines.find((c) => c.what.includes("<img"))!;
    expect(line.what).toStartWith('Webhook to hooks.example.test failed 3 times: <img src=x');
    expect(line.what).toContain("a repository admin can redeliver or pause it in Settings › Webhooks");
    expect(line.link).toBe(`/${repo.ownerName}/${repo.name}`);
    expect(candidatesFor(app.db, bob, Date.now(), noChips).some((c) => c.what.startsWith("Webhook to "))).toBe(false);
  });
  test("the rendered card escapes the reason and the host", () => {
    const evil = hook("https://hooks.example.test/evil2");
    for (let i = 0; i < 3; i++) fail(evil, '<img src=x onerror="alert(1)">');
    // A host cannot hold markup once parsed, but a payload is only data: forge one.
    app.bus.append({
      kind: "webhook.failing",
      repoId: repo.id,
      payload: { webhookId: "w", since: "s", host: "<b>evil</b>.test", reason: "<script>x()</script>", failures: 3 },
    });
    const cards = readFeed(app.db, alice, { now: Date.now(), limit: 100 })
      .items.filter((c) => c.what.startsWith("Webhook to "))
      .map((c) => feedCard(c).text);
    expect(cards.length).toBeGreaterThanOrEqual(2);
    const all = cards.join("\n");
    expect(all).toContain("&lt;img src=x onerror=");
    expect(all).toContain("&lt;b&gt;evil&lt;/b&gt;.test");
    expect(all).toContain("&lt;script&gt;");
    expect(all).not.toContain("<img src=x");
    expect(all).not.toContain("<b>evil");
    expect(all).not.toContain("<script>");
  });
  test("a suspended writer hears nothing", () => {
    const event = failing()[0]!;
    expect(audienceFilter(app.db, alice)(event)).toBe(true);
    app.db.run("UPDATE user SET suspended_at = ?2 WHERE id = ?1", [bob.id, new Date().toISOString()]);
    // Bob owns a repository of his own with a hook of his own.
    const suspended = getUserById(app.db, bob.id)!;
    app.db.run("UPDATE repo SET owner_id = ?2 WHERE id = ?1", [repo.id, bob.id]);
    const ownerRepo = getRepoById(app.db, repo.id)!;
    expect(canWrite(app.db, ownerRepo, suspended)).toBe(true);
    expect(audienceFilter(app.db, suspended)(event)).toBe(false);
    app.db.run("UPDATE repo SET owner_id = ?2 WHERE id = ?1", [repo.id, alice.id]);
    app.db.run("UPDATE user SET suspended_at = NULL WHERE id = ?1", [bob.id]);
  });
});
		@@ -0,0 +1,12 @@
	1	# A failing webhook is said once
	2
	3	When a webhook's third delivery in a row ends `failed`, one `webhook.failing`
	4	event goes on the bus (`src/core/webhook_streak.ts`, called from the delivery
	5	worker in `src/core/webhooks.ts`). Once per streak: a success ends it, and a
	6	fourth failure or a redelivery that fails again adds nothing.
	7
	8	Writers of the repository see one line in their feed — "Webhook to
	9	example.com failed 3 times: answered HTTP 500" — linking to the repository, and naming Settings ›
	10	Webhooks, where redeliver and pause are (`webhook-redeliver.md`,
	11	`webhook-pause.md`). Readers do not: the event's audience is `writers`
	12	(`src/core/event_audience.ts`). It names the host, never the address.
		@@ -181,6 +181,9 @@ const READER_KINDS: ReadonlySet<string> = new Set([
181	181	* `test/event-audience-coverage.test.ts` fails on an emitted kind that is.
182	182	*/
183	183	export function classifiedAudience(kind: string): EventAudience \| undefined {
	184	// A hook that keeps failing is for those who can act on it: the host and
	185	// the reason, never the address (`webhook_streak.ts`).
	186	if (kind === "webhook.failing") return "writers";
184	187	if (
185	188	kind.startsWith("webhook.") \|\|
186	189	kind.startsWith("app.env_") \|\|
		@@ -298,7 +301,11 @@ export function audienceFilter(
298	301	const known = writers.get(repoId);
299	302	if (known !== undefined) return known;
300	303	const repo = getRepoById(db, repoId);
301		const yes = repo !== undefined && canWrite(db, repo, viewer);
	304	const yes =
	305	repo !== undefined &&
	306	viewer !== undefined &&
	307	isActive(viewer) &&
	308	canWrite(db, repo, viewer);
302	309	writers.set(repoId, yes);
303	310	return yes;
304	311	};
		@@ -0,0 +1,77 @@
	1	/**
	2	* A webhook that keeps failing, said once.
	3	*
	4	* Redeliver and pause exist, but nobody is told to use them. When a hook's
	5	* third delivery in a row ends `failed`, one `webhook.failing` event goes on
	6	* the bus for the repository's writers: once per streak, not per failure. A
	7	* delivery that ends `ok` ends the streak; the next three failures are a new
	8	* one. The streak is named by its oldest failed delivery (`since`), so a
	9	* redelivery that fails again, or a fourth failure, finds the event already
	10	* written and adds nothing.
	11	*/
	12
	13	import type { Database } from "bun:sqlite";
	14	import type { EventBus } from "./events.ts";
	15
	16	/** Failed deliveries in a row that make a streak worth saying. */
	17	export const STREAK = 3;
	18
	19	/** The longest reason the event carries. */
	20	const REASON_MAX = 200;
	21
	22	/** The host a hook posts to; the whole address (often a secret) is not for writers. */
	23	export function hostOf(url: string): string {
	24	try {
	25	return new URL(url).hostname;
	26	} catch {
	27	return "an unreadable address";
	28	}
	29	}
	30
	31	/** Call after a delivery of this hook has ended `failed`. */
	32	export function noteFailure(
	33	db: Database,
	34	bus: EventBus,
	35	webhookId: string,
	36	reason: string,
	37	): void {
	38	const hook = db
	39	.query<{ repo_id: string; url: string }, [string]>(
	40	"SELECT repo_id, url FROM webhook WHERE id = ?1",
	41	)
	42	.get(webhookId);
	43	if (hook === null) return;
	44	// The streak is every failure since the newest success.
	45	const failed = db
	46	.query<{ id: string }, [string]>(
	47	`SELECT id FROM webhook_delivery
	48	WHERE webhook_id = ?1 AND state = 'failed'
	49	AND id > COALESCE((SELECT MAX(id) FROM webhook_delivery
	50	WHERE webhook_id = ?1 AND state = 'ok'), '')
	51	ORDER BY id`,
	52	)
	53	.all(webhookId);
	54	if (failed.length < STREAK) return;
	55	// Its oldest failure names it, however long it grows.
	56	const since = failed[0]!.id;
	57	const said = db
	58	.query(
	59	`SELECT 1 FROM event
	60	WHERE kind = 'webhook.failing' AND repo_id = ?1
	61	AND json_extract(payload, '$.webhookId') = ?2
	62	AND json_extract(payload, '$.since') = ?3 LIMIT 1`,
	63	)
	64	.get(hook.repo_id, webhookId, since);
	65	if (said !== null) return;
	66	bus.append({
	67	kind: "webhook.failing",
	68	repoId: hook.repo_id,
	69	payload: {
	70	webhookId,
	71	since,
	72	host: hostOf(hook.url),
	73	reason: reason.slice(0, REASON_MAX),
	74	failures: STREAK,
	75	},
	76	});
	77	}
		@@ -19,7 +19,8 @@ import {
19	19	type OutboundOptions,
20	20	} from "../kernel/outbound.ts";
21	21	import { err, ok, type Result } from "../kernel/result.ts";
22		import type { ForgeEvent } from "./events.ts";
	22	import { EventBus, type ForgeEvent } from "./events.ts";
	23	import { noteFailure } from "./webhook_streak.ts";
23	24
24	25	const MAX_ATTEMPTS = 6;
25	26	const DELIVER_TIMEOUT_MS = 10_000;
		@@ -167,6 +168,9 @@ export function kindMatches(spec: string, kind: string): boolean {
167	168	*/
168	169	export function enqueueForEvent(db: Database, event: ForgeEvent): number {
169	170	if (event.repoId === null) return 0;
	171	// The news that a hook is failing is for writers, not for hooks: sent to the
	172	// failing one it would be one more doomed delivery in the streak it reports.
	173	if (event.kind === "webhook.failing") return 0;
170	174	// A hook's URL is somewhere else: nothing about a sealed report goes
171	175	// there, not even that it moved (`sealed.ts`).
172	176	if (eventSealed(db, event)) return 0;
		@@ -300,7 +304,6 @@ interface DueRow {
300	304	created_at: string;
301	305	}
302	306
303		/** Deliver all due pending deliveries. Returns how many were attempted. */
304	307	/**
305	308	* How long a claimed delivery is its sender's before another may try it:
306	309	* well past the longest one send may take, so a claim never lapses while
		@@ -309,11 +312,18 @@ interface DueRow {
309	312	export const CLAIM_MS = 60_000;
310	313	export { DELIVER_TIMEOUT_MS };
311	314
	315	/**
	316	* Deliver all due pending deliveries. Returns how many were attempted.
	317	* `bus` is optional only for one-shot callers (tests, scripts) with no live
	318	* listeners: it then writes a bus of its own, which stores the event and wakes
	319	* no one. The worker (`runDelivery`) always passes the forge's bus.
	320	*/
312	321	export async function deliverDue(
313	322	db: Database,
314	323	log: Log,
315	324	now = new Date(),
316	325	outbound: OutboundOptions = {},
	326	bus: EventBus = new EventBus(db),
317	327	): Promise<number> {
318	328	const nowIso = now.toISOString();
319	329	const events = db
		@@ -365,7 +375,7 @@ export async function deliverDue(
365	375	);
366	376	if (claimed.changes !== 1) continue;
367	377	try {
368		await deliverOne(db, log, row, now, outbound);
	378	await deliverOne(db, log, row, now, outbound, bus);
369	379	} catch (error) {
370	380	// One malformed delivery must not take the batch with it. Without
371	381	// this, a single unsendable row stops every delivery behind it — and
		@@ -377,7 +387,7 @@ export async function deliverDue(
377	387	delivery: row.id,
378	388	error: String(error),
379	389	});
380		backOff(db, log, row, 0, now);
	390	backOff(db, log, row, 0, now, bus);
381	391	}
382	392	}
383	393	return due.length;
		@@ -394,6 +404,7 @@ async function deliverOne(
394	404	row: DueRow,
395	405	now: Date,
396	406	outbound: OutboundOptions,
	407	bus: EventBus,
397	408	): Promise<void> {
398	409	// A ping's body is its payload as it stands: { kind, repo, hook, at }.
399	410	const body =
		@@ -440,6 +451,7 @@ async function deliverOne(
440	451	db.query(
441	452	`UPDATE ${tableOf(row)} SET state = 'failed', attempts = ?1, last_status = NULL, last_error = ?2 WHERE id = ?3`,
442	453	).run(attempt, error.message, row.id);
	454	if (!row.ping) noteFailure(db, bus, row.webhook_id, error.message);
443	455	return;
444	456	}
445	457	log.warn("webhook delivery error", {
		@@ -455,7 +467,7 @@ async function deliverOne(
455	467	).run(attempt, status, row.id);
456	468	return;
457	469	}
458		backOff(db, log, row, status, now);
	470	backOff(db, log, row, status, now, bus);
459	471	}
460	472
461	473	/** Record an attempt and schedule the next one — or give up. */
		@@ -465,6 +477,7 @@ function backOff(
465	477	row: DueRow,
466	478	status: number,
467	479	now: Date,
	480	bus: EventBus,
468	481	): void {
469	482	const attempt = row.attempts + 1;
470	483	if (attempt >= MAX_ATTEMPTS) {
		@@ -476,6 +489,13 @@ function backOff(
476	489	attempts: attempt,
477	490	status,
478	491	});
	492	if (!row.ping)
	493	noteFailure(
	494	db,
	495	bus,
	496	row.webhook_id,
	497	status === 0 ? "no response" : `answered HTTP ${status}`,
	498	);
479	499	return;
480	500	}
481	501	// Exponential backoff: ~10s, 40s, 90s, … capped implicitly by attempt count.
		@@ -591,11 +611,15 @@ const delivering = new WeakSet<Database>();
591	611	/** For tests and for anyone who wants to know why nothing is being sent. */
592	612	export const deliveryBusy = (db: Database): boolean => delivering.has(db);
593	613
594		/** A never-throwing wrapper for the worker loop's fire-and-forget calls. */
595		export function runDelivery(db: Database, log: Log): void {
	614	/**
	615	* A never-throwing wrapper for the worker loop's fire-and-forget calls. The
	616	* bus is required: it is the forge's own, so the `webhook.failing` event
	617	* wakes live streams and not only the outbox.
	618	*/
	619	export function runDelivery(db: Database, log: Log, bus: EventBus): void {
596	620	if (delivering.has(db)) return;
597	621	delivering.add(db);
598		deliverDue(db, log)
	622	deliverDue(db, log, new Date(), {}, bus)
599	623	.catch((error) => {
600	624	// The worker must survive any delivery batch failing — a thrown rejection
601	625	// here would take the process down (a lesson paid for elsewhere).
		@@ -189,6 +189,7 @@ const WORDS: Record<string, string> = {
189	189	"webhook.deleted": "removed a webhook",
190	190	"webhook.paused": "paused a webhook",
191	191	"webhook.resumed": "resumed a webhook",
	192	"webhook.failing": "had a webhook fail three deliveries in a row",
192	193	"origin.connected": "connected an origin",
193	194	"origin.disconnected": "disconnected an origin",
194	195	"origin.needs_reauth": "needs to reconnect an origin",
		@@ -731,6 +731,25 @@ function fromEvent(
731	731	? `#${String(number)} now waits for #${String(p["blockedBy"])}`
732	732	: `#${String(number)} no longer waits for #${String(p["blockedBy"])}`,
733	733	};
	734	// A hook that failed three deliveries in a row: writers only (`webhook_streak.ts`).
	735	case "webhook.failing": {
		sentinelcrew 2d ago The link target /settings/hooks is admin-only, so a writer who is not an admin gets a refusal. Link somewhere the writer can open, or say who must act. Also add a test that renders this line to HTML and shows the reason is escaped.
	736	const host = typeof p["host"] === "string" ? p["host"] : "a webhook";
	737	const reason = typeof p["reason"] === "string" ? p["reason"] : "";
	738	const failures = typeof p["failures"] === "number" ? p["failures"] : 3;
	739	const what = clip(
	740	`Webhook to ${host} failed ${failures} times${reason === "" ? "" : `: ${reason}`}`,
	741	300,
	742	);
	743	// Settings › Webhooks is for admins, and a writer need not be one: the
	744	// line links to the repository and says who can redeliver or pause.
	745	return {
	746	...common,
	747	kind: "map",
	748	link: base(repo),
	749	what: `${what} (a repository admin can redeliver or pause it in Settings › Webhooks)`,
	750	urgency: 3,
	751	};
	752	}
734	753	case "crew.governed":
735	754	return {
736	755	...common,
		@@ -583,11 +583,11 @@ const notifier = startNotifier(app, { forgeUrl });
583	583	// Expired sessions and run tokens ride the same tick: cheap DELETEs, never
584	584	// worth a worker.
585	585	const deliveryTimer = setInterval(() => {
586		runDelivery(app.db, app.log);
	586	runDelivery(app.db, app.log, app.bus);
587	587	pruneSessions(app.db);
588	588	pruneTokens(app.db);
589	589	}, 5_000);
590		runDelivery(app.db, app.log); // sweep anything queued while we were down
	590	runDelivery(app.db, app.log, app.bus); // sweep anything queued while we were down
591	591	pruneSessions(app.db);
592	592	pruneTokens(app.db);
593	593
		@@ -734,7 +734,7 @@ describe("backpressure", () => {
734	734	}
735	735
736	736	expect(deliveryBusy(app.db)).toBe(false);
737		runDelivery(app.db, app.log);
	737	runDelivery(app.db, app.log, app.bus);
738	738	expect(deliveryBusy(app.db)).toBe(true);
739	739	// Every tick that lands while a batch is in flight is a no-op, which is
740	740	// what keeps a slow endpoint from being attempted twice per delivery.
		@@ -744,7 +744,7 @@ describe("backpressure", () => {
744	744	[]
745	745	>("SELECT COALESCE(SUM(attempts), 0) AS n FROM webhook_delivery")
746	746	.get()!.n;
747		for (let i = 0; i < 10; i++) runDelivery(app.db, app.log);
	747	for (let i = 0; i < 10; i++) runDelivery(app.db, app.log, app.bus);
748	748	expect(attemptsBefore).toBe(0);
749	749
750	750	const deadline = Date.now() + 15_000;
		@@ -0,0 +1,186 @@
	1	/**
	2	* A hook that fails three deliveries in a row is said once, to the people who
	3	* can write the repository: one `webhook.failing` per streak, ended by a
	4	* success, never shown to readers, and never trusted as markup.
	5	*/
	6	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	7	import { mkdtemp, rm } from "node:fs/promises";
	8	import { tmpdir } from "node:os";
	9	import { join } from "node:path";
	10	import { createApp, type App } from "../src/app.ts";
	11	import { createRepo } from "../src/actions/repos.ts";
	12	import { audienceFilter } from "../src/core/event_audience.ts";
	13	import { listEvents } from "../src/core/events.ts";
	14	import type { Repo } from "../src/core/repos.ts";
	15	import { createUser, type User } from "../src/core/users.ts";
	16	import { id } from "../src/kernel/id.ts";
	17	import { createWebhook, deliverDue, enqueueForEvent } from "../src/core/webhooks.ts";
	18	import { noteFailure } from "../src/core/webhook_streak.ts";
	19	import { readFeed } from "../src/feed/feed.ts";
	20	import { feedCard } from "../src/web/studio/feed.ts";
	21	import { getUserById } from "../src/core/users.ts";
	22	import { canWrite, getRepoById } from "../src/core/repos.ts";
	23	import { candidatesFor } from "../src/feed/items.ts";
	24	import { noChips } from "../src/core/feed_chips.ts";
	25	import { createLog } from "../src/kernel/log.ts";
	26	import { promote } from "./helpers/members.ts";
	27
	28	describe("a failing webhook", () => {
	29	let dir: string;
	30	let app: App;
	31	let alice: User;
	32	let bob: User;
	33	let repo: Repo;
	34	let hookId: string;
	35
	36	const mk = async (username: string, first = false) => {
	37	const r = await createUser(app.db, {
	38	origin: "seed", trusted: true, username, email: `${username}@forge.local`,
	39	password: "a-good-passphrase", adminIfFirst: first,
	40	});
	41	if (!r.ok) throw new Error(r.error);
	42	promote(app.db, r.value);
	43	return r.value;
	44	};
	45	const hook = (url: string): string => {
	46	const r = createWebhook(app.db, { repoId: repo.id, url, kinds: "*", secret: "s3cret-signing-key-0123456789" });
	47	if (!r.ok) throw new Error(r.error);
	48	return r.value.webhook.id;
	49	};
	50	/** A delivery that has ended in `state`, ids in the order made. */
	51	const ended = (webhookId: string, state: "ok" \| "failed"): string => {
	52	const ev = app.bus.append({ kind: "push", repoId: repo.id, actorId: alice.id, payload: {} });
	53	// The bus has already queued it for every matching hook; settle that row.
	54	app.db.run(
	55	`INSERT INTO webhook_delivery (id, webhook_id, event_id, state, attempts) VALUES (?1, ?2, ?3, ?4, 6)
	56	ON CONFLICT (webhook_id, event_id) DO UPDATE SET state = ?4, attempts = 6`,
	57	[id(), webhookId, ev.id, state],
	58	);
	59	const did = app.db
	60	.query<{ id: string }, [string, string]>("SELECT id FROM webhook_delivery WHERE webhook_id = ?1 AND event_id = ?2")
	61	.get(webhookId, ev.id)!.id;
	62	return did;
	63	};
	64	const failing = () => listEvents(app.db, { kinds: ["webhook.failing"] }).filter((e) => e.repoId === repo.id);
	65	const fail = (webhookId: string, reason = "answered HTTP 500") => {
	66	ended(webhookId, "failed");
	67	noteFailure(app.db, app.bus, webhookId, reason);
	68	};
	69
	70	beforeAll(async () => {
	71	dir = await mkdtemp(join(tmpdir(), "forge-hookfail-"));
	72	app = createApp({ dataDir: dir, port: 0 });
	73	alice = await mk("alice", true);
	74	bob = await mk("bob");
	75	const made = await createRepo(app, alice, { name: "open", visibility: "public" });
	76	if (!made.ok) throw new Error(made.error);
	77	repo = made.value;
	78	hookId = hook("https://hooks.example.test/services/s3cr3t-in-the-path");
	79	});
	80	afterAll(async () => {
	81	app.close();
	82	await rm(dir, { recursive: true, force: true });
	83	});
	84
	85	test("says nothing before the third failure, then once, and not again on a fourth", () => {
	86	fail(hookId);
	87	fail(hookId);
	88	expect(failing()).toHaveLength(0);
	89	fail(hookId, "answered HTTP 502");
	90	expect(failing()).toHaveLength(1);
	91	const [event] = failing();
	92	expect(event!.payload.host).toBe("hooks.example.test");
	93	expect(event!.payload.reason).toBe("answered HTTP 502");
	94	expect(JSON.stringify(event!.payload)).not.toContain("s3cr3t");
	95	fail(hookId);
	96	fail(hookId);
	97	expect(failing()).toHaveLength(1);
	98	});
	99
	100	test("the news is not queued as a delivery to any hook", () => {
	101	const [event] = failing();
	102	const queued = app.db
	103	.query<{ n: number }, [string]>("SELECT count(*) n FROM webhook_delivery WHERE event_id = ?1")
	104	.get(event!.id)!.n;
	105	expect(queued).toBe(0);
	106	});
	107
	108	test("a success ends the streak, and three more failures are a new one", () => {
	109	ended(hookId, "ok");
	110	fail(hookId);
	111	fail(hookId);
	112	expect(failing()).toHaveLength(1);
	113	fail(hookId);
	114	expect(failing()).toHaveLength(2);
	115	});
	116
	117	test("the worker reports a refused delivery's reason on the third", async () => {
	118	const other = hook("https://hooks.example.test/other");
	119	// A private address is refused at once, with its reason, and never retried.
	120	app.db.run("UPDATE webhook SET url = 'http://127.0.0.1/x' WHERE id = ?1", [other]);
	121	const before = failing().length;
	122	for (let i = 0; i < 3; i++) {
	123	const ev = app.bus.append({ kind: "push", repoId: repo.id, actorId: alice.id, payload: {} });
	124	enqueueForEvent(app.db, ev);
	125	}
	126	await deliverDue(app.db, createLog("test"), new Date(), {}, app.bus);
	127	const now = failing();
	128	expect(now.length).toBe(before + 1);
	129	expect(now[0]!.payload.host).toBe("127.0.0.1");
	130	expect(String(now[0]!.payload.reason)).toContain("refused");
	131	});
	132
	133	test("writers hear it, readers do not", () => {
	134	const event = failing()[0]!;
	135	expect(audienceFilter(app.db, alice)(event)).toBe(true);
	136	expect(audienceFilter(app.db, bob)(event)).toBe(false);
	137	expect(audienceFilter(app.db, undefined)(event)).toBe(false);
	138	});
	139
	140	test("the line links to the repository and says an admin acts; readers get none", () => {
	141	const evil = hook("https://hooks.example.test/evil");
	142	for (let i = 0; i < 3; i++) fail(evil, '<img src=x onerror="alert(1)">');
	143	const lines = candidatesFor(app.db, alice, Date.now(), noChips).filter((c) => c.what.startsWith("Webhook to "));
	144	const line = lines.find((c) => c.what.includes("<img"))!;
	145	expect(line.what).toStartWith('Webhook to hooks.example.test failed 3 times: <img src=x');
	146	expect(line.what).toContain("a repository admin can redeliver or pause it in Settings › Webhooks");
	147	expect(line.link).toBe(`/${repo.ownerName}/${repo.name}`);
	148	expect(candidatesFor(app.db, bob, Date.now(), noChips).some((c) => c.what.startsWith("Webhook to "))).toBe(false);
	149	});
	150
	151	test("the rendered card escapes the reason and the host", () => {
	152	const evil = hook("https://hooks.example.test/evil2");
	153	for (let i = 0; i < 3; i++) fail(evil, '<img src=x onerror="alert(1)">');
	154	// A host cannot hold markup once parsed, but a payload is only data: forge one.
	155	app.bus.append({
	156	kind: "webhook.failing",
	157	repoId: repo.id,
	158	payload: { webhookId: "w", since: "s", host: "<b>evil</b>.test", reason: "<script>x()</script>", failures: 3 },
	159	});
	160	const cards = readFeed(app.db, alice, { now: Date.now(), limit: 100 })
	161	.items.filter((c) => c.what.startsWith("Webhook to "))
	162	.map((c) => feedCard(c).text);
	163	expect(cards.length).toBeGreaterThanOrEqual(2);
	164	const all = cards.join("\n");
	165	expect(all).toContain("<img src=x onerror=");
	166	expect(all).toContain("<b>evil</b>.test");
	167	expect(all).toContain("<script>");
	168	expect(all).not.toContain("<img src=x");
	169	expect(all).not.toContain("<b>evil");
	170	expect(all).not.toContain("<script>");
	171	});
	172
	173	test("a suspended writer hears nothing", () => {
	174	const event = failing()[0]!;
	175	expect(audienceFilter(app.db, alice)(event)).toBe(true);
	176	app.db.run("UPDATE user SET suspended_at = ?2 WHERE id = ?1", [bob.id, new Date().toISOString()]);
	177	// Bob owns a repository of his own with a hook of his own.
	178	const suspended = getUserById(app.db, bob.id)!;
	179	app.db.run("UPDATE repo SET owner_id = ?2 WHERE id = ?1", [repo.id, bob.id]);
	180	const ownerRepo = getRepoById(app.db, repo.id)!;
	181	expect(canWrite(app.db, ownerRepo, suspended)).toBe(true);
	182	expect(audienceFilter(app.db, suspended)(event)).toBe(false);
	183	app.db.run("UPDATE repo SET owner_id = ?2 WHERE id = ?1", [repo.id, alice.id]);
	184	app.db.run("UPDATE user SET suspended_at = NULL WHERE id = ?1", [bob.id]);
	185	});
	186	});