feat(orb): pull-mode relay delivery for NAT/tailnet self-hosts

src/github/webhook.ts

@@ -35,60 +35,66 @@ export async function handleGitHubWebhook(c: Context<{ Bindings: Env }>): Promis
  *  webhook receiver above AND the Orb relay receiver below (they verify the body differently — GitHub's HMAC vs the
  *  Orb relay HMAC — then share everything after). */
 export async function enqueueVerifiedWebhook(c: Context<{ Bindings: Env }>, deliveryId: string, eventName: string, rawBody: string): Promise<Response> {
+  const result = await enqueueWebhookByEnv(c.env, deliveryId, eventName, rawBody);
+  switch (result) {
+    case "invalid_json":
+      return c.json({ error: "invalid_json" }, 400);
+    case "duplicate":
+      return c.json({ ok: true, deliveryId, eventName, status: "duplicate" }, 202);
+    case "enqueue_failed":
+      return c.json({ error: "enqueue_failed", deliveryId }, 500);
+    default:
+      return c.json({ ok: true, deliveryId, eventName, status: "queued" }, 202);
+  }
+}
+
+export type EnqueueWebhookResult = "queued" | "duplicate" | "invalid_json" | "enqueue_failed";
+
+/** Env-based core of the webhook enqueue (parse → dedup → record → WEBHOOKS lane), with NO Hono Context. Shared by
+ *  the request-context receiver above AND the pull-mode relay drain loop (server.ts), which has no Context. Returns
+ *  a status the caller maps to a response / an ack decision. */
+export async function enqueueWebhookByEnv(env: Env, deliveryId: string, eventName: string, rawBody: string): Promise<EnqueueWebhookResult> {
   let payload: GitHubWebhookPayload;
   try {
     payload = JSON.parse(rawBody) as GitHubWebhookPayload;
   } catch {
-    return c.json({ error: "invalid_json" }, 400);
+    return "invalid_json";
   }
 
   const payloadHash = await sha256Hex(rawBody);
-  const existingEvent = await getWebhookEvent(c.env, deliveryId);
+  const existingEvent = await getWebhookEvent(env, deliveryId);
   // Suppress redelivery of an already-processed event (on success its payloadHash is overwritten to a
   // "processed" sentinel, so a hash match alone misses it and the event re-runs its side effects) or one
   // still in flight with the same payload. "error" rows are never suppressed so a failed enqueue/processing
   // can still be retried (#789).
   if (existingEvent && existingEvent.status !== "error" && (existingEvent.status === "processed" || existingEvent.payloadHash === payloadHash)) {
-    return c.json({ ok: true, deliveryId, eventName, status: "duplicate" }, 202);
+    return "duplicate";
   }
 
-  await recordWebhookEvent(c.env, {
+  const eventRow = {
     deliveryId,
     eventName,
     action: payload.action,
     installationId: payload.installation?.id,
     repositoryFullName: payload.repository?.full_name,
     payloadHash,
-    status: "queued",
-  });
-
-  const message: JobMessage = {
-    type: "github-webhook",
-    deliveryId,
-    eventName,
-    payload,
   };
+  await recordWebhookEvent(env, { ...eventRow, status: "queued" });
+
+  const message: JobMessage = { type: "github-webhook", deliveryId, eventName, payload };
   try {
     // Send to the dedicated WEBHOOKS lane (not the shared JOBS queue) so a maintenance burst on JOBS can never
     // starve real GitHub events into the DLQ. (#audit-webhook-queue)
-    await c.env.WEBHOOKS.send(message);
+    await env.WEBHOOKS.send(message);
   } catch {
-    // Enqueue failed: flip the event to "error" so the dedup guard above lets GitHub redeliver,
-    // and return 500 so GitHub retries instead of treating the webhook as handled (#786). This also covers the
-    // deploy-ordering case where the WEBHOOKS queue is not yet provisioned — no event is lost.
-    await recordWebhookEvent(c.env, {
-      deliveryId,
-      eventName,
-      action: payload.action,
-      installationId: payload.installation?.id,
-      repositoryFullName: payload.repository?.full_name,
-      payloadHash,
-      status: "error",
-    });
-    return c.json({ error: "enqueue_failed", deliveryId }, 500);
+    // Enqueue failed: flip the event to "error" so the dedup guard above lets GitHub redeliver / the next pull
+    // re-deliver, instead of treating the webhook as handled (#786). Also covers the deploy-ordering case where
+    // the WEBHOOKS queue is not yet provisioned — no event is lost.
+    await recordWebhookEvent(env, { ...eventRow, status: "error" });
+    return "enqueue_failed";
   }
 
-  return c.json({ ok: true, deliveryId, eventName, status: "queued" }, 202);
+  return "queued";
 }
 
 /** The brokered self-host's relay RECEIVER. The central Orb forwards an event here, HMAC-signed (x-orb-signature-

src/orb/broker-client.ts

@@ -81,21 +81,57 @@ export async function fetchBrokeredInstallationToken(
  *  relay until the next boot — it never blocks startup or throws. The relay URL is the container's public origin +
  *  /v1/orb/relay (the receiver); the Orb SSRF-validates it, so PUBLIC_API_ORIGIN must be a real public https host. */
 export async function registerOrbRelayTarget(
-  env: { ORB_ENROLLMENT_SECRET?: string | undefined; ORB_BROKER_URL?: string | undefined; PUBLIC_API_ORIGIN?: string | undefined },
+  env: { ORB_ENROLLMENT_SECRET?: string | undefined; ORB_BROKER_URL?: string | undefined; PUBLIC_API_ORIGIN?: string | undefined; ORB_RELAY_MODE?: string | undefined },
   fetchImpl: typeof fetch = fetch,
 ): Promise<"registered" | "skipped" | "failed"> {
-  if (!isOrbBrokerMode(env) || !env.PUBLIC_API_ORIGIN) return "skipped";
-  const relayUrl = `${env.PUBLIC_API_ORIGIN.replace(/\/+$/, "")}/v1/orb/relay`;
+  if (!isOrbBrokerMode(env)) return "skipped";
+  // Pull mode (#secure-relay): the engine DRAINS events outbound from the Orb, so NO inbound endpoint is exposed —
+  // the right fit for a NAT/tailnet self-host (a public push URL would otherwise be unreachable). Push mode needs a
+  // public relay URL the Orb can reach.
+  const mode = env.ORB_RELAY_MODE === "pull" ? "pull" : "push";
+  if (mode === "push" && !env.PUBLIC_API_ORIGIN) return "skipped";
+  const relayUrl = mode === "push" ? `${env.PUBLIC_API_ORIGIN!.replace(/\/+$/, "")}/v1/orb/relay` : "";
   try {
     const base = orbBrokerBaseUrl(env);
     const res = await fetchImpl(`${base}/v1/orb/relay/register`, {
       method: "POST",
       headers: { authorization: `Bearer ${env.ORB_ENROLLMENT_SECRET}`, "content-type": "application/json" }, // present — isOrbBrokerMode required it
-      body: JSON.stringify({ relayUrl }),
+      body: JSON.stringify({ relayUrl, mode }),
       signal: AbortSignal.timeout(10_000),
     });
     return res.ok ? "registered" : "failed";
   } catch {
     return "failed";
   }
 }
+
+/** Pull-mode drain (#secure-relay): fetch this install's queued events from the Orb, acking the previous batch's
+ *  delivery ids so the Orb deletes them. Lets a NAT/tailnet engine receive events WITHOUT exposing an inbound
+ *  endpoint. BEST-EFFORT — returns [] on a non-broker / unsafe-URL / non-ok / thrown case; the next tick retries. */
+export async function drainOrbRelay(
+  env: { ORB_ENROLLMENT_SECRET?: string | undefined; ORB_BROKER_URL?: string | undefined },
+  ack: string[] = [],
+  fetchImpl: typeof fetch = fetch,
+): Promise<{ deliveryId: string; eventName: string; rawBody: string }[]> {
+  if (!isOrbBrokerMode(env)) return [];
+  try {
+    const base = orbBrokerBaseUrl(env);
+    const res = await fetchImpl(`${base}/v1/orb/relay/pull`, {
+      method: "POST",
+      headers: { authorization: `Bearer ${env.ORB_ENROLLMENT_SECRET}`, "content-type": "application/json" },
+      body: JSON.stringify({ ack }),
+      signal: AbortSignal.timeout(15_000),
+    });
+    if (!res.ok) return [];
+    const body = (await res.json()) as { events?: Array<{ deliveryId?: unknown; eventName?: unknown; rawBody?: unknown }> };
+    const out: { deliveryId: string; eventName: string; rawBody: string }[] = [];
+    for (const e of body.events ?? []) {
+      if (typeof e.deliveryId === "string" && typeof e.eventName === "string" && typeof e.rawBody === "string") {
+        out.push({ deliveryId: e.deliveryId, eventName: e.eventName, rawBody: e.rawBody });
+      }
+    }
+    return out;
+  } catch {
+    return [];
+  }
+}

src/server.ts

@@ -735,12 +735,14 @@ async function main(): Promise<void> {
   void runOrbExport(); // flush any pending events at startup
   setInterval(runOrbExport, 3_600_000); // then hourly
 
-  // Brokered self-host: register our public relay URL with the central Orb so it forwards this install's events
-  // here (best-effort, fire-and-forget — a no-op unless ORB_ENROLLMENT_SECRET + PUBLIC_API_ORIGIN are set).
+  // Brokered self-host: register our relay target with the central Orb (best-effort, fire-and-forget). PUSH mode
+  // (default) registers a public relay URL the Orb POSTs to; PULL mode (ORB_RELAY_MODE=pull) registers no URL and
+  // the drain loop below pulls events outbound — the right fit behind NAT/tailnet (no inbound endpoint exposed).
   void registerOrbRelayTarget({
     ORB_ENROLLMENT_SECRET: process.env.ORB_ENROLLMENT_SECRET,
     ORB_BROKER_URL: process.env.ORB_BROKER_URL,
     PUBLIC_API_ORIGIN: process.env.PUBLIC_API_ORIGIN,
+    ORB_RELAY_MODE: process.env.ORB_RELAY_MODE,
   })
     .then((r) => {
       if (r === "registered") {
@@ -753,6 +755,32 @@ async function main(): Promise<void> {
     })
     .catch((error) => captureError(error, { kind: "orb_relay_register" }));
 
+  // Pull-mode relay drain (#secure-relay): when ORB_RELAY_MODE=pull, the engine DRAINS its events from the Orb on a
+  // timer instead of exposing an inbound endpoint — the right fit behind NAT/tailnet. Acks the previous batch so the
+  // Orb deletes delivered events; best-effort (a failed tick retries next interval). Each event enqueues into the
+  // same WEBHOOKS lane the push receiver uses.
+  if (process.env.ORB_RELAY_MODE === "pull" && process.env.ORB_ENROLLMENT_SECRET) {
+    const { drainOrbRelay } = await import("./orb/broker-client");
+    const { enqueueWebhookByEnv } = await import("./github/webhook");
+    let pendingAck: string[] = [];
+    const drainRelay = async (): Promise<void> => {
+      const events = await drainOrbRelay(
+        { ORB_ENROLLMENT_SECRET: process.env.ORB_ENROLLMENT_SECRET, ORB_BROKER_URL: process.env.ORB_BROKER_URL },
+        pendingAck,
+      );
+      pendingAck = [];
+      for (const ev of events) {
+        const result = await enqueueWebhookByEnv(env, ev.deliveryId, ev.eventName, ev.rawBody);
+        // Ack everything durably handled (queued / duplicate / invalid_json) so the Orb deletes it; retry only a
+        // real enqueue failure on the next pull (don't ack → the Orb keeps it).
+        if (result !== "enqueue_failed") pendingAck.push(ev.deliveryId);
+      }
+      if (events.length > 0) console.log(JSON.stringify({ event: "orb_relay_drained", count: events.length }));
+    };
+    void drainRelay();
+    setInterval(() => void drainRelay().catch((error) => captureError(error, { kind: "orb_relay_drain" })), 15_000);
+  }
+
   // Graceful shutdown: stop accepting HTTP, let the queue finish, close the backend.
   let shuttingDown = false;
   const shutdown = async (signal: string): Promise<void> => {

test/unit/orb-broker-client.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { fetchBrokeredInstallationToken, isOrbBrokerMode, registerOrbRelayTarget } from "../../src/orb/broker-client";
+import { drainOrbRelay, fetchBrokeredInstallationToken, isOrbBrokerMode, registerOrbRelayTarget } from "../../src/orb/broker-client";
 
 /** A fetch stub that records the URL + init and returns a fixed response. */
 function captureFetch(resp: Response): { fetchImpl: typeof fetch; calls: { url: string; init?: RequestInit | undefined }[] } {
@@ -106,7 +106,7 @@ describe("registerOrbRelayTarget", () => {
     expect(await registerOrbRelayTarget({ ORB_ENROLLMENT_SECRET: "orbsec_x", PUBLIC_API_ORIGIN: "https://me.example/", ORB_BROKER_URL: "https://broker.example/" }, fetchImpl)).toBe("registered");
     expect(calls[0]?.url).toBe("https://broker.example/v1/orb/relay/register"); // ORB_BROKER_URL trailing slash stripped
     expect((calls[0]?.init?.headers as Record<string, string>).authorization).toBe("Bearer orbsec_x");
-    expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({ relayUrl: "https://me.example/v1/orb/relay" }); // PUBLIC_API_ORIGIN trailing slash stripped
+    expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({ relayUrl: "https://me.example/v1/orb/relay", mode: "push" }); // PUBLIC_API_ORIGIN trailing slash stripped
   });
 
   it("uses the default broker base when ORB_BROKER_URL is unset", async () => {
@@ -128,4 +128,44 @@ describe("registerOrbRelayTarget", () => {
     expect(await registerOrbRelayTarget(cfg, (async () => new Response("no", { status: 403 })) as typeof fetch)).toBe("failed");
     expect(await registerOrbRelayTarget(cfg, (async () => { throw new Error("down"); }) as typeof fetch)).toBe("failed");
   });
+
+  it("pull mode (ORB_RELAY_MODE=pull) registers with NO relay URL and works without a public origin (NAT/tailnet)", async () => {
+    const { fetchImpl, calls } = captureFetch(new Response("ok"));
+    // No PUBLIC_API_ORIGIN — push would skip, but pull doesn't need an inbound URL.
+    expect(await registerOrbRelayTarget({ ORB_ENROLLMENT_SECRET: "s", ORB_RELAY_MODE: "pull" }, fetchImpl)).toBe("registered");
+    expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({ relayUrl: "", mode: "pull" });
+  });
+});
+
+describe("drainOrbRelay (pull-mode drain)", () => {
+  it("returns [] when not in broker mode (no enrollment secret)", async () => {
+    expect(await drainOrbRelay({})).toEqual([]);
+  });
+
+  it("POSTs the ack list, parses returned events, and filters malformed ones", async () => {
+    const { fetchImpl, calls } = captureFetch(
+      Response.json({
+        events: [
+          { deliveryId: "d1", eventName: "pull_request", rawBody: "{\"a\":1}" },
+          { deliveryId: "d2", eventName: "check_suite", rawBody: "{}" },
+          { deliveryId: "bad", eventName: "x" }, // no rawBody → filtered out
+        ],
+      }),
+    );
+    const out = await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s" }, ["prev-1"], fetchImpl);
+    expect(out).toEqual([
+      { deliveryId: "d1", eventName: "pull_request", rawBody: "{\"a\":1}" },
+      { deliveryId: "d2", eventName: "check_suite", rawBody: "{}" },
+    ]);
+    expect(calls[0]?.url).toBe("https://gittensory-api.aethereal.dev/v1/orb/relay/pull");
+    expect((calls[0]?.init?.headers as Record<string, string>).authorization).toBe("Bearer s");
+    expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({ ack: ["prev-1"] });
+  });
+
+  it("tolerates a missing events array (?? [] arm) and returns [] on non-ok / thrown / unsafe-URL", async () => {
+    expect(await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s" }, [], (async () => Response.json({})) as typeof fetch)).toEqual([]);
+    expect(await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s" }, [], (async () => new Response("no", { status: 403 })) as typeof fetch)).toEqual([]);
+    expect(await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s" }, [], (async () => { throw new Error("down"); }) as typeof fetch)).toEqual([]);
+    expect(await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s", ORB_BROKER_URL: "http://broker.example" }, [], (async () => { throw new Error("unsafe should not fetch"); }) as typeof fetch)).toEqual([]);
+  });
 });