diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index d863c23ad..12441a15d 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -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 " 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 diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index ceae7fc56..6248cd321 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -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 diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index 84b51ec5b..db82204b3 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs @@ -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) diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index e26dbb925..45110705e 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/src/Simplex/Messaging/Agent/Store.hs @@ -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. diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 7cb48d350..56aa2ccf0 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -20,20 +20,82 @@ module Simplex.Messaging.Agent.Store.SQLite ( SQLiteStore (..), - AgentStoreMonad, createSQLiteStore, connectSQLiteStore, + + -- * Queues and connections + createRcvConn, + createSndConn, + getConn, + getRcvConn, + deleteConn, + upgradeRcvConnToDuplex, + upgradeSndConnToDuplex, + setRcvQueueStatus, + setRcvQueueConfirmedE2E, + setSndQueueStatus, + getRcvQueue, + -- RcvQueue notifier key and ID + setRcvQueueNotifierKey, + setRcvQueueNotifierId, + -- Confirmations + createConfirmation, + acceptConfirmation, + getAcceptedConfirmation, + removeConfirmations, + setHandshakeVersion, + -- Invitations - sent via Contact connections + createInvitation, + getInvitation, + acceptInvitation, + deleteInvitation, + -- Messages + updateRcvIds, + createRcvMsg, + updateSndIds, + createSndMsg, + getPendingMsgData, + getPendingMsgs, + setMsgUserAck, + getLastMsg, + deleteMsg, + -- Double ratchet persistence + createRatchetX3dhKeys, + getRatchetX3dhKeys, + createRatchet, + getRatchet, + getSkippedMsgKeys, + updateRatchet, + -- Notification device token persistence + createNtfToken, + getDeviceNtfToken, + updateNtfTokenRegistration, + updateNtfToken, + removeNtfToken, + -- Notification subscription persistence + getNtfSubscription, + createNtfSubscription, + markNtfSubscriptionForDeletion, + updateNtfSubscription, + setNullNtfSubscriptionAction, + deleteNtfSubscription, + getNextNtfSubAction, + getNextNtfSubSMPAction, + getActiveNtfToken, + getNtfRcvQueue, + + -- * utilities withConnection, withTransaction, firstRow, + firstRow', + maybeFirstRow, ) where import Control.Concurrent (threadDelay) -import Control.Concurrent.STM -import Control.Exception (bracket) +import Control.Concurrent.STM (stateTVar) import Control.Monad.Except -import Control.Monad.IO.Unlift (MonadUnliftIO) import Crypto.Random (ChaChaDRG, randomBytesGenerate) import Data.Bifunctor (first, second) import Data.ByteString (ByteString) @@ -47,11 +109,12 @@ import Data.Text (Text) import qualified Data.Text as T import Data.Text.Encoding (decodeLatin1, encodeUtf8) import Data.Time.Clock (UTCTime, getCurrentTime) -import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), SQLError, ToRow, field, (:.) (..)) +import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), Query (..), SQLError, ToRow, field, (:.) (..)) import qualified Database.SQLite.Simple as DB import Database.SQLite.Simple.FromField import Database.SQLite.Simple.QQ (sql) import Database.SQLite.Simple.ToField (ToField (..)) +import qualified Database.SQLite3 as SQLite3 import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.Store import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration) @@ -65,13 +128,15 @@ import Simplex.Messaging.Notifications.Protocol (DeviceToken (..), NtfTknStatus import Simplex.Messaging.Parsers (blobFieldParser, fromTextField_) import Simplex.Messaging.Protocol (MsgBody, MsgFlags, NotifierId, NtfPrivateSignKey, NtfPublicVerifyKey, ProtocolServer (..), RcvDhSecret) import qualified Simplex.Messaging.Protocol as SMP -import Simplex.Messaging.Util (bshow, eitherToMaybe, liftIOEither) +import Simplex.Messaging.Util (bshow, eitherToMaybe, ($>>=), (<$$>)) import Simplex.Messaging.Version import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist) import System.Exit (exitFailure) import System.FilePath (takeDirectory) import System.IO (hFlush, stdout) +import UnliftIO.Exception (bracket) import qualified UnliftIO.Exception as E +import UnliftIO.STM -- * SQLite Store implementation @@ -130,24 +195,27 @@ connectSQLiteStore dbFilePath = do connectDB :: FilePath -> IO DB.Connection connectDB path = do dbConn <- DB.open path - DB.execute_ dbConn "PRAGMA foreign_keys = ON;" - -- DB.execute_ dbConn "PRAGMA trusted_schema = OFF;" - DB.execute_ dbConn "PRAGMA secure_delete = ON;" - DB.execute_ dbConn "PRAGMA auto_vacuum = FULL;" + SQLite3.exec (DB.connectionHandle dbConn) . fromQuery $ + [sql| + PRAGMA foreign_keys = ON; + -- PRAGMA trusted_schema = OFF; + PRAGMA secure_delete = ON; + PRAGMA auto_vacuum = FULL; + |] -- _printPragmas dbConn path pure dbConn -_printPragmas :: DB.Connection -> FilePath -> IO () -_printPragmas db path = do - foreign_keys <- DB.query_ db "PRAGMA foreign_keys;" :: IO [[Int]] - print $ path <> " foreign_keys: " <> show foreign_keys - -- when run via sqlite-simple query for trusted_schema seems to return empty list - trusted_schema <- DB.query_ db "PRAGMA trusted_schema;" :: IO [[Int]] - print $ path <> " trusted_schema: " <> show trusted_schema - secure_delete <- DB.query_ db "PRAGMA secure_delete;" :: IO [[Int]] - print $ path <> " secure_delete: " <> show secure_delete - auto_vacuum <- DB.query_ db "PRAGMA auto_vacuum;" :: IO [[Int]] - print $ path <> " auto_vacuum: " <> show auto_vacuum +-- _printPragmas :: DB.Connection -> FilePath -> IO () +-- _printPragmas db path = do +-- foreign_keys <- DB.query_ db "PRAGMA foreign_keys;" :: IO [[Int]] +-- print $ path <> " foreign_keys: " <> show foreign_keys +-- -- when run via sqlite-simple query for trusted_schema seems to return empty list +-- trusted_schema <- DB.query_ db "PRAGMA trusted_schema;" :: IO [[Int]] +-- print $ path <> " trusted_schema: " <> show trusted_schema +-- secure_delete <- DB.query_ db "PRAGMA secure_delete;" :: IO [[Int]] +-- print $ path <> " secure_delete: " <> show secure_delete +-- auto_vacuum <- DB.query_ db "PRAGMA auto_vacuum;" :: IO [[Int]] +-- print $ path <> " auto_vacuum: " <> show auto_vacuum checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a) checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err) @@ -164,7 +232,7 @@ withConnection SQLiteStore {dbConnection} = (atomically . putTMVar dbConnection) withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a -withTransaction st action = withConnection st $ loop 100 100_000 +withTransaction st action = withConnection st $ loop 500 2_000_000 where loop :: Int -> Int -> DB.Connection -> IO a loop t tLim db = @@ -176,694 +244,621 @@ withTransaction st action = withConnection st $ loop 100 100_000 else E.throwIO e createConn_ :: - (MonadUnliftIO m, MonadError StoreError m) => - SQLiteStore -> TVar ChaChaDRG -> ConnData -> - (DB.Connection -> ByteString -> IO ()) -> - m ByteString -createConn_ st gVar cData create = - liftIOEither . checkConstraint SEConnDuplicate . withTransaction st $ \db -> - case cData of - ConnData {connId = ""} -> createWithRandomId gVar $ create db - ConnData {connId} -> create db connId $> Right connId + (ByteString -> IO ()) -> + IO (Either StoreError ByteString) +createConn_ gVar cData create = checkConstraint SEConnDuplicate $ case cData of + ConnData {connId = ""} -> createWithRandomId gVar create + ConnData {connId} -> create connId $> Right connId -type AgentStoreMonad m = (MonadUnliftIO m, MonadError StoreError m, MonadAgentStore SQLiteStore m) +createRcvConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> IO (Either StoreError ConnId) +createRcvConn db gVar cData q@RcvQueue {server} cMode = + createConn_ gVar cData $ \connId -> do + upsertServer_ db server + DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, duplex_handshake) VALUES (?, ?, ?, ?)" (connId, cMode, connAgentVersion cData, duplexHandshake cData) + insertRcvQueue_ db connId q -instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteStore m where - createRcvConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> m ConnId - createRcvConn st gVar cData q@RcvQueue {server} cMode = - createConn_ st gVar cData $ \db connId -> do +createSndConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> SndQueue -> IO (Either StoreError ConnId) +createSndConn db gVar cData q@SndQueue {server} = + createConn_ gVar cData $ \connId -> do + upsertServer_ db server + DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, duplex_handshake) VALUES (?, ?, ?, ?)" (connId, SCMInvitation, connAgentVersion cData, duplexHandshake cData) + insertSndQueue_ db connId q + +getRcvConn :: DB.Connection -> SMPServer -> SMP.RecipientId -> IO (Either StoreError SomeConn) +getRcvConn db ProtocolServer {host, port} rcvId = + DB.queryNamed + db + [sql| + SELECT q.conn_id + FROM rcv_queues q + WHERE q.host = :host AND q.port = :port AND q.rcv_id = :rcv_id; + |] + [":host" := host, ":port" := port, ":rcv_id" := rcvId] + >>= \case + [Only connId] -> getConn db connId + _ -> pure $ Left SEConnNotFound + +deleteConn :: DB.Connection -> ConnId -> IO () +deleteConn db connId = + DB.executeNamed + db + "DELETE FROM connections WHERE conn_id = :conn_id;" + [":conn_id" := connId] + +upgradeRcvConnToDuplex :: DB.Connection -> ConnId -> SndQueue -> IO (Either StoreError ()) +upgradeRcvConnToDuplex db connId sq@SndQueue {server} = + getConn db connId $>>= \case + (SomeConn _ RcvConnection {}) -> do upsertServer_ db server - DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, duplex_handshake) VALUES (?, ?, ?, ?)" (connId, cMode, connAgentVersion cData, duplexHandshake cData) - insertRcvQueue_ db connId q + insertSndQueue_ db connId sq + pure $ Right () + (SomeConn c _) -> pure . Left . SEBadConnType $ connType c - createSndConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> SndQueue -> m ConnId - createSndConn st gVar cData q@SndQueue {server} = - createConn_ st gVar cData $ \db connId -> do +upgradeSndConnToDuplex :: DB.Connection -> ConnId -> RcvQueue -> IO (Either StoreError ()) +upgradeSndConnToDuplex db connId rq@RcvQueue {server} = + getConn db connId >>= \case + Right (SomeConn _ SndConnection {}) -> do upsertServer_ db server - DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, duplex_handshake) VALUES (?, ?, ?, ?)" (connId, SCMInvitation, connAgentVersion cData, duplexHandshake cData) - insertSndQueue_ db connId q + insertRcvQueue_ db connId rq + pure $ Right () + Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c + _ -> pure $ Left SEConnNotFound - getConn :: SQLiteStore -> ConnId -> m SomeConn - getConn st connId = - liftIOEither . withTransaction st $ \db -> - getConn_ db connId +setRcvQueueStatus :: DB.Connection -> RcvQueue -> QueueStatus -> IO () +setRcvQueueStatus db RcvQueue {rcvId, server = ProtocolServer {host, port}} status = + -- ? return error if queue does not exist? + DB.executeNamed + db + [sql| + UPDATE rcv_queues + SET status = :status + WHERE host = :host AND port = :port AND rcv_id = :rcv_id; + |] + [":status" := status, ":host" := host, ":port" := port, ":rcv_id" := rcvId] - getRcvConn :: SQLiteStore -> SMPServer -> SMP.RecipientId -> m SomeConn - getRcvConn st ProtocolServer {host, port} rcvId = - liftIOEither . withTransaction st $ \db -> - DB.queryNamed +setRcvQueueConfirmedE2E :: DB.Connection -> RcvQueue -> C.DhSecretX25519 -> IO () +setRcvQueueConfirmedE2E db RcvQueue {rcvId, server = ProtocolServer {host, port}} e2eDhSecret = + DB.executeNamed + db + [sql| + UPDATE rcv_queues + SET e2e_dh_secret = :e2e_dh_secret, + status = :status + WHERE host = :host AND port = :port AND rcv_id = :rcv_id + |] + [ ":status" := Confirmed, + ":e2e_dh_secret" := e2eDhSecret, + ":host" := host, + ":port" := port, + ":rcv_id" := rcvId + ] + +setSndQueueStatus :: DB.Connection -> SndQueue -> QueueStatus -> IO () +setSndQueueStatus db SndQueue {sndId, server = ProtocolServer {host, port}} status = + -- ? return error if queue does not exist? + DB.executeNamed + db + [sql| + UPDATE snd_queues + SET status = :status + WHERE host = :host AND port = :port AND snd_id = :snd_id; + |] + [":status" := status, ":host" := host, ":port" := port, ":snd_id" := sndId] + +getRcvQueue :: DB.Connection -> ConnId -> IO (Either StoreError RcvQueue) +getRcvQueue db connId = + maybe (Left SEConnNotFound) Right <$> getRcvQueueByConnId_ db connId + +setRcvQueueNotifierKey :: DB.Connection -> ConnId -> NtfPublicVerifyKey -> NtfPrivateSignKey -> IO () +setRcvQueueNotifierKey db connId ntfPublicKey ntfPrivateKey = + DB.execute + db + [sql| + UPDATE rcv_queues + SET ntf_public_key = ?, ntf_private_key = ? + WHERE conn_id = ? + |] + (ntfPublicKey, ntfPrivateKey, connId) + +setRcvQueueNotifierId :: DB.Connection -> ConnId -> NotifierId -> IO () +setRcvQueueNotifierId db connId nId = + DB.execute + db + [sql| + UPDATE rcv_queues + SET ntf_id = ? + WHERE conn_id = ? + |] + (nId, connId) + +createConfirmation :: DB.Connection -> TVar ChaChaDRG -> NewConfirmation -> IO (Either StoreError ConfirmationId) +createConfirmation db gVar NewConfirmation {connId, senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues}, ratchetState} = + createWithRandomId gVar $ \confirmationId -> + DB.execute + db + [sql| + INSERT INTO conn_confirmations + (confirmation_id, conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, smp_reply_queues, accepted) VALUES (?, ?, ?, ?, ?, ?, ?, 0); + |] + (confirmationId, connId, senderKey, e2ePubKey, ratchetState, connInfo, smpReplyQueues) + +acceptConfirmation :: DB.Connection -> ConfirmationId -> ConnInfo -> IO (Either StoreError AcceptedConfirmation) +acceptConfirmation db confirmationId ownConnInfo = do + DB.executeNamed + db + [sql| + UPDATE conn_confirmations + SET accepted = 1, + own_conn_info = :own_conn_info + WHERE confirmation_id = :confirmation_id; + |] + [ ":own_conn_info" := ownConnInfo, + ":confirmation_id" := confirmationId + ] + firstRow confirmation SEConfirmationNotFound $ + DB.query + db + [sql| + SELECT conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, smp_reply_queues + FROM conn_confirmations + WHERE confirmation_id = ?; + |] + (Only confirmationId) + where + confirmation (connId, senderKey, e2ePubKey, ratchetState, connInfo, smpReplyQueues_) = + AcceptedConfirmation + { confirmationId, + connId, + senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = fromMaybe [] smpReplyQueues_}, + ratchetState, + ownConnInfo + } + +getAcceptedConfirmation :: DB.Connection -> ConnId -> IO (Either StoreError AcceptedConfirmation) +getAcceptedConfirmation db connId = + firstRow confirmation SEConfirmationNotFound $ + DB.query + db + [sql| + SELECT confirmation_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, smp_reply_queues, own_conn_info + FROM conn_confirmations + WHERE conn_id = ? AND accepted = 1; + |] + (Only connId) + where + confirmation (confirmationId, senderKey, e2ePubKey, ratchetState, connInfo, smpReplyQueues_, ownConnInfo) = + AcceptedConfirmation + { confirmationId, + connId, + senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = fromMaybe [] smpReplyQueues_}, + ratchetState, + ownConnInfo + } + +removeConfirmations :: DB.Connection -> ConnId -> IO () +removeConfirmations db connId = + DB.executeNamed + db + [sql| + DELETE FROM conn_confirmations + WHERE conn_id = :conn_id; + |] + [":conn_id" := connId] + +setHandshakeVersion :: DB.Connection -> ConnId -> Version -> Bool -> IO () +setHandshakeVersion db connId aVersion duplexHS = + DB.execute db "UPDATE connections SET smp_agent_version = ?, duplex_handshake = ? WHERE conn_id = ?" (aVersion, duplexHS, connId) + +createInvitation :: DB.Connection -> TVar ChaChaDRG -> NewInvitation -> IO (Either StoreError InvitationId) +createInvitation db gVar NewInvitation {contactConnId, connReq, recipientConnInfo} = + createWithRandomId gVar $ \invitationId -> + DB.execute + db + [sql| + INSERT INTO conn_invitations + (invitation_id, contact_conn_id, cr_invitation, recipient_conn_info, accepted) VALUES (?, ?, ?, ?, 0); + |] + (invitationId, contactConnId, connReq, recipientConnInfo) + +getInvitation :: DB.Connection -> InvitationId -> IO (Either StoreError Invitation) +getInvitation db invitationId = + firstRow invitation SEInvitationNotFound $ + DB.query + db + [sql| + SELECT contact_conn_id, cr_invitation, recipient_conn_info, own_conn_info, accepted + FROM conn_invitations + WHERE invitation_id = ? + AND accepted = 0 + |] + (Only invitationId) + where + invitation (contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted) = + Invitation {invitationId, contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted} + +acceptInvitation :: DB.Connection -> InvitationId -> ConnInfo -> IO () +acceptInvitation db invitationId ownConnInfo = + DB.executeNamed + db + [sql| + UPDATE conn_invitations + SET accepted = 1, + own_conn_info = :own_conn_info + WHERE invitation_id = :invitation_id + |] + [ ":own_conn_info" := ownConnInfo, + ":invitation_id" := invitationId + ] + +deleteInvitation :: DB.Connection -> ConnId -> InvitationId -> IO (Either StoreError ()) +deleteInvitation db contactConnId invId = + getConn db contactConnId $>>= \case + SomeConn SCContact _ -> + Right <$> DB.execute db "DELETE FROM conn_invitations WHERE contact_conn_id = ? AND invitation_id = ?" (contactConnId, invId) + _ -> pure $ Left SEConnNotFound + +updateRcvIds :: DB.Connection -> ConnId -> IO (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash) +updateRcvIds db connId = do + (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash) <- retrieveLastIdsAndHashRcv_ db connId + let internalId = InternalId $ unId lastInternalId + 1 + internalRcvId = InternalRcvId $ unRcvId lastInternalRcvId + 1 + updateLastIdsRcv_ db connId internalId internalRcvId + pure (internalId, internalRcvId, lastExternalSndId, lastRcvHash) + +createRcvMsg :: DB.Connection -> ConnId -> RcvMsgData -> IO () +createRcvMsg db connId rcvMsgData = do + insertRcvMsgBase_ db connId rcvMsgData + insertRcvMsgDetails_ db connId rcvMsgData + updateHashRcv_ db connId rcvMsgData + +updateSndIds :: DB.Connection -> ConnId -> IO (InternalId, InternalSndId, PrevSndMsgHash) +updateSndIds db connId = do + (lastInternalId, lastInternalSndId, prevSndHash) <- retrieveLastIdsAndHashSnd_ db connId + let internalId = InternalId $ unId lastInternalId + 1 + internalSndId = InternalSndId $ unSndId lastInternalSndId + 1 + updateLastIdsSnd_ db connId internalId internalSndId + pure (internalId, internalSndId, prevSndHash) + +createSndMsg :: DB.Connection -> ConnId -> SndMsgData -> IO () +createSndMsg db connId sndMsgData = do + insertSndMsgBase_ db connId sndMsgData + insertSndMsgDetails_ db connId sndMsgData + updateHashSnd_ db connId sndMsgData + +getPendingMsgData :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (Maybe RcvQueue, PendingMsgData)) +getPendingMsgData db connId msgId = do + rq_ <- getRcvQueueByConnId_ db connId + (rq_,) <$$> firstRow pendingMsgData SEMsgNotFound getMsgData_ + where + getMsgData_ = + DB.query db [sql| - SELECT q.conn_id - FROM rcv_queues q - WHERE q.host = :host AND q.port = :port AND q.rcv_id = :rcv_id; + SELECT m.msg_type, m.msg_flags, m.msg_body, m.internal_ts + FROM messages m + JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id + WHERE m.conn_id = ? AND m.internal_id = ? |] - [":host" := host, ":port" := port, ":rcv_id" := rcvId] - >>= \case - [Only connId] -> getConn_ db connId - _ -> pure $ Left SEConnNotFound + (connId, msgId) + pendingMsgData :: (AgentMessageType, MsgFlags, MsgBody, InternalTs) -> PendingMsgData + pendingMsgData (msgType, msgFlags, msgBody, internalTs) = PendingMsgData {msgId, msgType, msgFlags, msgBody, internalTs} - deleteConn :: SQLiteStore -> ConnId -> m () - deleteConn st connId = - liftIO . withTransaction st $ \db -> - DB.executeNamed - db - "DELETE FROM connections WHERE conn_id = :conn_id;" - [":conn_id" := connId] +getPendingMsgs :: DB.Connection -> ConnId -> IO [InternalId] +getPendingMsgs db connId = + map fromOnly + <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_id = ?" (Only connId) - upgradeRcvConnToDuplex :: SQLiteStore -> ConnId -> SndQueue -> m () - upgradeRcvConnToDuplex st connId sq@SndQueue {server} = - liftIOEither . withTransaction st $ \db -> - getConn_ db connId >>= \case - Right (SomeConn _ RcvConnection {}) -> do - upsertServer_ db server - insertSndQueue_ db connId sq - pure $ Right () - Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c - _ -> pure $ Left SEConnNotFound +setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError SMP.MsgId) +setMsgUserAck db connId agentMsgId = do + DB.execute db "UPDATE rcv_messages SET user_ack = ? WHERE conn_id = ? AND internal_id = ?" (True, connId, agentMsgId) + firstRow fromOnly SEMsgNotFound $ + DB.query db "SELECT broker_id FROM rcv_messages WHERE conn_id = ? AND internal_id = ?" (connId, agentMsgId) - upgradeSndConnToDuplex :: SQLiteStore -> ConnId -> RcvQueue -> m () - upgradeSndConnToDuplex st connId rq@RcvQueue {server} = - liftIOEither . withTransaction st $ \db -> - getConn_ db connId >>= \case - Right (SomeConn _ SndConnection {}) -> do - upsertServer_ db server - insertRcvQueue_ db connId rq - pure $ Right () - Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c - _ -> pure $ Left SEConnNotFound +getLastMsg :: DB.Connection -> ConnId -> SMP.MsgId -> IO (Maybe RcvMsg) +getLastMsg db connId msgId = + maybeFirstRow rcvMsg $ + DB.query + db + [sql| + SELECT + r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, + m.msg_body, r.user_ack + FROM rcv_messages r + JOIN messages m ON r.internal_id = m.internal_id + JOIN connections c ON r.conn_id = c.conn_id AND c.last_internal_msg_id = r.internal_id + WHERE r.conn_id = ? AND r.broker_id = ? + |] + (connId, msgId) + where + rcvMsg (agentMsgId, internalTs, brokerId, brokerTs, sndMsgId, integrity, msgBody, userAck) = + let msgMeta = MsgMeta {recipient = (agentMsgId, internalTs), broker = (brokerId, brokerTs), sndMsgId, integrity} + in RcvMsg {internalId = InternalId agentMsgId, msgMeta, msgBody, userAck} - setRcvQueueStatus :: SQLiteStore -> RcvQueue -> QueueStatus -> m () - setRcvQueueStatus st RcvQueue {rcvId, server = ProtocolServer {host, port}} status = - -- ? throw error if queue does not exist? - liftIO . withTransaction st $ \db -> - DB.executeNamed +deleteMsg :: DB.Connection -> ConnId -> InternalId -> IO () +deleteMsg db connId msgId = + DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId) + +createRatchetX3dhKeys :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> IO () +createRatchetX3dhKeys db connId x3dhPrivKey1 x3dhPrivKey2 = + DB.execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2) + +getRatchetX3dhKeys :: DB.Connection -> ConnId -> IO (Either StoreError (C.PrivateKeyX448, C.PrivateKeyX448)) +getRatchetX3dhKeys db connId = + fmap hasKeys $ + firstRow id SEX3dhKeysNotFound $ + DB.query db "SELECT x3dh_priv_key_1, x3dh_priv_key_2 FROM ratchets WHERE conn_id = ?" (Only connId) + where + hasKeys = \case + Right (Just k1, Just k2) -> Right (k1, k2) + _ -> Left SEX3dhKeysNotFound + +createRatchet :: DB.Connection -> ConnId -> RatchetX448 -> IO () +createRatchet db connId rc = + DB.executeNamed + db + [sql| + INSERT INTO ratchets (conn_id, ratchet_state) + VALUES (:conn_id, :ratchet_state) + ON CONFLICT (conn_id) DO UPDATE SET + ratchet_state = :ratchet_state, + x3dh_priv_key_1 = NULL, + x3dh_priv_key_2 = NULL + |] + [":conn_id" := connId, ":ratchet_state" := rc] + +getRatchet :: DB.Connection -> ConnId -> IO (Either StoreError RatchetX448) +getRatchet db connId = + firstRow' ratchet SERatchetNotFound $ DB.query db "SELECT ratchet_state FROM ratchets WHERE conn_id = ?" (Only connId) + where + ratchet = maybe (Left SERatchetNotFound) Right . fromOnly + +getSkippedMsgKeys :: DB.Connection -> ConnId -> IO SkippedMsgKeys +getSkippedMsgKeys db connId = + skipped <$> DB.query db "SELECT header_key, msg_n, msg_key FROM skipped_messages WHERE conn_id = ?" (Only connId) + where + skipped ms = foldl' addSkippedKey M.empty ms + addSkippedKey smks (hk, msgN, mk) = M.alter (Just . addMsgKey) hk smks + where + addMsgKey = maybe (M.singleton msgN mk) (M.insert msgN mk) + +updateRatchet :: DB.Connection -> ConnId -> RatchetX448 -> SkippedMsgDiff -> IO () +updateRatchet db connId rc skipped = do + DB.execute db "UPDATE ratchets SET ratchet_state = ? WHERE conn_id = ?" (rc, connId) + case skipped of + SMDNoChange -> pure () + SMDRemove hk msgN -> + DB.execute db "DELETE FROM skipped_messages WHERE conn_id = ? AND header_key = ? AND msg_n = ?" (connId, hk, msgN) + SMDAdd smks -> + forM_ (M.assocs smks) $ \(hk, mks) -> + 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) + +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} = do + upsertNtfServer_ db srv + DB.execute + db + [sql| + INSERT INTO ntf_tokens + (provider, device_token, ntf_host, ntf_port, tkn_id, tkn_pub_key, tkn_priv_key, tkn_pub_dh_key, tkn_priv_dh_key, tkn_dh_secret, tkn_status, tkn_action) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + |] + (provider, token, host, port, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret, ntfTknStatus, ntfTknAction) + +getDeviceNtfToken :: DB.Connection -> DeviceToken -> IO (Maybe NtfToken, [NtfToken]) +getDeviceNtfToken db t = do + tokens <- + map ntfToken + <$> DB.query_ db [sql| - UPDATE rcv_queues - SET status = :status - WHERE host = :host AND port = :port AND rcv_id = :rcv_id; + SELECT s.ntf_host, s.ntf_port, s.ntf_key_hash, + t.provider, t.device_token, t.tkn_id, t.tkn_pub_key, t.tkn_priv_key, t.tkn_pub_dh_key, t.tkn_priv_dh_key, t.tkn_dh_secret, t.tkn_status, t.tkn_action + FROM ntf_tokens t + JOIN ntf_servers s USING (ntf_host, ntf_port) |] - [":status" := status, ":host" := host, ":port" := port, ":rcv_id" := rcvId] + pure . first listToMaybe $ partition ((t ==) . deviceToken) tokens + where + ntfToken ((host, port, keyHash) :. (provider, dt, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret, ntfTknStatus, ntfTknAction)) = + let ntfServer = ProtocolServer {host, port, keyHash} + ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey) + in NtfToken {deviceToken = DeviceToken provider dt, ntfServer, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys, ntfDhSecret, ntfTknStatus, ntfTknAction} - setRcvQueueConfirmedE2E :: SQLiteStore -> RcvQueue -> C.DhSecretX25519 -> m () - setRcvQueueConfirmedE2E st RcvQueue {rcvId, server = ProtocolServer {host, port}} e2eDhSecret = - liftIO . withTransaction st $ \db -> - DB.executeNamed - db - [sql| - UPDATE rcv_queues - SET e2e_dh_secret = :e2e_dh_secret, - status = :status - WHERE host = :host AND port = :port AND rcv_id = :rcv_id - |] - [ ":status" := Confirmed, - ":e2e_dh_secret" := e2eDhSecret, - ":host" := host, - ":port" := port, - ":rcv_id" := rcvId - ] +updateNtfTokenRegistration :: DB.Connection -> NtfToken -> NtfTokenId -> C.DhSecretX25519 -> IO () +updateNtfTokenRegistration db NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} tknId ntfDhSecret = do + updatedAt <- getCurrentTime + DB.execute + db + [sql| + UPDATE ntf_tokens + SET tkn_id = ?, tkn_dh_secret = ?, tkn_status = ?, tkn_action = ?, updated_at = ? + WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ? + |] + (tknId, ntfDhSecret, NTRegistered, Nothing :: Maybe NtfTknAction, updatedAt, provider, token, host, port) - setSndQueueStatus :: SQLiteStore -> SndQueue -> QueueStatus -> m () - setSndQueueStatus st SndQueue {sndId, server = ProtocolServer {host, port}} status = - -- ? throw error if queue does not exist? - liftIO . withTransaction st $ \db -> - DB.executeNamed - db - [sql| - UPDATE snd_queues - SET status = :status - WHERE host = :host AND port = :port AND snd_id = :snd_id; - |] - [":status" := status, ":host" := host, ":port" := port, ":snd_id" := sndId] +updateNtfToken :: DB.Connection -> NtfToken -> NtfTknStatus -> Maybe NtfTknAction -> IO () +updateNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} tknStatus tknAction = do + updatedAt <- getCurrentTime + DB.execute + db + [sql| + UPDATE ntf_tokens + SET tkn_status = ?, tkn_action = ?, updated_at = ? + WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ? + |] + (tknStatus, tknAction, updatedAt, provider, token, host, port) - getRcvQueue :: SQLiteStore -> ConnId -> m RcvQueue - getRcvQueue st connId = - liftIOEither . withTransaction st $ \db -> do - rq_ <- getRcvQueueByConnId_ db connId - pure $ maybe (Left SEConnNotFound) Right rq_ +removeNtfToken :: DB.Connection -> NtfToken -> IO () +removeNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} = + DB.execute + db + [sql| + DELETE FROM ntf_tokens + WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ? + |] + (provider, token, host, port) - setRcvQueueNotifierKey :: SQLiteStore -> ConnId -> NtfPublicVerifyKey -> NtfPrivateSignKey -> m () - setRcvQueueNotifierKey st connId ntfPublicKey ntfPrivateKey = - liftIO . withTransaction st $ \db -> - DB.execute - db - [sql| - UPDATE rcv_queues - SET ntf_public_key = ?, ntf_private_key = ? - WHERE conn_id = ? - |] - (ntfPublicKey, ntfPrivateKey, connId) +getNtfSubscription :: DB.Connection -> ConnId -> IO (Maybe NtfSubscription) +getNtfSubscription db connId = + maybeFirstRow ntfSubscription $ + DB.query + db + [sql| + SELECT s.host, s.port, s.key_hash, ns.ntf_host, ns.ntf_port, ns.ntf_key_hash, + nsb.smp_ntf_id, nsb.ntf_sub_id, nsb.ntf_sub_status, nsb.ntf_sub_action_ts + FROM ntf_subscriptions nsb + JOIN servers s ON s.host = nsb.smp_host AND s.port = nsb.smp_port + JOIN ntf_servers ns USING (ntf_host, ntf_port) + WHERE nsb.conn_id = ? + |] + (Only connId) + where + ntfSubscription (smpHost, smpPort, smpKeyHash, ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs) = + let smpServer = SMPServer smpHost smpPort smpKeyHash + ntfServer = ProtocolServer ntfHost ntfPort ntfKeyHash + in NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs} - setRcvQueueNotifierId :: SQLiteStore -> ConnId -> NotifierId -> m () - setRcvQueueNotifierId st connId nId = - liftIO . withTransaction st $ \db -> - DB.execute - db - [sql| - UPDATE rcv_queues - SET ntf_id = ? - WHERE conn_id = ? - |] - (nId, connId) +createNtfSubscription :: DB.Connection -> NtfSubscription -> NtfSubOrSMPAction -> IO () +createNtfSubscription db NtfSubscription {connId, smpServer = (SMPServer host port _), ntfQueueId, ntfServer = (SMPServer ntfHost ntfPort _), ntfSubId, ntfSubStatus, ntfSubActionTs} ntfAction = + DB.execute + db + [sql| + INSERT INTO ntf_subscriptions + (conn_id, smp_host, smp_port, smp_ntf_id, ntf_host, ntf_port, ntf_sub_id, + ntf_sub_status, ntf_sub_action, ntf_sub_smp_action, ntf_sub_action_ts) + VALUES (?,?,?,?,?,?,?,?,?,?,?) + |] + ( (connId, host, port, ntfQueueId, ntfHost, ntfPort, ntfSubId) + :. (ntfSubStatus, ntfSubAction, ntfSubSMPAction, ntfSubActionTs) + ) + where + (ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction ntfAction - createConfirmation :: SQLiteStore -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId - createConfirmation st gVar NewConfirmation {connId, senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues}, ratchetState} = - liftIOEither . withTransaction st $ \db -> - createWithRandomId gVar $ \confirmationId -> +markNtfSubscriptionForDeletion :: DB.Connection -> ConnId -> IO (Either StoreError ()) +markNtfSubscriptionForDeletion _db _rcvQueue = pure $ Left SENotImplemented + +updateNtfSubscription :: DB.Connection -> ConnId -> NtfSubscription -> NtfSubOrSMPAction -> IO () +updateNtfSubscription db connId NtfSubscription {ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs} ntfAction = do + r <- maybeFirstRow fromOnly $ DB.query db "SELECT updated_by_supervisor FROM ntf_subscriptions WHERE conn_id = ?" (Only connId) + forM_ r $ \updatedBySupervisor -> do + updatedAt <- getCurrentTime + if updatedBySupervisor + then DB.execute db [sql| - INSERT INTO conn_confirmations - (confirmation_id, conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, smp_reply_queues, accepted) VALUES (?, ?, ?, ?, ?, ?, ?, 0); + UPDATE ntf_subscriptions + SET smp_ntf_id = ?, ntf_sub_id = ?, ntf_sub_status = ?, updated_by_supervisor = ?, updated_at = ? + WHERE conn_id = ? |] - (confirmationId, connId, senderKey, e2ePubKey, ratchetState, connInfo, smpReplyQueues) - - acceptConfirmation :: SQLiteStore -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation - acceptConfirmation st confirmationId ownConnInfo = - liftIOEither . withTransaction st $ \db -> do - DB.executeNamed - db - [sql| - UPDATE conn_confirmations - SET accepted = 1, - own_conn_info = :own_conn_info - WHERE confirmation_id = :confirmation_id; - |] - [ ":own_conn_info" := ownConnInfo, - ":confirmation_id" := confirmationId - ] - firstRow confirmation SEConfirmationNotFound $ - DB.query - db - [sql| - SELECT conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, smp_reply_queues - FROM conn_confirmations - WHERE confirmation_id = ?; - |] - (Only confirmationId) - where - confirmation (connId, senderKey, e2ePubKey, ratchetState, connInfo, smpReplyQueues_) = - AcceptedConfirmation - { confirmationId, - connId, - senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = fromMaybe [] smpReplyQueues_}, - ratchetState, - ownConnInfo - } - - getAcceptedConfirmation :: SQLiteStore -> ConnId -> m AcceptedConfirmation - getAcceptedConfirmation st connId = - liftIOEither . withTransaction st $ \db -> - firstRow confirmation SEConfirmationNotFound $ - DB.query - db - [sql| - SELECT confirmation_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, smp_reply_queues, own_conn_info - FROM conn_confirmations - WHERE conn_id = ? AND accepted = 1; - |] - (Only connId) - where - confirmation (confirmationId, senderKey, e2ePubKey, ratchetState, connInfo, smpReplyQueues_, ownConnInfo) = - AcceptedConfirmation - { confirmationId, - connId, - senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = fromMaybe [] smpReplyQueues_}, - ratchetState, - ownConnInfo - } - - removeConfirmations :: SQLiteStore -> ConnId -> m () - removeConfirmations st connId = - liftIO . withTransaction st $ \db -> - DB.executeNamed - db - [sql| - DELETE FROM conn_confirmations - WHERE conn_id = :conn_id; - |] - [":conn_id" := connId] - - setHandshakeVersion :: SQLiteStore -> ConnId -> Version -> Bool -> m () - setHandshakeVersion st connId aVersion duplexHS = - liftIO . withTransaction st $ \db -> - DB.execute db "UPDATE connections SET smp_agent_version = ?, duplex_handshake = ? WHERE conn_id = ?" (aVersion, duplexHS, connId) - - createInvitation :: SQLiteStore -> TVar ChaChaDRG -> NewInvitation -> m InvitationId - createInvitation st gVar NewInvitation {contactConnId, connReq, recipientConnInfo} = - liftIOEither . withTransaction st $ \db -> - createWithRandomId gVar $ \invitationId -> + (ntfQueueId, ntfSubId, ntfSubStatus, False, updatedAt, connId) + else DB.execute db [sql| - INSERT INTO conn_invitations - (invitation_id, contact_conn_id, cr_invitation, recipient_conn_info, accepted) VALUES (?, ?, ?, ?, 0); + UPDATE ntf_subscriptions + SET smp_ntf_id = ?, ntf_sub_id = ?, ntf_sub_status = ?, ntf_sub_action = ?, ntf_sub_smp_action = ?, ntf_sub_action_ts = ?, updated_by_supervisor = ?, updated_at = ? + WHERE conn_id = ? |] - (invitationId, contactConnId, connReq, recipientConnInfo) + (ntfQueueId, ntfSubId, ntfSubStatus, ntfSubAction, ntfSubSMPAction, ntfSubActionTs, False, updatedAt, connId) + where + (ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction ntfAction - getInvitation :: SQLiteStore -> InvitationId -> m Invitation - getInvitation st invitationId = - liftIOEither . withTransaction st $ \db -> - firstRow invitation SEInvitationNotFound $ - DB.query - db - [sql| - SELECT contact_conn_id, cr_invitation, recipient_conn_info, own_conn_info, accepted - FROM conn_invitations - WHERE invitation_id = ? - AND accepted = 0 - |] - (Only invitationId) - where - invitation (contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted) = - Invitation {invitationId, contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted} - - acceptInvitation :: SQLiteStore -> InvitationId -> ConnInfo -> m () - acceptInvitation st invitationId ownConnInfo = - liftIO . withTransaction st $ \db -> do - DB.executeNamed - db - [sql| - UPDATE conn_invitations - SET accepted = 1, - own_conn_info = :own_conn_info - WHERE invitation_id = :invitation_id - |] - [ ":own_conn_info" := ownConnInfo, - ":invitation_id" := invitationId - ] - - deleteInvitation :: SQLiteStore -> ConnId -> InvitationId -> m () - deleteInvitation st contactConnId invId = - liftIOEither . withTransaction st $ \db -> - runExceptT $ - ExceptT (getConn_ db contactConnId) >>= \case - SomeConn SCContact _ -> - liftIO $ DB.execute db "DELETE FROM conn_invitations WHERE contact_conn_id = ? AND invitation_id = ?" (contactConnId, invId) - _ -> throwError SEConnNotFound - - updateRcvIds :: SQLiteStore -> ConnId -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash) - updateRcvIds st connId = - liftIO . withTransaction st $ \db -> do - (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash) <- retrieveLastIdsAndHashRcv_ db connId - let internalId = InternalId $ unId lastInternalId + 1 - internalRcvId = InternalRcvId $ unRcvId lastInternalRcvId + 1 - updateLastIdsRcv_ db connId internalId internalRcvId - pure (internalId, internalRcvId, lastExternalSndId, lastRcvHash) - - createRcvMsg :: SQLiteStore -> ConnId -> RcvMsgData -> m () - createRcvMsg st connId rcvMsgData = - liftIO . withTransaction st $ \db -> do - insertRcvMsgBase_ db connId rcvMsgData - insertRcvMsgDetails_ db connId rcvMsgData - updateHashRcv_ db connId rcvMsgData - - updateSndIds :: SQLiteStore -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash) - updateSndIds st connId = - liftIO . withTransaction st $ \db -> do - (lastInternalId, lastInternalSndId, prevSndHash) <- retrieveLastIdsAndHashSnd_ db connId - let internalId = InternalId $ unId lastInternalId + 1 - internalSndId = InternalSndId $ unSndId lastInternalSndId + 1 - updateLastIdsSnd_ db connId internalId internalSndId - pure (internalId, internalSndId, prevSndHash) - - createSndMsg :: SQLiteStore -> ConnId -> SndMsgData -> m () - createSndMsg st connId sndMsgData = - liftIO . withTransaction st $ \db -> do - insertSndMsgBase_ db connId sndMsgData - insertSndMsgDetails_ db connId sndMsgData - updateHashSnd_ db connId sndMsgData - - getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, PendingMsgData) - getPendingMsgData st connId msgId = - liftIOEither . withTransaction st $ \db -> runExceptT $ do - rq_ <- liftIO $ getRcvQueueByConnId_ db connId - msgData <- - ExceptT . firstRow pendingMsgData SEMsgNotFound $ - DB.query - db - [sql| - SELECT m.msg_type, m.msg_flags, m.msg_body, m.internal_ts - FROM messages m - JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id - WHERE m.conn_id = ? AND m.internal_id = ? - |] - (connId, msgId) - pure (rq_, msgData) - where - pendingMsgData :: (AgentMessageType, MsgFlags, MsgBody, InternalTs) -> PendingMsgData - pendingMsgData (msgType, msgFlags, msgBody, internalTs) = PendingMsgData {msgId, msgType, msgFlags, msgBody, internalTs} - - getPendingMsgs :: SQLiteStore -> ConnId -> m [InternalId] - getPendingMsgs st connId = - liftIO . withTransaction st $ \db -> - map fromOnly - <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_id = ?" (Only connId) - - setMsgUserAck :: SQLiteStore -> ConnId -> InternalId -> m SMP.MsgId - setMsgUserAck st connId agentMsgId = - liftIOEither . withTransaction st $ \db -> do - DB.execute db "UPDATE rcv_messages SET user_ack = ? WHERE conn_id = ? AND internal_id = ?" (True, connId, agentMsgId) - firstRow fromOnly SEMsgNotFound $ - DB.query db "SELECT broker_id FROM rcv_messages WHERE conn_id = ? AND internal_id = ?" (connId, agentMsgId) - - getLastMsg :: SQLiteStore -> ConnId -> SMP.MsgId -> m (Maybe RcvMsg) - getLastMsg st connId msgId = - liftIO . withTransaction st $ \db -> - fmap rcvMsg . listToMaybe - <$> DB.query - db - [sql| - SELECT - r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, - m.msg_body, r.user_ack - FROM rcv_messages r - JOIN messages m ON r.internal_id = m.internal_id - JOIN connections c ON r.conn_id = c.conn_id AND c.last_internal_msg_id = r.internal_id - WHERE r.conn_id = ? AND r.broker_id = ? - |] - (connId, msgId) - where - rcvMsg (agentMsgId, internalTs, brokerId, brokerTs, sndMsgId, integrity, msgBody, userAck) = - let msgMeta = MsgMeta {recipient = (agentMsgId, internalTs), broker = (brokerId, brokerTs), sndMsgId, integrity} - in RcvMsg {internalId = InternalId agentMsgId, msgMeta, msgBody, userAck} - - deleteMsg :: SQLiteStore -> ConnId -> InternalId -> m () - deleteMsg st connId msgId = - liftIO . withTransaction st $ \db -> - DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId) - - createRatchetX3dhKeys :: SQLiteStore -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> m () - createRatchetX3dhKeys st connId x3dhPrivKey1 x3dhPrivKey2 = - liftIO . withTransaction st $ \db -> - DB.execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2) - - getRatchetX3dhKeys :: SQLiteStore -> ConnId -> m (C.PrivateKeyX448, C.PrivateKeyX448) - getRatchetX3dhKeys st connId = - liftIOEither . withTransaction st $ \db -> - fmap hasKeys $ - firstRow id SEX3dhKeysNotFound $ - DB.query db "SELECT x3dh_priv_key_1, x3dh_priv_key_2 FROM ratchets WHERE conn_id = ?" (Only connId) - where - hasKeys = \case - Right (Just k1, Just k2) -> Right (k1, k2) - _ -> Left SEX3dhKeysNotFound - - createRatchet :: SQLiteStore -> ConnId -> RatchetX448 -> m () - createRatchet st connId rc = - liftIO . withTransaction st $ \db -> do - DB.executeNamed - db - [sql| - INSERT INTO ratchets (conn_id, ratchet_state) - VALUES (:conn_id, :ratchet_state) - ON CONFLICT (conn_id) DO UPDATE SET - ratchet_state = :ratchet_state, - x3dh_priv_key_1 = NULL, - x3dh_priv_key_2 = NULL - |] - [":conn_id" := connId, ":ratchet_state" := rc] - - getRatchet :: SQLiteStore -> ConnId -> m RatchetX448 - getRatchet st connId = - liftIOEither . withTransaction st $ \db -> - ratchet - <$> DB.query db "SELECT ratchet_state FROM ratchets WHERE conn_id = ?" (Only connId) - where - ratchet (Only (Just rc) : _) = Right rc - ratchet _ = Left SERatchetNotFound - - getSkippedMsgKeys :: SQLiteStore -> ConnId -> m SkippedMsgKeys - getSkippedMsgKeys st connId = - liftIO . withTransaction st $ \db -> - skipped <$> DB.query db "SELECT header_key, msg_n, msg_key FROM skipped_messages WHERE conn_id = ?" (Only connId) - where - skipped ms = foldl' addSkippedKey M.empty ms - addSkippedKey smks (hk, msgN, mk) = M.alter (Just . addMsgKey) hk smks - where - addMsgKey = maybe (M.singleton msgN mk) (M.insert msgN mk) - - updateRatchet :: SQLiteStore -> ConnId -> RatchetX448 -> SkippedMsgDiff -> m () - updateRatchet st connId rc skipped = - liftIO . withTransaction st $ \db -> do - DB.execute db "UPDATE ratchets SET ratchet_state = ? WHERE conn_id = ?" (rc, connId) - case skipped of - SMDNoChange -> pure () - SMDRemove hk msgN -> - DB.execute db "DELETE FROM skipped_messages WHERE conn_id = ? AND header_key = ? AND msg_n = ?" (connId, hk, msgN) - SMDAdd smks -> - forM_ (M.assocs smks) $ \(hk, mks) -> - 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) - - createNtfToken :: SQLiteStore -> NtfToken -> m () - createNtfToken st NtfToken {deviceToken = DeviceToken provider token, ntfServer = srv@ProtocolServer {host, port}, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey), ntfDhSecret, ntfTknStatus, ntfTknAction} = - liftIO . withTransaction st $ \db -> do - upsertNtfServer_ db srv - DB.execute - db - [sql| - INSERT INTO ntf_tokens - (provider, device_token, ntf_host, ntf_port, tkn_id, tkn_pub_key, tkn_priv_key, tkn_pub_dh_key, tkn_priv_dh_key, tkn_dh_secret, tkn_status, tkn_action) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - |] - (provider, token, host, port, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret, ntfTknStatus, ntfTknAction) - - getDeviceNtfToken :: SQLiteStore -> DeviceToken -> m (Maybe NtfToken, [NtfToken]) - getDeviceNtfToken st t = - liftIO . withTransaction st $ \db -> do - tokens <- - map ntfToken - <$> DB.query_ - db - [sql| - SELECT s.ntf_host, s.ntf_port, s.ntf_key_hash, - t.provider, t.device_token, t.tkn_id, t.tkn_pub_key, t.tkn_priv_key, t.tkn_pub_dh_key, t.tkn_priv_dh_key, t.tkn_dh_secret, t.tkn_status, t.tkn_action - FROM ntf_tokens t - JOIN ntf_servers s USING (ntf_host, ntf_port) - |] - pure . first listToMaybe $ partition ((t ==) . deviceToken) tokens - where - ntfToken ((host, port, keyHash) :. (provider, dt, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret, ntfTknStatus, ntfTknAction)) = - let ntfServer = ProtocolServer {host, port, keyHash} - ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey) - in NtfToken {deviceToken = DeviceToken provider dt, ntfServer, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys, ntfDhSecret, ntfTknStatus, ntfTknAction} - - updateNtfTokenRegistration :: SQLiteStore -> NtfToken -> NtfTokenId -> C.DhSecretX25519 -> m () - updateNtfTokenRegistration st NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} tknId ntfDhSecret = - liftIO . withTransaction st $ \db -> do +setNullNtfSubscriptionAction :: DB.Connection -> ConnId -> IO () +setNullNtfSubscriptionAction db connId = do + r <- maybeFirstRow fromOnly $ DB.query db "SELECT updated_by_supervisor FROM ntf_subscriptions WHERE conn_id = ?" (Only connId) + forM_ r $ \updatedBySupervisor -> + unless updatedBySupervisor $ do updatedAt <- getCurrentTime DB.execute db [sql| - UPDATE ntf_tokens - SET tkn_id = ?, tkn_dh_secret = ?, tkn_status = ?, tkn_action = ?, updated_at = ? - WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ? + UPDATE ntf_subscriptions + SET ntf_sub_action = ?, ntf_sub_smp_action = ?, ntf_sub_action_ts = ?, updated_by_supervisor = ?, updated_at = ? + WHERE conn_id = ? |] - (tknId, ntfDhSecret, NTRegistered, Nothing :: Maybe NtfTknAction, updatedAt, provider, token, host, port) + (Nothing :: Maybe NtfSubAction, Nothing :: Maybe NtfSubSMPAction, Nothing :: Maybe UTCTime, False, updatedAt, connId) - updateNtfToken :: SQLiteStore -> NtfToken -> NtfTknStatus -> Maybe NtfTknAction -> m () - updateNtfToken st NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} tknStatus tknAction = - liftIO . withTransaction st $ \db -> do - updatedAt <- getCurrentTime - DB.execute +deleteNtfSubscription :: DB.Connection -> ConnId -> IO (Either StoreError ()) +deleteNtfSubscription _db _connId = pure $ Left SENotImplemented + +getNextNtfSubAction :: DB.Connection -> NtfServer -> IO (Maybe (NtfSubscription, NtfSubAction, RcvQueue)) +getNextNtfSubAction db ntfServer@(ProtocolServer ntfHost ntfPort _) = do + maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \(ntfSub@NtfSubscription {connId}, ntfSubAction) -> do + DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId) + (ntfSub,ntfSubAction,) <$$> getRcvQueueByConnId_ db connId + where + getNtfSubAction_ = + DB.query db [sql| - UPDATE ntf_tokens - SET tkn_status = ?, tkn_action = ?, updated_at = ? - WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ? + SELECT ns.conn_id, s.host, s.port, s.key_hash, + ns.smp_ntf_id, ns.ntf_sub_id, ns.ntf_sub_status, ns.ntf_sub_action_ts, ns.ntf_sub_action + FROM ntf_subscriptions ns + JOIN servers s ON s.host = ns.smp_host AND s.port = ns.smp_port + WHERE ns.ntf_host = ? AND ns.ntf_port = ? AND ns.ntf_sub_action IS NOT NULL + ORDER BY ns.ntf_sub_action_ts ASC + LIMIT 1 |] - (tknStatus, tknAction, updatedAt, provider, token, host, port) + (ntfHost, ntfPort) + ntfSubscription (connId, smpHost, smpPort, smpKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs, ntfSubAction) = + let smpServer = SMPServer smpHost smpPort smpKeyHash + in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs}, ntfSubAction) - removeNtfToken :: SQLiteStore -> NtfToken -> m () - removeNtfToken st NtfToken {deviceToken = DeviceToken provider token, ntfServer = ProtocolServer {host, port}} = - liftIO . withTransaction st $ \db -> - DB.execute +getNextNtfSubSMPAction :: DB.Connection -> SMPServer -> IO (Maybe (NtfSubscription, NtfSubSMPAction, RcvQueue)) +getNextNtfSubSMPAction db smpServer@(SMPServer smpHost smpPort _) = do + maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \(ntfSub@NtfSubscription {connId}, ntfSubAction) -> do + DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId) + (ntfSub,ntfSubAction,) <$$> getRcvQueueByConnId_ db connId + where + getNtfSubAction_ = + DB.query db [sql| - DELETE FROM ntf_tokens - WHERE provider = ? AND device_token = ? AND ntf_host = ? AND ntf_port = ? + SELECT ns.conn_id, s.ntf_host, s.ntf_port, s.ntf_key_hash, + ns.smp_ntf_id, ns.ntf_sub_id, ns.ntf_sub_status, ns.ntf_sub_action_ts, ns.ntf_sub_smp_action + FROM ntf_subscriptions ns + JOIN ntf_servers s USING (ntf_host, ntf_port) + WHERE ns.smp_host = ? AND ns.smp_port = ? AND ns.ntf_sub_smp_action IS NOT NULL + ORDER BY ns.ntf_sub_action_ts ASC + LIMIT 1 |] - (provider, token, host, port) + (smpHost, smpPort) + ntfSubscription (connId, ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs, ntfSubAction) = + let ntfServer = ProtocolServer ntfHost ntfPort ntfKeyHash + in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs}, ntfSubAction) - getNtfSubscription :: SQLiteStore -> ConnId -> m (Maybe NtfSubscription) - getNtfSubscription st connId = - liftIO . withTransaction st $ \db -> - maybeFirstRow ntfSubscription $ - DB.query - db - [sql| - SELECT s.host, s.port, s.key_hash, ns.ntf_host, ns.ntf_port, ns.ntf_key_hash, - nsb.smp_ntf_id, nsb.ntf_sub_id, nsb.ntf_sub_status, nsb.ntf_sub_action_ts - FROM ntf_subscriptions nsb - JOIN servers s ON s.host = nsb.smp_host AND s.port = nsb.smp_port - JOIN ntf_servers ns USING (ntf_host, ntf_port) - WHERE nsb.conn_id = ? - |] - (Only connId) - where - ntfSubscription (smpHost, smpPort, smpKeyHash, ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs) = - let smpServer = SMPServer smpHost smpPort smpKeyHash - ntfServer = ProtocolServer ntfHost ntfPort ntfKeyHash - in NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs} +getActiveNtfToken :: DB.Connection -> IO (Maybe NtfToken) +getActiveNtfToken db = + maybeFirstRow ntfToken $ + DB.query + db + [sql| + SELECT s.ntf_host, s.ntf_port, s.ntf_key_hash, + t.provider, t.device_token, t.tkn_id, t.tkn_pub_key, t.tkn_priv_key, t.tkn_pub_dh_key, t.tkn_priv_dh_key, t.tkn_dh_secret, t.tkn_status, t.tkn_action + FROM ntf_tokens t + JOIN ntf_servers s USING (ntf_host, ntf_port) + WHERE t.tkn_status = ? + |] + (Only NTActive) + where + ntfToken ((host, port, keyHash) :. (provider, dt, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret, ntfTknStatus, ntfTknAction)) = + let ntfServer = ProtocolServer {host, port, keyHash} + ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey) + in NtfToken {deviceToken = DeviceToken provider dt, ntfServer, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys, ntfDhSecret, ntfTknStatus, ntfTknAction} - createNtfSubscription :: SQLiteStore -> NtfSubscription -> NtfSubOrSMPAction -> m () - createNtfSubscription st NtfSubscription {connId, smpServer = (SMPServer host port _), ntfQueueId, ntfServer = (SMPServer ntfHost ntfPort _), ntfSubId, ntfSubStatus, ntfSubActionTs} ntfAction = - liftIO . withTransaction st $ \db -> - DB.execute - db - [sql| - INSERT INTO ntf_subscriptions - (conn_id, smp_host, smp_port, smp_ntf_id, ntf_host, ntf_port, ntf_sub_id, - ntf_sub_status, ntf_sub_action, ntf_sub_smp_action, ntf_sub_action_ts) - VALUES (?,?,?,?,?,?,?,?,?,?,?) - |] - ( (connId, host, port, ntfQueueId, ntfHost, ntfPort, ntfSubId) - :. (ntfSubStatus, ntfSubAction, ntfSubSMPAction, ntfSubActionTs) - ) - where - (ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction ntfAction - - markNtfSubscriptionForDeletion :: SQLiteStore -> ConnId -> m () - markNtfSubscriptionForDeletion _st _rcvQueue = throwError SENotImplemented - - updateNtfSubscription :: SQLiteStore -> ConnId -> NtfSubscription -> NtfSubOrSMPAction -> m () - updateNtfSubscription st connId NtfSubscription {ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs} ntfAction = - liftIO . withTransaction st $ \db -> do - r <- maybeFirstRow fromOnly $ DB.query db "SELECT updated_by_supervisor FROM ntf_subscriptions WHERE conn_id = ?" (Only connId) - forM_ r $ \updatedBySupervisor -> do - updatedAt <- getCurrentTime - if updatedBySupervisor - then - DB.execute - db - [sql| - UPDATE ntf_subscriptions - SET smp_ntf_id = ?, ntf_sub_id = ?, ntf_sub_status = ?, updated_by_supervisor = ?, updated_at = ? - WHERE conn_id = ? - |] - (ntfQueueId, ntfSubId, ntfSubStatus, False, updatedAt, connId) - else - DB.execute - db - [sql| - UPDATE ntf_subscriptions - SET smp_ntf_id = ?, ntf_sub_id = ?, ntf_sub_status = ?, ntf_sub_action = ?, ntf_sub_smp_action = ?, ntf_sub_action_ts = ?, updated_by_supervisor = ?, updated_at = ? - WHERE conn_id = ? - |] - (ntfQueueId, ntfSubId, ntfSubStatus, ntfSubAction, ntfSubSMPAction, ntfSubActionTs, False, updatedAt, connId) - where - (ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction ntfAction - - setNullNtfSubscriptionAction :: SQLiteStore -> ConnId -> m () - setNullNtfSubscriptionAction st connId = - liftIO . withTransaction st $ \db -> do - r <- maybeFirstRow fromOnly $ DB.query db "SELECT updated_by_supervisor FROM ntf_subscriptions WHERE conn_id = ?" (Only connId) - forM_ r $ \updatedBySupervisor -> - unless updatedBySupervisor $ do - updatedAt <- getCurrentTime - DB.execute - db - [sql| - UPDATE ntf_subscriptions - SET ntf_sub_action = ?, ntf_sub_smp_action = ?, ntf_sub_action_ts = ?, updated_by_supervisor = ?, updated_at = ? - WHERE conn_id = ? - |] - (Nothing :: Maybe NtfSubAction, Nothing :: Maybe NtfSubSMPAction, Nothing :: Maybe UTCTime, False, updatedAt, connId) - - deleteNtfSubscription :: SQLiteStore -> ConnId -> m () - deleteNtfSubscription _st _connId = throwError SENotImplemented - - getNextNtfSubAction :: SQLiteStore -> NtfServer -> m (Maybe (NtfSubscription, NtfSubAction, RcvQueue)) - getNextNtfSubAction st ntfServer@(ProtocolServer ntfHost ntfPort _) = - liftIO . withTransaction st $ \db -> do - r <- - maybeFirstRow ntfSubscription $ - DB.query - db - [sql| - SELECT ns.conn_id, s.host, s.port, s.key_hash, - ns.smp_ntf_id, ns.ntf_sub_id, ns.ntf_sub_status, ns.ntf_sub_action_ts, ns.ntf_sub_action - FROM ntf_subscriptions ns - JOIN servers s ON s.host = ns.smp_host AND s.port = ns.smp_port - WHERE ns.ntf_host = ? AND ns.ntf_port = ? AND ns.ntf_sub_action IS NOT NULL - ORDER BY ns.ntf_sub_action_ts ASC - LIMIT 1 - |] - (ntfHost, ntfPort) - case r of - Just (ntfSub@NtfSubscription {connId}, ntfSubAction) -> do - DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId) - rq_ <- getRcvQueueByConnId_ db connId - pure $ (\rq -> Just (ntfSub, ntfSubAction, rq)) =<< rq_ - Nothing -> pure Nothing - where - ntfSubscription (connId, smpHost, smpPort, smpKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs, ntfSubAction) = - let smpServer = SMPServer smpHost smpPort smpKeyHash - in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs}, ntfSubAction) - - getNextNtfSubSMPAction :: SQLiteStore -> SMPServer -> m (Maybe (NtfSubscription, NtfSubSMPAction, RcvQueue)) - getNextNtfSubSMPAction st smpServer@(SMPServer smpHost smpPort _) = - liftIO . withTransaction st $ \db -> do - r <- - maybeFirstRow ntfSubscription $ - DB.query - db - [sql| - SELECT ns.conn_id, s.ntf_host, s.ntf_port, s.ntf_key_hash, - ns.smp_ntf_id, ns.ntf_sub_id, ns.ntf_sub_status, ns.ntf_sub_action_ts, ns.ntf_sub_smp_action - FROM ntf_subscriptions ns - JOIN ntf_servers s USING (ntf_host, ntf_port) - WHERE ns.smp_host = ? AND ns.smp_port = ? AND ns.ntf_sub_smp_action IS NOT NULL - ORDER BY ns.ntf_sub_action_ts ASC - LIMIT 1 - |] - (smpHost, smpPort) - case r of - Just (ntfSub@NtfSubscription {connId}, ntfSubAction) -> do - DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId) - rq_ <- getRcvQueueByConnId_ db connId - pure $ (\rq -> Just (ntfSub, ntfSubAction, rq)) =<< rq_ - Nothing -> pure Nothing - where - ntfSubscription (connId, ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs, ntfSubAction) = - let ntfServer = ProtocolServer ntfHost ntfPort ntfKeyHash - in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs}, ntfSubAction) - - getActiveNtfToken :: SQLiteStore -> m (Maybe NtfToken) - getActiveNtfToken st = - liftIO . withTransaction st $ \db -> - maybeFirstRow ntfToken $ - DB.query - db - [sql| - SELECT s.ntf_host, s.ntf_port, s.ntf_key_hash, - t.provider, t.device_token, t.tkn_id, t.tkn_pub_key, t.tkn_priv_key, t.tkn_pub_dh_key, t.tkn_priv_dh_key, t.tkn_dh_secret, t.tkn_status, t.tkn_action - FROM ntf_tokens t - JOIN ntf_servers s USING (ntf_host, ntf_port) - WHERE t.tkn_status = ? - |] - (Only NTActive) - where - ntfToken ((host, port, keyHash) :. (provider, dt, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhPubKey, ntfDhPrivKey, ntfDhSecret, ntfTknStatus, ntfTknAction)) = - let ntfServer = ProtocolServer {host, port, keyHash} - ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey) - in NtfToken {deviceToken = DeviceToken provider dt, ntfServer, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys, ntfDhSecret, ntfTknStatus, ntfTknAction} - - getNtfRcvQueue :: SQLiteStore -> SMPQueueNtf -> m (ConnId, RcvDhSecret) - getNtfRcvQueue st SMPQueueNtf {smpServer = (SMPServer host port _), notifierId} = - liftIOEither . withTransaction st $ \db -> do - firstRow id SEConnNotFound $ - DB.query - db - [sql| - SELECT conn_id, rcv_dh_secret - FROM rcv_queues - WHERE host = ? AND port = ? AND ntf_id = ? - |] - (host, port, notifierId) +getNtfRcvQueue :: DB.Connection -> SMPQueueNtf -> IO (Either StoreError (ConnId, RcvDhSecret)) +getNtfRcvQueue db SMPQueueNtf {smpServer = (SMPServer host port _), notifierId} = + firstRow id SEConnNotFound $ + DB.query + db + [sql| + SELECT conn_id, rcv_dh_secret + FROM rcv_queues + WHERE host = ? AND port = ? AND ntf_id = ? + |] + (host, port, notifierId) -- * Auxiliary helpers @@ -929,9 +924,8 @@ firstRow f e a = second f . listToEither e <$> a maybeFirstRow :: Functor f => (a -> b) -> f [a] -> f (Maybe b) maybeFirstRow f q = fmap f . listToMaybe <$> q --- TODO move from simplex-chat --- firstRow' :: (a -> Either e b) -> e -> IO [a] -> IO (Either e b) --- firstRow' f e a = (f <=< listToEither e) <$> a +firstRow' :: (a -> Either e b) -> e -> IO [a] -> IO (Either e b) +firstRow' f e a = (f <=< listToEither e) <$> a {- ORMOLU_DISABLE -} -- SQLite.Simple only has these up to 10 fields, which is insufficient for some of our queries @@ -1015,8 +1009,8 @@ insertSndQueue_ dbConn connId SndQueue {..} = do -- * getConn helpers -getConn_ :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn) -getConn_ dbConn connId = +getConn :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn) +getConn dbConn connId = getConnData_ dbConn connId >>= \case Nothing -> pure $ Left SEConnNotFound Just (connData, cMode) -> do diff --git a/src/Simplex/Messaging/Util.hs b/src/Simplex/Messaging/Util.hs index 8940f239b..5c0876196 100644 --- a/src/Simplex/Messaging/Util.hs +++ b/src/Simplex/Messaging/Util.hs @@ -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 #-} diff --git a/tests/AgentTests/SQLiteTests.hs b/tests/AgentTests/SQLiteTests.hs index 47a818503..2afc1d316 100644 --- a/tests/AgentTests/SQLiteTests.hs +++ b/tests/AgentTests/SQLiteTests.hs @@ -11,7 +11,6 @@ module AgentTests.SQLiteTests (storeTests) where import Control.Concurrent.Async (concurrently_) import Control.Concurrent.STM import Control.Monad (replicateM_) -import Control.Monad.Except (ExceptT, runExceptT) import Crypto.Random (drgNew) import Data.ByteString.Char8 (ByteString) import qualified Data.Text as T @@ -55,29 +54,23 @@ createStore = do createSQLiteStore (testDB <> show r) Migrations.app True removeStore :: SQLiteStore -> IO () -removeStore store = do - close store - removeFile $ dbFilePath store +removeStore db = do + close db + removeFile $ dbFilePath db where close :: SQLiteStore -> IO () close st = mapM_ DB.close =<< atomically (tryTakeTMVar $ dbConnection st) -returnsResult :: (Eq a, Eq e, Show a, Show e) => ExceptT e IO a -> a -> Expectation -action `returnsResult` r = runExceptT action `shouldReturn` Right r - -throwsError :: (Eq a, Eq e, Show a, Show e) => ExceptT e IO a -> e -> Expectation -action `throwsError` e = runExceptT action `shouldReturn` Left e - -- TODO add null port tests storeTests :: Spec storeTests = do withStore2 $ do describe "stress test" testConcurrentWrites withStore $ do - describe "store setup" $ do + describe "db setup" $ do testCompiledThreadsafe testForeignKeysEnabled - describe "store methods" $ do + describe "db methods" $ do describe "Queue and Connection management" $ do describe "createRcvConn" $ do testCreateRcvConn @@ -112,28 +105,29 @@ testConcurrentWrites :: SpecWith (SQLiteStore, SQLiteStore) testConcurrentWrites = it "should complete multiple concurrent write transactions w/t sqlite busy errors" $ \(s1, s2) -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createRcvConn s1 g cData1 rcvQueue1 SCMInvitation + _ <- withTransaction s1 $ \db -> + createRcvConn db g cData1 rcvQueue1 SCMInvitation let ConnData {connId} = cData1 concurrently_ (runTest s1 connId) (runTest s2 connId) where - runTest :: SQLiteStore -> ConnId -> IO (Either StoreError ()) - runTest store connId = runExceptT . replicateM_ 100 $ do - (internalId, internalRcvId, _, _) <- updateRcvIds store connId + runTest :: SQLiteStore -> ConnId -> IO () + runTest st connId = replicateM_ 100 . withTransaction st $ \db -> do + (internalId, internalRcvId, _, _) <- updateRcvIds db connId let rcvMsgData = mkRcvMsgData internalId internalRcvId 0 "0" "hash_dummy" - createRcvMsg store connId rcvMsgData + createRcvMsg db connId rcvMsgData testCompiledThreadsafe :: SpecWith SQLiteStore testCompiledThreadsafe = - it "compiled sqlite library should be threadsafe" . withStoreConnection $ \db -> do + it "compiled sqlite library should be threadsafe" . withStoreTransaction $ \db -> do compileOptions <- DB.query_ db "pragma COMPILE_OPTIONS;" :: IO [[T.Text]] compileOptions `shouldNotContain` [["THREADSAFE=0"]] -withStoreConnection :: (DB.Connection -> IO a) -> SQLiteStore -> IO a -withStoreConnection = flip withConnection +withStoreTransaction :: (DB.Connection -> IO a) -> SQLiteStore -> IO a +withStoreTransaction = flip withTransaction testForeignKeysEnabled :: SpecWith SQLiteStore testForeignKeysEnabled = - it "foreign keys should be enabled" . withStoreConnection $ \db -> do + it "foreign keys should be enabled" . withStoreTransaction $ \db -> do let inconsistentQuery = [sql| INSERT INTO snd_queues @@ -186,125 +180,125 @@ sndQueue1 = testCreateRcvConn :: SpecWith SQLiteStore testCreateRcvConn = - it "should create RcvConnection and add SndQueue" $ \store -> do + it "should create RcvConnection and add SndQueue" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - createRcvConn store g cData1 rcvQueue1 SCMInvitation - `returnsResult` "conn1" - getConn store "conn1" - `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1) - upgradeRcvConnToDuplex store "conn1" sndQueue1 - `returnsResult` () - getConn store "conn1" - `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1) + createRcvConn db g cData1 rcvQueue1 SCMInvitation + `shouldReturn` Right "conn1" + getConn db "conn1" + `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1)) + upgradeRcvConnToDuplex db "conn1" sndQueue1 + `shouldReturn` Right () + getConn db "conn1" + `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1)) testCreateRcvConnRandomId :: SpecWith SQLiteStore testCreateRcvConnRandomId = - it "should create RcvConnection and add SndQueue with random ID" $ \store -> do + it "should create RcvConnection and add SndQueue with random ID" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - Right connId <- runExceptT $ createRcvConn store g cData1 {connId = ""} rcvQueue1 SCMInvitation - getConn store connId - `returnsResult` SomeConn SCRcv (RcvConnection cData1 {connId} rcvQueue1) - upgradeRcvConnToDuplex store connId sndQueue1 - `returnsResult` () - getConn store connId - `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1) + Right connId <- createRcvConn db g cData1 {connId = ""} rcvQueue1 SCMInvitation + getConn db connId + `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 {connId} rcvQueue1)) + upgradeRcvConnToDuplex db connId sndQueue1 + `shouldReturn` Right () + getConn db connId + `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1)) testCreateRcvConnDuplicate :: SpecWith SQLiteStore testCreateRcvConnDuplicate = - it "should throw error on attempt to create duplicate RcvConnection" $ \store -> do + it "should throw error on attempt to create duplicate RcvConnection" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1 SCMInvitation - createRcvConn store g cData1 rcvQueue1 SCMInvitation - `throwsError` SEConnDuplicate + _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation + createRcvConn db g cData1 rcvQueue1 SCMInvitation + `shouldReturn` Left (SEConnDuplicate) testCreateSndConn :: SpecWith SQLiteStore testCreateSndConn = - it "should create SndConnection and add RcvQueue" $ \store -> do + it "should create SndConnection and add RcvQueue" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - createSndConn store g cData1 sndQueue1 - `returnsResult` "conn1" - getConn store "conn1" - `returnsResult` SomeConn SCSnd (SndConnection cData1 sndQueue1) - upgradeSndConnToDuplex store "conn1" rcvQueue1 - `returnsResult` () - getConn store "conn1" - `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1) + createSndConn db g cData1 sndQueue1 + `shouldReturn` Right "conn1" + getConn db "conn1" + `shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sndQueue1)) + upgradeSndConnToDuplex db "conn1" rcvQueue1 + `shouldReturn` Right () + getConn db "conn1" + `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1)) testCreateSndConnRandomID :: SpecWith SQLiteStore testCreateSndConnRandomID = - it "should create SndConnection and add RcvQueue with random ID" $ \store -> do + it "should create SndConnection and add RcvQueue with random ID" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - Right connId <- runExceptT $ createSndConn store g cData1 {connId = ""} sndQueue1 - getConn store connId - `returnsResult` SomeConn SCSnd (SndConnection cData1 {connId} sndQueue1) - upgradeSndConnToDuplex store connId rcvQueue1 - `returnsResult` () - getConn store connId - `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1) + Right connId <- createSndConn db g cData1 {connId = ""} sndQueue1 + getConn db connId + `shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 {connId} sndQueue1)) + upgradeSndConnToDuplex db connId rcvQueue1 + `shouldReturn` Right () + getConn db connId + `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} rcvQueue1 sndQueue1)) testCreateSndConnDuplicate :: SpecWith SQLiteStore testCreateSndConnDuplicate = - it "should throw error on attempt to create duplicate SndConnection" $ \store -> do + it "should throw error on attempt to create duplicate SndConnection" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createSndConn store g cData1 sndQueue1 - createSndConn store g cData1 sndQueue1 - `throwsError` SEConnDuplicate + _ <- createSndConn db g cData1 sndQueue1 + createSndConn db g cData1 sndQueue1 + `shouldReturn` Left (SEConnDuplicate) testGetRcvConn :: SpecWith SQLiteStore testGetRcvConn = - it "should get connection using rcv queue id and server" $ \store -> do + it "should get connection using rcv queue id and server" . withStoreTransaction $ \db -> do let smpServer = SMPServer "smp.simplex.im" "5223" testKeyHash let recipientId = "1234" g <- newTVarIO =<< drgNew - _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1 SCMInvitation - getRcvConn store smpServer recipientId - `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1) + _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation + getRcvConn db smpServer recipientId + `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1)) testDeleteRcvConn :: SpecWith SQLiteStore testDeleteRcvConn = - it "should create RcvConnection and delete it" $ \store -> do + it "should create RcvConnection and delete it" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1 SCMInvitation - getConn store "conn1" - `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1) - deleteConn store "conn1" - `returnsResult` () + _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation + getConn db "conn1" + `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1)) + deleteConn db "conn1" + `shouldReturn` () -- TODO check queues are deleted as well - getConn store "conn1" - `throwsError` SEConnNotFound + getConn db "conn1" + `shouldReturn` Left (SEConnNotFound) testDeleteSndConn :: SpecWith SQLiteStore testDeleteSndConn = - it "should create SndConnection and delete it" $ \store -> do + it "should create SndConnection and delete it" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createSndConn store g cData1 sndQueue1 - getConn store "conn1" - `returnsResult` SomeConn SCSnd (SndConnection cData1 sndQueue1) - deleteConn store "conn1" - `returnsResult` () + _ <- createSndConn db g cData1 sndQueue1 + getConn db "conn1" + `shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sndQueue1)) + deleteConn db "conn1" + `shouldReturn` () -- TODO check queues are deleted as well - getConn store "conn1" - `throwsError` SEConnNotFound + getConn db "conn1" + `shouldReturn` Left (SEConnNotFound) testDeleteDuplexConn :: SpecWith SQLiteStore testDeleteDuplexConn = - it "should create DuplexConnection and delete it" $ \store -> do + it "should create DuplexConnection and delete it" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1 SCMInvitation - _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1 - getConn store "conn1" - `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1) - deleteConn store "conn1" - `returnsResult` () + _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation + _ <- upgradeRcvConnToDuplex db "conn1" sndQueue1 + getConn db "conn1" + `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1)) + deleteConn db "conn1" + `shouldReturn` () -- TODO check queues are deleted as well - getConn store "conn1" - `throwsError` SEConnNotFound + getConn db "conn1" + `shouldReturn` Left (SEConnNotFound) testUpgradeRcvConnToDuplex :: SpecWith SQLiteStore testUpgradeRcvConnToDuplex = - it "should throw error on attempt to add SndQueue to SndConnection or DuplexConnection" $ \store -> do + it "should throw error on attempt to add SndQueue to SndConnection or DuplexConnection" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createSndConn store g cData1 sndQueue1 + _ <- createSndConn db g cData1 sndQueue1 let anotherSndQueue = SndQueue { server = SMPServer "smp.simplex.im" "5223" testKeyHash, @@ -315,17 +309,17 @@ testUpgradeRcvConnToDuplex = e2eDhSecret = testDhSecret, status = New } - upgradeRcvConnToDuplex store "conn1" anotherSndQueue - `throwsError` SEBadConnType CSnd - _ <- runExceptT $ upgradeSndConnToDuplex store "conn1" rcvQueue1 - upgradeRcvConnToDuplex store "conn1" anotherSndQueue - `throwsError` SEBadConnType CDuplex + upgradeRcvConnToDuplex db "conn1" anotherSndQueue + `shouldReturn` Left (SEBadConnType CSnd) + _ <- upgradeSndConnToDuplex db "conn1" rcvQueue1 + upgradeRcvConnToDuplex db "conn1" anotherSndQueue + `shouldReturn` Left (SEBadConnType CDuplex) testUpgradeSndConnToDuplex :: SpecWith SQLiteStore testUpgradeSndConnToDuplex = - it "should throw error on attempt to add RcvQueue to RcvConnection or DuplexConnection" $ \store -> do + it "should throw error on attempt to add RcvQueue to RcvConnection or DuplexConnection" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1 SCMInvitation + _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation let anotherRcvQueue = RcvQueue { server = SMPServer "smp.simplex.im" "5223" testKeyHash, @@ -340,52 +334,52 @@ testUpgradeSndConnToDuplex = ntfPrivateKey = Nothing, notifierId = Nothing } - upgradeSndConnToDuplex store "conn1" anotherRcvQueue - `throwsError` SEBadConnType CRcv - _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1 - upgradeSndConnToDuplex store "conn1" anotherRcvQueue - `throwsError` SEBadConnType CDuplex + upgradeSndConnToDuplex db "conn1" anotherRcvQueue + `shouldReturn` Left (SEBadConnType CRcv) + _ <- upgradeRcvConnToDuplex db "conn1" sndQueue1 + upgradeSndConnToDuplex db "conn1" anotherRcvQueue + `shouldReturn` Left (SEBadConnType CDuplex) testSetRcvQueueStatus :: SpecWith SQLiteStore testSetRcvQueueStatus = - it "should update status of RcvQueue" $ \store -> do + it "should update status of RcvQueue" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1 SCMInvitation - getConn store "conn1" - `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1) - setRcvQueueStatus store rcvQueue1 Confirmed - `returnsResult` () - getConn store "conn1" - `returnsResult` SomeConn SCRcv (RcvConnection cData1 rcvQueue1 {status = Confirmed}) + _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation + getConn db "conn1" + `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1)) + setRcvQueueStatus db rcvQueue1 Confirmed + `shouldReturn` () + getConn db "conn1" + `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rcvQueue1 {status = Confirmed})) testSetSndQueueStatus :: SpecWith SQLiteStore testSetSndQueueStatus = - it "should update status of SndQueue" $ \store -> do + it "should update status of SndQueue" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createSndConn store g cData1 sndQueue1 - getConn store "conn1" - `returnsResult` SomeConn SCSnd (SndConnection cData1 sndQueue1) - setSndQueueStatus store sndQueue1 Confirmed - `returnsResult` () - getConn store "conn1" - `returnsResult` SomeConn SCSnd (SndConnection cData1 sndQueue1 {status = Confirmed}) + _ <- createSndConn db g cData1 sndQueue1 + getConn db "conn1" + `shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sndQueue1)) + setSndQueueStatus db sndQueue1 Confirmed + `shouldReturn` () + getConn db "conn1" + `shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sndQueue1 {status = Confirmed})) testSetQueueStatusDuplex :: SpecWith SQLiteStore testSetQueueStatusDuplex = - it "should update statuses of RcvQueue and SndQueue in DuplexConnection" $ \store -> do + it "should update statuses of RcvQueue and SndQueue in DuplexConnection" . withStoreTransaction $ \db -> do g <- newTVarIO =<< drgNew - _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1 SCMInvitation - _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1 - getConn store "conn1" - `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1) - setRcvQueueStatus store rcvQueue1 Secured - `returnsResult` () - getConn store "conn1" - `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1) - setSndQueueStatus store sndQueue1 Confirmed - `returnsResult` () - getConn store "conn1" - `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1 {status = Confirmed}) + _ <- createRcvConn db g cData1 rcvQueue1 SCMInvitation + _ <- upgradeRcvConnToDuplex db "conn1" sndQueue1 + getConn db "conn1" + `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1)) + setRcvQueueStatus db rcvQueue1 Secured + `shouldReturn` () + getConn db "conn1" + `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1)) + setSndQueueStatus db sndQueue1 Confirmed + `shouldReturn` () + getConn db "conn1" + `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 {status = Secured} sndQueue1 {status = Confirmed})) hw :: ByteString hw = encodeUtf8 "Hello world!" @@ -411,22 +405,24 @@ mkRcvMsgData internalId internalRcvId externalSndId brokerId internalHash = externalPrevSndHash = "hash_from_sender" } -testCreateRcvMsg_ :: SQLiteStore -> PrevExternalSndId -> PrevRcvMsgHash -> ConnId -> RcvMsgData -> Expectation -testCreateRcvMsg_ st expectedPrevSndId expectedPrevHash connId rcvMsgData@RcvMsgData {..} = do +testCreateRcvMsg_ :: DB.Connection -> PrevExternalSndId -> PrevRcvMsgHash -> ConnId -> RcvMsgData -> Expectation +testCreateRcvMsg_ db expectedPrevSndId expectedPrevHash connId rcvMsgData@RcvMsgData {..} = do let MsgMeta {recipient = (internalId, _)} = msgMeta - updateRcvIds st connId - `returnsResult` (InternalId internalId, internalRcvId, expectedPrevSndId, expectedPrevHash) - createRcvMsg st connId rcvMsgData - `returnsResult` () + updateRcvIds db connId + `shouldReturn` (InternalId internalId, internalRcvId, expectedPrevSndId, expectedPrevHash) + createRcvMsg db connId rcvMsgData + `shouldReturn` () testCreateRcvMsg :: SpecWith SQLiteStore testCreateRcvMsg = it "should reserve internal ids and create a RcvMsg" $ \st -> do g <- newTVarIO =<< drgNew let ConnData {connId} = cData1 - _ <- runExceptT $ createRcvConn st g cData1 rcvQueue1 SCMInvitation - testCreateRcvMsg_ st 0 "" connId $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "hash_dummy" - testCreateRcvMsg_ st 1 "hash_dummy" connId $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "new_hash_dummy" + _ <- withTransaction st $ \db -> do + createRcvConn db g cData1 rcvQueue1 SCMInvitation + withTransaction st $ \db -> do + testCreateRcvMsg_ db 0 "" connId $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "hash_dummy" + testCreateRcvMsg_ db 1 "hash_dummy" connId $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "new_hash_dummy" mkSndMsgData :: InternalId -> InternalSndId -> MsgHash -> SndMsgData mkSndMsgData internalId internalSndId internalHash = @@ -441,32 +437,36 @@ mkSndMsgData internalId internalSndId internalHash = prevMsgHash = internalHash } -testCreateSndMsg_ :: SQLiteStore -> PrevSndMsgHash -> ConnId -> SndMsgData -> Expectation -testCreateSndMsg_ store expectedPrevHash connId sndMsgData@SndMsgData {..} = do - updateSndIds store connId - `returnsResult` (internalId, internalSndId, expectedPrevHash) - createSndMsg store connId sndMsgData - `returnsResult` () +testCreateSndMsg_ :: DB.Connection -> PrevSndMsgHash -> ConnId -> SndMsgData -> Expectation +testCreateSndMsg_ db expectedPrevHash connId sndMsgData@SndMsgData {..} = do + updateSndIds db connId + `shouldReturn` (internalId, internalSndId, expectedPrevHash) + createSndMsg db connId sndMsgData + `shouldReturn` () testCreateSndMsg :: SpecWith SQLiteStore testCreateSndMsg = - it "should create a SndMsg and return InternalId and PrevSndMsgHash" $ \store -> do + it "should create a SndMsg and return InternalId and PrevSndMsgHash" $ \st -> do g <- newTVarIO =<< drgNew let ConnData {connId} = cData1 - _ <- runExceptT $ createSndConn store g cData1 sndQueue1 - testCreateSndMsg_ store "" connId $ mkSndMsgData (InternalId 1) (InternalSndId 1) "hash_dummy" - testCreateSndMsg_ store "hash_dummy" connId $ mkSndMsgData (InternalId 2) (InternalSndId 2) "new_hash_dummy" + _ <- withTransaction st $ \db -> do + createSndConn db g cData1 sndQueue1 + withTransaction st $ \db -> do + testCreateSndMsg_ db "" connId $ mkSndMsgData (InternalId 1) (InternalSndId 1) "hash_dummy" + testCreateSndMsg_ db "hash_dummy" connId $ mkSndMsgData (InternalId 2) (InternalSndId 2) "new_hash_dummy" testCreateRcvAndSndMsgs :: SpecWith SQLiteStore testCreateRcvAndSndMsgs = - it "should create multiple RcvMsg and SndMsg, correctly ordering internal Ids and returning previous state" $ \store -> do - g <- newTVarIO =<< drgNew + it "should create multiple RcvMsg and SndMsg, correctly ordering internal Ids and returning previous state" $ \st -> do let ConnData {connId} = cData1 - _ <- runExceptT $ createRcvConn store g cData1 rcvQueue1 SCMInvitation - _ <- runExceptT $ upgradeRcvConnToDuplex store "conn1" sndQueue1 - testCreateRcvMsg_ store 0 "" connId $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "rcv_hash_1" - testCreateRcvMsg_ store 1 "rcv_hash_1" connId $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "rcv_hash_2" - testCreateSndMsg_ store "" connId $ mkSndMsgData (InternalId 3) (InternalSndId 1) "snd_hash_1" - testCreateRcvMsg_ store 2 "rcv_hash_2" connId $ mkRcvMsgData (InternalId 4) (InternalRcvId 3) 3 "3" "rcv_hash_3" - testCreateSndMsg_ store "snd_hash_1" connId $ mkSndMsgData (InternalId 5) (InternalSndId 2) "snd_hash_2" - testCreateSndMsg_ store "snd_hash_2" connId $ mkSndMsgData (InternalId 6) (InternalSndId 3) "snd_hash_3" + _ <- withTransaction st $ \db -> do + g <- newTVarIO =<< drgNew + createRcvConn db g cData1 rcvQueue1 SCMInvitation + withTransaction st $ \db -> do + _ <- upgradeRcvConnToDuplex db "conn1" sndQueue1 + testCreateRcvMsg_ db 0 "" connId $ mkRcvMsgData (InternalId 1) (InternalRcvId 1) 1 "1" "rcv_hash_1" + testCreateRcvMsg_ db 1 "rcv_hash_1" connId $ mkRcvMsgData (InternalId 2) (InternalRcvId 2) 2 "2" "rcv_hash_2" + testCreateSndMsg_ db "" connId $ mkSndMsgData (InternalId 3) (InternalSndId 1) "snd_hash_1" + testCreateRcvMsg_ db 2 "rcv_hash_2" connId $ mkRcvMsgData (InternalId 4) (InternalRcvId 3) 3 "3" "rcv_hash_3" + testCreateSndMsg_ db "snd_hash_1" connId $ mkSndMsgData (InternalId 5) (InternalSndId 2) "snd_hash_2" + testCreateSndMsg_ db "snd_hash_2" connId $ mkSndMsgData (InternalId 6) (InternalSndId 3) "snd_hash_3"