mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-06 23:40:18 +00:00
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
This commit is contained in:
Evgeny Poberezkin
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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} =
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user