6.0.0.4

smp server: fix server-info, additional stats, allow expiring inactive clients which have prohibit subscriptions only (iOS NSE clients) (#1237 )
* smp server: fix server-info * fix * faster saving messages * remove comment * move ProhibitSub out of TVar * subscription stats * stabilize test * order Co-authored-by: spaced4ndy <[email protected]> * more notification stats * count ntf stats * update server-info --------- Co-authored-by: spaced4ndy <[email protected]>
2026-10-12 11:39:56 +00:00 · 2024-07-28 15:03:32 +01:00 · 2024-07-25 16:06:40 +01:00 · 2024-07-25 13:07:28 +01:00 · 2024-07-23 08:27:54 +01:00 · 2024-07-22 15:43:38 +01:00
26 changed files with 703 additions and 358 deletions
@@ -1,5 +1,5 @@
 name: simplexmq
-version: 6.0.0.3
+version: 6.0.0.4
 synopsis: SimpleXMQ message broker
 description: |
  This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack

 name:           simplexmq
-version:        6.0.0.3
+version:        6.0.0.4
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                <./docs/Simplex-Messaging-Client.html client> and
@@ -43,7 +43,7 @@ import Data.Either (partitionEithers, rights)
 import Data.Int (Int64)
 import Data.List (foldl', partition, sortOn)
 import qualified Data.List.NonEmpty as L
-import Data.Map (Map)
+import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Maybe (mapMaybe)
 import qualified Data.Set as S
@@ -43,7 +43,7 @@ import Data.Int (Int64)
 import Data.List (foldl', sortOn)
 import Data.List.NonEmpty (NonEmpty (..), nonEmpty)
 import qualified Data.List.NonEmpty as L
-import Data.Map (Map)
+import Data.Map.Strict (Map)
 import qualified Data.Map as M
 import Data.Maybe (fromMaybe, listToMaybe)
 import qualified Data.Text as T
@@ -52,7 +52,7 @@ import Data.Int (Int64)
 import Data.List (foldl', sortOn)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
-import Data.Map (Map)
+import Data.Map.Strict (Map)
 import qualified Data.Map as M
 import Data.Maybe (fromMaybe)
 import Data.String
@@ -106,6 +106,7 @@ module Simplex.Messaging.Agent
    rcConnectHost,
    rcConnectCtrl,
    rcDiscoverCtrl,
+    getAgentSubsTotal,
    getAgentServersSummary,
    resetAgentServersStats,
    foregroundAgent,
@@ -237,25 +238,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 +376,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 +1033,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 +1041,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)
@@ -2214,7 +2202,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
    processSubOk :: RcvQueue -> TVar [ConnId] -> AM ()
    processSubOk rq@RcvQueue {connId} upConnIds =
      atomically . whenM (isPendingSub connId) $ do
-        addSubscription c rq
+        addSubscription c sessId rq
        modifyTVar' upConnIds (connId :)
    processSubErr :: RcvQueue -> SMPClientError -> AM ()
    processSubErr rq@RcvQueue {connId} e = do
@@ -93,6 +93,7 @@ module Simplex.Messaging.Agent.Client
    AgentServersSummary (..),
    ServerSessions (..),
    SMPServerSubs (..),
+    getAgentSubsTotal,
    getAgentServersSummary,
    getAgentSubscriptions,
    slowNetworkConfig,
@@ -145,6 +146,7 @@ module Simplex.Messaging.Agent.Client
    incXFTPServerStat,
    incXFTPServerStat',
    incXFTPServerSizeStat,
+    incNtfServerStat,
    AgentWorkersDetails (..),
    getAgentWorkersDetails,
    AgentWorkersSummary (..),
@@ -304,8 +306,8 @@ data AgentClient = AgentClient
    userNetworkInfo :: TVar UserNetworkInfo,
    userNetworkUpdated :: TVar (Maybe UTCTime),
    subscrConns :: TVar (Set ConnId),
-    activeSubs :: TRcvQueues,
-    pendingSubs :: TRcvQueues,
+    activeSubs :: TRcvQueues (SessionId, RcvQueue),
+    pendingSubs :: TRcvQueues RcvQueue,
    removedSubs :: TMap (UserId, SMPServer, SMP.RecipientId) SMPClientError,
    workerSeq :: TVar Int,
    smpDeliveryWorkers :: TMap SndQAddr (Worker, TMVar ()),
@@ -330,6 +332,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 +491,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 +532,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a
        agentEnv,
        smpServersStats,
        xftpServersStats,
+        ntfServersStats,
        srvStatsStartedAt
      }

@@ -672,11 +677,13 @@ smpClientDisconnected c@AgentClient {active, smpClients, smpProxiedRelays} tSess
    -- because we can have a race condition when a new current client could have already
    -- made subscriptions active, and the old client would be processing diconnection later.
    removeClientAndSubs :: IO ([RcvQueue], [ConnId])
-    removeClientAndSubs = atomically $ ifM currentActiveClient removeSubs $ pure ([], [])
+    removeClientAndSubs = atomically $ do
+      removeSessVar v tSess smpClients
+      ifM (readTVar active) removeSubs (pure ([], []))
      where
-        currentActiveClient = (&&) <$> removeSessVar' v tSess smpClients <*> readTVar active
+        sessId = sessionId $ thParams client
        removeSubs = do
-          (qs, cs) <- RQ.getDelSessQueues tSess $ activeSubs c
+          (qs, cs) <- RQ.getDelSessQueues tSess sessId $ activeSubs c
          RQ.batchAddQueues (pendingSubs c) qs
          -- this removes proxied relays that this client created sessions to
          destSrvs <- M.keys <$> readTVar prs
@@ -1342,8 +1349,8 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode sender
      qUri = SMPQueueUri vRange $ SMPQueueAddress srv sndId e2eDhKey sndSecure
  pure (rq, qUri, tSess, sessId)

-processSubResult :: AgentClient -> RcvQueue -> Either SMPClientError () -> STM ()
-processSubResult c rq@RcvQueue {userId, server, connId} = \case
+processSubResult :: AgentClient -> SessionId -> RcvQueue -> Either SMPClientError () -> STM ()
+processSubResult c sessId rq@RcvQueue {userId, server, connId} = \case
  Left e ->
    unless (temporaryClientError e) $ do
      incSMPServerStat c userId server connSubErrs
@@ -1351,7 +1358,7 @@ processSubResult c rq@RcvQueue {userId, server, connId} = \case
  Right () ->
    ifM
      (hasPendingSubscription c connId)
-      (incSMPServerStat c userId server connSubscribed >> addSubscription c rq)
+      (incSMPServerStat c userId server connSubscribed >> addSubscription c sessId rq)
      (incSMPServerStat c userId server connSubIgnored)

 temporaryAgentError :: AgentErrorType -> Bool
@@ -1422,7 +1429,7 @@ subscribeQueues c qs = do
        sessId = sessionId $ thParams smp
        hasTempErrors = any (either temporaryClientError (const False) . snd)
        processSubResults :: NonEmpty (RcvQueue, Either SMPClientError ()) -> STM ()
-        processSubResults = mapM_ $ uncurry $ processSubResult c
+        processSubResults = mapM_ $ uncurry $ processSubResult c sessId
        resubscribe = resubscribeSMPSession c tSess `runReaderT` env

 activeClientSession :: AgentClient -> SMPTransportSession -> SessionId -> STM Bool
@@ -1461,10 +1468,10 @@ sendBatch smpCmdFunc smp qs = L.zip qs <$> smpCmdFunc smp (L.map queueCreds qs)
  where
    queueCreds RcvQueue {rcvPrivateKey, rcvId} = (rcvPrivateKey, rcvId)

-addSubscription :: AgentClient -> RcvQueue -> STM ()
-addSubscription c rq@RcvQueue {connId} = do
+addSubscription :: AgentClient -> SessionId -> RcvQueue -> STM ()
+addSubscription c sessId rq@RcvQueue {connId} = do
  modifyTVar' (subscrConns c) $ S.insert connId
-  RQ.addQueue rq $ activeSubs c
+  RQ.addQueue (sessId, rq) $ activeSubs c
  RQ.deleteQueue rq $ pendingSubs c

 failSubscription :: AgentClient -> RcvQueue -> SMPClientError -> STM ()
@@ -1483,7 +1490,7 @@ addNewQueueSubscription c rq tSess sessId = do
    atomically $
      ifM
        (activeClientSession c tSess sessId)
-        (True <$ addSubscription c rq)
+        (True <$ addSubscription c sessId rq)
        (False <$ addPendingSubscription c rq)
  unless same $ resubscribeSMPSession c tSess

@@ -1954,13 +1961,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 +1976,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)

@@ -2009,10 +2020,30 @@ data ServerSessions = ServerSessions
  }
  deriving (Show)

+getAgentSubsTotal :: AgentClient -> [UserId] -> IO (SMPServerSubs, Bool)
+getAgentSubsTotal c userIds = do
+  ssActive <- getSubsCount activeSubs
+  ssPending <- getSubsCount pendingSubs
+  sess <- hasSession . M.toList =<< readTVarIO (smpClients c)
+  pure (SMPServerSubs {ssActive, ssPending}, sess)
+  where
+    getSubsCount :: (AgentClient -> TRcvQueues q) -> IO Int
+    getSubsCount subs = M.foldrWithKey' addSub 0 <$> readTVarIO (getRcvQueues $ subs c)
+    addSub :: (UserId, SMPServer, SMP.RecipientId) -> q -> Int -> Int
+    addSub (userId, _, _) _ cnt = if userId `elem` userIds then cnt + 1 else cnt
+    hasSession :: [(SMPTransportSession, SMPClientVar)] -> IO Bool
+    hasSession = \case
+      [] -> pure False
+      (s : ss) -> ifM (isConnected s) (pure True) (hasSession ss)
+    isConnected ((userId, _, _), SessionVar {sessionVar})
+      | userId `elem` userIds = atomically $ maybe False isRight <$> tryReadTMVar sessionVar
+      | otherwise = pure False
+
 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 +2051,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
@@ -2076,6 +2110,7 @@ getAgentSubscriptions c = do
  removedSubscriptions <- getRemovedSubs
  pure $ SubscriptionsInfo {activeSubscriptions, pendingSubscriptions, removedSubscriptions}
  where
+    getSubs :: (AgentClient -> TRcvQueues q) -> IO [SubInfo]
    getSubs sel = map (`subInfo` Nothing) . M.keys <$> readTVarIO (getRcvQueues $ sel c)
    getRemovedSubs = map (uncurry subInfo . second Just) . M.assocs <$> readTVarIO (removedSubs c)
    subInfo :: (UserId, SMPServer, SMP.RecipientId) -> Maybe SMPClientError -> SubInfo
@@ -51,7 +51,7 @@ import Data.ByteArray (ScrubbedBytes)
 import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
-import Data.Map (Map)
+import Data.Map.Strict (Map)
 import Data.Maybe (fromMaybe)
 import Data.Time.Clock (NominalDiffTime, nominalDay)
 import Data.Time.Clock.System (SystemTime (..))
@@ -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
@@ -155,8 +155,8 @@ import Data.Int (Int64)
 import Data.Kind (Type)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
-import Data.Map (Map)
-import qualified Data.Map as M
+import Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
 import Data.Maybe (fromMaybe, isJust)
 import Data.Text (Text)
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
@@ -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} =
@@ -1,7 +1,9 @@
+{-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE LambdaCase #-}

 module Simplex.Messaging.Agent.TRcvQueues
  ( TRcvQueues (getRcvQueues, getConnections),
+    Queue (..),
    empty,
    clear,
    deleteConn,
@@ -11,7 +13,6 @@ module Simplex.Messaging.Agent.TRcvQueues
    deleteQueue,
    getSessQueues,
    getDelSessQueues,
-    qKey,
  )
 where

@@ -25,46 +26,51 @@ import Simplex.Messaging.Agent.Store (RcvQueue, StoredRcvQueue (..))
 import Simplex.Messaging.Protocol (RecipientId, SMPServer)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Transport
+
+class Queue q where
+  connId' :: q -> ConnId
+  qKey :: q -> (UserId, SMPServer, ConnId)

 -- the fields in this record have the same data with swapped keys for lookup efficiency,
 -- and all methods must maintain this invariant.
-data TRcvQueues = TRcvQueues
-  { getRcvQueues :: TMap (UserId, SMPServer, RecipientId) RcvQueue,
+data TRcvQueues q = TRcvQueues
+  { getRcvQueues :: TMap (UserId, SMPServer, RecipientId) q,
    getConnections :: TMap ConnId (NonEmpty (UserId, SMPServer, RecipientId))
  }

-empty :: STM TRcvQueues
+empty :: STM (TRcvQueues q)
 empty = TRcvQueues <$> TM.empty <*> TM.empty

-clear :: TRcvQueues -> STM ()
+clear :: TRcvQueues q -> STM ()
 clear (TRcvQueues qs cs) = TM.clear qs >> TM.clear cs

-deleteConn :: ConnId -> TRcvQueues -> STM ()
+deleteConn :: ConnId -> TRcvQueues q -> STM ()
 deleteConn cId (TRcvQueues qs cs) =
  TM.lookupDelete cId cs >>= \case
    Just ks -> modifyTVar' qs $ \qs' -> foldl' (flip M.delete) qs' ks
    Nothing -> pure ()

-hasConn :: ConnId -> TRcvQueues -> STM Bool
+hasConn :: ConnId -> TRcvQueues q -> STM Bool
 hasConn cId (TRcvQueues _ cs) = TM.member cId cs

-addQueue :: RcvQueue -> TRcvQueues -> STM ()
+addQueue :: Queue q => q -> TRcvQueues q -> STM ()
 addQueue rq (TRcvQueues qs cs) = do
  TM.insert k rq qs
-  TM.alter addQ (connId rq) cs
+  TM.alter addQ (connId' rq) cs
  where
    addQ = Just . maybe (k :| []) (k <|)
    k = qKey rq

 -- Save time by aggregating modifyTVar
-batchAddQueues :: Foldable t => TRcvQueues -> t RcvQueue -> STM ()
+batchAddQueues :: (Foldable t, Queue q) => TRcvQueues q -> t q -> STM ()
 batchAddQueues (TRcvQueues qs cs) rqs = do
  modifyTVar' qs $ \now -> foldl' (\rqs' rq -> M.insert (qKey rq) rq rqs') now rqs
-  modifyTVar' cs $ \now -> foldl' (\cs' rq -> M.alter (addQ $ qKey rq) (connId rq) cs') now rqs
+  modifyTVar' cs $ \now -> foldl' (\cs' rq -> M.alter (addQ $ qKey rq) (connId' rq) cs') now rqs
  where
    addQ k = Just . maybe (k :| []) (k <|)

-deleteQueue :: RcvQueue -> TRcvQueues -> STM ()
+deleteQueue :: RcvQueue -> TRcvQueues RcvQueue -> STM ()
 deleteQueue rq (TRcvQueues qs cs) = do
  TM.delete k qs
  TM.update delQ (connId rq) cs
@@ -72,21 +78,22 @@ deleteQueue rq (TRcvQueues qs cs) = do
    delQ = L.nonEmpty . L.filter (/= k)
    k = qKey rq

-getSessQueues :: (UserId, SMPServer, Maybe ConnId) -> TRcvQueues -> STM [RcvQueue]
+getSessQueues :: (UserId, SMPServer, Maybe ConnId) -> TRcvQueues RcvQueue -> STM [RcvQueue]
 getSessQueues tSess (TRcvQueues qs _) = M.foldl' addQ [] <$> readTVar qs
  where
    addQ qs' rq = if rq `isSession` tSess then rq : qs' else qs'

-getDelSessQueues :: (UserId, SMPServer, Maybe ConnId) -> TRcvQueues -> STM ([RcvQueue], [ConnId])
-getDelSessQueues tSess (TRcvQueues qs cs) = do
+getDelSessQueues :: (UserId, SMPServer, Maybe ConnId) -> SessionId -> TRcvQueues (SessionId, RcvQueue) -> STM ([RcvQueue], [ConnId])
+getDelSessQueues tSess sessId' (TRcvQueues qs cs) = do
  (removedQs, qs'') <- (\qs' -> M.foldl' delQ ([], qs') qs') <$> readTVar qs
  writeTVar qs $! qs''
  removedConns <- stateTVar cs $ \cs' -> foldl' delConn ([], cs') removedQs
  pure (removedQs, removedConns)
  where
-    delQ acc@(removed, qs') rq
-      | rq `isSession` tSess = (rq : removed, M.delete (qKey rq) qs')
+    delQ acc@(removed, qs') (sessId, rq)
+      | rq `isSession` tSess && sessId == sessId' = (rq : removed, M.delete (qKey rq) qs')
      | otherwise = acc
+    delConn :: ([ConnId], M.Map ConnId (NonEmpty (UserId, SMPServer, ConnId))) -> RcvQueue -> ([ConnId], M.Map ConnId (NonEmpty (UserId, SMPServer, ConnId)))
    delConn (removed, cs') rq = M.alterF f cId cs'
      where
        cId = connId rq
@@ -100,5 +107,10 @@ isSession :: RcvQueue -> (UserId, SMPServer, Maybe ConnId) -> Bool
 isSession rq (uId, srv, connId_) =
  userId rq == uId && server rq == srv && maybe True (connId rq ==) connId_

-qKey :: RcvQueue -> (UserId, SMPServer, ConnId)
-qKey rq = (userId rq, server rq, connId rq)
+instance Queue RcvQueue where
+  connId' = connId
+  qKey rq = (userId rq, server rq, connId rq)
+
+instance Queue (SessionId, RcvQueue) where
+  connId' = connId . snd
+  qKey = qKey . snd
@@ -100,7 +100,7 @@ data SMPClientAgent = SMPClientAgent
    randomDrg :: TVar ChaChaDRG,
    smpClients :: TMap SMPServer SMPClientVar,
    smpSessions :: TMap SessionId (OwnServer, SMPClient),
-    srvSubs :: TMap SMPServer (TMap SMPSub C.APrivateAuthKey),
+    srvSubs :: TMap SMPServer (TMap SMPSub (SessionId, C.APrivateAuthKey)),
    pendingSrvSubs :: TMap SMPServer (TMap SMPSub C.APrivateAuthKey),
    smpSubWorkers :: TMap SMPServer (SessionVar (Async ())),
    workerSeq :: TVar Int
@@ -204,14 +204,17 @@ connectClient ca@SMPClientAgent {agentCfg, smpClients, smpSessions, msgQ, random

    removeClientAndSubs :: SMPClient -> IO (Maybe (Map SMPSub C.APrivateAuthKey))
    removeClientAndSubs smp = atomically $ do
+      TM.delete sessId smpSessions
      removeSessVar v srv smpClients
-      TM.delete (sessionId $ thParams smp) smpSessions
-      TM.lookupDelete srv (srvSubs ca) >>= mapM updateSubs
+      TM.lookup srv (srvSubs ca) >>= mapM updateSubs
      where
+        sessId = sessionId $ thParams smp
        updateSubs sVar = do
-          ss <- readTVar sVar
-          addSubs_ (pendingSrvSubs ca) srv ss
-          pure ss
+          -- removing subscriptions that have matching sessionId to disconnected client
+          -- and keep the other ones (they can be made by the new client)
+          pending <- M.map snd <$> stateTVar sVar (M.partition ((sessId ==) . fst))
+          addSubs_ (pendingSrvSubs ca) srv pending
+          pure pending

    serverDown :: Map SMPSub C.APrivateAuthKey -> IO ()
    serverDown ss = unless (M.null ss) $ do
@@ -256,9 +259,9 @@ reconnectSMPClient ca@SMPClientAgent {agentCfg} srv cs =
    subscribe_ smp SPNotifier nSubs
    subscribe_ smp SPRecipient rSubs
  where
-    groupSub :: Map SMPSub C.APrivateAuthKey -> (SMPSub, C.APrivateAuthKey) -> ([(QueueId, C.APrivateAuthKey)], [(QueueId, C.APrivateAuthKey)]) -> ([(QueueId, C.APrivateAuthKey)], [(QueueId, C.APrivateAuthKey)])
-    groupSub currSubs (s@(party, qId), k) (nSubs, rSubs)
-      | M.member s currSubs = (nSubs, rSubs)
+    groupSub :: Map SMPSub (SessionId, C.APrivateAuthKey) -> (SMPSub, C.APrivateAuthKey) -> ([(QueueId, C.APrivateAuthKey)], [(QueueId, C.APrivateAuthKey)]) -> ([(QueueId, C.APrivateAuthKey)], [(QueueId, C.APrivateAuthKey)])
+    groupSub currSubs (s@(party, qId), k) acc@(nSubs, rSubs)
+      | M.member s currSubs = acc
      | otherwise = case party of
          SPNotifier -> (s' : nSubs, rSubs)
          SPRecipient -> (nSubs, s' : rSubs)
@@ -346,17 +349,18 @@ smpSubscribeQueues party ca smp srv subs = do
      when tempErrs $ reconnectClient ca srv
    Nothing -> reconnectClient ca srv
  where
-    processSubscriptions :: NonEmpty (Either SMPClientError ()) -> STM (Bool, [(QueueId, SMPClientError)], [(QueueId, C.APrivateAuthKey)], [QueueId])
+    processSubscriptions :: NonEmpty (Either SMPClientError ()) -> STM (Bool, [(QueueId, SMPClientError)], [(QueueId, (SessionId, C.APrivateAuthKey))], [QueueId])
    processSubscriptions rs = do
      pending <- maybe (pure M.empty) readTVar =<< TM.lookup srv (pendingSrvSubs ca)
      let acc@(_, _, oks, notPending) = foldr (groupSub pending) (False, [], [], []) (L.zip subs rs)
      unless (null oks) $ addSubscriptions ca srv party oks
      unless (null notPending) $ removePendingSubs ca srv party notPending
      pure acc
-    groupSub :: Map SMPSub C.APrivateAuthKey -> ((QueueId, C.APrivateAuthKey), Either SMPClientError ()) -> (Bool, [(QueueId, SMPClientError)], [(QueueId, C.APrivateAuthKey)], [QueueId]) -> (Bool, [(QueueId, SMPClientError)], [(QueueId, C.APrivateAuthKey)], [QueueId])
-    groupSub pending (s@(qId, _), r) acc@(!tempErrs, finalErrs, oks, notPending) = case r of
+    sessId = sessionId $ thParams smp
+    groupSub :: Map SMPSub C.APrivateAuthKey -> ((QueueId, C.APrivateAuthKey), Either SMPClientError ()) -> (Bool, [(QueueId, SMPClientError)], [(QueueId, (SessionId, C.APrivateAuthKey))], [QueueId]) -> (Bool, [(QueueId, SMPClientError)], [(QueueId, (SessionId, C.APrivateAuthKey))], [QueueId])
+    groupSub pending ((qId, pk), r) acc@(!tempErrs, finalErrs, oks, notPending) = case r of
      Right ()
-        | M.member (party, qId) pending -> (tempErrs, finalErrs, s : oks, qId : notPending)
+        | M.member (party, qId) pending -> (tempErrs, finalErrs, (qId, (sessId, pk)) : oks, qId : notPending)
        | otherwise -> acc
      Left e
        | temporaryClientError e -> (True, finalErrs, oks, notPending)
@@ -379,7 +383,7 @@ showServer :: SMPServer -> ByteString
 showServer ProtocolServer {host, port} =
  strEncode host <> B.pack (if null port then "" else ':' : port)

-addSubscriptions :: SMPClientAgent -> SMPServer -> SMPSubParty -> [(QueueId, C.APrivateAuthKey)] -> STM ()
+addSubscriptions :: SMPClientAgent -> SMPServer -> SMPSubParty -> [(QueueId, (SessionId, C.APrivateAuthKey))] -> STM ()
 addSubscriptions = addSubsList_ . srvSubs
 {-# INLINE addSubscriptions #-}

@@ -387,12 +391,12 @@ addPendingSubs :: SMPClientAgent -> SMPServer -> SMPSubParty -> [(QueueId, C.APr
 addPendingSubs = addSubsList_ . pendingSrvSubs
 {-# INLINE addPendingSubs #-}

-addSubsList_ :: TMap SMPServer (TMap SMPSub C.APrivateAuthKey) -> SMPServer -> SMPSubParty -> [(QueueId, C.APrivateAuthKey)] -> STM ()
+addSubsList_ :: TMap SMPServer (TMap SMPSub s) -> SMPServer -> SMPSubParty -> [(QueueId, s)] -> STM ()
 addSubsList_ subs srv party ss = addSubs_ subs srv ss'
  where
    ss' = M.fromList $ map (first (party,)) ss

-addSubs_ :: TMap SMPServer (TMap SMPSub C.APrivateAuthKey) -> SMPServer -> Map SMPSub C.APrivateAuthKey -> STM ()
+addSubs_ :: TMap SMPServer (TMap SMPSub s) -> SMPServer -> Map SMPSub s -> STM ()
 addSubs_ subs srv ss =
  TM.lookup srv subs >>= \case
    Just m -> TM.union ss m
@@ -402,7 +406,7 @@ removeSubscription :: SMPClientAgent -> SMPServer -> SMPSub -> STM ()
 removeSubscription = removeSub_ . srvSubs
 {-# INLINE removeSubscription #-}

-removeSub_ :: TMap SMPServer (TMap SMPSub C.APrivateAuthKey) -> SMPServer -> SMPSub -> STM ()
+removeSub_ :: TMap SMPServer (TMap SMPSub s) -> SMPServer -> SMPSub -> STM ()
 removeSub_ subs srv s = TM.lookup srv subs >>= mapM_ (TM.delete s)

 removePendingSubs :: SMPClientAgent -> SMPServer -> SMPSubParty -> [QueueId] -> STM ()
@@ -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 (ConnId, NotificationsMode (..), UserId)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Notifications.Protocol
@@ -48,6 +48,7 @@ data NtfToken = NtfToken
    ntfServer :: NtfServer,
    ntfTokenId :: Maybe NtfTokenId,
    -- TODO combine keys to key pair as the types should match
+
    -- | key used by the ntf server to verify transmissions
    ntfPubKey :: C.APublicAuthKey,
    -- | key used by the ntf client to sign transmissions
@@ -79,17 +80,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 +178,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 +188,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,
@@ -37,6 +37,7 @@ module Simplex.Messaging.Server
  )
 where

+import Control.Concurrent.STM.TQueue (flushTQueue)
 import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
@@ -47,6 +48,7 @@ import Crypto.Random
 import Control.Monad.STM (retry)
 import Data.Bifunctor (first)
 import Data.ByteString.Base64 (encode)
+import qualified Data.ByteString.Builder as BLD
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
@@ -54,6 +56,7 @@ import Data.Either (fromRight, partitionEithers)
 import Data.Functor (($>))
 import Data.Int (Int64)
 import qualified Data.IntMap.Strict as IM
+import qualified Data.IntSet as IS
 import Data.List (intercalate, mapAccumR)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
@@ -69,6 +72,7 @@ import Data.Type.Equality
 import GHC.Stats (getRTSStats)
 import GHC.TypeLits (KnownNat)
 import Network.Socket (ServiceName, Socket, socketToHandle)
+import Numeric.Natural (Natural)
 import Simplex.Messaging.Agent.Lock
 import Simplex.Messaging.Client (ProtocolClient (thParams), ProtocolClientError (..), SMPClient, SMPClientError, forwardSMPTransmission, smpProxyError, temporaryClientError)
 import Simplex.Messaging.Client.Agent (OwnServer, SMPClientAgent (..), SMPClientAgentEvent (..), closeSMPClientAgent, getSMPServerClient'', isOwnServer, lookupSMPServerClient, getConnectedSMPServerClient)
@@ -231,7 +235,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
      initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime
      liftIO $ putStrLn $ "server stats log enabled: " <> statsFilePath
      liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
-      ss@ServerStats {fromTime, qCreated, qSecured, qDeletedAll, qDeletedNew, qDeletedSecured, qSub, qSubNoMsg, qSubAuth, qSubDuplicate, qSubProhibited, msgSent, msgSentAuth, msgSentQuota, msgSentLarge, msgRecv, msgRecvGet, msgGet, msgGetNoMsg, msgGetAuth, msgGetDuplicate, msgGetProhibited, msgExpired, activeQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount, pRelays, pRelaysOwn, pMsgFwds, pMsgFwdsOwn, pMsgFwdsRecv} <- asks serverStats
+      ss@ServerStats {fromTime, qCreated, qSecured, qDeletedAll, qDeletedNew, qDeletedSecured, qSub, qSubNoMsg, qSubAuth, qSubDuplicate, qSubProhibited, ntfCreated, ntfDeleted, ntfSub, ntfSubAuth, ntfSubDuplicate, msgSent, msgSentAuth, msgSentQuota, msgSentLarge, msgRecv, msgRecvGet, msgGet, msgGetNoMsg, msgGetAuth, msgGetDuplicate, msgGetProhibited, msgExpired, activeQueues, subscribedQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount, pRelays, pRelaysOwn, pMsgFwds, pMsgFwdsOwn, pMsgFwdsRecv}
+        <- asks serverStats
+      QueueStore {queues, notifiers} <- asks queueStore
      let interval = 1000000 * logInterval
      forever $ do
        withFile statsFilePath AppendMode $ \h -> liftIO $ do
@@ -248,6 +254,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
          qSubAuth' <- atomically $ swapTVar qSubAuth 0
          qSubDuplicate' <- atomically $ swapTVar qSubDuplicate 0
          qSubProhibited' <- atomically $ swapTVar qSubProhibited 0
+          ntfCreated' <- atomically $ swapTVar ntfCreated 0
+          ntfDeleted' <- atomically $ swapTVar ntfDeleted 0
+          ntfSub' <- atomically $ swapTVar ntfSub 0
+          ntfSubAuth' <- atomically $ swapTVar ntfSubAuth 0
+          ntfSubDuplicate' <- atomically $ swapTVar ntfSubDuplicate 0
          msgSent' <- atomically $ swapTVar msgSent 0
          msgSentAuth' <- atomically $ swapTVar msgSentAuth 0
          msgSentQuota' <- atomically $ swapTVar msgSentQuota 0
@@ -261,6 +272,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
          msgGetProhibited' <- atomically $ swapTVar msgGetProhibited 0
          msgExpired' <- atomically $ swapTVar msgExpired 0
          ps <- atomically $ periodStatCounts activeQueues ts
+          psSub <- atomically $ periodStatCounts subscribedQueues ts
          msgSentNtf' <- atomically $ swapTVar msgSentNtf 0
          msgRecvNtf' <- atomically $ swapTVar msgRecvNtf 0
          psNtf <- atomically $ periodStatCounts activeQueuesNtf ts
@@ -273,6 +285,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
          pMsgFwdsOwn' <- atomically $ getResetProxyStatsData pMsgFwdsOwn
          pMsgFwdsRecv' <- atomically $ swapTVar pMsgFwdsRecv 0
          qCount' <- readTVarIO qCount
+          qCount'' <- M.size <$> readTVarIO queues
+          ntfCount' <- M.size <$> readTVarIO notifiers
          msgCount' <- readTVarIO msgCount
          hPutStrLn h $
            intercalate
@@ -318,7 +332,17 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                       show msgGetNoMsg',
                       show msgGetAuth',
                       show msgGetDuplicate',
-                       show msgGetProhibited'
+                       show msgGetProhibited',
+                       dayCount psSub,
+                       weekCount psSub,
+                       monthCount psSub,
+                       show qCount'',
+                       show ntfCreated',
+                       show ntfDeleted',
+                       show ntfSub',
+                       show ntfSubAuth',
+                       show ntfSubDuplicate',
+                       show ntfCount'
                     ]
              )
        liftIO $ threadDelay' interval
@@ -395,27 +419,33 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                  hPutStrLn h . B.unpack $ B.intercalate "," [bshow cid, encode sessionId, connected', strEncode createdAt, rcvActiveAt', sndActiveAt', bshow age, subscriptions']
              CPStats -> withUserRole $ do
                ss <- unliftIO u $ asks serverStats
-                let putStat :: Show a => ByteString -> (ServerStats -> TVar a) -> IO ()
-                    putStat label var = readTVarIO (var ss) >>= \v -> B.hPutStr h $ label <> ": " <> bshow v <> "\n"
-                    putProxyStat :: ByteString -> (ServerStats -> ProxyStats) -> IO ()
+                let getStat :: (ServerStats -> TVar a) -> IO a
+                    getStat var = readTVarIO (var ss)
+                    putStat :: Show a => String -> (ServerStats -> TVar a) -> IO ()
+                    putStat label var = getStat var >>= \v -> hPutStrLn h $ label <> ": " <> show v
+                    putProxyStat :: String -> (ServerStats -> ProxyStats) -> IO ()
                    putProxyStat label var = do
                      ProxyStatsData {_pRequests, _pSuccesses, _pErrorsConnect, _pErrorsCompat, _pErrorsOther} <- atomically $ getProxyStatsData $ var ss
-                      B.hPutStr h $ label <> ": requests=" <> bshow _pRequests <> ", successes=" <> bshow _pSuccesses <> ", errorsConnect=" <> bshow _pErrorsConnect <> ", errorsCompat=" <> bshow _pErrorsCompat <> ", errorsOther=" <> bshow _pErrorsOther <> "\n"
+                      hPutStrLn h $ label <> ": requests=" <> show _pRequests <> ", successes=" <> show _pSuccesses <> ", errorsConnect=" <> show _pErrorsConnect <> ", errorsCompat=" <> show _pErrorsCompat <> ", errorsOther=" <> show _pErrorsOther
                putStat "fromTime" fromTime
                putStat "qCreated" qCreated
                putStat "qSecured" qSecured
                putStat "qDeletedAll" qDeletedAll
                putStat "qDeletedNew" qDeletedNew
                putStat "qDeletedSecured" qDeletedSecured
-                readTVarIO (day $ activeQueues ss) >>= \v -> B.hPutStr h $ "dayMsgQueues" <> ": " <> bshow (S.size v) <> "\n"
+                getStat (day . activeQueues) >>= \v -> hPutStrLn h $ "daily active queues: " <> show (S.size v)
+                getStat (day . subscribedQueues) >>= \v -> hPutStrLn h $ "daily subscribed queues: " <> show (S.size v)
+                putStat "qSub" qSub
+                putStat "qSubNoMsg" qSubNoMsg
+                subs <- (,,) <$> getStat qSubAuth <*> getStat qSubDuplicate <*> getStat qSubProhibited
+                hPutStrLn h $ "other SUB events (auth, duplicate, prohibited): " <> show subs
                putStat "msgSent" msgSent
                putStat "msgRecv" msgRecv
                putStat "msgRecvGet" msgRecvGet
                putStat "msgGet" msgGet
-                putStat "msgGetNoMsg" msgGetNoMsg
-                putStat "msgGetAuth" msgGetAuth
-                putStat "msgGetDuplicate" msgGetDuplicate
-                putStat "msgGetProhibited" msgGetProhibited
+                putStat "msgGetNoMsg" msgGet
+                gets <- (,,) <$> getStat msgGetAuth <*> getStat msgGetDuplicate <*> getStat msgGetProhibited
+                hPutStrLn h $ "other GET events (auth, duplicate, prohibited): " <> show gets
                putStat "msgSentNtf" msgSentNtf
                putStat "msgRecvNtf" msgRecvNtf
                putStat "qCount" qCount
@@ -474,40 +504,77 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                  activeClients <- readTVarIO clients
                  hPutStrLn h $ "Clients: " <> show (IM.size activeClients)
                  when (r == CPRAdmin) $ do
-                    (smpSubCnt, smpSubCntByGroup, smpClCnt) <- countClientSubs subscriptions countSMPSubs activeClients
-                    (ntfSubCnt, _, ntfClCnt) <- countClientSubs ntfSubscriptions (\_ -> pure (0, 0, 0, 0)) activeClients
-                    hPutStrLn h $ "SMP subscriptions (via clients, slow): " <> show smpSubCnt
+                    clQs <- clientTBQueueLengths activeClients
+                    hPutStrLn h $ "Client queues (rcvQ, sndQ, msgQ): " <> show clQs
+                    (smpSubCnt, smpSubCntByGroup, smpClCnt, smpClQs) <- countClientSubs subscriptions (Just countSMPSubs) activeClients
+                    hPutStrLn h $ "SMP subscriptions (via clients): " <> show smpSubCnt
                    hPutStrLn h $ "SMP subscriptions (by group: NoSub, SubPending, SubThread, ProhibitSub): " <> show smpSubCntByGroup
-                    hPutStrLn h $ "SMP subscribed clients (via clients, slow): " <> show smpClCnt
-                    hPutStrLn h $ "Ntf subscriptions (via clients, slow): " <> show ntfSubCnt
-                    hPutStrLn h $ "Ntf subscribed clients (via clients, slow): " <> show ntfClCnt
-                  activeSubs <- readTVarIO subscribers
-                  activeNtfSubs <- readTVarIO notifiers
-                  hPutStrLn h $ "SMP subscriptions: " <> show (M.size activeSubs)
-                  hPutStrLn h $ "SMP subscribed clients: " <> show (countSubClients activeSubs)
-                  hPutStrLn h $ "Ntf subscriptions: " <> show (M.size activeNtfSubs)
-                  hPutStrLn h $ "Ntf subscribed clients: " <> show (countSubClients activeNtfSubs)
+                    hPutStrLn h $ "SMP subscribed clients (via clients): " <> show smpClCnt
+                    hPutStrLn h $ "SMP subscribed clients queues (via clients, rcvQ, sndQ, msgQ): " <> show smpClQs
+                    (ntfSubCnt, _, ntfClCnt, ntfClQs) <- countClientSubs ntfSubscriptions Nothing activeClients
+                    hPutStrLn h $ "Ntf subscriptions (via clients): " <> show ntfSubCnt
+                    hPutStrLn h $ "Ntf subscribed clients (via clients): " <> show ntfClCnt
+                    hPutStrLn h $ "Ntf subscribed clients queues (via clients, rcvQ, sndQ, msgQ): " <> show ntfClQs
+                  putActiveClientsInfo "SMP" subscribers
+                  putActiveClientsInfo "Ntf" notifiers
                  where
-                    countClientSubs :: (Client -> TMap QueueId a) -> (M.Map QueueId a -> IO (Int, Int, Int, Int)) -> IM.IntMap Client -> IO (Int, (Int, Int, Int, Int), Int)
-                    countClientSubs subSel countSubs = foldM addSubs (0, (0, 0, 0, 0), 0)
+                    putActiveClientsInfo :: String -> TMap QueueId Client -> IO ()
+                    putActiveClientsInfo protoName clients = do
+                      activeSubs <- readTVarIO clients
+                      hPutStrLn h $ protoName <> " subscriptions: " <> show (M.size activeSubs)
+                      clCnt <- if r == CPRAdmin then putClientQueues activeSubs else pure $ countSubClients activeSubs
+                      hPutStrLn h $ protoName <> " subscribed clients: " <> show clCnt
                      where
-                        addSubs :: (Int, (Int, Int, Int, Int), Int) -> Client -> IO (Int, (Int, Int, Int, Int), Int)
-                        addSubs (subCnt, (c1, c2, c3, c4), clCnt) cl = do
+                        putClientQueues :: M.Map QueueId Client -> IO Int
+                        putClientQueues subs = do
+                          let cls = differentClients subs
+                          clQs <- clientTBQueueLengths cls
+                          hPutStrLn h $ protoName <> " subscribed clients queues (rcvQ, sndQ, msgQ): " <> show clQs
+                          pure $ length cls
+                        differentClients :: M.Map QueueId Client -> [Client]
+                        differentClients = fst . M.foldl' addClient ([], IS.empty)
+                          where
+                            addClient acc@(cls, clSet) cl@Client {clientId}
+                              | IS.member clientId clSet = acc
+                              | otherwise = (cl : cls, IS.insert clientId clSet)
+                        countSubClients :: M.Map QueueId Client -> Int
+                        countSubClients = IS.size . M.foldr' (IS.insert . clientId) IS.empty
+                    countClientSubs :: (Client -> TMap QueueId a) -> Maybe (M.Map QueueId a -> IO (Int, Int, Int, Int)) -> IM.IntMap Client -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
+                    countClientSubs subSel countSubs_ = foldM addSubs (0, (0, 0, 0, 0), 0, (0, 0, 0))
+                      where
+                        addSubs :: (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural)) -> Client -> IO (Int, (Int, Int, Int, Int), Int, (Natural, Natural, Natural))
+                        addSubs (!subCnt, cnts@(!c1, !c2, !c3, !c4), !clCnt, !qs) cl = do
                          subs <- readTVarIO $ subSel cl
-                          (c1', c2', c3', c4') <- countSubs subs
+                          cnts' <- case countSubs_ of
+                            Nothing -> pure cnts
+                            Just countSubs -> do
+                              (c1', c2', c3', c4') <- countSubs subs
+                              pure (c1 + c1', c2 + c2', c3 + c3', c4 + c4')
                          let cnt = M.size subs
-                              cnts' = (c1 + c1', c2 + c2', c3 + c3', c4 + c4')
-                          pure (subCnt + cnt, cnts', clCnt + if cnt == 0 then 0 else 1)
+                              clCnt' = if cnt == 0 then clCnt else clCnt + 1
+                          qs' <- if cnt == 0 then pure qs else addQueueLengths qs cl
+                          pure (subCnt + cnt, cnts', clCnt', qs')
+                    clientTBQueueLengths :: Foldable t => t Client -> IO (Natural, Natural, Natural)
+                    clientTBQueueLengths = foldM addQueueLengths (0, 0, 0)
+                    addQueueLengths (!rl, !sl, !ml) cl = do
+                      (rl', sl', ml') <- queueLengths cl
+                      pure (rl + rl', sl + sl', ml + ml')
+                    queueLengths Client {rcvQ, sndQ, msgQ} = do
+                      rl <- atomically $ lengthTBQueue rcvQ
+                      sl <- atomically $ lengthTBQueue sndQ
+                      ml <- atomically $ lengthTBQueue msgQ
+                      pure (rl, sl, ml)
                    countSMPSubs :: M.Map QueueId Sub -> IO (Int, Int, Int, Int)
                    countSMPSubs = foldM countSubs (0, 0, 0, 0)
                      where
-                        countSubs (c1, c2, c3, c4) Sub {subThread} =
-                          readTVarIO subThread >>= \st -> pure $ case st of
-                            NoSub -> (c1 + 1, c2, c3, c4)
-                            SubPending -> (c1, c2 + 1, c3, c4)
-                            SubThread _ -> (c1, c2, c3 + 1, c4)
-                            ProhibitSub -> (c1, c2, c3, c4 + 1)
-                    countSubClients = S.size . M.foldr' (S.insert . clientId) S.empty
+                        countSubs (c1, c2, c3, c4) Sub {subThread} = case subThread of
+                          ServerSub t -> do
+                            st <- readTVarIO t
+                            pure $ case st of
+                              NoSub -> (c1 + 1, c2, c3, c4)
+                              SubPending -> (c1, c2 + 1, c3, c4)
+                              SubThread _ -> (c1, c2, c3 + 1, c4)
+                          ProhibitSub -> pure (c1, c2, c3, c4 + 1)
              CPDelete queueId' -> withUserRole $ unliftIO u $ do
                st <- asks queueStore
                ms <- asks msgStore
@@ -562,7 +629,8 @@ runClientTransport h@THandle {params = thParams@THandleParams {thVersion, sessio
  where
    disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport h (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
    disconnectThread_ _ _ = []
-    noSubscriptions c = atomically $ (&&) <$> TM.null (subscriptions c) <*> TM.null (ntfSubscriptions c)
+    noSubscriptions c = atomically $ (&&) <$> TM.null (ntfSubscriptions c) <*> (not . hasSubs <$> readTVar (subscriptions c))
+    hasSubs = any $ (\case ServerSub _ -> True; ProhibitSub -> False) . subThread

 clientDisconnected :: Client -> M ()
 clientDisconnected c@Client {clientId, subscriptions, ntfSubscriptions, connected, sessionId, endThreads} = do
@@ -590,10 +658,12 @@ sameClientId :: Client -> Client -> Bool
 sameClientId Client {clientId} Client {clientId = cId'} = clientId == cId'

 cancelSub :: Sub -> IO ()
-cancelSub s =
-  readTVarIO (subThread s) >>= \case
-    SubThread t -> liftIO $ deRefWeak t >>= mapM_ killThread
-    _ -> pure ()
+cancelSub s = case subThread s of
+  ServerSub st ->
+    readTVarIO st >>= \case
+      SubThread t -> liftIO $ deRefWeak t >>= mapM_ killThread
+      _ -> pure ()
+  ProhibitSub -> pure ()

 receive :: Transport c => THandleSMP c 'TServer -> Client -> M ()
 receive h@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiveAt, sessionId} = do
@@ -618,6 +688,7 @@ receive h@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiv
                case cmd of
                  Cmd _ SEND {} -> incStat $ msgSentAuth stats
                  Cmd _ SUB -> incStat $ qSubAuth stats
+                  Cmd _ NSUB -> incStat $ ntfSubAuth stats
                  Cmd _ GET -> incStat $ msgGetAuth stats
                  _ -> pure ()
                pure $ Left (corrId, entId, ERR AUTH)
@@ -849,10 +920,10 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
      incStat $ sel ps
      when own $ incStat $ sel psOwn
    processCommand :: (Maybe QueueRec, Transmission Cmd) -> M (Maybe (Transmission BrokerMsg))
-    processCommand (qr_, (corrId, queueId, cmd)) = case cmd of
-      Cmd SProxiedClient command -> processProxiedCmd (corrId, queueId, command)
+    processCommand (qr_, (corrId, entId, cmd)) = case cmd of
+      Cmd SProxiedClient command -> processProxiedCmd (corrId, entId, command)
      Cmd SSender command -> Just <$> case command of
-        SKEY sKey -> (corrId,queueId,) <$> case qr_ of
+        SKEY sKey -> (corrId,entId,) <$> case qr_ of
          Just QueueRec {sndSecure, recipientId}
            | sndSecure -> secureQueue_ "SKEY" recipientId sKey
            | otherwise -> pure $ ERR AUTH
@@ -868,15 +939,15 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
            ifM
              allowNew
              (createQueue st rKey dhKey subMode sndSecure)
-              (pure (corrId, queueId, ERR AUTH))
+              (pure (corrId, entId, ERR AUTH))
            where
              allowNew = do
                ServerConfig {allowNewQueues, newQueueBasicAuth} <- asks config
                pure $ allowNewQueues && maybe True ((== auth) . Just) newQueueBasicAuth
-          SUB -> withQueue (`subscribeQueue` queueId)
+          SUB -> withQueue (`subscribeQueue` entId)
          GET -> withQueue getMessage
          ACK msgId -> withQueue (`acknowledgeMsg` msgId)
-          KEY sKey -> (corrId,queueId,) <$> case qr_ of
+          KEY sKey -> (corrId,entId,) <$> case qr_ of
            Just QueueRec {recipientId} -> secureQueue_ "KEY" recipientId sKey
            Nothing -> pure $ ERR INTERNAL
          NKEY nKey dhKey -> addQueueNotifier_ st nKey dhKey
@@ -901,7 +972,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                    status = QueueActive,
                    sndSecure
                  }
-          (corrId,queueId,) <$> addQueueRetry 3 qik qRec
+          (corrId,entId,) <$> addQueueRetry 3 qik qRec
          where
            addQueueRetry ::
              Int -> ((RecipientId, SenderId) -> QueueIdsKeys) -> ((RecipientId, SenderId) -> QueueRec) -> M BrokerMsg
@@ -946,34 +1017,36 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
        addQueueNotifier_ st notifierKey dhKey = time "NKEY" $ do
          (rcvPublicDhKey, privDhKey) <- atomically . C.generateKeyPair =<< asks random
          let rcvNtfDhSecret = C.dh' dhKey privDhKey
-          (corrId,queueId,) <$> addNotifierRetry 3 rcvPublicDhKey rcvNtfDhSecret
+          (corrId,entId,) <$> addNotifierRetry 3 rcvPublicDhKey rcvNtfDhSecret
          where
            addNotifierRetry :: Int -> RcvNtfPublicDhKey -> RcvNtfDhSecret -> M BrokerMsg
            addNotifierRetry 0 _ _ = pure $ ERR INTERNAL
            addNotifierRetry n rcvPublicDhKey rcvNtfDhSecret = do
              notifierId <- randomId =<< asks (queueIdBytes . config)
              let ntfCreds = NtfCreds {notifierId, notifierKey, rcvNtfDhSecret}
-              atomically (addQueueNotifier st queueId ntfCreds) >>= \case
+              atomically (addQueueNotifier st entId ntfCreds) >>= \case
                Left DUPLICATE_ -> addNotifierRetry (n - 1) rcvPublicDhKey rcvNtfDhSecret
                Left e -> pure $ ERR e
                Right _ -> do
-                  withLog $ \s -> logAddNotifier s queueId ntfCreds
+                  withLog $ \s -> logAddNotifier s entId ntfCreds
+                  incStat . ntfCreated =<< asks serverStats
                  pure $ NID notifierId rcvPublicDhKey

        deleteQueueNotifier_ :: QueueStore -> M (Transmission BrokerMsg)
        deleteQueueNotifier_ st = do
-          withLog (`logDeleteNotifier` queueId)
-          atomically (deleteQueueNotifier st queueId) >>= \case
+          withLog (`logDeleteNotifier` entId)
+          atomically (deleteQueueNotifier st entId) >>= \case
            Right () -> do
              -- Possibly, the same should be done if the queue is suspended, but currently we do not use it
-              atomically $ writeTQueue ntfSubscribedQ (queueId, clnt, False)
+              atomically $ writeTQueue ntfSubscribedQ (entId, clnt, False)
+              incStat . ntfDeleted =<< asks serverStats
              pure ok
            Left e -> pure $ err e

        suspendQueue_ :: QueueStore -> M (Transmission BrokerMsg)
        suspendQueue_ st = do
-          withLog (`logSuspendQueue` queueId)
-          okResp <$> atomically (suspendQueue st queueId)
+          withLog (`logSuspendQueue` entId)
+          okResp <$> atomically (suspendQueue st entId)

        subscribeQueue :: QueueRec -> RecipientId -> M (Transmission BrokerMsg)
        subscribeQueue qr rId = do
@@ -981,7 +1054,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
            Nothing -> newSub >>= deliver True
            Just s@Sub {subThread} -> do
              stats <- asks serverStats
-              readTVarIO subThread >>= \case
+              case subThread of
                ProhibitSub -> do
                  -- cannot use SUB in the same connection where GET was used
                  incStat $ qSubProhibited stats
@@ -1003,15 +1076,16 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
              when inc $ do
                stats <- asks serverStats
                incStat $ (if isJust msg_ then qSub else qSubNoMsg) stats
+                atomically $ updatePeriodStats (subscribedQueues stats) rId
              deliverMessage "SUB" qr rId sub msg_

        getMessage :: QueueRec -> M (Transmission BrokerMsg)
        getMessage qr = time "GET" $ do
-          atomically (TM.lookup queueId subscriptions) >>= \case
+          atomically (TM.lookup entId subscriptions) >>= \case
            Nothing ->
              atomically newSub >>= (`getMessage_` Nothing)
            Just s@Sub {subThread} ->
-              readTVarIO subThread >>= \case
+              case subThread of
                ProhibitSub ->
                  atomically (tryTakeTMVar $ delivered s)
                    >>= getMessage_ s
@@ -1019,16 +1093,16 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                _ -> do
                  stats <- asks serverStats
                  incStat $ msgGetProhibited stats
-                  pure (corrId, queueId, ERR $ CMD PROHIBITED)
+                  pure (corrId, entId, ERR $ CMD PROHIBITED)
          where
            newSub :: STM Sub
            newSub = do
-              s <- newSubscription ProhibitSub
-              TM.insert queueId s subscriptions
+              s <- newProhibitedSub
+              TM.insert entId s subscriptions
              pure s
            getMessage_ :: Sub -> Maybe MsgId -> M (Transmission BrokerMsg)
            getMessage_ s delivered_ = do
-              q <- getStoreMsgQueue "GET" queueId
+              q <- getStoreMsgQueue "GET" entId
              stats <- asks serverStats
              (statCnt, r) <-
                atomically $
@@ -1036,8 +1110,8 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                    Just msg ->
                      let encMsg = encryptMsg qr msg
                          cnt = if isJust delivered_ then msgGetDuplicate else msgGet
-                       in setDelivered s msg $> (cnt, (corrId, queueId, MSG encMsg))
-                    _ -> pure (msgGetNoMsg, (corrId, queueId, OK))
+                       in setDelivered s msg $> (cnt, (corrId, entId, MSG encMsg))
+                    _ -> pure (msgGetNoMsg, (corrId, entId, OK))
              incStat $ statCnt stats
              pure r

@@ -1045,20 +1119,28 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
        withQueue action = maybe (pure $ err AUTH) action qr_

        subscribeNotifications :: M (Transmission BrokerMsg)
-        subscribeNotifications = time "NSUB" . atomically $ do
-          unlessM (TM.member queueId ntfSubscriptions) $ do
-            writeTQueue ntfSubscribedQ (queueId, clnt, True)
-            TM.insert queueId () ntfSubscriptions
+        subscribeNotifications = do
+          statCount <-
+            time "NSUB" . atomically $ do
+              ifM
+                (TM.member entId ntfSubscriptions)
+                (pure ntfSubDuplicate)
+                (newSub $> ntfSub)
+          incStat . statCount =<< asks serverStats
          pure ok
+          where
+            newSub = do
+              writeTQueue ntfSubscribedQ (entId, clnt, True)
+              TM.insert entId () ntfSubscriptions

        acknowledgeMsg :: QueueRec -> MsgId -> M (Transmission BrokerMsg)
        acknowledgeMsg qr msgId = time "ACK" $ do
-          atomically (TM.lookup queueId subscriptions) >>= \case
+          atomically (TM.lookup entId subscriptions) >>= \case
            Nothing -> pure $ err NO_MSG
            Just sub ->
              atomically (getDelivered sub) >>= \case
                Just st -> do
-                  q <- getStoreMsgQueue "ACK" queueId
+                  q <- getStoreMsgQueue "ACK" entId
                  case st of
                    ProhibitSub -> do
                      deletedMsg_ <- atomically $ tryDelMsg q msgId
@@ -1067,14 +1149,14 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                    _ -> do
                      (deletedMsg_, msg_) <- atomically $ tryDelPeekMsg q msgId
                      mapM_ (updateStats False) deletedMsg_
-                      deliverMessage "ACK" qr queueId sub msg_
+                      deliverMessage "ACK" qr entId sub msg_
                _ -> pure $ err NO_MSG
          where
-            getDelivered :: Sub -> STM (Maybe SubscriptionThread)
+            getDelivered :: Sub -> STM (Maybe ServerSub)
            getDelivered Sub {delivered, subThread} = do
              tryTakeTMVar delivered $>>= \msgId' ->
                if msgId == msgId' || B.null msgId
-                  then Just <$> readTVar subThread
+                  then pure $ Just subThread
                  else putTMVar delivered msgId' $> Nothing
            updateStats :: Bool -> Message -> M ()
            updateStats isGet = \case
@@ -1084,10 +1166,10 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                incStat $ msgRecv stats
                when isGet $ incStat $ msgRecvGet stats
                atomically $ modifyTVar' (msgCount stats) (subtract 1)
-                atomically $ updatePeriodStats (activeQueues stats) queueId
+                atomically $ updatePeriodStats (activeQueues stats) entId
                when (notification msgFlags) $ do
                  incStat $ msgRecvNtf stats
-                  atomically $ updatePeriodStats (activeQueuesNtf stats) queueId
+                  atomically $ updatePeriodStats (activeQueuesNtf stats) entId

        sendMessage :: QueueRec -> MsgFlags -> MsgBody -> M (Transmission BrokerMsg)
        sendMessage qr msgFlags msgBody
@@ -1163,26 +1245,28 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                deliverToSub =
                  TM.lookup rId subscribers
                    $>>= \rc@Client {subscriptions = subs, sndQ = q} -> TM.lookup rId subs
-                    $>>= \s@Sub {subThread, delivered} -> readTVar subThread >>= \case
-                      NoSub ->
-                        tryTakeTMVar delivered >>= \case
-                          Just _ -> pure Nothing -- if a message was already delivered, should not deliver more
-                          Nothing ->
-                            ifM
-                              (isFullTBQueue q)
-                              (writeTVar subThread SubPending $> Just (rc, s))
-                              (deliver q s $> Nothing)
-                      _ -> pure Nothing
+                    $>>= \s@Sub {subThread, delivered} -> case subThread of
+                      ProhibitSub -> pure Nothing
+                      ServerSub st -> readTVar st >>= \case
+                        NoSub ->
+                          tryTakeTMVar delivered >>= \case
+                            Just _ -> pure Nothing -- if a message was already delivered, should not deliver more
+                            Nothing ->
+                              ifM
+                                (isFullTBQueue q)
+                                (writeTVar st SubPending $> Just (rc, s, st))
+                                (deliver q s $> Nothing)
+                        _ -> pure Nothing
                deliver q s = do
                  let encMsg = encryptMsg qr msg
                  writeTBQueue q [(CorrId "", rId, MSG encMsg)]
                  void $ setDelivered s msg
-                forkDeliver (rc@Client {sndQ = q}, s@Sub {subThread, delivered}) = do
+                forkDeliver (rc@Client {sndQ = q}, s@Sub {delivered}, st) = do
                  t <- mkWeakThreadId =<< forkIO deliverThread
-                  atomically . modifyTVar' subThread $ \case
+                  atomically . modifyTVar' st $ \case
                    -- this case is needed because deliverThread can exit before it
                    SubPending -> SubThread t
-                    st -> st
+                    st' -> st'
                  where
                    deliverThread = do
                      labelMyThread $ B.unpack ("client $" <> encode sessionId) <> " deliver/SEND"
@@ -1192,7 +1276,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                            Just _ -> pure () -- if a message was already delivered, should not deliver more
                            Nothing -> do
                              deliver q s
-                              writeTVar subThread NoSub
+                              writeTVar st NoSub

            trySendNotification :: NtfCreds -> Message -> TVar ChaChaDRG -> STM (Maybe Bool)
            trySendNotification NtfCreds {notifierId, rcvNtfDhSecret} msg ntfNonceDrg =
@@ -1270,7 +1354,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
                      VRFailed -> Left (corrId', entId', ERR AUTH)
        deliverMessage :: T.Text -> QueueRec -> RecipientId -> Sub -> Maybe Message -> M (Transmission BrokerMsg)
        deliverMessage name qr rId s@Sub {subThread} msg_ = time (name <> " deliver") . atomically $
-          readTVar subThread >>= \case
+          case subThread of
            ProhibitSub -> pure resp
            _ -> case msg_ of
              Just msg ->
@@ -1281,7 +1365,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi
            resp = (corrId, rId, OK)

        time :: T.Text -> M a -> M a
-        time name = timed name queueId
+        time name = timed name entId

        encryptMsg :: QueueRec -> Message -> RcvMessage
        encryptMsg qr msg = encrypt . encodeRcvMsgBody $ case msg of
@@ -1304,42 +1388,44 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi

        delQueueAndMsgs :: QueueStore -> M (Transmission BrokerMsg)
        delQueueAndMsgs st = do
-          withLog (`logDeleteQueue` queueId)
+          withLog (`logDeleteQueue` entId)
          ms <- asks msgStore
-          atomically (deleteQueue st queueId $>>= \q -> delMsgQueue ms queueId $> Right q) >>= \case
+          atomically (deleteQueue st entId $>>= \q -> delMsgQueue ms entId $> Right q) >>= \case
            Right q -> do
              -- Possibly, the same should be done if the queue is suspended, but currently we do not use it
-              atomically $ writeTQueue subscribedQ (queueId, clnt, False)
-              atomically $ writeTQueue ntfSubscribedQ (queueId, clnt, False)
+              atomically $ writeTQueue subscribedQ (entId, clnt, False)
+              atomically $ writeTQueue ntfSubscribedQ (entId, clnt, False)
              updateDeletedStats q
              pure ok
            Left e -> pure $ err e

        getQueueInfo :: QueueRec -> M (Transmission BrokerMsg)
        getQueueInfo QueueRec {senderKey, notifier} = do
-          q@MsgQueue {size} <- getStoreMsgQueue "getQueueInfo" queueId
+          q@MsgQueue {size} <- getStoreMsgQueue "getQueueInfo" entId
          info <- atomically $ do
-            qiSub <- TM.lookup queueId subscriptions >>= mapM mkQSub
+            qiSub <- TM.lookup entId subscriptions >>= mapM mkQSub
            qiSize <- readTVar size
            qiMsg <- toMsgInfo <$$> tryPeekMsg q
            pure QueueInfo {qiSnd = isJust senderKey, qiNtf = isJust notifier, qiSub, qiSize, qiMsg}
-          pure (corrId, queueId, INFO info)
+          pure (corrId, entId, INFO info)
          where
            mkQSub Sub {subThread, delivered} = do
-              st <- readTVar subThread
-              let qSubThread = case st of
+              qSubThread <- case subThread of
+                ServerSub t -> do
+                  st <- readTVar t
+                  pure $ case st of
                    NoSub -> QNoSub
                    SubPending -> QSubPending
                    SubThread _ -> QSubThread
-                    ProhibitSub -> QProhibitSub
+                ProhibitSub -> pure QProhibitSub
              qDelivered <- decodeLatin1 . encode <$$> tryReadTMVar delivered
              pure QSub {qSubThread, qDelivered}

        ok :: Transmission BrokerMsg
-        ok = (corrId, queueId, OK)
+        ok = (corrId, entId, OK)

        err :: ErrorType -> Transmission BrokerMsg
-        err e = (corrId, queueId, ERR e)
+        err e = (corrId, entId, ERR e)

        okResp :: Either ErrorType () -> Transmission BrokerMsg
        okResp = either err $ const ok
@@ -1383,13 +1469,16 @@ saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessag
      logInfo $ "saving messages to file " <> T.pack f
      ms <- asks msgStore
      liftIO . withFile f WriteMode $ \h ->
-        readTVarIO ms >>= mapM_ (saveQueueMsgs ms h) . M.keys
+        readTVarIO ms >>= mapM_ (saveQueueMsgs h) . M.assocs
      logInfo "messages saved"
      where
-        getMessages = if keepMsgs then snapshotMsgQueue else flushMsgQueue
-        saveQueueMsgs ms h rId =
-          atomically (getMessages ms rId)
-            >>= mapM_ (B.hPutStrLn h . strEncode . MLRv3 rId)
+        saveQueueMsgs h (rId, q) = BLD.hPutBuilder h . encodeMessages rId =<< atomically (getMessages $ msgQueue q)
+        getMessages = if keepMsgs then snapshotTQueue else flushTQueue
+        snapshotTQueue q = do
+          msgs <- flushTQueue q
+          mapM_ (writeTQueue q) msgs
+          pure msgs
+        encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')

 restoreServerMessages :: M Int
 restoreServerMessages =
@@ -1,12 +1,15 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StrictData #-}

 module Simplex.Messaging.Server.Env.STM where

 import Control.Concurrent (ThreadId)
+import Control.Logger.Simple
+import Control.Monad
 import Control.Monad.IO.Unlift
 import Crypto.Random
 import Data.ByteString.Char8 (ByteString)
@@ -17,6 +20,7 @@ import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Maybe (isJust, isNothing)
+import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Clock.System (SystemTime)
 import Data.X509.Validation (Fingerprint (..))
@@ -104,7 +108,7 @@ defaultMessageExpiration =
 defaultInactiveClientExpiration :: ExpirationConfig
 defaultInactiveClientExpiration =
  ExpirationConfig
-    { ttl = 43200, -- seconds, 12 hours
+    { ttl = 21600, -- seconds, 6 hours
      checkInterval = 3600 -- seconds, 1 hours
    }

@@ -162,10 +166,12 @@ data Client = Client
    sndActiveAt :: TVar SystemTime
  }

-data SubscriptionThread = NoSub | SubPending | SubThread (Weak ThreadId) | ProhibitSub
+data ServerSub = ServerSub (TVar SubscriptionThread) | ProhibitSub
+
+data SubscriptionThread = NoSub | SubPending | SubThread (Weak ThreadId)

 data Sub = Sub
-  { subThread :: TVar SubscriptionThread,
+  { subThread :: ServerSub, -- Nothing value indicates that sub
    delivered :: TMVar MsgId
  }

@@ -197,16 +203,24 @@ newClient nextClientId qSize thVersion sessionId createdAt = do
 newSubscription :: SubscriptionThread -> STM Sub
 newSubscription st = do
  delivered <- newEmptyTMVar
-  subThread <- newTVar st
+  subThread <- ServerSub <$> newTVar st
  return Sub {subThread, delivered}

+newProhibitedSub :: STM Sub
+newProhibitedSub = do
+  delivered <- newEmptyTMVar
+  return Sub {subThread = ProhibitSub, delivered}
+
 newEnv :: ServerConfig -> IO Env
 newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile, smpAgentCfg, transportConfig, information, messageExpiration} = do
  server <- atomically newServer
  queueStore <- atomically newQueueStore
  msgStore <- atomically newMsgStore
  random <- liftIO C.newRandom
-  storeLog <- restoreQueues queueStore `mapM` storeLogFile
+  storeLog <-
+    forM storeLogFile $ \f -> do
+      logInfo $ "restoring queues from file " <> T.pack f
+      restoreQueues queueStore f
  tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig)
  Fingerprint fp <- loadFingerprint caCertificateFile
  let serverIdentity = KeyHash fp
@@ -14,8 +14,6 @@ module Simplex.Messaging.Server.MsgStore.STM
    getMsgQueue,
    delMsgQueue,
    delMsgQueueSize,
-    flushMsgQueue,
-    snapshotMsgQueue,
    writeMsg,
    tryPeekMsg,
    peekMsg,
@@ -25,7 +23,6 @@ module Simplex.Messaging.Server.MsgStore.STM
  )
 where

-import Control.Concurrent.STM.TQueue (flushTQueue)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
 import Data.Int (Int64)
@@ -64,17 +61,6 @@ delMsgQueue st rId = TM.delete rId st
 delMsgQueueSize :: STMMsgStore -> RecipientId -> STM Int
 delMsgQueueSize st rId = TM.lookupDelete rId st >>= maybe (pure 0) (\MsgQueue {size} -> readTVar size)

-flushMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
-flushMsgQueue st rId = TM.lookupDelete rId st >>= maybe (pure []) (flushTQueue . msgQueue)
-
-snapshotMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
-snapshotMsgQueue st rId = TM.lookup rId st >>= maybe (pure []) (snapshotTQueue . msgQueue)
-  where
-    snapshotTQueue q = do
-      msgs <- flushTQueue q
-      mapM_ (writeTQueue q) msgs
-      pure msgs
-
 writeMsg :: MsgQueue -> Message -> STM (Maybe (Message, Bool))
 writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} !msg = do
  canWrt <- readTVar canWrite
@@ -31,6 +31,11 @@ data ServerStats = ServerStats
    qSubAuth :: TVar Int,
    qSubDuplicate :: TVar Int,
    qSubProhibited :: TVar Int,
+    ntfCreated :: TVar Int,
+    ntfDeleted :: TVar Int,
+    ntfSub :: TVar Int,
+    ntfSubAuth :: TVar Int,
+    ntfSubDuplicate :: TVar Int,
    msgSent :: TVar Int,
    msgSentAuth :: TVar Int,
    msgSentQuota :: TVar Int,
@@ -44,6 +49,7 @@ data ServerStats = ServerStats
    msgGetProhibited :: TVar Int,
    msgExpired :: TVar Int,
    activeQueues :: PeriodStats RecipientId,
+    subscribedQueues :: PeriodStats RecipientId,
    msgSentNtf :: TVar Int, -- sent messages with NTF flag
    msgRecvNtf :: TVar Int, -- received messages with NTF flag
    activeQueuesNtf :: PeriodStats RecipientId,
@@ -71,6 +77,11 @@ data ServerStatsData = ServerStatsData
    _qSubAuth :: Int,
    _qSubDuplicate :: Int,
    _qSubProhibited :: Int,
+    _ntfCreated :: Int,
+    _ntfDeleted :: Int,
+    _ntfSub :: Int,
+    _ntfSubAuth :: Int,
+    _ntfSubDuplicate :: Int,
    _msgSent :: Int,
    _msgSentAuth :: Int,
    _msgSentQuota :: Int,
@@ -84,6 +95,7 @@ data ServerStatsData = ServerStatsData
    _msgGetProhibited :: Int,
    _msgExpired :: Int,
    _activeQueues :: PeriodStatsData RecipientId,
+    _subscribedQueues :: PeriodStatsData RecipientId,
    _msgSentNtf :: Int,
    _msgRecvNtf :: Int,
    _activeQueuesNtf :: PeriodStatsData RecipientId,
@@ -113,6 +125,11 @@ newServerStats ts = do
  qSubAuth <- newTVar 0
  qSubDuplicate <- newTVar 0
  qSubProhibited <- newTVar 0
+  ntfCreated <- newTVar 0
+  ntfDeleted <- newTVar 0
+  ntfSub <- newTVar 0
+  ntfSubAuth <- newTVar 0
+  ntfSubDuplicate <- newTVar 0
  msgSent <- newTVar 0
  msgSentAuth <- newTVar 0
  msgSentQuota <- newTVar 0
@@ -126,6 +143,7 @@ newServerStats ts = do
  msgGetProhibited <- newTVar 0
  msgExpired <- newTVar 0
  activeQueues <- newPeriodStats
+  subscribedQueues <- newPeriodStats
  msgSentNtf <- newTVar 0
  msgRecvNtf <- newTVar 0
  activeQueuesNtf <- newPeriodStats
@@ -152,6 +170,11 @@ newServerStats ts = do
        qSubAuth,
        qSubDuplicate,
        qSubProhibited,
+        ntfCreated,
+        ntfDeleted,
+        ntfSub,
+        ntfSubAuth,
+        ntfSubDuplicate,
        msgSent,
        msgSentAuth,
        msgSentQuota,
@@ -165,6 +188,7 @@ newServerStats ts = do
        msgGetProhibited,
        msgExpired,
        activeQueues,
+        subscribedQueues,
        msgSentNtf,
        msgRecvNtf,
        activeQueuesNtf,
@@ -193,6 +217,11 @@ getServerStatsData s = do
  _qSubAuth <- readTVar $ qSubAuth s
  _qSubDuplicate <- readTVar $ qSubDuplicate s
  _qSubProhibited <- readTVar $ qSubProhibited s
+  _ntfCreated <- readTVar $ ntfCreated s
+  _ntfDeleted <- readTVar $ ntfDeleted s
+  _ntfSub <- readTVar $ ntfSub s
+  _ntfSubAuth <- readTVar $ ntfSubAuth s
+  _ntfSubDuplicate <- readTVar $ ntfSubDuplicate s
  _msgSent <- readTVar $ msgSent s
  _msgSentAuth <- readTVar $ msgSentAuth s
  _msgSentQuota <- readTVar $ msgSentQuota s
@@ -206,6 +235,7 @@ getServerStatsData s = do
  _msgGetProhibited <- readTVar $ msgGetProhibited s
  _msgExpired <- readTVar $ msgExpired s
  _activeQueues <- getPeriodStatsData $ activeQueues s
+  _subscribedQueues <- getPeriodStatsData $ subscribedQueues s
  _msgSentNtf <- readTVar $ msgSentNtf s
  _msgRecvNtf <- readTVar $ msgRecvNtf s
  _activeQueuesNtf <- getPeriodStatsData $ activeQueuesNtf s
@@ -232,6 +262,11 @@ getServerStatsData s = do
        _qSubAuth,
        _qSubDuplicate,
        _qSubProhibited,
+        _ntfCreated,
+        _ntfDeleted,
+        _ntfSub,
+        _ntfSubAuth,
+        _ntfSubDuplicate,
        _msgSent,
        _msgSentAuth,
        _msgSentQuota,
@@ -245,6 +280,7 @@ getServerStatsData s = do
        _msgGetProhibited,
        _msgExpired,
        _activeQueues,
+        _subscribedQueues,
        _msgSentNtf,
        _msgRecvNtf,
        _activeQueuesNtf,
@@ -273,6 +309,11 @@ setServerStats s d = do
  writeTVar (qSubAuth s) $! _qSubAuth d
  writeTVar (qSubDuplicate s) $! _qSubDuplicate d
  writeTVar (qSubProhibited s) $! _qSubProhibited d
+  writeTVar (ntfCreated s) $! _ntfCreated d
+  writeTVar (ntfDeleted s) $! _ntfDeleted d
+  writeTVar (ntfSub s) $! _ntfSub d
+  writeTVar (ntfSubAuth s) $! _ntfSubAuth d
+  writeTVar (ntfSubDuplicate s) $! _ntfSubDuplicate d
  writeTVar (msgSent s) $! _msgSent d
  writeTVar (msgSentAuth s) $! _msgSentAuth d
  writeTVar (msgSentQuota s) $! _msgSentQuota d
@@ -286,6 +327,7 @@ setServerStats s d = do
  writeTVar (msgGetProhibited s) $! _msgGetProhibited d
  writeTVar (msgExpired s) $! _msgExpired d
  setPeriodStats (activeQueues s) (_activeQueues d)
+  setPeriodStats (subscribedQueues s) (_subscribedQueues d)
  writeTVar (msgSentNtf s) $! _msgSentNtf d
  writeTVar (msgRecvNtf s) $! _msgRecvNtf d
  setPeriodStats (activeQueuesNtf s) (_activeQueuesNtf d)
@@ -315,6 +357,11 @@ instance StrEncoding ServerStatsData where
        "qSubAuth=" <> strEncode (_qSubAuth d),
        "qSubDuplicate=" <> strEncode (_qSubDuplicate d),
        "qSubProhibited=" <> strEncode (_qSubProhibited d),
+        "ntfCreated=" <> strEncode (_ntfCreated d),
+        "ntfDeleted=" <> strEncode (_ntfDeleted d),
+        "ntfSub=" <> strEncode (_ntfSub d),
+        "ntfSubAuth=" <> strEncode (_ntfSubAuth d),
+        "ntfSubDuplicate=" <> strEncode (_ntfSubDuplicate d),
        "msgSent=" <> strEncode (_msgSent d),
        "msgSentAuth=" <> strEncode (_msgSentAuth d),
        "msgSentQuota=" <> strEncode (_msgSentQuota d),
@@ -334,6 +381,8 @@ instance StrEncoding ServerStatsData where
        "msgNtfLost=" <> strEncode (_msgNtfLost d),
        "activeQueues:",
        strEncode (_activeQueues d),
+        "subscribedQueues:",
+        strEncode (_subscribedQueues d),
        "activeQueuesNtf:",
        strEncode (_activeQueuesNtf d),
        "pRelays:",
@@ -359,6 +408,11 @@ instance StrEncoding ServerStatsData where
    _qSubAuth <- opt "qSubAuth="
    _qSubDuplicate <- opt "qSubDuplicate="
    _qSubProhibited <- opt "qSubProhibited="
+    _ntfCreated <- opt "ntfCreated="
+    _ntfDeleted <- opt "ntfDeleted="
+    _ntfSub <- opt "ntfSub="
+    _ntfSubAuth <- opt "ntfSubAuth="
+    _ntfSubDuplicate <- opt "ntfSubDuplicate="
    _msgSent <- "msgSent=" *> strP <* A.endOfLine
    _msgSentAuth <- opt "msgSentAuth="
    _msgSentQuota <- opt "msgSentQuota="
@@ -384,6 +438,10 @@ instance StrEncoding ServerStatsData where
          _week <- "weekMsgQueues=" *> strP <* A.endOfLine
          _month <- "monthMsgQueues=" *> strP <* optional A.endOfLine
          pure PeriodStatsData {_day, _week, _month}
+    _subscribedQueues <-
+      optional ("subscribedQueues:" <* A.endOfLine) >>= \case
+        Just _ -> strP <* optional A.endOfLine
+        _ -> pure newPeriodStatsData
    _activeQueuesNtf <-
      optional ("activeQueuesNtf:" <* A.endOfLine) >>= \case
        Just _ -> strP <* optional A.endOfLine
@@ -406,6 +464,11 @@ instance StrEncoding ServerStatsData where
          _qSubAuth,
          _qSubDuplicate,
          _qSubProhibited,
+          _ntfCreated,
+          _ntfDeleted,
+          _ntfSub,
+          _ntfSubAuth,
+          _ntfSubDuplicate,
          _msgSent,
          _msgSentAuth,
          _msgSentQuota,
@@ -424,6 +487,7 @@ instance StrEncoding ServerStatsData where
          _msgNtfNoSub,
          _msgNtfLost,
          _activeQueues,
+          _subscribedQueues,
          _activeQueuesNtf,
          _pRelays,
          _pRelaysOwn,
@@ -5,9 +5,6 @@
 module Simplex.Messaging.Session where

 import Control.Concurrent.STM
-import Control.Monad
-import Data.Composition ((.:.))
-import Data.Functor (($>))
 import Data.Time (UTCTime)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
@@ -31,14 +28,10 @@ getSessVar sessSeq sessKey vs sessionVarTs = maybe (Left <$> newSessionVar) (pur
      pure v

 removeSessVar :: Ord k => SessionVar a -> k -> TMap k (SessionVar a) -> STM ()
-removeSessVar = void .:. removeSessVar'
-{-# INLINE removeSessVar #-}
-
-removeSessVar' :: Ord k => SessionVar a -> k -> TMap k (SessionVar a) -> STM Bool
-removeSessVar' v sessKey vs =
+removeSessVar v sessKey vs =
  TM.lookup sessKey vs >>= \case
-    Just v' | sessionVarId v == sessionVarId v' -> TM.delete sessKey vs $> True
-    _ -> pure False
+    Just v' | sessionVarId v == sessionVarId v' -> TM.delete sessKey vs
+    _ -> pure ()

 tryReadSessVar :: Ord k => k -> TMap k (SessionVar a) -> STM (Maybe a)
 tryReadSessVar sessKey vs = TM.lookup sessKey vs $>>= (tryReadTMVar . sessionVar)
@@ -38,6 +38,7 @@ module AgentTests.FunctionalAPITests
    rfGet,
    sfGet,
    nGet,
+    getInAnyOrder,
    (##>),
    (=##>),
    pattern CON,
@@ -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
@@ -1,6 +1,7 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
+{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}

 module CoreTests.TRcvQueuesTests where
@@ -30,13 +31,13 @@ tRcvQueuesTests = do
  describe "queue transfer" $ do
    it "getDelSessQueues-batchAddQueues preserves total length" removeSubsTest

-checkDataInvariant :: RQ.TRcvQueues -> IO Bool
+checkDataInvariant :: RQ.Queue q => RQ.TRcvQueues q -> IO Bool
 checkDataInvariant trq = atomically $ do
  conns <- readTVar $ RQ.getConnections trq
  qs <- readTVar $ RQ.getRcvQueues trq
  -- three invariant checks
-  let inv1 = all (\cId -> (S.fromList . L.toList <$> M.lookup cId conns) == Just (M.keysSet (M.filter (\q -> connId q == cId) qs))) (M.keys conns)
-      inv2 = all (\(k, q) -> maybe False ((k `elem`) . L.toList) (M.lookup (connId q) conns)) (M.assocs qs)
+  let inv1 = all (\cId -> (S.fromList . L.toList <$> M.lookup cId conns) == Just (M.keysSet (M.filter (\q -> RQ.connId' q == cId) qs))) (M.keys conns)
+      inv2 = all (\(k, q) -> maybe False ((k `elem`) . L.toList) (M.lookup (RQ.connId' q) conns)) (M.assocs qs)
      inv3 = all (\(k, q) -> RQ.qKey q == k) (M.assocs qs)
  pure $ inv1 && inv2 && inv3

@@ -76,7 +77,7 @@ batchIdempotentTest = do
  atomically $ RQ.batchAddQueues trq qs
  checkDataInvariant trq `shouldReturn` True
  readTVarIO (RQ.getRcvQueues trq) `shouldReturn` qs'
-  fmap L.nub <$> readTVarIO (RQ.getConnections trq) `shouldReturn`cs' -- connections get duplicated, but that doesn't appear to affect anybody
+  fmap L.nub <$> readTVarIO (RQ.getConnections trq) `shouldReturn` cs' -- connections get duplicated, but that doesn't appear to affect anybody

 deleteConnTest :: IO ()
 deleteConnTest = do
@@ -112,23 +113,23 @@ getDelSessQueuesTest :: IO ()
 getDelSessQueuesTest = do
  trq <- atomically RQ.empty
  let qs =
-        [ dummyRQ 0 "smp://1234-w==@alpha" "c1",
-          dummyRQ 0 "smp://1234-w==@alpha" "c2",
-          dummyRQ 0 "smp://1234-w==@beta" "c3",
-          dummyRQ 1 "smp://1234-w==@beta" "c4"
+        [ ("1", dummyRQ 0 "smp://1234-w==@alpha" "c1"),
+          ("1", dummyRQ 0 "smp://1234-w==@alpha" "c2"),
+          ("1", dummyRQ 0 "smp://1234-w==@beta" "c3"),
+          ("1", dummyRQ 1 "smp://1234-w==@beta" "c4")
        ]
  atomically $ RQ.batchAddQueues trq qs
  checkDataInvariant trq `shouldReturn` True
  -- no user
-  atomically (RQ.getDelSessQueues (2, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
+  atomically (RQ.getDelSessQueues (2, "smp://1234-w==@alpha", Nothing) "1" trq) `shouldReturn` ([], [])
  checkDataInvariant trq `shouldReturn` True
  -- wrong user
-  atomically (RQ.getDelSessQueues (1, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
+  atomically (RQ.getDelSessQueues (1, "smp://1234-w==@alpha", Nothing) "1" trq) `shouldReturn` ([], [])
  checkDataInvariant trq `shouldReturn` True
  -- connections intact
  atomically (RQ.hasConn "c1" trq) `shouldReturn` True
  atomically (RQ.hasConn "c2" trq) `shouldReturn` True
-  atomically (RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"], ["c1", "c2"])
+  atomically (RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) "1" trq) `shouldReturn` ([dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"], ["c1", "c2"])
  checkDataInvariant trq `shouldReturn` True
  -- connections gone
  atomically (RQ.hasConn "c1" trq) `shouldReturn` False
@@ -141,29 +142,29 @@ removeSubsTest :: IO ()
 removeSubsTest = do
  aq <- atomically RQ.empty
  let qs =
-        [ dummyRQ 0 "smp://1234-w==@alpha" "c1",
-          dummyRQ 0 "smp://1234-w==@alpha" "c2",
-          dummyRQ 0 "smp://1234-w==@beta" "c3",
-          dummyRQ 1 "smp://1234-w==@beta" "c4"
+        [ ("1", dummyRQ 0 "smp://1234-w==@alpha" "c1"),
+          ("1", dummyRQ 0 "smp://1234-w==@alpha" "c2"),
+          ("1", dummyRQ 0 "smp://1234-w==@beta" "c3"),
+          ("1", dummyRQ 1 "smp://1234-w==@beta" "c4")
        ]
  atomically $ RQ.batchAddQueues aq qs

  pq <- atomically RQ.empty
  atomically (totalSize aq pq) `shouldReturn` (4, 4)

-  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) aq >>= RQ.batchAddQueues pq . fst
+  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) "1" aq >>= RQ.batchAddQueues pq . map ("1",) . fst
  atomically (totalSize aq pq) `shouldReturn` (4, 4)

-  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@beta", Just "non-existent") aq >>= RQ.batchAddQueues pq . fst
+  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@beta", Just "non-existent") "1" aq >>= RQ.batchAddQueues pq . map ("1",) . fst
  atomically (totalSize aq pq) `shouldReturn` (4, 4)

-  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@localhost", Nothing) aq >>= RQ.batchAddQueues pq . fst
+  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@localhost", Nothing) "1" aq >>= RQ.batchAddQueues pq . map ("1",) . fst
  atomically (totalSize aq pq) `shouldReturn` (4, 4)

-  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@beta", Just "c3") aq >>= RQ.batchAddQueues pq . fst
+  atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@beta", Just "c3") "1" aq >>= RQ.batchAddQueues pq . map ("1",) . fst
  atomically (totalSize aq pq) `shouldReturn` (4, 4)

-totalSize :: RQ.TRcvQueues -> RQ.TRcvQueues -> STM (Int, Int)
+totalSize :: RQ.TRcvQueues q -> RQ.TRcvQueues q -> STM (Int, Int)
 totalSize a b = do
  qsizeA <- M.size <$> readTVar (RQ.getRcvQueues a)
  qsizeB <- M.size <$> readTVar (RQ.getRcvQueues b)
@@ -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"
@@ -361,11 +361,15 @@ agentViaProxyRetryOffline = do
          -- proxy relay down
          4 <- msgId <$> A.sendMessage bob aliceId pqEnc noMsgFlags msg2
          bob `down` aliceId
-        withServer2 $ \_ -> runRight_ $ do
-          bob `up` aliceId
-          get bob ##> ("", aliceId, A.SENT (baseId + 4) bProxySrv)
-          get alice =##> \case ("", c, Msg' _ pq msg2') -> c == bobId && pq == pqEnc && msg2 == msg2'; _ -> False
-          ackMessage alice bobId (baseId + 4) Nothing
+        withServer2 $ \_ -> do
+          getInAnyOrder
+            bob
+            [ \case ("", "", AEvt SAENone (UP _ [c])) -> c == aliceId; _ -> False,
+              \case ("", c, AEvt SAEConn (A.SENT mId srv)) -> c == aliceId && mId == baseId + 4 && srv == bProxySrv; _ -> False
+            ]
+          runRight_ $ do
+            get alice =##> \case ("", c, Msg' _ pq msg2') -> c == bobId && pq == pqEnc && msg2 == msg2'; _ -> False
+            ackMessage alice bobId (baseId + 4) Nothing
  where
    withServer :: (ThreadId -> IO a) -> IO a
    withServer = withServer_ testStoreLogFile testStoreMsgsFile testPort
@@ -610,7 +610,7 @@ testRestoreMessages at@(ATransport t) =

    logSize testStoreLogFile `shouldReturn` 2
    logSize testStoreMsgsFile `shouldReturn` 5
-    logSize testServerStatsBackupFile `shouldReturn` 62
+    logSize testServerStatsBackupFile `shouldReturn` 71
    Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile
    checkStats stats1 [rId] 5 1

@@ -628,7 +628,7 @@ testRestoreMessages at@(ATransport t) =
    logSize testStoreLogFile `shouldReturn` 1
    -- the last message is not removed because it was not ACK'd
    logSize testStoreMsgsFile `shouldReturn` 3
-    logSize testServerStatsBackupFile `shouldReturn` 62
+    logSize testServerStatsBackupFile `shouldReturn` 71
    Right stats2 <- strDecode <$> B.readFile testServerStatsBackupFile
    checkStats stats2 [rId] 5 3

@@ -647,7 +647,7 @@ testRestoreMessages at@(ATransport t) =

    logSize testStoreLogFile `shouldReturn` 1
    logSize testStoreMsgsFile `shouldReturn` 0
-    logSize testServerStatsBackupFile `shouldReturn` 62
+    logSize testServerStatsBackupFile `shouldReturn` 71
    Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
    checkStats stats3 [rId] 5 5
Author	SHA1	Message	Date
Evgeny Poberezkin	2de16cfae8	6.0.0.4	2024-07-28 15:03:32 +01:00
Evgeny Poberezkinandspaced4ndy	5fa3c149e9	smp server: fix server-info, additional stats, allow expiring inactive clients which have prohibit subscriptions only (iOS NSE clients) (#1237 ) * smp server: fix server-info * fix * faster saving messages * remove comment * move ProhibitSub out of TVar * subscription stats * stabilize test * order Co-authored-by: spaced4ndy <[email protected]> * more notification stats * count ntf stats * update server-info --------- Co-authored-by: spaced4ndy <[email protected]>	2024-07-25 16:06:40 +01:00
Evgeny Poberezkin	9093c7b120	agent, ntf server: only mark subscriptions as pending that were created by the disconnected client (#1242 ) * ntf server: only mark subscriptions as pending if the disconnected client is current * add sessionId to subscribed queue * add sessionId to subscriptions in ntf server agent * fix	2024-07-25 13:07:28 +01:00
Evgeny Poberezkin	47ae921b86	use strict Map (#1241 )	2024-07-23 08:27:54 +01:00
spaced4ndyandEvgeny Poberezkin	c605156302	agent: getAgentSubsTotal api (#1238 ) * agent: getAgentSubsTotal api * move * export * count sessions * simplify * simplify 2 --------- Co-authored-by: Evgeny Poberezkin <[email protected]>	2024-07-22 15:43:38 +01:00
Evgeny Poberezkin	051bf38bc7	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	2024-07-22 15:42:34 +01:00
Evgeny Poberezkin	8423c636a8	rename queueId to entId (can be message queue, proxied session and data blob ID to be added) (#1240 )	2024-07-22 12:01:38 +01:00
Evgeny Poberezkin	cc5732f41f	smp server: stats for commands retrieving messages when notifications are received (#1236 ) * more stats * fix stats	2024-07-20 14:10:10 +01:00