​

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:
         ▼                                   ▼
┌──────────────────────────┐      ┌──────────────────────────────┐
│ Subscriber                  │      │ Subscriber                      │
│  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 and 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 and spec updates

• This architecture doc.
• specs/ai-gateway/governance/anomaly-detection.feature updated to drop poller language; subscriber framing throughout.
• AnomalyAlert Prisma model + migration 20260427020000_add_anomaly_alert/ doc-comment updated to reference the subscriber 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 subscriber

• 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).
• Manual check: curl → 202 → row visible in CH (same as today, just via event-sourced path).

​

C2: AnomalyAlert and anomaly subscriber 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 subscriber for spend_spike only first (cleanest mapping to the existing CostUSD field).
• Wire into api.activityMonitor.recentAnomalies (replaces current [] stub).
• Manual check: create a rule in the anomaly rules UI → curl a 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 subscriber pattern is proven.

​

What we keep from the v0 receiver code

• IngestionSourceService (CRUD and auth): unchanged.
• gateway_activity_events CH schema (migration 00019_*): unchanged. The map projection writes the same columns.
• OTel and webhook normalisers (normalizers/otel.ts etc.): unchanged. The map projection calls them now instead of the receiver handler.
• All receiver auth, sourceId-mismatch, and 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 subscriber.

​

Test strategy per slice

​

Cross-references

• PR #3351: event-driven trace triggers via subscriber (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).
• Governance architecture: top-level governance architecture. This doc is the activity-monitor deep-dive linked from the “Activity Monitor (Tier C/D)” block.