diff --git a/docs/plan/message-bus.md b/docs/plan/message-bus.md index 04a361da..ded7f55f 100644 --- a/docs/plan/message-bus.md +++ b/docs/plan/message-bus.md @@ -75,18 +75,41 @@ smoke suite plus manual multi-user E2E. Live today: - **Ticket tested** — S10 (happy path), S11 (single-use replay rejected). -Deferred and still open — see the Roadmap section and the +Active — still under Phase A, ordered by priority: + +- **Job dashboard live** (next) — `JobRegistry` publishes step + progress + terminal state on `job:{id}`; the admin jobs view + subscribes and drops its polling. Small; same shape as folder-live. + Value: an operator who triggers a long-running job (backend + migration, thumbnail import, etc.) can navigate to another admin + page and come back without losing progress visibility. +- **Notifications table + bell** (E) — topic + producer + auto-sub + land here. Same pattern as `:authz`. Larger; unblocks Phase-B + `@mentions`. + +Deferred — see the Roadmap section's `## Deferred` block and the `project_message_bus_reconnect_gap` memory: -- **Notifications table + bell** (E) — topic + producer + auto-sub - land here. Same pattern as `:authz`. -- **Presence** (Phase B) — `folder:{id}:presence` topic + awareness - frames. +- **Workspace UX** (was Phase B) — presence, comments, reactions, + `@mentions`, `NotificationService` as bus subscriber. +- **Infrastructure payoff** (was Phase C) — sync-client push + invalidation, album live, slideshow sync. - **Yjs collab** — `docs/plan/markdown-collab.md`, depends on the binary-frame routing this plan sketches but doesn't ship. - **Broker replicator** (Postgres LISTEN/NOTIFY or Redis) — for multi-instance and durable event log. `BusReplicator` port declared, `NoopReplicator` wired today. +- **Session-resume tokens** — `rt.subscribe { since: N }` + a + server-side per-topic ring buffer with sequence numbers. + Replaces "full refetch on reconnect" with delta replay. Pairs + with the collab editor slice (Yjs) where refetch cost is high. +- **SharedWorker for multi-tab dedup** — one WS per user per + browser profile, shared across every same-origin tab. Turns "5 + tabs open" into 1 WS instead of 5. Ship when the "Live WS + sessions" admin card sits persistently at N × user count. +- **Web Push for offline delivery** — pairs with Slice E + (notifications bell). Delivers to closed browsers via FCM / + Mozilla autopush / Apple Push through a service worker. ## Non-goals @@ -1219,9 +1242,13 @@ Ships the infrastructure and the two most visible consumers together. `useReconnect` composable → folder view refetches after WS comes back. Bridges the in-memory-bus "events lost during outage" gap (see `project_message_bus_reconnect_gap` memory). -- **Job dashboard live** — TODO. `JobRegistry` publishes step - progress and terminal state; FE job dashboard subscribes and - replaces polling. +- **Job dashboard live** — TODO (next slice). `JobRegistry` + publishes step progress and terminal state on `job:{id}`; FE + job dashboard subscribes and replaces polling. Operator value: + once a long-running job is triggered (backend migration, thumb + import, blobs consistency…), the admin can navigate to another + page and come back without losing progress visibility — the WS + push keeps whatever component is subscribed up-to-date. - **Notifications table + bell** — TODO (Slice E). New `notifications` table + `NotificationService` port; initial ingesters for `share-granted`, `new-login-from-new-device`, @@ -1238,83 +1265,159 @@ Deliverables sized ~4 weeks end-to-end. Slice D (folder-live) and Slice F (ticket flow) landed 2026-09-11. Slices E + collab are the open work in Phase A. -### Phase B — Presence + comments +### Deferred — everything below is on the shelf -Everything that turns OxiCloud from a file store into a shared -workspace. +None of these ship on a fixed date; each is triggered by a concrete +consumer need. Grouped by theme (workspace UX, infrastructure +payoff, replicator) so the reader still sees the connective tissue, +but there is no commitment to sequencing. + +#### Workspace UX (formerly "Phase B") + +Turns OxiCloud from a file store into a shared workspace. Ship when +a specific feature here graduates from "would be nice" to "the +product needs it". - **Presence topics** — `folder:{id}:presence`, `file:{id}:presence`. Awareness-style: joined/left/cursor. Ephemeral, not persisted. -- **FE presence UI**: "N people viewing" badge in folder header; +- **FE presence UI** — "N people viewing" badge in folder header; avatar rail; hover to highlight; "someone is previewing this photo right now" in the lightbox. - **Comments on any file** — new `comments` table (threaded, per file, supports reactions), `CommentService` port, `file:{id}:comments` topic for live delivery. -- **@mentions**: mention autocomplete in the comment editor; +- **@mentions** — mention autocomplete in the comment editor; mention → notification into the mentioned user's `user:{u}:notifications` topic + `notifications` row + optional email (reuses existing `MagicLinkMailer`-style templating). -- **Reactions**: 👍❤️🎉 on comments and on files themselves; live - fan-out on the same `file:{id}:comments` topic. -- **Comment resolutions**: Google-Docs-style thread markers. -- **NotificationService consumes bus events** — up to Phase A the - bus's publish calls sit inline in each mutation site - (`FolderService::create_folder_with_perms`, - `FileUploadService::upload_file_streaming`, and — once folder-live - rounds out — the delete / rename / move sites for both files and - folders). That is the right shape and stays: the bus is - location-keyed (`Topic::Folder(id)`, subscriber-scoped) and - belongs at the mutation site. - When Phase B ships, notifications sit on the **same axis** (also - location + actor + subscriber-driven) — not the FileLifecycleHook - axis (which is server-internal, content-keyed, fan-out-to-all). - So `NotificationService` becomes an in-process subscriber to the - bus itself: it registers a `bus.subscribe(...)` on the topics it - cares about (`folder:{id}`, `file:{id}`, share-grant events), - translates relevant events into `notif.notifications` rows, and - re-publishes on `user:{u}:notifications`. No new dispatcher, no - new hook trait, no changes to existing mutation sites — the bus IS - the mutation-event pipeline for anything subscriber-driven. - Contrast with `FileLifecycleHook` (`src/application/ports/file_lifecycle.rs`): - that stays focused on content transitions (blob_hash, content_type) - and fires unconditionally to server-side workers (thumbnails, - audio metadata, plugins). Bus and lifecycle-hook are complementary - — same triggering moment, orthogonal fan-out shape and payload - discipline. Do NOT try to unify them; the two axes are genuinely - different (all-vs-subscribed × content-vs-location). +- **Reactions** — 👍❤️🎉 on comments and files. Live fan-out on the + same `file:{id}:comments` topic. +- **Comment resolutions** — Google-Docs-style thread markers. +- **`NotificationService` as a bus subscriber** (architectural + pivot). Up to Phase A the bus's publish calls sit inline in each + mutation site (`FolderService::create_folder_with_perms`, + `FileUploadService::upload_file_streaming`, and the delete / + rename / move sites for both files and folders). That is the + right shape and stays: the bus is location-keyed + (`Topic::Folder(id)`, subscriber-scoped) and belongs at the + mutation site. When notifications ship, they sit on the **same + axis** (also location + actor + subscriber-driven) — not the + `FileLifecycleHook` axis (which is server-internal, content-keyed, + fan-out-to-all). So `NotificationService` becomes an in-process + subscriber to the bus itself: it registers a `bus.subscribe(...)` + on the topics it cares about (`folder:{id}`, `file:{id}`, + share-grant events), translates relevant events into + `notif.notifications` rows, and re-publishes on + `user:{u}:notifications`. No new dispatcher, no new hook trait, + no changes to existing mutation sites — the bus IS the + mutation-event pipeline for anything subscriber-driven. Contrast + with `FileLifecycleHook` (`src/application/ports/file_lifecycle.rs`): + that stays focused on content transitions (blob_hash, + content_type) and fires unconditionally to server-side workers + (thumbnails, audio metadata, plugins). Bus and lifecycle-hook are + complementary — same triggering moment, orthogonal fan-out shape + and payload discipline. Do NOT try to unify them; the two axes + are genuinely different (all-vs-subscribed × content-vs-location). -Deliverables sized ~3 weeks after Phase A. +#### Infrastructure payoff (formerly "Phase C") -### Phase C — Sync client push + album live +Where the bus starts paying for itself on operator cost. Ship when +sync-client PROPFIND traffic or the album-viewing experience +becomes a real bottleneck. -Where the bus starts paying for itself on infrastructure cost too. - -- **Sync-client push invalidation**: WebDAV / NextCloud DAV handlers - publish `file:{id}` and `folder:{id}` deltas after commit. Sync - clients get a lightweight `Sync-Invalidate` mechanism (or a - dedicated WS endpoint for headless clients) so they refetch only - changed paths instead of polling PROPFIND. Cuts a large chunk of - Nextcloud-style client chatter. -- **Album live updates**: `folder:{album_id}` reused — as photos are - added to an album, everyone viewing sees them appear. -- **Slideshow sync**: one presenter picks "Present"; other viewers of - the album can opt-in to follow the presenter's current frame. +- **Sync-client push invalidation** — WebDAV / NextCloud DAV + handlers publish `file:{id}` and `folder:{id}` deltas after + commit. Sync clients get a lightweight `Sync-Invalidate` + mechanism (or a dedicated WS endpoint for headless clients) so + they refetch only changed paths instead of polling PROPFIND. + Cuts a large chunk of Nextcloud-style client chatter. +- **Album live updates** — `folder:{album_id}` reused; as photos + are added to an album, everyone viewing sees them appear. +- **Slideshow sync** — one presenter picks "Present"; other viewers + of the album can opt-in to follow the presenter's current frame. Uses `folder:{album_id}` with a `presenter_frame` event kind. -Deliverables sized ~2–3 weeks after Phase B. +#### Multi-instance & broker -### Later — multi-instance & broker - -Only invoked when the deployment actually needs it. Nothing above -depends on these landing on any fixed date. +Only invoked when the deployment actually needs it. Also the +mitigation for the "events lost during outage window" gap (see +`project_message_bus_reconnect_gap` memory) if durable replay +becomes important for collab or sync-push. - **`PgListenReplicator`** — ship when we run more than one server - instance. Same port, no consumer changes. + instance. Same `BusReplicator` port, no consumer changes. - **`BrokerReplicator`** for RabbitMQ or NATS — ship when either cross-datacenter fan-out or a shared broker with other services matters. Same port, no consumer changes. +#### Session-resume tokens (wire-protocol extension) + +Replaces today's "full refetch on reconnect" workaround with a +delta-replay protocol: the client remembers the sequence number of +the last event it processed per topic; on reconnect, it says +"resume from N" and the server replays every event since N. The +canonical shape across the industry — Discord's `OP 6 Resume`, +Slack's sync API, Firestore's `resume_token`, Notion's sync-token +pattern. Cheaper than a REST refetch for high-fan-out topics (Yjs +CRDT deltas, notification streams) where the "catch-up" would +otherwise pull megabytes of state the client mostly already has. + +Requires: + +- **Server-side**: per-topic bounded ring buffer with monotonic + sequence numbers. Bounded because we're not building a durable + log — a hold-back of the last N events per topic is enough for + the common "closed laptop for 10 min" case. A resume request + older than the retention window falls through to a client-side + full refetch (same code path today's `onReconnect` uses), so + the client never fails hard — just degrades. +- **Wire**: `rt.subscribe` gains an optional `since: number` param + and the ack carries the current sequence number. `rt.event` + gains a `seq` field the client stores as `last_seq[topic]`. +- **Client**: `MessageBusClient` persists `last_seq[topic]` and + replays it on `#onOpen`'s subscribe-replay. `onReconnect` + handlers keep their fallback-to-refetch role for the + older-than-retention case. + +Meaningful for the collab editor slice (Yjs) and for future +sync-client push. Not worth doing before either of those lands — +folder-view refetch is a folder-page fetch (small); Yjs +"refetch" would be the whole doc snapshot (potentially large). +See `project_message_bus_reconnect_gap` memory (option 2). + +#### Client-side connection efficiency + +Optimizations to how the SPA holds its WebSocket. Independent of +server changes; ship when the per-user connection count actually +becomes a load concern. Today's grace-period tab-hidden close +(closes the WS after 60 s hidden, reconnects on visibility return) +covers the low-hanging fruit; both items below layer on top. + +- **SharedWorker for multi-tab dedup** — one WebSocket per user per + browser profile, shared across every same-origin tab via a + `SharedWorker`. All tabs `postMessage` through the worker + instead of holding independent `WebSocket` instances. Slack / + Gmail / Google Docs all do this. Turns "user has 5 folder tabs + open" from 5 WS into 1. Refactor cost: `MessageBusClient` moves + behind the worker boundary; every `useTopic` call becomes an + RPC to the worker instead of a direct method call. Payoff + scales with per-user tab count — worth doing if operators see + the "Live WS sessions" admin card sitting persistently at 5× + the user count. Not worth it otherwise; grace-close already + handles the common "background tab" case at ~30% of this + refactor's complexity. +- **Web Push for truly-offline delivery** — service-worker-backed + push notifications delivered by the browser vendor (FCM for + Chrome, Mozilla autopush for Firefox, Apple Push for Safari) + even when the user has no OxiCloud tab open. Complements the + WS-based notification stream: WS delivers to foreground tabs; + Web Push delivers to closed browsers. Requires server-side + push-subscription store, per-vendor endpoint delivery (usually + `web-push` crate), and a service worker in the FE. Meaningful + UX win alongside Slice E (notifications bell) — a share + arriving while the user is away actually reaches them. + Deferred until Slice E ships; the two form a natural pair. + ## What this bus does NOT replace - Message queue / job queue — jobs stay in `job_registry`; bus just diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte index d30f1367..f14ac53c 100644 --- a/frontend/src/lib/components/AdminJobsPanel.svelte +++ b/frontend/src/lib/components/AdminJobsPanel.svelte @@ -35,6 +35,8 @@ purgeJobRuns } from '$lib/api/endpoints/adminJobs'; import type { Finding, JobParam, JobSummary, RunSummary, RunStatus } from '$lib/api/types'; + import { messageBus } from '$lib/message-bus/client.svelte'; + import { serverConfig } from '$lib/stores/serverConfig.svelte'; // ─── State ──────────────────────────────────────────────────────── @@ -222,6 +224,68 @@ return () => stopPolling(); }); + // ─── Live updates via the message bus ───────────────────────────── + // + // Subscribes to `job:{name}` for every registered job so a run's + // start/end flips this panel's state within a network hop instead + // of waiting up to POLL_MS for the next poll tick. The 5s polling + // stays as fallback — messages that arrive while the tab was hidden + // (Page Visibility grace-close in `messageBus`) are lost, and + // polling reconciles. + // + // Progress publishes aren't wired yet (deferred — see + // `docs/plan/message-bus.md`), so the handler treats + // `job_run_progress` as a benign no-op and simply refetches the + // row's runs when a run ends. When per-handler progress emits + // land, this composable is where `onProgress` will map into the + // runs table without a poll round-trip. + // + // Keyed on the SORTED name set — the polling refresh reassigns + // `jobs` on every tick with a fresh array, which would tear down + // and rebuild every sub if the effect keyed on `jobs` identity. + // The registered set is fixed at server boot, so this stable key + // stops the effect churning. + const jobNameKey = $derived( + jobs + ?.map((j) => j.name) + .sort() + .join('|') ?? '' + ); + + $effect(() => { + // Don't attempt to open a socket if the server has the bus + // disabled — the WS route is unmounted (404) and the circuit + // breaker would just count failures. + if (!serverConfig.features.message_bus) return; + if (!jobNameKey) return; + const names = jobNameKey.split('|').filter(Boolean); + const releases = names.map((name) => + messageBus.subscribe( + `job:${name}`, + (params) => { + // `job_run_progress` currently has no publisher — + // treat any incoming variant defensively. + if (params.event === 'job_run_started') { + void loadJobs(); + if (expandedJob === name) void loadRuns(name); + } else if (params.event === 'job_run_ended') { + void loadJobs(); + if (expandedJob === name) void loadRuns(name); + } + }, + () => { + // Server-side eviction — admin role revoked or bus + // disabled mid-session. Nothing surgical to do; the + // next poll will reflect whatever changed and the + // operator's UI will render normally. + } + ) + ); + return () => { + for (const release of releases) release(); + }; + }); + async function loadRuns(name: string) { runsLoadingByJob = { ...runsLoadingByJob, [name]: true }; runsErrorByJob = { ...runsErrorByJob, [name]: '' }; diff --git a/frontend/src/lib/composables/useJobTopic.svelte.ts b/frontend/src/lib/composables/useJobTopic.svelte.ts new file mode 100644 index 00000000..5b65a691 --- /dev/null +++ b/frontend/src/lib/composables/useJobTopic.svelte.ts @@ -0,0 +1,76 @@ +// Admin-job-dashboard sugar around `useTopic`. +// +// Subscribes to `job:{name}` and dispatches the three `rt.event` +// variants — `job_run_started`, `job_run_progress`, `job_run_ended` +// — to per-verb handlers. Admin-only server-side (Class 3, see +// `application/ports/message_bus_ports.rs::required_perm`); a +// non-admin caller sees `topic_forbidden` on subscribe and the +// subscription is dropped. +// +// This composable mirrors `useFolderTopic` but is deliberately +// separate: the two share nothing beyond `useTopic`, and merging +// them would smear two AuthZ classes (ResourceRead vs RoleAdmin) +// into one call surface. + +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 JobRunStartedData from '$lib/generated/message-bus/JobRunStartedData'; +import type JobRunProgressData from '$lib/generated/message-bus/JobRunProgressData'; +import type JobRunEndedData from '$lib/generated/message-bus/JobRunEndedData'; + +/** + * Optional per-verb handlers for a single job's run stream. Any + * subset is accepted; unhandled verbs fall through silently. + * + * `onEnded` is the canonical "the server is done publishing on this + * topic for now" signal — the admin dashboard uses it to switch a + * row back to "idle" and stop expecting progress updates. The + * subscription itself stays open (jobs can run again), so callers + * that want a one-shot pattern should track that in their own state. + */ +export interface JobTopicHandlers { + onStarted?: (data: JobRunStartedData) => void; + onProgress?: (data: JobRunProgressData) => void; + onEnded?: (data: JobRunEndedData) => void; + /** Server evicted the subscription — admin role revoked, or + * the message bus itself was disabled mid-session. */ + onRevoked?: (params: RtRevokedParams) => void; +} + +/** + * Subscribe to `job:{name}` and dispatch each `rt.event` + * notification to the matching per-verb handler. + * + * `name` 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 so the subscription follows + * the currently-selected job. + */ +export function useJobTopic( + name: string | null | (() => string | null), + handlers: JobTopicHandlers +): void { + const topic = () => { + const n = typeof name === 'function' ? name() : name; + return n ? `job:${n}` : null; + }; + useTopic(topic, (params) => dispatch(params, handlers), handlers.onRevoked); +} + +function dispatch(params: RtEventParams, handlers: JobTopicHandlers): 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 'job_run_started': + handlers.onStarted?.(params.data as JobRunStartedData); + return; + case 'job_run_progress': + handlers.onProgress?.(params.data as JobRunProgressData); + return; + case 'job_run_ended': + handlers.onEnded?.(params.data as JobRunEndedData); + return; + } +} diff --git a/frontend/src/lib/generated/message-bus/JobRunEndedData.ts b/frontend/src/lib/generated/message-bus/JobRunEndedData.ts new file mode 100644 index 00000000..1eef442b --- /dev/null +++ b/frontend/src/lib/generated/message-bus/JobRunEndedData.ts @@ -0,0 +1,9 @@ +// AUTO-GENERATED — do not edit by hand. +// Regenerate with `just asyncapi-ts`. +interface JobRunEndedData { + ended_at: string; + name: string; + reason?: string | null; + success: boolean; +} +export type { JobRunEndedData as default }; diff --git a/frontend/src/lib/generated/message-bus/JobRunProgressData.ts b/frontend/src/lib/generated/message-bus/JobRunProgressData.ts new file mode 100644 index 00000000..fc237968 --- /dev/null +++ b/frontend/src/lib/generated/message-bus/JobRunProgressData.ts @@ -0,0 +1,9 @@ +// AUTO-GENERATED — do not edit by hand. +// Regenerate with `just asyncapi-ts`. +interface JobRunProgressData { + message?: string | null; + name: string; + step?: number | null; + total?: number | null; +} +export type { JobRunProgressData as default }; diff --git a/frontend/src/lib/generated/message-bus/JobRunStartedData.ts b/frontend/src/lib/generated/message-bus/JobRunStartedData.ts new file mode 100644 index 00000000..c2c13851 --- /dev/null +++ b/frontend/src/lib/generated/message-bus/JobRunStartedData.ts @@ -0,0 +1,8 @@ +// AUTO-GENERATED — do not edit by hand. +// Regenerate with `just asyncapi-ts`. +interface JobRunStartedData { + actor: string; + name: string; + started_at: string; +} +export type { JobRunStartedData as default }; diff --git a/frontend/src/lib/generated/message-bus/RtEventKind.ts b/frontend/src/lib/generated/message-bus/RtEventKind.ts index 65aa0797..ff42f7f7 100644 --- a/frontend/src/lib/generated/message-bus/RtEventKind.ts +++ b/frontend/src/lib/generated/message-bus/RtEventKind.ts @@ -6,6 +6,9 @@ enum RtEventKind { FOLDER_CREATED = 'folder_created', FOLDER_RENAMED = 'folder_renamed', FOLDER_MOVED = 'folder_moved', - FOLDER_DELETED = 'folder_deleted' + FOLDER_DELETED = 'folder_deleted', + JOB_RUN_STARTED = 'job_run_started', + JOB_RUN_PROGRESS = 'job_run_progress', + JOB_RUN_ENDED = 'job_run_ended' } export type { RtEventKind as default }; diff --git a/frontend/src/lib/generated/message-bus/RtEventParams.ts b/frontend/src/lib/generated/message-bus/RtEventParams.ts index 576ca986..38602e61 100644 --- a/frontend/src/lib/generated/message-bus/RtEventParams.ts +++ b/frontend/src/lib/generated/message-bus/RtEventParams.ts @@ -6,6 +6,9 @@ import type FolderCreatedData from './FolderCreatedData'; import type FolderRenamedData from './FolderRenamedData'; import type FolderMovedData from './FolderMovedData'; import type FolderDeletedData from './FolderDeletedData'; +import type JobRunStartedData from './JobRunStartedData'; +import type JobRunProgressData from './JobRunProgressData'; +import type JobRunEndedData from './JobRunEndedData'; import type RtEventKind from './RtEventKind'; // AUTO-GENERATED — do not edit by hand. // Regenerate with `just asyncapi-ts`. @@ -18,7 +21,10 @@ interface RtEventParams { | FolderCreatedData | FolderRenamedData | FolderMovedData - | FolderDeletedData; + | FolderDeletedData + | JobRunStartedData + | JobRunProgressData + | JobRunEndedData; event: RtEventKind; topic: string; } diff --git a/src/application/ports/message_bus_ports.rs b/src/application/ports/message_bus_ports.rs index ecdfe6ff..a69bf99c 100644 --- a/src/application/ports/message_bus_ports.rs +++ b/src/application/ports/message_bus_ports.rs @@ -53,7 +53,7 @@ use crate::common::errors::DomainError; /// Encodes to a stable dotted wire key that maps naturally onto RabbitMQ /// topic-exchange routing keys or NATS subjects when the [`BusReplicator`] /// seam is filled in later. -#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)] +#[derive(Clone, PartialEq, Eq, Hash, Debug)] pub enum Topic { /// A folder's mutation stream — file/subfolder created/deleted/renamed/ /// moved in or out. Consumed by the folder view for live refresh. @@ -63,6 +63,17 @@ pub enum Topic { /// subscribe the caller and evict stale subs when its events fire once /// the eviction wiring lands (Phase-A follow-up). UserAuthz(Uuid), + + /// A named background job's run lifecycle — start / progress / + /// end. Consumed by the admin job dashboard so operators who + /// trigger a long-running job (backend migration, thumb import…) + /// can navigate to other admin pages without losing progress + /// visibility. AuthZ: **admin-only** (Class 3 role-scoped). + /// Non-admins get `topic_forbidden` — indistinguishable on the + /// wire from an unknown topic. Job names are stable + /// scheduler-registered strings (e.g. `backend_migration`, + /// `thumb_derived_import`); the topic string is `job:`. + Job(String), } impl Topic { @@ -72,6 +83,7 @@ impl Topic { match self { Topic::Folder(id) => format!("folder:{id}"), Topic::UserAuthz(id) => format!("user:{id}:authz"), + Topic::Job(name) => format!("job:{name}"), } } @@ -90,6 +102,22 @@ impl Topic { let id = Uuid::parse_str(id_str).map_err(|_| ParseTopicErr::BadUuid)?; return Ok(Topic::UserAuthz(id)); } + if let Some(name) = s.strip_prefix("job:") { + // Job names are scheduler-registered short slugs — see + // `infrastructure/scheduler/registry.rs`. Validate here + // only that the name is non-empty and consists of + // `[a-z0-9_-]` chars — reject anything else as + // `Unknown` (indistinguishable to the caller from a + // topic shape we've never heard of). + if !name.is_empty() + && name + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_' || b == b'-') + { + return Ok(Topic::Job(name.to_string())); + } + return Err(ParseTopicErr::Unknown); + } Err(ParseTopicErr::Unknown) } @@ -107,6 +135,7 @@ impl Topic { resource: BusResource::Folder(*id), }, Topic::UserAuthz(id) => AuthzCheck::IdentityMatch { user_id: *id }, + Topic::Job(_) => AuthzCheck::RoleAdmin, } } } @@ -152,8 +181,12 @@ pub enum AuthzCheck { /// Class 2 — Identity-scoped. `caller_id` must equal `user_id`. /// No admin bypass — privacy is a hard rule. IdentityMatch { user_id: Uuid }, - // Class 3 (role-scoped `admin:*`) and the bespoke `job:{id}` check - // land with their topic variants. + + /// Class 3 — Role-scoped. Caller must hold the admin role. Used + /// by `Topic::Job(_)` today; future `admin:*` topics land here. + /// Non-admin subscriber gets `topic_forbidden` on the wire — + /// same anti-enum shape as unknown-topic denial. + RoleAdmin, } // ════════════════════════════════════════════════════════════════════════════ @@ -257,6 +290,50 @@ pub enum MessageBusEvent { /// the payload extends with additional resource classes — see the /// plan's Phase-B roadmap. AuthzChanged { affected_folders: Vec }, + + /// A background job's run started. Published on + /// [`Topic::Job`]. `started_at` is server wall-clock (RFC 3339 + /// serialised by serde). Admin dashboard's job-list view uses + /// this to flip a row from "idle" to "running" without a + /// polling round-trip. + JobRunStarted { + name: String, + started_at: chrono::DateTime, + actor: Uuid, + }, + + /// A background job made progress. Published at most every + /// 3 seconds per job (throttled at the publish site — see + /// scheduler engine). `step` / `total` populate an operator- + /// facing progress bar; `message` is a one-line free-form + /// status. All three are optional because different jobs have + /// different progress semantics (some know the total up front, + /// some don't; some can render a step count, some just have a + /// running status message). + JobRunProgress { + name: String, + step: Option, + total: Option, + message: Option, + }, + + /// A background job's run ended. `success = true` for a normal + /// completion; `false` for failure / cancelled / paused with + /// unhandled outcome. `reason` populates the "click for + /// details" flow on the admin dashboard: the notification (Slice + /// E) will link to `/admin/jobs/` on the `false` branch, + /// where the full outcome and paused-run state live. + /// + /// Deliberately NOT a rich outcome enum — the admin panel is one + /// click away and holds the full detail; the bus event just + /// needs to say "done, ok or not". Adding a new outcome nuance + /// server-side does NOT churn the wire. + JobRunEnded { + name: String, + success: bool, + reason: Option, + ended_at: chrono::DateTime, + }, } // ════════════════════════════════════════════════════════════════════════════ @@ -420,6 +497,33 @@ mod tests { assert_eq!(Topic::parse(&wire).unwrap(), t); } + #[test] + fn job_topic_roundtrip() { + let t = Topic::Job("backend_migration".to_string()); + let wire = t.to_wire_key(); + assert_eq!(wire, "job:backend_migration"); + assert_eq!(Topic::parse(&wire).unwrap(), t); + } + + #[test] + fn job_topic_rejects_bad_name_chars() { + // Job names come from the scheduler registry — a stable + // `[a-z0-9_-]` alphabet. Anything else is `Unknown` (same + // wire response as an unrecognised topic shape). + assert_eq!(Topic::parse("job:"), Err(ParseTopicErr::Unknown)); + assert_eq!(Topic::parse("job:UPPER"), Err(ParseTopicErr::Unknown)); + assert_eq!(Topic::parse("job:with.dot"), Err(ParseTopicErr::Unknown)); + assert_eq!(Topic::parse("job:with space"), Err(ParseTopicErr::Unknown)); + } + + #[test] + fn required_perm_job_is_role_admin() { + assert_eq!( + Topic::Job("thumb_derived_import".to_string()).required_perm(), + AuthzCheck::RoleAdmin + ); + } + #[test] fn parse_rejects_bad_uuid() { assert_eq!( @@ -547,6 +651,32 @@ mod tests { }, "authz_changed", ), + ( + MessageBusEvent::JobRunStarted { + name: "backend_migration".into(), + started_at: chrono::DateTime::::from_timestamp(0, 0).unwrap(), + actor: Uuid::nil(), + }, + "job_run_started", + ), + ( + MessageBusEvent::JobRunProgress { + name: "backend_migration".into(), + step: Some(10), + total: Some(100), + message: Some("phase 2".into()), + }, + "job_run_progress", + ), + ( + MessageBusEvent::JobRunEnded { + name: "backend_migration".into(), + success: true, + reason: None, + ended_at: chrono::DateTime::::from_timestamp(0, 0).unwrap(), + }, + "job_run_ended", + ), ]; for (ev, expected) in cases { let json = serde_json::to_value(ev).unwrap(); diff --git a/src/bin/generate-asyncapi.rs b/src/bin/generate-asyncapi.rs index 3df67207..52bd67e8 100644 --- a/src/bin/generate-asyncapi.rs +++ b/src/bin/generate-asyncapi.rs @@ -140,6 +140,21 @@ fn channels() -> Value { "SubscribeRequest": { "$ref": "#/components/messages/RtSubscribeRequest" }, "UnsubscribeRequest": { "$ref": "#/components/messages/RtUnsubscribeRequest" }, } + }, + "Job": { + "address": "job:{jobName}", + "description": "A named background job's run lifecycle — Started / Progress / Ended. Consumed by the admin dashboard so operators who trigger a long-running job (backend migration, thumb import…) can navigate off the admin page and come back without losing progress. AuthZ: admin-only (Class 3 role-scoped) — non-admin gets `topic_forbidden`, indistinguishable on the wire from an unknown topic.", + "parameters": { + "jobName": { "description": "Scheduler-registered short slug (e.g. `backend_migration`); `[a-z0-9_-]` chars only" } + }, + "messages": { + "SubscribeRequest": { "$ref": "#/components/messages/RtSubscribeRequest" }, + "UnsubscribeRequest": { "$ref": "#/components/messages/RtUnsubscribeRequest" }, + "SubscribedResponse": { "$ref": "#/components/messages/RtSubscribedResponse" }, + "ErrorResponse": { "$ref": "#/components/messages/RtErrorResponse" }, + "JobEvent": { "$ref": "#/components/messages/RtFolderEventNotification" }, + "RevokedNotification": { "$ref": "#/components/messages/RtRevokedNotification" }, + } } }) } @@ -301,6 +316,9 @@ fn components() -> Value { "FolderRenamedData": folder_renamed_schema(), "FolderMovedData": folder_moved_schema(), "FolderDeletedData": folder_deleted_schema(), + "JobRunStartedData": job_run_started_schema(), + "JobRunProgressData": job_run_progress_schema(), + "JobRunEndedData": job_run_ended_schema(), }, // How the client authenticates. Handler side is `auth_middleware` // — the same middleware every `/api/*` request goes through, so @@ -580,6 +598,7 @@ fn event_kind_schema() -> Value { "enum": [ "file_created", "file_renamed", "file_moved", "file_deleted", "folder_created", "folder_renamed", "folder_moved", "folder_deleted", + "job_run_started", "job_run_progress", "job_run_ended", ], }) } @@ -596,6 +615,9 @@ fn event_data_union_schema() -> Value { ref_schema("FolderRenamedData"), ref_schema("FolderMovedData"), ref_schema("FolderDeletedData"), + ref_schema("JobRunStartedData"), + ref_schema("JobRunProgressData"), + ref_schema("JobRunEndedData"), ] }) } @@ -710,6 +732,52 @@ fn folder_deleted_schema() -> Value { }) } +// ─────────────────── Job event data payloads ───────────────────── +// Published on `Topic::Job(name)`. AuthZ is Class-3 (admin-only) — +// non-admins get `topic_forbidden` on subscribe, so these payloads +// only ever reach admin subscribers. See `handlers/rt_ws.rs`. + +fn job_run_started_schema() -> Value { + json!({ + "type": "object", + "description": "A background job's run started. `name` matches the scheduler-registered job name (e.g. `backend_migration`). `actor` is `00000000-0000-0000-0000-000000000000` today — the scheduler doesn't yet thread the trigger caller through.", + "required": ["name", "started_at", "actor"], + "properties": { + "name": { "type": "string" }, + "started_at": { "type": "string", "format": "date-time" }, + "actor": { "type": "string", "format": "uuid" }, + } + }) +} + +fn job_run_progress_schema() -> Value { + json!({ + "type": "object", + "description": "A background job made progress. Throttled at the publish site to at most one per 3 s per job (see scheduler engine). `step` / `total` populate a progress bar; all three fields are optional because different jobs report different granularities.", + "required": ["name"], + "properties": { + "name": { "type": "string" }, + "step": { "type": ["integer", "null"], "minimum": 0 }, + "total": { "type": ["integer", "null"], "minimum": 0 }, + "message": { "type": ["string", "null"] }, + } + }) +} + +fn job_run_ended_schema() -> Value { + json!({ + "type": "object", + "description": "A background job's run ended. `success = true` for a normal completion; `false` for failure / timeout / cancelled / paused-with-unhandled-outcome. `reason` populates the toast text on the `false` branch and links to `/admin/jobs/` for the full outcome. Consumer typically drops its subscription on receipt (job is done).", + "required": ["name", "success", "ended_at"], + "properties": { + "name": { "type": "string" }, + "success": { "type": "boolean" }, + "reason": { "type": ["string", "null"] }, + "ended_at": { "type": "string", "format": "date-time" }, + } + }) +} + /// `rt.revoked` notification body — server tells the client that a /// specific subscription has been evicted. `topic` is the wire-form /// string the client originally subscribed to. `reason` is the stable diff --git a/src/common/di.rs b/src/common/di.rs index fec2e0ad..0efe2a65 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1759,6 +1759,17 @@ impl AppServiceFactory { Arc::new(crate::application::ports::message_bus_ports::NoopReplicator), ); + // Wire the bus into the JobRegistry so `dispatch` (both the + // periodic supervisor and the manual `trigger` paths) can + // publish `JobRunStarted` / `JobRunEnded` on `Topic::Job(name)`. + // Set here — after both the bus and the registry are + // constructed — via `OnceLock`. Silent no-op on subsequent + // calls; unit tests that build a registry without a bus just + // skip this. + let bus_for_jobs: Arc = + bus.clone(); + core.job_registry.set_message_bus(bus_for_jobs); + // WebSocket ticket store — see `rt_ticket_store` module doc for // why this exists (DPoP-bound sessions can't be re-proofed on // a browser-issued WS upgrade). Reaper task runs for the app diff --git a/src/infrastructure/scheduler/engine.rs b/src/infrastructure/scheduler/engine.rs index 88e18cf4..57206c40 100644 --- a/src/infrastructure/scheduler/engine.rs +++ b/src/infrastructure/scheduler/engine.rs @@ -94,8 +94,11 @@ async fn run(registry: Arc) { // Fire and forget from the supervisor's perspective — we // don't care about the outcome, `dispatch` records it on the // entry and emits the log line itself. Periodic ticks never - // force — that's an admin-trigger-only affordance. - let _ = dispatch(&name, entry, &JobRunArgs::default()).await; + // force — that's an admin-trigger-only affordance. Pass the + // bus reference so periodic runs also publish job events + // (same reasoning as the manual-trigger path). + let bus = registry.message_bus_snapshot(); + let _ = dispatch(&name, entry, &JobRunArgs::default(), bus).await; } } @@ -117,7 +120,12 @@ async fn run(registry: Arc) { /// `args` is passed through to `JobHandler::run`. The supervisor's /// periodic ticks pass `JobRunArgs::default()`; the admin trigger /// endpoint forwards parsed query params such as `?force=true`. -pub(super) async fn dispatch(name: &str, entry: Arc, args: &JobRunArgs) -> JobOutcome { +pub(super) async fn dispatch( + name: &str, + entry: Arc, + args: &JobRunArgs, + bus: Option>, +) -> JobOutcome { // Try to acquire the single-permit gate. `try_acquire` is // non-blocking — if held, we know the previous run is still // executing and skip this tick. @@ -161,6 +169,26 @@ pub(super) async fn dispatch(name: &str, entry: Arc, args: &JobRunArgs let started_wall = Utc::now(); let start_instant = Instant::now(); + // Publish `JobRunStarted` on `Topic::Job(name)` so the admin + // job dashboard's live tab receives a "started" tick without + // polling. Silent no-op when the bus isn't wired (test setup) + // or when nobody is subscribed. `actor` is `Uuid::nil()` today + // because the scheduler doesn't carry the trigger caller + // through — the periodic supervisor has no caller, and the + // admin trigger endpoints don't thread it in. When they do, + // swap to the real UUID. + if let Some(bus) = bus.as_ref() { + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; + bus.publish( + &Topic::Job(name.to_string()), + MessageBusEvent::JobRunStarted { + name: name.to_string(), + started_at: started_wall, + actor: uuid::Uuid::nil(), + }, + ); + } + // Spawn so panics land as `JoinError::is_panic()` instead of // unwinding into the supervisor loop. Args cloned into the spawn // scope so the borrow doesn't outlive the caller. @@ -214,6 +242,33 @@ pub(super) async fn dispatch(name: &str, entry: Arc, args: &JobRunArgs // the diagnostic `cause` field. log_outcome(name, &outcome, cause, elapsed_ms); + // Publish `JobRunEnded` on `Topic::Job(name)`. This is the + // signal the FE watches for to terminate its subscription + // (`useJobTopic` unsubscribes on `onEnded`). `success = false` + // covers timeout, panic, handler error — the admin dashboard + // renders the row as failed and the "click for details" + // notification (Slice E) will link to `/admin/jobs/`. + // Silent no-op when the bus isn't wired. + if let Some(bus) = bus.as_ref() { + use crate::application::ports::message_bus_ports::{MessageBusEvent, Topic}; + let success = outcome.is_ok(); + let reason = match &outcome { + crate::infrastructure::scheduler::types::JobOutcome::Err { message } => { + Some(message.clone()) + } + _ => None, + }; + bus.publish( + &Topic::Job(name.to_string()), + MessageBusEvent::JobRunEnded { + name: name.to_string(), + success, + reason, + ended_at: Utc::now(), + }, + ); + } + drop(permit); outcome } @@ -421,16 +476,15 @@ mod tests { // Kick off dispatch 1 in the background — it holds the permit // for ~200 ms. let entry_bg = entry.clone(); - let bg = - tokio::spawn( - async move { dispatch("overrun", entry_bg, &JobRunArgs::default()).await }, - ); + let bg = tokio::spawn(async move { + dispatch("overrun", entry_bg, &JobRunArgs::default(), None).await + }); // Give dispatch 1 time to grab the permit. tokio::time::sleep(Duration::from_millis(50)).await; // Dispatch 2 should observe the permit taken and skip. - dispatch("overrun", entry.clone(), &JobRunArgs::default()).await; + dispatch("overrun", entry.clone(), &JobRunArgs::default(), None).await; // Only dispatch 1's handler should have actually run so far. assert_eq!(calls.load(Ordering::SeqCst), 1); @@ -458,7 +512,7 @@ mod tests { .await; let entry = registry.get("slow").await.unwrap(); - dispatch("slow", entry.clone(), &JobRunArgs::default()).await; + dispatch("slow", entry.clone(), &JobRunArgs::default(), None).await; // The timeout fired; last_outcome must be Err. let state = entry.state.lock().unwrap(); diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs index 7692d982..9573c831 100644 --- a/src/infrastructure/scheduler/registry.rs +++ b/src/infrastructure/scheduler/registry.rs @@ -60,12 +60,24 @@ pub(super) struct JobState { /// native services `register()` during DI wiring. pub struct JobRegistry { entries: RwLock>>, + /// Message bus — used by `dispatch` (via `trigger`) to publish + /// `JobRunStarted` / `JobRunProgress` / `JobRunEnded` on + /// `Topic::Job(name)` so the admin dashboard can render live + /// progress without polling. `OnceLock` because it's set exactly + /// once at DI time (after both the registry and the bus are + /// constructed) and read from many concurrent triggers; `Arc` + /// keeps consumers cheap. `None` before wiring (unit tests + /// exercise the registry without a bus). + message_bus: std::sync::OnceLock< + std::sync::Arc, + >, } impl JobRegistry { pub fn new() -> Self { Self { entries: RwLock::new(HashMap::new()), + message_bus: std::sync::OnceLock::new(), } } @@ -294,7 +306,33 @@ impl JobRegistry { /// that just want a plain run pass `JobRunArgs::default()`. pub async fn trigger(self: &Arc, name: &str, args: &JobRunArgs) -> Option { let entry = self.get(name).await?; - Some(super::engine::dispatch(name, entry, args).await) + // Pass the bus reference through to `dispatch` so start / end + // events publish on `Topic::Job(name)`. `Option::cloned()` + // returns a fresh `Arc` clone (or None) — negligible. + let bus = self.message_bus.get().cloned(); + Some(super::engine::dispatch(name, entry, args, bus).await) + } + + /// Wire the message bus. Called once from DI after both the + /// registry and the bus are constructed. Idempotent: a second + /// call is a silent no-op (`OnceLock::set` returns `Err`), so + /// test setups that call this more than once don't panic. + pub fn set_message_bus( + &self, + bus: std::sync::Arc, + ) { + let _ = self.message_bus.set(bus); + } + + /// Snapshot the currently-wired bus (if any). `None` when + /// `set_message_bus` hasn't been called yet — every test setup + /// that skips DI wiring, and the very early boot before the + /// bus is constructed. Called by the periodic supervisor and + /// by `trigger` so both paths publish job events identically. + pub(super) fn message_bus_snapshot( + &self, + ) -> Option> { + self.message_bus.get().cloned() } } diff --git a/src/infrastructure/services/in_process_message_bus.rs b/src/infrastructure/services/in_process_message_bus.rs index 324f3665..b97fb8fc 100644 --- a/src/infrastructure/services/in_process_message_bus.rs +++ b/src/infrastructure/services/in_process_message_bus.rs @@ -112,8 +112,13 @@ impl InProcessMessageBus { /// (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 { + // `topic.clone()` because `Topic::Job(String)` isn't `Copy`. + // The clone is a String alloc on the cold path (first ever + // subscriber for a topic) and free on the hot path (existing + // entry — `entry` doesn't need to move the key when the + // entry is already present). self.topics - .entry(*topic) + .entry(topic.clone()) .or_insert_with(|| broadcast::channel(BROADCAST_RING_CAPACITY).0) .clone() } @@ -194,6 +199,9 @@ fn event_kind(event: &MessageBusEvent) -> &'static str { MessageBusEvent::FolderMoved { .. } => "folder_moved", MessageBusEvent::FolderDeleted { .. } => "folder_deleted", MessageBusEvent::AuthzChanged { .. } => "authz_changed", + MessageBusEvent::JobRunStarted { .. } => "job_run_started", + MessageBusEvent::JobRunProgress { .. } => "job_run_progress", + MessageBusEvent::JobRunEnded { .. } => "job_run_ended", } } diff --git a/src/interfaces/api/handlers/rt_ws.rs b/src/interfaces/api/handlers/rt_ws.rs index 7a8655da..8608b155 100644 --- a/src/interfaces/api/handlers/rt_ws.rs +++ b/src/interfaces/api/handlers/rt_ws.rs @@ -338,6 +338,48 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc { + match crate::interfaces::middleware::user::resolve_live_role( + auth.auth_application_service.as_ref(), + caller_id, + "user", + ) + .await + { + crate::interfaces::middleware::user::LiveRole::Active(role) => role.to_string(), + crate::interfaces::middleware::user::LiveRole::Revoked => { + // Account revoked between ticket-issue and now. + // Terminate the session immediately — dropping + // `socket` at end of scope closes the WS cleanly + // (no explicit `.close()` needed; that would + // require pulling `SinkExt` into scope for one + // line). + tracing::info!( + target: "audit", + event = "message_bus.session_rejected", + reason = "account_revoked", + caller_id = %caller_id, + "👮🏻‍♂️ WS session rejected — account revoked", + ); + drop(socket); + return; + } + } + } + None => "user".to_string(), + }; + // Outbound queue — every path that produces a client-bound frame // enqueues here; the writer half of the select drains. Also // carries internal `EvictFolders` control signals from the @@ -450,7 +492,7 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc { if let Some(reply) = - handle_text_frame(&txt, caller_id, &state, &mut subs, &out_tx).await + handle_text_frame(&txt, caller_id, &caller_role, &state, &mut subs, &out_tx).await && socket.send(Message::Text(reply.into())).await.is_err() { break; } @@ -487,6 +529,7 @@ async fn handle_session(mut socket: WebSocket, caller_id: Uuid, state: Arc, subs: &mut HashMap, out_tx: &mpsc::Sender, @@ -516,9 +559,9 @@ async fn handle_text_frame( }; match method.as_str() { - "rt.subscribe" => { - Some(handle_subscribe(id, req.params, caller_id, state, subs, out_tx).await) - } + "rt.subscribe" => Some( + handle_subscribe(id, req.params, caller_id, caller_role, state, subs, out_tx).await, + ), "rt.unsubscribe" => Some(handle_unsubscribe(id, req.params, subs)), "rt.ping" => Some(success_response(id, serde_json::json!({ "pong": true }))), _ => Some(error_response( @@ -534,6 +577,7 @@ async fn handle_subscribe( id: Value, params: Value, caller_id: Uuid, + caller_role: &str, state: &Arc, subs: &mut HashMap, out_tx: &mpsc::Sender, @@ -622,6 +666,21 @@ async fn handle_subscribe( ); } } + AuthzCheck::RoleAdmin => { + // Class 3 — role-scoped. Caller must be admin. `caller_role` + // was snapshotted at session start (see `handle_session`), + // so no per-subscribe DB hit. A demotion mid-session + // takes effect on the caller's next reconnect. + if caller_role != "admin" { + audit_denied(caller_id, &topic_str, "role_denied"); + return error_response( + id, + error_code::TOPIC_FORBIDDEN, + "topic_forbidden", + Some(serde_json::json!({ "topic": topic_str })), + ); + } + } } // AuthZ passed — install the subscription and spawn a reader task diff --git a/tests/api/rt_bus_check.sh b/tests/api/rt_bus_check.sh index a3a5880d..0ecd0122 100755 --- a/tests/api/rt_bus_check.sh +++ b/tests/api/rt_bus_check.sh @@ -52,6 +52,18 @@ # again. Guards replay: a captured token # outside its 30 s TTL, or one already # consumed, MUST fail the upgrade with 401. +# S12 Job topic admin-only — user1 (non-admin) subscribes to +# `job:`; server must reject +# with `topic_forbidden` (audit reason +# `role_denied`). Guards the Class-3 +# role-scoped AuthZ gate on +# `Topic::Job` — admin-only, no bypass, +# anti-enumeration parity with unknown +# topics. The allow side is covered by +# the Rust `required_perm` + dispatch +# unit tests; the seeded suite has no +# admin token, and minting one here +# would pollute state for other files. # # Exit non-zero on any failure — run.sh treats that as a suite failure. # ───────────────────────────────────────────────────────────────────────────── @@ -641,4 +653,30 @@ set -e || die "S11: expected exit 2 (connect refused), got $reuse_exit" log "S11 OK" -log "All eleven message-bus scenarios passed." +# ── Scenario 12 — Job topic is admin-only ─────────────────────────────────── +# `Topic::Job("")` maps to `AuthzCheck::RoleAdmin` in +# `application/ports/message_bus_ports.rs::required_perm`, and +# `handle_subscribe` denies any caller whose snapshotted role at +# session open is not "admin". user1 is a plain account, so this +# subscribe MUST land on the deny arm. +# +# The wire response uses `topic_forbidden` (same shape as an unknown +# topic — anti-enumeration: a non-admin cannot probe which job names +# are registered). The audit reason `role_denied` is asserted at the +# Rust unit-test layer. +# +# If this ever accepts and delivers events, someone weakened the +# Class-3 gate (dropped the role-snapshot check in the Job arm, +# widened `required_perm`, or reused a permissive dispatch branch). +log "S12: user1 subscribes to job:whatever; expect topic_forbidden." +if ! "$HELPER_BIN" expect-denied \ + --url "$ws_url" \ + --token "$user1_token" \ + --subscribe "job:whatever" \ + --reason topic_forbidden \ + --timeout 3s; then + die "S12: user1 was NOT denied on job topic (admin gate broken?)" +fi +log "S12 OK" + +log "All twelve message-bus scenarios passed."