cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(msg-bus): prefer MessageBus as Realtime

Browse Source
This commit is contained in:
Edouard Vanbelle
2026-09-11 00:25:25 +02:00
parent 1d280c161c
commit 7918fff47b
51 changed files with 295 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/application/ports/realtime_ports.rs → src/application/ports/message_bus_ports.rs
+32 -32
View File
@@ -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<Box<...>>` shape.
pub type BusStream = Pin<Box<dyn Stream<Item = RealtimeEvent> + Send>>;
pub type BusStream = Pin<Box<dyn Stream<Item = MessageBusEvent> + 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(),
src/application/ports/mod.rs
+1 -1
View File
@@ -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;
src/application/services/file_management_service.rs
+15 -15
View File
@@ -57,12 +57,12 @@ pub struct FileManagementService {
/// (stub/test builders); production DI wires it in.
storage_usage:
Option<Arc<crate::application::services::storage_usage_service::StorageUsageService>>,
/// 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<Arc<dyn crate::application::ports::realtime_ports::RealtimeBus>>,
bus: Option<Arc<dyn crate::application::ports::message_bus_ports::MessageBus>>,
/// 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<dyn crate::application::ports::realtime_ports::RealtimeBus>,
bus: Arc<dyn crate::application::ports::message_bus_ports::MessageBus>,
) -> 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(),
src/application/services/file_upload_service.rs
+8 -8
View File
@@ -56,13 +56,13 @@ pub struct FileUploadService {
/// (`create_file_from_owned_blob_with_perms`); `None` in minimal test
/// wiring.
instant_upload: Option<InstantUploadDeps>,
/// 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<Arc<dyn crate::application::ports::realtime_ports::RealtimeBus>>,
bus: Option<Arc<dyn crate::application::ports::message_bus_ports::MessageBus>>,
}
/// 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<dyn crate::application::ports::realtime_ports::RealtimeBus>,
bus: Arc<dyn crate::application::ports::message_bus_ports::MessageBus>,
) -> 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,
src/application/services/folder_service.rs
+16 -16
View File
@@ -49,13 +49,13 @@ pub struct FolderService {
/// on cross-drive MOVE. Silently skipped when unwired (stubs).
storage_usage:
Option<Arc<crate::application::services::storage_usage_service::StorageUsageService>>,
/// 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<Arc<dyn crate::application::ports::realtime_ports::RealtimeBus>>,
bus: Option<Arc<dyn crate::application::ports::message_bus_ports::MessageBus>>,
}
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<dyn crate::application::ports::realtime_ports::RealtimeBus>,
bus: Arc<dyn crate::application::ports::message_bus_ports::MessageBus>,
) -> 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,
src/bin/generate-asyncapi.rs
+7 -7
View File
@@ -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",
src/bin/opaque-hurl-helper.rs
+1 -1
View File
@@ -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
src/bin/rt-hurl-helper.rs
+1 -1
View File
@@ -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:
src/common/di.rs
+15 -15
View File
@@ -703,12 +703,12 @@ impl AppServiceFactory {
resource_access_hook: Option<
Arc<dyn crate::application::ports::resource_access_hook::ResourceAccessHook>,
>,
bus: &Arc<crate::infrastructure::services::in_process_realtime_bus::InProcessRealtimeBus>,
bus: &Arc<crate::infrastructure::services::in_process_message_bus::InProcessMessageBus>,
) -> 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<dyn crate::application::ports::realtime_ports::RealtimeBus> =
let bus_trait: Arc<dyn crate::application::ports::message_bus_ports::MessageBus> =
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<crate::application::services::external_mount_router::MountRouter>,
/// 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<dyn RealtimeBus>`) so the
/// Stored as the concrete type (not `Arc<dyn MessageBus>`) so the
/// GC task's `Weak<Self>` lifecycle is legible from di.rs. Consumers
/// that only need the trait obtain it via
/// `Arc::clone(&state.bus) as Arc<dyn RealtimeBus>`.
/// `Arc::clone(&state.bus) as Arc<dyn MessageBus>`.
pub bus: Arc<
crate::infrastructure::services::in_process_realtime_bus::InProcessRealtimeBus,
crate::infrastructure::services::in_process_message_bus::InProcessMessageBus,
>,
pub auth_service: Option<AuthServices>,
/// OPAQUE aPAKE substrate (RFC 9807). Populated only when
src/infrastructure/services/in_process_realtime_bus.rs → src/infrastructure/services/in_process_message_bus.rs
+20 -20
View File
@@ -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<InProcessRealtimeBus>` (or `Arc<dyn RealtimeBus>`).
/// Callers hold `Arc<InProcessMessageBus>` (or `Arc<dyn MessageBus>`).
/// The struct owns its topic map and — when constructed via
/// [`InProcessRealtimeBus::with_replicator`] — an [`Arc<dyn BusReplicator>`]
/// [`InProcessMessageBus::with_replicator`] — an [`Arc<dyn BusReplicator>`]
/// that gets fed every local publish for outbound broker forwarding.
pub struct InProcessRealtimeBus {
topics: DashMap<Topic, broadcast::Sender<RealtimeEvent>>,
pub struct InProcessMessageBus {
topics: DashMap<Topic, broadcast::Sender<MessageBusEvent>>,
replicator: Arc<dyn BusReplicator>,
}
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<RealtimeEvent> {
fn sender_for(&self, topic: &Topic) -> broadcast::Sender<MessageBusEvent> {
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> {
InProcessRealtimeBus::with_replicator(Arc::new(NoopReplicator))
fn make_bus() -> Arc<InProcessMessageBus> {
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 {
src/infrastructure/services/mod.rs
+1 -1
View File
@@ -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;
src/interfaces/api/handlers/grant_handler.rs
+4 -4
View File
@@ -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],
},
);
src/interfaces/api/handlers/rt_ws.rs
+16 -16
View File
@@ -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<String, Sub>)
///
/// 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<AppState>,
) {
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<Value>) -> 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(),
src/interfaces/api/routes.rs
+1 -1
View File
@@ -674,7 +674,7 @@ pub fn create_api_routes(app_state: &Arc<AppState>) -> Router<Arc<AppState>> {
.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