feat(orb): forward registered installs' events to the brokered self-host

src/orb/relay.ts

@@ -6,7 +6,28 @@
 // (the encryption key is a separate secret).
 import { hashToken } from "../auth/security";
 import { isSafeHttpUrl } from "../review/content-lane/safe-url";
-import { encryptSecret } from "../utils/crypto";
+import { decryptSecret, encryptSecret } from "../utils/crypto";
+
+// The events a brokered container needs to review/act on. Installation-lifecycle + other Orb-internal events are
+// deliberately NOT forwarded (the container runs under the CENTRAL Orb App, not its own, so it must not treat
+// those as its own installation state).
+const RELAY_FORWARD_EVENTS = new Set([
+  "pull_request",
+  "pull_request_review",
+  "pull_request_review_comment",
+  "check_run",
+  "check_suite",
+  "issue_comment",
+  "issues",
+]);
+
+/** HMAC-SHA256 hex over the raw event body — the relay signature BOTH sides compute (the Orb with the decrypted
+ *  enrollment secret, the container with its own ORB_ENROLLMENT_SECRET). Web Crypto (worker + node). */
+export async function relaySignature(secret: string, body: string): Promise<string> {
+  const key = await crypto.subtle.importKey("raw", new TextEncoder().encode(secret), { name: "HMAC", hash: "SHA-256" }, false, ["sign"]);
+  const sig = await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(body));
+  return [...new Uint8Array(sig)].map((b) => b.toString(16).padStart(2, "0")).join("");
+}
 
 export type RegisterResult =
   | { ok: true; installationId: number }
@@ -38,3 +59,40 @@ export async function registerOrbRelay(env: Env, secret: string, relayUrl: strin
     .run();
   return { ok: true, installationId: row.installation_id };
 }
+
+/** Forward a webhook event to the brokered self-host registered for this installation. BEST-EFFORT + fail-safe:
+ *  a non-forwardable event, no registered relay, or ANY error returns without throwing (the Orb's webhook 202
+ *  stands; reliability hardening — a retry queue for a down container — is a follow-up). The body is HMAC-signed
+ *  with the container's enrollment secret (decrypted from the stored ciphertext); the container verifies with its
+ *  own ORB_ENROLLMENT_SECRET, so only the genuine Orb can drive it. */
+export async function forwardOrbEvent(
+  env: Env,
+  args: { eventName: string; installationId: number | null | undefined; deliveryId: string; rawBody: string },
+  fetchImpl: typeof fetch = fetch,
+): Promise<"forwarded" | "skipped" | "failed"> {
+  if (!args.installationId || !RELAY_FORWARD_EVENTS.has(args.eventName)) return "skipped";
+  const row = await env.DB
+    .prepare("SELECT relay_url, relay_secret_enc, relay_secret_iv, relay_secret_salt FROM orb_enrollments WHERE installation_id = ? AND state = 'enrolled' AND revoked_at IS NULL AND relay_url IS NOT NULL")
+    .bind(args.installationId)
+    .first<{ relay_url: string; relay_secret_enc: string; relay_secret_iv: string; relay_secret_salt: string | null }>();
+  if (!row || !env.TOKEN_ENCRYPTION_SECRET) return "skipped";
+  try {
+    const secret = await decryptSecret(row.relay_secret_enc, row.relay_secret_iv, env.TOKEN_ENCRYPTION_SECRET, row.relay_secret_salt);
+    const signature = await relaySignature(secret, args.rawBody);
+    const res = await fetchImpl(row.relay_url, {
+      method: "POST",
+      headers: {
+        "content-type": "application/json",
+        "x-github-event": args.eventName,
+        "x-github-delivery": args.deliveryId,
+        "x-orb-signature-256": `sha256=${signature}`,
+        "user-agent": "gittensory-orb/0.1",
+      },
+      body: args.rawBody,
+      signal: AbortSignal.timeout(10_000),
+    });
+    return res.ok ? "forwarded" : "failed";
+  } catch {
+    return "failed"; // a down / unreachable container (or a decrypt/sign error) must never fail the Orb's 202
+  }
+}

src/orb/webhook.ts

@@ -12,6 +12,7 @@ import type { GitHubWebhookPayload } from "../types";
 import { sha256Hex, verifyGitHubSignature } from "../utils/crypto";
 import { upsertOrbInstallation } from "./installations";
 import { recordOrbPrOutcome } from "./outcomes";
+import { forwardOrbEvent } from "./relay";
 
 const DEFAULT_MAX_ORB_WEBHOOK_BODY_BYTES = 1024 * 1024;
 
@@ -77,6 +78,9 @@ export async function handleOrbWebhook(c: Context<{ Bindings: Env }>): Promise<R
   }
 
   await recordOrbWebhookEvent(c.env, { ...eventMeta, status: "received" });
+  // Forward the event to a brokered self-host registered for this installation (best-effort, fail-safe — a down
+  // container never fails the 202; a non-forwardable event / no registered relay is a fast no-op).
+  await forwardOrbEvent(c.env, { eventName, installationId: payload.installation?.id, deliveryId, rawBody });
   return c.json({ ok: true, deliveryId, eventName, status: "received" }, 202);
 }
 

test/integration/orb-relay.test.ts

@@ -1,7 +1,7 @@
 import { describe, expect, it } from "vitest";
 import { createApp } from "../../src/api/routes";
 import { issueOrbEnrollment } from "../../src/orb/broker";
-import { registerOrbRelay } from "../../src/orb/relay";
+import { forwardOrbEvent, registerOrbRelay, relaySignature } from "../../src/orb/relay";
 import { createTestEnv, type TestD1Database } from "../helpers/d1";
 
 const db = (e: Env) => e.DB as unknown as TestD1Database;
@@ -93,3 +93,61 @@ describe("POST /v1/orb/relay/register", () => {
     expect((await app.request("/v1/orb/relay/register", { method: "POST", headers: { authorization: `Bearer ${s3}` }, body: JSON.stringify({ relayUrl: "https://x.example/relay" }) }, noEnc)).status).toBe(500);
   });
 });
+
+describe("relaySignature", () => {
+  it("is a deterministic 64-hex HMAC both sides can recompute (and key-dependent)", async () => {
+    expect(await relaySignature("s", "body")).toBe(await relaySignature("s", "body"));
+    expect(await relaySignature("s", "body")).not.toBe(await relaySignature("other", "body"));
+    expect(await relaySignature("s", "body")).toMatch(/^[0-9a-f]{64}$/);
+  });
+});
+
+describe("forwardOrbEvent", () => {
+  const capture = (resp: Response) => {
+    const calls: { url: string; init?: RequestInit | undefined }[] = [];
+    const fetchImpl = ((u: RequestInfo | URL, init?: RequestInit) => {
+      calls.push({ url: String(u), init });
+      return Promise.resolve(resp);
+    }) as typeof fetch;
+    return { fetchImpl, calls };
+  };
+
+  it("SKIPS a non-forwardable event, a missing installation, and an enrolled install with no relay registered", async () => {
+    const e = brokeredEnv();
+    expect(await forwardOrbEvent(e, { eventName: "installation", installationId: 1, deliveryId: "d", rawBody: "{}" })).toBe("skipped");
+    expect(await forwardOrbEvent(e, { eventName: "pull_request", installationId: null, deliveryId: "d", rawBody: "{}" })).toBe("skipped");
+    await enroll(e, 801);
+    expect(await forwardOrbEvent(e, { eventName: "pull_request", installationId: 801, deliveryId: "d", rawBody: "{}" })).toBe("skipped"); // enrolled, no relay
+  });
+
+  it("FORWARDS a registered install's event, HMAC-signed with the container's secret (the container can verify)", async () => {
+    const e = brokeredEnv();
+    const secret = await enroll(e, 800);
+    await registerOrbRelay(e, secret, "https://c.example/v1/orb/relay");
+    const { fetchImpl, calls } = capture(new Response("ok"));
+    const body = '{"action":"opened","number":7}';
+    expect(await forwardOrbEvent(e, { eventName: "pull_request", installationId: 800, deliveryId: "del-1", rawBody: body }, fetchImpl)).toBe("forwarded");
+    expect(calls[0]?.url).toBe("https://c.example/v1/orb/relay");
+    const h = calls[0]?.init?.headers as Record<string, string>;
+    expect(h["x-github-event"]).toBe("pull_request");
+    expect(h["x-github-delivery"]).toBe("del-1");
+    expect(h["x-orb-signature-256"]).toBe(`sha256=${await relaySignature(secret, body)}`); // matches what the container recomputes
+    expect(calls[0]?.init?.body).toBe(body);
+  });
+
+  it("returns FAILED (never throws) on a non-ok response or a thrown fetch — the Orb 202 always stands", async () => {
+    const e = brokeredEnv();
+    const secret = await enroll(e, 802);
+    await registerOrbRelay(e, secret, "https://c.example/v1/orb/relay");
+    expect(await forwardOrbEvent(e, { eventName: "pull_request", installationId: 802, deliveryId: "d", rawBody: "{}" }, (() => Promise.resolve(new Response("no", { status: 503 }))) as typeof fetch)).toBe("failed");
+    expect(await forwardOrbEvent(e, { eventName: "pull_request", installationId: 802, deliveryId: "d", rawBody: "{}" }, (() => Promise.reject(new Error("down"))) as typeof fetch)).toBe("failed");
+  });
+
+  it("SKIPS when the server's encryption secret is gone (can't decrypt the stored secret)", async () => {
+    const e = brokeredEnv();
+    const secret = await enroll(e, 803);
+    await registerOrbRelay(e, secret, "https://c.example/v1/orb/relay");
+    const noKey = { ...e, TOKEN_ENCRYPTION_SECRET: undefined } as unknown as Env; // same DB, key removed
+    expect(await forwardOrbEvent(noKey, { eventName: "pull_request", installationId: 803, deliveryId: "d", rawBody: "{}" })).toBe("skipped");
+  });
+});