Oxicloud/docs/architecture/message-bus-and-notifications.md

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

• 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

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.

// 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

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):

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:

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.

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).

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 — full design rationale, roadmap, and deferred slices (Yjs collab, broker replicator, SharedWorker, Web Push).
• Plan doc: docs/plan/templated-messages.md — schema-ownership rule, rendering shape, notification routing decision.
• ReBAC Authorization — the engine every Class-1 topic gate calls.
• Background jobs — notifications_cleanup is one of them; Topic::Job publishes on job lifecycle.