refactor

2026-07-09 03:42:26 +00:00 · 2026-05-25 13:26:29 +00:00
parent 3042be3168
commit 151feece26
6 changed files with 90 additions and 101 deletions
@@ -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
@@ -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
@@ -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
@@ -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 #-}
@@ -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
@@ -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)