feat(notification): add persistent notification

src/application/ports/message_bus_ports.rs

@@ -64,6 +64,16 @@ pub enum Topic {
     /// the eviction wiring lands (Phase-A follow-up).
     UserAuthz(Uuid),
 
+    /// A user's private notifications channel — poked when a
+    /// [`MessageBusEvent::NotificationReceived`] event fires. The WS
+    /// handler auto-subscribes each session at session open (same
+    /// pattern as [`Topic::UserAuthz`]). Payload is a thin fact
+    /// (`notification_id` + `kind`); the client refetches the row from
+    /// `GET /api/notifications` for the details. AuthZ: **strict
+    /// identity match** — no admin bypass, direct UUID equality,
+    /// anti-enumeration parity with [`Topic::UserAuthz`].
+    UserNotifications(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…)
@@ -83,6 +93,7 @@ impl Topic {
         match self {
             Topic::Folder(id) => format!("folder:{id}"),
             Topic::UserAuthz(id) => format!("user:{id}:authz"),
+            Topic::UserNotifications(id) => format!("user:{id}:notifications"),
             Topic::Job(name) => format!("job:{name}"),
         }
     }
@@ -97,10 +108,14 @@ impl Topic {
             return Ok(Topic::Folder(id));
         }
         if let Some(rest) = s.strip_prefix("user:")
-            && let Some((id_str, "authz")) = rest.rsplit_once(':')
+            && let Some((id_str, suffix)) = rest.rsplit_once(':')
         {
             let id = Uuid::parse_str(id_str).map_err(|_| ParseTopicErr::BadUuid)?;
-            return Ok(Topic::UserAuthz(id));
+            return match suffix {
+                "authz" => Ok(Topic::UserAuthz(id)),
+                "notifications" => Ok(Topic::UserNotifications(id)),
+                _ => Err(ParseTopicErr::Unknown),
+            };
         }
         if let Some(name) = s.strip_prefix("job:") {
             // Job names are scheduler-registered short slugs — see
@@ -135,6 +150,7 @@ impl Topic {
                 resource: BusResource::Folder(*id),
             },
             Topic::UserAuthz(id) => AuthzCheck::IdentityMatch { user_id: *id },
+            Topic::UserNotifications(id) => AuthzCheck::IdentityMatch { user_id: *id },
             Topic::Job(_) => AuthzCheck::RoleAdmin,
         }
     }
@@ -291,6 +307,25 @@ pub enum MessageBusEvent {
     /// plan's Phase-B roadmap.
     AuthzChanged { affected_folders: Vec<Uuid> },
 
+    /// A new notification was created for the caller — publishes on
+    /// [`Topic::UserNotifications`]. Payload is deliberately thin: the
+    /// FE learns "there's something new to look at" and calls
+    /// `GET /api/notifications` to load the row. Same recovery path a
+    /// missed push takes on next mount, so the wire event stays a
+    /// pure poke — no fields the bell needs to render on its own.
+    ///
+    /// `kind` is the notification's registered kind slug
+    /// (`share_granted`, `job_completed_for_you`,
+    /// `new_login_from_new_device`, `storage_quota_threshold`, …).
+    /// The FE may use it to route the toast (high-priority kinds pop
+    /// a toast; low-priority ones just bump the badge) but never
+    /// treats it as authoritative — the DB row is the truth.
+    NotificationReceived {
+        notification_id: Uuid,
+        kind: String,
+        created_at: chrono::DateTime<chrono::Utc>,
+    },
+
     /// 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
@@ -497,6 +532,15 @@ mod tests {
         assert_eq!(Topic::parse(&wire).unwrap(), t);
     }
 
+    #[test]
+    fn user_notifications_topic_roundtrip() {
+        let id = Uuid::new_v4();
+        let t = Topic::UserNotifications(id);
+        let wire = t.to_wire_key();
+        assert_eq!(wire, format!("user:{id}:notifications"));
+        assert_eq!(Topic::parse(&wire).unwrap(), t);
+    }
+
     #[test]
     fn job_topic_roundtrip() {
         let t = Topic::Job("backend_migration".to_string());
@@ -539,7 +583,12 @@ mod tests {
         assert_eq!(
             Topic::parse(&format!("user:{}", Uuid::new_v4())),
             Err(ParseTopicErr::Unknown),
-            "user:<uuid> without :authz suffix is not a known topic in MVP"
+            "user:<uuid> without a known suffix (:authz, :notifications) is not a known topic"
+        );
+        assert_eq!(
+            Topic::parse(&format!("user:{}:whatever", Uuid::new_v4())),
+            Err(ParseTopicErr::Unknown),
+            "an unrecognised suffix rejects — no partial match on the prefix"
         );
     }
 
@@ -563,6 +612,18 @@ mod tests {
         );
     }
 
+    #[test]
+    fn required_perm_user_notifications_is_identity_match() {
+        // Same strict-privacy gate as :authz — no admin bypass, direct
+        // UUID equality, anti-enum parity. A regression here would
+        // let admins snoop on other users' notification streams.
+        let id = Uuid::new_v4();
+        assert_eq!(
+            Topic::UserNotifications(id).required_perm(),
+            AuthzCheck::IdentityMatch { user_id: id }
+        );
+    }
+
     #[test]
     fn event_serializes_with_snake_case_discriminator() {
         // The `#[serde(tag = "event")]` shape is the WS wire contract for
@@ -651,6 +712,14 @@ mod tests {
                 },
                 "authz_changed",
             ),
+            (
+                MessageBusEvent::NotificationReceived {
+                    notification_id: Uuid::nil(),
+                    kind: "share_granted".into(),
+                    created_at: chrono::DateTime::<chrono::Utc>::from_timestamp(0, 0).unwrap(),
+                },
+                "notification_received",
+            ),
             (
                 MessageBusEvent::JobRunStarted {
                     name: "backend_migration".into(),

src/application/services/mod.rs

@@ -25,6 +25,7 @@ pub mod mount_registry;
 pub mod music_service;
 pub mod nextcloud_file_id_service;
 pub mod nextcloud_login_flow_service;
+pub mod notification_application_service;
 pub mod people_service;
 pub mod places_service;
 pub mod recent_service;

src/application/services/notification_application_service.rs

@@ -0,0 +1,127 @@
+//! Orchestrates persistent notifications.
+//!
+//! `create()` is the single ingester entry point:
+//!
+//! 1. Insert the row via [`NotificationRepository::create`].
+//! 2. Publish a thin `NotificationReceived` event on
+//!    `user:{user_id}:notifications` so subscribed sessions refetch
+//!    immediately.
+//!
+//! The DB row is the truth (see `docs/plan/message-bus.md § Slice E`).
+//! The bus is best-effort — a subscriber offline at publish time
+//! recovers on next `GET /api/notifications`. Publish happens AFTER
+//! the DB write succeeds, never inside a transaction — the plan's
+//! "publish after commit" invariant.
+//!
+//! Reads (`list_for_user`, `count_unread_for_user`) and state changes
+//! (`mark_read`, `mark_all_read`, `delete`) back the REST endpoints in
+//! `interfaces/api/handlers/notifications.rs`. Every mutating method
+//! is scoped on `user_id` at the SQL layer; the service does not run
+//! its own AuthZ check because the identity is by construction
+//! (`caller_id == user_id`, extracted from the auth middleware).
+
+use std::sync::Arc;
+
+use chrono::Utc;
+use uuid::Uuid;
+
+use crate::application::ports::message_bus_ports::{MessageBus, MessageBusEvent, Topic};
+use crate::common::errors::DomainError;
+use crate::domain::entities::notification::{NewNotification, Notification};
+use crate::domain::repositories::notification_repository::{
+    NotificationListFilter, NotificationRepository,
+};
+
+pub struct NotificationApplicationService {
+    repo: Arc<dyn NotificationRepository>,
+    bus: Arc<dyn MessageBus>,
+}
+
+impl NotificationApplicationService {
+    pub fn new(repo: Arc<dyn NotificationRepository>, bus: Arc<dyn MessageBus>) -> Self {
+        Self { repo, bus }
+    }
+
+    /// Insert a row for `new_notif` and publish a thin bus event.
+    /// Returns the persisted row. This is the ingester-facing method
+    /// — called from `ShareService::create_grant`,
+    /// `AuthApplicationService` (new-device login),
+    /// `SchedulerEngine` (job completed for actor), and the quota
+    /// threshold hook.
+    pub async fn create(&self, new_notif: NewNotification) -> Result<Notification, DomainError> {
+        let row = self.repo.create(&new_notif).await?;
+
+        // Publish AFTER the row is durable. Silent no-op if the bus
+        // is disabled at boot (`OXICLOUD_MESSAGEBUS_ENABLE=false`) —
+        // the WS route is unmounted so the publish just hits a dead
+        // sender. The FE bell still works: it reads from the DB on
+        // mount. See plan § "Slice E".
+        self.bus.publish(
+            &Topic::UserNotifications(row.user_id),
+            MessageBusEvent::NotificationReceived {
+                notification_id: row.id,
+                kind: row.kind.clone(),
+                created_at: row.created_at,
+            },
+        );
+
+        Ok(row)
+    }
+
+    /// List notifications for `user_id` newest-first. Default limit at
+    /// this layer is 50 rows (the repo caps at 500 defensively).
+    pub async fn list_for_user(
+        &self,
+        user_id: Uuid,
+        filter: NotificationListFilter,
+    ) -> Result<Vec<Notification>, DomainError> {
+        self.repo.list_for_user(user_id, &filter).await
+    }
+
+    /// Unread badge count.
+    pub async fn count_unread_for_user(&self, user_id: Uuid) -> Result<i64, DomainError> {
+        self.repo.count_unread_for_user(user_id).await
+    }
+
+    /// Mark one notification as read. Returns `true` if the row
+    /// transitioned unread → read (i.e. was owned by `caller_id` and
+    /// was previously unread). Returns `false` for already-read,
+    /// missing, or misowned rows — indistinguishable at the wire so
+    /// enumeration doesn't leak.
+    pub async fn mark_read(
+        &self,
+        notification_id: Uuid,
+        caller_id: Uuid,
+    ) -> Result<bool, DomainError> {
+        self.repo
+            .mark_read(notification_id, caller_id, Utc::now())
+            .await
+    }
+
+    /// Bulk mark-all-read. Returns rows updated.
+    pub async fn mark_all_read(&self, caller_id: Uuid) -> Result<u64, DomainError> {
+        self.repo
+            .mark_all_read_for_user(caller_id, Utc::now())
+            .await
+    }
+
+    /// Hard-delete one row. Same anti-enumeration semantics as
+    /// [`mark_read`] — returns `false` for missing / misowned.
+    pub async fn delete(
+        &self,
+        notification_id: Uuid,
+        caller_id: Uuid,
+    ) -> Result<bool, DomainError> {
+        self.repo.delete_by_id(notification_id, caller_id).await
+    }
+
+    /// Retention job entry point. Called by `notifications_cleanup`
+    /// on its daily cadence — deletes read rows older than `cutoff`.
+    /// Unread rows are always preserved.
+    pub async fn purge_read_before_cutoff(
+        &self,
+        cutoff: chrono::DateTime<Utc>,
+    ) -> Result<u64, DomainError> {
+        self.repo.purge_read_before(cutoff).await
+    }
+}