feat(notifications): in-app notification center backend (phase d.3)

ROADMAP.md

@@ -34,7 +34,7 @@ Forward-looking work for clean-stack. **All SOTA 2026, outside DDD** (DDD reserv
 | **Phase D.5 — Email delivery queue** | **Aug 2026** | `email_message` durable queue + `EmailDeliveryWorker` + `@packages/emails` React Email templates + `sendTemplateBatch` + retention sweep + `email.delivery.exhausted` → 55 total. As-built in [`docs/HISTORY.md`](docs/HISTORY.md). |
 | **Option / Result convention back-fill** | **Aug 2026** | `Result.ok` overloads close the `Result.ok<string>() → undefined` hole; ports across consents, billing, rate-limiter, webhooks, outbox and audit express absence as `Option<T>` instead of `T \| null`, with `null` stopping at the store boundary. No wire format or hash-chain change. As-built in [`docs/HISTORY.md`](docs/HISTORY.md). |
 | **Phase C.3 — Admin & impersonation** | **Aug 2026** | `modules/admin/` (back) + `features/admin-users/` + `features/admin-orgs/` (front) — audited ban / unban / role-change / force-password-reset / revoke-sessions; justified impersonation (reason required, ticketRef optional); two-layer server-side blocklist (BetterAuth hook + 11 `denyImpersonated` Hono routes, incl. `POST /me/policies/accept`); non-dismissable banner with live countdown; transparency email to impersonated user. "Admin" nav entry (MFA-gated, links to `/admin/users`). Legal acceptance gate disabled during impersonation. All UI in English. `APP_URL` promoted to required. 7 `admin.*` events (`compliance`) → **62 total**. As-built in [`docs/HISTORY.md`](docs/HISTORY.md). |
-| **Phase C.4 — API tokens / PATs** | **Aug 2026** | `modules/api-token/` (back) + `features/api-tokens/` (front) — `clean_` + 44-char base58 + CRC32 checksum; HMAC-SHA256 + pepper rotation; org-scoped + expirable; `/settings/tokens` CRUD with `denyImpersonated` on writes (blocklist 11 → 13); `/api/v1` sub-app outside `AppType`; cascade revocation on membership loss; `POST /api/token-scanning/github` (ECDSA P-256); `visibility-map.ts` introduces explicit public/internal classification (28 public + 37 internal). 3 new events → **65 total / 28 public / 37 internal**. As-built in [`docs/HISTORY.md`](docs/HISTORY.md). |
+| **Phase C.4 — API tokens / PATs** | **Aug 2026** | `modules/api-token/` (back) + `features/api-tokens/` (front) — `clean_` + 44-char base58 + CRC32 checksum; HMAC-SHA256 + pepper rotation; org-scoped + expirable; `/settings/tokens` CRUD with `denyImpersonated` on writes (blocklist 11 → 13); `/api/v1` sub-app outside `AppType`; cascade revocation on membership loss; `POST /api/token-scanning/github` (ECDSA P-256); `visibility-map.ts` introduces explicit public/internal classification (28 public + 37 internal). 3 new events → **65 total / 28 public / 37 internal**. As-built in [`docs/HISTORY.md`](docs/HISTORY.md). As-built corrections: 2 preference audit events added post-ship (catalog now **67 / 28 public / 39 internal**). |
 
 ---
 
@@ -77,8 +77,8 @@ As-built record + all decisions in [`docs/HISTORY.md`](docs/HISTORY.md). Per-are
 ### M4 — Operate the product + paying customers
 
 - **C.3** Admin & impersonation ✅ **shipped** (Aug 2026 — `modules/admin/` back + `features/admin-users/` + `features/admin-orgs/` front; audited ban/unban/role/reset/revoke-sessions, justified impersonation, two-layer blocklist (BetterAuth hook + 11 `denyImpersonated` routes incl. policy acceptance), MFA-gated "Admin" nav → `/admin/users`, legal gate disabled during impersonation, UI in English, `APP_URL` required, live banner, transparency email; 7 `admin.*` events → **62 total**. As-built in [`docs/HISTORY.md`](docs/HISTORY.md).)
-- **C.4** API tokens / PATs ✅ **shipped** (Aug 2026 — `modules/api-token/` back + `features/api-tokens/` front; `clean_` + 44-char base58 + 6-char CRC32 checksum; HMAC-SHA256 + pepper rotation (`API_TOKEN_PEPPER` / `API_TOKEN_PEPPER_PREVIOUS` / `API_TOKEN_PEPPER_VERSION`); `/settings/tokens` CRUD (name + scope picker + expiry) with `denyImpersonated` on writes; `/api/v1` sub-app outside `AppType` (token-auth only, no session middleware); cascade revocation on membership loss; `POST /api/token-scanning/github` (ECDSA P-256); 3 events (`api_token.created`, `api_token.revoked`, `api_token.used`) → **65 total / 28 public / 37 internal**. As-built in [`docs/HISTORY.md`](docs/HISTORY.md).)
-- **D.2** OpenAPI auto-docs (`@hono/zod-openapi` + Scalar UI at `/api/docs`) — after PATs ship (customers integrate).
+- **C.4** API tokens / PATs ✅ **shipped** (Aug 2026 — `modules/api-token/` back + `features/api-tokens/` front; `clean_` + 44-char base58 + 6-char CRC32 checksum; HMAC-SHA256 + pepper rotation (`API_TOKEN_PEPPER` / `API_TOKEN_PEPPER_PREVIOUS` / `API_TOKEN_PEPPER_VERSION`); `/settings/tokens` CRUD (name + scope picker + expiry) with `denyImpersonated` on writes; `/api/v1` sub-app outside `AppType` (token-auth only, no session middleware); cascade revocation on membership loss; `POST /api/token-scanning/github` (ECDSA P-256); 3 events (`api_token.created`, `api_token.revoked`, `api_token.used`) → **65 total / 28 public / 37 internal**. As-built in [`docs/HISTORY.md`](docs/HISTORY.md). As-built corrections: 2 preference audit events added post-ship (catalog now **67 / 28 public / 39 internal**).)
+- **D.2** OpenAPI auto-docs `[deferred]` (2026-08-07) — **deferred until the API is actually opened to third parties**. The SOTA options (`@hono/zod-openapi` route rewrite, `hono-openapi` decorators) all demand restructuring every route registration to carry doc metadata, for a surface that has no external consumer yet. Docs that nobody reads still pay the drift tax on every route change. Revisit when a clone publishes `/api/v1` externally — the `zValidator(...)` schemas that make auto-derivation possible aren't going anywhere.
 - **D.3** In-app notification center — `<Bell />` + `/settings/notifications`. Handler = 1-line `onEvent(...)` via event-driven foundation.
 - **D.5** Email delivery ✅ **shipped** (Aug 2026 — see ✅ table; `email_message` durable queue + `EmailDeliveryWorker` polling + `@packages/emails` in-repo React Email templates + `sendTemplateBatch` + retention sweep. 1 new internal event → **55 total / 50 subscribable / 5 internal**. As-built in [`docs/HISTORY.md`](docs/HISTORY.md).)
 
@@ -283,11 +283,21 @@ HIPAA tooling, real-time WebSocket/SSE bus, third-party app marketplace, A/B tes
 
 **Why**: transactional emails are async; users miss them. An in-app inbox is the SaaS-default pattern (Linear, GitHub, Stripe). Persistent, mark-as-read, deep-linked.
 
-- [ ] DB schema `notification(id, userId FK, organizationId FK nullable, kind, payload jsonb, readAt nullable, createdAt)`.
-- [ ] Bell icon in app shell with unread count badge — TanStack Query subscription + `BroadcastChannel` for cross-tab sync (reuse `auth-broadcast` pattern).
-- [ ] `/settings/notifications` — preferences per category (security / billing / mentions / digests), per channel (email vs in-app vs both).
-- [ ] Domain event handler pattern: `OrganizationInvitationSent → InAppNotificationHandler` writes a notification row + dispatches WS-style refetch on the recipient's bell query.
-- [ ] Out of scope: native push (mobile / browser). Phase F.
+**Design settled 2026-08-07** (full spec local, not versioned — `docs/superpowers/` is gitignored). SOTA baseline: Knock / Novu / Courier / SuprSend converge on six primitives (workflow engine, three-level preferences, batching + digest, critical bypass, throttling with dedup, inbox over feed). Two deliberate divergences: SSE instead of WebSocket, and a typed `Record<EventType, Config>` instead of a workflow DSL — the outbox already resolves fan-out, which is the only reason those platforms need a DSL at all.
+
+- [x] **Fan-out = `NotificationFanoutSubscriber implements OutboxSubscriber`**, in the dispatch TX beside `AuditEventSubscriber` / `WebhookFanoutSubscriber` — **not** the `onEvent(...)` post-commit handler this spec originally suggested. **Why**: `onEvent` is best-effort and isolated, so a lost notification fails silently; and batching needs a transactional write (two concurrent events on one batch key without a lock produce two batches). Constraint: one `INSERT ... SELECT` joining members × preferences, never N inserts in a loop.
+- [x] **Recipients resolved by capability, never by role tuple.** `type Audience = "self" | "actor" | "org:all" | { can: OrgPermissions }`. **Why**: `audience: "org:admins"` is exactly the hardcoded tuple org-scoping rule #6 forbids — it duplicates a decision owned by `@packages/access-control` and drifts the moment a role is added. Costs nothing at runtime: roles are static code, so `ORG_ROLES.filter(r => authorizeRole(r, perms))` resolves once at boot, leaving `WHERE member.role = ANY($1)`. **Trap**: `billing:["manage"]` is owner-only (`access-control/src/index.ts:33`) — a notification asks *who needs to know*, not *who may act*, so `read` is almost always the right level.
+- [x] **`notification-map.ts`** — third projection of the catalog after `visibility-map` (webhooks) and `retention-map` (purge). Absent event = no notification (the default: most of the 67 events are audit-only). `forced: true` short-circuits preferences *and* batching, before either is evaluated (the SOTA critical bypass).
+- [x] **Batching splits per channel.** In-app writes the row immediately and groups on read by `groupKey` (Linear's "X and 3 others"); email batches on write via `emailPendingAt` / `emailSentAt` columns. **Why no `notification_batch` table**: SaaS platforms need one because they don't own their customers' storage. We do — so the batch stays a query instead of state that can desync. Frequency preference (`immediate` / `hourly` / `daily`) supplies the window, which makes the scheduled digest the same cron with a longer one.
+- [x] **Throttling = partial unique index** `(userId, dedupKey) WHERE dedup_key IS NOT NULL`, with the window baked into the key (`<eventType>:<resourceId>:<hourBucket>`). Dedup happens at insert inside the TX, so it's concurrency-correct; a counter would need an extra lock for the same guarantee.
+- [x] Tables: `notification(id, userId, organizationId?, category, eventType, groupKey, dedupKey?, payload, readAt, emailPendingAt, emailSentAt, createdAt)` + `notification_preference(scope 'user'|'org', scopeId, category, channel, enabled, frequency, locked, updatedAt)`. Partial index `ON notification (user_id) WHERE read_at IS NULL` for the unread count. **`organizationId` nullable is a documented exception to org-scoping rule #3** — a notification is user-scoped by nature (`user.password_changed` belongs to no org).
+- [x] **Preference cascade**: org lock → user preference → map default, with `forced` bypassing all three. Org-level is the SOTA B2B differentiator (Knock) and near-free here.
+- [x] **SSE stream carries a signal, never data.** `GET /notifications/stream` (Hono `streamSSE`) + Postgres trigger `pg_notify('notification_created', user_id)`, mirroring `outbox-dispatcher.service.ts:104`. **Why signal-only**: a reconnect just fires `invalidateQueries`, which by definition catches up — killing `Last-Event-ID`, replay, and merge logic in one stroke, and degrading naturally to polling if the stream dies. **`NotificationStreamHub` holds one `LISTEN` connection per instance**, never one per client (that exhausts the pool at a few hundred connected users); multi-instance works broker-free since `pg_notify` broadcasts to every listener. Heartbeat 25 s (Caddy timeouts). Client uses `fetch` + `ReadableStream`, **not `EventSource`** — it can't carry an `Authorization` header, which would break F.1's Capacitor bearer.
+- [ ] Front: `<NotificationBell />` in `app-shell`, `/settings/notifications` in `settingsLayout`; org defaults as a card inside `/settings/organization` (a route under `orgScopeLayout` would collide — it flattens children under `settings/`). Polling survives only as fallback: `refetchInterval: streamConnected ? false : 30_000`. **Promotes `auth-broadcast.ts` into `createBroadcastChannel<T>(name)`** — 2nd occurrence triggers rule #2.
+- [x] Crons on the existing `/internal/*` rail: `flush-notification-emails` (1 min — so `immediate` means "next tick"; true instant is the SSE's job) + `sweep-notifications` (**read rows only** — an unread notification outlives retention, same logic as D.5's `failed` rows).
+- [x] **Notification creation emits no event.** D.3 consumes the catalog for fan-out — a `notification.created` event would loop with its own subscriber. Preference *mutations* (PUT /preferences, PUT /org-preferences) do emit audit events (`notification.preference.updated`, `notification.org_preference.updated`) because they are persistent state changes, not read projections. Catalog **67 / 28 public / 39 internal**.
+- [ ] Out of scope: native push (mobile / browser, Phase F), generalized real-time bus, workflow DSL.
+- [ ] **Resend Broadcasts/Audiences rejected**: marketing one-to-many, orthogonal to event-driven one-to-one. Topics (Resend-side email preferences) also rejected — it would hand a product decision to the vendor and covers only one channel, when D.3 exists precisely to arbitrate *between* channels. Revisit Broadcasts at E.2 (newsletter / marketing digests).
 
 ---
 
@@ -382,9 +392,11 @@ Shipping a status page before there are customer integrations is theatre.
 
 ---
 
-## OpenAPI schema docs — **Phase D.2**
+## OpenAPI schema docs — **Phase D.2** ⏸️ DEFERRED (2026-08-07)
 
-**Why**: the moment Phase C.12 (PATs) ships, customers will integrate. They need typed docs. Manual maintenance = drift = support tickets.
+**Deferred**: the API isn't open to third parties yet, and every SOTA approach (`@hono/zod-openapi`'s `createRoute` rewrite, `hono-openapi`'s per-route decorators) requires restructuring route registration across the codebase to carry doc metadata. That cost buys nothing until an external consumer exists, and unread docs still drift on every route change. **Activation trigger**: a clone publishes `/api/v1` to third parties. The `zValidator(...)` schemas the derivation depends on stay in place, so nothing rots meanwhile.
+
+**Why (original)**: the moment Phase C.12 (PATs) ships, customers will integrate. They need typed docs. Manual maintenance = drift = support tickets.
 
 - [ ] `@hono/zod-openapi` middleware to auto-derive OpenAPI 3.1 spec from existing `zValidator(...)` calls + route registrations.
 - [ ] `/api/docs` route serves Scalar UI (lightweight, Stripe-aesthetic).

apps/api/src/container.ts

@@ -7,6 +7,7 @@ import { auditLogModule } from "./modules/audit-log/module";
 import { billingModule } from "./modules/billing/module";
 import { consentModule } from "./modules/consents/module";
 import { healthModule } from "./modules/health/module";
+import { notificationsModule } from "./modules/notifications/module";
 import { policyModule } from "./modules/policies/module";
 import { quotaModule } from "./modules/quotas/module";
 import { rgpdModule } from "./modules/rgpd/module";
@@ -32,6 +33,8 @@ import { QueuedEmailService } from "./shared/services/email.service";
 import { EmailDeliveryWorker } from "./shared/services/email-delivery-worker.service";
 import { HibpPasswordBreachService } from "./shared/services/hibp-password-breach.service";
 import { NoOpInstrumentation } from "./shared/services/noop-instrumentation";
+import { NotificationFanoutSubscriber } from "./shared/services/notification-fanout-subscriber";
+import { NotificationStreamHub } from "./shared/services/notification-stream-hub";
 import { OutboxDispatcher } from "./shared/services/outbox-dispatcher.service";
 import {
   RateLimiterFlexibleAdapter,
@@ -54,9 +57,11 @@ declare module "inwire" {
     IRateLimiter: IRateLimiter;
     AuditEventSubscriber: AuditEventSubscriber;
     WebhookFanoutSubscriber: WebhookFanoutSubscriber;
+    NotificationFanoutSubscriber: NotificationFanoutSubscriber;
     OutboxDispatcher: OutboxDispatcher;
     BackupCodeUsedNotifier: EventHandler;
     EmailDeliveryWorker: EmailDeliveryWorker;
+    NotificationStreamHub: NotificationStreamHub;
   }
 }
 
@@ -105,13 +110,14 @@ export const di = container()
   )
   .add("AuditEventSubscriber", (c) => new AuditEventSubscriber(c.IInstrumentation))
   .add("WebhookFanoutSubscriber", (c) => new WebhookFanoutSubscriber(c.IInstrumentation))
+  .add("NotificationFanoutSubscriber", (c) => new NotificationFanoutSubscriber(c.IInstrumentation))
   .add("BackupCodeUsedNotifier", (c) => backupCodeUsedNotifier({ IEmailService: c.IEmailService }))
   .add(
     "OutboxDispatcher",
     (c) =>
       new OutboxDispatcher(
         c.IOutboxRepository,
-        [c.AuditEventSubscriber, c.WebhookFanoutSubscriber],
+        [c.AuditEventSubscriber, c.WebhookFanoutSubscriber, c.NotificationFanoutSubscriber],
         logger,
         env.DATABASE_URL,
         c.IInstrumentation,
@@ -128,4 +134,6 @@ export const di = container()
   .addModule(consentModule)
   .addModule(quotaModule)
   .addModule(billingModule)
+  .addModule(notificationsModule)
+  .add("NotificationStreamHub", () => new NotificationStreamHub(logger, env.DATABASE_URL))
   .build();

apps/api/src/index.ts

@@ -18,6 +18,7 @@ import { billingRoutes } from "./modules/billing/routes";
 import { consentRoutes } from "./modules/consents/routes";
 import { healthInternalRoutes } from "./modules/health/internal.routes";
 import { healthRoutes } from "./modules/health/routes";
+import { notificationsRoutes } from "./modules/notifications/routes";
 import { policyRoutes } from "./modules/policies/routes";
 import { rgpdInternalRoutes } from "./modules/rgpd/internal.routes";
 import { rgpdMeRoutes } from "./modules/rgpd/routes";
@@ -26,9 +27,11 @@ import { webhooksRoutes } from "./modules/webhooks/routes";
 import { createPublicApiV1 } from "./public-api";
 import { env } from "./shared/env";
 import { cspReportCors, makeCspReportApp } from "./shared/internal-routes/csp-report.route";
+import { flushNotificationEmailsRoutes } from "./shared/internal-routes/flush-notification-emails.route";
 import { sweepAuditLogRoutes } from "./shared/internal-routes/sweep-audit-log.route";
 import { sweepConsentsRoutes } from "./shared/internal-routes/sweep-consents.route";
 import { sweepEmailMessagesRoutes } from "./shared/internal-routes/sweep-email-messages.route";
+import { sweepNotificationsRoutes } from "./shared/internal-routes/sweep-notifications.route";
 import { sweepOutboxRoutes } from "./shared/internal-routes/sweep-outbox.route";
 import { sweepWebhookDeliveryRoutes } from "./shared/internal-routes/sweep-webhook-delivery.route";
 import { logger } from "./shared/logger";
@@ -194,6 +197,8 @@ app.route("/internal", sweepAuditLogRoutes);
 app.route("/internal", sweepWebhookDeliveryRoutes);
 app.route("/internal", sweepConsentsRoutes);
 app.route("/internal", sweepEmailMessagesRoutes);
+app.route("/internal", sweepNotificationsRoutes);
+app.route("/internal", flushNotificationEmailsRoutes);
 
 app.route(
   "/api/v1",
@@ -242,7 +247,8 @@ const routes = app
   .route("/settings/tokens", apiTokenRoutes)
   .route("/settings/webhooks", webhooksRoutes)
   .route("/consents", consentRoutes)
-  .route("/billing", billingRoutes);
+  .route("/billing", billingRoutes)
+  .route("/notifications", notificationsRoutes);
 
 app.onError(createErrorHandler(di.IInstrumentation));
 
@@ -252,6 +258,7 @@ await di.preload();
 await di.OutboxDispatcher.start(di as unknown as Record<string, unknown>);
 await di.WebhookDeliveryWorker.start();
 await di.EmailDeliveryWorker.start();
+await di.NotificationStreamHub.start();
 lifecycleState.markStarted();
 
 const SHUTDOWN_STEP_TIMEOUT_MS = 25_000;
@@ -280,6 +287,7 @@ const shutdown = async (signal: string) => {
     stopWithTimeout("webhookDeliveryWorker", () => di.WebhookDeliveryWorker.stop()),
     stopWithTimeout("emailDeliveryWorker", () => di.EmailDeliveryWorker.stop()),
     stopWithTimeout("outboxDispatcher", () => di.OutboxDispatcher.stop()),
+    stopWithTimeout("notificationStreamHub", () => di.NotificationStreamHub.stop()),
   ]);
   process.exit(0);
 };