add CountableCallback

Author: shenjinti

crates/restsend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "restsend-sdk"
-version = "1.0.7"
+version = "1.0.9"
 edition = "2021"
 
 [lib]

crates/restsend/src/callback.rs

@@ -50,7 +50,13 @@ pub trait RsCallback: Send + Sync {
     fn on_conversations_updated(&self, conversations: Vec<Conversation>) {}
     fn on_conversation_removed(&self, conversation_id: String) {}
 }
-
+#[allow(unused_variables)]
+#[export_wasm_or_ffi(#[uniffi::export(callback_interface)])]
+pub trait CountableCallback: Send + Sync {
+    fn is_countable(&self, content: Content) -> bool {
+        !content.unreadable
+    }
+}
 #[allow(unused_variables)]
 #[export_wasm_or_ffi(#[uniffi::export(callback_interface)])]
 pub trait UploadCallback: Send + Sync {

crates/restsend/src/client/conversation.rs

@@ -91,7 +91,13 @@ impl Client {
             .ok()?;
 
         for c in r.items.iter() {
-            if !c.content.unreadable {
+            let is_countable =
+                if let Some(cb) = self.store.countable_callback.lock().unwrap().as_ref() {
+                    cb.is_countable(c.content.clone())
+                } else {
+                    !c.content.unreadable
+                };
+            if is_countable {
                 conversation.last_message_at = c.created_at.clone();
                 conversation.last_message = Some(c.content.clone());
                 conversation.last_sender_id = c.sender_id.clone();
@@ -117,7 +123,12 @@ impl Client {
         ensure_conversation_last_version: Option<bool>,
     ) {
         let st = now_millis();
-        let limit = if limit == 0 { MAX_LOGS_LIMIT/2 } else { limit }.min(MAX_LOGS_LIMIT);
+        let limit = if limit == 0 {
+            MAX_LOGS_LIMIT / 2
+        } else {
+            limit
+        }
+        .min(MAX_LOGS_LIMIT);
         let conversation = self
             .store
             .get_conversation(
@@ -127,9 +138,13 @@ impl Client {
             )
             .await
             .unwrap_or_default();
-        
+
         let store_st = now_millis();
-        match self.store.get_chat_logs(&topic_id, conversation.start_seq, last_seq, limit).await {
+        match self
+            .store
+            .get_chat_logs(&topic_id, conversation.start_seq, last_seq, limit)
+            .await
+        {
             Ok((local_logs, need_fetch)) => {
                 info!(
                     "sync_chat_logs local_logs.len: {} start_seq: {} last_seq: {:?} limit: {} local_logs.start_sort_value:{} local_logs.end_sort_value:{} need_fetch:{} store_cost:{:?} total_cost:{:?}",
@@ -165,10 +180,7 @@ impl Client {
         }
     }
 
-    pub async fn save_chat_logs(
-        &self,
-       logs: &Vec<ChatLog>,
-    )  -> Result<()> {
+    pub async fn save_chat_logs(&self, logs: &Vec<ChatLog>) -> Result<()> {
         self.store.save_chat_logs(logs).await
     }
 
@@ -437,17 +449,32 @@ impl Client {
                 None => continue,
             };
 
+            if self.store.countable_callback.lock().unwrap().is_some() {
+                conversation.unread = 0
+            }
+
             for c in lr.items.iter() {
                 if c.seq <= conversation.last_seq {
                     continue;
                 }
                 conversation.last_seq = c.seq;
-                if !c.content.unreadable {
+
+                let is_countable =
+                    if let Some(cb) = self.store.countable_callback.lock().unwrap().as_ref() {
+                        cb.is_countable(c.content.clone())
+                    } else {
+                        !c.content.unreadable
+                    };
+
+                if is_countable {
                     conversation.updated_at = c.created_at.clone();
                     conversation.last_message_at = c.created_at.clone();
                     conversation.last_message = Some(c.content.clone());
                     conversation.last_sender_id = c.sender_id.clone();
                     conversation.last_message_seq = Some(c.seq);
+                    if conversation.last_read_seq < c.seq {
+                        conversation.unread += 1;
+                    }
                 }
                 break;
             }
@@ -457,6 +484,11 @@ impl Client {
         if let Some(cb) = self.store.callback.lock().unwrap().as_ref() {
             cb.on_conversations_updated(updated_conversations.clone());
         }
+        // sync to store
+        let t = self.store.message_storage.table::<Conversation>().await;
+        for c in updated_conversations.iter_mut() {
+            t.set("", &c.topic_id, Some(c)).await.ok();
+        }
         Ok(())
     }
 
@@ -534,10 +566,7 @@ impl Client {
     ) -> Result<Conversation> {
         self.store.set_conversation_extra(&topic_id, extra).await
     }
-    pub async fn clear_conversation(
-        &self,
-        topic_id: String
-    ) -> Result<()> {
+    pub async fn clear_conversation(&self, topic_id: String) -> Result<()> {
         self.store.clear_conversation(&topic_id).await
     }
 }

crates/restsend/src/client/mod.rs

@@ -3,7 +3,7 @@ use self::{
     store::{ClientStore, ClientStoreRef},
 };
 use crate::{
-    callback::{DownloadCallback, RsCallback},
+    callback::{CountableCallback, DownloadCallback, RsCallback},
     error::ClientError,
     media::{build_download_url, download_file},
     models::{AuthInfo, User},
@@ -92,6 +92,9 @@ impl Client {
     pub fn set_callback(&self, callback: Option<Box<dyn RsCallback>>) {
         *self.store.callback.lock().unwrap() = callback;
     }
+    pub fn set_countable_callback(&self, callback: Option<Box<dyn CountableCallback>>) {
+        *self.store.countable_callback.lock().unwrap() = callback;
+    }
 
     pub async fn get_user(&self, user_id: String, blocking: bool) -> Option<User> {
         self.store.get_user(&user_id, blocking).await

crates/restsend/src/client/store/conversations.rs

@@ -228,7 +228,10 @@ impl ClientStore {
 
             if conversation.unread == 0 {
                 let start_seq = conversation.start_seq.max(conversation.last_read_seq);
-                let diff = conversation.last_seq - start_seq;
+                let diff = conversation
+                    .last_message_seq
+                    .unwrap_or(conversation.last_seq)
+                    - start_seq;
                 conversation.unread = diff.max(0);
             }
 

crates/restsend/src/client/store/mod.rs

@@ -1,5 +1,5 @@
 use self::attachments::UploadTask;
-use crate::callback::RsCallback;
+use crate::callback::{CountableCallback, RsCallback};
 use crate::models::Attachment;
 use crate::storage::Storage;
 use crate::utils::{elapsed, now_millis};
@@ -99,6 +99,7 @@ type PendingRequests = Arc<Mutex<HashMap<String, PendingRequest>>>;
 
 pub(super) type ClientStoreRef = Arc<ClientStore>;
 pub(super) type CallbackRef = Arc<Mutex<Option<Box<dyn RsCallback>>>>;
+pub(super) type CountableCallbackRef = Arc<Mutex<Option<Box<dyn CountableCallback>>>>;
 pub(super) struct ClientStore {
     user_id: String,
     endpoint: String,
@@ -110,6 +111,7 @@ pub(super) struct ClientStore {
     removed_conversations: Mutex<HashMap<String, i64>>,
     pub(crate) message_storage: Arc<Storage>,
     pub(crate) callback: CallbackRef,
+    pub(crate) countable_callback: CountableCallbackRef,
     incoming_logs: Mutex<HashMap<String, Vec<String>>>,
 }
 
@@ -132,6 +134,7 @@ impl ClientStore {
             removed_conversations: Mutex::new(HashMap::new()),
             message_storage: Arc::new(Storage::new(db_path)),
             callback: Arc::new(Mutex::new(None)),
+            countable_callback: Arc::new(Mutex::new(None)),
             incoming_logs: Mutex::new(HashMap::new()),
         }
     }