ntf: batch ntf worker command processing (#1337)

* ntf: batch ntf worker command processing

* remove comment

* change batch size

* wip

* catch

* refactor

* refactor

* batch check

* refactor

* reschedule

* increase ntfSubCheckInterval

* first check interval

* check more statuses

* refactor, remove foldr'

* refactor 2

* refactor client

* refactor 3

* ntf server: improve support for batched commands (#1340)

* re-create ntf subscriptions on NTF AUTH errors

* name

---------

Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>

Author: spaced4ndy

src/Simplex/Messaging/Agent/Client.hs

@@ -70,7 +70,9 @@ module Simplex.Messaging.Agent.Client
     agentNtfDeleteToken,
     agentNtfEnableCron,
     agentNtfCreateSubscription,
+    agentNtfCreateSubscriptions,
     agentNtfCheckSubscription,
+    agentNtfCheckSubscriptions,
     agentNtfDeleteSubscription,
     agentXFTPDownloadChunk,
     agentXFTPNewChunk,
@@ -153,6 +155,7 @@ module Simplex.Messaging.Agent.Client
     incXFTPServerStat',
     incXFTPServerSizeStat,
     incNtfServerStat,
+    incNtfServerStat',
     AgentWorkersDetails (..),
     getAgentWorkersDetails,
     AgentWorkersSummary (..),
@@ -1730,10 +1733,34 @@ agentNtfCreateSubscription :: AgentClient -> NtfTokenId -> NtfToken -> SMPQueueN
 agentNtfCreateSubscription c tknId NtfToken {ntfServer, ntfPrivKey} smpQueue nKey =
   withNtfClient c ntfServer tknId "SNEW" $ \ntf -> ntfCreateSubscription ntf ntfPrivKey (NewNtfSub tknId smpQueue nKey)
 
-agentNtfCheckSubscription :: AgentClient -> NtfSubscriptionId -> NtfToken -> AM NtfSubStatus
-agentNtfCheckSubscription c subId NtfToken {ntfServer, ntfPrivKey} =
+agentNtfCreateSubscriptions :: AgentClient -> NtfToken -> NonEmpty (NewNtfEntity 'Subscription) -> AM' (NonEmpty (Either AgentErrorType NtfSubscriptionId))
+agentNtfCreateSubscriptions = withNtfBatch "SNEW" ntfCreateSubscriptions
+
+agentNtfCheckSubscription :: AgentClient -> NtfToken -> NtfSubscriptionId -> AM NtfSubStatus
+agentNtfCheckSubscription c NtfToken {ntfServer, ntfPrivKey} subId =
   withNtfClient c ntfServer subId "SCHK" $ \ntf -> ntfCheckSubscription ntf ntfPrivKey subId
 
+agentNtfCheckSubscriptions :: AgentClient -> NtfToken -> NonEmpty NtfSubscriptionId -> AM' (NonEmpty (Either AgentErrorType NtfSubStatus))
+agentNtfCheckSubscriptions = withNtfBatch "SCHK" ntfCheckSubscriptions
+
+-- This batch sends all commands to one ntf server (client can only use one server at a time)
+withNtfBatch ::
+  ByteString ->
+  (NtfClient -> C.APrivateAuthKey -> NonEmpty a -> IO (NonEmpty (Either NtfClientError r))) ->
+  AgentClient ->
+  NtfToken ->
+  NonEmpty a ->
+  AM' (NonEmpty (Either AgentErrorType r))
+withNtfBatch cmdStr action c NtfToken {ntfServer, ntfPrivKey} subs = do
+  let tSess = (0, ntfServer, Nothing)
+  tryAgentError' (getNtfServerClient c tSess) >>= \case
+    Left e -> pure $ L.map (\_ -> Left e) subs
+    Right ntf -> liftIO $ do
+      logServer' "-->" c ntfServer (bshow (length subs) <> " subscriptions") cmdStr
+      L.map agentError <$> action ntf ntfPrivKey subs
+      where
+        agentError = first $ protocolClientError NTF $ clientServer ntf
+
 agentNtfDeleteSubscription :: AgentClient -> NtfSubscriptionId -> NtfToken -> AM ()
 agentNtfDeleteSubscription c subId NtfToken {ntfServer, ntfPrivKey} =
   withNtfClient c ntfServer subId "SDEL" $ \ntf -> ntfDeleteSubscription ntf ntfPrivKey subId
@@ -2058,9 +2085,13 @@ incXFTPServerStat_ = incServerStat (\AgentClient {xftpServersStats = s} -> s) ne
 {-# 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
+incNtfServerStat c userId srv sel = incNtfServerStat' c userId srv sel 1
 {-# INLINE incNtfServerStat #-}
 
+incNtfServerStat' :: AgentClient -> UserId -> NtfServer -> (AgentNtfServerStats -> TVar Int) -> Int -> STM ()
+incNtfServerStat' = incServerStat (\AgentClient {ntfServersStats = s} -> s) newAgentNtfServerStats
+{-# 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

src/Simplex/Messaging/Agent/Env/SQLite.hs

@@ -150,6 +150,7 @@ data AgentConfig = AgentConfig
     deleteErrorCount :: Int,
     ntfCron :: Word16,
     ntfBatchSize :: Int,
+    ntfSubFirstCheckInterval :: NominalDiffTime,
     ntfSubCheckInterval :: NominalDiffTime,
     caCertificateFile :: FilePath,
     privateKeyFile :: FilePath,
@@ -219,8 +220,9 @@ defaultAgentConfig =
       xftpMaxRecipientsPerRequest = 200,
       deleteErrorCount = 10,
       ntfCron = 20, -- minutes
-      ntfBatchSize = 200,
-      ntfSubCheckInterval = nominalDay,
+      ntfBatchSize = 150,
+      ntfSubFirstCheckInterval = nominalDay,
+      ntfSubCheckInterval = 3 * nominalDay,
       -- CA certificate private key is not needed for initialization
       -- ! we do not generate these
       caCertificateFile = "/etc/opt/simplex-agent/ca.crt",