From ecd66ad7f356a67c742a9b4183ebfe087ba5bb73 Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Wed, 6 Dec 2023 20:10:56 +0400 Subject: [PATCH] schema, refactor group processing for batched messages --- simplex-chat.cabal | 1 + src/Simplex/Chat.hs | 282 +++++++++--------- .../M20231206_recreate_msg_deliveries.hs | 85 ++++++ src/Simplex/Chat/Protocol.hs | 13 +- src/Simplex/Chat/Store/Migrations.hs | 4 +- 5 files changed, 241 insertions(+), 144 deletions(-) create mode 100644 src/Simplex/Chat/Migrations/M20231206_recreate_msg_deliveries.hs diff --git a/simplex-chat.cabal b/simplex-chat.cabal index a53c15c78a..669b2785bd 100644 --- a/simplex-chat.cabal +++ b/simplex-chat.cabal @@ -125,6 +125,7 @@ library Simplex.Chat.Migrations.M20231113_group_forward Simplex.Chat.Migrations.M20231114_remote_control Simplex.Chat.Migrations.M20231126_remote_ctrl_address + Simplex.Chat.Migrations.M20231206_recreate_msg_deliveries Simplex.Chat.Mobile Simplex.Chat.Mobile.File Simplex.Chat.Mobile.Shared diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index 475845c74c..3a8b9a5ce7 100644 --- a/src/Simplex/Chat.hs +++ b/src/Simplex/Chat.hs @@ -3340,35 +3340,39 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do withAckMessage' agentConnId conn msgMeta $ directMsgReceived ct conn msgMeta msgRcpt CONF confId _ connInfo -> do - ChatMessage {chatVRange, chatMsgEvent} <- parseChatMessage conn connInfo - conn' <- updatePeerChatVRange conn chatVRange - case chatMsgEvent of - -- confirming direct connection with a member - XGrpMemInfo _memId _memProfile -> do - -- TODO check member ID - -- TODO update member profile - -- [async agent commands] no continuation needed, but command should be asynchronous for stability - allowAgentConnectionAsync user conn' confId XOk - XInfo profile -> do - ct' <- processContactProfileUpdate ct profile False `catchChatError` const (pure ct) - -- [incognito] send incognito profile - incognitoProfile <- forM customUserProfileId $ \profileId -> withStore $ \db -> getProfileById db userId profileId - let p = userProfileToSend user (fromLocalProfile <$> incognitoProfile) (Just ct') - allowAgentConnectionAsync user conn' confId $ XInfo p - void $ withStore' $ \db -> resetMemberContactFields db ct' - _ -> messageError "CONF for existing contact must have x.grp.mem.info or x.info" + parseChatMessage conn connInfo >>= \case + [ChatMessage {chatVRange, chatMsgEvent}] -> do + conn' <- updatePeerChatVRange conn chatVRange + case chatMsgEvent of + -- confirming direct connection with a member + XGrpMemInfo _memId _memProfile -> do + -- TODO check member ID + -- TODO update member profile + -- [async agent commands] no continuation needed, but command should be asynchronous for stability + allowAgentConnectionAsync user conn' confId XOk + XInfo profile -> do + ct' <- processContactProfileUpdate ct profile False `catchChatError` const (pure ct) + -- [incognito] send incognito profile + incognitoProfile <- forM customUserProfileId $ \profileId -> withStore $ \db -> getProfileById db userId profileId + let p = userProfileToSend user (fromLocalProfile <$> incognitoProfile) (Just ct') + allowAgentConnectionAsync user conn' confId $ XInfo p + void $ withStore' $ \db -> resetMemberContactFields db ct' + _ -> messageError "CONF for existing contact must have x.grp.mem.info or x.info" + _ -> messageError "CONF: batching not supported" INFO connInfo -> do - ChatMessage {chatVRange, chatMsgEvent} <- parseChatMessage conn connInfo - _conn' <- updatePeerChatVRange conn chatVRange - case chatMsgEvent of - XGrpMemInfo _memId _memProfile -> do - -- TODO check member ID - -- TODO update member profile - pure () - XInfo profile -> - void $ processContactProfileUpdate ct profile False - XOk -> pure () - _ -> messageError "INFO for existing contact must have x.grp.mem.info, x.info or x.ok" + parseChatMessage conn connInfo >>= \case + [ChatMessage {chatVRange, chatMsgEvent}] -> do + _conn' <- updatePeerChatVRange conn chatVRange + case chatMsgEvent of + XGrpMemInfo _memId _memProfile -> do + -- TODO check member ID + -- TODO update member profile + pure () + XInfo profile -> + void $ processContactProfileUpdate ct profile False + XOk -> pure () + _ -> messageError "INFO for existing contact must have x.grp.mem.info, x.info or x.ok" + _ -> messageError "INFO: batching not supported" CON -> withStore' (\db -> getViaGroupMember db user ct) >>= \case Nothing -> do @@ -3495,40 +3499,44 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do _ -> throwChatError $ CECommandError "unexpected cmdFunction" CRContactUri _ -> throwChatError $ CECommandError "unexpected ConnectionRequestUri type" CONF confId _ connInfo -> do - ChatMessage {chatVRange, chatMsgEvent} <- parseChatMessage conn connInfo - conn' <- updatePeerChatVRange conn chatVRange - case memberCategory m of - GCInviteeMember -> - case chatMsgEvent of - XGrpAcpt memId - | sameMemberId memId m -> do - withStore $ \db -> liftIO $ updateGroupMemberStatus db userId m GSMemAccepted - -- [async agent commands] no continuation needed, but command should be asynchronous for stability - allowAgentConnectionAsync user conn' confId XOk - | otherwise -> messageError "x.grp.acpt: memberId is different from expected" - _ -> messageError "CONF from invited member must have x.grp.acpt" - _ -> + parseChatMessage conn connInfo >>= \case + [ChatMessage {chatVRange, chatMsgEvent}] -> do + conn' <- updatePeerChatVRange conn chatVRange + case memberCategory m of + GCInviteeMember -> + case chatMsgEvent of + XGrpAcpt memId + | sameMemberId memId m -> do + withStore $ \db -> liftIO $ updateGroupMemberStatus db userId m GSMemAccepted + -- [async agent commands] no continuation needed, but command should be asynchronous for stability + allowAgentConnectionAsync user conn' confId XOk + | otherwise -> messageError "x.grp.acpt: memberId is different from expected" + _ -> messageError "CONF from invited member must have x.grp.acpt" + _ -> + case chatMsgEvent of + XGrpMemInfo memId _memProfile + | sameMemberId memId m -> do + -- TODO update member profile + -- [async agent commands] no continuation needed, but command should be asynchronous for stability + allowAgentConnectionAsync user conn' confId $ XGrpMemInfo membership.memberId (fromLocalProfile $ memberProfile membership) + | otherwise -> messageError "x.grp.mem.info: memberId is different from expected" + _ -> messageError "CONF from member must have x.grp.mem.info" + _ -> messageError "CONF: batching not supported" + INFO connInfo -> do + parseChatMessage conn connInfo >>= \case + [ChatMessage {chatVRange, chatMsgEvent}] -> do + _conn' <- updatePeerChatVRange conn chatVRange case chatMsgEvent of XGrpMemInfo memId _memProfile | sameMemberId memId m -> do -- TODO update member profile - -- [async agent commands] no continuation needed, but command should be asynchronous for stability - allowAgentConnectionAsync user conn' confId $ XGrpMemInfo membership.memberId (fromLocalProfile $ memberProfile membership) + pure () | otherwise -> messageError "x.grp.mem.info: memberId is different from expected" - _ -> messageError "CONF from member must have x.grp.mem.info" - INFO connInfo -> do - ChatMessage {chatVRange, chatMsgEvent} <- parseChatMessage conn connInfo - _conn' <- updatePeerChatVRange conn chatVRange - case chatMsgEvent of - XGrpMemInfo memId _memProfile - | sameMemberId memId m -> do - -- TODO update member profile - pure () - | otherwise -> messageError "x.grp.mem.info: memberId is different from expected" - XInfo _ -> pure () -- sent when connecting via group link - XOk -> pure () - _ -> messageError "INFO from member must have x.grp.mem.info, x.info or x.ok" - pure () + XInfo _ -> pure () -- sent when connecting via group link + XOk -> pure () + _ -> messageError "INFO from member must have x.grp.mem.info, x.info or x.ok" + pure () + _ -> messageError "INFO: batching not supported" CON -> do withStore' $ \db -> do updateGroupMemberStatus db userId m GSMemConnected @@ -3617,40 +3625,24 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do void $ sendDirectMessage imConn (XGrpMemCon m.memberId) (GroupId groupId) _ -> messageWarning "sendXGrpMemCon: member category GCPreMember or GCPostMember is expected" MSG msgMeta _msgFlags msgBody -> do - -- TODO [batch send] process events in loop - -- - catch error on each iteration - -- - don't forward - -- - acknowledge after all events are processed + checkIntegrityCreateItem (CDGroupRcv gInfo m) msgMeta `catchChatError` \_ -> pure () cmdId <- createAckCmd conn - tryChatError (processChatMessage cmdId) >>= \case - Right (ACMsg _ chatMsg, withRcpt) -> do + tryChatError (parseAChatMessage conn msgMeta msgBody) >>= \case + Right aChatMsgs -> do + -- TODO [batch send] workaround for UNIQUE(connection_id, agent_msg_id) in msg_deliveries + -- - recreate table without UNIQUE constraint? + -- - many-to-many table? + forM_ aChatMsgs $ \(ACMsg _ chatMsg) -> do + processEvent cmdId chatMsg `catchChatError` \e -> toView $ CRChatError (Just user) e + withRcpt <- checkSendRcpt aChatMsgs `catchChatError` \_ -> pure False ackMsg agentConnId cmdId msgMeta $ if withRcpt then Just "" else Nothing - when (membership.memberRole >= GRAdmin) $ forwardMsg_ chatMsg + when (membership.memberRole >= GRAdmin) $ case aChatMsgs of + [ACMsg _ chatMsg] -> forwardMsg_ chatMsg + _ -> pure () Left e -> ackMsg agentConnId cmdId msgMeta Nothing >> throwError e where - processChatMessage :: Int64 -> m (AChatMessage, Bool) - processChatMessage cmdId = do - msg@(ACMsg _ chatMsg) <- parseAChatMessage conn msgMeta msgBody - checkIntegrity chatMsg `catchChatError` \_ -> pure () - (msg,) <$> processEvent cmdId chatMsg brokerTs = metaBrokerTs msgMeta - checkIntegrity :: ChatMessage e -> m () - checkIntegrity ChatMessage {chatMsgEvent} = do - when checkForEvent $ checkIntegrityCreateItem (CDGroupRcv gInfo m) msgMeta - where - checkForEvent = case chatMsgEvent of - XMsgNew _ -> True - XFileCancel _ -> True - XFileAcptInv {} -> True - XGrpMemNew _ -> True - XGrpMemRole {} -> True - XGrpMemDel _ -> True - XGrpLeave -> True - XGrpDel -> True - XGrpInfo _ -> True - XGrpDirectInv {} -> True - _ -> False - processEvent :: MsgEncodingI e => CommandId -> ChatMessage e -> m Bool + processEvent :: MsgEncodingI e => CommandId -> ChatMessage e -> m () processEvent cmdId chatMsg = do (m', conn', msg@RcvMessage {chatMsgEvent = ACME _ event}) <- saveGroupRcvMsg user groupId m conn msgMeta cmdId msgBody chatMsg updateChatLock "groupMessage" event @@ -3683,15 +3675,17 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do XInfoProbeOk probe -> xInfoProbeOk (COMGroupMember m') probe BFileChunk sharedMsgId chunk -> bFileChunkGroup gInfo sharedMsgId chunk msgMeta _ -> messageError $ "unsupported message: " <> T.pack (show event) - checkSendRcpt event - checkSendRcpt :: ChatMsgEvent e -> m Bool - checkSendRcpt event = do + checkSendRcpt :: [AChatMessage] -> m Bool + checkSendRcpt aChatMsgs = do currentMemCount <- withStore' $ \db -> getGroupCurrentMembersCount db user gInfo let GroupInfo {chatSettings = ChatSettings {sendRcpts}} = gInfo pure $ fromMaybe (sendRcptsSmallGroups user) sendRcpts - && hasDeliveryReceipt (toCMEventTag event) + && any aChatMsgHasReceipt aChatMsgs && currentMemCount <= smallGroupsRcptsMemLimit + where + aChatMsgHasReceipt (ACMsg _ ChatMessage {chatMsgEvent}) = + hasDeliveryReceipt (toCMEventTag chatMsgEvent) forwardMsg_ :: MsgEncodingI e => ChatMessage e -> m () forwardMsg_ chatMsg = forM_ (forwardedGroupMsg chatMsg) $ \chatMsg' -> do @@ -3792,17 +3786,19 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do -- SMP CONF for SndFileConnection happens for direct file protocol -- when recipient of the file "joins" connection created by the sender CONF confId _ connInfo -> do - ChatMessage {chatVRange, chatMsgEvent} <- parseChatMessage conn connInfo - conn' <- updatePeerChatVRange conn chatVRange - case chatMsgEvent of - -- TODO save XFileAcpt message - XFileAcpt name - | name == fileName -> do - withStore' $ \db -> updateSndFileStatus db ft FSAccepted - -- [async agent commands] no continuation needed, but command should be asynchronous for stability - allowAgentConnectionAsync user conn' confId XOk - | otherwise -> messageError "x.file.acpt: fileName is different from expected" - _ -> messageError "CONF from file connection must have x.file.acpt" + parseChatMessage conn connInfo >>= \case + [ChatMessage {chatVRange, chatMsgEvent}] -> do + conn' <- updatePeerChatVRange conn chatVRange + case chatMsgEvent of + -- TODO save XFileAcpt message + XFileAcpt name + | name == fileName -> do + withStore' $ \db -> updateSndFileStatus db ft FSAccepted + -- [async agent commands] no continuation needed, but command should be asynchronous for stability + allowAgentConnectionAsync user conn' confId XOk + | otherwise -> messageError "x.file.acpt: fileName is different from expected" + _ -> messageError "CONF from file connection must have x.file.acpt" + _ -> messageError "CONF: batching not supported" CON -> do ci <- withStore $ \db -> do liftIO $ updateSndFileStatus db ft FSConnected @@ -3861,11 +3857,13 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do -- when sender of the file "joins" connection created by the recipient -- (sender doesn't create connections for all group members) CONF confId _ connInfo -> do - ChatMessage {chatVRange, chatMsgEvent} <- parseChatMessage conn connInfo - conn' <- updatePeerChatVRange conn chatVRange - case chatMsgEvent of - XOk -> allowAgentConnectionAsync user conn' confId XOk -- [async agent commands] no continuation needed, but command should be asynchronous for stability - _ -> pure () + parseChatMessage conn connInfo >>= \case + [ChatMessage {chatVRange, chatMsgEvent}] -> do + conn' <- updatePeerChatVRange conn chatVRange + case chatMsgEvent of + XOk -> allowAgentConnectionAsync user conn' confId XOk -- [async agent commands] no continuation needed, but command should be asynchronous for stability + _ -> pure () + _ -> messageError "CONF: batching not supported" CON -> startReceivingFile user fileId MSG meta _ msgBody -> do parseFileChunk msgBody >>= receiveFileChunk ft (Just conn) meta @@ -3922,12 +3920,14 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do processUserContactRequest :: ACommand 'Agent e -> ConnectionEntity -> Connection -> UserContact -> m () processUserContactRequest agentMsg connEntity conn UserContact {userContactLinkId} = case agentMsg of REQ invId _ connInfo -> do - ChatMessage {chatVRange, chatMsgEvent} <- parseChatMessage conn connInfo - case chatMsgEvent of - XContact p xContactId_ -> profileContactRequest invId chatVRange p xContactId_ - XInfo p -> profileContactRequest invId chatVRange p Nothing - -- TODO show/log error, other events in contact request - _ -> pure () + parseChatMessage conn connInfo >>= \case + [ChatMessage {chatVRange, chatMsgEvent}] -> do + case chatMsgEvent of + XContact p xContactId_ -> profileContactRequest invId chatVRange p xContactId_ + XInfo p -> profileContactRequest invId chatVRange p Nothing + -- TODO show/log error, other events in contact request + _ -> pure () + _ -> messageError "REQ: batching not supported" MERR _ err -> do toView $ CRChatError (Just user) (ChatErrorAgent err $ Just connEntity) incAuthErrCounter connEntity conn err @@ -4955,19 +4955,21 @@ processAgentMessageConn user@User {userId} corrId agentConnId agentMessage = do saveConnInfo :: Connection -> ConnInfo -> m Connection saveConnInfo activeConn connInfo = do - ChatMessage {chatVRange, chatMsgEvent} <- parseChatMessage activeConn connInfo - conn' <- updatePeerChatVRange activeConn chatVRange - case chatMsgEvent of - XInfo p -> do - ct <- withStore $ \db -> createDirectContact db user conn' p - toView $ CRContactConnecting user ct - pure conn' - XGrpLinkInv glInv -> do - (gInfo, host) <- withStore $ \db -> createGroupInvitedViaLink db user conn' glInv - toView $ CRGroupLinkConnecting user gInfo host - pure conn' - -- TODO show/log error, other events in SMP confirmation - _ -> pure conn' + parseChatMessage activeConn connInfo >>= \case + [ChatMessage {chatVRange, chatMsgEvent}] -> do + conn' <- updatePeerChatVRange activeConn chatVRange + case chatMsgEvent of + XInfo p -> do + ct <- withStore $ \db -> createDirectContact db user conn' p + toView $ CRContactConnecting user ct + pure conn' + XGrpLinkInv glInv -> do + (gInfo, host) <- withStore $ \db -> createGroupInvitedViaLink db user conn' glInv + toView $ CRGroupLinkConnecting user gInfo host + pure conn' + -- TODO show/log error, other events in SMP confirmation + _ -> pure conn' + _ -> throwChatError $ CEException "saveConnInfo: batching not supported" xGrpMemNew :: GroupInfo -> GroupMember -> MemberInfo -> RcvMessage -> UTCTime -> m () xGrpMemNew gInfo m memInfo@(MemberInfo memId memRole _ memberProfile) msg brokerTs = do @@ -5347,11 +5349,11 @@ sendFileInline_ FileTransferMeta {filePath, chunkSize} sharedMsgId sendMsg = else sendChunks (chunkNo + 1) rest chSize = fromIntegral chunkSize -parseChatMessage :: ChatMonad m => Connection -> ByteString -> m (ChatMessage 'Json) +parseChatMessage :: ChatMonad m => Connection -> ByteString -> m [ChatMessage 'Json] parseChatMessage conn = parseChatMessage_ conn Nothing {-# INLINE parseChatMessage #-} -parseAChatMessage :: ChatMonad m => Connection -> MsgMeta -> ByteString -> m AChatMessage +parseAChatMessage :: ChatMonad m => Connection -> MsgMeta -> ByteString -> m [AChatMessage] parseAChatMessage conn msgMeta = parseChatMessage_ conn (Just msgMeta) {-# INLINE parseAChatMessage #-} @@ -5536,7 +5538,7 @@ createSndMessage chatMsgEvent connOrGroupId = do gVar <- asks idsDrg ChatConfig {chatVRange} <- asks config withStore $ \db -> createNewSndMessage db gVar connOrGroupId $ \sharedMsgId -> - let msgBody = strEncode ChatMessage {chatVRange, msgId = Just sharedMsgId, chatMsgEvent} + let msgBody = strEncode [ChatMessage {chatVRange, msgId = Just sharedMsgId, chatMsgEvent}] in NewMessage {chatMsgEvent, msgBody} sendBatchedDirectMessages :: (MsgEncodingI e, ChatMonad m) => Connection -> [ChatMsgEvent e] -> ConnOrGroupId -> m (SndMessage, Int64) @@ -5559,16 +5561,16 @@ createBatchedSndMessage events connOrGroupId = do -- - * return list of SndMessages? it's not necessary for current use cases withStore $ \db -> createNewSndMessage db gVar connOrGroupId $ \sharedMsgId -> let chatMsgEvent = XOk -- dummy - msgBody = strEncode ChatMessage {chatVRange, msgId = Just sharedMsgId, chatMsgEvent} + msgBody = strEncode [ChatMessage {chatVRange, msgId = Just sharedMsgId, chatMsgEvent}] in NewMessage {chatMsgEvent, msgBody} directMessage :: (MsgEncodingI e, ChatMonad m) => ChatMsgEvent e -> m ByteString directMessage chatMsgEvent = do ChatConfig {chatVRange} <- asks config - pure $ strEncode ChatMessage {chatVRange, msgId = Nothing, chatMsgEvent} + pure $ strEncode [ChatMessage {chatVRange, msgId = Nothing, chatMsgEvent}] deliverMessage :: ChatMonad m => Connection -> CMEventTag e -> MsgBody -> MessageId -> m Int64 -deliverMessage conn@Connection {connId} cmEventTag msgBody msgId = do +deliverMessage conn cmEventTag msgBody msgId = do let msgFlags = MsgFlags {notification = hasNotification cmEventTag} deliverMessage' conn msgFlags msgBody msgId @@ -5652,13 +5654,17 @@ sendPendingGroupMessages user GroupMember {groupMemberId, localDisplayName} conn saveDirectRcvMSG :: ChatMonad m => Connection -> MsgMeta -> CommandId -> MsgBody -> m (Connection, RcvMessage) saveDirectRcvMSG conn@Connection {connId} agentMsgMeta agentAckCmdId msgBody = do - ACMsg _ ChatMessage {chatVRange, msgId = sharedMsgId_, chatMsgEvent} <- parseAChatMessage conn agentMsgMeta msgBody - conn' <- updatePeerChatVRange conn chatVRange - let agentMsgId = fst $ recipient agentMsgMeta - newMsg = NewMessage {chatMsgEvent, msgBody} - rcvMsgDelivery = RcvMsgDelivery {connId, agentMsgId, agentMsgMeta, agentAckCmdId} - msg <- withStore $ \db -> createNewMessageAndRcvMsgDelivery db (ConnectionId connId) newMsg sharedMsgId_ rcvMsgDelivery Nothing - pure (conn', msg) + parseAChatMessage conn agentMsgMeta msgBody >>= \case + [ACMsg _ ChatMessage {chatVRange, msgId = sharedMsgId_, chatMsgEvent}] -> do + conn' <- updatePeerChatVRange conn chatVRange + let agentMsgId = fst $ recipient agentMsgMeta + newMsg = NewMessage {chatMsgEvent, msgBody} + rcvMsgDelivery = RcvMsgDelivery {connId, agentMsgId, agentMsgMeta, agentAckCmdId} + msg <- withStore $ \db -> createNewMessageAndRcvMsgDelivery db (ConnectionId connId) newMsg sharedMsgId_ rcvMsgDelivery Nothing + pure (conn', msg) + _ -> + -- TODO [batch send] refactor direct message processing same as groups + error "saveDirectRcvMSG: batching not supported" saveGroupRcvMsg :: (MsgEncodingI e, ChatMonad m) => User -> GroupId -> GroupMember -> Connection -> MsgMeta -> CommandId -> MsgBody -> ChatMessage e -> m (GroupMember, Connection, RcvMessage) saveGroupRcvMsg user groupId authorMember conn@Connection {connId} agentMsgMeta agentAckCmdId msgBody ChatMessage {chatVRange, msgId = sharedMsgId_, chatMsgEvent} = do diff --git a/src/Simplex/Chat/Migrations/M20231206_recreate_msg_deliveries.hs b/src/Simplex/Chat/Migrations/M20231206_recreate_msg_deliveries.hs new file mode 100644 index 0000000000..6477f42b2c --- /dev/null +++ b/src/Simplex/Chat/Migrations/M20231206_recreate_msg_deliveries.hs @@ -0,0 +1,85 @@ +{-# LANGUAGE QuasiQuotes #-} + +module Simplex.Chat.Migrations.M20231206_recreate_msg_deliveries where + +import Database.SQLite.Simple (Query) +import Database.SQLite.Simple.QQ (sql) + +m20231206_recreate_msg_deliveries :: Query +m20231206_recreate_msg_deliveries = + [sql| +DROP INDEX msg_delivery_events_msg_delivery_id; +DROP TABLE msg_delivery_events; + +DROP INDEX idx_msg_deliveries_message_id; +DROP INDEX idx_msg_deliveries_agent_ack_cmd_id; + +CREATE TABLE new_msg_deliveries( + msg_delivery_id INTEGER PRIMARY KEY, + message_id INTEGER NOT NULL REFERENCES messages ON DELETE CASCADE, -- non UNIQUE for group messages + connection_id INTEGER NOT NULL REFERENCES connections ON DELETE CASCADE, + agent_msg_id INTEGER, -- internal agent message ID (NULL while pending), non UNIQUE for batched messages + agent_msg_meta TEXT, -- JSON with timestamps etc. sent in MSG, NULL for sent + chat_ts TEXT NOT NULL DEFAULT(datetime('now')), + created_at TEXT CHECK(created_at NOT NULL), + updated_at TEXT CHECK(updated_at NOT NULL), + agent_ack_cmd_id INTEGER, -- broker_ts for received, created_at for sent + delivery_status TEXT -- MsgDeliveryStatus +); + +INSERT INTO new_msg_deliveries + (msg_delivery_id, message_id, connection_id, agent_msg_id, agent_msg_meta, chat_ts, created_at, updated_at, agent_ack_cmd_id) +SELECT + msg_delivery_id, message_id, connection_id, agent_msg_id, agent_msg_meta, chat_ts, created_at, updated_at, agent_ack_cmd_id + FROM msg_deliveries; + +DROP TABLE msg_deliveries; +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); +|] + +down_m20231206_recreate_msg_deliveries :: Query +down_m20231206_recreate_msg_deliveries = + [sql| +DROP INDEX idx_msg_deliveries_message_id; +DROP INDEX idx_msg_deliveries_agent_ack_cmd_id; +DROP INDEX idx_msg_deliveries_agent_msg_id; + +CREATE TABLE old_msg_deliveries( + msg_delivery_id INTEGER PRIMARY KEY, + message_id INTEGER NOT NULL REFERENCES messages ON DELETE CASCADE, -- non UNIQUE for group messages + connection_id INTEGER NOT NULL REFERENCES connections ON DELETE CASCADE, + agent_msg_id INTEGER, -- internal agent message ID(NULL while pending) + agent_msg_meta TEXT, -- JSON with timestamps etc. sent in MSG, NULL for sent + chat_ts TEXT NOT NULL DEFAULT(datetime('now')), + created_at TEXT CHECK(created_at NOT NULL), + updated_at TEXT CHECK(updated_at NOT NULL), + agent_ack_cmd_id INTEGER, -- broker_ts for received, created_at for sent + UNIQUE(connection_id, agent_msg_id) +); + +INSERT INTO old_msg_deliveries + (msg_delivery_id, message_id, connection_id, agent_msg_id, agent_msg_meta, chat_ts, created_at, updated_at, agent_ack_cmd_id) +SELECT + msg_delivery_id, message_id, connection_id, agent_msg_id, agent_msg_meta, chat_ts, created_at, updated_at, agent_ack_cmd_id + FROM msg_deliveries; + +DROP TABLE msg_deliveries; +ALTER TABLE old_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 TABLE msg_delivery_events ( + msg_delivery_event_id INTEGER PRIMARY KEY, + msg_delivery_id INTEGER NOT NULL REFERENCES msg_deliveries ON DELETE CASCADE, -- non UNIQUE for multiple events per msg delivery + delivery_status TEXT NOT NULL, -- see MsgDeliveryStatus for allowed values + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')), + UNIQUE (msg_delivery_id, delivery_status) +); +CREATE INDEX msg_delivery_events_msg_delivery_id ON msg_delivery_events(msg_delivery_id); +|] diff --git a/src/Simplex/Chat/Protocol.hs b/src/Simplex/Chat/Protocol.hs index 30248f6126..27784b4561 100644 --- a/src/Simplex/Chat/Protocol.hs +++ b/src/Simplex/Chat/Protocol.hs @@ -471,18 +471,21 @@ data ExtMsgContent = ExtMsgContent {content :: MsgContent, file :: Maybe FileInv $(JQ.deriveJSON defaultJSON ''QuotedMsg) -instance MsgEncodingI e => StrEncoding (ChatMessage e) where +instance MsgEncodingI e => StrEncoding [ChatMessage e] where strEncode msg = case chatToAppMessage msg of AMJson m -> LB.toStrict $ J.encode m AMBinary m -> strEncode m strP = (\(ACMsg _ m) -> checkEncoding m) <$?> strP -instance StrEncoding AChatMessage where - strEncode (ACMsg _ m) = strEncode m +instance StrEncoding [AChatMessage] where + strEncode [(ACMsg _ m)] = strEncode m strP = A.peekChar' >>= \case - '{' -> ACMsg SJson <$> ((appJsonToCM <=< J.eitherDecodeStrict') <$?> A.takeByteString) - _ -> ACMsg SBinary <$> (appBinaryToCM <$?> strP) + '{' -> (:[]) <$> parseItem + -- '[' -> + _ -> (:[]) . ACMsg SBinary <$> (appBinaryToCM <$?> strP) + where + parseItem = ACMsg SJson <$> ((appJsonToCM <=< J.eitherDecodeStrict') <$?> A.takeByteString) parseMsgContainer :: J.Object -> JT.Parser MsgContainer parseMsgContainer v = diff --git a/src/Simplex/Chat/Store/Migrations.hs b/src/Simplex/Chat/Store/Migrations.hs index 31d0525dba..f143299e28 100644 --- a/src/Simplex/Chat/Store/Migrations.hs +++ b/src/Simplex/Chat/Store/Migrations.hs @@ -91,6 +91,7 @@ import Simplex.Chat.Migrations.M20231107_indexes import Simplex.Chat.Migrations.M20231113_group_forward import Simplex.Chat.Migrations.M20231114_remote_control import Simplex.Chat.Migrations.M20231126_remote_ctrl_address +import Simplex.Chat.Migrations.M20231206_recreate_msg_deliveries import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration (..)) schemaMigrations :: [(String, Query, Maybe Query)] @@ -181,7 +182,8 @@ schemaMigrations = ("20231107_indexes", m20231107_indexes, Just down_m20231107_indexes), ("20231113_group_forward", m20231113_group_forward, Just down_m20231113_group_forward), ("20231114_remote_control", m20231114_remote_control, Just down_m20231114_remote_control), - ("20231126_remote_ctrl_address", m20231126_remote_ctrl_address, Just down_m20231126_remote_ctrl_address) + ("20231126_remote_ctrl_address", m20231126_remote_ctrl_address, Just down_m20231126_remote_ctrl_address), + ("20231206_recreate_msg_deliveries", m20231206_recreate_msg_deliveries, Just down_m20231206_recreate_msg_deliveries) ] -- | The list of migrations in ascending order by date