collect and send message history

src/Simplex/Chat.hs

@@ -3601,26 +3601,12 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do
                 void $ sendDirectMessage conn (XGrpMemIntro $ memberInfo (reMember intro)) (GroupId groupId)
                 withStore' $ \db -> updateIntroStatus db introId GMIntroSent
               sendHistory =
-                when (isCompatibleRange (memberChatVRange' m) batchSendVRange) $
-                  pure ()
-                -- - build list of messages forming latest history, possible approaches:
-                --   - retrieve events from 'messages' table as is and filter based on event type
-                --     - downside is that they are periodically cleaned up,
-                --       but practically it won't matter for recent history of active group
-                --     - 'msg_deliveries.chat_ts' to be used as event ts for received messages,
-                --       'messages.created_at' for sent
-                --     - simple - 'msg_body' contains full event which can be forwarded
-                --     - unclear how many records to load to reach required number of messages,
-                --       so either load iteratively or with a reserve
-                --   - reconstruct events from 'chat_items' table
-                --     - it's not periodically cleaned up, but chat items can be deleted
-                --       manually or scheduled for deletion by user
-                --     - 'item_ts' can be used as timestamp of original message
-                --     - more complex - custom reconstruction of reactions, file descriptions, etc.
-                --   - in any case use XGrpMsgForward to wrap messages together with their original ts and sender id
-                -- - send messages in batches (sendBatchedDirectMessages)
+                when (isCompatibleRange (memberChatVRange' m) batchSendVRange) $ do
+                  (errs, events) <- withStore' $ \db -> getGroupMsgsHistory db user gInfo 300
+                  toView $ CRChatErrors (Just user) (map ChatErrorStore errs)
+                  forM_ (L.nonEmpty events) $ \events' ->
+                    sendBatchedDirectMessages conn events' (GroupId groupId)
           _ -> do
-            -- TODO notify member who forwarded introduction - question - where it is stored? There is via_contact but probably there should be via_member in group_members table
             let memCategory = memberCategory m
             withStore' (\db -> getViaGroupContact db user m) >>= \case
               Nothing -> do
@@ -5240,6 +5226,10 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do
         processForwardedMsg :: GroupMember -> ChatMessage 'Json -> m ()
         processForwardedMsg author chatMsg = do
           let body = LB.toStrict $ J.encode msg
+          -- TODO [batch send] save broker ts (here - msgTs) on messages received via forward to include in history
+          -- consider adding chat_ts to messages table: as of now, messages received via forward
+          -- wouldn't have a record in msg_deliveries to retrieve it from;
+          -- practically it's not a problem if members join via a single same admin (e.g. directory bot)
           rcvMsg@RcvMessage {chatMsgEvent = ACME _ event} <- saveGroupFwdRcvMsg user groupId m author body chatMsg
           case event of
             XMsgNew mc -> memberCanSend author $ newGroupContentMessage gInfo author mc rcvMsg msgTs

src/Simplex/Chat/Migrations/M20231208_recreate_msg_deliveries.hs

@@ -39,11 +39,16 @@ ALTER TABLE new_msg_deliveries RENAME TO msg_deliveries;
 CREATE INDEX idx_msg_deliveries_message_id ON "msg_deliveries"(message_id);
 CREATE INDEX idx_msg_deliveries_agent_ack_cmd_id ON "msg_deliveries"(connection_id, agent_ack_cmd_id);
 CREATE INDEX idx_msg_deliveries_agent_msg_id ON "msg_deliveries"(connection_id, agent_msg_id);
+
+-- for message history retrieval
+CREATE INDEX idx_messages_group_id_created_at ON messages(group_id, created_at);
 |]
 
 down_m20231208_recreate_msg_deliveries :: Query
 down_m20231208_recreate_msg_deliveries =
   [sql|
+DROP INDEX idx_messages_group_id_created_at;
+
 DROP INDEX idx_msg_deliveries_message_id;
 DROP INDEX idx_msg_deliveries_agent_ack_cmd_id;
 DROP INDEX idx_msg_deliveries_agent_msg_id;

src/Simplex/Chat/Migrations/chat_schema.sql

@@ -811,3 +811,7 @@ CREATE INDEX idx_msg_deliveries_agent_msg_id ON "msg_deliveries"(
   connection_id,
   agent_msg_id
 );
+CREATE INDEX idx_messages_group_id_created_at ON messages(
+  group_id,
+  created_at
+);

src/Simplex/Chat/Protocol.hs

@@ -31,6 +31,7 @@ import qualified Data.ByteString.Char8 as B
 import Data.ByteString.Internal (c2w, w2c)
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Maybe (fromMaybe)
+import Data.Set (Set)
 import Data.String
 import Data.Text (Text)
 import qualified Data.Text as T
@@ -149,7 +150,7 @@ instance StrEncoding AppMessageBinary where
     pure AppMessageBinary {tag, msgId, body}
 
 newtype SharedMsgId = SharedMsgId ByteString
-  deriving (Eq, Show)
+  deriving (Eq, Ord, Show)
 
 instance FromField SharedMsgId where fromField f = SharedMsgId <$> fromField f
 
@@ -289,6 +290,24 @@ forwardedGroupMsg msg@ChatMessage {chatMsgEvent} = case encoding @e of
   SJson | isForwardedGroupMsg chatMsgEvent -> Just msg
   _ -> Nothing
 
+msgIncludedInHistory :: forall e. MsgEncodingI e => ChatMessage e -> Set SharedMsgId -> Maybe (ChatMessage 'Json)
+msgIncludedInHistory msg@ChatMessage {chatMsgEvent} accMsgIds = case encoding @e of
+  SJson -> case chatMsgEvent of
+    XMsgNew mc -> case mcExtMsgContent mc of
+      ExtMsgContent {file = Just FileInvitation {fileInline = Just _}} -> Nothing
+      _ -> Just msg
+    XMsgFileDescr msgId _ -> refIncluded msgId
+    XMsgUpdate {msgId} -> refIncluded msgId
+    XMsgDel msgId _ -> refIncluded msgId
+    XMsgReact {msgId} -> refIncluded msgId
+    XFileCancel msgId -> refIncluded msgId
+    _ -> Nothing
+    where
+      refIncluded msgId
+        | msgId `elem` accMsgIds = Just msg
+        | otherwise = Nothing
+  _ -> Nothing
+
 data MsgReaction = MREmoji {emoji :: MREmojiChar} | MRUnknown {tag :: Text, json :: J.Object}
   deriving (Eq, Show)
 

src/Simplex/Chat/Store/Messages.hs

@@ -7,6 +7,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
@@ -99,6 +100,7 @@ module Simplex.Chat.Store.Messages
     updateGroupSndStatus,
     getGroupSndStatuses,
     getGroupSndStatusCounts,
+    getGroupMsgsHistory,
   )
 where
 
@@ -108,11 +110,13 @@ import Control.Monad.IO.Class
 import Crypto.Random (ChaChaDRG)
 import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
-import Data.Either (fromRight, rights)
+import Data.Either (fromRight, partitionEithers, rights)
 import Data.Int (Int64)
 import Data.List (sortBy)
 import Data.Maybe (fromMaybe, isJust, mapMaybe)
 import Data.Ord (Down (..), comparing)
+import Data.Set (Set)
+import qualified Data.Set as S
 import Data.Text (Text)
 import Data.Time (addUTCTime)
 import Data.Time.Clock (UTCTime (..), getCurrentTime)
@@ -2075,3 +2079,88 @@ getGroupSndStatusCounts db itemId =
       GROUP BY group_snd_item_status
     |]
     (Only itemId)
+
+data JsonMessageRecord = JsonMessageRecord
+  { messageId :: Int64,
+    msgSent :: Bool,
+    chatMsgEvent :: CMEventTag 'Json,
+    chatMsg :: ChatMessage 'Json,
+    createdAt :: UTCTime,
+    updatedAt :: UTCTime,
+    connectionId :: Maybe Int64,
+    groupId :: Maybe Int64,
+    sharedMsgId :: Maybe SharedMsgId,
+    sharedMsgIdUser :: Maybe Bool,
+    authorGroupMemberId :: Maybe GroupMemberId,
+    forwardedByGroupMemberId :: Maybe GroupMemberId
+  }
+
+type JsonMessageRow = (Int64, Bool, CMEventTag 'Json, ByteString, UTCTime, UTCTime, Maybe Int64, Maybe Int64, Maybe SharedMsgId, Maybe Bool, Maybe GroupMemberId, Maybe GroupMemberId)
+
+toJsonMessageRecord :: JsonMessageRow -> Either StoreError JsonMessageRecord
+toJsonMessageRecord (messageId, msgSent, chatMsgEvent, msgBody, createdAt, updatedAt, connectionId, groupId, sharedMsgId, sharedMsgIdUser, authorGroupMemberId, forwardedByGroupMemberId) =
+  case parseChatMessages msgBody of
+    [Right (ACMsg SJson chatMsg)] -> Right JsonMessageRecord {..}
+    _ -> Left $ SEParseJsonMessageError messageId
+
+getGroupMsgsHistory :: DB.Connection -> User -> GroupInfo -> Int64 -> IO ([StoreError], [ChatMsgEvent 'Json])
+getGroupMsgsHistory db user GroupInfo {groupId, membership} numMsgsToSearch = do
+  (errs, msgRecs) <- partitionEithers <$> retrieveMessages
+  (errs', evts) <- collectForwardEvents msgRecs S.empty errs []
+  pure (errs <> errs', evts)
+  where
+    retrieveMessages :: IO [Either StoreError JsonMessageRecord]
+    retrieveMessages =
+      map toJsonMessageRecord
+        <$> DB.query
+          db
+          [sql|
+            SELECT
+              message_id, msg_sent, chat_msg_event, msg_body, created_at, updated_at, connection_id, group_id,
+              shared_msg_id, shared_msg_id_user, author_group_member_id, forwarded_by_group_member_id
+            FROM messages
+            WHERE group_id = ?
+              AND chat_msg_event IN (?,?,?,?,?,?)
+            ORDER BY created_at DESC
+            LIMIT ?
+          |]
+          -- see msgIncludedInHistory in Protocol
+          (groupId, XMsgNew_, XMsgFileDescr_, XMsgUpdate_, XMsgDel_, XMsgReact_, XFileCancel_, numMsgsToSearch)
+    collectForwardEvents ::
+      [JsonMessageRecord] ->
+      Set SharedMsgId ->
+      [StoreError] ->
+      [ChatMsgEvent 'Json] ->
+      IO ([StoreError], [ChatMsgEvent 'Json])
+    collectForwardEvents [] _ errs evts = pure (errs, evts)
+    collectForwardEvents (mrec : mrecs) msgIds errs evts = do
+      let JsonMessageRecord {messageId, msgSent, chatMsg, createdAt, sharedMsgId, authorGroupMemberId} = mrec
+      case msgIncludedInHistory chatMsg msgIds of
+        Nothing -> collectForwardEvents mrecs msgIds errs evts
+        Just chatMsg'
+          | msgSent -> do
+              let GroupMember {memberId} = membership
+                  evt = XGrpMsgForward memberId chatMsg' createdAt -- TODO round to seconds
+                  msgIds' = maybe msgIds (`S.insert` msgIds) sharedMsgId
+              collectForwardEvents mrecs msgIds' errs (evt : evts)
+          | otherwise ->
+              case authorGroupMemberId of
+                Just amId -> do
+                  runExceptT (getGroupMember db user groupId amId) >>= \case
+                    Right _author@GroupMember {memberId} -> do
+                      brokerTs <- getDeliveryTs messageId
+                      -- message received via forward wouldn't have delivery record to retrieve chat_ts from,
+                      -- see comment in processForwardedMsg in Chat
+                      let msgTs = fromMaybe createdAt brokerTs
+                          evt = XGrpMsgForward memberId chatMsg' msgTs
+                          msgIds' = maybe msgIds (`S.insert` msgIds) sharedMsgId
+                      collectForwardEvents mrecs msgIds' errs (evt : evts)
+                    Left err ->
+                      collectForwardEvents mrecs msgIds (err : errs) evts
+                _ -> do
+                  let err = SEInternalError $ "message " <> show messageId <> " has no author group member id (can be due to older app version)"
+                  collectForwardEvents mrecs msgIds (err : errs) evts
+    getDeliveryTs :: Int64 -> IO (Maybe UTCTime)
+    getDeliveryTs messageId =
+      maybeFirstRow fromOnly $
+        DB.query db "SELECT chat_ts FROM msg_deliveries WHERE message_id = ? LIMIT 1" (Only messageId)

src/Simplex/Chat/Store/Shared.hs

@@ -99,6 +99,7 @@ data StoreError
   | SEContactNotFoundByFileId {fileId :: FileTransferId}
   | SENoGroupSndStatus {itemId :: ChatItemId, groupMemberId :: GroupMemberId}
   | SEDuplicateGroupMessage {groupId :: Int64, sharedMsgId :: SharedMsgId, authorGroupMemberId :: Maybe GroupMemberId, forwardedByGroupMemberId :: Maybe GroupMemberId}
+  | SEParseJsonMessageError {messageId :: Int64}
   | SERemoteHostNotFound {remoteHostId :: RemoteHostId}
   | SERemoteHostUnknown -- attempting to store KnownHost without a known fingerprint
   | SERemoteHostDuplicateCA