feat(orb): central Orb App webhook receiver (POST /v1/orb/webhook)

migrations/0063_orb_webhook_events.sql

@@ -0,0 +1,20 @@
+-- Gittensory Orb central GitHub App (#1255) — webhook delivery dedup + audit for POST /v1/orb/webhook.
+-- The central Orb App is a SEPARATE GitHub App from the review app, with its OWN webhook secret and its OWN
+-- delivery IDs, so it gets its OWN dedup table (not webhook_events) — a GitHub delivery_id is only unique per
+-- App, so sharing one table across two Apps could collide. This receiver just verifies + records (PR1);
+-- install-registry + PR-outcome processing land in later PRs.
+CREATE TABLE IF NOT EXISTS orb_webhook_events (
+  delivery_id TEXT PRIMARY KEY NOT NULL,
+  event_name TEXT NOT NULL,
+  action TEXT,
+  installation_id INTEGER,
+  repository_full_name TEXT,
+  payload_hash TEXT NOT NULL,
+  -- 'received' (recorded, not yet processed) | 'processed' | 'error'. Processing is added in a later PR.
+  status TEXT NOT NULL DEFAULT 'received',
+  received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  processed_at TEXT
+);
+
+CREATE INDEX IF NOT EXISTS orb_webhook_events_status_idx ON orb_webhook_events(status);
+CREATE INDEX IF NOT EXISTS orb_webhook_events_installation_idx ON orb_webhook_events(installation_id);

src/api/routes.ts

@@ -123,6 +123,7 @@ import {
 } from "../github/commands";
 import { handleGitHubWebhook } from "../github/webhook";
 import { handleOrbIngest, readOrbIngestBody } from "../orb/ingest";
+import { handleOrbWebhook } from "../orb/webhook";
 import { computeFleetAnalytics } from "../orb/analytics";
 import { handleMcpRequest } from "../mcp/server";
 import { buildOpenApiSpec } from "../openapi/spec";
@@ -2864,6 +2865,11 @@ export function createApp() {
 
   app.post("/v1/github/webhook", handleGitHubWebhook);
 
+  // Gittensory Orb central GitHub App (#1255) — inbound webhook for the ONE shared Orb App maintainers install.
+  // Verifies the Orb App's OWN webhook secret, dedups, and records install + PR/review events (the homepage
+  // fleet-metrics data spine). Separate App + secret from the review-app /v1/github/webhook above.
+  app.post("/v1/orb/webhook", handleOrbWebhook);
+
   // Gittensory Orb (#1255) — central fleet-calibration collector. Receives anonymized, reversal-aware
   // outcome batches from self-hosted instances. No auth required: all data is HMAC-anonymized by the sender;
   // dedup is enforced via UNIQUE(instance_id, repo_hash, pr_hash) in orb_signals. Rate-limited (strict, #1254).
@@ -4866,6 +4872,7 @@ function requiresApiToken(path: string): boolean {
   if (path === "/v1/drafts" || path.startsWith("/v1/drafts/")) return false;
   if (path.startsWith("/v1/auth/")) return false;
   if (path === "/v1/github/webhook") return false;
+  if (path === "/v1/orb/webhook") return false;
   if (path === "/v1/orb/ingest") return false;
   if (path.startsWith("/v1/internal/")) return false;
   return path.startsWith("/v1/");

src/auth/rate-limit.ts

@@ -96,6 +96,9 @@ export async function enforceRateLimit(c: Context<{ Bindings: Env }>, routeClass
 
 export function routeClassForPath(path: string): RateLimitClass {
   if (path === "/v1/github/webhook") return "strict";
+  // Orb central-App inbound webhook — same class as the review-app webhook above (GitHub delivers from a
+  // narrow IP range; the per-IP strict cap is proven for /v1/github/webhook and #1292 reserves headroom).
+  if (path === "/v1/orb/webhook") return "strict";
   // Orb telemetry ingest: unauthenticated + write, accepting anonymized batches from untrusted
   // self-host instances. Strict (10/min per IP) caps abuse — legitimate instances export hourly.
   if (path === "/v1/orb/ingest") return "strict";

src/env.d.ts

@@ -38,6 +38,9 @@ declare global {
     ADMIN_GITHUB_LOGINS?: string;
     GITHUB_WEBHOOK_SECRET: string;
     GITHUB_WEBHOOK_MAX_BODY_BYTES?: string;
+    /** Webhook secret for the central Gittensory Orb GitHub App (#1255) — distinct from the review app's
+     *  GITHUB_WEBHOOK_SECRET. Verifies inbound POST /v1/orb/webhook deliveries. Inject as a wrangler secret. */
+    ORB_GITHUB_WEBHOOK_SECRET?: string;
     GITHUB_APP_PRIVATE_KEY: string;
     GITHUB_APP_ID: string;
     GITHUB_APP_SLUG: string;

src/orb/webhook.ts

@@ -0,0 +1,114 @@
+// Gittensory Orb central GitHub App (#1255) — inbound webhook receiver (POST /v1/orb/webhook).
+//
+// The central Orb App is a SEPARATE GitHub App that maintainers INSTALL (one shared app, like
+// das-github-mirror's). GitHub delivers its install + PR/review events here, to gittensory-api. This is the
+// data spine for the homepage fleet metrics (reviews initiated / merged / closed / reversals).
+//
+// PR1 scope: receive + verify (the Orb App's OWN webhook secret) + dedup + record. NO processing yet — the
+// install registry and PR-outcome aggregation land in later PRs, reading from orb_webhook_events. This mirrors
+// the proven src/github/webhook.ts handler verbatim; only the secret + dedup table differ.
+import type { Context } from "hono";
+import type { GitHubWebhookPayload } from "../types";
+import { sha256Hex, verifyGitHubSignature } from "../utils/crypto";
+
+const DEFAULT_MAX_ORB_WEBHOOK_BODY_BYTES = 1024 * 1024;
+
+export async function handleOrbWebhook(c: Context<{ Bindings: Env }>): Promise<Response> {
+  const deliveryId = c.req.header("x-github-delivery") ?? null;
+  const eventName = c.req.header("x-github-event") ?? null;
+  const signature = c.req.header("x-hub-signature-256") ?? null;
+  if (!deliveryId || !eventName) {
+    return c.json({ error: "missing_github_headers" }, 400);
+  }
+
+  const maxBodyBytes = parsePositiveInt(c.env.GITHUB_WEBHOOK_MAX_BODY_BYTES) ?? DEFAULT_MAX_ORB_WEBHOOK_BODY_BYTES;
+  const contentLength = parsePositiveInt(c.req.header("content-length"));
+  if (contentLength !== null && contentLength > maxBodyBytes) {
+    return c.json({ error: "payload_too_large", maxBytes: maxBodyBytes }, 413);
+  }
+
+  const rawBody = await readBodyWithLimit(c.req.raw, maxBodyBytes);
+  if (rawBody === null) {
+    return c.json({ error: "payload_too_large", maxBytes: maxBodyBytes }, 413);
+  }
+  // The Orb App's OWN webhook secret — distinct from the review app's GITHUB_WEBHOOK_SECRET. Absent secret →
+  // verifyGitHubSignature returns false → 401 (fail-closed), so this route is inert until the secret is injected.
+  const verified = await verifyGitHubSignature(rawBody, signature, c.env.ORB_GITHUB_WEBHOOK_SECRET ?? "");
+  if (!verified) {
+    return c.json({ error: "invalid_signature" }, 401);
+  }
+
+  let payload: GitHubWebhookPayload;
+  try {
+    payload = JSON.parse(rawBody) as GitHubWebhookPayload;
+  } catch {
+    return c.json({ error: "invalid_json" }, 400);
+  }
+
+  const payloadHash = await sha256Hex(rawBody);
+  const existing = await getOrbWebhookEvent(c.env, deliveryId);
+  // Suppress redelivery of an already-recorded delivery (same payload) or a processed one; "error" rows are
+  // never suppressed so a failed record can be retried — same semantics as the review-app handler (#789).
+  if (existing && existing.status !== "error" && (existing.status === "processed" || existing.payloadHash === payloadHash)) {
+    return c.json({ ok: true, deliveryId, eventName, status: "duplicate" }, 202);
+  }
+
+  await recordOrbWebhookEvent(c.env, {
+    deliveryId,
+    eventName,
+    action: payload.action ?? null,
+    installationId: payload.installation?.id ?? null,
+    repositoryFullName: payload.repository?.full_name ?? null,
+    payloadHash,
+  });
+
+  return c.json({ ok: true, deliveryId, eventName, status: "received" }, 202);
+}
+
+async function getOrbWebhookEvent(env: Env, deliveryId: string): Promise<{ payloadHash: string; status: string } | null> {
+  const row = await env.DB.prepare("SELECT payload_hash AS payloadHash, status FROM orb_webhook_events WHERE delivery_id = ?")
+    .bind(deliveryId)
+    .first<{ payloadHash: string; status: string }>();
+  return row ?? null;
+}
+
+async function recordOrbWebhookEvent(
+  env: Env,
+  e: { deliveryId: string; eventName: string; action: string | null; installationId: number | null; repositoryFullName: string | null; payloadHash: string },
+): Promise<void> {
+  await env.DB.prepare(
+    `INSERT INTO orb_webhook_events (delivery_id, event_name, action, installation_id, repository_full_name, payload_hash, status)
+     VALUES (?, ?, ?, ?, ?, ?, 'received')
+     ON CONFLICT(delivery_id) DO UPDATE SET
+       status = 'received', payload_hash = excluded.payload_hash, action = excluded.action,
+       installation_id = excluded.installation_id, repository_full_name = excluded.repository_full_name`,
+  )
+    .bind(e.deliveryId, e.eventName, e.action, e.installationId, e.repositoryFullName, e.payloadHash)
+    .run();
+}
+
+function parsePositiveInt(value: string | null | undefined): number | null {
+  if (!value) return null;
+  const parsed = Number.parseInt(value, 10);
+  if (!Number.isFinite(parsed) || parsed <= 0) return null;
+  return parsed;
+}
+
+async function readBodyWithLimit(request: Request, maxBytes: number): Promise<string | null> {
+  const stream = request.body;
+  if (!stream) return "";
+  const reader = stream.getReader();
+  const decoder = new TextDecoder();
+  const chunks: string[] = [];
+  let total = 0;
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    if (!value) continue;
+    total += value.byteLength;
+    if (total > maxBytes) return null;
+    chunks.push(decoder.decode(value, { stream: true }));
+  }
+  chunks.push(decoder.decode());
+  return chunks.join("");
+}

test/integration/orb-webhook.test.ts

@@ -0,0 +1,161 @@
+import type { Context } from "hono";
+import { describe, expect, it } from "vitest";
+import { createApp } from "../../src/api/routes";
+import { handleOrbWebhook } from "../../src/orb/webhook";
+import { createTestEnv, type TestD1Database } from "../helpers/d1";
+
+const SECRET = "orb-test-secret";
+const env = (over: Record<string, string> = {}): Env => createTestEnv({ ORB_GITHUB_WEBHOOK_SECRET: SECRET, ...over });
+
+async function sign(body: string, secret: string): Promise<string> {
+  const key = await crypto.subtle.importKey("raw", new TextEncoder().encode(secret), { name: "HMAC", hash: "SHA-256" }, false, ["sign"]);
+  const signed = await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(body));
+  return `sha256=${[...new Uint8Array(signed)].map((b) => b.toString(16).padStart(2, "0")).join("")}`;
+}
+
+function ctx(e: Env, headers: Record<string, string | null>, request: Request): Context<{ Bindings: Env }> {
+  return {
+    req: { raw: request, header: (n: string) => headers[n.toLowerCase()] ?? null },
+    env: e,
+    json: (payload: unknown, status?: number) => Response.json(payload, status === undefined ? undefined : { status }),
+  } as unknown as Context<{ Bindings: Env }>;
+}
+
+async function post(
+  e: Env,
+  body: string,
+  opts: { delivery?: string | null; event?: string | null; sig?: string; headers?: Record<string, string | null>; request?: Request } = {},
+): Promise<Response> {
+  const headers: Record<string, string | null> = {
+    "x-github-delivery": opts.delivery === undefined ? "d1" : opts.delivery,
+    "x-github-event": opts.event === undefined ? "installation" : opts.event,
+    "x-hub-signature-256": opts.sig ?? (await sign(body, SECRET)),
+    ...opts.headers,
+  };
+  const request = opts.request ?? new Request("https://collector/v1/orb/webhook", { method: "POST", body });
+  return handleOrbWebhook(ctx(e, headers, request));
+}
+
+const INSTALL = JSON.stringify({ action: "created", installation: { id: 42 }, repository: { full_name: "JSONbored/gittensory" } });
+const row = (e: Env, delivery: string) =>
+  (e.DB as unknown as TestD1Database).prepare("SELECT event_name, action, installation_id, repository_full_name, status FROM orb_webhook_events WHERE delivery_id=?").bind(delivery).first<{ event_name: string; action: string; installation_id: number; repository_full_name: string; status: string }>();
+
+describe("handleOrbWebhook (POST /v1/orb/webhook)", () => {
+  it("400 when the GitHub delivery or event header is missing", async () => {
+    expect((await post(env(), INSTALL, { delivery: null as unknown as string })).status).toBe(400);
+    expect((await post(env(), INSTALL, { event: null as unknown as string })).status).toBe(400);
+  });
+
+  it("401 on an invalid signature (and 401 when the Orb secret is absent — fail-closed)", async () => {
+    expect((await post(env(), INSTALL, { sig: "sha256=deadbeef" })).status).toBe(401);
+    const noSecret = createTestEnv(); // ORB_GITHUB_WEBHOOK_SECRET unset
+    expect((await post(noSecret, INSTALL)).status).toBe(401);
+  });
+
+  it("401 when the signature header is absent entirely", async () => {
+    expect((await post(env(), INSTALL, { headers: { "x-hub-signature-256": null } })).status).toBe(401);
+  });
+
+  it("ignores a non-numeric content-length and processes normally", async () => {
+    const res = await post(env(), INSTALL, { delivery: "cl-abc", headers: { "content-length": "abc" } });
+    expect(res.status).toBe(202);
+  });
+
+  it("413 when content-length exceeds the cap", async () => {
+    const res = await post(env(), INSTALL, { headers: { "content-length": "99999999" } });
+    expect(res.status).toBe(413);
+  });
+
+  it("413 when the streamed body exceeds the cap (no content-length declared)", async () => {
+    const res = await post(env({ GITHUB_WEBHOOK_MAX_BODY_BYTES: "16" }), "x".repeat(40));
+    expect(res.status).toBe(413);
+  });
+
+  it("400 on a signed-but-non-JSON body", async () => {
+    expect((await post(env(), "not json")).status).toBe(400);
+  });
+
+  it("202 + records the install event (action/installation/repo extracted)", async () => {
+    const e = env();
+    const res = await post(e, INSTALL, { delivery: "ok-1" });
+    expect(res.status).toBe(202);
+    await expect(res.json()).resolves.toMatchObject({ status: "received", eventName: "installation" });
+    expect(await row(e, "ok-1")).toMatchObject({ action: "created", installation_id: 42, repository_full_name: "JSONbored/gittensory", status: "received" });
+  });
+
+  it("stores null fields for a payload with no action/installation/repository (e.g. ping)", async () => {
+    const e = env();
+    await post(e, JSON.stringify({ zen: "keep it logically awesome" }), { delivery: "ping-1", event: "ping" });
+    expect(await row(e, "ping-1")).toMatchObject({ action: null, installation_id: null, repository_full_name: null });
+  });
+
+  it("dedups a redelivery of the same (delivery, payload) as a duplicate", async () => {
+    const e = env();
+    expect((await post(e, INSTALL, { delivery: "dup-1" })).status).toBe(202);
+    const second = await post(e, INSTALL, { delivery: "dup-1" });
+    expect(second.status).toBe(202);
+    await expect(second.json()).resolves.toMatchObject({ status: "duplicate" });
+  });
+
+  it("re-records a redelivery whose payload CHANGED (same delivery id, different hash)", async () => {
+    const e = env();
+    await post(e, INSTALL, { delivery: "chg-1" });
+    const changed = JSON.stringify({ action: "deleted", installation: { id: 42 } });
+    const res = await post(e, changed, { delivery: "chg-1" });
+    await expect(res.json()).resolves.toMatchObject({ status: "received" }); // not a duplicate
+    expect((await row(e, "chg-1"))?.action).toBe("deleted");
+  });
+
+  it("treats an already-processed delivery as a duplicate regardless of payload", async () => {
+    const e = env();
+    await (e.DB as unknown as TestD1Database)
+      .prepare("INSERT INTO orb_webhook_events (delivery_id, event_name, payload_hash, status) VALUES ('proc-1','installation','oldhash','processed')")
+      .run();
+    const res = await post(e, INSTALL, { delivery: "proc-1" });
+    await expect(res.json()).resolves.toMatchObject({ status: "duplicate" });
+  });
+
+  it("does NOT suppress an 'error' row — the delivery is retried", async () => {
+    const e = env();
+    await (e.DB as unknown as TestD1Database)
+      .prepare("INSERT INTO orb_webhook_events (delivery_id, event_name, payload_hash, status) VALUES ('err-1','installation','oldhash','error')")
+      .run();
+    const res = await post(e, INSTALL, { delivery: "err-1" });
+    await expect(res.json()).resolves.toMatchObject({ status: "received" }); // re-recorded, not suppressed
+    expect((await row(e, "err-1"))?.status).toBe("received");
+  });
+
+  it("treats a missing request body as empty (→ 400 invalid JSON after a valid empty-body signature)", async () => {
+    const e = env();
+    const sig = await sign("", SECRET);
+    const res = await post(e, "", { request: new Request("https://collector/v1/orb/webhook", { method: "POST" }), sig });
+    expect(res.status).toBe(400); // empty body verifies, then fails JSON.parse
+  });
+
+  it("skips undefined stream chunks while reading the body", async () => {
+    const e = env();
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(undefined as unknown as Uint8Array);
+        controller.close();
+      },
+    });
+    const request = { body } as unknown as Request;
+    const res = await post(e, "", { request, sig: "sha256=bad" });
+    expect(res.status).toBe(401); // empty (undefined-skipped) body + bad sig
+  });
+});
+
+describe("POST /v1/orb/webhook route (through the app middleware)", () => {
+  const app = createApp();
+
+  it("is token-exempt + rate-classified, routing to the handler (401 on a bad signature)", async () => {
+    // Exercises requiresApiToken (exempt) + routeClassForPath (strict) for the new path, then the handler.
+    const res = await app.request(
+      "/v1/orb/webhook",
+      { method: "POST", headers: { "x-github-delivery": "rt-1", "x-github-event": "ping", "x-hub-signature-256": "sha256=bad" }, body: "{}" },
+      createTestEnv({ ORB_GITHUB_WEBHOOK_SECRET: "s" }),
+    );
+    expect(res.status).toBe(401);
+  });
+});