refactor

cabal.project

@@ -12,7 +12,7 @@ constraints: zip +disable-bzip2 +disable-zstd
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/simplexmq.git
-    tag: c7886926870e97fa592d51fa36a2cdec49296388
+    tag: 5021f78a7e231fb3764f3bc3865c6902dcaebebc
 
 source-repository-package
     type: git

scripts/nix/sha256map.nix

@@ -1,5 +1,5 @@
 {
-  "https://github.com/simplex-chat/simplexmq.git"."c7886926870e97fa592d51fa36a2cdec49296388" = "1r3nibcgw3whl0q3ssyr1606x4ilqphhzqyihi3aw4nw5fmz226h";
+  "https://github.com/simplex-chat/simplexmq.git"."5021f78a7e231fb3764f3bc3865c6902dcaebebc" = "0yrcdc07q86kaw35y3pk52chgmxhv5yc348wd2fw89p7lfqcgwwy";
   "https://github.com/simplex-chat/hs-socks.git"."a30cc7a79a08d8108316094f8f2f82a0c5e1ac51" = "0yasvnr7g91k76mjkamvzab2kvlb1g5pspjyjn2fr6v83swjhj38";
   "https://github.com/simplex-chat/direct-sqlcipher.git"."f814ee68b16a9447fbb467ccc8f29bdd3546bfd9" = "1ql13f4kfwkbaq7nygkxgw84213i0zm7c1a8hwvramayxl38dq5d";
   "https://github.com/simplex-chat/sqlcipher-simple.git"."a46bd361a19376c5211f1058908fc0ae6bf42446" = "1z0r78d8f0812kxbgsm735qf6xx8lvaz27k1a0b4a2m0sshpd5gl";

src/Simplex/Chat.hs

@@ -20,7 +20,6 @@ import Control.Applicative (optional, (<|>))
 import Control.Concurrent.STM (retry)
 import Control.Logger.Simple
 import Control.Monad
-import Simplex.Chat.Stats
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Reader
@@ -72,6 +71,7 @@ import Simplex.Chat.ProfileGenerator (generateRandomProfile)
 import Simplex.Chat.Protocol
 import Simplex.Chat.Remote
 import Simplex.Chat.Remote.Types
+import Simplex.Chat.Stats
 import Simplex.Chat.Store
 import Simplex.Chat.Store.AppSettings
 import Simplex.Chat.Store.Connections
@@ -95,7 +95,7 @@ import Simplex.FileTransfer.Protocol (FileParty (..), FilePartyI)
 import qualified Simplex.FileTransfer.Transport as XFTP
 import Simplex.FileTransfer.Types (FileErrorType (..), RcvFileId, SndFileId)
 import Simplex.Messaging.Agent as Agent
-import Simplex.Messaging.Agent.Client (AgentStatsKey (..), SubInfo (..), agentClientStore, getAgentQueuesInfo, getAgentWorkersDetails, getAgentWorkersSummary, getNetworkConfig', ipAddressProtected, withLockMap)
+import Simplex.Messaging.Agent.Client (AgentStatsKey (..), SMPServerSubs (..), SubInfo (..), agentClientStore, getAgentQueuesInfo, getAgentWorkersDetails, getAgentWorkersSummary, getNetworkConfig', ipAddressProtected, withLockMap)
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..), createAgentStore, defaultAgentConfig)
 import Simplex.Messaging.Agent.Lock (withLock)
 import Simplex.Messaging.Agent.Protocol
@@ -113,7 +113,7 @@ import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (base64P)
-import Simplex.Messaging.Protocol (AProtoServerWithAuth (..), AProtocolType (..), EntityId, ErrorType (..), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth (..), ProtocolTypeI, SProtocolType (..), SubscriptionMode (..), UserProtocol, XFTPServer, userProtocol, ProtocolServer)
+import Simplex.Messaging.Protocol (AProtoServerWithAuth (..), AProtocolType (..), EntityId, ErrorType (..), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth (..), ProtocolServer, ProtocolTypeI, SProtocolType (..), SubscriptionMode (..), UserProtocol, XFTPServer, userProtocol)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import qualified Simplex.Messaging.TMap as TM
@@ -309,12 +309,12 @@ newChatController
       agentServers :: ChatConfig -> IO InitialAgentServers
       agentServers config@ChatConfig {defaultServers = defServers@DefaultAgentServers {ntf, netCfg}} = do
         users <- withTransaction chatStore getUsers
-        smp' <- getUserServers users SPSMP
-        xftp' <- getUserServers users SPXFTP
+        smp' <- getUsersServers users SPSMP
+        xftp' <- getUsersServers users SPXFTP
         pure InitialAgentServers {smp = smp', xftp = xftp', ntf, netCfg}
         where
-          getUserServers :: forall p. (ProtocolTypeI p, UserProtocol p) => [User] -> SProtocolType p -> IO (Map UserId (NonEmpty (ProtoServerWithAuth p)))
-          getUserServers users protocol = case users of
+          getUsersServers :: forall p. (ProtocolTypeI p, UserProtocol p) => [User] -> SProtocolType p -> IO (Map UserId (NonEmpty (ProtoServerWithAuth p)))
+          getUsersServers users protocol = case users of
             [] -> pure $ M.fromList [(1, cfgServers protocol defServers)]
             _ -> M.fromList <$> initialServers
             where
@@ -1309,19 +1309,21 @@ processChatCommand' vr = \case
         withStore (\db -> Just <$> getConnectionEntity db vr user agentConnId) `catchChatError` (\e -> toView (CRChatError (Just user) e) $> Nothing)
     pure CRNtfMessages {user_, connEntity_, msgTs = msgTs', ntfMessages = map ntfMsgInfo msgs}
   APIGetUserProtoServers userId (AProtocolType p) -> withUserId userId $ \user -> withServerProtocol p $ do
-    ChatConfig {defaultServers} <- asks config
-    servers <- withStore' (`getProtocolServers` user)
-    let defServers = cfgServers p defaultServers
-        servers' = fromMaybe (L.map toServerCfg defServers) $ nonEmpty servers
-    pure $ CRUserProtoServers user $ AUPS $ UserProtoServers p servers' defServers
-    where
-      toServerCfg server = ServerCfg {server, preset = True, tested = Nothing, enabled = SEEnabled}
+    (srvCfgs, defServers) <- getUserServers user p
+    pure $ CRUserProtoServers user $ AUPS $ UserProtoServers p srvCfgs defServers
   GetUserProtoServers aProtocol -> withUser $ \User {userId} ->
     processChatCommand $ APIGetUserProtoServers userId aProtocol
-  APISetUserProtoServers userId (APSC p (ProtoServersConfig servers)) -> withUserId userId $ \user -> withServerProtocol p $ do
-    withStore $ \db -> overwriteProtocolServers db user servers
+  APISetUserProtoServers userId (APSC p (ProtoServersConfig srvsCfg)) -> withUserId userId $ \user -> withServerProtocol p $ do
+    withStore $ \db -> overwriteProtocolServers db user srvsCfg
     cfg <- asks config
-    lift $ withAgent' $ \a -> setProtocolServers a (aUserId user) $ activeAgentServers cfg p servers
+    lift $ withAgent' $ \a -> setProtocolServers a (aUserId user) $ activeAgentServers cfg p srvsCfg
+    ok user
+  APIAddKnownProtoServer userId (AProtoServerWithAuth p newSrv) -> withUserId userId $ \user -> withServerProtocol p $ do
+    (srvCfgs, _) <- getUserServers user p
+    when (protoServer newSrv `elem` map (\ServerCfg {server} -> protoServer server) (L.toList srvCfgs)) $
+      throwChatError (CECommandError "server already configured")
+    let newSrvCfg = ServerCfg {server = newSrv, preset = False, tested = Nothing, enabled = SEKnown}
+    withStore $ \db -> overwriteProtocolServers db user (L.toList srvCfgs <> [newSrvCfg])
     ok user
   SetUserProtoServers serversConfig -> withUser $ \User {userId} ->
     processChatCommand $ APISetUserProtoServers userId serversConfig
@@ -2256,18 +2258,28 @@ processChatCommand' vr = \case
   GetAgentServersSummary userId -> withUserId userId $ \user -> do
     agentServersSummary <- lift $ withAgent' getAgentServersSummary
     users <- withStore' getUsers
-    smpServers <- getUserServers user SPSMP
-    xftpServers <- getUserServers user SPXFTP
+    smpServers <- getUserSrvs user SPSMP
+    xftpServers <- getUserSrvs user SPXFTP
     let presentedServersSummary = toPresentedServersSummary agentServersSummary users user smpServers xftpServers
     pure $ CRAgentServersSummary user presentedServersSummary
     where
-      getUserServers :: forall p. (ProtocolTypeI p, UserProtocol p) => User -> SProtocolType p -> CM [ProtocolServer p]
-      getUserServers users protocol = do
-        ChatConfig {defaultServers} <- asks config
-        let defServers = cfgServers protocol defaultServers
-        servers <- map (\ServerCfg {server} -> server) <$> withStore' (`getProtocolServers` users)
-        let srvs = if null servers then L.toList defServers else servers
-        pure $ map protoServer srvs
+      getUserSrvs :: forall p. (ProtocolTypeI p, UserProtocol p) => User -> SProtocolType p -> CM [ProtocolServer p]
+      getUserSrvs user protocol = do
+        (srvCfgs, _) <- getUserServers user protocol
+        pure $ map (\ServerCfg {server} -> protoServer server) (L.toList srvCfgs)
+  ResetAgentServersStats -> withAgent resetAgentServersStats >> ok_
+  GetAgentSubsSummary userId -> withUserId userId $ \user -> do
+    agentSubsSummary <- lift $ withAgent' getAgentSubsSummary
+    users <- withStore' getUsers
+    let subs = countSubs agentSubsSummary user users
+    pure $ CRAgentSubsSummary user subs
+    where
+      countSubs agentSubsSummary user users = do
+        M.foldrWithKey' addSubs SMPServerSubs {ssActive = 0, ssPending = 0} agentSubsSummary
+        where
+          addSubs auId subs subsAcc
+            | countUserInAllStats (AgentUserId auId) user users = addSMPSubs subs subsAcc
+            | otherwise = subsAcc
   GetAgentWorkers -> lift $ CRAgentWorkersSummary <$> withAgent' getAgentWorkersSummary
   GetAgentWorkersDetails -> lift $ CRAgentWorkersDetails <$> withAgent' getAgentWorkersDetails
   GetAgentStats -> lift $ CRAgentStats . map stat <$> withAgent' getAgentStats
@@ -2845,6 +2857,16 @@ processChatCommand' vr = \case
       msgInfo <- withStore' (`getLastRcvMsgInfo` connId)
       CRQueueInfo user msgInfo <$> withAgent (`getConnectionQueueInfo` acId)
 
+getUserServers :: forall p. (ProtocolTypeI p, UserProtocol p) => User -> SProtocolType p -> CM (NonEmpty (ServerCfg p), NonEmpty (ProtoServerWithAuth p))
+getUserServers user p = do
+  ChatConfig {defaultServers} <- asks config
+  servers <- withStore' (`getProtocolServers` user)
+  let defServers = cfgServers p defaultServers
+      userServers = fromMaybe (L.map defServerCfg defServers) $ nonEmpty servers
+  pure (userServers, defServers)
+  where
+    defServerCfg server = ServerCfg {server, preset = True, tested = Nothing, enabled = SEEnabled}
+
 contactCITimed :: Contact -> CM (Maybe CITimed)
 contactCITimed ct = sndContactCITimed False ct Nothing
 
@@ -3229,10 +3251,8 @@ receiveViaCompleteFD user fileId RcvFileDescr {fileDescrText, fileDescrComplete}
       S.toList $ S.fromList $ concatMap (\FD.FileChunk {replicas} -> map (\FD.FileChunkReplica {server} -> server) replicas) chunks
     getUnknownSrvs :: [XFTPServer] -> CM [XFTPServer]
     getUnknownSrvs srvs = do
-      ChatConfig {defaultServers = DefaultAgentServers {xftp = defXftp}} <- asks config
-      storedSrvs <- map (\ServerCfg {server} -> protoServer server) <$> withStore' (`getProtocolServers` user)
-      let defXftp' = L.map protoServer defXftp
-          knownSrvs = fromMaybe defXftp' $ nonEmpty storedSrvs
+      (srvCfgs, _) <- getUserServers user SPXFTP
+      let knownSrvs = map (\ServerCfg {server} -> protoServer server) (L.toList srvCfgs)
       pure $ filter (`notElem` knownSrvs) srvs
     ipProtectedForSrvs :: [XFTPServer] -> CM Bool
     ipProtectedForSrvs srvs = do
@@ -7428,6 +7448,7 @@ chatCommandP =
       "/xftp test " *> (TestProtoServer . AProtoServerWithAuth SPXFTP <$> strP),
       "/ntf test " *> (TestProtoServer . AProtoServerWithAuth SPNTF <$> strP),
       "/_servers " *> (APISetUserProtoServers <$> A.decimal <* A.space <*> srvCfgP),
+      "/_known server " *> (APIAddKnownProtoServer <$> A.decimal <* A.space <*> strP),
       "/smp " *> (SetUserProtoServers . APSC SPSMP . ProtoServersConfig . map toServerCfg <$> protocolServersP),
       "/smp default" $> SetUserProtoServers (APSC SPSMP $ ProtoServersConfig []),
       "/xftp " *> (SetUserProtoServers . APSC SPXFTP . ProtoServersConfig . map toServerCfg <$> protocolServersP),
@@ -7627,6 +7648,8 @@ chatCommandP =
       "/debug locks" $> DebugLocks,
       "/debug event " *> (DebugEvent <$> jsonP),
       "/get servers summary " *> (GetAgentServersSummary <$> A.decimal),
+      "/reset servers stats" $> ResetAgentServersStats,
+      "/get subs summary " *> (GetAgentSubsSummary <$> A.decimal),
       "/get stats" $> GetAgentStats,
       "/reset stats" $> ResetAgentStats,
       "/get subs" $> GetAgentSubs,

src/Simplex/Chat/Controller.hs

@@ -69,7 +69,7 @@ import Simplex.Chat.Types.UITheme
 import Simplex.Chat.Util (liftIOEither)
 import Simplex.FileTransfer.Description (FileDescriptionURI)
 import Simplex.Messaging.Agent (AgentClient, SubscriptionsInfo)
-import Simplex.Messaging.Agent.Client (AgentLocks, AgentQueuesInfo (..), AgentWorkersDetails (..), AgentWorkersSummary (..), ProtocolTestFailure, UserNetworkInfo)
+import Simplex.Messaging.Agent.Client (AgentLocks, AgentQueuesInfo (..), AgentWorkersDetails (..), AgentWorkersSummary (..), ProtocolTestFailure, SMPServerSubs, UserNetworkInfo)
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, NetworkConfig)
 import Simplex.Messaging.Agent.Lock
 import Simplex.Messaging.Agent.Protocol
@@ -344,6 +344,7 @@ data ChatCommand
   | APIGetUserProtoServers UserId AProtocolType
   | GetUserProtoServers AProtocolType
   | APISetUserProtoServers UserId AProtoServersConfig
+  | APIAddKnownProtoServer UserId AProtoServerWithAuth
   | SetUserProtoServers AProtoServersConfig
   | APITestProtoServer UserId AProtoServerWithAuth
   | TestProtoServer AProtoServerWithAuth
@@ -507,6 +508,8 @@ data ChatCommand
   | DebugLocks
   | DebugEvent ChatResponse
   | GetAgentServersSummary UserId
+  | ResetAgentServersStats
+  | GetAgentSubsSummary UserId
   | GetAgentStats
   | ResetAgentStats
   | GetAgentSubs
@@ -759,6 +762,7 @@ data ChatResponse
   | CRSlowSQLQueries {chatQueries :: [SlowSQLQuery], agentQueries :: [SlowSQLQuery]}
   | CRDebugLocks {chatLockName :: Maybe String, chatEntityLocks :: Map String String, agentLocks :: AgentLocks}
   | CRAgentServersSummary {user :: User, serversSummary :: PresentedServersSummary}
+  | CRAgentSubsSummary {user :: User, subsSummary :: SMPServerSubs}
   | CRAgentStats {agentStats :: [[String]]}
   | CRAgentWorkersDetails {agentWorkersDetails :: AgentWorkersDetails}
   | CRAgentWorkersSummary {agentWorkersSummary :: AgentWorkersSummary}

src/Simplex/Chat/Stats.hs

@@ -5,7 +5,6 @@
 
 module Simplex.Chat.Stats where
 
-import Control.Applicative ((<|>))
 import qualified Data.Aeson.TH as J
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
@@ -113,8 +112,7 @@ toPresentedServersSummary agentSummary users currentUser userSMPSrvs userXFTPSrv
     }
   where
     AgentServersSummary {statsStartedAt, smpServersSessions, smpServersSubs, smpServersStats, xftpServersSessions, xftpServersStats, xftpRcvInProgress, xftpSndInProgress, xftpDelInProgress} = agentSummary
-    countUserInAll auId = auId == aUserId currentUser || auId `notElem` hiddenUserIds
-    hiddenUserIds = map aUserId $ filter (isJust . viewPwdHash) users
+    countUserInAll auId = countUserInAllStats (AgentUserId auId) currentUser users
     smpSummsIntoCategories :: Map SMPServer SMPServerSummary -> ([SMPServerSummary], [SMPServerSummary], [SMPServerSummary])
     smpSummsIntoCategories = foldr partitionSummary ([], [], [])
       where
@@ -213,12 +211,6 @@ toPresentedServersSummary agentSummary users currentUser userSMPSrvs userXFTPSrv
           ssErrors = ssErrors ss1 + ssErrors ss2,
           ssConnecting = ssConnecting ss1 + ssConnecting ss2
         }
-    addSMPSubs :: SMPServerSubs -> SMPServerSubs -> SMPServerSubs
-    addSMPSubs ss1 ss2 =
-      SMPServerSubs
-        { ssActive = ssActive ss1 + ssActive ss2,
-          ssPending = ssPending ss1 + ssPending ss2
-        }
     addSMPStats :: AgentSMPServerStatsData -> AgentSMPServerStatsData -> AgentSMPServerStatsData
     addSMPStats sd1 sd2 =
       AgentSMPServerStatsData
@@ -236,6 +228,8 @@ toPresentedServersSummary agentSummary users currentUser userSMPSrvs userXFTPSrv
           _recvDuplicates = _recvDuplicates sd1 + _recvDuplicates sd2,
           _recvCryptoErrs = _recvCryptoErrs sd1 + _recvCryptoErrs sd2,
           _recvErrs = _recvErrs sd1 + _recvErrs sd2,
+          _ackMsgs = _ackMsgs sd1 + _ackMsgs sd2,
+          _ackAttempts = _ackAttempts sd1 + _ackAttempts sd2,
           _connCreated = _connCreated sd1 + _connCreated sd2,
           _connSecured = _connSecured sd1 + _connSecured sd2,
           _connCompleted = _connCompleted sd1 + _connCompleted sd2,
@@ -259,6 +253,19 @@ toPresentedServersSummary agentSummary users currentUser userSMPSrvs userXFTPSrv
           _deleteErrs = _deleteErrs sd1 + _deleteErrs sd2
         }
 
+countUserInAllStats :: AgentUserId -> User -> [User] -> Bool
+countUserInAllStats (AgentUserId auId) currentUser users =
+  auId == aUserId currentUser || auId `notElem` hiddenUserIds
+  where
+    hiddenUserIds = map aUserId $ filter (isJust . viewPwdHash) users
+
+addSMPSubs :: SMPServerSubs -> SMPServerSubs -> SMPServerSubs
+addSMPSubs ss1 ss2 =
+  SMPServerSubs
+    { ssActive = ssActive ss1 + ssActive ss2,
+      ssPending = ssPending ss1 + ssPending ss2
+    }
+
 $(J.deriveJSON defaultJSON ''SMPServerSummary)
 
 $(J.deriveJSON defaultJSON ''XFTPServerSummary)

src/Simplex/Chat/View.hs

@@ -51,7 +51,7 @@ import Simplex.Chat.Types.Preferences
 import Simplex.Chat.Types.Shared
 import Simplex.Chat.Types.UITheme
 import qualified Simplex.FileTransfer.Transport as XFTP
-import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), SubscriptionsInfo (..))
+import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), SMPServerSubs (..), SubscriptionsInfo (..))
 import Simplex.Messaging.Agent.Env.SQLite (NetworkConfig (..))
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store.SQLite.DB (SlowQueryStats (..))
@@ -366,6 +366,7 @@ responseToView hu@(currentRH, user_) ChatConfig {logLevel, showReactions, showRe
       "agent locks: " <> viewJSON agentLocks
     ]
   CRAgentServersSummary u serversSummary -> ttyUser u ["agent servers summary: " <> viewJSON serversSummary]
+  CRAgentSubsSummary u SMPServerSubs {ssActive, ssPending} -> ttyUser u ["agent subscriptions summary: active = " <> sShow ssActive <> ", pending = " <> sShow ssPending]
   CRAgentStats stats -> map (plain . intercalate ",") stats
   CRAgentSubs {activeSubs, pendingSubs, removedSubs} ->
     [plain $ "Subscriptions: active = " <> show (sum activeSubs) <> ", pending = " <> show (sum pendingSubs) <> ", removed = " <> show (sum $ M.map length removedSubs)]