feat(orb): record terminal PR outcomes + global merged/closed aggregate

migrations/0067_orb_pr_outcomes.sql

@@ -0,0 +1,16 @@
+-- Gittensory Orb central GitHub App (#1255) — terminal pull-request outcomes (merged | closed) observed via
+-- the central App's webhook. The raw material for the global "proof of power" homepage counter (total merged /
+-- closed across ALL registered maintainer repos, das-github-mirror style). Aggregated only over REGISTERED
+-- installations. Idempotent on (repo, pr_number): a redelivery or a reopen→close cycle overwrites the latest
+-- terminal state. occurred_at is always written explicitly (CURRENT_TIMESTAMP in VALUES); the column default is
+-- a fallback only.
+CREATE TABLE IF NOT EXISTS orb_pr_outcomes (
+  repository_full_name TEXT NOT NULL,
+  pr_number INTEGER NOT NULL,
+  installation_id INTEGER,
+  outcome TEXT NOT NULL,                       -- 'merged' | 'closed'
+  occurred_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (repository_full_name, pr_number)
+);
+CREATE INDEX IF NOT EXISTS orb_pr_outcomes_installation_idx ON orb_pr_outcomes(installation_id);
+CREATE INDEX IF NOT EXISTS orb_pr_outcomes_outcome_idx ON orb_pr_outcomes(outcome);

src/orb/outcomes.ts

@@ -0,0 +1,49 @@
+// Gittensory Orb central GitHub App (#1255) — terminal PR-outcome capture + the global aggregate.
+//
+// recordOrbPrOutcome runs synchronously from the verified webhook receiver: a `pull_request` `closed` event
+// records whether the PR was merged or closed (no merge) into orb_pr_outcomes, keyed on (repo, pr_number) so a
+// redelivery or reopen→close cycle overwrites the latest terminal state. getOrbGlobalStats sums it across only
+// REGISTERED installations — the das-github-mirror-style "total merged / closed" feeding the homepage counter.
+import type { GitHubWebhookPayload } from "../types";
+
+export async function recordOrbPrOutcome(env: Env, eventName: string, payload: GitHubWebhookPayload): Promise<void> {
+  if (eventName !== "pull_request" || payload.action !== "closed") return; // only a terminal close carries an outcome
+  const pr = payload.pull_request;
+  const repo = payload.repository?.full_name;
+  if (!pr?.number || !repo) return;
+  // merged_at is set iff the PR was merged; a close without it is a plain close (rejected / abandoned).
+  const outcome = pr.merged_at ? "merged" : "closed";
+  await env.DB.prepare(
+    `INSERT INTO orb_pr_outcomes (repository_full_name, pr_number, installation_id, outcome, occurred_at)
+     VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
+     ON CONFLICT(repository_full_name, pr_number) DO UPDATE SET
+       installation_id = excluded.installation_id, outcome = excluded.outcome, occurred_at = CURRENT_TIMESTAMP`,
+  )
+    .bind(repo, pr.number, payload.installation?.id ?? null, outcome)
+    .run();
+}
+
+export interface OrbGlobalStats {
+  merged: number;
+  closed: number;
+  total: number;
+}
+
+/**
+ * The public global aggregate: merged / closed / total terminal PR outcomes across REGISTERED installations
+ * only (registered = 1) — an install that hasn't been opted in never contributes to the public counter. SUM over
+ * no matching rows is NULL, so each total is nullish-guarded to 0 (fail-safe on an empty/cold table).
+ */
+export async function getOrbGlobalStats(env: Env): Promise<OrbGlobalStats> {
+  const row = await env.DB.prepare(
+    `SELECT
+       SUM(CASE WHEN o.outcome = 'merged' THEN 1 ELSE 0 END) AS merged,
+       SUM(CASE WHEN o.outcome = 'closed' THEN 1 ELSE 0 END) AS closed,
+       COUNT(*) AS total
+     FROM orb_pr_outcomes o
+     JOIN orb_github_installations i ON i.installation_id = o.installation_id AND i.registered = 1`,
+  ).first<{ merged: number | null; closed: number | null; total: number | null }>();
+  /* v8 ignore next -- an aggregate query always returns exactly one row; this guards the nullable .first() type only */
+  if (!row) return { merged: 0, closed: 0, total: 0 };
+  return { merged: row.merged ?? 0, closed: row.closed ?? 0, total: row.total ?? 0 };
+}

src/orb/webhook.ts

@@ -11,6 +11,7 @@ import type { Context } from "hono";
 import type { GitHubWebhookPayload } from "../types";
 import { sha256Hex, verifyGitHubSignature } from "../utils/crypto";
 import { upsertOrbInstallation } from "./installations";
+import { recordOrbPrOutcome } from "./outcomes";
 
 const DEFAULT_MAX_ORB_WEBHOOK_BODY_BYTES = 1024 * 1024;
 
@@ -63,11 +64,13 @@ export async function handleOrbWebhook(c: Context<{ Bindings: Env }>): Promise<R
     payloadHash,
   };
 
-  // Maintain the installation registry from `installation` lifecycle events BEFORE recording, so a failed
-  // upsert is flipped to "error" + 500 and GitHub redelivers (the dedup guard only suppresses non-error rows).
-  // No-op for every other event in PR2 — PR/review-outcome processing lands in a later queue-backed PR.
+  // Maintain the installation registry from `installation` lifecycle events, and record terminal PR outcomes
+  // from `pull_request closed` events, BEFORE recording the webhook row — so a failed write is flipped to
+  // "error" + 500 and GitHub redelivers (the dedup guard only suppresses non-error rows). Each is a no-op for
+  // every unrelated event.
   try {
     await upsertOrbInstallation(c.env, eventName, payload);
+    await recordOrbPrOutcome(c.env, eventName, payload);
   } catch {
     await recordOrbWebhookEvent(c.env, { ...eventMeta, status: "error" });
     return c.json({ error: "processing_failed", deliveryId }, 500);

test/integration/orb-outcomes.test.ts

@@ -0,0 +1,63 @@
+import { describe, expect, it } from "vitest";
+import { getOrbGlobalStats, recordOrbPrOutcome } from "../../src/orb/outcomes";
+import { createTestEnv, type TestD1Database } from "../helpers/d1";
+
+const db = (e: Env) => e.DB as unknown as TestD1Database;
+
+const closedPr = (repo: string, number: number, mergedAt: string | null, installationId = 100) =>
+  ({
+    action: "closed",
+    pull_request: { number, state: "closed", merged_at: mergedAt },
+    repository: { full_name: repo },
+    installation: { id: installationId },
+  }) as never;
+
+const registerInstall = (e: Env, id: number, registered: number) =>
+  db(e).prepare("INSERT INTO orb_github_installations (installation_id, registered) VALUES (?, ?)").bind(id, registered).run();
+
+describe("recordOrbPrOutcome", () => {
+  it("records a merged PR as outcome 'merged' (with the installation id)", async () => {
+    const e = createTestEnv();
+    await recordOrbPrOutcome(e, "pull_request", closedPr("acme/widgets", 7, "2026-06-24T00:00:00Z"));
+    const row = await db(e).prepare("SELECT outcome, installation_id FROM orb_pr_outcomes WHERE repository_full_name=? AND pr_number=?").bind("acme/widgets", 7).first<{ outcome: string; installation_id: number }>();
+    expect(row).toMatchObject({ outcome: "merged", installation_id: 100 });
+  });
+
+  it("records a closed-not-merged PR as outcome 'closed'", async () => {
+    const e = createTestEnv();
+    await recordOrbPrOutcome(e, "pull_request", closedPr("acme/widgets", 8, null));
+    expect((await db(e).prepare("SELECT outcome FROM orb_pr_outcomes WHERE pr_number=8").first<{ outcome: string }>())?.outcome).toBe("closed");
+  });
+
+  it("is a no-op for a non-pull_request event, a non-closed action, or a missing pr/repo", async () => {
+    const e = createTestEnv();
+    await recordOrbPrOutcome(e, "installation", closedPr("a/b", 1, null)); // wrong event
+    await recordOrbPrOutcome(e, "pull_request", { action: "opened", pull_request: { number: 2, state: "open", merged_at: null }, repository: { full_name: "a/b" } } as never); // not closed
+    await recordOrbPrOutcome(e, "pull_request", { action: "closed", repository: { full_name: "a/b" } } as never); // no pr
+    await recordOrbPrOutcome(e, "pull_request", { action: "closed", pull_request: { number: 3, state: "closed", merged_at: null } } as never); // no repo
+    expect((await db(e).prepare("SELECT COUNT(*) AS n FROM orb_pr_outcomes").first<{ n: number }>())?.n).toBe(0);
+  });
+
+  it("overwrites the terminal state on a re-close (idempotent on repo + pr)", async () => {
+    const e = createTestEnv();
+    await recordOrbPrOutcome(e, "pull_request", closedPr("acme/widgets", 9, null)); // closed
+    await recordOrbPrOutcome(e, "pull_request", closedPr("acme/widgets", 9, "2026-06-24T01:00:00Z")); // reopened → merged
+    expect((await db(e).prepare("SELECT outcome FROM orb_pr_outcomes WHERE pr_number=9").first<{ outcome: string }>())?.outcome).toBe("merged");
+  });
+});
+
+describe("getOrbGlobalStats", () => {
+  it("returns zeros on an empty/cold table (nullish SUM guard)", async () => {
+    expect(await getOrbGlobalStats(createTestEnv())).toEqual({ merged: 0, closed: 0, total: 0 });
+  });
+
+  it("aggregates merged/closed across REGISTERED installations only", async () => {
+    const e = createTestEnv();
+    await registerInstall(e, 100, 1); // registered
+    await registerInstall(e, 200, 0); // recorded but NOT opted in
+    await recordOrbPrOutcome(e, "pull_request", closedPr("acme/a", 1, "2026-06-24T00:00:00Z", 100)); // merged, registered
+    await recordOrbPrOutcome(e, "pull_request", closedPr("acme/b", 2, null, 100)); // closed, registered
+    await recordOrbPrOutcome(e, "pull_request", closedPr("acme/c", 3, "2026-06-24T00:00:00Z", 200)); // merged, UNREGISTERED → excluded
+    expect(await getOrbGlobalStats(e)).toEqual({ merged: 1, closed: 1, total: 2 });
+  });
+});