Extract admin webhook helpers

app/admin.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.models import WebhookEvent
+
+WEBHOOK_OUTCOME_SCAN_ORDER = {
+    "missing_submitter": 0,
+    "bounty_not_found": 1,
+    "exhausted_bounty": 2,
+    "duplicate_delivery": 3,
+    "delivery_payload_mismatch": 4,
+    "already_paid": 5,
+    "paid": 6,
+}
+
+
+def normalize_webhook_status_filter(status: str | None) -> str | None:
+    if status is None:
+        return None
+    normalized = status.strip().lower()
+    return normalized or None
+
+
+def list_webhook_events(
+    session: Session, status: str | None = None, limit: int = 50
+) -> list[WebhookEvent]:
+    normalized_status = normalize_webhook_status_filter(status)
+    query = select(WebhookEvent)
+    if normalized_status is not None:
+        query = query.where(func.lower(WebhookEvent.processed_status) == normalized_status)
+    return list(
+        session.scalars(
+            query.order_by(WebhookEvent.created_at.desc(), WebhookEvent.delivery_id.desc()).limit(
+                limit
+            )
+        ).all()
+    )
+
+
+def webhook_event_to_dict(event: WebhookEvent) -> dict[str, Any]:
+    return {
+        "delivery_id": event.delivery_id,
+        "event_type": event.event_type,
+        "processed_status": event.processed_status,
+        "payload_hash": event.payload_hash,
+        "created_at": event.created_at.isoformat(),
+    }
+
+
+def webhook_events_to_dict(events: list[WebhookEvent]) -> list[dict[str, Any]]:
+    return [webhook_event_to_dict(event) for event in events]
+
+
+def webhook_status_summary(session: Session) -> list[dict[str, Any]]:
+    status_expr = func.lower(WebhookEvent.processed_status)
+    count_expr = func.count(WebhookEvent.delivery_id)
+    rows = session.execute(
+        select(status_expr, count_expr)
+        .group_by(status_expr)
+        .order_by(count_expr.desc(), status_expr.asc())
+    ).all()
+    summary = [
+        {"processed_status": str(status), "count": int(count)} for status, count in rows if status
+    ]
+    return sorted(
+        summary,
+        key=lambda item: (
+            WEBHOOK_OUTCOME_SCAN_ORDER.get(str(item["processed_status"]), 100),
+            -int(item["count"]),
+            str(item["processed_status"]),
+        ),
+    )

app/main.py

@@ -20,6 +20,12 @@
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from app.admin import (
+    list_webhook_events,
+    normalize_webhook_status_filter,
+    webhook_events_to_dict,
+    webhook_status_summary,
+)
 from app.config import Settings, get_settings
 from app.db import create_schema, session_scope
 from app.ledger.reconciliation import payout_reconciliation_summary, reconcile_accepted_payouts
@@ -51,7 +57,6 @@
     Proof,
     Submission,
     Wallet,
-    WebhookEvent,
 )
 from app.serializers import (
     accepted_work_for_account,
@@ -111,15 +116,6 @@
 DEFAULT_ATTEMPT_TTL_SECONDS = 24 * 60 * 60
 MIN_ATTEMPT_TTL_SECONDS = 60
 MAX_ATTEMPT_TTL_SECONDS = 7 * 24 * 60 * 60
-WEBHOOK_OUTCOME_SCAN_ORDER = {
-    "missing_submitter": 0,
-    "bounty_not_found": 1,
-    "exhausted_bounty": 2,
-    "duplicate_delivery": 3,
-    "delivery_payload_mismatch": 4,
-    "already_paid": 5,
-    "paid": 6,
-}
 
 
 def _request_was_forwarded_https(request: Request) -> bool:
@@ -256,27 +252,6 @@ def _existing_payout_proof_for_submission(
     )
 
 
-def webhook_status_summary(session: Session) -> list[dict[str, Any]]:
-    status_expr = func.lower(WebhookEvent.processed_status)
-    count_expr = func.count(WebhookEvent.delivery_id)
-    rows = session.execute(
-        select(status_expr, count_expr)
-        .group_by(status_expr)
-        .order_by(count_expr.desc(), status_expr.asc())
-    ).all()
-    summary = [
-        {"processed_status": str(status), "count": int(count)} for status, count in rows if status
-    ]
-    return sorted(
-        summary,
-        key=lambda item: (
-            WEBHOOK_OUTCOME_SCAN_ORDER.get(str(item["processed_status"]), 100),
-            -int(item["count"]),
-            str(item["processed_status"]),
-        ),
-    )
-
-
 def _host_without_port(request: Request) -> str:
     return request.headers.get("host", "").split(":", 1)[0].lower()
 
@@ -656,28 +631,8 @@ def api_admin_webhook_events(
         admin_login: str = Depends(require_admin_token),
     ) -> list[dict[str, Any]]:
         del admin_login
-        normalized_status = status.strip().lower() if status is not None else None
-        if normalized_status == "":
-            normalized_status = None
         with session_scope(db_url) as session:
-            query = select(WebhookEvent)
-            if normalized_status is not None:
-                query = query.where(func.lower(WebhookEvent.processed_status) == normalized_status)
-            events = session.scalars(
-                query.order_by(
-                    WebhookEvent.created_at.desc(), WebhookEvent.delivery_id.desc()
-                ).limit(limit)
-            ).all()
-            return [
-                {
-                    "delivery_id": event.delivery_id,
-                    "event_type": event.event_type,
-                    "processed_status": event.processed_status,
-                    "payload_hash": event.payload_hash,
-                    "created_at": event.created_at.isoformat(),
-                }
-                for event in events
-            ]
+            return webhook_events_to_dict(list_webhook_events(session, status, limit))
 
     @app.post("/api/v1/bounties")
     async def api_create_bounty(
@@ -1452,16 +1407,9 @@ def admin_page(
             if _oauth_configured(settings):
                 return RedirectResponse("/auth/github/login?next=/admin", status_code=302)
             raise HTTPException(status_code=503, detail="GitHub OAuth is not configured")
-        normalized_status = webhook_status.strip().lower() if webhook_status is not None else ""
+        normalized_status = normalize_webhook_status_filter(webhook_status) or ""
         with session_scope(db_url) as session:
-            query = select(WebhookEvent)
-            if normalized_status:
-                query = query.where(func.lower(WebhookEvent.processed_status) == normalized_status)
-            webhook_events = session.scalars(
-                query.order_by(
-                    WebhookEvent.created_at.desc(), WebhookEvent.delivery_id.desc()
-                ).limit(webhook_limit)
-            ).all()
+            webhook_events = list_webhook_events(session, normalized_status, webhook_limit)
             webhook_summary = webhook_status_summary(session)
         return templates.TemplateResponse(
             request,

tests/test_admin_helpers.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from app.admin import (
+    list_webhook_events,
+    normalize_webhook_status_filter,
+    webhook_events_to_dict,
+    webhook_status_summary,
+)
+from app.db import create_schema, session_scope
+from app.models import WebhookEvent
+
+
+def _event(
+    delivery_id: str,
+    status: str,
+    created_at: datetime,
+    event_type: str = "pull_request",
+    payload_hash: str | None = None,
+) -> WebhookEvent:
+    return WebhookEvent(
+        delivery_id=delivery_id,
+        event_type=event_type,
+        payload_hash=payload_hash or delivery_id.ljust(64, "0")[:64],
+        processed_status=status,
+        created_at=created_at,
+    )
+
+
+def test_normalize_webhook_status_filter_trims_and_lowers() -> None:
+    assert normalize_webhook_status_filter(None) is None
+    assert normalize_webhook_status_filter("   ") is None
+    assert normalize_webhook_status_filter(" Missing_Submitter ") == "missing_submitter"
+
+
+def test_list_webhook_events_filters_and_serializes_safe_fields(sqlite_url: str) -> None:
+    create_schema(sqlite_url)
+    base_time = datetime(2026, 5, 25, 12, 0, tzinfo=UTC)
+    with session_scope(sqlite_url) as session:
+        session.add(_event("delivery-paid", "paid", base_time))
+        session.add(
+            _event("delivery-missing-old", "missing_submitter", base_time - timedelta(hours=1))
+        )
+        session.add(
+            _event("delivery-missing-new", "Missing_Submitter", base_time + timedelta(hours=1))
+        )
+
+    with session_scope(sqlite_url) as session:
+        events = list_webhook_events(session, " Missing_Submitter ", limit=10)
+        serialized = webhook_events_to_dict(events)
+
+    assert [event.delivery_id for event in events] == [
+        "delivery-missing-new",
+        "delivery-missing-old",
+    ]
+    assert serialized == [
+        {
+            "delivery_id": "delivery-missing-new",
+            "event_type": "pull_request",
+            "processed_status": "Missing_Submitter",
+            "payload_hash": "delivery-missing-new".ljust(64, "0")[:64],
+            "created_at": (base_time + timedelta(hours=1)).replace(tzinfo=None).isoformat(),
+        },
+        {
+            "delivery_id": "delivery-missing-old",
+            "event_type": "pull_request",
+            "processed_status": "missing_submitter",
+            "payload_hash": "delivery-missing-old".ljust(64, "0")[:64],
+            "created_at": (base_time - timedelta(hours=1)).replace(tzinfo=None).isoformat(),
+        },
+    ]
+
+
+def test_webhook_status_summary_uses_admin_scan_order(sqlite_url: str) -> None:
+    create_schema(sqlite_url)
+    base_time = datetime(2026, 5, 25, 12, 0, tzinfo=UTC)
+    with session_scope(sqlite_url) as session:
+        session.add(_event("delivery-paid-1", "paid", base_time))
+        session.add(_event("delivery-paid-2", "paid", base_time + timedelta(minutes=1)))
+        session.add(
+            _event("delivery-missing", "Missing_Submitter", base_time + timedelta(minutes=2))
+        )
+        session.add(_event("delivery-custom-1", "custom_status", base_time + timedelta(minutes=3)))
+        session.add(_event("delivery-custom-2", "custom_status", base_time + timedelta(minutes=4)))
+        session.add(_event("delivery-custom-3", "custom_status", base_time + timedelta(minutes=5)))
+
+    with session_scope(sqlite_url) as session:
+        summary = webhook_status_summary(session)
+
+    assert summary == [
+        {"processed_status": "missing_submitter", "count": 1},
+        {"processed_status": "paid", "count": 2},
+        {"processed_status": "custom_status", "count": 3},
+    ]