From 151feece262b7fe7f460abc69f86c61ab6dc40ef Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Mon, 25 May 2026 13:26:29 +0000 Subject: [PATCH] refactor --- src/Simplex/Messaging/Agent.hs | 6 +- src/Simplex/Messaging/Agent/Client.hs | 6 +- src/Simplex/Messaging/Client/Agent.hs | 8 +- src/Simplex/Messaging/Notifications/Server.hs | 154 +++++++++--------- .../Messaging/Notifications/Server/Env.hs | 15 +- src/Simplex/Messaging/Server/Env/STM.hs | 2 +- 6 files changed, 90 insertions(+), 101 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 8801a68f0..b8adb8315 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -145,7 +145,6 @@ where import Control.Applicative ((<|>)) import Control.Concurrent.STM (retry) -import Data.IORef import Control.Logger.Simple import Control.Monad import Control.Monad.Except @@ -272,13 +271,10 @@ getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp, netC currentTs <- liftIO getCurrentTime notices <- liftIO $ withTransaction store (`getClientNotices` presetServers) `catchAll_` pure [] env <- ask - cRef <- liftIO $ newIORef (error "agent client not initialized") - let processMsg t = do - c <- readIORef cRef + let processMsg c t = agentOperationBracket c AORcvNetwork waitUntilActive (processSMPTransmissions c t) `runReaderT` env `catchOwn` \e -> atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ "subscriber error: " <> show e) c@AgentClient {acThread} <- liftIO $ newAgentClient clientId initServers currentTs notices processMsg env - liftIO $ writeIORef cRef c t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c) atomically . writeTVar acThread . Just =<< mkWeakThreadId t pure c diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index c2ef42c29..e70004a56 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -338,7 +338,7 @@ data AgentClient = AgentClient { acThread :: TVar (Maybe (Weak ThreadId)), active :: TVar Bool, subQ :: TBQueue ATransmission, - processServerMsg :: ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO (), + processServerMsg :: AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO (), smpServers :: TMap UserId (UserServers 'PSMP), smpClients :: TMap SMPTransportSession SMPClientVar, useClientServices :: TMap UserId Bool, @@ -505,7 +505,7 @@ data UserNetworkType = UNNone | UNCellular | UNWifi | UNEthernet | UNOther deriving (Eq, Show) -- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's. -newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Map (Maybe SMPServer) (Maybe SystemSeconds) -> (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Env -> IO AgentClient +newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Map (Maybe SMPServer) (Maybe SystemSeconds) -> (AgentClient -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Env -> IO AgentClient newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg, useServices, presetDomains, presetServers} currentTs notices processServerMsg agentEnv = do let cfg = config agentEnv qSize = tbqSize cfg @@ -732,7 +732,7 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq Nothing -> Left $ BROKER (B.unpack $ strEncode srv) TIMEOUT smpConnectClient :: AgentClient -> NetworkRequestMode -> SMPTransportSession -> TMap SMPServer ProxiedRelayVar -> SMPClientVar -> AM SMPConnectedClient -smpConnectClient c@AgentClient {smpClients, proxySessTs, presetDomains} nm tSess@(userId, srv, _) prs v = +smpConnectClient c@AgentClient {processServerMsg, smpClients, proxySessTs, presetDomains} nm tSess@(userId, srv, _) prs v = newProtocolClient c tSess smpClients connectClient v `catchAllErrors` \e -> lift (resubscribeSMPSession c tSess) >> throwE e where diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index a1699009d..f035a800f 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/src/Simplex/Messaging/Client/Agent.hs @@ -138,7 +138,7 @@ data SMPClientAgent p = SMPClientAgent dbService :: Maybe DBService, active :: TVar Bool, startedAt :: UTCTime, - processMsg :: ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO (), + processMsg :: SMPClientAgent p -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO (), agentQ :: TBQueue SMPClientAgentEvent, randomDrg :: TVar ChaChaDRG, smpClients :: TMap SMPServer SMPClientVar, @@ -158,7 +158,7 @@ data SMPClientAgent p = SMPClientAgent type OwnServer = Bool -newSMPClientAgent :: SParty p -> SMPClientAgentConfig -> (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Maybe DBService -> TVar ChaChaDRG -> IO (SMPClientAgent p) +newSMPClientAgent :: SParty p -> SMPClientAgentConfig -> (SMPClientAgent p -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Maybe DBService -> TVar ChaChaDRG -> IO (SMPClientAgent p) newSMPClientAgent agentParty agentCfg@SMPClientAgentConfig {agentQSize} processMsg dbService randomDrg = do active <- newTVarIO True startedAt <- getCurrentTime @@ -256,7 +256,7 @@ isOwnServer SMPClientAgent {agentCfg} ProtocolServer {host} = -- | Run an SMP client for SMPClientVar connectClient :: SMPClientAgent p -> SMPServer -> SMPClientVar -> IO (Either SMPClientError SMPClient) -connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, processMsg, randomDrg, startedAt} srv v = case dbService of +connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, processMsg, randomDrg, startedAt} srv v = case dbService of Just dbs -> runExceptT $ do creds <- ExceptT $ getCredentials dbs srv smp <- ExceptT $ getClient cfg {serviceCredentials = Just creds} @@ -266,7 +266,7 @@ connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, p Nothing -> getClient cfg where cfg = smpCfg agentCfg - getClient cfg' = getProtocolClient randomDrg NRMBackground (1, srv, Nothing) cfg' [] (Just processMsg) startedAt clientDisconnected + getClient cfg' = getProtocolClient randomDrg NRMBackground (1, srv, Nothing) cfg' [] (Just $ processMsg ca) startedAt clientDisconnected clientDisconnected :: SMPClient -> IO () clientDisconnected smp = do diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index a8579c425..e4d20acff 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -101,11 +101,7 @@ runNtfServer cfg = do runNtfServerBlocking started cfg runNtfServerBlocking :: TMVar Bool -> NtfServerConfig -> IO () -runNtfServerBlocking started cfg = runReaderT (ntfServer cfg started) =<< newNtfServerEnv cfg processMsg - where - processMsg envRef t = do - env <- readIORef envRef - receiveSMPMessage env t +runNtfServerBlocking started cfg = runReaderT (ntfServer cfg started) =<< newNtfServerEnv cfg receiveSMPMessage type M a = ReaderT NtfEnv IO a @@ -529,42 +525,37 @@ subscribeNtfs NtfSubscriber {smpSubscribers, subscriberSeq, smpAgent = ca} st sm void $ updateSubStatus st srvId' nId NSPending subscribeQueuesNtfs ca smpServer' [sub] -receiveSMPMessage :: NtfEnv -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO () -receiveSMPMessage env ((_, srv@(SMPServer (h :| _) _ _), _), THandleParams {sessionId}, ts) = - (`runReaderT` env) $ do - st <- asks store - ps <- asks pushServer - stats <- asks serverStats - let ca = smpAgent $ subscriber env - forM_ ts $ \(ntfId, t) -> case t of - STUnexpectedError e -> logError $ "SMP client unexpected error: " <> tshow e -- uncorrelated response, should not happen - STResponse {} -> pure () -- it was already reported as timeout error - STEvent msgOrErr -> do - let smpQueue = SMPQueueNtf srv ntfId - case msgOrErr of - Right (SMP.NMSG nmsgNonce encNMsgMeta) -> do - ntfTs <- liftIO getSystemTime - liftIO $ updatePeriodStats (activeSubs stats) ntfId - let newNtf = PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} - srvHost = safeDecodeUtf8 $ strEncode h - isOwn = isOwnServer ca srv - liftIO (addTokenLastNtf st newNtf) >>= \case - Right (tkn, lastNtfs) -> do - pushNotification ps (Just srvHost) isOwn tkn $ PNMessage lastNtfs - liftIO $ incNtfStat_ stats ntfReceived - when isOwn $ liftIO $ incServerStat srvHost (ntfReceivedOwn stats) - Left AUTH -> liftIO $ do - incNtfStat_ stats ntfReceivedAuth - when isOwn $ incServerStat srvHost (ntfReceivedAuthOwn stats) - Left _ -> pure () - Right SMP.END -> - whenM (atomically $ activeClientSession' ca sessionId srv) $ - void $ liftIO $ updateSrvSubStatus st smpQueue NSEnd - Right SMP.DELD -> - void $ liftIO $ updateSrvSubStatus st smpQueue NSDeleted - Right (SMP.ERR e) -> logError $ "SMP server error: " <> tshow e - Right _ -> logError "SMP server unexpected response" - Left e -> logError $ "SMP client error: " <> tshow e +receiveSMPMessage :: NtfPostgresStore -> NtfPushServer -> NtfServerStats -> SMPClientAgent 'NotifierService -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO () +receiveSMPMessage st ps stats ca ((_, srv@(SMPServer (h :| _) _ _), _), THandleParams {sessionId}, ts) = + forM_ ts $ \(ntfId, t) -> case t of + STUnexpectedError e -> logError $ "SMP client unexpected error: " <> tshow e -- uncorrelated response, should not happen + STResponse {} -> pure () -- it was already reported as timeout error + STEvent msgOrErr -> do + let smpQueue = SMPQueueNtf srv ntfId + case msgOrErr of + Right (SMP.NMSG nmsgNonce encNMsgMeta) -> do + ntfTs <- getSystemTime + updatePeriodStats (activeSubs stats) ntfId + let newNtf = PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} + srvHost = safeDecodeUtf8 $ strEncode h + isOwn = isOwnServer ca srv + addTokenLastNtf st newNtf >>= \case + Right (tkn, lastNtfs) -> do + pushNotification st stats ps (Just srvHost) isOwn tkn $ PNMessage lastNtfs + incNtfStat_ stats ntfReceived + when isOwn $ incServerStat srvHost (ntfReceivedOwn stats) + Left AUTH -> do + incNtfStat_ stats ntfReceivedAuth + when isOwn $ incServerStat srvHost (ntfReceivedAuthOwn stats) + Left _ -> pure () + Right SMP.END -> + whenM (atomically $ activeClientSession' ca sessionId srv) $ + void $ updateSrvSubStatus st smpQueue NSEnd + Right SMP.DELD -> + void $ updateSrvSubStatus st smpQueue NSDeleted + Right (SMP.ERR e) -> logError $ "SMP server error: " <> tshow e + Right _ -> logError "SMP server unexpected response" + Left e -> logError $ "SMP client error: " <> tshow e ntfSubscriber :: NtfSubscriber -> M () ntfSubscriber NtfSubscriber {smpAgent = ca@SMPClientAgent {agentQ}} = do @@ -639,55 +630,53 @@ logSubStatus srv event n updated = showServer' :: SMPServer -> Text showServer' = decodeLatin1 . strEncode . host -pushNotification :: NtfPushServer -> Maybe T.Text -> OwnServer -> NtfTknRec -> PushNotification -> M () -pushNotification s srvHost_ isOwn tkn@NtfTknRec {token = DeviceToken pp _} ntf = do - q <- getOrCreatePushWorker s (srvHost_, pp) isOwn +pushNotification :: NtfPostgresStore -> NtfServerStats -> NtfPushServer -> Maybe T.Text -> OwnServer -> NtfTknRec -> PushNotification -> IO () +pushNotification st stats s srvHost_ isOwn tkn@NtfTknRec {token = DeviceToken pp _} ntf = do + q <- getOrCreatePushWorker st stats s (srvHost_, pp) isOwn atomically $ writeTBQueue q (tkn, ntf) -getOrCreatePushWorker :: NtfPushServer -> (Maybe T.Text, PushProvider) -> OwnServer -> M (TBQueue (NtfTknRec, PushNotification)) -getOrCreatePushWorker s@NtfPushServer {pushWorkers, pushWorkerSeq, pushQSize} key@(srvHost_, _) isOwn = do - ts <- liftIO getCurrentTime +getOrCreatePushWorker :: NtfPostgresStore -> NtfServerStats -> NtfPushServer -> (Maybe T.Text, PushProvider) -> OwnServer -> IO (TBQueue (NtfTknRec, PushNotification)) +getOrCreatePushWorker st stats s@NtfPushServer {pushWorkers, pushWorkerSeq, pushQSize} key@(srvHost_, _) isOwn = do + ts <- getCurrentTime atomically (getSessVar pushWorkerSeq key pushWorkers ts) >>= \case Left v -> do - q <- liftIO $ newTBQueueIO pushQSize - tId <- mkWeakThreadId =<< forkIO (runPushWorker s srvHost_ isOwn q) + q <- newTBQueueIO pushQSize + tId <- mkWeakThreadId =<< forkIO (runPushWorker st stats s srvHost_ isOwn q) atomically $ putTMVar (sessionVar v) PushWorker {workerQ = q, workerThreadId = tId} pure q Right v -> workerQ <$> atomically (readTMVar $ sessionVar v) -runPushWorker :: NtfPushServer -> Maybe T.Text -> OwnServer -> TBQueue (NtfTknRec, PushNotification) -> M () -runPushWorker s srvHost_ isOwn q = forever $ do +runPushWorker :: NtfPostgresStore -> NtfServerStats -> NtfPushServer -> Maybe T.Text -> OwnServer -> TBQueue (NtfTknRec, PushNotification) -> IO () +runPushWorker st stats s srvHost_ isOwn q = forever $ do (tkn@NtfTknRec {ntfTknId, token = t@(DeviceToken pp _), tknStatus}, ntf) <- atomically (readTBQueue q) - liftIO $ logDebug $ "sending push notification to " <> T.pack (show pp) - st <- asks store + logDebug $ "sending push notification to " <> T.pack (show pp) case ntf of PNVerification _ -> - liftIO (deliverNotification st pp tkn ntf) >>= \case + deliverNotification st pp tkn ntf >>= \case Right _ -> do - void $ liftIO $ setTknStatusConfirmed st tkn - incNtfStatT t ntfVrfDelivered - Left _ -> incNtfStatT t ntfVrfFailed + void $ setTknStatusConfirmed st tkn + incNtfStatT_ stats t ntfVrfDelivered + Left _ -> incNtfStatT_ stats t ntfVrfFailed PNCheckMessages -> - liftIO (deliverNotification st pp tkn ntf) >>= \case + deliverNotification st pp tkn ntf >>= \case Right _ -> do - void $ liftIO $ updateTokenCronSentAt st ntfTknId . systemSeconds =<< getSystemTime - incNtfStatT t ntfCronDelivered - Left _ -> incNtfStatT t ntfCronFailed + void $ updateTokenCronSentAt st ntfTknId . systemSeconds =<< getSystemTime + incNtfStatT_ stats t ntfCronDelivered + Left _ -> incNtfStatT_ stats t ntfCronFailed PNMessage {} -> checkActiveTkn tknStatus $ do - stats <- asks serverStats - liftIO $ updatePeriodStats (activeTokens stats) ntfTknId - liftIO (deliverNotification st pp tkn ntf) >>= \case + updatePeriodStats (activeTokens stats) ntfTknId + deliverNotification st pp tkn ntf >>= \case Left _ -> do - incNtfStatT t ntfFailed - when isOwn $ liftIO $ mapM_ (`incServerStat` ntfFailedOwn stats) srvHost_ + incNtfStatT_ stats t ntfFailed + when isOwn $ mapM_ (`incServerStat` ntfFailedOwn stats) srvHost_ Right () -> do - incNtfStatT t ntfDelivered - when isOwn $ liftIO $ mapM_ (`incServerStat` ntfDeliveredOwn stats) srvHost_ + incNtfStatT_ stats t ntfDelivered + when isOwn $ mapM_ (`incServerStat` ntfDeliveredOwn stats) srvHost_ where - checkActiveTkn :: NtfTknStatus -> M () -> M () + checkActiveTkn :: NtfTknStatus -> IO () -> IO () checkActiveTkn status action | status == NTActive = action - | otherwise = liftIO $ logError "bad notification token status" + | otherwise = logError "bad notification token status" deliverNotification :: NtfPostgresStore -> PushProvider -> NtfTknRec -> PushNotification -> IO (Either PushProviderError ()) deliverNotification st pp tkn@NtfTknRec {ntfTknId} ntf' = do (deliver, clientVar) <- getPushClient s pp @@ -730,13 +719,13 @@ pushWorkersQLength workers = do periodicNtfsThread :: NtfPushServer -> M () periodicNtfsThread s = do st <- asks store + stats <- asks serverStats ntfsInterval <- asks $ periodicNtfsInterval . config let interval = 1000000 * ntfsInterval - UnliftIO unlift <- askUnliftIO liftIO $ forever $ do threadDelay interval now <- systemSeconds <$> getSystemTime - cnt <- withPeriodicNtfTokens st now $ \tkn -> unlift $ pushNotification s Nothing False tkn PNCheckMessages + cnt <- withPeriodicNtfTokens st now $ \tkn -> pushNotification st stats s Nothing False tkn PNCheckMessages logNote $ "Scheduled periodic notifications: " <> tshow cnt runNtfClientTransport :: Transport c => THandleNTF c 'TServer -> M () @@ -826,14 +815,16 @@ verifyNtfTransmission st thAuth (tAuth, authorized, (corrId, entId, cmd)) = case e -> VRFailed e client :: NtfServerClient -> NtfSubscriber -> NtfPushServer -> M () -client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} ps = +client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} ps = do + st <- asks store + stats <- asks serverStats forever $ atomically (readTBQueue rcvQ) - >>= mapM processCommand + >>= mapM (processCommand st stats) >>= atomically . writeTBQueue sndQ where - processCommand :: NtfRequest -> M (Transmission NtfResponse) - processCommand = \case + processCommand :: NtfPostgresStore -> NtfServerStats -> NtfRequest -> M (Transmission NtfResponse) + processCommand st stats = \case NtfReqNew corrId (ANE SToken newTkn@(NewNtfTkn token _ dhPubKey)) -> (corrId,NoEntity,) <$> do logDebug "TNEW - new token" (srvDhPubKey, srvDhPrivKey) <- atomically . C.generateKeyPair =<< asks random @@ -843,7 +834,7 @@ client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} ps = ts <- liftIO $ getSystemDate let tkn = mkNtfTknRec tknId newTkn srvDhPrivKey dhSecret regCode ts withNtfStore (`addNtfToken` tkn) $ \_ -> do - pushNotification ps Nothing False tkn $ PNVerification regCode + liftIO $ pushNotification st stats ps Nothing False tkn $ PNVerification regCode incNtfStatT token ntfVrfQueued incNtfStatT token tknCreated pure $ NRTknId tknId srvDhPubKey @@ -859,7 +850,7 @@ client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} ps = | otherwise -> withNtfStore (\st -> updateTknStatus st tkn NTRegistered) $ \_ -> sendVerification where sendVerification = do - pushNotification ps Nothing False tkn $ PNVerification tknRegCode + liftIO $ pushNotification st stats ps Nothing False tkn $ PNVerification tknRegCode incNtfStatT token ntfVrfQueued pure $ NRTknId ntfTknId $ C.publicKey tknDhPrivKey TVFY code -- this allows repeated verification for cases when client connection dropped before server response @@ -877,7 +868,7 @@ client NtfServerClient {rcvQ, sndQ} ns@NtfSubscriber {smpAgent = ca} ps = regCode <- getRegCode let tkn' = tkn {token = token', tknStatus = NTRegistered, tknRegCode = regCode} withNtfStore (`replaceNtfToken` tkn') $ \_ -> do - pushNotification ps Nothing False tkn' $ PNVerification regCode + liftIO $ pushNotification st stats ps Nothing False tkn' $ PNVerification regCode incNtfStatT token ntfVrfQueued incNtfStatT token tknReplaced pure NROk @@ -949,6 +940,11 @@ incNtfStatT (DeviceToken PPApnsNull _) _ = pure () incNtfStatT _ statSel = incNtfStat statSel {-# INLINE incNtfStatT #-} +incNtfStatT_ :: NtfServerStats -> DeviceToken -> (NtfServerStats -> IORef Int) -> IO () +incNtfStatT_ _ (DeviceToken PPApnsNull _) _ = pure () +incNtfStatT_ stats _ statSel = incNtfStat_ stats statSel +{-# INLINE incNtfStatT_ #-} + incNtfStat :: (NtfServerStats -> IORef Int) -> M () incNtfStat statSel = asks serverStats >>= liftIO . (`incNtfStat_` statSel) {-# INLINE incNtfStat #-} diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 54349f600..7ece78609 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/src/Simplex/Messaging/Notifications/Server/Env.hs @@ -28,7 +28,6 @@ module Simplex.Messaging.Notifications.Server.Env ) where import Control.Concurrent (ThreadId) -import Data.IORef import qualified Control.Exception as E import Control.Logger.Simple import Control.Monad @@ -120,20 +119,18 @@ data NtfEnv = NtfEnv serverStats :: NtfServerStats } -newNtfServerEnv :: NtfServerConfig -> (IORef NtfEnv -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> IO NtfEnv +newNtfServerEnv :: NtfServerConfig -> (NtfPostgresStore -> NtfPushServer -> NtfServerStats -> SMPClientAgent 'NotifierService -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> IO NtfEnv newNtfServerEnv config@NtfServerConfig {pushQSize, smpAgentCfg, apnsConfig, dbStoreConfig, ntfCredentials, useServiceCreds} mkProcessMsg = do random <- C.newRandom store <- newNtfDbStore dbStoreConfig tlsServerCreds <- loadServerCredential ntfCredentials XV.Fingerprint fp <- loadFingerprint ntfCredentials - let dbService = if useServiceCreds then Just $ mkDbService random store else Nothing - envRef <- newIORef $ error "NtfEnv not initialized" - subscriber <- newNtfSubscriber smpAgentCfg (mkProcessMsg envRef) dbService random pushServer <- newNtfPushServer pushQSize apnsConfig serverStats <- newNtfServerStats =<< getCurrentTime - let env = NtfEnv {config, subscriber, pushServer, store, random, tlsServerCreds, serverIdentity = C.KeyHash fp, serverStats} - writeIORef envRef env - pure env + let dbService = if useServiceCreds then Just $ mkDbService random store else Nothing + processMsg = mkProcessMsg store pushServer serverStats + subscriber <- newNtfSubscriber smpAgentCfg processMsg dbService random + pure NtfEnv {config, subscriber, pushServer, store, random, tlsServerCreds, serverIdentity = C.KeyHash fp, serverStats} where mkDbService g st = DBService {getCredentials, updateServiceId} where @@ -162,7 +159,7 @@ data NtfSubscriber = NtfSubscriber type SMPSubscriberVar = SessionVar SMPSubscriber -newNtfSubscriber :: SMPClientAgentConfig -> (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Maybe DBService -> TVar ChaChaDRG -> IO NtfSubscriber +newNtfSubscriber :: SMPClientAgentConfig -> (SMPClientAgent 'NotifierService -> ServerTransmissionBatch SMPVersion ErrorType BrokerMsg -> IO ()) -> Maybe DBService -> TVar ChaChaDRG -> IO NtfSubscriber newNtfSubscriber smpAgentCfg processMsg dbService random = do smpSubscribers <- TM.emptyIO subscriberSeq <- newTVarIO 0 diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index 3879fd8ee..123ccd545 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/src/Simplex/Messaging/Server/Env/STM.hs @@ -706,7 +706,7 @@ mkJournalStoreConfig queueStoreCfg storePath msgQueueQuota maxJournalMsgCount ma newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO ProxyAgent newSMPProxyAgent smpAgentCfg random = do - smpAgent <- newSMPClientAgent SSender smpAgentCfg (\_ -> pure ()) Nothing random + smpAgent <- newSMPClientAgent SSender smpAgentCfg (\_ _ -> pure ()) Nothing random pure ProxyAgent {smpAgent} readWriteQueueStore :: forall q. StoreQueueClass q => Bool -> (RecipientId -> QueueRec -> IO q) -> FilePath -> STMQueueStore q -> IO (StoreLog 'WriteMode)