docs(message-bus + notifications): add arch doc
This commit is contained in:
@@ -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" },
|
||||
],
|
||||
},
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<Topic, tokio::broadcast::Sender<Event>> │
|
||||
└──────────────────────────┬──────────────────────────────────┘
|
||||
│
|
||||
│ (optional replicator seam)
|
||||
▼
|
||||
NoopReplicator (v1)
|
||||
PgListenReplicator (deferred)
|
||||
BrokerReplicator (RabbitMQ / NATS, deferred)
|
||||
│
|
||||
▼
|
||||
┌─────────────────────────────────────────────────────────────┐
|
||||
│ WS handler (interfaces/api/handlers/rt_ws.rs) │
|
||||
│ │
|
||||
│ One session per socket: │
|
||||
│ HashMap<wire_key, Sub> + 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 <jwt>` 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.<uuid>` | 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=<lastReceivedAt>&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<String>, // snapshot at grant time
|
||||
pub resource_path: Option<String>, // storage path snapshot
|
||||
pub navigate_folder_id: Option<Uuid>, // FE routing target for drives
|
||||
pub role: String,
|
||||
pub expires_at: Option<DateTime<Utc>>,
|
||||
}
|
||||
```
|
||||
|
||||
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.
|
||||
Reference in New Issue
Block a user