Docs / notification-service.md · mirrored from the repo

Notification Service — Architecture & Design

Status: design / planned. Phase 1 (in-app SSE) is the first implementation target. Phases 2–3 are the extension roadmap — adapters slot in behind the same port without touching the schema or the core delivery logic.

1. Goals

• Phase 1 (now): In-app push notifications for back-office (BO) users via SSE. The bell icon stub already exists in apps/console/src/components/Header.tsx.
• Phase 2: Email (Resend) + Slack webhook for BO ops alerts.
• Phase 3: SMS + Web Push for consumer-app users (same port, new adapters, no schema migration).
• Reusable across both trust domains: BO users (Cloudflare Access) and consumer users (Better Auth on Workers). Differentiated only by recipient_type.

2. Architecture: Two Layers

Layer 1 — NotificationPort (packages/notification)

A provider-agnostic interface, mirroring the ModelPort/MemoryPort pattern from AGENT-KERNEL.md. Every delivery channel is an adapter behind this port.

// packages/notification/src/types.ts

export type NotificationChannel = 'in_app' | 'email' | 'sms' | 'slack' | 'web_push';

export type NotificationEventType =
  | 'agent.run.failed'
  | 'agent.run.completed'
  | 'agent.mode.changed'
  | 'approval.required'
  | 'approval.submitted';
  // extend as new system events emerge

export interface Notification {
  id: string;
  recipient_id: string;        // email (BO) or consumer user id
  recipient_type: 'bo_user' | 'consumer_user';
  channel: NotificationChannel;
  event_type: NotificationEventType | string;
  title: string;
  body: string;
  data?: Record<string, unknown>; // structured payload for UI deep-linking
  created_at: string;
  read_at?: string | null;        // in_app only
  delivered_at?: string | null;
}

export interface NotificationPort {
  send(n: Omit<Notification, 'id' | 'created_at'>): Promise<void>;
}

The InAppAdapter writes to D1 notifications and is the only adapter in Phase 1. Future adapters (ResendAdapter, SlackAdapter, TwilioAdapter) implement the same interface and are selected by the service layer based on notification_preferences.

Layer 2 — Delivery Channels (phased)

Why Resend over SendGrid/SES: Resend has a first-class CF Workers SDK that doesn’t require TCP sockets (which Workers can’t open). Consistent with “Cloudflare by default.”

3. D1 Schema (services/console-api/migrations/0007_notifications.sql)

-- Notification inbox — source of truth for all channels; read-state for in_app.
CREATE TABLE IF NOT EXISTS notifications (
  id             TEXT PRIMARY KEY,
  recipient_id   TEXT NOT NULL,
  recipient_type TEXT NOT NULL DEFAULT 'bo_user'
                   CHECK (recipient_type IN ('bo_user', 'consumer_user')),
  channel        TEXT NOT NULL DEFAULT 'in_app',
  event_type     TEXT NOT NULL,
  title          TEXT NOT NULL,
  body           TEXT NOT NULL,
  data           TEXT,            -- JSON
  read_at        TEXT,            -- NULL = unread (in_app only)
  delivered_at   TEXT,
  created_at     TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_notif_recipient ON notifications(recipient_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_notif_unread    ON notifications(recipient_id, read_at)
  WHERE read_at IS NULL;

-- Per-user channel preferences (opt-in/opt-out per event_type × channel).
-- Default = in_app enabled for all event types (no row needed to enable).
CREATE TABLE IF NOT EXISTS notification_preferences (
  user_id    TEXT NOT NULL,
  event_type TEXT NOT NULL,
  channel    TEXT NOT NULL,
  enabled    INTEGER NOT NULL DEFAULT 1,
  PRIMARY KEY (user_id, event_type, channel)
);

4. In-App Delivery: SSE Stream (Phase 1)

No Durable Objects needed at BO scale. The approach is a polling SSE stream:

1. Client opens GET /api/notifications/stream (requires notification:read permission, resolved via withIdentity()).
2. Worker opens a ReadableStream and polls D1 for new unread rows every ~3 s, sending each as data: <json>\n\n.
3. c.executionCtx.waitUntil() keeps the Worker alive while the stream is open.

API routes (services/console-api/src/index.ts)

GET  /api/notifications/stream    → SSE, pushes new notifications in real time
GET  /api/notifications           → paginated list, unread-first (query: limit, before)
POST /api/notifications/:id/read  → mark one read
POST /api/notifications/read-all  → mark all read for the authenticated user

Frontend (apps/console/src/)

5. Event Emission

A notifyBOUsers(db, event_type, title, body, data?) helper (in console-api/src/index.ts):

1. Queries active BO users from users (status = ‘active’).
2. Respects notification_preferences (default: in_app on for all).
3. Batch-inserts into notifications.
4. Runs inside c.executionCtx.waitUntil() — never blocks the HTTP response.

Natural trigger points

6. RBAC

Add notification:read to packages/rbac/src/permissions.ts. All roles receive it (mirrors the audit:read pattern — every authenticated user can read their own notifications).

Mirror the addition in the seed section of migrations/0001_rbac.sql (comment only; D1 is the runtime authority).

7. Consumer-User Extensibility

The recipient_type: 'bo_user' | 'consumer_user' column and the NotificationPort interface are the extension seams. When consumer auth (Better Auth on Workers) ships:

• Pass the consumer user_id as recipient_id with recipient_type: 'consumer_user'.
• Wire web_push/sms adapters behind the same port.
• No schema migration needed.

8. Secrets (Phase 2+)

Provider credentials go into the CF Secrets Store (already used for Langfuse, resolved via the readSecret() helper in console-api). No new pattern — just new entries:

RESEND_API_KEY     → Secrets Store
SLACK_WEBHOOK_URL  → Secrets Store (or KV for per-channel config at runtime)
TWILIO_AUTH_TOKEN  → Secrets Store (Phase 3)

9. Files Summary

10. Verification Checklist

• docker compose up — stack boots clean
• Trigger an agent run → run.finished (failed) → bell shows badge + notification in dropdown
• Mark as read → badge clears
• Flip agent mode → agent.mode.changed notification appears without page refresh
• Confirm D1 notifications rows via wrangler d1 execute
• SSE stream stays open across multiple events (no reconnects)
• notification:read gating: 401 without identity, 403 with wrong role