schema, refactor group processing for batched messages

2026-07-09 16:41:47 +00:00 · 2023-12-06 20:10:56 +04:00
parent fad0e4181f
commit ecd66ad7f3
5 changed files with 241 additions and 144 deletions
@@ -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
@@ -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
@@ -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);
+|]
@@ -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 =
@@ -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