core: update statuses of all batched messages on SENT, RCVD (#4888)

* core: update statuses of all batched messages on SENT, RCVD * wip * update all * refactor
2024-12-17 17:20:21 +01:00 · 2024-09-17 23:50:26 +04:00
parent 665d9dcd00
commit 17b55c51c5
4 changed files with 90 additions and 73 deletions
@@ -4545,10 +4545,13 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
          void $ continueSending connEntity conn
          sentMsgDeliveryEvent conn msgId
          checkSndInlineFTComplete conn msgId
-          ci_ <- withStore $ \db -> do
-            ci_ <- updateDirectItemStatus' db ct conn msgId (CISSndSent SSPComplete)
-            forM ci_ $ \ci -> liftIO $ setDirectSndChatItemViaProxy db user ct ci (isJust proxy)
-          forM_ ci_ $ \ci -> toView $ CRChatItemStatusUpdated user (AChatItem SCTDirect SMDSnd (DirectChat ct) ci)
+          cis <- withStore $ \db -> do
+            cis <- updateDirectItemsStatus' db ct conn msgId (CISSndSent SSPComplete)
+            liftIO $ forM cis $ \ci -> setDirectSndChatItemViaProxy db user ct ci (isJust proxy)
+          let acis = map ctItem cis
+          unless (null acis) $ toView $ CRChatItemsStatusesUpdated user acis
+          where
+            ctItem = AChatItem SCTDirect SMDSnd (DirectChat ct)
        SWITCH qd phase cStats -> do
          toView $ CRContactSwitch user ct (SwitchProgress qd phase cStats)
          when (phase `elem` [SPStarted, SPCompleted]) $ case qd of
@@ -4604,7 +4607,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
          processConnMERR connEntity conn err
        MERRS msgIds err -> do
          -- error cannot be AUTH error here
-          updateDirectItemsStatus ct conn (L.toList msgIds) (CISSndError $ agentSndError err)
+          updateDirectItemsStatusMsgs ct conn (L.toList msgIds) (CISSndError $ agentSndError err)
          toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
        ERR err -> do
          toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
@@ -4958,7 +4961,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
        continued <- continueSending connEntity conn
        sentMsgDeliveryEvent conn msgId
        checkSndInlineFTComplete conn msgId
-        updateGroupItemStatus gInfo m conn msgId GSSSent (Just $ isJust proxy)
+        updateGroupItemsStatus gInfo m conn msgId GSSSent (Just $ isJust proxy)
        when continued $ sendPendingGroupMessages user m conn
      SWITCH qd phase cStats -> do
        toView $ CRGroupMemberSwitch user gInfo m (SwitchProgress qd phase cStats)
@@ -5002,10 +5005,10 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
        continued <- continueSending connEntity conn
        when continued $ sendPendingGroupMessages user m conn
      MWARN msgId err -> do
-        withStore' $ \db -> updateGroupItemErrorStatus db msgId (groupMemberId' m) (GSSWarning $ agentSndError err)
+        withStore' $ \db -> updateGroupItemsErrorStatus db msgId (groupMemberId' m) (GSSWarning $ agentSndError err)
        processConnMWARN connEntity conn err
      MERR msgId err -> do
-        withStore' $ \db -> updateGroupItemErrorStatus db msgId (groupMemberId' m) (GSSError $ agentSndError err)
+        withStore' $ \db -> updateGroupItemsErrorStatus db msgId (groupMemberId' m) (GSSError $ agentSndError err)
        -- group errors are silenced to reduce load on UI event log
        -- toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
        processConnMERR connEntity conn err
@@ -5013,7 +5016,7 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
        let newStatus = GSSError $ agentSndError err
        -- error cannot be AUTH error here
        withStore' $ \db -> forM_ msgIds $ \msgId ->
-          updateGroupItemErrorStatus db msgId (groupMemberId' m) newStatus `catchAll_` pure ()
+          updateGroupItemsErrorStatus db msgId (groupMemberId' m) newStatus `catchAll_` pure ()
        toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
      ERR err -> do
        toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity)
@@ -5021,10 +5024,10 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
      -- TODO add debugging output
      _ -> pure ()
      where
-        updateGroupItemErrorStatus :: DB.Connection -> AgentMsgId -> GroupMemberId -> GroupSndStatus -> IO ()
-        updateGroupItemErrorStatus db msgId groupMemberId newStatus = do
-          chatItemId_ <- getChatItemIdByAgentMsgId db connId msgId
-          forM_ chatItemId_ $ \itemId -> updateGroupMemSndStatus' db itemId groupMemberId newStatus
+        updateGroupItemsErrorStatus :: DB.Connection -> AgentMsgId -> GroupMemberId -> GroupSndStatus -> IO ()
+        updateGroupItemsErrorStatus db msgId groupMemberId newStatus = do
+          itemIds <- getChatItemIdsByAgentMsgId db connId msgId
+          forM_ itemIds $ \itemId -> updateGroupMemSndStatus' db itemId groupMemberId newStatus

    agentMsgDecryptError :: AgentCryptoError -> (MsgDecryptError, Word32)
    agentMsgDecryptError = \case
@@ -6696,43 +6699,42 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
        withStore' $ \db -> updateSndMsgDeliveryStatus db connId agentMsgId $ MDSSndRcvd msgRcptStatus
        updateDirectItemStatus ct conn agentMsgId $ CISSndRcvd msgRcptStatus SSPComplete

-    -- TODO [batch send] update status of all messages in batch
-    -- - this is for when we implement identifying inactive connections
-    -- - regular messages sent in batch would all be marked as delivered by a single receipt
-    -- - repeat for directMsgReceived if same logic is applied to direct messages
-    -- - getChatItemIdByAgentMsgId to return [ChatItemId]
    groupMsgReceived :: GroupInfo -> GroupMember -> Connection -> MsgMeta -> NonEmpty MsgReceipt -> CM ()
    groupMsgReceived gInfo m conn@Connection {connId} msgMeta msgRcpts = do
      checkIntegrityCreateItem (CDGroupRcv gInfo m) msgMeta `catchChatError` \_ -> pure ()
      forM_ msgRcpts $ \MsgReceipt {agentMsgId, msgRcptStatus} -> do
        withStore' $ \db -> updateSndMsgDeliveryStatus db connId agentMsgId $ MDSSndRcvd msgRcptStatus
-        updateGroupItemStatus gInfo m conn agentMsgId (GSSRcvd msgRcptStatus) Nothing
+        updateGroupItemsStatus gInfo m conn agentMsgId (GSSRcvd msgRcptStatus) Nothing

-    updateDirectItemsStatus :: Contact -> Connection -> [AgentMsgId] -> CIStatus 'MDSnd -> CM ()
-    updateDirectItemsStatus ct conn msgIds newStatus = do
-      cis_ <- withStore' $ \db -> forM msgIds $ \msgId -> runExceptT $ updateDirectItemStatus' db ct conn msgId newStatus
-      -- only send the last expired item event to view
-      case reverse $ catMaybes $ rights cis_ of
-        ci : _ -> toView $ CRChatItemStatusUpdated user (AChatItem SCTDirect SMDSnd (DirectChat ct) ci)
-        _ -> pure ()
+    -- Searches chat items for many agent message IDs and updates their status
+    updateDirectItemsStatusMsgs :: Contact -> Connection -> [AgentMsgId] -> CIStatus 'MDSnd -> CM ()
+    updateDirectItemsStatusMsgs ct conn msgIds newStatus = do
+      cis <- withStore' $ \db -> forM msgIds $ \msgId -> runExceptT $ updateDirectItemsStatus' db ct conn msgId newStatus
+      let acis = map ctItem $ concat $ rights cis
+      unless (null acis) $ toView $ CRChatItemsStatusesUpdated user acis
+      where
+        ctItem = AChatItem SCTDirect SMDSnd (DirectChat ct)

    updateDirectItemStatus :: Contact -> Connection -> AgentMsgId -> CIStatus 'MDSnd -> CM ()
    updateDirectItemStatus ct conn msgId newStatus = do
-      ci_ <- withStore $ \db -> updateDirectItemStatus' db ct conn msgId newStatus
-      forM_ ci_ $ \ci -> toView $ CRChatItemStatusUpdated user (AChatItem SCTDirect SMDSnd (DirectChat ct) ci)
+      cis <- withStore $ \db -> updateDirectItemsStatus' db ct conn msgId newStatus
+      let acis = map ctItem cis
+      unless (null acis) $ toView $ CRChatItemsStatusesUpdated user acis
+      where
+        ctItem = AChatItem SCTDirect SMDSnd (DirectChat ct)

-    updateDirectItemStatus' :: DB.Connection -> Contact -> Connection -> AgentMsgId -> CIStatus 'MDSnd -> ExceptT StoreError IO (Maybe (ChatItem 'CTDirect 'MDSnd))
-    updateDirectItemStatus' db ct@Contact {contactId} Connection {connId} msgId newStatus =
-      liftIO (getDirectChatItemByAgentMsgId db user contactId connId msgId) >>= \case
-        Just (CChatItem SMDSnd ChatItem {meta = CIMeta {itemStatus = CISSndRcvd _ _}}) -> pure Nothing
-        Just (CChatItem SMDSnd ChatItem {meta = CIMeta {itemId, itemStatus}})
-          | itemStatus == newStatus -> pure Nothing
-          | otherwise -> Just <$> updateDirectChatItemStatus db user ct itemId newStatus
-        _ -> pure Nothing
-
-    updateGroupMemSndStatus :: ChatItemId -> GroupMemberId -> GroupSndStatus -> CM Bool
-    updateGroupMemSndStatus itemId groupMemberId newStatus =
-      withStore' $ \db -> updateGroupMemSndStatus' db itemId groupMemberId newStatus
+    updateDirectItemsStatus' :: DB.Connection -> Contact -> Connection -> AgentMsgId -> CIStatus 'MDSnd -> ExceptT StoreError IO [ChatItem 'CTDirect 'MDSnd]
+    updateDirectItemsStatus' db ct@Contact {contactId} Connection {connId} msgId newStatus = do
+      items <- liftIO $ getDirectChatItemsByAgentMsgId db user contactId connId msgId
+      catMaybes <$> mapM updateItem items
+      where
+        updateItem :: CChatItem 'CTDirect -> ExceptT StoreError IO (Maybe (ChatItem 'CTDirect 'MDSnd))
+        updateItem = \case
+          (CChatItem SMDSnd ChatItem {meta = CIMeta {itemStatus = CISSndRcvd _ _}}) -> pure Nothing
+          (CChatItem SMDSnd ChatItem {meta = CIMeta {itemId, itemStatus}})
+            | itemStatus == newStatus -> pure Nothing
+            | otherwise -> Just <$> updateDirectChatItemStatus db user ct itemId newStatus
+          _ -> pure Nothing

    updateGroupMemSndStatus' :: DB.Connection -> ChatItemId -> GroupMemberId -> GroupSndStatus -> IO Bool
    updateGroupMemSndStatus' db itemId groupMemberId newStatus =
@@ -6743,20 +6745,29 @@ processAgentMessageConn vr user@User {userId} corrId agentConnId agentMessage =
          | otherwise -> updateGroupSndStatus db itemId groupMemberId newStatus $> True
        _ -> pure False

-    updateGroupItemStatus :: GroupInfo -> GroupMember -> Connection -> AgentMsgId -> GroupSndStatus -> Maybe Bool -> CM ()
-    updateGroupItemStatus gInfo@GroupInfo {groupId} GroupMember {groupMemberId} Connection {connId} msgId newMemStatus viaProxy_ =
-      withStore' (\db -> getGroupChatItemByAgentMsgId db user groupId connId msgId) >>= \case
-        Just (CChatItem SMDSnd ChatItem {meta = CIMeta {itemStatus = CISSndRcvd _ SSPComplete}}) -> pure ()
-        Just (CChatItem SMDSnd ChatItem {meta = CIMeta {itemId, itemStatus}}) -> do
-          forM_ viaProxy_ $ \viaProxy -> withStore' $ \db -> setGroupSndViaProxy db itemId groupMemberId viaProxy
-          memStatusChanged <- updateGroupMemSndStatus itemId groupMemberId newMemStatus
-          when memStatusChanged $ do
-            memStatusCounts <- withStore' (`getGroupSndStatusCounts` itemId)
-            let newStatus = membersGroupItemStatus memStatusCounts
-            when (newStatus /= itemStatus) $ do
-              chatItem <- withStore $ \db -> updateGroupChatItemStatus db user gInfo itemId newStatus
-              toView $ CRChatItemStatusUpdated user (AChatItem SCTGroup SMDSnd (GroupChat gInfo) chatItem)
-        _ -> pure ()
+    updateGroupItemsStatus :: GroupInfo -> GroupMember -> Connection -> AgentMsgId -> GroupSndStatus -> Maybe Bool -> CM ()
+    updateGroupItemsStatus gInfo@GroupInfo {groupId} GroupMember {groupMemberId} Connection {connId} msgId newMemStatus viaProxy_ = do
+      items <- withStore' (\db -> getGroupChatItemsByAgentMsgId db user groupId connId msgId)
+      cis <- catMaybes <$> withStore (\db -> mapM (updateItem db) items)
+      let acis = map gItem cis
+      unless (null acis) $ toView $ CRChatItemsStatusesUpdated user acis
+      where
+        gItem = AChatItem SCTGroup SMDSnd (GroupChat gInfo)
+        updateItem :: DB.Connection -> CChatItem 'CTGroup -> ExceptT StoreError IO (Maybe (ChatItem 'CTGroup 'MDSnd))
+        updateItem db = \case
+          (CChatItem SMDSnd ChatItem {meta = CIMeta {itemStatus = CISSndRcvd _ SSPComplete}}) -> pure Nothing
+          (CChatItem SMDSnd ChatItem {meta = CIMeta {itemId, itemStatus}}) -> do
+            forM_ viaProxy_ $ \viaProxy -> liftIO $ setGroupSndViaProxy db itemId groupMemberId viaProxy
+            memStatusChanged <- liftIO $ updateGroupMemSndStatus' db itemId groupMemberId newMemStatus
+            if memStatusChanged
+              then do
+                memStatusCounts <- liftIO $ getGroupSndStatusCounts db itemId
+                let newStatus = membersGroupItemStatus memStatusCounts
+                if newStatus /= itemStatus
+                  then Just <$> updateGroupChatItemStatus db user gInfo itemId newStatus
+                  else pure Nothing
+              else pure Nothing
+          _ -> pure Nothing

 createContactPQSndItem :: User -> Contact -> Connection -> PQEncryption -> CM (Contact, Connection)
 createContactPQSndItem user ct conn@Connection {pqSndEnabled} pqSndEnabled' =
@@ -600,7 +600,7 @@ data ChatResponse
  | CRGroupMemberCode {user :: User, groupInfo :: GroupInfo, member :: GroupMember, connectionCode :: Text}
  | CRConnectionVerified {user :: User, verified :: Bool, expectedCode :: Text}
  | CRNewChatItems {user :: User, chatItems :: [AChatItem]}
-  | CRChatItemStatusUpdated {user :: User, chatItem :: AChatItem}
+  | CRChatItemsStatusesUpdated {user :: User, chatItems :: [AChatItem]}
  | CRChatItemUpdated {user :: User, chatItem :: AChatItem}
  | CRChatItemNotChanged {user :: User, chatItem :: AChatItem}
  | CRChatItemReaction {user :: User, added :: Bool, reaction :: ACIReaction}
@@ -75,16 +75,16 @@ module Simplex.Chat.Store.Messages
    getGroupCIReactions,
    getGroupReactions,
    setGroupReaction,
-    getChatItemIdByAgentMsgId,
+    getChatItemIdsByAgentMsgId,
    getDirectChatItem,
    getDirectCIWithReactions,
    getDirectChatItemBySharedMsgId,
-    getDirectChatItemByAgentMsgId,
+    getDirectChatItemsByAgentMsgId,
    getGroupChatItem,
    getGroupCIWithReactions,
    getGroupChatItemBySharedMsgId,
    getGroupMemberCIBySharedMsgId,
-    getGroupChatItemByAgentMsgId,
+    getGroupChatItemsByAgentMsgId,
    getGroupMemberChatItemLast,
    getLocalChatItem,
    updateLocalChatItem',
@@ -1624,19 +1624,18 @@ getAllChatItems db vr user@User {userId} pagination search_ = do
          |]
          (userId, search, beforeTs, beforeTs, beforeId, count)

-getChatItemIdByAgentMsgId :: DB.Connection -> Int64 -> AgentMsgId -> IO (Maybe ChatItemId)
-getChatItemIdByAgentMsgId db connId msgId =
-  fmap join . maybeFirstRow fromOnly $
-    DB.query
+getChatItemIdsByAgentMsgId :: DB.Connection -> Int64 -> AgentMsgId -> IO [ChatItemId]
+getChatItemIdsByAgentMsgId db connId msgId =
+  map fromOnly
+    <$> DB.query
      db
      [sql|
        SELECT chat_item_id
        FROM chat_item_messages
-        WHERE message_id = (
+        WHERE message_id IN (
          SELECT message_id
          FROM msg_deliveries
          WHERE connection_id = ? AND agent_msg_id = ?
-          LIMIT 1
        )
      |]
      (connId, msgId)
@@ -1780,10 +1779,10 @@ getDirectChatItemBySharedMsgId db user@User {userId} contactId sharedMsgId = do
  itemId <- getDirectChatItemIdBySharedMsgId_ db userId contactId sharedMsgId
  getDirectChatItem db user contactId itemId

-getDirectChatItemByAgentMsgId :: DB.Connection -> User -> ContactId -> Int64 -> AgentMsgId -> IO (Maybe (CChatItem 'CTDirect))
-getDirectChatItemByAgentMsgId db user contactId connId msgId = do
-  itemId_ <- getChatItemIdByAgentMsgId db connId msgId
-  maybe (pure Nothing) (fmap eitherToMaybe . runExceptT . getDirectChatItem db user contactId) itemId_
+getDirectChatItemsByAgentMsgId :: DB.Connection -> User -> ContactId -> Int64 -> AgentMsgId -> IO [CChatItem 'CTDirect]
+getDirectChatItemsByAgentMsgId db user contactId connId msgId = do
+  itemIds <- getChatItemIdsByAgentMsgId db connId msgId
+  catMaybes <$> mapM (fmap eitherToMaybe . runExceptT . getDirectChatItem db user contactId) itemIds

 getDirectChatItemIdBySharedMsgId_ :: DB.Connection -> UserId -> Int64 -> SharedMsgId -> ExceptT StoreError IO Int64
 getDirectChatItemIdBySharedMsgId_ db userId contactId sharedMsgId =
@@ -2036,10 +2035,10 @@ getGroupMemberCIBySharedMsgId db user@User {userId} groupId memberId sharedMsgId
        (GCUserMember, userId, groupId, memberId, sharedMsgId)
  getGroupChatItem db user groupId itemId

-getGroupChatItemByAgentMsgId :: DB.Connection -> User -> GroupId -> Int64 -> AgentMsgId -> IO (Maybe (CChatItem 'CTGroup))
-getGroupChatItemByAgentMsgId db user groupId connId msgId = do
-  itemId_ <- getChatItemIdByAgentMsgId db connId msgId
-  maybe (pure Nothing) (fmap eitherToMaybe . runExceptT . getGroupChatItem db user groupId) itemId_
+getGroupChatItemsByAgentMsgId :: DB.Connection -> User -> GroupId -> Int64 -> AgentMsgId -> IO [CChatItem 'CTGroup]
+getGroupChatItemsByAgentMsgId db user groupId connId msgId = do
+  itemIds <- getChatItemIdsByAgentMsgId db connId msgId
+  catMaybes <$> mapM (fmap eitherToMaybe . runExceptT . getGroupChatItem db user groupId) itemIds

 getGroupChatItem :: DB.Connection -> User -> Int64 -> ChatItemId -> ExceptT StoreError IO (CChatItem 'CTGroup)
 getGroupChatItem db User {userId, userContactId} groupId itemId = ExceptT $ do
@@ -134,7 +134,14 @@ responseToView hu@(currentRH, user_) ChatConfig {logLevel, showReactions, showRe
  CRChatItems u _ chatItems -> ttyUser u $ concatMap (\(AChatItem _ _ chat item) -> viewChatItem chat item True ts tz <> viewItemReactions item) chatItems
  CRChatItemInfo u ci ciInfo -> ttyUser u $ viewChatItemInfo ci ciInfo tz
  CRChatItemId u itemId -> ttyUser u [plain $ maybe "no item" show itemId]
-  CRChatItemStatusUpdated u ci -> ttyUser u $ viewChatItemStatusUpdated ci ts tz testView showReceipts
+  CRChatItemsStatusesUpdated u chatItems
+    | length chatItems <= 20 ->
+        concatMap
+          (\ci -> ttyUser u $ viewChatItemStatusUpdated ci ts tz testView showReceipts)
+          chatItems
+    | testView && showReceipts ->
+        ttyUser u [sShow (length chatItems) <> " message statuses updated"]
+    | otherwise -> []
  CRChatItemUpdated u (AChatItem _ _ chat item) -> ttyUser u $ unmuted u chat item $ viewItemUpdate chat item liveItems ts tz
  CRChatItemNotChanged u ci -> ttyUser u $ viewItemNotChanged ci
  CRChatItemsDeleted u deletions byUser timed -> case deletions of