From e714f4037aa79bdb635888f61333fd1be0224bdf Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Wed, 13 Dec 2023 21:44:29 +0400 Subject: [PATCH] collect and send message history --- src/Simplex/Chat.hs | 28 ++---- .../M20231208_recreate_msg_deliveries.hs | 5 + src/Simplex/Chat/Migrations/chat_schema.sql | 4 + src/Simplex/Chat/Protocol.hs | 21 ++++- src/Simplex/Chat/Store/Messages.hs | 91 ++++++++++++++++++- src/Simplex/Chat/Store/Shared.hs | 1 + 6 files changed, 129 insertions(+), 21 deletions(-) diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index c48b3feba7..2f6b792901 100644 --- a/src/Simplex/Chat.hs +++ b/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 diff --git a/src/Simplex/Chat/Migrations/M20231208_recreate_msg_deliveries.hs b/src/Simplex/Chat/Migrations/M20231208_recreate_msg_deliveries.hs index 2d1617ba5d..784fffe1fb 100644 --- a/src/Simplex/Chat/Migrations/M20231208_recreate_msg_deliveries.hs +++ b/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; diff --git a/src/Simplex/Chat/Migrations/chat_schema.sql b/src/Simplex/Chat/Migrations/chat_schema.sql index 59fb2bae59..eac85841b4 100644 --- a/src/Simplex/Chat/Migrations/chat_schema.sql +++ b/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 +); diff --git a/src/Simplex/Chat/Protocol.hs b/src/Simplex/Chat/Protocol.hs index 0d927878ea..bd03e5d3f8 100644 --- a/src/Simplex/Chat/Protocol.hs +++ b/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) diff --git a/src/Simplex/Chat/Store/Messages.hs b/src/Simplex/Chat/Store/Messages.hs index 7bfc9c6d05..863b5d551f 100644 --- a/src/Simplex/Chat/Store/Messages.hs +++ b/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) diff --git a/src/Simplex/Chat/Store/Shared.hs b/src/Simplex/Chat/Store/Shared.hs index 6fb77b5484..68152dc087 100644 --- a/src/Simplex/Chat/Store/Shared.hs +++ b/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