From 051bf38bc75414fa58fa5004619594b0f674b591 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Mon, 22 Jul 2024 15:42:34 +0100 Subject: [PATCH] agent: remove delays in notification processing, add notification stats (#1235) * agent: remove delays in notification processing, add notification stats * do not replace token after failed check * more stats * refactor * fix * backwards compatible JSON parsing * retry deleting subscription on temporary error * remove attempts to get multiple notification messages * fix JSON decoding to be backwards compatible --- src/Simplex/Messaging/Agent.hs | 37 ++-- src/Simplex/Messaging/Agent/Client.hs | 40 ++-- src/Simplex/Messaging/Agent/Env/SQLite.hs | 6 - .../Messaging/Agent/NtfSubSupervisor.hs | 108 +++++------ src/Simplex/Messaging/Agent/Stats.hs | 176 +++++++++++++++++- src/Simplex/Messaging/Agent/Store/SQLite.hs | 29 +-- src/Simplex/Messaging/Notifications/Types.hs | 18 +- tests/AgentTests/NotificationTests.hs | 2 +- tests/SMPAgentClient.hs | 2 - 9 files changed, 285 insertions(+), 133 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index a0079c883..f26c46925 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -237,25 +237,25 @@ logServersStats c = do liftIO $ threadDelay' int saveServersStats :: AgentClient -> AM' () -saveServersStats c@AgentClient {subQ, smpServersStats, xftpServersStats} = do +saveServersStats c@AgentClient {subQ, smpServersStats, xftpServersStats, ntfServersStats} = do sss <- mapM (lift . getAgentSMPServerStats) =<< readTVarIO smpServersStats xss <- mapM (lift . getAgentXFTPServerStats) =<< readTVarIO xftpServersStats - let stats = AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss} + nss <- mapM (lift . getAgentNtfServerStats) =<< readTVarIO ntfServersStats + let stats = AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss, ntfServersStats = OptionalMap nss} tryAgentError' (withStore' c (`updateServersStats` stats)) >>= \case Left e -> atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e) Right () -> pure () restoreServersStats :: AgentClient -> AM' () -restoreServersStats c@AgentClient {smpServersStats, xftpServersStats, srvStatsStartedAt} = do +restoreServersStats c@AgentClient {smpServersStats, xftpServersStats, ntfServersStats, srvStatsStartedAt} = do tryAgentError' (withStore c getServersStats) >>= \case Left e -> atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e) Right (startedAt, Nothing) -> atomically $ writeTVar srvStatsStartedAt startedAt - Right (startedAt, Just AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss}) -> do + Right (startedAt, Just AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss, ntfServersStats = OptionalMap nss}) -> do atomically $ writeTVar srvStatsStartedAt startedAt - sss' <- mapM (atomically . newAgentSMPServerStats') sss - atomically $ writeTVar smpServersStats sss' - xss' <- mapM (atomically . newAgentXFTPServerStats') xss - atomically $ writeTVar xftpServersStats xss' + atomically . writeTVar smpServersStats =<< mapM (atomically . newAgentSMPServerStats') sss + atomically . writeTVar xftpServersStats =<< mapM (atomically . newAgentXFTPServerStats') xss + atomically . writeTVar ntfServersStats =<< mapM (atomically . newAgentNtfServerStats') nss disconnectAgentClient :: AgentClient -> IO () disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAgent = xa}} = do @@ -375,7 +375,7 @@ getConnectionMessage c = withAgentEnv c . getConnectionMessage' c {-# INLINE getConnectionMessage #-} -- | Get connection message for received notification -getNotificationMessage :: AgentClient -> C.CbNonce -> ByteString -> AE (NotificationInfo, [SMPMsgMeta]) +getNotificationMessage :: AgentClient -> C.CbNonce -> ByteString -> AE (NotificationInfo, Maybe SMPMsgMeta) getNotificationMessage c = withAgentEnv c .: getNotificationMessage' c {-# INLINE getNotificationMessage #-} @@ -1032,7 +1032,7 @@ getConnectionMessage' c connId = do SndConnection _ _ -> throwE $ CONN SIMPLEX NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionMessage: NewConnection" -getNotificationMessage' :: AgentClient -> C.CbNonce -> ByteString -> AM (NotificationInfo, [SMPMsgMeta]) +getNotificationMessage' :: AgentClient -> C.CbNonce -> ByteString -> AM (NotificationInfo, Maybe SMPMsgMeta) getNotificationMessage' c nonce encNtfInfo = do withStore' c getActiveNtfToken >>= \case Just NtfToken {ntfDhSecret = Just dhSecret} -> do @@ -1040,22 +1040,9 @@ getNotificationMessage' c nonce encNtfInfo = do PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} <- liftEither (parse strP (INTERNAL "error parsing PNMessageData") ntfData) (ntfConnId, rcvNtfDhSecret) <- withStore c (`getNtfRcvQueue` smpQueue) ntfMsgMeta <- (eitherToMaybe . smpDecode <$> agentCbDecrypt rcvNtfDhSecret nmsgNonce encNMsgMeta) `catchAgentError` \_ -> pure Nothing - maxMsgs <- asks $ ntfMaxMessages . config - (NotificationInfo {ntfConnId, ntfTs, ntfMsgMeta},) <$> getNtfMessages ntfConnId ntfMsgMeta maxMsgs + msgMeta <- getConnectionMessage' c ntfConnId + pure (NotificationInfo {ntfConnId, ntfTs, ntfMsgMeta}, msgMeta) _ -> throwE $ CMD PROHIBITED "getNotificationMessage" - where - getNtfMessages ntfConnId nMeta = getMsg - where - getMsg 0 = pure [] - getMsg n = - getConnectionMessage' c ntfConnId >>= \case - Just m - | lastMsg m -> pure [m] - | otherwise -> (m :) <$> getMsg (n - 1) - Nothing -> pure [] - lastMsg SMP.SMPMsgMeta {msgId, msgTs, msgFlags} = case nMeta of - Just SMP.NMsgMeta {msgId = msgId', msgTs = msgTs'} -> msgId == msgId' || msgTs > msgTs' - Nothing -> SMP.notification msgFlags -- | Send message to the connection (SEND command) in Reader monad sendMessage' :: AgentClient -> ConnId -> PQEncryption -> MsgFlags -> MsgBody -> AM (AgentMsgId, PQEncryption) diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 43b3b8064..3a7616499 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -145,6 +145,7 @@ module Simplex.Messaging.Agent.Client incXFTPServerStat, incXFTPServerStat', incXFTPServerSizeStat, + incNtfServerStat, AgentWorkersDetails (..), getAgentWorkersDetails, AgentWorkersSummary (..), @@ -330,6 +331,7 @@ data AgentClient = AgentClient agentEnv :: Env, smpServersStats :: TMap (UserId, SMPServer) AgentSMPServerStats, xftpServersStats :: TMap (UserId, XFTPServer) AgentXFTPServerStats, + ntfServersStats :: TMap (UserId, NtfServer) AgentNtfServerStats, srvStatsStartedAt :: TVar UTCTime } @@ -488,6 +490,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a smpSubWorkers <- TM.empty smpServersStats <- TM.empty xftpServersStats <- TM.empty + ntfServersStats <- TM.empty srvStatsStartedAt <- newTVar currentTs return AgentClient @@ -528,6 +531,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a agentEnv, smpServersStats, xftpServersStats, + ntfServersStats, srvStatsStartedAt } @@ -1954,13 +1958,7 @@ incSMPServerStat :: AgentClient -> UserId -> SMPServer -> (AgentSMPServerStats - incSMPServerStat c userId srv sel = incSMPServerStat' c userId srv sel 1 incSMPServerStat' :: AgentClient -> UserId -> SMPServer -> (AgentSMPServerStats -> TVar Int) -> Int -> STM () -incSMPServerStat' AgentClient {smpServersStats} userId srv sel n = do - TM.lookup (userId, srv) smpServersStats >>= \case - Just v -> modifyTVar' (sel v) (+ n) - Nothing -> do - newStats <- newAgentSMPServerStats - modifyTVar' (sel newStats) (+ n) - TM.insert (userId, srv) newStats smpServersStats +incSMPServerStat' = incServerStat (\AgentClient {smpServersStats = s} -> s) newAgentSMPServerStats incXFTPServerStat :: AgentClient -> UserId -> XFTPServer -> (AgentXFTPServerStats -> TVar Int) -> STM () incXFTPServerStat c userId srv sel = incXFTPServerStat_ c userId srv sel 1 @@ -1975,24 +1973,34 @@ incXFTPServerSizeStat = incXFTPServerStat_ {-# INLINE incXFTPServerSizeStat #-} incXFTPServerStat_ :: Num n => AgentClient -> UserId -> XFTPServer -> (AgentXFTPServerStats -> TVar n) -> n -> STM () -incXFTPServerStat_ AgentClient {xftpServersStats} userId srv sel n = do - TM.lookup (userId, srv) xftpServersStats >>= \case +incXFTPServerStat_ = incServerStat (\AgentClient {xftpServersStats = s} -> s) newAgentXFTPServerStats +{-# INLINE incXFTPServerStat_ #-} + +incNtfServerStat :: AgentClient -> UserId -> NtfServer -> (AgentNtfServerStats -> TVar Int) -> STM () +incNtfServerStat c userId srv sel = incServerStat (\AgentClient {ntfServersStats = s} -> s) newAgentNtfServerStats c userId srv sel 1 +{-# INLINE incNtfServerStat #-} + +incServerStat :: Num n => (AgentClient -> TMap (UserId, ProtocolServer p) s) -> STM s -> AgentClient -> UserId -> ProtocolServer p -> (s -> TVar n) -> n -> STM () +incServerStat statsSel mkNewStats c userId srv sel n = do + TM.lookup (userId, srv) (statsSel c) >>= \case Just v -> modifyTVar' (sel v) (+ n) Nothing -> do - newStats <- newAgentXFTPServerStats + newStats <- mkNewStats modifyTVar' (sel newStats) (+ n) - TM.insert (userId, srv) newStats xftpServersStats + TM.insert (userId, srv) newStats (statsSel c) data AgentServersSummary = AgentServersSummary { smpServersStats :: Map (UserId, SMPServer) AgentSMPServerStatsData, xftpServersStats :: Map (UserId, XFTPServer) AgentXFTPServerStatsData, + ntfServersStats :: Map (UserId, NtfServer) AgentNtfServerStatsData, statsStartedAt :: UTCTime, smpServersSessions :: Map (UserId, SMPServer) ServerSessions, smpServersSubs :: Map (UserId, SMPServer) SMPServerSubs, xftpServersSessions :: Map (UserId, XFTPServer) ServerSessions, xftpRcvInProgress :: [XFTPServer], xftpSndInProgress :: [XFTPServer], - xftpDelInProgress :: [XFTPServer] + xftpDelInProgress :: [XFTPServer], + ntfServersSessions :: Map (UserId, NtfServer) ServerSessions } deriving (Show) @@ -2010,9 +2018,10 @@ data ServerSessions = ServerSessions deriving (Show) getAgentServersSummary :: AgentClient -> IO AgentServersSummary -getAgentServersSummary c@AgentClient {smpServersStats, xftpServersStats, srvStatsStartedAt, agentEnv} = do +getAgentServersSummary c@AgentClient {smpServersStats, xftpServersStats, ntfServersStats, srvStatsStartedAt, agentEnv} = do sss <- mapM getAgentSMPServerStats =<< readTVarIO smpServersStats xss <- mapM getAgentXFTPServerStats =<< readTVarIO xftpServersStats + nss <- mapM getAgentNtfServerStats =<< readTVarIO ntfServersStats statsStartedAt <- readTVarIO srvStatsStartedAt smpServersSessions <- countSessions =<< readTVarIO (smpClients c) smpServersSubs <- getServerSubs @@ -2020,17 +2029,20 @@ getAgentServersSummary c@AgentClient {smpServersStats, xftpServersStats, srvStat xftpRcvInProgress <- catMaybes <$> getXFTPWorkerSrvs xftpRcvWorkers xftpSndInProgress <- catMaybes <$> getXFTPWorkerSrvs xftpSndWorkers xftpDelInProgress <- getXFTPWorkerSrvs xftpDelWorkers + ntfServersSessions <- countSessions =<< readTVarIO (ntfClients c) pure AgentServersSummary { smpServersStats = sss, xftpServersStats = xss, + ntfServersStats = nss, statsStartedAt, smpServersSessions, smpServersSubs, xftpServersSessions, xftpRcvInProgress, xftpSndInProgress, - xftpDelInProgress + xftpDelInProgress, + ntfServersSessions } where getServerSubs = do diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 0f88508b9..d21d0fb5e 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -148,10 +148,7 @@ data AgentConfig = AgentConfig xftpMaxRecipientsPerRequest :: Int, deleteErrorCount :: Int, ntfCron :: Word16, - ntfWorkerDelay :: Int, - ntfSMPWorkerDelay :: Int, ntfSubCheckInterval :: NominalDiffTime, - ntfMaxMessages :: Int, caCertificateFile :: FilePath, privateKeyFile :: FilePath, certificateFile :: FilePath, @@ -220,10 +217,7 @@ defaultAgentConfig = xftpMaxRecipientsPerRequest = 200, deleteErrorCount = 10, ntfCron = 20, -- minutes - ntfWorkerDelay = 100000, -- microseconds - ntfSMPWorkerDelay = 500000, -- microseconds ntfSubCheckInterval = nominalDay, - ntfMaxMessages = 3, -- CA certificate private key is not needed for initialization -- ! we do not generate these caCertificateFile = "/etc/opt/simplex-agent/ca.crt", diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index a239768b0..689e5fcce 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs @@ -20,8 +20,8 @@ where import Control.Logger.Simple (logError, logInfo) import Control.Monad -import Control.Monad.Except import Control.Monad.Reader +import Control.Monad.Trans.Except import Data.Bifunctor (first) import qualified Data.Map.Strict as M import Data.Text (Text) @@ -31,6 +31,7 @@ import Simplex.Messaging.Agent.Client import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Protocol (AEvent (..), AEvt (..), AgentErrorType (..), BrokerErrorType (..), ConnId, NotificationsMode (..), SAEntity (..)) import Simplex.Messaging.Agent.RetryInterval +import Simplex.Messaging.Agent.Stats import Simplex.Messaging.Agent.Store import Simplex.Messaging.Agent.Store.SQLite import qualified Simplex.Messaging.Crypto as C @@ -40,7 +41,7 @@ import Simplex.Messaging.Protocol (NtfServer, SMPServer, sameSrvAddr) import Simplex.Messaging.Util (diffToMicroseconds, threadDelay', tshow, unlessM) import System.Random (randomR) import UnliftIO -import UnliftIO.Concurrent (forkIO, threadDelay) +import UnliftIO.Concurrent (forkIO) import qualified UnliftIO.Exception as E runNtfSupervisor :: AgentClient -> AM' () @@ -64,7 +65,7 @@ processNtfSub c (connId, cmd) = do logInfo $ "processNtfSub - connId = " <> tshow connId <> " - cmd = " <> tshow cmd case cmd of NSCCreate -> do - (a, RcvQueue {server = smpServer, clientNtfCreds}) <- withStore c $ \db -> runExceptT $ do + (a, RcvQueue {userId, server = smpServer, clientNtfCreds}) <- withStore c $ \db -> runExceptT $ do a <- liftIO $ getNtfSubscription db connId q <- ExceptT $ getPrimaryRcvQueue db connId pure (a, q) @@ -74,12 +75,12 @@ processNtfSub c (connId, cmd) = do withTokenServer $ \ntfServer -> do case clientNtfCreds of Just ClientNtfCreds {notifierId} -> do - let newSub = newNtfSubscription connId smpServer (Just notifierId) ntfServer NASKey - withStore c $ \db -> createNtfSubscription db newSub $ NtfSubNTFAction NSACreate + let newSub = newNtfSubscription userId connId smpServer (Just notifierId) ntfServer NASKey + withStore c $ \db -> createNtfSubscription db newSub $ NSANtf NSACreate lift . void $ getNtfNTFWorker True c ntfServer Nothing -> do - let newSub = newNtfSubscription connId smpServer Nothing ntfServer NASNew - withStore c $ \db -> createNtfSubscription db newSub $ NtfSubSMPAction NSASmpKey + let newSub = newNtfSubscription userId connId smpServer Nothing ntfServer NASNew + withStore c $ \db -> createNtfSubscription db newSub $ NSASMP NSASmpKey lift . void $ getNtfSMPWorker True c smpServer (Just (sub@NtfSubscription {ntfSubStatus, ntfServer = subNtfServer, smpServer = smpServer', ntfQueueId}, action_)) -> do case (clientNtfCreds, ntfQueueId) of @@ -99,24 +100,24 @@ processNtfSub c (connId, cmd) = do if ntfSubStatus == NASNew || ntfSubStatus == NASOff || ntfSubStatus == NASDeleted then resetSubscription else withTokenServer $ \ntfServer -> do - withStore' c $ \db -> supervisorUpdateNtfSub db sub {ntfServer} (NtfSubNTFAction NSACreate) + withStore' c $ \db -> supervisorUpdateNtfSub db sub {ntfServer} (NSANtf NSACreate) lift . void $ getNtfNTFWorker True c ntfServer | otherwise -> case action of - NtfSubNTFAction _ -> lift . void $ getNtfNTFWorker True c subNtfServer - NtfSubSMPAction _ -> lift . void $ getNtfSMPWorker True c smpServer + NSANtf _ -> lift . void $ getNtfNTFWorker True c subNtfServer + NSASMP _ -> lift . void $ getNtfSMPWorker True c smpServer rotate :: AM () rotate = do - withStore' c $ \db -> supervisorUpdateNtfSub db sub (NtfSubNTFAction NSARotate) + withStore' c $ \db -> supervisorUpdateNtfSub db sub (NSANtf NSARotate) lift . void $ getNtfNTFWorker True c subNtfServer resetSubscription :: AM () resetSubscription = withTokenServer $ \ntfServer -> do let sub' = sub {ntfQueueId = Nothing, ntfServer, ntfSubId = Nothing, ntfSubStatus = NASNew} - withStore' c $ \db -> supervisorUpdateNtfSub db sub' (NtfSubSMPAction NSASmpKey) + withStore' c $ \db -> supervisorUpdateNtfSub db sub' (NSASMP NSASmpKey) lift . void $ getNtfSMPWorker True c smpServer NSCDelete -> do sub_ <- withStore' c $ \db -> do - supervisorUpdateNtfAction db connId (NtfSubNTFAction NSADelete) + supervisorUpdateNtfAction db connId (NSANtf NSADelete) getNtfSubscription db connId logInfo $ "processNtfSub, NSCDelete - sub_ = " <> tshow sub_ case sub_ of @@ -126,7 +127,7 @@ processNtfSub c (connId, cmd) = do withStore' c (`getPrimaryRcvQueue` connId) >>= \case Right rq@RcvQueue {server = smpServer} -> do logInfo $ "processNtfSub, NSCSmpDelete - rq = " <> tshow rq - withStore' c $ \db -> supervisorUpdateNtfAction db connId (NtfSubSMPAction NSASmpDelete) + withStore' c $ \db -> supervisorUpdateNtfAction db connId (NSASMP NSASmpDelete) lift . void $ getNtfSMPWorker True c smpServer _ -> notifyInternalError c connId "NSCSmpDelete - no rcv queue" NSCNtfWorker ntfServer -> lift . void $ getNtfNTFWorker True c ntfServer @@ -146,12 +147,10 @@ withTokenServer :: (NtfServer -> AM ()) -> AM () withTokenServer action = lift getNtfToken >>= mapM_ (\NtfToken {ntfServer} -> action ntfServer) runNtfWorker :: AgentClient -> NtfServer -> Worker -> AM () -runNtfWorker c srv Worker {doWork} = do - delay <- asks $ ntfWorkerDelay . config +runNtfWorker c srv Worker {doWork} = forever $ do waitForWork doWork ExceptT $ agentOperationBracket c AONtfNetwork throwWhenInactive $ runExceptT runNtfOperation - threadDelay delay where runNtfOperation :: AM () runNtfOperation = @@ -164,66 +163,68 @@ runNtfWorker c srv Worker {doWork} = do processSub nextSub `catchAgentError` retryOnError c "NtfWorker" loop (workerInternalError c connId . show) processSub :: (NtfSubscription, NtfSubNTFAction, NtfActionTs) -> AM () - processSub (sub@NtfSubscription {connId, smpServer, ntfSubId}, action, actionTs) = do + processSub (sub@NtfSubscription {userId, connId, smpServer, ntfSubId}, action, actionTs) = do ts <- liftIO getCurrentTime unlessM (lift $ rescheduleAction doWork ts actionTs) $ case action of NSACreate -> lift getNtfToken >>= \case - Just tkn@NtfToken {ntfTokenId = Just tknId, ntfTknStatus = NTActive, ntfMode = NMInstant} -> do + Just tkn@NtfToken {ntfServer, ntfTokenId = Just tknId, ntfTknStatus = NTActive, ntfMode = NMInstant} -> do RcvQueue {clientNtfCreds} <- withStore c (`getPrimaryRcvQueue` connId) case clientNtfCreds of Just ClientNtfCreds {ntfPrivateKey, notifierId} -> do + atomically $ incNtfServerStat c userId ntfServer ntfCreateAttempts nSubId <- agentNtfCreateSubscription c tknId tkn (SMPQueueNtf smpServer notifierId) ntfPrivateKey + atomically $ incNtfServerStat c userId ntfServer ntfCreated -- possible improvement: smaller retry until Active, less frequently (daily?) once Active let actionTs' = addUTCTime 30 ts withStore' c $ \db -> - updateNtfSubscription db sub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew} (NtfSubNTFAction NSACheck) actionTs' + updateNtfSubscription db sub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew} (NSANtf NSACheck) actionTs' _ -> workerInternalError c connId "NSACreate - no notifier queue credentials" _ -> workerInternalError c connId "NSACreate - no active token" NSACheck -> lift getNtfToken >>= \case - Just tkn -> + Just tkn@NtfToken {ntfServer} -> case ntfSubId of - Just nSubId -> + Just nSubId -> do + atomically $ incNtfServerStat c userId ntfServer ntfCheckAttempts agentNtfCheckSubscription c nSubId tkn >>= \case NSAuth -> do - lift (getNtfServer c) >>= \case - Just ntfServer -> do - withStore' c $ \db -> - updateNtfSubscription db sub {ntfServer, ntfQueueId = Nothing, ntfSubId = Nothing, ntfSubStatus = NASNew} (NtfSubSMPAction NSASmpKey) ts - ns <- asks ntfSupervisor - atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCNtfSMPWorker smpServer) - _ -> workerInternalError c connId "NSACheck - failed to reset subscription, notification server not configured" + withStore' c $ \db -> + updateNtfSubscription db sub {ntfServer, ntfQueueId = Nothing, ntfSubId = Nothing, ntfSubStatus = NASNew} (NSASMP NSASmpKey) ts + ns <- asks ntfSupervisor + atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCNtfSMPWorker smpServer) status -> updateSubNextCheck ts status + atomically $ incNtfServerStat c userId ntfServer ntfChecked Nothing -> workerInternalError c connId "NSACheck - no subscription ID" _ -> workerInternalError c connId "NSACheck - no active token" - NSADelete -> case ntfSubId of - Just nSubId -> - (lift getNtfToken >>= mapM_ (agentNtfDeleteSubscription c nSubId)) - `agentFinally` continueDeletion - _ -> continueDeletion - where - continueDeletion = do - let sub' = sub {ntfSubId = Nothing, ntfSubStatus = NASOff} - withStore' c $ \db -> updateNtfSubscription db sub' (NtfSubSMPAction NSASmpDelete) ts - ns <- asks ntfSupervisor - atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCNtfSMPWorker smpServer) - NSARotate -> case ntfSubId of - Just nSubId -> - (lift getNtfToken >>= mapM_ (agentNtfDeleteSubscription c nSubId)) - `agentFinally` deleteCreate - _ -> deleteCreate - where - deleteCreate = do - withStore' c $ \db -> deleteNtfSubscription db connId - ns <- asks ntfSupervisor - atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCCreate) + NSADelete -> + deleteNtfSub $ do + let sub' = sub {ntfSubId = Nothing, ntfSubStatus = NASOff} + withStore' c $ \db -> updateNtfSubscription db sub' (NSASMP NSASmpDelete) ts + ns <- asks ntfSupervisor + atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCNtfSMPWorker smpServer) + NSARotate -> + deleteNtfSub $ do + withStore' c $ \db -> deleteNtfSubscription db connId + ns <- asks ntfSupervisor + atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCCreate) where + deleteNtfSub continue = case ntfSubId of + Just nSubId -> + lift getNtfToken >>= \case + Just tkn@NtfToken {ntfServer} -> do + atomically $ incNtfServerStat c userId ntfServer ntfDelAttempts + tryAgentError (agentNtfDeleteSubscription c nSubId tkn) >>= \case + Left e | temporaryOrHostError e -> throwE e + _ -> continue + atomically $ incNtfServerStat c userId ntfServer ntfDeleted + Nothing -> continue + _ -> continue updateSubNextCheck ts toStatus = do checkInterval <- asks $ ntfSubCheckInterval . config let nextCheckTs = addUTCTime checkInterval ts - updateSub (NASCreated toStatus) (NtfSubNTFAction NSACheck) nextCheckTs + updateSub (NASCreated toStatus) (NSANtf NSACheck) nextCheckTs updateSub toStatus toAction actionTs' = withStore' c $ \db -> updateNtfSubscription db sub {ntfSubStatus = toStatus} toAction actionTs' @@ -231,12 +232,10 @@ runNtfWorker c srv Worker {doWork} = do runNtfSMPWorker :: AgentClient -> SMPServer -> Worker -> AM () runNtfSMPWorker c srv Worker {doWork} = do env <- ask - delay <- asks $ ntfSMPWorkerDelay . config forever $ do waitForWork doWork ExceptT . liftIO . agentOperationBracket c AONtfNetwork throwWhenInactive $ runReaderT (runExceptT runNtfSMPOperation) env - threadDelay delay where runNtfSMPOperation = withWork c doWork (`getNextNtfSubSMPAction` srv) $ @@ -264,11 +263,12 @@ runNtfSMPWorker c srv Worker {doWork} = do let rcvNtfDhSecret = C.dh' rcvNtfSrvPubDhKey rcvNtfPrivDhKey withStore' c $ \db -> do setRcvQueueNtfCreds db connId $ Just ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret} - updateNtfSubscription db sub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey} (NtfSubNTFAction NSACreate) ts + updateNtfSubscription db sub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey} (NSANtf NSACreate) ts ns <- asks ntfSupervisor atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer) _ -> workerInternalError c connId "NSASmpKey - no active token" NSASmpDelete -> do + -- TODO should we remove it after successful removal from the server? rq_ <- withStore' c $ \db -> do setRcvQueueNtfCreds db connId Nothing getPrimaryRcvQueue db connId diff --git a/src/Simplex/Messaging/Agent/Stats.hs b/src/Simplex/Messaging/Agent/Stats.hs index 424052d74..d4663bfb1 100644 --- a/src/Simplex/Messaging/Agent/Stats.hs +++ b/src/Simplex/Messaging/Agent/Stats.hs @@ -1,17 +1,20 @@ {-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE TemplateHaskell #-} module Simplex.Messaging.Agent.Stats where +import Data.Aeson (FromJSON (..), FromJSONKey, ToJSON (..)) import qualified Data.Aeson.TH as J import Data.Int (Int64) -import Data.Map (Map) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as M import Database.SQLite.Simple.FromField (FromField (..)) import Database.SQLite.Simple.ToField (ToField (..)) import Simplex.Messaging.Agent.Protocol (UserId) import Simplex.Messaging.Parsers (defaultJSON, fromTextField_) -import Simplex.Messaging.Protocol (SMPServer, XFTPServer) +import Simplex.Messaging.Protocol (SMPServer, XFTPServer, NtfServer) import Simplex.Messaging.Util (decodeJSON, encodeJSON) import UnliftIO.STM @@ -44,7 +47,12 @@ data AgentSMPServerStats = AgentSMPServerStats connSubscribed :: TVar Int, -- total successful subscription connSubAttempts :: TVar Int, -- subscription attempts connSubIgnored :: TVar Int, -- subscription results ignored (client switched to different session or it was not pending) - connSubErrs :: TVar Int -- permanent subscription errors (temporary accounted for in attempts) + connSubErrs :: TVar Int, -- permanent subscription errors (temporary accounted for in attempts) + -- notifications stats + ntfKey :: TVar Int, + ntfKeyAttempts :: TVar Int, + ntfKeyDeleted :: TVar Int, + ntfKeyDeleteAttempts :: TVar Int } data AgentSMPServerStatsData = AgentSMPServerStatsData @@ -75,10 +83,17 @@ data AgentSMPServerStatsData = AgentSMPServerStatsData _connSubscribed :: Int, _connSubAttempts :: Int, _connSubIgnored :: Int, - _connSubErrs :: Int + _connSubErrs :: Int, + _ntfKey :: OptionalInt, + _ntfKeyAttempts :: OptionalInt, + _ntfKeyDeleted :: OptionalInt, + _ntfKeyDeleteAttempts :: OptionalInt } deriving (Show) +newtype OptionalInt = OInt {toInt :: Int} + deriving (Num, Show, ToJSON) + newAgentSMPServerStats :: STM AgentSMPServerStats newAgentSMPServerStats = do sentDirect <- newTVar 0 @@ -109,6 +124,10 @@ newAgentSMPServerStats = do connSubAttempts <- newTVar 0 connSubIgnored <- newTVar 0 connSubErrs <- newTVar 0 + ntfKey <- newTVar 0 + ntfKeyAttempts <- newTVar 0 + ntfKeyDeleted <- newTVar 0 + ntfKeyDeleteAttempts <- newTVar 0 pure AgentSMPServerStats { sentDirect, @@ -138,7 +157,11 @@ newAgentSMPServerStats = do connSubscribed, connSubAttempts, connSubIgnored, - connSubErrs + connSubErrs, + ntfKey, + ntfKeyAttempts, + ntfKeyDeleted, + ntfKeyDeleteAttempts } newAgentSMPServerStatsData :: AgentSMPServerStatsData @@ -171,7 +194,11 @@ newAgentSMPServerStatsData = _connSubscribed = 0, _connSubAttempts = 0, _connSubIgnored = 0, - _connSubErrs = 0 + _connSubErrs = 0, + _ntfKey = 0, + _ntfKeyAttempts = 0, + _ntfKeyDeleted = 0, + _ntfKeyDeleteAttempts = 0 } newAgentSMPServerStats' :: AgentSMPServerStatsData -> STM AgentSMPServerStats @@ -204,6 +231,10 @@ newAgentSMPServerStats' s = do connSubAttempts <- newTVar $ _connSubAttempts s connSubIgnored <- newTVar $ _connSubIgnored s connSubErrs <- newTVar $ _connSubErrs s + ntfKey <- newTVar $ toInt $ _ntfKey s + ntfKeyAttempts <- newTVar $ toInt $ _ntfKeyAttempts s + ntfKeyDeleted <- newTVar $ toInt $ _ntfKeyDeleted s + ntfKeyDeleteAttempts <- newTVar $ toInt $ _ntfKeyDeleteAttempts s pure AgentSMPServerStats { sentDirect, @@ -233,7 +264,11 @@ newAgentSMPServerStats' s = do connSubscribed, connSubAttempts, connSubIgnored, - connSubErrs + connSubErrs, + ntfKey, + ntfKeyAttempts, + ntfKeyDeleted, + ntfKeyDeleteAttempts } -- as this is used to periodically update stats in db, @@ -268,6 +303,10 @@ getAgentSMPServerStats s = do _connSubAttempts <- readTVarIO $ connSubAttempts s _connSubIgnored <- readTVarIO $ connSubIgnored s _connSubErrs <- readTVarIO $ connSubErrs s + _ntfKey <- OInt <$> readTVarIO (ntfKey s) + _ntfKeyAttempts <- OInt <$> readTVarIO (ntfKeyAttempts s) + _ntfKeyDeleted <- OInt <$> readTVarIO (ntfKeyDeleted s) + _ntfKeyDeleteAttempts <- OInt <$> readTVarIO (ntfKeyDeleteAttempts s) pure AgentSMPServerStatsData { _sentDirect, @@ -297,7 +336,11 @@ getAgentSMPServerStats s = do _connSubscribed, _connSubAttempts, _connSubIgnored, - _connSubErrs + _connSubErrs, + _ntfKey, + _ntfKeyAttempts, + _ntfKeyDeleted, + _ntfKeyDeleteAttempts } addSMPStatsData :: AgentSMPServerStatsData -> AgentSMPServerStatsData -> AgentSMPServerStatsData @@ -330,7 +373,11 @@ addSMPStatsData sd1 sd2 = _connSubscribed = _connSubscribed sd1 + _connSubscribed sd2, _connSubAttempts = _connSubAttempts sd1 + _connSubAttempts sd2, _connSubIgnored = _connSubIgnored sd1 + _connSubIgnored sd2, - _connSubErrs = _connSubErrs sd1 + _connSubErrs sd2 + _connSubErrs = _connSubErrs sd1 + _connSubErrs sd2, + _ntfKey = _ntfKey sd1 + _ntfKey sd2, + _ntfKeyAttempts = _ntfKeyAttempts sd1 + _ntfKeyAttempts sd2, + _ntfKeyDeleted = _ntfKeyDeleted sd1 + _ntfKeyDeleted sd2, + _ntfKeyDeleteAttempts = _ntfKeyDeleteAttempts sd1 + _ntfKeyDeleteAttempts sd2 } data AgentXFTPServerStats = AgentXFTPServerStats @@ -490,18 +537,127 @@ addXFTPStatsData sd1 sd2 = _deleteErrs = _deleteErrs sd1 + _deleteErrs sd2 } +data AgentNtfServerStats = AgentNtfServerStats + { ntfCreated :: TVar Int, + ntfCreateAttempts :: TVar Int, + ntfChecked :: TVar Int, + ntfCheckAttempts :: TVar Int, + ntfDeleted :: TVar Int, + ntfDelAttempts :: TVar Int + } + +data AgentNtfServerStatsData = AgentNtfServerStatsData + { _ntfCreated :: Int, + _ntfCreateAttempts :: Int, + _ntfChecked :: Int, + _ntfCheckAttempts :: Int, + _ntfDeleted :: Int, + _ntfDelAttempts :: Int + } + deriving (Show) + +newAgentNtfServerStats :: STM AgentNtfServerStats +newAgentNtfServerStats = do + ntfCreated <- newTVar 0 + ntfCreateAttempts <- newTVar 0 + ntfChecked <- newTVar 0 + ntfCheckAttempts <- newTVar 0 + ntfDeleted <- newTVar 0 + ntfDelAttempts <- newTVar 0 + pure + AgentNtfServerStats + { ntfCreated, + ntfCreateAttempts, + ntfChecked, + ntfCheckAttempts, + ntfDeleted, + ntfDelAttempts + } + +newAgentNtfServerStatsData :: AgentNtfServerStatsData +newAgentNtfServerStatsData = + AgentNtfServerStatsData + { _ntfCreated = 0, + _ntfCreateAttempts = 0, + _ntfChecked = 0, + _ntfCheckAttempts = 0, + _ntfDeleted = 0, + _ntfDelAttempts = 0 + } + +newAgentNtfServerStats' :: AgentNtfServerStatsData -> STM AgentNtfServerStats +newAgentNtfServerStats' s = do + ntfCreated <- newTVar $ _ntfCreated s + ntfCreateAttempts <- newTVar $ _ntfCreateAttempts s + ntfChecked <- newTVar $ _ntfChecked s + ntfCheckAttempts <- newTVar $ _ntfCheckAttempts s + ntfDeleted <- newTVar $ _ntfDeleted s + ntfDelAttempts <- newTVar $ _ntfDelAttempts s + pure + AgentNtfServerStats + { ntfCreated, + ntfCreateAttempts, + ntfChecked, + ntfCheckAttempts, + ntfDeleted, + ntfDelAttempts + } + +getAgentNtfServerStats :: AgentNtfServerStats -> IO AgentNtfServerStatsData +getAgentNtfServerStats s = do + _ntfCreated <- readTVarIO $ ntfCreated s + _ntfCreateAttempts <- readTVarIO $ ntfCreateAttempts s + _ntfChecked <- readTVarIO $ ntfChecked s + _ntfCheckAttempts <- readTVarIO $ ntfCheckAttempts s + _ntfDeleted <- readTVarIO $ ntfDeleted s + _ntfDelAttempts <- readTVarIO $ ntfDelAttempts s + pure + AgentNtfServerStatsData + { _ntfCreated, + _ntfCreateAttempts, + _ntfChecked, + _ntfCheckAttempts, + _ntfDeleted, + _ntfDelAttempts + } + +addNtfStatsData :: AgentNtfServerStatsData -> AgentNtfServerStatsData -> AgentNtfServerStatsData +addNtfStatsData sd1 sd2 = + AgentNtfServerStatsData + { _ntfCreated = _ntfCreated sd1 + _ntfCreated sd2, + _ntfCreateAttempts = _ntfCreateAttempts sd1 + _ntfCreateAttempts sd2, + _ntfChecked = _ntfChecked sd1 + _ntfChecked sd2, + _ntfCheckAttempts = _ntfCheckAttempts sd1 + _ntfCheckAttempts sd2, + _ntfDeleted = _ntfDeleted sd1 + _ntfDeleted sd2, + _ntfDelAttempts = _ntfDelAttempts sd1 + _ntfDelAttempts sd2 + } + -- Type for gathering both smp and xftp stats across all users and servers, -- to then be persisted to db as a single json. data AgentPersistedServerStats = AgentPersistedServerStats { smpServersStats :: Map (UserId, SMPServer) AgentSMPServerStatsData, - xftpServersStats :: Map (UserId, XFTPServer) AgentXFTPServerStatsData + xftpServersStats :: Map (UserId, XFTPServer) AgentXFTPServerStatsData, + ntfServersStats :: OptionalMap (UserId, NtfServer) AgentNtfServerStatsData } deriving (Show) +instance FromJSON OptionalInt where + parseJSON v = OInt <$> parseJSON v + omittedField = Just (OInt 0) + +newtype OptionalMap k v = OptionalMap (Map k v) + deriving (Show, ToJSON) + +instance (FromJSONKey k, Ord k, FromJSON v) => FromJSON (OptionalMap k v) where + parseJSON v = OptionalMap <$> parseJSON v + omittedField = Just (OptionalMap M.empty) + $(J.deriveJSON defaultJSON ''AgentSMPServerStatsData) $(J.deriveJSON defaultJSON ''AgentXFTPServerStatsData) +$(J.deriveJSON defaultJSON ''AgentNtfServerStatsData) + $(J.deriveJSON defaultJSON ''AgentPersistedServerStats) instance ToField AgentPersistedServerStats where diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 0727343e7..574137883 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -1457,23 +1457,24 @@ getNtfSubscription db connId = DB.query db [sql| - SELECT s.host, s.port, COALESCE(nsb.smp_server_key_hash, s.key_hash), ns.ntf_host, ns.ntf_port, ns.ntf_key_hash, + SELECT c.user_id, s.host, s.port, COALESCE(nsb.smp_server_key_hash, 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, nsb.ntf_sub_smp_action, nsb.ntf_sub_action_ts FROM ntf_subscriptions nsb + JOIN connections c USING (conn_id) 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, ntfAction_, smpAction_, actionTs_) = + ntfSubscription ((userId, smpHost, smpPort, smpKeyHash, ntfHost, ntfPort, ntfKeyHash ) :. (ntfQueueId, ntfSubId, ntfSubStatus, ntfAction_, smpAction_, actionTs_)) = let smpServer = SMPServer smpHost smpPort smpKeyHash ntfServer = NtfServer ntfHost ntfPort ntfKeyHash action = case (ntfAction_, smpAction_, actionTs_) of - (Just ntfAction, Nothing, Just actionTs) -> Just (NtfSubNTFAction ntfAction, actionTs) - (Nothing, Just smpAction, Just actionTs) -> Just (NtfSubSMPAction smpAction, actionTs) + (Just ntfAction, Nothing, Just actionTs) -> Just (NSANtf ntfAction, actionTs) + (Nothing, Just smpAction, Just actionTs) -> Just (NSASMP smpAction, actionTs) _ -> Nothing - in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus}, action) + in (NtfSubscription {userId, connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus}, action) createNtfSubscription :: DB.Connection -> NtfSubscription -> NtfSubAction -> IO (Either StoreError ()) createNtfSubscription db ntfSubscription action = runExceptT $ do @@ -1607,18 +1608,19 @@ getNextNtfSubNTFAction db ntfServer@(NtfServer ntfHost ntfPort _) = DB.query db [sql| - SELECT s.host, s.port, COALESCE(ns.smp_server_key_hash, s.key_hash), + SELECT c.user_id, s.host, s.port, COALESCE(ns.smp_server_key_hash, 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 connections c USING (conn_id) JOIN servers s ON s.host = ns.smp_host AND s.port = ns.smp_port WHERE ns.conn_id = ? |] (Only connId) where err = SEInternal $ "ntf subscription " <> bshow connId <> " returned []" - ntfSubAction (smpHost, smpPort, smpKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, actionTs, action) = + ntfSubAction (userId, smpHost, smpPort, smpKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, actionTs, action) = let smpServer = SMPServer smpHost smpPort smpKeyHash - ntfSubscription = NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus} + ntfSubscription = NtfSubscription {userId, connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus} in (ntfSubscription, action, actionTs) markNtfSubActionNtfFailed_ :: DB.Connection -> ConnId -> IO () @@ -1650,18 +1652,19 @@ getNextNtfSubSMPAction db smpServer@(SMPServer smpHost smpPort _) = DB.query db [sql| - SELECT s.ntf_host, s.ntf_port, s.ntf_key_hash, + SELECT c.user_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 connections c USING (conn_id) JOIN ntf_servers s USING (ntf_host, ntf_port) WHERE ns.conn_id = ? |] (Only connId) where err = SEInternal $ "ntf subscription " <> bshow connId <> " returned []" - ntfSubAction (ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, actionTs, action) = + ntfSubAction (userId, ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, actionTs, action) = let ntfServer = NtfServer ntfHost ntfPort ntfKeyHash - ntfSubscription = NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus} + ntfSubscription = NtfSubscription {userId, connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus} in (ntfSubscription, action, actionTs) markNtfSubActionSMPFailed_ :: DB.Connection -> ConnId -> IO () @@ -2272,8 +2275,8 @@ randomId :: TVar ChaChaDRG -> Int -> IO ByteString randomId gVar n = atomically $ U.encode <$> C.randomBytes n gVar ntfSubAndSMPAction :: NtfSubAction -> (Maybe NtfSubNTFAction, Maybe NtfSubSMPAction) -ntfSubAndSMPAction (NtfSubNTFAction action) = (Just action, Nothing) -ntfSubAndSMPAction (NtfSubSMPAction action) = (Nothing, Just action) +ntfSubAndSMPAction (NSANtf action) = (Just action, Nothing) +ntfSubAndSMPAction (NSASMP action) = (Nothing, Just action) createXFTPServer_ :: DB.Connection -> XFTPServer -> IO Int64 createXFTPServer_ db newSrv@ProtocolServer {host, port, keyHash} = diff --git a/src/Simplex/Messaging/Notifications/Types.hs b/src/Simplex/Messaging/Notifications/Types.hs index 4465f8767..97f9e6bdd 100644 --- a/src/Simplex/Messaging/Notifications/Types.hs +++ b/src/Simplex/Messaging/Notifications/Types.hs @@ -11,7 +11,7 @@ import Data.Text.Encoding (decodeLatin1, encodeUtf8) import Data.Time (UTCTime) import Database.SQLite.Simple.FromField (FromField (..)) import Database.SQLite.Simple.ToField (ToField (..)) -import Simplex.Messaging.Agent.Protocol (ConnId, NotificationsMode (..)) +import Simplex.Messaging.Agent.Protocol (UserId, ConnId, NotificationsMode (..)) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding import Simplex.Messaging.Notifications.Protocol @@ -79,17 +79,17 @@ newNtfToken deviceToken ntfServer (ntfPubKey, ntfPrivKey) ntfDhKeys ntfMode = ntfMode } -data NtfSubAction = NtfSubNTFAction NtfSubNTFAction | NtfSubSMPAction NtfSubSMPAction +data NtfSubAction = NSANtf NtfSubNTFAction | NSASMP NtfSubSMPAction deriving (Show) isDeleteNtfSubAction :: NtfSubAction -> Bool isDeleteNtfSubAction = \case - NtfSubNTFAction a -> case a of + NSANtf a -> case a of NSACreate -> False NSACheck -> False NSADelete -> True NSARotate -> True - NtfSubSMPAction a -> case a of + NSASMP a -> case a of NSASmpKey -> False NSASmpDelete -> True @@ -177,7 +177,8 @@ instance FromField NtfAgentSubStatus where fromField = fromTextField_ $ either ( instance ToField NtfAgentSubStatus where toField = toField . decodeLatin1 . smpEncode data NtfSubscription = NtfSubscription - { connId :: ConnId, + { userId :: UserId, + connId :: ConnId, smpServer :: SMPServer, ntfQueueId :: Maybe NotifierId, ntfServer :: NtfServer, @@ -186,10 +187,11 @@ data NtfSubscription = NtfSubscription } deriving (Show) -newNtfSubscription :: ConnId -> SMPServer -> Maybe NotifierId -> NtfServer -> NtfAgentSubStatus -> NtfSubscription -newNtfSubscription connId smpServer ntfQueueId ntfServer ntfSubStatus = +newNtfSubscription :: UserId -> ConnId -> SMPServer -> Maybe NotifierId -> NtfServer -> NtfAgentSubStatus -> NtfSubscription +newNtfSubscription userId connId smpServer ntfQueueId ntfServer ntfSubStatus = NtfSubscription - { connId, + { userId, + connId, smpServer, ntfQueueId, ntfServer, diff --git a/tests/AgentTests/NotificationTests.hs b/tests/AgentTests/NotificationTests.hs index 41ed8c730..cc79faeca 100644 --- a/tests/AgentTests/NotificationTests.hs +++ b/tests/AgentTests/NotificationTests.hs @@ -511,7 +511,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} baseId ali -- aliceNtf client doesn't have subscription and is allowed to get notification message withAgent 3 aliceCfg initAgentServers testDB $ \aliceNtf -> runRight_ $ do - (_, [SMPMsgMeta {msgFlags = MsgFlags True}]) <- getNotificationMessage aliceNtf nonce message + (_, Just SMPMsgMeta {msgFlags = MsgFlags True}) <- getNotificationMessage aliceNtf nonce message pure () threadDelay 1000000 diff --git a/tests/SMPAgentClient.hs b/tests/SMPAgentClient.hs index e0de57466..0bb050cbe 100644 --- a/tests/SMPAgentClient.hs +++ b/tests/SMPAgentClient.hs @@ -72,8 +72,6 @@ agentCfg = ntfCfg = defaultNTFClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS), networkConfig}, reconnectInterval = fastRetryInterval, persistErrorInterval = 1, - ntfWorkerDelay = 100, - ntfSMPWorkerDelay = 100, caCertificateFile = "tests/fixtures/ca.crt", privateKeyFile = "tests/fixtures/server.key", certificateFile = "tests/fixtures/server.crt"