mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 08:42:27 +00:00
withStore runs in a single transaction (#408)
* remove store typeclass * make withStore run single transaction * combine 2 store calls * refactor * add exports * fix catch * correction
This commit is contained in:
Evgeny Poberezkin authored and GitHub committed
1 parent
ec70429106
commit
d1db7d6f79
7 files changed
+945
-1017
No files matched your search
@@ -81,13 +81,14 @@ import qualified Data.Text as T
|
||||
import Data.Time.Clock
|
||||
import Data.Time.Clock.System (systemToUTCTime)
|
||||
import Data.Word (Word16)
|
||||
import qualified Database.SQLite.Simple as DB
|
||||
import Simplex.Messaging.Agent.Client
|
||||
import Simplex.Messaging.Agent.Env.SQLite
|
||||
import Simplex.Messaging.Agent.NtfSubSupervisor
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import Simplex.Messaging.Agent.RetryInterval
|
||||
import Simplex.Messaging.Agent.Store
|
||||
import Simplex.Messaging.Agent.Store.SQLite (AgentStoreMonad, SQLiteStore)
|
||||
import Simplex.Messaging.Agent.Store.SQLite
|
||||
import Simplex.Messaging.Client (ProtocolClient (..), ServerTransmission)
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import qualified Simplex.Messaging.Crypto.Ratchet as CR
|
||||
@@ -100,7 +101,7 @@ import Simplex.Messaging.Parsers (parse)
|
||||
import Simplex.Messaging.Protocol (BrokerMsg, ErrorType (AUTH), MsgBody, MsgFlags, NMsgMeta (..))
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import qualified Simplex.Messaging.TMap as TM
|
||||
import Simplex.Messaging.Util (bshow, liftError, tryError, unlessM, whenM, ($>>=))
|
||||
import Simplex.Messaging.Util (bshow, liftE, liftError, tryError, unlessM, whenM, ($>>=))
|
||||
import Simplex.Messaging.Version
|
||||
import System.Random (randomR)
|
||||
import UnliftIO.Async (async, race_)
|
||||
@@ -255,7 +256,7 @@ newConn c connId cMode = do
|
||||
g <- asks idsDrg
|
||||
agentVersion <- asks $ smpAgentVersion . config
|
||||
let cData = ConnData {connId, connAgentVersion = agentVersion, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent
|
||||
connId' <- withStore c $ \st -> createRcvConn st g cData rq cMode
|
||||
connId' <- withStore c $ \db -> createRcvConn db g cData rq cMode
|
||||
addSubscription c rq connId'
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ sendNtfSubCommand ns (connId', NSCCreate)
|
||||
@@ -265,7 +266,7 @@ newConn c connId cMode = do
|
||||
SCMContact -> pure (connId', CRContactUri crData)
|
||||
SCMInvitation -> do
|
||||
(pk1, pk2, e2eRcvParams) <- liftIO $ CR.generateE2EParams CR.e2eEncryptVersion
|
||||
withStore c $ \st -> createRatchetX3dhKeys st connId' pk1 pk2
|
||||
withStore' c $ \db -> createRatchetX3dhKeys db connId' pk1 pk2
|
||||
pure (connId', CRInvitationUri crData $ toVersionRangeT e2eRcvParams CR.e2eEncryptVRange)
|
||||
|
||||
joinConn :: AgentMonad m => AgentClient -> ConnId -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
@@ -284,9 +285,9 @@ joinConn c connId (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2
|
||||
g <- asks idsDrg
|
||||
let duplexHS = connAgentVersion /= 1
|
||||
cData = ConnData {connId, connAgentVersion, duplexHandshake = Just duplexHS}
|
||||
connId' <- withStore c $ \st -> do
|
||||
connId' <- createSndConn st g cData sq
|
||||
createRatchet st connId' rc
|
||||
connId' <- withStore c $ \db -> runExceptT $ do
|
||||
connId' <- ExceptT $ createSndConn db g cData sq
|
||||
liftIO $ createRatchet db connId' rc
|
||||
pure connId'
|
||||
let cData' = (cData :: ConnData) {connId = connId'}
|
||||
tryError (confirmQueue aVersion c connId' sq cInfo $ Just e2eSndParams) >>= \case
|
||||
@@ -295,7 +296,7 @@ joinConn c connId (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2
|
||||
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')
|
||||
withStore' c (`deleteConn` connId')
|
||||
throwError e
|
||||
_ -> throwError $ AGENT A_VERSION
|
||||
joinConn c connId (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInfo = do
|
||||
@@ -317,18 +318,20 @@ createReplyQueue c connId = do
|
||||
-- TODO reply queue version should be the same as send queue, ignoring it in v1
|
||||
let qInfo = toVersionT qUri SMP.smpClientVersion
|
||||
addSubscription c rq connId
|
||||
withStore c $ \st -> upgradeSndConnToDuplex st connId rq
|
||||
withStore c $ \db -> upgradeSndConnToDuplex db connId rq
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ sendNtfSubCommand ns (connId, NSCCreate)
|
||||
pure qInfo
|
||||
|
||||
-- | Approve confirmation (LET command) in Reader monad
|
||||
allowConnection' :: AgentMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
|
||||
allowConnection' c connId confId ownConnInfo = do
|
||||
allowConnection' c connId confId ownConnInfo =
|
||||
withStore c (`getConn` connId) >>= \case
|
||||
SomeConn _ (RcvConnection cData rq) -> do
|
||||
AcceptedConfirmation {senderConf, ratchetState} <- withStore c $ \st -> acceptConfirmation st confId ownConnInfo
|
||||
withStore c $ \st -> createRatchet st connId ratchetState
|
||||
AcceptedConfirmation {senderConf} <- withStore c $ \db -> runExceptT $ do
|
||||
conf <- ExceptT $ acceptConfirmation db confId ownConnInfo
|
||||
liftIO $ createRatchet db connId $ ratchetState (conf :: AcceptedConfirmation)
|
||||
pure conf
|
||||
processConfirmation c rq senderConf
|
||||
mapM_ (connectReplyQueues c cData ownConnInfo) (L.nonEmpty $ smpReplyQueues senderConf)
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
@@ -339,21 +342,21 @@ acceptContact' c connId invId ownConnInfo = do
|
||||
Invitation {contactConnId, connReq} <- withStore c (`getInvitation` invId)
|
||||
withStore c (`getConn` contactConnId) >>= \case
|
||||
SomeConn _ ContactConnection {} -> do
|
||||
withStore c $ \st -> acceptInvitation st invId ownConnInfo
|
||||
withStore' c $ \db -> acceptInvitation db invId ownConnInfo
|
||||
joinConn c connId connReq ownConnInfo
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
|
||||
-- | Reject contact (RJCT command) in Reader monad
|
||||
rejectContact' :: AgentMonad m => AgentClient -> ConnId -> InvitationId -> m ()
|
||||
rejectContact' c contactConnId invId =
|
||||
withStore c $ \st -> deleteInvitation st contactConnId invId
|
||||
withStore c $ \db -> deleteInvitation db contactConnId invId
|
||||
|
||||
processConfirmation :: AgentMonad m => AgentClient -> RcvQueue -> SMPConfirmation -> m ()
|
||||
processConfirmation c rq@RcvQueue {e2ePrivKey} SMPConfirmation {senderKey, e2ePubKey} = do
|
||||
let dhSecret = C.dh' e2ePubKey e2ePrivKey
|
||||
withStore c $ \st -> setRcvQueueConfirmedE2E st rq dhSecret
|
||||
withStore' c $ \db -> setRcvQueueConfirmedE2E db rq dhSecret
|
||||
secureQueue c rq senderKey
|
||||
withStore c $ \st -> setRcvQueueStatus st rq Secured
|
||||
withStore' c $ \db -> setRcvQueueStatus db rq Secured
|
||||
|
||||
-- | Subscribe to receive connection messages (SUB command) in Reader monad
|
||||
subscribeConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
|
||||
@@ -394,11 +397,11 @@ getConnectionMessage' c connId = do
|
||||
|
||||
getNotificationMessage' :: forall m. AgentMonad m => AgentClient -> ByteString -> C.CbNonce -> m (Maybe (SMP.MsgId, MsgFlags))
|
||||
getNotificationMessage' c encMessageInfo nonce = do
|
||||
withStore c getActiveNtfToken >>= \case
|
||||
withStore' c getActiveNtfToken >>= \case
|
||||
Just NtfToken {ntfDhSecret = Just dhSecret} -> do
|
||||
ntfData <- agentCbDecrypt dhSecret nonce encMessageInfo
|
||||
PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} <- liftEither (parse strP (INTERNAL "error parsing PNMessageData") ntfData)
|
||||
(connId, rcvDhSecret) <- withStore c $ \st -> getNtfRcvQueue st smpQueue
|
||||
(connId, rcvDhSecret) <- withStore c (`getNtfRcvQueue` smpQueue)
|
||||
nMsgMeta <- agentCbDecrypt rcvDhSecret nmsgNonce encNMsgMeta `catchError` const (pure "")
|
||||
let nMsgMetaParsed = parse smpP (INTERNAL "error parsing NMsgMeta") nMsgMeta
|
||||
case nMsgMetaParsed of
|
||||
@@ -426,18 +429,18 @@ enqueueMessage c cData@ConnData {connId, connAgentVersion} sq msgFlags aMessage
|
||||
pure $ unId msgId
|
||||
where
|
||||
storeSentMsg :: m InternalId
|
||||
storeSentMsg = withStore c $ \st -> do
|
||||
storeSentMsg = withStore c $ \db -> runExceptT $ do
|
||||
internalTs <- liftIO getCurrentTime
|
||||
(internalId, internalSndId, prevMsgHash) <- updateSndIds st connId
|
||||
(internalId, internalSndId, prevMsgHash) <- liftIO $ updateSndIds db connId
|
||||
let privHeader = APrivHeader (unSndId internalSndId) prevMsgHash
|
||||
agentMsg = AgentMessage privHeader aMessage
|
||||
agentMsgStr = smpEncode agentMsg
|
||||
internalHash = C.sha256Hash agentMsgStr
|
||||
encAgentMessage <- agentRatchetEncrypt st connId agentMsgStr e2eEncUserMsgLength
|
||||
encAgentMessage <- agentRatchetEncrypt db connId agentMsgStr e2eEncUserMsgLength
|
||||
let msgBody = smpEncode $ AgentMsgEnvelope {agentVersion = connAgentVersion, encAgentMessage}
|
||||
msgType = agentMessageType agentMsg
|
||||
msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody, internalHash, prevMsgHash}
|
||||
createSndMsg st connId msgData
|
||||
liftIO $ createSndMsg db connId msgData
|
||||
pure internalId
|
||||
|
||||
resumeMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> m ()
|
||||
@@ -447,7 +450,7 @@ resumeMsgDelivery c cData@ConnData {connId} sq@SndQueue {server, sndId} = do
|
||||
async (runSmpQueueMsgDelivery c cData sq)
|
||||
>>= \a -> atomically (TM.insert qKey a $ smpQueueMsgDeliveries c)
|
||||
unlessM connQueued $
|
||||
withStore c (`getPendingMsgs` connId)
|
||||
withStore' c (`getPendingMsgs` connId)
|
||||
>>= queuePendingMsgs c connId sq
|
||||
where
|
||||
queueDelivering qKey = atomically $ TM.member qKey (smpQueueMsgDeliveries c)
|
||||
@@ -477,7 +480,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
|
||||
msgId <- atomically $ readTQueue mq
|
||||
atomically $ beginAgentOperation c AONetwork
|
||||
let mId = unId msgId
|
||||
withStore c (\st -> E.try $ getPendingMsgData st connId msgId) >>= \case
|
||||
E.try (withStore c $ \db -> getPendingMsgData db connId msgId) >>= \case
|
||||
Left (e :: E.SomeException) ->
|
||||
notify $ MERR mId (INTERNAL $ show e)
|
||||
Right (rq_, PendingMsgData {msgType, msgBody, msgFlags, internalTs}) ->
|
||||
@@ -521,12 +524,13 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
|
||||
Right () -> do
|
||||
case msgType of
|
||||
AM_CONN_INFO -> do
|
||||
withStore c $ \st -> setSndQueueStatus st sq Confirmed
|
||||
when (isJust rq_) $ withStore c (`removeConfirmations` connId)
|
||||
withStore' c $ \db -> do
|
||||
setSndQueueStatus db sq Confirmed
|
||||
when (isJust rq_) $ removeConfirmations db connId
|
||||
-- TODO possibly notification flag should be ON for one of the parties, to result in contact connected notification
|
||||
unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO
|
||||
AM_HELLO_ -> do
|
||||
withStore c $ \st -> setSndQueueStatus st sq Active
|
||||
withStore' c $ \db -> setSndQueueStatus db sq Active
|
||||
case rq_ of
|
||||
-- party initiating connection (in v1)
|
||||
Just RcvQueue {status} ->
|
||||
@@ -551,7 +555,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
|
||||
delMsg msgId
|
||||
where
|
||||
delMsg :: InternalId -> m ()
|
||||
delMsg msgId = withStore c $ \st -> deleteMsg st connId msgId
|
||||
delMsg msgId = withStore' c $ \db -> deleteMsg db connId msgId
|
||||
notify :: ACommand 'Agent -> m ()
|
||||
notify cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
|
||||
notifyDel :: InternalId -> ACommand 'Agent -> m ()
|
||||
@@ -574,11 +578,11 @@ ackMessage' c connId msgId = do
|
||||
ack :: RcvQueue -> m ()
|
||||
ack rq = do
|
||||
let mId = InternalId msgId
|
||||
srvMsgId <- withStore c $ \st -> setMsgUserAck st connId mId
|
||||
srvMsgId <- withStore c $ \db -> setMsgUserAck db connId mId
|
||||
sendAck c rq srvMsgId `catchError` \case
|
||||
SMP SMP.NO_MSG -> pure ()
|
||||
e -> throwError e
|
||||
withStore c $ \st -> deleteMsg st connId mId
|
||||
withStore' c $ \db -> deleteMsg db connId mId
|
||||
|
||||
-- | Suspend SMP agent connection (OFF command) in Reader monad
|
||||
suspendConnection' :: AgentMonad m => AgentClient -> ConnId -> m ()
|
||||
@@ -596,7 +600,7 @@ deleteConnection' c connId =
|
||||
SomeConn _ (DuplexConnection _ rq _) -> delete rq
|
||||
SomeConn _ (RcvConnection _ rq) -> delete rq
|
||||
SomeConn _ (ContactConnection _ rq) -> delete rq
|
||||
SomeConn _ (SndConnection _ _) -> withStore c (`deleteConn` connId)
|
||||
SomeConn _ (SndConnection _ _) -> withStore' c (`deleteConn` connId)
|
||||
where
|
||||
delete :: RcvQueue -> m ()
|
||||
delete rq = do
|
||||
@@ -605,7 +609,7 @@ deleteConnection' c connId =
|
||||
atomically $ do
|
||||
removeSubscription c connId
|
||||
sendNtfSubCommand ns (connId, NSCDelete)
|
||||
withStore c (`deleteConn` connId)
|
||||
withStore' c (`deleteConn` connId)
|
||||
|
||||
-- | Change servers to be used for creating new queues, in Reader monad
|
||||
setSMPServers' :: AgentMonad m => AgentClient -> NonEmpty SMPServer -> m ()
|
||||
@@ -614,7 +618,7 @@ setSMPServers' c servers = do
|
||||
|
||||
registerNtfToken' :: forall m. AgentMonad m => AgentClient -> DeviceToken -> m NtfTknStatus
|
||||
registerNtfToken' c deviceToken =
|
||||
withStore c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
withStore' c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
(Just tkn@NtfToken {ntfTokenId, ntfTknStatus, ntfTknAction}, prevTokens) -> do
|
||||
mapM_ (deleteToken_ c) prevTokens
|
||||
ns <- asks ntfSupervisor
|
||||
@@ -634,7 +638,7 @@ registerNtfToken' c deviceToken =
|
||||
-- agentNtfCheckToken c tknId tkn >>= \case
|
||||
(Just tknId, Just NTADelete) -> do
|
||||
agentNtfDeleteToken c tknId tkn
|
||||
withStore c $ \st -> removeNtfToken st tkn
|
||||
withStore' c (`removeNtfToken` tkn)
|
||||
atomically $ nsRemoveNtfToken ns
|
||||
pure NTExpired
|
||||
_ -> pure ntfTknStatus
|
||||
@@ -646,7 +650,7 @@ registerNtfToken' c deviceToken =
|
||||
tknKeys <- liftIO $ C.generateSignatureKeyPair a
|
||||
dhKeys <- liftIO C.generateKeyPair'
|
||||
let tkn = newNtfToken deviceToken ntfServer tknKeys dhKeys
|
||||
withStore c $ \st -> createNtfToken st tkn
|
||||
withStore' c (`createNtfToken` tkn)
|
||||
registerToken tkn
|
||||
pure NTRegistered
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
@@ -656,14 +660,14 @@ registerNtfToken' c deviceToken =
|
||||
registerToken tkn@NtfToken {ntfPubKey, ntfDhKeys = (pubDhKey, privDhKey)} = do
|
||||
(tknId, srvPubDhKey) <- agentNtfRegisterToken c tkn ntfPubKey pubDhKey
|
||||
let dhSecret = C.dh' srvPubDhKey privDhKey
|
||||
withStore c $ \st -> updateNtfTokenRegistration st tkn tknId dhSecret
|
||||
withStore' c $ \db -> updateNtfTokenRegistration db tkn tknId dhSecret
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ nsUpdateToken ns tkn
|
||||
|
||||
-- TODO decrypt verification code
|
||||
verifyNtfToken' :: AgentMonad m => AgentClient -> DeviceToken -> ByteString -> C.CbNonce -> m ()
|
||||
verifyNtfToken' c deviceToken code nonce =
|
||||
withStore c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
withStore' c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
(Just tkn@NtfToken {ntfTokenId = Just tknId, ntfDhSecret = Just dhSecret}, _) -> do
|
||||
code' <- liftEither . bimap cryptoError NtfRegCode $ C.cbDecrypt dhSecret nonce code
|
||||
void . withToken c tkn (Just (NTConfirmed, NTAVerify code')) (NTActive, Just NTACheck) $ do
|
||||
@@ -673,7 +677,7 @@ verifyNtfToken' c deviceToken code nonce =
|
||||
enableNtfCron' :: AgentMonad m => AgentClient -> DeviceToken -> Word16 -> m ()
|
||||
enableNtfCron' c deviceToken interval = do
|
||||
when (interval < 20) . throwError $ CMD PROHIBITED
|
||||
withStore c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
withStore' c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
(Just tkn@NtfToken {ntfTokenId = Just tknId, ntfTknStatus = NTActive}, _) ->
|
||||
void . withToken c tkn (Just (NTActive, NTACron interval)) (cronSuccess interval) $
|
||||
agentNtfEnableCron c tknId tkn interval
|
||||
@@ -686,13 +690,13 @@ cronSuccess interval
|
||||
|
||||
checkNtfToken' :: AgentMonad m => AgentClient -> DeviceToken -> m NtfTknStatus
|
||||
checkNtfToken' c deviceToken =
|
||||
withStore c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
withStore' c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
(Just tkn@NtfToken {ntfTokenId = Just tknId}, _) -> agentNtfCheckToken c tknId tkn
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
|
||||
deleteNtfToken' :: AgentMonad m => AgentClient -> DeviceToken -> m ()
|
||||
deleteNtfToken' c deviceToken =
|
||||
withStore c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
withStore' c (`getDeviceNtfToken` deviceToken) >>= \case
|
||||
(Just tkn, _) -> deleteToken_ c tkn
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
|
||||
@@ -701,30 +705,30 @@ deleteToken_ c tkn@NtfToken {ntfTokenId, ntfTknStatus} = do
|
||||
ns <- asks ntfSupervisor
|
||||
forM_ ntfTokenId $ \tknId -> do
|
||||
let ntfTknAction = Just NTADelete
|
||||
withStore c $ \st -> updateNtfToken st tkn ntfTknStatus ntfTknAction
|
||||
withStore' c $ \db -> updateNtfToken db tkn ntfTknStatus ntfTknAction
|
||||
atomically $ nsUpdateToken ns tkn {ntfTknStatus, ntfTknAction}
|
||||
agentNtfDeleteToken c tknId tkn `catchError` \case
|
||||
NTF AUTH -> pure ()
|
||||
e -> throwError e
|
||||
withStore c $ \st -> removeNtfToken st tkn
|
||||
withStore' c $ \db -> removeNtfToken db tkn
|
||||
atomically $ nsRemoveNtfToken ns
|
||||
|
||||
withToken :: AgentMonad m => AgentClient -> NtfToken -> Maybe (NtfTknStatus, NtfTknAction) -> (NtfTknStatus, Maybe NtfTknAction) -> m a -> m NtfTknStatus
|
||||
withToken c tkn@NtfToken {deviceToken} from_ (toStatus, toAction_) f = do
|
||||
ns <- asks ntfSupervisor
|
||||
forM_ from_ $ \(status, action) -> do
|
||||
withStore c $ \st -> updateNtfToken st tkn status (Just action)
|
||||
withStore' c $ \db -> updateNtfToken db tkn status (Just action)
|
||||
atomically $ nsUpdateToken ns tkn {ntfTknStatus = status, ntfTknAction = Just action}
|
||||
tryError f >>= \case
|
||||
Right _ -> do
|
||||
withStore c $ \st -> updateNtfToken st tkn toStatus toAction_
|
||||
withStore' c $ \db -> updateNtfToken db tkn toStatus toAction_
|
||||
let updatedToken = tkn {ntfTknStatus = toStatus, ntfTknAction = toAction_}
|
||||
if toStatus == NTActive
|
||||
then initializeNtfSubQ c updatedToken
|
||||
else atomically $ nsUpdateToken ns updatedToken
|
||||
pure toStatus
|
||||
Left e@(NTF AUTH) -> do
|
||||
withStore c $ \st -> removeNtfToken st tkn
|
||||
withStore' c $ \db -> removeNtfToken db tkn
|
||||
atomically $ nsRemoveNtfToken ns
|
||||
void $ registerNtfToken' c deviceToken
|
||||
throwError e
|
||||
@@ -774,7 +778,7 @@ subscriber c@AgentClient {msgQ} = forever $ do
|
||||
|
||||
processSMPTransmission :: forall m. AgentMonad m => AgentClient -> ServerTransmission BrokerMsg -> m ()
|
||||
processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd) =
|
||||
withStore c (\st -> getRcvConn st srv rId) >>= \case
|
||||
withStore c (\db -> getRcvConn db srv rId) >>= \case
|
||||
SomeConn _ conn@(DuplexConnection cData rq _) -> processSMP conn cData rq
|
||||
SomeConn _ conn@(RcvConnection cData rq) -> processSMP conn cData rq
|
||||
SomeConn _ conn@(ContactConnection cData rq) -> processSMP conn cData rq
|
||||
@@ -803,19 +807,19 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd)
|
||||
(SMP.PHEmpty, AgentMsgEnvelope _ encAgentMsg) ->
|
||||
tryError agentClientMsg >>= \case
|
||||
Right (Just (msgId, msgMeta, aMessage)) -> case aMessage of
|
||||
HELLO -> helloMsg >> ack >> withStore c (\st -> deleteMsg st connId msgId)
|
||||
REPLY cReq -> replyMsg cReq >> ack >> withStore c (\st -> deleteMsg st connId msgId)
|
||||
HELLO -> helloMsg >> ack >> withStore' c (\db -> deleteMsg db connId msgId)
|
||||
REPLY cReq -> replyMsg cReq >> ack >> withStore' c (\db -> deleteMsg db connId msgId)
|
||||
-- note that there is no ACK sent for A_MSG, it is sent with agent's user ACK command
|
||||
A_MSG body -> do
|
||||
logServer "<--" c srv rId "MSG <MSG>"
|
||||
notify $ MSG msgMeta msgFlags body
|
||||
Right _ -> prohibited >> ack
|
||||
Left e@(AGENT A_DUPLICATE) -> do
|
||||
withStore c (\st -> getLastMsg st connId srvMsgId) >>= \case
|
||||
withStore' c (\db -> getLastMsg db connId srvMsgId) >>= \case
|
||||
Just RcvMsg {internalId, msgMeta, msgBody = agentMsgBody, userAck}
|
||||
| userAck -> do
|
||||
ack
|
||||
withStore c $ \st -> deleteMsg st connId internalId
|
||||
withStore' c $ \db -> deleteMsg db connId internalId
|
||||
| otherwise -> do
|
||||
liftEither (parse smpP (AGENT A_MESSAGE) agentMsgBody) >>= \case
|
||||
AgentMessage _ (A_MSG body) -> do
|
||||
@@ -826,20 +830,20 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd)
|
||||
Left e -> throwError e
|
||||
where
|
||||
agentClientMsg :: m (Maybe (InternalId, MsgMeta, AMessage))
|
||||
agentClientMsg = withStore c $ \st -> do
|
||||
agentMsgBody <- agentRatchetDecrypt st connId encAgentMsg
|
||||
agentClientMsg = withStore c $ \db -> runExceptT $ do
|
||||
agentMsgBody <- agentRatchetDecrypt db connId encAgentMsg
|
||||
liftEither (parse smpP (SEAgentError $ AGENT A_MESSAGE) agentMsgBody) >>= \case
|
||||
agentMsg@(AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage) -> do
|
||||
let msgType = agentMessageType agentMsg
|
||||
internalHash = C.sha256Hash agentMsgBody
|
||||
internalTs <- liftIO getCurrentTime
|
||||
(internalId, internalRcvId, prevExtSndId, prevRcvMsgHash) <- updateRcvIds st connId
|
||||
(internalId, internalRcvId, prevExtSndId, prevRcvMsgHash) <- liftIO $ updateRcvIds db connId
|
||||
let integrity = checkMsgIntegrity prevExtSndId sndMsgId prevRcvMsgHash prevMsgHash
|
||||
recipient = (unId internalId, internalTs)
|
||||
broker = (srvMsgId, systemToUTCTime srvTs)
|
||||
msgMeta = MsgMeta {integrity, recipient, broker, sndMsgId}
|
||||
rcvMsg = RcvMsgData {msgMeta, msgType, msgFlags, msgBody = agentMsgBody, internalRcvId, internalHash, externalPrevSndHash = prevMsgHash}
|
||||
createRcvMsg st connId rcvMsg
|
||||
liftIO $ createRcvMsg db connId rcvMsg
|
||||
pure $ Just (internalId, msgMeta, aMessage)
|
||||
_ -> pure Nothing
|
||||
_ -> prohibited >> ack
|
||||
@@ -900,7 +904,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd)
|
||||
New -> case (conn, e2eEncryption) of
|
||||
-- party initiating connection
|
||||
(RcvConnection {}, Just e2eSndParams) -> do
|
||||
(pk1, rcDHRs) <- withStore c $ \st -> getRatchetX3dhKeys st connId
|
||||
(pk1, rcDHRs) <- withStore c $ (`getRatchetX3dhKeys` connId)
|
||||
let rc = CR.initRcvRatchet rcDHRs $ CR.x3dhRcv pk1 rcDHRs e2eSndParams
|
||||
(agentMsgBody_, rc', skipped) <- liftError cryptoError $ CR.rcDecrypt rc M.empty encConnInfo
|
||||
case (agentMsgBody_, skipped) of
|
||||
@@ -915,14 +919,14 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd)
|
||||
processConf connInfo senderConf duplexHS = do
|
||||
let newConfirmation = NewConfirmation {connId, senderConf, ratchetState = rc'}
|
||||
g <- asks idsDrg
|
||||
confId <- withStore c $ \st -> do
|
||||
setHandshakeVersion st connId agentVersion duplexHS
|
||||
createConfirmation st g newConfirmation
|
||||
confId <- withStore c $ \db -> do
|
||||
setHandshakeVersion db connId agentVersion duplexHS
|
||||
createConfirmation db g newConfirmation
|
||||
notify $ CONF confId connInfo
|
||||
_ -> prohibited
|
||||
-- party accepting connection
|
||||
(DuplexConnection _ _ sq, Nothing) -> do
|
||||
withStore c (\st -> agentRatchetDecrypt st connId encConnInfo) >>= parseMessage >>= \case
|
||||
withStore c (\db -> runExceptT $ agentRatchetDecrypt db connId encConnInfo) >>= parseMessage >>= \case
|
||||
AgentConnInfo connInfo -> do
|
||||
notify $ INFO connInfo
|
||||
processConfirmation c rq $ SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = []}
|
||||
@@ -937,7 +941,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd)
|
||||
case status of
|
||||
Active -> prohibited
|
||||
_ -> do
|
||||
withStore c $ \st -> setRcvQueueStatus st rq Active
|
||||
withStore' c $ \db -> setRcvQueueStatus db rq Active
|
||||
case conn of
|
||||
DuplexConnection _ _ sq@SndQueue {status = sndStatus}
|
||||
-- `sndStatus == Active` when HELLO was previously sent, and this is the reply HELLO
|
||||
@@ -970,7 +974,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd)
|
||||
ContactConnection {} -> do
|
||||
g <- asks idsDrg
|
||||
let newInv = NewInvitation {contactConnId = connId, connReq, recipientConnInfo = cInfo}
|
||||
invId <- withStore c $ \st -> createInvitation st g newInv
|
||||
invId <- withStore c $ \db -> createInvitation db g newInv
|
||||
notify $ REQ invId cInfo
|
||||
_ -> prohibited
|
||||
|
||||
@@ -990,7 +994,7 @@ connectReplyQueues c cData@ConnData {connId} ownConnInfo (qInfo :| _) = do
|
||||
Nothing -> throwError $ AGENT A_VERSION
|
||||
Just qInfo' -> do
|
||||
sq <- newSndQueue qInfo'
|
||||
withStore c $ \st -> upgradeRcvConnToDuplex st connId sq
|
||||
withStore c $ \db -> upgradeRcvConnToDuplex db connId sq
|
||||
enqueueConfirmation c cData sq ownConnInfo Nothing
|
||||
|
||||
confirmQueue :: forall m. AgentMonad m => Compatible Version -> AgentClient -> ConnId -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
|
||||
@@ -998,12 +1002,12 @@ confirmQueue (Compatible agentVersion) c connId sq connInfo e2eEncryption = do
|
||||
aMessage <- mkAgentMessage agentVersion
|
||||
msg <- mkConfirmation aMessage
|
||||
sendConfirmation c sq msg
|
||||
withStore c $ \st -> setSndQueueStatus st sq Confirmed
|
||||
withStore' c $ \db -> setSndQueueStatus db sq Confirmed
|
||||
where
|
||||
mkConfirmation :: AgentMessage -> m MsgBody
|
||||
mkConfirmation aMessage = withStore c $ \st -> do
|
||||
void $ updateSndIds st connId
|
||||
encConnInfo <- agentRatchetEncrypt st connId (smpEncode aMessage) e2eEncConnInfoLength
|
||||
mkConfirmation aMessage = withStore c $ \db -> runExceptT $ do
|
||||
void . liftIO $ updateSndIds db connId
|
||||
encConnInfo <- agentRatchetEncrypt db connId (smpEncode aMessage) e2eEncConnInfoLength
|
||||
pure . smpEncode $ AgentConfirmation {agentVersion, e2eEncryption, encConnInfo}
|
||||
mkAgentMessage :: Version -> m AgentMessage
|
||||
mkAgentMessage 1 = pure $ AgentConnInfo connInfo
|
||||
@@ -1018,34 +1022,34 @@ enqueueConfirmation c cData@ConnData {connId, connAgentVersion} sq connInfo e2eE
|
||||
queuePendingMsgs c connId sq [msgId]
|
||||
where
|
||||
storeConfirmation :: m InternalId
|
||||
storeConfirmation = withStore c $ \st -> do
|
||||
storeConfirmation = withStore c $ \db -> runExceptT $ do
|
||||
internalTs <- liftIO getCurrentTime
|
||||
(internalId, internalSndId, prevMsgHash) <- updateSndIds st connId
|
||||
(internalId, internalSndId, prevMsgHash) <- liftIO $ updateSndIds db connId
|
||||
let agentMsg = AgentConnInfo connInfo
|
||||
agentMsgStr = smpEncode agentMsg
|
||||
internalHash = C.sha256Hash agentMsgStr
|
||||
encConnInfo <- agentRatchetEncrypt st connId agentMsgStr e2eEncConnInfoLength
|
||||
encConnInfo <- agentRatchetEncrypt db connId agentMsgStr e2eEncConnInfoLength
|
||||
let msgBody = smpEncode $ AgentConfirmation {agentVersion = connAgentVersion, e2eEncryption, encConnInfo}
|
||||
msgType = agentMessageType agentMsg
|
||||
msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash}
|
||||
createSndMsg st connId msgData
|
||||
liftIO $ createSndMsg db connId msgData
|
||||
pure internalId
|
||||
|
||||
-- encoded AgentMessage -> encoded EncAgentMessage
|
||||
agentRatchetEncrypt :: AgentStoreMonad m => SQLiteStore -> ConnId -> ByteString -> Int -> m ByteString
|
||||
agentRatchetEncrypt st connId msg paddedLen = do
|
||||
rc <- getRatchet st connId
|
||||
(encMsg, rc') <- liftError (SEAgentError . cryptoError) $ CR.rcEncrypt rc paddedLen msg
|
||||
updateRatchet st connId rc' CR.SMDNoChange
|
||||
agentRatchetEncrypt :: DB.Connection -> ConnId -> ByteString -> Int -> ExceptT StoreError IO ByteString
|
||||
agentRatchetEncrypt db connId msg paddedLen = do
|
||||
rc <- ExceptT $ getRatchet db connId
|
||||
(encMsg, rc') <- liftE (SEAgentError . cryptoError) $ CR.rcEncrypt rc paddedLen msg
|
||||
liftIO $ updateRatchet db connId rc' CR.SMDNoChange
|
||||
pure encMsg
|
||||
|
||||
-- encoded EncAgentMessage -> encoded AgentMessage
|
||||
agentRatchetDecrypt :: AgentStoreMonad m => SQLiteStore -> ConnId -> ByteString -> m ByteString
|
||||
agentRatchetDecrypt st connId encAgentMsg = do
|
||||
rc <- getRatchet st connId
|
||||
skipped <- getSkippedMsgKeys st connId
|
||||
(agentMsgBody_, rc', skippedDiff) <- liftError (SEAgentError . cryptoError) $ CR.rcDecrypt rc skipped encAgentMsg
|
||||
updateRatchet st connId rc' skippedDiff
|
||||
agentRatchetDecrypt :: DB.Connection -> ConnId -> ByteString -> ExceptT StoreError IO ByteString
|
||||
agentRatchetDecrypt db connId encAgentMsg = do
|
||||
rc <- ExceptT $ getRatchet db connId
|
||||
skipped <- liftIO $ getSkippedMsgKeys db connId
|
||||
(agentMsgBody_, rc', skippedDiff) <- liftE (SEAgentError . cryptoError) $ CR.rcDecrypt rc skipped encAgentMsg
|
||||
liftIO $ updateRatchet db connId rc' skippedDiff
|
||||
liftEither $ first (SEAgentError . cryptoError) agentMsgBody_
|
||||
|
||||
newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> m SndQueue
|
||||
|
||||
@@ -50,6 +50,7 @@ module Simplex.Messaging.Agent.Client
|
||||
endAgentOperation,
|
||||
notifyAgentPhaseChanged,
|
||||
withStore,
|
||||
withStore',
|
||||
)
|
||||
where
|
||||
|
||||
@@ -72,11 +73,12 @@ import Data.Set (Set)
|
||||
import Data.Text.Encoding
|
||||
import Data.Word (Word16)
|
||||
import Database.SQLite.Simple (SQLError)
|
||||
import qualified Database.SQLite.Simple as DB
|
||||
import Simplex.Messaging.Agent.Env.SQLite
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import Simplex.Messaging.Agent.RetryInterval
|
||||
import Simplex.Messaging.Agent.Store
|
||||
import Simplex.Messaging.Agent.Store.SQLite (AgentStoreMonad, SQLiteStore (..))
|
||||
import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore (..), withTransaction)
|
||||
import Simplex.Messaging.Client
|
||||
import Simplex.Messaging.Client.Agent ()
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
@@ -644,20 +646,19 @@ notifyAgentPhaseChanged AgentClient {subQ, agentEnv = Env {agentPhase, agentOper
|
||||
writeTBQueue subQ ("", "", PHASE p)
|
||||
writeTVar agentPhase (p, True)
|
||||
|
||||
withStore :: AgentMonad m => AgentClient -> (forall m'. AgentStoreMonad m' => SQLiteStore -> m' a) -> m a
|
||||
withStore' :: AgentMonad m => AgentClient -> (DB.Connection -> IO a) -> m a
|
||||
withStore' c action = withStore c $ fmap Right . action
|
||||
|
||||
withStore :: AgentMonad m => AgentClient -> (DB.Connection -> IO (Either StoreError a)) -> m a
|
||||
withStore c action = do
|
||||
st <- asks store
|
||||
atomically $ beginAgentOperation c AODatabase
|
||||
r <- runExceptT (action st `E.catch` handleInternal)
|
||||
r <- liftIO $ withTransaction st action `E.catch` handleInternal
|
||||
atomically $ endAgentOperation c AODatabase
|
||||
case r of
|
||||
Right res -> pure res
|
||||
Left e -> throwError $ storeError e
|
||||
liftEither $ first storeError r
|
||||
where
|
||||
-- TODO when parsing exception happens in store, the agent hangs;
|
||||
-- changing SQLError to SomeException does not help
|
||||
handleInternal :: (MonadError StoreError m') => SQLError -> m' a
|
||||
handleInternal e = throwError . SEInternal $ bshow e
|
||||
handleInternal :: SQLError -> IO (Either StoreError a)
|
||||
handleInternal = pure . Left . SEInternal . bshow
|
||||
storeError :: StoreError -> AgentErrorType
|
||||
storeError = \case
|
||||
SEConnNotFound -> CONN NOT_FOUND
|
||||
|
||||
@@ -18,7 +18,7 @@ where
|
||||
import Control.Concurrent.Async (Async, uninterruptibleCancel)
|
||||
import Control.Concurrent.STM (stateTVar)
|
||||
import Control.Monad
|
||||
import Control.Monad.Except (runExceptT)
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.IO.Unlift (MonadUnliftIO)
|
||||
import Control.Monad.Reader
|
||||
import Data.Bifunctor (first)
|
||||
@@ -30,6 +30,7 @@ import Simplex.Messaging.Agent.Env.SQLite
|
||||
import Simplex.Messaging.Agent.Protocol (ConnId)
|
||||
import qualified Simplex.Messaging.Agent.Protocol as AP
|
||||
import Simplex.Messaging.Agent.Store
|
||||
import Simplex.Messaging.Agent.Store.SQLite
|
||||
import Simplex.Messaging.Client.Agent ()
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Notifications.Client
|
||||
@@ -58,18 +59,20 @@ processNtfSub c (connId, cmd) = do
|
||||
case cmd of
|
||||
NSCCreate -> do
|
||||
-- TODO merge getNtfSubscription and getRcvQueue into one method to read both in same transaction?
|
||||
sub_ <- withStore c $ \st -> getNtfSubscription st connId
|
||||
RcvQueue {notifierId, server = smpServer} <- withStore c $ \st -> getRcvQueue st connId
|
||||
(sub_, RcvQueue {notifierId, server = smpServer}) <- withStore c $ \db -> runExceptT $ do
|
||||
sub_ <- liftIO $ getNtfSubscription db connId
|
||||
q <- ExceptT $ getRcvQueue db connId
|
||||
pure (sub_, q)
|
||||
case (sub_, ntfServer_) of
|
||||
(Nothing, Just ntfServer) -> do
|
||||
currentTime <- liftIO getCurrentTime
|
||||
case notifierId of
|
||||
(Just nId) -> do
|
||||
let newSub = newNtfSubscription connId smpServer (Just nId) ntfServer NASKey currentTime
|
||||
withStore c $ \st -> createNtfSubscription st newSub (NtfSubAction NSACreate)
|
||||
withStore' c $ \db -> createNtfSubscription db newSub (NtfSubAction NSACreate)
|
||||
_ -> do
|
||||
let newSub = newNtfSubscription connId smpServer Nothing ntfServer NASNew currentTime
|
||||
withStore c $ \st -> createNtfSubscription st newSub (NtfSubSMPAction NSAKey)
|
||||
withStore' c $ \db -> createNtfSubscription db newSub (NtfSubSMPAction NSAKey)
|
||||
-- TODO optimize?
|
||||
-- TODO - read action in getNtfSubscription and decide which worker to create
|
||||
-- TODO - SMP worker can create Ntf worker on NKEY completion
|
||||
@@ -87,7 +90,7 @@ processNtfSub c (connId, cmd) = do
|
||||
_ -> pure () -- error - notification server not configured
|
||||
NSCDelete -> do
|
||||
-- TODO delete notifier ID and Key from SMP server (SDEL, then NDEL)
|
||||
withStore c $ \st -> markNtfSubscriptionForDeletion st connId
|
||||
withStore c (`markNtfSubscriptionForDeletion` connId)
|
||||
case ntfServer_ of
|
||||
(Just ntfServer) -> addNtfWorker ntfServer
|
||||
_ -> pure ()
|
||||
@@ -116,7 +119,7 @@ runNtfWorker c srv doWork = forever $ do
|
||||
void . atomically $ readTMVar doWork
|
||||
getNtfToken >>= \case
|
||||
Just tkn@NtfToken {ntfTokenId = Just tknId, ntfTknStatus} -> do
|
||||
nextSub_ <- withStore c (`getNextNtfSubAction` srv)
|
||||
nextSub_ <- withStore' c (`getNextNtfSubAction` srv)
|
||||
ts <- liftIO getCurrentTime
|
||||
case nextSub_ of
|
||||
Nothing -> noWorkToDo
|
||||
@@ -128,8 +131,8 @@ runNtfWorker c srv doWork = forever $ do
|
||||
| ntfTknStatus == NTActive -> do
|
||||
nSubId <- agentNtfCreateSubscription c tknId tkn (SMPQueueNtf smpServer nId) ntfPrivKey
|
||||
let actionTs = addUTCTime 30 ts
|
||||
withStore c $ \st ->
|
||||
updateNtfSubscription st connId ntfSub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew, ntfSubActionTs = actionTs} (NtfSubAction NSACheck)
|
||||
withStore' c $ \db ->
|
||||
updateNtfSubscription db connId ntfSub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew, ntfSubActionTs = actionTs} (NtfSubAction NSACheck)
|
||||
| otherwise -> ntfInternalError c connId "NSACreate - token not active"
|
||||
_ -> ntfInternalError c connId "NSACreate - no notifier key or ID"
|
||||
NSACheck -> case ntfSubId of
|
||||
@@ -148,8 +151,8 @@ runNtfWorker c srv doWork = forever $ do
|
||||
let nextCheckTs = addUTCTime checkInterval ts
|
||||
updateSub (NASCreated toStatus) (NtfSubAction NSACheck) nextCheckTs
|
||||
updateSub toStatus toAction actionTs =
|
||||
withStore c $ \st ->
|
||||
updateNtfSubscription st connId ntfSub {ntfSubStatus = toStatus, ntfSubActionTs = actionTs} toAction
|
||||
withStore' c $ \db ->
|
||||
updateNtfSubscription db connId ntfSub {ntfSubStatus = toStatus, ntfSubActionTs = actionTs} toAction
|
||||
_ -> noWorkToDo
|
||||
delay <- asks $ ntfWorkerThrottle . config
|
||||
liftIO $ threadDelay delay
|
||||
@@ -161,7 +164,7 @@ runNtfSMPWorker c srv doWork = forever $ do
|
||||
void . atomically $ readTMVar doWork
|
||||
getNtfToken >>= \case
|
||||
Just NtfToken {ntfTknStatus} -> do
|
||||
nextSub_ <- withStore c (`getNextNtfSubSMPAction` srv)
|
||||
nextSub_ <- withStore' c (`getNextNtfSubSMPAction` srv)
|
||||
ts <- liftIO getCurrentTime
|
||||
case nextSub_ of
|
||||
Nothing -> noWorkToDo
|
||||
@@ -176,15 +179,15 @@ runNtfSMPWorker c srv doWork = forever $ do
|
||||
_ -> do
|
||||
C.SignAlg a <- asks (cmdSignAlg . config)
|
||||
(ntfPubKey, ntfPrivKey) <- liftIO $ C.generateSignatureKeyPair a
|
||||
withStore c $ \st -> setRcvQueueNotifierKey st connId ntfPubKey ntfPrivKey
|
||||
withStore' c $ \db -> setRcvQueueNotifierKey db connId ntfPubKey ntfPrivKey
|
||||
enableNotificationsWithNKey ntfPubKey
|
||||
| otherwise -> ntfInternalError c connId "NSAKey - token not active"
|
||||
where
|
||||
enableNotificationsWithNKey ntfPubKey = do
|
||||
nId <- enableQueueNotifications c rq ntfPubKey
|
||||
withStore c $ \st -> do
|
||||
setRcvQueueNotifierId st connId nId
|
||||
updateNtfSubscription st connId ntfSub {ntfQueueId = Just nId, ntfSubStatus = NASKey, ntfSubActionTs = ts} (NtfSubAction NSACreate)
|
||||
withStore' c $ \db -> do
|
||||
setRcvQueueNotifierId db connId nId
|
||||
updateNtfSubscription db connId ntfSub {ntfQueueId = Just nId, ntfSubStatus = NASKey, ntfSubActionTs = ts} (NtfSubAction NSACreate)
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer)
|
||||
_ -> noWorkToDo
|
||||
@@ -211,7 +214,7 @@ diffInMicros a b = (`div` 1000000) . fromInteger . fromPico . nominalDiffTimeToS
|
||||
|
||||
ntfInternalError :: AgentMonad m => AgentClient -> ConnId -> String -> m ()
|
||||
ntfInternalError c@AgentClient {subQ} connId internalErrStr = do
|
||||
withStore c $ \st -> setNullNtfSubscriptionAction st connId
|
||||
withStore' c $ \db -> setNullNtfSubscriptionAction db connId
|
||||
atomically $ writeTBQueue subQ ("", connId, AP.ERR $ AP.INTERNAL internalErrStr)
|
||||
|
||||
getNtfToken :: AgentMonad m => m (Maybe NtfToken)
|
||||
|
||||
@@ -9,9 +9,7 @@
|
||||
|
||||
module Simplex.Messaging.Agent.Store where
|
||||
|
||||
import Control.Concurrent.STM (TVar)
|
||||
import Control.Exception (Exception)
|
||||
import Crypto.Random (ChaChaDRG)
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import Data.Int (Int64)
|
||||
import Data.Kind (Type)
|
||||
@@ -19,9 +17,7 @@ import Data.Time (UTCTime)
|
||||
import Data.Type.Equality
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Crypto.Ratchet (RatchetX448, SkippedMsgDiff, SkippedMsgKeys)
|
||||
import Simplex.Messaging.Notifications.Client
|
||||
import Simplex.Messaging.Notifications.Protocol (DeviceToken, NtfTknStatus, NtfTokenId, SMPQueueNtf)
|
||||
import Simplex.Messaging.Crypto.Ratchet (RatchetX448)
|
||||
import Simplex.Messaging.Protocol
|
||||
( MsgBody,
|
||||
MsgFlags,
|
||||
@@ -36,80 +32,6 @@ import Simplex.Messaging.Protocol
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Version
|
||||
|
||||
-- * Store management
|
||||
|
||||
-- | Store class type. Defines store access methods for implementations.
|
||||
class Monad m => MonadAgentStore s m where
|
||||
-- Queue and Connection management
|
||||
createRcvConn :: s -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> m ConnId
|
||||
createSndConn :: s -> TVar ChaChaDRG -> ConnData -> SndQueue -> m ConnId
|
||||
getConn :: s -> ConnId -> m SomeConn
|
||||
getRcvConn :: s -> SMPServer -> SMP.RecipientId -> m SomeConn
|
||||
deleteConn :: s -> ConnId -> m ()
|
||||
upgradeRcvConnToDuplex :: s -> ConnId -> SndQueue -> m ()
|
||||
upgradeSndConnToDuplex :: s -> ConnId -> RcvQueue -> m ()
|
||||
setRcvQueueStatus :: s -> RcvQueue -> QueueStatus -> m ()
|
||||
setRcvQueueConfirmedE2E :: s -> RcvQueue -> C.DhSecretX25519 -> m ()
|
||||
setSndQueueStatus :: s -> SndQueue -> QueueStatus -> m ()
|
||||
getRcvQueue :: s -> ConnId -> m RcvQueue
|
||||
|
||||
-- RcvQueue notifier key and ID
|
||||
setRcvQueueNotifierKey :: s -> ConnId -> NtfPublicVerifyKey -> NtfPrivateSignKey -> m ()
|
||||
setRcvQueueNotifierId :: s -> ConnId -> NotifierId -> m ()
|
||||
|
||||
-- Confirmations
|
||||
createConfirmation :: s -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId
|
||||
acceptConfirmation :: s -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation
|
||||
getAcceptedConfirmation :: s -> ConnId -> m AcceptedConfirmation
|
||||
removeConfirmations :: s -> ConnId -> m ()
|
||||
setHandshakeVersion :: s -> ConnId -> Version -> Bool -> m ()
|
||||
|
||||
-- Invitations - sent via Contact connections
|
||||
createInvitation :: s -> TVar ChaChaDRG -> NewInvitation -> m InvitationId
|
||||
getInvitation :: s -> InvitationId -> m Invitation
|
||||
acceptInvitation :: s -> InvitationId -> ConnInfo -> m ()
|
||||
deleteInvitation :: s -> ConnId -> InvitationId -> m ()
|
||||
|
||||
-- Msg management
|
||||
updateRcvIds :: s -> ConnId -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)
|
||||
createRcvMsg :: s -> ConnId -> RcvMsgData -> m ()
|
||||
updateSndIds :: s -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)
|
||||
createSndMsg :: s -> ConnId -> SndMsgData -> m ()
|
||||
getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, PendingMsgData)
|
||||
getPendingMsgs :: s -> ConnId -> m [InternalId]
|
||||
setMsgUserAck :: s -> ConnId -> InternalId -> m MsgId
|
||||
getLastMsg :: s -> ConnId -> SMP.MsgId -> m (Maybe RcvMsg)
|
||||
deleteMsg :: s -> ConnId -> InternalId -> m ()
|
||||
|
||||
-- Double ratchet persistence
|
||||
createRatchetX3dhKeys :: s -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> m ()
|
||||
getRatchetX3dhKeys :: s -> ConnId -> m (C.PrivateKeyX448, C.PrivateKeyX448)
|
||||
createRatchet :: s -> ConnId -> RatchetX448 -> m ()
|
||||
getRatchet :: s -> ConnId -> m RatchetX448
|
||||
getSkippedMsgKeys :: s -> ConnId -> m SkippedMsgKeys
|
||||
updateRatchet :: s -> ConnId -> RatchetX448 -> SkippedMsgDiff -> m ()
|
||||
|
||||
-- Notification device token persistence
|
||||
createNtfToken :: s -> NtfToken -> m ()
|
||||
getDeviceNtfToken :: s -> DeviceToken -> m (Maybe NtfToken, [NtfToken])
|
||||
updateNtfTokenRegistration :: s -> NtfToken -> NtfTokenId -> C.DhSecretX25519 -> m ()
|
||||
updateNtfToken :: s -> NtfToken -> NtfTknStatus -> Maybe NtfTknAction -> m ()
|
||||
removeNtfToken :: s -> NtfToken -> m ()
|
||||
|
||||
-- Notification subscription persistence
|
||||
getNtfSubscription :: s -> ConnId -> m (Maybe NtfSubscription)
|
||||
createNtfSubscription :: s -> NtfSubscription -> NtfSubOrSMPAction -> m ()
|
||||
markNtfSubscriptionForDeletion :: s -> ConnId -> m ()
|
||||
updateNtfSubscription :: s -> ConnId -> NtfSubscription -> NtfSubOrSMPAction -> m ()
|
||||
setNullNtfSubscriptionAction :: s -> ConnId -> m ()
|
||||
deleteNtfSubscription :: s -> ConnId -> m ()
|
||||
getNextNtfSubAction :: s -> NtfServer -> m (Maybe (NtfSubscription, NtfSubAction, RcvQueue))
|
||||
getNextNtfSubSMPAction :: s -> SMPServer -> m (Maybe (NtfSubscription, NtfSubSMPAction, RcvQueue))
|
||||
|
||||
-- Notification decryption
|
||||
getActiveNtfToken :: s -> m (Maybe NtfToken)
|
||||
getNtfRcvQueue :: s -> SMPQueueNtf -> m (ConnId, RcvDhSecret)
|
||||
|
||||
-- * Queue types
|
||||
|
||||
-- | A receive queue. SMP queue through which the agent receives messages from a sender.
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -56,6 +56,10 @@ tryE :: Monad m => ExceptT e m a -> ExceptT e m (Either e a)
|
||||
tryE m = (Right <$> m) `catchE` (pure . Left)
|
||||
{-# INLINE tryE #-}
|
||||
|
||||
liftE :: (e -> e') -> ExceptT e IO a -> ExceptT e' IO a
|
||||
liftE f a = ExceptT $ first f <$> runExceptT a
|
||||
{-# INLINE liftE #-}
|
||||
|
||||
ifM :: Monad m => m Bool -> m a -> m a -> m a
|
||||
ifM ba t f = ba >>= \b -> if b then t else f
|
||||
{-# INLINE ifM #-}
|
||||
|
||||
Reference in new issue
Block a user