From 821f76b471b813f15f08f28cb9a1040aa3f11247 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Fri, 11 Sep 2026 00:59:49 +0200 Subject: [PATCH] feat(msg-bus): wire message bus on frontend --- frontend/AGENTS.md | 16 +- frontend/src/hooks.client.ts | 3 +- .../lib/composables/useFolderTopic.svelte.ts | 96 ++++ .../src/lib/composables/useTopic.svelte.ts | 40 ++ frontend/src/lib/message-bus/client.svelte.ts | 418 ++++++++++++++++++ frontend/src/lib/message-bus/frames.test.ts | 107 +++++ frontend/src/lib/message-bus/frames.ts | 110 +++++ .../src/routes/files/[...path]/+page.svelte | 57 +++ 8 files changed, 842 insertions(+), 5 deletions(-) create mode 100644 frontend/src/lib/composables/useFolderTopic.svelte.ts create mode 100644 frontend/src/lib/composables/useTopic.svelte.ts create mode 100644 frontend/src/lib/message-bus/client.svelte.ts create mode 100644 frontend/src/lib/message-bus/frames.test.ts create mode 100644 frontend/src/lib/message-bus/frames.ts diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index f972b1f0..cc478099 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -71,8 +71,16 @@ protocol reasons: 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` +- Subsystem dir: `$lib/message-bus/` — reactive client (`client.svelte.ts`, + a `MessageBusClient` singleton owning the WebSocket, refcounted topic + subs, and reconnect), frame builders (`frames.ts`), error-code + constants (`error-codes.ts`). Mirrors the `$lib/auth/` and + `$lib/upload/` subsystem-dir pattern rather than living in + `$lib/stores/` — the client is subsystem-scoped plumbing that only + the message-bus composables reach for, not a global reactive store + read from route decisions like `session`. - 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-*` + bus name in the file), `$lib/composables/useFolderTopic.svelte.ts` + (folder-view sugar with per-verb handlers). +- Logger namespace: `oxi:message-bus`. +- localStorage keys (if any): `oxi-message-bus-*`. diff --git a/frontend/src/hooks.client.ts b/frontend/src/hooks.client.ts index 5329bb23..c8c35418 100644 --- a/frontend/src/hooks.client.ts +++ b/frontend/src/hooks.client.ts @@ -14,7 +14,8 @@ import { seedNonceFromCookie } from '$lib/auth/dpop-proof'; // needing to import anything. // // Log levels — namespaces used today: `oxi:upload` (delta + direct -// upload pipeline). Levels: 'trace' | 'debug' | 'info' | 'warn' | 'error' | 'silent'. +// upload pipeline), `oxi:message-bus` (WebSocket client + `useTopic`). +// Levels: 'trace' | 'debug' | 'info' | 'warn' | 'error' | 'silent'. // Choices persist to `localStorage['loglevel:']` via loglevel. // // oxi.setLogLevel('oxi:upload', 'debug') // deep dive diff --git a/frontend/src/lib/composables/useFolderTopic.svelte.ts b/frontend/src/lib/composables/useFolderTopic.svelte.ts new file mode 100644 index 00000000..747bec37 --- /dev/null +++ b/frontend/src/lib/composables/useFolderTopic.svelte.ts @@ -0,0 +1,96 @@ +// Folder-view sugar around `useTopic`. +// +// Discriminates the `rt.event` union at the composable boundary so +// each consumer supplies per-verb handlers with correctly-typed +// payloads. Adding a new event kind in Rust regenerates +// `RtEventKind` — the switch below fails to type-check until every +// arm is handled, keeping the FE exhaustive. + +import { useTopic } from './useTopic.svelte'; +import type RtEventParams from '$lib/generated/message-bus/RtEventParams'; +import type RtRevokedParams from '$lib/generated/message-bus/RtRevokedParams'; +import type FileCreatedData from '$lib/generated/message-bus/FileCreatedData'; +import type FileRenamedData from '$lib/generated/message-bus/FileRenamedData'; +import type FileMovedData from '$lib/generated/message-bus/FileMovedData'; +import type FileDeletedData from '$lib/generated/message-bus/FileDeletedData'; +import type FolderCreatedData from '$lib/generated/message-bus/FolderCreatedData'; +import type FolderRenamedData from '$lib/generated/message-bus/FolderRenamedData'; +import type FolderMovedData from '$lib/generated/message-bus/FolderMovedData'; +import type FolderDeletedData from '$lib/generated/message-bus/FolderDeletedData'; + +/** + * Optional per-verb handlers. Any subset is accepted; unhandled verbs + * are silently ignored. Fires only when the folder view actually cares + * about that kind — leave a handler undefined to opt out. + * + * Callers commonly bind ONE `refresh` function to every handler (see + * `routes/files/[...path]/+page.svelte`) rather than reason about + * surgical mutations — that keeps the folder listing consistent + * with server-side sort/pagination without maintaining a second + * mutation path. + */ +export interface FolderTopicHandlers { + onFileCreated?: (data: FileCreatedData) => void; + onFileRenamed?: (data: FileRenamedData) => void; + onFileMoved?: (data: FileMovedData) => void; + onFileDeleted?: (data: FileDeletedData) => void; + onFolderCreated?: (data: FolderCreatedData) => void; + onFolderRenamed?: (data: FolderRenamedData) => void; + onFolderMoved?: (data: FolderMovedData) => void; + onFolderDeleted?: (data: FolderDeletedData) => void; + /** Grant revoked or folder deleted — the subscription is gone + * server-side. Reasonable UX: toast + navigate away. */ + onRevoked?: (params: RtRevokedParams) => void; +} + +/** + * Subscribe to `folder:{folderId}` and dispatch each `rt.event` + * notification to the matching per-verb handler. + * + * `folderId` accepts the same shapes as `useTopic`'s `topic` — a + * plain string, a nullable string (null = don't subscribe yet), or a + * getter that reads from reactive state (route param) so the + * subscription follows the current folder. + */ +export function useFolderTopic( + folderId: string | null | (() => string | null), + handlers: FolderTopicHandlers +): void { + const topic = () => { + const id = typeof folderId === 'function' ? folderId() : folderId; + return id ? `folder:${id}` : null; + }; + useTopic(topic, (params) => dispatch(params, handlers), handlers.onRevoked); +} + +function dispatch(params: RtEventParams, handlers: FolderTopicHandlers): void { + // The generated `RtEventKind` string-enum values match the Rust + // `#[serde(rename_all = "snake_case")]` variants exactly — see + // `application/ports/message_bus_ports.rs::MessageBusEvent`. + switch (params.event) { + case 'file_created': + handlers.onFileCreated?.(params.data as FileCreatedData); + return; + case 'file_renamed': + handlers.onFileRenamed?.(params.data as FileRenamedData); + return; + case 'file_moved': + handlers.onFileMoved?.(params.data as FileMovedData); + return; + case 'file_deleted': + handlers.onFileDeleted?.(params.data as FileDeletedData); + return; + case 'folder_created': + handlers.onFolderCreated?.(params.data as FolderCreatedData); + return; + case 'folder_renamed': + handlers.onFolderRenamed?.(params.data as FolderRenamedData); + return; + case 'folder_moved': + handlers.onFolderMoved?.(params.data as FolderMovedData); + return; + case 'folder_deleted': + handlers.onFolderDeleted?.(params.data as FolderDeletedData); + return; + } +} diff --git a/frontend/src/lib/composables/useTopic.svelte.ts b/frontend/src/lib/composables/useTopic.svelte.ts new file mode 100644 index 00000000..cf5fe4a9 --- /dev/null +++ b/frontend/src/lib/composables/useTopic.svelte.ts @@ -0,0 +1,40 @@ +// Svelte 5 rune wrapper around `messageBus.subscribe`. +// +// Call from a component's initialisation phase — `$effect` handles the +// mount/unmount lifecycle so the caller never sees the underlying +// WebSocket or the refcount plumbing. Two subscribers of the same +// topic share one wire subscription automatically (refcount lives in +// `MessageBusClient`). +// +// Reactive `topic`: pass a `$derived` or a getter and the composable +// re-subscribes when it changes. Static `topic`: pass a plain string. + +import { messageBus } from '$lib/message-bus/client.svelte'; +import type RtEventParams from '$lib/generated/message-bus/RtEventParams'; +import type RtRevokedParams from '$lib/generated/message-bus/RtRevokedParams'; + +/** + * Subscribe to `topic` for the lifetime of the calling component. + * + * Accepts `topic` as either a plain string or a getter — pass a + * function returning the current topic when it's reactive (e.g. + * derived from a route param) and `$effect` will re-subscribe when + * the returned value changes. A `null` value means "not subscribed + * right now" — useful during route load before the folder id is known. + * + * `onRevoked` fires when the server evicts the subscription (grant + * revoked, folder deleted, etc.); by then the local state is already + * cleared, so the handler can safely re-subscribe or navigate away. + */ +export function useTopic( + topic: string | null | (() => string | null), + onEvent: (params: RtEventParams) => void, + onRevoked?: (params: RtRevokedParams) => void +): void { + $effect(() => { + const resolved = typeof topic === 'function' ? topic() : topic; + if (!resolved) return; + const release = messageBus.subscribe(resolved, onEvent, onRevoked); + return () => release(); + }); +} diff --git a/frontend/src/lib/message-bus/client.svelte.ts b/frontend/src/lib/message-bus/client.svelte.ts new file mode 100644 index 00000000..58d1045c --- /dev/null +++ b/frontend/src/lib/message-bus/client.svelte.ts @@ -0,0 +1,418 @@ +// Message-bus WebSocket client — one connection per tab. +// +// Owns the single `/api/rt/ws` connection, refcounted per-topic +// subscriptions, JSON-RPC request/response correlation, and reconnect +// with jittered exponential backoff. Consumers reach for this through +// the `useTopic` / `useFolderTopic` composables and never see the +// connection directly. +// +// Related files: +// * `frames.ts` — JSON-RPC framing (pure functions). +// * `error-codes.ts` — named constants for `RtErrorObject.code`. +// * `$lib/composables/useTopic.svelte.ts` — per-component lifecycle. +// * `$lib/generated/message-bus/` — wire DTOs (Modelina, auto). +// +// Auth: same-origin WS carries the session cookie automatically. DPoP- +// required deployments need the ticket flow (Phase F, deferred); the +// unauthenticated close is surfaced through `state = 'disconnected'` +// and the console logger so users can diagnose without a redeploy. + +import log from 'loglevel'; + +import { RtErrorCode } from './error-codes'; +import { + parseIncoming, + pingFrame, + subscribeFrame, + unsubscribeFrame, + type IncomingFrame +} from './frames'; +import type RtEventParams from '$lib/generated/message-bus/RtEventParams'; +import type RtRevokedParams from '$lib/generated/message-bus/RtRevokedParams'; + +/** Logger namespace — matches `frontend/AGENTS.md § Logging`. Users + * tune with `oxi.setLogLevel('oxi:message-bus', 'debug')`. */ +const busLog = log.getLogger('oxi:message-bus'); + +/** Reactive connection state. `idle` before the first `subscribe`; + * `connecting` while the handshake is in flight; `connected` once + * the server has accepted the upgrade; `disconnected` after any + * close (reconnect fires from the client). */ +export type ConnectionState = 'idle' | 'connecting' | 'connected' | 'disconnected'; + +/** Callback invoked for every `rt.event` notification on a topic. */ +export type EventHandler = (params: RtEventParams) => void; + +/** Callback invoked when the server sends `rt.revoked` for a topic — + * the subscription is already gone server-side by the time the frame + * arrives; the client removes it from the local refcount map and + * fires this so the consumer can toast / redirect / whatever. */ +export type RevokedHandler = (params: RtRevokedParams) => void; + +/** Handle returned by `subscribe`. Call to release one refcount on the + * topic; the client unsubscribes over the wire only when the last + * refcount drops. Idempotent — calling twice from the same subscriber + * is safe (second call is a no-op). */ +export type UnsubscribeHandle = () => void; + +/** + * Shape returned by a rejected JSON-RPC call. Structurally a superset + * of `RtErrorObject` — every server-side error slots in, and this + * type also lets the client raise synthetic errors (`ws_closed`, + * `send_failed`, `not_connected`) whose `message` is a plain string + * outside the wire's `RtErrorMessage` enum. + */ +export interface MessageBusError { + code: number; + message: string; + data?: unknown; +} + +/** Reconnect backoff — 250 ms doubling with full jitter, capped at 30 s. + * Same shape as the HTTP retry we use in the fetch interceptor. */ +const RECONNECT_MIN_MS = 250; +const RECONNECT_MAX_MS = 30_000; + +interface SubEntry { + count: number; + handlers: Set; + revokedHandlers: Set; + /** True once the server has ack'd `rt.subscribe`. Used by + * reconnect: on wire-up we re-send every already-ack'd topic. */ + acked: boolean; +} + +interface PendingCall { + resolve: (result: unknown) => void; + reject: (error: MessageBusError) => void; +} + +export class MessageBusClient { + /** Reactive connection state — exposed for a debug indicator or + * Playwright test. Not consumed by the composables directly. */ + state = $state('idle'); + /** Last observed round-trip in ms, updated on each `rt.pong`. + * `null` until the first ping completes. */ + latencyMs = $state(null); + + #ws: WebSocket | null = null; + /** Backoff for the NEXT reconnect attempt. Reset to + * `RECONNECT_MIN_MS` on every successful open. */ + #backoffMs = RECONNECT_MIN_MS; + /** setTimeout handle for a scheduled reconnect. Cleared on + * explicit `close()` so we don't reconnect after teardown. */ + #reconnectTimer: ReturnType | null = null; + + /** `topic` → `{count, handlers, revokedHandlers, acked}`. Refcount + * drives the wire: first refcount ⇒ send `rt.subscribe`; last drop + * ⇒ send `rt.unsubscribe`. Plain `Map` (not `SvelteMap`) — this is + * internal plumbing keyed by topic string; a reactive collection + * would re-run every component's `$effect` on unrelated + * subscribes. */ + // eslint-disable-next-line svelte/prefer-svelte-reactivity + #subs = new Map(); + /** Pending JSON-RPC requests keyed by id. Same rationale as + * `#subs` — internal state, not reactive. */ + // eslint-disable-next-line svelte/prefer-svelte-reactivity + #pending = new Map(); + #nextId = 1; + + /** URL for the WebSocket. Injectable so tests can point at a mock. */ + #url: string; + /** WebSocket constructor. Injectable for the same reason. */ + #WebSocketCtor: typeof WebSocket; + + constructor(opts?: { url?: string; WebSocketCtor?: typeof WebSocket }) { + // Default to same-origin `/api/rt/ws`. `location` is unavailable + // in SSR; the client is instantiated lazily on first `subscribe` + // so this executes in the browser. + const defaultUrl = () => { + const proto = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + return `${proto}//${window.location.host}/api/rt/ws`; + }; + this.#url = opts?.url ?? (typeof window !== 'undefined' ? defaultUrl() : ''); + this.#WebSocketCtor = opts?.WebSocketCtor ?? WebSocket; + } + + /** + * Refcounted subscribe. Adds `onEvent` (and optional `onRevoked`) + * to the local handlers for `topic`, sends `rt.subscribe` on the + * wire only for the first ref, and returns an unsubscribe fn that + * drops that same ref (last ref out sends `rt.unsubscribe`). + */ + subscribe(topic: string, onEvent: EventHandler, onRevoked?: RevokedHandler): UnsubscribeHandle { + let entry = this.#subs.get(topic); + if (!entry) { + // Plain Sets: internal callback registries, not reactive. Same + // rationale as `#subs` / `#pending` — see the doc there. + entry = { + count: 0, + // eslint-disable-next-line svelte/prefer-svelte-reactivity + handlers: new Set(), + // eslint-disable-next-line svelte/prefer-svelte-reactivity + revokedHandlers: new Set(), + acked: false + }; + this.#subs.set(topic, entry); + } + entry.count += 1; + entry.handlers.add(onEvent); + if (onRevoked) entry.revokedHandlers.add(onRevoked); + + // Kick the connection if nothing is holding it yet, otherwise + // send `rt.subscribe` if this is the first ref on this topic. + if (this.state === 'idle' || this.state === 'disconnected') { + this.#connect(); + } else if (entry.count === 1 && this.state === 'connected') { + this.#sendSubscribe(topic).catch((err) => + busLog.warn('subscribe failed', { topic, error: err }) + ); + } + + let released = false; + return () => { + if (released) return; + released = true; + this.#releaseOne(topic, onEvent, onRevoked); + }; + } + + /** Force a fresh reconnect — for a live-updates toggle or a manual + * "reconnect" button. Rare; not part of the normal flow. */ + reconnect(): void { + if (this.#ws) this.#ws.close(); + this.#backoffMs = RECONNECT_MIN_MS; + this.#scheduleReconnect(0); + } + + /** Tear down. Currently only meaningful in tests — the singleton + * lives for the lifetime of the tab. */ + close(): void { + if (this.#reconnectTimer !== null) { + clearTimeout(this.#reconnectTimer); + this.#reconnectTimer = null; + } + if (this.#ws) { + this.#ws.close(); + this.#ws = null; + } + this.state = 'idle'; + this.#subs.clear(); + this.#pending.clear(); + } + + // ─────────────────────── connection lifecycle ──────────────────── + + #connect(): void { + if (this.state === 'connecting' || this.state === 'connected') return; + this.state = 'connecting'; + busLog.debug('connecting', { url: this.#url }); + let ws: WebSocket; + try { + ws = new this.#WebSocketCtor(this.#url); + } catch (err) { + busLog.warn('WebSocket ctor threw — reconnect scheduled', { error: err }); + this.state = 'disconnected'; + this.#scheduleReconnect(); + return; + } + this.#ws = ws; + ws.onopen = () => this.#onOpen(); + ws.onmessage = (ev) => this.#onMessage(ev); + ws.onerror = (ev) => busLog.debug('ws error event', { ev }); + ws.onclose = (ev) => this.#onClose(ev); + } + + #onOpen(): void { + busLog.debug('connected'); + this.state = 'connected'; + this.#backoffMs = RECONNECT_MIN_MS; + // Replay every already-known topic. `entry.acked` is reset here + // because the fresh connection has no server-side memory of + // prior subscriptions. + for (const [topic, entry] of this.#subs) { + entry.acked = false; + this.#sendSubscribe(topic).catch((err) => + busLog.warn('resubscribe failed', { topic, error: err }) + ); + } + } + + #onMessage(ev: MessageEvent): void { + if (typeof ev.data !== 'string') { + // Binary frames are the Yjs sync protocol (Phase G) — not in + // scope yet. Silently drop; a future collab store will + // receive them via a separate handler. + busLog.debug('binary frame dropped (Phase G)'); + return; + } + const frame = parseIncoming(ev.data); + this.#dispatch(frame); + } + + #dispatch(frame: IncomingFrame): void { + switch (frame.kind) { + case 'event': { + const entry = this.#subs.get(frame.params.topic); + if (!entry) { + busLog.debug('event for unknown topic', { topic: frame.params.topic }); + return; + } + for (const handler of entry.handlers) { + try { + handler(frame.params); + } catch (err) { + busLog.warn('event handler threw', { topic: frame.params.topic, error: err }); + } + } + break; + } + case 'revoked': { + const entry = this.#subs.get(frame.params.topic); + if (!entry) { + busLog.debug('revoked for unknown topic', { topic: frame.params.topic }); + return; + } + busLog.warn('subscription revoked', { + topic: frame.params.topic, + reason: frame.params.reason + }); + // Server-side sub is already gone; drop local state + // BEFORE firing consumer handlers so any handler that + // re-subscribes gets a fresh entry with `count = 1`. + const revokedHandlers = [...entry.revokedHandlers]; + this.#subs.delete(frame.params.topic); + for (const handler of revokedHandlers) { + try { + handler(frame.params); + } catch (err) { + busLog.warn('revoked handler threw', { topic: frame.params.topic, error: err }); + } + } + break; + } + case 'success': { + const pending = this.#pending.get(frame.id); + if (!pending) return; + this.#pending.delete(frame.id); + pending.resolve(frame.result); + break; + } + case 'error': { + busLog.warn('rt.error', { id: frame.id, error: frame.error }); + if (frame.id === null) return; + const pending = this.#pending.get(frame.id); + if (!pending) return; + this.#pending.delete(frame.id); + pending.reject(frame.error); + break; + } + case 'ignore': { + busLog.warn('ignored frame', { reason: frame.reason, raw: frame.raw }); + break; + } + } + } + + #onClose(ev: CloseEvent): void { + busLog.debug('close', { code: ev.code, reason: ev.reason }); + this.#ws = null; + this.state = 'disconnected'; + // Reject every pending call — the caller sees a synthetic + // error rather than hanging. Reconnect will re-issue the + // subscribe via `#onOpen`, not by resolving these. + const closed: MessageBusError = { code: RtErrorCode.INTERNAL_ERROR, message: 'ws_closed' }; + for (const pending of this.#pending.values()) pending.reject(closed); + this.#pending.clear(); + // Only reconnect if we still have subscribers waiting. + if (this.#subs.size > 0) this.#scheduleReconnect(); + } + + #scheduleReconnect(overrideMs?: number): void { + if (this.#reconnectTimer !== null) return; + const delay = overrideMs ?? this.#backoffMs; + // Full jitter — random in [0, backoff]. Prevents thundering + // herd if the server was momentarily overloaded. + const jittered = Math.floor(Math.random() * (delay + 1)); + busLog.warn('reconnect scheduled', { attemptBackoffMs: delay, jitteredMs: jittered }); + this.#reconnectTimer = setTimeout(() => { + this.#reconnectTimer = null; + this.#backoffMs = Math.min(this.#backoffMs * 2, RECONNECT_MAX_MS); + this.#connect(); + }, jittered); + } + + // ─────────────────────── request/response ──────────────────────── + + #sendSubscribe(topic: string): Promise { + return this.#call((id) => subscribeFrame(id, topic)).then((result) => { + const entry = this.#subs.get(topic); + if (entry) entry.acked = true; + busLog.debug('subscribed', { topic, result }); + }); + } + + #sendUnsubscribe(topic: string): Promise { + // Fire-and-forget — the server accepts idempotently. Not chained + // on the promise because by the time we send this the caller + // has already cleaned up its local state. + return this.#call((id) => unsubscribeFrame(id, topic)).then(() => { + busLog.debug('unsubscribed', { topic }); + }); + } + + /** Public latency probe. Sends `rt.ping` and updates `latencyMs`. */ + async ping(): Promise { + const started = performance.now(); + await this.#call((id) => pingFrame(id)); + const elapsed = Math.round(performance.now() - started); + this.latencyMs = elapsed; + return elapsed; + } + + #call(makeFrame: (id: number) => object): Promise { + if (this.state !== 'connected' || !this.#ws) { + const err: MessageBusError = { + code: RtErrorCode.INTERNAL_ERROR, + message: 'not_connected' + }; + return Promise.reject(err); + } + const id = this.#nextId++; + const frame = makeFrame(id); + return new Promise((resolve, reject) => { + this.#pending.set(id, { resolve, reject }); + try { + this.#ws!.send(JSON.stringify(frame)); + } catch (err) { + this.#pending.delete(id); + busLog.warn('send failed', { id, error: err }); + reject({ code: RtErrorCode.INTERNAL_ERROR, message: 'send_failed' }); + } + }); + } + + // ─────────────────────── refcount teardown ──────────────────────── + + #releaseOne(topic: string, onEvent: EventHandler, onRevoked?: RevokedHandler): void { + const entry = this.#subs.get(topic); + if (!entry) return; + entry.handlers.delete(onEvent); + if (onRevoked) entry.revokedHandlers.delete(onRevoked); + entry.count -= 1; + if (entry.count > 0) return; + this.#subs.delete(topic); + if (this.state === 'connected' && entry.acked) { + void this.#sendUnsubscribe(topic).catch(() => { + // Server drops idempotently; nothing to do if it errors. + }); + } + } +} + +/** + * Process-wide singleton — one WebSocket per tab. Lazy: nothing opens + * until the first `subscribe`. Exported for `useTopic` to consume; + * app code should reach for the composables instead. + */ +export const messageBus = new MessageBusClient(); diff --git a/frontend/src/lib/message-bus/frames.test.ts b/frontend/src/lib/message-bus/frames.test.ts new file mode 100644 index 00000000..75bd84d1 --- /dev/null +++ b/frontend/src/lib/message-bus/frames.test.ts @@ -0,0 +1,107 @@ +import { describe, it, expect } from 'vitest'; +import { parseIncoming, pingFrame, subscribeFrame, unsubscribeFrame } from './frames'; + +describe('frame builders', () => { + it('subscribeFrame produces a valid JSON-RPC 2.0 request', () => { + expect(subscribeFrame(7, 'folder:abc')).toEqual({ + jsonrpc: '2.0', + id: 7, + method: 'rt.subscribe', + params: { topic: 'folder:abc' } + }); + }); + + it('unsubscribeFrame mirrors the subscribe shape', () => { + expect(unsubscribeFrame(8, 'folder:abc')).toEqual({ + jsonrpc: '2.0', + id: 8, + method: 'rt.unsubscribe', + params: { topic: 'folder:abc' } + }); + }); + + it('pingFrame omits params entirely (matches the wire spec)', () => { + const frame = pingFrame(9); + expect(frame).toEqual({ jsonrpc: '2.0', id: 9, method: 'rt.ping' }); + expect('params' in frame).toBe(false); + }); +}); + +describe('parseIncoming', () => { + it('recognises an `rt.event` notification', () => { + const raw = JSON.stringify({ + jsonrpc: '2.0', + method: 'rt.event', + params: { + topic: 'folder:abc', + event: 'file_created', + data: { file_id: 'x', name: 'a.txt', parent_id: 'abc', actor: 'me' } + } + }); + const result = parseIncoming(raw); + expect(result.kind).toBe('event'); + if (result.kind === 'event') { + expect(result.params.topic).toBe('folder:abc'); + expect(result.params.event).toBe('file_created'); + } + }); + + it('recognises an `rt.revoked` notification', () => { + const raw = JSON.stringify({ + jsonrpc: '2.0', + method: 'rt.revoked', + params: { topic: 'folder:abc', reason: 'grant_revoked' } + }); + const result = parseIncoming(raw); + expect(result.kind).toBe('revoked'); + if (result.kind === 'revoked') expect(result.params.topic).toBe('folder:abc'); + }); + + it('recognises a success response', () => { + const raw = JSON.stringify({ + jsonrpc: '2.0', + id: 42, + result: { subscribed: 'folder:abc' } + }); + const result = parseIncoming(raw); + expect(result.kind).toBe('success'); + if (result.kind === 'success') { + expect(result.id).toBe(42); + expect(result.result).toEqual({ subscribed: 'folder:abc' }); + } + }); + + it('recognises an error response and preserves the code', () => { + const raw = JSON.stringify({ + jsonrpc: '2.0', + id: 42, + error: { code: -32001, message: 'no_read', data: { topic: 'folder:xyz' } } + }); + const result = parseIncoming(raw); + expect(result.kind).toBe('error'); + if (result.kind === 'error') { + expect(result.id).toBe(42); + expect(result.error.code).toBe(-32001); + expect(result.error.message).toBe('no_read'); + } + }); + + it('collapses malformed frames to `ignore` with a stable reason key', () => { + expect(parseIncoming('not-json').kind).toBe('ignore'); + expect(parseIncoming('[]').kind).toBe('ignore'); + expect(parseIncoming(JSON.stringify({ jsonrpc: '1.0', method: 'rt.event' })).kind).toBe( + 'ignore' + ); + expect(parseIncoming(JSON.stringify({ jsonrpc: '2.0', method: 'rt.unknown' })).kind).toBe( + 'ignore' + ); + }); + + it('never throws — always returns a discriminated result', () => { + // Random shapes that used to trigger throws in earlier drafts. + const cases: string[] = ['', 'null', '42', '{}', '{"jsonrpc":"2.0"}']; + for (const c of cases) { + expect(() => parseIncoming(c)).not.toThrow(); + } + }); +}); diff --git a/frontend/src/lib/message-bus/frames.ts b/frontend/src/lib/message-bus/frames.ts new file mode 100644 index 00000000..800a1988 --- /dev/null +++ b/frontend/src/lib/message-bus/frames.ts @@ -0,0 +1,110 @@ +// JSON-RPC 2.0 frame builders + parsers for the message bus. +// +// Pure functions — no I/O, no state, no side effects. Sits between +// `client.svelte.ts` (owns the WebSocket + subscription refcounts) and +// the generated wire DTOs under `$lib/generated/message-bus/`. Keeping +// the framing logic isolated makes it directly unit-testable and keeps +// `client.svelte.ts` focused on lifecycle. +// +// One-way import direction: this file reads from `$lib/generated/…`; +// nothing under `generated/` imports from here. + +import type RtSubscribeRequestBody from '$lib/generated/message-bus/RtSubscribeRequestBody'; +import type RtUnsubscribeRequestBody from '$lib/generated/message-bus/RtUnsubscribeRequestBody'; +import type RtPingRequestBody from '$lib/generated/message-bus/RtPingRequestBody'; +import type RtEventParams from '$lib/generated/message-bus/RtEventParams'; +import type RtRevokedParams from '$lib/generated/message-bus/RtRevokedParams'; +import type RtErrorObject from '$lib/generated/message-bus/RtErrorObject'; + +/** + * Discriminated result of parsing one text frame off the wire. + * + * A well-formed frame lands as `event`, `revoked`, `success`, or + * `error`. Anything the client should silently drop (a malformed + * payload, an unknown notification method, a frame with the wrong + * `jsonrpc` version) collapses to `ignore` with a `reason` so the + * logger can surface it at `warn` without the caller having to + * distinguish. + */ +export type IncomingFrame = + | { kind: 'event'; params: RtEventParams } + | { kind: 'revoked'; params: RtRevokedParams } + | { kind: 'success'; id: number; result: unknown } + | { kind: 'error'; id: number | null; error: RtErrorObject } + | { kind: 'ignore'; reason: string; raw: unknown }; + +/** JSON-RPC subscribe request. `id` correlates the eventual success/error. */ +export function subscribeFrame(id: number, topic: string): RtSubscribeRequestBody { + return { + jsonrpc: '2.0', + id, + method: 'rt.subscribe', + params: { topic } + }; +} + +/** JSON-RPC unsubscribe request. */ +export function unsubscribeFrame(id: number, topic: string): RtUnsubscribeRequestBody { + return { + jsonrpc: '2.0', + id, + method: 'rt.unsubscribe', + params: { topic } + }; +} + +/** JSON-RPC application-level ping. The server also issues protocol-level + * RFC 6455 Pings on its own timer (keepalive); this request is available + * for the client to probe round-trip latency on demand. */ +export function pingFrame(id: number): RtPingRequestBody { + return { jsonrpc: '2.0', id, method: 'rt.ping' }; +} + +/** + * Parse one inbound text frame. Never throws — every unrecoverable + * shape maps to `{kind: 'ignore', reason, raw}` so the caller can log + * once and move on. The caller decides whether an ignored frame is + * noise (double-ping) or a bug (unknown method). + */ +export function parseIncoming(raw: string): IncomingFrame { + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return { kind: 'ignore', reason: 'not_json', raw }; + } + if (!isJsonObject(parsed)) { + return { kind: 'ignore', reason: 'not_object', raw }; + } + if (parsed.jsonrpc !== '2.0') { + return { kind: 'ignore', reason: 'wrong_jsonrpc_version', raw }; + } + + // Notification (server → client, no id). + if (typeof parsed.method === 'string') { + if (parsed.method === 'rt.event' && isJsonObject(parsed.params)) { + return { kind: 'event', params: parsed.params as unknown as RtEventParams }; + } + if (parsed.method === 'rt.revoked' && isJsonObject(parsed.params)) { + return { kind: 'revoked', params: parsed.params as unknown as RtRevokedParams }; + } + return { kind: 'ignore', reason: `unknown_method:${parsed.method}`, raw }; + } + + // Response to one of our requests. + const id = typeof parsed.id === 'number' ? parsed.id : null; + if (parsed.error !== undefined) { + if (!isJsonObject(parsed.error)) { + return { kind: 'ignore', reason: 'error_not_object', raw }; + } + return { kind: 'error', id, error: parsed.error as unknown as RtErrorObject }; + } + if (parsed.result !== undefined && id !== null) { + return { kind: 'success', id, result: parsed.result }; + } + return { kind: 'ignore', reason: 'malformed_response', raw }; +} + +function isJsonObject(v: unknown): v is Record { + return typeof v === 'object' && v !== null && !Array.isArray(v); +} diff --git a/frontend/src/routes/files/[...path]/+page.svelte b/frontend/src/routes/files/[...path]/+page.svelte index e4d9056f..3b4cf19b 100644 --- a/frontend/src/routes/files/[...path]/+page.svelte +++ b/frontend/src/routes/files/[...path]/+page.svelte @@ -52,6 +52,8 @@ type GroupByDef as RLGroupByDef } from '$lib/components/ResourceList.svelte'; import { lazyComponent } from '$lib/composables/lazyComponent.svelte'; + import { useFolderTopic } from '$lib/composables/useFolderTopic.svelte'; + import log from 'loglevel'; import { t } from '$lib/i18n/index.svelte'; import { confirmDialog, promptDialog } from '$lib/stores/dialogs.svelte'; import { drives as drivesStore } from '$lib/stores/drives.svelte'; @@ -61,6 +63,11 @@ import { dateBucket, sizeBucket, typeLabel } from '$lib/stores/files.svelte'; import { replaceSet } from '$lib/utils/sets'; + // Message-bus logger. Users can tune with + // oxi.setLogLevel('oxi:message-bus', 'debug') + // See `frontend/AGENTS.md § Logging`. + const busLog = log.getLogger('oxi:message-bus'); + // File preview and the WOPI editor are heavy and only appear on demand, so // their modules load the first time the user opens one (see the effects that // call `.load()` when `viewerOpen` / `wopiOpen` flip true). @@ -430,6 +437,56 @@ } } + // ── Live folder updates (message bus) ──────────────────────────── + // Subscribe to `folder:{currentId}` and refresh when another tab — + // or another user with a share — mutates something in this folder. + // The refresh call is coalesced through `#reloadScheduled` so a + // burst of events (e.g. a multi-file upload) collapses to a single + // fetch. Local mutations trigger `reload()` themselves, so events + // authored by this same user are dropped as echo (the `actor` on + // the event is the caller UUID from the server). + // + // See `docs/plan/message-bus.md § D` and the `useFolderTopic` + // composable for the wiring. + let reloadScheduled = false; + function scheduleLiveReload(actor: string): void { + // Actor echo: this same session's mutations already updated the + // listing through their own success path, so a re-fetch would + // only cost a round-trip. Other tabs of the same user still see + // the change (they render from their own state, not this one). + if (session.user?.id && actor === session.user.id) return; + if (reloadScheduled) return; + reloadScheduled = true; + // Coalesce a burst; 100 ms is enough for the tail of a multi- + // event upload without feeling laggy. + setTimeout(() => { + reloadScheduled = false; + void reload(); + }, 100); + } + useFolderTopic(() => currentId, { + onFileCreated: (d) => scheduleLiveReload(d.actor), + onFileRenamed: (d) => scheduleLiveReload(d.actor), + onFileMoved: (d) => scheduleLiveReload(d.actor), + onFileDeleted: (d) => scheduleLiveReload(d.actor), + onFolderCreated: (d) => scheduleLiveReload(d.actor), + onFolderRenamed: (d) => scheduleLiveReload(d.actor), + onFolderMoved: (d) => scheduleLiveReload(d.actor), + onFolderDeleted: (d) => scheduleLiveReload(d.actor), + onRevoked: (params) => { + // The subscription is already gone server-side. Notify the + // user and send them back to their home so they don't sit + // on a stale folder view with no way to know why updates + // stopped. + ui.notify( + t('files.folder_access_revoked', 'Your access to this folder was revoked.'), + 'warning' + ); + busLog.warn('folder access revoked', { topic: params.topic, reason: params.reason }); + void goto(resolve('/files')); + } + }); + function openFolder(folder: FolderItem) { // Canonical single-id URL. Legacy `/files/A/B/C` still resolves // (canonicalize-on-load rewrites it inside `load()`), but new