feat(orb): add POST /v1/orb/ingest central collector endpoint

apps/gittensory-ui/public/openapi.json

@@ -15216,6 +15216,26 @@
           }
         ]
       }
+    },
+    "/v1/orb/ingest": {
+      "post": {
+        "responses": {
+          "200": {
+            "description": "Batch accepted; returns { accepted: number }"
+          },
+          "400": {
+            "description": "Malformed JSON or invalid payload shape"
+          }
+        },
+        "security": [
+          {
+            "GittensoryBearer": []
+          },
+          {
+            "GittensorySessionCookie": []
+          }
+        ]
+      }
     }
   },
   "servers": [

migrations/0058_orb_signals.sql

@@ -0,0 +1,16 @@
+-- Gittensory Orb (#1219): central collector store. Receives anonymized outcome signal batches
+-- from self-hosted instances running exportOrbBatch. repo_hash and pr_hash are HMAC-anonymized
+-- by the sender — no repo names, owner identifiers, or PR content is stored here.
+CREATE TABLE IF NOT EXISTS orb_signals (
+  id              INTEGER PRIMARY KEY,
+  instance_id     TEXT    NOT NULL,
+  repo_hash       TEXT    NOT NULL,
+  pr_hash         TEXT    NOT NULL,
+  outcome         TEXT    NOT NULL CHECK (outcome IN ('merged', 'closed')),
+  gate_verdict    TEXT,
+  time_to_close_ms INTEGER,
+  sent_at         TEXT,
+  received_at     TEXT    NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  UNIQUE (instance_id, pr_hash)
+);
+CREATE INDEX IF NOT EXISTS orb_signals_instance ON orb_signals (instance_id, received_at);

src/api/routes.ts

@@ -122,6 +122,7 @@ import {
   type GittensoryMentionCommandName,
 } from "../github/commands";
 import { handleGitHubWebhook } from "../github/webhook";
+import { handleOrbIngest } from "../orb/ingest";
 import { handleMcpRequest } from "../mcp/server";
 import { buildOpenApiSpec } from "../openapi/spec";
 import { generateSignalSnapshots } from "../queue/processors";
@@ -2862,6 +2863,17 @@ export function createApp() {
 
   app.post("/v1/github/webhook", handleGitHubWebhook);
 
+  // Gittensory Orb (#1219) — central collector. Receives anonymized outcome signal batches
+  // from self-hosted instances. No auth required: all data is HMAC-anonymized by the sender;
+  // dedup is enforced via UNIQUE(instance_id, pr_hash) in orb_signals.
+  app.post("/v1/orb/ingest", async (c) => {
+    const body = await c.req.text().catch(() => null);
+    if (!body) return c.json({ error: "invalid_request" }, 400);
+    const result = await handleOrbIngest(body, c.env.DB);
+    if ("error" in result) return c.json(result, 400);
+    return c.json(result, 200);
+  });
+
   // Convergence (ops / observability, flag GITTENSORY_REVIEW_OPS). Cross-repo review-OUTCOME aggregate (gate-block
   // ledger + recommendation/slop calibration) for an operator dashboard. Bearer-gated by the `/v1/internal/*`
   // middleware above (INTERNAL_JOB_TOKEN). Flag-OFF (default) → 404, so the endpoint does not exist and the
@@ -4808,6 +4820,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/ingest") return false;
   if (path.startsWith("/v1/internal/")) return false;
   return path.startsWith("/v1/");
 }

src/openapi/spec.ts

@@ -664,6 +664,14 @@ export function buildOpenApiSpec() {
       401: { description: "Invalid webhook signature" },
     },
   });
+  registry.registerPath({
+    method: "post",
+    path: "/v1/orb/ingest",
+    responses: {
+      200: { description: "Batch accepted; returns { accepted: number }" },
+      400: { description: "Malformed JSON or invalid payload shape" },
+    },
+  });
   registry.registerPath({
     method: "get",
     path: "/v1/auth/github/start",

src/orb/ingest.ts

@@ -0,0 +1,81 @@
+// Gittensory Orb (#1219) — central collector receiver.
+// Accepts anonymized outcome signal batches from self-hosted instances running exportOrbBatch.
+// No raw repo names, owner identifiers, or PR content is accepted or stored — only HMAC-anonymized
+// hashes + aggregate outcome metadata (verdict, timing).
+
+const MAX_BATCH = 500;
+const VALID_OUTCOMES = new Set(["merged", "closed"]);
+
+interface OrbIngestEvent {
+  repo_hash: string;
+  pr_hash: string;
+  outcome: string;
+  gate_verdict?: string | null;
+  time_to_close_ms?: number | null;
+  created_at?: string | null;
+}
+
+interface OrbIngestPayload {
+  instance_id: string;
+  events: OrbIngestEvent[];
+}
+
+export type OrbIngestResult = { accepted: number } | { error: string };
+
+export async function handleOrbIngest(body: string, db: D1Database): Promise<OrbIngestResult> {
+  let payload: unknown;
+  try {
+    payload = JSON.parse(body);
+  } catch {
+    return { error: "invalid_json" };
+  }
+
+  if (
+    typeof (payload as OrbIngestPayload)?.instance_id !== "string" ||
+    !Array.isArray((payload as OrbIngestPayload)?.events)
+  ) {
+    return { error: "invalid_payload" };
+  }
+
+  const { instance_id, events } = payload as OrbIngestPayload;
+  if (!instance_id || events.length === 0) {
+    return { error: "invalid_payload" };
+  }
+
+  const batch = events.slice(0, MAX_BATCH);
+  let accepted = 0;
+
+  for (const event of batch) {
+    if (
+      typeof event.repo_hash !== "string" || !event.repo_hash ||
+      typeof event.pr_hash !== "string" || !event.pr_hash ||
+      !VALID_OUTCOMES.has(event.outcome)
+    ) {
+      continue;
+    }
+
+    try {
+      const result = await db
+        .prepare(
+          `INSERT OR IGNORE INTO orb_signals
+           (instance_id, repo_hash, pr_hash, outcome, gate_verdict, time_to_close_ms, sent_at)
+           VALUES (?, ?, ?, ?, ?, ?, ?)`,
+        )
+        .bind(
+          instance_id,
+          event.repo_hash,
+          event.pr_hash,
+          event.outcome,
+          typeof event.gate_verdict === "string" ? event.gate_verdict : null,
+          typeof event.time_to_close_ms === "number" ? event.time_to_close_ms : null,
+          typeof event.created_at === "string" ? event.created_at : null,
+        )
+        .run();
+      if (result.meta.changes > 0) accepted++;
+    } catch {
+      // best-effort — skip rows that violate constraints or hit transient errors
+    }
+  }
+
+  return { accepted };
+}