Add webhook monitoring dashboard views

dashboard/src/main.test.tsx

@@ -35,6 +35,9 @@ import {
   selectedJobIdFromPath,
   shouldRefreshJobForSessionEvent,
   urlBase64ToUint8Array,
+  webhookDeliveriesPath,
+  webhookQuerySelection,
+  webhookTimeseriesPath,
 } from "./main";
 
 describe("dashboard routing and API query helpers", () => {
@@ -79,6 +82,16 @@ describe("dashboard routing and API query helpers", () => {
     expect(isWebhooksPath("/webhooks/github")).toBe(false);
   });
 
+  it("loads only the active webhook dataset and builds bounded queries", () => {
+    expect(webhookQuerySelection("overview")).toEqual({ timeseries: true, hooks: false, deliveries: false });
+    expect(webhookQuerySelection("hooks")).toEqual({ timeseries: false, hooks: true, deliveries: false });
+    expect(webhookQuerySelection("deliveries")).toEqual({ timeseries: false, hooks: false, deliveries: true });
+    expect(webhookTimeseriesPath("2026-09-01T00:00:00Z", "2026-10-01T00:00:00Z"))
+      .toBe("/api/webhooks/github/timeseries?from=2026-09-01T00%3A00%3A00Z&to=2026-10-01T00%3A00%3A00Z&bucket=day");
+    expect(webhookDeliveriesPath(50, "next page"))
+      .toBe("/api/webhooks/github/deliveries?limit=50&cursor=next+page");
+  });
+
   it("recognizes only canonical job detail routes", () => {
     expect(selectedJobIdFromPath("/jobs/45")).toBe(45);
     expect(selectedJobIdFromPath("/jobs/45/")).toBe(45);
@@ -114,14 +127,36 @@ describe("dashboard routing and API query helpers", () => {
   });
 
   it("renders the webhook status exported by the backend", () => {
-    render(<WebhookPage status={{ mode: "shadow", configured: true, receipts: { observed: 7 }, duplicate_deliveries: 2, cross_source_matches: 3 }} loading={false} error={null} onRefresh={vi.fn()} />);
+    render(<WebhookPage summary={{ mode: "shadow", configured: true, receipts: { observed: 7 }, duplicate_deliveries: 2, cross_source_matches: 3 }} timeseries={[]} section="overview" summaryLoading={false} sectionLoading={false} error={null} onSectionChange={vi.fn()} onOlderDeliveries={vi.fn()} onNewestDeliveries={vi.fn()} onRefresh={vi.fn()} />);
 
     expect(screen.getByRole("heading", { name: "GitHub webhooks" })).toBeInTheDocument();
     expect(screen.getByText("shadow")).toBeInTheDocument();
     expect(screen.getByText("observed")).toBeInTheDocument();
     expect(screen.getAllByText("7").length).toBeGreaterThan(0);
   });
 
+  it("navigates webhook hook inventory and delivery details", async () => {
+    const user = userEvent.setup();
+    const onSectionChange = vi.fn();
+    const summary = { mode: "shadow", configured: true, receipts: { observed: 7 }, duplicate_deliveries: 2, cross_source_matches: 3 };
+    const hooks = [{ id: "42", target: "gisce", target_type: "organization" as const, active: true, events: ["issue_comment"], status: "receiving" as const, last_ping_at: "2026-10-02T10:00:00Z", last_event_at: "2026-10-02T10:05:00Z" }];
+    const deliveries = [{ delivery_id: "delivery-1", created_at: "2026-10-02T10:05:00Z", event_name: "issue_comment", action: "created", repository: "gisce/github-agent-bridge", status: "observed" }];
+    const common = { summary, summaryLoading: false, sectionLoading: false, error: null, onSectionChange, onOlderDeliveries: vi.fn(), onNewestDeliveries: vi.fn(), onRefresh: vi.fn() };
+    const { rerender } = render(<WebhookPage {...common} section="overview" />);
+
+    await user.click(screen.getByRole("tab", { name: /Hooks/ }));
+    expect(onSectionChange).toHaveBeenCalledWith("hooks");
+    rerender(<WebhookPage {...common} hooks={hooks} section="hooks" />);
+    expect(screen.getByText("organization · #42")).toBeInTheDocument();
+    expect(screen.getByText("receiving")).toBeInTheDocument();
+
+    await user.click(screen.getByRole("tab", { name: /Deliveries/ }));
+    expect(onSectionChange).toHaveBeenCalledWith("deliveries");
+    rerender(<WebhookPage {...common} deliveries={deliveries} section="deliveries" />);
+    expect(screen.getByText("issue_comment · created")).toBeInTheDocument();
+    expect(screen.getByText("gisce/github-agent-bridge")).toBeInTheDocument();
+  });
+
   it("uses client-side navigation for dashboard section links", async () => {
     const user = userEvent.setup();
     const onNavigate = vi.fn();

docs/ingestion.md

@@ -78,17 +78,34 @@ organization webhook under **Organization settings → Webhooks**:
 
 A repository webhook covers only that repository. An organization webhook
 covers repositories in that organization and is the recommended deployment.
-Multiple organizations may use the same endpoint when each owner has its own
-entry in `GITHUB_AGENT_BRIDGE_WEBHOOK_SECRETS_BY_OWNER`.
+Multiple organizations and multiple hooks may use the same endpoint when each
+owner has its own entry in `GITHUB_AGENT_BRIDGE_WEBHOOK_SECRETS_BY_OWNER`.
+GitHub's `X-GitHub-Hook-ID` header keeps their inventory and activity separate.
 
 The endpoint stores only routing metadata, a SHA-256 payload hash, and the
 canonical event key in `webhook_shadow_receipts`; it deliberately stores no raw
-payload and never creates a queue job. Authenticated operators can inspect
-counts, duplicate deliveries, and cross-source event-key matches at
-`GET /api/webhooks/github/status`. Here “operators” means users authorized as
-dashboard administrators through `GITHUB_AGENT_BRIDGE_DASHBOARD_ADMIN_USERS`
-or `GITHUB_AGENT_BRIDGE_DASHBOARD_ADMIN_TEAMS`; other authenticated dashboard
-users receive HTTP 403 and unauthenticated requests receive HTTP 401.
+payload and never creates a queue job. The monitoring API is split so opening
+the overview does not load hook inventory or delivery history:
+
+- `GET /api/webhooks/github/summary` returns mode, receipt counts, retries, and
+  cross-source matches. The previous `/status` path remains a summary-only
+  compatibility alias.
+- `GET /api/webhooks/github/timeseries?from=<iso>&to=<iso>&bucket=day|hour`
+  returns only the requested interval. Daily ranges are capped at 366 days and
+  hourly ranges at 31 days; omitted bounds default to the configured retention
+  window ending now, capped at the daily maximum.
+- `GET /api/webhooks/github/hooks` returns the per-hook inventory.
+- `GET /api/webhooks/github/deliveries?limit=<1-100>&cursor=<opaque>` returns a
+  cursor-paginated delivery page.
+
+The dashboard loads these resources lazily per tab and caches them separately.
+Here “operators” means users authorized as dashboard administrators through
+`GITHUB_AGENT_BRIDGE_DASHBOARD_ADMIN_USERS` or
+`GITHUB_AGENT_BRIDGE_DASHBOARD_ADMIN_TEAMS`; every monitoring endpoint returns
+HTTP 403 to other authenticated users and HTTP 401 to unauthenticated users.
+Receipt details are retained for 30 days by default and pruned during ingestion;
+set `GITHUB_AGENT_BRIDGE_WEBHOOK_RETENTION_DAYS` to change that window. Raw
+payloads are never retained.
 
 The maintained event inventory and support levels are in
 [`webhook-events.md`](webhook-events.md).

src/github_agent_bridge/backend.py

@@ -4,12 +4,14 @@
 import asyncio
 import base64
 from contextlib import asynccontextmanager, suppress
+from datetime import UTC, datetime, timedelta
 import json
 import os
 import secrets
 import shlex
 import sqlite3
 import sys
+import threading
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -68,6 +70,8 @@
 GITHUB_TEAMS_URL = "https://api.github.com/user/teams"
 PROJECT_REPOSITORY_URL = "https://github.com/gisce/github-agent-bridge"
 SESSION_VERSION = 1
+WEBHOOK_TIMESERIES_MAX_DAYS = 366
+WEBHOOK_TIMESERIES_MAX_HOURLY_DAYS = 31
 
 
 def _knowledge_actor(item: dict[str, Any]) -> str:
@@ -98,6 +102,43 @@ def _mark_manageable_knowledge(items: list[dict[str, Any]], profile: dict[str, A
     return [{**item, "can_manage": _knowledge_item_owned_by(item, login)} for item in items]
 
 
+def _parse_webhook_datetime(value: str, parameter: str) -> datetime:
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"invalid_{parameter}_datetime",
+        ) from exc
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=UTC)
+    return parsed.astimezone(UTC)
+
+
+def _webhook_datetime_value(value: datetime) -> str:
+    return value.replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _encode_webhook_delivery_cursor(created_at: str, delivery_id: str) -> str:
+    payload = json.dumps([created_at, delivery_id], separators=(",", ":")).encode("utf-8")
+    return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=")
+
+
+def _decode_webhook_delivery_cursor(cursor: str) -> tuple[str, str]:
+    try:
+        padded = cursor + "=" * (-len(cursor) % 4)
+        payload = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8"))
+    except (ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid_cursor") from exc
+    if not (
+        isinstance(payload, list)
+        and len(payload) == 2
+        and all(isinstance(value, str) and value for value in payload)
+    ):
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid_cursor")
+    return payload[0], payload[1]
+
+
 class DashboardConfig:
     def __init__(
         self,
@@ -118,6 +159,7 @@ def __init__(
         webhook_secrets: tuple[str, ...] | None = None,
         webhook_secrets_by_owner: dict[str, tuple[str, ...]] | None = None,
         webhook_max_bytes: int | None = None,
+        webhook_retention_days: int | None = None,
     ) -> None:
         self.db = Path(db).expanduser()
         self.secret_key = secret_key or os.getenv("GITHUB_AGENT_BRIDGE_DASHBOARD_SECRET_KEY", "")
@@ -140,6 +182,7 @@ def __init__(
         )
         self.webhook_secrets_by_owner = webhook_secrets_by_owner if webhook_secrets_by_owner is not None else _webhook_secrets_by_owner_env()
         self.webhook_max_bytes = webhook_max_bytes or int(os.getenv("GITHUB_AGENT_BRIDGE_WEBHOOK_MAX_BYTES", "1048576"))
+        self.webhook_retention_days = webhook_retention_days or int(os.getenv("GITHUB_AGENT_BRIDGE_WEBHOOK_RETENTION_DAYS", "30"))
 
     @property
     def oauth_ready(self) -> bool:
@@ -535,6 +578,18 @@ async def lifespan(app: FastAPI):
     app = FastAPI(title="GitHub Agent Bridge Dashboard API", lifespan=lifespan)
     app.state.dashboard_config = config
     app.state.dashboard_shutdown_event = shutdown_event
+    webhook_schema_lock = threading.Lock()
+    webhook_schema_ready = False
+
+    def ensure_webhook_schema() -> None:
+        nonlocal webhook_schema_ready
+        if webhook_schema_ready:
+            return
+        with webhook_schema_lock:
+            if not webhook_schema_ready:
+                JobQueue(config.db)
+                webhook_schema_ready = True
+
     assets_dir = config.static_dir / "assets"
     if assets_dir.exists():
         app.mount("/assets", StaticFiles(directory=assets_dir), name="dashboard-assets")
@@ -605,6 +660,7 @@ async def github_webhook_shadow(request: Request) -> dict[str, Any]:
             raise HTTPException(status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, detail="application_json_required")
         delivery_id = request.headers.get("x-github-delivery", "").strip()
         event_name = request.headers.get("x-github-event", "").strip()
+        hook_id = request.headers.get("x-github-hook-id", "").strip() or None
         signature = request.headers.get("x-hub-signature-256", "").strip()
         if not delivery_id or not event_name:
             raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="github_headers_required")
@@ -633,36 +689,136 @@ async def github_webhook_shadow(request: Request) -> dict[str, Any]:
             webhook_secrets = config.webhook_secrets
         if not verify_signature(raw_payload, signature, webhook_secrets):
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid_signature")
-        JobQueue(config.db)
+        ensure_webhook_schema()
         receipt = persist_shadow_delivery(
             config.db,
             delivery_id=delivery_id,
             event_name=event_name,
             raw_payload=raw_payload,
+            hook_id=hook_id,
+            retention_days=config.webhook_retention_days,
         )
         return {"mode": "shadow", "status": receipt.status, "event_key": receipt.event_key}
 
     @app.get("/api/webhooks/github/status")
-    def github_webhook_shadow_status(_: dict[str, Any] = Depends(current_admin_profile)) -> dict[str, Any]:
-        JobQueue(config.db)
+    @app.get("/api/webhooks/github/summary")
+    def github_webhook_shadow_summary(_: dict[str, Any] = Depends(current_admin_profile)) -> dict[str, Any]:
+        ensure_webhook_schema()
         with sqlite3.connect(config.db) as con:
-            rows = con.execute(
-                "SELECT status,COUNT(*) FROM webhook_shadow_receipts GROUP BY status"
-            ).fetchall()
-            cross_source = con.execute(
-                "SELECT COUNT(DISTINCT w.event_key) FROM webhook_shadow_receipts w "
-                "JOIN ingest_receipts i ON i.event_key=w.event_key WHERE w.event_key IS NOT NULL"
-            ).fetchone()[0]
-            duplicate_deliveries = con.execute(
-                "SELECT COALESCE(SUM(duplicate_count),0) FROM webhook_shadow_receipts"
-            ).fetchone()[0]
-        counts = {row[0]: row[1] for row in rows}
+            row = con.execute(
+                "WITH receipt_summary AS ("
+                " SELECT SUM(CASE WHEN status='observed' THEN 1 ELSE 0 END) observed,"
+                " SUM(CASE WHEN status='unsupported' THEN 1 ELSE 0 END) unsupported,"
+                " COALESCE(SUM(duplicate_count),0) duplicate_deliveries"
+                " FROM webhook_shadow_receipts"
+                "), cross_source AS ("
+                " SELECT COUNT(DISTINCT w.event_key) matches FROM webhook_shadow_receipts w"
+                " WHERE w.event_key IS NOT NULL AND EXISTS ("
+                "  SELECT 1 FROM ingest_receipts i WHERE i.event_key=w.event_key"
+                " )"
+                ") SELECT COALESCE(observed,0),COALESCE(unsupported,0),"
+                "duplicate_deliveries,matches FROM receipt_summary CROSS JOIN cross_source"
+            ).fetchone()
+        counts = {
+            name: count
+            for name, count in (("observed", row[0]), ("unsupported", row[1]))
+            if count
+        }
         return {
             "mode": "shadow",
             "configured": bool(config.webhook_secrets or config.webhook_secrets_by_owner),
             "receipts": counts,
-            "duplicate_deliveries": duplicate_deliveries,
-            "cross_source_matches": cross_source,
+            "duplicate_deliveries": row[2],
+            "cross_source_matches": row[3],
+        }
+
+    @app.get("/api/webhooks/github/timeseries")
+    def github_webhook_shadow_timeseries(
+        from_value: str | None = Query(None, alias="from"),
+        to_value: str | None = Query(None, alias="to"),
+        bucket: str = Query("day", pattern="^(hour|day)$"),
+        _: dict[str, Any] = Depends(current_admin_profile),
+    ) -> dict[str, Any]:
+        ensure_webhook_schema()
+        end = _parse_webhook_datetime(to_value, "to") if to_value else datetime.now(UTC)
+        default_days = min(config.webhook_retention_days, WEBHOOK_TIMESERIES_MAX_DAYS)
+        start = _parse_webhook_datetime(from_value, "from") if from_value else end - timedelta(days=default_days)
+        maximum = WEBHOOK_TIMESERIES_MAX_HOURLY_DAYS if bucket == "hour" else WEBHOOK_TIMESERIES_MAX_DAYS
+        if start >= end:
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid_time_range")
+        if end - start > timedelta(days=maximum):
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="time_range_too_large")
+        start_value = _webhook_datetime_value(start)
+        end_value = _webhook_datetime_value(end)
+        bucket_expression = "substr(created_at,1,13) || ':00:00Z'" if bucket == "hour" else "substr(created_at,1,10)"
+        with sqlite3.connect(config.db) as con:
+            rows = con.execute(
+                f"SELECT {bucket_expression}, "
+                "SUM(CASE WHEN status='observed' THEN 1 ELSE 0 END), "
+                "SUM(duplicate_count), SUM(CASE WHEN status='unsupported' THEN 1 ELSE 0 END) "
+                "FROM webhook_shadow_receipts WHERE created_at>=? AND created_at<? "
+                "GROUP BY 1 ORDER BY 1",
+                (start_value, end_value),
+            ).fetchall()
+        return {
+            "from": start_value,
+            "to": end_value,
+            "bucket": bucket,
+            "points": [
+                {"bucket": row[0], "observed": row[1], "duplicate": row[2], "unsupported": row[3]}
+                for row in rows
+            ],
+        }
+
+    @app.get("/api/webhooks/github/hooks")
+    def github_webhook_shadow_hooks(_: dict[str, Any] = Depends(current_admin_profile)) -> dict[str, Any]:
+        ensure_webhook_schema()
+        with sqlite3.connect(config.db) as con:
+            hook_rows = con.execute(
+                "SELECT hook_id,target,target_type,active,events_json,last_ping_at,last_event_at "
+                "FROM webhook_hooks ORDER BY target_type,target,hook_id"
+            ).fetchall()
+        return {
+            "hooks": [
+                {"id": row[0], "target": row[1], "target_type": row[2], "active": bool(row[3]),
+                 "events": json.loads(row[4]), "last_ping_at": row[5], "last_event_at": row[6],
+                 "status": ("inactive" if not row[3] else "receiving" if row[6] else "never_seen")}
+                for row in hook_rows
+            ],
+        }
+
+    @app.get("/api/webhooks/github/deliveries")
+    def github_webhook_shadow_deliveries(
+        limit: int = Query(50, ge=1, le=100),
+        cursor: str | None = Query(None),
+        _: dict[str, Any] = Depends(current_admin_profile),
+    ) -> dict[str, Any]:
+        ensure_webhook_schema()
+        where = ""
+        parameters: list[Any] = []
+        if cursor:
+            cursor_created_at, cursor_delivery_id = _decode_webhook_delivery_cursor(cursor)
+            where = "WHERE created_at<? OR (created_at=? AND delivery_id<?)"
+            parameters.extend((cursor_created_at, cursor_created_at, cursor_delivery_id))
+        parameters.append(limit + 1)
+        with sqlite3.connect(config.db) as con:
+            delivery_rows = con.execute(
+                "SELECT delivery_id,hook_id,event_name,action,event_key,repository,status,created_at "
+                f"FROM webhook_shadow_receipts {where} "
+                "ORDER BY created_at DESC,delivery_id DESC LIMIT ?",
+                parameters,
+            ).fetchall()
+        page = delivery_rows[:limit]
+        next_cursor = None
+        if len(delivery_rows) > limit and page:
+            next_cursor = _encode_webhook_delivery_cursor(page[-1][7], page[-1][0])
+        return {
+            "deliveries": [
+                {"delivery_id": row[0], "hook_id": row[1], "event_name": row[2], "action": row[3],
+                 "event_key": row[4], "repository": row[5], "status": row[6], "created_at": row[7]}
+                for row in page
+            ],
+            "next_cursor": next_cursor,
         }
 
     def dashboard_index() -> FileResponse: