store messages (#166)

Co-authored-by: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com>

src/Simplex/Chat.hs

@@ -51,6 +51,7 @@ import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), defaultAgentConfig)
 import Simplex.Messaging.Agent.Protocol
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Parsers (parseAll)
+import Simplex.Messaging.Protocol (MsgBody)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Util (bshow, raceAny_, tryError)
 import System.Exit (exitFailure, exitSuccess)
@@ -262,7 +263,7 @@ processChatCommand user@User {userId, profile} = \case
     (agentConnId, cReq) <- withAgent (`createConnection` SCMInvitation)
     GroupMember {memberId} <- withStore $ \st -> createContactGroupMember st gVar user groupId contact memRole agentConnId
     let msg = XGrpInv $ GroupInvitation (userMemberId, userRole) (memberId, memRole) cReq groupProfile
-    sendDirectMessage (contactConnId contact) msg
+    sendDirectMessage (contactConn contact) msg
     showSentGroupInvitation gName cName
     setActive $ ActiveG gName
   JoinGroup gName -> do
@@ -306,7 +307,6 @@ processChatCommand user@User {userId, profile} = \case
     showGroupMembers group
   ListGroups -> withStore (`getUserGroupNames` userId) >>= showGroupsList
   SendGroupMessage gName msg -> do
-    -- TODO save sent messages
     -- TODO save pending message delivery for members without connections
     Group {members, membership} <- withStore $ \st -> getGroup st user gName
     unless (memberActive membership) $ chatError CEGroupMemberUserRemoved
@@ -320,7 +320,7 @@ processChatCommand user@User {userId, profile} = \case
     let fileInv = FileInvitation {fileName = takeFileName f, fileSize, fileConnReq}
     SndFileTransfer {fileId} <- withStore $ \st ->
       createSndFileTransfer st userId contact f fileInv agentConnId chSize
-    sendDirectMessage (contactConnId contact) $ XFile fileInv
+    sendDirectMessage (contactConn contact) $ XFile fileInv
     showSentFileInfo fileId
     setActive $ ActiveC cName
   SendGroupFile gName f -> do
@@ -332,8 +332,9 @@ processChatCommand user@User {userId, profile} = \case
       (connId, fileConnReq) <- withAgent (`createConnection` SCMInvitation)
       pure (m, connId, FileInvitation {fileName, fileSize, fileConnReq})
     fileId <- withStore $ \st -> createSndGroupFileTransfer st userId group ms f fileSize chSize
+    -- TODO sendGroupMessage - same file invitation to all
     forM_ ms $ \(m, _, fileInv) ->
-      traverse (`sendDirectMessage` XFile fileInv) $ memberConnId m
+      traverse (`sendDirectMessage` XFile fileInv) $ memberConn m
     showSentFileInfo fileId
     setActive $ ActiveG gName
   ReceiveFile fileId filePath_ -> do
@@ -361,7 +362,7 @@ processChatCommand user@User {userId, profile} = \case
     user' <- withStore $ \st -> updateUserProfile st user p
     asks currentUser >>= atomically . (`writeTVar` user')
     contacts <- withStore (`getUserContacts` user)
-    forM_ contacts $ \ct -> sendDirectMessage (contactConnId ct) $ XInfo p
+    forM_ contacts $ \ct -> sendDirectMessage (contactConn ct) $ XInfo p
     showUserProfileUpdated user user'
   ShowProfile -> showUserProfile profile
   QuitChat -> liftIO exitSuccess
@@ -375,7 +376,7 @@ processChatCommand user@User {userId, profile} = \case
     sendMessageCmd cName msg = do
       contact <- withStore $ \st -> getContact st userId cName
       let msgEvent = XMsgNew $ MsgContent MTText [] [MsgContentBody {contentType = SimplexContentType XCText, contentData = msg}]
-      sendDirectMessage (contactConnId contact) msgEvent
+      sendDirectMessage (contactConn contact) msgEvent
       setActive $ ActiveC cName
     contactMember :: Contact -> [GroupMember] -> Maybe GroupMember
     contactMember Contact {contactId} =
@@ -530,21 +531,27 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
           allowAgentConnection conn confId $ XInfo profile
         INFO connInfo ->
           saveConnInfo conn connInfo
-        MSG meta _ ->
+        MSG meta msgBody -> do
+          _ <- saveRcvMSG conn meta msgBody
           withAckMessage agentConnId meta $ pure ()
+          ackMsgDeliveryEvent conn meta
+        SENT msgId ->
+          sentMsgDeliveryEvent conn msgId
         _ -> pure ()
       Just ct@Contact {localDisplayName = c} -> case agentMsg of
-        MSG meta msgBody -> withAckMessage agentConnId meta $ do
-          ChatMessage {chatMsgEvent} <- liftEither $ parseChatMessage msgBody
-          case chatMsgEvent of
-            XMsgNew (MsgContent MTText [] body) -> newTextMessage c meta $ find (isSimplexContentType XCText) body
-            XFile fInv -> processFileInvitation ct meta fInv
-            XInfo p -> xInfo ct p
-            XGrpInv gInv -> processGroupInvitation ct gInv
-            XInfoProbe probe -> xInfoProbe ct probe
-            XInfoProbeCheck probeHash -> xInfoProbeCheck ct probeHash
-            XInfoProbeOk probe -> xInfoProbeOk ct probe
-            _ -> pure ()
+        MSG meta msgBody -> do
+          chatMsgEvent <- saveRcvMSG conn meta msgBody
+          withAckMessage agentConnId meta $
+            case chatMsgEvent of
+              XMsgNew (MsgContent MTText [] body) -> newTextMessage c meta $ find (isSimplexContentType XCText) body
+              XFile fInv -> processFileInvitation ct meta fInv
+              XInfo p -> xInfo ct p
+              XGrpInv gInv -> processGroupInvitation ct gInv
+              XInfoProbe probe -> xInfoProbe ct probe
+              XInfoProbeCheck probeHash -> xInfoProbeCheck ct probeHash
+              XInfoProbeOk probe -> xInfoProbeOk ct probe
+              _ -> pure ()
+          ackMsgDeliveryEvent conn meta
         CONF confId connInfo -> do
           -- confirming direct connection with a member
           ChatMessage {chatMsgEvent} <- liftEither $ parseChatMessage connInfo
@@ -576,6 +583,8 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
               when (memberIsReady m) $ do
                 notifyMemberConnected gName m
                 when (memberCategory m == GCPreMember) $ probeMatchingContacts ct
+        SENT msgId ->
+          sentMsgDeliveryEvent conn msgId
         END -> do
           showContactAnotherClient c
           showToast (c <> "> ") "connected to another client"
@@ -641,7 +650,7 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
             intros <- withStore $ \st -> createIntroductions st group m
             sendGroupMessage members . XGrpMemNew $ memberInfo m
             forM_ intros $ \intro -> do
-              sendDirectMessage agentConnId . XGrpMemIntro . memberInfo $ reMember intro
+              sendDirectMessage conn . XGrpMemIntro . memberInfo $ reMember intro
               withStore $ \st -> updateIntroStatus st intro GMIntroSent
           _ -> do
             -- TODO send probe and decide whether to use existing contact connection or the new contact connection
@@ -654,20 +663,24 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
                 when (contactIsReady ct) $ do
                   notifyMemberConnected gName m
                   when (memberCategory m == GCPreMember) $ probeMatchingContacts ct
-      MSG meta msgBody -> withAckMessage agentConnId meta $ do
-        ChatMessage {chatMsgEvent} <- liftEither $ parseChatMessage msgBody
-        case chatMsgEvent of
-          XMsgNew (MsgContent MTText [] body) ->
-            newGroupTextMessage gName m meta $ find (isSimplexContentType XCText) body
-          XFile fInv -> processGroupFileInvitation gName m meta fInv
-          XGrpMemNew memInfo -> xGrpMemNew gName m memInfo
-          XGrpMemIntro memInfo -> xGrpMemIntro gName m memInfo
-          XGrpMemInv memId introInv -> xGrpMemInv gName m memId introInv
-          XGrpMemFwd memInfo introInv -> xGrpMemFwd gName m memInfo introInv
-          XGrpMemDel memId -> xGrpMemDel gName m memId
-          XGrpLeave -> xGrpLeave gName m
-          XGrpDel -> xGrpDel gName m
-          _ -> messageError $ "unsupported message: " <> T.pack (show chatMsgEvent)
+      MSG meta msgBody -> do
+        chatMsgEvent <- saveRcvMSG conn meta msgBody
+        withAckMessage agentConnId meta $
+          case chatMsgEvent of
+            XMsgNew (MsgContent MTText [] body) ->
+              newGroupTextMessage gName m meta $ find (isSimplexContentType XCText) body
+            XFile fInv -> processGroupFileInvitation gName m meta fInv
+            XGrpMemNew memInfo -> xGrpMemNew gName m memInfo
+            XGrpMemIntro memInfo -> xGrpMemIntro conn gName m memInfo
+            XGrpMemInv memId introInv -> xGrpMemInv gName m memId introInv
+            XGrpMemFwd memInfo introInv -> xGrpMemFwd gName m memInfo introInv
+            XGrpMemDel memId -> xGrpMemDel gName m memId
+            XGrpLeave -> xGrpLeave gName m
+            XGrpDel -> xGrpDel gName m
+            _ -> messageError $ "unsupported message: " <> T.pack (show chatMsgEvent)
+        ackMsgDeliveryEvent conn meta
+      SENT msgId ->
+        sentMsgDeliveryEvent conn msgId
       _ -> pure ()
 
     processSndFileConn :: ACommand 'Agent -> Connection -> SndFileTransfer -> m ()
@@ -676,6 +689,7 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
         CONF confId connInfo -> do
           ChatMessage {chatMsgEvent} <- liftEither $ parseChatMessage connInfo
           case chatMsgEvent of
+            -- TODO save XFileAcpt message
             XFileAcpt name
               | name == fileName -> do
                 withStore $ \st -> updateSndFileStatus st ft FSAccepted
@@ -755,6 +769,14 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
     withAckMessage cId MsgMeta {recipient = (msgId, _)} action =
       action `E.finally` withAgent (\a -> ackMessage a cId msgId `catchError` \_ -> pure ())
 
+    ackMsgDeliveryEvent :: Connection -> MsgMeta -> m ()
+    ackMsgDeliveryEvent Connection {connId} MsgMeta {recipient = (msgId, _)} =
+      withStore $ \st -> createRcvMsgDeliveryEvent st connId msgId MDSRcvAcknowledged
+
+    sentMsgDeliveryEvent :: Connection -> AgentMsgId -> m ()
+    sentMsgDeliveryEvent Connection {connId} msgId =
+      withStore $ \st -> createSndMsgDeliveryEvent st connId msgId MDSSndSent
+
     badRcvFileChunk :: RcvFileTransfer -> String -> m ()
     badRcvFileChunk ft@RcvFileTransfer {fileStatus} err =
       case fileStatus of
@@ -773,13 +795,13 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
     probeMatchingContacts ct = do
       gVar <- asks idsDrg
       (probe, probeId) <- withStore $ \st -> createSentProbe st gVar userId ct
-      sendDirectMessage (contactConnId ct) $ XInfoProbe probe
+      sendDirectMessage (contactConn ct) $ XInfoProbe probe
       cs <- withStore (\st -> getMatchingContacts st userId ct)
       let probeHash = C.sha256Hash probe
       forM_ cs $ \c -> sendProbeHash c probeHash probeId `catchError` const (pure ())
       where
         sendProbeHash c probeHash probeId = do
-          sendDirectMessage (contactConnId c) $ XInfoProbeCheck probeHash
+          sendDirectMessage (contactConn c) $ XInfoProbeCheck probeHash
           withStore $ \st -> createSentProbeHash st userId probeId c
 
     messageWarning :: Text -> m ()
@@ -792,7 +814,7 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
     newTextMessage c meta = \case
       Just MsgContentBody {contentData = bs} -> do
         let text = safeDecodeUtf8 bs
-        showReceivedMessage c (snd $ broker meta) (msgPlain text) (integrity meta)
+        showReceivedMessage c (snd $ broker meta) (msgPlain text) (integrity (meta :: MsgMeta))
         showToast (c <> "> ") text
         setActive $ ActiveC c
       _ -> messageError "x.msg.new: no expected message body"
@@ -801,7 +823,7 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
     newGroupTextMessage gName GroupMember {localDisplayName = c} meta = \case
       Just MsgContentBody {contentData = bs} -> do
         let text = safeDecodeUtf8 bs
-        showReceivedGroupMessage gName c (snd $ broker meta) (msgPlain text) (integrity meta)
+        showReceivedGroupMessage gName c (snd $ broker meta) (msgPlain text) (integrity (meta :: MsgMeta))
         showToast ("#" <> gName <> " " <> c <> "> ") text
         setActive $ ActiveG gName
       _ -> messageError "x.msg.new: no expected message body"
@@ -811,14 +833,14 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
       -- TODO chunk size has to be sent as part of invitation
       chSize <- asks $ fileChunkSize . config
       ft <- withStore $ \st -> createRcvFileTransfer st userId contact fInv chSize
-      showReceivedMessage c (snd $ broker meta) (receivedFileInvitation ft) (integrity meta)
+      showReceivedMessage c (snd $ broker meta) (receivedFileInvitation ft) (integrity (meta :: MsgMeta))
       setActive $ ActiveC c
 
     processGroupFileInvitation :: GroupName -> GroupMember -> MsgMeta -> FileInvitation -> m ()
     processGroupFileInvitation gName m@GroupMember {localDisplayName = c} meta fInv = do
       chSize <- asks $ fileChunkSize . config
       ft <- withStore $ \st -> createRcvGroupFileTransfer st userId m fInv chSize
-      showReceivedGroupMessage gName c (snd $ broker meta) (receivedFileInvitation ft) (integrity meta)
+      showReceivedGroupMessage gName c (snd $ broker meta) (receivedFileInvitation ft) (integrity (meta :: MsgMeta))
       setActive $ ActiveG gName
 
     processGroupInvitation :: Contact -> GroupInvitation -> m ()
@@ -846,7 +868,7 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
     probeMatch :: Contact -> Contact -> ByteString -> m ()
     probeMatch c1@Contact {profile = p1} c2@Contact {profile = p2} probe =
       when (p1 == p2) $ do
-        sendDirectMessage (contactConnId c1) $ XInfoProbeOk probe
+        sendDirectMessage (contactConn c1) $ XInfoProbeOk probe
         mergeContacts c1 c2
 
     xInfoProbeOk :: Contact -> ByteString -> m ()
@@ -859,9 +881,6 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
       withStore $ \st -> mergeContactRecords st userId to from
       showContactsMerged to from
 
-    parseChatMessage :: ByteString -> Either ChatError ChatMessage
-    parseChatMessage msgBody = first ChatErrorMessage (parseAll rawChatMessageP msgBody >>= toChatMessage)
-
     saveConnInfo :: Connection -> ConnInfo -> m ()
     saveConnInfo activeConn connInfo = do
       ChatMessage {chatMsgEvent} <- liftEither $ parseChatMessage connInfo
@@ -881,8 +900,8 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
             newMember <- withStore $ \st -> createNewGroupMember st user group memInfo GCPostMember GSMemAnnounced
             showJoinedGroupMemberConnecting gName m newMember
 
-    xGrpMemIntro :: GroupName -> GroupMember -> MemberInfo -> m ()
-    xGrpMemIntro gName m memInfo@(MemberInfo memId _ _) =
+    xGrpMemIntro :: Connection -> GroupName -> GroupMember -> MemberInfo -> m ()
+    xGrpMemIntro conn gName m memInfo@(MemberInfo memId _ _) =
       case memberCategory m of
         GCHostMember -> do
           group <- withStore $ \st -> getGroup st user gName
@@ -893,7 +912,7 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
               (directConnId, directConnReq) <- withAgent (`createConnection` SCMInvitation)
               newMember <- withStore $ \st -> createIntroReMember st user group m memInfo groupConnId directConnId
               let msg = XGrpMemInv memId IntroInvitation {groupConnReq, directConnReq}
-              sendDirectMessage agentConnId msg
+              sendDirectMessage conn msg
               withStore $ \st -> updateGroupMemberStatus st userId newMember GSMemIntroInvited
         _ -> messageError "x.grp.mem.intro can be only sent by host member"
 
@@ -908,8 +927,8 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
               intro <- withStore $ \st -> saveIntroInvitation st reMember m introInv
               case activeConn (reMember :: GroupMember) of
                 Nothing -> pure () -- this is not an error, introduction will be forwarded once the member is connected
-                Just Connection {agentConnId = reAgentConnId} -> do
-                  sendDirectMessage reAgentConnId $ XGrpMemFwd (memberInfo m) introInv
+                Just reConn -> do
+                  sendDirectMessage reConn $ XGrpMemFwd (memberInfo m) introInv
                   withStore $ \st -> updateIntroStatus st intro GMIntroInvForwarded
         _ -> messageError "x.grp.mem.inv can be only sent by invitee member"
 
@@ -964,6 +983,9 @@ processAgentMessage user@User {userId, profile} agentConnId agentMessage = do
       mapM_ deleteMemberConnection ms
       showGroupDeleted gName m
 
+parseChatMessage :: ByteString -> Either ChatError ChatMessage
+parseChatMessage msgBody = first ChatErrorMessage (parseAll rawChatMessageP msgBody >>= toChatMessage)
+
 sendFileChunk :: ChatMonad m => SndFileTransfer -> m ()
 sendFileChunk ft@SndFileTransfer {fileId, fileStatus, agentConnId} =
   unless (fileStatus == FSComplete || fileStatus == FSCancelled) $
@@ -1069,22 +1091,42 @@ deleteMemberConnection m@GroupMember {activeConn} = do
   -- withStore $ \st -> deleteGroupMemberConnection st userId m
   forM_ activeConn $ \conn -> withStore $ \st -> updateConnectionStatus st conn ConnDeleted
 
-sendDirectMessage :: ChatMonad m => ConnId -> ChatMsgEvent -> m ()
-sendDirectMessage agentConnId chatMsgEvent =
-  void . withAgent $ \a -> sendMessage a agentConnId $ directMessage chatMsgEvent
+sendDirectMessage :: ChatMonad m => Connection -> ChatMsgEvent -> m ()
+sendDirectMessage conn chatMsgEvent = do
+  let msgBody = directMessage chatMsgEvent
+      newMsg = NewMessage {direction = MDSnd, chatMsgEventType = toChatEventType chatMsgEvent, msgBody}
+  -- can be done in transaction after sendMessage, probably shouldn't
+  msgId <- withStore $ \st -> createNewMessage st newMsg
+  deliverMessage conn msgBody msgId
 
 directMessage :: ChatMsgEvent -> ByteString
 directMessage chatMsgEvent =
   serializeRawChatMessage $
     rawChatMessage ChatMessage {chatMsgId = Nothing, chatMsgEvent, chatDAG = Nothing}
 
+deliverMessage :: ChatMonad m => Connection -> MsgBody -> MessageId -> m ()
+deliverMessage Connection {connId, agentConnId} msgBody msgId = do
+  agentMsgId <- withAgent $ \a -> sendMessage a agentConnId msgBody
+  let sndMsgDelivery = SndMsgDelivery {connId, agentMsgId}
+  withStore $ \st -> createSndMsgDelivery st sndMsgDelivery msgId
+
 sendGroupMessage :: ChatMonad m => [GroupMember] -> ChatMsgEvent -> m ()
 sendGroupMessage members chatMsgEvent = do
-  let msg = directMessage chatMsgEvent
+  let msgBody = directMessage chatMsgEvent
+      newMsg = NewMessage {direction = MDSnd, chatMsgEventType = toChatEventType chatMsgEvent, msgBody}
+  msgId <- withStore $ \st -> createNewMessage st newMsg
   -- TODO once scheduled delivery is implemented memberActive should be changed to memberCurrent
-  withAgent $ \a ->
-    forM_ (filter memberActive members) $
-      traverse (\connId -> sendMessage a connId msg) . memberConnId
+  forM_ (map memberConn $ filter memberActive members) $
+    traverse (\conn -> deliverMessage conn msgBody msgId)
+
+saveRcvMSG :: ChatMonad m => Connection -> MsgMeta -> MsgBody -> m ChatMsgEvent
+saveRcvMSG Connection {connId} agentMsgMeta msgBody = do
+  ChatMessage {chatMsgEvent} <- liftEither $ parseChatMessage msgBody
+  let newMsg = NewMessage {direction = MDRcv, chatMsgEventType = toChatEventType chatMsgEvent, msgBody}
+      agentMsgId = fst $ recipient agentMsgMeta
+      rcvMsgDelivery = RcvMsgDelivery {connId, agentMsgId, agentMsgMeta}
+  withStore $ \st -> createNewMessageAndRcvMsgDelivery st newMsg rcvMsgDelivery
+  pure chatMsgEvent
 
 allowAgentConnection :: ChatMonad m => Connection -> ConfirmationId -> ChatMsgEvent -> m ()
 allowAgentConnection conn@Connection {agentConnId} confId msg = do

src/Simplex/Chat/Protocol.hs

@@ -103,6 +103,30 @@ data ChatMessage = ChatMessage
   }
   deriving (Eq, Show)
 
+toChatEventType :: ChatMsgEvent -> Text
+toChatEventType = \case
+  XMsgNew _ -> "x.msg.new"
+  XFile _ -> "x.file"
+  XFileAcpt _ -> "x.file.acpt"
+  XInfo _ -> "x.info"
+  XContact _ _ -> "x.con"
+  XGrpInv _ -> "x.grp.inv"
+  XGrpAcpt _ -> "x.grp.acpt"
+  XGrpMemNew _ -> "x.grp.mem.new"
+  XGrpMemIntro _ -> "x.grp.mem.intro"
+  XGrpMemInv _ _ -> "x.grp.mem.inv"
+  XGrpMemFwd _ _ -> "x.grp.mem.fwd"
+  XGrpMemInfo _ _ -> "x.grp.mem.info"
+  XGrpMemCon _ -> "x.grp.mem.con"
+  XGrpMemConAll _ -> "x.grp.mem.con.all"
+  XGrpMemDel _ -> "x.grp.mem.del"
+  XGrpLeave -> "x.grp.leave"
+  XGrpDel -> "x.grp.del"
+  XInfoProbe _ -> "x.info.probe"
+  XInfoProbeCheck _ -> "x.info.probe.check"
+  XInfoProbeOk _ -> "x.info.probe.ok"
+  XOk -> "x.ok"
+
 toChatMessage :: RawChatMessage -> Either String ChatMessage
 toChatMessage RawChatMessage {chatMsgId, chatMsgEvent, chatMsgParams, chatMsgBody} = do
   (chatDAG, body) <- getDAG <$> mapM toMsgBodyContent chatMsgBody
@@ -161,9 +185,9 @@ toChatMessage RawChatMessage {chatMsgId, chatMsgEvent, chatMsgParams, chatMsgBod
     ("x.info.probe", [probe]) -> do
       chatMsg . XInfoProbe =<< B64.decode probe
     ("x.info.probe.check", [probeHash]) -> do
-      chatMsg =<< (XInfoProbeCheck <$> B64.decode probeHash)
+      chatMsg . XInfoProbeCheck =<< B64.decode probeHash
     ("x.info.probe.ok", [probe]) -> do
-      chatMsg =<< (XInfoProbeOk <$> B64.decode probe)
+      chatMsg . XInfoProbeOk =<< B64.decode probe
     ("x.ok", []) ->
       chatMsg XOk
     _ -> Left $ "bad syntax or unsupported event " <> B.unpack chatMsgEvent
@@ -216,17 +240,17 @@ rawChatMessage :: ChatMessage -> RawChatMessage
 rawChatMessage ChatMessage {chatMsgId, chatMsgEvent, chatDAG} =
   case chatMsgEvent of
     XMsgNew MsgContent {messageType = t, files, content} ->
-      rawMsg "x.msg.new" (rawMsgType t : toRawFiles files) content
+      rawMsg (rawMsgType t : toRawFiles files) content
     XFile FileInvitation {fileName, fileSize, fileConnReq} ->
-      rawMsg "x.file" [encodeUtf8 $ T.pack fileName, bshow fileSize, serializeConnReq' fileConnReq] []
+      rawMsg [encodeUtf8 $ T.pack fileName, bshow fileSize, serializeConnReq' fileConnReq] []
     XFileAcpt fileName ->
-      rawMsg "x.file.acpt" [encodeUtf8 $ T.pack fileName] []
+      rawMsg [encodeUtf8 $ T.pack fileName] []
     XInfo profile ->
-      rawMsg "x.info" [] [jsonBody profile]
+      rawMsg [] [jsonBody profile]
     XContact profile Nothing ->
-      rawMsg "x.con" [] [jsonBody profile]
+      rawMsg [] [jsonBody profile]
     XContact profile (Just MsgContent {messageType = t, files, content}) ->
-      rawMsg "x.con" (rawMsgType t : toRawFiles files) (jsonBody profile : content)
+      rawMsg (rawMsgType t : toRawFiles files) (jsonBody profile : content)
     XGrpInv (GroupInvitation (fromMemId, fromRole) (memId, role) cReq groupProfile) ->
       let params =
             [ B64.encode fromMemId,
@@ -235,17 +259,17 @@ rawChatMessage ChatMessage {chatMsgId, chatMsgEvent, chatDAG} =
               serializeMemberRole role,
               serializeConnReq' cReq
             ]
-       in rawMsg "x.grp.inv" params [jsonBody groupProfile]
+       in rawMsg params [jsonBody groupProfile]
     XGrpAcpt memId ->
-      rawMsg "x.grp.acpt" [B64.encode memId] []
+      rawMsg [B64.encode memId] []
     XGrpMemNew (MemberInfo memId role profile) ->
       let params = [B64.encode memId, serializeMemberRole role]
-       in rawMsg "x.grp.mem.new" params [jsonBody profile]
+       in rawMsg params [jsonBody profile]
     XGrpMemIntro (MemberInfo memId role profile) ->
-      rawMsg "x.grp.mem.intro" [B64.encode memId, serializeMemberRole role] [jsonBody profile]
+      rawMsg [B64.encode memId, serializeMemberRole role] [jsonBody profile]
     XGrpMemInv memId IntroInvitation {groupConnReq, directConnReq} ->
       let params = [B64.encode memId, serializeConnReq' groupConnReq, serializeConnReq' directConnReq]
-       in rawMsg "x.grp.mem.inv" params []
+       in rawMsg params []
     XGrpMemFwd (MemberInfo memId role profile) IntroInvitation {groupConnReq, directConnReq} ->
       let params =
             [ B64.encode memId,
@@ -253,30 +277,31 @@ rawChatMessage ChatMessage {chatMsgId, chatMsgEvent, chatDAG} =
               serializeConnReq' groupConnReq,
               serializeConnReq' directConnReq
             ]
-       in rawMsg "x.grp.mem.fwd" params [jsonBody profile]
+       in rawMsg params [jsonBody profile]
     XGrpMemInfo memId profile ->
-      rawMsg "x.grp.mem.info" [B64.encode memId] [jsonBody profile]
+      rawMsg [B64.encode memId] [jsonBody profile]
     XGrpMemCon memId ->
-      rawMsg "x.grp.mem.con" [B64.encode memId] []
+      rawMsg [B64.encode memId] []
     XGrpMemConAll memId ->
-      rawMsg "x.grp.mem.con.all" [B64.encode memId] []
+      rawMsg [B64.encode memId] []
     XGrpMemDel memId ->
-      rawMsg "x.grp.mem.del" [B64.encode memId] []
+      rawMsg [B64.encode memId] []
     XGrpLeave ->
-      rawMsg "x.grp.leave" [] []
+      rawMsg [] []
     XGrpDel ->
-      rawMsg "x.grp.del" [] []
+      rawMsg [] []
     XInfoProbe probe ->
-      rawMsg "x.info.probe" [B64.encode probe] []
+      rawMsg [B64.encode probe] []
     XInfoProbeCheck probeHash ->
-      rawMsg "x.info.probe.check" [B64.encode probeHash] []
+      rawMsg [B64.encode probeHash] []
     XInfoProbeOk probe ->
-      rawMsg "x.info.probe.ok" [B64.encode probe] []
+      rawMsg [B64.encode probe] []
     XOk ->
-      rawMsg "x.ok" [] []
+      rawMsg [] []
   where
-    rawMsg :: ByteString -> [ByteString] -> [MsgContentBody] -> RawChatMessage
-    rawMsg event chatMsgParams body =
+    rawMsg :: [ByteString] -> [MsgContentBody] -> RawChatMessage
+    rawMsg chatMsgParams body = do
+      let event = encodeUtf8 $ toChatEventType chatMsgEvent
       RawChatMessage {chatMsgId, chatMsgEvent = event, chatMsgParams, chatMsgBody = rawWithDAG body}
     rawContentInfo :: (ContentType, Int) -> (RawContentType, Int)
     rawContentInfo (t, size) = (rawContentType t, size)

src/Simplex/Chat/Store.hs

@@ -90,6 +90,11 @@ module Simplex.Chat.Store
     getFileTransfer,
     getFileTransferProgress,
     getOnboarding,
+    createNewMessage,
+    createSndMsgDelivery,
+    createNewMessageAndRcvMsgDelivery,
+    createSndMsgDeliveryEvent,
+    createRcvMsgDeliveryEvent,
   )
 where
 
@@ -118,7 +123,7 @@ import qualified Database.SQLite.Simple as DB
 import Database.SQLite.Simple.QQ (sql)
 import Simplex.Chat.Protocol
 import Simplex.Chat.Types
-import Simplex.Messaging.Agent.Protocol (AParty (..), AgentMsgId, ConnId, InvitationId)
+import Simplex.Messaging.Agent.Protocol (AParty (..), AgentMsgId, ConnId, InvitationId, MsgMeta (..))
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore (..), createSQLiteStore, withTransaction)
 import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration (..))
 import qualified Simplex.Messaging.Crypto as C
@@ -742,7 +747,6 @@ mergeContactRecords st userId Contact {contactId = toContactId} Contact {contact
     DB.execute db "UPDATE connections SET contact_id = ? WHERE contact_id = ? AND user_id = ?" (toContactId, fromContactId, userId)
     DB.execute db "UPDATE connections SET via_contact = ? WHERE via_contact = ? AND user_id = ?" (toContactId, fromContactId, userId)
     DB.execute db "UPDATE group_members SET invited_by = ? WHERE invited_by = ? AND user_id = ?" (toContactId, fromContactId, userId)
-    DB.execute db "UPDATE messages SET contact_id = ? WHERE contact_id = ?" (toContactId, fromContactId)
     DB.executeNamed
       db
       [sql|
@@ -1613,6 +1617,101 @@ getOnboarding st userId =
     headOrZero [] = 0
     headOrZero (n : _) = fromOnly n
 
+createNewMessage :: MonadUnliftIO m => SQLiteStore -> NewMessage -> m MessageId
+createNewMessage st newMsg =
+  liftIO . withTransaction st $ \db ->
+    createNewMessage_ db newMsg
+
+createSndMsgDelivery :: MonadUnliftIO m => SQLiteStore -> SndMsgDelivery -> MessageId -> m ()
+createSndMsgDelivery st sndMsgDelivery messageId =
+  liftIO . withTransaction st $ \db -> do
+    msgDeliveryId <- createSndMsgDelivery_ db sndMsgDelivery messageId
+    createMsgDeliveryEvent_ db msgDeliveryId MDSSndAgent
+
+createNewMessageAndRcvMsgDelivery :: MonadUnliftIO m => SQLiteStore -> NewMessage -> RcvMsgDelivery -> m ()
+createNewMessageAndRcvMsgDelivery st newMsg rcvMsgDelivery =
+  liftIO . withTransaction st $ \db -> do
+    messageId <- createNewMessage_ db newMsg
+    msgDeliveryId <- createRcvMsgDelivery_ db rcvMsgDelivery messageId
+    createMsgDeliveryEvent_ db msgDeliveryId MDSRcvAgent
+
+createSndMsgDeliveryEvent :: StoreMonad m => SQLiteStore -> Int64 -> AgentMsgId -> MsgDeliveryStatus 'MDSnd -> m ()
+createSndMsgDeliveryEvent st connId agentMsgId sndMsgDeliveryStatus =
+  liftIOEither . withTransaction st $ \db -> runExceptT $ do
+    msgDeliveryId <- ExceptT $ getMsgDeliveryId_ db connId agentMsgId
+    liftIO $ createMsgDeliveryEvent_ db msgDeliveryId sndMsgDeliveryStatus
+
+createRcvMsgDeliveryEvent :: StoreMonad m => SQLiteStore -> Int64 -> AgentMsgId -> MsgDeliveryStatus 'MDRcv -> m ()
+createRcvMsgDeliveryEvent st connId agentMsgId rcvMsgDeliveryStatus =
+  liftIOEither . withTransaction st $ \db -> runExceptT $ do
+    msgDeliveryId <- ExceptT $ getMsgDeliveryId_ db connId agentMsgId
+    liftIO $ createMsgDeliveryEvent_ db msgDeliveryId rcvMsgDeliveryStatus
+
+createNewMessage_ :: DB.Connection -> NewMessage -> IO MessageId
+createNewMessage_ db NewMessage {direction, chatMsgEventType, msgBody} = do
+  createdAt <- getCurrentTime
+  DB.execute
+    db
+    [sql|
+      INSERT INTO messages
+        (msg_sent, chat_msg_event, msg_body, created_at) VALUES (?,?,?,?);
+    |]
+    (direction, chatMsgEventType, msgBody, createdAt)
+  insertedRowId db
+
+createSndMsgDelivery_ :: DB.Connection -> SndMsgDelivery -> MessageId -> IO Int64
+createSndMsgDelivery_ db SndMsgDelivery {connId, agentMsgId} messageId = do
+  chatTs <- getCurrentTime
+  DB.execute
+    db
+    [sql|
+      INSERT INTO msg_deliveries
+        (message_id, connection_id, agent_msg_id, agent_msg_meta, chat_ts)
+      VALUES (?,?,?,NULL,?);
+    |]
+    (messageId, connId, agentMsgId, chatTs)
+  insertedRowId db
+
+createRcvMsgDelivery_ :: DB.Connection -> RcvMsgDelivery -> MessageId -> IO Int64
+createRcvMsgDelivery_ db RcvMsgDelivery {connId, agentMsgId, agentMsgMeta} messageId = do
+  DB.execute
+    db
+    [sql|
+      INSERT INTO msg_deliveries
+        (message_id, connection_id, agent_msg_id, agent_msg_meta, chat_ts)
+      VALUES (?,?,?,?,?);
+    |]
+    (messageId, connId, agentMsgId, msgMetaJson agentMsgMeta, snd $ broker agentMsgMeta)
+  insertedRowId db
+
+createMsgDeliveryEvent_ :: DB.Connection -> Int64 -> MsgDeliveryStatus d -> IO ()
+createMsgDeliveryEvent_ db msgDeliveryId msgDeliveryStatus = do
+  createdAt <- getCurrentTime
+  DB.execute
+    db
+    [sql|
+      INSERT INTO msg_delivery_events
+        (msg_delivery_id, delivery_status, created_at) VALUES (?,?,?);
+    |]
+    (msgDeliveryId, msgDeliveryStatus, createdAt)
+
+getMsgDeliveryId_ :: DB.Connection -> Int64 -> AgentMsgId -> IO (Either StoreError Int64)
+getMsgDeliveryId_ db connId agentMsgId =
+  toMsgDeliveryId
+    <$> DB.query
+      db
+      [sql|
+        SELECT msg_delivery_id
+        FROM msg_deliveries m
+        WHERE m.connection_id = ? AND m.agent_msg_id == ?
+        LIMIT 1;
+      |]
+      (connId, agentMsgId)
+  where
+    toMsgDeliveryId :: [Only Int64] -> Either StoreError Int64
+    toMsgDeliveryId [Only msgDeliveryId] = Right msgDeliveryId
+    toMsgDeliveryId _ = Left $ SENoMsgDelivery connId agentMsgId
+
 -- | Saves unique local display name based on passed displayName, suffixed with _N if required.
 -- This function should be called inside transaction.
 withLocalDisplayName :: forall a. DB.Connection -> UserId -> Text -> (Text -> IO a) -> IO (Either StoreError a)
@@ -1689,4 +1788,5 @@ data StoreError
   | SEIntroNotFound
   | SEUniqueID
   | SEInternal ByteString
+  | SENoMsgDelivery Int64 AgentMsgId
   deriving (Show, Exception)

src/Simplex/Chat/Types.hs

@@ -1,19 +1,27 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
 
 module Simplex.Chat.Types where
 
 import Data.Aeson (FromJSON, ToJSON)
 import qualified Data.Aeson as J
+import qualified Data.ByteString.Base64 as B64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Int (Int64)
 import Data.Text (Text)
+import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (UTCTime)
+import Data.Type.Equality
 import Data.Typeable (Typeable)
 import Database.SQLite.Simple (ResultError (..), SQLData (..))
 import Database.SQLite.Simple.FromField (FieldParser, FromField (..), returnError)
@@ -21,8 +29,9 @@ import Database.SQLite.Simple.Internal (Field (..))
 import Database.SQLite.Simple.Ok (Ok (Ok))
 import Database.SQLite.Simple.ToField (ToField (..))
 import GHC.Generics
-import Simplex.Messaging.Agent.Protocol (ConnId, ConnectionMode (..), ConnectionRequest, InvitationId)
+import Simplex.Messaging.Agent.Protocol (AgentMsgId, ConnId, ConnectionMode (..), ConnectionRequest, InvitationId, MsgMeta (..), serializeMsgIntegrity)
 import Simplex.Messaging.Agent.Store.SQLite (fromTextField_)
+import Simplex.Messaging.Protocol (MsgBody)
 
 class IsContact a where
   contactId' :: a -> Int64
@@ -58,6 +67,9 @@ data Contact = Contact
   }
   deriving (Eq, Show)
 
+contactConn :: Contact -> Connection
+contactConn = activeConn
+
 contactConnId :: Contact -> ConnId
 contactConnId Contact {activeConn = Connection {agentConnId}} = agentConnId
 
@@ -153,6 +165,9 @@ data GroupMember = GroupMember
   }
   deriving (Eq, Show)
 
+memberConn :: GroupMember -> Maybe Connection
+memberConn = activeConn
+
 memberConnId :: GroupMember -> Maybe ConnId
 memberConnId GroupMember {activeConn} = case activeConn of
   Just Connection {agentConnId} -> Just agentConnId
@@ -526,3 +541,125 @@ data Onboarding = Onboarding
     filesSentCount :: Int,
     addressCount :: Int
   }
+
+data NewMessage = NewMessage
+  { direction :: MsgDirection,
+    chatMsgEventType :: Text,
+    msgBody :: MsgBody
+  }
+
+type MessageId = Int64
+
+data MsgDirection = MDRcv | MDSnd
+
+data SMsgDirection (d :: MsgDirection) where
+  SMDRcv :: SMsgDirection 'MDRcv
+  SMDSnd :: SMsgDirection 'MDSnd
+
+instance TestEquality SMsgDirection where
+  testEquality SMDRcv SMDRcv = Just Refl
+  testEquality SMDSnd SMDSnd = Just Refl
+  testEquality _ _ = Nothing
+
+class MsgDirectionI (d :: MsgDirection) where
+  msgDirection :: SMsgDirection d
+
+instance MsgDirectionI 'MDRcv where msgDirection = SMDRcv
+
+instance MsgDirectionI 'MDSnd where msgDirection = SMDSnd
+
+instance ToField MsgDirection where toField = toField . msgDirectionInt
+
+msgDirectionInt :: MsgDirection -> Int
+msgDirectionInt = \case
+  MDRcv -> 0
+  MDSnd -> 1
+
+msgDirectionIntP :: Int -> Maybe MsgDirection
+msgDirectionIntP = \case
+  0 -> Just MDRcv
+  1 -> Just MDSnd
+  _ -> Nothing
+
+data SndMsgDelivery = SndMsgDelivery
+  { connId :: Int64,
+    agentMsgId :: AgentMsgId
+  }
+
+data RcvMsgDelivery = RcvMsgDelivery
+  { connId :: Int64,
+    agentMsgId :: AgentMsgId,
+    agentMsgMeta :: MsgMeta
+  }
+
+data MsgMetaJ = MsgMetaJ
+  { integrity :: Text,
+    rcvId :: Int64,
+    rcvTs :: UTCTime,
+    serverId :: Text,
+    serverTs :: UTCTime,
+    sndId :: Int64
+  }
+  deriving (Generic, Eq, Show)
+
+instance ToJSON MsgMetaJ where toEncoding = J.genericToEncoding J.defaultOptions
+
+instance FromJSON MsgMetaJ
+
+msgMetaToJson :: MsgMeta -> MsgMetaJ
+msgMetaToJson MsgMeta {integrity, recipient = (rcvId, rcvTs), broker = (serverId, serverTs), sender = (sndId, _)} =
+  MsgMetaJ
+    { integrity = (decodeLatin1 . serializeMsgIntegrity) integrity,
+      rcvId,
+      rcvTs,
+      serverId = (decodeLatin1 . B64.encode) serverId,
+      serverTs,
+      sndId
+    }
+
+msgMetaJson :: MsgMeta -> Text
+msgMetaJson = decodeLatin1 . LB.toStrict . J.encode . msgMetaToJson
+
+data MsgDeliveryStatus (d :: MsgDirection) where
+  MDSRcvAgent :: MsgDeliveryStatus 'MDRcv
+  MDSRcvAcknowledged :: MsgDeliveryStatus 'MDRcv
+  MDSSndPending :: MsgDeliveryStatus 'MDSnd
+  MDSSndAgent :: MsgDeliveryStatus 'MDSnd
+  MDSSndSent :: MsgDeliveryStatus 'MDSnd
+  MDSSndReceived :: MsgDeliveryStatus 'MDSnd
+  MDSSndRead :: MsgDeliveryStatus 'MDSnd
+
+data AMsgDeliveryStatus = forall d. AMDS (SMsgDirection d) (MsgDeliveryStatus d)
+
+instance (Typeable d, MsgDirectionI d) => FromField (MsgDeliveryStatus d) where
+  fromField = fromTextField_ msgDeliveryStatusT'
+
+instance ToField (MsgDeliveryStatus d) where toField = toField . serializeMsgDeliveryStatus
+
+serializeMsgDeliveryStatus :: MsgDeliveryStatus d -> Text
+serializeMsgDeliveryStatus = \case
+  MDSRcvAgent -> "rcv_agent"
+  MDSRcvAcknowledged -> "rcv_acknowledged"
+  MDSSndPending -> "snd_pending"
+  MDSSndAgent -> "snd_agent"
+  MDSSndSent -> "snd_sent"
+  MDSSndReceived -> "snd_received"
+  MDSSndRead -> "snd_read"
+
+msgDeliveryStatusT :: Text -> Maybe AMsgDeliveryStatus
+msgDeliveryStatusT = \case
+  "rcv_agent" -> Just $ AMDS SMDRcv MDSRcvAgent
+  "rcv_acknowledged" -> Just $ AMDS SMDRcv MDSRcvAcknowledged
+  "snd_pending" -> Just $ AMDS SMDSnd MDSSndPending
+  "snd_agent" -> Just $ AMDS SMDSnd MDSSndAgent
+  "snd_sent" -> Just $ AMDS SMDSnd MDSSndSent
+  "snd_received" -> Just $ AMDS SMDSnd MDSSndReceived
+  "snd_read" -> Just $ AMDS SMDSnd MDSSndRead
+  _ -> Nothing
+
+msgDeliveryStatusT' :: forall d. MsgDirectionI d => Text -> Maybe (MsgDeliveryStatus d)
+msgDeliveryStatusT' s =
+  msgDeliveryStatusT s >>= \(AMDS d st) ->
+    case testEquality d (msgDirection @d) of
+      Just Refl -> Just st
+      _ -> Nothing