agent store: accept internal ts and return internal id (#58)

src/Simplex/Messaging/Agent.hs

@@ -164,10 +164,10 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) =
         _ -> throwError PROHIBITED -- NOT_READY ?
       where
         sendMsg sq = do
-          withStore $ \st -> createSndMsg st connAlias msgBody
-          sendAgentMessage c sq $ A_MSG msgBody
-          -- TODO respond $ SENT aMsgId
-          respond $ SENT 0
+          senderTs <- liftIO getCurrentTime
+          senderId <- withStore $ \st -> createSndMsg st connAlias msgBody senderTs
+          sendAgentMessage c sq senderTs $ A_MSG msgBody
+          respond $ SENT senderId
 
     suspendConnection :: m ()
     suspendConnection =
@@ -195,7 +195,8 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) =
     sendReplyQInfo srv sq = do
       (rq, qInfo) <- newReceiveQueue c srv connAlias
       withStore $ \st -> upgradeSndConnToDuplex st connAlias rq
-      sendAgentMessage c sq $ REPLY qInfo
+      senderTs <- liftIO getCurrentTime
+      sendAgentMessage c sq senderTs $ REPLY qInfo
 
     respond :: ACommand 'Agent -> m ()
     respond resp = atomically $ writeTBQueue sndQ (corrId, connAlias, resp)
@@ -250,17 +251,12 @@ processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do
             A_MSG body -> do
               logServer "<--" c srv rId "MSG <MSG>"
               -- TODO check message status
-              withStore $ \st -> createRcvMsg st connAlias body senderMsgId senderTimestamp srvMsgId srvTs
-              -- TODO either:
-              -- TODO   - pass MSG{.., msg = generatedRcvMsg} to notify,
-              -- TODO   - retrieve internalTs from generatedRcvMsg,
-              -- TODO   - pass recipientTs to createRcvMsg (second cleanest),
-              -- TODO   - remove m_recipient from MSG (might be the cleanest?).
               recipientTs <- liftIO getCurrentTime
+              recipientId <- withStore $ \st -> createRcvMsg st connAlias body recipientTs senderMsgId senderTimestamp srvMsgId srvTs
               notify connAlias $
                 MSG
                   { m_status = MsgOk,
-                    m_recipient = (0, recipientTs),
+                    m_recipient = (recipientId, recipientTs),
                     m_sender = (senderMsgId, senderTimestamp),
                     m_broker = (srvMsgId, srvTs),
                     m_body = body

src/Simplex/Messaging/Agent/Client.hs

@@ -224,7 +224,9 @@ sendHello c SndQueue {server, sndId, sndPrivateKey, encryptKey} verifyKey = do
     send 20 msg
   where
     mkHello :: AckMode -> m ByteString
-    mkHello ackMode = mkAgentMessage encryptKey $ HELLO verifyKey ackMode
+    mkHello ackMode = do
+      senderTs <- liftIO getCurrentTime
+      mkAgentMessage encryptKey senderTs $ HELLO verifyKey ackMode
 
     send :: Int -> ByteString -> SMPClient -> ExceptT SMPClientError IO ()
     send 0 _ _ = throwE SMPResponseTimeout -- TODO different error
@@ -255,22 +257,20 @@ deleteQueue c RcvQueue {server, rcvId, rcvPrivateKey} =
   withLogSMP c server rcvId "DEL" $ \smp ->
     deleteSMPQueue smp rcvPrivateKey rcvId
 
-sendAgentMessage :: AgentMonad m => AgentClient -> SndQueue -> AMessage -> m ()
-sendAgentMessage c SndQueue {server, sndId, sndPrivateKey, encryptKey} agentMsg = do
-  msg <- mkAgentMessage encryptKey agentMsg
+sendAgentMessage :: AgentMonad m => AgentClient -> SndQueue -> SenderTimestamp -> AMessage -> m ()
+sendAgentMessage c SndQueue {server, sndId, sndPrivateKey, encryptKey} senderTs agentMsg = do
+  msg <- mkAgentMessage encryptKey senderTs agentMsg
   withLogSMP c server sndId "SEND <message>" $ \smp ->
     sendSMPMessage smp (Just sndPrivateKey) sndId msg
 
-mkAgentMessage :: (MonadUnliftIO m, MonadError AgentErrorType m) => EncryptionKey -> AMessage -> m ByteString
-mkAgentMessage encKey agentMessage = do
-  ts <- liftIO getCurrentTime
-  liftError CRYPTO $ C.encrypt encKey $ msg ts
-  where
-    msg ts =
-      serializeSMPMessage
-        SMPMessage
-          { senderMsgId = 0,
-            senderTimestamp = ts,
-            previousMsgHash = "1234", -- TODO hash of the previous message
-            agentMessage
-          }
+mkAgentMessage :: (MonadUnliftIO m, MonadError AgentErrorType m) => EncryptionKey -> SenderTimestamp -> AMessage -> m ByteString
+mkAgentMessage encKey senderTs agentMessage = do
+  let msg =
+        serializeSMPMessage
+          SMPMessage
+            { senderMsgId = 0,
+              senderTimestamp = senderTs,
+              previousMsgHash = "1234", -- TODO hash of the previous message
+              agentMessage
+            }
+  liftError CRYPTO $ C.encrypt encKey msg

src/Simplex/Messaging/Agent/Store.hs

@@ -40,8 +40,8 @@ class Monad m => MonadAgentStore s m where
   setSndQueueStatus :: s -> SndQueue -> QueueStatus -> m ()
 
   -- Msg management
-  createRcvMsg :: s -> ConnAlias -> MsgBody -> ExternalSndId -> ExternalSndTs -> BrokerId -> BrokerTs -> m ()
-  createSndMsg :: s -> ConnAlias -> MsgBody -> m ()
+  createRcvMsg :: s -> ConnAlias -> MsgBody -> InternalTs -> ExternalSndId -> ExternalSndTs -> BrokerId -> BrokerTs -> m InternalId
+  createSndMsg :: s -> ConnAlias -> MsgBody -> InternalTs -> m InternalId
   getMsg :: s -> ConnAlias -> InternalId -> m Msg
 
 -- * Queue types
@@ -150,7 +150,7 @@ data RcvMsg = RcvMsg
 
 type InternalRcvId = Int64
 
-type ExternalSndId = Integer
+type ExternalSndId = Int64
 
 type ExternalSndTs = UTCTime
 

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

@@ -22,7 +22,6 @@ import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Data.Maybe (fromMaybe)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8)
-import Data.Time (getCurrentTime)
 import Database.SQLite.Simple as DB
 import Database.SQLite.Simple.FromField
 import Database.SQLite.Simple.Internal (Field (..))
@@ -108,15 +107,15 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
     liftIO $
       updateSndQueueStatus dbConn sndQueue status
 
-  createRcvMsg :: SQLiteStore -> ConnAlias -> MsgBody -> ExternalSndId -> ExternalSndTs -> BrokerId -> BrokerTs -> m ()
-  createRcvMsg SQLiteStore {dbConn} connAlias msgBody externalSndId externalSndTs brokerId brokerTs =
+  createRcvMsg :: SQLiteStore -> ConnAlias -> MsgBody -> InternalTs -> ExternalSndId -> ExternalSndTs -> BrokerId -> BrokerTs -> m InternalId
+  createRcvMsg SQLiteStore {dbConn} connAlias msgBody internalTs externalSndId externalSndTs brokerId brokerTs =
     liftIOEither $
-      insertRcvMsg dbConn connAlias msgBody externalSndId externalSndTs brokerId brokerTs
+      insertRcvMsg dbConn connAlias msgBody internalTs externalSndId externalSndTs brokerId brokerTs
 
-  createSndMsg :: SQLiteStore -> ConnAlias -> MsgBody -> m ()
-  createSndMsg SQLiteStore {dbConn} connAlias msgBody =
+  createSndMsg :: SQLiteStore -> ConnAlias -> MsgBody -> InternalTs -> m InternalId
+  createSndMsg SQLiteStore {dbConn} connAlias msgBody internalTs =
     liftIOEither $
-      insertSndMsg dbConn connAlias msgBody
+      insertSndMsg dbConn connAlias msgBody internalTs
 
   getMsg :: SQLiteStore -> ConnAlias -> InternalId -> m Msg
   getMsg _st _connAlias _id = throwError SENotImplemented
@@ -444,12 +443,13 @@ insertRcvMsg ::
   DB.Connection ->
   ConnAlias ->
   MsgBody ->
+  InternalTs ->
   ExternalSndId ->
   ExternalSndTs ->
   BrokerId ->
   BrokerTs ->
-  IO (Either StoreError ())
-insertRcvMsg dbConn connAlias msgBody externalSndId externalSndTs brokerId brokerTs =
+  IO (Either StoreError InternalId)
+insertRcvMsg dbConn connAlias msgBody internalTs externalSndId externalSndTs brokerId brokerTs =
   DB.withTransaction dbConn $ do
     queues <- retrieveConnQueues_ dbConn connAlias
     case queues of
@@ -457,10 +457,10 @@ insertRcvMsg dbConn connAlias msgBody externalSndId externalSndTs brokerId broke
         (lastInternalId, lastInternalRcvId) <- retrieveLastInternalIdsRcv_ dbConn connAlias
         let internalId = lastInternalId + 1
         let internalRcvId = lastInternalRcvId + 1
-        insertRcvMsgBase_ dbConn connAlias internalId internalRcvId msgBody
+        insertRcvMsgBase_ dbConn connAlias internalId internalTs internalRcvId msgBody
         insertRcvMsgDetails_ dbConn connAlias internalRcvId internalId externalSndId externalSndTs brokerId brokerTs
         updateLastInternalIdsRcv_ dbConn connAlias internalId internalRcvId
-        return $ Right ()
+        return $ Right internalId
       (Nothing, Just _sndQ) -> return $ Left (SEBadConnType CSnd)
       _ -> return $ Left SEBadConn
 
@@ -477,9 +477,8 @@ retrieveLastInternalIdsRcv_ dbConn connAlias = do
       [":conn_alias" := connAlias]
   return (lastInternalId, lastInternalRcvId)
 
-insertRcvMsgBase_ :: DB.Connection -> ConnAlias -> InternalId -> InternalRcvId -> MsgBody -> IO ()
-insertRcvMsgBase_ dbConn connAlias internalId internalRcvId msgBody = do
-  internalTs <- liftIO getCurrentTime
+insertRcvMsgBase_ :: DB.Connection -> ConnAlias -> InternalId -> InternalTs -> InternalRcvId -> MsgBody -> IO ()
+insertRcvMsgBase_ dbConn connAlias internalId internalTs internalRcvId msgBody = do
   DB.executeNamed
     dbConn
     [sql|
@@ -542,8 +541,8 @@ updateLastInternalIdsRcv_ dbConn connAlias newInternalId newInternalRcvId =
 
 -- * createSndMsg helpers
 
-insertSndMsg :: DB.Connection -> ConnAlias -> MsgBody -> IO (Either StoreError ())
-insertSndMsg dbConn connAlias msgBody =
+insertSndMsg :: DB.Connection -> ConnAlias -> MsgBody -> InternalTs -> IO (Either StoreError InternalId)
+insertSndMsg dbConn connAlias msgBody internalTs =
   DB.withTransaction dbConn $ do
     queues <- retrieveConnQueues_ dbConn connAlias
     case queues of
@@ -551,10 +550,10 @@ insertSndMsg dbConn connAlias msgBody =
         (lastInternalId, lastInternalSndId) <- retrieveLastInternalIdsSnd_ dbConn connAlias
         let internalId = lastInternalId + 1
         let internalSndId = lastInternalSndId + 1
-        insertSndMsgBase_ dbConn connAlias internalId internalSndId msgBody
+        insertSndMsgBase_ dbConn connAlias internalId internalTs internalSndId msgBody
         insertSndMsgDetails_ dbConn connAlias internalSndId internalId
         updateLastInternalIdsSnd_ dbConn connAlias internalId internalSndId
-        return $ Right ()
+        return $ Right internalId
       (Just _rcvQ, Nothing) -> return $ Left (SEBadConnType CRcv)
       _ -> return $ Left SEBadConn
 
@@ -571,9 +570,8 @@ retrieveLastInternalIdsSnd_ dbConn connAlias = do
       [":conn_alias" := connAlias]
   return (lastInternalId, lastInternalSndId)
 
-insertSndMsgBase_ :: DB.Connection -> ConnAlias -> InternalId -> InternalSndId -> MsgBody -> IO ()
-insertSndMsgBase_ dbConn connAlias internalId internalSndId msgBody = do
-  internalTs <- liftIO getCurrentTime
+insertSndMsgBase_ :: DB.Connection -> ConnAlias -> InternalId -> InternalTs -> InternalSndId -> MsgBody -> IO ()
+insertSndMsgBase_ dbConn connAlias internalId internalTs internalSndId msgBody = do
   DB.executeNamed
     dbConn
     [sql|

src/Simplex/Messaging/Agent/Transmission.hs

@@ -20,6 +20,7 @@ import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
+import Data.Int (Int64)
 import Data.Kind
 import Data.Time.Clock (UTCTime)
 import Data.Time.ISO8601
@@ -109,8 +110,8 @@ type Message = ByteString
 data SMPMessage
   = SMPConfirmation SenderPublicKey
   | SMPMessage
-      { senderMsgId :: Integer,
-        senderTimestamp :: UTCTime,
+      { senderMsgId :: AgentMsgId,
+        senderTimestamp :: SenderTimestamp,
         previousMsgHash :: ByteString,
         agentMessage :: AMessage
       }
@@ -229,7 +230,9 @@ data QueueDirection = SND | RCV deriving (Show)
 data QueueStatus = New | Confirmed | Secured | Active | Disabled
   deriving (Eq, Show, Read)
 
-type AgentMsgId = Integer
+type AgentMsgId = Int64
+
+type SenderTimestamp = UTCTime
 
 data MsgStatus = MsgOk | MsgError MsgErrorType
   deriving (Eq, Show)

tests/AgentTests/SQLiteTests.hs

@@ -290,40 +290,42 @@ testSetSndQueueStatusNoQueue = do
 
 testCreateRcvMsg :: SpecWith SQLiteStore
 testCreateRcvMsg = do
-  it "should create a RcvMsg" $ \store -> do
+  it "should create a RcvMsg and return InternalId" $ \store -> do
     createRcvConn store rcvQueue1
       `returnsResult` ()
     -- TODO getMsg to check message
     let ts = UTCTime (fromGregorian 2021 02 24) (secondsToDiffTime 0)
-    createRcvMsg store "conn1" (encodeUtf8 "Hello world!") 1 ts "1" ts
-      `returnsResult` ()
+    createRcvMsg store "conn1" (encodeUtf8 "Hello world!") ts 1 ts "1" ts
+      `returnsResult` (1 :: InternalId)
 
 testCreateRcvMsgNoQueue :: SpecWith SQLiteStore
 testCreateRcvMsgNoQueue = do
   it "should throw error on attempt to create a RcvMsg w/t a RcvQueue" $ \store -> do
     let ts = UTCTime (fromGregorian 2021 02 24) (secondsToDiffTime 0)
-    createRcvMsg store "conn1" (encodeUtf8 "Hello world!") 1 ts "1" ts
+    createRcvMsg store "conn1" (encodeUtf8 "Hello world!") ts 1 ts "1" ts
       `throwsError` SEBadConn
     createSndConn store sndQueue1
       `returnsResult` ()
-    createRcvMsg store "conn1" (encodeUtf8 "Hello world!") 1 ts "1" ts
+    createRcvMsg store "conn1" (encodeUtf8 "Hello world!") ts 1 ts "1" ts
       `throwsError` SEBadConnType CSnd
 
 testCreateSndMsg :: SpecWith SQLiteStore
 testCreateSndMsg = do
-  it "should create a SndMsg" $ \store -> do
+  it "should create a SndMsg and return InternalId" $ \store -> do
     createSndConn store sndQueue1
       `returnsResult` ()
     -- TODO getMsg to check message
-    createSndMsg store "conn1" (encodeUtf8 "Hello world!")
-      `returnsResult` ()
+    let ts = UTCTime (fromGregorian 2021 02 24) (secondsToDiffTime 0)
+    createSndMsg store "conn1" (encodeUtf8 "Hello world!") ts
+      `returnsResult` (1 :: InternalId)
 
 testCreateSndMsgNoQueue :: SpecWith SQLiteStore
 testCreateSndMsgNoQueue = do
   it "should throw error on attempt to create a SndMsg w/t a SndQueue" $ \store -> do
-    createSndMsg store "conn1" (encodeUtf8 "Hello world!")
+    let ts = UTCTime (fromGregorian 2021 02 24) (secondsToDiffTime 0)
+    createSndMsg store "conn1" (encodeUtf8 "Hello world!") ts
       `throwsError` SEBadConn
     createRcvConn store rcvQueue1
       `returnsResult` ()
-    createSndMsg store "conn1" (encodeUtf8 "Hello world!")
+    createSndMsg store "conn1" (encodeUtf8 "Hello world!") ts
       `throwsError` SEBadConnType CRcv