From 3af34dea8b7594c352b07c87854c059da0e43f22 Mon Sep 17 00:00:00 2001 From: Efim Poberezkin Date: Sat, 27 Feb 2021 15:06:51 +0400 Subject: [PATCH] agent store: accept internal ts and return internal id (#58) --- src/Simplex/Messaging/Agent.hs | 20 +++++------ src/Simplex/Messaging/Agent/Client.hs | 34 +++++++++--------- src/Simplex/Messaging/Agent/Store.hs | 6 ++-- src/Simplex/Messaging/Agent/Store/SQLite.hs | 40 ++++++++++----------- src/Simplex/Messaging/Agent/Transmission.hs | 9 +++-- tests/AgentTests/SQLiteTests.hs | 22 ++++++------ 6 files changed, 65 insertions(+), 66 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 3b1ce4115..4dfee0849 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/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 " -- 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 diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 37594e95c..b49c294d7 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/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 " $ \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 diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index ff8144204..cf5fbf678 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 573da9f78..9912c304c 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/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| diff --git a/src/Simplex/Messaging/Agent/Transmission.hs b/src/Simplex/Messaging/Agent/Transmission.hs index d7d20e53a..5563c58a5 100644 --- a/src/Simplex/Messaging/Agent/Transmission.hs +++ b/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) diff --git a/tests/AgentTests/SQLiteTests.hs b/tests/AgentTests/SQLiteTests.hs index f0ff453c1..92b122ace 100644 --- a/tests/AgentTests/SQLiteTests.hs +++ b/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