diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a925ef9e..af9dcc55 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,7 +26,7 @@ jobs: wasm: ${{ steps.filter.outputs.wasm }} plugins: ${{ steps.filter.outputs.plugins }} migrations: ${{ steps.filter.outputs.migrations }} - realtime_spec: ${{ steps.filter.outputs.realtime_spec }} + message_bus_spec: ${{ steps.filter.outputs.message_bus_spec }} steps: - uses: actions/checkout@v4 - uses: dorny/paths-filter@v3 @@ -53,12 +53,12 @@ jobs: - 'src/application/adapters/plugin_user_lifecycle_hook.rs' migrations: - 'migrations/**' - realtime_spec: - - 'src/application/ports/realtime_ports.rs' + message_bus_spec: + - 'src/application/ports/message_bus_ports.rs' - 'src/bin/generate-asyncapi.rs' - 'resources/gen/asyncapi.json' - - 'frontend/scripts/gen-realtime-types.mjs' - - 'frontend/src/lib/generated/realtime/**' + - 'frontend/scripts/gen-message-bus-types.mjs' + - 'frontend/src/lib/generated/message-bus/**' - 'frontend/package.json' frontend-check: @@ -92,18 +92,18 @@ jobs: # scratch, then fails the PR if either output drifts from what was # committed. Same discipline as the OpenAPI + wasm-fixture approach # elsewhere in this file — the wire spec is a compile-time artefact - # of the Rust source (`realtime_ports.rs`), and the TS DTOs are a - # compile-time artefact of the spec, so both must be reproducible. + # of the Rust source (`message_bus_ports.rs`), and the TS DTOs are + # a compile-time artefact of the spec, so both must be reproducible. # - # Scoped by the `realtime_spec` path filter so a PR that doesn't + # Scoped by the `message_bus_spec` path filter so a PR that doesn't # touch the wire (or its generator scripts, or the Modelina version) # skips this job entirely. Needs BOTH Rust and Node toolchains, so # it's slightly heavier than a single-toolchain job — the filter # keeps it off the hot path. - realtime-spec-drift: - name: Realtime spec — AsyncAPI + TypeScript DTO drift + message-bus-spec-drift: + name: Message-bus spec — AsyncAPI + TypeScript DTO drift needs: changes - if: needs.changes.outputs.realtime_spec == 'true' + if: needs.changes.outputs.message_bus_spec == 'true' runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -112,7 +112,7 @@ jobs: - uses: dtolnay/rust-toolchain@stable - uses: Swatinem/rust-cache@v2 # Node for `just asyncapi-ts` — Modelina projects the spec into - # the FE `src/lib/generated/realtime/` folder. + # the FE `src/lib/generated/message-bus/` folder. - name: Setup Node uses: actions/setup-node@v4 with: @@ -127,7 +127,7 @@ jobs: run: cargo run --features dev_tools --bin generate-asyncapi - name: Regenerate TS DTOs (Modelina) working-directory: frontend - run: npm run gen:realtime + run: npm run gen:message-bus - name: Fail if committed files drifted # A non-empty diff means a contributor edited the Rust wire # source (or Modelina config) without regenerating, or hand- @@ -136,9 +136,9 @@ jobs: run: | if ! git diff --exit-code \ resources/gen/asyncapi.json \ - frontend/src/lib/generated/realtime/; then + frontend/src/lib/generated/message-bus/; then echo "" - echo "::error::Realtime spec drift: the committed files differ from what the" + echo "::error::Message-bus spec drift: the committed files differ from what the" echo "::error::generator produces from source. Run \`just asyncapi-ts\` locally" echo "::error::and commit the result — that recipe re-runs both stages." exit 1 diff --git a/Cargo.toml b/Cargo.toml index 049f814e..ecf09c4a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -271,7 +271,7 @@ name = "generate-asyncapi" path = "src/bin/generate-asyncapi.rs" # Dev-only: regenerates `resources/gen/asyncapi.json` — the WS surface's # analogue of openapi.json. Constructed from the same `error_code` -# constants + `Topic`/`RealtimeEvent` shapes the server uses, so the +# constants + `Topic`/`MessageBusEvent` shapes the server uses, so the # spec stays in sync with the implementation by construction. Same # gating as `generate-openapi`. Invoked by `just asyncapi`. required-features = ["dev_tools"] @@ -309,7 +309,7 @@ required-features = ["load_seed_bin"] [[bin]] name = "rt-hurl-helper" path = "src/bin/rt-hurl-helper.rs" -# Test-suite WebSocket client for the realtime message bus. Hurl is +# Test-suite WebSocket client for the message bus. Hurl is # HTTP-only and cannot drive a WS handshake or read frames; this bin # supplies the two modes the smoke test needs — `subscribe-and-collect` # (background subscriber that captures events to JSON) and diff --git a/docs/config/env.md b/docs/config/env.md index 7ca660da..d9b4c385 100644 --- a/docs/config/env.md +++ b/docs/config/env.md @@ -355,7 +355,7 @@ Today's shipped locales: `ar, de, en, es, fa, fr, hi, it, ja, ko, nl, pl, pt, ru Example: `OXICLOUD_TRUST_PROXY_CIDR=127.0.0.1/32,10.0.0.0/8,172.16.0.0/12` -## Realtime WebSocket +## Message bus WebSocket | Variable | Default | Description | |---|---|---| diff --git a/docs/plan/message-bus.md b/docs/plan/message-bus.md index aab0be0f..c58250ae 100644 --- a/docs/plan/message-bus.md +++ b/docs/plan/message-bus.md @@ -1,4 +1,4 @@ -# Plan — Realtime message bus over WebSocket +# Plan — Message bus over WebSocket ## Context @@ -10,7 +10,7 @@ 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 +This plan introduces a single message 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 @@ -41,13 +41,13 @@ almost no extra scaffolding. │ CollabSessionService.apply() ─────────────────▶ bus.publish(...) │ │ │ └───────────────────────────────┬──────────────────────────────────────┘ - │ publish(&Topic, RealtimeEvent) + │ publish(&Topic, MessageBusEvent) ▼ ┌──────────────────────────────────────────────────────────────────────┐ -│ REALTIME BUS (RealtimeBus trait — application/ports) │ +│ MESSAGE BUS (MessageBus trait — application/ports) │ │ │ -│ InProcessRealtimeBus (v1) │ -│ DashMap> │ +│ InProcessMessageBus (v1) │ +│ DashMap> │ │ │ └──────┬───────────────────────────────────────────────────────────────┘ │ @@ -59,7 +59,7 @@ almost no extra scaffolding. │ │ - v2: PgListenReplicator (pg_notify) │ │ │ - v3: BrokerReplicator (RabbitMQ / NATS) │ │ │ │ - │ │ Sits BESIDE InProcessRealtimeBus, forwards │ + │ │ Sits BESIDE InProcessMessageBus, forwards │ │ │ local publishes outbound + inbound events │ │ │ from the broker back into local publish. │ │ └──────────────────────────────────────────────────────┘ @@ -67,7 +67,7 @@ almost no extra scaffolding. ┌──────────────────────────────────────────────────────────────────────┐ │ WS HANDLER (interfaces/api/handlers/rt_ws.rs) │ │ │ -│ One RealtimeSession per WS: HashSet + outbound mpsc │ +│ One BusSession 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 │ @@ -77,12 +77,12 @@ almost no extra scaffolding. **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 +`MessageBus`. A future `BrokerReplicator` publishes outbound + injects inbound. Zero touch to callers. ## Backend components -### 1. Port + event types (`application/ports/realtime_ports.rs`) +### 1. Port + event types (`application/ports/message_bus_ports.rs`) ```rust // Topic is a typed enum, not a string. Prevents typos, gives @@ -127,7 +127,7 @@ pub enum PrincipalRef { #[derive(Serialize, Deserialize)] #[serde(tag = "event", rename_all = "snake_case")] -pub enum RealtimeEvent { +pub enum MessageBusEvent { // 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 }, @@ -168,21 +168,21 @@ pub enum RealtimeEvent { } #[async_trait] -pub trait RealtimeBus: Send + Sync { +pub trait MessageBus: Send + Sync { /// Fire-and-forget. SYNC (not async) — services must not await /// under a DB transaction. - fn publish(&self, topic: &Topic, event: RealtimeEvent); + fn publish(&self, topic: &Topic, event: MessageBusEvent); /// Returns a Stream so the impl can change (broadcast, mpsc, /// pg listener) without churn. - fn subscribe(&self, topic: &Topic) -> Pin + Send>>; + fn subscribe(&self, topic: &Topic) -> Pin + Send>>; } -/// Kept SEPARATE from RealtimeBus so v2/v3 wiring is drop-in. +/// Kept SEPARATE from MessageBus 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); + fn on_local_publish(&self, topic: &Topic, event: &MessageBusEvent); /// Long-running consumer task: reads remote messages and /// re-publishes locally. Started by DI, returns on shutdown. @@ -203,7 +203,7 @@ paths**: | 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 +group-expansion boundary. `MessageBus` only fans out topics that already exist as concrete `user:*` streams. Post-commit sequence for `ShareService::grant(file=F, principal=Group(G), role=R)`: @@ -262,9 +262,9 @@ Coalescing: `NotificationService::create` de-dupes on 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`) +### 2. In-process impl (`infrastructure/services/in_process_message_bus.rs`) -- `DashMap>`, capacity 256 per topic. +- `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 @@ -275,7 +275,7 @@ notification, not two. ### 3. Replicator scaffolding (day-1) - `NoopReplicator` in v1. Wired in DI as `Arc`. -- `InProcessRealtimeBus::publish` calls +- `InProcessMessageBus::publish` calls `replicator.on_local_publish(...)` **after** local fan-out. Futures: @@ -313,7 +313,7 @@ Futures: - 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 + - Spawn `BusSession` actor: owns `HashSet`, outbound `mpsc::Sender` (bounded 512), one reader task per subscribed topic. - Per-frame: @@ -324,7 +324,7 @@ Futures: 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, + `event = "message_bus.subscribe_denied"`. Allow → subscribe on bus, ack. - `unsubscribe`: drop the reader task for that topic. - `ping/pong` for keepalive. @@ -345,14 +345,14 @@ 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 +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`) +### 1. Singleton client (`lib/stores/message-bus.svelte.ts`) - Fetches a ticket via `POST /api/rt/ticket` (through `apiFetch`, so DPoP is applied). @@ -390,9 +390,10 @@ Two wire formats share the same WS connection: - **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). +Method namespace for our JSON-RPC methods: `rt.*` — a short opaque +prefix reserved for message-bus methods. Prevents collisions if we +ever expose additional RPCs on the same WS (not planned, but the +namespace costs nothing). ### JSON-RPC frames (control + events) @@ -511,7 +512,7 @@ implementation by construction — no hand-written spec that drifts. - **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 + the same Rust `MessageBusEvent` 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 @@ -529,8 +530,8 @@ implementation by construction — no hand-written spec that drifts. 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` + spec from `Topic`, `MessageBusEvent`, `AuthzCheck`, and the JSON-RPC + method/error tables — all live in `application/ports/message_bus_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). @@ -618,8 +619,8 @@ it once, avoid hand-maintaining a growing catalog of message types. stays; only the message DTOs come from codegen. - **Wiring:** - `frontend/package.json` dev-dep: `@asyncapi/modelina`. - - Script `frontend/scripts/gen-realtime-types.mjs` invokes Modelina, - writes to `frontend/src/lib/generated/realtime/`. + - Script `frontend/scripts/gen-message-bus-types.mjs` invokes Modelina, + writes to `frontend/src/lib/generated/message-bus/`. - `just asyncapi-ts` recipe alongside `just asyncapi`. - CI dirty-tree check — regenerate on every build, fail if `git diff` on the generated folder is non-empty. Same discipline as @@ -747,10 +748,10 @@ confirms the anti-enumeration collapse rules the wire honours. - **Connect reject** — `event = "auth.rt_ticket_rejected"`, `reason ∈ {expired, unknown, ip_mismatch, replay}`. -- **Subscribe deny** — `event = "realtime.subscribe_denied"`, `reason` +- **Subscribe deny** — `event = "message_bus.subscribe_denied"`, `reason` from the audit column above, plus `caller_id`, `topic`. Emitted BEFORE the wire `denied` frame. -- **Evict** — `event = "realtime.subscription_evicted"`, +- **Evict** — `event = "message_bus.subscription_evicted"`, `reason ∈ {grant_revoked, resource_deleted, admin_kick, group_membership_lost}`, plus `caller_id`, `topic`. - **Collab edit rejected** — `event = "collab.write_denied"`, @@ -796,7 +797,7 @@ larger (notifications table, presence, collab) rides on top later. ### Scope in -- `RealtimeBus` port + `InProcessRealtimeBus`. +- `MessageBus` port + `InProcessMessageBus`. - 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 @@ -954,7 +955,7 @@ subscribe to but didn't. 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, +`message_bus.subscribe_denied` with `reason ∈ {no_read, no_such_resource}`. ``` @@ -1046,7 +1047,7 @@ this baseline once the baseline is green. Ships the infrastructure and the two most visible consumers together. -- Bus port + `InProcessRealtimeBus` + `NoopReplicator` + WS handler +- Bus port + `InProcessMessageBus` + `NoopReplicator` + WS handler + ticket endpoint. - Frontend singleton + `useTopic` composable. - Topics live: `folder:{id}`, `user:{u}:notifications`, `job:{id}`, diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 4ff903a0..f972b1f0 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -9,3 +9,70 @@ Prefix `oxi-`, kebab-case separators. Example: `oxi-view-mode`. Enforced by `$lib/utils/localStoragePrefs::wipeAppKeys()` which sweeps every `oxi-*` key on user-account switches — any other prefix leaks the previous user's state into the new one. + +## Logging — `loglevel` with `oxi:*` namespaces + +**Never use bare `console.debug/info/warn/error` in `$lib` or route code.** +Route through the shared [`loglevel`](https://github.com/pimterry/loglevel) +logger so users and support can dial verbosity per subsystem from the +browser console without a redeploy. + +```ts +import log from 'loglevel'; + +const bus = log.getLogger('oxi:message-bus'); +bus.debug('subscribed', { topic }); +bus.warn('reconnect scheduled', { attempt, backoffMs }); +bus.error('unexpected frame', { raw }); +``` + +Convention: + +- **Namespace = `oxi:`** in kebab-case. One namespace per + subsystem/module boundary — e.g. `oxi:upload` (delta + direct + uploader), `oxi:message-bus` (WS client + `useTopic`). Do not create + finer-grained per-file namespaces; users tune subsystems, not files. +- **Level is user-controlled** via the DevTools helper installed in + `src/hooks.client.ts`: + ```js + oxi.setLogLevel('oxi:message-bus', 'debug'); + oxi.listLogLevels(); + ``` + Choices persist to `localStorage['loglevel:']`. Default is + loglevel's `warn` — production stays quiet unless the user opts in. +- **Add every new namespace to the DevTools comment block** in + `hooks.client.ts` (the `Log levels — namespaces used today: …` line) + so users have a discoverable list. +- **No `console.log` at all** — Stylelint/ESLint don't flag it, but the + codebase convention does. `console.error` is only acceptable in + boot-time paths (`hooks.client.ts`, generator scripts, worker + bootstraps) where the shared logger isn't reachable yet. +- **Workers can't `import log` from a static path** — see + `lib/api/endpoints/deltaUpload.ts`: the worker `postMessage`s a + `{type: 'log', level, msg, extra}` envelope and the main thread relays + it through the shared logger. Mirror this pattern for any new worker. + +## Message bus naming + +The realtime channel is the **message bus** everywhere — backend port +`MessageBus`, plan doc `docs/plan/message-bus.md`, generated DTOs under +`$lib/generated/message-bus/`, FE store/composables named accordingly. +Only two things keep the older `rt`/`Rt` shorthand, and both for wire- +protocol reasons: + +- **JSON-RPC method prefix** — `rt.subscribe`, `rt.event`, `rt.revoked`, + `rt.ping`, `rt.error`. The prefix is opaque wire vocabulary and does + not have to expand to "realtime"; treat it as a short namespace tag + reserved for message-bus methods. +- **Generated type names** — `RtSubscribeParams`, `RtEventBody`, etc. + Modelina keys off the AsyncAPI schema names, which mirror the JSON-RPC + method names. + +When adding FE code around the bus, use `message-bus` in file names, +store names, and logger namespaces: + +- Store: `$lib/stores/message-bus.svelte.ts` +- Composables: `$lib/composables/useTopic.svelte.ts` (topic-generic — no + bus name in the file) +- Logger namespace: `oxi:message-bus` +- localStorage keys (if any): `oxi-message-bus-*` diff --git a/frontend/package.json b/frontend/package.json index 4515fee8..1cf4c204 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -18,7 +18,7 @@ "test:unit": "LANG=C vitest run", "test:unit:watch": "LANG=C vitest", "test:unit:coverage": "rm -rf ../tests/e2e/.nyc_output_unit && LANG=C COVERAGE=1 vitest run", - "gen:realtime": "node scripts/gen-realtime-types.mjs" + "gen:message-bus": "node scripts/gen-message-bus-types.mjs" }, "devDependencies": { "@asyncapi/modelina": "^5.5.0", diff --git a/frontend/scripts/gen-realtime-types.mjs b/frontend/scripts/gen-message-bus-types.mjs similarity index 91% rename from frontend/scripts/gen-realtime-types.mjs rename to frontend/scripts/gen-message-bus-types.mjs index 468c7cf5..1d763aa9 100644 --- a/frontend/scripts/gen-realtime-types.mjs +++ b/frontend/scripts/gen-message-bus-types.mjs @@ -1,5 +1,5 @@ #!/usr/bin/env node -// Realtime bus — TypeScript DTOs generated from `resources/gen/asyncapi.json`. +// Message bus — TypeScript DTOs generated from `resources/gen/asyncapi.json`. // // Sits on the same axis as `resources/gen/openapi.json`: the wire spec // (authored by `cargo run --features dev_tools --bin generate-asyncapi`) @@ -7,14 +7,14 @@ // interfaces so `lib/composables/useTopic.ts` and every folder-view // switch statement is compile-time exhaustive over the `rt.event` variants. // -// Regenerate: `just asyncapi-ts` (or `npm run gen:realtime`). +// Regenerate: `just asyncapi-ts` (or `npm run gen:message-bus`). // CI is expected to run the same command and fail if the working tree is // dirty afterwards — same discipline `just openapi` follows. // // Design notes: // * `modelType: 'interface'` — plain records, not classes-with-getters. // Matches the FE codebase style (see `lib/api/types.ts`). -// * Output goes to `src/lib/generated/realtime/` — a directory reserved +// * Output goes to `src/lib/generated/message-bus/` — a directory reserved // for auto-generated files. Never hand-edit anything inside. // * Every file gets a `AUTO-GENERATED` banner via a preset so a stray // edit is obvious at review time. @@ -32,13 +32,13 @@ import { TypeScriptFileGenerator } from '@asyncapi/modelina'; const execFile = promisify(execFileCb); // Anchor everything on this script's location so `just asyncapi-ts` from -// the repo root and `npm run gen:realtime` from the frontend both work. +// the repo root and `npm run gen:message-bus` from the frontend both work. const __dirname = dirname(fileURLToPath(import.meta.url)); const frontendRoot = resolve(__dirname, '..'); const repoRoot = resolve(frontendRoot, '..'); const specPath = resolve(repoRoot, 'resources/gen/asyncapi.json'); -const outputDir = resolve(frontendRoot, 'src/lib/generated/realtime'); +const outputDir = resolve(frontendRoot, 'src/lib/generated/message-bus'); // Load the spec. Failing here means the wire spec hasn't been generated // yet — hint the operator at the right command. @@ -47,7 +47,7 @@ try { spec = JSON.parse(await readFile(specPath, 'utf8')); } catch (err) { console.error( - `gen-realtime-types: cannot read ${specPath}: ${err.message}\n` + + `gen-message-bus-types: cannot read ${specPath}: ${err.message}\n` + `\nDid you run \`just asyncapi\` first? The Rust generator writes\n` + `resources/gen/asyncapi.json; this script consumes it.` ); @@ -77,7 +77,7 @@ const generator = new TypeScriptFileGenerator({ const banner = '// AUTO-GENERATED — do not edit by hand.\n' + '// Regenerate with `just asyncapi-ts` (which runs\n' + - '// `node frontend/scripts/gen-realtime-types.mjs`).\n' + + '// `node frontend/scripts/gen-message-bus-types.mjs`).\n' + '// Source of truth: resources/gen/asyncapi.json,\n' + '// authored by the Rust `generate-asyncapi` binary.\n'; return `${banner}${content}`; @@ -171,7 +171,7 @@ for (const f of files) { const anonymous = files.filter((f) => f.endsWith('.ts') && /^AnonymousSchema_/i.test(f)); if (anonymous.length > 0) { console.error( - `gen-realtime-types: FAIL — Modelina produced ${anonymous.length} ` + + `gen-message-bus-types: FAIL — Modelina produced ${anonymous.length} ` + `AnonymousSchema_N file(s):` ); for (const f of anonymous) console.error(` - ${f}`); @@ -198,7 +198,7 @@ try { }); } catch (err) { console.error( - `gen-realtime-types: prettier --write failed: ${err.message}\n` + + `gen-message-bus-types: prettier --write failed: ${err.message}\n` + `The generated files may still be usable but will fail\n` + `\`npm run check\` on the prettier step. Fix prettier setup\n` + `(is @prettier installed in frontend/node_modules?) then\n` + @@ -208,7 +208,7 @@ try { } console.log( - `gen-realtime-types: wrote ${models.length} model(s) to ${outputDir}` + + `gen-message-bus-types: wrote ${models.length} model(s) to ${outputDir}` + ` (rewrote ${rewritten} for verbatimModuleSyntax, 0 AnonymousSchema,` + ` prettier-formatted)` ); diff --git a/frontend/src/lib/generated/realtime/FileCreatedData.ts b/frontend/src/lib/generated/message-bus/FileCreatedData.ts similarity index 100% rename from frontend/src/lib/generated/realtime/FileCreatedData.ts rename to frontend/src/lib/generated/message-bus/FileCreatedData.ts diff --git a/frontend/src/lib/generated/realtime/FileDeletedData.ts b/frontend/src/lib/generated/message-bus/FileDeletedData.ts similarity index 100% rename from frontend/src/lib/generated/realtime/FileDeletedData.ts rename to frontend/src/lib/generated/message-bus/FileDeletedData.ts diff --git a/frontend/src/lib/generated/realtime/FileMovedData.ts b/frontend/src/lib/generated/message-bus/FileMovedData.ts similarity index 100% rename from frontend/src/lib/generated/realtime/FileMovedData.ts rename to frontend/src/lib/generated/message-bus/FileMovedData.ts diff --git a/frontend/src/lib/generated/realtime/FileRenamedData.ts b/frontend/src/lib/generated/message-bus/FileRenamedData.ts similarity index 100% rename from frontend/src/lib/generated/realtime/FileRenamedData.ts rename to frontend/src/lib/generated/message-bus/FileRenamedData.ts diff --git a/frontend/src/lib/generated/realtime/Folder.ts b/frontend/src/lib/generated/message-bus/Folder.ts similarity index 100% rename from frontend/src/lib/generated/realtime/Folder.ts rename to frontend/src/lib/generated/message-bus/Folder.ts diff --git a/frontend/src/lib/generated/realtime/FolderCreatedData.ts b/frontend/src/lib/generated/message-bus/FolderCreatedData.ts similarity index 100% rename from frontend/src/lib/generated/realtime/FolderCreatedData.ts rename to frontend/src/lib/generated/message-bus/FolderCreatedData.ts diff --git a/frontend/src/lib/generated/realtime/FolderDeletedData.ts b/frontend/src/lib/generated/message-bus/FolderDeletedData.ts similarity index 100% rename from frontend/src/lib/generated/realtime/FolderDeletedData.ts rename to frontend/src/lib/generated/message-bus/FolderDeletedData.ts diff --git a/frontend/src/lib/generated/realtime/FolderMovedData.ts b/frontend/src/lib/generated/message-bus/FolderMovedData.ts similarity index 100% rename from frontend/src/lib/generated/realtime/FolderMovedData.ts rename to frontend/src/lib/generated/message-bus/FolderMovedData.ts diff --git a/frontend/src/lib/generated/realtime/FolderRenamedData.ts b/frontend/src/lib/generated/message-bus/FolderRenamedData.ts similarity index 100% rename from frontend/src/lib/generated/realtime/FolderRenamedData.ts rename to frontend/src/lib/generated/message-bus/FolderRenamedData.ts diff --git a/frontend/src/lib/generated/realtime/RtErrorCode.ts b/frontend/src/lib/generated/message-bus/RtErrorCode.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtErrorCode.ts rename to frontend/src/lib/generated/message-bus/RtErrorCode.ts diff --git a/frontend/src/lib/generated/realtime/RtErrorMessage.ts b/frontend/src/lib/generated/message-bus/RtErrorMessage.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtErrorMessage.ts rename to frontend/src/lib/generated/message-bus/RtErrorMessage.ts diff --git a/frontend/src/lib/generated/realtime/RtErrorObject.ts b/frontend/src/lib/generated/message-bus/RtErrorObject.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtErrorObject.ts rename to frontend/src/lib/generated/message-bus/RtErrorObject.ts diff --git a/frontend/src/lib/generated/realtime/RtErrorResponseBody.ts b/frontend/src/lib/generated/message-bus/RtErrorResponseBody.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtErrorResponseBody.ts rename to frontend/src/lib/generated/message-bus/RtErrorResponseBody.ts diff --git a/frontend/src/lib/generated/realtime/RtEventKind.ts b/frontend/src/lib/generated/message-bus/RtEventKind.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtEventKind.ts rename to frontend/src/lib/generated/message-bus/RtEventKind.ts diff --git a/frontend/src/lib/generated/realtime/RtEventParams.ts b/frontend/src/lib/generated/message-bus/RtEventParams.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtEventParams.ts rename to frontend/src/lib/generated/message-bus/RtEventParams.ts diff --git a/frontend/src/lib/generated/realtime/RtFolderEventBody.ts b/frontend/src/lib/generated/message-bus/RtFolderEventBody.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtFolderEventBody.ts rename to frontend/src/lib/generated/message-bus/RtFolderEventBody.ts diff --git a/frontend/src/lib/generated/realtime/RtPingRequestBody.ts b/frontend/src/lib/generated/message-bus/RtPingRequestBody.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtPingRequestBody.ts rename to frontend/src/lib/generated/message-bus/RtPingRequestBody.ts diff --git a/frontend/src/lib/generated/realtime/RtPongResponseBody.ts b/frontend/src/lib/generated/message-bus/RtPongResponseBody.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtPongResponseBody.ts rename to frontend/src/lib/generated/message-bus/RtPongResponseBody.ts diff --git a/frontend/src/lib/generated/realtime/RtPongResult.ts b/frontend/src/lib/generated/message-bus/RtPongResult.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtPongResult.ts rename to frontend/src/lib/generated/message-bus/RtPongResult.ts diff --git a/frontend/src/lib/generated/realtime/RtRevokedBody.ts b/frontend/src/lib/generated/message-bus/RtRevokedBody.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtRevokedBody.ts rename to frontend/src/lib/generated/message-bus/RtRevokedBody.ts diff --git a/frontend/src/lib/generated/realtime/RtRevokedParams.ts b/frontend/src/lib/generated/message-bus/RtRevokedParams.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtRevokedParams.ts rename to frontend/src/lib/generated/message-bus/RtRevokedParams.ts diff --git a/frontend/src/lib/generated/realtime/RtRevokedReason.ts b/frontend/src/lib/generated/message-bus/RtRevokedReason.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtRevokedReason.ts rename to frontend/src/lib/generated/message-bus/RtRevokedReason.ts diff --git a/frontend/src/lib/generated/realtime/RtSubscribeParams.ts b/frontend/src/lib/generated/message-bus/RtSubscribeParams.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtSubscribeParams.ts rename to frontend/src/lib/generated/message-bus/RtSubscribeParams.ts diff --git a/frontend/src/lib/generated/realtime/RtSubscribeRequestBody.ts b/frontend/src/lib/generated/message-bus/RtSubscribeRequestBody.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtSubscribeRequestBody.ts rename to frontend/src/lib/generated/message-bus/RtSubscribeRequestBody.ts diff --git a/frontend/src/lib/generated/realtime/RtSuccessResponseBody.ts b/frontend/src/lib/generated/message-bus/RtSuccessResponseBody.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtSuccessResponseBody.ts rename to frontend/src/lib/generated/message-bus/RtSuccessResponseBody.ts diff --git a/frontend/src/lib/generated/realtime/RtUnsubscribeParams.ts b/frontend/src/lib/generated/message-bus/RtUnsubscribeParams.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtUnsubscribeParams.ts rename to frontend/src/lib/generated/message-bus/RtUnsubscribeParams.ts diff --git a/frontend/src/lib/generated/realtime/RtUnsubscribeRequestBody.ts b/frontend/src/lib/generated/message-bus/RtUnsubscribeRequestBody.ts similarity index 100% rename from frontend/src/lib/generated/realtime/RtUnsubscribeRequestBody.ts rename to frontend/src/lib/generated/message-bus/RtUnsubscribeRequestBody.ts diff --git a/justfile b/justfile index 5ef8b434..1608d376 100644 --- a/justfile +++ b/justfile @@ -195,8 +195,8 @@ openapi: cargo run --features dev_tools --bin generate-openapi # Regenerate `resources/gen/asyncapi.json` — the WS surface's spec, -# analogue of openapi.json. Built from the `Topic`, `RealtimeEvent`, -# and `error_code` constants in `application/ports/realtime_ports.rs` +# analogue of openapi.json. Built from the `Topic`, `MessageBusEvent`, +# and `error_code` constants in `application/ports/message_bus_ports.rs` # so the spec stays in sync with the wire by construction. asyncapi: cargo run --features dev_tools --bin generate-asyncapi @@ -210,34 +210,34 @@ asyncapi: # .ts files (idempotent — same input → same output, CI dirty-tree # check catches genuine drift). # -# Output lands in `frontend/src/lib/generated/realtime/`; consumers +# Output lands in `frontend/src/lib/generated/message-bus/`; consumers # import from there but never edit those files. asyncapi-ts: asyncapi - cd frontend && npm run gen:realtime + cd frontend && npm run gen:message-bus -# Local mirror of the `realtime-spec-drift` CI job. Regenerates both -# artefacts and fails if the committed files differ from the fresh -# generator output. Included in `pre-pull-request` so developers -# catch drift BEFORE pushing — the CI job is belt-and-braces, not the -# only defence. +# Local mirror of the `message-bus-spec-drift` CI job. Regenerates +# both artefacts and fails if the committed files differ from the +# fresh generator output. Included in `pre-pull-request` so +# developers catch drift BEFORE pushing — the CI job is +# belt-and-braces, not the only defence. # # Depends on `asyncapi-ts` which itself depends on `asyncapi`, so the # whole chain runs; then we assert on `git diff --exit-code` over # the two paths we care about. -check-realtime-spec: asyncapi-ts +check-message-bus-spec: asyncapi-ts #!/usr/bin/env bash set -euo pipefail if ! git diff --exit-code \ resources/gen/asyncapi.json \ - frontend/src/lib/generated/realtime/; then + frontend/src/lib/generated/message-bus/; then echo "" - echo "❌ realtime spec drift: committed files differ from the fresh" + echo "❌ message-bus spec drift: committed files differ from the fresh" echo " generator output. Fix:" - echo " git add resources/gen/asyncapi.json frontend/src/lib/generated/realtime/" - echo " git commit -m 'chore(rt): regenerate spec + DTOs'" + echo " git add resources/gen/asyncapi.json frontend/src/lib/generated/message-bus/" + echo " git commit -m 'chore(bus): regenerate spec + DTOs'" exit 1 fi - echo "✅ realtime spec: committed files match generator output" + echo "✅ message-bus spec: committed files match generator output" db: docker compose up -d postgres @@ -374,7 +374,7 @@ fe-dev: asyncapi-ts # build the SPA (Phase 0: -> frontend/build; Phase 5: -> static-dist). # `asyncapi-ts` prerequisite (which itself depends on `asyncapi`) -# guarantees `frontend/src/lib/generated/realtime/*.ts` is in sync +# guarantees `frontend/src/lib/generated/message-bus/*.ts` is in sync # with the Rust-side wire spec before Vite compiles — no stale-DTO # window in local dev. CI still runs a dirty-tree check on the # generated files as belt-and-braces. @@ -406,7 +406,7 @@ fe-check: asyncapi-ts cd frontend && npm run check # Vitest unit/component tests. Same asyncapi-ts prereq — tests that -# import from `lib/generated/realtime` need it fresh. +# import from `lib/generated/message-bus` need it fresh. fe-test: asyncapi-ts cd frontend && npm run test:unit @@ -447,4 +447,4 @@ test-docker-tags: # Check and test everything # recommanded before pull request -pre-pull-request: test-docker-tags check fe-check audit check-migrations check-realtime-spec test test-integration fe-test build test-bundle test-api fe-build-e2e front-test +pre-pull-request: test-docker-tags check fe-check audit check-migrations check-message-bus-spec test test-integration fe-test build test-bundle test-api fe-build-e2e front-test diff --git a/src/application/ports/realtime_ports.rs b/src/application/ports/message_bus_ports.rs similarity index 93% rename from src/application/ports/realtime_ports.rs rename to src/application/ports/message_bus_ports.rs index 3e0e2c39..ecdfe6ff 100644 --- a/src/application/ports/realtime_ports.rs +++ b/src/application/ports/message_bus_ports.rs @@ -1,13 +1,13 @@ -//! Realtime message-bus port — the seam every service publishes through and -//! every WS session subscribes on. +//! 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 +//! - [`MessageBus`] 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 +//! [`MessageBus`]; a real replicator plugs into the in-process impl without //! touching consumers. Day-1 impl is [`NoopReplicator`]. //! //! # MVP scope @@ -17,10 +17,10 @@ //! to folders the caller can't `Read`: //! //! - Topics: [`Topic::Folder`] and [`Topic::UserAuthz`] -//! - Events: [`RealtimeEvent::FileCreated`], [`RealtimeEvent::FileRenamed`], -//! [`RealtimeEvent::FileMoved`], [`RealtimeEvent::FileDeleted`], -//! [`RealtimeEvent::FolderCreated`], [`RealtimeEvent::FolderRenamed`], -//! [`RealtimeEvent::FolderMoved`], [`RealtimeEvent::FolderDeleted`] +//! - Events: [`MessageBusEvent::FileCreated`], [`MessageBusEvent::FileRenamed`], +//! [`MessageBusEvent::FileMoved`], [`MessageBusEvent::FileDeleted`], +//! [`MessageBusEvent::FolderCreated`], [`MessageBusEvent::FolderRenamed`], +//! [`MessageBusEvent::FolderMoved`], [`MessageBusEvent::FolderDeleted`] //! //! 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}`, @@ -47,7 +47,7 @@ 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 +/// A topic on the message 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 @@ -157,7 +157,7 @@ pub enum AuthzCheck { } // ════════════════════════════════════════════════════════════════════════════ -// RealtimeEvent — the payload +// MessageBusEvent — the payload // ════════════════════════════════════════════════════════════════════════════ /// A fact that has just become true. Emitted by services AFTER commit, @@ -176,7 +176,7 @@ pub enum AuthzCheck { /// per project convention. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(tag = "event", rename_all = "snake_case")] -pub enum RealtimeEvent { +pub enum MessageBusEvent { /// A file was created inside `parent_id`. FileCreated { file_id: Uuid, @@ -315,7 +315,7 @@ pub mod error_code { } // ════════════════════════════════════════════════════════════════════════════ -// RealtimeBus — the port +// MessageBus — the port // ════════════════════════════════════════════════════════════════════════════ /// The local-facing message bus. Fire-and-forget publish, stream subscribe. @@ -326,11 +326,11 @@ pub mod error_code { /// /// `subscribe` returns a `Stream` so the impl can change (broadcast, mpsc, /// pg listener) without churn at the consumer. -pub trait RealtimeBus: Send + Sync + 'static { +pub trait MessageBus: 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); + fn publish(&self, topic: &Topic, event: MessageBusEvent); /// Subscribe to `topic`. The returned stream yields events until the /// subscriber is dropped or the impl kicks it out (e.g. for lagging @@ -338,15 +338,15 @@ pub trait RealtimeBus: Send + Sync + 'static { fn subscribe(&self, topic: &Topic) -> BusStream; } -/// Boxed stream returned by [`RealtimeBus::subscribe`]. Aliased so +/// Boxed stream returned by [`MessageBus::subscribe`]. Aliased so /// consumers don't need to spell out the `Pin>` shape. -pub type BusStream = Pin + Send>>; +pub type BusStream = Pin + Send>>; // ════════════════════════════════════════════════════════════════════════════ // BusReplicator — the multi-instance seam (day-1 noop) // ════════════════════════════════════════════════════════════════════════════ -/// Cross-instance replicator. Sits BESIDE [`RealtimeBus`], not in front of +/// Cross-instance replicator. Sits BESIDE [`MessageBus`], 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. @@ -358,7 +358,7 @@ 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); + fn on_local_publish(&self, topic: &Topic, event: &MessageBusEvent); /// Long-running consumer task: reads remote messages and re-publishes /// locally. Returns when `shutdown` is notified — DI calls @@ -382,7 +382,7 @@ pub struct NoopReplicator; #[async_trait::async_trait] impl BusReplicator for NoopReplicator { - fn on_local_publish(&self, _topic: &Topic, _event: &RealtimeEvent) { + fn on_local_publish(&self, _topic: &Topic, _event: &MessageBusEvent) { // Intentionally empty. Local fan-out already happened in the bus. } @@ -466,9 +466,9 @@ mod tests { // every variant's discriminator with a snapshot so an accidental // rename fails the test instead of silently breaking clients — // the AsyncAPI spec's `event` enum mirrors these exact strings. - let cases: &[(RealtimeEvent, &str)] = &[ + let cases: &[(MessageBusEvent, &str)] = &[ ( - RealtimeEvent::FileCreated { + MessageBusEvent::FileCreated { file_id: Uuid::nil(), name: "notes.md".into(), parent_id: Uuid::nil(), @@ -477,7 +477,7 @@ mod tests { "file_created", ), ( - RealtimeEvent::FileRenamed { + MessageBusEvent::FileRenamed { file_id: Uuid::nil(), old_name: "a.md".into(), new_name: "b.md".into(), @@ -487,7 +487,7 @@ mod tests { "file_renamed", ), ( - RealtimeEvent::FileMoved { + MessageBusEvent::FileMoved { file_id: Uuid::nil(), name: "a.md".into(), from: Uuid::nil(), @@ -497,7 +497,7 @@ mod tests { "file_moved", ), ( - RealtimeEvent::FileDeleted { + MessageBusEvent::FileDeleted { file_id: Uuid::nil(), parent_id: Uuid::nil(), actor: Uuid::nil(), @@ -505,7 +505,7 @@ mod tests { "file_deleted", ), ( - RealtimeEvent::FolderCreated { + MessageBusEvent::FolderCreated { folder_id: Uuid::nil(), name: "docs".into(), parent_id: Uuid::nil(), @@ -514,7 +514,7 @@ mod tests { "folder_created", ), ( - RealtimeEvent::FolderRenamed { + MessageBusEvent::FolderRenamed { folder_id: Uuid::nil(), old_name: "old".into(), new_name: "new".into(), @@ -524,7 +524,7 @@ mod tests { "folder_renamed", ), ( - RealtimeEvent::FolderMoved { + MessageBusEvent::FolderMoved { folder_id: Uuid::nil(), name: "docs".into(), from: Uuid::nil(), @@ -534,7 +534,7 @@ mod tests { "folder_moved", ), ( - RealtimeEvent::FolderDeleted { + MessageBusEvent::FolderDeleted { folder_id: Uuid::nil(), parent_id: Uuid::nil(), actor: Uuid::nil(), @@ -542,7 +542,7 @@ mod tests { "folder_deleted", ), ( - RealtimeEvent::AuthzChanged { + MessageBusEvent::AuthzChanged { affected_folders: vec![Uuid::nil()], }, "authz_changed", @@ -562,14 +562,14 @@ mod tests { let file_id = Uuid::new_v4(); let parent_id = Uuid::new_v4(); let actor = Uuid::new_v4(); - let original = RealtimeEvent::FileCreated { + let original = MessageBusEvent::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(); + let decoded: MessageBusEvent = serde_json::from_str(&json).unwrap(); assert_eq!(decoded, original); } @@ -618,7 +618,7 @@ mod tests { // on_local_publish is a no-op that should not panic or spawn work. repl.on_local_publish( &Topic::Folder(Uuid::nil()), - &RealtimeEvent::FileCreated { + &MessageBusEvent::FileCreated { file_id: Uuid::nil(), name: "x".into(), parent_id: Uuid::nil(), diff --git a/src/application/ports/mod.rs b/src/application/ports/mod.rs index a246d661..3a2496c7 100644 --- a/src/application/ports/mod.rs +++ b/src/application/ports/mod.rs @@ -18,11 +18,11 @@ pub mod file_lifecycle; pub mod file_ports; pub mod folder_ports; pub mod inbound; +pub mod message_bus_ports; 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/services/file_management_service.rs b/src/application/services/file_management_service.rs index cdad81bf..b004e294 100644 --- a/src/application/services/file_management_service.rs +++ b/src/application/services/file_management_service.rs @@ -57,12 +57,12 @@ pub struct FileManagementService { /// (stub/test builders); production DI wires it in. storage_usage: Option>, - /// Realtime message bus. When wired, delete / rename / move - /// mutations publish their corresponding `RealtimeEvent` on + /// Message bus. When wired, delete / rename / move + /// mutations publish their corresponding `MessageBusEvent` on /// `Topic::Folder(parent_id)` (both source AND destination for /// move) after the DB commit. `None` silently no-ops the publish /// path — same pattern as `bus` on FileUploadService. - bus: Option>, + bus: Option>, /// Read repository — needed by the mutation publish path /// (delete / rename / move) to snapshot the file's pre-mutation /// parent folder BEFORE the write commits: delete removes the row, @@ -102,11 +102,11 @@ impl FileManagementService { } } - /// Wire the realtime message bus. When set, delete / rename / move + /// Wire the message bus. When set, delete / rename / move /// mutations publish on the affected folder topics after commit. - pub fn with_realtime_bus( + pub fn with_message_bus( mut self, - bus: Arc, + bus: Arc, ) -> Self { self.bus = Some(bus); self @@ -208,7 +208,7 @@ impl FileManagementService { } /// Snapshot the (uuid, name, parent-folder-uuid) of a file BEFORE - /// a mutation, so the realtime publish path has a stable + /// a mutation, so the message-bus publish path has a stable /// `Topic::Folder(parent)` to address even after the write commits /// (delete removes the row; move rewrites `folder_id`). /// @@ -239,10 +239,10 @@ impl FileManagementService { /// file, mount, unwired `file_read`). fn publish_file_deleted(&self, caller_id: Uuid, snapshot: Option<(Uuid, String, Uuid)>) { if let (Some(bus), Some((file_uuid, _name, parent_uuid))) = (&self.bus, snapshot) { - use crate::application::ports::realtime_ports::{RealtimeEvent, Topic}; + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; bus.publish( &Topic::Folder(parent_uuid), - RealtimeEvent::FileDeleted { + MessageBusEvent::FileDeleted { file_id: file_uuid, parent_id: parent_uuid, actor: caller_id, @@ -539,7 +539,7 @@ impl FileManagementUseCase for FileManagementService { let dto = self.move_file(file_id, folder_id, caller_id).await?; - // Realtime fan-out on BOTH source and destination folder + // Bus fan-out on BOTH source and destination folder // topics. Subscribers to the source see the file "gone" from // their view; subscribers to the destination see it "appear". // Silent no-op when the bus isn't wired, the source snapshot @@ -552,8 +552,8 @@ impl FileManagementUseCase for FileManagementService { && let Ok(dest_uuid) = Uuid::parse_str(dest_str) && source_uuid != dest_uuid { - use crate::application::ports::realtime_ports::{RealtimeEvent, Topic}; - let event = RealtimeEvent::FileMoved { + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; + let event = MessageBusEvent::FileMoved { file_id: file_uuid, name, from: source_uuid, @@ -674,7 +674,7 @@ impl FileManagementUseCase for FileManagementService { let dto = self.rename_file(file_id, new_name, caller_id).await?; - // Realtime publish AFTER commit. Silent no-op when the bus + // Bus publish AFTER commit. Silent no-op when the bus // isn't wired, the pre-fetch failed (old_name = None), or the // file has no folder (`dto.folder_id = None` — drive-root). if let (Some(bus), Some(old_name), Some(parent_str)) = @@ -682,10 +682,10 @@ impl FileManagementUseCase for FileManagementService { && let (Ok(file_uuid), Ok(parent_uuid)) = (Uuid::parse_str(&dto.id), Uuid::parse_str(parent_str)) { - use crate::application::ports::realtime_ports::{RealtimeEvent, Topic}; + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; bus.publish( &Topic::Folder(parent_uuid), - RealtimeEvent::FileRenamed { + MessageBusEvent::FileRenamed { file_id: file_uuid, old_name, new_name: dto.name.clone(), diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index 1a10258d..1195f73e 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -56,13 +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` + /// 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>, + bus: Option>, } /// Everything the instant-upload path needs beyond the upload service's own @@ -117,13 +117,13 @@ impl FileUploadService { self } - /// Wire the realtime message bus. Enables live folder-view updates: + /// Wire the 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( + pub fn with_message_bus( mut self, - bus: Arc, + bus: Arc, ) -> Self { self.bus = Some(bus); self @@ -476,7 +476,7 @@ impl FileUploadUseCase for FileUploadService { // "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 + // Bus 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). @@ -484,10 +484,10 @@ impl FileUploadUseCase for FileUploadService { && 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}; + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; bus.publish( &Topic::Folder(parent_uuid), - RealtimeEvent::FileCreated { + MessageBusEvent::FileCreated { file_id: file_uuid, name: dto.name.clone(), parent_id: parent_uuid, diff --git a/src/application/services/folder_service.rs b/src/application/services/folder_service.rs index e52723e8..93c4bca2 100644 --- a/src/application/services/folder_service.rs +++ b/src/application/services/folder_service.rs @@ -49,13 +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` + /// 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>, + bus: Option>, } impl FolderService { @@ -77,12 +77,12 @@ impl FolderService { } } - /// Wire the realtime message bus. Enables live folder-view updates: + /// Wire the 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( + pub fn with_message_bus( mut self, - bus: Arc, + bus: Arc, ) -> Self { self.bus = Some(bus); self @@ -406,10 +406,10 @@ impl FolderUseCase for FolderService { parent_uuid_for_publish, Uuid::parse_str(folder.id()), ) { - use crate::application::ports::realtime_ports::{RealtimeEvent, Topic}; + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; bus.publish( &Topic::Folder(parent_uuid), - RealtimeEvent::FolderCreated { + MessageBusEvent::FolderCreated { folder_id: folder_uuid, name: folder.name().to_owned(), parent_id: parent_uuid, @@ -813,7 +813,7 @@ impl FolderUseCase for FolderService { drive_repo.invalidate_default_drive_all(); } - // Realtime publish AFTER commit. Root folders (`parent_id() = None`) + // Bus publish AFTER commit. Root folders (`parent_id() = None`) // have no parent folder topic to publish on — the drive's // display-name change is handled by the readable/default-drive // cache invalidations above, not the bus. Silent no-op if the @@ -822,10 +822,10 @@ impl FolderUseCase for FolderService { && let (Ok(folder_uuid), Ok(parent_uuid)) = (Uuid::parse_str(renamed.id()), Uuid::parse_str(parent_str)) { - use crate::application::ports::realtime_ports::{RealtimeEvent, Topic}; + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; bus.publish( &Topic::Folder(parent_uuid), - RealtimeEvent::FolderRenamed { + MessageBusEvent::FolderRenamed { folder_id: folder_uuid, old_name: folder.name().to_owned(), new_name: renamed.name().to_owned(), @@ -984,7 +984,7 @@ impl FolderUseCase for FolderService { ) })?; - // Realtime fan-out on BOTH source and destination folder + // Bus fan-out on BOTH source and destination folder // topics. Same shape as `FileMoved` — subscribers to either // see the event exactly once. Silent no-op when the bus isn't // wired, the source snapshot failed, or the destination is @@ -995,8 +995,8 @@ impl FolderUseCase for FolderService { (Uuid::parse_str(folder.id()), Uuid::parse_str(dest_str)) && source_uuid != dest_uuid { - use crate::application::ports::realtime_ports::{RealtimeEvent, Topic}; - let event = RealtimeEvent::FolderMoved { + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; + let event = MessageBusEvent::FolderMoved { folder_id: folder_uuid, name: folder.name().to_owned(), from: source_uuid, @@ -1106,15 +1106,15 @@ impl FolderUseCase for FolderService { self.file_lifecycle.on_file_deleted(file_id); } - // Realtime publish AFTER the DELETE commits. Root folders + // Bus publish AFTER the DELETE commits. Root folders // (no parent) can't be deleted through this endpoint per the // mount / drive-root guards above, so `publish_snapshot` is // effectively always Some for regular deletes. if let (Some(bus), Some((folder_uuid, parent_uuid))) = (&self.bus, publish_snapshot) { - use crate::application::ports::realtime_ports::{RealtimeEvent, Topic}; + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; bus.publish( &Topic::Folder(parent_uuid), - RealtimeEvent::FolderDeleted { + MessageBusEvent::FolderDeleted { folder_id: folder_uuid, parent_id: parent_uuid, actor: caller_id, diff --git a/src/bin/generate-asyncapi.rs b/src/bin/generate-asyncapi.rs index 4db396a3..91337f84 100644 --- a/src/bin/generate-asyncapi.rs +++ b/src/bin/generate-asyncapi.rs @@ -1,7 +1,7 @@ -//! AsyncAPI 3.0 spec generator for the realtime message bus. +//! AsyncAPI 3.0 spec generator for the message bus. //! //! Mirrors `generate-openapi.rs`: constructs the spec from the same -//! Rust enums the server uses (`Topic`, `RealtimeEvent`, JSON-RPC +//! Rust enums the server uses (`Topic`, `MessageBusEvent`, JSON-RPC //! error codes) and writes `resources/gen/asyncapi.json`. //! //! This is the first-PR MVP surface — the two topics and two events @@ -24,7 +24,7 @@ use std::fs; use std::path::PathBuf; -use oxicloud::application::ports::realtime_ports::error_code; +use oxicloud::application::ports::message_bus_ports::error_code; use serde_json::{Value, json}; fn main() { @@ -48,7 +48,7 @@ fn build_asyncapi() -> Value { json!({ "asyncapi": "3.0.0", "info": { - "title": "OxiCloud realtime message bus", + "title": "OxiCloud message bus", "version": env!("CARGO_PKG_VERSION"), "description": r#" JSON-RPC 2.0 over WebSocket for control + events, Yjs sync protocol for @@ -68,7 +68,7 @@ Phase C (sync-client push, album live) extend the same channels — see "host": "{host}", "pathname": "/api/rt/ws", "protocol": "wss", - "description": "OxiCloud realtime bus WebSocket endpoint. Text frames are JSON-RPC 2.0. Binary frames (out of AsyncAPI scope) are Yjs sync protocol for the collab editor — see `docs/plan/markdown-collab.md`.", + "description": "OxiCloud message bus WebSocket endpoint. Text frames are JSON-RPC 2.0. Binary frames (out of AsyncAPI scope) are Yjs sync protocol for the collab editor — see `docs/plan/markdown-collab.md`.", "variables": { "host": { "description": "Server host — replace with the deployment domain", @@ -437,7 +437,7 @@ fn rpc_pong_result_schema() -> Value { fn rpc_error_response_schema() -> Value { // The `code`/`message` catalog is the stable public vocabulary — // any change here IS a wire break. Every entry mirrors - // `application/ports/realtime_ports.rs::error_code`. The inner + // `application/ports/message_bus_ports.rs::error_code`. The inner // error object is hoisted to `RtErrorObject` so Modelina emits a // named type instead of `AnonymousSchema_N`. json!({ @@ -533,7 +533,7 @@ fn event_params_schema() -> Value { fn event_kind_schema() -> Value { json!({ "type": "string", - "description": "Discriminator for the `data` payload. Mirrors the `#[serde(tag = \"event\", rename_all = \"snake_case\")]` variants of the Rust `RealtimeEvent` enum — a new event kind is a new enum variant on both sides.", + "description": "Discriminator for the `data` payload. Mirrors the `#[serde(tag = \"event\", rename_all = \"snake_case\")]` variants of the Rust `MessageBusEvent` enum — a new event kind is a new enum variant on both sides.", "enum": [ "file_created", "file_renamed", "file_moved", "file_deleted", "folder_created", "folder_renamed", "folder_moved", "folder_deleted", diff --git a/src/bin/opaque-hurl-helper.rs b/src/bin/opaque-hurl-helper.rs index ed8325f4..de18f0ce 100644 --- a/src/bin/opaque-hurl-helper.rs +++ b/src/bin/opaque-hurl-helper.rs @@ -402,7 +402,7 @@ async fn main() -> ExitCode { Err(e) => return fail(format!("/api/admin/sessions network: {e}")), } - // ── OPAQUE-minted JWT works against the realtime WS ───────────── + // ── OPAQUE-minted JWT works against the WebSocket ───────────── // // Regression guard: `auth_middleware` doesn't inspect how a JWT // was minted, so an OPAQUE-issued access_token must Just Work on diff --git a/src/bin/rt-hurl-helper.rs b/src/bin/rt-hurl-helper.rs index 36bc446b..136efbbf 100644 --- a/src/bin/rt-hurl-helper.rs +++ b/src/bin/rt-hurl-helper.rs @@ -1,4 +1,4 @@ -//! WebSocket-side smoke-test helper for the realtime message bus. +//! WebSocket-side smoke-test helper for the message bus. //! //! Hurl is HTTP-only — it can't do a WS upgrade, let alone read frames //! for later assertion. This binary is the WS half of the smoke test: diff --git a/src/common/di.rs b/src/common/di.rs index 7377bdaa..78b62c2f 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -703,12 +703,12 @@ impl AppServiceFactory { resource_access_hook: Option< Arc, >, - bus: &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 = + let bus_trait: Arc = bus.clone(); // Main services @@ -732,9 +732,9 @@ impl AppServiceFactory { // already runs. Without this, a Move that would push the // destination past its cap succeeds silently. .with_storage_usage(storage_usage.clone()) - // Realtime fan-out on `create_folder_with_perms` — the + // Bus fan-out on `create_folder_with_perms` — the // parent-folder subscribers see new sub-folders live. - .with_realtime_bus(bus_trait.clone()), + .with_message_bus(bus_trait.clone()), ); // Built before the upload/management services so the plugin lifecycle @@ -782,10 +782,10 @@ impl AppServiceFactory { core.dedup_service.clone(), storage_usage.clone(), ) - // Realtime fan-out — every successful `upload_file_streaming` + // Bus 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()); + .with_message_bus(bus_trait.clone()); if let Some(hook) = resource_access_hook.clone() { svc = svc.with_resource_access_hook(hook); } @@ -826,11 +826,11 @@ impl AppServiceFactory { // Destination-drive quota pre-check on cross-drive file // MOVE. Same rationale as the folder side above. .with_storage_usage(storage_usage.clone()) - // Realtime fan-out on delete / rename / move — each hook + // Bus fan-out on delete / rename / move — each hook // publishes on the affected folder topic (move fans out on // BOTH source and destination) so folder-view subscribers // see the mutation live. - .with_realtime_bus(bus_trait.clone()); + .with_message_bus(bus_trait.clone()); if let Some(hook) = resource_access_hook.clone() { svc = svc.with_resource_access_hook(hook); } @@ -1794,15 +1794,15 @@ impl AppServiceFactory { crate::application::services::external_mount_router::MountRouter::new(mount_registry), ); - // Realtime bus: single instance for the app lifetime, wired + // Message 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 bus = crate::infrastructure::services::in_process_message_bus::InProcessMessageBus::with_replicator( + Arc::new(crate::application::ports::message_bus_ports::NoopReplicator), ); let mut apps = self.create_application_services( @@ -3209,19 +3209,19 @@ pub struct AppState { /// method (which still owns the authorization check). pub mount_router: Arc, - /// Realtime message bus. Always present — an empty bus (no + /// 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 + /// 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`. + /// `Arc::clone(&state.bus) as Arc`. pub bus: Arc< - crate::infrastructure::services::in_process_realtime_bus::InProcessRealtimeBus, + crate::infrastructure::services::in_process_message_bus::InProcessMessageBus, >, pub auth_service: Option, /// OPAQUE aPAKE substrate (RFC 9807). Populated only when diff --git a/src/infrastructure/services/in_process_realtime_bus.rs b/src/infrastructure/services/in_process_message_bus.rs similarity index 91% rename from src/infrastructure/services/in_process_realtime_bus.rs rename to src/infrastructure/services/in_process_message_bus.rs index afd0f7cb..e92200d1 100644 --- a/src/infrastructure/services/in_process_realtime_bus.rs +++ b/src/infrastructure/services/in_process_message_bus.rs @@ -1,4 +1,4 @@ -//! In-process `RealtimeBus` — one `broadcast::Sender` per active topic, +//! In-process `MessageBus` — 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`]. //! @@ -28,8 +28,8 @@ use futures::StreamExt; use tokio::sync::broadcast; use tokio_stream::wrappers::BroadcastStream; -use crate::application::ports::realtime_ports::{ - BusReplicator, BusStream, RealtimeBus, RealtimeEvent, Topic, +use crate::application::ports::message_bus_ports::{ + BusReplicator, BusStream, MessageBus, MessageBusEvent, Topic, }; /// Per-topic ring-buffer size for slow subscribers. When a subscriber lags @@ -44,20 +44,20 @@ pub const BROADCAST_RING_CAPACITY: usize = 256; /// long enough that GC overhead stays trivial. pub const GC_INTERVAL: Duration = Duration::from_secs(60); -/// The in-process implementation of [`RealtimeBus`]. +/// The in-process implementation of [`MessageBus`]. /// -/// Callers hold `Arc` (or `Arc`). +/// Callers hold `Arc` (or `Arc`). /// The struct owns its topic map and — when constructed via -/// [`InProcessRealtimeBus::with_replicator`] — an [`Arc`] +/// [`InProcessMessageBus::with_replicator`] — an [`Arc`] /// that gets fed every local publish for outbound broker forwarding. -pub struct InProcessRealtimeBus { - topics: DashMap>, +pub struct InProcessMessageBus { + topics: DashMap>, replicator: Arc, } -impl InProcessRealtimeBus { +impl InProcessMessageBus { /// Construct with a replicator. In v1 that's a - /// [`crate::application::ports::realtime_ports::NoopReplicator`]; when + /// [`crate::application::ports::message_bus_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 @@ -111,7 +111,7 @@ impl InProcessRealtimeBus { /// 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 { + fn sender_for(&self, topic: &Topic) -> broadcast::Sender { self.topics .entry(*topic) .or_insert_with(|| broadcast::channel(BROADCAST_RING_CAPACITY).0) @@ -119,8 +119,8 @@ impl InProcessRealtimeBus { } } -impl RealtimeBus for InProcessRealtimeBus { - fn publish(&self, topic: &Topic, event: RealtimeEvent) { +impl MessageBus for InProcessMessageBus { + fn publish(&self, topic: &Topic, event: MessageBusEvent) { // 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 @@ -162,23 +162,23 @@ impl RealtimeBus for InProcessRealtimeBus { #[cfg(test)] mod tests { use super::*; - use crate::application::ports::realtime_ports::NoopReplicator; + use crate::application::ports::message_bus_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 make_bus() -> Arc { + InProcessMessageBus::with_replicator(Arc::new(NoopReplicator)) } fn folder_topic() -> Topic { Topic::Folder(Uuid::new_v4()) } - fn file_created(parent_id: Uuid) -> RealtimeEvent { - RealtimeEvent::FileCreated { + fn file_created(parent_id: Uuid) -> MessageBusEvent { + MessageBusEvent::FileCreated { file_id: Uuid::new_v4(), name: "a.txt".into(), parent_id, @@ -299,7 +299,7 @@ mod tests { } #[async_trait::async_trait] impl BusReplicator for CountingReplicator { - fn on_local_publish(&self, _topic: &Topic, _event: &RealtimeEvent) { + fn on_local_publish(&self, _topic: &Topic, _event: &MessageBusEvent) { self.count.fetch_add(1, Ordering::SeqCst); } async fn run( @@ -314,7 +314,7 @@ mod tests { let counter = Arc::new(CountingReplicator { count: AtomicUsize::new(0), }); - let bus = InProcessRealtimeBus::with_replicator(Arc::clone(&counter) as Arc<_>); + let bus = InProcessMessageBus::with_replicator(Arc::clone(&counter) as Arc<_>); let topic = folder_topic(); let _sub = bus.subscribe(&topic); let parent = match topic { diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 1fb189c3..bb17125c 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -27,7 +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 in_process_message_bus; pub mod jwt_service; pub mod last_seen_tracker; pub mod local_blob_backend; diff --git a/src/interfaces/api/handlers/grant_handler.rs b/src/interfaces/api/handlers/grant_handler.rs index 93dda8f1..b302fc62 100644 --- a/src/interfaces/api/handlers/grant_handler.rs +++ b/src/interfaces/api/handlers/grant_handler.rs @@ -505,7 +505,7 @@ pub async fn revoke_grant( "🗑️ grant revoked", ); - // Realtime eviction cascade — the revoke committed, so any WS + // Message-bus eviction cascade — the revoke committed, so any WS // session that had the affected user auto-subscribed to // `user:{u}:authz` gets an AuthzChanged event and drops any live // subscriptions to the affected resource. Silent no-op when the @@ -514,11 +514,11 @@ pub async fn revoke_grant( // membership expansion ships). Folder resources only for MVP; // File/Drive topics don't exist yet. if let (Subject::User(target_user), Resource::Folder(folder_id)) = (subject, resource) { - use crate::application::ports::realtime_ports::{RealtimeBus, RealtimeEvent, Topic}; - RealtimeBus::publish( + use crate::application::ports::message_bus_ports::{MessageBus, MessageBusEvent, Topic}; + MessageBus::publish( state.bus.as_ref(), &Topic::UserAuthz(target_user), - RealtimeEvent::AuthzChanged { + MessageBusEvent::AuthzChanged { affected_folders: vec![folder_id], }, ); diff --git a/src/interfaces/api/handlers/rt_ws.rs b/src/interfaces/api/handlers/rt_ws.rs index d96ee7ad..fb8329a5 100644 --- a/src/interfaces/api/handlers/rt_ws.rs +++ b/src/interfaces/api/handlers/rt_ws.rs @@ -1,4 +1,4 @@ -//! Realtime bus WebSocket handler — the endpoint every WS session +//! Message bus WebSocket handler — the endpoint every WS session //! multiplexes over. See `docs/plan/message-bus.md § Wire protocol`. //! //! # Wire @@ -50,8 +50,8 @@ use tokio::time::MissedTickBehavior; 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::application::ports::message_bus_ports::{ + AuthzCheck, BusResource, MessageBus, MessageBusEvent, ParseTopicErr, Topic, error_code, }; use crate::common::di::AppState; use crate::domain::services::authorization::{Permission, Resource, Subject}; @@ -192,7 +192,7 @@ impl Drop for Sub { /// the socket. /// - `EvictFolders` — internal control signal. The reader for the /// session's auto-subscribed `user:{caller}:authz` topic translates -/// inbound [`RealtimeEvent::AuthzChanged`] events into this rather +/// inbound [`MessageBusEvent::AuthzChanged`] events into this rather /// than a client-visible frame. Main loop walks its subs, drops any /// whose resource is in the list, and emits one `rt.revoked` frame /// per evicted topic. @@ -522,7 +522,7 @@ fn handle_unsubscribe(id: Value, params: Value, subs: &mut HashMap) /// /// The reader interprets bus events differently by topic class: /// -/// - For `Topic::UserAuthz(_)`: an incoming `RealtimeEvent::AuthzChanged` +/// - For `Topic::UserAuthz(_)`: an incoming `MessageBusEvent::AuthzChanged` /// is translated to `SessionOut::EvictFolders(affected)` — the main /// loop then walks the sub set and drops matching topics. Any other /// event kind on this topic is ignored (defensive; shouldn't happen @@ -536,7 +536,7 @@ fn install_subscription( state: &Arc, ) { let topic_wire = topic.to_wire_key(); - let mut stream = RealtimeBus::subscribe(state.bus.as_ref(), &topic); + let mut stream = MessageBus::subscribe(state.bus.as_ref(), &topic); let out_tx_task = out_tx.clone(); let translate_authz = matches!(topic, Topic::UserAuthz(_)); // Clone for the reader closure; keep the original to key `subs`. @@ -546,7 +546,7 @@ fn install_subscription( while let Some(event) = stream.next().await { let message = if translate_authz { match event { - RealtimeEvent::AuthzChanged { affected_folders } => { + MessageBusEvent::AuthzChanged { affected_folders } => { SessionOut::EvictFolders(affected_folders) } // The authz topic only carries AuthzChanged in @@ -597,15 +597,15 @@ fn error_response(id: Value, code: i32, message: &str, data: Option) -> S /// Build an `rt.event` JSON-RPC notification for a bus event. /// -/// Payload discipline (see plan): thin facts only. The `RealtimeEvent`'s +/// Payload discipline (see plan): thin facts only. The `MessageBusEvent`'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 { +fn event_notification(topic_wire: &str, event: &MessageBusEvent) -> 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_json = serde_json::to_value(event).expect("MessageBusEvent always serializes"); let (event_name, data) = split_event_discriminator(event_json); let params = serde_json::json!({ @@ -622,7 +622,7 @@ fn event_notification(topic_wire: &str, event: &RealtimeEvent) -> String { .expect("RpcNotification always serializes") } -/// Given a `RealtimeEvent` serialised as `{ "event": "file_created", ...rest }`, +/// Given a `MessageBusEvent` 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). @@ -658,11 +658,11 @@ fn revoked_notification(topic_wire: &str, reason: &'static str) -> String { fn audit_denied(caller_id: Uuid, topic: &str, reason: &'static str) { tracing::info!( target: "audit", - event = "realtime.subscribe_denied", + event = "message_bus.subscribe_denied", reason = reason, caller_id = %caller_id, topic = %topic, - "👮🏻‍♂️ realtime subscribe rejected", + "👮🏻‍♂️ message-bus subscribe rejected", ); } @@ -672,11 +672,11 @@ fn audit_denied(caller_id: Uuid, topic: &str, reason: &'static str) { fn audit_evicted(caller_id: Uuid, topic: &str, reason: &'static str) { tracing::info!( target: "audit", - event = "realtime.subscription_evicted", + event = "message_bus.subscription_evicted", reason = reason, caller_id = %caller_id, topic = %topic, - "🚫 realtime subscription evicted", + "🚫 message-bus subscription evicted", ); } @@ -720,7 +720,7 @@ mod tests { #[test] fn event_notification_shape() { - let event = RealtimeEvent::FileCreated { + let event = MessageBusEvent::FileCreated { file_id: Uuid::nil(), name: "notes.md".into(), parent_id: Uuid::nil(), diff --git a/src/interfaces/api/routes.rs b/src/interfaces/api/routes.rs index 2f0e11a2..5cc5967d 100644 --- a/src/interfaces/api/routes.rs +++ b/src/interfaces/api/routes.rs @@ -674,7 +674,7 @@ 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 + // Message 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 diff --git a/tests/api/rt_bus_check.sh b/tests/api/rt_bus_check.sh index 350d6eb1..06e99018 100755 --- a/tests/api/rt_bus_check.sh +++ b/tests/api/rt_bus_check.sh @@ -1,6 +1,6 @@ #!/usr/bin/env bash # ───────────────────────────────────────────────────────────────────────────── -# Realtime bus smoke test — the parts Hurl can't drive. +# Message bus smoke test — the parts Hurl can't drive. # # Hurl is HTTP-only and cannot open a WebSocket, so the WS half of the test # runs through `rt-hurl-helper` (a small Rust bin gated on `test_utils`). @@ -496,4 +496,4 @@ sub_count=$(jq -r '.subscribed | length' "$out_s8") log "S8 OK" -log "All eight realtime-bus scenarios passed." +log "All eight message-bus scenarios passed." diff --git a/tests/api/run.sh b/tests/api/run.sh index 5c4e073d..6dfe31c9 100755 --- a/tests/api/run.sh +++ b/tests/api/run.sh @@ -267,7 +267,7 @@ bash "$API_DIR/thumb_import_check.sh" bash "$API_DIR/storage_cleanup_check.sh" -# ── 5. Realtime message bus — WebSocket smoke test ────────────────────── +# ── 5. Message bus — WebSocket smoke test ─────────────────────────────── # Runs BEFORE the OPAQUE helper so its user registration + login uses # the legacy password path (opaque_substrate.hurl migrates the admin # account, but by running first this check is unaffected by whatever @@ -275,9 +275,9 @@ bash "$API_DIR/storage_cleanup_check.sh" # positive delivery, topic isolation, AuthZ denial on subscribe, # anti-enumeration parity. See `tests/api/rt_bus_check.sh` and # `docs/plan/message-bus.md`. -log "Running realtime-bus smoke test..." +log "Running message-bus smoke test..." BUILD_TARGET="$BUILD_TARGET" bash "$REPO_ROOT/tests/api/rt_bus_check.sh" \ - || die "realtime-bus smoke test failed" + || die "message-bus smoke test failed" # ── 6. OPAQUE crypto handshake — the parts Hurl can't drive ───────────── # Full OPAQUE register + login handshake against the running server,