feat: communicate store changes from background processing (#342)

A background processing task fetches new messages and stores them in the
database. This task is triggered by push notifications.

This commit introduces a store notifications queue. The background task
stores in the queue any changes made to the store in the background.
When the foreground application reconnects to the websocket, it dequeues
the pending store notifications and communicates them to the UI so it
can update its state.

Fixes #337

Author: boxdot

apiclient/src/qs_api/ws.rs

@@ -3,10 +3,10 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 use core::time;
-use std::time::Duration;
+use std::{pin::pin, time::Duration};
 
 use base64::{engine::general_purpose, Engine as _};
-use futures_util::{pin_mut, SinkExt, StreamExt};
+use futures_util::{SinkExt, StreamExt};
 use http::{HeaderValue, Request};
 use phnxtypes::{
     codec::PhnxCodec,
@@ -116,7 +116,7 @@ impl QsWebSocket {
         let mut interval = tokio::time::interval(Duration::from_secs(1));
 
         // Pin the stream
-        pin_mut!(ws_stream);
+        let mut ws_stream = pin!(ws_stream);
 
         // Initialize the connection status
         let mut connection_status = ConnectionStatus::new();

applogic/src/api/user_cubit.rs

@@ -10,7 +10,7 @@ use std::time::Duration;
 use anyhow::bail;
 use flutter_rust_bridge::frb;
 use phnxapiclient::qs_api::ws::WsEvent;
-use phnxcoreclient::{clients::CoreUser, ConversationId};
+use phnxcoreclient::{clients::CoreUser, store::Store, ConversationId};
 use phnxcoreclient::{Asset, UserProfile};
 use phnxtypes::identifiers::QualifiedUserName;
 use phnxtypes::messages::client_ds::QsWsMessage;
@@ -318,7 +318,15 @@ fn spawn_polling(core_user: CoreUser, cancel: CancellationToken) {
 
 async fn handle_websocket_message(event: WsEvent, core_user: &CoreUser) {
     match event {
-        WsEvent::ConnectedEvent => info!("connected to websocket"),
+        WsEvent::ConnectedEvent => {
+            info!("connected to websocket");
+            // After (re)connecting, dequeue any pending store notifications that might have been
+            // enqueued by the push notifications background processing task.
+            match core_user.dequeue_notification().await {
+                Ok(notification) => core_user.notify(notification),
+                Err(error) => error!(%error, "Failed to dequeue store notification"),
+            }
+        }
         WsEvent::DisconnectedEvent => info!("disconnected from websocket"),
         WsEvent::MessageEvent(QsWsMessage::Event(event)) => {
             warn!("ignoring websocket event: {event:?}")

applogic/src/background_execution/processing.rs

@@ -2,6 +2,7 @@
 //
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+use phnxcoreclient::store::Store;
 use std::panic::{self, AssertUnwindSafe};
 use tokio::runtime::Builder;
 use tracing::{error, info};
@@ -76,6 +77,9 @@ pub(crate) async fn retrieve_messages(path: String) -> NotificationBatch {
         }
     };
 
+    // capture store notification in below store calls
+    let pending_store_notifications = user.user.subscribe_iter();
+
     let notifications = match user.fetch_all_messages().await {
         Ok(fetched_messages) => {
             info!("All messages fetched");
@@ -100,6 +104,12 @@ pub(crate) async fn retrieve_messages(path: String) -> NotificationBatch {
 
     let badge_count = user.global_unread_messages_count().await;
 
+    for store_notification in pending_store_notifications {
+        if let Err(error) = Store::enqueue_notification(&user.user, &store_notification).await {
+            error!(%error, "Failed to enqueue store notification");
+        }
+    }
+
     NotificationBatch {
         badge_count,
         removals: Vec::new(),

coreclient/migrations/V3__store_notifications_queue.rs

@@ -0,0 +1,9 @@
+// SPDX-FileCopyrightText: 2025 Phoenix R&D GmbH <[email protected]>
+//
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use crate::store::StoreNotification;
+
+pub fn migration() -> String {
+    StoreNotification::CREATE_TABLE_STATEMENT.to_string()
+}

coreclient/src/clients/mod.rs

@@ -274,21 +274,49 @@ impl CoreUser {
     }
 
     pub(crate) fn send_store_notification(&self, notification: StoreNotification) {
-        if !notification.ops.is_empty() {
+        if !notification.is_empty() {
             self.inner.store_notifications_tx.notify(notification);
         }
     }
 
+    /// Subscribes to store notifications.
+    ///
+    /// All notifications sent after this function was called are observed as items of the returned
+    /// stream.
     pub(crate) fn subscribe_to_store_notifications(
         &self,
     ) -> impl Stream<Item = Arc<StoreNotification>> + Send + 'static {
         self.inner.store_notifications_tx.subscribe()
     }
 
+    /// Subcribes to pending store notifications.
+    ///
+    /// Unlike `subscribe_to_store_notifications`, this function does not remove stored
+    /// notifications from the persisted queue.
+    pub(crate) fn subscribe_iter_to_store_notifications(
+        &self,
+    ) -> impl Iterator<Item = Arc<StoreNotification>> + Send + 'static {
+        self.inner.store_notifications_tx.subscribe_iter()
+    }
+
     pub(crate) fn store_notifier(&self) -> StoreNotifier {
         StoreNotifier::new(self.inner.store_notifications_tx.clone())
     }
 
+    pub(crate) async fn enqueue_store_notification(
+        &self,
+        notification: &StoreNotification,
+    ) -> Result<()> {
+        notification.enqueue(&mut *self.inner.connection.lock().await)?;
+        Ok(())
+    }
+
+    pub(crate) async fn dequeue_store_notification(&self) -> Result<StoreNotification> {
+        Ok(StoreNotification::dequeue(
+            &mut *self.inner.connection.lock().await,
+        )?)
+    }
+
     pub async fn set_own_user_profile(&self, mut user_profile: UserProfile) -> Result<()> {
         if user_profile.user_name() != &self.user_name() {
             bail!("Can't set user profile for users other than the current user.",);

coreclient/src/store/impl.rs

@@ -167,4 +167,16 @@ impl Store for CoreUser {
     fn subscribe(&self) -> impl Stream<Item = Arc<StoreNotification>> + Send + 'static {
         self.subscribe_to_store_notifications()
     }
+
+    fn subscribe_iter(&self) -> impl Iterator<Item = Arc<StoreNotification>> + Send + 'static {
+        self.subscribe_iter_to_store_notifications()
+    }
+
+    async fn enqueue_notification(&self, notification: &StoreNotification) -> StoreResult<()> {
+        self.enqueue_store_notification(notification).await
+    }
+
+    async fn dequeue_notification(&self) -> StoreResult<StoreNotification> {
+        self.dequeue_store_notification().await
+    }
 }

coreclient/src/store/mod.rs

@@ -19,6 +19,7 @@ pub(crate) use notification::{StoreNotificationsSender, StoreNotifier};
 
 mod r#impl;
 mod notification;
+mod persistence;
 
 /// The result type of a failable [`Store`] method
 pub type StoreResult<T> = anyhow::Result<T>;
@@ -133,4 +134,10 @@ pub trait LocalStore {
     fn notify(&self, notification: StoreNotification);
 
     fn subscribe(&self) -> impl Stream<Item = Arc<StoreNotification>> + Send + 'static;
+
+    fn subscribe_iter(&self) -> impl Iterator<Item = Arc<StoreNotification>> + Send + 'static;
+
+    async fn enqueue_notification(&self, notification: &StoreNotification) -> StoreResult<()>;
+
+    async fn dequeue_notification(&self) -> StoreResult<StoreNotification>;
 }

coreclient/src/store/notification.rs

@@ -108,7 +108,13 @@ impl StoreNotificationsSender {
 
     /// Sends a notification to all current subscribers.
     pub(crate) fn notify(&self, notification: impl Into<Arc<StoreNotification>>) {
-        let _no_receivers = self.tx.send(notification.into());
+        let notification = notification.into();
+        debug!(
+            num_receivers = self.tx.receiver_count(),
+            ?notification,
+            "StoreNotificationsSender::notify"
+        );
+        let _no_receivers = self.tx.send(notification);
     }
 
     /// Creates a new subscription to the notifications.
@@ -126,6 +132,31 @@ impl StoreNotificationsSender {
             }
         })
     }
+
+    /// Returns all pending notifications.
+    ///
+    /// The pending notifications are the notifications captured starting at the call to this function.
+    /// Getting the next item from the iterator gets the next pending notification is there is any,
+    /// otherwise it returns `None`. Therefore, the iterator is not fused.
+    ///
+    /// This is useful for capturing all pending notifications synchronously.
+    pub(crate) fn subscribe_iter(
+        &self,
+    ) -> impl Iterator<Item = Arc<StoreNotification>> + Send + 'static {
+        let mut rx = self.tx.subscribe();
+        std::iter::from_fn(move || loop {
+            match rx.try_recv() {
+                Ok(notification) => return Some(notification),
+                Err(broadcast::error::TryRecvError::Lagged(n)) => {
+                    error!(n, "store notifications lagged");
+                    continue;
+                }
+                Err(
+                    broadcast::error::TryRecvError::Closed | broadcast::error::TryRecvError::Empty,
+                ) => return None,
+            }
+        })
+    }
 }
 
 impl Default for StoreNotificationsSender {
@@ -139,6 +170,7 @@ impl Default for StoreNotificationsSender {
 /// Bundles all changes to the store, that is, all entities that have been added, updated or
 /// removed.
 #[derive(Debug, Default)]
+#[cfg_attr(test, derive(PartialEq, Eq))]
 pub struct StoreNotification {
     pub ops: BTreeMap<StoreEntityId, StoreOperation>,
 }
@@ -147,6 +179,10 @@ impl StoreNotification {
     fn empty() -> Self {
         Self::default()
     }
+
+    pub fn is_empty(&self) -> bool {
+        self.ops.is_empty()
+    }
 }
 
 /// Operation which was performed in a [`super::Store`]
@@ -168,3 +204,76 @@ pub enum StoreEntityId {
     Conversation(ConversationId),
     Message(ConversationMessageId),
 }
+
+impl StoreEntityId {
+    pub(crate) fn kind(&self) -> StoreEntityKind {
+        match self {
+            StoreEntityId::User(_) => StoreEntityKind::User,
+            StoreEntityId::Conversation(_) => StoreEntityKind::Conversation,
+            StoreEntityId::Message(_) => StoreEntityKind::Message,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub(crate) enum StoreEntityKind {
+    User = 0,
+    Conversation = 1,
+    Message = 2,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn subscribe_iter() {
+        let tx = StoreNotificationsSender::new();
+
+        let ops_1: BTreeMap<_, _> = [(
+            StoreEntityId::User("alice@localhost".parse().unwrap()),
+            StoreOperation::Add,
+        )]
+        .into_iter()
+        .collect();
+
+        let ops_2: BTreeMap<_, _> = [(
+            StoreEntityId::User("bob@localhost".parse().unwrap()),
+            StoreOperation::Update,
+        )]
+        .into_iter()
+        .collect();
+
+        let ops_3: BTreeMap<_, _> = [(
+            StoreEntityId::User("eve@localhost".parse().unwrap()),
+            StoreOperation::Remove,
+        )]
+        .into_iter()
+        .collect();
+
+        let ops_4: BTreeMap<_, _> = [(
+            StoreEntityId::User("mallory@localhost".parse().unwrap()),
+            StoreOperation::Add,
+        )]
+        .into_iter()
+        .collect();
+
+        tx.notify(StoreNotification {
+            ops: ops_1.into_iter().collect(),
+        });
+
+        let mut iter = tx.subscribe_iter();
+
+        tx.notify(StoreNotification { ops: ops_2.clone() });
+
+        // first notification is not observed, because it was sent before the subscription
+        assert_eq!(iter.next().unwrap().ops, ops_2);
+        assert_eq!(iter.next(), None);
+
+        tx.notify(StoreNotification { ops: ops_3.clone() });
+        assert_eq!(iter.next().unwrap().ops, ops_3);
+        tx.notify(StoreNotification { ops: ops_4.clone() });
+        assert_eq!(iter.next().unwrap().ops, ops_4);
+        assert_eq!(iter.next(), None);
+    }
+}