From 1b824cb45cb708b8fff9248210210a41108e2b5e Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Thu, 10 Sep 2026 00:02:30 +0200 Subject: [PATCH] feat(msg-bus): prepare engine --- Cargo.lock | 72 +- Cargo.toml | 2 +- docs/plan/message-bus.md | 1053 +++++++++++++++++ src/application/ports/mod.rs | 1 + src/application/ports/realtime_ports.rs | 493 ++++++++ .../services/file_upload_service.rs | 42 + src/application/services/folder_service.rs | 48 + src/common/di.rs | 45 +- .../services/in_process_realtime_bus.rs | 347 ++++++ src/infrastructure/services/mod.rs | 1 + src/interfaces/api/handlers/mod.rs | 1 + src/interfaces/api/handlers/rt_ws.rs | 552 +++++++++ src/interfaces/api/routes.rs | 10 + 13 files changed, 2650 insertions(+), 17 deletions(-) create mode 100644 docs/plan/message-bus.md create mode 100644 src/application/ports/realtime_ports.rs create mode 100644 src/infrastructure/services/in_process_realtime_bus.rs create mode 100644 src/interfaces/api/handlers/rt_ws.rs diff --git a/Cargo.lock b/Cargo.lock index 1dd449b5..870a7493 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -166,7 +166,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -177,7 +177,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -765,12 +765,13 @@ dependencies = [ [[package]] name = "axum" -version = "0.8.9" +version = "0.8.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +checksum = "8b52af3cb4058c895d37317bb27508dccc8e5f2d39454016b297bf4a400597b8" dependencies = [ "axum-core", "axum-macros", + "base64 0.22.1", "bytes", "form_urlencoded", "futures-util", @@ -790,8 +791,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1 0.10.6", "sync_wrapper", "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", @@ -1998,6 +2001,12 @@ dependencies = [ "parking_lot_core", ] +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "datasketches" version = "0.2.0" @@ -2289,7 +2298,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3320,7 +3329,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.4", + "socket2 0.5.10", "tokio", "tower-service", "tracing", @@ -3671,7 +3680,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4419,7 +4428,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -5265,7 +5274,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls 0.23.40", - "socket2 0.6.4", + "socket2 0.5.10", "thiserror 2.0.18", "tokio", "tracing", @@ -5302,7 +5311,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.4", + "socket2 0.5.10", "tracing", "windows-sys 0.60.2", ] @@ -5820,7 +5829,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6343,7 +6352,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -6855,7 +6864,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7086,6 +7095,18 @@ dependencies = [ "tokio-util", ] +[[package]] +name = "tokio-tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -7363,6 +7384,23 @@ version = "0.25.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d2df906b07856748fa3f6e0ad0cbaa047052d4a7dd609e231c4f72cee8c36f31" +[[package]] +name = "tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +dependencies = [ + "bytes", + "data-encoding", + "http 1.4.0", + "httparse", + "log", + "rand 0.9.4", + "sha1 0.10.6", + "thiserror 2.0.18", + "utf-8", +] + [[package]] name = "type1-encoding-parser" version = "0.1.1" @@ -7539,6 +7577,12 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8-ranges" version = "1.0.5" @@ -8323,7 +8367,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index aeb0fa2d..250ec8a5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,7 +25,7 @@ bin-dir = "oxicloud-{ version }-{ target }/{ bin }{ binary-ext }" [dependencies] mimalloc = { version = "0.1.52", default-features = false } -axum = { version = "0.8.9", features = ["multipart", "http1", "http2", "tokio", "macros"] } +axum = { version = "0.8.8", features = ["multipart", "http1", "http2", "tokio", "macros", "ws"] } # "process" was previously enabled implicitly through aws-config's feature # unification; ffmpeg_video_frame_service needs it, so declare it ourselves. tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "io-util", "net", "time", "sync", "fs", "process", "signal"] } diff --git a/docs/plan/message-bus.md b/docs/plan/message-bus.md new file mode 100644 index 00000000..56ed4845 --- /dev/null +++ b/docs/plan/message-bus.md @@ -0,0 +1,1053 @@ +# Plan — Realtime message bus over WebSocket + +## Context + +OxiCloud today has no server-push channel. Every "live-ish" surface +(folder listing, job dashboard, share dialog, admin session count) is +either stale-until-refresh or polled by the SPA. That leaves a whole +category of features unreachable — collab editing, presence, +notifications, sync-client push invalidation — and it makes existing +surfaces feel dated compared to Google Drive, Notion, Nextcloud, and +M365. + +This plan introduces a single realtime bus over WebSocket that any +service can publish facts to and any client can subscribe to. Collab +editing is one consumer on top; folder-live updates, notifications, +job progress, presence, and sync-client push invalidation follow with +almost no extra scaffolding. + +## Non-goals + +- Persistent event log with "you missed these" replay. Durable state + lives in real tables (`notifications`, `collab.doc_sessions`, …); + the bus is a live-delivery optimization, always best-effort. +- Chat / DM / voice / video / screen share. Explicitly out of scope + for OxiCloud — that's Nextcloud Talk territory, not a file-server + job. +- Wildcard subscriptions (`folder:*`). Breaks per-subscribe AuthZ and + makes revocation semantics fuzzy. +- Cross-user subscriptions. Privacy + AuthZ risk. Admins subscribe to + `admin:*` topics, never to another user's private feed. + +## Architecture — 3 layers, clean seams + +``` +┌──────────────────────────────────────────────────────────────────────┐ +│ SERVICE LAYER │ +│ │ +│ FileMgmtService.create_file() ── after commit ──▶ bus.publish(...) │ +│ ShareService.grant() ── after commit ──▶ bus.publish(...) │ +│ JobRegistry step progress ─────────────────▶ bus.publish(...) │ +│ CollabSessionService.apply() ─────────────────▶ bus.publish(...) │ +│ │ +└───────────────────────────────┬──────────────────────────────────────┘ + │ publish(&Topic, RealtimeEvent) + ▼ +┌──────────────────────────────────────────────────────────────────────┐ +│ REALTIME BUS (RealtimeBus trait — application/ports) │ +│ │ +│ InProcessRealtimeBus (v1) │ +│ DashMap> │ +│ │ +└──────┬───────────────────────────────────────────────────────────────┘ + │ + │ ┌──────────────────────────────────────────────────────┐ + │ │ REPLICATOR (optional, feature-flagged) │ + │ │ │ + │ │ BusReplicator trait ── separate port │ + │ │ - v1: NoopReplicator (single-instance) │ + │ │ - v2: PgListenReplicator (pg_notify) │ + │ │ - v3: BrokerReplicator (RabbitMQ / NATS) │ + │ │ │ + │ │ Sits BESIDE InProcessRealtimeBus, forwards │ + │ │ local publishes outbound + inbound events │ + │ │ from the broker back into local publish. │ + │ └──────────────────────────────────────────────────────┘ + ▼ +┌──────────────────────────────────────────────────────────────────────┐ +│ WS HANDLER (interfaces/api/handlers/rt_ws.rs) │ +│ │ +│ One RealtimeSession per WS: HashSet + outbound mpsc │ +│ - subscribe/unsubscribe frames → bus.subscribe(topic) │ +│ - each subscribed stream drains into the outbound mpsc │ +│ - AuthZ at subscribe (once), evict on grant-revoked │ +│ │ +└──────────────────────────────────────────────────────────────────────┘ +``` + +**The seam that keeps RabbitMQ/NATS doors open is the replicator, not +the bus.** Services and the WS handler only ever see the local +`RealtimeBus`. A future `BrokerReplicator` publishes outbound + injects +inbound. Zero touch to callers. + +## Backend components + +### 1. Port + event types (`application/ports/realtime_ports.rs`) + +```rust +// Topic is a typed enum, not a string. Prevents typos, gives +// exhaustive matching for the AuthZ gate, encodes stably to +// wire keys for any broker (RabbitMQ topic exchange, NATS subject). +pub enum Topic { + Folder(FolderId), + File(FileId), + Drive(DriveId), + UserNotifications(UserId), + UserAuthz(UserId), + UserSessions(UserId), + UserUploads(UserId), + Job(JobId), + Collab(FileId), + CollabAwareness(FileId), + FolderPresence(FolderId), + FilePresence(FileId), + FileComments(FileId), + Calendar(CalendarId), + AddressBook(AddressBookId), + AdminSessions, // admin-only + AdminAudit, // admin-only, sampled +} + +impl Topic { + pub fn to_wire_key(&self) -> String; // stable dotted form + pub fn parse(s: &str) -> Result; + pub fn required_perm(&self) -> AuthzCheck; // used by the AuthZ gate +} + +/// Wire mirror of `domain::services::authorization::Subject`. +/// Kept as its own type so the bus payload schema can evolve +/// independently of the domain enum. +#[derive(Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum PrincipalRef { + User { id: UserId }, + Group { id: GroupId }, + Token { id: TokenId }, // anonymous public share link +} + +#[derive(Serialize, Deserialize)] +#[serde(tag = "event", rename_all = "snake_case")] +pub enum RealtimeEvent { + // Folder / File verbs — thin facts only, client refetches details. + FileCreated { file_id: FileId, name: String, parent_id: FolderId, actor: UserId }, + FileDeleted { file_id: FileId, parent_id: FolderId, actor: UserId }, + FileRenamed { file_id: FileId, old_name: String, new_name: String, actor: UserId }, + FileMoved { file_id: FileId, from: FolderId, to: FolderId, actor: UserId }, + FolderCreated { folder_id: FolderId, parent_id: FolderId, actor: UserId }, + // Jobs + JobStep { job_id: JobId, step: u32, message: String }, + JobFinished { job_id: JobId, outcome: JobOutcome }, + // Notifications + Notification { notification_id: NotifId, kind: NotifKind }, + // Sharing — principal is any Subject (user, group, or public-link token). + // `affected_users` is populated by the producer only for the group case, + // so consumers of `file:{id}:shares` don't need to expand membership. + ShareGranted { + file_id: FileId, + principal: PrincipalRef, + role: GrantRole, + affected_users: Option>, + }, + ShareRevoked { + file_id: FileId, + principal: PrincipalRef, + affected_users: Option>, + }, + // Group membership changes — cascade grants to/from the affected user. + GroupMemberAdded { group_id: GroupId, user_id: UserId, actor: UserId }, + GroupMemberRemoved { group_id: GroupId, user_id: UserId, actor: UserId }, + // AuthZ eviction / re-evaluation signal + AuthzChanged { affected: Vec }, + // Presence (Phase B) + PresenceJoined { user_id: UserId, display_name: String, color: String }, + PresenceLeft { user_id: UserId }, + PresenceCursor { user_id: UserId, position: PresencePosition }, + // Collab (binary bytes on the wire, kept opaque in the enum) + CrdtUpdate { bytes: Bytes }, + // ... one variant per verb; enum > strings per project convention. +} + +#[async_trait] +pub trait RealtimeBus: Send + Sync { + /// Fire-and-forget. SYNC (not async) — services must not await + /// under a DB transaction. + fn publish(&self, topic: &Topic, event: RealtimeEvent); + + /// Returns a Stream so the impl can change (broadcast, mpsc, + /// pg listener) without churn. + fn subscribe(&self, topic: &Topic) -> Pin + Send>>; +} + +/// Kept SEPARATE from RealtimeBus so v2/v3 wiring is drop-in. +#[async_trait] +pub trait BusReplicator: Send + Sync { + /// Called whenever the local bus publishes; may forward to broker. + fn on_local_publish(&self, topic: &Topic, event: &RealtimeEvent); + + /// Long-running consumer task: reads remote messages and + /// re-publishes locally. Started by DI, returns on shutdown. + async fn run(self: Arc, shutdown: CancellationToken) -> Result<(), BusErr>; +} +``` + +#### Group principals and fan-out + +Grants target any `Subject` — `User(Uuid)`, `Group(Uuid)`, or +`Token(Uuid)` (public share link). A single `ShareGranted` event +therefore has **two distinct audiences with different delivery +paths**: + +| Audience | Topic | Payload use | +|---|---|---| +| Share dialog on the resource (anyone with `Share` watching it) | `file:{F}:shares` | Thin fact: "new grantee X with role R"; UI refetches grant list | +| Each affected user (persistent "shared with you") | `user:{member}:notifications` (one publish per member) | Becomes a `notif.notifications` row via `NotificationService::create` | + +The bus **never expands groups**. `NotificationService` is the +group-expansion boundary. `RealtimeBus` only fans out topics that +already exist as concrete `user:*` streams. + +Post-commit sequence for `ShareService::grant(file=F, principal=Group(G), role=R)`: + +``` +1. Insert grant row, COMMIT. +2. members = GroupService::expand_transitive(G) // via closure table +3. bus.publish(Topic::File(F).shares(), + ShareGranted { + principal: PrincipalRef::Group { id: G }, + role: R, + affected_users: Some(members.iter().copied().collect()), + }) // share-dialog fan-out +4. for member in &members { + notification_service.create(NewNotification { + recipient_id: *member, + kind: "share.granted_via_group_membership", + subject_type: "file", subject_id: F, + actor_id: caller, + data: json!({ "via_group": G, "role": R }), + }); + // create() inserts row AND publishes to user:{member}:notifications + } +5. for member in &members { + bus.publish(Topic::UserAuthz(*member), + AuthzChanged { affected: vec![Resource::File(F).into()] }); + } // WS handler re-checks subs +``` + +`Token(_)` principals (public share links) fan out to `file:{F}:shares` +and to a single `user:{creator}:notifications` (kind +`share.link_created` / `share.link_revoked`). No `AuthzChanged` — token +holders don't have WS sessions in this model. + +`GroupMemberAdded { group_id, user_id }` triggers the mirror cascade: +enumerate the group's grants → synthesize one +`share.granted_via_group_membership` notification per affected resource +for the new user → one `AuthzChanged` for the resource set. `Removed` +runs the revocation mirror. + +Bounded fan-out is baked in from day 1 (see § Limits & backpressure): +groups larger than `max_notification_fanout` (default 1000) drop the +per-user notifications with an `event = "notification.fanout_truncated"` +audit line; the `file:{F}:shares` event still fires, so the share dialog +stays accurate, and the recipients discover the grant via UI on next +visit. This case is realistically the "all-employees" scenario where +individual bell pings are noise anyway. + +Membership snapshotting: `GroupService::expand_transitive` runs at the +transaction boundary — we do NOT re-expand at publish time, because a +concurrent membership edit would then leak or duplicate deliveries. The +`members` set is captured then closed over into the post-commit block. + +Coalescing: `NotificationService::create` de-dupes on +`(recipient, kind, subject_type, subject_id, day)` within a short +window. Alice in both `G1` and `G2`, both granted `F`, gets one +notification, not two. + +### 2. In-process impl (`infrastructure/services/in_process_realtime_bus.rs`) + +- `DashMap>`, capacity 256 per topic. +- `subscribe` creates the entry lazily; wraps `Receiver` in + `BroadcastStream` (converts `Lagged` into a stream-level marker; WS + handler kills that session with a `revoked` frame, reason + `slow_consumer`). +- Background GC: when a topic's `receiver_count() == 0` for >60 s, + drop the sender. + +### 3. Replicator scaffolding (day-1) + +- `NoopReplicator` in v1. Wired in DI as `Arc`. +- `InProcessRealtimeBus::publish` calls + `replicator.on_local_publish(...)` **after** local fan-out. + +Futures: + +- **v2 — `PgListenReplicator`.** `pg_notify('oxi_rt', serde_json::to_string(event))` outbound; dedicated `sqlx::PgListener` connection inbound. Payload cap ~8 KB fine because events are thin. No new deployed service — reuses the existing PG. +- **v3 — `BrokerReplicator`.** + - **RabbitMQ:** topic exchange `oxi_rt`, per-server exclusive auto-delete queue bound to `#` (or per-topic bindings for broker-side filtering). Non-durable messages, no user-level queues. + - **NATS:** subject hierarchy = `oxi.rt.folder.{id}`, `oxi.rt.job.{id}`, etc. `Topic::to_wire_key()` maps directly. Core NATS (no JetStream) — ephemeral is the point. + +**Invariant for any broker impl:** no user- or session-scoped state at the broker. Servers hold sessions; the broker is stateless fan-out. Keeps replicator swaps painless and prevents per-user queue leaks. + +### 4. WS handler (`interfaces/api/handlers/rt_ws.rs`) + +- Route `GET /api/rt/ws`. +- **Auth strategy — three accepted paths, all reuse the existing + auth middleware:** + - **Browser session cookie** (`oxicloud_access` or whichever cookie + the auth middleware validates on REST). The WS upgrade request + carries cookies by default; the same `auth_middleware` + + `CurrentUserId` extractor produces `caller_id`. No new code path. + - **Bearer JWT via `Sec-WebSocket-Protocol`** — + `Sec-WebSocket-Protocol: oxi.rt.v1, authorization.bearer.`. + Standard workaround for browser `WebSocket` (which can't set an + `Authorization` header) and native clients like our + `rt-hurl-helper`. The handler reads the second subprotocol + element, validates the JWT via the same path as the REST auth + middleware, and echoes back the base subprotocol name in the + handshake response. + - **Ticket flow** (deferred until DPoP deployments matter) — + `POST /api/rt/ticket` issues a 30-second one-shot ticket, URL is + `/api/rt/ws?ticket=…`. NOT MVP; added when DPoP-strict + deployments make cookie/bearer over WS awkward. Callers today + have no functional need for it. +- On upgrade: + - Extract `caller_id` from the auth mechanism above. + - Auto-subscribe to `user:{caller}:notifications`, + `user:{caller}:authz`, `user:{caller}:sessions`. + - Spawn `RealtimeSession` actor: owns `HashSet`, outbound + `mpsc::Sender` (bounded 512), one reader task per + subscribed topic. +- Per-frame: + - `subscribe`: dispatch on `topic.required_perm() -> AuthzCheck` + and run the matching gate. Three classes exist — + resource-scoped (typically `Read`, sometimes `Share` / + `Comment`), identity-scoped (`caller_id == subject_uuid`, no + admin bypass), role-scoped (`caller.role == Admin`); plus the + bespoke job-originator-or-admin check for `job:{id}`. Full + matrix in **§ AuthZ model**. Deny → `denied` frame + audit + `event = "realtime.subscribe_denied"`. Allow → subscribe on bus, + ack. + - `unsubscribe`: drop the reader task for that topic. + - `ping/pong` for keepalive. + - Binary CRDT frame: route to `CollabSessionService` (see + `docs/plan/markdown-collab.md`), not the generic path. +- On `user:{caller}:authz` event: walk session's subs, re-check each, + evict any that lost access (`revoked` frame with reason + `grant_revoked`). +- On resource-delete: bus publishes `AuthzChanged` for affected → same + eviction path. +- Outbound queue full → close WS with 1013 "try again later"; client + reconnects, refetches, resubs. + +### 5. Service integration — publish AFTER commit + +Rule: **`bus.publish` is called after the DB transaction commits, +never before, never inside**. If publish were inside the tx, a +rollback would still fan out to clients. If publish were async and +awaited, a slow subscriber could hold the tx open. + +Pattern: services return `(result, Vec)` from the tx +boundary; the calling layer publishes after commit. Or a +`TxCommitHook` queues events and flushes on commit. Pick one, apply +everywhere. + +## Frontend components + +### 1. Singleton client (`lib/stores/realtime.svelte.ts`) + +- Fetches a ticket via `POST /api/rt/ticket` (through `apiFetch`, so + DPoP is applied). +- Opens `wss:///api/rt/ws?ticket=…`. +- **Refcounted subscriptions**: + `subs: Map }>`. +- On subscribe by first component: send frame; on last unsubscribe: + send frame. +- On reconnect: reissue ticket, re-establish WS, re-send `subscribe` + for every live topic — components don't care. +- Backoff: exponential (250 ms → 30 s), full-jitter. +- Health: `$state({ connected, latencyMs, subscribedTopics })` + exposed for a debug indicator. + +### 2. Composable (`lib/composables/useTopic.ts`) + +```ts +useTopic(`folder:${folderId}`, (evt) => { /* mutate local $state */ }); +``` + +Handles `$effect` lifecycle (subscribe on mount, unsubscribe on +destroy). Zero connection awareness in components. + +## Wire protocol + +Two wire formats share the same WS connection: + +- **Control + notifications: JSON-RPC 2.0** — universally recognized, + no library needed on either side, standard `id`-correlated + responses, standard `error` object shape, id-less notifications for + server-pushed events. Adopts the same well-known framing as + Ethereum node WS APIs, LSP-over-WS, and countless other services; + costs ~30 bytes/message over a bespoke shape and buys instant "oh, + it's JSON-RPC" recognition + off-the-shelf client compat. +- **CRDT binary frames: Yjs sync protocol** — de-facto standard in the + Yjs ecosystem, kept as-is because it's the reason we picked Yjs. + +Method namespace for our JSON-RPC methods: `rt.*` (short for +realtime). Prevents collisions if we ever expose additional RPCs on +the same WS (not planned, but the namespace costs nothing). + +### JSON-RPC frames (control + events) + +```jsonc +// c → s (requests — id-correlated) +{ "jsonrpc": "2.0", "id": 42, "method": "rt.subscribe", + "params": { "topic": "folder:abc" } } +{ "jsonrpc": "2.0", "id": 43, "method": "rt.unsubscribe", + "params": { "topic": "folder:abc" } } +{ "jsonrpc": "2.0", "id": 44, "method": "rt.ping" } + +// s → c (responses to requests — same id) +{ "jsonrpc": "2.0", "id": 42, + "result": { "subscribed": "folder:abc" } } +{ "jsonrpc": "2.0", "id": 43, + "result": { "unsubscribed": "folder:abc" } } +{ "jsonrpc": "2.0", "id": 44, + "result": { "pong": true } } + +// s → c (denials — same id, standard JSON-RPC error object) +{ "jsonrpc": "2.0", "id": 42, + "error": { "code": -32001, "message": "no_read", + "data": { "topic": "drive:xyz" } } } + +// s → c (events — id-less = JSON-RPC notification) +{ "jsonrpc": "2.0", "method": "rt.event", + "params": { + "topic": "folder:abc", + "event": "file_created", + "data": { "file_id": "…", "name": "notes.md", + "parent_id": "abc" }, + "actor": { "user_id": "…" }, + "ts": "2026-09-08T20:12:00Z" + }} + +// s → c (server-initiated eviction — also a notification) +{ "jsonrpc": "2.0", "method": "rt.revoked", + "params": { "topic": "folder:abc", "reason": "grant_revoked" } } +``` + +### Yjs binary frames (CRDT — Phase A collab consumer) + +``` +[1 byte kind][16 bytes doc_id][payload…] + 0x01 = Yjs update → collab:{doc_id} + 0x02 = Yjs awareness → collab:{doc_id}:awareness + 0x03 = Yjs sync-step → collab:{doc_id} +``` + +The WS handler classifies incoming frames by the `MessageType` +(text/binary). Text frames are JSON-RPC; binary frames are Yjs sync +protocol routed to `CollabSessionService` (see +`docs/plan/markdown-collab.md`). + +### JSON-RPC error codes (stable — never repurpose) + +Uses the JSON-RPC 2.0 "server-defined" range `-32000` to `-32099`, +per spec (`-32700..=-32000` is the reserved-by-spec block; `-32000` +downward is application-defined). + +| `code` | `message` | Meaning | Audit `reason` variants | +|---|---|---|---| +| `-32001` | `"no_read"` | Resource-scoped topic, caller lacks Read (or resource doesn't exist — indistinguishable to caller by design). Anti-enum invariant. | `no_read`, `no_such_resource` | +| `-32002` | `"no_share"` | Resource-scoped topic requiring `Share`, caller has Read but not Share. Applies to `file:{id}:shares` (Phase B). | `no_share` | +| `-32003` | `"no_comment"` | Resource-scoped topic requiring `Comment` (`file:{id}:comments` Phase B). | `no_comment` | +| `-32004` | `"topic_forbidden"` | Identity-scoped mismatch OR unknown/malformed topic. Same wire code regardless of whether the target user exists — anti-enum. | `identity_mismatch`, `unknown_topic`, `not_admin` | +| `-32005` | `"sub_limit"` | Per-connection sub cap hit. | `sub_limit` | +| `-32006` | `"rate_limited"` | Subscribe-frame token bucket exhausted. | `rate_limited` | +| `-32007` | `"no_edit"` | CRDT edit frame from a caller without `Edit`. Emitted as a `rt.write_denied` notification (not tied to a request `id`). | `no_edit` | +| `-32603` | `"internal_error"` | Standard JSON-RPC internal error — server-side failure the client should retry. | — (server log) | +| `-32600` | `"invalid_request"` | Malformed JSON-RPC envelope (missing `method`, wrong `jsonrpc` version). Standard JSON-RPC. | `bad_envelope` | +| `-32601` | `"method_not_found"` | Method outside the `rt.*` allowlist. Standard JSON-RPC. | `unknown_method` | +| `-32602` | `"invalid_params"` | Method known but `params` shape wrong (missing `topic`, unparseable). Standard JSON-RPC. | `bad_params` | + +Codes `-32001..=-32007` are our application-defined vocabulary; the +`-326xx` range is JSON-RPC's own standard set and we honour it for +envelope-level problems. Both are stable — a new denial cause gets a +new code, we never repurpose an existing one, per project convention. + +### Sec-WebSocket-Protocol subprotocol advertisement + +Client's WS handshake sends: +`Sec-WebSocket-Protocol: oxi.rt.v1, authorization.bearer.` + +Server accepts the handshake with `Sec-WebSocket-Protocol: oxi.rt.v1` +(the bearer half is consumed for auth, not echoed). The `v1` gives +us a bump-when-we-break contract handle; adding new methods stays +backward-compatible under `oxi.rt.v1`. + +### Payload discipline + +Event payloads are **thin facts** (IDs + actor + verb). Never full +DTOs — client refetches details via REST if it needs them. Keeps the +AuthZ surface small (thin payloads can't leak fields the caller +couldn't already read via REST for that resource) and makes the pg +NOTIFY 8 KB cap a non-issue. + +## AsyncAPI generation + +Mirror OpenAPI's role for the REST surface. The WS surface gets a +machine-readable AsyncAPI 3.0 document generated from the same Rust +enums the server uses, so the wire contract stays in sync with +implementation by construction — no hand-written spec that drifts. + +### What it documents + +- **Server info + subprotocol** — `oxi.rt.v1` under + `Sec-WebSocket-Protocol`, connect URL, auth mechanisms. +- **Channels** — one per topic-kind (`folder`, `file`, `job`, + `user-notifications`, `collab`, …), parameterized by their id: + `folder/{folderId}`, `job/{jobId}`, etc. +- **Operations per channel:** + - `send` — client subscribe / unsubscribe via `rt.subscribe` / + `rt.unsubscribe` (JSON-RPC request messages). + - `receive` — server events via `rt.event` notifications. +- **Message schemas** — the JSON-RPC envelope and one schema per + `event` variant (`file_created`, `folder_created`, + `share_granted`, `notification`, …). Generated via `schemars` from + the same Rust `RealtimeEvent` enum the server publishes, so the + schema is authoritative, not aspirational. +- **Error object shape + `code`/`message` catalog** — the JSON-RPC + error table above becomes an AsyncAPI-declared `errors` block on + the subscribe operation. +- **Binary frame schema** — a `application/octet-stream` message + binding for the Yjs sync protocol frames, with a text description + of the `[kind][doc_id][payload]` layout. AsyncAPI schemas can't + fully describe the Yjs framing (it's out-of-band from the JSON + envelope), so we document the structure in prose alongside a + placeholder schema — same tradeoff every WS spec makes with binary + bodies. + +### Generator — `cargo run --bin generate-asyncapi` + +Follows the same shape as `generate-openapi`: + +- New binary `src/bin/generate_asyncapi.rs` that constructs the + spec from `Topic`, `RealtimeEvent`, `AuthzCheck`, and the JSON-RPC + method/error tables — all live in `application/ports/realtime_ports.rs` + as the single source of truth. +- Uses `schemars` for JSON Schema of each event variant (already + compatible with `serde` derives; no re-annotation needed). +- Emits `resources/gen/asyncapi.yaml` (YAML for human-diffability, + same choice AsyncAPI tooling defaults to). +- Add `just asyncapi` recipe alongside `just openapi`. +- CI check: same as the OpenAPI check — regenerate on every build, + fail if the working tree is dirty after regeneration. Keeps spec + and code from drifting. + +### Consumers + +- **Docs site** — AsyncAPI has a first-class HTML renderer + (`@asyncapi/html-template` or the Studio playground). Point the + docs at `resources/gen/asyncapi.yaml` and the WS surface has the + same discoverability as `openapi.json`. +- **Client SDK generation (later)** — `@asyncapi/generator` produces + typed clients (TS, Go, Python, Java). Not needed for v1, but the + door is open when a third-party integration asks for one. +- **Contract testing (later)** — the spec doubles as a contract the + smoke tests can assert against; `rt-hurl-helper` could validate + incoming events against the schema before asserting on values. + Cheap follow-up. + +### Scope for the first PR + +- Generator produces spec covering the Phase-A-MVP surface only + (`rt.subscribe` / `rt.unsubscribe` / `rt.ping` methods, + `rt.event` / `rt.revoked` notifications, `Folder(id)` and + `UserAuthz(u)` topics, `FileCreated` / `FolderCreated` events, + the error-code table). +- Adding a new topic/event/method later is an enum variant + serde + derive → regenerate → commit. Same discipline as OpenAPI. + +## AuthZ model (audit rules per AGENTS.md) + +### The subscribe gate + +"At least Read on the resource" is the **default** for resource-scoped +topics, but not the whole story. Every topic variant declares its own +gate via `Topic::required_perm() -> AuthzCheck`. Three classes exist — +the WS handler dispatches on the returned enum, it does not assume a +single check applies everywhere. + +#### Class 1 — Resource-scoped (majority) + +Default gate: `AuthorizationEngine::require(caller, resource, Read)`. + +| Topic | Resource | Permission | +|---|---|---| +| `folder:{id}` | folder | `Read` | +| `folder:{id}:presence` | folder | `Read` | +| `file:{id}` | file | `Read` | +| `file:{id}:presence` | file | `Read` | +| `collab:{file_id}` | file | `Read` (Reader = view + own cursor; edits gate separately, see below) | +| `collab:{file_id}:awareness` | file | `Read` | +| `drive:{id}` | drive | `Read` (drive membership) | +| `calendar:{id}` | calendar | `Read` | +| `addressbook:{id}` | address book | `Read` | + +Two Phase-B resource topics use a **stricter** permission because the +topic itself would leak enumeration metadata a Reader can't otherwise +see today: + +| Topic | Actual permission | Why not Read | +|---|---|---| +| `file:{id}:shares` | `Share` (Owner-tier) | Reader sees the file's content, not who else has access. The share list is management metadata; the REST share endpoints already gate this way. | +| `file:{id}:comments` | Whatever the REST comments API decides — `Read` if comments are public to Readers; `Comment` if commenter-tier only | Consistency with REST. The bus does not invent a new policy. | + +#### Class 2 — Identity-scoped + +Gate: `caller_id == subject_uuid`. Plain equality. **No admin bypass** +— an admin cannot subscribe to `user:{other}:notifications`. Privacy is +a hard rule; cross-user monitoring uses admin topics, never a user's +private stream. + +| Topic | Gate | +|---|---| +| `user:{u}:notifications` | caller == u | +| `user:{u}:authz` | caller == u | +| `user:{u}:sessions` | caller == u | +| `user:{u}:uploads` | caller == u | +| `user:{u}:trash` | caller == u | + +Auto-subscribed topics (`user:{caller}:*`) at connect go through the +same check for consistency — the caller identity is derived from the +validated ticket, so this is by construction, but the code path must +not short-circuit. + +#### Class 3 — Role-scoped + +Gate: `caller.role == Admin` (or specific admin sub-role once we +introduce them). + +| Topic | Gate | +|---|---| +| `admin:sessions` | admin role | +| `admin:audit` | admin role | + +#### One non-resource topic — bespoke check + +| Topic | Gate | +|---|---| +| `job:{id}` | `jobs.created_by == caller` **OR** admin role. Jobs are not in the AuthZ engine's resource set; the check lives in `Topic::required_perm()` and queries the job registry. | + +### Enforcement rules + +1. **AuthZ at subscribe time, not per event.** Fan-out is hot; + subscribe is the choke point. Checking every event against every + subscriber's grants would burn CPU on busy topics. +2. **Evict on grant loss** — do NOT keep re-checking to preserve a + sub. The write path publishes `AuthzChanged { affected }` to + `user:{u}:authz`; the WS handler walks that session's + `HashSet` and drops any sub whose resource intersects + `affected`. Same eviction path for resource-delete, group-member + removal, and admin kicks. +3. **Anti-enumeration on denials.** Per the graduated-denial + convention (see `authz_require_graduated_denial`), the wire reason + collapses cases the caller cannot distinguish; the audit line + records the truth. +4. **CRDT edit frames re-check on the write side.** A Reader can hold + a `collab:{file}` sub (view + cursor); their `0x01` update frames + are dropped by the WS handler with `collab.write_denied` audit + (`reason = no_edit`). Verified once per session and re-verified on + `user:{caller}:authz` events. + +### Wire-reason vocabulary (stable — never repurpose) + +The wire uses JSON-RPC 2.0 `error` objects — see **§ Wire protocol → +JSON-RPC error codes** for the full `code`/`message`/audit-`reason` +mapping. That table is the authoritative one; this section +cross-references its audit-reason column for the AuthZ dispatch and +confirms the anti-enumeration collapse rules the wire honours. + +### Audit-line convention + +- **Connect reject** — `event = "auth.rt_ticket_rejected"`, + `reason ∈ {expired, unknown, ip_mismatch, replay}`. +- **Subscribe deny** — `event = "realtime.subscribe_denied"`, `reason` + from the audit column above, plus `caller_id`, `topic`. Emitted + BEFORE the wire `denied` frame. +- **Evict** — `event = "realtime.subscription_evicted"`, + `reason ∈ {grant_revoked, resource_deleted, admin_kick, group_membership_lost}`, + plus `caller_id`, `topic`. +- **Collab edit rejected** — `event = "collab.write_denied"`, + `reason ∈ {no_edit, session_evicted, external_write_conflict}`. +- **Notification fanout truncated** — `event = + "notification.fanout_truncated"`, `reason = "over_max_fanout"`, + `resource_id`, `principal`, `member_count`. + +Every audit line uses `target: "audit"` per project convention. Wire +reasons are the compressed public vocabulary; audit reasons are the +uncompressed private truth. + +## Limits & backpressure + +| Limit | Default | Rationale | +|---|---|---| +| Subs per connection | 128 | Prevents runaway/malicious pinning of server memory | +| Subscribe frames/sec/conn | 50 | Token bucket, prevents storm-subscribing | +| Outbound mpsc slots/conn | 512 | Full → close WS 1013 | +| Broadcast ring slots/topic | 256 | Slow subscriber → lag → close WS + audit | +| Max event size | 8 KB | Fail-fast dev assertion; keeps pg NOTIFY cap a non-issue | +| Ticket TTL | 30 s | Short window, one-shot | +| `max_notification_fanout` | 1000 recipients / event | Beyond this, drop per-user notifications + audit `notification.fanout_truncated`; the `file:{id}:shares` event still fires. Covers the "all-employees" group case where individual bell pings would be noise. | + +## Failure modes + +- WS drop mid-session → client reconnects, ticket flow again, + re-subscribes. Server discards session state. +- Publish under load → `broadcast::Sender::send` never blocks; slow + subs lag out. Never let the publish path stall. +- Ticket replay → ticket is one-shot in-memory; second use is + `denied` + audit. +- Post-commit publish failing → log a warning and move on. Do NOT + retry into a queue; ephemeral events are best-effort by design. +- Replicator down (v2+) → local bus keeps working for same-instance + subs; log the outage; alert. + +## First PR — MVP scope and hurl smoke test + +The smallest slice that proves fan-out works, topics are isolated, +and the AuthZ gate rejects unauthorized subscribes. Everything +larger (notifications table, presence, collab) rides on top later. + +### Scope in + +- `RealtimeBus` port + `InProcessRealtimeBus`. +- WS handler at `GET /api/rt/ws` with `subscribe` / `unsubscribe` / + `ping` frames only (no CRDT binary frames yet). +- Auth: reuse existing `auth_middleware` — session cookie for + browsers OR bearer JWT via `Sec-WebSocket-Protocol: + oxi.rt.v1, authorization.bearer.` for programmatic clients. + Ticket flow deferred. +- Topics: `Folder(id)` (Class 1 — Resource-scoped, `Read`) and + `UserAuthz(u)` (Class 2 — Identity-scoped, auto-subscribed at + connect). No other topics accepted in MVP; parser returns + `Unknown` → `denied` with `reason = topic_forbidden`. +- Events: `FileCreated`, `FolderCreated`. Publish hooks added in + `FolderService::create_folder_with_perms` and + `FileManagementService`'s file-create path (upload / chunked + upload commit — publish AFTER commit only). + +### Scope out (later PRs, not this one) + +- Delete / rename / move publishes (same pattern, verified after + create works). +- `user:{u}:notifications` topic, notifications table, bell UI. +- `job:{id}` topic, `collab:{id}` binary frames. +- Grant-revocation eviction (still enforced structurally via + `Topic::required_perm` at subscribe, but no live evict-on-change + wiring — that comes with the `AuthzChanged` publish hook in a + follow-up). +- Ticket flow, rate limiting on subscribe frames, slow-subscriber + metrics. +- Frontend integration (`useTopic`, folder-view autorefresh). +- `PgListenReplicator` — v2 multi-instance. + +### Test surface — `rt-hurl-helper` (follows existing convention) + +The api-test suite is entirely HTTP via hurl and cannot drive +WebSocket. Precedent for auxiliary Rust binaries exists in +`opaque-hurl-helper` and `dpop-hurl-helper` (both built with +`--features test_utils`, both invoked from `tests/api/run.sh` +outside the main hurl block). The bus test follows the same +pattern. + +**New binary:** `src/bin/rt_hurl_helper.rs`, gated on +`test_utils`. Sole new crate dependency: +`tokio-tungstenite` — added under `[dependencies.tokio-tungstenite] +optional = true` and pulled in by the `test_utils` feature so the +release binary is unaffected. Never ships in production. + +**CLI shape:** + +``` +oxi-rt-hurl-helper [flags] + + subscribe-and-collect # runs in background alongside hurl + --url ws://.../api/rt/ws + --token JWT # bearer, passed via Sec-WebSocket-Protocol + --subscribe TOPIC # may repeat + --expect-events N # exit 0 when N events arrive + --timeout DURATION # overall cap, default 3s + --output PATH # write JSON summary on exit + + expect-denied # runs synchronously + --url ws://.../api/rt/ws + --token JWT + --subscribe TOPIC + --reason KEY # expected denial reason, default: any + --timeout DURATION # default 2s +``` + +Exit codes: `0` = expectation met, `1` = expectation failed +(wrong event, unexpected event, timeout without hitting the target, +denied when expecting event, or vice versa), `2` = protocol error +/ connect failure. + +Output JSON schema (for post-mortem assertions in shell): + +```jsonc +{ + "subscribed": ["folder:"], + "denied": [], + "events": [ { "topic": "folder:", "event": "file_created", + "data": { "file_id": "…", "name": "…", + "parent_id": "", "actor": "…" }, + "ts": "2026-…" } ], + "timed_out": false, + "protocol_err": null +} +``` + +### Coverage — four scenarios, each in the same test file + +Orchestrated by a single `tests/api/rt_bus_check.sh` invoked from +`tests/api/run.sh` after the main hurl block. Follows the +`refcount_cascade` / `thumb_import_check` patterns already in place. + +**Scenario 1 — Positive delivery** (fan-out works) + +``` +setup.hurl: + - user1 logs in → capture $USER1_TOKEN + - user1 creates folder A → capture $FOLDER_A + +shell: + rt-hurl-helper subscribe-and-collect \ + --token $USER1_TOKEN --subscribe folder:$FOLDER_A \ + --expect-events 1 --timeout 3s --output /tmp/rt_s1.json & + sleep 0.3 # give the subscribe frame time to ack + +actions.hurl: + - user1 creates a file in $FOLDER_A + +wait rt-hurl-helper +``` + +Assertion (jq on `/tmp/rt_s1.json`): +- `.timed_out == false` +- `.events | length == 1` +- `.events[0].event == "file_created"` +- `.events[0].data.parent_id == $FOLDER_A` + +**Scenario 2 — Topic isolation** (no event on unsubscribed folder) + +Verifies: a user subscribed only to folder A does NOT receive +events for actions in folder B, even when the user has full access +to both. + +``` +setup.hurl: + - user1 creates folder B → capture $FOLDER_B (folder A from S1 reused) + +shell: + rt-hurl-helper subscribe-and-collect \ + --token $USER1_TOKEN --subscribe folder:$FOLDER_A \ + --expect-events 1 --timeout 3s --output /tmp/rt_s2.json & + sleep 0.3 + +actions.hurl: + # First: create a file in B — user1 has full access, but we're + # not subscribed to B, so nothing should arrive on the helper. + - user1 creates a file in $FOLDER_B + # Second: create a file in A — this triggers the helper's exit. + - user1 creates a file in $FOLDER_A + +wait +``` + +Assertion: +- `.events | length == 1` +- `.events[0].data.parent_id == $FOLDER_A` ← NOT B +- no event with `parent_id == $FOLDER_B` present + +The key invariant this locks in: **the server fans out per topic, +not per user or per drive**. A subscriber to `folder:A` sees only +`folder:A` events, even for topics they'd have permission to +subscribe to but didn't. + +**Scenario 3 — AuthZ denial** (subscribe rejected on missing Read) + +Verifies: a user without `Read` on a folder cannot subscribe to +its topic. Denial wire reason is `no_read`; audit line records +`realtime.subscribe_denied` with `reason ∈ {no_read, +no_such_resource}`. + +``` +setup.hurl: + - user2 registers and logs in → capture $USER2_TOKEN + - (user2 has no grant on $FOLDER_A, which is user1's private folder) + +shell: + rt-hurl-helper expect-denied \ + --token $USER2_TOKEN --subscribe folder:$FOLDER_A \ + --reason no_read --timeout 2s + # exit 0 = denied frame received with reason=no_read +``` + +Assertion is the helper's exit code (`0` pass, `1` fail). No +`/tmp` output file needed for a binary pass/fail. + +**Scenario 4 — Anti-enumeration parity** (nonexistent folder ≡ no +access, from the caller's POV) + +Verifies: subscribing to a folder that does not exist returns the +**same** wire reason as subscribing to a folder the caller can't +Read. Protects against a folder-enumeration oracle. + +``` +shell: + rt-hurl-helper expect-denied \ + --token $USER2_TOKEN --subscribe folder:00000000-0000-0000-0000-000000000000 \ + --reason no_read --timeout 2s + # exit 0 = same wire reason as scenario 3 +``` + +Assertion: exit code 0. The audit line (checked out-of-band if we +wire log capture) records `reason = "no_such_resource"` — but the +wire reason is `no_read`, matching scenario 3. This is the +graduated-denial invariant from `authz_require_graduated_denial`. + +### How the scenarios chain + +All four run in one shell script, one WS connection is opened per +scenario for isolation (a helper invocation = a fresh WS). No +state carries between scenarios except the folder ids and tokens +captured in `setup.hurl`. Total wall-clock ≤ 10 s including +sleeps. + +### Cleanup + +Follows the existing api-test convention (per project memory +`api_tests`): + +- Shared DB is dropped between full test-suite runs by + `tests/common/stop-db.sh`. +- Storage is wiped at run start. +- No per-scenario teardown; folders A and B persist for the rest + of the run — no test that runs after this cares about them. + +### justfile / CI hook + +Add to the existing `test-api` recipe list of Rust helper builds +(there's already a compile step for `opaque-hurl-helper` / +`dpop-hurl-helper`); the new binary joins the same +`--features test_utils` build. `tests/api/run.sh` gets one line — +`./rt_bus_check.sh || die "rt bus smoke failed"` — inserted after +the main hurl block, before the existing storage-cleanup / thumb +checks. + +### What this coverage locks in + +- Subscribe path AuthZ gate is real (S3, S4). +- Anti-enumeration parity between "no perm" and "no resource" (S4) + — the invariant the plan promises. +- Fan-out is topic-scoped, not user-scoped (S2). +- Publish-after-commit produces exactly one event per action (S1), + not zero (rollback lost the publish) and not multiple (retry / + double-hook). +- End-to-end wire format is stable (S1 asserts on + `event = "file_created"` string). + +Everything else in the plan — evict-on-revoke, slow-subscriber +kick, rate limiting, ticket flow, PgListen replicator — is +follow-up test work with its own scenarios, layered on top of +this baseline once the baseline is green. + +--- + +## Roadmap + +### Phase A — Foundation (bus + notifications + MD collab) + +Ships the infrastructure and the two most visible consumers together. + +- Bus port + `InProcessRealtimeBus` + `NoopReplicator` + WS handler + + ticket endpoint. +- Frontend singleton + `useTopic` composable. +- Topics live: `folder:{id}`, `user:{u}:notifications`, `job:{id}`, + `collab:{file_id}`, `collab:{file_id}:awareness`. +- **Folder-live updates**: `FolderService` / `FileManagementService` + publish `file.created` / `file.deleted` / `file.renamed` / + `file.moved` after commit; FE folder view subscribes and mutates + local state — no manual refresh. +- **Job dashboard live**: `JobRegistry` publishes step progress and + terminal state; FE job dashboard subscribes and replaces the + current polling. +- **Notifications table + bell**: new `notifications` table + + `NotificationService` port; initial ingesters for `share-granted`, + `new-login-from-new-device`, `job-completed-for-you`, + `storage-quota-threshold`. FE bell with unread count, slide-out + panel, toast pop on receive. +- **MD collab editor**: see companion plan + `docs/plan/markdown-collab.md` — depends on this phase's WS + handler + binary frame routing. + +Deliverables sized ~4 weeks end-to-end. + +### Phase B — Presence + comments + +Everything that turns OxiCloud from a file store into a shared +workspace. + +- **Presence topics** — `folder:{id}:presence`, `file:{id}:presence`. + Awareness-style: joined/left/cursor. Ephemeral, not persisted. +- **FE presence UI**: "N people viewing" badge in folder header; + avatar rail; hover to highlight; "someone is previewing this photo + right now" in the lightbox. +- **Comments on any file** — new `comments` table (threaded, per + file, supports reactions), `CommentService` port, + `file:{id}:comments` topic for live delivery. +- **@mentions**: mention autocomplete in the comment editor; + mention → notification into the mentioned user's + `user:{u}:notifications` topic + `notifications` row + optional + email (reuses existing `MagicLinkMailer`-style templating). +- **Reactions**: 👍❤️🎉 on comments and on files themselves; live + fan-out on the same `file:{id}:comments` topic. +- **Comment resolutions**: Google-Docs-style thread markers. + +Deliverables sized ~3 weeks after Phase A. + +### Phase C — Sync client push + album live + +Where the bus starts paying for itself on infrastructure cost too. + +- **Sync-client push invalidation**: WebDAV / NextCloud DAV handlers + publish `file:{id}` and `folder:{id}` deltas after commit. Sync + clients get a lightweight `Sync-Invalidate` mechanism (or a + dedicated WS endpoint for headless clients) so they refetch only + changed paths instead of polling PROPFIND. Cuts a large chunk of + Nextcloud-style client chatter. +- **Album live updates**: `folder:{album_id}` reused — as photos are + added to an album, everyone viewing sees them appear. +- **Slideshow sync**: one presenter picks "Present"; other viewers of + the album can opt-in to follow the presenter's current frame. + Uses `folder:{album_id}` with a `presenter_frame` event kind. + +Deliverables sized ~2–3 weeks after Phase B. + +### Later — multi-instance & broker + +Only invoked when the deployment actually needs it. Nothing above +depends on these landing on any fixed date. + +- **`PgListenReplicator`** — ship when we run more than one server + instance. Same port, no consumer changes. +- **`BrokerReplicator`** for RabbitMQ or NATS — ship when either + cross-datacenter fan-out or a shared broker with other services + matters. Same port, no consumer changes. + +## What this bus does NOT replace + +- Message queue / job queue — jobs stay in `job_registry`; bus just + carries their progress live. +- Audit log — stays `tracing target: "audit"`. +- Email — `NotificationService`'s deliverer for offline users. +- Durable per-user "inbox" — the `notifications` table is the source + of truth; bus is the live-delivery optimization. diff --git a/src/application/ports/mod.rs b/src/application/ports/mod.rs index 53c8e8db..a246d661 100644 --- a/src/application/ports/mod.rs +++ b/src/application/ports/mod.rs @@ -22,6 +22,7 @@ pub mod music_ports; pub mod opaque_ports; pub mod outbound; pub mod plugin_ports; +pub mod realtime_ports; pub mod recent_ports; pub mod resource_access_hook; pub mod share_ports; diff --git a/src/application/ports/realtime_ports.rs b/src/application/ports/realtime_ports.rs new file mode 100644 index 00000000..a3bd1165 --- /dev/null +++ b/src/application/ports/realtime_ports.rs @@ -0,0 +1,493 @@ +//! Realtime message-bus port — the seam every service publishes through and +//! every WS session subscribes on. +//! +//! # Design (see `docs/plan/message-bus.md`) +//! +//! - [`RealtimeBus`] is the **local-facing** trait: services publish, the WS +//! handler subscribes. It never involves the network. +//! - [`BusReplicator`] is the OPTIONAL seam that mirrors local publishes to +//! and from a broker (pg `LISTEN/NOTIFY`, RabbitMQ, NATS). Callers see only +//! [`RealtimeBus`]; a real replicator plugs into the in-process impl without +//! touching consumers. Day-1 impl is [`NoopReplicator`]. +//! +//! # MVP scope +//! +//! Ships the smallest slice that lets the smoke test verify a folder +//! subscription receives file/folder-created events and rejects subscribes +//! to folders the caller can't `Read`: +//! +//! - Topics: [`Topic::Folder`] and [`Topic::UserAuthz`] +//! - Events: [`RealtimeEvent::FileCreated`], [`RealtimeEvent::FolderCreated`] +//! +//! Adding a variant is a one-line change plus a match arm in `to_wire_key` / +//! `parse` / `required_perm`. Other topics (`file:{id}`, `job:{id}`, +//! `collab:{id}`, `user:{u}:notifications`, …) land with their producers in +//! Phase-A follow-ups. +//! +//! # Wire protocol +//! +//! JSON-RPC 2.0 for control + events (text frames), Yjs sync protocol for +//! CRDT (binary frames). This module owns the JSON-RPC error-code +//! vocabulary; see [`error_code`]. + +use std::pin::Pin; +use std::sync::Arc; + +use futures::Stream; +use serde::{Deserialize, Serialize}; +use tokio::sync::Notify; +use uuid::Uuid; + +use crate::common::errors::DomainError; + +// ════════════════════════════════════════════════════════════════════════════ +// Topic — a typed key on the bus +// ════════════════════════════════════════════════════════════════════════════ + +/// A topic on the realtime bus. Typed enum, not a string — prevents typos +/// and gives exhaustive matching in the AuthZ dispatch and the wire encoder. +/// +/// Encodes to a stable dotted wire key that maps naturally onto RabbitMQ +/// topic-exchange routing keys or NATS subjects when the [`BusReplicator`] +/// seam is filled in later. +#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)] +pub enum Topic { + /// A folder's mutation stream — file/subfolder created/deleted/renamed/ + /// moved in or out. Consumed by the folder view for live refresh. + Folder(Uuid), + + /// A user's private authz-change channel. The WS handler will auto- + /// subscribe the caller and evict stale subs when its events fire once + /// the eviction wiring lands (Phase-A follow-up). + UserAuthz(Uuid), +} + +impl Topic { + /// Stable dotted wire form used by the JSON-RPC control frames and any + /// future broker routing keys. Reverse of [`Topic::parse`]. + pub fn to_wire_key(&self) -> String { + match self { + Topic::Folder(id) => format!("folder:{id}"), + Topic::UserAuthz(id) => format!("user:{id}:authz"), + } + } + + /// Parse a wire-form topic string. Rejects unknown shapes with a stable + /// error kind so the WS handler can respond with a JSON-RPC error object + /// (`topic_forbidden` for unknown topic shapes, `no_read` for known + /// shapes the caller can't reach — the latter after the AuthZ check). + pub fn parse(s: &str) -> Result { + if let Some(rest) = s.strip_prefix("folder:") { + let id = Uuid::parse_str(rest).map_err(|_| ParseTopicErr::BadUuid)?; + return Ok(Topic::Folder(id)); + } + if let Some(rest) = s.strip_prefix("user:") + && let Some((id_str, "authz")) = rest.rsplit_once(':') + { + let id = Uuid::parse_str(id_str).map_err(|_| ParseTopicErr::BadUuid)?; + return Ok(Topic::UserAuthz(id)); + } + Err(ParseTopicErr::Unknown) + } + + /// Which permission check the WS handler must run before allowing a + /// subscribe. Three classes per plan (see + /// `docs/plan/message-bus.md § AuthZ model`): + /// + /// - Resource-scoped: default `Read` on the resource (Phase-B adds + /// `Share`/`Comment` for the stricter topics). + /// - Identity-scoped: `caller_id == subject_uuid`. No admin bypass. + /// - Role-scoped / bespoke: not represented in this MVP. + pub fn required_perm(&self) -> AuthzCheck { + match self { + Topic::Folder(id) => AuthzCheck::ResourceRead { + resource: BusResource::Folder(*id), + }, + Topic::UserAuthz(id) => AuthzCheck::IdentityMatch { user_id: *id }, + } + } +} + +/// Parse failure for a wire-form topic string. Kept small — the WS handler +/// maps every variant to `topic_forbidden` on the wire (both a bad UUID and +/// an unknown shape are indistinguishable from the caller's perspective). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ParseTopicErr { + /// The prefix was recognized but the UUID inside didn't parse. + BadUuid, + /// The topic string didn't match any known shape (typo, or a topic + /// that isn't in this MVP). + Unknown, +} + +// ════════════════════════════════════════════════════════════════════════════ +// AuthzCheck — the gate class the WS handler dispatches on +// ════════════════════════════════════════════════════════════════════════════ + +/// Resource kinds the bus knows how to gate on. Deliberately a small closed +/// enum, not the full `domain::authorization::Resource` — the bus does not +/// need every resource type in the domain, and keeping this separate avoids +/// dragging domain-shaped churn into the port. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum BusResource { + Folder(Uuid), + // File(Uuid), Drive(Uuid), Calendar(Uuid), AddressBook(Uuid) land with + // their topic variants. +} + +/// The check the WS handler must run at subscribe time. Split into the three +/// classes described in `docs/plan/message-bus.md § AuthZ model`, so a new +/// topic variant with a new gate shape is a compile error at the dispatch +/// site rather than a runtime "unhandled" bug. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum AuthzCheck { + /// Class 1 — Resource-scoped, default gate is Read on the resource. + /// Extend to `ResourceShare`/`ResourceComment` when the Phase-B topics + /// (`file:{id}:shares`, `file:{id}:comments`) land. + ResourceRead { resource: BusResource }, + + /// Class 2 — Identity-scoped. `caller_id` must equal `user_id`. + /// No admin bypass — privacy is a hard rule. + IdentityMatch { user_id: Uuid }, + // Class 3 (role-scoped `admin:*`) and the bespoke `job:{id}` check + // land with their topic variants. +} + +// ════════════════════════════════════════════════════════════════════════════ +// RealtimeEvent — the payload +// ════════════════════════════════════════════════════════════════════════════ + +/// A fact that has just become true. Emitted by services AFTER commit, +/// never inside a DB transaction — a rollback would otherwise fan out a +/// lie. +/// +/// Payloads are **thin facts** (ids + actor + verb): the client refetches +/// details via REST when it needs them. This keeps the AuthZ surface small +/// (thin payloads can't leak fields the caller couldn't already read via +/// REST) and keeps events well under the ~8 KB pg NOTIFY cap when the +/// `PgListenReplicator` seam is filled in later. +/// +/// Wire form uses `#[serde(tag = "event", rename_all = "snake_case")]`; +/// discriminator strings are the JSON-RPC notification `event` field. New +/// denial cause / new event = new variant, never repurpose an existing one, +/// per project convention. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "event", rename_all = "snake_case")] +pub enum RealtimeEvent { + /// A file was created inside `parent_id`. + FileCreated { + file_id: Uuid, + name: String, + parent_id: Uuid, + actor: Uuid, + }, + /// A sub-folder was created inside `parent_id`. + FolderCreated { + folder_id: Uuid, + name: String, + parent_id: Uuid, + actor: Uuid, + }, +} + +// ════════════════════════════════════════════════════════════════════════════ +// JSON-RPC 2.0 error codes — stable, never repurpose +// ════════════════════════════════════════════════════════════════════════════ + +/// JSON-RPC 2.0 `error.code` values used on the WS wire. Follows the spec's +/// "server-defined" range `-32000` to `-32099` for our application-defined +/// codes; the standard `-326xx` envelope codes are re-exported here too so +/// the WS handler has one place to reach for. +/// +/// See `docs/plan/message-bus.md § JSON-RPC error codes` for the +/// wire-`message`/audit-`reason` mapping. +pub mod error_code { + /// Resource-scoped topic, caller lacks Read (or resource doesn't exist — + /// indistinguishable to caller by design). Anti-enum invariant. + pub const NO_READ: i32 = -32001; + + /// Resource-scoped topic requiring `Share`, caller has Read but not + /// Share. Applies to `file:{id}:shares` (Phase B). + pub const NO_SHARE: i32 = -32002; + + /// Resource-scoped topic requiring `Comment` (`file:{id}:comments` + /// Phase B). + pub const NO_COMMENT: i32 = -32003; + + /// Identity-scoped mismatch OR unknown/malformed topic. Same wire code + /// regardless of whether the target user exists — anti-enum. + pub const TOPIC_FORBIDDEN: i32 = -32004; + + /// Per-connection sub cap hit. + pub const SUB_LIMIT: i32 = -32005; + + /// Subscribe-frame token bucket exhausted. + pub const RATE_LIMITED: i32 = -32006; + + /// CRDT edit frame from a caller without `Edit`. Emitted as an + /// `rt.write_denied` notification (not tied to a request `id`). + pub const NO_EDIT: i32 = -32007; + + // ────────────────────── JSON-RPC 2.0 standard codes ───────────────────── + // Re-exported so the WS handler doesn't reach for two constant lists. + + /// Server-side failure the client should retry. + pub const INTERNAL_ERROR: i32 = -32603; + + /// Malformed JSON-RPC envelope (missing `method`, wrong `jsonrpc` + /// version). + pub const INVALID_REQUEST: i32 = -32600; + + /// Method outside the `rt.*` allowlist. + pub const METHOD_NOT_FOUND: i32 = -32601; + + /// Method known but `params` shape wrong (missing `topic`, unparseable). + pub const INVALID_PARAMS: i32 = -32602; +} + +// ════════════════════════════════════════════════════════════════════════════ +// RealtimeBus — the port +// ════════════════════════════════════════════════════════════════════════════ + +/// The local-facing message bus. Fire-and-forget publish, stream subscribe. +/// +/// `publish` is intentionally synchronous — services must not `await` under +/// a DB transaction (a slow subscriber could hold the tx open) and services +/// should not care whether fan-out is happening in a background task or not. +/// +/// `subscribe` returns a `Stream` so the impl can change (broadcast, mpsc, +/// pg listener) without churn at the consumer. +pub trait RealtimeBus: Send + Sync + 'static { + /// Fan an event out to every current subscriber of `topic`. Never + /// blocks; slow subscribers are dropped by the impl (they'll reconnect + /// and refetch). + fn publish(&self, topic: &Topic, event: RealtimeEvent); + + /// Subscribe to `topic`. The returned stream yields events until the + /// subscriber is dropped or the impl kicks it out (e.g. for lagging + /// too far behind). + fn subscribe(&self, topic: &Topic) -> BusStream; +} + +/// Boxed stream returned by [`RealtimeBus::subscribe`]. Aliased so +/// consumers don't need to spell out the `Pin>` shape. +pub type BusStream = Pin + Send>>; + +// ════════════════════════════════════════════════════════════════════════════ +// BusReplicator — the multi-instance seam (day-1 noop) +// ════════════════════════════════════════════════════════════════════════════ + +/// Cross-instance replicator. Sits BESIDE [`RealtimeBus`], not in front of +/// it — the bus does the local fan-out; the replicator forwards outbound +/// publishes to the broker (pg NOTIFY, RabbitMQ, NATS) and injects inbound +/// broker messages back into the local bus. +/// +/// V1 ships [`NoopReplicator`]. The trait is declared today so wiring the +/// day the second impl arrives is drop-in. +#[async_trait::async_trait] +pub trait BusReplicator: Send + Sync + 'static { + /// Called by the local bus for every publish. Fire-and-forget — must not + /// block or await; forwarding to the broker happens on a background task + /// owned by the impl. + fn on_local_publish(&self, topic: &Topic, event: &RealtimeEvent); + + /// Long-running consumer task: reads remote messages and re-publishes + /// locally. Returns when `shutdown` is notified — DI calls + /// `shutdown.notify_one()` on graceful shutdown. + /// + /// **Shutdown semantics:** use `Notify::notify_one` (not + /// `notify_waiters`) at the signalling site: `notify_one` stores a + /// permit if no waiter is currently parked, so signal-before-park is + /// safe. `notify_waiters` silently drops signals sent before parking + /// and creates a race. This constrains the impl to a single-waiter + /// shutdown handle; multi-task replicators must spin their own + /// `CancellationToken`-style fan-out internally. + async fn run(self: Arc, shutdown: Arc) -> Result<(), DomainError>; +} + +/// Day-1 replicator: does nothing. Wired unconditionally so callers hold +/// `Arc` uniformly. Swapped for a real impl when +/// multi-instance deployment matters. +#[derive(Default)] +pub struct NoopReplicator; + +#[async_trait::async_trait] +impl BusReplicator for NoopReplicator { + fn on_local_publish(&self, _topic: &Topic, _event: &RealtimeEvent) { + // Intentionally empty. Local fan-out already happened in the bus. + } + + async fn run(self: Arc, shutdown: Arc) -> Result<(), DomainError> { + // Park until shutdown so the DI-managed handle stays alive with the + // same lifecycle as a future real replicator. + shutdown.notified().await; + Ok(()) + } +} + +// ════════════════════════════════════════════════════════════════════════════ +// Tests +// ════════════════════════════════════════════════════════════════════════════ + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn folder_topic_roundtrip() { + let id = Uuid::new_v4(); + let t = Topic::Folder(id); + let wire = t.to_wire_key(); + assert_eq!(wire, format!("folder:{id}")); + assert_eq!(Topic::parse(&wire).unwrap(), t); + } + + #[test] + fn user_authz_topic_roundtrip() { + let id = Uuid::new_v4(); + let t = Topic::UserAuthz(id); + let wire = t.to_wire_key(); + assert_eq!(wire, format!("user:{id}:authz")); + assert_eq!(Topic::parse(&wire).unwrap(), t); + } + + #[test] + fn parse_rejects_bad_uuid() { + assert_eq!( + Topic::parse("folder:not-a-uuid"), + Err(ParseTopicErr::BadUuid) + ); + } + + #[test] + fn parse_rejects_unknown_shape() { + assert_eq!(Topic::parse(""), Err(ParseTopicErr::Unknown)); + assert_eq!(Topic::parse("unknown:x"), Err(ParseTopicErr::Unknown)); + assert_eq!( + Topic::parse(&format!("user:{}", Uuid::new_v4())), + Err(ParseTopicErr::Unknown), + "user: without :authz suffix is not a known topic in MVP" + ); + } + + #[test] + fn required_perm_folder_is_resource_read() { + let id = Uuid::new_v4(); + assert_eq!( + Topic::Folder(id).required_perm(), + AuthzCheck::ResourceRead { + resource: BusResource::Folder(id) + } + ); + } + + #[test] + fn required_perm_user_authz_is_identity_match() { + let id = Uuid::new_v4(); + assert_eq!( + Topic::UserAuthz(id).required_perm(), + AuthzCheck::IdentityMatch { user_id: id } + ); + } + + #[test] + fn event_serializes_with_snake_case_discriminator() { + // The `#[serde(tag = "event")]` shape is the WS wire contract for + // the `rt.event` JSON-RPC notification's `params.event` field. Pin + // it with a snapshot so accidental rename of the enum variant + // fails the test instead of silently breaking clients. + let ev = RealtimeEvent::FileCreated { + file_id: Uuid::nil(), + name: "notes.md".into(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }; + let json = serde_json::to_value(&ev).unwrap(); + assert_eq!(json["event"], "file_created"); + assert_eq!(json["name"], "notes.md"); + + let ev = RealtimeEvent::FolderCreated { + folder_id: Uuid::nil(), + name: "docs".into(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }; + let json = serde_json::to_value(&ev).unwrap(); + assert_eq!(json["event"], "folder_created"); + } + + #[test] + fn event_roundtrip() { + let file_id = Uuid::new_v4(); + let parent_id = Uuid::new_v4(); + let actor = Uuid::new_v4(); + let original = RealtimeEvent::FileCreated { + file_id, + name: "a.txt".into(), + parent_id, + actor, + }; + let json = serde_json::to_string(&original).unwrap(); + let decoded: RealtimeEvent = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded, original); + } + + #[test] + fn error_codes_stay_in_the_defined_ranges() { + // Application-defined codes live in the JSON-RPC "server-defined" + // range -32000..=-32099. Standard envelope codes live in + // -32700..=-32600. A refactor that moves a value out of its range + // is a wire-break — pin it here. + for code in [ + error_code::NO_READ, + error_code::NO_SHARE, + error_code::NO_COMMENT, + error_code::TOPIC_FORBIDDEN, + error_code::SUB_LIMIT, + error_code::RATE_LIMITED, + error_code::NO_EDIT, + ] { + assert!( + (-32099..=-32000).contains(&code), + "app-defined code {code} outside -32099..=-32000" + ); + } + for code in [ + error_code::INTERNAL_ERROR, + error_code::INVALID_REQUEST, + error_code::METHOD_NOT_FOUND, + error_code::INVALID_PARAMS, + ] { + assert!( + (-32700..=-32600).contains(&code), + "standard code {code} outside -32700..=-32600" + ); + } + } + + #[tokio::test] + async fn noop_replicator_parks_until_notified() { + let repl = Arc::new(NoopReplicator); + let shutdown = Arc::new(Notify::new()); + let handle = tokio::spawn({ + let repl = Arc::clone(&repl); + let shutdown = Arc::clone(&shutdown); + async move { BusReplicator::run(repl, shutdown).await } + }); + // on_local_publish is a no-op that should not panic or spawn work. + repl.on_local_publish( + &Topic::Folder(Uuid::nil()), + &RealtimeEvent::FileCreated { + file_id: Uuid::nil(), + name: "x".into(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }, + ); + // `notify_one` (not `notify_waiters`) so the signal survives if the + // spawned task hasn't yet reached `.notified().await` — permit + // queues instead of being dropped. See BusReplicator docs. + shutdown.notify_one(); + handle.await.unwrap().unwrap(); + } +} diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index 9b26acb7..1a10258d 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -56,6 +56,13 @@ pub struct FileUploadService { /// (`create_file_from_owned_blob_with_perms`); `None` in minimal test /// wiring. instant_upload: Option, + /// Realtime message bus. When wired, `upload_file_streaming` + /// publishes a `FileCreated` event on `Topic::Folder(parent_id)` + /// after the DB commit — subscribers see the new file appear in + /// their live folder view. Optional so stub / test factories can + /// build the service without a bus; a `None` bus is a silent no-op + /// on the publish path. + bus: Option>, } /// Everything the instant-upload path needs beyond the upload service's own @@ -78,6 +85,7 @@ impl FileUploadService { resource_access_hook: None, authorization: None, instant_upload: None, + bus: None, } } @@ -95,6 +103,7 @@ impl FileUploadService { resource_access_hook: None, authorization: None, instant_upload: None, + bus: None, } } @@ -108,6 +117,18 @@ impl FileUploadService { self } + /// Wire the realtime message bus. Enables live folder-view updates: + /// after `upload_file_streaming` commits, a `FileCreated` event + /// fires on `Topic::Folder(parent_id)` — subscribers see the new + /// file appear without polling. + pub fn with_realtime_bus( + mut self, + bus: Arc, + ) -> Self { + self.bus = Some(bus); + self + } + /// Wires the authorization engine, dedup index and quota service that /// power the instant-upload path. /// @@ -454,6 +475,27 @@ impl FileUploadUseCase for FileUploadService { // The caller just created this file — surface it in Recent so the // "I just uploaded X" UX matches the pre-SvelteKit behaviour. self.notify_file_accessed(caller_id, &dto.id); + + // Realtime fan-out AFTER commit — subscribers to the parent + // folder's topic see the new file appear live. Silent no-op if + // the bus isn't wired (stubs / tests) or the file landed at + // drive-root (no folder id → nothing to publish on). + if let (Some(bus), Some(parent_folder_id)) = (&self.bus, dto.folder_id.as_deref()) + && let (Ok(parent_uuid), Ok(file_uuid)) = + (Uuid::parse_str(parent_folder_id), Uuid::parse_str(&dto.id)) + { + use crate::application::ports::realtime_ports::{RealtimeEvent, Topic}; + bus.publish( + &Topic::Folder(parent_uuid), + RealtimeEvent::FileCreated { + file_id: file_uuid, + name: dto.name.clone(), + parent_id: parent_uuid, + actor: caller_id, + }, + ); + } + Ok(dto) } diff --git a/src/application/services/folder_service.rs b/src/application/services/folder_service.rs index ac2ed941..34917be4 100644 --- a/src/application/services/folder_service.rs +++ b/src/application/services/folder_service.rs @@ -49,6 +49,13 @@ pub struct FolderService { /// on cross-drive MOVE. Silently skipped when unwired (stubs). storage_usage: Option>, + /// Realtime message bus. When wired, `create_folder_with_perms` + /// publishes a `FolderCreated` event on `Topic::Folder(parent_id)` + /// after the DB commit — subscribers see the new folder appear in + /// their live folder view. Optional so stub / test factories can + /// build the service without a bus; a `None` bus is a silent no-op + /// on the publish path (no fan-out, no audit). + bus: Option>, } impl FolderService { @@ -66,9 +73,21 @@ impl FolderService { file_lifecycle, drive_repo: None, storage_usage: None, + bus: None, } } + /// Wire the realtime message bus. Enables live folder-view updates: + /// after `create_folder_with_perms` commits, a `FolderCreated` event + /// fires on `Topic::Folder(parent_id)`. Off in stubs / tests. + pub fn with_realtime_bus( + mut self, + bus: Arc, + ) -> Self { + self.bus = Some(bus); + self + } + /// Borrow the external-mount classifier (handlers branch on this before /// treating an id as a native UUID). pub fn mount_router(&self) -> &MountRouter { @@ -366,10 +385,39 @@ impl FolderUseCase for FolderService { ) .await?; + // Snapshot the parent UUID before the move so the post-commit + // publish can address `Topic::Folder(parent_uuid)` without + // re-borrowing `dto.parent_id` (which is moved into + // `create_folder`). + let parent_uuid_for_publish = Uuid::parse_str(parent_id).ok(); + let folder = self .folder_storage .create_folder(dto.name, dto.parent_id, caller_id) .await?; + + // Publish AFTER commit — never before, never inside the write. + // Silent no-op if the bus isn't wired (stubs/tests) or the + // parent uuid didn't parse (won't happen — AuthZ above already + // parsed it — but the None-fallthrough keeps the publish path + // infallible). + if let (Some(bus), Some(parent_uuid), Ok(folder_uuid)) = ( + &self.bus, + parent_uuid_for_publish, + Uuid::parse_str(folder.id()), + ) { + use crate::application::ports::realtime_ports::{RealtimeEvent, Topic}; + bus.publish( + &Topic::Folder(parent_uuid), + RealtimeEvent::FolderCreated { + folder_id: folder_uuid, + name: folder.name().to_owned(), + parent_id: parent_uuid, + actor: caller_id, + }, + ); + } + Ok(FolderDto::from(folder)) } diff --git a/src/common/di.rs b/src/common/di.rs index d65377de..b6768800 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -703,7 +703,14 @@ impl AppServiceFactory { resource_access_hook: Option< Arc, >, + bus: &Arc, ) -> ApplicationServices { + // Upcast the concrete bus once — service builders take the + // trait object so the wire remains stable across future bus + // impls. + let bus_trait: Arc = + bus.clone(); + // Main services let folder_service = Arc::new( FolderService::new( @@ -724,7 +731,10 @@ impl AppServiceFactory { // MOVE. Reuses the `check_drive_quota` the upload path // already runs. Without this, a Move that would push the // destination past its cap succeeds silently. - .with_storage_usage(storage_usage.clone()), + .with_storage_usage(storage_usage.clone()) + // Realtime fan-out on `create_folder_with_perms` — the + // parent-folder subscribers see new sub-folders live. + .with_realtime_bus(bus_trait.clone()), ); // Built before the upload/management services so the plugin lifecycle @@ -771,7 +781,11 @@ impl AppServiceFactory { authz.clone(), core.dedup_service.clone(), storage_usage.clone(), - ); + ) + // Realtime fan-out — every successful `upload_file_streaming` + // publishes a `FileCreated` event on the parent folder's + // topic so open folder views refresh live. + .with_realtime_bus(bus_trait.clone()); if let Some(hook) = resource_access_hook.clone() { svc = svc.with_resource_access_hook(hook); } @@ -1775,6 +1789,17 @@ impl AppServiceFactory { crate::application::services::external_mount_router::MountRouter::new(mount_registry), ); + // Realtime bus: single instance for the app lifetime, wired + // with a no-op replicator (multi-instance broker is a follow-up + // per `docs/plan/message-bus.md § Roadmap`). Constructed here + // so `create_application_services` can hand it to services that + // publish after their DB commits (`FolderService`, + // `FileUploadService`, …). Spawns its own GC task in + // `with_replicator` — no supervisor setup required. + let bus = crate::infrastructure::services::in_process_realtime_bus::InProcessRealtimeBus::with_replicator( + Arc::new(crate::application::ports::realtime_ports::NoopReplicator), + ); + let mut apps = self.create_application_services( &core, &repos, @@ -1786,6 +1811,7 @@ impl AppServiceFactory { plugin_dispatch.clone(), mount_router.clone(), Some(resource_access_hook.clone()), + &bus, ); // 5. Share service @@ -2284,6 +2310,7 @@ impl AppServiceFactory { db_pool: Some(pool.clone()), maintenance_pool: Some(maintenance_pool), mount_router, + bus, auth_service: auth_services, opaque_service, opaque_repo, @@ -3177,6 +3204,20 @@ pub struct AppState { /// method (which still owns the authorization check). pub mount_router: Arc, + /// Realtime message bus. Always present — an empty bus (no + /// subscribers, no publishes) costs a single `DashMap` allocation. + /// The WS handler reads `subscribe`; service publish hooks + /// (`FolderService::create_folder_with_perms`, + /// `FileManagementService`'s file-create commit) call `publish` + /// AFTER their DB transaction commits. + /// + /// Stored as the concrete type (not `Arc`) so the + /// GC task's `Weak` lifecycle is legible from di.rs. Consumers + /// that only need the trait obtain it via + /// `Arc::clone(&state.bus) as Arc`. + pub bus: Arc< + crate::infrastructure::services::in_process_realtime_bus::InProcessRealtimeBus, + >, pub auth_service: Option, /// OPAQUE aPAKE substrate (RFC 9807). Populated only when /// [`OpaqueConfig::effective_mode`] is not `Off` — that method diff --git a/src/infrastructure/services/in_process_realtime_bus.rs b/src/infrastructure/services/in_process_realtime_bus.rs new file mode 100644 index 00000000..afd0f7cb --- /dev/null +++ b/src/infrastructure/services/in_process_realtime_bus.rs @@ -0,0 +1,347 @@ +//! In-process `RealtimeBus` — one `broadcast::Sender` per active topic, +//! held in a [`DashMap`] keyed by [`Topic`]. Publish is fire-and-forget, +//! subscribe returns a `Stream` backed by [`BroadcastStream`]. +//! +//! # Slow-subscriber policy +//! +//! `tokio::sync::broadcast` drops the oldest queued message when a +//! subscriber can't keep up (ring is bounded to [`BROADCAST_RING_CAPACITY`]). +//! When a subscriber's stream sees a `Lagged` marker, the WS handler kills +//! that session with a JSON-RPC `rt.revoked` notification (reason +//! `slow_consumer`) and lets the client reconnect + refetch. That policy +//! lives in the handler; this module just surfaces the `Lagged` variant. +//! +//! # Topic GC +//! +//! When the last subscriber of a topic drops, `broadcast::Sender::receiver_count` +//! falls to zero. New publishes on that topic still succeed (they hit the +//! now-orphaned sender), but the entry stays in the map. A background GC +//! task periodically sweeps entries whose `receiver_count == 0`. Kept +//! simple: no ref-count tracking, no watchdog — a sweep every +//! [`GC_INTERVAL`] is enough for our fan-out volume. + +use std::sync::Arc; +use std::time::Duration; + +use dashmap::DashMap; +use futures::StreamExt; +use tokio::sync::broadcast; +use tokio_stream::wrappers::BroadcastStream; + +use crate::application::ports::realtime_ports::{ + BusReplicator, BusStream, RealtimeBus, RealtimeEvent, Topic, +}; + +/// Per-topic ring-buffer size for slow subscribers. When a subscriber lags +/// past this, the broadcast channel starts dropping the oldest messages and +/// signals `Lagged`. Sized generously — fan-out volume per topic is low +/// (folder mutations, job step ticks) so pressure comes from a genuinely +/// dead consumer, not from a normal traffic spike. +pub const BROADCAST_RING_CAPACITY: usize = 256; + +/// How often the GC task sweeps empty topics. Short enough that a burst of +/// short-lived subs (folder navigations) doesn't grow the map indefinitely, +/// long enough that GC overhead stays trivial. +pub const GC_INTERVAL: Duration = Duration::from_secs(60); + +/// The in-process implementation of [`RealtimeBus`]. +/// +/// Callers hold `Arc` (or `Arc`). +/// The struct owns its topic map and — when constructed via +/// [`InProcessRealtimeBus::with_replicator`] — an [`Arc`] +/// that gets fed every local publish for outbound broker forwarding. +pub struct InProcessRealtimeBus { + topics: DashMap>, + replicator: Arc, +} + +impl InProcessRealtimeBus { + /// Construct with a replicator. In v1 that's a + /// [`crate::application::ports::realtime_ports::NoopReplicator`]; when + /// multi-instance ships, it becomes the pg-NOTIFY or broker impl. + /// + /// The GC task holds a [`Weak`] handle so it exits naturally when the + /// last outer `Arc` drops — matches OxiCloud's DI convention that + /// background tasks are dropped on runtime shutdown, no explicit + /// signal needed. + pub fn with_replicator(replicator: Arc) -> Arc { + let bus = Arc::new(Self { + topics: DashMap::new(), + replicator, + }); + bus.spawn_gc(); + bus + } + + /// Spawn the periodic GC task. Holds `Weak` so it does not keep + /// the bus alive past the last outer `Arc` drop; the next + /// upgrade-and-sweep call after that returns `None` and the loop + /// exits. + fn spawn_gc(self: &Arc) { + let weak = Arc::downgrade(self); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(GC_INTERVAL); + // First tick fires immediately; skip it so we don't sweep an + // empty map on startup. + ticker.tick().await; + loop { + ticker.tick().await; + match weak.upgrade() { + Some(bus) => bus.gc_empty_topics(), + None => break, + } + } + }); + } + + /// Remove topics whose broadcast sender has no live receivers. Called + /// on the GC ticker. + fn gc_empty_topics(&self) { + self.topics + .retain(|_topic, sender| sender.receiver_count() > 0); + } + + /// For tests + observability: how many topics currently have a + /// broadcast sender in the map. + pub fn active_topic_count(&self) -> usize { + self.topics.len() + } + + /// Get-or-insert the broadcast sender for `topic`, returning a fresh + /// receiver. Used by both `publish` (for the sender) and `subscribe` + /// (for the receiver) — one code path for the map insert avoids a race + /// where publish creates a sender concurrent subscribers miss. + fn sender_for(&self, topic: &Topic) -> broadcast::Sender { + self.topics + .entry(*topic) + .or_insert_with(|| broadcast::channel(BROADCAST_RING_CAPACITY).0) + .clone() + } +} + +impl RealtimeBus for InProcessRealtimeBus { + fn publish(&self, topic: &Topic, event: RealtimeEvent) { + // Feed the replicator FIRST — if it were called after local fan-out, + // an unwind on a broken subscriber could skip broker forwarding. + // `on_local_publish` is a sync fire-and-forget contract; slow + // replicators must background their I/O themselves. + self.replicator.on_local_publish(topic, &event); + + // If nobody is subscribed, don't allocate a sender just to drop + // its message. `broadcast::Sender::send` returns Err when there + // are no receivers — cheaper still to short-circuit here. + if let Some(sender) = self.topics.get(topic) { + // `send` never blocks; it drops the oldest when the ring is + // full, signalling `Lagged` on that subscriber's next recv. + let _ = sender.send(event); + } + // Else: no active subs. Event is lost by design (see + // `docs/plan/message-bus.md § Failure modes`). + } + + fn subscribe(&self, topic: &Topic) -> BusStream { + let receiver = self.sender_for(topic).subscribe(); + // `BroadcastStream` yields `Result`; + // filter out the `Lagged` variant here and terminate the stream on + // it so the WS handler sees a clean "the stream ended" signal + // rather than having to match on the error. The handler is + // responsible for emitting the `rt.revoked` notification with + // `slow_consumer` reason on such a termination. + let stream = BroadcastStream::new(receiver).take_while(|item| { + let keep = item.is_ok(); + async move { keep } + }); + Box::pin(stream.filter_map(|item| async move { item.ok() })) + } +} + +// ════════════════════════════════════════════════════════════════════════════ +// Tests +// ════════════════════════════════════════════════════════════════════════════ + +#[cfg(test)] +mod tests { + use super::*; + use crate::application::ports::realtime_ports::NoopReplicator; + use futures::StreamExt; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + use tokio::sync::Notify; + use uuid::Uuid; + + fn make_bus() -> Arc { + InProcessRealtimeBus::with_replicator(Arc::new(NoopReplicator)) + } + + fn folder_topic() -> Topic { + Topic::Folder(Uuid::new_v4()) + } + + fn file_created(parent_id: Uuid) -> RealtimeEvent { + RealtimeEvent::FileCreated { + file_id: Uuid::new_v4(), + name: "a.txt".into(), + parent_id, + actor: Uuid::new_v4(), + } + } + + /// Positive fan-out: a subscriber to a topic receives an event + /// published on that same topic. + #[tokio::test] + async fn subscriber_receives_publish_on_same_topic() { + let bus = make_bus(); + let topic = folder_topic(); + let mut stream = bus.subscribe(&topic); + + // Give the subscriber a moment to install (broadcast::Sender::send + // silently fails against a not-yet-installed receiver; the + // subscribe() call above is synchronous but the receiver still + // needs to be registered on the sender's side before publish). + let parent = match topic { + Topic::Folder(id) => id, + _ => unreachable!(), + }; + let event = file_created(parent); + bus.publish(&topic, event.clone()); + + let received = tokio::time::timeout(Duration::from_millis(200), stream.next()) + .await + .expect("event should arrive within 200ms") + .expect("stream must yield at least once"); + assert_eq!(received, event); + } + + /// Topic isolation: a subscriber to folder A does not receive events + /// published on folder B. This is the invariant the smoke test's + /// Scenario 2 asserts end-to-end; verifying it in-unit here catches + /// bugs early. + #[tokio::test] + async fn subscriber_does_not_receive_other_topic() { + let bus = make_bus(); + let topic_a = folder_topic(); + let topic_b = folder_topic(); + assert_ne!(topic_a, topic_b); + + let mut stream_a = bus.subscribe(&topic_a); + + let parent_b = match topic_b { + Topic::Folder(id) => id, + _ => unreachable!(), + }; + let event_b = file_created(parent_b); + bus.publish(&topic_b, event_b); + + // The A subscriber must NOT see B's event. Poll with a short + // timeout — if the isolation is broken we'll see the event; if + // it holds we'll time out. + let result = tokio::time::timeout(Duration::from_millis(100), stream_a.next()).await; + assert!( + result.is_err(), + "subscriber to topic A must not observe events published on topic B \ + (got {:?})", + result.ok().flatten() + ); + } + + /// Multiple subscribers to the same topic all see each publish. + #[tokio::test] + async fn multi_subscriber_fanout() { + let bus = make_bus(); + let topic = folder_topic(); + let mut s1 = bus.subscribe(&topic); + let mut s2 = bus.subscribe(&topic); + + let parent = match topic { + Topic::Folder(id) => id, + _ => unreachable!(), + }; + let event = file_created(parent); + bus.publish(&topic, event.clone()); + + let r1 = tokio::time::timeout(Duration::from_millis(200), s1.next()) + .await + .unwrap() + .unwrap(); + let r2 = tokio::time::timeout(Duration::from_millis(200), s2.next()) + .await + .unwrap() + .unwrap(); + assert_eq!(r1, event); + assert_eq!(r2, event); + } + + /// Publishing with no subscribers is a no-op (does not panic, does not + /// grow the map into an orphan-sender state we later have to sweep). + #[tokio::test] + async fn publish_with_no_subscribers_is_noop() { + let bus = make_bus(); + let topic = folder_topic(); + let parent = match topic { + Topic::Folder(id) => id, + _ => unreachable!(), + }; + bus.publish(&topic, file_created(parent)); + assert_eq!( + bus.active_topic_count(), + 0, + "publish without any subscribe must not insert into topics map" + ); + } + + /// The replicator sees every publish, exactly once per publish. + /// Locks in the "feed replicator FIRST" contract without asserting on + /// broker semantics we don't control from unit-tests. + #[tokio::test] + async fn replicator_is_notified_on_publish() { + struct CountingReplicator { + count: AtomicUsize, + } + #[async_trait::async_trait] + impl BusReplicator for CountingReplicator { + fn on_local_publish(&self, _topic: &Topic, _event: &RealtimeEvent) { + self.count.fetch_add(1, Ordering::SeqCst); + } + async fn run( + self: Arc, + shutdown: Arc, + ) -> Result<(), crate::common::errors::DomainError> { + shutdown.notified().await; + Ok(()) + } + } + + let counter = Arc::new(CountingReplicator { + count: AtomicUsize::new(0), + }); + let bus = InProcessRealtimeBus::with_replicator(Arc::clone(&counter) as Arc<_>); + let topic = folder_topic(); + let _sub = bus.subscribe(&topic); + let parent = match topic { + Topic::Folder(id) => id, + _ => unreachable!(), + }; + bus.publish(&topic, file_created(parent)); + bus.publish(&topic, file_created(parent)); + bus.publish(&topic, file_created(parent)); + assert_eq!(counter.count.load(Ordering::SeqCst), 3); + } + + /// Dropping the last subscriber leaves the topic sender orphaned until + /// the GC sweeps it. We don't wait for the timer here (that would make + /// the test slow); instead we call `gc_empty_topics` directly to + /// verify the sweep does what it promises. + #[tokio::test] + async fn gc_removes_empty_topics() { + let bus = make_bus(); + let topic = folder_topic(); + { + let _sub = bus.subscribe(&topic); + assert_eq!(bus.active_topic_count(), 1); + } + // Subscriber dropped. Sender is still in the map, but receiver + // count is zero — the sweep should reclaim it. + bus.gc_empty_topics(); + assert_eq!(bus.active_topic_count(), 0); + } +} diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 1522429e..1fb189c3 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -27,6 +27,7 @@ pub mod files_consistency_service; pub mod folders_consistency_service; pub mod grant_cleanup_service; pub mod image_transcode_service; +pub mod in_process_realtime_bus; pub mod jwt_service; pub mod last_seen_tracker; pub mod local_blob_backend; diff --git a/src/interfaces/api/handlers/mod.rs b/src/interfaces/api/handlers/mod.rs index a6a945cf..8177ee35 100644 --- a/src/interfaces/api/handlers/mod.rs +++ b/src/interfaces/api/handlers/mod.rs @@ -23,6 +23,7 @@ pub mod opaque_auth_handler; pub mod people_handler; pub mod photos_handler; pub mod recent_handler; +pub mod rt_ws; pub mod search_handler; pub mod share_handler; pub mod subject_group_handler; diff --git a/src/interfaces/api/handlers/rt_ws.rs b/src/interfaces/api/handlers/rt_ws.rs new file mode 100644 index 00000000..839ff66b --- /dev/null +++ b/src/interfaces/api/handlers/rt_ws.rs @@ -0,0 +1,552 @@ +//! Realtime bus WebSocket handler — the endpoint every WS session +//! multiplexes over. See `docs/plan/message-bus.md § Wire protocol`. +//! +//! # Wire +//! +//! JSON-RPC 2.0 for control + events (text frames). Binary frames are +//! reserved for the Yjs sync protocol (collab editor, Phase A follow-up) +//! and are IGNORED in MVP. +//! +//! Methods accepted in MVP: +//! - `rt.subscribe { topic }` → `{ subscribed: "" }` or JSON-RPC error. +//! - `rt.unsubscribe { topic }` → `{ unsubscribed: "" }`. +//! - `rt.ping` → `{ pong: true }`. +//! +//! Server-initiated notifications: +//! - `rt.event { topic, event, data, actor, ts }` — an event published to +//! a topic the caller is subscribed to. +//! +//! # Auth +//! +//! Route sits under `protected_api` (see `src/interfaces/api/routes.rs`) +//! so `auth_middleware` runs first. Cookie AND `Authorization: Bearer` +//! paths both produce a `CurrentUserId` extension the handler extracts. +//! Browser-side subprotocol bearer (`Sec-WebSocket-Protocol: +//! authorization.bearer.`) is a Phase-A follow-up — the MVP relies +//! on the Authorization header, which programmatic clients (the +//! `rt-hurl-helper` smoke test) set directly. +//! +//! # Limits +//! +//! Per-connection outbound `mpsc::Sender` bounded to 512 — full → close. +//! Per-connection subscription cap 128. Frame-size cap not enforced in +//! MVP; the underlying tokio-tungstenite default is 64 MiB which is more +//! than adequate for JSON-RPC control traffic. + +use std::collections::HashMap; +use std::sync::Arc; + +use axum::extract::State; +use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; +use axum::response::Response; +use futures::StreamExt; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use uuid::Uuid; + +use crate::application::ports::authorization_ports::AuthorizationEngine; +use crate::application::ports::realtime_ports::{ + AuthzCheck, BusResource, ParseTopicErr, RealtimeBus, RealtimeEvent, Topic, error_code, +}; +use crate::common::di::AppState; +use crate::domain::services::authorization::{Permission, Resource, Subject}; +use crate::interfaces::middleware::auth::CurrentUserId; + +/// Max simultaneous subscriptions on a single WS session. Beyond this the +/// server responds `-32005 sub_limit` and the client is expected to +/// unsubscribe before subscribing to another topic. +const MAX_SUBSCRIPTIONS_PER_CONNECTION: usize = 128; + +/// Outbound mpsc capacity per session. When full → we close the WS +/// (client reconnects, refetches). Sized so a subscriber blocked on the +/// socket layer doesn't back-pressure into the bus's broadcast ring. +const OUTBOUND_CHANNEL_CAPACITY: usize = 512; + +// ════════════════════════════════════════════════════════════════════════════ +// JSON-RPC 2.0 envelope types +// ════════════════════════════════════════════════════════════════════════════ + +/// Marker constant for the `jsonrpc` field. +const JSONRPC_V2: &str = "2.0"; + +/// Inbound JSON-RPC envelope — deserialize-tolerant so a client can +/// send `rt.ping` without `params`, or a notification without an `id`. +/// +/// Response/notification serialization uses the more strongly-typed +/// [`RpcResponse`] and [`RpcNotification`] types below. +#[derive(Debug, Deserialize)] +struct RpcRequest { + #[serde(rename = "jsonrpc")] + _jsonrpc: Option, + /// `id` is `None` for notifications (which the client-side of MVP + /// never sends). We accept it in the shape but do not treat missing + /// `id` as a permitted request — every `rt.*` method requires + /// `id`-correlated responses in MVP. + id: Option, + method: Option, + #[serde(default)] + params: Value, +} + +/// Server → client response for a request (both success and error use +/// this shape; exactly one of `result`/`error` is populated per spec). +#[derive(Debug, Serialize)] +struct RpcResponse<'a> { + jsonrpc: &'static str, + id: Value, + #[serde(skip_serializing_if = "Option::is_none")] + result: Option, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option>, +} + +#[derive(Debug, Serialize)] +struct RpcError<'a> { + code: i32, + message: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + data: Option, +} + +/// Server → client notification (id-less). Emitted for `rt.event` (bus +/// fan-out) and `rt.revoked` (subscription eviction). +#[derive(Debug, Serialize)] +struct RpcNotification<'a> { + jsonrpc: &'static str, + method: &'a str, + params: Value, +} + +// ════════════════════════════════════════════════════════════════════════════ +// Handler entrypoint +// ════════════════════════════════════════════════════════════════════════════ + +/// `GET /api/rt/ws` — WS upgrade handler. Sits under `protected_api` so +/// [`CurrentUserId`] resolves against a valid session before we reach +/// `on_upgrade`. +/// +/// Returns whatever `WebSocketUpgrade::on_upgrade` produces (an HTTP 101 +/// Switching Protocols with the WebSocket handshake headers). +pub async fn rt_ws_handler( + ws: WebSocketUpgrade, + CurrentUserId(caller_id): CurrentUserId, + State(state): State>, +) -> Response { + ws.on_upgrade(move |socket| handle_session(socket, caller_id, state)) +} + +// ════════════════════════════════════════════════════════════════════════════ +// Session loop +// ════════════════════════════════════════════════════════════════════════════ + +/// A per-topic subscription: the join handle for the reader task that +/// drains the bus stream into `out_tx`. Dropping does NOT abort a spawned +/// tokio task — we must call `.abort()` explicitly on unsubscribe. +struct Sub { + reader: JoinHandle<()>, +} + +impl Drop for Sub { + fn drop(&mut self) { + // Belt-and-braces: if `remove()` bypasses `.abort()` for some + // future call path, dropping still stops the reader. + self.reader.abort(); + } +} + +async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc) { + // Outbound queue — every path that produces a text frame for the + // client enqueues here; the writer half of the select drains. + let (out_tx, mut out_rx) = mpsc::channel::(OUTBOUND_CHANNEL_CAPACITY); + + // Active subscriptions on this session. Keyed by the wire-form topic + // string so an incoming `rt.unsubscribe` with the same string is + // recognised without re-parsing. + let mut subs: HashMap = HashMap::new(); + + loop { + tokio::select! { + // biased: process outbound before inbound so an event burst + // doesn't get overtaken by a control-frame handshake. + biased; + + outbound = out_rx.recv() => { + match outbound { + Some(text) => { + if socket.send(Message::Text(text.into())).await.is_err() { + break; + } + } + None => break, // out_tx dropped — unreachable but safe + } + } + + incoming = socket.recv() => { + match incoming { + Some(Ok(Message::Text(txt))) => { + if let Some(reply) = + handle_text_frame(&txt, caller_id, &state, &mut subs, &out_tx).await + && socket.send(Message::Text(reply.into())).await.is_err() { + break; + } + } + Some(Ok(Message::Binary(_))) => { + // Reserved for Yjs sync protocol frames (collab + // editor, Phase A follow-up). Silently ignored in + // MVP so a future client that speaks binary + // frames on the same connection isn't rejected. + } + Some(Ok(Message::Ping(_) | Message::Pong(_))) => { + // Handled by axum's WebSocket state machine. + } + Some(Ok(Message::Close(_))) | Some(Err(_)) | None => break, + } + } + } + } + + // Session cleanup: abort every subscription reader task. + subs.clear(); +} + +// ════════════════════════════════════════════════════════════════════════════ +// Frame handling +// ════════════════════════════════════════════════════════════════════════════ + +/// Parse one inbound text frame as a JSON-RPC 2.0 request and dispatch +/// it. Returns the response string to send back (empty option if the +/// dispatch already enqueued via `out_tx`). +async fn handle_text_frame( + text: &str, + caller_id: Uuid, + state: &Arc, + subs: &mut HashMap, + out_tx: &mpsc::Sender, +) -> Option { + // Parse envelope. On malformed JSON: reply with an id-less error per + // JSON-RPC 2.0 (id = null when the request couldn't be parsed). + let req: RpcRequest = match serde_json::from_str(text) { + Ok(r) => r, + Err(_) => { + return Some(error_response( + Value::Null, + error_code::INVALID_REQUEST, + "invalid_request", + None, + )); + } + }; + + let id = req.id.unwrap_or(Value::Null); + let Some(method) = req.method else { + return Some(error_response( + id, + error_code::INVALID_REQUEST, + "invalid_request", + None, + )); + }; + + match method.as_str() { + "rt.subscribe" => { + Some(handle_subscribe(id, req.params, caller_id, state, subs, out_tx).await) + } + "rt.unsubscribe" => Some(handle_unsubscribe(id, req.params, subs)), + "rt.ping" => Some(success_response(id, serde_json::json!({ "pong": true }))), + _ => Some(error_response( + id, + error_code::METHOD_NOT_FOUND, + "method_not_found", + Some(serde_json::json!({ "method": method })), + )), + } +} + +async fn handle_subscribe( + id: Value, + params: Value, + caller_id: Uuid, + state: &Arc, + subs: &mut HashMap, + out_tx: &mpsc::Sender, +) -> String { + // Extract topic. + let topic_str = match params.get("topic").and_then(Value::as_str) { + Some(s) => s.to_owned(), + None => { + return error_response( + id, + error_code::INVALID_PARAMS, + "invalid_params", + Some(serde_json::json!({ "missing": "topic" })), + ); + } + }; + + // Guard against runaway subscribers pinning server memory. + if subs.len() >= MAX_SUBSCRIPTIONS_PER_CONNECTION && !subs.contains_key(&topic_str) { + audit_denied(caller_id, &topic_str, "sub_limit"); + return error_response( + id, + error_code::SUB_LIMIT, + "sub_limit", + Some(serde_json::json!({ "topic": topic_str })), + ); + } + + // Idempotent: re-subscribing to an already-active topic acks with + // no side effects. Client reconnect logic can replay its topic set + // without dedup. + if subs.contains_key(&topic_str) { + return success_response(id, serde_json::json!({ "subscribed": topic_str })); + } + + // Parse topic. + let topic = match Topic::parse(&topic_str) { + Ok(t) => t, + Err(ParseTopicErr::BadUuid) | Err(ParseTopicErr::Unknown) => { + // Both parse failures collapse to `topic_forbidden` on the + // wire — the caller cannot distinguish "unknown shape" from + // "shape known but resource doesn't exist" without hinting + // an enumeration oracle. + audit_denied(caller_id, &topic_str, "unknown_topic"); + return error_response( + id, + error_code::TOPIC_FORBIDDEN, + "topic_forbidden", + Some(serde_json::json!({ "topic": topic_str })), + ); + } + }; + + // AuthZ dispatch — one match arm per gate class. Adding a new topic + // variant with a new gate shape is a compile error here. + match topic.required_perm() { + AuthzCheck::ResourceRead { resource } => { + let domain_resource = match resource { + BusResource::Folder(uuid) => Resource::Folder(uuid), + }; + if state + .authorization + .require(Subject::User(caller_id), Permission::Read, domain_resource) + .await + .is_err() + { + // Anti-enum: "no such resource" and "no read" collapse + // to the same wire code. Audit records the truth. + audit_denied(caller_id, &topic_str, "no_read"); + return error_response( + id, + error_code::NO_READ, + "no_read", + Some(serde_json::json!({ "topic": topic_str })), + ); + } + } + AuthzCheck::IdentityMatch { user_id } => { + if user_id != caller_id { + audit_denied(caller_id, &topic_str, "identity_mismatch"); + return error_response( + id, + error_code::TOPIC_FORBIDDEN, + "topic_forbidden", + Some(serde_json::json!({ "topic": topic_str })), + ); + } + } + } + + // AuthZ passed — install the subscription and spawn a reader task + // that forwards bus events to the outbound channel as `rt.event` + // notifications. + let stream = RealtimeBus::subscribe(state.bus.as_ref(), &topic); + let topic_wire = topic_str.clone(); + let out_tx_task = out_tx.clone(); + let reader = tokio::spawn(async move { + let mut stream = stream; + while let Some(event) = stream.next().await { + let notification = event_notification(&topic_wire, &event); + if out_tx_task.send(notification).await.is_err() { + // Session's outbound channel closed — receiver dropped. + break; + } + } + }); + subs.insert(topic_str.clone(), Sub { reader }); + + success_response(id, serde_json::json!({ "subscribed": topic_str })) +} + +fn handle_unsubscribe(id: Value, params: Value, subs: &mut HashMap) -> String { + let Some(topic_str) = params.get("topic").and_then(Value::as_str) else { + return error_response( + id, + error_code::INVALID_PARAMS, + "invalid_params", + Some(serde_json::json!({ "missing": "topic" })), + ); + }; + // Idempotent: removing a topic the session isn't subscribed to is + // still a success ack, per plan. + subs.remove(topic_str); + success_response(id, serde_json::json!({ "unsubscribed": topic_str })) +} + +// ════════════════════════════════════════════════════════════════════════════ +// Envelope helpers +// ════════════════════════════════════════════════════════════════════════════ + +fn success_response(id: Value, result: Value) -> String { + serde_json::to_string(&RpcResponse { + jsonrpc: JSONRPC_V2, + id, + result: Some(result), + error: None, + }) + .expect("RpcResponse always serializes") +} + +fn error_response(id: Value, code: i32, message: &str, data: Option) -> String { + serde_json::to_string(&RpcResponse { + jsonrpc: JSONRPC_V2, + id, + result: None, + error: Some(RpcError { + code, + message, + data, + }), + }) + .expect("RpcResponse always serializes") +} + +/// Build an `rt.event` JSON-RPC notification for a bus event. +/// +/// Payload discipline (see plan): thin facts only. The `RealtimeEvent`'s +/// own `#[serde(tag = "event")]` shape provides `event` + variant fields +/// under one flat object; we lift them into `params.data` alongside a +/// `topic` selector for the client. +fn event_notification(topic_wire: &str, event: &RealtimeEvent) -> String { + // Serialize the event to extract `event` (discriminator) and the + // remaining fields as `data`. Two-step to avoid re-inventing the + // enum's discriminator string here. + let event_json = serde_json::to_value(event).expect("RealtimeEvent always serializes"); + let (event_name, data) = split_event_discriminator(event_json); + + let params = serde_json::json!({ + "topic": topic_wire, + "event": event_name, + "data": data, + }); + + serde_json::to_string(&RpcNotification { + jsonrpc: JSONRPC_V2, + method: "rt.event", + params, + }) + .expect("RpcNotification always serializes") +} + +/// Given a `RealtimeEvent` serialised as `{ "event": "file_created", ...rest }`, +/// split into `(event_name, rest)`. Falls back to `("unknown", full)` if +/// the shape doesn't match (defensive — shouldn't happen given the enum +/// derive, but a future untagged variant would land here). +fn split_event_discriminator(mut event_json: Value) -> (String, Value) { + if let Some(obj) = event_json.as_object_mut() + && let Some(Value::String(name)) = obj.remove("event") + { + return (name, Value::Object(obj.clone())); + } + ("unknown".to_owned(), event_json) +} + +// ════════════════════════════════════════════════════════════════════════════ +// Audit +// ════════════════════════════════════════════════════════════════════════════ + +fn audit_denied(caller_id: Uuid, topic: &str, reason: &'static str) { + tracing::info!( + target: "audit", + event = "realtime.subscribe_denied", + reason = reason, + caller_id = %caller_id, + topic = %topic, + "👮🏻‍♂️ realtime subscribe rejected", + ); +} + +// ════════════════════════════════════════════════════════════════════════════ +// Tests +// ════════════════════════════════════════════════════════════════════════════ + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn success_response_shape() { + let s = success_response( + Value::Number(42.into()), + serde_json::json!({ "subscribed": "folder:x" }), + ); + let v: Value = serde_json::from_str(&s).unwrap(); + assert_eq!(v["jsonrpc"], "2.0"); + assert_eq!(v["id"], 42); + assert_eq!(v["result"]["subscribed"], "folder:x"); + assert!(v.get("error").is_none()); + } + + #[test] + fn error_response_shape() { + let s = error_response( + Value::Number(7.into()), + error_code::NO_READ, + "no_read", + Some(serde_json::json!({ "topic": "folder:x" })), + ); + let v: Value = serde_json::from_str(&s).unwrap(); + assert_eq!(v["jsonrpc"], "2.0"); + assert_eq!(v["id"], 7); + assert_eq!(v["error"]["code"], error_code::NO_READ); + assert_eq!(v["error"]["message"], "no_read"); + assert_eq!(v["error"]["data"]["topic"], "folder:x"); + assert!(v.get("result").is_none()); + } + + #[test] + fn event_notification_shape() { + let event = RealtimeEvent::FileCreated { + file_id: Uuid::nil(), + name: "notes.md".into(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }; + let s = event_notification("folder:abc", &event); + let v: Value = serde_json::from_str(&s).unwrap(); + assert_eq!(v["jsonrpc"], "2.0"); + assert_eq!(v["method"], "rt.event"); + assert_eq!(v["params"]["topic"], "folder:abc"); + assert_eq!(v["params"]["event"], "file_created"); + assert_eq!(v["params"]["data"]["name"], "notes.md"); + // The discriminator field must have been lifted OUT of `data` — + // otherwise the client sees `data.event` alongside the + // top-level `event`, which is confusing and violates the plan's + // wire shape. + assert!(v["params"]["data"].get("event").is_none()); + } + + #[test] + fn split_event_discriminator_extracts_and_removes() { + let input = serde_json::json!({ + "event": "file_created", + "file_id": "00000000-0000-0000-0000-000000000000", + "name": "x", + }); + let (name, rest) = split_event_discriminator(input); + assert_eq!(name, "file_created"); + assert!(rest.get("event").is_none()); + assert_eq!(rest["name"], "x"); + } +} diff --git a/src/interfaces/api/routes.rs b/src/interfaces/api/routes.rs index 10ad9281..2f0e11a2 100644 --- a/src/interfaces/api/routes.rs +++ b/src/interfaces/api/routes.rs @@ -674,6 +674,16 @@ pub fn create_api_routes(app_state: &Arc) -> Router> { .with_state(app_state.clone()); router = router.nest("/users", users_router); + // Realtime bus WebSocket. Auth (session cookie or bearer JWT) via + // the same `auth_middleware` the rest of `/api/*` gets; the handler + // extracts `CurrentUserId` from the extension the middleware + // installs. See `docs/plan/message-bus.md` and the module doc on + // `rt_ws` for the JSON-RPC 2.0 wire. + router = router.route( + "/rt/ws", + get(crate::interfaces::api::handlers::rt_ws::rt_ws_handler).with_state(app_state.clone()), + ); + // Collector for any unknown `/api/*` path. Without this, an // unmatched API URL falls through Axum's matcher to the // ServeDir fallback and is logged under `http::web` — wrong