diff --git a/docs/.vitepress/config.mts b/docs/.vitepress/config.mts index d0daecd3..29d794da 100644 --- a/docs/.vitepress/config.mts +++ b/docs/.vitepress/config.mts @@ -176,6 +176,10 @@ export default defineConfig({ { text: "Authentication model", link: "/architecture/auth-model" }, { text: "Magic-link auth", link: "/architecture/magic-link-auth" }, { text: "Background jobs", link: "/architecture/jobs" }, + { + text: "Message bus & notifications", + link: "/architecture/message-bus-and-notifications", + }, { text: "UI diagnostics", link: "/architecture/ui-diagnostics" }, ], }, diff --git a/docs/architecture/index.md b/docs/architecture/index.md index 2ed52f5d..439d49fb 100644 --- a/docs/architecture/index.md +++ b/docs/architecture/index.md @@ -77,3 +77,4 @@ src/ - [Backend Storage →](/architecture/backend-storage) - [Derived and Attached Blobs →](/architecture/derived-and-attached-blobs) — thumbnails, transcodes and uploaded previews: why content-keyed and file-keyed artifacts need separate tables - [Background Jobs →](/architecture/jobs) +- [Message Bus & Notifications →](/architecture/message-bus-and-notifications) — real-time WebSocket bus (topics, AuthZ scopes, tab-visibility grace-close), persistent notifications (bell), AsyncAPI vs OpenAPI schema ownership diff --git a/docs/architecture/message-bus-and-notifications.md b/docs/architecture/message-bus-and-notifications.md new file mode 100644 index 00000000..ad18e26e --- /dev/null +++ b/docs/architecture/message-bus-and-notifications.md @@ -0,0 +1,645 @@ +# Message Bus & Persistent Notifications + +OxiCloud has two coupled subsystems that together power its real-time +UX — a **live message bus** over WebSocket for "something just +happened, refresh your view", and a **persistent notifications +table** for "you need to know about this even if you weren't +online." This document explains how both work, how they authenticate +and authorize subscribers, and why the frontend deliberately drops +the WebSocket while a tab is hidden. + +Design docs the shipped code implements: [`docs/plan/message-bus.md`](https://github.com/oxicloud/oxicloud/blob/main/docs/plan/message-bus.md) ++ [`docs/plan/templated-messages.md`](https://github.com/oxicloud/oxicloud/blob/main/docs/plan/templated-messages.md). + +--- + +## The two channels + +| | **Message bus (WebSocket)** | **Persistent notifications (REST + DB)** | +|---|---|---| +| Purpose | "Something changed, refresh your view" | "You need to know about this — later is fine" | +| Transport | JSON-RPC 2.0 over `/api/rt/ws` | `GET/POST/DELETE /api/notifications/*` + `notif.notifications` table | +| Delivery | Best-effort, in-memory, no replay | Durable, per-user rows, survive reboot / offline | +| Payload | Thin "poke" facts (id + verb) | Full per-kind DTO with all render data | +| Loss on disconnect | Yes (events during outage window are dropped) | No (rows are the source of truth) | +| Schema owner | AsyncAPI (`resources/gen/asyncapi.json`) | OpenAPI (`resources/gen/openapi.json`) | + +The two work together: an ingester that wants to notify a user +writes **both** — the DB row (for durability + the bell's history) +AND publishes a bus event on `user:{u}:notifications` (so online +sessions refetch instantly instead of waiting for the next +mount). The wire event on that topic is a **pure poke** — empty +`data: {}`. The row's real content only ever crosses the REST wire. + +--- + +## Schema ownership — AsyncAPI vs OpenAPI + +The bus and the REST endpoints have separate wire specs. The rule +the codebase adopts to keep them from drifting: + +> **AsyncAPI defines the envelope + transport for clients. +> OpenAPI defines the payload.** + +Concretely: + +| Type | Home | How it stays in sync with Rust | +|---|---|---| +| **Bus events** (`MessageBusEvent`, subscribe / unsubscribe frames, revoked notifications, envelope shape) | **AsyncAPI** — `resources/gen/asyncapi.json` | Hand-written in `src/bin/generate-asyncapi.rs` via `json!` macros; kept in lockstep with `MessageBusEvent`'s serde shape. Small drift risk — Rust is truth. | +| **REST DTOs** (response bodies, request bodies, per-kind notification payloads) | **OpenAPI** — `resources/gen/openapi.json` | `#[derive(utoipa::ToSchema)]` on the Rust struct. Utoipa walks `#[utoipa::path(...)]` handlers + registered schemas. **No drift possible** — projection is derived from Rust. | +| **Types on both wires** (rare; none today) | Would live as one Rust struct with both derives, or wait for single-source codegen | — | + +### Why this split, not one unified spec + +The instinct is to put the notification payload schema in AsyncAPI +alongside the bus event that triggers a refetch. It looks cleaner +until you realize the payload **never travels on the bus wire** — +the bus event is `NotificationReceived` with empty `data: {}`, a +pure cache-invalidation poke. The FE fetches the payload from +`GET /api/notifications`, which is REST → OpenAPI's territory. +Putting the payload schema in AsyncAPI would mean "documenting +this shape on a transport it doesn't travel on" — a conceptual +stretch that adds a drift risk for zero gain. + +### Rejected alternatives + +- **Dual-spec (same type declared in both AsyncAPI + OpenAPI).** + Guaranteed drift unless both come from a single codegen. Nothing + in the tooling today produces both, so we'd hand-maintain two + copies of every shared type. Bug factory. +- **Cross-spec `$ref`** — AsyncAPI 3.0 allows `"$ref": + "openapi.json#/components/schemas/Foo"`, and utoipa's + `components(schemas(...))` can publish orphan types (no + `#[utoipa::path]` reference) so OpenAPI advertises "internal" + schemas. Technically workable but: Modelina + Swagger UI + Redoc + handle external refs inconsistently, OpenAPI stops being "the + REST contract" and becomes "a general schema registry", + reviewers get confused. Legal, fragile, avoided. +- **Bus event carries the full payload** (revert the pure-poke + design). Would put per-kind payload schemas in AsyncAPI as + `MessageBusEvent::NotificationReceived { granter_id, resource_id, + … }`. Rejected because the FE has to REST-fetch anyway (bell + reads from DB for history + persistence), so the fields on the + wire are dead weight — same-content overlap between the two + specs, no consumer benefit. +- **Session-resume tokens** (`rt.subscribe { since: N }` + + server-side ring buffer). Would let the bus deliver missed rows + directly on reconnect, saving one REST round-trip. Rejected for + **backward compatibility with `OXICLOUD_MESSAGEBUS_ENABLE=false`**: + ops who disable the WS rely on the bell falling back to REST; + bus-only replay would leave those deployments with no catch-up + path. The REST `?after=` cursor works in every mode (bus on, bus + off, network gap); the bus stays purely "instant-poke". + +### What this looks like in the tree + +- `src/application/ports/message_bus_ports.rs` — `MessageBusEvent` + enum (Rust source of truth for bus wire shapes). +- `src/bin/generate-asyncapi.rs` — projects those Rust variants + into `resources/gen/asyncapi.json`. +- `src/domain/entities/notification.rs` — `SharegrantedPayload` + and its siblings, `#[derive(ToSchema)]`, source of truth for + REST payload shapes. +- `src/interfaces/api/mod.rs` — utoipa `#[openapi(components(schemas(SharegrantedPayload, ...)))]` + registers the payload in OpenAPI even though `NotificationDto.payload` + stays `serde_json::Value` on the response type. (FE type-narrows + on `row.kind` and casts to the right shape.) +- Nothing lives in both specs today. + +### Adding a new bus event + +1. Add a variant to `MessageBusEvent`. +2. Add the variant to `generate-asyncapi.rs`'s `event_kind` enum + and (if the variant has payload fields) a schema function. +3. Regenerate AsyncAPI + FE DTOs via `just asyncapi` + + `npm run gen:message-bus`. +4. **Do not** add the variant to OpenAPI. Bus events don't + travel on REST. + +### Adding a new notification kind's payload + +1. Add a Rust struct in `domain/entities/notification.rs` with + `#[derive(Serialize, Deserialize, ToSchema)]`. +2. Register it in `src/interfaces/api/mod.rs`'s + `components(schemas(...))` list. +3. Regenerate OpenAPI via `just openapi`. +4. **Do not** add the struct to AsyncAPI. Notification payloads + only cross the REST wire. + +--- + +## Message bus + +### Architecture + +``` +┌─────────────────────────────────────────────────────────────┐ +│ SERVICE LAYER │ +│ │ +│ ShareService.grant() ── after commit ──▶ bus.publish(…) │ +│ FileMgmtService.…() ── after commit ──▶ bus.publish(…) │ +│ NotificationService ── after commit ──▶ bus.publish(…) │ +│ Scheduler engine ── on run start/end ──▶ bus.publish │ +└──────────────────────────┬──────────────────────────────────┘ + ▼ +┌─────────────────────────────────────────────────────────────┐ +│ MessageBus port (application/ports/message_bus_ports.rs) │ +│ │ +│ InProcessMessageBus │ +│ DashMap> │ +└──────────────────────────┬──────────────────────────────────┘ + │ + │ (optional replicator seam) + ▼ + NoopReplicator (v1) + PgListenReplicator (deferred) + BrokerReplicator (RabbitMQ / NATS, deferred) + │ + ▼ +┌─────────────────────────────────────────────────────────────┐ +│ WS handler (interfaces/api/handlers/rt_ws.rs) │ +│ │ +│ One session per socket: │ +│ HashMap + outbound mpsc │ +│ │ +│ rt.subscribe / rt.unsubscribe frames │ +│ rt.event / rt.revoked / rt.pong notifications │ +└─────────────────────────────────────────────────────────────┘ +``` + +### Topics — a typed enum, not a string + +```rust +pub enum Topic { + Folder(Uuid), // "folder:{uuid}" + UserAuthz(Uuid), // "user:{uuid}:authz" + UserNotifications(Uuid), // "user:{uuid}:notifications" + Job(String), // "job:{name}" +} +``` + +Defined in `application/ports/message_bus_ports.rs`. Encoded to a +stable dotted wire form; parsed back with strict validation. The +wire form doubles as a routing key for future broker replicators +(RabbitMQ topic exchanges, NATS subjects). + +### Events + +`MessageBusEvent` (same module) is the discriminated union of every +payload a publisher can produce — `FileCreated`, `FolderMoved`, +`AuthzChanged`, `JobRunStarted / Progress / Ended`, +`NotificationReceived`, etc. Serde tags with `#[serde(tag = "event", +rename_all = "snake_case")]`, so the wire is +`{"event": "file_created", "file_id": "...", "actor": "..."}`. + +Payloads are **thin facts**: the ID of the changed resource + the +actor + the verb. Clients refetch details via REST if they need +them. Keeps the AuthZ surface small (thin payloads can't leak +fields the caller couldn't already read via REST) and keeps events +well under any future broker's message-size cap. + +### Wire protocol + +JSON-RPC 2.0 over text frames. Full protocol in [`docs/plan/message-bus.md § Wire protocol`](https://github.com/oxicloud/oxicloud/blob/main/docs/plan/message-bus.md). + +```jsonc +// Client → server +{"jsonrpc":"2.0","id":1,"method":"rt.subscribe","params":{"topic":"folder:abc-…"}} + +// Server → client (ack) +{"jsonrpc":"2.0","id":1,"result":{"subscribed":"folder:abc-…"}} + +// Server → client (push, id-less notification) +{"jsonrpc":"2.0","method":"rt.event","params":{ + "topic":"folder:abc-…", + "event":"file_created", + "data":{"file_id":"…","name":"…","parent_id":"…","actor":"…"} +}} +``` + +### Authentication for the WebSocket upgrade + +Two paths, both accepted by the same handler: + +| Client kind | Path | Why | +|---|---|---| +| **Programmatic** (CLI, test helper) | `Authorization: Bearer ` on the upgrade | The `new WebSocket()` API in browsers can attach `Sec-WebSocket-Protocol` but NOT arbitrary headers, so browsers can't do this. | +| **Browser** | `POST /api/rt/ticket` (with the full DPoP + CSRF middleware chain) mints a one-shot 30-second opaque UUID; the browser opens the WS with `Sec-WebSocket-Protocol: oxi.ticket.` | DPoP-bound sessions cannot attach a `DPoP:` header to `new WebSocket()`. The ticket flow moves the DPoP check to a normal POST that DOES support headers, and the WS upgrade just redeems the opaque token. | + +Tickets are single-use, TTL 30 s, stored in a `RtTicketStore` +(in-memory). Redemption removes the entry — replay is impossible. + +The WS route is deliberately mounted **outside** the +`protected_api` middleware stack — otherwise the DPoP-required +layer would 401 every browser on upgrade before the ticket flow +could kick in. + +--- + +## The three AuthZ scopes + +`Topic::required_perm(&self) -> AuthzCheck` dispatches every +subscribe attempt into exactly one of three classes. This is the +authoritative diagram of what the WS handler enforces: + +``` + ┌──────────────────────────────────────────┐ + │ Topic::required_perm() │ + └─────┬─────────────┬─────────────┬────────┘ + │ │ │ + ResourceRead │ IdentityMatch │ RoleAdmin + ▼ ▼ ▼ + ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ + │ Class 1 │ │ Class 2 │ │ Class 3 │ + │ per-resource│ │ per-user │ │ per-session │ + │ │ │ │ │ (role) │ + └─────────────┘ └─────────────┘ └─────────────┘ +``` + +### Class 1 — Per-resource (`AuthzCheck::ResourceRead`) + +**Topics:** `folder:{id}`, and (future) `file:{id}`, +`drive:{id}`, `calendar:{id}`, `addressbook:{id}`. + +**Rule:** the caller must hold **`Read`** on the resource via the +same `AuthorizationEngine` that guards every REST endpoint. Owner +short-circuits pass; direct grants pass; group-mediated grants +pass; drive-membership cascades pass. Everything else is denied. + +**Wire response on denial:** JSON-RPC error object with +`code = -32001`, `message = "no_read"`. Same shape whether the +resource doesn't exist OR the caller lacks the grant — **anti- +enumeration invariant**. Audit reason (`no_read` / +`no_such_resource`) distinguishes internally. + +**On grant revocation:** the WS handler auto-subscribes each +session to `user:{caller}:authz` (Class 2 below). When a +`MessageBusEvent::AuthzChanged { affected_folders }` fires, the +session's reader translates it to an internal `EvictFolders` +signal → the main loop walks the sub set and drops any Class-1 +subscription whose resource was affected, emitting a client-visible +`rt.revoked` notification per evicted topic. Same pattern applies +to any Class-1 topic when the AuthZ model widens beyond folders. + +### Class 2 — Per-user, strict privacy (`AuthzCheck::IdentityMatch`) + +**Topics:** `user:{u}:authz`, `user:{u}:notifications`. + +**Rule:** direct UUID equality — `caller_id == u`. **No admin +bypass, no group indirection, no owner short-circuit.** Admins +cannot subscribe to other users' `:authz` or `:notifications` +streams; that's a privacy invariant, not a mere policy choice. + +**Wire response on mismatch:** `topic_forbidden` — the **same +wire shape as an unknown topic**. An attacker probing +`user:{someone_else_uuid}:authz` cannot distinguish "user exists +but not me" from "no such user." + +**Auto-subscription:** the WS handler auto-subscribes every session +to its own `user:{caller}:authz` AND `user:{caller}:notifications` +at session open. No `rt.subscribe` frame is needed from the client +for these — they're always active for the caller's own UUID. + +### Class 3 — Per-session role (`AuthzCheck::RoleAdmin`) + +**Topics:** `job:{name}` today. Future `admin:*` topics land here. + +**Rule:** the session's snapshotted role at open time must be +`admin`. The handler resolves `caller_role` once during session +setup via `resolve_live_role` and stores it on the session state — +no per-subscribe DB round-trip. + +**Wire response on non-admin:** `topic_forbidden` — same anti-enum +shape as Class 2. A non-admin probing job topics cannot enumerate +which jobs are registered. + +**Why snapshot at session open, not per subscribe:** admin role +loss is rare + trivially recoverable (the user closes the tab and +reopens, hitting the fresh role check). Per-subscribe checks would +be an extra DB round-trip on every frame with no meaningful +security gain — the WS session itself was authenticated at upgrade +time under the current role. + +### Adding a new topic + +Every new topic variant must decide which class it belongs to at +`Topic::required_perm`. The compiler enforces exhaustiveness — a +new variant with no branch fails to build, which is deliberate. New +topics get audited before shipping precisely because the +`required_perm` match forces the author to state the class +explicitly. + +--- + +## Tab-visibility grace-close — reducing idle connections + +Every open browser tab holds one WebSocket to the server. A user +with five tabs open holds five sockets. A user who leaves a tab +open all day but only uses one holds five sockets, four of them +serving nothing. + +The frontend closes the WebSocket **after 60 seconds of tab +hidden** and reopens it when the tab becomes visible again. The +subscription state is preserved locally through the outage — every +subscriber's release handle stays live, the reactive store still +holds the last-known list — but the wire is silent while the tab +is hidden. + +### Implementation + +Frontend `MessageBusClient` (`frontend/src/lib/message-bus/client.svelte.ts`) +attaches a `visibilitychange` listener on construction: + +- **Tab hidden** → starts a 60-second timer. +- **Timer fires while still hidden** → close the WS via a + `#closeForHidden` path that sets state to `disconnected` but + preserves `#subs` for later replay. The `#onClose` handler is + guarded by `#tabIsHidden()` — an auto-reconnect won't fire while + the tab remains hidden. +- **Tab visible again** → cancels the timer if it hadn't fired + yet; if the WS was closed, kicks off a normal reconnect. +- **On reopen**, the WS handler auto-subscribes to `:authz` and + `:notifications` again, and the client replays every entry in + `#subs` as `rt.subscribe` frames. From the user's POV, the state + is identical to what they left behind. + +### What this trades + +- **Saved**: N-1 idle sockets per user with N tabs open, over the + hidden-tab window. Meaningful at scale (100 users × 5 tabs × 8 + idle hours = 4000 idle-tab-hours of connection state to keep + alive per day). +- **Lost**: bus events published during the 60-second delay + (transient) + the whole grace-close window (indefinite while + hidden) are dropped for that session. **Recovery**: on reopen, + every consumer that cares refetches. See "Reconnect catch-up" + below. + +### Why 60 seconds + +Short enough that leaving a tab for a coffee break doesn't burn +the connection. Long enough that the momentary focus-shifts +users do all day (Cmd-Tab to another app, back within seconds) +don't churn the socket. Not tunable per-user — the value is +hard-coded in `HIDDEN_GRACE_MS`. + +### Reconnect catch-up — the `?after=` cursor + +For consumers whose state can't be reconstructed by refetching a +current listing (specifically: **notifications**, whose bell must +show rows that landed during the outage), the FE issues a delta +fetch: + +- Store tracks `#lastReceivedAt` — the newest `created_at` seen + before disconnect. +- On `messageBus.onReconnect(...)` fire, calls + `GET /api/notifications?after=&limit=100`. +- Merges the returned rows into local state via `mergeById` — + duplicates are resolved with **incoming wins** (server value + overrides local, so a `read_at` flip on another device shows + up correctly). + +The server-side `?after=` predicate is **strict `>`** — a row at +exactly `lastReceivedAt` is excluded. This makes the WS push (which +delivers a row at time T) and the delta fetch (which asks for +"anything after T") non-overlapping by construction. `mergeById` +handles the case where the two paths race and both deliver the +same row. + +For consumers whose state IS a current listing (folder view: the +files/subfolders in a folder), reconnect just refetches the listing +via the normal REST endpoint. `useReconnect` composable exposes +`onReconnect(cb)` as a one-liner for that pattern. + +--- + +## Persistent notifications (the bell) + +### Data model + +```sql +notif.notifications ( + id UUID PRIMARY KEY, + user_id UUID NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE, + kind TEXT NOT NULL, -- 'share_granted' | 'new_login_from_new_device' | … + payload JSONB NOT NULL, -- per-kind shape (see below) + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + read_at TIMESTAMPTZ -- NULL = unread +) +``` + +Two indexes: + +- `(user_id, created_at DESC) INCLUDE (read_at, kind)` covers the + bell's list query + the mark-all-read filter. +- `(read_at) WHERE read_at IS NOT NULL` — partial, tiny on healthy + DBs; feeds the retention job's DELETE. + +### Ingesters + +An ingester is a code path that calls +`NotificationApplicationService::create(NewNotification)`. The +service atomically: + +1. `INSERT INTO notif.notifications RETURNING …` — durable row. +2. `bus.publish(Topic::UserNotifications(user_id), NotificationReceived)` — the fast-path poke. + +Publish happens **after** the DB write succeeds, never inside a +transaction — a rolled-back INSERT would otherwise fan out a lie. + +**Currently shipped ingester:** `share_granted` in +`interfaces/api/handlers/grant_handler.rs::create_grant`. Fires +after `authz.set_role(...)` succeeds. Fans out to every resolved +recipient user: + +- `Subject::User(id)` → one row for that user. +- `Subject::Group(id)` → one row per transitive member (via + `SubjectGroupService::list_transitive_users`). +- `Subject::Token(_)` → no row (anonymous share links have no + target user). + +Self-shares (owner grants themselves via a group they belong to) +skip. Failure to write is best-effort — a warn log; the grant row +stays durable in `role_grants`, the recipient can still discover +the share via `/api/grants/shared-with-me`. + +**Planned but not wired** (each needs a prerequisite subsystem +listed in [`docs/plan/message-bus.md § Deferred`](https://github.com/oxicloud/oxicloud/blob/main/docs/plan/message-bus.md)): + +| Kind | Prerequisite | +|---|---| +| `new_login_from_new_device` | Device-fingerprint tracking table | +| `job_completed_for_you` | Scheduler engine threading the trigger caller's `caller_id` through `dispatch()` | +| `storage_quota_threshold` | Per-user usage/quota comparator with threshold-crossing detection | + +### Payload shape — typed per kind + +The `payload` column is JSONB (schema-free at the DB layer). Each +kind's Rust shape lives in `domain/entities/notification.rs` with +`#[derive(Serialize, Deserialize, ToSchema)]`. OpenAPI picks up the +struct automatically. Adding a new field is additive on JSONB — no +migration. + +Example — `share_granted`: + +```rust +pub struct SharegrantedPayload { + pub granter_id: Uuid, + pub resource_type: String, // 'folder' | 'file' | 'drive' | … + pub resource_id: Uuid, + pub resource_name: Option, // snapshot at grant time + pub resource_path: Option, // storage path snapshot + pub navigate_folder_id: Option, // FE routing target for drives + pub role: String, + pub expires_at: Option>, +} +``` + +The name/path fields are **snapshotted at grant time**. If the +folder is later renamed or moved, the notification still reflects +what it was called when the share happened. Same principle as +email invitations or activity feeds: the record is a fact about +what was true at the moment, not a live pointer. + +### Schema-ownership rule — AsyncAPI vs OpenAPI + +> **AsyncAPI defines the envelope + transport for clients. +> OpenAPI defines the payload.** + +The bus event `MessageBusEvent::NotificationReceived` is a **unit +variant** — it serializes to `{"event":"notification_received", +"data":{}}` with no fields on the wire. The topic identifies the +semantic; the FE responds by refetching from REST. + +The payload's shape lives in OpenAPI via `ToSchema` on +`SharegrantedPayload` (and its future siblings), auto-derived from +Rust. AsyncAPI never sees these types — payloads don't travel on +the bus wire. + +**Why this split** — it eliminates schema drift between the two +specs. A payload edit changes Rust → OpenAPI updates on regen +(mechanical). AsyncAPI stays stable (hand-written, but it never +touches payloads). Same rule applies to any future bus consumer +that also has a REST DTO — Rust is the source of truth; each spec +projects the parts of Rust that travel on its transport. + +Design rationale + rejected alternatives (dual-spec, cross-`$ref`, +per-kind Svelte components) in [`docs/plan/templated-messages.md § Schema ownership`](https://github.com/oxicloud/oxicloud/blob/main/docs/plan/templated-messages.md). + +### REST surface — `/api/notifications` + +| Method | Path | Purpose | +|---|---|---| +| `GET` | `/api/notifications` | List newest-first. Query params: `unread` (bool), `before` / `after` (cursors), `limit` | +| `GET` | `/api/notifications/unread` | Badge-only fast path (returns just `unread_count`) | +| `POST` | `/api/notifications/{id}/read` | Mark one row read (idempotent, always 204) | +| `POST` | `/api/notifications/read-all` | Bulk mark-all-read (returns rows updated) | +| `DELETE` | `/api/notifications/{id}` | Hard-delete one row (idempotent, always 204) | + +**Anti-enumeration:** mark-read and delete always respond 204 — +whether the row existed and belonged to the caller, or didn't +exist at all, or belonged to someone else. Every mutating endpoint +scopes on `caller_id` at the SQL layer; the response shape is +identical across the three outcomes. + +### Retention + +The `notifications_cleanup` scheduled job (daily, same tier as +`trash_cleanup`) DELETEs read rows older than +`OXICLOUD_NOTIFICATIONS_RETENTION_DAYS` (default 30). **Unread +rows are preserved unconditionally** — a user offline for a month +still sees the share-granted notice when they log back in. + +Runtime override via the job's `retention_days` parameter on the +admin panel's trigger — the env default seeds it, the panel +overrides at trigger time. + +### Frontend rendering + +`frontend/src/lib/composables/useNotifications.svelte.ts` owns the +module-scoped store — one instance per SPA session. Exposes: + +- `notifications.items` — reactive list (newest first) +- `notifications.unread` — reactive badge count +- `notifications.refresh()` / `refreshDelta()` / `markRead(id)` / + `markAllRead()` / `delete(id)` + +`NotificationRow.svelte` handles the actual rendering — one file, +one `switch` on `row.kind`, one rich template per shipped kind +(`share_granted` today; the others fall back to a generic string). +Extraction into per-kind components is deferred until a single +kind's block exceeds ~30 lines or two kinds start needing the same +sub-component (see [`docs/plan/templated-messages.md § Rendering`](https://github.com/oxicloud/oxicloud/blob/main/docs/plan/templated-messages.md)). + +### Notification click routing + +| `resource_type` | Route | Data used | +|---|---|---| +| `folder` | `/files/{resource_id}` | `resource_id` | +| `file` | `/shared-with-me?file={resource_id}` | `resource_id` — the `/files/{uuid}` route requires a **folder** id, and a file-scoped grant may not include parent-folder access. `/shared-with-me` is the guaranteed-accessible home for every recipient of a `share_granted`, and its `?file=` deep link opens the inline `FileViewer`. | +| `drive` | `/files/{navigate_folder_id}` | Drives have no browsable URL of their own; `navigate_folder_id` is the drive's `root_folder_id`, enriched at ingest via `DriveRepository::get_by_id`. | +| `calendar` / `address_book` / `playlist` | no link (bold text) | Not addressable via `/files/*`. | + +The bell also fires a **transient toast** (via the existing +`ui.notify(...)` mechanism) on every fresh row that arrives via +delta — the toast fades out after ~4 s while the persistent row +stays in the bell's history section. Same bell icon, same badge +count, no duplicate UX. + +--- + +## Feature flags & config + +| Env var | Default | Effect | +|---|---|---| +| `OXICLOUD_MESSAGEBUS_ENABLE` | `true` | Master switch. When `false`, the `/api/rt/ws` and `/api/rt/ticket` routes are **not registered** at boot (Axum returns 404), and the FE `useNotifications` composable skips the WS setup entirely. The bell falls back to REST-only mode — polling on mount, delta on manual refresh. Zero client-side error spam. | +| `OXICLOUD_MESSAGEBUS_KEEPALIVE_SECONDS` | `30` | Server-initiated RFC 6455 Ping interval on each WS connection. Prevents intermediate proxies (nginx, Traefik, Cloudflare) from reaping the TCP session as idle. | +| `OXICLOUD_NOTIFICATIONS_RETENTION_DAYS` | `30` | Retention window for read notifications. Unread rows are always preserved. The `notifications_cleanup` job clamps to a minimum of 1 day. | + +Client discovers all of these via `GET /api/config` — no in-band +"does the server support the bus?" probe needed. The FE reads +`serverConfig.features.message_bus` at boot and skips WS setup +entirely when it's false. + +--- + +## Failure modes + +| Scenario | Behavior | +|---|---| +| Bus is disabled server-side (`OXICLOUD_MESSAGEBUS_ENABLE=false`) | `/api/rt/ws` returns 404. `useTopic` in the FE returns early. Bell works via REST only. | +| Network drops mid-session | Client-side jittered exponential backoff (250 ms → 30 s cap, 20-failure circuit breaker). On reconnect, WS handler re-auto-subscribes to `:authz` + `:notifications`; `useReconnect` composable fires `onReconnect` callbacks so views refetch. | +| Server restart | Same as network drop — the WS breaks, client backs off, reconnects when server is back. Events published during the outage are lost (no persistent event log by design); consumers refetch. | +| Tab hidden > 60 s | WS closed via `visibilitychange` grace-close. State preserved locally. On visibility return, reconnect + replay subscriptions. | +| Publish before commit | Not allowed. Every publish site is documented as "after commit". A publish inside a transaction that rolls back would fan out a lie. | +| Broker replicator failure (future) | The local `InProcessMessageBus` publishes still succeed — the replicator is beside the bus, not in front. Broker-hop failures affect multi-instance fanout but never local delivery. | + +--- + +## Testing + +The api-test suite exercises the full stack end-to-end via `rt-hurl-helper` (a small Rust binary gated on `test_utils`) — Hurl alone can't drive a WebSocket. Sixteen scenarios in `tests/api/rt_bus_check.sh`: + +- Positive delivery, topic isolation +- AuthZ denial (Class 1 folder), unknown-topic anti-enum +- Server keepalive, delete emits, move fan-out +- Grant-revoke eviction (`AuthzChanged` → `rt.revoked`) +- Cross-user identity gate (Class 2) +- Ticket happy path + single-use replay refused +- Admin-only job topic (Class 3) +- `notification_received` wire push, DB row via `GET /api/notifications` +- Cross-user notifications identity gate (Class 2, notifications topic) +- `?after=` cursor with strict-`>` boundary invariant + +`mergeById` — the FE's WS/reconnect race dedup — has its own +Vitest with 5 covered cases (empty, non-overlap, exact-dup, +stale-read-at overwrite, mixed overlap). + +--- + +## Further reading + +- Plan doc: [`docs/plan/message-bus.md`](https://github.com/oxicloud/oxicloud/blob/main/docs/plan/message-bus.md) — full design rationale, roadmap, and deferred slices (Yjs collab, broker replicator, SharedWorker, Web Push). +- Plan doc: [`docs/plan/templated-messages.md`](https://github.com/oxicloud/oxicloud/blob/main/docs/plan/templated-messages.md) — schema-ownership rule, rendering shape, notification routing decision. +- [ReBAC Authorization](/architecture/rebac-authorization) — the engine every Class-1 topic gate calls. +- [Background jobs](/architecture/jobs) — `notifications_cleanup` is one of them; `Topic::Job` publishes on job lifecycle.