diff --git a/ROADMAP.md b/ROADMAP.md index 28d331ef..3c5b1e52 100644 --- a/ROADMAP.md +++ b/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() → undefined` hole; ports across consents, billing, rate-limiter, webhooks, outbox and audit express absence as `Option` 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 — `` + `/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` 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 (`::`). 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: `` 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(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). diff --git a/apps/api/src/container.ts b/apps/api/src/container.ts index 76062e4c..d1ce59c8 100644 --- a/apps/api/src/container.ts +++ b/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(); diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index 24c645c5..9bfe0db5 100644 --- a/apps/api/src/index.ts +++ b/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); 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); }; diff --git a/apps/api/src/modules/admin/__TESTS__/admin-impersonation.routes.test.ts b/apps/api/src/modules/admin/__TESTS__/admin-impersonation.routes.test.ts index e1f39e34..157238eb 100644 --- a/apps/api/src/modules/admin/__TESTS__/admin-impersonation.routes.test.ts +++ b/apps/api/src/modules/admin/__TESTS__/admin-impersonation.routes.test.ts @@ -68,6 +68,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, emailSchema: {}, schema: {}, TransactionService: class {}, @@ -90,7 +98,7 @@ mock.module("@packages/drizzle", () => ({ like: mk("like"), count: mk("count"), arrayContains: mk("arrayContains"), - sql: Object.assign(mk("sql"), { raw: mk("sql.raw") }), + sql: Object.assign(mk("sql"), { raw: mk("sql.raw"), identifier: () => ({}) }), })); mock.module("hono/bun", () => ({ diff --git a/apps/api/src/modules/admin/__TESTS__/admin-org-query.test.ts b/apps/api/src/modules/admin/__TESTS__/admin-org-query.test.ts index ebf9103f..f3a7a5e7 100644 --- a/apps/api/src/modules/admin/__TESTS__/admin-org-query.test.ts +++ b/apps/api/src/modules/admin/__TESTS__/admin-org-query.test.ts @@ -106,6 +106,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, emailSchema: {}, schema: {}, TransactionService: class {}, @@ -128,7 +136,7 @@ mock.module("@packages/drizzle", () => ({ like: mk("like"), count: mk("count"), arrayContains: mk("arrayContains"), - sql: Object.assign(mk("sql"), { raw: mk("sql.raw") }), + sql: Object.assign(mk("sql"), { raw: mk("sql.raw"), identifier: () => ({}) }), })); const { AdminQueryService } = await import("../application/services/admin-query.service"); diff --git a/apps/api/src/modules/admin/__TESTS__/admin-query.service.test.ts b/apps/api/src/modules/admin/__TESTS__/admin-query.service.test.ts index e5af24b7..3333bbac 100644 --- a/apps/api/src/modules/admin/__TESTS__/admin-query.service.test.ts +++ b/apps/api/src/modules/admin/__TESTS__/admin-query.service.test.ts @@ -119,6 +119,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, emailSchema: {}, schema: {}, TransactionService: class {}, @@ -141,7 +149,7 @@ mock.module("@packages/drizzle", () => ({ like: mk("like"), count: mk("count"), arrayContains: mk("arrayContains"), - sql: Object.assign(mk("sql"), { raw: mk("sql.raw") }), + sql: Object.assign(mk("sql"), { raw: mk("sql.raw"), identifier: () => ({}) }), })); const { AdminQueryService } = await import("../application/services/admin-query.service"); diff --git a/apps/api/src/modules/admin/__TESTS__/drizzle-admin-user.store.test.ts b/apps/api/src/modules/admin/__TESTS__/drizzle-admin-user.store.test.ts index 6225e3f4..2daf9ada 100644 --- a/apps/api/src/modules/admin/__TESTS__/drizzle-admin-user.store.test.ts +++ b/apps/api/src/modules/admin/__TESTS__/drizzle-admin-user.store.test.ts @@ -106,6 +106,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, emailSchema: {}, schema: {}, TransactionService: class {}, @@ -128,7 +136,7 @@ mock.module("@packages/drizzle", () => ({ like: mk("like"), count: mk("count"), arrayContains: mk("arrayContains"), - sql: Object.assign(mk("sql"), { raw: mk("sql.raw") }), + sql: Object.assign(mk("sql"), { raw: mk("sql.raw"), identifier: () => ({}) }), })); const { DrizzleAdminUserStore } = await import( diff --git a/apps/api/src/modules/api-token/__TESTS__/drizzle-api-token.repository.test.ts b/apps/api/src/modules/api-token/__TESTS__/drizzle-api-token.repository.test.ts index 010ed0dd..50a605db 100644 --- a/apps/api/src/modules/api-token/__TESTS__/drizzle-api-token.repository.test.ts +++ b/apps/api/src/modules/api-token/__TESTS__/drizzle-api-token.repository.test.ts @@ -65,6 +65,7 @@ mock.module("@packages/drizzle", () => ({ arrayContains: () => ({}), sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { raw: () => ({}), + identifier: () => ({}), }), apiTokenSchema: { apiToken: { @@ -125,6 +126,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, emailSchema: {}, schema: {}, TransactionService: class {}, diff --git a/apps/api/src/modules/notifications/__TESTS__/notification-preference.service.test.ts b/apps/api/src/modules/notifications/__TESTS__/notification-preference.service.test.ts new file mode 100644 index 00000000..7e4b9a92 --- /dev/null +++ b/apps/api/src/modules/notifications/__TESTS__/notification-preference.service.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, test } from "bun:test"; +import { Option, Result } from "@packages/ddd-kit"; +import type { + INotificationStore, + PreferenceRecord, + PreferenceScope, +} from "../application/ports/notification.port"; +import { NotificationPreferenceService } from "../application/services/notification-preference.service"; + +function storeWith(preferences: PreferenceRecord[]): INotificationStore { + return { + listPreferences: async (scope: PreferenceScope, scopeId: string) => + Result.ok(preferences.filter((p) => p.scope === scope && p.scopeId === scopeId)), + } as unknown as INotificationStore; +} + +const pref = (over: Partial): PreferenceRecord => ({ + scope: "user", + scopeId: "u1", + category: "billing", + channel: "email", + enabled: true, + frequency: "immediate", + locked: false, + ...over, +}); + +describe("NotificationPreferenceService.resolve", () => { + test("sans preference, retombe sur le defaut actif", async () => { + const service = new NotificationPreferenceService(storeWith([])); + const result = await service.resolve("u1", Option.none(), "billing", "email"); + expect(result.getValue()).toBe(true); + }); + + test("la preference user prime sur le defaut", async () => { + const service = new NotificationPreferenceService(storeWith([pref({ enabled: false })])); + const result = await service.resolve("u1", Option.none(), "billing", "email"); + expect(result.getValue()).toBe(false); + }); + + test("un lock org ecrase la preference user", async () => { + const service = new NotificationPreferenceService( + storeWith([ + pref({ enabled: true }), + pref({ scope: "org", scopeId: "org-1", enabled: false, locked: true }), + ]), + ); + const result = await service.resolve("u1", Option.some("org-1"), "billing", "email"); + expect(result.getValue()).toBe(false); + }); + + test("une preference org non verrouillee ne prime pas sur le choix user", async () => { + const service = new NotificationPreferenceService( + storeWith([ + pref({ enabled: true }), + pref({ scope: "org", scopeId: "org-1", enabled: false, locked: false }), + ]), + ); + const result = await service.resolve("u1", Option.some("org-1"), "billing", "email"); + expect(result.getValue()).toBe(true); + }); +}); diff --git a/apps/api/src/modules/notifications/__TESTS__/routes.test.ts b/apps/api/src/modules/notifications/__TESTS__/routes.test.ts new file mode 100644 index 00000000..c1dbcbdb --- /dev/null +++ b/apps/api/src/modules/notifications/__TESTS__/routes.test.ts @@ -0,0 +1,320 @@ +import { describe, expect, it, mock } from "bun:test"; +import { Option, Result } from "@packages/ddd-kit"; +import type { + NotificationError, + NotificationRecord, + PreferenceRecord, +} from "../application/ports/notification.port"; +import { listQuerySchema, markReadSchema, preferenceSchema } from "../notifications.schema"; + +// ── Schema tests ─────────────────────────────────────────────────────────── + +describe("schemas de notification", () => { + it("limit par defaut a 20 et plafonne a 50", () => { + expect(listQuerySchema.parse({}).limit).toBe(20); + expect(listQuerySchema.safeParse({ limit: 51 }).success).toBe(false); + }); + + it("markRead refuse un tableau vide", () => { + expect(markReadSchema.safeParse({ ids: [] }).success).toBe(false); + }); + + it("preference refuse une categorie inconnue", () => { + expect( + preferenceSchema.safeParse({ category: "inexistante", channel: "email", enabled: true }) + .success, + ).toBe(false); + }); +}); + +// ── Route tests ──────────────────────────────────────────────────────────── + +const NOTIFICATION: NotificationRecord = { + id: "notif-1", + userId: "user-1", + organizationId: Option.none(), + category: "billing", + eventType: "billing.subscription.created", + groupKey: Option.none(), + payload: {}, + readAt: Option.none(), + createdAt: new Date("2024-01-01"), +}; + +const PREFERENCE: PreferenceRecord = { + scope: "user", + scopeId: "user-1", + category: "billing", + channel: "email", + enabled: true, + frequency: "immediate", + locked: false, +}; + +const mockList = mock( + async (): Promise> => Result.ok([NOTIFICATION]), +); +const mockUnreadCount = mock(async (): Promise> => Result.ok(3)); +const mockMarkRead = mock(async (): Promise> => Result.ok()); +const mockMarkAllRead = mock(async (): Promise> => Result.ok()); +const mockListPreferences = mock( + async (): Promise> => Result.ok([PREFERENCE]), +); +const mockUpsertPreference = mock( + async (): Promise> => Result.ok(), +); + +const mockEnqueue = mock(async () => {}); + +mock.module("../../../container", () => ({ + di: { + INotificationStore: { + list: mockList, + unreadCount: mockUnreadCount, + markRead: mockMarkRead, + markAllRead: mockMarkAllRead, + listPreferences: mockListPreferences, + upsertPreference: mockUpsertPreference, + }, + IOutboxRepository: { + enqueue: mockEnqueue, + }, + }, +})); + +let currentSession: Record = {}; +let allowOrgPermission = true; + +mock.module("../../../shared/middleware/auth.middleware", () => ({ + // biome-ignore lint/suspicious/noExplicitAny: test stub + requireAuth: async (c: any, next: () => Promise) => { + c.set("user", { id: "user-1" }); + c.set("session", currentSession); + await next(); + }, + AuthVariables: {}, +})); + +mock.module("../../../shared/middleware/org.middleware", () => ({ + // biome-ignore lint/suspicious/noExplicitAny: test stub + requireOrg: async (c: any, next: () => Promise) => { + const orgId = (c.get("session") as Record)?.activeOrganizationId; + if (!orgId) { + const { HTTPException } = await import("hono/http-exception"); + throw new HTTPException(403, { message: "No active organization" }); + } + c.set("orgId", orgId); + await next(); + }, + requireOrgPermission: () => async (_c: unknown, next: () => Promise) => { + if (!allowOrgPermission) { + const { HTTPException } = await import("hono/http-exception"); + throw new HTTPException(403, { message: "Insufficient permission" }); + } + await next(); + }, +})); + +const { notificationsRoutes } = await import("../routes"); +const { Hono } = await import("hono"); +const { createErrorHandler } = await import("../../../shared/middleware/error.middleware"); +const { NoOpInstrumentation } = await import("../../../shared/services/noop-instrumentation"); + +function makeApp() { + const app = new Hono<{ Variables: { requestId: string } }>(); + app.use("*", async (c, next) => { + c.set("requestId", "req-test"); + await next(); + }); + app.onError(createErrorHandler(new NoOpInstrumentation())); + app.route("/notifications", notificationsRoutes); + return app; +} + +describe("GET /notifications - list", () => { + it("renvoie les items avec les Options serialises en null", async () => { + currentSession = {}; + const app = makeApp(); + const res = await app.request("/notifications"); + expect(res.status).toBe(200); + // biome-ignore lint/suspicious/noExplicitAny: test assertion + const body = (await res.json()) as any; + expect(Array.isArray(body.items)).toBe(true); + expect(body.items[0].organizationId).toBeNull(); + expect(body.items[0].groupKey).toBeNull(); + expect(body.items[0].readAt).toBeNull(); + }); +}); + +describe("GET /notifications/unread-count", () => { + it("renvoie le compte des non lues", async () => { + currentSession = {}; + const app = makeApp(); + const res = await app.request("/notifications/unread-count"); + expect(res.status).toBe(200); + // biome-ignore lint/suspicious/noExplicitAny: test assertion + const body = (await res.json()) as any; + expect(body.count).toBe(3); + }); +}); + +describe("POST /notifications/read - mark-read", () => { + it("retourne ok quand les ids sont valides", async () => { + currentSession = {}; + mockMarkRead.mockClear(); + const app = makeApp(); + const res = await app.request("/notifications/read", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ids: ["notif-1"] }), + }); + expect(res.status).toBe(200); + expect(mockMarkRead).toHaveBeenCalledTimes(1); + }); + + it("rejette une session impersonnifiee (403)", async () => { + currentSession = { impersonatedBy: "admin-99" }; + mockMarkRead.mockClear(); + const app = makeApp(); + const res = await app.request("/notifications/read", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ids: ["notif-1"] }), + }); + expect(res.status).toBe(403); + expect(mockMarkRead).not.toHaveBeenCalled(); + }); + + it("rejette un tableau vide (400)", async () => { + currentSession = {}; + const app = makeApp(); + const res = await app.request("/notifications/read", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ids: [] }), + }); + expect(res.status).toBe(400); + }); +}); + +describe("POST /notifications/read-all", () => { + it("rejette une session impersonnifiee (403)", async () => { + currentSession = { impersonatedBy: "admin-99" }; + mockMarkAllRead.mockClear(); + const app = makeApp(); + const res = await app.request("/notifications/read-all", { method: "POST" }); + expect(res.status).toBe(403); + expect(mockMarkAllRead).not.toHaveBeenCalled(); + }); + + it("retourne ok pour une session normale", async () => { + currentSession = {}; + const app = makeApp(); + const res = await app.request("/notifications/read-all", { method: "POST" }); + expect(res.status).toBe(200); + }); +}); + +describe("GET /notifications/preferences", () => { + it("renvoie les preferences utilisateur", async () => { + currentSession = {}; + const app = makeApp(); + const res = await app.request("/notifications/preferences"); + expect(res.status).toBe(200); + // biome-ignore lint/suspicious/noExplicitAny: test assertion + const body = (await res.json()) as any; + expect(Array.isArray(body.items)).toBe(true); + }); +}); + +describe("PUT /notifications/preferences", () => { + it("sauvegarde la preference et retourne ok", async () => { + currentSession = {}; + mockUpsertPreference.mockClear(); + const app = makeApp(); + const res = await app.request("/notifications/preferences", { + method: "PUT", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ category: "billing", channel: "email", enabled: false }), + }); + expect(res.status).toBe(200); + expect(mockUpsertPreference).toHaveBeenCalledTimes(1); + }); + + it("rejette une session impersonnifiee (403)", async () => { + currentSession = { impersonatedBy: "admin-99" }; + const app = makeApp(); + const res = await app.request("/notifications/preferences", { + method: "PUT", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ category: "billing", channel: "email", enabled: false }), + }); + expect(res.status).toBe(403); + }); +}); + +describe("GET /notifications/org-preferences", () => { + it("exige un org actif (403 sans org)", async () => { + currentSession = {}; + allowOrgPermission = true; + const app = makeApp(); + const res = await app.request("/notifications/org-preferences"); + expect(res.status).toBe(403); + }); + + it("rejette un membre sans la capability organization:update (403)", async () => { + currentSession = { activeOrganizationId: "org-1" }; + allowOrgPermission = false; + const app = makeApp(); + const res = await app.request("/notifications/org-preferences"); + expect(res.status).toBe(403); + allowOrgPermission = true; + }); + + it("renvoie les preferences org quand la capability est presente", async () => { + currentSession = { activeOrganizationId: "org-1" }; + allowOrgPermission = true; + const app = makeApp(); + const res = await app.request("/notifications/org-preferences"); + expect(res.status).toBe(200); + // biome-ignore lint/suspicious/noExplicitAny: test assertion + const body = (await res.json()) as any; + expect(Array.isArray(body.items)).toBe(true); + }); +}); + +describe("PUT /notifications/org-preferences", () => { + it("sauvegarde la preference org et retourne ok", async () => { + currentSession = { activeOrganizationId: "org-1" }; + mockUpsertPreference.mockClear(); + const app = makeApp(); + const res = await app.request("/notifications/org-preferences", { + method: "PUT", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + category: "security", + channel: "in_app", + enabled: true, + locked: true, + }), + }); + expect(res.status).toBe(200); + expect(mockUpsertPreference).toHaveBeenCalledTimes(1); + }); + + it("rejette une session impersonnifiee (403)", async () => { + currentSession = { activeOrganizationId: "org-1", impersonatedBy: "admin-99" }; + const app = makeApp(); + const res = await app.request("/notifications/org-preferences", { + method: "PUT", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + category: "security", + channel: "in_app", + enabled: true, + locked: true, + }), + }); + expect(res.status).toBe(403); + }); +}); diff --git a/apps/api/src/modules/notifications/application/ports/notification.port.ts b/apps/api/src/modules/notifications/application/ports/notification.port.ts new file mode 100644 index 00000000..55a1c5d4 --- /dev/null +++ b/apps/api/src/modules/notifications/application/ports/notification.port.ts @@ -0,0 +1,53 @@ +import type { Option, Result } from "@packages/ddd-kit"; +import type { + NotificationChannel, + NotificationFrequency, + NotificationPreferenceScope, +} from "@packages/events"; + +export type { NotificationChannel, NotificationFrequency }; +export type PreferenceScope = NotificationPreferenceScope; + +export type NotificationRecord = { + id: string; + userId: string; + organizationId: Option; + category: string; + eventType: string; + groupKey: Option; + payload: unknown; + readAt: Option; + createdAt: Date; +}; + +export type PreferenceRecord = { + scope: PreferenceScope; + scopeId: string; + category: string; + channel: NotificationChannel; + enabled: boolean; + frequency: NotificationFrequency; + locked: boolean; +}; + +export type PreferenceInput = PreferenceRecord; + +export type NotificationError = + | { code: "NOTIFICATION_PROVIDER_FAILURE"; message: string } + | { code: "NOTIFICATION_WRITE_PROVIDER_FAILURE"; message: string }; + +export interface INotificationStore { + list( + userId: string, + cursor: Option, + limit: number, + ): Promise>; + unreadCount(userId: string): Promise>; + markRead(userId: string, ids: string[], now: Date): Promise>; + markAllRead(userId: string, now: Date): Promise>; + listPreferences( + scope: PreferenceScope, + scopeId: string, + ): Promise>; + upsertPreference(input: PreferenceInput): Promise>; +} diff --git a/apps/api/src/modules/notifications/application/services/notification-preference.service.ts b/apps/api/src/modules/notifications/application/services/notification-preference.service.ts new file mode 100644 index 00000000..d4555c49 --- /dev/null +++ b/apps/api/src/modules/notifications/application/services/notification-preference.service.ts @@ -0,0 +1,34 @@ +import { type Option, Result } from "@packages/ddd-kit"; +import type { + INotificationStore, + NotificationChannel, + NotificationError, +} from "../ports/notification.port"; + +export class NotificationPreferenceService { + constructor(private readonly store: INotificationStore) {} + + async resolve( + userId: string, + organizationId: Option, + category: string, + channel: NotificationChannel, + ): Promise> { + if (organizationId.isSome()) { + const orgPreferences = await this.store.listPreferences("org", organizationId.unwrap()); + if (orgPreferences.isFailure) return Result.fail(orgPreferences.getError()); + const locked = orgPreferences + .getValue() + .find((p) => p.category === category && p.channel === channel && p.locked); + if (locked) return Result.ok(locked.enabled); + } + + const userPreferences = await this.store.listPreferences("user", userId); + if (userPreferences.isFailure) return Result.fail(userPreferences.getError()); + const own = userPreferences + .getValue() + .find((p) => p.category === category && p.channel === channel); + + return Result.ok(own ? own.enabled : true); + } +} diff --git a/apps/api/src/modules/notifications/infrastructure/repositories/drizzle-notification.store.ts b/apps/api/src/modules/notifications/infrastructure/repositories/drizzle-notification.store.ts new file mode 100644 index 00000000..d884fec4 --- /dev/null +++ b/apps/api/src/modules/notifications/infrastructure/repositories/drizzle-notification.store.ts @@ -0,0 +1,233 @@ +import { Option, Result } from "@packages/ddd-kit"; +import { + and, + count, + db, + desc, + eq, + inArray, + isNull, + lt, + notificationSchema, +} from "@packages/drizzle"; +import type { IInstrumentation } from "../../../../shared/ports/instrumentation.port"; +import type { + INotificationStore, + NotificationError, + NotificationRecord, + PreferenceInput, + PreferenceRecord, + PreferenceScope, +} from "../../application/ports/notification.port"; + +const dbAttrs = { "db.system.name": "postgresql" } as const; + +function readFailure(err: unknown): NotificationError { + return { + code: "NOTIFICATION_PROVIDER_FAILURE", + message: err instanceof Error ? err.message : "unknown", + }; +} + +function writeFailure(err: unknown): NotificationError { + return { + code: "NOTIFICATION_WRITE_PROVIDER_FAILURE", + message: err instanceof Error ? err.message : "unknown", + }; +} + +export class DrizzleNotificationStore implements INotificationStore { + constructor(private readonly instrumentation: IInstrumentation) {} + + async list( + userId: string, + cursor: Option, + limit: number, + ): Promise> { + return this.instrumentation.startSpan({ name: "DrizzleNotificationStore > list" }, async () => { + try { + const n = notificationSchema.notification; + const where = cursor.isSome() + ? and(eq(n.userId, userId), lt(n.createdAt, new Date(cursor.unwrap()))) + : eq(n.userId, userId); + const query = db.select().from(n).where(where).orderBy(desc(n.createdAt)).limit(limit); + const rows = await this.instrumentation.startSpan( + { name: query.toSQL().sql, op: "db.query", attributes: dbAttrs }, + () => query.execute(), + ); + return Result.ok(rows.map((row) => this.toRecord(row))); + } catch (err) { + this.instrumentation.capture(err); + return Result.fail(readFailure(err)); + } + }); + } + + async unreadCount(userId: string): Promise> { + return this.instrumentation.startSpan( + { name: "DrizzleNotificationStore > unreadCount" }, + async () => { + try { + const n = notificationSchema.notification; + const query = db + .select({ count: count() }) + .from(n) + .where(and(eq(n.userId, userId), isNull(n.readAt))); + const rows = await this.instrumentation.startSpan( + { name: query.toSQL().sql, op: "db.query", attributes: dbAttrs }, + () => query.execute(), + ); + return Result.ok(rows[0]?.count ?? 0); + } catch (err) { + this.instrumentation.capture(err); + return Result.fail(readFailure(err)); + } + }, + ); + } + + async markRead( + userId: string, + ids: string[], + now: Date, + ): Promise> { + return this.instrumentation.startSpan( + { name: "DrizzleNotificationStore > markRead" }, + async () => { + try { + const n = notificationSchema.notification; + const query = db + .update(n) + .set({ readAt: now }) + .where(and(eq(n.userId, userId), inArray(n.id, ids))); + await this.instrumentation.startSpan( + { name: query.toSQL().sql, op: "db.query", attributes: dbAttrs }, + () => query.execute(), + ); + return Result.ok(); + } catch (err) { + this.instrumentation.capture(err); + return Result.fail(writeFailure(err)); + } + }, + ); + } + + async markAllRead(userId: string, now: Date): Promise> { + return this.instrumentation.startSpan( + { name: "DrizzleNotificationStore > markAllRead" }, + async () => { + try { + const n = notificationSchema.notification; + const query = db + .update(n) + .set({ readAt: now }) + .where(and(eq(n.userId, userId), isNull(n.readAt))); + await this.instrumentation.startSpan( + { name: query.toSQL().sql, op: "db.query", attributes: dbAttrs }, + () => query.execute(), + ); + return Result.ok(); + } catch (err) { + this.instrumentation.capture(err); + return Result.fail(writeFailure(err)); + } + }, + ); + } + + async listPreferences( + scope: PreferenceScope, + scopeId: string, + ): Promise> { + return this.instrumentation.startSpan( + { name: "DrizzleNotificationStore > listPreferences" }, + async () => { + try { + const p = notificationSchema.notificationPreference; + const query = db + .select() + .from(p) + .where(and(eq(p.scope, scope), eq(p.scopeId, scopeId))); + const rows = await this.instrumentation.startSpan( + { name: query.toSQL().sql, op: "db.query", attributes: dbAttrs }, + () => query.execute(), + ); + return Result.ok(rows.map((row) => this.toPreference(row))); + } catch (err) { + this.instrumentation.capture(err); + return Result.fail(readFailure(err)); + } + }, + ); + } + + async upsertPreference(input: PreferenceInput): Promise> { + return this.instrumentation.startSpan( + { name: "DrizzleNotificationStore > upsertPreference" }, + async () => { + try { + const p = notificationSchema.notificationPreference; + const query = db + .insert(p) + .values({ + id: crypto.randomUUID(), + scope: input.scope, + scopeId: input.scopeId, + category: input.category, + channel: input.channel, + enabled: input.enabled, + frequency: input.frequency, + locked: input.locked, + updatedAt: new Date(), + }) + .onConflictDoUpdate({ + target: [p.scope, p.scopeId, p.category, p.channel], + set: { + enabled: input.enabled, + frequency: input.frequency, + locked: input.locked, + updatedAt: new Date(), + }, + }); + await this.instrumentation.startSpan( + { name: query.toSQL().sql, op: "db.query", attributes: dbAttrs }, + () => query.execute(), + ); + return Result.ok(); + } catch (err) { + this.instrumentation.capture(err); + return Result.fail(writeFailure(err)); + } + }, + ); + } + + private toRecord(row: typeof notificationSchema.notification.$inferSelect): NotificationRecord { + return { + id: row.id, + userId: row.userId, + organizationId: Option.fromNullable(row.organizationId), + category: row.category, + eventType: row.eventType, + groupKey: Option.fromNullable(row.groupKey), + payload: row.payload, + readAt: Option.fromNullable(row.readAt), + createdAt: row.createdAt, + }; + } + + private toPreference( + row: typeof notificationSchema.notificationPreference.$inferSelect, + ): PreferenceRecord { + return { + scope: row.scope, + scopeId: row.scopeId, + category: row.category, + channel: row.channel, + enabled: row.enabled, + frequency: row.frequency, + locked: row.locked, + }; + } +} diff --git a/apps/api/src/modules/notifications/module.ts b/apps/api/src/modules/notifications/module.ts new file mode 100644 index 00000000..84131187 --- /dev/null +++ b/apps/api/src/modules/notifications/module.ts @@ -0,0 +1,23 @@ +import { defineModule } from "inwire"; +import type { INotificationStore } from "./application/ports/notification.port"; +import { NotificationPreferenceService } from "./application/services/notification-preference.service"; +import { DrizzleNotificationStore } from "./infrastructure/repositories/drizzle-notification.store"; + +declare module "inwire" { + interface AppDeps { + INotificationStore: INotificationStore; + NotificationPreferenceService: NotificationPreferenceService; + } +} + +export const notificationsModule = defineModule()((b) => + b + .add( + "INotificationStore", + (c): INotificationStore => new DrizzleNotificationStore(c.IInstrumentation), + ) + .add( + "NotificationPreferenceService", + (c) => new NotificationPreferenceService(c.INotificationStore), + ), +); diff --git a/apps/api/src/modules/notifications/notifications.schema.ts b/apps/api/src/modules/notifications/notifications.schema.ts new file mode 100644 index 00000000..8b5c734d --- /dev/null +++ b/apps/api/src/modules/notifications/notifications.schema.ts @@ -0,0 +1,26 @@ +import { + NOTIFICATION_CATEGORIES, + NOTIFICATION_CHANNELS, + NOTIFICATION_FREQUENCIES, +} from "@packages/events"; +import { z } from "zod"; + +export const listQuerySchema = z.object({ + cursor: z.string().optional(), + limit: z.coerce.number().int().min(1).max(50).default(20), +}); + +export const markReadSchema = z.object({ + ids: z.array(z.string().min(1)).min(1).max(100), +}); + +export const preferenceSchema = z.object({ + category: z.enum(NOTIFICATION_CATEGORIES), + channel: z.enum(NOTIFICATION_CHANNELS), + enabled: z.boolean(), + frequency: z.enum(NOTIFICATION_FREQUENCIES).default("immediate"), +}); + +export const orgPreferenceSchema = preferenceSchema.extend({ + locked: z.boolean().default(false), +}); diff --git a/apps/api/src/modules/notifications/routes.ts b/apps/api/src/modules/notifications/routes.ts new file mode 100644 index 00000000..73c0ba3a --- /dev/null +++ b/apps/api/src/modules/notifications/routes.ts @@ -0,0 +1,176 @@ +import { AppErrorException, Option } from "@packages/ddd-kit"; +import { EventTypes } from "@packages/events"; +import { Hono } from "hono"; +import { HTTPException } from "hono/http-exception"; +import { streamSSE } from "hono/streaming"; +import { di } from "../../container"; +import { emitEvent } from "../../shared/event-emitter"; +import { logger } from "../../shared/logger"; +import { type AuthVariables, requireAuth } from "../../shared/middleware/auth.middleware"; +import { denyImpersonated } from "../../shared/middleware/deny-impersonated.middleware"; +import { requireOrg, requireOrgPermission } from "../../shared/middleware/org.middleware"; +import { MAX_STREAMS_PER_USER } from "../../shared/services/notification-stream-hub"; +import { zV } from "../../shared/validator"; +import { + listQuerySchema, + markReadSchema, + orgPreferenceSchema, + preferenceSchema, +} from "./notifications.schema"; + +export const notificationsRoutes = new Hono<{ Variables: AuthVariables }>() + .get("/", requireAuth, zV("query", listQuerySchema), async (c) => { + const { cursor, limit } = c.req.valid("query"); + const userId = c.get("user").id; + const result = await di.INotificationStore.list( + userId, + Option.fromNullable(cursor ?? null), + limit, + ); + if (result.isFailure) throw new AppErrorException(result.getError()); + const notifications = result.getValue(); + const lastItem = notifications.at(-1); + const nextCursor = + notifications.length === limit && lastItem ? lastItem.createdAt.toISOString() : null; + return c.json({ + items: notifications.map((n) => ({ + ...n, + organizationId: n.organizationId.toNull(), + groupKey: n.groupKey.toNull(), + readAt: n.readAt.toNull(), + })), + nextCursor, + }); + }) + .get("/unread-count", requireAuth, async (c) => { + const userId = c.get("user").id; + const result = await di.INotificationStore.unreadCount(userId); + if (result.isFailure) throw new AppErrorException(result.getError()); + return c.json({ count: result.getValue() }); + }) + .post("/read", requireAuth, denyImpersonated, zV("json", markReadSchema), async (c) => { + const { ids } = c.req.valid("json"); + const userId = c.get("user").id; + const result = await di.INotificationStore.markRead(userId, ids, new Date()); + if (result.isFailure) throw new AppErrorException(result.getError()); + return c.json({ ok: true as const }); + }) + .post("/read-all", requireAuth, denyImpersonated, async (c) => { + const userId = c.get("user").id; + const result = await di.INotificationStore.markAllRead(userId, new Date()); + if (result.isFailure) throw new AppErrorException(result.getError()); + return c.json({ ok: true as const }); + }) + .get("/preferences", requireAuth, async (c) => { + const userId = c.get("user").id; + const result = await di.INotificationStore.listPreferences("user", userId); + if (result.isFailure) throw new AppErrorException(result.getError()); + return c.json({ items: result.getValue() }); + }) + .put("/preferences", requireAuth, denyImpersonated, zV("json", preferenceSchema), async (c) => { + const body = c.req.valid("json"); + const userId = c.get("user").id; + const result = await di.INotificationStore.upsertPreference({ + scope: "user", + scopeId: userId, + category: body.category, + channel: body.channel, + enabled: body.enabled, + frequency: body.frequency, + locked: false, + }); + if (result.isFailure) throw new AppErrorException(result.getError()); + await emitEvent( + di.IOutboxRepository, + EventTypes.NOTIFICATION_PREFERENCE_UPDATED, + "notification_preference", + userId, + { + userId, + category: body.category, + channel: body.channel, + enabled: body.enabled, + frequency: body.frequency, + }, + ); + return c.json({ ok: true as const }); + }) + .get( + "/org-preferences", + requireAuth, + requireOrg, + requireOrgPermission({ organization: ["update"] }), + async (c) => { + const orgId = c.get("orgId"); + const result = await di.INotificationStore.listPreferences("org", orgId); + if (result.isFailure) throw new AppErrorException(result.getError()); + return c.json({ items: result.getValue() }); + }, + ) + .put( + "/org-preferences", + requireAuth, + requireOrg, + requireOrgPermission({ organization: ["update"] }), + denyImpersonated, + zV("json", orgPreferenceSchema), + async (c) => { + const body = c.req.valid("json"); + const userId = c.get("user").id; + const orgId = c.get("orgId"); + const result = await di.INotificationStore.upsertPreference({ + scope: "org", + scopeId: orgId, + category: body.category, + channel: body.channel, + enabled: body.enabled, + frequency: body.frequency, + locked: body.locked, + }); + if (result.isFailure) throw new AppErrorException(result.getError()); + await emitEvent( + di.IOutboxRepository, + EventTypes.NOTIFICATION_ORG_PREFERENCE_UPDATED, + "notification_preference", + orgId, + { + organizationId: orgId, + actorUserId: userId, + category: body.category, + channel: body.channel, + enabled: body.enabled, + frequency: body.frequency, + locked: body.locked, + }, + { organizationId: orgId }, + ); + return c.json({ ok: true as const }); + }, + ) + .get("/stream", requireAuth, (c) => { + const userId = c.get("user").id; + const hub = di.NotificationStreamHub; + + if (hub.subscriberCount(userId) >= MAX_STREAMS_PER_USER) { + throw new HTTPException(429, { message: "NOTIFICATION_STREAM_LIMIT" }); + } + + return streamSSE(c, async (stream) => { + const unsubscribe = hub.subscribe(userId, () => { + stream.writeSSE({ event: "notification", data: "1" }).catch((err) => { + logger.warn({ err }, "notification sse write failed"); + }); + }); + + stream.onAbort(unsubscribe); + + try { + while (!stream.closed) { + await stream.writeSSE({ event: "ping", data: "" }); + await stream.sleep(25_000); + } + } finally { + unsubscribe(); + } + }); + }); diff --git a/apps/api/src/modules/webhooks/__TESTS__/drizzle-webhook-delivery.repository.test.ts b/apps/api/src/modules/webhooks/__TESTS__/drizzle-webhook-delivery.repository.test.ts index 4f8c0b07..75da08da 100644 --- a/apps/api/src/modules/webhooks/__TESTS__/drizzle-webhook-delivery.repository.test.ts +++ b/apps/api/src/modules/webhooks/__TESTS__/drizzle-webhook-delivery.repository.test.ts @@ -62,6 +62,7 @@ mock.module("@packages/drizzle", () => ({ arrayContains: () => ({}), sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { raw: () => ({}), + identifier: () => ({}), }), outboxSchema: { outboxEvent: {} }, auditLogSchema: { auditLog: {} }, @@ -117,6 +118,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, })); // All imports AFTER mock.module to ensure mocks are in place before module resolution. diff --git a/apps/api/src/modules/webhooks/__TESTS__/drizzle-webhook-endpoint.repository.test.ts b/apps/api/src/modules/webhooks/__TESTS__/drizzle-webhook-endpoint.repository.test.ts index 5c780068..0dc92b30 100644 --- a/apps/api/src/modules/webhooks/__TESTS__/drizzle-webhook-endpoint.repository.test.ts +++ b/apps/api/src/modules/webhooks/__TESTS__/drizzle-webhook-endpoint.repository.test.ts @@ -78,6 +78,7 @@ mock.module("@packages/drizzle", () => ({ arrayContains: () => ({}), sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { raw: () => ({}), + identifier: () => ({}), }), outboxSchema: { outboxEvent: {} }, auditLogSchema: { auditLog: {} }, @@ -127,6 +128,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, })); const { DrizzleWebhookEndpointRepository } = await import( diff --git a/apps/api/src/modules/webhooks/__TESTS__/webhook-delivery-worker.service.test.ts b/apps/api/src/modules/webhooks/__TESTS__/webhook-delivery-worker.service.test.ts index 6542d0b8..48db41db 100644 --- a/apps/api/src/modules/webhooks/__TESTS__/webhook-delivery-worker.service.test.ts +++ b/apps/api/src/modules/webhooks/__TESTS__/webhook-delivery-worker.service.test.ts @@ -85,6 +85,7 @@ mock.module("@packages/drizzle", () => ({ arrayContains: () => ({}), sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { raw: () => ({}), + identifier: () => ({}), }), schema: {}, authSchema: {}, @@ -98,6 +99,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, webhooksSchema: { webhookEndpoint: { id: "id", diff --git a/apps/api/src/shared/__TESTS__/csp-report.route.test.ts b/apps/api/src/shared/__TESTS__/csp-report.route.test.ts index ce1471d7..07368006 100644 --- a/apps/api/src/shared/__TESTS__/csp-report.route.test.ts +++ b/apps/api/src/shared/__TESTS__/csp-report.route.test.ts @@ -29,6 +29,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, inArray: () => {}, eq: () => {}, lt: () => {}, @@ -36,7 +44,7 @@ mock.module("@packages/drizzle", () => ({ asc: () => {}, desc: () => {}, and: () => {}, - sql: () => {}, + sql: Object.assign(() => {}, { raw: () => ({}), identifier: () => ({}) }), })); const { cspReportCors, makeCspReportApp } = await import("../internal-routes/csp-report.route"); diff --git a/apps/api/src/shared/__TESTS__/drizzle-audit.service.test.ts b/apps/api/src/shared/__TESTS__/drizzle-audit.service.test.ts index ece1ff2a..41bd0bad 100644 --- a/apps/api/src/shared/__TESTS__/drizzle-audit.service.test.ts +++ b/apps/api/src/shared/__TESTS__/drizzle-audit.service.test.ts @@ -51,6 +51,7 @@ mock.module("@packages/drizzle", () => ({ arrayContains: () => ({}), sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { raw: () => ({}), + identifier: () => ({}), }), outboxSchema: { outboxEvent: {} }, auditLogSchema: { @@ -79,6 +80,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, })); // ── Imports after mocks ──────────────────────────────────────────────────── diff --git a/apps/api/src/shared/__TESTS__/drizzle-email-queue.service.test.ts b/apps/api/src/shared/__TESTS__/drizzle-email-queue.service.test.ts index ddd74645..3c409e8a 100644 --- a/apps/api/src/shared/__TESTS__/drizzle-email-queue.service.test.ts +++ b/apps/api/src/shared/__TESTS__/drizzle-email-queue.service.test.ts @@ -36,6 +36,7 @@ mock.module("@packages/drizzle", () => ({ }, { raw: () => ({}), + identifier: () => ({}), }, ), outboxSchema: { @@ -74,6 +75,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, })); const { DrizzleEmailQueue } = await import("../services/drizzle-email-queue.service"); diff --git a/apps/api/src/shared/__TESTS__/drizzle-outbox.service.test.ts b/apps/api/src/shared/__TESTS__/drizzle-outbox.service.test.ts index a4a3a17f..f1f49386 100644 --- a/apps/api/src/shared/__TESTS__/drizzle-outbox.service.test.ts +++ b/apps/api/src/shared/__TESTS__/drizzle-outbox.service.test.ts @@ -56,6 +56,7 @@ mock.module("@packages/drizzle", () => ({ arrayContains: () => ({}), sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { raw: () => ({}), + identifier: () => ({}), }), outboxSchema: { outboxEvent: { @@ -93,6 +94,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, apiTokenSchema: { apiToken: { id: {}, diff --git a/apps/api/src/shared/__TESTS__/email-delivery-worker.service.test.ts b/apps/api/src/shared/__TESTS__/email-delivery-worker.service.test.ts index 37b367f6..28bc6bc2 100644 --- a/apps/api/src/shared/__TESTS__/email-delivery-worker.service.test.ts +++ b/apps/api/src/shared/__TESTS__/email-delivery-worker.service.test.ts @@ -23,7 +23,10 @@ mock.module("@packages/drizzle", () => ({ like: (...a: unknown[]) => a, count: (...a: unknown[]) => a, arrayContains: (...a: unknown[]) => a, - sql: Object.assign((s: TemplateStringsArray) => s.join(""), { raw: () => ({}) }), + sql: Object.assign((s: TemplateStringsArray) => s.join(""), { + raw: () => ({}), + identifier: () => ({}), + }), outboxSchema: { outboxEvent: { id: {}, @@ -60,6 +63,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, })); mock.module("@packages/emails", () => ({ diff --git a/apps/api/src/shared/__TESTS__/org.middleware.test.ts b/apps/api/src/shared/__TESTS__/org.middleware.test.ts index fb37a4fa..0b42fd76 100644 --- a/apps/api/src/shared/__TESTS__/org.middleware.test.ts +++ b/apps/api/src/shared/__TESTS__/org.middleware.test.ts @@ -34,6 +34,7 @@ mock.module("@packages/drizzle", () => ({ arrayContains: () => ({}), sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { raw: () => ({}), + identifier: () => ({}), }), schema: { member: { role: {}, organizationId: {}, userId: {} } }, outboxSchema: { outboxEvent: {} }, @@ -50,6 +51,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, })); const { requireOrg, requireOrgPermission } = await import("../middleware/org.middleware"); diff --git a/apps/api/src/shared/__TESTS__/rate-limiter-flexible.adapter.test.ts b/apps/api/src/shared/__TESTS__/rate-limiter-flexible.adapter.test.ts index f499e926..ed530745 100644 --- a/apps/api/src/shared/__TESTS__/rate-limiter-flexible.adapter.test.ts +++ b/apps/api/src/shared/__TESTS__/rate-limiter-flexible.adapter.test.ts @@ -54,6 +54,7 @@ mock.module("@packages/drizzle", () => ({ arrayContains: () => ({}), sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { raw: () => ({}), + identifier: () => ({}), }), outboxSchema: { outboxEvent: {} }, auditLogSchema: { auditLog: {} }, @@ -70,6 +71,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, })); // ── Mock rate-limiter-flexible ────────────────────────────────────────────── diff --git a/apps/api/src/shared/__TESTS__/webhook-fanout-subscriber.test.ts b/apps/api/src/shared/__TESTS__/webhook-fanout-subscriber.test.ts index d3e266b1..2001a930 100644 --- a/apps/api/src/shared/__TESTS__/webhook-fanout-subscriber.test.ts +++ b/apps/api/src/shared/__TESTS__/webhook-fanout-subscriber.test.ts @@ -77,6 +77,7 @@ mock.module("@packages/drizzle", () => ({ alias: () => ({}), sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { raw: () => ({}), + identifier: () => ({}), }), schema: {}, authSchema: {}, @@ -90,6 +91,14 @@ mock.module("@packages/drizzle", () => ({ }, policiesSchema: {}, consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, webhooksSchema: { webhookEndpoint: { id: "id", diff --git a/apps/api/src/shared/env.ts b/apps/api/src/shared/env.ts index e8cff297..68ca566d 100644 --- a/apps/api/src/shared/env.ts +++ b/apps/api/src/shared/env.ts @@ -66,6 +66,7 @@ const envSchema = z.object({ WEBHOOK_AUTO_DISABLE_MIN_FAILURES: z.coerce.number().int().positive().default(2), WEBHOOK_RESPONSE_CAPTURE_BYTES: z.coerce.number().int().positive().default(4096), CONSENT_RETENTION_DAYS: z.coerce.number().int().positive().default(365), + NOTIFICATION_RETENTION_DAYS: z.coerce.number().int().positive().default(30), GIT_SHA: z.string().optional(), BUILD_TIME: z.string().optional(), SHUTDOWN_GRACE_PERIOD_MS: z.coerce.number().int().min(0).default(15_000), diff --git a/apps/api/src/shared/internal-routes/__TESTS__/flush-notification-emails.test.ts b/apps/api/src/shared/internal-routes/__TESTS__/flush-notification-emails.test.ts new file mode 100644 index 00000000..4d95c54d --- /dev/null +++ b/apps/api/src/shared/internal-routes/__TESTS__/flush-notification-emails.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, test } from "bun:test"; +import { buildDigests, digestIdempotencyKey } from "../flush-notification-emails.route"; + +const row = (userId: string, category: string, id: string) => ({ + id, + userId, + category, + eventType: "billing.payment.failed", + email: `${userId}@example.com`, + payload: {}, +}); + +describe("buildDigests", () => { + test("groupe par utilisateur et par categorie", () => { + const digests = buildDigests([ + row("u1", "billing", "n1"), + row("u1", "billing", "n2"), + row("u1", "org", "n3"), + row("u2", "billing", "n4"), + ]); + + expect(digests).toHaveLength(3); + const billingU1 = digests.find((d) => d.userId === "u1" && d.category === "billing"); + expect(billingU1?.notificationIds).toEqual(["n1", "n2"]); + }); + + test("un lot vide ne produit aucun digest", () => { + expect(buildDigests([])).toEqual([]); + }); +}); + +describe("digestIdempotencyKey", () => { + test("produit une cle de longueur constante quel que soit le nombre d'ids", async () => { + const key2 = await digestIdempotencyKey(["n1", "n2"]); + const key500 = await digestIdempotencyKey(Array.from({ length: 500 }, (_, i) => `n${i}`)); + expect(key2).toHaveLength(64); + expect(key500).toHaveLength(64); + }); +}); diff --git a/apps/api/src/shared/internal-routes/__TESTS__/sweep-notifications.test.ts b/apps/api/src/shared/internal-routes/__TESTS__/sweep-notifications.test.ts new file mode 100644 index 00000000..1ad6a3c1 --- /dev/null +++ b/apps/api/src/shared/internal-routes/__TESTS__/sweep-notifications.test.ts @@ -0,0 +1,113 @@ +import { mock } from "bun:test"; + +// Expose the FULL export surface — bun's mock.module leaks across files. +// and/isNotNull/lt pass their column argument through so hasColumnName can inspect the filter. +mock.module("@packages/drizzle", () => ({ + db: {}, + eq: () => ({}), + and: (...args: unknown[]) => ({ queryChunks: args }), + or: (...args: unknown[]) => ({ queryChunks: args }), + isNull: (col: unknown) => ({ queryChunks: [col] }), + isNotNull: (col: unknown) => ({ queryChunks: [col] }), + lt: (col: unknown) => ({ queryChunks: [col] }), + lte: (col: unknown) => ({ queryChunks: [col] }), + gt: (col: unknown) => ({ queryChunks: [col] }), + gte: (col: unknown) => ({ queryChunks: [col] }), + not: (col: unknown) => ({ queryChunks: [col] }), + asc: () => ({}), + desc: () => ({}), + like: () => ({}), + inArray: () => ({}), + count: () => ({}), + arrayContains: () => ({}), + sql: Object.assign((_strings: TemplateStringsArray, ..._values: unknown[]) => ({}), { + raw: () => ({}), + identifier: () => ({}), + }), + outboxSchema: { + outboxEvent: { + id: {}, + eventType: {}, + dispatchedAt: {}, + nextAttemptAt: {}, + occurredAt: {}, + attempts: {}, + }, + }, + auditLogSchema: { + auditLog: { + actorId: {}, + actorType: {}, + organizationId: {}, + action: {}, + targetType: {}, + targetId: {}, + occurredAt: {}, + retention: {}, + id: {}, + }, + }, + webhooksSchema: { webhookDelivery: {}, webhookEndpoint: {} }, + multiTenantSchema: {}, + authSchema: {}, + schema: {}, + trackEventsOnSuccess: () => {}, + TransactionService: class {}, + rateLimitSchema: { rateLimitRecord: { key: {}, points: {}, expire: {} } }, + billingSchema: {}, + quotaUsageSchema: { + quotaUsage: { organizationId: {}, resource: {}, periodStart: {}, used: {}, updatedAt: {} }, + }, + policiesSchema: {}, + consentSchema: {}, + emailSchema: { emailMessage: { id: {}, status: {}, sentAt: {}, createdAt: {} } }, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, + apiTokenSchema: { + apiToken: { + id: {}, + userId: {}, + organizationId: {}, + name: {}, + scopes: {}, + tokenHmac: {}, + pepperVersion: {}, + tokenStart: {}, + lastUsedAt: {}, + expiresAt: {}, + revokedAt: {}, + revokedReason: {}, + createdAt: {}, + updatedAt: {}, + }, + }, +})); + +import { describe, expect, test } from "bun:test"; + +const { buildPurgeFilter } = await import("../sweep-notifications.route"); + +function hasColumnName(obj: unknown, target: string, seen = new WeakSet()): boolean { + if (!obj || typeof obj !== "object") return false; + if (seen.has(obj as object)) return false; + seen.add(obj as object); + if ((obj as Record).name === target) return true; + return Object.values(obj as object).some((v) => + Array.isArray(v) + ? v.some((i) => hasColumnName(i, target, seen)) + : hasColumnName(v, target, seen), + ); +} + +describe("sweep-notifications", () => { + test("le filtre de purge exige une notification lue", () => { + const filter = buildPurgeFilter(new Date("2026-01-01T00:00:00Z")); + expect(hasColumnName(filter, "read_at")).toBe(true); + }); +}); diff --git a/apps/api/src/shared/internal-routes/flush-notification-emails.route.ts b/apps/api/src/shared/internal-routes/flush-notification-emails.route.ts new file mode 100644 index 00000000..c20ec9ee --- /dev/null +++ b/apps/api/src/shared/internal-routes/flush-notification-emails.route.ts @@ -0,0 +1,156 @@ +// `/internal/flush-notification-emails` — gated by signed HMAC + optional private-network (env-driven). Never exposed to public traffic. + +import { + and, + authSchema, + count, + db, + eq, + inArray, + isNull, + lte, + notificationSchema, + sql, +} from "@packages/drizzle"; +import { Hono } from "hono"; +import type { PinoLogger } from "hono-pino"; +import { z } from "zod"; +import { di } from "../../container"; +import { zV } from "../validator"; +import { internalLayers } from "./internal-layers"; + +type HonoEnv = { Variables: { logger: PinoLogger } }; + +const bodySchema = z + .object({ + batchSize: z.number().int().min(1).max(5000).optional(), + dryRun: z.boolean().optional(), + }) + .default({}); + +export type PendingRow = { + id: string; + userId: string; + category: string; + eventType: string; + email: string; + payload: unknown; +}; + +export type DigestGroup = { + userId: string; + email: string; + category: string; + notificationIds: string[]; + items: { eventType: string; payload: unknown }[]; +}; + +export function buildDigests(rows: PendingRow[]): DigestGroup[] { + const map = new Map(); + for (const row of rows) { + const key = `${row.userId}:${row.category}`; + const group = map.get(key); + if (group) { + group.notificationIds.push(row.id); + group.items.push({ eventType: row.eventType, payload: row.payload }); + } else { + map.set(key, { + userId: row.userId, + email: row.email, + category: row.category, + notificationIds: [row.id], + items: [{ eventType: row.eventType, payload: row.payload }], + }); + } + } + return [...map.values()]; +} + +export async function digestIdempotencyKey(ids: string[]): Promise { + const raw = ids.slice().sort().join("|"); + const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(raw)); + return Array.from(new Uint8Array(digest)) + .map((b) => b.toString(16).padStart(2, "0")) + .join(""); +} + +const DEFAULT_BATCH_SIZE = 500; + +export const flushNotificationEmailsRoutes = new Hono() + .use("*", ...internalLayers) + .post("/flush-notification-emails", zV("json", bodySchema), async (c) => { + const { batchSize = DEFAULT_BATCH_SIZE, dryRun = false } = c.req.valid("json"); + const now = new Date(); + const logger = c.var.logger; + const n = notificationSchema.notification; + const u = authSchema.user; + + if (dryRun) { + const rows = await db + .select({ eligible: count() }) + .from(n) + .where(and(lte(n.emailPendingAt, now), isNull(n.emailSentAt))); + const eligible = rows[0]?.eligible ?? 0; + logger.info({ eligible }, "flush-notification-emails dry-run"); + return c.json({ dryRun: true, eligible, flushed: 0, notifications: 0 }); + } + + const { flushed, notifications } = await db.transaction(async (tx) => { + await tx.execute(sql`SET LOCAL statement_timeout = '30s'`); + await tx.execute(sql`SET LOCAL lock_timeout = '500ms'`); + await tx.execute(sql`SET LOCAL idle_in_transaction_session_timeout = '10s'`); + + const rows = await tx + .select({ + id: n.id, + userId: n.userId, + category: n.category, + eventType: n.eventType, + email: u.email, + payload: n.payload, + }) + .from(n) + .innerJoin(u, eq(n.userId, u.id)) + .where(and(lte(n.emailPendingAt, now), isNull(n.emailSentAt))) + .limit(batchSize) + .for("update", { skipLocked: true }); + + if (rows.length === 0) return { flushed: 0, notifications: 0 }; + + const digests = buildDigests(rows as PendingRow[]); + + const idempotencyKey = await digestIdempotencyKey(rows.map((r) => r.id)); + + const enqueued = await di.IEmailService.sendTemplateBatch( + "notification_digest", + digests.map((d) => ({ + to: d.email, + variables: { + category: d.category, + itemCount: String(d.items.length), + itemsSummary: d.items.map((i) => i.eventType).join(", "), + }, + })), + { + tx: tx as unknown as import("../transaction").ITransaction, + idempotencyKey, + }, + ); + + if (enqueued.isFailure) { + logger.error({ err: enqueued.getError() }, "flush-notification-emails enqueue failed"); + throw new Error(enqueued.getError().message); + } + + const notificationIds = rows.map((r) => r.id); + await tx.update(n).set({ emailSentAt: now }).where(inArray(n.id, notificationIds)); + + logger.info( + { flushed: digests.length, notifications: notificationIds.length }, + "flush-notification-emails done", + ); + return { flushed: digests.length, notifications: notificationIds.length }; + }); + + return c.json({ dryRun: false, flushed, notifications }); + }); diff --git a/apps/api/src/shared/internal-routes/sweep-notifications.route.ts b/apps/api/src/shared/internal-routes/sweep-notifications.route.ts new file mode 100644 index 00000000..014ed006 --- /dev/null +++ b/apps/api/src/shared/internal-routes/sweep-notifications.route.ts @@ -0,0 +1,54 @@ +import { and, count, db, inArray, isNotNull, lt, notificationSchema, sql } from "@packages/drizzle"; +import { Hono } from "hono"; +import type { PinoLogger } from "hono-pino"; +import { env } from "../env"; +import { zV } from "../validator"; +import { internalLayers } from "./internal-layers"; +import { runRetentionSweep, type SweepBody, sweepBodySchema } from "./sweep-runner"; + +type HonoEnv = { Variables: { logger: PinoLogger } }; + +export function buildPurgeFilter(cutoff: Date) { + const n = notificationSchema.notification; + return and(isNotNull(n.readAt), lt(n.createdAt, cutoff)); +} + +async function countEligible(cutoff: Date): Promise { + const n = notificationSchema.notification; + const rows = await db.select({ count: count() }).from(n).where(buildPurgeFilter(cutoff)); + return rows[0]?.count ?? 0; +} + +async function purgeBatch(cutoff: Date, batchSize: number): Promise { + const n = notificationSchema.notification; + return db.transaction(async (tx) => { + await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); + await tx.execute(sql`SET LOCAL lock_timeout = '500ms'`); + await tx.execute(sql`SET LOCAL idle_in_transaction_session_timeout = '10s'`); + + const subq = tx + .select({ id: n.id }) + .from(n) + .where(buildPurgeFilter(cutoff)) + .orderBy(n.createdAt) + .limit(batchSize) + .for("update", { skipLocked: true }); + + const deleted = await tx.delete(n).where(inArray(n.id, subq)).returning({ id: n.id }); + return deleted.length; + }); +} + +export const sweepNotificationsRoutes = new Hono() + .use("*", ...internalLayers) + .post("/sweep-notifications", zV("json", sweepBodySchema), async (c) => { + const response = await runRetentionSweep({ + body: c.req.valid("json") as SweepBody, + retentionDays: env.NOTIFICATION_RETENTION_DAYS, + purgeBatch, + countEligible, + logger: c.var.logger, + label: "sweep-notifications", + }); + return c.json(response); + }); diff --git a/apps/api/src/shared/services/__TESTS__/notification-fanout-subscriber.test.ts b/apps/api/src/shared/services/__TESTS__/notification-fanout-subscriber.test.ts new file mode 100644 index 00000000..1a31e675 --- /dev/null +++ b/apps/api/src/shared/services/__TESTS__/notification-fanout-subscriber.test.ts @@ -0,0 +1,85 @@ +import { describe, expect, mock, test } from "bun:test"; +import { Option } from "@packages/ddd-kit"; +import type { OutboxRecord } from "../../ports/outbox.port"; +import { NoOpInstrumentation } from "../noop-instrumentation"; +import { NotificationFanoutSubscriber } from "../notification-fanout-subscriber"; + +const event = (eventType: string, payload: unknown, orgId?: string): OutboxRecord => ({ + id: "01J000000000000000000000", + eventType, + aggregateId: "agg-1", + aggregateType: "user", + organizationId: orgId ? Option.some(orgId) : Option.none(), + payload, + metadata: {} as OutboxRecord["metadata"], + occurredAt: new Date("2026-08-07T10:00:00Z"), + attempts: 0, +}); + +function fakeTx() { + const calls: string[] = []; + let lastInsertValues: Record | null = null; + const tx = { + insert: mock(() => { + calls.push("insert"); + return { + values: mock((vals: Record) => { + lastInsertValues = vals; + return { + onConflictDoNothing: mock(() => ({ + execute: mock(async () => undefined), + toSQL: () => ({ sql: "insert into notification" }), + })), + }; + }), + select: mock(() => ({})), + }; + }), + }; + return { tx, calls, getLastValues: () => lastInsertValues }; +} + +describe("NotificationFanoutSubscriber", () => { + test("ignore un event absent du map", async () => { + const { tx, calls } = fakeTx(); + const subscriber = new NotificationFanoutSubscriber(new NoOpInstrumentation()); + + await subscriber.handle(event("api_token.used", { userId: "u1" }), tx as never); + + expect(calls).toEqual([]); + }); + + test("ignore un event dont l'audience ne resout personne", async () => { + const { tx, calls } = fakeTx(); + const subscriber = new NotificationFanoutSubscriber(new NoOpInstrumentation()); + + await subscriber.handle(event("billing.payment.failed", {}), tx as never); + + expect(calls).toEqual([]); + }); + + test("insere pour un event self notifiable", async () => { + const { tx, calls } = fakeTx(); + const subscriber = new NotificationFanoutSubscriber(new NoOpInstrumentation()); + + await subscriber.handle(event("user.password_changed", { userId: "u1" }), tx as never); + + expect(calls).toEqual(["insert"]); + }); + + test("emailPendingAt est toujours non-null, meme pour un event force", async () => { + const { tx, getLastValues } = fakeTx(); + const subscriber = new NotificationFanoutSubscriber(new NoOpInstrumentation()); + + const occurredAt = new Date("2026-08-07T10:00:00Z"); + await subscriber.handle( + { ...event("user.password_changed", { userId: "u1" }), occurredAt }, + tx as never, + ); + + const values = getLastValues(); + expect(values).not.toBeNull(); + expect((values as Record).emailPendingAt).not.toBeNull(); + expect((values as Record).emailPendingAt).toEqual(occurredAt); + }); +}); diff --git a/apps/api/src/shared/services/__TESTS__/notification-stream-hub.test.ts b/apps/api/src/shared/services/__TESTS__/notification-stream-hub.test.ts new file mode 100644 index 00000000..8760c2a9 --- /dev/null +++ b/apps/api/src/shared/services/__TESTS__/notification-stream-hub.test.ts @@ -0,0 +1,72 @@ +import { describe, expect, test } from "bun:test"; +import { logger } from "../../logger"; +import { MAX_STREAMS_PER_USER, NotificationStreamHub } from "../notification-stream-hub"; + +describe("NotificationStreamHub", () => { + test("distribue un signal au seul destinataire concerne", () => { + const hub = new NotificationStreamHub(logger, "postgres://unused"); + const recu: string[] = []; + + hub.subscribe("u1", () => recu.push("u1")); + hub.subscribe("u2", () => recu.push("u2")); + + hub.dispatchSignal("u1"); + + expect(recu).toEqual(["u1"]); + }); + + test("la desinscription libere le handle", () => { + const hub = new NotificationStreamHub(logger, "postgres://unused"); + const unsubscribe = hub.subscribe("u1", () => {}); + + expect(hub.subscriberCount("u1")).toBe(1); + unsubscribe(); + expect(hub.subscriberCount("u1")).toBe(0); + }); + + test("plusieurs onglets du meme user recoivent chacun le signal", () => { + const hub = new NotificationStreamHub(logger, "postgres://unused"); + let appels = 0; + + hub.subscribe("u1", () => appels++); + hub.subscribe("u1", () => appels++); + hub.dispatchSignal("u1"); + + expect(appels).toBe(2); + }); + + test("un signal pour un user sans abonne ne touche pas les autres", () => { + const hub = new NotificationStreamHub(logger, "postgres://unused"); + let appels = 0; + hub.subscribe("u1", () => appels++); + + hub.dispatchSignal("inconnu"); + + expect(appels).toBe(0); + expect(hub.subscriberCount("u1")).toBe(1); + }); + + test("un handle qui jette n'empeche pas les autres de recevoir", () => { + const hub = new NotificationStreamHub(logger, "postgres://unused"); + const recus: string[] = []; + hub.subscribe("u1", () => { + throw new Error("onglet mort"); + }); + hub.subscribe("u1", () => recus.push("second")); + + hub.dispatchSignal("u1"); + + expect(recus).toEqual(["second"]); + }); + + test("le compteur declenche le plafond apres MAX_STREAMS_PER_USER abonnements", () => { + const hub = new NotificationStreamHub(logger, "postgres://unused"); + + for (let i = 0; i < MAX_STREAMS_PER_USER; i++) { + expect(hub.subscriberCount("u1") >= MAX_STREAMS_PER_USER).toBe(false); + hub.subscribe("u1", () => {}); + } + + expect(hub.subscriberCount("u1") >= MAX_STREAMS_PER_USER).toBe(true); + }); +}); diff --git a/apps/api/src/shared/services/__TESTS__/notification-trigger.test.ts b/apps/api/src/shared/services/__TESTS__/notification-trigger.test.ts new file mode 100644 index 00000000..b33fa5ba --- /dev/null +++ b/apps/api/src/shared/services/__TESTS__/notification-trigger.test.ts @@ -0,0 +1,104 @@ +import { describe, expect, mock, test } from "bun:test"; + +const fakeSql = Object.assign( + (strings: TemplateStringsArray) => ({ + toSQL: () => ({ sql: strings.join(""), params: [] as unknown[] }), + }), + { + raw: (_s: string) => ({}), + identifier: (_s: string) => ({}), + }, +); + +mock.module("@packages/drizzle", () => ({ + db: {}, + sql: fakeSql, + eq: () => ({}), + and: () => ({}), + or: () => ({}), + isNull: () => ({}), + isNotNull: () => ({}), + lt: () => ({}), + lte: () => ({}), + gt: () => ({}), + gte: () => ({}), + not: () => ({}), + asc: () => ({}), + desc: () => ({}), + like: () => ({}), + inArray: () => ({}), + count: () => ({}), + arrayContains: () => ({}), + outboxSchema: { outboxEvent: {} }, + auditLogSchema: { auditLog: {} }, + webhooksSchema: { webhookDelivery: {} }, + multiTenantSchema: {}, + authSchema: {}, + schema: {}, + trackEventsOnSuccess: () => {}, + TransactionService: class {}, + rateLimitSchema: { rateLimitRecord: { key: {}, points: {}, expire: {} } }, + billingSchema: {}, + quotaUsageSchema: { + quotaUsage: { organizationId: {}, resource: {}, periodStart: {}, used: {}, updatedAt: {} }, + }, + policiesSchema: {}, + consentSchema: {}, + notificationSchema: { + notification: { + dedupKey: { name: "dedup_key" }, + readAt: { name: "read_at" }, + createdAt: { name: "created_at" }, + id: {}, + }, + }, + apiTokenSchema: {}, +})); + +import { ensureNotificationTrigger } from "../notification-trigger"; + +describe("ensureNotificationTrigger", () => { + function makeClient() { + const executed: Array<{ toSQL: () => { sql: string } }> = []; + const client = { + execute: mock(async (query: { toSQL: () => { sql: string } }) => { + executed.push(query); + }), + }; + return { client, executed }; + } + + test("appelle execute et emet le DDL attendu", async () => { + const { client, executed } = makeClient(); + + await ensureNotificationTrigger(client as never); + + expect(client.execute).toHaveBeenCalledTimes(2); + + const fnDdl = executed[0]?.toSQL().sql ?? ""; + const triggerDdl = executed[1]?.toSQL().sql ?? ""; + + expect(fnDdl).toContain("CREATE OR REPLACE"); + expect(fnDdl).toContain("notification_created"); + expect(fnDdl).toContain("NEW.user_id"); + + expect(triggerDdl).toContain("CREATE OR REPLACE"); + expect(triggerDdl).toContain("notification_notify_trigger"); + }); + + test("est idempotent - deux appels ne levent pas d'erreur", async () => { + const { client } = makeClient(); + + await ensureNotificationTrigger(client as never); + await ensureNotificationTrigger(client as never); + + expect(client.execute).toHaveBeenCalledTimes(4); + + const sqls = (client.execute as ReturnType).mock.calls.map( + (c) => (c[0] as { toSQL: () => { sql: string } }).toSQL().sql, + ); + for (const s of sqls) { + expect(s).toContain("CREATE OR REPLACE"); + } + }); +}); diff --git a/apps/api/src/shared/services/__TESTS__/resolve-audience.test.ts b/apps/api/src/shared/services/__TESTS__/resolve-audience.test.ts new file mode 100644 index 00000000..3b0df4d3 --- /dev/null +++ b/apps/api/src/shared/services/__TESTS__/resolve-audience.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, test } from "bun:test"; +import { Option } from "@packages/ddd-kit"; +import type { OutboxRecord } from "../../ports/outbox.port"; +import { resolveAudience } from "../resolve-audience"; + +const baseEvent = (payload: unknown, orgId?: string): OutboxRecord => ({ + id: "01J000000000000000000000", + eventType: "org.member.joined", + aggregateId: "agg-1", + aggregateType: "organization", + organizationId: orgId ? Option.some(orgId) : Option.none(), + payload, + metadata: {} as OutboxRecord["metadata"], + occurredAt: new Date("2026-08-07T10:00:00Z"), + attempts: 0, +}); + +describe("resolveAudience", () => { + test("self cible le userId du payload", () => { + const target = resolveAudience("self", baseEvent({ userId: "user-1" })); + expect(target).toEqual({ kind: "user", userId: "user-1" }); + }); + + test("actor cible actorUserId en priorite sur userId", () => { + const target = resolveAudience("actor", baseEvent({ userId: "sujet", actorUserId: "acteur" })); + expect(target).toEqual({ kind: "user", userId: "acteur" }); + }); + + test("actor respecte la priorite complete de extractActor", () => { + expect( + resolveAudience("actor", baseEvent({ userId: "sujet", ownerUserId: "proprio" })), + ).toEqual({ kind: "user", userId: "proprio" }); + expect( + resolveAudience( + "actor", + baseEvent({ userId: "sujet", ownerUserId: "proprio", inviterUserId: "invitant" }), + ), + ).toEqual({ kind: "user", userId: "invitant" }); + }); + + test("org:all cible toute l'org", () => { + const target = resolveAudience("org:all", baseEvent({}, "org-1")); + expect(target).toEqual({ kind: "org", organizationId: "org-1", roles: "all" }); + }); + + test("une capability se resout en liste de roles", () => { + const target = resolveAudience({ can: { billing: ["read"] } }, baseEvent({}, "org-1")); + expect(target).toEqual({ + kind: "org", + organizationId: "org-1", + roles: ["owner", "admin"], + }); + }); + + test("une audience org sans organizationId ne cible personne", () => { + expect(resolveAudience("org:all", baseEvent({}))).toBeNull(); + expect(resolveAudience({ can: { billing: ["read"] } }, baseEvent({}))).toBeNull(); + }); + + test("self sans userId exploitable ne cible personne", () => { + expect(resolveAudience("self", baseEvent({ foo: "bar" }))).toBeNull(); + }); +}); diff --git a/apps/api/src/shared/services/email-delivery-worker.service.ts b/apps/api/src/shared/services/email-delivery-worker.service.ts index 42a10d89..1767a692 100644 --- a/apps/api/src/shared/services/email-delivery-worker.service.ts +++ b/apps/api/src/shared/services/email-delivery-worker.service.ts @@ -29,6 +29,7 @@ export const TEMPLATE_IDS: Record = { delete_completed: "", change_email: "", backup_code_used: "", + notification_digest: "", }; type BatchEntry = Record; diff --git a/apps/api/src/shared/services/notification-fanout-subscriber.ts b/apps/api/src/shared/services/notification-fanout-subscriber.ts new file mode 100644 index 00000000..cccd740e --- /dev/null +++ b/apps/api/src/shared/services/notification-fanout-subscriber.ts @@ -0,0 +1,121 @@ +import { uuidv7 } from "@packages/ddd-kit"; +import { + and, + eq, + inArray, + multiTenantSchema, + notificationSchema, + sql, + type Transaction, +} from "@packages/drizzle"; +import { notificationConfigOf } from "@packages/events"; +import type { IInstrumentation } from "../ports/instrumentation.port"; +import type { OutboxRecord } from "../ports/outbox.port"; +import type { OutboxSubscriber } from "./outbox-subscriber"; +import { resolveAudience } from "./resolve-audience"; + +function dedupKeyFor(event: OutboxRecord, window: "hour" | "day" | undefined): string | null { + if (!window) return null; + const iso = event.occurredAt.toISOString(); + const bucket = window === "hour" ? iso.slice(0, 13) : iso.slice(0, 10); + return `${event.eventType}:${event.aggregateId}:${bucket}`; +} + +export class NotificationFanoutSubscriber implements OutboxSubscriber { + readonly name = "notification-fanout"; + + constructor(private readonly instrumentation: IInstrumentation) {} + + async handle(event: OutboxRecord, tx: Transaction): Promise { + return this.instrumentation.startSpan( + { name: "NotificationFanoutSubscriber > handle" }, + async () => { + try { + const config = notificationConfigOf(event.eventType); + if (!config) return; + + const target = resolveAudience(config.audience, event); + if (!target) return; + + const n = notificationSchema.notification; + const shared = { + organizationId: event.organizationId.isSome() ? event.organizationId.unwrap() : null, + category: config.category, + eventType: event.eventType, + groupKey: config.groupBy ? `${event.eventType}:${event.aggregateId}` : null, + dedupKey: dedupKeyFor(event, config.dedupWindow), + payload: event.payload, + emailPendingAt: event.occurredAt, + }; + + const conflictWhere = sql`${sql.identifier(n.dedupKey.name)} IS NOT NULL`; + + if (target.kind === "user") { + const insertQuery = tx + .insert(n) + .values({ id: uuidv7(), userId: target.userId, ...shared }) + .onConflictDoNothing({ target: [n.userId, n.dedupKey], where: conflictWhere }); + await this.instrumentation.startSpan( + { + name: insertQuery.toSQL().sql, + op: "db.query", + attributes: { "db.system.name": "postgresql" }, + }, + () => insertQuery.execute(), + ); + return; + } + + const m = multiTenantSchema.member; + const memberFilter = + target.roles === "all" + ? eq(m.organizationId, target.organizationId) + : and(eq(m.organizationId, target.organizationId), inArray(m.role, target.roles)); + + const orgInsertSql = sql` + INSERT INTO ${n} ( + ${sql.identifier(n.id.name)}, + ${sql.identifier(n.userId.name)}, + ${sql.identifier(n.organizationId.name)}, + ${sql.identifier(n.category.name)}, + ${sql.identifier(n.eventType.name)}, + ${sql.identifier(n.groupKey.name)}, + ${sql.identifier(n.dedupKey.name)}, + ${sql.identifier(n.payload.name)}, + ${sql.identifier(n.emailPendingAt.name)} + ) + SELECT + gen_random_uuid()::text, + ${m.userId}, + ${shared.organizationId}, + ${shared.category}, + ${shared.eventType}, + ${shared.groupKey}, + ${shared.dedupKey}, + ${JSON.stringify(shared.payload)}::jsonb, + ${shared.emailPendingAt} + FROM ${m} + WHERE ${memberFilter} + ON CONFLICT ( + ${sql.identifier(n.userId.name)}, + ${sql.identifier(n.dedupKey.name)} + ) WHERE ${conflictWhere} DO NOTHING + `; + + const rawQuery = tx.execute(orgInsertSql); + await this.instrumentation.startSpan( + { + name: rawQuery.getQuery().sql, + op: "db.query", + attributes: { "db.system.name": "postgresql" }, + }, + () => rawQuery.execute(), + ); + } catch (err) { + this.instrumentation.capture(err); + throw err; + } + }, + ); + } +} diff --git a/apps/api/src/shared/services/notification-stream-hub.ts b/apps/api/src/shared/services/notification-stream-hub.ts new file mode 100644 index 00000000..6cd3a59d --- /dev/null +++ b/apps/api/src/shared/services/notification-stream-hub.ts @@ -0,0 +1,117 @@ +import { db } from "@packages/drizzle"; +import { Client } from "pg"; +import type { Logger } from "../logger"; +import { ensureNotificationTrigger, NOTIFICATION_NOTIFY_CHANNEL } from "./notification-trigger"; + +const RECONNECT_BACKOFF_MS = 1_000; +const RECONNECT_MAX_BACKOFF_MS = 30_000; + +export const MAX_STREAMS_PER_USER = 5; + +export class NotificationStreamHub { + private listenClient: Client | null = null; + private readonly subscribers = new Map void>>(); + private stopping = false; + private started = false; + private reconnectBackoff = RECONNECT_BACKOFF_MS; + + constructor( + private readonly logger: Logger, + private readonly databaseUrl: string, + ) {} + + subscribe(userId: string, onSignal: () => void): () => void { + const existing = this.subscribers.get(userId) ?? new Set<() => void>(); + existing.add(onSignal); + this.subscribers.set(userId, existing); + return () => { + const handles = this.subscribers.get(userId); + if (!handles) return; + handles.delete(onSignal); + if (handles.size === 0) this.subscribers.delete(userId); + }; + } + + subscriberCount(userId: string): number { + return this.subscribers.get(userId)?.size ?? 0; + } + + dispatchSignal(userId: string): void { + const handles = this.subscribers.get(userId); + if (!handles) return; + for (const handle of handles) { + try { + handle(); + } catch (err) { + this.logger.warn({ err }, "notification stream handle failed"); + } + } + } + + async start(): Promise { + if (this.started) return; + this.started = true; + this.stopping = false; + await ensureNotificationTrigger(db); + await this.connectListener(); + this.logger.info("notification stream hub started"); + } + + async stop(): Promise { + this.started = false; + this.stopping = true; + if (this.listenClient) { + try { + await this.listenClient.end(); + } catch (err) { + this.logger.warn({ err }, "notification stream hub client end failed"); + } + this.listenClient = null; + } + this.subscribers.clear(); + this.logger.info("notification stream hub stopped"); + } + + private async connectListener(): Promise { + if (this.stopping) return; + const client = new Client({ + connectionString: this.databaseUrl, + keepAlive: true, + keepAliveInitialDelayMillis: 30_000, + }); + client.on("notification", (msg) => { + this.dispatchSignal(msg.payload ?? ""); + }); + client.on("error", (err: Error) => { + this.logger.warn({ err }, "notification stream hub listener error, will reconnect"); + this.scheduleReconnect(); + }); + client.on("end", () => { + if (this.stopping) return; + this.logger.warn("notification stream hub listener ended, will reconnect"); + this.scheduleReconnect(); + }); + + try { + await client.connect(); + await client.query(`LISTEN ${NOTIFICATION_NOTIFY_CHANNEL}`); + this.listenClient = client; + this.reconnectBackoff = RECONNECT_BACKOFF_MS; + this.logger.debug("notification stream hub listener connected"); + } catch (err) { + this.logger.warn({ err }, "notification stream hub listener initial connect failed"); + client.removeAllListeners(); + await client.end().catch(() => {}); + this.scheduleReconnect(); + } + } + + private scheduleReconnect(): void { + if (this.stopping) return; + const delay = this.reconnectBackoff; + this.reconnectBackoff = Math.min(this.reconnectBackoff * 2, RECONNECT_MAX_BACKOFF_MS); + setTimeout(() => { + void this.connectListener(); + }, delay); + } +} diff --git a/apps/api/src/shared/services/notification-trigger.ts b/apps/api/src/shared/services/notification-trigger.ts new file mode 100644 index 00000000..be8c75ad --- /dev/null +++ b/apps/api/src/shared/services/notification-trigger.ts @@ -0,0 +1,19 @@ +import { type DbClient, sql } from "@packages/drizzle"; + +export const NOTIFICATION_NOTIFY_CHANNEL = "notification_created"; + +export async function ensureNotificationTrigger(client: DbClient): Promise { + await client.execute(sql` + CREATE OR REPLACE FUNCTION notification_notify() RETURNS trigger AS $$ + BEGIN + PERFORM pg_notify('notification_created', NEW.user_id::text); + RETURN NEW; + END; + $$ LANGUAGE plpgsql; + `); + await client.execute(sql` + CREATE OR REPLACE TRIGGER notification_notify_trigger + AFTER INSERT ON notification + FOR EACH ROW EXECUTE FUNCTION notification_notify() + `); +} diff --git a/apps/api/src/shared/services/resolve-audience.ts b/apps/api/src/shared/services/resolve-audience.ts new file mode 100644 index 00000000..345943c6 --- /dev/null +++ b/apps/api/src/shared/services/resolve-audience.ts @@ -0,0 +1,41 @@ +import { type OrgRole, rolesWith } from "@packages/access-control"; +import type { Audience } from "@packages/events"; +import type { OutboxRecord } from "../ports/outbox.port"; + +export type AudienceTarget = + | { kind: "user"; userId: string } + | { kind: "org"; organizationId: string; roles: OrgRole[] | "all" }; + +function readUserId(payload: unknown, keys: readonly string[]): string | null { + if (typeof payload !== "object" || payload === null) return null; + const record = payload as Record; + for (const key of keys) { + const value = record[key]; + if (typeof value === "string" && value.length > 0) return value; + } + return null; +} + +export function resolveAudience(audience: Audience, event: OutboxRecord): AudienceTarget | null { + if (audience === "self") { + const userId = readUserId(event.payload, ["userId", "ownerUserId"]); + return userId ? { kind: "user", userId } : null; + } + + if (audience === "actor") { + const userId = readUserId(event.payload, [ + "actorUserId", + "inviterUserId", + "ownerUserId", + "userId", + ]); + return userId ? { kind: "user", userId } : null; + } + + if (event.organizationId.isNone()) return null; + const organizationId = event.organizationId.unwrap(); + + if (audience === "org:all") return { kind: "org", organizationId, roles: "all" }; + + return { kind: "org", organizationId, roles: rolesWith(audience.can) }; +} diff --git a/apps/app/src/shared/api/mutations/notifications.ts b/apps/app/src/shared/api/mutations/notifications.ts new file mode 100644 index 00000000..0e2263c0 --- /dev/null +++ b/apps/app/src/shared/api/mutations/notifications.ts @@ -0,0 +1,59 @@ +import { mutationOptions } from "@tanstack/react-query"; +import type { InferRequestType, InferResponseType } from "hono/client"; +import { api } from "../api-client"; +import { throwApiError } from "../errors/api-error"; + +const $markRead = api.notifications.read.$post; +const $markAllRead = api.notifications["read-all"].$post; +const $updatePreference = api.notifications.preferences.$put; +const $updateOrgPreference = api.notifications["org-preferences"].$put; + +type MarkReadInput = InferRequestType["json"]; +type UpdatePreferenceInput = InferRequestType["json"]; +type UpdateOrgPreferenceInput = InferRequestType["json"]; + +type OkResponse = InferResponseType; + +async function markReadFn(input: MarkReadInput): Promise { + const res = await $markRead({ json: input }); + if (!res.ok) await throwApiError(res, "Failed to mark notifications as read"); + return res.json(); +} + +async function markAllReadFn(): Promise { + const res = await $markAllRead({}); + if (!res.ok) await throwApiError(res, "Failed to mark all notifications as read"); + return res.json(); +} + +async function updatePreferenceFn(input: UpdatePreferenceInput): Promise { + const res = await $updatePreference({ json: input }); + if (!res.ok) await throwApiError(res, "Failed to update notification preference"); + return res.json(); +} + +async function updateOrgPreferenceFn(input: UpdateOrgPreferenceInput): Promise { + const res = await $updateOrgPreference({ json: input }); + if (!res.ok) await throwApiError(res, "Failed to update org notification preference"); + return res.json(); +} + +export const markReadMutationOptions = mutationOptions({ + mutationKey: ["notifications", "mark-read"] as const, + mutationFn: markReadFn, +}); + +export const markAllReadMutationOptions = mutationOptions({ + mutationKey: ["notifications", "mark-all-read"] as const, + mutationFn: markAllReadFn, +}); + +export const updatePreferenceMutationOptions = mutationOptions({ + mutationKey: ["notifications", "update-preference"] as const, + mutationFn: updatePreferenceFn, +}); + +export const updateOrgPreferenceMutationOptions = mutationOptions({ + mutationKey: ["notifications", "update-org-preference"] as const, + mutationFn: updateOrgPreferenceFn, +}); diff --git a/apps/app/src/shared/api/queries/__tests__/notifications.test.ts b/apps/app/src/shared/api/queries/__tests__/notifications.test.ts new file mode 100644 index 00000000..5091af3f --- /dev/null +++ b/apps/app/src/shared/api/queries/__tests__/notifications.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, test } from "vitest"; +import { + notificationPreferencesQueryOptions, + notificationsQueryOptions, + orgNotificationPreferencesQueryOptions, + unreadCountQueryOptions, +} from "../notifications"; + +describe("cles de query notifications", () => { + test("la liste et le compteur ont des cles distinctes", () => { + expect(notificationsQueryOptions().queryKey).not.toEqual(unreadCountQueryOptions.queryKey); + }); + + test("le curseur fait partie de la cle de liste", () => { + const sansCurseur = notificationsQueryOptions().queryKey; + const avecCurseur = notificationsQueryOptions("2026-08-13T10:00:00Z").queryKey; + expect(avecCurseur).not.toEqual(sansCurseur); + }); + + test("toutes les cles partagent le prefixe notifications", () => { + expect(notificationsQueryOptions().queryKey[0]).toBe("notifications"); + expect(unreadCountQueryOptions.queryKey[0]).toBe("notifications"); + expect(notificationPreferencesQueryOptions.queryKey[0]).toBe("notifications"); + expect(orgNotificationPreferencesQueryOptions.queryKey[0]).toBe("notifications"); + }); +}); diff --git a/apps/app/src/shared/api/queries/notifications.ts b/apps/app/src/shared/api/queries/notifications.ts new file mode 100644 index 00000000..6322088f --- /dev/null +++ b/apps/app/src/shared/api/queries/notifications.ts @@ -0,0 +1,54 @@ +import { queryOptions } from "@tanstack/react-query"; +import type { InferResponseType } from "hono/client"; +import { api } from "../api-client"; +import { throwApiError } from "../errors/api-error"; + +const $list = api.notifications.$get; +const $unreadCount = api.notifications["unread-count"].$get; +const $preferences = api.notifications.preferences.$get; +const $orgPreferences = api.notifications["org-preferences"].$get; + +export type NotificationsResponse = InferResponseType; +export type Notification = NotificationsResponse["items"][number]; + +export type UnreadCountResponse = InferResponseType; + +export type NotificationPreferencesResponse = InferResponseType; +export type NotificationPreference = NotificationPreferencesResponse["items"][number]; + +export const notificationsQueryOptions = (cursor?: string) => + queryOptions({ + queryKey: ["notifications", "list", cursor ?? null] as const, + queryFn: async ({ signal }) => { + const res = await $list({ query: cursor ? { cursor } : {} }, { init: { signal } }); + if (!res.ok) await throwApiError(res, "Failed to load notifications"); + return (await res.json()) as NotificationsResponse; + }, + }); + +export const unreadCountQueryOptions = queryOptions({ + queryKey: ["notifications", "unread-count"] as const, + queryFn: async ({ signal }) => { + const res = await $unreadCount({}, { init: { signal } }); + if (!res.ok) await throwApiError(res, "Failed to load unread count"); + return (await res.json()) as UnreadCountResponse; + }, +}); + +export const notificationPreferencesQueryOptions = queryOptions({ + queryKey: ["notifications", "preferences"] as const, + queryFn: async ({ signal }) => { + const res = await $preferences({}, { init: { signal } }); + if (!res.ok) await throwApiError(res, "Failed to load notification preferences"); + return (await res.json()) as NotificationPreferencesResponse; + }, +}); + +export const orgNotificationPreferencesQueryOptions = queryOptions({ + queryKey: ["notifications", "org-preferences"] as const, + queryFn: async ({ signal }) => { + const res = await $orgPreferences({}, { init: { signal } }); + if (!res.ok) await throwApiError(res, "Failed to load org notification preferences"); + return (await res.json()) as NotificationPreferencesResponse; + }, +}); diff --git a/apps/app/src/shared/auth/auth-broadcast.ts b/apps/app/src/shared/auth/auth-broadcast.ts index 2be15c27..7505b9e4 100644 --- a/apps/app/src/shared/auth/auth-broadcast.ts +++ b/apps/app/src/shared/auth/auth-broadcast.ts @@ -1,17 +1,15 @@ +import { createBroadcastChannel } from "../hooks/use-broadcast-channel"; + type AuthEvent = { type: "session-changed" }; -const channel = - typeof BroadcastChannel !== "undefined" ? new BroadcastChannel("clean-stack-auth") : null; +const channel = createBroadcastChannel("clean-stack-auth"); export function broadcastAuthChange(): void { - channel?.postMessage({ type: "session-changed" } satisfies AuthEvent); + channel.post({ type: "session-changed" }); } export function onAuthChange(handler: () => void): () => void { - if (!channel) return () => {}; - const listener = (event: MessageEvent) => { - if (event.data.type === "session-changed") handler(); - }; - channel.addEventListener("message", listener); - return () => channel.removeEventListener("message", listener); + return channel.subscribe((event) => { + if (event.type === "session-changed") handler(); + }); } diff --git a/apps/app/src/shared/hooks/__tests__/use-broadcast-channel.test.ts b/apps/app/src/shared/hooks/__tests__/use-broadcast-channel.test.ts new file mode 100644 index 00000000..a506d1bb --- /dev/null +++ b/apps/app/src/shared/hooks/__tests__/use-broadcast-channel.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, test, vi } from "vitest"; +import { createBroadcastChannel } from "../use-broadcast-channel"; + +describe("createBroadcastChannel", () => { + test("livre le message aux abonnes", async () => { + const publisher = createBroadcastChannel<{ type: string }>("test-livraison"); + const receiver = createBroadcastChannel<{ type: string }>("test-livraison"); + + const received = new Promise((resolve, reject) => { + const guard = setTimeout( + () => reject(new Error("message non recu dans le delai imparti")), + 500, + ); + receiver.subscribe((m) => { + clearTimeout(guard); + resolve(m.type); + }); + }); + + publisher.post({ type: "ping" }); + + await expect(received).resolves.toBe("ping"); + }); + + test("le desabonnement arrete la livraison", async () => { + const publisher = createBroadcastChannel<{ type: string }>("test-desabo"); + const receiver = createBroadcastChannel<{ type: string }>("test-desabo"); + const handler = vi.fn(); + const unsubscribe = receiver.subscribe(handler); + + unsubscribe(); + publisher.post({ type: "ping" }); + + await new Promise((r) => setTimeout(r, 100)); + + expect(handler).not.toHaveBeenCalled(); + }); + + test("sans BroadcastChannel dans l'environnement, post et subscribe sont inoffensifs", () => { + const original = globalThis.BroadcastChannel; + // @ts-expect-error suppression volontaire pour simuler un environnement sans support + globalThis.BroadcastChannel = undefined; + + const channel = createBroadcastChannel<{ type: string }>("test-absent"); + const unsubscribe = channel.subscribe(() => {}); + + expect(() => channel.post({ type: "ping" })).not.toThrow(); + expect(() => unsubscribe()).not.toThrow(); + + globalThis.BroadcastChannel = original; + }); +}); diff --git a/apps/app/src/shared/hooks/use-broadcast-channel.ts b/apps/app/src/shared/hooks/use-broadcast-channel.ts new file mode 100644 index 00000000..91895401 --- /dev/null +++ b/apps/app/src/shared/hooks/use-broadcast-channel.ts @@ -0,0 +1,15 @@ +export function createBroadcastChannel(name: string) { + const channel = typeof BroadcastChannel !== "undefined" ? new BroadcastChannel(name) : null; + + return { + post(message: T): void { + channel?.postMessage(message); + }, + subscribe(handler: (message: T) => void): () => void { + if (!channel) return () => {}; + const listener = (event: MessageEvent) => handler(event.data); + channel.addEventListener("message", listener); + return () => channel.removeEventListener("message", listener); + }, + }; +} diff --git a/docs/CRON.md b/docs/CRON.md index 81baf2e9..dd6dc67f 100644 --- a/docs/CRON.md +++ b/docs/CRON.md @@ -8,6 +8,8 @@ internal endpoints (`POST /internal/`); you wire your own scheduler. | Endpoint | Body | What it does | |---|---|---| | `POST /internal/rgpd-sweep` | `{ batchSize?: number; dryRun?: boolean }` | Wipes accounts whose 7-day grace window has elapsed (`pendingDeletionUntil <= now AND deletedAt IS NULL`). Idempotent, returns `{ processed, succeeded, failed, dryRun }`. | +| `POST /internal/flush-notification-emails` | `{ batchSize?: number; dryRun?: boolean }` | Groups pending notification emails into per-user/category digests and enqueues them for delivery. Recommended cadence: every minute. Note: "immediate" frequency means "at the next cron tick" — true real-time delivery is handled by the SSE event stream, not email. | +| `POST /internal/sweep-notifications` | `{ batchSize?: number; dryRun?: boolean }` | Purges read notifications older than `NOTIFICATION_RETENTION_DAYS` (default 30d). Unread notifications are never purged regardless of age. Recommended cadence: daily. | ## Authentication diff --git a/docs/EVENTS.md b/docs/EVENTS.md index 559ffc5b..5fa3012f 100644 --- a/docs/EVENTS.md +++ b/docs/EVENTS.md @@ -282,9 +282,32 @@ Three surfaces consume the visibility map simultaneously: Changing a public event type string or removing a payload field requires a changelog entry and a deprecation window. Promoting a new event to public is permanent — plan for it in the PR review. +## Notifications — catalogue de notifiabilité + +`packages/events/src/notification-map.ts` est la **troisième projection** du catalogue d'événements, après `visibility-map.ts` (webhooks) et `retention-map.ts` (purge). + +```ts +// notification-map.ts +export const NOTIFICATION_MAP = { + "billing.payment.failed": { audience: { can: { billing: ["read"] } }, category: "billing", forced: true }, + "org.member.joined": { audience: { can: { organization: ["update"] } }, category: "org", groupBy: "resource" }, + // ... +} satisfies Partial>; +``` + +**Ce que cette projection projette.** Pour chaque type d'événement listé, elle déclare : +- `audience` — qui doit recevoir la notification : `"self"` (l'utilisateur concerné), `"actor"`, `"org:all"`, ou `{ can: OrgPermissions }` (les membres dont le rôle porte la capability). La résolution capability→roles est faite par `rolesWith(audience.can)` (`@packages/access-control`) et reste cohérente avec le reste des gates de l'app. +- `category` — regroupement UI (`security`, `org`, `billing`, `activity`). +- `forced?: true` — contourne les préférences utilisateur et le batching email (bypass critique du SOTA). +- `groupBy` et `dedupWindow` — fenêtre de déduplication et clé de regroupement lecteur (style Linear "X et 3 autres"). + +**Pourquoi elle ne crée aucun événement.** Une notification est une projection de lecture d'un événement déjà audité et déjà émis dans l'outbox. Émettre un `notification.created` créerait une boucle : son propre abonné (`NotificationFanoutSubscriber`) déclencherait à nouveau l'insertion. La création de notifications n'émet aucun événement — une notification est une projection d'un événement déjà audité, et émettre un `notification.created` créerait une boucle avec son propre abonné. En revanche, les mutations de préférences (`notification.preference.updated`, `notification.org_preference.updated`) émettent bien des événements car ce sont des changements d'état persistant. Le catalogue est à **67 événements / 28 publics / 39 internes**. + +**`NotificationFanoutSubscriber`** est l'abonné outbox qui lit cette projection (aux côtés de `AuditEventSubscriber` et `WebhookFanoutSubscriber`). Il tourne dans la même transaction que `markDispatched` — une notification perdue ne passe pas inaperçue. Les événements absents du catalogue ne génèrent aucune notification (le comportement par défaut : la plupart des 67 événements sont audit-only). + ## BetterAuth bridge — what fires what -The boilerplate emits **65 events** (28 public + 37 internal) automatically. Sources: 23 from `apps/api/src/auth.ts` covering BetterAuth lifecycles, 5 from `modules/rgpd/`, 3 from `modules/uploads/`, **7 from `modules/webhooks/`** (3 CRUD + 4 internal: test, secret_rotated, disabled, exhausted), 1 from `modules/policies/`, **2 from `modules/consents/`**, 5 from security (3 middleware/endpoint + 2 abuse-prevention hooks in `auth.ts`), **4 from `modules/billing/`**, **1 from quota middleware**, **1 from audit-log operator** (`security.operator.audit_accessed`), **1 from email delivery worker** (`email.delivery.exhausted`), **7 from `modules/admin/`** (Phase C.3 — 5 actions + 2 impersonation lifecycle), **3 from `modules/api-token/`** (Phase C.4 — created, revoked, used). Source of truth: `packages/events/src/event-types.ts` + `packages/events/src/visibility-map.ts`. **Internal events** skip `WebhookFanoutSubscriber` — they flow to `audit_log` and in-process handlers only. +The boilerplate emits **67 events** (28 public + 39 internal) automatically. Sources: 23 from `apps/api/src/auth.ts` covering BetterAuth lifecycles, 5 from `modules/rgpd/`, 3 from `modules/uploads/`, **7 from `modules/webhooks/`** (3 CRUD + 4 internal: test, secret_rotated, disabled, exhausted), 1 from `modules/policies/`, **2 from `modules/consents/`**, 5 from security (3 middleware/endpoint + 2 abuse-prevention hooks in `auth.ts`), **4 from `modules/billing/`**, **1 from quota middleware**, **1 from audit-log operator** (`security.operator.audit_accessed`), **1 from email delivery worker** (`email.delivery.exhausted`), **7 from `modules/admin/`** (Phase C.3 — 5 actions + 2 impersonation lifecycle), **3 from `modules/api-token/`** (Phase C.4 — created, revoked, used), **2 from `modules/notifications/`** (Phase D.3 — preference.updated, org_preference.updated). Source of truth: `packages/events/src/event-types.ts` + `packages/events/src/visibility-map.ts`. **Internal events** skip `WebhookFanoutSubscriber` — they flow to `audit_log` and in-process handlers only. ### Via `databaseHooks` (TX-bound, captures all flows) - `USER_CREATED` — `databaseHooks.user.create.after` @@ -414,7 +437,7 @@ The guard lives in `DrizzleOutboxRepository.enqueue` (the single porte d'entrée | Path | Role | |---|---| -| `packages/events/src/{event-types,payloads,retention-map}.ts` | Central catalog (65 events: 28 public + 37 internal) | +| `packages/events/src/{event-types,payloads,retention-map}.ts` | Central catalog (67 events: 28 public + 39 internal) | | `packages/events/src/visibility-map.ts` | Allowlist: `"public"` = customer contract, `"internal"` = operational signal. Drives fanout, picker, and public catalog simultaneously. | | `packages/events/src/{descriptions,json-schema}.ts` | Human-readable descriptions + `jsonSchemaForEvent` (Zod 4 `z.toJSONSchema`) — consumed by public catalog + `EventTypePicker` | | `packages/ddd-kit/src/events/{event-collector,on-event,outbox-mapping}.ts` | ALS collector + handler factory + CloudEvents mapping | diff --git a/docs/FEATURES.md b/docs/FEATURES.md index cd74582a..2a7a6fe9 100644 --- a/docs/FEATURES.md +++ b/docs/FEATURES.md @@ -350,9 +350,9 @@ Machine-to-machine access with scoped, expirable tokens. Tokens are shown once a - `forms/token-form.tsx` — name + scope checkboxes + optional expiry. Created token shown once via ``. - `api/api-tokens.{queries,mutations}.ts` — list + create + revoke. -**Event visibility** `packages/events/src/visibility-map.ts` — 65-event catalog with explicit `public` / `internal` classification (28 public / 37 internal). Three consumers: `WebhookFanoutSubscriber` (only fans out public events), `/developers/events` catalog page (only lists public events), and webhook subscription picker (only offers public events). +**Event visibility** `packages/events/src/visibility-map.ts` — 67-event catalog with explicit `public` / `internal` classification (28 public / 39 internal). Three consumers: `WebhookFanoutSubscriber` (only fans out public events), `/developers/events` catalog page (only lists public events), and webhook subscription picker (only offers public events). -**Events** (3, `operational` retention): `api_token.created` (public), `api_token.revoked` (public), `api_token.used` (internal, sampled via bucket) → **65 total / 28 public / 37 internal**. +**Events** (3, `operational` retention): `api_token.created` (public), `api_token.revoked` (public), `api_token.used` (internal, sampled via bucket) → **67 total / 28 public / 39 internal**. **Email** `packages/emails/src/components/api-token-leaked.tsx` — template sent to the token owner when GitHub Secret Scanning reports a match. diff --git a/docs/REMOVABILITY.md b/docs/REMOVABILITY.md index cb3a41dd..bbae8d08 100644 --- a/docs/REMOVABILITY.md +++ b/docs/REMOVABILITY.md @@ -128,7 +128,7 @@ Reference cartography for removing the Personal Access Tokens module. Walk the 6 | Axis | Touch-points | |---|---| | 1. **Back code** | `trash apps/api/src/modules/api-token/`. `container.ts` — remove `.addModule(apiTokenModule)` and the `import`. `index.ts` — remove `/api/v1` mount, `/api/token-scanning/*` scanning mount, and `apiTokenRoutes` + `createApiTokenScanningRoutes` imports. Remove `apps/api/src/public-api/` entirely (sub-app, no other consumers). | -| 2. **Events** | `packages/events/src/event-types.ts` — 3 types: `api_token.created`, `api_token.revoked`, `api_token.used`. `payloads.ts` — 3 payload types. `retention-map.ts` — 3 entries. `visibility-map.ts` — 3 entries (`public`, `public`, `internal`). Event count: 65 → 62. | +| 2. **Events** | `packages/events/src/event-types.ts` — 3 types: `api_token.created`, `api_token.revoked`, `api_token.used`. `payloads.ts` — 3 payload types. `retention-map.ts` — 3 entries. `visibility-map.ts` — 3 entries (`public`, `public`, `internal`). Event count: 67 → 64. | | 3. **Shared ports / middleware** | `apps/api/src/shared/middleware/api-token.middleware.ts` — delete (no other consumers). `apps/api/src/shared/crypto/api-token.ts` — delete (only used by `api-token.middleware.ts` and `scanning.routes.ts`). `apps/api/src/shared/middleware/auth.middleware.ts` — remove the `/api/v1/` path exclusion from `sessionMiddleware` (the sub-app bypass). `apps/api/src/shared/middleware/rate-limit.policies.ts` — remove `API_TOKEN_POLICY` + `API_TOKEN_IP_POLICY`. `apps/api/src/auth-queries.ts` — verify `findUserById` is still consumed by other modules (webhooks, admin) before removing. | | 4. **Env vars** | `apps/api/src/shared/env.ts` — remove `API_TOKEN_PEPPER`, `API_TOKEN_PEPPER_PREVIOUS`, `API_TOKEN_PREFIX`, `API_TOKEN_MAX_EXPIRY_DAYS`, `API_TOKEN_LAST_USED_BUCKET_MIN`, `API_TOKEN_PEPPER_VERSION` and the production boot guard that requires `API_TOKEN_PEPPER`. `apps/api/.env.example` — same 6 keys. | | 5. **Schema** | `packages/drizzle/src/schema/api-token.ts` — delete. Remove barrel re-export from `packages/drizzle/src/index.ts`. Run `pnpm db:generate` — expect 1 `DROP TABLE api_token` migration. Read the SQL before committing. | diff --git a/packages/access-control/package.json b/packages/access-control/package.json index 1f3d238e..8a7107fc 100644 --- a/packages/access-control/package.json +++ b/packages/access-control/package.json @@ -7,13 +7,17 @@ ".": "./src/index.ts" }, "scripts": { - "type-check": "tsc --noEmit" + "type-check": "tsc --noEmit", + "test": "vitest run", + "test:watch": "vitest watch" }, "dependencies": { "better-auth": "^1.6.20" }, "devDependencies": { + "@packages/test": "workspace:*", "@packages/typescript-config": "workspace:*", - "typescript": "^6.0.3" + "typescript": "^6.0.3", + "vitest": "^4.1.9" } } diff --git a/packages/access-control/src/__tests__/roles-with.test.ts b/packages/access-control/src/__tests__/roles-with.test.ts new file mode 100644 index 00000000..9beada01 --- /dev/null +++ b/packages/access-control/src/__tests__/roles-with.test.ts @@ -0,0 +1,20 @@ +import { describe, expect, test } from "vitest"; +import { rolesWith } from "../index"; + +describe("rolesWith", () => { + test("billing:read couvre owner et admin", () => { + expect(rolesWith({ billing: ["read"] }).sort()).toEqual(["admin", "owner"]); + }); + + test("billing:manage ne couvre que owner", () => { + expect(rolesWith({ billing: ["manage"] })).toEqual(["owner"]); + }); + + test("organization:update couvre owner et admin, pas member", () => { + expect(rolesWith({ organization: ["update"] })).not.toContain("member"); + }); + + test("une capability inconnue de tous ne renvoie aucun role", () => { + expect(rolesWith({ billing: ["read", "manage"], apiToken: ["revoke"] })).toEqual(["owner"]); + }); +}); diff --git a/packages/access-control/src/index.ts b/packages/access-control/src/index.ts index 1119c4c0..018cc20b 100644 --- a/packages/access-control/src/index.ts +++ b/packages/access-control/src/index.ts @@ -69,6 +69,10 @@ export function authorizeRole( return policy.authorize(permissions, connector).success; } +export function rolesWith(permissions: OrgPermissions): OrgRole[] { + return ORG_ROLES.filter((role) => authorizeRole(role, permissions)); +} + export const ac = _ac as unknown as AccessControl; export const roles = _roles; diff --git a/packages/access-control/vitest.config.ts b/packages/access-control/vitest.config.ts new file mode 100644 index 00000000..31ff5616 --- /dev/null +++ b/packages/access-control/vitest.config.ts @@ -0,0 +1,4 @@ +import baseConfig from "@packages/test/base-vitest.config"; +import { defineConfig, mergeConfig } from "vitest/config"; + +export default mergeConfig(baseConfig, defineConfig({})); diff --git a/packages/drizzle/package.json b/packages/drizzle/package.json index 56fbaed7..b0007c5e 100644 --- a/packages/drizzle/package.json +++ b/packages/drizzle/package.json @@ -21,6 +21,7 @@ "dependencies": { "@packages/cookie-consent": "workspace:*", "@packages/ddd-kit": "workspace:*", + "@packages/events": "workspace:*", "@packages/policies": "workspace:*", "dotenv": "^17.4.2", "drizzle-orm": "^0.45.2", diff --git a/packages/drizzle/src/index.ts b/packages/drizzle/src/index.ts index fd36c2e5..1c8d503a 100644 --- a/packages/drizzle/src/index.ts +++ b/packages/drizzle/src/index.ts @@ -34,6 +34,7 @@ export * as auditLogSchema from "./schema/audit-log"; export * as authSchema from "./schema/auth"; export * as consentSchema from "./schema/consent"; export * as multiTenantSchema from "./schema/multi-tenant"; +export * as notificationSchema from "./schema/notification"; export const schema = { ...authSchema, ...multiTenantSchema }; export * as billingSchema from "./schema/billing"; export * as emailSchema from "./schema/email"; diff --git a/packages/drizzle/src/schema/notification.ts b/packages/drizzle/src/schema/notification.ts new file mode 100644 index 00000000..47ad09c1 --- /dev/null +++ b/packages/drizzle/src/schema/notification.ts @@ -0,0 +1,65 @@ +import { + NOTIFICATION_CHANNELS, + NOTIFICATION_FREQUENCIES, + NOTIFICATION_PREFERENCE_SCOPES, +} from "@packages/events"; +import { sql } from "drizzle-orm"; +import { boolean, index, jsonb, pgTable, text, timestamp, uniqueIndex } from "drizzle-orm/pg-core"; +import { user } from "./auth"; +import { organization } from "./multi-tenant"; + +export const notification = pgTable( + "notification", + { + id: text("id").primaryKey(), + userId: text("user_id") + .notNull() + .references(() => user.id, { onDelete: "cascade" }), + organizationId: text("organization_id").references(() => organization.id, { + onDelete: "cascade", + }), + category: text("category").notNull(), + eventType: text("event_type").notNull(), + groupKey: text("group_key"), + dedupKey: text("dedup_key"), + payload: jsonb("payload").$type().notNull(), + readAt: timestamp("read_at"), + emailPendingAt: timestamp("email_pending_at"), + emailSentAt: timestamp("email_sent_at"), + createdAt: timestamp("created_at").defaultNow().notNull(), + }, + (table) => [ + index("notification_unread_idx").on(table.userId).where(sql`${table.readAt} IS NULL`), + index("notification_feed_idx").on(table.userId, table.createdAt), + uniqueIndex("notification_dedup_uidx") + .on(table.userId, table.dedupKey) + .where(sql`${table.dedupKey} IS NOT NULL`), + index("notification_email_pending_idx") + .on(table.emailPendingAt) + .where(sql`${table.emailSentAt} IS NULL AND ${table.emailPendingAt} IS NOT NULL`), + index("notification_sweep_idx").on(table.createdAt).where(sql`${table.readAt} IS NOT NULL`), + ], +); + +export const notificationPreference = pgTable( + "notification_preference", + { + id: text("id").primaryKey(), + scope: text("scope", { enum: NOTIFICATION_PREFERENCE_SCOPES }).notNull(), + scopeId: text("scope_id").notNull(), + category: text("category").notNull(), + channel: text("channel", { enum: NOTIFICATION_CHANNELS }).notNull(), + enabled: boolean("enabled").notNull(), + frequency: text("frequency", { enum: NOTIFICATION_FREQUENCIES }).notNull().default("immediate"), + locked: boolean("locked").notNull().default(false), + updatedAt: timestamp("updated_at").defaultNow().notNull(), + }, + (table) => [ + uniqueIndex("notification_preference_uidx").on( + table.scope, + table.scopeId, + table.category, + table.channel, + ), + ], +); diff --git a/packages/emails/src/__tests__/render.test.ts b/packages/emails/src/__tests__/render.test.ts index a23e8185..4d769b8a 100644 --- a/packages/emails/src/__tests__/render.test.ts +++ b/packages/emails/src/__tests__/render.test.ts @@ -30,6 +30,11 @@ const STUB_VARS = { tokenName: "CI token", revokedAt: "6 août 2026 à 10:00", }, + notification_digest: { + category: "billing", + itemCount: "2", + itemsSummary: "billing.payment.failed, billing.payment.failed", + }, } as const satisfies Record; describe("renderTemplate", () => { diff --git a/packages/emails/src/components/notification-digest.tsx b/packages/emails/src/components/notification-digest.tsx new file mode 100644 index 00000000..7566570f --- /dev/null +++ b/packages/emails/src/components/notification-digest.tsx @@ -0,0 +1,18 @@ +import { Heading, Text } from "react-email"; +import type { EmailTemplates } from "../templates"; +import { EmailLayout } from "./layout"; + +type NotificationDigestVars = EmailTemplates["notification_digest"]; +interface NotificationDigestProps extends NotificationDigestVars {} + +export function NotificationDigest({ category, itemCount, itemsSummary }: NotificationDigestProps) { + return ( + + + {itemCount} new {category} notification{itemCount === "1" ? "" : "s"} + + Here is a summary of your recent {category} activity: + {itemsSummary} + + ); +} diff --git a/packages/emails/src/render.tsx b/packages/emails/src/render.tsx index 58d153f9..7c15488a 100644 --- a/packages/emails/src/render.tsx +++ b/packages/emails/src/render.tsx @@ -9,6 +9,7 @@ import { DeleteCompleted } from "./components/delete-completed"; import { DeleteRequested } from "./components/delete-requested"; import { ImpersonationStarted } from "./components/impersonation-started"; import { MagicLink } from "./components/magic-link"; +import { NotificationDigest } from "./components/notification-digest"; import { OrgInvitation } from "./components/org-invitation"; import { ResetPassword } from "./components/reset-password"; import { VerifyEmail } from "./components/verify-email"; @@ -47,6 +48,11 @@ const TEMPLATES: { [K in EmailTemplateKey]: TemplateEntry } = { component: ApiTokenLeaked, subject: () => "Your API token was automatically revoked", }, + notification_digest: { + component: NotificationDigest, + subject: (v) => + `${v.itemCount} new ${v.category} notification${v.itemCount === "1" ? "" : "s"}`, + }, }; export const EMAIL_TEMPLATE_KEYS = Object.keys(TEMPLATES) as EmailTemplateKey[]; diff --git a/packages/emails/src/templates.ts b/packages/emails/src/templates.ts index 491b9120..e4f15328 100644 --- a/packages/emails/src/templates.ts +++ b/packages/emails/src/templates.ts @@ -21,6 +21,11 @@ export type EmailTemplates = { tokenName: string; revokedAt: string; }; + notification_digest: { + category: string; + itemCount: string; + itemsSummary: string; + }; }; export type EmailTemplateKey = keyof EmailTemplates; diff --git a/packages/events/package.json b/packages/events/package.json index e9fe1e3c..ac49771d 100644 --- a/packages/events/package.json +++ b/packages/events/package.json @@ -16,6 +16,7 @@ "test:watch": "vitest watch" }, "dependencies": { + "@packages/access-control": "workspace:*", "zod": "^4.4.3" }, "devDependencies": { diff --git a/packages/events/src/__tests__/notification-map.test.ts b/packages/events/src/__tests__/notification-map.test.ts new file mode 100644 index 00000000..4112f850 --- /dev/null +++ b/packages/events/src/__tests__/notification-map.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, test } from "vitest"; +import { ALL_EVENT_TYPES } from "../event-types"; +import { isNotifiable, NOTIFICATION_MAP, notificationConfigOf } from "../notification-map"; + +describe("NOTIFICATION_MAP", () => { + test("les events d'audit purs ne sont pas notifiables", () => { + expect(isNotifiable("api_token.used")).toBe(false); + expect(isNotifiable("security.csp.violation")).toBe(false); + expect(isNotifiable("webhook.delivery.exhausted")).toBe(false); + }); + + test("toute cle du map est un event connu", () => { + for (const key of Object.keys(NOTIFICATION_MAP)) { + expect(ALL_EVENT_TYPES).toContain(key); + } + }); + + test("les events security sont forced", () => { + const config = notificationConfigOf("user.password_changed"); + expect(config?.forced).toBe(true); + expect(config?.category).toBe("security"); + }); + + test("billing.payment.failed cible billing:read et non manage", () => { + const config = notificationConfigOf("billing.payment.failed"); + expect(config?.audience).toEqual({ can: { billing: ["read"] } }); + }); + + test("aucun event forced ne cible org:all", () => { + for (const [type, config] of Object.entries(NOTIFICATION_MAP)) { + if (!config.forced) continue; + expect(config.audience, `${type} forced vers toute l'org`).not.toBe("org:all"); + } + }); + + test("un event forced n'est jamais batche par dedupWindow", () => { + for (const [type, config] of Object.entries(NOTIFICATION_MAP)) { + if (!config.forced) continue; + expect(config.dedupWindow, `${type} forced avec une fenetre de dedup`).toBeUndefined(); + } + }); +}); diff --git a/packages/events/src/__tests__/webhook-events.test.ts b/packages/events/src/__tests__/webhook-events.test.ts index 8eb9ba41..f667546a 100644 --- a/packages/events/src/__tests__/webhook-events.test.ts +++ b/packages/events/src/__tests__/webhook-events.test.ts @@ -9,8 +9,8 @@ import { import { RETENTION_MAP } from "../retention-map"; describe("webhook SOTA events", () => { - it("catalog contains exactly 65 event types", () => { - expect(ALL_EVENT_TYPES).toHaveLength(65); + it("catalog contains exactly 67 event types", () => { + expect(ALL_EVENT_TYPES).toHaveLength(67); }); it("declares the 4 new event constants", () => { diff --git a/packages/events/src/event-descriptions.ts b/packages/events/src/event-descriptions.ts index 076749b2..1d7a5d5e 100644 --- a/packages/events/src/event-descriptions.ts +++ b/packages/events/src/event-descriptions.ts @@ -75,6 +75,10 @@ export const EVENT_DESCRIPTIONS: Record = { [EventTypes.API_TOKEN_CREATED]: "A personal access token was created.", [EventTypes.API_TOKEN_REVOKED]: "A personal access token was revoked.", [EventTypes.API_TOKEN_USED]: "A personal access token was used to authenticate a request.", + [EventTypes.NOTIFICATION_PREFERENCE_UPDATED]: + "A user updated their personal notification preferences.", + [EventTypes.NOTIFICATION_ORG_PREFERENCE_UPDATED]: + "An organization administrator updated the organization notification preferences.", }; export function descriptionFor(eventType: string): string { diff --git a/packages/events/src/event-types.ts b/packages/events/src/event-types.ts index 35e14d65..fd54fecf 100644 --- a/packages/events/src/event-types.ts +++ b/packages/events/src/event-types.ts @@ -66,6 +66,8 @@ export const EventTypes = { ADMIN_USER_PASSWORD_RESET: "admin.user.password_reset", ADMIN_USER_SESSIONS_REVOKED: "admin.user.sessions_revoked", EMAIL_DELIVERY_EXHAUSTED: "email.delivery.exhausted", + NOTIFICATION_PREFERENCE_UPDATED: "notification.preference.updated", + NOTIFICATION_ORG_PREFERENCE_UPDATED: "notification.org_preference.updated", } as const; export type EventType = (typeof EventTypes)[keyof typeof EventTypes]; diff --git a/packages/events/src/index.ts b/packages/events/src/index.ts index f7808cd4..542d5bb3 100644 --- a/packages/events/src/index.ts +++ b/packages/events/src/index.ts @@ -1,6 +1,7 @@ export * from "./event-descriptions"; export * from "./event-types"; export * from "./json-schema"; +export * from "./notification-map"; export * from "./payloads"; export * from "./retention-map"; export * from "./visibility-map"; diff --git a/packages/events/src/notification-map.ts b/packages/events/src/notification-map.ts new file mode 100644 index 00000000..53e9be72 --- /dev/null +++ b/packages/events/src/notification-map.ts @@ -0,0 +1,77 @@ +import type { OrgPermissions } from "@packages/access-control"; +import type { EventType } from "./event-types"; + +export type Audience = "self" | "actor" | "org:all" | { can: OrgPermissions }; + +export const NOTIFICATION_CHANNELS = ["in_app", "email"] as const; +export type NotificationChannel = (typeof NOTIFICATION_CHANNELS)[number]; + +export const NOTIFICATION_FREQUENCIES = ["immediate", "hourly", "daily"] as const; +export type NotificationFrequency = (typeof NOTIFICATION_FREQUENCIES)[number]; + +export const NOTIFICATION_PREFERENCE_SCOPES = ["user", "org"] as const; +export type NotificationPreferenceScope = (typeof NOTIFICATION_PREFERENCE_SCOPES)[number]; + +export const NOTIFICATION_CATEGORIES = ["security", "org", "billing", "activity"] as const; +export type NotificationCategory = (typeof NOTIFICATION_CATEGORIES)[number]; + +export type NotificationConfig = { + audience: Audience; + category: NotificationCategory; + forced?: boolean; + groupBy?: "actor" | "resource"; + dedupWindow?: "hour" | "day"; +}; + +const _map = { + "user.password_changed": { audience: "self", category: "security", forced: true }, + "user.mfa.enabled": { audience: "self", category: "security", forced: true }, + "user.mfa.disabled": { audience: "self", category: "security", forced: true }, + "user.mfa.backup_code_used": { audience: "self", category: "security", forced: true }, + "user.passkey.added": { audience: "self", category: "security", forced: true }, + "user.passkey.removed": { audience: "self", category: "security", forced: true }, + "user.email.change_requested": { audience: "self", category: "security", forced: true }, + "user.export.completed": { audience: "self", category: "activity" }, + "user.deletion.requested": { audience: "self", category: "security", forced: true }, + "user.deletion.cancelled": { audience: "self", category: "security", forced: true }, + "org.member.joined": { + audience: { can: { organization: ["update"] } }, + category: "org", + groupBy: "resource", + }, + "org.member.removed": { audience: { can: { organization: ["update"] } }, category: "org" }, + "org.member.role_changed": { audience: "self", category: "org" }, + "org.member.invited": { audience: { can: { organization: ["update"] } }, category: "org" }, + "billing.payment.failed": { + audience: { can: { billing: ["read"] } }, + category: "billing", + forced: true, + }, + "billing.subscription.created": { audience: { can: { billing: ["read"] } }, category: "billing" }, + "billing.subscription.updated": { audience: { can: { billing: ["read"] } }, category: "billing" }, + "billing.subscription.cancelled": { + audience: { can: { billing: ["read"] } }, + category: "billing", + forced: true, + }, + "billing.quota.exceeded": { + audience: { can: { billing: ["read"] } }, + category: "billing", + dedupWindow: "day", + }, + "webhook.endpoint.disabled": { + audience: { can: { webhooks: ["read"] } }, + category: "activity", + }, + "api_token.created": { audience: "self", category: "security", forced: true }, +} as const satisfies Partial>; + +export const NOTIFICATION_MAP: Partial> = _map; + +export function notificationConfigOf(eventType: string): NotificationConfig | undefined { + return (NOTIFICATION_MAP as Record)[eventType]; +} + +export function isNotifiable(eventType: string): boolean { + return notificationConfigOf(eventType) !== undefined; +} diff --git a/packages/events/src/payloads.ts b/packages/events/src/payloads.ts index 096a9a5b..2a2bbb07 100644 --- a/packages/events/src/payloads.ts +++ b/packages/events/src/payloads.ts @@ -450,6 +450,28 @@ export const ApiTokenUsedPayload = z.object({ }); export type ApiTokenUsedPayload = z.infer; +export const NotificationPreferenceUpdatedPayload = UserRef.extend({ + category: z.string(), + channel: z.string(), + enabled: z.boolean(), + frequency: z.string(), +}); +export type NotificationPreferenceUpdatedPayload = z.infer< + typeof NotificationPreferenceUpdatedPayload +>; + +export const NotificationOrgPreferenceUpdatedPayload = OrgRef.extend({ + actorUserId: z.string(), + category: z.string(), + channel: z.string(), + enabled: z.boolean(), + frequency: z.string(), + locked: z.boolean(), +}); +export type NotificationOrgPreferenceUpdatedPayload = z.infer< + typeof NotificationOrgPreferenceUpdatedPayload +>; + export const PayloadByEventType = { [EventTypes.USER_CREATED]: UserCreatedPayload, [EventTypes.USER_SIGNED_IN]: UserSignedInPayload, @@ -516,4 +538,6 @@ export const PayloadByEventType = { [EventTypes.API_TOKEN_CREATED]: ApiTokenCreatedPayload, [EventTypes.API_TOKEN_REVOKED]: ApiTokenRevokedPayload, [EventTypes.API_TOKEN_USED]: ApiTokenUsedPayload, + [EventTypes.NOTIFICATION_PREFERENCE_UPDATED]: NotificationPreferenceUpdatedPayload, + [EventTypes.NOTIFICATION_ORG_PREFERENCE_UPDATED]: NotificationOrgPreferenceUpdatedPayload, } as const; diff --git a/packages/events/src/retention-map.ts b/packages/events/src/retention-map.ts index 1a5d7da9..ae75b45f 100644 --- a/packages/events/src/retention-map.ts +++ b/packages/events/src/retention-map.ts @@ -68,6 +68,8 @@ export const RETENTION_MAP: Record = { [EventTypes.API_TOKEN_CREATED]: "compliance", [EventTypes.API_TOKEN_REVOKED]: "compliance", [EventTypes.API_TOKEN_USED]: "operational", + [EventTypes.NOTIFICATION_PREFERENCE_UPDATED]: "compliance", + [EventTypes.NOTIFICATION_ORG_PREFERENCE_UPDATED]: "compliance", }; export function retentionFor(eventType: string): RetentionPolicy { diff --git a/packages/events/src/visibility-map.ts b/packages/events/src/visibility-map.ts index 7f85702a..74d3812c 100644 --- a/packages/events/src/visibility-map.ts +++ b/packages/events/src/visibility-map.ts @@ -68,6 +68,8 @@ export const VISIBILITY = { "admin.user.password_reset": "internal", "admin.user.sessions_revoked": "internal", "email.delivery.exhausted": "internal", + "notification.preference.updated": "internal", + "notification.org_preference.updated": "internal", } satisfies Record; export function isPublicEvent(eventType: EventType): boolean { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e6152315..355d3d05 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -269,12 +269,18 @@ importers: specifier: ^1.6.20 version: 1.6.20(@opentelemetry/api@1.9.1)(drizzle-kit@0.31.10)(drizzle-orm@0.45.2(@opentelemetry/api@1.9.1)(@types/pg@8.20.0)(bun-types@1.3.14)(kysely@0.29.2)(pg@8.22.0))(pg@8.22.0)(react-dom@19.2.7(react@19.2.7))(react@19.2.7)(vitest@4.1.9) devDependencies: + '@packages/test': + specifier: workspace:* + version: link:../test '@packages/typescript-config': specifier: workspace:* version: link:../typescript-config typescript: specifier: ^6.0.3 version: 6.0.3 + vitest: + specifier: ^4.1.9 + version: 4.1.9(@opentelemetry/api@1.9.1)(@types/node@26.0.1)(@vitest/coverage-v8@4.1.9)(vite@8.1.0(@types/node@26.0.1)(esbuild@0.28.1)(jiti@2.7.0)(tsx@4.22.4)(yaml@2.9.0)) packages/cookie-consent: devDependencies: @@ -321,6 +327,9 @@ importers: '@packages/ddd-kit': specifier: workspace:* version: link:../ddd-kit + '@packages/events': + specifier: workspace:* + version: link:../events '@packages/policies': specifier: workspace:* version: link:../policies @@ -377,6 +386,9 @@ importers: packages/events: dependencies: + '@packages/access-control': + specifier: workspace:* + version: link:../access-control zod: specifier: ^4.4.3 version: 4.4.3