​

Activity Monitor — event-sourcing architecture

​

Why this exists

​

The pipeline

┌────────────────────────────────────────────────────────────────┐
│  /api/ingest/otel/:sourceId  /api/ingest/webhook/:sourceId      │
│  (Hono routes — auth, validate sourceId, parse body)            │
└────────────────────────┬───────────────────────────────────────┘
                         │
                         ▼  RecordActivityEventCommand
                ┌────────────────────┐
                │  event_log (CH)    │
                │  ActivityEventReceived           │
                └────────┬───────────┘
                         │
                         ▼  pipeline: activity-monitor-processing
                         │  aggregateType: "activity_event"
                         │
   ┌─────────────────────┴────────────────────────┐
   │                                              │
   ▼                                              ▼
┌──────────────────────────┐      ┌──────────────────────────────┐
│ Map projection           │      │ Fold projection              │
│  activityEventStorage    │      │  anomalyWindow               │
│  → gateway_activity_events│     │  → per-tenant rolling totals │
│  (CH, OCSF + AOS shape)  │      │   (in-memory + Redis cache)  │
└────────┬─────────────────┘      └──────────┬───────────────────┘
         │                                   │
         │ wakes:                            │ wakes:
         ▼                                   ▼
┌──────────────────────────┐      ┌──────────────────────────────┐
│ Reactor                  │      │ Reactor                      │
│  activityEventBroadcast  │      │  anomalyDetection            │
│  (real-time UI push for  │      │  - load active AnomalyRules  │
│   /governance dashboard) │      │  - evaluate per-rule type    │
└──────────────────────────┘      │  - if trigger:               │
                                  │    • upsert AnomalyAlert     │
                                  │    • dispatch via shared     │
                                  │      triggerActionDispatch   │
                                  └──────────────────────────────┘

​

Why a dedicated pipeline (not bolted onto trace-processing)

1. Activity events get represented as fake single-span traces (lossy + confusing — trace_summaries would mix gateway-proxied traces and per-event activity rows under the same TenantId).
2. trace_summaries grows a discriminator column and the fold projection becomes branchy.

​

Aggregate identity

aggregateType:  "activity_event"
aggregateId:    EventId  (one event = one aggregate, no fold across events)
tenantId:       IngestionSource.id  (matches gateway_activity_events.TenantId)

​

Slicing the redesign

​

C0 — this doc + spec updates

• This architecture doc.
• specs/ai-gateway/governance/anomaly-detection.feature updated to drop poller language; reactor framing throughout.
• AnomalyAlert Prisma model + migration 20260427020000_add_anomaly_alert/ doc-comment updated to reference the reactor as producer.
• Existing receivers continue to write CH directly until C1 lands — this slice is doc-only so the team can review the architecture before more code moves.

​

C1 — receiver → event_log → projection reactor

• New event schema: ActivityEventReceived with the OCSF-normalised ActivityEventRow shape.
• New command: RecordActivityEventCommand wired into the pipeline.
• Refactor /api/ingest/otel/:sourceId and /api/ingest/webhook/:sourceId to call the command instead of writing CH directly.
• Map projection activityEventStorage writes to gateway_activity_events (replaces today’s direct insert).
• Dogfood: curl → 202 → row visible in CH (same as today, just via event-sourced path).

​

C2 — AnomalyAlert + anomaly reactor for one rule type

• Apply the AnomalyAlert migration that’s already drafted but doesn’t ship behaviour yet.
• Add anomalyWindow fold projection (per-tenant rolling totals).
• Add anomalyDetection reactor for spend_spike only first (cleanest mapping to the existing CostUSD field).
• Wire into api.activityMonitor.recentAnomalies (replaces current [] stub).
• Dogfood: create rule in Alexis’s UI → curl violating event → alert appears on /governance within ~30s.

​

C3 — Dispatch destinations

• Generic webhook + log-only first (matches PR #3351’s triggerActionDispatch shape).
• Slack / PagerDuty / SIEM / email follow as per-destination adapter slices once the reactor pattern is proven.

​

What we keep from the v0 receiver code

• IngestionSourceService (CRUD + auth) — unchanged.
• gateway_activity_events CH schema (migration 00019_*) — unchanged. The map projection writes the same columns.
• OTel + webhook normalisers (normalizers/otel.ts etc.) — unchanged. They get called from the map projection now instead of the receiver handler.
• All receiver auth + sourceId-mismatch + 24h secret rotation grace — unchanged.

​

What we drop from the v0 receiver code

• The direct ActivityEventRepository.insert(...) call from the receiver handler. The receiver instead enqueues an event into the pipeline; the map projection does the actual CH insert.
• The poller-based AnomalyEvaluatorService design that was sketched but never shipped. Replaced by the anomaly reactor.

​

Test strategy per slice

​

Cross-references

• PR #3351 — feat: event-driven trace triggers via reactor (the pattern this redesign learns from).
• anomaly-detection.feature — user-facing contract, updated for event-sourcing.
• anomaly-rules.feature — configuration entity (already shipped, unchanged).
• activity-monitor.feature — admin UI contract (already shipped; pipeline section adds in C1).
• architecture.md — top-level governance architecture; this doc is the activity-monitor deep-dive linked from the “Activity Monitor (Tier C/D)” block.