core: optimize marking chat items as read, batch API (#4843)

* core: optimize marking chat items as read

* tests, ui types

* ios: fix api

* refactor

apps/ios/Shared/Model/SimpleXAPI.swift

@@ -1000,6 +1000,10 @@ func apiChatRead(type: ChatType, id: Int64, itemRange: (Int64, Int64)) async thr
     try await sendCommandOkResp(.apiChatRead(type: type, id: id, itemRange: itemRange))
 }
 
+func apiChatItemsRead(type: ChatType, id: Int64, itemIds: [Int64]) async throws {
+    try await sendCommandOkResp(.apiChatItemsRead(type: type, id: id, itemIds: itemIds))
+}
+
 func apiChatUnread(type: ChatType, id: Int64, unreadChat: Bool) async throws {
     try await sendCommandOkResp(.apiChatUnread(type: type, id: id, unreadChat: unreadChat))
 }

apps/ios/SimpleXChat/APITypes.swift

@@ -129,6 +129,7 @@ public enum ChatCommand {
     // WebRTC calls /
     case apiGetNetworkStatuses
     case apiChatRead(type: ChatType, id: Int64, itemRange: (Int64, Int64))
+    case apiChatItemsRead(type: ChatType, id: Int64, itemIds: [Int64])
     case apiChatUnread(type: ChatType, id: Int64, unreadChat: Bool)
     case receiveFile(fileId: Int64, userApprovedRelays: Bool, encrypted: Bool?, inline: Bool?)
     case setFileToReceive(fileId: Int64, userApprovedRelays: Bool, encrypted: Bool?)
@@ -293,6 +294,7 @@ public enum ChatCommand {
             case let .apiCallStatus(contact, callStatus): return "/_call status @\(contact.apiId) \(callStatus.rawValue)"
             case .apiGetNetworkStatuses: return "/_network_statuses"
             case let .apiChatRead(type, id, itemRange: (from, to)): return "/_read chat \(ref(type, id)) from=\(from) to=\(to)"
+            case let .apiChatItemsRead(type, id, itemIds): return "/_read chat items \(ref(type, id)) \(joinedIds(itemIds))"
             case let .apiChatUnread(type, id, unreadChat): return "/_unread chat \(ref(type, id)) \(onOff(unreadChat))"
             case let .receiveFile(fileId, userApprovedRelays, encrypt, inline): return "/freceive \(fileId)\(onOffParam("approved_relays", userApprovedRelays))\(onOffParam("encrypt", encrypt))\(onOffParam("inline", inline))"
             case let .setFileToReceive(fileId, userApprovedRelays, encrypt): return "/_set_file_to_receive \(fileId)\(onOffParam("approved_relays", userApprovedRelays))\(onOffParam("encrypt", encrypt))"
@@ -434,6 +436,7 @@ public enum ChatCommand {
             case .apiCallStatus: return "apiCallStatus"
             case .apiGetNetworkStatuses: return "apiGetNetworkStatuses"
             case .apiChatRead: return "apiChatRead"
+            case .apiChatItemsRead: return "apiChatItemsRead"
             case .apiChatUnread: return "apiChatUnread"
             case .receiveFile: return "receiveFile"
             case .setFileToReceive: return "setFileToReceive"
@@ -462,6 +465,10 @@ public enum ChatCommand {
         "\(type.rawValue)\(id)"
     }
 
+    func joinedIds(_ ids: [Int64]) -> String {
+        ids.map { "\($0)" }.joined(separator: ",")
+    }
+    
     func protoServersStr(_ servers: [ServerCfg]) -> String {
         encodeJSON(ProtoServersConfig(servers: servers))
     }

apps/multiplatform/common/src/commonMain/kotlin/chat/simplex/common/model/SimpleXAPI.kt

@@ -1480,6 +1480,13 @@ object ChatController {
     return false
   }
 
+  suspend fun apiChatItemsRead(rh: Long?, type: ChatType, id: Long, itemIds: List<Long>): Boolean {
+    val r = sendCmd(rh, CC.ApiChatItemsRead(type, id, itemIds))
+    if (r is CR.CmdOk) return true
+    Log.e(TAG, "apiChatItemsRead bad response: ${r.responseType} ${r.details}")
+    return false
+  }
+
   suspend fun apiChatUnread(rh: Long?, type: ChatType, id: Long, unreadChat: Boolean): Boolean {
     val r = sendCmd(rh, CC.ApiChatUnread(type, id, unreadChat))
     if (r is CR.CmdOk) return true
@@ -2967,6 +2974,7 @@ sealed class CC {
   class ApiAcceptContact(val incognito: Boolean, val contactReqId: Long): CC()
   class ApiRejectContact(val contactReqId: Long): CC()
   class ApiChatRead(val type: ChatType, val id: Long, val range: ItemRange): CC()
+  class ApiChatItemsRead(val type: ChatType, val id: Long, val itemIds: List<Long>): CC()
   class ApiChatUnread(val type: ChatType, val id: Long, val unreadChat: Boolean): CC()
   class ReceiveFile(val fileId: Long, val userApprovedRelays: Boolean, val encrypt: Boolean, val inline: Boolean?): CC()
   class CancelFile(val fileId: Long): CC()
@@ -3123,6 +3131,7 @@ sealed class CC {
     is ApiCallStatus -> "/_call status @${contact.apiId} ${callStatus.value}"
     is ApiGetNetworkStatuses -> "/_network_statuses"
     is ApiChatRead -> "/_read chat ${chatRef(type, id)} from=${range.from} to=${range.to}"
+    is ApiChatItemsRead -> "/_read chat items ${chatRef(type, id)} ${itemIds.joinToString(",")}"
     is ApiChatUnread -> "/_unread chat ${chatRef(type, id)} ${onOff(unreadChat)}"
     is ReceiveFile ->
       "/freceive $fileId" +
@@ -3266,6 +3275,7 @@ sealed class CC {
     is ApiCallStatus -> "apiCallStatus"
     is ApiGetNetworkStatuses -> "apiGetNetworkStatuses"
     is ApiChatRead -> "apiChatRead"
+    is ApiChatItemsRead -> "apiChatItemsRead"
     is ApiChatUnread -> "apiChatUnread"
     is ReceiveFile -> "receiveFile"
     is CancelFile -> "cancelFile"

src/Simplex/Chat.hs

@@ -55,9 +55,9 @@ import Data.Text.Encoding (decodeLatin1, encodeUtf8)
 import Data.Time (NominalDiffTime, addUTCTime, defaultTimeLocale, formatTime)
 import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime, nominalDay, nominalDiffTimeToSeconds)
 import Data.Time.Clock.System (systemToUTCTime)
-import Data.Word (Word32)
 import qualified Data.UUID as UUID
 import qualified Data.UUID.V4 as V4
+import Data.Word (Word32)
 import qualified Database.SQLite.Simple as SQL
 import Simplex.Chat.Archive
 import Simplex.Chat.Call
@@ -115,7 +115,7 @@ import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (base64P)
-import Simplex.Messaging.Protocol (AProtoServerWithAuth (..), AProtocolType (..), EntityId, ErrorType (..), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth (..), ProtocolServer, ProtocolType (..), ProtocolTypeI (..), SProtocolType (..), SubscriptionMode (..), UserProtocol, XFTPServer, userProtocol)
+import Simplex.Messaging.Protocol (AProtoServerWithAuth (..), AProtocolType (..), ErrorType (..), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth (..), ProtocolServer, ProtocolType (..), ProtocolTypeI (..), SProtocolType (..), SubscriptionMode (..), UserProtocol, XFTPServer, userProtocol)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import qualified Simplex.Messaging.TMap as TM
@@ -1114,26 +1114,24 @@ processChatCommand' vr = \case
                 when (size' > 0) $ copyChunks r w size'
   APIUserRead userId -> withUserId userId $ \user -> withFastStore' (`setUserChatsRead` user) >> ok user
   UserRead -> withUser $ \User {userId} -> processChatCommand $ APIUserRead userId
-  APIChatRead (ChatRef cType chatId) fromToIds -> withUser $ \_ -> case cType of
+  APIChatRead chatRef@(ChatRef cType chatId) fromToIds -> withUser $ \_ -> case cType of
     CTDirect -> do
       user <- withFastStore $ \db -> getUserByContactId db chatId
-      timedItems <- withFastStore' $ \db -> getDirectUnreadTimedItems db user chatId fromToIds
       ts <- liftIO getCurrentTime
-      forM_ timedItems $ \(itemId, ttl) -> do
-        let deleteAt = addUTCTime (realToFrac ttl) ts
-        withFastStore' $ \db -> setDirectChatItemDeleteAt db user chatId itemId deleteAt
-        startProximateTimedItemThread user (ChatRef CTDirect chatId, itemId) deleteAt
-      withFastStore' $ \db -> updateDirectChatItemsRead db user chatId fromToIds
+      timedItems <- withFastStore' $ \db -> do
+        timedItems <- getDirectUnreadTimedItems db user chatId fromToIds
+        updateDirectChatItemsRead db user chatId fromToIds
+        setDirectChatItemsDeleteAt db user chatId timedItems ts
+      forM_ timedItems $ \(itemId, deleteAt) -> startProximateTimedItemThread user (chatRef, itemId) deleteAt
       ok user
     CTGroup -> do
-      user@User {userId} <- withFastStore $ \db -> getUserByGroupId db chatId
-      timedItems <- withFastStore' $ \db -> getGroupUnreadTimedItems db user chatId fromToIds
+      user <- withFastStore $ \db -> getUserByGroupId db chatId
       ts <- liftIO getCurrentTime
-      forM_ timedItems $ \(itemId, ttl) -> do
-        let deleteAt = addUTCTime (realToFrac ttl) ts
-        withFastStore' $ \db -> setGroupChatItemDeleteAt db user chatId itemId deleteAt
-        startProximateTimedItemThread user (ChatRef CTGroup chatId, itemId) deleteAt
-      withFastStore' $ \db -> updateGroupChatItemsRead db userId chatId fromToIds
+      timedItems <- withFastStore' $ \db -> do
+        timedItems <- getGroupUnreadTimedItems db user chatId fromToIds
+        updateGroupChatItemsRead db user chatId fromToIds
+        setGroupChatItemsDeleteAt db user chatId timedItems ts
+      forM_ timedItems $ \(itemId, deleteAt) -> startProximateTimedItemThread user (chatRef, itemId) deleteAt
       ok user
     CTLocal -> do
       user <- withFastStore $ \db -> getUserByNoteFolderId db chatId
@@ -1141,6 +1139,24 @@ processChatCommand' vr = \case
       ok user
     CTContactRequest -> pure $ chatCmdError Nothing "not supported"
     CTContactConnection -> pure $ chatCmdError Nothing "not supported"
+  APIChatItemsRead chatRef@(ChatRef cType chatId) itemIds -> withUser $ \_ -> case cType of
+    CTDirect -> do
+      user <- withFastStore $ \db -> getUserByContactId db chatId
+      timedItems <- withFastStore' $ \db -> do
+        timedItems <- updateDirectChatItemsReadList db user chatId itemIds
+        setDirectChatItemsDeleteAt db user chatId timedItems =<< getCurrentTime
+      forM_ timedItems $ \(itemId, deleteAt) -> startProximateTimedItemThread user (chatRef, itemId) deleteAt
+      ok user
+    CTGroup -> do
+      user <- withFastStore $ \db -> getUserByGroupId db chatId
+      timedItems <- withFastStore' $ \db -> do
+        timedItems <- updateGroupChatItemsReadList db user chatId itemIds
+        setGroupChatItemsDeleteAt db user chatId timedItems =<< getCurrentTime
+      forM_ timedItems $ \(itemId, deleteAt) -> startProximateTimedItemThread user (chatRef, itemId) deleteAt
+      ok user
+    CTLocal -> pure $ chatCmdError Nothing "not supported"
+    CTContactRequest -> pure $ chatCmdError Nothing "not supported"
+    CTContactConnection -> pure $ chatCmdError Nothing "not supported"
   APIChatUnread (ChatRef cType chatId) unreadChat -> withUser $ \user -> case cType of
     CTDirect -> do
       withFastStore $ \db -> do
@@ -7873,6 +7889,7 @@ chatCommandP =
       "/_read user " *> (APIUserRead <$> A.decimal),
       "/read user" $> UserRead,
       "/_read chat " *> (APIChatRead <$> chatRefP <*> optional (A.space *> ((,) <$> ("from=" *> A.decimal) <* A.space <*> ("to=" *> A.decimal)))),
+      "/_read chat items " *> (APIChatItemsRead <$> chatRefP <*> _strP),
       "/_unread chat " *> (APIChatUnread <$> chatRefP <* A.space <*> onOffP),
       "/_delete " *> (APIDeleteChat <$> chatRefP <*> chatDeleteMode),
       "/_clear chat " *> (APIClearChat <$> chatRefP),

src/Simplex/Chat/Controller.hs

@@ -302,6 +302,7 @@ data ChatCommand
   | APIUserRead UserId
   | UserRead
   | APIChatRead ChatRef (Maybe (ChatItemId, ChatItemId))
+  | APIChatItemsRead ChatRef (NonEmpty ChatItemId)
   | APIChatUnread ChatRef Bool
   | APIDeleteChat ChatRef ChatDeleteMode -- currently delete mode settings are only applied to direct chats
   | APIClearChat ChatRef

src/Simplex/Chat/Store/Messages.hs

@@ -60,10 +60,12 @@ module Simplex.Chat.Store.Messages
     deleteLocalChatItem,
     updateDirectChatItemsRead,
     getDirectUnreadTimedItems,
-    setDirectChatItemDeleteAt,
+    updateDirectChatItemsReadList,
+    setDirectChatItemsDeleteAt,
     updateGroupChatItemsRead,
     getGroupUnreadTimedItems,
-    setGroupChatItemDeleteAt,
+    updateGroupChatItemsReadList,
+    setGroupChatItemsDeleteAt,
     updateLocalChatItemsRead,
     getChatRefViaItemId,
     getChatItemVersions,
@@ -126,7 +128,9 @@ import Data.ByteString.Char8 (ByteString)
 import Data.Either (fromRight, rights)
 import Data.Int (Int64)
 import Data.List (sortBy)
-import Data.Maybe (fromMaybe, isJust, mapMaybe)
+import Data.List.NonEmpty (NonEmpty)
+import qualified Data.List.NonEmpty as L
+import Data.Maybe (catMaybes, fromMaybe, isJust, mapMaybe)
 import Data.Ord (Down (..), comparing)
 import Data.Text (Text)
 import qualified Data.Text as T
@@ -1339,15 +1343,27 @@ getDirectUnreadTimedItems db User {userId} contactId itemsRange_ = case itemsRan
       |]
       (userId, contactId, CISRcvNew)
 
-setDirectChatItemDeleteAt :: DB.Connection -> User -> ContactId -> ChatItemId -> UTCTime -> IO ()
-setDirectChatItemDeleteAt db User {userId} contactId chatItemId deleteAt =
+updateDirectChatItemsReadList :: DB.Connection -> User -> ContactId -> NonEmpty ChatItemId -> IO [(ChatItemId, Int)]
+updateDirectChatItemsReadList db user contactId itemIds = do
+  catMaybes . L.toList <$> mapM getUpdateDirectItem itemIds
+  where
+    getUpdateDirectItem chatItemId = do
+      let itemsRange = Just (chatItemId, chatItemId)
+      timedItem <- maybeFirstRow id $ getDirectUnreadTimedItems db user contactId itemsRange
+      updateDirectChatItemsRead db user contactId itemsRange
+      pure timedItem
+
+setDirectChatItemsDeleteAt :: DB.Connection -> User -> ContactId -> [(ChatItemId, Int)] -> UTCTime -> IO [(ChatItemId, UTCTime)]
+setDirectChatItemsDeleteAt db User {userId} contactId itemIds currentTs = forM itemIds $ \(chatItemId, ttl) -> do
+  let deleteAt = addUTCTime (realToFrac ttl) currentTs
   DB.execute
     db
     "UPDATE chat_items SET timed_delete_at = ? WHERE user_id = ? AND contact_id = ? AND chat_item_id = ?"
     (deleteAt, userId, contactId, chatItemId)
+  pure (chatItemId, deleteAt)
 
-updateGroupChatItemsRead :: DB.Connection -> UserId -> GroupId -> Maybe (ChatItemId, ChatItemId) -> IO ()
-updateGroupChatItemsRead db userId groupId itemsRange_ = do
+updateGroupChatItemsRead :: DB.Connection -> User -> GroupId -> Maybe (ChatItemId, ChatItemId) -> IO ()
+updateGroupChatItemsRead db User {userId} groupId itemsRange_ = do
   currentTs <- getCurrentTime
   case itemsRange_ of
     Just (fromItemId, toItemId) ->
@@ -1392,12 +1408,24 @@ getGroupUnreadTimedItems db User {userId} groupId itemsRange_ = case itemsRange_
       |]
       (userId, groupId, CISRcvNew)
 
-setGroupChatItemDeleteAt :: DB.Connection -> User -> GroupId -> ChatItemId -> UTCTime -> IO ()
-setGroupChatItemDeleteAt db User {userId} groupId chatItemId deleteAt =
+updateGroupChatItemsReadList :: DB.Connection -> User -> GroupId -> NonEmpty ChatItemId -> IO [(ChatItemId, Int)]
+updateGroupChatItemsReadList db user groupId itemIds = do
+  catMaybes . L.toList <$> mapM getUpdateGroupItem itemIds
+  where
+    getUpdateGroupItem chatItemId = do
+      let itemsRange = Just (chatItemId, chatItemId)
+      timedItem <- maybeFirstRow id $ getGroupUnreadTimedItems db user groupId itemsRange
+      updateGroupChatItemsRead db user groupId itemsRange
+      pure timedItem
+
+setGroupChatItemsDeleteAt :: DB.Connection -> User -> GroupId -> [(ChatItemId, Int)] -> UTCTime -> IO [(ChatItemId, UTCTime)]
+setGroupChatItemsDeleteAt db User {userId} groupId itemIds currentTs = forM itemIds $ \(chatItemId, ttl) -> do
+  let deleteAt = addUTCTime (realToFrac ttl) currentTs
   DB.execute
     db
     "UPDATE chat_items SET timed_delete_at = ? WHERE user_id = ? AND group_id = ? AND chat_item_id = ?"
     (deleteAt, userId, groupId, chatItemId)
+  pure (chatItemId, deleteAt)
 
 updateLocalChatItemsRead :: DB.Connection -> User -> NoteFolderId -> Maybe (ChatItemId, ChatItemId) -> IO ()
 updateLocalChatItemsRead db User {userId} noteFolderId itemsRange_ = do

tests/ChatTests/Direct.hs

@@ -3,6 +3,7 @@
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE PostfixOperators #-}
 {-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 
 module ChatTests.Direct where
 
@@ -22,6 +23,7 @@ import Simplex.Chat.AppSettings (defaultAppSettings)
 import qualified Simplex.Chat.AppSettings as AS
 import Simplex.Chat.Call
 import Simplex.Chat.Controller (ChatConfig (..))
+import Simplex.Chat.Messages (ChatItemId)
 import Simplex.Chat.Options (ChatOpts (..))
 import Simplex.Chat.Protocol (supportedChatVRange)
 import Simplex.Chat.Store (agentStoreFile, chatStoreFile)
@@ -38,6 +40,7 @@ chatDirectTests :: SpecWith FilePath
 chatDirectTests = do
   describe "direct messages" $ do
     describe "add contact and send/receive messages" testAddContact
+    it "mark multiple messages as read" testMarkReadDirect
     it "clear chat with contact" testContactClear
     it "deleting contact deletes profile" testDeleteContactDeletesProfile
     it "delete contact keeping conversation" testDeleteContactKeepConversation
@@ -212,6 +215,22 @@ testAddContact = versionTestMatrix2 runTestAddContact
             then chatFeatures
             else (0, e2eeInfoNoPQStr) : tail chatFeatures
 
+testMarkReadDirect :: HasCallStack => FilePath -> IO ()
+testMarkReadDirect = testChat2 aliceProfile bobProfile $ \alice bob -> do
+  connectUsers alice bob
+  alice #> "@bob 1"
+  alice #> "@bob 2"
+  alice #> "@bob 3"
+  alice #> "@bob 4"
+  bob <# "alice> 1"
+  bob <# "alice> 2"
+  bob <# "alice> 3"
+  bob <# "alice> 4"
+  bob ##> "/last_item_id"
+  i :: ChatItemId <- read <$> getTermLine bob
+  let itemIds = intercalate "," $ map show [i - 3 .. i]
+  bob #$> ("/_read chat items @2 " <> itemIds, id, "ok")
+
 testDuplicateContactsSeparate :: HasCallStack => FilePath -> IO ()
 testDuplicateContactsSeparate =
   testChat2 aliceProfile bobProfile $

tests/ChatTests/Groups.hs

@@ -16,6 +16,7 @@ import Data.List (intercalate, isInfixOf)
 import qualified Data.Text as T
 import Database.SQLite.Simple (Only (..))
 import Simplex.Chat.Controller (ChatConfig (..))
+import Simplex.Chat.Messages (ChatItemId)
 import Simplex.Chat.Options
 import Simplex.Chat.Protocol (supportedChatVRange)
 import Simplex.Chat.Store (agentStoreFile, chatStoreFile)
@@ -34,6 +35,7 @@ chatGroupTests :: SpecWith FilePath
 chatGroupTests = do
   describe "chat groups" $ do
     describe "add contacts, create group and send/receive messages" testGroupMatrix
+    it "mark multiple messages as read" testMarkReadGroup
     it "v1: add contacts, create group and send/receive messages" testGroup
     it "v1: add contacts, create group and send/receive messages, check messages" testGroupCheckMessages
     it "send large message" testGroupLargeMessage
@@ -355,6 +357,22 @@ testGroupShared alice bob cath checkMessages directConnections = do
       alice #$> ("/_unread chat #1 on", id, "ok")
       alice #$> ("/_unread chat #1 off", id, "ok")
 
+testMarkReadGroup :: HasCallStack => FilePath -> IO ()
+testMarkReadGroup = testChat2 aliceProfile bobProfile $ \alice bob -> do
+  createGroup2 "team" alice bob
+  alice #> "#team 1"
+  alice #> "#team 2"
+  alice #> "#team 3"
+  alice #> "#team 4"
+  bob <# "#team alice> 1"
+  bob <# "#team alice> 2"
+  bob <# "#team alice> 3"
+  bob <# "#team alice> 4"
+  bob ##> "/last_item_id"
+  i :: ChatItemId <- read <$> getTermLine bob
+  let itemIds = intercalate "," $ map show [i - 3 .. i]
+  bob #$> ("/_read chat items #1 " <> itemIds, id, "ok")
+
 testGroupLargeMessage :: HasCallStack => FilePath -> IO ()
 testGroupLargeMessage =
   testChat2 aliceProfile bobProfile $