feat(msg-bus): add file and folder mutation notoficaton + tests

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(())
     }
 }