From d850e9c1003eb5193789677812a1df76aca8fca1 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Thu, 10 Sep 2026 07:15:09 +0200 Subject: [PATCH] feat(msg-bus): add file and folder mutation notoficaton + tests --- docs/plan/message-bus.md | 25 +++ src/application/ports/realtime_ports.rs | 165 +++++++++++++++--- .../services/file_management_service.rs | 163 ++++++++++++++++- src/application/services/folder_service.rs | 83 +++++++++ src/bin/generate-asyncapi.rs | 101 ++++++++++- src/common/di.rs | 7 +- tests/api/rt_bus_check.sh | 112 +++++++++++- 7 files changed, 626 insertions(+), 30 deletions(-) diff --git a/docs/plan/message-bus.md b/docs/plan/message-bus.md index 56df8f13..bd730303 100644 --- a/docs/plan/message-bus.md +++ b/docs/plan/message-bus.md @@ -1049,6 +1049,31 @@ workspace. - **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). Deliverables sized ~3 weeks after Phase A. diff --git a/src/application/ports/realtime_ports.rs b/src/application/ports/realtime_ports.rs index a3bd1165..85a10019 100644 --- a/src/application/ports/realtime_ports.rs +++ b/src/application/ports/realtime_ports.rs @@ -17,7 +17,10 @@ //! to folders the caller can't `Read`: //! //! - Topics: [`Topic::Folder`] and [`Topic::UserAuthz`] -//! - Events: [`RealtimeEvent::FileCreated`], [`RealtimeEvent::FolderCreated`] +//! - Events: [`RealtimeEvent::FileCreated`], [`RealtimeEvent::FileRenamed`], +//! [`RealtimeEvent::FileMoved`], [`RealtimeEvent::FileDeleted`], +//! [`RealtimeEvent::FolderCreated`], [`RealtimeEvent::FolderRenamed`], +//! [`RealtimeEvent::FolderMoved`], [`RealtimeEvent::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}`, @@ -181,6 +184,36 @@ pub enum RealtimeEvent { parent_id: Uuid, actor: Uuid, }, + /// A file was renamed. `parent_id` unchanged — same folder. + FileRenamed { + file_id: Uuid, + old_name: String, + new_name: String, + parent_id: Uuid, + actor: Uuid, + }, + /// A file was moved between folders. Fanned out on BOTH the source + /// and destination folder topics — subscribers to either see the + /// event once. `from` / `to` are the folder UUIDs; a move + /// involving a drive root would be `Option` in a future + /// variant, but MVP mutations all address a real folder. + FileMoved { + file_id: Uuid, + name: String, + from: Uuid, + to: Uuid, + actor: Uuid, + }, + /// A file was deleted (trashed OR permanently removed — the wire + /// doesn't distinguish, and clients treat both as "disappears from + /// the folder view"). `parent_id` is the folder the file used to + /// live in — snapshotted before the delete since the row may be + /// gone by publish time. + FileDeleted { + file_id: Uuid, + parent_id: Uuid, + actor: Uuid, + }, /// A sub-folder was created inside `parent_id`. FolderCreated { folder_id: Uuid, @@ -188,6 +221,30 @@ pub enum RealtimeEvent { parent_id: Uuid, actor: Uuid, }, + /// A folder was renamed. `parent_id` unchanged. + FolderRenamed { + folder_id: Uuid, + old_name: String, + new_name: String, + parent_id: Uuid, + actor: Uuid, + }, + /// A folder was moved between parents. Fanned out on BOTH source + /// and destination folder topics. + FolderMoved { + folder_id: Uuid, + name: String, + from: Uuid, + to: Uuid, + actor: Uuid, + }, + /// A folder was deleted (trashed or permanent — see `FileDeleted` + /// for the same wire-collapse rationale). + FolderDeleted { + folder_id: Uuid, + parent_id: Uuid, + actor: Uuid, + }, } // ════════════════════════════════════════════════════════════════════════════ @@ -394,26 +451,92 @@ mod tests { fn event_serializes_with_snake_case_discriminator() { // The `#[serde(tag = "event")]` shape is the WS wire contract for // the `rt.event` JSON-RPC notification's `params.event` field. Pin - // it with a snapshot so accidental rename of the enum variant - // fails the test instead of silently breaking clients. - let ev = RealtimeEvent::FileCreated { - file_id: Uuid::nil(), - name: "notes.md".into(), - parent_id: Uuid::nil(), - actor: Uuid::nil(), - }; - let json = serde_json::to_value(&ev).unwrap(); - assert_eq!(json["event"], "file_created"); - assert_eq!(json["name"], "notes.md"); - - let ev = RealtimeEvent::FolderCreated { - folder_id: Uuid::nil(), - name: "docs".into(), - parent_id: Uuid::nil(), - actor: Uuid::nil(), - }; - let json = serde_json::to_value(&ev).unwrap(); - assert_eq!(json["event"], "folder_created"); + // 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)] = &[ + ( + RealtimeEvent::FileCreated { + file_id: Uuid::nil(), + name: "notes.md".into(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }, + "file_created", + ), + ( + RealtimeEvent::FileRenamed { + file_id: Uuid::nil(), + old_name: "a.md".into(), + new_name: "b.md".into(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }, + "file_renamed", + ), + ( + RealtimeEvent::FileMoved { + file_id: Uuid::nil(), + name: "a.md".into(), + from: Uuid::nil(), + to: Uuid::nil(), + actor: Uuid::nil(), + }, + "file_moved", + ), + ( + RealtimeEvent::FileDeleted { + file_id: Uuid::nil(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }, + "file_deleted", + ), + ( + RealtimeEvent::FolderCreated { + folder_id: Uuid::nil(), + name: "docs".into(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }, + "folder_created", + ), + ( + RealtimeEvent::FolderRenamed { + folder_id: Uuid::nil(), + old_name: "old".into(), + new_name: "new".into(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }, + "folder_renamed", + ), + ( + RealtimeEvent::FolderMoved { + folder_id: Uuid::nil(), + name: "docs".into(), + from: Uuid::nil(), + to: Uuid::nil(), + actor: Uuid::nil(), + }, + "folder_moved", + ), + ( + RealtimeEvent::FolderDeleted { + folder_id: Uuid::nil(), + parent_id: Uuid::nil(), + actor: Uuid::nil(), + }, + "folder_deleted", + ), + ]; + for (ev, expected) in cases { + let json = serde_json::to_value(ev).unwrap(); + assert_eq!( + json["event"], *expected, + "wire discriminator mismatch for {ev:?}" + ); + } } #[test] diff --git a/src/application/services/file_management_service.rs b/src/application/services/file_management_service.rs index 95352b55..cdad81bf 100644 --- a/src/application/services/file_management_service.rs +++ b/src/application/services/file_management_service.rs @@ -5,7 +5,7 @@ use crate::application::ports::authorization_ports::AuthorizationEngine; use crate::application::ports::file_lifecycle::FileLifecycleHook; use crate::application::ports::file_ports::FileManagementUseCase; use crate::application::ports::resource_access_hook::ResourceAccessHook; -use crate::application::ports::storage_ports::{CopyFolderTreeResult, FileWritePort}; +use crate::application::ports::storage_ports::{CopyFolderTreeResult, FileReadPort, FileWritePort}; use crate::application::ports::trash_ports::TrashUseCase; use crate::application::services::external_mount_router::{MountRouter, ResolvedId}; use crate::application::services::mount_dto::{audit_mount_write, mount_file_dto, mount_parent_id}; @@ -57,6 +57,20 @@ pub struct FileManagementService { /// (stub/test builders); production DI wires it in. storage_usage: Option>, + /// Realtime message bus. When wired, delete / rename / move + /// mutations publish their corresponding `RealtimeEvent` on + /// `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>, + /// 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, + /// move rewrites `folder_id`. Without it we couldn't publish on + /// the correct `Topic::Folder(parent)` (delete) or fan out on the + /// source-side folder (move). Optional so stubs stay minimal; when + /// unwired, the affected publishes silently no-op. + file_read: Option>, } impl FileManagementService { @@ -68,7 +82,7 @@ impl FileManagementService { pub fn with_trash( file_repository: Arc, trash_service: Option>, - _file_read: Option>, + file_read: Option>, _folder_repo: Option>, content_cache: Option>, authz: Arc, @@ -83,9 +97,21 @@ impl FileManagementService { resource_access_hook: None, drive_repo: None, storage_usage: None, + bus: None, + file_read, } } + /// Wire the realtime message bus. When set, delete / rename / move + /// mutations publish on the affected folder topics after commit. + pub fn with_realtime_bus( + mut self, + bus: Arc, + ) -> Self { + self.bus = Some(bus); + self + } + /// Sets the lifecycle hook dispatcher (thumbnails, audio metadata, …). pub fn with_file_lifecycle_hook(mut self, hook: Arc) -> Self { self.file_lifecycle_hook = Some(hook); @@ -181,6 +207,50 @@ impl FileManagementService { self } + /// Snapshot the (uuid, name, parent-folder-uuid) of a file BEFORE + /// a mutation, so the realtime publish path has a stable + /// `Topic::Folder(parent)` to address even after the write commits + /// (delete removes the row; move rewrites `folder_id`). + /// + /// Returns `None` when: + /// - `file_read` is unwired (stub / test builder), + /// - the file can't be read (already gone, permission failure — + /// the caller is responsible for AuthZ, this is only a + /// best-effort snapshot), + /// - the file is at drive-root (no parent folder, nothing to + /// publish on), + /// - the id can't be parsed as a `Uuid` (mount id or malformed). + /// + /// All `None` paths silently skip the publish — never fail the + /// mutation. The bus is best-effort. + async fn snapshot_for_publish(&self, file_id: &str) -> Option<(Uuid, String, Uuid)> { + let file_read = self.file_read.as_ref()?; + let file = file_read.get_file(file_id).await.ok()?; + let parts = file.into_parts(); + let file_uuid = Uuid::parse_str(&parts.id).ok()?; + let parent_uuid = Uuid::parse_str(parts.folder_id.as_deref()?).ok()?; + Some((file_uuid, parts.name, parent_uuid)) + } + + /// Publish `FileDeleted` on the file's parent folder topic. Called + /// by both the trash and permanent-delete paths so subscribers see + /// one event regardless of which happened. Silent no-op when the + /// bus isn't wired or the pre-mutation snapshot failed (drive-root + /// 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}; + bus.publish( + &Topic::Folder(parent_uuid), + RealtimeEvent::FileDeleted { + file_id: file_uuid, + parent_id: parent_uuid, + actor: caller_id, + }, + ); + } + } + /// Engine check for a file resource. Parses the id into a `Uuid` and /// requires the specified permission. async fn require_file_perm( @@ -462,8 +532,41 @@ impl FileManagementUseCase for FileManagementService { } } + // Snapshot source parent BEFORE the write — after `move_file` + // the row's `folder_id` reflects the destination, so we'd lose + // the from-side for the fan-out. + let source_snapshot = self.snapshot_for_publish(file_id).await; + let dto = self.move_file(file_id, folder_id, caller_id).await?; + // Realtime 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 + // failed (drive-root file, mount), or the destination is + // drive-root (`dto.folder_id = None`). Any of those cases + // matches the "no interested subscribers" invariant so + // silently skipping is honest. + if let (Some(bus), Some((file_uuid, name, source_uuid)), Some(dest_str)) = + (&self.bus, source_snapshot, dto.folder_id.as_deref()) + && 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 { + file_id: file_uuid, + name, + from: source_uuid, + to: dest_uuid, + actor: caller_id, + }; + // Publish twice — subscribers to either folder see the + // event exactly once because they're only subscribed to + // one of the two topics. + bus.publish(&Topic::Folder(source_uuid), event.clone()); + bus.publish(&Topic::Folder(dest_uuid), event); + } + // Cross-drive move invalidates the file's `owner_cache` entry // in the authz engine — the cache assumed drive_id stability // that no longer holds. Without this call the drive-role @@ -559,7 +662,39 @@ impl FileManagementUseCase for FileManagementService { } self.require_file_perm(file_id, Permission::Update, caller_id) .await?; - self.rename_file(file_id, new_name, caller_id).await + + // Snapshot old_name pre-rename so the publish carries both + // sides of the transition. `parent_id` is the same before and + // after (rename doesn't move) so we can safely reuse it from + // the post-mutation DTO. + let old_name = self + .snapshot_for_publish(file_id) + .await + .map(|(_, name, _)| name); + + let dto = self.rename_file(file_id, new_name, caller_id).await?; + + // Realtime 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)) = + (&self.bus, old_name, dto.folder_id.as_deref()) + && 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}; + bus.publish( + &Topic::Folder(parent_uuid), + RealtimeEvent::FileRenamed { + file_id: file_uuid, + old_name, + new_name: dto.name.clone(), + parent_id: parent_uuid, + actor: caller_id, + }, + ); + } + Ok(dto) } async fn delete_file_with_perms(&self, id: &str, caller_id: Uuid) -> Result<(), DomainError> { @@ -572,7 +707,15 @@ impl FileManagementUseCase for FileManagementService { } self.require_file_perm(id, Permission::Delete, caller_id) .await?; - self.delete_file(id).await + + // Snapshot the pre-delete parent so the publish path has a + // `Topic::Folder(parent)` to address — the row is gone by the + // time `delete_file` returns. + let snapshot = self.snapshot_for_publish(id).await; + + self.delete_file(id).await?; + self.publish_file_deleted(caller_id, snapshot); + Ok(()) } /// Smart delete: trash-first with dedup reference cleanup. @@ -597,6 +740,15 @@ impl FileManagementUseCase for FileManagementService { self.require_file_perm(id, Permission::Delete, caller_id) .await?; + + // Snapshot the pre-mutation parent so both the trash and the + // fallback permanent-delete path can publish `FileDeleted` on + // the right folder topic. Trash leaves the row in place but + // `is_trashed=TRUE` makes it disappear from folder listings — + // subscribers should see the same "gone from this folder" + // event either way. + let snapshot = self.snapshot_for_publish(id).await; + // Step 1: Try trash (soft delete — file row stays, blob stays referenced) if let Some(trash) = &self.trash_service { info!("Moving file to trash: {}", id); @@ -610,6 +762,7 @@ impl FileManagementUseCase for FileManagementService { // Do NOT decrement blob ref here — the file row still exists // (is_trashed = TRUE). The trigger will decrement when the // row is actually DELETEd during trash emptying. + self.publish_file_deleted(caller_id, snapshot); return Ok(true); // trashed } Err(err) => { @@ -625,7 +778,7 @@ impl FileManagementUseCase for FileManagementService { // Step 2: Permanent delete — trigger handles blob ref_count self.delete_file(id).await?; - + self.publish_file_deleted(caller_id, snapshot); Ok(false) // permanently deleted } diff --git a/src/application/services/folder_service.rs b/src/application/services/folder_service.rs index 34917be4..e52723e8 100644 --- a/src/application/services/folder_service.rs +++ b/src/application/services/folder_service.rs @@ -813,6 +813,28 @@ impl FolderUseCase for FolderService { drive_repo.invalidate_default_drive_all(); } + // Realtime 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 + // bus isn't wired. + if let (Some(bus), Some(parent_str)) = (&self.bus, folder.parent_id()) + && 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}; + bus.publish( + &Topic::Folder(parent_uuid), + RealtimeEvent::FolderRenamed { + folder_id: folder_uuid, + old_name: folder.name().to_owned(), + new_name: renamed.name().to_owned(), + parent_id: parent_uuid, + actor: caller_id, + }, + ); + } + Ok(FolderDto::from(renamed)) } @@ -938,6 +960,18 @@ impl FolderUseCase for FolderService { } } + // Snapshot source parent BEFORE the move — the post-move + // `folder.parent_id()` is the destination. Best-effort: if the + // lookup fails or the folder has no parent (root — can't be + // moved anyway per drive_semantics), the publish path below + // silently skips. + let source_parent_uuid = self + .folder_storage + .get_folder(id) + .await + .ok() + .and_then(|f| f.parent_id().and_then(|p| Uuid::parse_str(p).ok())); + let parent_ref = dto.parent_id.as_deref(); let folder = self .folder_storage @@ -950,6 +984,29 @@ impl FolderUseCase for FolderService { ) })?; + // Realtime 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 + // drive-root (`folder.parent_id() = None`). + if let (Some(bus), Some(source_uuid), Some(dest_str)) = + (&self.bus, source_parent_uuid, folder.parent_id()) + && let (Ok(folder_uuid), Ok(dest_uuid)) = + (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 { + folder_id: folder_uuid, + name: folder.name().to_owned(), + from: source_uuid, + to: dest_uuid, + actor: caller_id, + }; + bus.publish(&Topic::Folder(source_uuid), event.clone()); + bus.publish(&Topic::Folder(dest_uuid), event); + } + // Cross-drive move flushes the authz engine's `owner_cache` // — every descendant's cached `Resource → drive_id` mapping // just got stale via the cascade trigger, and we don't (yet) @@ -1028,6 +1085,16 @@ impl FolderUseCase for FolderService { .await .unwrap_or_default(); + // Pre-delete snapshot for the bus publish — post-DELETE the + // row is gone and we can't recover `parent_id`. Best-effort; + // failures fall through to a silent skip below. + let publish_snapshot: Option<(Uuid, Uuid)> = + self.folder_storage.get_folder(id).await.ok().and_then(|f| { + let folder_uuid = Uuid::parse_str(f.id()).ok()?; + let parent_uuid = Uuid::parse_str(f.parent_id()?).ok()?; + Some((folder_uuid, parent_uuid)) + }); + self.folder_storage.delete_folder(id).await.map_err(|e| { DomainError::internal_error( "FolderStorage", @@ -1039,6 +1106,22 @@ impl FolderUseCase for FolderService { self.file_lifecycle.on_file_deleted(file_id); } + // Realtime 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}; + bus.publish( + &Topic::Folder(parent_uuid), + RealtimeEvent::FolderDeleted { + folder_id: folder_uuid, + parent_id: parent_uuid, + actor: caller_id, + }, + ); + } + Ok(()) } } diff --git a/src/bin/generate-asyncapi.rs b/src/bin/generate-asyncapi.rs index 93e93bbd..8da99838 100644 --- a/src/bin/generate-asyncapi.rs +++ b/src/bin/generate-asyncapi.rs @@ -246,7 +246,13 @@ fn components() -> Value { "RtErrorResponseBody": rpc_error_response_schema(), "RtFolderEventBody": folder_event_notification_schema(), "FileCreatedData": file_created_schema(), + "FileRenamedData": file_renamed_schema(), + "FileMovedData": file_moved_schema(), + "FileDeletedData": file_deleted_schema(), "FolderCreatedData": folder_created_schema(), + "FolderRenamedData": folder_renamed_schema(), + "FolderMovedData": folder_moved_schema(), + "FolderDeletedData": folder_deleted_schema(), }, // How the client authenticates. Handler side is `auth_middleware` // — the same middleware every `/api/*` request goes through, so @@ -388,12 +394,21 @@ fn folder_event_notification_schema() -> Value { "topic": { "type": "string" }, "event": { "type": "string", - "enum": ["file_created", "folder_created"], + "enum": [ + "file_created", "file_renamed", "file_moved", "file_deleted", + "folder_created", "folder_renamed", "folder_moved", "folder_deleted", + ], }, "data": { "oneOf": [ { "$ref": "#/components/schemas/FileCreatedData" }, + { "$ref": "#/components/schemas/FileRenamedData" }, + { "$ref": "#/components/schemas/FileMovedData" }, + { "$ref": "#/components/schemas/FileDeletedData" }, { "$ref": "#/components/schemas/FolderCreatedData" }, + { "$ref": "#/components/schemas/FolderRenamedData" }, + { "$ref": "#/components/schemas/FolderMovedData" }, + { "$ref": "#/components/schemas/FolderDeletedData" }, ] } } @@ -415,6 +430,48 @@ fn file_created_schema() -> Value { }) } +fn file_renamed_schema() -> Value { + json!({ + "type": "object", + "required": ["file_id", "old_name", "new_name", "parent_id", "actor"], + "properties": { + "file_id": { "type": "string", "format": "uuid" }, + "old_name": { "type": "string" }, + "new_name": { "type": "string" }, + "parent_id": { "type": "string", "format": "uuid" }, + "actor": { "type": "string", "format": "uuid" }, + } + }) +} + +fn file_moved_schema() -> Value { + json!({ + "type": "object", + "description": "Emitted on BOTH the source (`from`) and destination (`to`) folder topics. Subscribers to either see the event exactly once because they're subscribed to only one of the two.", + "required": ["file_id", "name", "from", "to", "actor"], + "properties": { + "file_id": { "type": "string", "format": "uuid" }, + "name": { "type": "string" }, + "from": { "type": "string", "format": "uuid" }, + "to": { "type": "string", "format": "uuid" }, + "actor": { "type": "string", "format": "uuid" }, + } + }) +} + +fn file_deleted_schema() -> Value { + json!({ + "type": "object", + "description": "The wire doesn't distinguish soft (trash) vs. permanent delete — clients treat both as \"disappears from the folder view\". `parent_id` is the folder the file used to live in.", + "required": ["file_id", "parent_id", "actor"], + "properties": { + "file_id": { "type": "string", "format": "uuid" }, + "parent_id": { "type": "string", "format": "uuid" }, + "actor": { "type": "string", "format": "uuid" }, + } + }) +} + fn folder_created_schema() -> Value { json!({ "type": "object", @@ -427,3 +484,45 @@ fn folder_created_schema() -> Value { } }) } + +fn folder_renamed_schema() -> Value { + json!({ + "type": "object", + "required": ["folder_id", "old_name", "new_name", "parent_id", "actor"], + "properties": { + "folder_id": { "type": "string", "format": "uuid" }, + "old_name": { "type": "string" }, + "new_name": { "type": "string" }, + "parent_id": { "type": "string", "format": "uuid" }, + "actor": { "type": "string", "format": "uuid" }, + } + }) +} + +fn folder_moved_schema() -> Value { + json!({ + "type": "object", + "description": "Emitted on BOTH the source (`from`) and destination (`to`) folder topics — same shape as `FileMoved`.", + "required": ["folder_id", "name", "from", "to", "actor"], + "properties": { + "folder_id": { "type": "string", "format": "uuid" }, + "name": { "type": "string" }, + "from": { "type": "string", "format": "uuid" }, + "to": { "type": "string", "format": "uuid" }, + "actor": { "type": "string", "format": "uuid" }, + } + }) +} + +fn folder_deleted_schema() -> Value { + json!({ + "type": "object", + "description": "Soft vs. permanent delete are indistinguishable on the wire.", + "required": ["folder_id", "parent_id", "actor"], + "properties": { + "folder_id": { "type": "string", "format": "uuid" }, + "parent_id": { "type": "string", "format": "uuid" }, + "actor": { "type": "string", "format": "uuid" }, + } + }) +} diff --git a/src/common/di.rs b/src/common/di.rs index b6768800..7377bdaa 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -825,7 +825,12 @@ impl AppServiceFactory { .with_drive_repo(drive_repo.clone()) // Destination-drive quota pre-check on cross-drive file // MOVE. Same rationale as the folder side above. - .with_storage_usage(storage_usage.clone()); + .with_storage_usage(storage_usage.clone()) + // Realtime 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()); if let Some(hook) = resource_access_hook.clone() { svc = svc.with_resource_access_hook(hook); } diff --git a/tests/api/rt_bus_check.sh b/tests/api/rt_bus_check.sh index 6a1a1a30..153f2684 100755 --- a/tests/api/rt_bus_check.sh +++ b/tests/api/rt_bus_check.sh @@ -7,7 +7,7 @@ # This script orchestrates it against a live oxicloud server: bootstraps # state with curl, exercises the bus, asserts on the helper's JSON output. # -# Five scenarios: +# Seven scenarios: # S1 Positive delivery — subscribe to folder A, upload into A, see event. # S2 Topic isolation — subscribe to folder A only, upload into B and # then A; must see A's event only. @@ -20,6 +20,14 @@ # frames from the server (proves the interval # fires), and the session still delivers an # event on the same subscription afterwards. +# S6 Delete emits — DELETE a pre-uploaded file → subscriber sees +# one `file_deleted` event with correct +# `file_id` + `parent_id` (snapshotted +# pre-delete since the row is gone by then). +# S7 Move fan-out — MOVE A→B while subscribed to BOTH topics on +# one session → observe TWO `file_moved` +# events (one via the A topic, one via B). +# Same file_id/from/to on both. # # Exit non-zero on any failure — run.sh treats that as a suite failure. # ───────────────────────────────────────────────────────────────────────────── @@ -263,4 +271,104 @@ fi || die "S5: parent_id mismatch after idle" log "S5 OK ($pings pings observed)" -log "All five realtime-bus scenarios passed." +# ── Scenario 6 — File delete emits `file_deleted` ─────────────────────────── +# Pre-create a file in folder A, then subscribe to `folder:$folder_a`, then +# DELETE the file. The subscription must observe exactly one +# `file_deleted` event — proves the delete publish hook fires and carries +# the correct `parent_id` (snapshotted pre-delete, since the row is gone +# by publish time). +log "S6: DELETE a file → subscriber observes file_deleted." +# Pre-create the file BEFORE the subscriber goes up, so S6 asserts on the +# delete event alone (S1 already covered the create-side). +s6_upload=$(curl -sS -X POST \ + -H "Authorization: Bearer $user1_token" \ + -F "folder_id=$folder_a" \ + -F "file=@$(mktemp -t rtbus_s6_body.XXXXXX);filename=s6.txt" \ + "$base_url/api/files/upload") +s6_file_id=$(printf '%s' "$s6_upload" | jq -r '.id') +[[ -n "$s6_file_id" && "$s6_file_id" != "null" ]] \ + || die "S6: pre-upload failed: $s6_upload" + +out_s6="$(mktemp -t rtbus_s6.XXXXXX)" +"$HELPER_BIN" subscribe-and-collect \ + --url "$ws_url" \ + --token "$user1_token" \ + --subscribe "folder:$folder_a" \ + --expect-events 1 \ + --timeout 5s \ + --output "$out_s6" & +helper_pid=$! +sleep 0.4 +# `DELETE /api/files/{id}` routes to `delete_and_cleanup_with_perms` — +# the trash-first path. Publish fires on BOTH the trash and the +# permanent-delete branch, so this covers whichever the test hits. +curl -sS -X DELETE \ + -H "Authorization: Bearer $user1_token" \ + "$base_url/api/files/$s6_file_id" > /dev/null +if ! wait "$helper_pid"; then + cat "$out_s6" >&2 || true + die "S6: helper did not observe the expected file_deleted event" +fi +[[ "$(jq -r '.events | length' "$out_s6")" == "1" ]] \ + || { cat "$out_s6"; die "S6: expected 1 event, got $(jq -r '.events | length' "$out_s6")"; } +[[ "$(jq -r '.events[0].event' "$out_s6")" == "file_deleted" ]] \ + || die "S6: wrong event: $(jq -r '.events[0].event' "$out_s6")" +[[ "$(jq -r '.events[0].data.file_id' "$out_s6")" == "$s6_file_id" ]] \ + || die "S6: file_id mismatch" +[[ "$(jq -r '.events[0].data.parent_id' "$out_s6")" == "$folder_a" ]] \ + || die "S6: parent_id mismatch" +log "S6 OK" + +# ── Scenario 7 — Move fans out on BOTH source and destination ─────────────── +# Pre-create a file in folder A, subscribe to BOTH `folder:$folder_a` and +# `folder:$folder_b` on ONE session, then MOVE the file A → B. The single +# session must observe TWO `file_moved` events — one delivered on the A +# topic, one on the B topic. Same file_id in both. Same event contents +# (from=A, to=B). Proves the plan's "fan out on both source AND +# destination" invariant. +# Broken publish (source-only or dest-only) would surface as 1 event. +# Broken publish-after-commit would surface as 0 events. +log "S7: MOVE fans out on both source AND destination folder topics." +s7_upload=$(curl -sS -X POST \ + -H "Authorization: Bearer $user1_token" \ + -F "folder_id=$folder_a" \ + -F "file=@$(mktemp -t rtbus_s7_body.XXXXXX);filename=s7.txt" \ + "$base_url/api/files/upload") +s7_file_id=$(printf '%s' "$s7_upload" | jq -r '.id') +[[ -n "$s7_file_id" && "$s7_file_id" != "null" ]] \ + || die "S7: pre-upload failed: $s7_upload" + +out_s7="$(mktemp -t rtbus_s7.XXXXXX)" +"$HELPER_BIN" subscribe-and-collect \ + --url "$ws_url" \ + --token "$user1_token" \ + --subscribe "folder:$folder_a" \ + --subscribe "folder:$folder_b" \ + --expect-events 2 \ + --timeout 5s \ + --output "$out_s7" & +helper_pid=$! +sleep 0.4 +# `PUT /api/files/{id}/move` — MoveFilePayload = { folder_id: }. +curl -sS -X PUT \ + -H "Authorization: Bearer $user1_token" \ + -H "Content-Type: application/json" \ + -d "$(printf '{"folder_id":"%s"}' "$folder_b")" \ + "$base_url/api/files/$s7_file_id/move" > /dev/null +if ! wait "$helper_pid"; then + cat "$out_s7" >&2 || true + die "S7: helper did not observe 2 file_moved events" +fi +# Both events same shape, same file_id, from = A, to = B. +[[ "$(jq -r '.events | length' "$out_s7")" == "2" ]] \ + || { cat "$out_s7"; die "S7: expected 2 events (fan-out on A + B), got $(jq -r '.events | length' "$out_s7")"; } +# Every event has event=file_moved, correct file_id/from/to. +if ! jq -e --arg fid "$s7_file_id" --arg from "$folder_a" --arg to "$folder_b" \ + '.events | all(.event == "file_moved" and .data.file_id == $fid and .data.from == $from and .data.to == $to)' \ + "$out_s7" > /dev/null; then + cat "$out_s7" + die "S7: event contents mismatch (expected file_moved, from=$folder_a, to=$folder_b)" +fi +log "S7 OK" + +log "All seven realtime-bus scenarios passed."