agent store: settle naming of Rcv and Snd entities (#55)

This commit is contained in:
Efim Poberezkin committed 2021-02-26 18:17:56 +04:00
1 parent a0c52033f7
commit 8134bb0eb9
5 files changed
+150 -147

No files matched your search

+14 -12
View File
@@ -147,8 +147,9 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) =
subscribeConnection =
withStore (`getConn` connAlias) >>= \case
SomeConn _ (DuplexConnection _ rq _) -> subscribe rq
SomeConn _ (ReceiveConnection _ rq) -> subscribe rq
-- TODO possibly there should be a separate error type trying to send the message to the connection without ReceiveQueue
SomeConn _ (RcvConnection _ rq) -> subscribe rq
-- TODO possibly there should be a separate error type trying
-- TODO to send the message to the connection without RcvQueue
_ -> throwError PROHIBITED
where
subscribe rq = subscribeQueue c rq connAlias >> respond OK
@@ -157,8 +158,9 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) =
sendMessage msgBody =
withStore (`getConn` connAlias) >>= \case
SomeConn _ (DuplexConnection _ _ sq) -> sendMsg sq
SomeConn _ (SendConnection _ sq) -> sendMsg sq
-- TODO possibly there should be a separate error type trying to send the message to the connection without SendQueue
SomeConn _ (SndConnection _ sq) -> sendMsg sq
-- TODO possibly there should be a separate error type trying
-- TODO to send the message to the connection without SndQueue
_ -> throwError PROHIBITED -- NOT_READY ?
where
sendMsg sq = do
@@ -171,7 +173,7 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) =
suspendConnection =
withStore (`getConn` connAlias) >>= \case
SomeConn _ (DuplexConnection _ rq _) -> suspend rq
SomeConn _ (ReceiveConnection _ rq) -> suspend rq
SomeConn _ (RcvConnection _ rq) -> suspend rq
_ -> throwError PROHIBITED
where
suspend rq = suspendQueue c rq >> respond OK
@@ -180,7 +182,7 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) =
deleteConnection =
withStore (`getConn` connAlias) >>= \case
SomeConn _ (DuplexConnection _ rq _) -> delete rq
SomeConn _ (ReceiveConnection _ rq) -> delete rq
SomeConn _ (RcvConnection _ rq) -> delete rq
_ -> throwError PROHIBITED
where
delete rq = do
@@ -189,7 +191,7 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) =
withStore (`deleteConn` connAlias)
respond OK
sendReplyQInfo :: SMPServer -> SendQueue -> m ()
sendReplyQInfo :: SMPServer -> SndQueue -> m ()
sendReplyQInfo srv sq = do
(rq, qInfo) <- newReceiveQueue c srv connAlias
withStore $ \st -> upgradeSndConnToDuplex st connAlias rq
@@ -208,7 +210,7 @@ subscriber c@AgentClient {msgQ} = forever $ do
processSMPTransmission :: forall m. AgentMonad m => AgentClient -> SMPServerTransmission -> m ()
processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do
rq@ReceiveQueue {connAlias, decryptKey, status} <- withStore $ \st -> getRcvQueue st srv rId
rq@RcvQueue {connAlias, decryptKey, status} <- withStore $ \st -> getRcvQueue st srv rId
case cmd of
SMP.MSG srvMsgId srvTs msgBody -> do
-- TODO deduplicate with previously received
@@ -239,7 +241,7 @@ processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do
sendAck c rq
REPLY qInfo -> do
logServer "<--" c srv rId "MSG <REPLY>"
-- TODO move senderKey inside SendQueue
-- TODO move senderKey inside SndQueue
(sq, senderKey, verifyKey) <- newSendQueue qInfo connAlias
withStore $ \st -> upgradeRcvConnToDuplex st connAlias sq
connectToSendQueue c sq senderKey verifyKey
@@ -274,7 +276,7 @@ processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do
notify :: ConnAlias -> ACommand 'Agent -> m ()
notify connAlias msg = atomically $ writeTBQueue sndQ ("", connAlias, msg)
connectToSendQueue :: AgentMonad m => AgentClient -> SendQueue -> SenderPublicKey -> VerificationKey -> m ()
connectToSendQueue :: AgentMonad m => AgentClient -> SndQueue -> SenderPublicKey -> VerificationKey -> m ()
connectToSendQueue c sq senderKey verifyKey = do
sendConfirmation c sq senderKey
withStore $ \st -> setSndQueueStatus st sq Confirmed
@@ -285,13 +287,13 @@ decryptMessage :: (MonadUnliftIO m, MonadError AgentErrorType m) => DecryptionKe
decryptMessage decryptKey msg = liftError CRYPTO $ C.decrypt decryptKey msg
newSendQueue ::
(MonadUnliftIO m, MonadReader Env m) => SMPQueueInfo -> ConnAlias -> m (SendQueue, SenderPublicKey, VerificationKey)
(MonadUnliftIO m, MonadReader Env m) => SMPQueueInfo -> ConnAlias -> m (SndQueue, SenderPublicKey, VerificationKey)
newSendQueue (SMPQueueInfo smpServer senderId encryptKey) connAlias = do
size <- asks $ rsaKeySize . config
(senderKey, sndPrivateKey) <- liftIO $ C.generateKeyPair size
(verifyKey, signKey) <- liftIO $ C.generateKeyPair size
let sndQueue =
SendQueue
SndQueue
{ server = smpServer,
sndId = senderId,
connAlias,
+20 -20
View File
@@ -145,7 +145,7 @@ withLogSMP c srv qId cmdStr action = do
logServer "<--" c srv qId "OK"
return res
newReceiveQueue :: AgentMonad m => AgentClient -> SMPServer -> ConnAlias -> m (ReceiveQueue, SMPQueueInfo)
newReceiveQueue :: AgentMonad m => AgentClient -> SMPServer -> ConnAlias -> m (RcvQueue, SMPQueueInfo)
newReceiveQueue c srv connAlias = do
size <- asks $ rsaKeySize . config
(recipientKey, rcvPrivateKey) <- liftIO $ C.generateKeyPair size
@@ -154,7 +154,7 @@ newReceiveQueue c srv connAlias = do
logServer "<--" c srv "" $ B.unwords ["IDS", logSecret rcvId, logSecret sId]
(encryptKey, decryptKey) <- liftIO $ C.generateKeyPair size
let rq =
ReceiveQueue
RcvQueue
{ server = srv,
rcvId,
connAlias,
@@ -168,14 +168,14 @@ newReceiveQueue c srv connAlias = do
addSubscription c rq connAlias
return (rq, SMPQueueInfo srv sId encryptKey)
subscribeQueue :: AgentMonad m => AgentClient -> ReceiveQueue -> ConnAlias -> m ()
subscribeQueue c rq@ReceiveQueue {server, rcvPrivateKey, rcvId} connAlias = do
subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnAlias -> m ()
subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connAlias = do
withLogSMP c server rcvId "SUB" $ \smp ->
subscribeSMPQueue smp rcvPrivateKey rcvId
addSubscription c rq connAlias
addSubscription :: MonadUnliftIO m => AgentClient -> ReceiveQueue -> ConnAlias -> m ()
addSubscription c ReceiveQueue {server} connAlias = atomically $ do
addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnAlias -> m ()
addSubscription c RcvQueue {server} connAlias = atomically $ do
modifyTVar (subscrConns c) $ M.insert connAlias server
modifyTVar (subscrSrvrs c) $ M.alter (Just . addSub) server
where
@@ -206,8 +206,8 @@ showServer srv = B.pack $ host srv <> maybe "" (":" <>) (port srv)
logSecret :: ByteString -> ByteString
logSecret bs = encode $ B.take 3 bs
sendConfirmation :: forall m. AgentMonad m => AgentClient -> SendQueue -> SenderPublicKey -> m ()
sendConfirmation c SendQueue {server, sndId, encryptKey} senderKey = do
sendConfirmation :: forall m. AgentMonad m => AgentClient -> SndQueue -> SenderPublicKey -> m ()
sendConfirmation c SndQueue {server, sndId, encryptKey} senderKey = do
msg <- mkConfirmation
withLogSMP c server sndId "SEND <KEY>" $ \smp ->
sendSMPMessage smp Nothing sndId msg
@@ -217,8 +217,8 @@ sendConfirmation c SendQueue {server, sndId, encryptKey} senderKey = do
let msg = serializeSMPMessage $ SMPConfirmation senderKey
liftError CRYPTO $ C.encrypt encryptKey msg
sendHello :: forall m. AgentMonad m => AgentClient -> SendQueue -> VerificationKey -> m ()
sendHello c SendQueue {server, sndId, sndPrivateKey, encryptKey} verifyKey = do
sendHello :: forall m. AgentMonad m => AgentClient -> SndQueue -> VerificationKey -> m ()
sendHello c SndQueue {server, sndId, sndPrivateKey, encryptKey} verifyKey = do
msg <- mkHello $ AckMode On
withLogSMP c server sndId "SEND <HELLO> (retrying)" $
send 20 msg
@@ -235,28 +235,28 @@ sendHello c SendQueue {server, sndId, sndPrivateKey, encryptKey} verifyKey = do
send (retry - 1) msg smp
e -> throwE e
secureQueue :: AgentMonad m => AgentClient -> ReceiveQueue -> SenderPublicKey -> m ()
secureQueue c ReceiveQueue {server, rcvId, rcvPrivateKey} senderKey =
secureQueue :: AgentMonad m => AgentClient -> RcvQueue -> SenderPublicKey -> m ()
secureQueue c RcvQueue {server, rcvId, rcvPrivateKey} senderKey =
withLogSMP c server rcvId "KEY <key>" $ \smp ->
secureSMPQueue smp rcvPrivateKey rcvId senderKey
sendAck :: AgentMonad m => AgentClient -> ReceiveQueue -> m ()
sendAck c ReceiveQueue {server, rcvId, rcvPrivateKey} =
sendAck :: AgentMonad m => AgentClient -> RcvQueue -> m ()
sendAck c RcvQueue {server, rcvId, rcvPrivateKey} =
withLogSMP c server rcvId "ACK" $ \smp ->
ackSMPMessage smp rcvPrivateKey rcvId
suspendQueue :: AgentMonad m => AgentClient -> ReceiveQueue -> m ()
suspendQueue c ReceiveQueue {server, rcvId, rcvPrivateKey} =
suspendQueue :: AgentMonad m => AgentClient -> RcvQueue -> m ()
suspendQueue c RcvQueue {server, rcvId, rcvPrivateKey} =
withLogSMP c server rcvId "OFF" $ \smp ->
suspendSMPQueue smp rcvPrivateKey rcvId
deleteQueue :: AgentMonad m => AgentClient -> ReceiveQueue -> m ()
deleteQueue c ReceiveQueue {server, rcvId, rcvPrivateKey} =
deleteQueue :: AgentMonad m => AgentClient -> RcvQueue -> m ()
deleteQueue c RcvQueue {server, rcvId, rcvPrivateKey} =
withLogSMP c server rcvId "DEL" $ \smp ->
deleteSMPQueue smp rcvPrivateKey rcvId
sendAgentMessage :: AgentMonad m => AgentClient -> SendQueue -> AMessage -> m ()
sendAgentMessage c SendQueue {server, sndId, sndPrivateKey, encryptKey} agentMsg = do
sendAgentMessage :: AgentMonad m => AgentClient -> SndQueue -> AMessage -> m ()
sendAgentMessage c SndQueue {server, sndId, sndPrivateKey, encryptKey} agentMsg = do
msg <- mkAgentMessage encryptKey agentMsg
withLogSMP c server sndId "SEND <message>" $ \smp ->
sendSMPMessage smp (Just sndPrivateKey) sndId msg
+24 -23
View File
@@ -29,15 +29,15 @@ import Simplex.Messaging.Types
-- | Store class type. Defines store access methods for implementations.
class Monad m => MonadAgentStore s m where
-- Queue and Connection management
createRcvConn :: s -> ReceiveQueue -> m ()
createSndConn :: s -> SendQueue -> m ()
createRcvConn :: s -> RcvQueue -> m ()
createSndConn :: s -> SndQueue -> m ()
getConn :: s -> ConnAlias -> m SomeConn
getRcvQueue :: s -> SMPServer -> SMP.RecipientId -> m ReceiveQueue
getRcvQueue :: s -> SMPServer -> SMP.RecipientId -> m RcvQueue
deleteConn :: s -> ConnAlias -> m ()
upgradeRcvConnToDuplex :: s -> ConnAlias -> SendQueue -> m ()
upgradeSndConnToDuplex :: s -> ConnAlias -> ReceiveQueue -> m ()
setRcvQueueStatus :: s -> ReceiveQueue -> QueueStatus -> m ()
setSndQueueStatus :: s -> SendQueue -> QueueStatus -> m ()
upgradeRcvConnToDuplex :: s -> ConnAlias -> SndQueue -> m ()
upgradeSndConnToDuplex :: s -> ConnAlias -> RcvQueue -> m ()
setRcvQueueStatus :: s -> RcvQueue -> QueueStatus -> m ()
setSndQueueStatus :: s -> SndQueue -> QueueStatus -> m ()
-- Msg management
createRcvMsg :: s -> ConnAlias -> MsgBody -> ExternalSndId -> ExternalSndTs -> BrokerId -> BrokerTs -> m ()
@@ -47,7 +47,7 @@ class Monad m => MonadAgentStore s m where
-- * Queue types
-- | A receive queue. SMP queue through which the agent receives messages from a sender.
data ReceiveQueue = ReceiveQueue
data RcvQueue = RcvQueue
{ server :: SMPServer,
rcvId :: SMP.RecipientId,
connAlias :: ConnAlias,
@@ -61,7 +61,7 @@ data ReceiveQueue = ReceiveQueue
deriving (Eq, Show)
-- | A send queue. SMP queue through which the agent sends messages to a recipient.
data SendQueue = SendQueue
data SndQueue = SndQueue
{ server :: SMPServer,
sndId :: SMP.SenderId,
connAlias :: ConnAlias,
@@ -75,30 +75,30 @@ data SendQueue = SendQueue
-- * Connection types
-- | Type of a connection.
data ConnType = CReceive | CSend | CDuplex deriving (Eq, Show)
data ConnType = CRcv | CSnd | CDuplex deriving (Eq, Show)
-- | Connection of a specific type.
--
-- - ReceiveConnection is a connection that only has a receive queue set up,
-- - RcvConnection is a connection that only has a receive queue set up,
-- typically created by a recipient initiating a duplex connection.
--
-- - SendConnection is a connection that only has a send queue set up, typically
-- - SndConnection is a connection that only has a send queue set up, typically
-- created by a sender joining a duplex connection through a recipient's invitation.
--
-- - DuplexConnection is a connection that has both receive and send queues set up,
-- typically created by upgrading a receive or a send connection with a missing queue.
data Connection (d :: ConnType) where
ReceiveConnection :: ConnAlias -> ReceiveQueue -> Connection CReceive
SendConnection :: ConnAlias -> SendQueue -> Connection CSend
DuplexConnection :: ConnAlias -> ReceiveQueue -> SendQueue -> Connection CDuplex
RcvConnection :: ConnAlias -> RcvQueue -> Connection CRcv
SndConnection :: ConnAlias -> SndQueue -> Connection CSnd
DuplexConnection :: ConnAlias -> RcvQueue -> SndQueue -> Connection CDuplex
deriving instance Eq (Connection d)
deriving instance Show (Connection d)
data SConnType :: ConnType -> Type where
SCReceive :: SConnType CReceive
SCSend :: SConnType CSend
SCRcv :: SConnType CRcv
SCSnd :: SConnType CSnd
SCDuplex :: SConnType CDuplex
deriving instance Eq (SConnType d)
@@ -106,8 +106,8 @@ deriving instance Eq (SConnType d)
deriving instance Show (SConnType d)
instance TestEquality SConnType where
testEquality SCReceive SCReceive = Just Refl
testEquality SCSend SCSend = Just Refl
testEquality SCRcv SCRcv = Just Refl
testEquality SCSnd SCSnd = Just Refl
testEquality SCDuplex SCDuplex = Just Refl
testEquality _ _ = Nothing
@@ -137,7 +137,7 @@ data RcvMsg = RcvMsg
externalSndTs :: ExternalSndTs,
brokerId :: BrokerId,
brokerTs :: BrokerTs,
rcvStatus :: RcvStatus,
rcvMsgStatus :: RcvMsgStatus,
-- | Timestamp of acknowledgement to broker, corresponds to `AcknowledgedToBroker` status.
-- Do not mix up with `brokerTs` - timestamp created at broker after it receives the message from sender.
ackBrokerTs :: AckBrokerTs,
@@ -158,7 +158,7 @@ type BrokerId = MsgId
type BrokerTs = UTCTime
data RcvStatus
data RcvMsgStatus
= Received
| AcknowledgedToBroker
| AcknowledgedToSender
@@ -173,7 +173,7 @@ data SndMsg = SndMsg
{ msgBase :: MsgBase,
-- | Id of the message sent / to be sent, as in its number in order of sending.
internalSndId :: InternalSndId,
sndStatus :: SndStatus,
sndMsgStatus :: SndMsgStatus,
-- | Timestamp of the message received by broker, corresponds to `Sent` status.
sentTs :: SentTs,
-- | Timestamp of the message received by recipient, corresponds to `Delivered` status.
@@ -183,7 +183,7 @@ data SndMsg = SndMsg
type InternalSndId = Int64
data SndStatus
data SndMsgStatus
= Created
| Sent
| Delivered
@@ -210,6 +210,7 @@ type InternalTs = UTCTime
-- * Store errors
-- TODO revise
data StoreError
= SEInternal
| SENotFound
+47 -47
View File
@@ -53,12 +53,12 @@ newSQLiteStore dbFilename = do
return SQLiteStore {dbFilename, dbConn}
instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteStore m where
createRcvConn :: SQLiteStore -> ReceiveQueue -> m ()
createRcvConn :: SQLiteStore -> RcvQueue -> m ()
createRcvConn SQLiteStore {dbConn} rcvQueue =
liftIO $
createRcvQueueAndConn dbConn rcvQueue
createSndConn :: SQLiteStore -> SendQueue -> m ()
createSndConn :: SQLiteStore -> SndQueue -> m ()
createSndConn SQLiteStore {dbConn} sndQueue =
liftIO $
createSndQueueAndConn dbConn sndQueue
@@ -70,11 +70,11 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
retrieveConnQueues dbConn connAlias
case queues of
(Just rcvQ, Just sndQ) -> return $ SomeConn SCDuplex (DuplexConnection connAlias rcvQ sndQ)
(Just rcvQ, Nothing) -> return $ SomeConn SCReceive (ReceiveConnection connAlias rcvQ)
(Nothing, Just sndQ) -> return $ SomeConn SCSend (SendConnection connAlias sndQ)
(Just rcvQ, Nothing) -> return $ SomeConn SCRcv (RcvConnection connAlias rcvQ)
(Nothing, Just sndQ) -> return $ SomeConn SCSnd (SndConnection connAlias sndQ)
_ -> throwError SEBadConn
getRcvQueue :: SQLiteStore -> SMPServer -> SMP.RecipientId -> m ReceiveQueue
getRcvQueue :: SQLiteStore -> SMPServer -> SMP.RecipientId -> m RcvQueue
getRcvQueue SQLiteStore {dbConn} SMPServer {host, port} rcvId = do
rcvQueue <-
liftIO $
@@ -88,22 +88,22 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
liftIO $
deleteConnCascade dbConn connAlias
upgradeRcvConnToDuplex :: SQLiteStore -> ConnAlias -> SendQueue -> m ()
upgradeRcvConnToDuplex :: SQLiteStore -> ConnAlias -> SndQueue -> m ()
upgradeRcvConnToDuplex SQLiteStore {dbConn} connAlias sndQueue =
liftIOEither $
updateRcvConnWithSndQueue dbConn connAlias sndQueue
upgradeSndConnToDuplex :: SQLiteStore -> ConnAlias -> ReceiveQueue -> m ()
upgradeSndConnToDuplex :: SQLiteStore -> ConnAlias -> RcvQueue -> m ()
upgradeSndConnToDuplex SQLiteStore {dbConn} connAlias rcvQueue =
liftIOEither $
updateSndConnWithRcvQueue dbConn connAlias rcvQueue
setRcvQueueStatus :: SQLiteStore -> ReceiveQueue -> QueueStatus -> m ()
setRcvQueueStatus :: SQLiteStore -> RcvQueue -> QueueStatus -> m ()
setRcvQueueStatus SQLiteStore {dbConn} rcvQueue status =
liftIO $
updateRcvQueueStatus dbConn rcvQueue status
setSndQueueStatus :: SQLiteStore -> SendQueue -> QueueStatus -> m ()
setSndQueueStatus :: SQLiteStore -> SndQueue -> QueueStatus -> m ()
setSndQueueStatus SQLiteStore {dbConn} sndQueue status =
liftIO $
updateSndQueueStatus dbConn sndQueue status
@@ -135,9 +135,9 @@ instance ToField QueueStatus where toField = toField . show
instance FromField QueueStatus where fromField = fromFieldToReadable_
instance ToField RcvStatus where toField = toField . show
instance ToField RcvMsgStatus where toField = toField . show
instance ToField SndStatus where toField = toField . show
instance ToField SndMsgStatus where toField = toField . show
fromFieldToReadable_ :: forall a. (Read a, E.Typeable a) => Field -> Ok a
fromFieldToReadable_ = \case
@@ -177,15 +177,15 @@ upsertServer_ dbConn SMPServer {host, port, keyHash} = do
-- * createRcvConn helpers
createRcvQueueAndConn :: DB.Connection -> ReceiveQueue -> IO ()
createRcvQueueAndConn :: DB.Connection -> RcvQueue -> IO ()
createRcvQueueAndConn dbConn rcvQueue =
DB.withTransaction dbConn $ do
upsertServer_ dbConn (server (rcvQueue :: ReceiveQueue))
upsertServer_ dbConn (server (rcvQueue :: RcvQueue))
insertRcvQueue_ dbConn rcvQueue
insertRcvConnection_ dbConn rcvQueue
insertRcvQueue_ :: DB.Connection -> ReceiveQueue -> IO ()
insertRcvQueue_ dbConn ReceiveQueue {..} = do
insertRcvQueue_ :: DB.Connection -> RcvQueue -> IO ()
insertRcvQueue_ dbConn RcvQueue {..} = do
let port_ = serializePort_ $ port server
DB.executeNamed
dbConn
@@ -207,8 +207,8 @@ insertRcvQueue_ dbConn ReceiveQueue {..} = do
":status" := status
]
insertRcvConnection_ :: DB.Connection -> ReceiveQueue -> IO ()
insertRcvConnection_ dbConn ReceiveQueue {server, rcvId, connAlias} = do
insertRcvConnection_ :: DB.Connection -> RcvQueue -> IO ()
insertRcvConnection_ dbConn RcvQueue {server, rcvId, connAlias} = do
let port_ = serializePort_ $ port server
DB.executeNamed
dbConn
@@ -224,15 +224,15 @@ insertRcvConnection_ dbConn ReceiveQueue {server, rcvId, connAlias} = do
-- * createSndConn helpers
createSndQueueAndConn :: DB.Connection -> SendQueue -> IO ()
createSndQueueAndConn :: DB.Connection -> SndQueue -> IO ()
createSndQueueAndConn dbConn sndQueue =
DB.withTransaction dbConn $ do
upsertServer_ dbConn (server (sndQueue :: SendQueue))
upsertServer_ dbConn (server (sndQueue :: SndQueue))
insertSndQueue_ dbConn sndQueue
insertSndConnection_ dbConn sndQueue
insertSndQueue_ :: DB.Connection -> SendQueue -> IO ()
insertSndQueue_ dbConn SendQueue {..} = do
insertSndQueue_ :: DB.Connection -> SndQueue -> IO ()
insertSndQueue_ dbConn SndQueue {..} = do
let port_ = serializePort_ $ port server
DB.executeNamed
dbConn
@@ -252,8 +252,8 @@ insertSndQueue_ dbConn SendQueue {..} = do
":status" := status
]
insertSndConnection_ :: DB.Connection -> SendQueue -> IO ()
insertSndConnection_ dbConn SendQueue {server, sndId, connAlias} = do
insertSndConnection_ :: DB.Connection -> SndQueue -> IO ()
insertSndConnection_ dbConn SndQueue {server, sndId, connAlias} = do
let port_ = serializePort_ $ port server
DB.executeNamed
dbConn
@@ -269,7 +269,7 @@ insertSndConnection_ dbConn SendQueue {server, sndId, connAlias} = do
-- * getConn helpers
retrieveConnQueues :: DB.Connection -> ConnAlias -> IO (Maybe ReceiveQueue, Maybe SendQueue)
retrieveConnQueues :: DB.Connection -> ConnAlias -> IO (Maybe RcvQueue, Maybe SndQueue)
retrieveConnQueues dbConn connAlias =
DB.withTransaction -- Avoid inconsistent state between queue reads
dbConn
@@ -277,13 +277,13 @@ retrieveConnQueues dbConn connAlias =
-- Separate transactionless version of retrieveConnQueues to be reused in other functions that already wrap
-- multiple statements in transaction - otherwise they'd be attempting to start a transaction within a transaction
retrieveConnQueues_ :: DB.Connection -> ConnAlias -> IO (Maybe ReceiveQueue, Maybe SendQueue)
retrieveConnQueues_ :: DB.Connection -> ConnAlias -> IO (Maybe RcvQueue, Maybe SndQueue)
retrieveConnQueues_ dbConn connAlias = do
rcvQ <- retrieveRcvQueueByConnAlias_ dbConn connAlias
sndQ <- retrieveSndQueueByConnAlias_ dbConn connAlias
return (rcvQ, sndQ)
retrieveRcvQueueByConnAlias_ :: DB.Connection -> ConnAlias -> IO (Maybe ReceiveQueue)
retrieveRcvQueueByConnAlias_ :: DB.Connection -> ConnAlias -> IO (Maybe RcvQueue)
retrieveRcvQueueByConnAlias_ dbConn connAlias = do
r <-
DB.queryNamed
@@ -300,10 +300,10 @@ retrieveRcvQueueByConnAlias_ dbConn connAlias = do
case r of
[(keyHash, host, port, rcvId, cAlias, rcvPrivateKey, sndId, sndKey, decryptKey, verifyKey, status)] -> do
let srv = SMPServer host (deserializePort_ port) keyHash
return . Just $ ReceiveQueue srv rcvId cAlias rcvPrivateKey sndId sndKey decryptKey verifyKey status
return . Just $ RcvQueue srv rcvId cAlias rcvPrivateKey sndId sndKey decryptKey verifyKey status
_ -> return Nothing
retrieveSndQueueByConnAlias_ :: DB.Connection -> ConnAlias -> IO (Maybe SendQueue)
retrieveSndQueueByConnAlias_ :: DB.Connection -> ConnAlias -> IO (Maybe SndQueue)
retrieveSndQueueByConnAlias_ dbConn connAlias = do
r <-
DB.queryNamed
@@ -320,12 +320,12 @@ retrieveSndQueueByConnAlias_ dbConn connAlias = do
case r of
[(keyHash, host, port, sndId, cAlias, sndPrivateKey, encryptKey, signKey, status)] -> do
let srv = SMPServer host (deserializePort_ port) keyHash
return . Just $ SendQueue srv sndId cAlias sndPrivateKey encryptKey signKey status
return . Just $ SndQueue srv sndId cAlias sndPrivateKey encryptKey signKey status
_ -> return Nothing
-- * getRcvQueue helper
retrieveRcvQueue :: DB.Connection -> HostName -> Maybe ServiceName -> SMP.RecipientId -> IO (Maybe ReceiveQueue)
retrieveRcvQueue :: DB.Connection -> HostName -> Maybe ServiceName -> SMP.RecipientId -> IO (Maybe RcvQueue)
retrieveRcvQueue dbConn host port rcvId = do
r <-
DB.queryNamed
@@ -342,7 +342,7 @@ retrieveRcvQueue dbConn host port rcvId = do
case r of
[(keyHash, hst, prt, rId, connAlias, rcvPrivateKey, sndId, sndKey, decryptKey, verifyKey, status)] -> do
let srv = SMPServer hst (deserializePort_ prt) keyHash
return . Just $ ReceiveQueue srv rId connAlias rcvPrivateKey sndId sndKey decryptKey verifyKey status
return . Just $ RcvQueue srv rId connAlias rcvPrivateKey sndId sndKey decryptKey verifyKey status
_ -> return Nothing
-- * deleteConn helper
@@ -356,22 +356,22 @@ deleteConnCascade dbConn connAlias =
-- * upgradeRcvConnToDuplex helpers
updateRcvConnWithSndQueue :: DB.Connection -> ConnAlias -> SendQueue -> IO (Either StoreError ())
updateRcvConnWithSndQueue :: DB.Connection -> ConnAlias -> SndQueue -> IO (Either StoreError ())
updateRcvConnWithSndQueue dbConn connAlias sndQueue =
DB.withTransaction dbConn $ do
queues <- retrieveConnQueues_ dbConn connAlias
case queues of
(Just _rcvQ, Nothing) -> do
upsertServer_ dbConn (server (sndQueue :: SendQueue))
upsertServer_ dbConn (server (sndQueue :: SndQueue))
insertSndQueue_ dbConn sndQueue
updateConnWithSndQueue_ dbConn connAlias sndQueue
return $ Right ()
(Nothing, Just _sndQ) -> return $ Left (SEBadConnType CSend)
(Nothing, Just _sndQ) -> return $ Left (SEBadConnType CSnd)
(Just _rcvQ, Just _sndQ) -> return $ Left (SEBadConnType CDuplex)
_ -> return $ Left SEBadConn
updateConnWithSndQueue_ :: DB.Connection -> ConnAlias -> SendQueue -> IO ()
updateConnWithSndQueue_ dbConn connAlias SendQueue {server, sndId} = do
updateConnWithSndQueue_ :: DB.Connection -> ConnAlias -> SndQueue -> IO ()
updateConnWithSndQueue_ dbConn connAlias SndQueue {server, sndId} = do
let port_ = serializePort_ $ port server
DB.executeNamed
dbConn
@@ -384,22 +384,22 @@ updateConnWithSndQueue_ dbConn connAlias SendQueue {server, sndId} = do
-- * upgradeSndConnToDuplex helpers
updateSndConnWithRcvQueue :: DB.Connection -> ConnAlias -> ReceiveQueue -> IO (Either StoreError ())
updateSndConnWithRcvQueue :: DB.Connection -> ConnAlias -> RcvQueue -> IO (Either StoreError ())
updateSndConnWithRcvQueue dbConn connAlias rcvQueue =
DB.withTransaction dbConn $ do
queues <- retrieveConnQueues_ dbConn connAlias
case queues of
(Nothing, Just _sndQ) -> do
upsertServer_ dbConn (server (rcvQueue :: ReceiveQueue))
upsertServer_ dbConn (server (rcvQueue :: RcvQueue))
insertRcvQueue_ dbConn rcvQueue
updateConnWithRcvQueue_ dbConn connAlias rcvQueue
return $ Right ()
(Just _rcvQ, Nothing) -> return $ Left (SEBadConnType CReceive)
(Just _rcvQ, Nothing) -> return $ Left (SEBadConnType CRcv)
(Just _rcvQ, Just _sndQ) -> return $ Left (SEBadConnType CDuplex)
_ -> return $ Left SEBadConn
updateConnWithRcvQueue_ :: DB.Connection -> ConnAlias -> ReceiveQueue -> IO ()
updateConnWithRcvQueue_ dbConn connAlias ReceiveQueue {server, rcvId} = do
updateConnWithRcvQueue_ :: DB.Connection -> ConnAlias -> RcvQueue -> IO ()
updateConnWithRcvQueue_ dbConn connAlias RcvQueue {server, rcvId} = do
let port_ = serializePort_ $ port server
DB.executeNamed
dbConn
@@ -413,8 +413,8 @@ updateConnWithRcvQueue_ dbConn connAlias ReceiveQueue {server, rcvId} = do
-- * setRcvQueueStatus helper
-- ? throw error if queue doesn't exist?
updateRcvQueueStatus :: DB.Connection -> ReceiveQueue -> QueueStatus -> IO ()
updateRcvQueueStatus dbConn ReceiveQueue {rcvId, server = SMPServer {host, port}} status =
updateRcvQueueStatus :: DB.Connection -> RcvQueue -> QueueStatus -> IO ()
updateRcvQueueStatus dbConn RcvQueue {rcvId, server = SMPServer {host, port}} status =
DB.executeNamed
dbConn
[sql|
@@ -427,8 +427,8 @@ updateRcvQueueStatus dbConn ReceiveQueue {rcvId, server = SMPServer {host, port}
-- * setSndQueueStatus helper
-- ? throw error if queue doesn't exist?
updateSndQueueStatus :: DB.Connection -> SendQueue -> QueueStatus -> IO ()
updateSndQueueStatus dbConn SendQueue {sndId, server = SMPServer {host, port}} status =
updateSndQueueStatus :: DB.Connection -> SndQueue -> QueueStatus -> IO ()
updateSndQueueStatus dbConn SndQueue {sndId, server = SMPServer {host, port}} status =
DB.executeNamed
dbConn
[sql|
@@ -461,7 +461,7 @@ insertRcvMsg dbConn connAlias msgBody externalSndId externalSndTs brokerId broke
insertRcvMsgDetails_ dbConn connAlias internalRcvId internalId externalSndId externalSndTs brokerId brokerTs
updateLastInternalIdsRcv_ dbConn connAlias internalId internalRcvId
return $ Right ()
(Nothing, Just _sndQ) -> return $ Left (SEBadConnType CSend)
(Nothing, Just _sndQ) -> return $ Left (SEBadConnType CSnd)
_ -> return $ Left SEBadConn
retrieveLastInternalIdsRcv_ :: DB.Connection -> ConnAlias -> IO (InternalId, InternalRcvId)
@@ -555,7 +555,7 @@ insertSndMsg dbConn connAlias msgBody =
insertSndMsgDetails_ dbConn connAlias internalSndId internalId
updateLastInternalIdsSnd_ dbConn connAlias internalId internalSndId
return $ Right ()
(Just _rcvQ, Nothing) -> return $ Left (SEBadConnType CReceive)
(Just _rcvQ, Nothing) -> return $ Left (SEBadConnType CRcv)
_ -> return $ Left SEBadConn
retrieveLastInternalIdsSnd_ :: DB.Connection -> ConnAlias -> IO (InternalId, InternalSndId)