ACK command, ackMessage, do not send ACK to SMP server on MSG (after the initial handshake) (#187)

src/Simplex/Messaging/Agent.hs

@@ -46,6 +46,7 @@ module Simplex.Messaging.Agent
     acceptConnection,
     subscribeConnection,
     sendMessage,
+    ackMessage,
     suspendConnection,
     deleteConnection,
   )
@@ -146,6 +147,9 @@ subscribeConnection c = withAgentEnv c . subscribeConnection' c
 sendMessage :: AgentErrorMonad m => AgentClient -> ConnId -> MsgBody -> m AgentMsgId
 sendMessage c = withAgentEnv c .: sendMessage' c
 
+ackMessage :: AgentErrorMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
+ackMessage c = withAgentEnv c .: ackMessage' c
+
 -- | Suspend SMP agent connection (OFF command)
 suspendConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ()
 suspendConnection c = withAgentEnv c . suspendConnection' c
@@ -236,6 +240,7 @@ processCommand c (connId, cmd) = case cmd of
   ACPT confId ownConnInfo -> acceptConnection' c connId confId ownConnInfo $> (connId, OK)
   SUB -> subscribeConnection' c connId $> (connId, OK)
   SEND msgBody -> (connId,) . MID <$> sendMessage' c connId msgBody
+  ACK msgId -> ackMessage' c connId msgId $> (connId, OK)
   OFF -> suspendConnection' c connId $> (connId, OK)
   DEL -> deleteConnection' c connId $> (connId, OK)
 
@@ -426,6 +431,20 @@ runSrvMsgDelivery c@AgentClient {subQ} srv = do
     notify :: ConnId -> ACommand 'Agent -> m ()
     notify connId cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
 
+ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
+ackMessage' c connId msgId = do
+  withStore (`getConn` connId) >>= \case
+    SomeConn _ (DuplexConnection _ rq _) -> ack rq
+    SomeConn _ (RcvConnection _ rq) -> ack rq
+    _ -> throwError $ CONN SIMPLEX
+  where
+    ack :: RcvQueue -> m ()
+    ack rq = do
+      let mId = InternalId msgId
+      withStore $ \st -> checkRcvMsg st connId mId
+      sendAck c rq
+      withStore $ \st -> updateRcvMsgAck st connId mId
+
 -- | Suspend SMP agent connection (OFF command) in Reader monad
 suspendConnection' :: AgentMonad m => AgentClient -> ConnId -> m ()
 suspendConnection' c connId =
@@ -491,14 +510,12 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
           let msgHash = C.sha256Hash msg
           case parseSMPMessage msg of
             Left e -> notify $ ERR e
-            Right (SMPConfirmation senderKey cInfo) -> smpConfirmation senderKey cInfo
+            Right (SMPConfirmation senderKey cInfo) -> smpConfirmation senderKey cInfo >> sendAck c rq
             Right SMPMessage {agentMessage, senderMsgId, senderTimestamp, previousMsgHash} ->
               case agentMessage of
-                HELLO verifyKey _ -> helloMsg verifyKey msgBody
-                REPLY qInfo -> replyMsg qInfo
+                HELLO verifyKey _ -> helloMsg verifyKey msgBody >> sendAck c rq
+                REPLY qInfo -> replyMsg qInfo >> sendAck c rq
                 A_MSG body -> agentClientMsg previousMsgHash (senderMsgId, senderTimestamp) (srvMsgId, srvTs) body msgHash
-          sendAck c rq
-          return ()
         SMP.END -> do
           removeSubscription c connId
           logServer "<--" c srv rId "END"

src/Simplex/Messaging/Agent/Protocol.hs

@@ -172,7 +172,7 @@ data ACommand (p :: AParty) where
   SENT :: AgentMsgId -> ACommand Agent
   MERR :: AgentMsgId -> AgentErrorType -> ACommand Agent
   MSG :: MsgMeta -> MsgBody -> ACommand Agent
-  -- ACK :: AgentMsgId -> ACommand Client
+  ACK :: AgentMsgId -> ACommand Client
   -- RCVD :: AgentMsgId -> ACommand Agent
   OFF :: ACommand Client
   DEL :: ACommand Client
@@ -469,6 +469,7 @@ commandP =
     <|> "SENT " *> sentResp
     <|> "MERR " *> msgErrResp
     <|> "MSG " *> message
+    <|> "ACK " *> ackCmd
     <|> "OFF" $> ACmd SClient OFF
     <|> "DEL" $> ACmd SClient DEL
     <|> "ERR " *> agentError
@@ -485,6 +486,7 @@ commandP =
     sentResp = ACmd SAgent . SENT <$> A.decimal
     msgErrResp = ACmd SAgent <$> (MERR <$> A.decimal <* A.space <*> agentErrorTypeP)
     message = ACmd SAgent <$> (MSG <$> msgMetaP <* A.space <*> A.takeByteString)
+    ackCmd = ACmd SClient . ACK <$> A.decimal
     msgMetaP = do
       integrity <- msgIntegrityP
       recipient <- " R=" *> partyMeta A.decimal
@@ -526,6 +528,7 @@ serializeCommand = \case
   MERR mId e -> "MERR " <> bshow mId <> " " <> serializeAgentError e
   MSG msgMeta msgBody ->
     "MSG " <> serializeMsgMeta msgMeta <> " " <> serializeBinary msgBody
+  ACK mId -> "ACK " <> bshow mId
   OFF -> "OFF"
   DEL -> "DEL"
   CON -> "CON"

src/Simplex/Messaging/Agent/Store.hs

@@ -60,6 +60,8 @@ class Monad m => MonadAgentStore s m where
   getPendingMsgData :: s -> ConnId -> InternalId -> m (SndQueue, MsgBody)
   getPendingMsgs :: s -> ConnId -> m [PendingMsg]
   getMsg :: s -> ConnId -> InternalId -> m Msg
+  checkRcvMsg :: s -> ConnId -> InternalId -> m ()
+  updateRcvMsgAck :: s -> ConnId -> InternalId -> m ()
 
 -- * Queue types
 

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -437,7 +437,35 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
         <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_alias = ? AND snd_status = ?" (connId, SndMsgCreated)
 
   getMsg :: SQLiteStore -> ConnId -> InternalId -> m Msg
-  getMsg _st _connAlias _id = throwError SENotImplemented
+  getMsg _st _connId _id = throwError SENotImplemented
+
+  checkRcvMsg :: SQLiteStore -> ConnId -> InternalId -> m ()
+  checkRcvMsg st connId msgId =
+    liftIOEither . withTransaction st $ \db ->
+      hasMsg
+        <$> DB.query
+          db
+          [sql|
+            SELECT conn_alias, internal_id
+            FROM rcv_messages
+            WHERE conn_alias = ? AND internal_id = ?
+          |]
+          (connId, msgId)
+    where
+      hasMsg :: [(ConnId, InternalId)] -> Either StoreError ()
+      hasMsg r = if null r then Left SEMsgNotFound else Right ()
+
+  updateRcvMsgAck :: SQLiteStore -> ConnId -> InternalId -> m ()
+  updateRcvMsgAck st connId msgId =
+    liftIO . withTransaction st $ \db -> do
+      DB.execute
+        db
+        [sql|
+          UPDATE rcv_messages
+          SET rcv_status = ?, ack_brocker_ts = datetime('now')
+          WHERE conn_alias = ? AND internal_id = ?
+        |]
+        (AcknowledgedToBroker, connId, msgId)
 
 -- * Auxiliary helpers