agent: delivery receipts (#752)

* rfc: delivery receipts * update doc * update rfc * implementation plan, types, schema * migration, update types * update types * rename migration * export MsgReceiptStatus, JSON encoding * update rfc, schema * correction Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> * skeleton of the implementation * more implementation (some tests fail) * more code, 1 test fails * fix encoding * refactor * refactor * test, fix * only send receipts in v3+, test * flip condition Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> * flip condition Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> * agent version 4 required to send receipts * fix test --------- Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com>
2026-09-17 05:55:04 +00:00 · 2023-07-13 22:33:48 +01:00
parent 745a144e0c
commit 58cb2855d2
14 changed files with 649 additions and 167 deletions
@@ -1,14 +0,0 @@
-module Simplex.FileTransfer where
-
-- TODO
-- Protocol
-- Store (in memory storage)
-- StoreLog (append only log)
-- FileDescription
-- Server
-- Client
-- Server/Main (server CLI)
-- Client/Main (client CLI)
--
-- Transport for HTTP2 ?
-- streaming Crypto
@@ -117,7 +117,7 @@ import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
-import Data.Maybe (fromMaybe, isJust, isNothing)
+import Data.Maybe (fromMaybe, isJust, isNothing, catMaybes)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock
@@ -152,7 +152,6 @@ import Simplex.Messaging.Util
 import Simplex.Messaging.Version
 import UnliftIO.Async (async, race_)
 import UnliftIO.Concurrent (forkFinally, forkIO, threadDelay)
-import qualified UnliftIO.Exception as E
 import UnliftIO.STM

 -- import GHC.Conc (unsafeIOToSTM)
@@ -203,8 +202,8 @@ acceptContactAsync :: AgentErrorMonad m => AgentClient -> ACorrId -> Bool -> Con
 acceptContactAsync c corrId enableNtfs = withAgentEnv c .: acceptContactAsync' c corrId enableNtfs

 -- | Acknowledge message (ACK command) asynchronously, no synchronous response
-ackMessageAsync :: forall m. AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> m ()
-ackMessageAsync c = withAgentEnv c .:. ackMessageAsync' c
+ackMessageAsync :: forall m. AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> m ()
+ackMessageAsync c = withAgentEnv c .:: ackMessageAsync' c

 -- | Switch connection to the new receive queue
 switchConnectionAsync :: AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> m ConnectionStats
@@ -264,8 +263,8 @@ resubscribeConnections c = withAgentEnv c . resubscribeConnections' c
 sendMessage :: AgentErrorMonad m => AgentClient -> ConnId -> MsgFlags -> MsgBody -> m AgentMsgId
 sendMessage c = withAgentEnv c .:. sendMessage' c

-ackMessage :: AgentErrorMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
-ackMessage c = withAgentEnv c .: ackMessage' c
+ackMessage :: AgentErrorMonad m => AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> m ()
+ackMessage c = withAgentEnv c .:. ackMessage' c

 -- | Switch connection to the new receive queue
 switchConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ConnectionStats
@@ -432,7 +431,7 @@ processCommand c (connId, APC e cmd) =
    RJCT invId -> rejectContact' c connId invId $> (connId, OK)
    SUB -> subscribeConnection' c connId $> (connId, OK)
    SEND msgFlags msgBody -> (connId,) . MID <$> sendMessage' c connId msgFlags msgBody
-    ACK msgId -> ackMessage' c connId msgId $> (connId, OK)
+    ACK msgId rcptInfo_ -> ackMessage' c connId msgId rcptInfo_ $> (connId, OK)
    SWCH -> switchConnection' c connId $> (connId, OK)
    OFF -> suspendConnection' c connId $> (connId, OK)
    DEL -> deleteConnection' c connId $> (connId, OK)
@@ -507,8 +506,8 @@ acceptContactAsync' c corrId enableNtfs invId ownConnInfo = do
        throwError err
    _ -> throwError $ CMD PROHIBITED

-ackMessageAsync' :: forall m. AgentMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> m ()
-ackMessageAsync' c corrId connId msgId = do
+ackMessageAsync' :: forall m. AgentMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> m ()
+ackMessageAsync' c corrId connId msgId rcptInfo_ = do
  SomeConn cType _ <- withStore c (`getConn` connId)
  case cType of
    SCDuplex -> enqueueAck
@@ -519,8 +518,11 @@ ackMessageAsync' c corrId connId msgId = do
  where
    enqueueAck :: m ()
    enqueueAck = do
-      (RcvQueue {server}, _) <- withStore c $ \db -> setMsgUserAck db connId $ InternalId msgId
-      enqueueCommand c corrId connId (Just server) . AClientCommand $ APC SAEConn $ ACK msgId
+      let mId = InternalId msgId
+      RcvMsg {msgType} <- withStore c $ \db -> getRcvMsg db connId mId
+      when (isJust rcptInfo_ && msgType /= AM_A_MSG_) $ throwError $ CMD PROHIBITED
+      (RcvQueue {server}, _) <- withStore c $ \db -> setMsgUserAck db connId mId
+      enqueueCommand c corrId connId (Just server) . AClientCommand $ APC SAEConn $ ACK msgId rcptInfo_

 deleteConnectionAsync' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
 deleteConnectionAsync' c connId = deleteConnectionsAsync' c [connId]
@@ -891,7 +893,7 @@ runCommandProcessing c@AgentClient {subQ} server_ = do
            void $ joinConnSrv c userId connId True enableNtfs cReq connInfo srv
            notify OK
        LET confId ownCInfo -> withServer' . tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK
-        ACK msgId -> withServer' . tryCommand $ ackMessage' c connId msgId >> notify OK
+        ACK msgId rcptInfo_ -> withServer' . tryCommand $ ackMessage' c connId msgId rcptInfo_ >> notify OK
        SWCH ->
          noServer . tryCommand . withConnLock c connId "switchConnection" $
            withStore c (`getConn` connId) >>= \case
@@ -1112,6 +1114,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                        _ -> connError msgId NOT_ACCEPTED
                  AM_REPLY_ -> notifyDel msgId err
                  AM_A_MSG_ -> notifyDel msgId err
+                  AM_A_RCVD_ -> notifyDel msgId err
                  AM_QCONT_ -> notifyDel msgId err
                  AM_QADD_ -> qError msgId "QADD: AUTH"
                  AM_QKEY_ -> qError msgId "QKEY: AUTH"
@@ -1166,6 +1169,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                      qInfo <- createReplyQueue c cData sq srv
                      void . enqueueMessage c cData sq SMP.noMsgFlags $ REPLY [qInfo]
                AM_A_MSG_ -> notify $ SENT mId
+                AM_A_RCVD_ -> pure ()
                AM_QCONT_ -> pure ()
                AM_QADD_ -> pure ()
                AM_QKEY_ -> do
@@ -1200,10 +1204,12 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                        _ -> internalErr msgId "sent QTEST: queue not in connection or not replacing another queue"
                    _ -> internalErr msgId "QTEST sent not in duplex connection"
                AM_EREADY_ -> pure ()
-              delMsg msgId
+              delMsgKeep (msgType == AM_A_MSG_) msgId
  where
    delMsg :: InternalId -> m ()
-    delMsg msgId = withStore' c $ \db -> deleteSndMsgDelivery db connId sq msgId
+    delMsg = delMsgKeep False
+    delMsgKeep :: Bool -> InternalId -> m ()
+    delMsgKeep keepForReceipt msgId = withStore' c $ \db -> deleteSndMsgDelivery db connId sq msgId keepForReceipt
    notify :: forall e. AEntityI e => ACommand 'Agent e -> m ()
    notify cmd = atomically $ writeTBQueue subQ ("", connId, APC (sAEntity @e) cmd)
    notifyDel :: AEntityI e => InternalId -> ACommand 'Agent e -> m ()
@@ -1220,22 +1226,38 @@ retrySndOp c loop = do
  atomically $ beginAgentOperation c AOSndNetwork
  loop

-ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
-ackMessage' c connId msgId = withConnLock c connId "ackMessage" $ do
+ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> m ()
+ackMessage' c connId msgId rcptInfo_ = withConnLock c connId "ackMessage" $ do
  SomeConn _ conn <- withStore c (`getConn` connId)
  case conn of
-    DuplexConnection {} -> ack
-    RcvConnection {} -> ack
+    DuplexConnection {} -> ack >> sendRcpt conn >> del
+    RcvConnection {} -> ack >> del
    SndConnection {} -> throwError $ CONN SIMPLEX
    ContactConnection {} -> throwError $ CMD PROHIBITED
    NewConnection _ -> throwError $ CMD PROHIBITED
  where
    ack :: m ()
    ack = do
-      let mId = InternalId msgId
-      (rq, srvMsgId) <- withStore c $ \db -> setMsgUserAck db connId mId
+      -- the stored message was delivered via a specific queue, the rest failed to decrypt and were already acknowledged
+      (rq, srvMsgId) <- withStore c $ \db -> setMsgUserAck db connId $ InternalId msgId
      ackQueueMessage c rq srvMsgId
-      withStore' c $ \db -> deleteMsg db connId mId
+    del :: m ()
+    del = withStore' c $ \db -> deleteMsg db connId $ InternalId msgId
+    sendRcpt :: Connection 'CDuplex -> m ()
+    sendRcpt (DuplexConnection cData _ sqs) = do
+      msg@RcvMsg {msgType, msgReceipt} <- withStore c $ \db -> getRcvMsg db connId $ InternalId msgId
+      case rcptInfo_ of
+        Just rcptInfo -> do
+          unless (msgType == AM_A_MSG_) $ throwError (CMD PROHIBITED)
+          when (messageRcptsSupported cData) $ do
+            let RcvMsg {msgMeta = MsgMeta {sndMsgId}, internalHash} = msg
+                rcpt = A_RCVD [AMessageReceipt {agentMsgId = sndMsgId, msgHash = internalHash, rcptInfo}]
+            void $ enqueueMessages c cData sqs SMP.MsgFlags {notification = False} rcpt
+        Nothing -> case (msgType, msgReceipt) of
+          -- only remove sent message if receipt hash was Ok, both to debug and for future redundancy
+          (AM_A_RCVD_, Just MsgReceipt {agentMsgId = sndMsgId, msgRcptStatus = MROk}) ->
+            withStore' c $ \db -> deleteDeliveredSndMsg db connId $ InternalId sndMsgId
+          _ -> pure ()

 switchConnection' :: AgentMonad m => AgentClient -> ConnId -> m ConnectionStats
 switchConnection' c connId =
@@ -1725,30 +1747,30 @@ cleanupManager c@AgentClient {subQ} = do
  delay <- asks (initialCleanupDelay . config)
  liftIO $ threadDelay' delay
  int <- asks (cleanupInterval . config)
+  ttl <- asks $ storedMsgDataTTL . config
  forever $ do
-    void . runExceptT $ do
-      deleteConns `catchAgentError` (notify "" . ERR)
-      deleteRcvMsgHashes `catchAgentError` (notify "" . ERR)
-      deleteProcessedRatchetKeyHashes `catchAgentError` (notify "" . ERR)
-      deleteRcvFilesExpired `catchAgentError` (notify "" . RFERR)
-      deleteRcvFilesDeleted `catchAgentError` (notify "" . RFERR)
-      deleteRcvFilesTmpPaths `catchAgentError` (notify "" . RFERR)
-      deleteSndFilesExpired `catchAgentError` (notify "" . SFERR)
-      deleteSndFilesDeleted `catchAgentError` (notify "" . SFERR)
-      deleteSndFilesPrefixPaths `catchAgentError` (notify "" . SFERR)
-      deleteExpiredReplicasForDeletion `catchAgentError` (notify "" . SFERR)
+    run ERR deleteConns
+    run ERR $ withStore' c (`deleteRcvMsgHashesExpired` ttl)
+    run ERR $ withStore' c (`deleteSndMsgsExpired` ttl)
+    run ERR $ withStore' c (`deleteRatchetKeyHashesExpired` ttl)
+    run RFERR deleteRcvFilesExpired
+    run RFERR deleteRcvFilesDeleted
+    run RFERR deleteRcvFilesTmpPaths
+    run SFERR deleteSndFilesExpired
+    run SFERR deleteSndFilesDeleted
+    run SFERR deleteSndFilesPrefixPaths
+    run SFERR deleteExpiredReplicasForDeletion
    liftIO $ threadDelay' int
  where
+    run :: forall e. AEntityI e => (AgentErrorType -> ACommand 'Agent e) -> ExceptT AgentErrorType m () -> m ()
+    run err a = do
+      void . runExceptT $ a `catchAgentError` (notify "" . err)
+      step <- asks $ cleanupStepInterval . config
+      liftIO $ threadDelay step
    deleteConns =
      withLock (deleteLock c) "cleanupManager" $ do
        void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c
        withStore' c deleteUsersWithoutConns >>= mapM_ (notify "" . DEL_USER)
-    deleteRcvMsgHashes = do
-      rcvMsgHashesTTL <- asks $ rcvMsgHashesTTL . config
-      withStore' c (`deleteRcvMsgHashesExpired` rcvMsgHashesTTL)
-    deleteProcessedRatchetKeyHashes = do
-      rkHashesTTL <- asks $ processedRatchetKeyHashesTTL . config
-      withStore' c (`deleteProcessedRatchetKeyHashesExpired` rkHashesTTL)
    deleteRcvFilesExpired = do
      rcvFilesTTL <- asks $ rcvFilesTTL . config
      rcvExpired <- withStore' c (`getRcvFilesExpired` rcvFilesTTL)
@@ -1855,6 +1877,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                              A_MSG body -> do
                                logServer "<--" c srv rId "MSG <MSG>"
                                notify $ MSG msgMeta msgFlags body
+                              A_RCVD rcpts -> qDuplex conn'' "RCVD" $ messagesRcvd rcpts msgMeta
                              QCONT addr -> qDuplexAckDel conn'' "QCONT" $ continueSending addr
                              QADD qs -> qDuplexAckDel conn'' "QADD" $ qAddMsg qs
                              QKEY qs -> qDuplexAckDel conn'' "QKEY" $ qKeyMsg qs
@@ -2078,6 +2101,28 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                  void $ tryPutTMVar qLock ()
              Nothing -> qError "QCONT: queue address not found"

+          messagesRcvd :: NonEmpty AMessageReceipt -> MsgMeta -> Connection 'CDuplex -> m ()
+          messagesRcvd rcpts msgMeta@MsgMeta {broker = (srvMsgId, _)} _ = do
+            logServer "<--" c srv rId "MSG <RCPT>"
+            rs <- forM rcpts $ \rcpt -> clientReceipt rcpt `catchAgentError` \e -> notify (ERR e) $> Nothing
+            case L.nonEmpty . catMaybes $ L.toList rs of
+              Just rs' -> notify $ RCVD msgMeta rs' -- client must ACK once processed
+              Nothing -> enqueueCmd $ ICAck rId srvMsgId
+            where
+              clientReceipt :: AMessageReceipt -> m (Maybe MsgReceipt)
+              clientReceipt AMessageReceipt {agentMsgId, msgHash} = do
+                let sndMsgId = InternalSndId agentMsgId
+                SndMsg {internalId = InternalId msgId, msgType, internalHash, msgReceipt} <- withStore c $ \db -> getSndMsgViaRcpt db connId sndMsgId
+                if msgType /= AM_A_MSG_
+                  then notify (ERR $ AGENT A_PROHIBITED) $> Nothing -- unexpected message type for receipt
+                  else case msgReceipt of
+                    Just MsgReceipt {msgRcptStatus = MROk} -> pure Nothing -- already notified with MROk status
+                    _ -> do
+                      let msgRcptStatus = if msgHash == internalHash then MROk else MRBadMsgHash
+                          rcpt = MsgReceipt {agentMsgId = msgId, msgRcptStatus}
+                      withStore' c $ \db -> updateSndMsgRcpt db connId sndMsgId rcpt
+                      pure $ Just rcpt
+
          -- processed by queue sender
          qAddMsg :: NonEmpty (SMPQueueUri, Maybe SndQAddr) -> Connection 'CDuplex -> m ()
          qAddMsg ((_, Nothing) :| _) _ = qError "adding queue without switching is not supported"
@@ -2195,7 +2240,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
              rkHash k1 k2 = C.sha256Hash $ C.pubKeyBytes k1 <> C.pubKeyBytes k2
              ratchetExists :: m Bool
              ratchetExists = withStore' c $ \db -> do
-                exists <- checkProcessedRatchetKeyHashExists db connId rkHashRcv
+                exists <- checkRatchetKeyHashExists db connId rkHashRcv
                unless exists $ addProcessedRatchetKeyHash db connId rkHashRcv
                pure exists
              getSendRatchetKeys :: m (C.PrivateKeyX448, C.PrivateKeyX448, C.PublicKeyX448, C.PublicKeyX448)
@@ -87,8 +87,8 @@ data AgentConfig = AgentConfig
    helloTimeout :: NominalDiffTime,
    initialCleanupDelay :: Int64,
    cleanupInterval :: Int64,
-    rcvMsgHashesTTL :: NominalDiffTime,
-    processedRatchetKeyHashesTTL :: NominalDiffTime,
+    cleanupStepInterval :: Int,
+    storedMsgDataTTL :: NominalDiffTime,
    rcvFilesTTL :: NominalDiffTime,
    sndFilesTTL :: NominalDiffTime,
    xftpNotifyErrsOnRetry :: Bool,
@@ -152,8 +152,8 @@ defaultAgentConfig =
      helloTimeout = 2 * nominalDay,
      initialCleanupDelay = 30 * 1000000, -- 30 seconds
      cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
-      rcvMsgHashesTTL = 30 * nominalDay,
-      processedRatchetKeyHashesTTL = 30 * nominalDay,
+      cleanupStepInterval = 200000, -- 200ms
+      storedMsgDataTTL = 21 * nominalDay,
      rcvFilesTTL = 2 * nominalDay,
      sndFilesTTL = nominalDay,
      xftpNotifyErrsOnRetry = True,
@@ -69,6 +69,10 @@ module Simplex.Messaging.Agent.Protocol
    AgentMessageType (..),
    APrivHeader (..),
    AMessage (..),
+    AMessageReceipt (..),
+    MsgReceipt (..),
+    MsgReceiptInfo,
+    MsgReceiptStatus (..),
    SndQAddr,
    SMPServer,
    pattern SMPServer,
@@ -211,7 +215,7 @@ import Text.Read
 import UnliftIO.Exception (Exception)

 currentSMPAgentVersion :: Version
-currentSMPAgentVersion = 3
+currentSMPAgentVersion = 4

 supportedSMPAgentVRange :: VersionRange
 supportedSMPAgentVRange = mkVersionRange 1 currentSMPAgentVersion
@@ -314,7 +318,7 @@ data ACommand (p :: AParty) (e :: AEntity) where
  JOIN :: Bool -> AConnectionRequestUri -> ConnInfo -> ACommand Client AEConn -- response OK
  CONF :: ConfirmationId -> [SMPServer] -> ConnInfo -> ACommand Agent AEConn -- ConnInfo is from sender, [SMPServer] will be empty only in v1 handshake
  LET :: ConfirmationId -> ConnInfo -> ACommand Client AEConn -- ConnInfo is from client
-  REQ :: InvitationId -> L.NonEmpty SMPServer -> ConnInfo -> ACommand Agent AEConn -- ConnInfo is from sender
+  REQ :: InvitationId -> NonEmpty SMPServer -> ConnInfo -> ACommand Agent AEConn -- ConnInfo is from sender
  ACPT :: InvitationId -> ConnInfo -> ACommand Client AEConn -- ConnInfo is from client
  RJCT :: InvitationId -> ACommand Client AEConn
  INFO :: ConnInfo -> ACommand Agent AEConn
@@ -332,7 +336,8 @@ data ACommand (p :: AParty) (e :: AEntity) where
  SENT :: AgentMsgId -> ACommand Agent AEConn
  MERR :: AgentMsgId -> AgentErrorType -> ACommand Agent AEConn
  MSG :: MsgMeta -> MsgFlags -> MsgBody -> ACommand Agent AEConn
-  ACK :: AgentMsgId -> ACommand Client AEConn
+  ACK :: AgentMsgId -> Maybe MsgReceiptInfo -> ACommand Client AEConn
+  RCVD :: MsgMeta -> NonEmpty MsgReceipt -> ACommand Agent AEConn 
  SWCH :: ACommand Client AEConn
  OFF :: ACommand Client AEConn
  DEL :: ACommand Client AEConn
@@ -392,6 +397,7 @@ data ACommandTag (p :: AParty) (e :: AEntity) where
  MERR_ :: ACommandTag Agent AEConn
  MSG_ :: ACommandTag Agent AEConn
  ACK_ :: ACommandTag Client AEConn
+  RCVD_ :: ACommandTag Agent AEConn
  SWCH_ :: ACommandTag Client AEConn
  OFF_ :: ACommandTag Client AEConn
  DEL_ :: ACommandTag Client AEConn
@@ -443,7 +449,8 @@ aCommandTag = \case
  SENT _ -> SENT_
  MERR {} -> MERR_
  MSG {} -> MSG_
-  ACK _ -> ACK_
+  ACK {} -> ACK_
+  RCVD {} -> RCVD_
  SWCH -> SWCH_
  OFF -> OFF_
  DEL -> DEL_
@@ -743,6 +750,25 @@ data MsgMeta = MsgMeta
  }
  deriving (Eq, Show)

+instance StrEncoding MsgMeta where
+  strEncode MsgMeta {integrity, recipient = (rmId, rTs), broker = (bmId, bTs), sndMsgId} =
+    B.unwords
+      [ strEncode integrity,
+        "R=" <> bshow rmId <> "," <> showTs rTs,
+        "B=" <> encode bmId <> "," <> showTs bTs,
+        "S=" <> bshow sndMsgId
+      ]
+    where
+      showTs = B.pack . formatISO8601Millis
+  strP = do
+    integrity <- strP
+    recipient <- " R=" *> partyMeta A.decimal
+    broker <- " B=" *> partyMeta base64P
+    sndMsgId <- " S=" *> A.decimal
+    pure MsgMeta {integrity, recipient, broker, sndMsgId}
+    where
+      partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P
+
 data SMPConfirmation = SMPConfirmation
  { -- | sender's public key to use for authentication of sender's commands at the recepient's server
    senderKey :: SndPublicVerifyKey,
@@ -815,7 +841,7 @@ data AgentMessage
  = AgentConnInfo ConnInfo
  | -- AgentConnInfoReply is only used in duplexHandshake mode (v2), allowing to include reply queue(s) in the initial confirmation.
    -- It makes REPLY message unnecessary.
-    AgentConnInfoReply (L.NonEmpty SMPQueueInfo) ConnInfo
+    AgentConnInfoReply (NonEmpty SMPQueueInfo) ConnInfo
  | AgentRatchetInfo ByteString
  | AgentMessage APrivHeader AMessage
  deriving (Show)
@@ -841,6 +867,7 @@ data AgentMessageType
  | AM_HELLO_
  | AM_REPLY_
  | AM_A_MSG_
+  | AM_A_RCVD_
  | AM_QCONT_
  | AM_QADD_
  | AM_QKEY_
@@ -857,6 +884,7 @@ instance Encoding AgentMessageType where
    AM_HELLO_ -> "H"
    AM_REPLY_ -> "R"
    AM_A_MSG_ -> "M"
+    AM_A_RCVD_ -> "V"
    AM_QCONT_ -> "QC"
    AM_QADD_ -> "QA"
    AM_QKEY_ -> "QK"
@@ -871,6 +899,7 @@ instance Encoding AgentMessageType where
      'H' -> pure AM_HELLO_
      'R' -> pure AM_REPLY_
      'M' -> pure AM_A_MSG_
+      'V' -> pure AM_A_RCVD_
      'Q' ->
        A.anyChar >>= \case
          'C' -> pure AM_QCONT_
@@ -896,6 +925,7 @@ agentMessageType = \case
    -- REPLY is only used in v1
    REPLY _ -> AM_REPLY_
    A_MSG _ -> AM_A_MSG_
+    A_RCVD {} -> AM_A_RCVD_
    QCONT _ -> AM_QCONT_
    QADD _ -> AM_QADD_
    QKEY _ -> AM_QKEY_
@@ -920,6 +950,7 @@ data AMsgType
  = HELLO_
  | REPLY_
  | A_MSG_
+  | A_RCVD_
  | QCONT_
  | QADD_
  | QKEY_
@@ -933,6 +964,7 @@ instance Encoding AMsgType where
    HELLO_ -> "H"
    REPLY_ -> "R"
    A_MSG_ -> "M"
+    A_RCVD_ -> "V"
    QCONT_ -> "QC"
    QADD_ -> "QA"
    QKEY_ -> "QK"
@@ -944,6 +976,7 @@ instance Encoding AMsgType where
      'H' -> pure HELLO_
      'R' -> pure REPLY_
      'M' -> pure A_MSG_
+      'V' -> pure A_RCVD_
      'Q' ->
        A.anyChar >>= \case
          'C' -> pure QCONT_
@@ -962,23 +995,60 @@ data AMessage
  = -- | the first message in the queue to validate it is secured
    HELLO
  | -- | reply queues information
-    REPLY (L.NonEmpty SMPQueueInfo)
+    REPLY (NonEmpty SMPQueueInfo)
  | -- | agent envelope for the client message
    A_MSG MsgBody
+  | -- | agent envelope for delivery receipt 
+    A_RCVD (NonEmpty AMessageReceipt)
  | -- | the message instructing the client to continue sending messages (after ERR QUOTA)
    QCONT SndQAddr
  | -- add queue to connection (sent by recipient), with optional address of the replaced queue
-    QADD (L.NonEmpty (SMPQueueUri, Maybe SndQAddr))
+    QADD (NonEmpty (SMPQueueUri, Maybe SndQAddr))
  | -- key to secure the added queues and agree e2e encryption key (sent by sender)
-    QKEY (L.NonEmpty (SMPQueueInfo, SndPublicVerifyKey))
+    QKEY (NonEmpty (SMPQueueInfo, SndPublicVerifyKey))
  | -- inform that the queues are ready to use (sent by recipient)
-    QUSE (L.NonEmpty (SndQAddr, Bool))
+    QUSE (NonEmpty (SndQAddr, Bool))
  | -- sent by the sender to test new queues and to complete switching
-    QTEST (L.NonEmpty SndQAddr)
+    QTEST (NonEmpty SndQAddr)
  | -- ratchet re-synchronization is complete, with last decrypted sender message id (recipient's `last_external_snd_msg_id`)
-    EREADY Int64
+    EREADY AgentMsgId
  deriving (Show)

+-- | this type is used to send as part of the protocol between different clients
+-- TODO possibly, rename fields and types referring to external and internal IDs to make them different
+data AMessageReceipt = AMessageReceipt
+  { agentMsgId :: AgentMsgId, -- this is an external snd message ID referenced by the message recipient
+    msgHash :: MsgHash,
+    rcptInfo :: MsgReceiptInfo
+  }
+  deriving (Show)
+
+-- | this type is used as part of agent protocol to communicate with the user application
+data MsgReceipt = MsgReceipt
+  { agentMsgId :: AgentMsgId, -- this is an internal agent message ID of received message
+    msgRcptStatus :: MsgReceiptStatus
+  }
+  deriving (Eq, Show)
+
+data MsgReceiptStatus = MROk | MRBadMsgHash
+  deriving (Eq, Show)
+
+instance StrEncoding MsgReceiptStatus where
+  strEncode = \case
+    MROk -> "ok"
+    MRBadMsgHash -> "badMsgHash"
+  strP =
+    A.takeWhile1 (/= ' ') >>= \ case
+      "ok" -> pure MROk
+      "badMsgHash" -> pure MRBadMsgHash
+      _ -> fail "bad MsgReceiptStatus"
+
+instance ToJSON MsgReceiptStatus where
+  toJSON = strToJSON
+  toEncoding = strToJEncoding
+
+type MsgReceiptInfo = ByteString
+
 type SndQAddr = (SMPServer, SMP.SenderId)

 instance Encoding AMessage where
@@ -986,6 +1056,7 @@ instance Encoding AMessage where
    HELLO -> smpEncode HELLO_
    REPLY smpQueues -> smpEncode (REPLY_, smpQueues)
    A_MSG body -> smpEncode (A_MSG_, Tail body)
+    A_RCVD mrs -> smpEncode (A_RCVD_, mrs)
    QCONT addr -> smpEncode (QCONT_, addr)
    QADD qs -> smpEncode (QADD_, qs)
    QKEY qs -> smpEncode (QKEY_, qs)
@@ -998,6 +1069,7 @@ instance Encoding AMessage where
        HELLO_ -> pure HELLO
        REPLY_ -> REPLY <$> smpP
        A_MSG_ -> A_MSG . unTail <$> smpP
+        A_RCVD_ -> A_RCVD <$> smpP
        QCONT_ -> QCONT <$> smpP
        QADD_ -> QADD <$> smpP
        QKEY_ -> QKEY <$> smpP
@@ -1005,6 +1077,21 @@ instance Encoding AMessage where
        QTEST_ -> QTEST <$> smpP
        EREADY_ -> EREADY <$> smpP

+instance Encoding AMessageReceipt where
+  smpEncode AMessageReceipt {agentMsgId, msgHash, rcptInfo} =
+    smpEncode (agentMsgId, msgHash, Large rcptInfo)
+  smpP = do
+    (agentMsgId, msgHash, Large rcptInfo) <- smpP
+    pure AMessageReceipt {agentMsgId, msgHash, rcptInfo}
+
+instance StrEncoding MsgReceipt where
+  strEncode MsgReceipt {agentMsgId, msgRcptStatus} =
+    strEncode agentMsgId <> ":" <> strEncode msgRcptStatus
+  strP = do
+    agentMsgId <- strP <* A.char ':'
+    msgRcptStatus <- strP
+    pure MsgReceipt {agentMsgId, msgRcptStatus}
+
 instance forall m. ConnectionModeI m => StrEncoding (ConnectionRequestUri m) where
  strEncode = \case
    CRInvitationUri crData e2eParams -> crEncode "invitation" crData (Just e2eParams)
@@ -1234,7 +1321,7 @@ deriving instance Show AConnectionRequestUri
 data ConnReqUriData = ConnReqUriData
  { crScheme :: ConnReqScheme,
    crAgentVRange :: VersionRange,
-    crSmpQueues :: L.NonEmpty SMPQueueUri,
+    crSmpQueues :: NonEmpty SMPQueueUri,
    crClientData :: Maybe CRClientData
  }
  deriving (Eq, Show)
@@ -1296,7 +1383,7 @@ instance StrEncoding MsgIntegrity where
  strP = "OK" $> MsgOk <|> "ERR " *> (MsgError <$> strP)
  strEncode = \case
    MsgOk -> "OK"
-    MsgError e -> "ERR" <> strEncode e
+    MsgError e -> "ERR " <> strEncode e

 instance ToJSON MsgIntegrity where
  toJSON = J.genericToJSON $ sumTypeJSON fstToLower
@@ -1316,7 +1403,7 @@ data MsgErrorType
 instance StrEncoding MsgErrorType where
  strP =
    "ID " *> (MsgBadId <$> A.decimal)
-      <|> "IDS " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)
+      <|> "NO_ID " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)
      <|> "HASH" $> MsgBadHash
      <|> "DUPLICATE" $> MsgDuplicate
  strEncode = \case
@@ -1557,6 +1644,7 @@ instance StrEncoding ACmdTag where
      "MERR" -> ct MERR_
      "MSG" -> ct MSG_
      "ACK" -> t ACK_
+      "RCVD" -> ct RCVD_
      "SWCH" -> t SWCH_
      "OFF" -> t OFF_
      "DEL" -> t DEL_
@@ -1611,6 +1699,7 @@ instance (APartyI p, AEntityI e) => StrEncoding (ACommandTag p e) where
    MERR_ -> "MERR"
    MSG_ -> "MSG"
    ACK_ -> "ACK"
+    RCVD_ -> "RCVD"
    SWCH_ -> "SWCH"
    OFF_ -> "OFF"
    DEL_ -> "DEL"
@@ -1654,7 +1743,7 @@ commandP binaryP =
          RJCT_ -> s (RJCT <$> A.takeByteString)
          SUB_ -> pure SUB
          SEND_ -> s (SEND <$> smpP <* A.space <*> binaryP)
-          ACK_ -> s (ACK <$> A.decimal)
+          ACK_ -> s (ACK <$> A.decimal <*> optional (A.space *> binaryP))
          SWCH_ -> pure SWCH
          OFF_ -> pure OFF
          DEL_ -> pure DEL
@@ -1676,7 +1765,8 @@ commandP binaryP =
          MID_ -> s (MID <$> A.decimal)
          SENT_ -> s (SENT <$> A.decimal)
          MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP)
-          MSG_ -> s (MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> binaryP)
+          MSG_ -> s (MSG <$> strP <* A.space <*> smpP <* A.space <*> binaryP)
+          RCVD_ -> s (RCVD <$> strP <* A.space <*> strP)
          DEL_RCVQ_ -> s (DEL_RCVQ <$> strP_ <*> strP_ <*> strP)
          DEL_CONN_ -> pure DEL_CONN
          DEL_USER_ -> s (DEL_USER <$> strP)
@@ -1701,13 +1791,6 @@ commandP binaryP =
       in case ds of
            [] -> Left "no sender file description"
            sd : rds -> SFDONE <$> strDecode (encodeUtf8 sd) <*> mapM (strDecode . encodeUtf8) rds
-    msgMetaP = do
-      integrity <- strP
-      recipient <- " R=" *> partyMeta A.decimal
-      broker <- " B=" *> partyMeta base64P
-      sndMsgId <- " S=" *> A.decimal
-      pure MsgMeta {integrity, recipient, broker, sndMsgId}
-    partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P

 parseCommand :: ByteString -> Either AgentErrorType ACmd
 parseCommand = parse (commandP A.takeByteString) $ CMD SYNTAX
@@ -1736,8 +1819,9 @@ serializeCommand = \case
  MID mId -> s (MID_, Str $ bshow mId)
  SENT mId -> s (SENT_, Str $ bshow mId)
  MERR mId e -> s (MERR_, Str $ bshow mId, e)
-  MSG msgMeta msgFlags msgBody -> B.unwords [s MSG_, serializeMsgMeta msgMeta, smpEncode msgFlags, serializeBinary msgBody]
-  ACK mId -> s (ACK_, Str $ bshow mId)
+  MSG msgMeta msgFlags msgBody -> B.unwords [s MSG_, s msgMeta, smpEncode msgFlags, serializeBinary msgBody]
+  ACK mId rcptInfo_ -> s (ACK_, Str $ bshow mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_
+  RCVD msgMeta rcpts -> s (RCVD_, msgMeta, rcpts)
  SWCH -> s SWCH_
  OFF -> s OFF_
  DEL -> s DEL_
@@ -1759,19 +1843,9 @@ serializeCommand = \case
  where
    s :: StrEncoding a => a -> ByteString
    s = strEncode
-    showTs :: UTCTime -> ByteString
-    showTs = B.pack . formatISO8601Millis
    connections :: [ConnId] -> ByteString
    connections = B.intercalate "," . map strEncode
    sfDone sd rds = B.intercalate fdSeparator $ strEncode sd : map strEncode rds
-    serializeMsgMeta :: MsgMeta -> ByteString
-    serializeMsgMeta MsgMeta {integrity, recipient = (rmId, rTs), broker = (bmId, bTs), sndMsgId} =
-      B.unwords
-        [ strEncode integrity,
-          "R=" <> bshow rmId <> "," <> showTs rTs,
-          "B=" <> encode bmId <> "," <> showTs bTs,
-          "S=" <> bshow sndMsgId
-        ]

 serializeBinary :: ByteString -> ByteString
 serializeBinary body = bshow (B.length body) <> "\n" <> body
@@ -315,6 +315,9 @@ ratchetSyncAllowed cData@ConnData {ratchetSyncState} =
 ratchetSyncSupported' :: ConnData -> Bool
 ratchetSyncSupported' ConnData {connAgentVersion} = connAgentVersion >= 3

+messageRcptsSupported :: ConnData -> Bool
+messageRcptsSupported ConnData {connAgentVersion} = connAgentVersion >= 4
+
 -- this function should be mirrored in the clients
 ratchetSyncSendProhibited :: ConnData -> Bool
 ratchetSyncSendProhibited ConnData {ratchetSyncState} =
@@ -506,7 +509,10 @@ data RcvMsgData = RcvMsgData
 data RcvMsg = RcvMsg
  { internalId :: InternalId,
    msgMeta :: MsgMeta,
+    msgType :: AgentMessageType,
    msgBody :: MsgBody,
+    internalHash :: MsgHash,
+    msgReceipt :: Maybe MsgReceipt, -- if this message is a delivery receipt
    userAck :: Bool
  }

@@ -521,6 +527,14 @@ data SndMsgData = SndMsgData
    prevMsgHash :: MsgHash
  }

+data SndMsg = SndMsg
+  { internalId :: InternalId,
+    internalSndId :: InternalSndId,
+    msgType :: AgentMessageType,
+    internalHash :: MsgHash,
+    msgReceipt :: Maybe MsgReceipt
+  }
+
 data PendingMsgData = PendingMsgData
  { msgId :: InternalId,
    msgType :: AgentMessageType,
@@ -56,8 +56,8 @@ module Simplex.Messaging.Agent.Store.SQLite
    getDeletedConnIds,
    setConnRatchetSync,
    addProcessedRatchetKeyHash,
-    checkProcessedRatchetKeyHashExists,
-    deleteProcessedRatchetKeyHashesExpired,
+    checkRatchetKeyHashExists,
+    deleteRatchetKeyHashesExpired,
    getRcvConn,
    getRcvQueueById,
    getSndQueueById,
@@ -99,16 +99,21 @@ module Simplex.Messaging.Agent.Store.SQLite
    updateSndIds,
    createSndMsg,
    createSndMsgDelivery,
+    getSndMsgViaRcpt,
+    updateSndMsgRcpt,
    getPendingMsgData,
    updatePendingMsgRIState,
    getPendingMsgs,
    deletePendingMsgs,
    setMsgUserAck,
+    getRcvMsg,
    getLastMsg,
    checkRcvMsgHashExists,
    deleteMsg,
+    deleteDeliveredSndMsg,
    deleteSndMsgDelivery,
    deleteRcvMsgHashesExpired,
+    deleteSndMsgsExpired,
    -- Double ratchet persistence
    createRatchetX3dhKeys,
    getRatchetX3dhKeys,
@@ -893,6 +898,31 @@ createSndMsgDelivery :: DB.Connection -> ConnId -> SndQueue -> InternalId -> IO
 createSndMsgDelivery db connId SndQueue {dbQueueId} msgId =
  DB.execute db "INSERT INTO snd_message_deliveries (conn_id, snd_queue_id, internal_id) VALUES (?, ?, ?)" (connId, dbQueueId, msgId)

+getSndMsgViaRcpt :: DB.Connection -> ConnId -> InternalSndId -> IO (Either StoreError SndMsg)
+getSndMsgViaRcpt db connId sndMsgId =
+  firstRow toSndMsg SEMsgNotFound $
+    DB.query
+      db
+      [sql|
+        SELECT s.internal_id, m.msg_type, s.internal_hash, s.rcpt_internal_id, s.rcpt_status
+        FROM snd_messages s
+        JOIN messages m ON s.internal_id = m.internal_id
+        WHERE s.conn_id = ? AND s.internal_snd_id = ?
+      |]
+      (connId, sndMsgId)
+  where
+    toSndMsg :: (InternalId, AgentMessageType, MsgHash, Maybe AgentMsgId, Maybe MsgReceiptStatus) -> SndMsg
+    toSndMsg (internalId, msgType, internalHash, rcptInternalId_, rcptStatus_) =
+      let msgReceipt = MsgReceipt <$> rcptInternalId_ <*> rcptStatus_
+       in SndMsg {internalId, internalSndId = sndMsgId, msgType, internalHash, msgReceipt}
+
+updateSndMsgRcpt :: DB.Connection -> ConnId -> InternalSndId -> MsgReceipt -> IO ()
+updateSndMsgRcpt db connId sndMsgId MsgReceipt {agentMsgId, msgRcptStatus} =
+  DB.execute
+    db
+    "UPDATE snd_messages SET rcpt_internal_id = ?, rcpt_status = ? WHERE conn_id = ? AND internal_snd_id = ?"
+    (agentMsgId, msgRcptStatus, connId, sndMsgId)
+
 getPendingMsgData :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (Maybe RcvQueue, PendingMsgData))
 getPendingMsgData db connId msgId = do
  rq_ <- L.head <$$> getRcvQueuesByConnId_ db connId
@@ -929,32 +959,51 @@ deletePendingMsgs db connId SndQueue {dbQueueId} =

 setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (RcvQueue, SMP.MsgId))
 setMsgUserAck db connId agentMsgId = runExceptT $ do
-  liftIO $ DB.execute db "UPDATE rcv_messages SET user_ack = ? WHERE conn_id = ? AND internal_id = ?" (True, connId, agentMsgId)
  (dbRcvId, srvMsgId) <-
    ExceptT . firstRow id SEMsgNotFound $
      DB.query db "SELECT rcv_queue_id, broker_id FROM rcv_messages WHERE conn_id = ? AND internal_id = ?" (connId, agentMsgId)
  rq <- ExceptT $ getRcvQueueById db connId dbRcvId
+  liftIO $ DB.execute db "UPDATE rcv_messages SET user_ack = ? WHERE conn_id = ? AND internal_id = ?" (True, connId, agentMsgId)
  pure (rq, srvMsgId)

-getLastMsg :: DB.Connection -> ConnId -> SMP.MsgId -> IO (Maybe RcvMsg)
-getLastMsg db connId msgId =
-  maybeFirstRow rcvMsg $
+getRcvMsg :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError RcvMsg)
+getRcvMsg db connId agentMsgId =
+  firstRow toRcvMsg SEMsgNotFound $
    DB.query
      db
      [sql|
        SELECT
-          r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity,
-          m.msg_body, r.user_ack
+          r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, r.internal_hash,
+          m.msg_type, m.msg_body, s.internal_id, s.rcpt_status, r.user_ack
+        FROM rcv_messages r
+        JOIN messages m ON r.internal_id = m.internal_id
+        LEFT JOIN snd_messages s ON s.rcpt_internal_id = r.internal_id
+        WHERE r.conn_id = ? AND r.internal_id = ?
+      |]
+      (connId, agentMsgId)
+
+getLastMsg :: DB.Connection -> ConnId -> SMP.MsgId -> IO (Maybe RcvMsg)
+getLastMsg db connId msgId =
+  maybeFirstRow toRcvMsg $
+    DB.query
+      db
+      [sql|
+        SELECT
+          r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, r.internal_hash,
+          m.msg_type, m.msg_body, s.internal_id, s.rcpt_status, r.user_ack
        FROM rcv_messages r
        JOIN messages m ON r.internal_id = m.internal_id
        JOIN connections c ON r.conn_id = c.conn_id AND c.last_internal_msg_id = r.internal_id
+        LEFT JOIN snd_messages s ON s.rcpt_internal_id = r.internal_id
        WHERE r.conn_id = ? AND r.broker_id = ?
      |]
      (connId, msgId)
-  where
-    rcvMsg (agentMsgId, internalTs, brokerId, brokerTs, sndMsgId, integrity, msgBody, userAck) =
-      let msgMeta = MsgMeta {recipient = (agentMsgId, internalTs), broker = (brokerId, brokerTs), sndMsgId, integrity}
-       in RcvMsg {internalId = InternalId agentMsgId, msgMeta, msgBody, userAck}
+
+toRcvMsg :: (Int64, InternalTs, BrokerId, BrokerTs, AgentMsgId, MsgIntegrity, MsgHash, AgentMessageType, MsgBody, Maybe AgentMsgId, Maybe MsgReceiptStatus, Bool) -> RcvMsg    
+toRcvMsg (agentMsgId, internalTs, brokerId, brokerTs, sndMsgId, integrity, internalHash, msgType, msgBody, rcptInternalId_, rcptStatus_, userAck) =
+  let msgMeta = MsgMeta {recipient = (agentMsgId, internalTs), broker = (brokerId, brokerTs), sndMsgId, integrity}
+      msgReceipt = MsgReceipt <$> rcptInternalId_ <*> rcptStatus_
+    in RcvMsg {internalId = InternalId agentMsgId, msgMeta, msgType, msgBody, internalHash, msgReceipt, userAck}

 checkRcvMsgHashExists :: DB.Connection -> ConnId -> ByteString -> IO Bool
 checkRcvMsgHashExists db connId hash = do
@@ -971,20 +1020,55 @@ deleteMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
 deleteMsg db connId msgId =
  DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)

-deleteSndMsgDelivery :: DB.Connection -> ConnId -> SndQueue -> InternalId -> IO ()
-deleteSndMsgDelivery db connId SndQueue {dbQueueId} msgId = do
+deleteMsgContent :: DB.Connection -> ConnId -> InternalId -> IO ()
+deleteMsgContent db connId msgId =
+  DB.execute db "UPDATE messages SET msg_body = x'' WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)
+
+deleteDeliveredSndMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
+deleteDeliveredSndMsg db connId msgId = do
+  cnt <- countPendingSndDeliveries_ db connId msgId
+  when (cnt == 0) $ deleteMsg db connId msgId
+
+deleteSndMsgDelivery :: DB.Connection -> ConnId -> SndQueue -> InternalId -> Bool -> IO ()
+deleteSndMsgDelivery db connId SndQueue {dbQueueId} msgId keepForReceipt = do
  DB.execute
    db
    "DELETE FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = ? AND internal_id = ?"
    (connId, dbQueueId, msgId)
-  (Only (cnt :: Int) : _) <- DB.query db "SELECT count(*) FROM snd_message_deliveries WHERE conn_id = ? AND internal_id = ?" (connId, msgId)
-  when (cnt == 0) $ deleteMsg db connId msgId
+  cnt <- countPendingSndDeliveries_ db connId msgId
+  when (cnt == 0) $ do
+    del <-
+      maybeFirstRow id (DB.query db "SELECT rcpt_internal_id, rcpt_status FROM snd_messages WHERE conn_id = ? AND internal_id = ?" (connId, msgId)) >>= \case
+        Just (Just (_ :: Int64), Just MROk) -> pure deleteMsg
+        _ -> pure $ if keepForReceipt then deleteMsgContent else deleteMsg
+    del db connId msgId
+
+countPendingSndDeliveries_ :: DB.Connection -> ConnId -> InternalId -> IO Int
+countPendingSndDeliveries_ db connId msgId = do
+  (Only cnt : _) <- DB.query db "SELECT count(*) FROM snd_message_deliveries WHERE conn_id = ? AND internal_id = ?" (connId, msgId)
+  pure cnt

 deleteRcvMsgHashesExpired :: DB.Connection -> NominalDiffTime -> IO ()
 deleteRcvMsgHashesExpired db ttl = do
  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
  DB.execute db "DELETE FROM encrypted_rcv_message_hashes WHERE created_at < ?" (Only cutoffTs)

+deleteSndMsgsExpired :: DB.Connection -> NominalDiffTime -> IO ()
+deleteSndMsgsExpired db ttl = do
+  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
+  DB.execute
+    db
+    [sql|
+      DELETE FROM messages
+      WHERE internal_id IN (
+        SELECT s.internal_id
+        FROM snd_messages s
+        JOIN messages m USING (internal_id)
+        WHERE m.internal_ts < ?
+      )
+    |]
+    (Only cutoffTs)
+
 createRatchetX3dhKeys :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> IO ()
 createRatchetX3dhKeys db connId x3dhPrivKey1 x3dhPrivKey2 =
  DB.execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2)
@@ -1504,6 +1588,10 @@ instance ToField AgentCommandTag where toField = toField . strEncode

 instance FromField AgentCommandTag where fromField = blobFieldParser strP

+instance ToField MsgReceiptStatus where toField = toField . decodeLatin1 . strEncode
+
+instance FromField MsgReceiptStatus where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8
+
 listToEither :: e -> [a] -> Either e a
 listToEither _ (x : _) = Right x
 listToEither e _ = Left e
@@ -1682,8 +1770,8 @@ addProcessedRatchetKeyHash :: DB.Connection -> ConnId -> ByteString -> IO ()
 addProcessedRatchetKeyHash db connId hash =
  DB.execute db "INSERT INTO processed_ratchet_key_hashes (conn_id, hash) VALUES (?,?)" (connId, hash)

-checkProcessedRatchetKeyHashExists :: DB.Connection -> ConnId -> ByteString -> IO Bool
-checkProcessedRatchetKeyHashExists db connId hash = do
+checkRatchetKeyHashExists :: DB.Connection -> ConnId -> ByteString -> IO Bool
+checkRatchetKeyHashExists db connId hash = do
  fromMaybe False
    <$> maybeFirstRow
      fromOnly
@@ -1693,8 +1781,8 @@ checkProcessedRatchetKeyHashExists db connId hash = do
          (connId, hash)
      )

-deleteProcessedRatchetKeyHashesExpired :: DB.Connection -> NominalDiffTime -> IO ()
-deleteProcessedRatchetKeyHashesExpired db ttl = do
+deleteRatchetKeyHashesExpired :: DB.Connection -> NominalDiffTime -> IO ()
+deleteRatchetKeyHashesExpired db ttl = do
  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
  DB.execute db "DELETE FROM processed_ratchet_key_hashes WHERE created_at < ?" (Only cutoffTs)

@@ -63,6 +63,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_r
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -91,7 +92,8 @@ schemaMigrations =
    ("m20230510_files_pending_replicas_indexes", m20230510_files_pending_replicas_indexes, Just down_m20230510_files_pending_replicas_indexes),
    ("m20230516_encrypted_rcv_message_hashes", m20230516_encrypted_rcv_message_hashes, Just down_m20230516_encrypted_rcv_message_hashes),
    ("m20230531_switch_status", m20230531_switch_status, Just down_m20230531_switch_status),
-    ("m20230615_ratchet_sync", m20230615_ratchet_sync, Just down_m20230615_ratchet_sync)
+    ("m20230615_ratchet_sync", m20230615_ratchet_sync, Just down_m20230615_ratchet_sync),
+    ("m20230701_delivery_receipts", m20230701_delivery_receipts, Just down_m20230701_delivery_receipts)
  ]

 -- | The list of migrations in ascending order by date
@@ -0,0 +1,24 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230701_delivery_receipts :: Query
+m20230701_delivery_receipts =
+  [sql|
+ALTER TABLE snd_messages ADD COLUMN rcpt_internal_id INTEGER;
+ALTER TABLE snd_messages ADD COLUMN rcpt_status TEXT;
+
+CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages(conn_id, rcpt_internal_id);
+|]
+
+down_m20230701_delivery_receipts :: Query
+down_m20230701_delivery_receipts =
+  [sql|
+DROP INDEX idx_snd_messages_rcpt_internal_id;
+
+ALTER TABLE snd_messages DROP COLUMN rcpt_internal_id;
+ALTER TABLE snd_messages DROP COLUMN rcpt_status;
+|]
@@ -119,6 +119,8 @@ CREATE TABLE snd_messages(
  previous_msg_hash BLOB NOT NULL DEFAULT x'',
  retry_int_slow INTEGER,
  retry_int_fast INTEGER,
+  rcpt_internal_id INTEGER,
+  rcpt_status TEXT,
  PRIMARY KEY(conn_id, internal_snd_id),
  FOREIGN KEY(conn_id, internal_id) REFERENCES messages
  ON DELETE CASCADE
@@ -461,3 +463,7 @@ CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hash
  conn_id,
  hash
 );
+CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages(
+  conn_id,
+  rcpt_internal_id
+);