mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-07-09 10:11:58 +00:00
Merge branch 'master' into ep/rfc-rotation
This commit is contained in:
Evgeny Poberezkin
@@ -53,6 +53,7 @@ library
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220811_onion_hosts
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220817_connection_ntfs
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220822_queue_rotation
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands
|
||||
Simplex.Messaging.Client
|
||||
Simplex.Messaging.Client.Agent
|
||||
Simplex.Messaging.Crypto
|
||||
|
||||
+201
-30
@@ -12,7 +12,6 @@
|
||||
{-# LANGUAGE RankNTypes #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# LANGUAGE TupleSections #-}
|
||||
{-# LANGUAGE TypeApplications #-}
|
||||
|
||||
-- |
|
||||
-- Module : Simplex.Messaging.Agent
|
||||
@@ -39,6 +38,10 @@ module Simplex.Messaging.Agent
|
||||
disconnectAgentClient,
|
||||
resumeAgentClient,
|
||||
withAgentLock,
|
||||
createConnectionAsync,
|
||||
joinConnectionAsync,
|
||||
allowConnectionAsync,
|
||||
ackMessageAsync,
|
||||
createConnection,
|
||||
joinConnection,
|
||||
allowConnection,
|
||||
@@ -144,13 +147,29 @@ resumeAgentClient c = atomically $ writeTVar (active c) True
|
||||
-- |
|
||||
type AgentErrorMonad m = (MonadUnliftIO m, MonadError AgentErrorType m)
|
||||
|
||||
-- | Create SMP agent connection (NEW command) asynchronously, synchronous response is new connection id
|
||||
createConnectionAsync :: forall m c. (AgentErrorMonad m, ConnectionModeI c) => AgentClient -> Bool -> SConnectionMode c -> m ConnId
|
||||
createConnectionAsync c enableNtfs cMode = withAgentEnv c $ newConnAsync c enableNtfs cMode
|
||||
|
||||
-- | Join SMP agent connection (JOIN command) asynchronously, synchronous response is new connection id
|
||||
joinConnectionAsync :: AgentErrorMonad m => AgentClient -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
joinConnectionAsync c enableNtfs = withAgentEnv c .: joinConnAsync c enableNtfs
|
||||
|
||||
-- | Allow connection to continue after CONF notification (LET command), no synchronous response
|
||||
allowConnectionAsync :: AgentErrorMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
|
||||
allowConnectionAsync c = withAgentEnv c .:. allowConnectionAsync' c
|
||||
|
||||
-- | Acknowledge message (ACK command) asynchronously, no synchronous response
|
||||
ackMessageAsync :: forall m. AgentErrorMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
|
||||
ackMessageAsync c = withAgentEnv c .: ackMessageAsync' c
|
||||
|
||||
-- | Create SMP agent connection (NEW command)
|
||||
createConnection :: AgentErrorMonad m => AgentClient -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
|
||||
createConnection c enableNtfs cMode = withAgentEnv c $ newConn c "" enableNtfs cMode
|
||||
createConnection c enableNtfs cMode = withAgentEnv c $ newConn c "" False enableNtfs cMode
|
||||
|
||||
-- | Join SMP agent connection (JOIN command)
|
||||
joinConnection :: AgentErrorMonad m => AgentClient -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
joinConnection c enableNtfs = withAgentEnv c .: joinConn c "" enableNtfs
|
||||
joinConnection c enableNtfs = withAgentEnv c .: joinConn c "" False enableNtfs
|
||||
|
||||
-- | Allow connection to continue after CONF notification (LET command)
|
||||
allowConnection :: AgentErrorMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
|
||||
@@ -290,8 +309,8 @@ client c@AgentClient {rcvQ, subQ} = forever $ do
|
||||
-- | execute any SMP agent command
|
||||
processCommand :: forall m. AgentMonad m => AgentClient -> (ConnId, ACommand 'Client) -> m (ConnId, ACommand 'Agent)
|
||||
processCommand c (connId, cmd) = case cmd of
|
||||
NEW (ACM cMode) -> second (INV . ACR cMode) <$> newConn c connId True cMode
|
||||
JOIN (ACR _ cReq) connInfo -> (,OK) <$> joinConn c connId True cReq connInfo
|
||||
NEW enableNtfs (ACM cMode) -> second (INV . ACR cMode) <$> newConn c connId False enableNtfs cMode
|
||||
JOIN enableNtfs (ACR _ cReq) connInfo -> (,OK) <$> joinConn c connId False enableNtfs cReq connInfo
|
||||
LET confId ownCInfo -> allowConnection' c connId confId ownCInfo $> (connId, OK)
|
||||
ACPT invId ownCInfo -> (,OK) <$> acceptContact' c connId True invId ownCInfo
|
||||
RJCT invId -> rejectContact' c connId invId $> (connId, OK)
|
||||
@@ -302,15 +321,56 @@ processCommand c (connId, cmd) = case cmd of
|
||||
DEL -> deleteConnection' c connId $> (connId, OK)
|
||||
CHK -> (connId,) . STAT <$> getConnectionServers' c connId
|
||||
|
||||
newConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
|
||||
newConn c connId enableNtfs cMode = do
|
||||
newConnAsync :: forall m c. (AgentMonad m, ConnectionModeI c) => AgentClient -> Bool -> SConnectionMode c -> m ConnId
|
||||
newConnAsync c enableNtfs cMode = do
|
||||
g <- asks idsDrg
|
||||
connAgentVersion <- asks $ maxVersion . smpAgentVRange . config
|
||||
let cData = ConnData {connId = "", connAgentVersion, enableNtfs, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent
|
||||
connId <- withStore c $ \db -> createNewConn db g cData cMode
|
||||
enqueueCommand c connId Nothing $ NEW enableNtfs (ACM cMode)
|
||||
pure connId
|
||||
|
||||
joinConnAsync :: AgentMonad m => AgentClient -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
joinConnAsync c enableNtfs cReqUri@(CRInvitationUri (ConnReqUriData _ agentVRange _) _) cInfo = do
|
||||
aVRange <- asks $ smpAgentVRange . config
|
||||
case agentVRange `compatibleVersion` aVRange of
|
||||
Just (Compatible connAgentVersion) -> do
|
||||
g <- asks idsDrg
|
||||
let duplexHS = connAgentVersion /= 1
|
||||
cData = ConnData {connId = "", connAgentVersion, enableNtfs, duplexHandshake = Just duplexHS}
|
||||
connId <- withStore c $ \db -> createNewConn db g cData SCMInvitation
|
||||
enqueueCommand c connId Nothing $ JOIN enableNtfs (ACR sConnectionMode cReqUri) cInfo
|
||||
pure connId
|
||||
_ -> throwError $ AGENT A_VERSION
|
||||
joinConnAsync _c _enableNtfs (CRContactUri _) _cInfo =
|
||||
throwError $ CMD PROHIBITED
|
||||
|
||||
allowConnectionAsync' :: AgentMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
|
||||
allowConnectionAsync' c connId confId ownConnInfo =
|
||||
withStore c (`getConn` connId) >>= \case
|
||||
SomeConn _ (RcvConnection _ RcvQueue {server}) ->
|
||||
enqueueCommand c connId (Just server) $ LET confId ownConnInfo
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
|
||||
ackMessageAsync' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
|
||||
ackMessageAsync' c connId msgId =
|
||||
withStore c (`getConn` connId) >>= \case
|
||||
SomeConn _ (DuplexConnection _ rq _ _ _) -> enqueueAck rq
|
||||
SomeConn _ (RcvConnection _ rq) -> enqueueAck rq
|
||||
SomeConn _ (SndConnection _ _) -> throwError $ CONN SIMPLEX
|
||||
SomeConn _ (ContactConnection _ _) -> throwError $ CMD PROHIBITED
|
||||
SomeConn _ (NewConnection _) -> throwError $ CMD PROHIBITED
|
||||
where
|
||||
enqueueAck :: RcvQueue -> m ()
|
||||
enqueueAck RcvQueue {server} = do
|
||||
enqueueCommand c connId (Just server) $ ACK msgId
|
||||
|
||||
newConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
|
||||
newConn c connId asyncMode enableNtfs cMode = do
|
||||
srv <- getAnySMPServer c
|
||||
clientVRange <- asks $ smpClientVRange . config
|
||||
(rq, qUri) <- newRcvQueue c srv clientVRange True
|
||||
g <- asks idsDrg
|
||||
connAgentVersion <- asks $ maxVersion . smpAgentVRange . config
|
||||
let cData = ConnData {connId, connAgentVersion, enableNtfs, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent
|
||||
connId' <- withStore c $ \db -> createRcvConn db g cData rq cMode
|
||||
connId' <- setUpConn asyncMode rq
|
||||
addSubscription c rq connId'
|
||||
when enableNtfs $ do
|
||||
ns <- asks ntfSupervisor
|
||||
@@ -323,9 +383,18 @@ newConn c connId enableNtfs cMode = do
|
||||
(pk1, pk2, e2eRcvParams) <- liftIO $ CR.generateE2EParams CR.e2eEncryptVersion
|
||||
withStore' c $ \db -> createRatchetX3dhKeys db connId' pk1 pk2
|
||||
pure (connId', CRInvitationUri crData $ toVersionRangeT e2eRcvParams CR.e2eEncryptVRange)
|
||||
where
|
||||
setUpConn True rq = do
|
||||
withStore c $ \db -> updateNewConnRcv db connId rq
|
||||
pure connId
|
||||
setUpConn False rq = do
|
||||
g <- asks idsDrg
|
||||
connAgentVersion <- asks $ maxVersion . smpAgentVRange . config
|
||||
let cData = ConnData {connId, connAgentVersion, enableNtfs, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent
|
||||
withStore c $ \db -> createRcvConn db g cData rq cMode
|
||||
|
||||
joinConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
joinConn c connId enableNtfs (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2eRcvParamsUri) cInfo = do
|
||||
joinConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
joinConn c connId asyncMode enableNtfs (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2eRcvParamsUri) cInfo = do
|
||||
aVRange <- asks $ smpAgentVRange . config
|
||||
clientVRange <- asks $ smpClientVRange . config
|
||||
case ( qUri `compatibleVersion` clientVRange,
|
||||
@@ -337,13 +406,9 @@ joinConn c connId enableNtfs (CRInvitationUri (ConnReqUriData _ agentVRange (qUr
|
||||
(_, rcDHRs) <- liftIO C.generateKeyPair'
|
||||
let rc = CR.initSndRatchet rcDHRr rcDHRs $ CR.x3dhSnd pk1 pk2 e2eRcvParams
|
||||
sq <- newSndQueue qInfo True
|
||||
g <- asks idsDrg
|
||||
let duplexHS = connAgentVersion /= 1
|
||||
cData = ConnData {connId, connAgentVersion, enableNtfs, duplexHandshake = Just duplexHS}
|
||||
connId' <- withStore c $ \db -> runExceptT $ do
|
||||
connId' <- ExceptT $ createSndConn db g cData sq
|
||||
liftIO $ createRatchet db connId' rc
|
||||
pure connId'
|
||||
connId' <- setUpConn asyncMode cData sq rc
|
||||
let cData' = (cData :: ConnData) {connId = connId'}
|
||||
tryError (confirmQueue aVersion c cData' sq cInfo $ Just e2eSndParams) >>= \case
|
||||
Right _ -> do
|
||||
@@ -351,20 +416,34 @@ joinConn c connId enableNtfs (CRInvitationUri (ConnReqUriData _ agentVRange (qUr
|
||||
pure connId'
|
||||
Left e -> do
|
||||
-- TODO recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md
|
||||
withStore' c (`deleteConn` connId')
|
||||
unless asyncMode $ withStore' c (`deleteConn` connId')
|
||||
throwError e
|
||||
where
|
||||
setUpConn True _ sq rc =
|
||||
withStore c $ \db -> runExceptT $ do
|
||||
ExceptT $ updateNewConnSnd db connId sq
|
||||
liftIO $ createRatchet db connId rc
|
||||
pure connId
|
||||
setUpConn False cData sq rc = do
|
||||
g <- asks idsDrg
|
||||
withStore c $ \db -> runExceptT $ do
|
||||
connId' <- ExceptT $ createSndConn db g cData sq
|
||||
liftIO $ createRatchet db connId' rc
|
||||
pure connId'
|
||||
_ -> throwError $ AGENT A_VERSION
|
||||
joinConn c connId enableNtfs (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInfo = do
|
||||
joinConn c connId False enableNtfs (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInfo = do
|
||||
aVRange <- asks $ smpAgentVRange . config
|
||||
clientVRange <- asks $ smpClientVRange . config
|
||||
case ( qUri `compatibleVersion` clientVRange,
|
||||
agentVRange `compatibleVersion` aVRange
|
||||
) of
|
||||
(Just qInfo, Just vrsn) -> do
|
||||
(connId', cReq) <- newConn c connId enableNtfs SCMInvitation
|
||||
(connId', cReq) <- newConn c connId False enableNtfs SCMInvitation
|
||||
sendInvitation c qInfo vrsn cReq cInfo
|
||||
pure connId'
|
||||
_ -> throwError $ AGENT A_VERSION
|
||||
joinConn _c _connId True _enableNtfs (CRContactUri _) _cInfo = do
|
||||
throwError $ CMD PROHIBITED
|
||||
|
||||
createReplyQueue :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> m SMPQueueInfo
|
||||
createReplyQueue c ConnData {connId, enableNtfs} SndQueue {server, smpClientVersion} = do
|
||||
@@ -398,7 +477,7 @@ acceptContact' c connId enableNtfs invId ownConnInfo = do
|
||||
withStore c (`getConn` contactConnId) >>= \case
|
||||
SomeConn _ ContactConnection {} -> do
|
||||
withStore' c $ \db -> acceptInvitation db invId ownConnInfo
|
||||
joinConn c connId enableNtfs connReq ownConnInfo `catchError` \err -> do
|
||||
joinConn c connId False enableNtfs connReq ownConnInfo `catchError` \err -> do
|
||||
withStore' c (`unacceptInvitation` invId)
|
||||
throwError err
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
@@ -425,14 +504,17 @@ subscribeConnection' c connId =
|
||||
resumeMsgDelivery c cData sq
|
||||
subscribe rq
|
||||
void . forkIO $ doRcvQueueAction c cData rq sq
|
||||
resumeConnCmds c connId
|
||||
SomeConn _ (SndConnection cData sq) -> do
|
||||
resumeMsgDelivery c cData sq
|
||||
case status (sq :: SndQueue) of
|
||||
Confirmed -> pure () -- TODO secure queue if this is a new server version
|
||||
Active -> throwError $ CONN SIMPLEX
|
||||
_ -> throwError $ INTERNAL "unexpected queue status"
|
||||
SomeConn _ (RcvConnection _ rq) -> subscribe rq
|
||||
SomeConn _ (ContactConnection _ rq) -> subscribe rq
|
||||
resumeConnCmds c connId
|
||||
SomeConn _ (RcvConnection _ rq) -> subscribe rq >> resumeConnCmds c connId
|
||||
SomeConn _ (ContactConnection _ rq) -> subscribe rq >> resumeConnCmds c connId
|
||||
SomeConn _ (NewConnection _) -> resumeConnCmds c connId
|
||||
where
|
||||
-- TODO sndQueueAction?
|
||||
subscribe :: RcvQueue -> m ()
|
||||
@@ -514,15 +596,15 @@ subscribeConnections' c connIds = do
|
||||
conns :: Map ConnId (Either StoreError SomeConn) <- M.fromList . zip connIds <$> withStore' c (forM connIds . getConn)
|
||||
let (errs, cs) = M.mapEither id conns
|
||||
errs' = M.map (Left . storeError) errs
|
||||
(sndQs, rcvQs) = M.mapEither rcvOrSndQueue cs
|
||||
sndRs = M.map (sndSubResult . fst) sndQs
|
||||
(subRs, rcvQs) = M.mapEither rcvQueueOrResult cs
|
||||
srvRcvQs :: Map SMPServer (Map ConnId (RcvQueue, ConnData)) = M.foldlWithKey' addRcvQueue M.empty rcvQs
|
||||
mapM_ (mapM_ (uncurry $ resumeMsgDelivery c) . sndQueue) cs
|
||||
forM_ (M.keys cs) $ resumeConnCmds c
|
||||
rcvRs <- mapConcurrently subscribe (M.assocs srvRcvQs)
|
||||
ns <- asks ntfSupervisor
|
||||
tkn <- readTVarIO (ntfTkn ns)
|
||||
when (instantNotifications tkn) . void . forkIO $ sendNtfCreate ns rcvRs
|
||||
let rs = M.unions $ errs' : sndRs : rcvRs
|
||||
let rs = M.unions $ errs' : subRs : rcvRs
|
||||
notifyResultError rs
|
||||
void . forkIO . forM_ cs $ \case
|
||||
SomeConn _ (DuplexConnection cData rq sq _ _) -> doRcvQueueAction c cData rq sq
|
||||
@@ -530,12 +612,13 @@ subscribeConnections' c connIds = do
|
||||
-- TODO secure Confirmed queues if this is a new server version
|
||||
pure rs
|
||||
where
|
||||
rcvOrSndQueue :: SomeConn -> Either (SndQueue, ConnData) (RcvQueue, ConnData)
|
||||
rcvOrSndQueue = \case
|
||||
rcvQueueOrResult :: SomeConn -> Either (Either AgentErrorType ()) (RcvQueue, ConnData)
|
||||
rcvQueueOrResult = \case
|
||||
SomeConn _ (DuplexConnection cData rq _ _ _) -> Right (rq, cData)
|
||||
SomeConn _ (SndConnection cData sq) -> Left (sq, cData)
|
||||
SomeConn _ (SndConnection _ sq) -> Left $ sndSubResult sq
|
||||
SomeConn _ (RcvConnection cData rq) -> Right (rq, cData)
|
||||
SomeConn _ (ContactConnection cData rq) -> Right (rq, cData)
|
||||
SomeConn _ (NewConnection _) -> Left (Right ())
|
||||
sndSubResult :: SndQueue -> Either AgentErrorType ()
|
||||
sndSubResult sq = case status (sq :: SndQueue) of
|
||||
Confirmed -> Right ()
|
||||
@@ -584,6 +667,7 @@ getConnectionMessage' c connId = do
|
||||
SomeConn _ (RcvConnection _ rq) -> getQueueMessage c rq
|
||||
SomeConn _ (ContactConnection _ rq) -> getQueueMessage c rq
|
||||
SomeConn _ SndConnection {} -> throwError $ CONN SIMPLEX
|
||||
SomeConn _ NewConnection {} -> throwError $ CMD PROHIBITED
|
||||
|
||||
getNotificationMessage' :: forall m. AgentMonad m => AgentClient -> C.CbNonce -> ByteString -> m (NotificationInfo, [SMPMsgMeta])
|
||||
getNotificationMessage' c nonce encNtfInfo = do
|
||||
@@ -623,6 +707,89 @@ sendMessage' c connId msgFlags msg =
|
||||
enqueueMsg :: ConnData -> SndQueue -> m AgentMsgId
|
||||
enqueueMsg cData sq = enqueueMessage c cData sq msgFlags $ A_MSG msg
|
||||
|
||||
-- / async command processing v v v
|
||||
|
||||
enqueueCommand :: forall m. AgentMonad m => AgentClient -> ConnId -> Maybe SMPServer -> ACommand 'Client -> m ()
|
||||
enqueueCommand c connId server aCommand = do
|
||||
resumeSrvCmds c server
|
||||
commandId <- withStore c $ \db -> runExceptT . liftIO $ createCommand db connId server aCommand
|
||||
queuePendingCommands c server [commandId]
|
||||
|
||||
resumeSrvCmds :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m ()
|
||||
resumeSrvCmds c server =
|
||||
unlessM (cmdProcessExists c server) $
|
||||
async (runCommandProcessing c server)
|
||||
>>= \a -> atomically (TM.insert server a $ asyncCmdProcesses c)
|
||||
|
||||
resumeConnCmds :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
|
||||
resumeConnCmds c connId =
|
||||
unlessM connQueued $
|
||||
withStore' c (`getPendingCommands` connId)
|
||||
>>= mapM_ (uncurry enqueueSrvCmds)
|
||||
where
|
||||
enqueueSrvCmds srv cmdIds = unlessM (cmdProcessExists c srv) $ do
|
||||
a <- async (runCommandProcessing c srv)
|
||||
atomically (TM.insert srv a $ asyncCmdProcesses c)
|
||||
queuePendingCommands c srv cmdIds
|
||||
connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connCmdsQueued c)
|
||||
|
||||
cmdProcessExists :: AgentMonad m => AgentClient -> Maybe SMPServer -> m Bool
|
||||
cmdProcessExists c srv = atomically $ TM.member srv (asyncCmdProcesses c)
|
||||
|
||||
queuePendingCommands :: AgentMonad m => AgentClient -> Maybe SMPServer -> [AsyncCmdId] -> m ()
|
||||
queuePendingCommands c server cmdIds = atomically $ do
|
||||
q <- getPendingCommandQ c server
|
||||
mapM_ (writeTQueue q) cmdIds
|
||||
|
||||
getPendingCommandQ :: AgentClient -> Maybe SMPServer -> STM (TQueue AsyncCmdId)
|
||||
getPendingCommandQ c server = do
|
||||
maybe newMsgQueue pure =<< TM.lookup server (asyncCmdQueues c)
|
||||
where
|
||||
newMsgQueue = do
|
||||
cq <- newTQueue
|
||||
TM.insert server cq $ asyncCmdQueues c
|
||||
pure cq
|
||||
|
||||
runCommandProcessing :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m ()
|
||||
runCommandProcessing c@AgentClient {subQ} server = do
|
||||
cq <- atomically $ getPendingCommandQ c server
|
||||
ri <- asks $ messageRetryInterval . config -- different retry interval?
|
||||
forever $ do
|
||||
atomically $ endAgentOperation c AOSndNetwork
|
||||
cmdId <- atomically $ readTQueue cq
|
||||
atomically $ beginAgentOperation c AOSndNetwork
|
||||
E.try (withStore c $ \db -> getPendingCommand db cmdId) >>= \case
|
||||
Left (e :: E.SomeException) ->
|
||||
notify "" $ ERR (INTERNAL $ show e)
|
||||
Right (connId, ACmd _ cmd) ->
|
||||
withRetryInterval ri $ \loop -> do
|
||||
resp <- tryError $ case cmd of
|
||||
NEW enableNtfs (ACM cMode) -> do
|
||||
(_, cReq) <- newConn c connId True enableNtfs cMode
|
||||
notify connId $ INV (ACR cMode cReq)
|
||||
JOIN enableNtfs (ACR _ cReq) connInfo -> void $ joinConn c connId True enableNtfs cReq connInfo
|
||||
LET confId ownCInfo -> allowConnection' c connId confId ownCInfo
|
||||
ACK msgId -> ackMessage' c connId msgId
|
||||
_ -> notify "" $ ERR (INTERNAL "")
|
||||
case resp of
|
||||
Left _ ->
|
||||
-- TODO retry NEW and JOIN on different server
|
||||
-- TODO depending on command, some errors shouldn't be retried
|
||||
retryCommand loop
|
||||
Right () -> do
|
||||
delCmd cmdId
|
||||
where
|
||||
delCmd :: AsyncCmdId -> m ()
|
||||
delCmd cmdId = withStore' c $ \db -> deleteCommand db cmdId
|
||||
notify :: ConnId -> ACommand 'Agent -> m ()
|
||||
notify connId cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
|
||||
retryCommand loop = do
|
||||
-- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
|
||||
atomically $ endAgentOperation c AOSndNetwork
|
||||
atomically $ beginAgentOperation c AOSndNetwork
|
||||
loop
|
||||
-- ^ ^ ^ async command processing /
|
||||
|
||||
enqueueMessage :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> MsgFlags -> AMessage -> m AgentMsgId
|
||||
enqueueMessage c cData@ConnData {connId, connAgentVersion} sq msgFlags aMessage = do
|
||||
resumeMsgDelivery c cData sq
|
||||
@@ -803,6 +970,7 @@ ackMessage' c connId msgId = do
|
||||
SomeConn _ (RcvConnection _ rq) -> ack rq
|
||||
SomeConn _ (SndConnection _ _) -> throwError $ CONN SIMPLEX
|
||||
SomeConn _ (ContactConnection _ _) -> throwError $ CMD PROHIBITED
|
||||
SomeConn _ (NewConnection _) -> throwError $ CMD PROHIBITED
|
||||
where
|
||||
ack :: RcvQueue -> m ()
|
||||
ack rq = do
|
||||
@@ -832,6 +1000,7 @@ suspendConnection' c connId =
|
||||
SomeConn _ (RcvConnection _ rq) -> suspendQueue c rq
|
||||
SomeConn _ (ContactConnection _ rq) -> suspendQueue c rq
|
||||
SomeConn _ (SndConnection _ _) -> throwError $ CONN SIMPLEX
|
||||
SomeConn _ (NewConnection _) -> throwError $ CMD PROHIBITED
|
||||
|
||||
-- | Delete SMP agent connection (DEL command) in Reader monad
|
||||
deleteConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
|
||||
@@ -841,6 +1010,7 @@ deleteConnection' c connId =
|
||||
SomeConn _ (RcvConnection _ rq) -> delete rq
|
||||
SomeConn _ (ContactConnection _ rq) -> delete rq
|
||||
SomeConn _ (SndConnection _ _) -> withStore' c (`deleteConn` connId)
|
||||
SomeConn _ (NewConnection _) -> withStore' c (`deleteConn` connId)
|
||||
where
|
||||
delete :: RcvQueue -> m ()
|
||||
delete rq = do
|
||||
@@ -861,6 +1031,7 @@ connectionStats conn = case conn of
|
||||
SndConnection _ sq -> ConnectionStats {rcvServers = [], sndServers = sndSrvs sq, nextRcvServers = [], nextSndServers = []}
|
||||
DuplexConnection _ rq sq nextRq_ nextSq_ -> ConnectionStats {rcvServers = rcvSrvs rq, sndServers = sndSrvs sq, nextRcvServers = maybe [] rcvSrvs nextRq_, nextSndServers = maybe [] sndSrvs nextSq_}
|
||||
ContactConnection _ rq -> ConnectionStats {rcvServers = rcvSrvs rq, sndServers = [], nextRcvServers = [], nextSndServers = []}
|
||||
NewConnection _ -> ConnectionStats {rcvServers = [], sndServers = [], nextRcvServers = [], nextSndServers = []}
|
||||
where
|
||||
rcvSrvs RcvQueue {server} = [server]
|
||||
sndSrvs SndQueue {server} = [server]
|
||||
|
||||
@@ -167,6 +167,9 @@ data AgentClient = AgentClient
|
||||
smpQueueMsgQueues :: TMap MsgDeliveryKey (TQueue InternalId),
|
||||
smpQueueMsgDeliveries :: TMap MsgDeliveryKey (Async ()),
|
||||
nextRcvQueueMsgs :: TMap (SMPServer, SMP.RecipientId) [ServerTransmission BrokerMsg],
|
||||
connCmdsQueued :: TMap ConnId Bool,
|
||||
asyncCmdQueues :: TMap (Maybe SMPServer) (TQueue AsyncCmdId),
|
||||
asyncCmdProcesses :: TMap (Maybe SMPServer) (Async ()),
|
||||
ntfNetworkOp :: TVar AgentOpState,
|
||||
rcvNetworkOp :: TVar AgentOpState,
|
||||
msgDeliveryOp :: TVar AgentOpState,
|
||||
@@ -220,6 +223,9 @@ newAgentClient InitialAgentServers {smp, ntf, netCfg} agentEnv = do
|
||||
smpQueueMsgQueues <- TM.empty
|
||||
smpQueueMsgDeliveries <- TM.empty
|
||||
nextRcvQueueMsgs <- TM.empty
|
||||
connCmdsQueued <- TM.empty
|
||||
asyncCmdQueues <- TM.empty
|
||||
asyncCmdProcesses <- TM.empty
|
||||
ntfNetworkOp <- newTVar $ AgentOpState False 0
|
||||
rcvNetworkOp <- newTVar $ AgentOpState False 0
|
||||
msgDeliveryOp <- newTVar $ AgentOpState False 0
|
||||
@@ -231,7 +237,7 @@ newAgentClient InitialAgentServers {smp, ntf, netCfg} agentEnv = do
|
||||
asyncClients <- newTVar []
|
||||
clientId <- stateTVar (clientCounter agentEnv) $ \i -> let i' = i + 1 in (i', i')
|
||||
lock <- newTMVar ()
|
||||
return AgentClient {active, rcvQ, subQ, msgQ, smpServers, smpClients, ntfServers, ntfClients, useNetworkConfig, subscrSrvrs, pendingSubscrSrvrs, subscrConns, activeSubscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, nextRcvQueueMsgs, ntfNetworkOp, rcvNetworkOp, msgDeliveryOp, sndNetworkOp, databaseOp, agentState, getMsgLocks, reconnections, asyncClients, clientId, agentEnv, lock}
|
||||
return AgentClient {active, rcvQ, subQ, msgQ, smpServers, smpClients, ntfServers, ntfClients, useNetworkConfig, subscrSrvrs, pendingSubscrSrvrs, subscrConns, activeSubscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, nextRcvQueueMsgs, connCmdsQueued, asyncCmdQueues, asyncCmdProcesses, ntfNetworkOp, rcvNetworkOp, msgDeliveryOp, sndNetworkOp, databaseOp, agentState, getMsgLocks, reconnections, asyncClients, clientId, agentEnv, lock}
|
||||
|
||||
agentDbPath :: AgentClient -> FilePath
|
||||
agentDbPath AgentClient {agentEnv = Env {store = SQLiteStore {dbFilePath}}} = dbFilePath
|
||||
|
||||
@@ -93,6 +93,8 @@ module Simplex.Messaging.Agent.Protocol
|
||||
serializeCommand,
|
||||
connMode,
|
||||
connMode',
|
||||
networkCommandP,
|
||||
dbCommandP,
|
||||
commandP,
|
||||
connModeT,
|
||||
serializeQueueStatus,
|
||||
@@ -220,9 +222,9 @@ type ConnInfo = ByteString
|
||||
|
||||
-- | Parameterized type for SMP agent protocol commands and responses from all participants.
|
||||
data ACommand (p :: AParty) where
|
||||
NEW :: AConnectionMode -> ACommand Client -- response INV
|
||||
NEW :: Bool -> AConnectionMode -> ACommand Client -- response INV
|
||||
INV :: AConnectionRequestUri -> ACommand Agent
|
||||
JOIN :: AConnectionRequestUri -> ConnInfo -> ACommand Client -- response OK
|
||||
JOIN :: Bool -> AConnectionRequestUri -> ConnInfo -> ACommand Client -- response OK
|
||||
CONF :: ConfirmationId -> [SMPServer] -> ConnInfo -> ACommand Agent -- ConnInfo is from sender, [SMPServer] will be empty only in v1 handshake
|
||||
LET :: ConfirmationId -> ConnInfo -> ACommand Client -- ConnInfo is from client
|
||||
REQ :: InvitationId -> L.NonEmpty SMPServer -> ConnInfo -> ACommand Agent -- ConnInfo is from sender
|
||||
@@ -1027,9 +1029,17 @@ instance Arbitrary BrokerErrorType where arbitrary = genericArbitraryU
|
||||
|
||||
instance Arbitrary SMPAgentError where arbitrary = genericArbitraryU
|
||||
|
||||
-- | SMP agent command and response parser for commands passed via network (only parses binary length)
|
||||
networkCommandP :: Parser ACmd
|
||||
networkCommandP = commandP A.takeByteString
|
||||
|
||||
-- | SMP agent command and response parser for commands stored in db (fully parses binary bodies)
|
||||
dbCommandP :: Parser ACmd
|
||||
dbCommandP = commandP $ A.take =<< (A.decimal <* "\n")
|
||||
|
||||
-- | SMP agent command and response parser
|
||||
commandP :: Parser ACmd
|
||||
commandP =
|
||||
commandP :: Parser ByteString -> Parser ACmd
|
||||
commandP parseByteString =
|
||||
"NEW " *> newCmd
|
||||
<|> "INV " *> invResp
|
||||
<|> "JOIN " *> joinCmd
|
||||
@@ -1060,25 +1070,25 @@ commandP =
|
||||
<|> "CON" $> ACmd SAgent CON
|
||||
<|> "OK" $> ACmd SAgent OK
|
||||
where
|
||||
newCmd = ACmd SClient . NEW <$> strP
|
||||
newCmd = ACmd SClient .: NEW <$> strP_ <*> strP
|
||||
invResp = ACmd SAgent . INV <$> strP
|
||||
joinCmd = ACmd SClient .: JOIN <$> strP_ <*> A.takeByteString
|
||||
confMsg = ACmd SAgent .:. CONF <$> A.takeTill (== ' ') <* A.space <*> strListP <* A.space <*> A.takeByteString
|
||||
letCmd = ACmd SClient .: LET <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString
|
||||
reqMsg = ACmd SAgent .:. REQ <$> A.takeTill (== ' ') <* A.space <*> strP_ <*> A.takeByteString
|
||||
acptCmd = ACmd SClient .: ACPT <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString
|
||||
joinCmd = ACmd SClient .:. JOIN <$> strP_ <*> strP_ <*> parseByteString
|
||||
confMsg = ACmd SAgent .:. CONF <$> A.takeTill (== ' ') <* A.space <*> strListP <* A.space <*> parseByteString
|
||||
letCmd = ACmd SClient .: LET <$> A.takeTill (== ' ') <* A.space <*> parseByteString
|
||||
reqMsg = ACmd SAgent .:. REQ <$> A.takeTill (== ' ') <* A.space <*> strP_ <*> parseByteString
|
||||
acptCmd = ACmd SClient .: ACPT <$> A.takeTill (== ' ') <* A.space <*> parseByteString
|
||||
rjctCmd = ACmd SClient . RJCT <$> A.takeByteString
|
||||
infoCmd = ACmd SAgent . INFO <$> A.takeByteString
|
||||
infoCmd = ACmd SAgent . INFO <$> parseByteString
|
||||
connectResp = ACmd SAgent .: CONNECT <$> strP_ <*> strP
|
||||
disconnectResp = ACmd SAgent .: DISCONNECT <$> strP_ <*> strP
|
||||
downResp = ACmd SAgent .: DOWN <$> strP_ <*> connections
|
||||
upResp = ACmd SAgent .: UP <$> strP_ <*> connections
|
||||
switchResp = ACmd SAgent .: SWITCH <$> strP_ <*> strP
|
||||
sendCmd = ACmd SClient .: SEND <$> smpP <* A.space <*> A.takeByteString
|
||||
sendCmd = ACmd SClient .: SEND <$> smpP <* A.space <*> parseByteString
|
||||
msgIdResp = ACmd SAgent . MID <$> A.decimal
|
||||
sentResp = ACmd SAgent . SENT <$> A.decimal
|
||||
msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> strP
|
||||
message = ACmd SAgent .:. MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> A.takeByteString
|
||||
message = ACmd SAgent .:. MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> parseByteString
|
||||
ackCmd = ACmd SClient . ACK <$> A.decimal
|
||||
statResp = ACmd SAgent . STAT <$> strP
|
||||
connections = strP `A.sepBy'` A.char ','
|
||||
@@ -1092,14 +1102,14 @@ commandP =
|
||||
agentError = ACmd SAgent . ERR <$> strP
|
||||
|
||||
parseCommand :: ByteString -> Either AgentErrorType ACmd
|
||||
parseCommand = parse commandP $ CMD SYNTAX
|
||||
parseCommand = parse (commandP A.takeByteString) $ CMD SYNTAX
|
||||
|
||||
-- | Serialize SMP agent command.
|
||||
serializeCommand :: ACommand p -> ByteString
|
||||
serializeCommand = \case
|
||||
NEW cMode -> "NEW " <> strEncode cMode
|
||||
NEW ntfs cMode -> B.unwords ["NEW", strEncode ntfs, strEncode cMode]
|
||||
INV cReq -> "INV " <> strEncode cReq
|
||||
JOIN cReq cInfo -> B.unwords ["JOIN", strEncode cReq, serializeBinary cInfo]
|
||||
JOIN ntfs cReq cInfo -> B.unwords ["JOIN", strEncode ntfs, strEncode cReq, serializeBinary cInfo]
|
||||
CONF confId srvs cInfo -> B.unwords ["CONF", confId, strEncodeList srvs, serializeBinary cInfo]
|
||||
LET confId cInfo -> B.unwords ["LET", confId, serializeBinary cInfo]
|
||||
REQ invId srvs cInfo -> B.unwords ["REQ", invId, strEncode srvs, serializeBinary cInfo]
|
||||
@@ -1178,7 +1188,7 @@ tGet party h = liftIO (tGetRaw h) >>= tParseLoadBody
|
||||
tConnId :: ARawTransmission -> ACommand p -> Either AgentErrorType (ACommand p)
|
||||
tConnId (_, connId, _) cmd = case cmd of
|
||||
-- NEW, JOIN and ACPT have optional connId
|
||||
NEW _ -> Right cmd
|
||||
NEW _ _ -> Right cmd
|
||||
JOIN {} -> Right cmd
|
||||
ACPT {} -> Right cmd
|
||||
-- ERROR response does not always have connId
|
||||
@@ -1196,7 +1206,7 @@ tGet party h = liftIO (tGetRaw h) >>= tParseLoadBody
|
||||
cmdWithMsgBody = \case
|
||||
SEND msgFlags body -> SEND msgFlags <$$> getBody body
|
||||
MSG msgMeta msgFlags body -> MSG msgMeta msgFlags <$$> getBody body
|
||||
JOIN qUri cInfo -> JOIN qUri <$$> getBody cInfo
|
||||
JOIN ntfs qUri cInfo -> JOIN ntfs qUri <$$> getBody cInfo
|
||||
CONF confId srvs cInfo -> CONF confId srvs <$$> getBody cInfo
|
||||
LET confId cInfo -> LET confId <$$> getBody cInfo
|
||||
REQ invId srvs cInfo -> REQ invId srvs <$$> getBody cInfo
|
||||
|
||||
@@ -117,7 +117,7 @@ data SndQueue = SndQueue
|
||||
-- * Connection types
|
||||
|
||||
-- | Type of a connection.
|
||||
data ConnType = CRcv | CSnd | CDuplex | CContact deriving (Eq, Show)
|
||||
data ConnType = CNew | CRcv | CSnd | CDuplex | CContact deriving (Eq, Show)
|
||||
|
||||
-- | Connection of a specific type.
|
||||
--
|
||||
@@ -130,6 +130,7 @@ data ConnType = CRcv | CSnd | CDuplex | CContact deriving (Eq, Show)
|
||||
-- - 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
|
||||
NewConnection :: ConnData -> Connection CNew
|
||||
RcvConnection :: ConnData -> RcvQueue -> Connection CRcv
|
||||
SndConnection :: ConnData -> SndQueue -> Connection CSnd
|
||||
DuplexConnection :: ConnData -> RcvQueue -> SndQueue -> Maybe RcvQueue -> Maybe SndQueue -> Connection CDuplex
|
||||
@@ -141,18 +142,21 @@ deriving instance Show (Connection d)
|
||||
|
||||
connData :: Connection d -> ConnData
|
||||
connData = \case
|
||||
NewConnection cData -> cData
|
||||
RcvConnection cData _ -> cData
|
||||
SndConnection cData _ -> cData
|
||||
DuplexConnection cData _ _ _ _ -> cData
|
||||
ContactConnection cData _ -> cData
|
||||
|
||||
data SConnType :: ConnType -> Type where
|
||||
SCNew :: SConnType CNew
|
||||
SCRcv :: SConnType CRcv
|
||||
SCSnd :: SConnType CSnd
|
||||
SCDuplex :: SConnType CDuplex
|
||||
SCContact :: SConnType CContact
|
||||
|
||||
connType :: SConnType c -> ConnType
|
||||
connType SCNew = CNew
|
||||
connType SCRcv = CRcv
|
||||
connType SCSnd = CSnd
|
||||
connType SCDuplex = CDuplex
|
||||
@@ -334,6 +338,8 @@ newtype InternalId = InternalId {unId :: Int64} deriving (Eq, Show)
|
||||
|
||||
type InternalTs = UTCTime
|
||||
|
||||
type AsyncCmdId = Int64
|
||||
|
||||
-- * Store errors
|
||||
|
||||
-- | Agent store error.
|
||||
@@ -355,6 +361,8 @@ data StoreError
|
||||
SEInvitationNotFound
|
||||
| -- | Message not found
|
||||
SEMsgNotFound
|
||||
| -- | Command not found
|
||||
SECmdNotFound
|
||||
| -- | Currently not used. The intention was to pass current expected queue status in methods,
|
||||
-- as we always know what it should be at any stage of the protocol,
|
||||
-- and in case it does not match use this error.
|
||||
|
||||
@@ -25,6 +25,9 @@ module Simplex.Messaging.Agent.Store.SQLite
|
||||
connectSQLiteStore,
|
||||
|
||||
-- * Queues and connections
|
||||
createNewConn,
|
||||
updateNewConnRcv,
|
||||
updateNewConnSnd,
|
||||
createRcvConn,
|
||||
createSndConn,
|
||||
getConn,
|
||||
@@ -74,6 +77,11 @@ module Simplex.Messaging.Agent.Store.SQLite
|
||||
getRatchet,
|
||||
getSkippedMsgKeys,
|
||||
updateRatchet,
|
||||
-- Async commands
|
||||
createCommand,
|
||||
getPendingCommands,
|
||||
getPendingCommand,
|
||||
deleteCommand,
|
||||
-- Notification device token persistence
|
||||
createNtfToken,
|
||||
getSavedNtfToken,
|
||||
@@ -113,9 +121,10 @@ import Data.Bifunctor (second)
|
||||
import Data.ByteString (ByteString)
|
||||
import qualified Data.ByteString.Base64.URL as U
|
||||
import Data.Char (toLower)
|
||||
import Data.Function (on)
|
||||
import Data.Functor (($>))
|
||||
import Data.Int (Int64)
|
||||
import Data.List (find, foldl')
|
||||
import Data.List (find, foldl', groupBy)
|
||||
import Data.List.NonEmpty (NonEmpty (..))
|
||||
import qualified Data.Map.Strict as M
|
||||
import Data.Maybe (fromMaybe, listToMaybe)
|
||||
@@ -267,6 +276,37 @@ createConn_ gVar cData create = checkConstraint SEConnDuplicate $ case cData of
|
||||
ConnData {connId = ""} -> createWithRandomId gVar create
|
||||
ConnData {connId} -> create connId $> Right connId
|
||||
|
||||
createNewConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> SConnectionMode c -> IO (Either StoreError ConnId)
|
||||
createNewConn db gVar cData@ConnData {connAgentVersion, enableNtfs, duplexHandshake} cMode =
|
||||
createConn_ gVar cData $ \connId -> do
|
||||
DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?, ?, ?, ?, ?)" (connId, cMode, connAgentVersion, enableNtfs, duplexHandshake)
|
||||
|
||||
updateNewConnRcv :: DB.Connection -> ConnId -> RcvQueue -> IO (Either StoreError ())
|
||||
updateNewConnRcv db connId rq@RcvQueue {server} =
|
||||
getConn db connId $>>= \case
|
||||
(SomeConn _ NewConnection {}) -> updateConn
|
||||
(SomeConn _ RcvConnection {}) -> updateConn -- to allow retries
|
||||
(SomeConn c _) -> pure . Left . SEBadConnType $ connType c
|
||||
where
|
||||
updateConn :: IO (Either StoreError ())
|
||||
updateConn = do
|
||||
upsertServer_ db server
|
||||
void $ insertRcvQueue_ db connId rq
|
||||
pure $ Right ()
|
||||
|
||||
updateNewConnSnd :: DB.Connection -> ConnId -> SndQueue -> IO (Either StoreError ())
|
||||
updateNewConnSnd db connId sq@SndQueue {server} =
|
||||
getConn db connId $>>= \case
|
||||
(SomeConn _ NewConnection {}) -> updateConn
|
||||
(SomeConn _ SndConnection {}) -> updateConn -- to allow retries
|
||||
(SomeConn c _) -> pure . Left . SEBadConnType $ connType c
|
||||
where
|
||||
updateConn :: IO (Either StoreError ())
|
||||
updateConn = do
|
||||
upsertServer_ db server
|
||||
void $ insertSndQueue_ db connId sq
|
||||
pure $ Right ()
|
||||
|
||||
createRcvConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> IO (Either StoreError ConnId)
|
||||
createRcvConn db gVar cData@ConnData {connAgentVersion, enableNtfs, duplexHandshake} q@RcvQueue {server} cMode =
|
||||
createConn_ gVar cData $ \connId -> do
|
||||
@@ -757,6 +797,54 @@ updateRatchet db connId rc skipped = do
|
||||
forM_ (M.assocs mks) $ \(msgN, mk) ->
|
||||
DB.execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, hk, msgN, mk)
|
||||
|
||||
createCommand :: DB.Connection -> ConnId -> Maybe SMPServer -> ACommand 'Client -> IO AsyncCmdId
|
||||
createCommand db connId (Just (SMPServer host port _)) command = do
|
||||
DB.execute
|
||||
db
|
||||
"INSERT INTO commands (host, port, conn_id, command) VALUES (?, ?, ?, ?)"
|
||||
(host, port, connId, serializeCommand command)
|
||||
insertedRowId db
|
||||
createCommand db connId Nothing command = do
|
||||
DB.execute
|
||||
db
|
||||
"INSERT INTO commands (conn_id, command) VALUES (?, ?)"
|
||||
(connId, command)
|
||||
insertedRowId db
|
||||
|
||||
insertedRowId :: DB.Connection -> IO Int64
|
||||
insertedRowId db = fromOnly . head <$> DB.query_ db "SELECT last_insert_rowid()"
|
||||
|
||||
getPendingCommands :: DB.Connection -> ConnId -> IO [(Maybe SMPServer, [AsyncCmdId])]
|
||||
getPendingCommands db connId = do
|
||||
map (\ids -> (fst $ head ids, map snd ids)) . groupBy ((==) `on` fst) . map srvCmdId
|
||||
<$> DB.query
|
||||
db
|
||||
[sql|
|
||||
SELECT c.host, c.port, s.key_hash, c.command_id
|
||||
FROM commands c
|
||||
LEFT JOIN servers s ON s.host = c.host AND s.port = c.port
|
||||
WHERE conn_id = ?
|
||||
ORDER BY c.host, c.port, c.command_id ASC
|
||||
|]
|
||||
(Only connId)
|
||||
where
|
||||
srvCmdId (host, port, keyHash, cmdId) = (SMPServer <$> host <*> port <*> keyHash, cmdId)
|
||||
|
||||
getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ConnId, ACmd))
|
||||
getPendingCommand db msgId = do
|
||||
firstRow pendingCmd SECmdNotFound $
|
||||
DB.query
|
||||
db
|
||||
"SELECT conn_id, command FROM commands WHERE command_id = ?"
|
||||
(Only msgId)
|
||||
where
|
||||
pendingCmd :: (ConnId, ACmd) -> (ConnId, ACmd)
|
||||
pendingCmd (connId, commandStr) = (connId, commandStr)
|
||||
|
||||
deleteCommand :: DB.Connection -> AsyncCmdId -> IO ()
|
||||
deleteCommand db cmdId =
|
||||
DB.execute db "DELETE FROM commands WHERE command_id = ?" (Only cmdId)
|
||||
|
||||
createNtfToken :: DB.Connection -> NtfToken -> IO ()
|
||||
createNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer = srv@ProtocolServer {host, port}, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey), ntfDhSecret, ntfTknStatus, ntfTknAction, ntfMode} = do
|
||||
upsertNtfServer_ db srv
|
||||
@@ -1127,6 +1215,10 @@ instance ToField SndQueueAction where toField = toField . textEncode
|
||||
|
||||
instance FromField SndQueueAction where fromField = fromTextField_ textDecode
|
||||
|
||||
instance ToField (ACommand p) where toField = toField . serializeCommand
|
||||
|
||||
instance FromField ACmd where fromField = blobFieldParser dbCommandP
|
||||
|
||||
listToEither :: e -> [a] -> Either e a
|
||||
listToEither _ (x : _) = Right x
|
||||
listToEither e _ = Left e
|
||||
@@ -1245,6 +1337,7 @@ getConn db connId =
|
||||
(Just rq, Nothing, CMInvitation) -> pure . Right $ SomeConn SCRcv (RcvConnection cData rq)
|
||||
(Nothing, Just sq, CMInvitation) -> pure . Right $ SomeConn SCSnd (SndConnection cData sq)
|
||||
(Just rq, Nothing, CMContact) -> pure . Right $ SomeConn SCContact (ContactConnection cData rq)
|
||||
(Nothing, Nothing, _) -> pure . Right $ SomeConn SCNew (NewConnection cData)
|
||||
_ -> pure $ Left SEConnNotFound
|
||||
|
||||
getConnData :: DB.Connection -> ConnId -> IO (Maybe (ConnData, ConnectionMode))
|
||||
|
||||
@@ -37,6 +37,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220625_v2_ntf_mode
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220811_onion_hosts
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220817_connection_ntfs
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220822_queue_rotation
|
||||
import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands
|
||||
import Simplex.Messaging.Encoding.String
|
||||
import Simplex.Messaging.Transport.Client (TransportHost)
|
||||
|
||||
@@ -52,7 +53,8 @@ schemaMigrations =
|
||||
("m20220625_v2_ntf_mode", m20220625_v2_ntf_mode),
|
||||
("m20220811_onion_hosts", m20220811_onion_hosts),
|
||||
("m20220817_connection_ntfs", m20220817_connection_ntfs),
|
||||
("m20220822_queue_rotation", m20220822_queue_rotation)
|
||||
("m20220822_queue_rotation", m20220822_queue_rotation),
|
||||
("m20220905_commands", m20220905_commands)
|
||||
]
|
||||
|
||||
-- | The list of migrations in ascending order by date
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
{-# LANGUAGE QuasiQuotes #-}
|
||||
|
||||
module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands where
|
||||
|
||||
import Database.SQLite.Simple (Query)
|
||||
import Database.SQLite.Simple.QQ (sql)
|
||||
|
||||
m20220905_commands :: Query
|
||||
m20220905_commands =
|
||||
[sql|
|
||||
CREATE TABLE commands (
|
||||
command_id INTEGER PRIMARY KEY,
|
||||
conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
|
||||
host TEXT,
|
||||
port TEXT,
|
||||
command TEXT NOT NULL,
|
||||
command_version INTEGER NOT NULL DEFAULT 1,
|
||||
FOREIGN KEY (host, port) REFERENCES servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE
|
||||
);
|
||||
|]
|
||||
@@ -212,3 +212,13 @@ CREATE UNIQUE INDEX idx_rcv_queue_id ON rcv_queues(rcv_queue_id);
|
||||
CREATE UNIQUE INDEX idx_next_rcv_queue_id ON rcv_queues(next_rcv_queue_id);
|
||||
CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(snd_queue_id);
|
||||
CREATE UNIQUE INDEX idx_next_snd_queue_id ON snd_queues(next_snd_queue_id);
|
||||
CREATE TABLE commands(
|
||||
command_id INTEGER PRIMARY KEY,
|
||||
conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
|
||||
host TEXT,
|
||||
port TEXT,
|
||||
command TEXT NOT NULL,
|
||||
command_version INTEGER NOT NULL DEFAULT 1,
|
||||
FOREIGN KEY(host, port) REFERENCES servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE
|
||||
);
|
||||
|
||||
+21
-21
@@ -131,9 +131,9 @@ pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} _ msgBody
|
||||
|
||||
testDuplexConnection :: Transport c => TProxy c -> c -> c -> IO ()
|
||||
testDuplexConnection _ alice bob = do
|
||||
("1", "bob", Right (INV cReq)) <- alice #: ("1", "bob", "NEW INV")
|
||||
("1", "bob", Right (INV cReq)) <- alice #: ("1", "bob", "NEW T INV")
|
||||
let cReq' = strEncode cReq
|
||||
bob #: ("11", "alice", "JOIN " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice", OK)
|
||||
bob #: ("11", "alice", "JOIN T " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice", OK)
|
||||
("", "bob", Right (CONF confId _ "bob's connInfo")) <- (alice <#:)
|
||||
alice #: ("2", "bob", "LET " <> confId <> " 16\nalice's connInfo") #> ("2", "bob", OK)
|
||||
bob <# ("", "alice", INFO "alice's connInfo")
|
||||
@@ -164,9 +164,9 @@ testDuplexConnection _ alice bob = do
|
||||
|
||||
testDuplexConnRandomIds :: Transport c => TProxy c -> c -> c -> IO ()
|
||||
testDuplexConnRandomIds _ alice bob = do
|
||||
("1", bobConn, Right (INV cReq)) <- alice #: ("1", "", "NEW INV")
|
||||
("1", bobConn, Right (INV cReq)) <- alice #: ("1", "", "NEW T INV")
|
||||
let cReq' = strEncode cReq
|
||||
("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN " <> cReq' <> " 14\nbob's connInfo")
|
||||
("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN T " <> cReq' <> " 14\nbob's connInfo")
|
||||
("", bobConn', Right (CONF confId _ "bob's connInfo")) <- (alice <#:)
|
||||
bobConn' `shouldBe` bobConn
|
||||
alice #: ("2", bobConn, "LET " <> confId <> " 16\nalice's connInfo") =#> \case ("2", c, OK) -> c == bobConn; _ -> False
|
||||
@@ -197,10 +197,10 @@ testDuplexConnRandomIds _ alice bob = do
|
||||
|
||||
testContactConnection :: Transport c => TProxy c -> c -> c -> c -> IO ()
|
||||
testContactConnection _ alice bob tom = do
|
||||
("1", "alice_contact", Right (INV cReq)) <- alice #: ("1", "alice_contact", "NEW CON")
|
||||
("1", "alice_contact", Right (INV cReq)) <- alice #: ("1", "alice_contact", "NEW T CON")
|
||||
let cReq' = strEncode cReq
|
||||
|
||||
bob #: ("11", "alice", "JOIN " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice", OK)
|
||||
bob #: ("11", "alice", "JOIN T " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice", OK)
|
||||
("", "alice_contact", Right (REQ aInvId _ "bob's connInfo")) <- (alice <#:)
|
||||
alice #: ("2", "bob", "ACPT " <> aInvId <> " 16\nalice's connInfo") #> ("2", "bob", OK)
|
||||
("", "alice", Right (CONF bConfId _ "alice's connInfo")) <- (bob <#:)
|
||||
@@ -213,7 +213,7 @@ testContactConnection _ alice bob tom = do
|
||||
bob <#= \case ("", "alice", Msg "hi") -> True; _ -> False
|
||||
bob #: ("13", "alice", "ACK 4") #> ("13", "alice", OK)
|
||||
|
||||
tom #: ("21", "alice", "JOIN " <> cReq' <> " 14\ntom's connInfo") #> ("21", "alice", OK)
|
||||
tom #: ("21", "alice", "JOIN T " <> cReq' <> " 14\ntom's connInfo") #> ("21", "alice", OK)
|
||||
("", "alice_contact", Right (REQ aInvId' _ "tom's connInfo")) <- (alice <#:)
|
||||
alice #: ("4", "tom", "ACPT " <> aInvId' <> " 16\nalice's connInfo") #> ("4", "tom", OK)
|
||||
("", "alice", Right (CONF tConfId _ "alice's connInfo")) <- (tom <#:)
|
||||
@@ -228,10 +228,10 @@ testContactConnection _ alice bob tom = do
|
||||
|
||||
testContactConnRandomIds :: Transport c => TProxy c -> c -> c -> IO ()
|
||||
testContactConnRandomIds _ alice bob = do
|
||||
("1", aliceContact, Right (INV cReq)) <- alice #: ("1", "", "NEW CON")
|
||||
("1", aliceContact, Right (INV cReq)) <- alice #: ("1", "", "NEW T CON")
|
||||
let cReq' = strEncode cReq
|
||||
|
||||
("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN " <> cReq' <> " 14\nbob's connInfo")
|
||||
("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN T " <> cReq' <> " 14\nbob's connInfo")
|
||||
("", aliceContact', Right (REQ aInvId _ "bob's connInfo")) <- (alice <#:)
|
||||
aliceContact' `shouldBe` aliceContact
|
||||
|
||||
@@ -251,9 +251,9 @@ testContactConnRandomIds _ alice bob = do
|
||||
|
||||
testRejectContactRequest :: Transport c => TProxy c -> c -> c -> IO ()
|
||||
testRejectContactRequest _ alice bob = do
|
||||
("1", "a_contact", Right (INV cReq)) <- alice #: ("1", "a_contact", "NEW CON")
|
||||
("1", "a_contact", Right (INV cReq)) <- alice #: ("1", "a_contact", "NEW T CON")
|
||||
let cReq' = strEncode cReq
|
||||
bob #: ("11", "alice", "JOIN " <> cReq' <> " 10\nbob's info") #> ("11", "alice", OK)
|
||||
bob #: ("11", "alice", "JOIN T " <> cReq' <> " 10\nbob's info") #> ("11", "alice", OK)
|
||||
("", "a_contact", Right (REQ aInvId _ "bob's info")) <- (alice <#:)
|
||||
-- RJCT must use correct contact connection
|
||||
alice #: ("2a", "bob", "RJCT " <> aInvId) #> ("2a", "bob", ERR $ CONN NOT_FOUND)
|
||||
@@ -282,7 +282,7 @@ testSubscription _ alice1 alice2 bob = do
|
||||
|
||||
testSubscrNotification :: Transport c => TProxy c -> (ThreadId, ThreadId) -> c -> IO ()
|
||||
testSubscrNotification t (server, _) client = do
|
||||
client #: ("1", "conn1", "NEW INV") =#> \case ("1", "conn1", INV {}) -> True; _ -> False
|
||||
client #: ("1", "conn1", "NEW T INV") =#> \case ("1", "conn1", INV {}) -> True; _ -> False
|
||||
client #:# "nothing should be delivered to client before the server is killed"
|
||||
killThread server
|
||||
client <# ("", "", DOWN testSMPServer ["conn1"])
|
||||
@@ -392,9 +392,9 @@ testConcurrentMsgDelivery :: Transport c => TProxy c -> c -> c -> IO ()
|
||||
testConcurrentMsgDelivery _ alice bob = do
|
||||
connect (alice, "alice") (bob, "bob")
|
||||
|
||||
("1", "bob2", Right (INV cReq)) <- alice #: ("1", "bob2", "NEW INV")
|
||||
("1", "bob2", Right (INV cReq)) <- alice #: ("1", "bob2", "NEW T INV")
|
||||
let cReq' = strEncode cReq
|
||||
bob #: ("11", "alice2", "JOIN " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice2", OK)
|
||||
bob #: ("11", "alice2", "JOIN T " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice2", OK)
|
||||
("", "bob2", Right (CONF _confId _ "bob's connInfo")) <- (alice <#:)
|
||||
-- below commands would be needed to accept bob's connection, but alice does not
|
||||
-- alice #: ("2", "bob", "LET " <> _confId <> " 16\nalice's connInfo") #> ("2", "bob", OK)
|
||||
@@ -431,9 +431,9 @@ testMsgDeliveryQuotaExceeded _ alice bob = do
|
||||
|
||||
connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO ()
|
||||
connect (h1, name1) (h2, name2) = do
|
||||
("c1", _, Right (INV cReq)) <- h1 #: ("c1", name2, "NEW INV")
|
||||
("c1", _, Right (INV cReq)) <- h1 #: ("c1", name2, "NEW T INV")
|
||||
let cReq' = strEncode cReq
|
||||
h2 #: ("c2", name1, "JOIN " <> cReq' <> " 5\ninfo2") #> ("c2", name1, OK)
|
||||
h2 #: ("c2", name1, "JOIN T " <> cReq' <> " 5\ninfo2") #> ("c2", name1, OK)
|
||||
("", _, Right (CONF connId _ "info2")) <- (h1 <#:)
|
||||
h1 #: ("c3", name2, "LET " <> connId <> " 5\ninfo1") #> ("c3", name2, OK)
|
||||
h2 <# ("", name1, INFO "info1")
|
||||
@@ -452,9 +452,9 @@ sendMessage (h1, name1) (h2, name2) msg = do
|
||||
|
||||
-- connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)
|
||||
-- connect' h1 h2 = do
|
||||
-- ("c1", conn2, Right (INV cReq)) <- h1 #: ("c1", "", "NEW INV")
|
||||
-- ("c1", conn2, Right (INV cReq)) <- h1 #: ("c1", "", "NEW T INV")
|
||||
-- let cReq' = strEncode cReq
|
||||
-- ("c2", conn1, Right OK) <- h2 #: ("c2", "", "JOIN " <> cReq' <> " 5\ninfo2")
|
||||
-- ("c2", conn1, Right OK) <- h2 #: ("c2", "", "JOIN T " <> cReq' <> " 5\ninfo2")
|
||||
-- ("", _, Right (REQ connId _ "info2")) <- (h1 <#:)
|
||||
-- h1 #: ("c3", conn2, "ACPT " <> connId <> " 5\ninfo1") =#> \case ("c3", c, OK) -> c == conn2; _ -> False
|
||||
-- h2 <# ("", conn1, INFO "info1")
|
||||
@@ -471,17 +471,17 @@ syntaxTests t = do
|
||||
describe "NEW" $ do
|
||||
describe "valid" $ do
|
||||
-- TODO: add tests with defined connection id
|
||||
it "with correct parameter" $ ("211", "", "NEW INV") >#>= \case ("211", _, "INV" : _) -> True; _ -> False
|
||||
it "with correct parameter" $ ("211", "", "NEW T INV") >#>= \case ("211", _, "INV" : _) -> True; _ -> False
|
||||
describe "invalid" $ do
|
||||
-- TODO: add tests with defined connection id
|
||||
it "with incorrect parameter" $ ("222", "", "NEW hi") >#> ("222", "", "ERR CMD SYNTAX")
|
||||
it "with incorrect parameter" $ ("222", "", "NEW T hi") >#> ("222", "", "ERR CMD SYNTAX")
|
||||
|
||||
describe "JOIN" $ do
|
||||
describe "valid" $ do
|
||||
it "using same server as in invitation" $
|
||||
( "311",
|
||||
"a",
|
||||
"JOIN https://simpex.chat/invitation#/?smp=smp%3A%2F%2F"
|
||||
"JOIN T https://simpex.chat/invitation#/?smp=smp%3A%2F%2F"
|
||||
<> urlEncode True "LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI="
|
||||
<> "%40localhost%3A5001%2F3456-w%3D%3D%23"
|
||||
<> urlEncode True sampleDhKey
|
||||
|
||||
@@ -117,6 +117,11 @@ functionalAPITests t = do
|
||||
describe "Switching receive queues" $
|
||||
xit "should switch to a new queue" $
|
||||
testSwitchRcvQueue t
|
||||
describe "Async agent commands" $ do
|
||||
it "should connect using async agent commands" $
|
||||
withSmpServer t testAsyncCommands
|
||||
it "should restore and complete async commands on restart" $
|
||||
testAsyncCommandsRestore t
|
||||
|
||||
testAgentClient :: IO ()
|
||||
testAgentClient = do
|
||||
@@ -579,6 +584,55 @@ testSwitchRcvQueue t = do
|
||||
liftIO $ print r3
|
||||
pure ()
|
||||
|
||||
testAsyncCommands :: IO ()
|
||||
testAsyncCommands = do
|
||||
alice <- getSMPAgentClient agentCfg initAgentServers
|
||||
bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
|
||||
Right () <- runExceptT $ do
|
||||
bobId <- createConnectionAsync alice True SCMInvitation
|
||||
("", _, INV (ACR _ qInfo)) <- get alice
|
||||
aliceId <- joinConnectionAsync bob True qInfo "bob's connInfo"
|
||||
("", _, CONF confId _ "bob's connInfo") <- get alice
|
||||
allowConnectionAsync alice bobId confId "alice's connInfo"
|
||||
get alice ##> ("", bobId, CON)
|
||||
get bob ##> ("", aliceId, INFO "alice's connInfo")
|
||||
get bob ##> ("", aliceId, CON)
|
||||
-- message IDs 1 to 3 get assigned to control messages, so first MSG is assigned ID 4
|
||||
1 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "hello"
|
||||
get alice ##> ("", bobId, SENT $ baseId + 1)
|
||||
2 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
|
||||
get alice ##> ("", bobId, SENT $ baseId + 2)
|
||||
get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
|
||||
ackMessageAsync bob aliceId $ baseId + 1
|
||||
get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
|
||||
ackMessageAsync bob aliceId $ baseId + 2
|
||||
3 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "hello too"
|
||||
get bob ##> ("", aliceId, SENT $ baseId + 3)
|
||||
4 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 1"
|
||||
get bob ##> ("", aliceId, SENT $ baseId + 4)
|
||||
get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
|
||||
ackMessageAsync alice bobId $ baseId + 3
|
||||
get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
|
||||
ackMessageAsync alice bobId $ baseId + 4
|
||||
pure ()
|
||||
where
|
||||
baseId = 3
|
||||
msgId = subtract baseId
|
||||
|
||||
testAsyncCommandsRestore :: ATransport -> IO ()
|
||||
testAsyncCommandsRestore t = do
|
||||
alice <- getSMPAgentClient agentCfg initAgentServers
|
||||
Right bobId <- runExceptT $ createConnectionAsync alice True SCMInvitation
|
||||
liftIO $ noMessages alice "alice doesn't receive INV because server is down"
|
||||
disconnectAgentClient alice
|
||||
alice' <- liftIO $ getSMPAgentClient agentCfg initAgentServers
|
||||
withSmpServerStoreLogOn t testPort $ \_ -> do
|
||||
Right () <- runExceptT $ do
|
||||
subscribeConnection alice' bobId
|
||||
("", _, INV _) <- get alice'
|
||||
pure ()
|
||||
pure ()
|
||||
|
||||
exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
|
||||
exchangeGreetings = exchangeGreetingsMsgId 4
|
||||
|
||||
|
||||
@@ -62,7 +62,7 @@ smpAgentTest _ cmd = runSmpAgentTest $ \(h :: c) -> tPutRaw h cmd >> get h
|
||||
where
|
||||
get h = do
|
||||
t@(_, _, cmdStr) <- tGetRaw h
|
||||
case parseAll commandP cmdStr of
|
||||
case parseAll networkCommandP cmdStr of
|
||||
Right (ACmd SAgent CONNECT {}) -> get h
|
||||
Right (ACmd SAgent DISCONNECT {}) -> get h
|
||||
_ -> pure t
|
||||
|
||||
Reference in New Issue
Block a user