v5.5.3

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

CHANGELOG.md

@@ -1,3 +1,54 @@
+# 5.5.3
+
+Agent:
+- notification token API also returns active notifications server.
+- support file descriptions with redirection and file URIs.
+
+Servers:
+- CLI commands for online key and certificate rotation.
+- Configure config and log paths via environment variables.
+
+# 5.5.2
+
+Extensible handshake for clients and SMP/NTF servers (ignore extra data).
+
+# 5.5.1
+
+SMP servers:
+- do not keep stats file open
+- additional stats about currently stored messages
+
+Agent:
+- support multiple notification servers (only one can be used at a time).
+- expire messages after "quota exceeded" error after 7 days (instead of 21 days previously).
+- stabilize message delivery, remove unnecessary subscription retries and traffic.
+- improve database performance for message delivery.
+- fix sockets/memory leak - a very old bug "activated" by improvements in v5.5.0.
+
+# 5.5.0
+
+Code:
+- compatible with GHC 8.10.7 to support compilation for armv7a.
+- migrate to `crypton` from deprecated `cryptonite` (the seed for DRG is now sha512-hashed).
+- use ChaChaDRG for all random IDs, keys and nonces, only using hashed entropy as seed.
+- more efficient transaction batching in SMP protocol client and server.
+
+Agent:
+- stabilize message reception and delivery, migrate message delivery to database queue.
+- additional event MSGNTF confirming that message received via notification is processed.
+- efficient processing of messages sent to multiple recipients with batched database transactions.
+- new worker abstraction for all queued tasks resilient to race conditions and some database errors.
+- many fixed race conditions.
+- background mode for iOS NSE.
+- additional error reporting to client on critical errors (to be show as alert in the clients).
+- functional api to get worker statistics.
+
+SMP/XFTP servers:
+- fix socket and memory leak on servers with high load (inactive clients without subscriptions are disconnected after set time of inactivity).
+- control port improvements.
+- fix statistics for stored queues, messages and files.
+- make writing to store log atomic (fixes a rare bug in XFTP server).
+
 # 5.4.0
 
 Migrate to GHC 9.6.3

apps/ntf-server/Main.hs

@@ -1,13 +1,14 @@
 module Main where
 
 import Control.Logger.Simple
+import Simplex.Messaging.Server.CLI (getEnvPath)
 import Simplex.Messaging.Notifications.Server.Main
 
-cfgPath :: FilePath
-cfgPath = "/etc/opt/simplex-notifications"
+defaultCfgPath :: FilePath
+defaultCfgPath = "/etc/opt/simplex-notifications"
 
-logPath :: FilePath
-logPath = "/var/opt/simplex-notifications"
+defaultLogPath :: FilePath
+defaultLogPath = "/var/opt/simplex-notifications"
 
 logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
@@ -15,4 +16,6 @@ logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
 main :: IO ()
 main = do
   setLogLevel LogDebug -- change to LogError in production
+  cfgPath <- getEnvPath "NTF_SERVER_CFG_PATH" defaultCfgPath
+  logPath <- getEnvPath "NTF_SERVER_LOG_PATH" defaultLogPath
   withGlobalLogging logCfg $ ntfServerCLI cfgPath logPath

apps/smp-server/Main.hs

@@ -3,8 +3,8 @@
 module Main where
 
 import Control.Logger.Simple
+import Simplex.Messaging.Server.CLI (getEnvPath)
 import Simplex.Messaging.Server.Main
-import System.Environment
 
 defaultCfgPath :: FilePath
 defaultCfgPath = "/etc/opt/simplex"
@@ -21,6 +21,3 @@ main = do
   cfgPath <- getEnvPath "SMP_SERVER_CFG_PATH" defaultCfgPath
   logPath <- getEnvPath "SMP_SERVER_LOG_PATH" defaultLogPath
   withGlobalLogging logCfg $ smpServerCLI cfgPath logPath
-    
-getEnvPath :: String -> FilePath -> IO FilePath
-getEnvPath name def = maybe def (\case "" -> def; f -> f) <$> lookupEnv name

apps/xftp-server/Main.hs

@@ -1,13 +1,14 @@
 module Main where
 
 import Control.Logger.Simple
+import Simplex.Messaging.Server.CLI (getEnvPath)
 import Simplex.FileTransfer.Server.Main
 
-cfgPath :: FilePath
-cfgPath = "/etc/opt/simplex-xftp"
+defaultCfgPath :: FilePath
+defaultCfgPath = "/etc/opt/simplex-xftp"
 
-logPath :: FilePath
-logPath = "/var/opt/simplex-xftp"
+defaultLogPath :: FilePath
+defaultLogPath = "/var/opt/simplex-xftp"
 
 logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
@@ -15,4 +16,6 @@ logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
 main :: IO ()
 main = do
   setLogLevel LogDebug -- change to LogError in production
+  cfgPath <- getEnvPath "XFTP_SERVER_CFG_PATH" defaultCfgPath
+  logPath <- getEnvPath "XFTP_SERVER_LOG_PATH" defaultLogPath
   withGlobalLogging logCfg $ xftpServerCLI cfgPath logPath

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.5.0.4
+version: 5.5.3.0
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

protocol/xftp-uri-request.txt

@@ -0,0 +1,16 @@
+Scheme name: xftp
+
+Status: Provisional
+
+Applications/protocols that use this scheme name:
+This scheme is used for URIs of XFTP (SimpleX File Transfer Protocol) servers,
+a client-server protocol for asynchronous file transfer via relays,
+preserving file meta-data (including size and name) and content privacy.
+
+Contact: Evgeny Poberezkin <ep@simplex.chat>
+
+Change controller: Evgeny Poberezkin <ep@simplex.chat>
+
+References:
+The syntax for server URIs in the provisional specification for SimpleX File Transfer Protocol:
+https://github.com/simplex-chat/simplexmq/blob/master/rfcs/2022-12-26-simplex-file-transfer.md#server-address-syntax

protocol/xrcp-uri-request.txt

@@ -0,0 +1,15 @@
+Scheme name: xrcp
+
+Status: Provisional
+
+Applications/protocols that use this scheme name:
+This scheme is used for URIs of controller sessions via SimpleX Remote Control protocol (XRCP),
+a protocol for remote access and management of hosts via insecure network.
+
+Contact: Evgeny Poberezkin <ep@simplex.chat>
+
+Change controller: Evgeny Poberezkin <ep@simplex.chat>
+
+References:
+The syntax for server URIs in the provisional specification for SimpleX File Transfer Protocol:
+https://github.com/simplex-chat/simplexmq/blob/master/rfcs/2023-10-25-remote-control.md#session-invitation

rfcs/2022-12-26-simplex-file-transfer.md

@@ -149,6 +149,19 @@ parts:
 
 This file description is sent to all recipients via normal messages, split to 15780 byte chunks if needed.
 
+### Server address syntax
+
+The server address is a URI with the following format:
+
+```abnf
+xftpServerURI = %s"xftp://" xftpServer
+xftpServer = serverIdentity "@" srvHost [":" port]
+srvHost = <hostname> ; RFC1123, RFC5891
+port = 1*DIGIT
+serverIdentity = base64url
+base64url = <base64url encoded binary> ; RFC4648, section 5
+```
+
 ### Receiving file
 
 Having received the description, the recipient will:

rfcs/2024-01-26-file-links.md

@@ -0,0 +1,73 @@
+# Sending large file descriptions
+
+It is desirable to provide a QR code/URI from which a file can be downloaded. This way files may be addressed outside a chat client.
+Currently the `xftp` CLI tool can generate YAML file descriptions that can be used to receive a file.
+It is possible to pass such a description as an URI, but descriptions for files larger than ~8 MBs (two 4 MB chunks) would give QR codes that are difficult to process.
+A user can manually upload description and get a shorter one. Typically descriptions for files that are up to ~20 GBs would still be small enough to not require another pass, and that is way beyond any current (or, reasonable, fwiw) limitations.
+
+It is possible to streamline this process, so any application using simplexmq agent can easily send file descriptions and follow redirects.
+A file description with a redirect contains an extra field with final file size and digest so it can be followed automatically.
+
+The flow would be like this:
+
+- Sending:
+  1. Upload file as usual with `xftpSendFile`, get recipient file descriptions in `SFDONE` message.
+  2. Upload one of the file descriptions with `xftpSendDescription`, get its redirect-description in its `SFDONE` message.
+  3. Wrap in `FileDescriptionURI` and use `strEncode` to get a QR-sized URI.
+  4. Show QR code / copy link.
+- Receiving:
+  1. Scan QR code / paste link.
+  2. Use `strDecode` and unwrap `FileDescriptionURI` to get `ValidFileDescription 'FRecipient`.
+  3. Download it as usual with `xftpReceiveFile`, getting `RFDONE` message when the file is fully received.
+
+It is not necesary to use redirect description if original description can be encoded to fit in 1002 characters. Beyond this size there is a significant jump in QR code complexity.
+It is possible to call `encodeFileDescriptionURI` right after upload to test if the URI fits and skip step 2.
+When `xftpReceiveFile` receives a decoded description that lacks `redirect` field, the procedure for downloading a file is the same as usual - download chunks and reassemble local file.
+
+## Agent changes
+
+### Sending
+
+Sending and receiving files in agent is a multi-step process mediated by DB entries in `snd_files` and `rcv_files` tables.
+
+`xftpSendDescription` is tasked with storing original description in a temporary locally-encrypted file, then creating upload task for it.
+
+It is necessary to preserve redirect metadata so it can be attached to descriptions in the `SFDONE` message sent by a worker:
+
+```sql
+ALTER TABLE snd_files ADD COLUMN redirect_size INTEGER;
+ALTER TABLE snd_files ADD COLUMN redirect_digest BLOB;
+```
+
+### Receiving
+
+`xftpReceiveFile` gets a file description as an argument and knows if it should follow redirect procedure or run an ordinary download.
+For redirects it will prepare a `RcvFile` for redirect and then a placeholder, for the final file.
+Agent messages would be sent using the entity ID of the final file, which is stored along with redirect metadata in `RcvFile` for the redirect.
+
+```sql
+ALTER TABLE rcv_files ADD COLUMN redirect_id INTEGER REFERENCES rcv_files ON DELETE CASCADE; -- for later updates
+ALTER TABLE rcv_files ADD COLUMN redirect_entity_id BLOB; -- for notifications
+ALTER TABLE rcv_files ADD COLUMN redirect_size INTEGER;
+ALTER TABLE rcv_files ADD COLUMN redirect_digest BLOB;
+```
+
+These additional fields will exist on the file that is a short description to receive an actual description of the final file.
+
+While a description YAML is being downloaded, the application will get `RFPROG` messages tagged for final entity, containing bytes downloaded so far and the total size from the original file.
+When the description is fully downloaded, the worker would decode description and check if the stated size and digest match the declared in redirect.
+Then it will replace placeholder description in `rcv_files` for destination file with the actual data from downloaded description.
+Finally, instead of sending `RFDONE` for redirect, it hands over work to chunk download worker, which will run exactly as if the user requested its download directly.
+An application will then receive `RFPROG` and `RFDONE` messages as usual.
+
+## URI encoding
+
+File description URIs use the same service schema `simplex:` or its `https://simplex.chat` (or any custom host) equivalent as do contact links and can be extracted from text and processed the same way.
+The path section is `/file` (with an optional trailing `/`).
+The payload is encoded in the "fragment" part of the link, using `#/?`, followed by a query string.
+File description is encoded first in a YAML document, then URL-encoded in under the key `d`.
+An application may want to pass extra parameters not necessary to download a file. Those go in the `_` key, encoded as a JSON dictionary.
+
+An example link:
+
+`simplex:/file#/?d=chunkSize%3A%2064kb%0Adigest%3A%20OtpnXkECTW4a18Eots2m3O22maeOCMqPUX4ulugIjgMEJfCpTYc_-T257Uw7s9bW_F0G5WBg5BioBWd4Z_OoCw%3D%3D%0Akey%3A%20rNR8_2SJuH7Qve43gV3zszL0R6oY5HSdRZT_paB-wfE%3D%0Anonce%3A%202oKwfK-w75nwyWp8_1Lv6QnQonIRtJmG%0Aparty%3A%20recipient%0Areplicas%3A%0A-%20chunks%3A%0A%20%20-%201%3ATdvaxMnG2Ph1e3QCx3-rpA%3D%3D%3AMC4CAQAwBQYDK2VwBCIEILdErEICvgrBCajDLTX2h3LXyMB7z5vrtLa3XVigJuf-%3ANS46KuYdgOWs6dUeMp7p2oF8rBQ9wQ2Ez6TW6Y6gHg0%3D%0A%20%20-%202%3AH5SRbtKYrXWVXTthrkeWzw%3D%3D%3AMC4CAQAwBQYDK2VwBCIEIGeEPNLt7lUGPfplwsoJLCDFnbIc5Hm31kz5X6rWXmgu%3A7QNRI-gvFx9UM-baXp3YVDli9pcfh3HGFKDhsA9JQHY%3D%0A%20%20-%203%3A_xjukkIl9WZFryUXT0h_TQ%3D%3D%3AMC4CAQAwBQYDK2VwBCIEIIRFBaL1HvUfePvKLuggwUrC_q_ZHd7v08IL9jhM7teC%3Aid2lgLMMjTGsR8SUogJuRdLoEHAc5SDQKFDqlZRSuEY%3D%0A%20%20server%3A%20xftp%3A%2F%2FLcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI%3D%40localhost%3A7002%0Asize%3A%20192kb%0A&_=%7B%22k%22:%22test%22%7D`

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.5.0.4
+version:        5.5.3.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -101,7 +101,8 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230829_crypto_files
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231222_command_created_at
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231225_failed_work_items
-      Simplex.Messaging.Agent.TAsyncs
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240121_message_delivery_indexes
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240124_file_redirect
       Simplex.Messaging.Agent.TRcvQueues
       Simplex.Messaging.Client
       Simplex.Messaging.Client.Agent
@@ -142,6 +143,7 @@ library
       Simplex.Messaging.Server.QueueStore.STM
       Simplex.Messaging.Server.Stats
       Simplex.Messaging.Server.StoreLog
+      Simplex.Messaging.ServiceScheme
       Simplex.Messaging.TMap
       Simplex.Messaging.Transport
       Simplex.Messaging.Transport.Buffer

src/Simplex/FileTransfer/Agent.hs

@@ -7,6 +7,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
@@ -19,6 +20,7 @@ module Simplex.FileTransfer.Agent
     xftpDeleteRcvFile',
     -- Sending files
     xftpSendFile',
+    xftpSendDescription',
     deleteSndFileInternal,
     deleteSndFileRemote,
   )
@@ -44,6 +46,7 @@ import Simplex.FileTransfer.Client.Main
 import Simplex.FileTransfer.Crypto
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..), SFileParty (..))
+import qualified Simplex.FileTransfer.Protocol as XFTP
 import Simplex.FileTransfer.Transport (XFTPRcvChunkSpec (..))
 import Simplex.FileTransfer.Types
 import Simplex.FileTransfer.Util (removePath, uniqueCombine)
@@ -57,6 +60,7 @@ import Simplex.Messaging.Crypto.File (CryptoFile (..), CryptoFileArgs)
 import qualified Simplex.Messaging.Crypto.File as CF
 import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding
+import Simplex.Messaging.Encoding.String (strDecode, strEncode)
 import Simplex.Messaging.Protocol (EntityId, XFTPServer)
 import Simplex.Messaging.Util (liftError, tshow, unlessM, whenM)
 import System.FilePath (takeFileName, (</>))
@@ -97,7 +101,7 @@ closeXFTPAgent a = do
     stopWorkers workers = atomically (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)
 
 xftpReceiveFile' :: AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe CryptoFileArgs -> m RcvFileId
-xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks}) cfArgs = do
+xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks, redirect}) cfArgs = do
   g <- asks random
   prefixPath <- getPrefixPath "rcv.xftp"
   createDirectory prefixPath
@@ -107,14 +111,25 @@ xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks}) cfA
   createDirectory =<< toFSFilePath relTmpPath
   createEmptyFile =<< toFSFilePath relSavePath
   let saveFile = CryptoFile relSavePath cfArgs
-  fId <- withStore c $ \db -> createRcvFile db g userId fd relPrefixPath relTmpPath saveFile
-  forM_ chunks downloadChunk
+  fId <- case redirect of
+    Nothing -> withStore c $ \db -> createRcvFile db g userId fd relPrefixPath relTmpPath saveFile
+    Just _ -> do
+      -- prepare description paths
+      let relTmpPathRedirect = relPrefixPath </> "xftp.redirect-encrypted"
+          relSavePathRedirect = relPrefixPath </> "xftp.redirect-decrypted"
+      createDirectory =<< toFSFilePath relTmpPathRedirect
+      createEmptyFile =<< toFSFilePath relSavePathRedirect
+      cfArgsRedirect <- atomically $ CF.randomArgs g
+      let saveFileRedirect = CryptoFile relSavePathRedirect $ Just cfArgsRedirect
+      -- create download tasks
+      withStore c $ \db -> createRcvFileRedirect db g userId fd relPrefixPath relTmpPathRedirect saveFileRedirect relTmpPath saveFile
+  forM_ chunks (downloadChunk c)
   pure fId
-  where
-    downloadChunk :: AgentMonad m => FileChunk -> m ()
-    downloadChunk FileChunk {replicas = (FileChunkReplica {server} : _)} = do
-      void $ getXFTPRcvWorker True c (Just server)
-    downloadChunk _ = throwError $ INTERNAL "no replicas"
+
+downloadChunk :: AgentMonad m => AgentClient -> FileChunk -> m ()
+downloadChunk c FileChunk {replicas = (FileChunkReplica {server} : _)} = do
+  void $ getXFTPRcvWorker True c (Just server)
+downloadChunk _ _ = throwError $ INTERNAL "no replicas"
 
 getPrefixPath :: AgentMonad m => String -> m FilePath
 getPrefixPath suffix = do
@@ -172,14 +187,17 @@ runXFTPRcvWorker c srv Worker {doWork} = do
           relChunkPath = fileTmpPath </> takeFileName chunkPath
       agentXFTPDownloadChunk c userId digest replica chunkSpec
       atomically $ waitUntilForeground c
-      (complete, progress) <- withStore c $ \db -> runExceptT $ do
+      (entityId, complete, progress) <- withStore c $ \db -> runExceptT $ do
         liftIO $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId relChunkPath
-        RcvFile {size = FileSize total, chunks} <- ExceptT $ getRcvFile db rcvFileId
+        RcvFile {size = FileSize currentSize, chunks, redirect} <- ExceptT $ getRcvFile db rcvFileId
         let rcvd = receivedSize chunks
             complete = all chunkReceived chunks
+            (entityId, total) = case redirect of
+              Nothing -> (rcvFileEntityId, currentSize)
+              Just RcvFileRedirect {redirectFileInfo = RedirectFileInfo {size = FileSize finalSize}, redirectEntityId} -> (redirectEntityId, finalSize)
         liftIO . when complete $ updateRcvFileStatus db rcvFileId RFSReceived
-        pure (complete, RFPROG rcvd total)
-      notify c rcvFileEntityId progress
+        pure (entityId, complete, RFPROG rcvd total)
+      notify c entityId progress
       when complete . void $
         getXFTPRcvWorker True c Nothing
       where
@@ -223,7 +241,7 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
         \f@RcvFile {rcvFileId, rcvFileEntityId, tmpPath} ->
           decryptFile f `catchAgentError` (rcvWorkerInternalError c rcvFileId rcvFileEntityId tmpPath . show)
     decryptFile :: RcvFile -> m ()
-    decryptFile RcvFile {rcvFileId, rcvFileEntityId, key, nonce, tmpPath, saveFile, status, chunks} = do
+    decryptFile RcvFile {rcvFileId, rcvFileEntityId, size, digest, key, nonce, tmpPath, saveFile, status, chunks, redirect} = do
       let CryptoFile savePath cfArgs = saveFile
       fsSavePath <- toFSFilePath savePath
       when (status == RFSDecrypting) $
@@ -231,12 +249,33 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
       withStore' c $ \db -> updateRcvFileStatus db rcvFileId RFSDecrypting
       chunkPaths <- getChunkPaths chunks
       encSize <- liftIO $ foldM (\s path -> (s +) . fromIntegral <$> getFileSize path) 0 chunkPaths
+      when (FileSize encSize /= size) $ throwError $ XFTP XFTP.SIZE
+      encDigest <- liftIO $ LC.sha512Hash <$> readChunks chunkPaths
+      when (FileDigest encDigest /= digest) $ throwError $ XFTP XFTP.DIGEST
       let destFile = CryptoFile fsSavePath cfArgs
       void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure destFile
-      notify c rcvFileEntityId $ RFDONE fsSavePath
-      forM_ tmpPath (removePath <=< toFSFilePath)
-      atomically $ waitUntilForeground c
-      withStore' c (`updateRcvFileComplete` rcvFileId)
+      case redirect of
+        Nothing -> do
+          notify c rcvFileEntityId $ RFDONE fsSavePath
+          forM_ tmpPath (removePath <=< toFSFilePath)
+          atomically $ waitUntilForeground c
+          withStore' c (`updateRcvFileComplete` rcvFileId)
+        Just RcvFileRedirect {redirectFileInfo, redirectDbId} -> do
+          let RedirectFileInfo {size = redirectSize, digest = redirectDigest} = redirectFileInfo
+          forM_ tmpPath (removePath <=< toFSFilePath)
+          atomically $ waitUntilForeground c
+          withStore' c (`updateRcvFileComplete` rcvFileId)
+          -- proceed with redirect
+          yaml <- liftError (INTERNAL . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `finally` (toFSFilePath fsSavePath >>= removePath)
+          next@FileDescription {chunks = nextChunks} <- case strDecode (LB.toStrict yaml) of
+            Left _ -> throwError . XFTP $ XFTP.REDIRECT "decode error"
+            Right (ValidFileDescription fd@FileDescription {size = dstSize, digest = dstDigest})
+              | dstSize /= redirectSize -> throwError . XFTP $ XFTP.REDIRECT "size mismatch"
+              | dstDigest /= redirectDigest -> throwError . XFTP $ XFTP.REDIRECT "digest mismatch"
+              | otherwise -> pure fd
+          -- register and download chunks from the actual file
+          withStore c $ \db -> updateRcvFileRedirect db redirectDbId next
+          forM_ nextChunks (downloadChunk c)
       where
         getChunkPaths :: [RcvFileChunk] -> m [FilePath]
         getChunkPaths [] = pure []
@@ -268,7 +307,23 @@ xftpSendFile' c userId file numRecipients = do
   key <- atomically $ C.randomSbKey g
   nonce <- atomically $ C.randomCbNonce g
   -- saving absolute filePath will not allow to restore file encryption after app update, but it's a short window
-  fId <- withStore c $ \db -> createSndFile db g userId file numRecipients relPrefixPath key nonce
+  fId <- withStore c $ \db -> createSndFile db g userId file numRecipients relPrefixPath key nonce Nothing
+  void $ getXFTPSndWorker True c Nothing
+  pure fId
+
+xftpSendDescription' :: forall m. AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Int -> m SndFileId
+xftpSendDescription' c userId (ValidFileDescription fdDirect@FileDescription {size, digest}) numRecipients = do
+  g <- asks random
+  prefixPath <- getPrefixPath "snd.xftp"
+  createDirectory prefixPath
+  let relPrefixPath = takeFileName prefixPath
+  let directYaml = prefixPath </> "direct.yaml"
+  cfArgs <- atomically $ CF.randomArgs g
+  let file = CryptoFile directYaml (Just cfArgs)
+  liftError (INTERNAL . show) $ CF.writeFile file (LB.fromStrict $ strEncode fdDirect)
+  key <- atomically $ C.randomSbKey g
+  nonce <- atomically $ C.randomCbNonce g
+  fId <- withStore c $ \db -> createSndFile db g userId file numRecipients relPrefixPath key nonce $ Just RedirectFileInfo {size, digest}
   void $ getXFTPSndWorker True c Nothing
   pure fId
 
@@ -423,15 +478,15 @@ runXFTPSndWorker c srv Worker {doWork} = do
         sndFileToDescrs :: SndFile -> m (ValidFileDescription 'FSender, [ValidFileDescription 'FRecipient])
         sndFileToDescrs SndFile {digest = Nothing} = throwError $ INTERNAL "snd file has no digest"
         sndFileToDescrs SndFile {chunks = []} = throwError $ INTERNAL "snd file has no chunks"
-        sndFileToDescrs SndFile {digest = Just digest, key, nonce, chunks = chunks@(fstChunk : _)} = do
+        sndFileToDescrs SndFile {digest = Just digest, key, nonce, chunks = chunks@(fstChunk : _), redirect} = do
           let chunkSize = FileSize $ sndChunkSize fstChunk
               size = FileSize $ sum $ map (fromIntegral . sndChunkSize) chunks
           -- snd description
           sndDescrChunks <- mapM toSndDescrChunk chunks
-          let fdSnd = FileDescription {party = SFSender, size, digest, key, nonce, chunkSize, chunks = sndDescrChunks}
+          let fdSnd = FileDescription {party = SFSender, size, digest, key, nonce, chunkSize, chunks = sndDescrChunks, redirect = Nothing}
           validFdSnd <- either (throwError . INTERNAL) pure $ validateFileDescription fdSnd
           -- rcv descriptions
-          let fdRcv = FileDescription {party = SFRecipient, size, digest, key, nonce, chunkSize, chunks = []}
+          let fdRcv = FileDescription {party = SFRecipient, size, digest, key, nonce, chunkSize, chunks = [], redirect}
               fdRcvs = createRcvFileDescriptions fdRcv chunks
           validFdRcvs <- either (throwError . INTERNAL) pure $ mapM validateFileDescription fdRcvs
           pure (validFdSnd, validFdRcvs)

src/Simplex/FileTransfer/Client/Main.hs

@@ -15,6 +15,7 @@ module Simplex.FileTransfer.Client.Main
     CLIError (..),
     xftpClientCLI,
     cliSendFile,
+    cliSendFileOpts,
     prepareChunkSizes,
     prepareChunkSpecs,
     maxFileSize,
@@ -297,8 +298,8 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
       withExceptT (CLIError . show) $ encryptFile srcFile fileHdr key nonce fileSize' encSize encPath
       digest <- liftIO $ LC.sha512Hash <$> LB.readFile encPath
       let chunkSpecs = prepareChunkSpecs encPath chunkSizes
-          fdRcv = FileDescription {party = SFRecipient, size = FileSize encSize, digest = FileDigest digest, key, nonce, chunkSize = FileSize defChunkSize, chunks = []}
-          fdSnd = FileDescription {party = SFSender, size = FileSize encSize, digest = FileDigest digest, key, nonce, chunkSize = FileSize defChunkSize, chunks = []}
+          fdRcv = FileDescription {party = SFRecipient, size = FileSize encSize, digest = FileDigest digest, key, nonce, chunkSize = FileSize defChunkSize, chunks = [], redirect = Nothing}
+          fdSnd = FileDescription {party = SFSender, size = FileSize encSize, digest = FileDigest digest, key, nonce, chunkSize = FileSize defChunkSize, chunks = [], redirect = Nothing}
       logInfo $ "encrypted file to " <> tshow encPath
       pure (encPath, fdRcv, fdSnd, chunkSpecs, encSize)
     uploadFile :: TVar ChaChaDRG -> [XFTPChunkSpec] -> TVar [Int64] -> Int64 -> ExceptT CLIError IO [SentFileChunk]
@@ -526,9 +527,8 @@ prepareChunkSizes size' = prepareSizes size'
   where
     (smallSize, bigSize)
       | size' > size34 chunkSize3 = (chunkSize2, chunkSize3)
-      | otherwise = (chunkSize1, chunkSize2)
-    --  | size' > size34 chunkSize2 = (chunkSize1, chunkSize2)
-    --  | otherwise = (chunkSize0, chunkSize1)
+      | size' > size34 chunkSize2 = (chunkSize1, chunkSize2)
+      | otherwise = (chunkSize0, chunkSize1)
     size34 sz = (fromIntegral sz * 3) `div` 4
     prepareSizes 0 = []
     prepareSizes size

src/Simplex/FileTransfer/Description.hs

@@ -12,6 +12,7 @@
 
 module Simplex.FileTransfer.Description
   ( FileDescription (..),
+    RedirectFileInfo (..),
     AFileDescription (..),
     ValidFileDescription, -- constructor is not exported, use pattern
     pattern ValidFileDescription,
@@ -30,12 +31,17 @@ module Simplex.FileTransfer.Description
     kb,
     mb,
     gb,
+    FileDescriptionURI (..),
+    FileClientData,
+    fileDescriptionURI,
+    qrSizeLimit,
   )
 where
 
 import Control.Applicative (optional)
 import Control.Monad ((<=<))
 import Data.Aeson (FromJSON, ToJSON)
+import qualified Data.Aeson as J
 import qualified Data.Aeson.TH as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
@@ -50,17 +56,21 @@ import Data.Map (Map)
 import qualified Data.Map as M
 import Data.Maybe (fromMaybe)
 import Data.String
+import Data.Text (Text)
+import Data.Text.Encoding (encodeUtf8)
 import Data.Word (Word32)
 import qualified Data.Yaml as Y
 import Database.SQLite.Simple.FromField (FromField (..))
 import Database.SQLite.Simple.ToField (ToField (..))
 import Simplex.FileTransfer.Chunks
 import Simplex.FileTransfer.Protocol
+import Simplex.Messaging.Agent.QueryString
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (defaultJSON, parseAll)
 import Simplex.Messaging.Protocol (XFTPServer)
-import Simplex.Messaging.Util (bshow, (<$?>))
+import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
+import Simplex.Messaging.Util (bshow, safeDecodeUtf8, (<$?>))
 
 data FileDescription (p :: FileParty) = FileDescription
   { party :: SFileParty p,
@@ -69,7 +79,14 @@ data FileDescription (p :: FileParty) = FileDescription
     key :: C.SbKey,
     nonce :: C.CbNonce,
     chunkSize :: FileSize Word32,
-    chunks :: [FileChunk]
+    chunks :: [FileChunk],
+    redirect :: Maybe RedirectFileInfo
+  }
+  deriving (Eq, Show)
+
+data RedirectFileInfo = RedirectFileInfo
+  { size :: FileSize Int64,
+    digest :: FileDigest
   }
   deriving (Eq, Show)
 
@@ -147,7 +164,8 @@ data YAMLFileDescription = YAMLFileDescription
     key :: C.SbKey,
     nonce :: C.CbNonce,
     chunkSize :: String,
-    replicas :: [YAMLServerReplicas]
+    replicas :: [YAMLServerReplicas],
+    redirect :: Maybe RedirectFileInfo
   }
   deriving (Eq, Show)
 
@@ -170,8 +188,16 @@ data FileServerReplica = FileServerReplica
 newtype FileSize a = FileSize {unFileSize :: a}
   deriving (Eq, Show)
 
+instance FromJSON a => FromJSON (FileSize a) where
+  parseJSON v = FileSize <$> Y.parseJSON v
+
+instance ToJSON a => ToJSON (FileSize a) where
+  toJSON = Y.toJSON . unFileSize
+
 $(J.deriveJSON defaultJSON ''YAMLServerReplicas)
 
+$(J.deriveJSON defaultJSON ''RedirectFileInfo)
+
 $(J.deriveJSON defaultJSON ''YAMLFileDescription)
 
 instance FilePartyI p => StrEncoding (ValidFileDescription p) where
@@ -204,7 +230,7 @@ validateFileDescription fd@FileDescription {size, chunks}
     chunksSize = fromIntegral . foldl' (\s FileChunk {chunkSize} -> s + unFileSize chunkSize) 0
 
 encodeFileDescription :: FileDescription p -> YAMLFileDescription
-encodeFileDescription FileDescription {party, size, digest, key, nonce, chunkSize, chunks} =
+encodeFileDescription FileDescription {party, size, digest, key, nonce, chunkSize, chunks, redirect} =
   YAMLFileDescription
     { party = toFileParty party,
       size = B.unpack $ strEncode size,
@@ -212,9 +238,39 @@ encodeFileDescription FileDescription {party, size, digest, key, nonce, chunkSiz
       key,
       nonce,
       chunkSize = B.unpack $ strEncode chunkSize,
-      replicas = encodeFileReplicas chunkSize chunks
+      replicas = encodeFileReplicas chunkSize chunks,
+      redirect
     }
 
+data FileDescriptionURI = FileDescriptionURI
+  { scheme :: ServiceScheme,
+    description :: ValidFileDescription 'FRecipient,
+    clientData :: Maybe FileClientData -- JSON-encoded extensions to pass in a link
+  }
+  deriving (Eq, Show)
+
+type FileClientData = Text
+
+fileDescriptionURI :: ValidFileDescription 'FRecipient -> FileDescriptionURI
+fileDescriptionURI vfd = FileDescriptionURI SSSimplex vfd mempty
+
+instance StrEncoding FileDescriptionURI where
+  strEncode FileDescriptionURI {scheme, description, clientData} = mconcat [strEncode scheme, "/file", "#/?", queryStr]
+    where
+      queryStr = strEncode $ QSP QEscape qs
+      qs = ("desc", strEncode description) : maybe [] (\cd -> [("data", encodeUtf8 cd)]) clientData
+  strP = do
+    scheme <- strP
+    _ <- "/file" <* optional (A.char '/') <* "#/?"
+    query <- strP
+    description <- queryParam "desc" query
+    let clientData = safeDecodeUtf8 <$> queryParamStr "data" query
+    pure FileDescriptionURI {scheme, description, clientData}
+
+-- | URL length in QR code before jumping up to a next size.
+qrSizeLimit :: Int
+qrSizeLimit = 1002 -- ~2 chunks in URLencoded YAML with some spare size for server hosts
+
 instance (Integral a, Show a) => StrEncoding (FileSize a) where
   strEncode (FileSize b)
     | b' /= 0 = bshow b
@@ -285,13 +341,13 @@ unfoldChunksToReplicas defChunkSize = concatMap chunkReplicas
        in FileServerReplica {chunkNo, server, replicaId, replicaKey, digest = digest', chunkSize = chunkSize'}
 
 decodeFileDescription :: YAMLFileDescription -> Either String AFileDescription
-decodeFileDescription YAMLFileDescription {party, size, digest, key, nonce, chunkSize, replicas} = do
+decodeFileDescription YAMLFileDescription {party, size, digest, key, nonce, chunkSize, replicas, redirect} = do
   size' <- strDecode $ B.pack size
   chunkSize' <- strDecode $ B.pack chunkSize
   replicas' <- decodeFileParts replicas
   chunks <- foldReplicasToChunks chunkSize' replicas'
   pure $ case aFileParty party of
-    AFP party' -> AFD FileDescription {party = party', size = size', digest, key, nonce, chunkSize = chunkSize', chunks}
+    AFP party' -> AFD FileDescription {party = party', size = size', digest, key, nonce, chunkSize = chunkSize', chunks, redirect}
   where
     decodeFileParts = fmap concat . mapM decodeYAMLServerReplicas
 

src/Simplex/FileTransfer/Protocol.hs

@@ -48,8 +48,7 @@ import Simplex.Messaging.Protocol
     encodeTransmission,
     messageTagP,
     tDecodeParseValidate,
-    tEncode,
-    tEncodeBatch,
+    tEncodeBatch1,
     tParse,
   )
 import Simplex.Messaging.Transport (SessionId, TransportError (..))
@@ -339,6 +338,8 @@ data XFTPErrorType
     HAS_FILE
   | -- | file IO error
     FILE_IO
+  | -- | bad redirect data
+    REDIRECT {redirectError :: String}
   | -- | internal server error
     INTERNAL
   | -- | used internally, never returned by the server (to be removed)
@@ -348,8 +349,12 @@ data XFTPErrorType
 instance StrEncoding XFTPErrorType where
   strEncode = \case
     CMD e -> "CMD " <> bshow e
+    REDIRECT e -> "REDIRECT " <> bshow e
     e -> bshow e
-  strP = "CMD " *> (CMD <$> parseRead1) <|> parseRead1
+  strP =
+    "CMD " *> (CMD <$> parseRead1)
+      <|> "REDIRECT " *> (REDIRECT <$> parseRead A.takeByteString)
+      <|> parseRead1
 
 instance Encoding XFTPErrorType where
   smpEncode = \case
@@ -364,6 +369,7 @@ instance Encoding XFTPErrorType where
     NO_FILE -> "NO_FILE"
     HAS_FILE -> "HAS_FILE"
     FILE_IO -> "FILE_IO"
+    REDIRECT err -> "REDIRECT " <> smpEncode err
     INTERNAL -> "INTERNAL"
     DUPLICATE_ -> "DUPLICATE_"
 
@@ -380,6 +386,7 @@ instance Encoding XFTPErrorType where
       "NO_FILE" -> pure NO_FILE
       "HAS_FILE" -> pure HAS_FILE
       "FILE_IO" -> pure FILE_IO
+      "REDIRECT" -> REDIRECT <$> _smpP
       "INTERNAL" -> pure INTERNAL
       "DUPLICATE_" -> pure DUPLICATE_
       _ -> fail "bad error type"
@@ -403,10 +410,8 @@ xftpEncodeTransmission sessionId pKey (corrId, fId, msg) = do
     signTransmission t = ((`C.sign` t) <$> pKey, t)
 
 -- this function uses batch syntax but puts only one transmission in the batch
-xftpEncodeBatch1 :: (Maybe C.ASignature, ByteString) -> Either TransportError ByteString
-xftpEncodeBatch1 (sig, t) =
-  let t' = tEncodeBatch 1 . smpEncode . Large $ tEncode (sig, t)
-   in first (const TELargeMsg) $ C.pad t' xftpBlockSize
+xftpEncodeBatch1 :: SentRawTransmission -> Either TransportError ByteString
+xftpEncodeBatch1 t = first (const TELargeMsg) $ C.pad (tEncodeBatch1 t) xftpBlockSize
 
 xftpDecodeTransmission :: ProtocolEncoding e c => SessionId -> ByteString -> Either XFTPErrorType (SignedTransmission e c)
 xftpDecodeTransmission sessionId t = do

src/Simplex/FileTransfer/Server/Main.hs

@@ -33,7 +33,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 xftpServerVersion :: String
-xftpServerVersion = "1.2.0.4"
+xftpServerVersion = "1.2.3.0"
 
 xftpServerCLI :: FilePath -> FilePath -> IO ()
 xftpServerCLI cfgPath logPath = do
@@ -42,6 +42,10 @@ xftpServerCLI cfgPath logPath = do
       doesFileExist iniFile >>= \case
         True -> exitError $ "Error: server is already initialized (" <> iniFile <> " exists).\nRun `" <> executableName <> " start`."
         _ -> initializeServer opts
+    OnlineCert certOpts ->
+      doesFileExist iniFile >>= \case
+        True -> genOnline cfgPath certOpts
+        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
     Start ->
       doesFileExist iniFile >>= \case
         True -> readIniFile iniFile >>= either exitError runServer
@@ -179,6 +183,7 @@ xftpServerCLI cfgPath logPath = do
 
 data CliCommand
   = Init InitOptions
+  | OnlineCert CertOptions
   | Start
   | Delete
 
@@ -196,6 +201,7 @@ cliCommandP :: FilePath -> FilePath -> FilePath -> Parser CliCommand
 cliCommandP cfgPath logPath iniFile =
   hsubparser
     ( command "init" (info (Init <$> initP) (progDesc $ "Initialize server - creates " <> cfgPath <> " and " <> logPath <> " directories and configuration files"))
+        <> command "cert" (info (OnlineCert <$> certOptionsP) (progDesc $ "Generate new online TLS server credentials (configuration: " <> iniFile <> ")"))
         <> command "start" (info (pure Start) (progDesc $ "Start server (configuration: " <> iniFile <> ")"))
         <> command "delete" (info (pure Delete) (progDesc "Delete configuration and log files"))
     )

src/Simplex/FileTransfer/Types.hs

@@ -47,6 +47,7 @@ data RcvFile = RcvFile
     key :: C.SbKey,
     nonce :: C.CbNonce,
     chunkSize :: FileSize Word32,
+    redirect :: Maybe RcvFileRedirect,
     chunks :: [RcvFileChunk],
     prefixPath :: FilePath,
     tmpPath :: Maybe FilePath,
@@ -108,6 +109,13 @@ data RcvFileChunkReplica = RcvFileChunkReplica
   }
   deriving (Eq, Show)
 
+data RcvFileRedirect = RcvFileRedirect
+  { redirectDbId :: DBRcvFileId,
+    redirectEntityId :: RcvFileId,
+    redirectFileInfo :: RedirectFileInfo
+  }
+  deriving (Eq, Show)
+
 -- Sending files
 
 type DBSndFileId = Int64
@@ -124,7 +132,8 @@ data SndFile = SndFile
     srcFile :: CryptoFile,
     prefixPath :: Maybe FilePath,
     status :: SndFileStatus,
-    deleted :: Bool
+    deleted :: Bool,
+    redirect :: Maybe RedirectFileInfo
   }
   deriving (Eq, Show)
 

src/Simplex/Messaging/Agent.hs

@@ -38,6 +38,7 @@ module Simplex.Messaging.Agent
     AgentErrorMonad,
     SubscriptionsInfo (..),
     getSMPAgentClient,
+    getSMPAgentClient_,
     disconnectAgentClient,
     resumeAgentClient,
     withConnLock,
@@ -92,6 +93,7 @@ module Simplex.Messaging.Agent
     xftpReceiveFile,
     xftpDeleteRcvFile,
     xftpSendFile,
+    xftpSendDescription,
     xftpDeleteSndFileInternal,
     xftpDeleteSndFileRemote,
     rcNewHostPairing,
@@ -136,7 +138,7 @@ import qualified Data.Text as T
 import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
 import Data.Word (Word16)
-import Simplex.FileTransfer.Agent (closeXFTPAgent, deleteSndFileInternal, deleteSndFileRemote, startXFTPWorkers, toFSFilePath, xftpDeleteRcvFile', xftpReceiveFile', xftpSendFile')
+import Simplex.FileTransfer.Agent (closeXFTPAgent, deleteSndFileInternal, deleteSndFileRemote, startXFTPWorkers, toFSFilePath, xftpDeleteRcvFile', xftpReceiveFile', xftpSendDescription', xftpSendFile')
 import Simplex.FileTransfer.Description (ValidFileDescription)
 import Simplex.FileTransfer.Protocol (FileParty (..))
 import Simplex.FileTransfer.Util (removePath)
@@ -162,6 +164,7 @@ import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Parsers (parse)
 import Simplex.Messaging.Protocol (BrokerMsg, EntityId, ErrorType (AUTH), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth, ProtocolTypeI (..), SMPMsgMeta, SProtocolType (..), SndPublicVerifyKey, SubscriptionMode (..), UserProtocol, XFTPServerWithAuth)
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
@@ -176,11 +179,15 @@ import UnliftIO.STM
 
 -- | Creates an SMP agent client instance
 getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
-getSMPAgentClient cfg initServers store backgroundMode =
+getSMPAgentClient = getSMPAgentClient_ 1
+{-# INLINE getSMPAgentClient #-}
+
+getSMPAgentClient_ :: (MonadRandom m, MonadUnliftIO m) => Int -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
+getSMPAgentClient_ clientId cfg initServers store backgroundMode =
   liftIO (newSMPAgentEnv cfg store) >>= runReaderT runAgent
   where
     runAgent = do
-      c <- getAgentClient initServers
+      c <- getAgentClient clientId initServers
       void $ runAgentThreads c `forkFinally` const (disconnectAgentClient c)
       pure c
     runAgentThreads c
@@ -370,7 +377,7 @@ checkNtfToken c = withAgentEnv c . checkNtfToken' c
 deleteNtfToken :: AgentErrorMonad m => AgentClient -> DeviceToken -> m ()
 deleteNtfToken c = withAgentEnv c . deleteNtfToken' c
 
-getNtfToken :: AgentErrorMonad m => AgentClient -> m (DeviceToken, NtfTknStatus, NotificationsMode)
+getNtfToken :: AgentErrorMonad m => AgentClient -> m (DeviceToken, NtfTknStatus, NotificationsMode, NtfServer)
 getNtfToken c = withAgentEnv c $ getNtfToken' c
 
 getNtfTokenData :: AgentErrorMonad m => AgentClient -> m NtfToken
@@ -395,6 +402,10 @@ xftpDeleteRcvFile c = withAgentEnv c . xftpDeleteRcvFile' c
 xftpSendFile :: AgentErrorMonad m => AgentClient -> UserId -> CryptoFile -> Int -> m SndFileId
 xftpSendFile c = withAgentEnv c .:. xftpSendFile' c
 
+-- | Send XFTP file
+xftpSendDescription :: AgentErrorMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Int -> m SndFileId
+xftpSendDescription c = withAgentEnv c .:. xftpSendDescription' c
+
 -- | Delete XFTP snd file internally (deletes work files from file system and db records)
 xftpDeleteSndFileInternal :: AgentErrorMonad m => AgentClient -> SndFileId -> m ()
 xftpDeleteSndFileInternal c = withAgentEnv c . deleteSndFileInternal c
@@ -461,8 +472,9 @@ withAgentEnv :: AgentClient -> ReaderT Env m a -> m a
 withAgentEnv c = (`runReaderT` agentEnv c)
 
 -- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's.
-getAgentClient :: AgentMonad' m => InitialAgentServers -> m AgentClient
-getAgentClient initServers = ask >>= atomically . newAgentClient initServers
+getAgentClient :: AgentMonad' m => Int -> InitialAgentServers -> m AgentClient
+getAgentClient clientId initServers = ask >>= atomically . newAgentClient clientId initServers
+{-# INLINE getAgentClient #-}
 
 logConnection :: MonadUnliftIO m => AgentClient -> Bool -> m ()
 logConnection c connected =
@@ -637,7 +649,7 @@ newRcvConnSrv c userId connId enableNtfs cMode clientData subMode srv = do
   when enableNtfs $ do
     ns <- asks ntfSupervisor
     atomically $ sendNtfSubCommand ns (connId, NSCCreate)
-  let crData = ConnReqUriData CRSSimplex smpAgentVRange [qUri] clientData
+  let crData = ConnReqUriData SSSimplex smpAgentVRange [qUri] clientData
   case cMode of
     SCMContact -> pure (connId, CRContactUri crData)
     SCMInvitation -> do
@@ -1136,7 +1148,7 @@ submitPendingMsg c cData sq = do
 
 runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> (Worker, TMVar ()) -> m ()
 runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, duplexHandshake} sq (Worker {doWork}, qLock) = do
-  ri <- asks $ messageRetryInterval . config
+  AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config
   forever $ do
     atomically $ endAgentOperation c AOSndNetwork
     waitForWork doWork
@@ -1160,7 +1172,9 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                 SMP SMP.QUOTA -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
-                  _ -> retrySndMsg RISlow
+                  _ -> do
+                    expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime
+                    if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RISlow
                 SMP SMP.AUTH -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
@@ -1169,8 +1183,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                     -- in duplexHandshake mode (v2) HELLO is only sent once, without retrying,
                     -- because the queue must be secured by the time the confirmation or the first HELLO is received
                     | duplexHandshake == Just True -> connErr
-                    | otherwise ->
-                        ifM (msgExpired helloTimeout) connErr (retrySndMsg RIFast)
+                    -- otherwise branch is not used in clients with v2+ of agent protocol (since June 2022)
+                    -- TODO remove in v6
+                    | otherwise -> do
+                        expireTs <- addUTCTime (-helloTimeout) <$> liftIO getCurrentTime
+                        if internalTs < expireTs then connErr else retrySndMsg RIFast
                     where
                       connErr = case rq_ of
                         -- party initiating connection
@@ -1190,14 +1207,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                   -- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server),
                   -- the message sending would be retried
                   | temporaryOrHostError e -> do
-                      let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
-                      ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg RIFast)
+                      let msgTimeout = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
+                      expireTs <- addUTCTime (-msgTimeout) <$> liftIO getCurrentTime
+                      if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RIFast
                   | otherwise -> notifyDel msgId err
               where
-                msgExpired timeoutSel = do
-                  msgTimeout <- asks $ timeoutSel . config
-                  currentTime <- liftIO getCurrentTime
-                  pure $ diffUTCTime currentTime internalTs > msgTimeout
                 retrySndMsg riMode = do
                   withStore' c $ \db -> updatePendingMsgRIState db connId msgId riState
                   retrySndOp c $ loop riMode
@@ -1273,6 +1287,13 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                     when (isJust rq_) $ removeConfirmations db connId
                   unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO
   where
+    notifyDelMsgs :: InternalId -> AgentErrorType -> UTCTime -> m ()
+    notifyDelMsgs msgId err expireTs = do
+      notifyDel msgId $ MERR (unId msgId) err
+      msgIds_ <- withStore' c $ \db -> getExpiredSndMessages db connId sq expireTs
+      forM_ (L.nonEmpty msgIds_) $ \msgIds -> do
+        notify $ MERRS (L.map unId msgIds) err
+        withStore' c $ \db -> forM_ msgIds $ \msgId' -> deleteSndMsgDelivery db connId sq msgId' False `catchAll_` pure ()
     delMsg :: InternalId -> m ()
     delMsg = delMsgKeep False
     delMsgKeep :: Bool -> InternalId -> m ()
@@ -1670,10 +1691,10 @@ deleteNtfToken' c deviceToken =
       deleteNtfSubs c NSCSmpDelete
     _ -> throwError $ CMD PROHIBITED
 
-getNtfToken' :: AgentMonad m => AgentClient -> m (DeviceToken, NtfTknStatus, NotificationsMode)
+getNtfToken' :: AgentMonad m => AgentClient -> m (DeviceToken, NtfTknStatus, NotificationsMode, NtfServer)
 getNtfToken' c =
   withStore' c getSavedNtfToken >>= \case
-    Just NtfToken {deviceToken, ntfTknStatus, ntfMode} -> pure (deviceToken, ntfTknStatus, ntfMode)
+    Just NtfToken {deviceToken, ntfTknStatus, ntfMode, ntfServer} -> pure (deviceToken, ntfTknStatus, ntfMode, ntfServer)
     _ -> throwError $ CMD PROHIBITED
 
 getNtfTokenData' :: AgentMonad m => AgentClient -> m NtfToken
@@ -1792,12 +1813,11 @@ getAgentMigrations' :: AgentMonad m => AgentClient -> m [UpMigration]
 getAgentMigrations' c = map upMigration <$> withStore' c (Migrations.getCurrent . DB.conn)
 
 debugAgentLocks' :: AgentMonad' m => AgentClient -> m AgentLocks
-debugAgentLocks' AgentClient {connLocks = cs, invLocks = is, reconnectLocks = rs, deleteLock = d} = do
+debugAgentLocks' AgentClient {connLocks = cs, invLocks = is, deleteLock = d} = do
   connLocks <- getLocks cs
   invLocks <- getLocks is
-  srvLocks <- getLocks rs
   delLock <- atomically $ tryReadTMVar d
-  pure AgentLocks {connLocks, invLocks, srvLocks, delLock}
+  pure AgentLocks {connLocks, invLocks, delLock}
   where
     getLocks ls = atomically $ M.mapKeys (B.unpack . strEncode) . M.mapMaybe id <$> (mapM tryReadTMVar =<< readTVar ls)
 
@@ -2042,7 +2062,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
               handleNotifyAck :: m () -> m ()
               handleNotifyAck m = m `catchAgentError` \e -> notify (ERR e) >> ack
           SMP.END ->
-            atomically (TM.lookup tSess smpClients $>>= tryReadTMVar >>= processEND)
+            atomically (TM.lookup tSess smpClients $>>= (tryReadTMVar . sessionVar) >>= processEND)
               >>= logServer "<--" c srv rId
             where
               processEND = \case

src/Simplex/Messaging/Agent/Client.hs

@@ -78,6 +78,7 @@ module Simplex.Messaging.Agent.Client
     agentDRG,
     getAgentSubscriptions,
     Worker (..),
+    SessionVar (..),
     SubscriptionsInfo (..),
     SubInfo (..),
     AgentOperation (..),
@@ -116,6 +117,10 @@ module Simplex.Messaging.Agent.Client
     getNextServer,
     withUserServers,
     withNextSrv,
+    AgentWorkersDetails (..),
+    getAgentWorkersDetails,
+    AgentWorkersSummary (..),
+    getAgentWorkersSummary,
   )
 where
 
@@ -150,6 +155,8 @@ import Data.Text.Encoding
 import Data.Time (UTCTime, defaultTimeLocale, formatTime, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Word (Word16)
+
+-- import GHC.Conc (unsafeIOToSTM)
 import Network.Socket (HostName)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientConfig (..), XFTPClientError)
 import qualified Simplex.FileTransfer.Client as X
@@ -165,7 +172,6 @@ import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore (..), withTransaction)
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
-import Simplex.Messaging.Agent.TAsyncs
 import Simplex.Messaging.Agent.TRcvQueues (TRcvQueues (getRcvQueues))
 import qualified Simplex.Messaging.Agent.TRcvQueues as RQ
 import Simplex.Messaging.Client
@@ -217,13 +223,18 @@ import UnliftIO.Exception (bracket)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
-type ClientVar msg = TMVar (Either AgentErrorType (Client msg))
+data SessionVar a = SessionVar
+  { sessionVar :: TMVar a,
+    sessionVarId :: Int
+  }
+
+type ClientVar msg = SessionVar (Either AgentErrorType (Client msg))
 
 type SMPClientVar = ClientVar SMP.BrokerMsg
 
 type NtfClientVar = ClientVar NtfResponse
 
-type XFTPClientVar = TMVar (Either AgentErrorType XFTPClient)
+type XFTPClientVar = ClientVar FileResponse
 
 type SMPTransportSession = TransportSession SMP.BrokerMsg
 
@@ -264,10 +275,8 @@ data AgentClient = AgentClient
     invLocks :: TMap ByteString Lock,
     -- lock to prevent concurrency between periodic and async connection deletions
     deleteLock :: Lock,
-    -- locks to prevent concurrent reconnections to SMP servers
-    reconnectLocks :: TMap SMPTransportSession Lock,
-    reconnections :: TAsyncs,
-    asyncClients :: TAsyncs,
+    -- smpSubWorkers for SMP servers sessions
+    smpSubWorkers :: TMap SMPTransportSession (SessionVar (Async ())),
     agentStats :: TMap AgentStatsKey (TVar Int),
     clientId :: Int,
     agentEnv :: Env
@@ -288,11 +297,11 @@ getAgentWorker' toW fromW name hasWork c key ws work = do
     whenExists w
       | hasWork = hasWorkToDo (toW w) $> w
       | otherwise = pure w
-    runWorker w = runWorkerAsync (toW w) . void $ runExceptT runWork
+    runWorker w = runWorkerAsync (toW w) runWork
       where
-        runWork :: ExceptT AgentErrorType m ()
-        runWork = tryAgentError (work w) >>= restartOrDelete
-        restartOrDelete :: Either AgentErrorType () -> ExceptT AgentErrorType m ()
+        runWork :: m ()
+        runWork = tryAgentError' (work w) >>= restartOrDelete
+        restartOrDelete :: Either AgentErrorType () -> m ()
         restartOrDelete e_ = do
           t <- liftIO getSystemTime
           maxRestarts <- asks $ maxWorkerRestartsPerMin . config
@@ -361,7 +370,6 @@ data AgentState = ASForeground | ASSuspending | ASSuspended
 data AgentLocks = AgentLocks
   { connLocks :: Map String String,
     invLocks :: Map String String,
-    srvLocks :: Map String String,
     delLock :: Maybe String
   }
   deriving (Show)
@@ -375,8 +383,8 @@ data AgentStatsKey = AgentStatsKey
   }
   deriving (Eq, Ord, Show)
 
-newAgentClient :: InitialAgentServers -> Env -> STM AgentClient
-newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
+newAgentClient :: Int -> InitialAgentServers -> Env -> STM AgentClient
+newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
   let qSize = tbqSize $ config agentEnv
   active <- newTVar True
   rcvQ <- newTBQueue qSize
@@ -407,11 +415,8 @@ newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
   connLocks <- TM.empty
   invLocks <- TM.empty
   deleteLock <- createLock
-  reconnectLocks <- TM.empty
-  reconnections <- newTAsyncs
-  asyncClients <- newTAsyncs
+  smpSubWorkers <- TM.empty
   agentStats <- TM.empty
-  clientId <- stateTVar (clientCounter agentEnv) $ \i -> let i' = i + 1 in (i', i')
   return
     AgentClient
       { active,
@@ -443,9 +448,7 @@ newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
         connLocks,
         invLocks,
         deleteLock,
-        reconnectLocks,
-        reconnections,
-        asyncClients,
+        smpSubWorkers,
         agentStats,
         clientId,
         agentEnv
@@ -496,26 +499,25 @@ instance ProtocolServerClient XFTPErrorType FileResponse where
 getSMPServerClient :: forall m. AgentMonad m => AgentClient -> SMPTransportSession -> m SMPClient
 getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getClientVar tSess smpClients)
-    >>= either newClient (waitForProtocolClient c tSess)
+  v <- atomically (getTSessVar c tSess smpClients)
+  either newClient (waitForProtocolClient c tSess) v
+    `catchAgentError` \e -> resubscribeSMPSession c tSess >> throwError e
   where
-    newClient v = do
-      tc <- newTVarIO 0
-      newProtocolClient c tSess smpClients connectClient (reconnectSMPClient 0 tc) v
-    connectClient :: m SMPClient
-    connectClient = do
+    newClient = newProtocolClient c tSess smpClients connectClient
+    connectClient :: SMPClientVar -> m SMPClient
+    connectClient v = do
       cfg <- getClientConfig c smpCfg
       u <- askUnliftIO
-      liftEitherError (protocolClientError SMP $ B.unpack $ strEncode srv) (getProtocolClient tSess cfg (Just msgQ) $ clientDisconnected u)
+      liftEitherError (protocolClientError SMP $ B.unpack $ strEncode srv) (getProtocolClient tSess cfg (Just msgQ) $ clientDisconnected u v)
 
-    clientDisconnected :: UnliftIO m -> SMPClient -> IO ()
-    clientDisconnected u client = do
+    clientDisconnected :: UnliftIO m -> SMPClientVar -> SMPClient -> IO ()
+    clientDisconnected u v client = do
       removeClientAndSubs >>= serverDown
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
       where
         removeClientAndSubs :: IO ([RcvQueue], [ConnId])
         removeClientAndSubs = atomically $ do
-          TM.delete tSess smpClients
+          removeTSessVar v tSess smpClients
           qs <- RQ.getDelSessQueues tSess $ activeSubs c
           mapM_ (`RQ.addQueue` pendingSubs c) qs
           let cs = S.fromList $ map qConnId qs
@@ -529,41 +531,55 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
           unless (null conns) $ notifySub "" $ DOWN srv conns
           unless (null qs) $ do
             atomically $ mapM_ (releaseGetLock c) qs
-            unliftIO u $ reconnectServer c tSess
+            unliftIO u $ resubscribeSMPSession c tSess
 
         notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
         notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
 
-reconnectServer :: AgentMonad m => AgentClient -> SMPTransportSession -> m ()
-reconnectServer c tSess = newAsyncAction tryReconnectSMPClient $ reconnections c
+resubscribeSMPSession :: AgentMonad' m => AgentClient -> SMPTransportSession -> m ()
+resubscribeSMPSession c@AgentClient {smpSubWorkers} tSess =
+  atomically getWorkerVar >>= mapM_ (either newSubWorker (\_ -> pure ()))
   where
-    tryReconnectSMPClient aId = do
+    getWorkerVar =
+      ifM
+        (null <$> getPending)
+        (pure Nothing) -- prevent race with cleanup and adding pending queues in another call
+        (Just <$> getTSessVar c tSess smpSubWorkers)
+    newSubWorker v = do
+      a <- async $ void (E.tryAny runSubWorker) >> atomically (cleanup v)
+      atomically $ putTMVar (sessionVar v) a
+    runSubWorker = do
       ri <- asks $ reconnectInterval . config
       timeoutCounts <- newTVarIO 0
-      withRetryIntervalCount ri $ \n _ loop ->
-        reconnectSMPClient n timeoutCounts c tSess `catchAgentError` const loop
-      atomically . removeAsyncAction aId $ reconnections c
+      withRetryInterval ri $ \_ loop -> do
+        pending <- atomically getPending
+        forM_ (L.nonEmpty pending) $ \qs -> do
+          void . tryAgentError' $ reconnectSMPClient timeoutCounts c tSess qs
+          loop
+    getPending = RQ.getSessQueues tSess $ pendingSubs c
+    cleanup :: SessionVar (Async ()) -> STM ()
+    cleanup v = do
+      -- Here we wait until TMVar is not empty to prevent worker cleanup happening before worker is added to TMVar.
+      -- Not waiting may result in terminated worker remaining in the map.
+      whenM (isEmptyTMVar $ sessionVar v) retry
+      removeTSessVar v tSess smpSubWorkers
 
-reconnectSMPClient :: forall m. AgentMonad m => Int -> TVar Int -> AgentClient -> SMPTransportSession -> m ()
-reconnectSMPClient n tc c tSess@(_, srv, _) = do
-  ts <- liftIO getCurrentTime
-  let label = unwords ["reconnect", show n, show ts]
-  withLockMap_ (reconnectLocks c) tSess label $ do
-    qs <- atomically (RQ.getSessQueues tSess $ pendingSubs c)
-    NetworkConfig {tcpTimeout} <- readTVarIO $ useNetworkConfig c
-    -- this allows 3x of timeout per batch of subscription (90 queues per batch empirically)
-    let t = (length qs `div` 90 + 1) * tcpTimeout * 3
-    t `timeout` mapM_ resubscribe (L.nonEmpty qs) >>= \case
-      Just _ -> atomically $ writeTVar tc 0
-      Nothing -> do
-        tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
-        maxTC <- asks $ maxSubscriptionTimeouts . config
-        let err = if tc' >= maxTC then CRITICAL True else INTERNAL
-            msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
-        atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err msg)
+reconnectSMPClient :: forall m. AgentMonad m => TVar Int -> AgentClient -> SMPTransportSession -> NonEmpty RcvQueue -> m ()
+reconnectSMPClient tc c tSess@(_, srv, _) qs = do
+  NetworkConfig {tcpTimeout} <- readTVarIO $ useNetworkConfig c
+  -- this allows 3x of timeout per batch of subscription (90 queues per batch empirically)
+  let t = (length qs `div` 90 + 1) * tcpTimeout * 3
+  t `timeout` resubscribe >>= \case
+    Just _ -> atomically $ writeTVar tc 0
+    Nothing -> do
+      tc' <- atomically $ stateTVar tc $ \i -> (i + 1, i + 1)
+      maxTC <- asks $ maxSubscriptionTimeouts . config
+      let err = if tc' >= maxTC then CRITICAL True else INTERNAL
+          msg = show tc' <> " consecutive subscription timeouts: " <> show (length qs) <> " queues, transport session: " <> show tSess
+      atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err msg)
   where
-    resubscribe :: NonEmpty RcvQueue -> m ()
-    resubscribe qs = do
+    resubscribe :: m ()
+    resubscribe = do
       cs <- atomically . RQ.getConns $ activeSubs c
       rs <- subscribeQueues c $ L.toList qs
       let (errs, okConns) = partitionEithers $ map (\(RcvQueue {connId}, r) -> bimap (connId,) (const connId) r) rs
@@ -572,26 +588,29 @@ reconnectSMPClient n tc c tSess@(_, srv, _) = do
         unless (null conns) $ notifySub "" $ UP srv conns
       let (tempErrs, finalErrs) = partition (temporaryAgentError . snd) errs
       liftIO $ mapM_ (\(connId, e) -> notifySub connId $ ERR e) finalErrs
-      mapM_ (throwError . snd) $ listToMaybe tempErrs
+      forM_ (listToMaybe tempErrs) $ \(_, err) -> do
+        when (null okConns && S.null cs && null finalErrs) . liftIO $
+          closeClient c smpClients tSess
+        throwError err
     notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
     notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
 
 getNtfServerClient :: forall m. AgentMonad m => AgentClient -> NtfTransportSession -> m NtfClient
 getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getClientVar tSess ntfClients)
+  atomically (getTSessVar c tSess ntfClients)
     >>= either
-      (newProtocolClient c tSess ntfClients connectClient $ \_ _ -> pure ())
+      (newProtocolClient c tSess ntfClients connectClient)
       (waitForProtocolClient c tSess)
   where
-    connectClient :: m NtfClient
-    connectClient = do
+    connectClient :: NtfClientVar -> m NtfClient
+    connectClient v = do
       cfg <- getClientConfig c ntfCfg
-      liftEitherError (protocolClientError NTF $ B.unpack $ strEncode srv) (getProtocolClient tSess cfg Nothing clientDisconnected)
+      liftEitherError (protocolClientError NTF $ B.unpack $ strEncode srv) (getProtocolClient tSess cfg Nothing $ clientDisconnected v)
 
-    clientDisconnected :: NtfClient -> IO ()
-    clientDisconnected client = do
-      atomically $ TM.delete tSess ntfClients
+    clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
+    clientDisconnected v client = do
+      atomically $ removeTSessVar v tSess ntfClients
       incClientStat c userId client "DISCONNECT" ""
       atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
@@ -599,78 +618,73 @@ getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = d
 getXFTPServerClient :: forall m. AgentMonad m => AgentClient -> XFTPTransportSession -> m XFTPClient
 getXFTPServerClient c@AgentClient {active, xftpClients, useNetworkConfig} tSess@(userId, srv, _) = do
   unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getClientVar tSess xftpClients)
+  atomically (getTSessVar c tSess xftpClients)
     >>= either
-      (newProtocolClient c tSess xftpClients connectClient $ \_ _ -> pure ())
+      (newProtocolClient c tSess xftpClients connectClient)
       (waitForProtocolClient c tSess)
   where
-    connectClient :: m XFTPClient
-    connectClient = do
+    connectClient :: XFTPClientVar -> m XFTPClient
+    connectClient v = do
       cfg <- asks $ xftpCfg . config
       xftpNetworkConfig <- readTVarIO useNetworkConfig
-      liftEitherError (protocolClientError XFTP $ B.unpack $ strEncode srv) (X.getXFTPClient tSess cfg {xftpNetworkConfig} clientDisconnected)
+      liftEitherError (protocolClientError XFTP $ B.unpack $ strEncode srv) (X.getXFTPClient tSess cfg {xftpNetworkConfig} $ clientDisconnected v)
 
-    clientDisconnected :: XFTPClient -> IO ()
-    clientDisconnected client = do
-      atomically $ TM.delete tSess xftpClients
+    clientDisconnected :: XFTPClientVar -> XFTPClient -> IO ()
+    clientDisconnected v client = do
+      atomically $ removeTSessVar v tSess xftpClients
       incClientStat c userId client "DISCONNECT" ""
       atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
       logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
 
-getClientVar :: forall a s. TransportSession s -> TMap (TransportSession s) (TMVar a) -> STM (Either (TMVar a) (TMVar a))
-getClientVar tSess clients = maybe (Left <$> newClientVar) (pure . Right) =<< TM.lookup tSess clients
+getTSessVar :: forall a s. AgentClient -> TransportSession s -> TMap (TransportSession s) (SessionVar a) -> STM (Either (SessionVar a) (SessionVar a))
+getTSessVar c tSess vs = maybe (Left <$> newSessionVar) (pure . Right) =<< TM.lookup tSess vs
   where
-    newClientVar :: STM (TMVar a)
-    newClientVar = do
-      var <- newEmptyTMVar
-      TM.insert tSess var clients
-      pure var
+    newSessionVar :: STM (SessionVar a)
+    newSessionVar = do
+      sessionVar <- newEmptyTMVar
+      sessionVarId <- stateTVar (workerSeq c) $ \next -> (next, next + 1)
+      let v = SessionVar {sessionVar, sessionVarId}
+      TM.insert tSess v vs
+      pure v
+
+removeTSessVar :: SessionVar a -> TransportSession msg -> TMap (TransportSession msg) (SessionVar a) -> STM ()
+removeTSessVar v tSess vs =
+  TM.lookup tSess vs
+    >>= mapM_ (\v' -> when (sessionVarId v == sessionVarId v') $ TM.delete tSess vs)
 
 waitForProtocolClient :: (AgentMonad m, ProtocolTypeI (ProtoType msg)) => AgentClient -> TransportSession msg -> ClientVar msg -> m (Client msg)
-waitForProtocolClient c (_, srv, _) clientVar = do
+waitForProtocolClient c (_, srv, _) v = do
   NetworkConfig {tcpConnectTimeout} <- readTVarIO $ useNetworkConfig c
-  client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar clientVar)
+  client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v)
   liftEither $ case client_ of
     Just (Right smpClient) -> Right smpClient
     Just (Left e) -> Left e
     Nothing -> Left $ BROKER (B.unpack $ strEncode srv) TIMEOUT
 
+-- clientConnected arg is only passed for SMP server
 newProtocolClient ::
   forall err msg m.
   (AgentMonad m, ProtocolTypeI (ProtoType msg), ProtocolServerClient err msg) =>
   AgentClient ->
   TransportSession msg ->
   TMap (TransportSession msg) (ClientVar msg) ->
-  m (Client msg) ->
-  (AgentClient -> TransportSession msg -> m ()) ->
+  (ClientVar msg -> m (Client msg)) ->
   ClientVar msg ->
   m (Client msg)
-newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient reconnectClient clientVar = tryConnectClient pure tryConnectAsync
-  where
-    tryConnectClient :: (Client msg -> m a) -> m () -> m a
-    tryConnectClient successAction retryAction =
-      tryError connectClient >>= \r -> case r of
-        Right client -> do
-          logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
-          atomically $ putTMVar clientVar r
-          liftIO $ incClientStat c userId client "CLIENT" "OK"
-          atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent CONNECT client)
-          successAction client
-        Left e -> do
-          liftIO $ incServerStat c userId srv "CLIENT" $ strEncode e
-          if temporaryAgentError e
-            then retryAction
-            else atomically $ do
-              putTMVar clientVar (Left e)
-              TM.delete tSess clients
-          throwError e
-    tryConnectAsync :: m ()
-    tryConnectAsync = newAsyncAction connectAsync $ asyncClients c
-    connectAsync :: Int -> m ()
-    connectAsync aId = do
-      ri <- asks $ reconnectInterval . config
-      withRetryInterval ri $ \_ loop -> void $ tryConnectClient (const $ reconnectClient c tSess) loop
-      atomically . removeAsyncAction aId $ asyncClients c
+newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
+  tryAgentError (connectClient v) >>= \case
+    Right client -> do
+      logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
+      atomically $ putTMVar (sessionVar v) (Right client)
+      liftIO $ incClientStat c userId client "CLIENT" "OK"
+      atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent CONNECT client)
+      pure client
+    Left e -> do
+      liftIO $ incServerStat c userId srv "CLIENT" $ strEncode e
+      atomically $ do
+        removeTSessVar v tSess clients
+        putTMVar (sessionVar v) (Left e)
+      throwError e -- signal error to caller
 
 hostEvent :: forall err msg. (ProtocolTypeI (ProtoType msg), ProtocolServerClient err msg) => (AProtocolType -> TransportHost -> ACommand 'Agent 'AENone) -> Client msg -> ACommand 'Agent 'AENone
 hostEvent event = event (AProtocolType $ protocolTypeI @(ProtoType msg)) . clientTransportHost
@@ -687,8 +701,7 @@ closeAgentClient c = liftIO $ do
   closeProtocolServerClients c smpClients
   closeProtocolServerClients c ntfClients
   closeProtocolServerClients c xftpClients
-  cancelActions . actions $ reconnections c
-  cancelActions . actions $ asyncClients c
+  atomically (swapTVar (smpSubWorkers c) M.empty) >>= mapM_ cancelReconnect
   clearWorkers smpDeliveryWorkers >>= mapM_ (cancelWorker . fst)
   clearWorkers asyncCmdWorkers >>= mapM_ cancelWorker
   clear connCmdsQueued
@@ -701,6 +714,8 @@ closeAgentClient c = liftIO $ do
     clearWorkers workers = atomically $ swapTVar (workers c) mempty
     clear :: Monoid m => (AgentClient -> TVar m) -> IO ()
     clear sel = atomically $ writeTVar (sel c) mempty
+    cancelReconnect :: SessionVar (Async ()) -> IO ()
+    cancelReconnect v = void . forkIO $ atomically (readTMVar $ sessionVar v) >>= uninterruptibleCancel
 
 cancelWorker :: Worker -> IO ()
 cancelWorker Worker {doWork, action} = do
@@ -728,9 +743,9 @@ closeClient c clientSel tSess =
   atomically (TM.lookupDelete tSess $ clientSel c) >>= mapM_ (closeClient_ c)
 
 closeClient_ :: ProtocolServerClient err msg => AgentClient -> ClientVar msg -> IO ()
-closeClient_ c cVar = do
+closeClient_ c v = do
   NetworkConfig {tcpConnectTimeout} <- readTVarIO $ useNetworkConfig c
-  tcpConnectTimeout `timeout` atomically (readTMVar cVar) >>= \case
+  tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v) >>= \case
     Just (Right client) -> closeProtocolServerClient client `catchAll_` pure ()
     _ -> pure ()
 
@@ -738,9 +753,6 @@ closeXFTPServerClient :: AgentMonad' m => AgentClient -> UserId -> XFTPServer ->
 closeXFTPServerClient c userId server (FileDigest chunkDigest) =
   mkTransportSession c userId server chunkDigest >>= liftIO . closeClient c xftpClients
 
-cancelActions :: (Foldable f, Monoid (f (Async ()))) => TVar (f (Async ())) -> IO ()
-cancelActions as = atomically (swapTVar as mempty) >>= mapM_ (forkIO . uninterruptibleCancel)
-
 withConnLock :: MonadUnliftIO m => AgentClient -> ConnId -> String -> m a -> m a
 withConnLock _ "" _ = id
 withConnLock AgentClient {connLocks} connId name = withLockMap_ connLocks connId name
@@ -991,7 +1003,7 @@ temporaryOrHostError = \case
   e -> temporaryAgentError e
 
 -- | Subscribe to queues. The list of results can have a different order.
-subscribeQueues :: forall m. AgentMonad m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
+subscribeQueues :: forall m. AgentMonad' m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
 subscribeQueues c qs = do
   (errs, qs') <- partitionEithers <$> mapM checkQueue qs
   forM_ qs' $ \rq@RcvQueue {connId} -> atomically $ do
@@ -1009,13 +1021,13 @@ subscribeQueues c qs = do
       rs <- sendBatch subscribeSMPQueues smp qs'
       mapM_ (uncurry $ processSubResult c) rs
       when (any temporaryClientError . lefts . map snd $ L.toList rs) . unliftIO u $
-        reconnectServer c (transportSession' smp)
+        resubscribeSMPSession c (transportSession' smp)
       pure rs
 
 type BatchResponses e r = (NonEmpty (RcvQueue, Either e r))
 
 -- statBatchSize is not used to batch the commands, only for traffic statistics
-sendTSessionBatches :: forall m q r. AgentMonad m => ByteString -> Int -> (q -> RcvQueue) -> (SMPClient -> NonEmpty q -> IO (BatchResponses SMPClientError r)) -> AgentClient -> [q] -> m [(RcvQueue, Either AgentErrorType r)]
+sendTSessionBatches :: forall m q r. AgentMonad' m => ByteString -> Int -> (q -> RcvQueue) -> (SMPClient -> NonEmpty q -> IO (BatchResponses SMPClientError r)) -> AgentClient -> [q] -> m [(RcvQueue, Either AgentErrorType r)]
 sendTSessionBatches statCmd statBatchSize toRQ action c qs =
   concatMap L.toList <$> (mapConcurrently sendClientBatch =<< batchQueues)
   where
@@ -1029,7 +1041,7 @@ sendTSessionBatches statCmd statBatchSize toRQ action c qs =
            in M.alter (Just . maybe [q] (q <|)) tSess m
     sendClientBatch :: (SMPTransportSession, NonEmpty q) -> m (BatchResponses AgentErrorType r)
     sendClientBatch (tSess@(userId, srv, _), qs') =
-      tryError (getSMPServerClient c tSess) >>= \case
+      tryAgentError' (getSMPServerClient c tSess) >>= \case
         Left e -> pure $ L.map ((,Left e) . toRQ) qs'
         Right smp -> liftIO $ do
           logServer "-->" c srv (bshow (length qs') <> " queues") statCmd
@@ -1130,7 +1142,7 @@ enableQueueNotifications c rq@RcvQueue {rcvId, rcvPrivateKey} notifierKey rcvNtf
   withSMPClient c rq "NKEY <nkey>" $ \smp ->
     enableSMPQueueNotifications smp rcvPrivateKey rcvId notifierKey rcvNtfPublicDhKey
 
-enableQueuesNtfs :: forall m. AgentMonad m => AgentClient -> [(RcvQueue, SMP.NtfPublicVerifyKey, SMP.RcvNtfPublicDhKey)] -> m [(RcvQueue, Either AgentErrorType (SMP.NotifierId, SMP.RcvNtfPublicDhKey))]
+enableQueuesNtfs :: forall m. AgentMonad' m => AgentClient -> [(RcvQueue, SMP.NtfPublicVerifyKey, SMP.RcvNtfPublicDhKey)] -> m [(RcvQueue, Either AgentErrorType (SMP.NotifierId, SMP.RcvNtfPublicDhKey))]
 enableQueuesNtfs = sendTSessionBatches "NKEY" 90 fst3 enableQueues_
   where
     fst3 (x, _, _) = x
@@ -1144,7 +1156,7 @@ disableQueueNotifications c rq@RcvQueue {rcvId, rcvPrivateKey} =
   withSMPClient c rq "NDEL" $ \smp ->
     disableSMPQueueNotifications smp rcvPrivateKey rcvId
 
-disableQueuesNtfs :: forall m. AgentMonad m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
+disableQueuesNtfs :: forall m. AgentMonad' m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
 disableQueuesNtfs = sendTSessionBatches "NDEL" 90 id $ sendBatch disableSMPQueuesNtfs
 
 sendAck :: AgentMonad m => AgentClient -> RcvQueue -> MsgId -> m ()
@@ -1171,7 +1183,7 @@ deleteQueue c rq@RcvQueue {rcvId, rcvPrivateKey} = do
   withSMPClient c rq "DEL" $ \smp ->
     deleteSMPQueue smp rcvPrivateKey rcvId
 
-deleteQueues :: forall m. AgentMonad m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
+deleteQueues :: forall m. AgentMonad' m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
 deleteQueues = sendTSessionBatches "DEL" 90 id $ sendBatch deleteSMPQueues
 
 sendAgentMessage :: AgentMonad m => AgentClient -> SndQueue -> MsgFlags -> ByteString -> m ()
@@ -1527,6 +1539,117 @@ getAgentSubscriptions c = do
     enc :: StrEncoding a => a -> Text
     enc = decodeLatin1 . strEncode
 
+data AgentWorkersDetails = AgentWorkersDetails
+  { smpClients_ :: [Text],
+    ntfClients_ :: [Text],
+    xftpClients_ :: [Text],
+    smpDeliveryWorkers_ :: Map Text WorkersDetails,
+    asyncCmdWorkers_ :: Map Text WorkersDetails,
+    smpSubWorkers_ :: [Text],
+    ntfWorkers_ :: Map Text WorkersDetails,
+    ntfSMPWorkers_ :: Map Text WorkersDetails,
+    xftpRcvWorkers_ :: Map Text WorkersDetails,
+    xftpSndWorkers_ :: Map Text WorkersDetails,
+    xftpDelWorkers_ :: Map Text WorkersDetails
+  }
+  deriving (Show)
+
+data WorkersDetails = WorkersDetails
+  { restarts :: Int,
+    hasWork :: Bool,
+    hasAction :: Bool
+  }
+  deriving (Show)
+
+getAgentWorkersDetails :: MonadIO m => AgentClient -> m AgentWorkersDetails
+getAgentWorkersDetails AgentClient {smpClients, ntfClients, xftpClients, smpDeliveryWorkers, asyncCmdWorkers, smpSubWorkers, agentEnv} = do
+  smpClients_ <- textKeys <$> readTVarIO smpClients
+  ntfClients_ <- textKeys <$> readTVarIO ntfClients
+  xftpClients_ <- textKeys <$> readTVarIO xftpClients
+  smpDeliveryWorkers_ <- workerStats . fmap fst =<< readTVarIO smpDeliveryWorkers
+  asyncCmdWorkers_ <- workerStats =<< readTVarIO asyncCmdWorkers
+  smpSubWorkers_ <- textKeys <$> readTVarIO smpSubWorkers
+  ntfWorkers_ <- workerStats =<< readTVarIO ntfWorkers
+  ntfSMPWorkers_ <- workerStats =<< readTVarIO ntfSMPWorkers
+  xftpRcvWorkers_ <- workerStats =<< readTVarIO xftpRcvWorkers
+  xftpSndWorkers_ <- workerStats =<< readTVarIO xftpSndWorkers
+  xftpDelWorkers_ <- workerStats =<< readTVarIO xftpDelWorkers
+  pure
+    AgentWorkersDetails
+      { smpClients_,
+        ntfClients_,
+        xftpClients_,
+        smpDeliveryWorkers_,
+        asyncCmdWorkers_,
+        smpSubWorkers_,
+        ntfWorkers_,
+        ntfSMPWorkers_,
+        xftpRcvWorkers_,
+        xftpSndWorkers_,
+        xftpDelWorkers_
+      }
+  where
+    textKeys :: StrEncoding k => Map k v -> [Text]
+    textKeys = map textKey . M.keys
+    textKey :: StrEncoding k => k -> Text
+    textKey = decodeASCII . strEncode
+    workerStats :: (StrEncoding k, MonadIO m) => Map k Worker -> m (Map Text WorkersDetails)
+    workerStats ws = fmap M.fromList . forM (M.toList ws) $ \(qa, Worker {restarts, doWork, action}) -> do
+      RestartCount {restartCount} <- readTVarIO restarts
+      hasWork <- atomically $ not <$> isEmptyTMVar doWork
+      hasAction <- atomically $ not <$> isEmptyTMVar action
+      pure (textKey qa, WorkersDetails {restarts = restartCount, hasWork, hasAction})
+    Env {ntfSupervisor, xftpAgent} = agentEnv
+    NtfSupervisor {ntfWorkers, ntfSMPWorkers} = ntfSupervisor
+    XFTPAgent {xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers} = xftpAgent
+
+data AgentWorkersSummary = AgentWorkersSummary
+  { smpClientsCount :: Int,
+    ntfClientsCount :: Int,
+    xftpClientsCount :: Int,
+    smpDeliveryWorkersCount :: Int,
+    asyncCmdWorkersCount :: Int,
+    smpSubWorkersCount :: Int,
+    ntfWorkersCount :: Int,
+    ntfSMPWorkersCount :: Int,
+    xftpRcvWorkersCount :: Int,
+    xftpSndWorkersCount :: Int,
+    xftpDelWorkersCount :: Int
+  }
+  deriving (Show)
+
+getAgentWorkersSummary :: MonadIO m => AgentClient -> m AgentWorkersSummary
+getAgentWorkersSummary AgentClient {smpClients, ntfClients, xftpClients, smpDeliveryWorkers, asyncCmdWorkers, smpSubWorkers, agentEnv} = do
+  smpClientsCount <- M.size <$> readTVarIO smpClients
+  ntfClientsCount <- M.size <$> readTVarIO ntfClients
+  xftpClientsCount <- M.size <$> readTVarIO xftpClients
+  smpDeliveryWorkersCount <- M.size <$> readTVarIO smpDeliveryWorkers
+  asyncCmdWorkersCount <- M.size <$> readTVarIO asyncCmdWorkers
+  smpSubWorkersCount <- M.size <$> readTVarIO smpSubWorkers
+  ntfWorkersCount <- M.size <$> readTVarIO ntfWorkers
+  ntfSMPWorkersCount <- M.size <$> readTVarIO ntfSMPWorkers
+  xftpRcvWorkersCount <- M.size <$> readTVarIO xftpRcvWorkers
+  xftpSndWorkersCount <- M.size <$> readTVarIO xftpSndWorkers
+  xftpDelWorkersCount <- M.size <$> readTVarIO xftpDelWorkers
+  pure
+    AgentWorkersSummary
+      { smpClientsCount,
+        ntfClientsCount,
+        xftpClientsCount,
+        smpDeliveryWorkersCount,
+        asyncCmdWorkersCount,
+        smpSubWorkersCount,
+        ntfWorkersCount,
+        ntfSMPWorkersCount,
+        xftpRcvWorkersCount,
+        xftpSndWorkersCount,
+        xftpDelWorkersCount
+      }
+  where
+    Env {ntfSupervisor, xftpAgent} = agentEnv
+    NtfSupervisor {ntfWorkers, ntfSMPWorkers} = ntfSupervisor
+    XFTPAgent {xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers} = xftpAgent
+
 $(J.deriveJSON defaultJSON ''AgentLocks)
 
 $(J.deriveJSON (enumJSON $ dropPrefix "TS") ''ProtocolTestStep)
@@ -1536,3 +1659,8 @@ $(J.deriveJSON defaultJSON ''ProtocolTestFailure)
 $(J.deriveJSON defaultJSON ''SubInfo)
 
 $(J.deriveJSON defaultJSON ''SubscriptionsInfo)
+
+$(J.deriveJSON defaultJSON ''WorkersDetails)
+$(J.deriveJSON defaultJSON {J.fieldLabelModifier = takeWhile (/= '_')} ''AgentWorkersDetails)
+
+$(J.deriveJSON defaultJSON ''AgentWorkersSummary)

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

@@ -19,6 +19,7 @@ module Simplex.Messaging.Agent.Env.SQLite
     defaultAgentConfig,
     defaultReconnectInterval,
     tryAgentError,
+    tryAgentError',
     catchAgentError,
     agentFinally,
     Env (..),
@@ -33,6 +34,7 @@ module Simplex.Messaging.Agent.Env.SQLite
   )
 where
 
+import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Reader
@@ -90,6 +92,7 @@ data AgentConfig = AgentConfig
     messageRetryInterval :: RetryInterval2,
     messageTimeout :: NominalDiffTime,
     helloTimeout :: NominalDiffTime,
+    quotaExceededTimeout :: NominalDiffTime,
     initialCleanupDelay :: Int64,
     cleanupInterval :: Int64,
     cleanupStepInterval :: Int,
@@ -112,8 +115,7 @@ data AgentConfig = AgentConfig
     certificateFile :: FilePath,
     e2eEncryptVRange :: VersionRange,
     smpAgentVRange :: VersionRange,
-    smpClientVRange :: VersionRange,
-    initialClientId :: Int
+    smpClientVRange :: VersionRange
   }
 
 defaultReconnectInterval :: RetryInterval
@@ -134,13 +136,10 @@ defaultMessageRetryInterval =
             maxInterval = 60_000000
           },
       riSlow =
-        -- TODO: these timeouts can be increased in v5.0 once most clients are updated
-        -- to resume sending on QCONT messages.
-        -- After that local message expiration period should be also increased.
         RetryInterval
-          { initialInterval = 60_000000,
+          { initialInterval = 180_000000, -- 3 minutes
             increaseAfter = 60_000000,
-            maxInterval = 3600_000000 -- 1 hour
+            maxInterval = 3 * 3600_000000 -- 3 hours
           }
     }
 
@@ -158,6 +157,7 @@ defaultAgentConfig =
       messageRetryInterval = defaultMessageRetryInterval,
       messageTimeout = 2 * nominalDay,
       helloTimeout = 2 * nominalDay,
+      quotaExceededTimeout = 7 * nominalDay,
       initialCleanupDelay = 30 * 1000000, -- 30 seconds
       cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
       cleanupStepInterval = 200000, -- 200ms
@@ -184,15 +184,13 @@ defaultAgentConfig =
       certificateFile = "/etc/opt/simplex-agent/agent.crt",
       e2eEncryptVRange = supportedE2EEncryptVRange,
       smpAgentVRange = supportedSMPAgentVRange,
-      smpClientVRange = supportedSMPClientVRange,
-      initialClientId = 0
+      smpClientVRange = supportedSMPClientVRange
     }
 
 data Env = Env
   { config :: AgentConfig,
     store :: SQLiteStore,
     random :: TVar ChaChaDRG,
-    clientCounter :: TVar Int,
     randomServer :: TVar StdGen,
     ntfSupervisor :: NtfSupervisor,
     xftpAgent :: XFTPAgent,
@@ -200,14 +198,13 @@ data Env = Env
   }
 
 newSMPAgentEnv :: AgentConfig -> SQLiteStore -> IO Env
-newSMPAgentEnv config@AgentConfig {initialClientId} store = do
+newSMPAgentEnv config store = do
   random <- C.newRandom
-  clientCounter <- newTVarIO initialClientId
   randomServer <- newTVarIO =<< liftIO newStdGen
   ntfSupervisor <- atomically . newNtfSubSupervisor $ tbqSize config
   xftpAgent <- atomically newXFTPAgent
   multicastSubscribers <- newTMVarIO 0
-  pure Env {config, store, random, clientCounter, randomServer, ntfSupervisor, xftpAgent, multicastSubscribers}
+  pure Env {config, store, random, randomServer, ntfSupervisor, xftpAgent, multicastSubscribers}
 
 createAgentStore :: FilePath -> ScrubbedBytes -> Bool -> MigrationConfirmation -> IO (Either MigrationError SQLiteStore)
 createAgentStore dbFilePath dbKey keepKey = createSQLiteStore dbFilePath dbKey keepKey Migrations.app
@@ -250,6 +247,11 @@ tryAgentError :: AgentMonad m => m a -> m (Either AgentErrorType a)
 tryAgentError = tryAllErrors mkInternal
 {-# INLINE tryAgentError #-}
 
+-- unlike runExceptT, this ensures we catch IO exceptions as well
+tryAgentError' :: AgentMonad' m => ExceptT AgentErrorType m a -> m (Either AgentErrorType a)
+tryAgentError' = fmap join . runExceptT . tryAgentError
+{-# INLINE tryAgentError' #-}
+
 catchAgentError :: AgentMonad m => m a -> (AgentErrorType -> m a) -> m a
 catchAgentError = catchAllErrors mkInternal
 {-# INLINE catchAgentError #-}

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -72,7 +72,7 @@ processNtfSub c (connId, cmd) = do
       logInfo $ "processNtfSub, NSCCreate - a = " <> tshow a
       case a of
         Nothing -> do
-          withNtfServer c $ \ntfServer -> do
+          withTokenServer $ \ntfServer -> do
             case clientNtfCreds of
               Just ClientNtfCreds {notifierId} -> do
                 let newSub = newNtfSubscription connId smpServer (Just notifierId) ntfServer NASKey
@@ -99,7 +99,7 @@ processNtfSub c (connId, cmd) = do
                 | isDeleteNtfSubAction action -> do
                     if ntfSubStatus == NASNew || ntfSubStatus == NASOff || ntfSubStatus == NASDeleted
                       then resetSubscription
-                      else withNtfServer c $ \ntfServer -> do
+                      else withTokenServer $ \ntfServer -> do
                         withStore' c $ \db -> supervisorUpdateNtfSub db sub {ntfServer} (NtfSubNTFAction NSACreate)
                         void $ getNtfNTFWorker True c ntfServer
                 | otherwise -> case action of
@@ -111,7 +111,7 @@ processNtfSub c (connId, cmd) = do
               void $ getNtfNTFWorker True c subNtfServer
             resetSubscription :: m ()
             resetSubscription =
-              withNtfServer c $ \ntfServer -> do
+              withTokenServer $ \ntfServer -> do
                 let sub' = sub {ntfQueueId = Nothing, ntfServer, ntfSubId = Nothing, ntfSubStatus = NASNew}
                 withStore' c $ \db -> supervisorUpdateNtfSub db sub' (NtfSubSMPAction NSASmpKey)
                 void $ getNtfSMPWorker True c smpServer
@@ -143,8 +143,8 @@ getNtfSMPWorker hasWork c server = do
   ws <- asks $ ntfSMPWorkers . ntfSupervisor
   getAgentWorker "ntf_smp" hasWork c server ws $ runNtfSMPWorker c server
 
-withNtfServer :: AgentMonad' m => AgentClient -> (NtfServer -> m ()) -> m ()
-withNtfServer c action = getNtfServer c >>= mapM_ action
+withTokenServer :: AgentMonad' m => (NtfServer -> m ()) -> m ()
+withTokenServer action = getNtfToken >>= mapM_ (\NtfToken {ntfServer} -> action ntfServer)
 
 runNtfWorker :: forall m. AgentMonad m => AgentClient -> NtfServer -> Worker -> m ()
 runNtfWorker c srv Worker {doWork} = do

src/Simplex/Messaging/Agent/Protocol.hs

@@ -97,7 +97,7 @@ module Simplex.Messaging.Agent.Protocol
     AConnectionRequestUri (..),
     ConnReqUriData (..),
     CRClientData,
-    ConnReqScheme (..),
+    ServiceScheme,
     simplexChat,
     AgentErrorType (..),
     CommandErrorType (..),
@@ -197,7 +197,6 @@ import Simplex.Messaging.Protocol
     SMPServer,
     SMPServerWithAuth,
     SndPublicVerifyKey,
-    SrvLoc (..),
     SubscriptionMode,
     legacyEncodeServer,
     legacyServerP,
@@ -208,6 +207,7 @@ import Simplex.Messaging.Protocol
     pattern SMPServer,
   )
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.ServiceScheme
 import Simplex.Messaging.Transport (Transport (..), TransportError, serializeTransportError, transportErrorP)
 import Simplex.Messaging.Transport.Client (TransportHost, TransportHosts_ (..))
 import Simplex.Messaging.Util
@@ -337,6 +337,7 @@ data ACommand (p :: AParty) (e :: AEntity) where
   MID :: AgentMsgId -> ACommand Agent AEConn
   SENT :: AgentMsgId -> ACommand Agent AEConn
   MERR :: AgentMsgId -> AgentErrorType -> ACommand Agent AEConn
+  MERRS :: NonEmpty AgentMsgId -> AgentErrorType -> ACommand Agent AEConn
   MSG :: MsgMeta -> MsgFlags -> MsgBody -> ACommand Agent AEConn
   MSGNTF :: SMPMsgMeta -> ACommand Agent AEConn
   ACK :: AgentMsgId -> Maybe MsgReceiptInfo -> ACommand Client AEConn
@@ -398,6 +399,7 @@ data ACommandTag (p :: AParty) (e :: AEntity) where
   MID_ :: ACommandTag Agent AEConn
   SENT_ :: ACommandTag Agent AEConn
   MERR_ :: ACommandTag Agent AEConn
+  MERRS_ :: ACommandTag Agent AEConn
   MSG_ :: ACommandTag Agent AEConn
   MSGNTF_ :: ACommandTag Agent AEConn
   ACK_ :: ACommandTag Client AEConn
@@ -452,6 +454,7 @@ aCommandTag = \case
   MID _ -> MID_
   SENT _ -> SENT_
   MERR {} -> MERR_
+  MERRS {} -> MERRS_
   MSG {} -> MSG_
   MSGNTF {} -> MSGNTF_
   ACK {} -> ACK_
@@ -1120,13 +1123,13 @@ instance forall m. ConnectionModeI m => StrEncoding (ConnectionRequestUri m) whe
 instance StrEncoding AConnectionRequestUri where
   strEncode (ACR _ cr) = strEncode cr
   strP = do
-    _crScheme :: ConnReqScheme <- strP
+    _crScheme :: ServiceScheme <- strP
     crMode <- A.char '/' *> crModeP <* optional (A.char '/') <* "#/?"
     query <- strP
     crAgentVRange <- queryParam "v" query
     crSmpQueues <- queryParam "smp" query
     let crClientData = safeDecodeUtf8 <$> queryParamStr "data" query
-    let crData = ConnReqUriData {crScheme = CRSSimplex, crAgentVRange, crSmpQueues, crClientData}
+    let crData = ConnReqUriData {crScheme = SSSimplex, crAgentVRange, crSmpQueues, crClientData}
     case crMode of
       CMInvitation -> do
         crE2eParams <- queryParam "e2e" query
@@ -1325,7 +1328,7 @@ instance Eq AConnectionRequestUri where
 deriving instance Show AConnectionRequestUri
 
 data ConnReqUriData = ConnReqUriData
-  { crScheme :: ConnReqScheme,
+  { crScheme :: ServiceScheme,
     crAgentVRange :: VersionRange,
     crSmpQueues :: NonEmpty SMPQueueUri,
     crClientData :: Maybe CRClientData
@@ -1334,20 +1337,6 @@ data ConnReqUriData = ConnReqUriData
 
 type CRClientData = Text
 
-data ConnReqScheme = CRSSimplex | CRSAppServer SrvLoc
-  deriving (Eq, Show)
-
-instance StrEncoding ConnReqScheme where
-  strEncode = \case
-    CRSSimplex -> "simplex:"
-    CRSAppServer srv -> "https://" <> strEncode srv
-  strP =
-    "simplex:" $> CRSSimplex
-      <|> "https://" *> (CRSAppServer <$> strP)
-
-simplexChat :: ConnReqScheme
-simplexChat = CRSAppServer $ SrvLoc "simplex.chat" ""
-
 -- | SMP queue status.
 data QueueStatus
   = -- | queue is created
@@ -1611,6 +1600,7 @@ instance StrEncoding ACmdTag where
       "MID" -> ct MID_
       "SENT" -> ct SENT_
       "MERR" -> ct MERR_
+      "MERRS" -> ct MERRS_
       "MSG" -> ct MSG_
       "MSGNTF" -> ct MSGNTF_
       "ACK" -> t ACK_
@@ -1667,6 +1657,7 @@ instance (APartyI p, AEntityI e) => StrEncoding (ACommandTag p e) where
     MID_ -> "MID"
     SENT_ -> "SENT"
     MERR_ -> "MERR"
+    MERRS_ -> "MERRS"
     MSG_ -> "MSG"
     MSGNTF_ -> "MSGNTF"
     ACK_ -> "ACK"
@@ -1736,6 +1727,7 @@ commandP binaryP =
           MID_ -> s (MID <$> A.decimal)
           SENT_ -> s (SENT <$> A.decimal)
           MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP)
+          MERRS_ -> s (MERRS <$> strP_ <*> strP)
           MSG_ -> s (MSG <$> strP <* A.space <*> smpP <* A.space <*> binaryP)
           MSGNTF_ -> s (MSGNTF <$> strP)
           RCVD_ -> s (RCVD <$> strP <* A.space <*> strP)
@@ -1788,12 +1780,13 @@ serializeCommand = \case
   SWITCH dir phase srvs -> s (SWITCH_, dir, phase, srvs)
   RSYNC rrState cryptoErr cstats -> s (RSYNC_, rrState, cryptoErr, cstats)
   SEND msgFlags msgBody -> B.unwords [s SEND_, smpEncode msgFlags, serializeBinary msgBody]
-  MID mId -> s (MID_, Str $ bshow mId)
-  SENT mId -> s (SENT_, Str $ bshow mId)
-  MERR mId e -> s (MERR_, Str $ bshow mId, e)
+  MID mId -> s (MID_, mId)
+  SENT mId -> s (SENT_, mId)
+  MERR mId e -> s (MERR_, mId, e)
+  MERRS mIds e -> s (MERRS_, mIds, e)
   MSG msgMeta msgFlags msgBody -> B.unwords [s MSG_, s msgMeta, smpEncode msgFlags, serializeBinary msgBody]
   MSGNTF smpMsgMeta -> s (MSGNTF_, smpMsgMeta)
-  ACK mId rcptInfo_ -> s (ACK_, Str $ bshow mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_
+  ACK mId rcptInfo_ -> s (ACK_, mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_
   RCVD msgMeta rcpts -> s (RCVD_, msgMeta, rcpts)
   SWCH -> s SWCH_
   OFF -> s OFF_

src/Simplex/Messaging/Agent/Server.hs

@@ -34,23 +34,25 @@ import UnliftIO.STM
 -- See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs
 runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> InitialAgentServers -> SQLiteStore -> m ()
 runSMPAgent t cfg initServers store =
-  runSMPAgentBlocking t cfg initServers store =<< newEmptyTMVarIO
+  runSMPAgentBlocking t cfg initServers store 0 =<< newEmptyTMVarIO
 
 -- | Runs an SMP agent as a TCP service using passed configuration with signalling.
 --
 -- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
 -- and when it is disconnected from the TCP socket once the server thread is killed (False).
-runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> InitialAgentServers -> SQLiteStore -> TMVar Bool -> m ()
-runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} initServers store started = do
+runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Int -> TMVar Bool -> m ()
+runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} initServers store initClientId started = do
   liftIO (newSMPAgentEnv cfg store) >>= runReaderT (smpAgent t)
   where
     smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()
     smpAgent _ = do
       -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
       tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+      clientId <- newTVarIO initClientId
       runTransportServer started tcpPort tlsServerParams defaultTransportServerConfig $ \(h :: c) -> do
         liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
-        c <- getAgentClient initServers
+        cId <- atomically $ stateTVar clientId $ \i -> (i + 1, i + 1)
+        c <- getAgentClient cId initServers
         logConnection c True
         race_ (connectClient h c) (runAgentClient c)
           `E.finally` disconnectAgentClient c

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

@@ -110,6 +110,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     getPendingQueueMsg,
     updatePendingMsgRIState,
     deletePendingMsgs,
+    getExpiredSndMessages,
     setMsgUserAck,
     getRcvMsg,
     getLastMsg,
@@ -163,6 +164,7 @@ module Simplex.Messaging.Agent.Store.SQLite
 
     -- Rcv files
     createRcvFile,
+    createRcvFileRedirect,
     getRcvFile,
     getRcvFileByEntityId,
     updateRcvChunkReplicaDelay,
@@ -170,6 +172,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     updateRcvFileStatus,
     updateRcvFileError,
     updateRcvFileComplete,
+    updateRcvFileRedirect,
     updateRcvFileNoTmpPath,
     updateRcvFileDeleted,
     deleteRcvFile',
@@ -230,6 +233,7 @@ import Data.ByteArray (ScrubbedBytes)
 import qualified Data.ByteArray as BA
 import Data.ByteString (ByteString)
 import qualified Data.ByteString.Base64.URL as U
+import qualified Data.ByteString.Char8 as B
 import Data.Char (toLower)
 import Data.Functor (($>))
 import Data.IORef
@@ -254,7 +258,7 @@ import qualified Database.SQLite3 as SQLite3
 import Network.Socket (ServiceName)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..))
 import Simplex.FileTransfer.Description
-import Simplex.FileTransfer.Protocol (FileParty (..))
+import Simplex.FileTransfer.Protocol (FileParty (..), SFileParty (..))
 import Simplex.FileTransfer.Types
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval (RI2State (..))
@@ -1041,6 +1045,33 @@ deletePendingMsgs :: DB.Connection -> ConnId -> SndQueue -> IO ()
 deletePendingMsgs db connId SndQueue {dbQueueId} =
   DB.execute db "DELETE FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = ?" (connId, dbQueueId)
 
+getExpiredSndMessages :: DB.Connection -> ConnId -> SndQueue -> UTCTime -> IO [InternalId]
+getExpiredSndMessages db connId SndQueue {dbQueueId} expireTs = do
+  -- type is Maybe InternalId because MAX always returns one row, possibly with NULL value
+  maxId :: [Maybe InternalId] <-
+    map fromOnly
+      <$> DB.query
+        db
+        [sql|
+          SELECT MAX(internal_id)
+          FROM messages
+          WHERE conn_id = ? AND internal_snd_id IS NOT NULL AND internal_ts < ?
+        |]
+        (connId, expireTs)
+  case maxId of
+    Just msgId : _ ->
+      map fromOnly
+        <$> DB.query
+          db
+          [sql|
+            SELECT internal_id
+            FROM snd_message_deliveries
+            WHERE conn_id = ? AND snd_queue_id = ? AND failed = 0 AND internal_id <= ?
+            ORDER BY internal_id ASC
+          |]
+          (connId, dbQueueId, msgId)
+    _ -> pure []
+
 setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (RcvQueue, SMP.MsgId))
 setMsgUserAck db connId agentMsgId = runExceptT $ do
   (dbRcvId, srvMsgId) <-
@@ -2235,38 +2266,66 @@ getXFTPServerId_ db ProtocolServer {host, port, keyHash} = do
     DB.query db "SELECT xftp_server_id FROM xftp_servers WHERE xftp_host = ? AND xftp_port = ? AND xftp_key_hash = ?" (host, port, keyHash)
 
 createRcvFile :: DB.Connection -> TVar ChaChaDRG -> UserId -> FileDescription 'FRecipient -> FilePath -> FilePath -> CryptoFile -> IO (Either StoreError RcvFileId)
-createRcvFile db gVar userId fd@FileDescription {chunks} prefixPath tmpPath (CryptoFile savePath cfArgs) = runExceptT $ do
-  (rcvFileEntityId, rcvFileId) <- ExceptT $ insertRcvFile fd
+createRcvFile db gVar userId fd@FileDescription {chunks} prefixPath tmpPath file = runExceptT $ do
+  (rcvFileEntityId, rcvFileId) <- ExceptT $ insertRcvFile db gVar userId fd prefixPath tmpPath file Nothing Nothing
   liftIO $
     forM_ chunks $ \fc@FileChunk {replicas} -> do
-      chunkId <- insertChunk fc rcvFileId
-      forM_ (zip [1 ..] replicas) $ \(rno, replica) -> insertReplica rno replica chunkId
+      chunkId <- insertRcvFileChunk db fc rcvFileId
+      forM_ (zip [1 ..] replicas) $ \(rno, replica) -> insertRcvFileChunkReplica db rno replica chunkId
   pure rcvFileEntityId
+
+createRcvFileRedirect :: DB.Connection -> TVar ChaChaDRG -> UserId -> FileDescription FRecipient -> FilePath -> FilePath -> CryptoFile -> FilePath -> CryptoFile -> IO (Either StoreError RcvFileId)
+createRcvFileRedirect _ _ _ FileDescription {redirect = Nothing} _ _ _ _ _ = pure $ Left $ SEInternal "createRcvFileRedirect called without redirect"
+createRcvFileRedirect db gVar userId redirectFd@FileDescription {chunks = redirectChunks, redirect = Just RedirectFileInfo {size, digest}} prefixPath redirectPath redirectFile dstPath dstFile = runExceptT $ do
+  (dstEntityId, dstId) <- ExceptT $ insertRcvFile db gVar userId dummyDst prefixPath dstPath dstFile Nothing Nothing
+  (_, redirectId) <- ExceptT $ insertRcvFile db gVar userId redirectFd prefixPath redirectPath redirectFile (Just dstId) (Just dstEntityId)
+  liftIO $
+    forM_ redirectChunks $ \fc@FileChunk {replicas} -> do
+      chunkId <- insertRcvFileChunk db fc redirectId
+      forM_ (zip [1 ..] replicas) $ \(rno, replica) -> insertRcvFileChunkReplica db rno replica chunkId
+  pure dstEntityId
   where
-    insertRcvFile :: FileDescription 'FRecipient -> IO (Either StoreError (RcvFileId, DBRcvFileId))
-    insertRcvFile FileDescription {size, digest, key, nonce, chunkSize} = runExceptT $ do
-      rcvFileEntityId <- ExceptT $
-        createWithRandomId gVar $ \rcvFileEntityId ->
-          DB.execute
-            db
-            "INSERT INTO rcv_files (rcv_file_entity_id, user_id, size, digest, key, nonce, chunk_size, prefix_path, tmp_path, save_path, save_file_key, save_file_nonce, status) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)"
-            ((rcvFileEntityId, userId, size, digest, key, nonce, chunkSize) :. (prefixPath, tmpPath, savePath, fileKey <$> cfArgs, fileNonce <$> cfArgs, RFSReceiving))
-      rcvFileId <- liftIO $ insertedRowId db
-      pure (rcvFileEntityId, rcvFileId)
-    insertChunk :: FileChunk -> DBRcvFileId -> IO Int64
-    insertChunk FileChunk {chunkNo, chunkSize, digest} rcvFileId = do
+    dummyDst = FileDescription
+      { party = SFRecipient,
+        size,
+        digest,
+        redirect = Nothing,
+        -- updated later with updateRcvFileRedirect
+        key = C.unsafeSbKey $ B.replicate 32 '#',
+        nonce = C.cbNonce "",
+        chunkSize = FileSize 0,
+        chunks = []
+      }
+
+insertRcvFile :: DB.Connection -> TVar ChaChaDRG -> UserId -> FileDescription 'FRecipient -> FilePath -> FilePath -> CryptoFile -> Maybe DBRcvFileId -> Maybe RcvFileId -> IO (Either StoreError (RcvFileId, DBRcvFileId))
+insertRcvFile db gVar userId FileDescription {size, digest, key, nonce, chunkSize, redirect} prefixPath tmpPath (CryptoFile savePath cfArgs) redirectId_ redirectEntityId_ = runExceptT $ do
+  let (redirectDigest_, redirectSize_) = case redirect of
+        Just RedirectFileInfo {digest = d, size = s} -> (Just d, Just s)
+        Nothing -> (Nothing, Nothing)
+  rcvFileEntityId <- ExceptT $
+    createWithRandomId gVar $ \rcvFileEntityId ->
       DB.execute
         db
-        "INSERT INTO rcv_file_chunks (rcv_file_id, chunk_no, chunk_size, digest) VALUES (?,?,?,?)"
-        (rcvFileId, chunkNo, chunkSize, digest)
-      insertedRowId db
-    insertReplica :: Int -> FileChunkReplica -> Int64 -> IO ()
-    insertReplica replicaNo FileChunkReplica {server, replicaId, replicaKey} chunkId = do
-      srvId <- createXFTPServer_ db server
-      DB.execute
-        db
-        "INSERT INTO rcv_file_chunk_replicas (replica_number, rcv_file_chunk_id, xftp_server_id, replica_id, replica_key) VALUES (?,?,?,?,?)"
-        (replicaNo, chunkId, srvId, replicaId, replicaKey)
+        "INSERT INTO rcv_files (rcv_file_entity_id, user_id, size, digest, key, nonce, chunk_size, prefix_path, tmp_path, save_path, save_file_key, save_file_nonce, status, redirect_id, redirect_entity_id, redirect_digest, redirect_size) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
+        ((rcvFileEntityId, userId, size, digest, key, nonce, chunkSize, prefixPath, tmpPath) :. (savePath, fileKey <$> cfArgs, fileNonce <$> cfArgs, RFSReceiving, redirectId_, redirectEntityId_, redirectDigest_, redirectSize_))
+  rcvFileId <- liftIO $ insertedRowId db
+  pure (rcvFileEntityId, rcvFileId)
+
+insertRcvFileChunk :: DB.Connection -> FileChunk -> DBRcvFileId -> IO Int64
+insertRcvFileChunk db FileChunk {chunkNo, chunkSize, digest} rcvFileId = do
+  DB.execute
+    db
+    "INSERT INTO rcv_file_chunks (rcv_file_id, chunk_no, chunk_size, digest) VALUES (?,?,?,?)"
+    (rcvFileId, chunkNo, chunkSize, digest)
+  insertedRowId db
+
+insertRcvFileChunkReplica :: DB.Connection -> Int -> FileChunkReplica -> Int64 -> IO ()
+insertRcvFileChunkReplica db replicaNo FileChunkReplica {server, replicaId, replicaKey} chunkId = do
+  srvId <- createXFTPServer_ db server
+  DB.execute
+    db
+    "INSERT INTO rcv_file_chunk_replicas (replica_number, rcv_file_chunk_id, xftp_server_id, replica_id, replica_key) VALUES (?,?,?,?,?)"
+    (replicaNo, chunkId, srvId, replicaId, replicaKey)
 
 getRcvFileByEntityId :: DB.Connection -> RcvFileId -> IO (Either StoreError RcvFile)
 getRcvFileByEntityId db rcvFileEntityId = runExceptT $ do
@@ -2290,17 +2349,21 @@ getRcvFile db rcvFileId = runExceptT $ do
         DB.query
           db
           [sql|
-            SELECT rcv_file_entity_id, user_id, size, digest, key, nonce, chunk_size, prefix_path, tmp_path, save_path, save_file_key, save_file_nonce, status, deleted
+            SELECT rcv_file_entity_id, user_id, size, digest, key, nonce, chunk_size, prefix_path, tmp_path, save_path, save_file_key, save_file_nonce, status, deleted, redirect_id, redirect_entity_id, redirect_size, redirect_digest
             FROM rcv_files
             WHERE rcv_file_id = ?
           |]
           (Only rcvFileId)
       where
-        toFile :: (RcvFileId, UserId, FileSize Int64, FileDigest, C.SbKey, C.CbNonce, FileSize Word32, FilePath, Maybe FilePath) :. (FilePath, Maybe C.SbKey, Maybe C.CbNonce, RcvFileStatus, Bool) -> RcvFile
-        toFile ((rcvFileEntityId, userId, size, digest, key, nonce, chunkSize, prefixPath, tmpPath) :. (savePath, saveKey_, saveNonce_, status, deleted)) =
+        toFile :: (RcvFileId, UserId, FileSize Int64, FileDigest, C.SbKey, C.CbNonce, FileSize Word32, FilePath, Maybe FilePath) :. (FilePath, Maybe C.SbKey, Maybe C.CbNonce, RcvFileStatus, Bool, Maybe DBRcvFileId, Maybe RcvFileId, Maybe (FileSize Int64), Maybe FileDigest) -> RcvFile
+        toFile ((rcvFileEntityId, userId, size, digest, key, nonce, chunkSize, prefixPath, tmpPath) :. (savePath, saveKey_, saveNonce_, status, deleted, redirectDbId, redirectEntityId, redirectSize_, redirectDigest_)) =
           let cfArgs = CFArgs <$> saveKey_ <*> saveNonce_
               saveFile = CryptoFile savePath cfArgs
-           in RcvFile {rcvFileId, rcvFileEntityId, userId, size, digest, key, nonce, chunkSize, prefixPath, tmpPath, saveFile, status, deleted, chunks = []}
+              redirect = RcvFileRedirect
+                <$> redirectDbId
+                <*> redirectEntityId
+                <*> (RedirectFileInfo <$> redirectSize_ <*> redirectDigest_)
+           in RcvFile {rcvFileId, rcvFileEntityId, userId, size, digest, key, nonce, chunkSize, redirect, prefixPath, tmpPath, saveFile, status, deleted, chunks = []}
     getChunks :: RcvFileId -> UserId -> FilePath -> IO [RcvFileChunk]
     getChunks rcvFileEntityId userId fileTmpPath = do
       chunks <-
@@ -2366,6 +2429,14 @@ updateRcvFileComplete db rcvFileId = do
   updatedAt <- getCurrentTime
   DB.execute db "UPDATE rcv_files SET tmp_path = NULL, status = ?, updated_at = ? WHERE rcv_file_id = ?" (RFSComplete, updatedAt, rcvFileId)
 
+updateRcvFileRedirect :: DB.Connection -> DBRcvFileId -> FileDescription 'FRecipient -> IO (Either StoreError ())
+updateRcvFileRedirect db rcvFileId FileDescription {key, nonce, chunkSize, chunks} = runExceptT $ do
+  updatedAt <- liftIO getCurrentTime
+  liftIO $ DB.execute db "UPDATE rcv_files SET key = ?, nonce = ?, chunk_size = ?, updated_at = ? WHERE rcv_file_id = ?" (key, nonce, chunkSize, updatedAt, rcvFileId)
+  liftIO $ forM_ chunks $ \fc@FileChunk {replicas} -> do
+    chunkId <- insertRcvFileChunk db fc rcvFileId
+    forM_ (zip [1 ..] replicas) $ \(rno, replica) -> insertRcvFileChunkReplica db rno replica chunkId
+
 updateRcvFileNoTmpPath :: DB.Connection -> DBRcvFileId -> IO ()
 updateRcvFileNoTmpPath db rcvFileId = do
   updatedAt <- getCurrentTime
@@ -2513,13 +2584,18 @@ getRcvFilesExpired db ttl = do
     |]
     (Only cutoffTs)
 
-createSndFile :: DB.Connection -> TVar ChaChaDRG -> UserId -> CryptoFile -> Int -> FilePath -> C.SbKey -> C.CbNonce -> IO (Either StoreError SndFileId)
-createSndFile db gVar userId (CryptoFile path cfArgs) numRecipients prefixPath key nonce =
+createSndFile :: DB.Connection -> TVar ChaChaDRG -> UserId -> CryptoFile -> Int -> FilePath -> C.SbKey -> C.CbNonce -> Maybe RedirectFileInfo -> IO (Either StoreError SndFileId)
+createSndFile db gVar userId (CryptoFile path cfArgs) numRecipients prefixPath key nonce redirect_ =
   createWithRandomId gVar $ \sndFileEntityId ->
     DB.execute
       db
-      "INSERT INTO snd_files (snd_file_entity_id, user_id, path, src_file_key, src_file_nonce, num_recipients, prefix_path, key, nonce, status) VALUES (?,?,?,?,?,?,?,?,?,?)"
-      (sndFileEntityId, userId, path, fileKey <$> cfArgs, fileNonce <$> cfArgs, numRecipients, prefixPath, key, nonce, SFSNew)
+      "INSERT INTO snd_files (snd_file_entity_id, user_id, path, src_file_key, src_file_nonce, num_recipients, prefix_path, key, nonce, status, redirect_size, redirect_digest) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)"
+      ((sndFileEntityId, userId, path, fileKey <$> cfArgs, fileNonce <$> cfArgs, numRecipients) :. (prefixPath, key, nonce, SFSNew, redirectSize_, redirectDigest_))
+  where
+    (redirectSize_, redirectDigest_) =
+      case redirect_ of
+        Nothing -> (Nothing, Nothing)
+        Just RedirectFileInfo {size, digest} -> (Just size, Just digest)
 
 getSndFileByEntityId :: DB.Connection -> SndFileId -> IO (Either StoreError SndFile)
 getSndFileByEntityId db sndFileEntityId = runExceptT $ do
@@ -2543,17 +2619,18 @@ getSndFile db sndFileId = runExceptT $ do
         DB.query
           db
           [sql|
-            SELECT snd_file_entity_id, user_id, path, src_file_key, src_file_nonce, num_recipients, digest, prefix_path, key, nonce, status, deleted
+            SELECT snd_file_entity_id, user_id, path, src_file_key, src_file_nonce, num_recipients, digest, prefix_path, key, nonce, status, deleted, redirect_size, redirect_digest
             FROM snd_files
             WHERE snd_file_id = ?
           |]
           (Only sndFileId)
       where
-        toFile :: (SndFileId, UserId, FilePath, Maybe C.SbKey, Maybe C.CbNonce, Int, Maybe FileDigest, Maybe FilePath, C.SbKey, C.CbNonce, SndFileStatus, Bool) -> SndFile
-        toFile (sndFileEntityId, userId, srcPath, srcKey_, srcNonce_, numRecipients, digest, prefixPath, key, nonce, status, deleted) =
+        toFile :: (SndFileId, UserId, FilePath, Maybe C.SbKey, Maybe C.CbNonce, Int, Maybe FileDigest, Maybe FilePath, C.SbKey, C.CbNonce) :. (SndFileStatus, Bool, Maybe (FileSize Int64), Maybe FileDigest) -> SndFile
+        toFile ((sndFileEntityId, userId, srcPath, srcKey_, srcNonce_, numRecipients, digest, prefixPath, key, nonce) :. (status, deleted, redirectSize_, redirectDigest_)) =
           let cfArgs = CFArgs <$> srcKey_ <*> srcNonce_
               srcFile = CryptoFile srcPath cfArgs
-           in SndFile {sndFileId, sndFileEntityId, userId, srcFile, numRecipients, digest, prefixPath, key, nonce, status, deleted, chunks = []}
+              redirect = RedirectFileInfo <$> redirectSize_ <*> redirectDigest_
+           in SndFile {sndFileId, sndFileEntityId, userId, srcFile, numRecipients, digest, prefixPath, key, nonce, status, deleted, redirect, chunks = []}
     getChunks :: SndFileId -> UserId -> Int -> FilePath -> IO [SndFileChunk]
     getChunks sndFileEntityId userId numRecipients filePrefixPath = do
       chunks <-

src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs

@@ -67,6 +67,8 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230814_indexes
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230829_crypto_files
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231222_command_created_at
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231225_failed_work_items
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240121_message_delivery_indexes
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240124_file_redirect
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -102,7 +104,9 @@ schemaMigrations =
     ("m20230814_indexes", m20230814_indexes, Just down_m20230814_indexes),
     ("m20230829_crypto_files", m20230829_crypto_files, Just down_m20230829_crypto_files),
     ("m20231222_command_created_at", m20231222_command_created_at, Just down_m20231222_command_created_at),
-    ("m20231225_failed_work_items", m20231225_failed_work_items, Just down_m20231225_failed_work_items)
+    ("m20231225_failed_work_items", m20231225_failed_work_items, Just down_m20231225_failed_work_items),
+    ("m20240121_message_delivery_indexes", m20240121_message_delivery_indexes, Just down_m20240121_message_delivery_indexes),
+    ("m20240124_file_redirect", m20240124_file_redirect, Just down_m20240124_file_redirect)
   ]
 
 -- | The list of migrations in ascending order by date
@@ -128,9 +132,12 @@ run st = \case
   where
     runUp Migration {name, up, down} = withTransaction' st $ \db -> do
       when (name == "m20220811_onion_hosts") $ updateServers db
-      insert db >> execSQL db up
+      insert db >> execSQL db up'
       where
         insert db = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
+        up'
+          | dbNew st && name == "m20230110_users" = fromQuery new_m20230110_users
+          | otherwise = up
         updateServers db = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
           let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
            in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230110_users.hs

@@ -27,3 +27,24 @@ UPDATE connections SET user_id = 1;
 
 PRAGMA ignore_check_constraints=OFF;
 |]
+
+-- This is executed in the new database
+-- It does not create new user record
+new_m20230110_users :: Query
+new_m20230110_users =
+  [sql|
+PRAGMA ignore_check_constraints=ON;
+
+CREATE TABLE users (
+  user_id INTEGER PRIMARY KEY AUTOINCREMENT
+);
+
+ALTER TABLE connections ADD COLUMN user_id INTEGER CHECK (user_id NOT NULL)
+REFERENCES users ON DELETE CASCADE;
+
+CREATE INDEX idx_connections_user ON connections(user_id);
+
+CREATE INDEX idx_commands_conn_id ON commands(conn_id);
+
+PRAGMA ignore_check_constraints=OFF;
+|]

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20240121_message_delivery_indexes.hs

@@ -0,0 +1,20 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240121_message_delivery_indexes where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20240121_message_delivery_indexes :: Query
+m20240121_message_delivery_indexes =
+    [sql|
+CREATE INDEX idx_messages_snd_expired ON messages(conn_id, internal_snd_id, internal_ts);
+CREATE INDEX idx_snd_message_deliveries_expired ON snd_message_deliveries(conn_id, snd_queue_id, failed, internal_id);
+|]
+
+down_m20240121_message_delivery_indexes :: Query
+down_m20240121_message_delivery_indexes =
+    [sql|
+DROP INDEX idx_messages_snd_expired;
+DROP INDEX idx_snd_message_deliveries_expired;
+|]

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20240124_file_redirect.hs

@@ -0,0 +1,34 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240124_file_redirect where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20240124_file_redirect :: Query
+m20240124_file_redirect =
+    [sql|
+ALTER TABLE snd_files ADD COLUMN redirect_size INTEGER;
+ALTER TABLE snd_files ADD COLUMN redirect_digest BLOB;
+
+ALTER TABLE rcv_files ADD COLUMN redirect_id INTEGER REFERENCES rcv_files ON DELETE CASCADE;
+ALTER TABLE rcv_files ADD COLUMN redirect_entity_id BLOB;
+ALTER TABLE rcv_files ADD COLUMN redirect_size INTEGER;
+ALTER TABLE rcv_files ADD COLUMN redirect_digest BLOB;
+
+CREATE INDEX idx_rcv_files_redirect_id on rcv_files(redirect_id);
+|]
+
+down_m20240124_file_redirect :: Query
+down_m20240124_file_redirect =
+    [sql|
+DROP INDEX idx_rcv_files_redirect_id;
+
+ALTER TABLE snd_files DROP COLUMN redirect_size;
+ALTER TABLE snd_files DROP COLUMN redirect_digest;
+
+ALTER TABLE rcv_files DROP COLUMN redirect_id;
+ALTER TABLE rcv_files DROP COLUMN redirect_entity_id;
+ALTER TABLE rcv_files DROP COLUMN redirect_size;
+ALTER TABLE rcv_files DROP COLUMN redirect_digest;
+|]

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -279,6 +279,10 @@ CREATE TABLE rcv_files(
   save_file_key BLOB,
   save_file_nonce BLOB,
   failed INTEGER DEFAULT 0,
+  redirect_id INTEGER REFERENCES rcv_files ON DELETE CASCADE,
+  redirect_entity_id BLOB,
+  redirect_size INTEGER,
+  redirect_digest BLOB,
   UNIQUE(rcv_file_entity_id)
 );
 CREATE TABLE rcv_file_chunks(
@@ -322,7 +326,9 @@ CREATE TABLE snd_files(
   ,
   src_file_key BLOB,
   src_file_nonce BLOB,
-  failed INTEGER DEFAULT 0
+  failed INTEGER DEFAULT 0,
+  redirect_size INTEGER,
+  redirect_digest BLOB
 );
 CREATE TABLE snd_file_chunks(
   snd_file_chunk_id INTEGER PRIMARY KEY,
@@ -497,3 +503,15 @@ CREATE INDEX idx_commands_server_commands ON commands(
 CREATE INDEX idx_rcv_files_status_created_at ON rcv_files(status, created_at);
 CREATE INDEX idx_snd_files_status_created_at ON snd_files(status, created_at);
 CREATE INDEX idx_snd_files_snd_file_entity_id ON snd_files(snd_file_entity_id);
+CREATE INDEX idx_messages_snd_expired ON messages(
+  conn_id,
+  internal_snd_id,
+  internal_ts
+);
+CREATE INDEX idx_snd_message_deliveries_expired ON snd_message_deliveries(
+  conn_id,
+  snd_queue_id,
+  failed,
+  internal_id
+);
+CREATE INDEX idx_rcv_files_redirect_id on rcv_files(redirect_id);

src/Simplex/Messaging/Agent/TAsyncs.hs

@@ -1,24 +0,0 @@
-module Simplex.Messaging.Agent.TAsyncs where
-
-import Control.Monad.IO.Unlift (MonadUnliftIO)
-import Simplex.Messaging.TMap (TMap)
-import qualified Simplex.Messaging.TMap as TM
-import UnliftIO.Async (Async, async)
-import UnliftIO.STM
-
-data TAsyncs = TAsyncs
-  { actionId :: TVar Int,
-    actions :: TMap Int (Async ())
-  }
-
-newTAsyncs :: STM TAsyncs
-newTAsyncs = TAsyncs <$> newTVar 0 <*> TM.empty
-
-newAsyncAction :: MonadUnliftIO m => (Int -> m ()) -> TAsyncs -> m ()
-newAsyncAction action as = do
-  aId <- atomically $ stateTVar (actionId as) $ \i -> (i + 1, i + 1)
-  a <- async $ action aId
-  atomically $ TM.insert aId a $ actions as
-
-removeAsyncAction :: Int -> TAsyncs -> STM ()
-removeAsyncAction aId = TM.delete aId . actions

src/Simplex/Messaging/Client.hs

@@ -72,9 +72,7 @@ module Simplex.Messaging.Client
     ClientCommand,
 
     -- * For testing
-    ClientBatch (..),
     PCTransmission,
-    batchClientTransmissions,
     mkTransmission,
     clientStub,
   )
@@ -99,10 +97,9 @@ import Data.Time.Clock (UTCTime, getCurrentTime)
 import Network.Socket (ServiceName)
 import Numeric.Natural
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, enumJSON)
-import Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Protocol
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
@@ -173,7 +170,7 @@ clientStub sessionId = do
             }
       }
 
-type SMPClient = ProtocolClient ErrorType SMP.BrokerMsg
+type SMPClient = ProtocolClient ErrorType BrokerMsg
 
 -- | Type for client command data
 type ClientCommand msg = (Maybe C.APrivateSignKey, EntityId, ProtoCommand msg)
@@ -349,12 +346,12 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
       action <-
         async $
           runTransportClient tcConfig (Just username) useHost port' (Just $ keyHash srv) (client t c cVar)
-            `finally` atomically (putTMVar cVar $ Left PCENetworkError)
+            `finally` atomically (tryPutTMVar cVar $ Left PCENetworkError)
       c_ <- tcpConnectTimeout `timeout` atomically (takeTMVar cVar)
-      pure $ case c_ of
-        Just (Right c') -> Right c' {action = Just action}
-        Just (Left e) -> Left e
-        Nothing -> Left PCENetworkError
+      case c_ of
+        Just (Right c') -> pure $ Right c' {action = Just action}
+        Just (Left e) -> pure $ Left e
+        Nothing -> cancel action $> Left PCENetworkError
 
     useTransport :: (ServiceName, ATransport)
     useTransport = case port srv of
@@ -634,7 +631,7 @@ type PCTransmission err msg = (SentRawTransmission, Request err msg)
 -- | Send multiple commands with batching and collect responses
 sendProtocolCommands :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
 sendProtocolCommands c@ProtocolClient {batch, blockSize} cs = do
-  bs <- batchClientTransmissions batch blockSize <$> mapM (mkTransmission c) cs
+  bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs
   validate . concat =<< mapM (sendBatch c) bs
   where
     validate :: [Response err msg] -> IO (NonEmpty (Response err msg))
@@ -651,58 +648,24 @@ sendProtocolCommands c@ProtocolClient {batch, blockSize} cs = do
 
 streamProtocolCommands :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> NonEmpty (ClientCommand msg) -> ([Response err msg] -> IO ()) -> IO ()
 streamProtocolCommands c@ProtocolClient {batch, blockSize} cs cb = do
-  bs <- batchClientTransmissions batch blockSize <$> mapM (mkTransmission c) cs
+  bs <- batchTransmissions' batch blockSize <$> mapM (mkTransmission c) cs
   mapM_ (cb <=< sendBatch c) bs
 
-sendBatch :: ProtocolClient err msg -> ClientBatch err msg -> IO [Response err msg]
+sendBatch :: ProtocolClient err msg -> TransportBatch (Request err msg) -> IO [Response err msg]
 sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do
   case b of
-    CBLargeTransmission Request {entityId} -> do
+    TBLargeTransmission Request {entityId} -> do
       putStrLn "send error: large message"
       pure [Response entityId $ Left $ PCETransportError TELargeMsg]
-    CBTransmissions s n rs -> do
-      when (n > 0) $ atomically $ writeTBQueue sndQ $ tEncodeBatch n s
-      mapConcurrently (getResponse c) rs
-    CBTransmission s r -> do
+    TBTransmissions s n rs
+      | n > 0 -> do
+          atomically $ writeTBQueue sndQ s
+          mapConcurrently (getResponse c) rs
+      | otherwise -> pure []
+    TBTransmission s r -> do
       atomically $ writeTBQueue sndQ s
       (: []) <$> getResponse c r
 
-data ClientBatch err msg
-  = -- ByteString in CBTransmissions does not include count byte, it is added by tEncodeBatch
-    CBTransmissions ByteString Int [Request err msg]
-  | CBTransmission ByteString (Request err msg)
-  | CBLargeTransmission (Request err msg)
-
--- | encodes and batches transmissions into blocks
-batchClientTransmissions :: forall err msg. Bool -> Int -> NonEmpty (PCTransmission err msg) -> [ClientBatch err msg]
-batchClientTransmissions batch blkSize
-  | batch = reverse . mkBatch []
-  | otherwise = map mkBatch1 . L.toList
-  where
-    mkBatch :: [ClientBatch err msg] -> NonEmpty (PCTransmission err msg) -> [ClientBatch err msg]
-    mkBatch bs ts =
-      let (b, ts_) = encodeBatch "" 0 [] ts
-          bs' = b : bs
-       in maybe bs' (mkBatch bs') ts_
-    mkBatch1 :: PCTransmission err msg -> ClientBatch err msg
-    mkBatch1 (t, r)
-      | B.length s <= blkSize - 2 = CBTransmission s r
-      | otherwise = CBLargeTransmission r
-      where
-        s = tEncode t
-    encodeBatch :: ByteString -> Int -> [Request err msg] -> NonEmpty (PCTransmission err msg) -> (ClientBatch err msg, Maybe (NonEmpty (PCTransmission err msg)))
-    encodeBatch s n rs ts@((t, r) :| ts_)
-      | B.length s' <= blkSize - 3 && n < 255 =
-          case L.nonEmpty ts_ of
-            Just ts' -> encodeBatch s' n' rs' ts'
-            Nothing -> (CBTransmissions s' n' (reverse rs'), Nothing)
-      | n == 0 = (CBLargeTransmission r, L.nonEmpty ts_)
-      | otherwise = (CBTransmissions s n (reverse rs), Just ts)
-      where
-        s' = s <> smpEncode (Large $ tEncode t)
-        n' = n + 1
-        rs' = r : rs
-
 -- | Send Protocol command
 sendProtocolCommand :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> Maybe C.APrivateSignKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
 sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, batch, blockSize} pKey entId cmd =
@@ -715,7 +678,7 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, batch, blockSize
       | otherwise = atomically (writeTBQueue sndQ s) >> response <$> getResponse c r
       where
         s
-          | batch = tEncodeBatch 1 . smpEncode . Large $ tEncode t
+          | batch = tEncodeBatch1 t
           | otherwise = tEncode t
 
 -- TODO switch to timeout or TimeManager that supports Int64

src/Simplex/Messaging/Crypto/Lazy.hs

@@ -41,7 +41,6 @@ import qualified Crypto.MAC.Poly1305 as Poly1305
 import Data.Bifunctor (first)
 import Data.ByteArray (ByteArrayAccess)
 import qualified Data.ByteArray as BA
-import qualified Data.ByteString as S
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
@@ -175,7 +174,7 @@ secretBoxTailTag sbProcess secret nonce msg = run <$> sbInit_ secret nonce
 
 -- passes lazy bytestring via initialized secret box returning the reversed list of chunks
 secretBoxLazy_ :: (SbState -> ByteString -> (ByteString, SbState)) -> SbState -> LazyByteString -> ([ByteString], SbState)
-secretBoxLazy_ sbProcess state = foldlChunks update ([], state)
+secretBoxLazy_ sbProcess state = LB.foldlChunks update ([], state)
   where
     update (cs, st) chunk = let (!c, !st') = sbProcess st chunk in (c : cs, st')
 
@@ -231,10 +230,3 @@ cryptoPassed :: CE.CryptoFailable b -> Either CryptoError b
 cryptoPassed = \case
   CE.CryptoPassed a -> Right a
   CE.CryptoFailed e -> Left $ CryptoPoly1305Error e
-
-foldlChunks :: (a -> S.ByteString -> a) -> a -> LazyByteString -> a
-foldlChunks f = go
-  where
-    go !a LB.Empty = a
-    go !a (LB.Chunk c cs) = go (f a c) cs
-{-# INLINE foldlChunks #-}

src/Simplex/Messaging/Encoding.hs

@@ -174,37 +174,37 @@ instance (Encoding a, Encoding b) => Encoding (a, b) where
   {-# INLINE smpP #-}
 
 instance (Encoding a, Encoding b, Encoding c) => Encoding (a, b, c) where
-  smpEncode (a, b, c) = smpEncode a <> smpEncode b <> smpEncode c
+  smpEncode (a, b, c) = B.concat [smpEncode a, smpEncode b, smpEncode c]
   {-# INLINE smpEncode #-}
   smpP = (,,) <$> smpP <*> smpP <*> smpP
   {-# INLINE smpP #-}
 
 instance (Encoding a, Encoding b, Encoding c, Encoding d) => Encoding (a, b, c, d) where
-  smpEncode (a, b, c, d) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d
+  smpEncode (a, b, c, d) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d]
   {-# INLINE smpEncode #-}
   smpP = (,,,) <$> smpP <*> smpP <*> smpP <*> smpP
   {-# INLINE smpP #-}
 
 instance (Encoding a, Encoding b, Encoding c, Encoding d, Encoding e) => Encoding (a, b, c, d, e) where
-  smpEncode (a, b, c, d, e) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d <> smpEncode e
+  smpEncode (a, b, c, d, e) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d, smpEncode e]
   {-# INLINE smpEncode #-}
   smpP = (,,,,) <$> smpP <*> smpP <*> smpP <*> smpP <*> smpP
   {-# INLINE smpP #-}
 
 instance (Encoding a, Encoding b, Encoding c, Encoding d, Encoding e, Encoding f) => Encoding (a, b, c, d, e, f) where
-  smpEncode (a, b, c, d, e, f) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d <> smpEncode e <> smpEncode f
+  smpEncode (a, b, c, d, e, f) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d, smpEncode e, smpEncode f]
   {-# INLINE smpEncode #-}
   smpP = (,,,,,) <$> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP
   {-# INLINE smpP #-}
 
 instance (Encoding a, Encoding b, Encoding c, Encoding d, Encoding e, Encoding f, Encoding g) => Encoding (a, b, c, d, e, f, g) where
-  smpEncode (a, b, c, d, e, f, g) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d <> smpEncode e <> smpEncode f <> smpEncode g
+  smpEncode (a, b, c, d, e, f, g) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d, smpEncode e, smpEncode f, smpEncode g]
   {-# INLINE smpEncode #-}
   smpP = (,,,,,,) <$> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP
   {-# INLINE smpP #-}
 
 instance (Encoding a, Encoding b, Encoding c, Encoding d, Encoding e, Encoding f, Encoding g, Encoding h) => Encoding (a, b, c, d, e, f, g, h) where
-  smpEncode (a, b, c, d, e, f, g, h) = smpEncode a <> smpEncode b <> smpEncode c <> smpEncode d <> smpEncode e <> smpEncode f <> smpEncode g <> smpEncode h
+  smpEncode (a, b, c, d, e, f, g, h) = B.concat [smpEncode a, smpEncode b, smpEncode c, smpEncode d, smpEncode e, smpEncode f, smpEncode g, smpEncode h]
   {-# INLINE smpEncode #-}
   smpP = (,,,,,,,) <$> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP <*> smpP
   {-# INLINE smpP #-}

src/Simplex/Messaging/Notifications/Server.hs

@@ -109,9 +109,9 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
       liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
       NtfServerStats {fromTime, tknCreated, tknVerified, tknDeleted, subCreated, subDeleted, ntfReceived, ntfDelivered, activeTokens, activeSubs} <- asks serverStats
       let interval = 1000000 * logInterval
-      withFile statsFilePath AppendMode $ \h -> liftIO $ do
-        hSetBuffering h LineBuffering
-        forever $ do
+      forever $ do
+        withFile statsFilePath AppendMode $ \h -> liftIO $ do
+          hSetBuffering h LineBuffering
           ts <- getCurrentTime
           fromTime' <- atomically $ swapTVar fromTime ts
           tknCreated' <- atomically $ swapTVar tknCreated 0
@@ -141,7 +141,7 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
                 weekCount sub,
                 monthCount sub
               ]
-          threadDelay' interval
+        liftIO $ threadDelay' interval
 
 resubscribe :: NtfSubscriber -> Map NtfSubscriptionId NtfSubData -> M ()
 resubscribe NtfSubscriber {newSubQ} subs = do
@@ -369,7 +369,7 @@ receive th NtfServerClient {rcvQ, sndQ, rcvActiveAt} = forever $ do
 send :: Transport c => THandle c -> NtfServerClient -> IO ()
 send h@THandle {thVersion = v} NtfServerClient {sndQ, sessionId, sndActiveAt} = forever $ do
   t <- atomically $ readTBQueue sndQ
-  void . liftIO $ tPut h Nothing [(Nothing, encodeTransmission v sessionId t)]
+  void . liftIO $ tPut h [(Nothing, encodeTransmission v sessionId t)]
   atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
 
 -- instance Show a => Show (TVar a) where

src/Simplex/Messaging/Notifications/Server/Main.hs

@@ -30,7 +30,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 ntfServerVersion :: String
-ntfServerVersion = "1.7.0.4"
+ntfServerVersion = "1.7.3.0"
 
 defaultSMPBatchDelay :: Int
 defaultSMPBatchDelay = 10000
@@ -42,6 +42,10 @@ ntfServerCLI cfgPath logPath =
       doesFileExist iniFile >>= \case
         True -> exitError $ "Error: server is already initialized (" <> iniFile <> " exists).\nRun `" <> executableName <> " start`."
         _ -> initializeServer opts
+    OnlineCert certOpts ->
+      doesFileExist iniFile >>= \case
+        True -> genOnline cfgPath certOpts
+        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
     Start ->
       doesFileExist iniFile >>= \case
         True -> readIniFile iniFile >>= either exitError runServer
@@ -143,6 +147,7 @@ ntfServerCLI cfgPath logPath =
 
 data CliCommand
   = Init InitOptions
+  | OnlineCert CertOptions
   | Start
   | Delete
 
@@ -158,6 +163,7 @@ cliCommandP :: FilePath -> FilePath -> FilePath -> Parser CliCommand
 cliCommandP cfgPath logPath iniFile =
   hsubparser
     ( command "init" (info (Init <$> initP) (progDesc $ "Initialize server - creates " <> cfgPath <> " and " <> logPath <> " directories and configuration files"))
+        <> command "cert" (info (OnlineCert <$> certOptionsP) (progDesc $ "Generate new online TLS server credentials (configuration: " <> iniFile <> ")"))
         <> command "start" (info (pure Start) (progDesc $ "Start server (configuration: " <> iniFile <> ")"))
         <> command "delete" (info (pure Delete) (progDesc "Delete configuration and log files"))
     )

src/Simplex/Messaging/Protocol.hs

@@ -144,8 +144,9 @@ module Simplex.Messaging.Protocol
     tParse,
     tDecodeParseValidate,
     tEncode,
-    tEncodeBatch,
+    tEncodeBatch1,
     batchTransmissions,
+    batchTransmissions',
 
     -- * exports for tests
     CommandTag (..),
@@ -154,7 +155,6 @@ module Simplex.Messaging.Protocol
 where
 
 import Control.Applicative (optional, (<|>))
-import Control.Concurrent (threadDelay)
 import Control.Monad.Except
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Aeson.TH as J
@@ -173,11 +173,12 @@ import Data.String
 import Data.Time.Clock.System (SystemTime (..))
 import Data.Type.Equality
 import GHC.TypeLits (ErrorMessage (..), TypeError, type (+))
-import Network.Socket (HostName, ServiceName)
+import Network.Socket (ServiceName)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
+import Simplex.Messaging.ServiceScheme
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (TransportHost, TransportHosts (..))
 import Simplex.Messaging.Util (bshow, eitherToMaybe, (<$?>))
@@ -915,16 +916,6 @@ serverStrP = do
   where
     portP = show <$> (A.char ':' *> (A.decimal :: Parser Int))
 
-data SrvLoc = SrvLoc HostName ServiceName
-  deriving (Eq, Ord, Show)
-
-instance StrEncoding SrvLoc where
-  strEncode (SrvLoc host port) = B.pack $ host <> if null port then "" else ':' : port
-  strP = SrvLoc <$> host <*> (port <|> pure "")
-    where
-      host = B.unpack <$> A.takeWhile1 (A.notInClass ":#,;/ ")
-      port = show <$> (A.char ':' *> (A.decimal :: Parser Int))
-
 -- | Transmission correlation ID.
 newtype CorrId = CorrId {bs :: ByteString} deriving (Eq, Ord, Show)
 
@@ -1283,14 +1274,14 @@ instance Encoding CommandError where
       _ -> fail "bad command error type"
 
 -- | Send signed SMP transmission to TCP transport.
-tPut :: Transport c => THandle c -> Maybe Int -> NonEmpty SentRawTransmission -> IO [Either TransportError ()]
-tPut th delay_ = fmap concat . mapM tPutBatch . batchTransmissions (batch th) (blockSize th)
+tPut :: Transport c => THandle c -> NonEmpty SentRawTransmission -> IO [Either TransportError ()]
+tPut th = fmap concat . mapM tPutBatch . batchTransmissions (batch th) (blockSize th)
   where
-    tPutBatch :: TransportBatch -> IO [Either TransportError ()]
+    tPutBatch :: TransportBatch () -> IO [Either TransportError ()]
     tPutBatch = \case
-      TBLargeTransmission -> [Left TELargeMsg] <$ putStrLn "tPut error: large message"
-      TBTransmissions n s -> replicate n <$> (tPutLog th (tEncodeBatch n s) <* mapM_ threadDelay delay_)
-      TBTransmission s -> (: []) <$> tPutLog th s
+      TBLargeTransmission _ -> [Left TELargeMsg] <$ putStrLn "tPut error: large message"
+      TBTransmissions s n _ -> replicate n <$> (tPutLog th s)
+      TBTransmission s _ -> (: []) <$> tPutLog th s
 
 tPutLog :: Transport c => THandle c -> ByteString -> IO (Either TransportError ())
 tPutLog th s = do
@@ -1300,44 +1291,52 @@ tPutLog th s = do
     _ -> pure ()
   pure r
 
--- ByteString does not include length byte, it is added by tEncodeBatch
-data TransportBatch = TBTransmissions Int ByteString | TBTransmission ByteString | TBLargeTransmission
+-- ByteString in TBTransmissions includes byte with transmissions count
+data TransportBatch r = TBTransmissions ByteString Int [r] | TBTransmission ByteString r | TBLargeTransmission r
+
+batchTransmissions :: Bool -> Int -> NonEmpty SentRawTransmission -> [TransportBatch ()]
+batchTransmissions batch bSize = batchTransmissions' batch bSize . L.map (,())
 
 -- | encodes and batches transmissions into blocks,
-batchTransmissions :: Bool -> Int -> NonEmpty SentRawTransmission -> [TransportBatch]
-batchTransmissions batch bSize
-  | batch = reverse . mkBatch [] . L.map tEncode
-  | otherwise = map (mkBatch1 . tEncode) . L.toList
+batchTransmissions' :: forall r. Bool -> Int -> NonEmpty (SentRawTransmission, r) -> [TransportBatch r]
+batchTransmissions' batch bSize
+  | batch = addBatch . foldr addTransmission ([], 0, 0, [], [])
+  | otherwise = map mkBatch1 . L.toList
   where
-    mkBatch :: [TransportBatch] -> NonEmpty ByteString -> [TransportBatch]
-    mkBatch rs ts =
-      let (n, s, ts_) = encodeBatch 0 "" ts
-          r = if n == 0 then TBLargeTransmission else TBTransmissions n s
-          rs' = r : rs
-       in case ts_ of
-            Just ts' -> mkBatch rs' ts'
-            _ -> rs'
-    mkBatch1 :: ByteString -> TransportBatch
-    mkBatch1 s = if B.length s > bSize - 2 then TBLargeTransmission else TBTransmission s
-    encodeBatch :: Int -> ByteString -> NonEmpty ByteString -> (Int, ByteString, Maybe (NonEmpty ByteString))
-    encodeBatch n s ts@(t :| ts_)
-      | n == 255 = (n, s, Just ts)
-      | otherwise =
-          let s' = s <> smpEncode (Large t)
-              n' = n + 1
-           in if B.length s' > bSize - 3 -- one byte is reserved for the number of messages in the batch
-                then (n,s,) $ if n == 0 then L.nonEmpty ts_ else Just ts
-                else case L.nonEmpty ts_ of
-                  Just ts' -> encodeBatch n' s' ts'
-                  _ -> (n', s', Nothing)
+    mkBatch1 :: (SentRawTransmission, r) -> TransportBatch r
+    mkBatch1 (t, r)
+      -- 2 bytes are reserved for pad size
+      | B.length s <= bSize - 2 = TBTransmission s r
+      | otherwise = TBLargeTransmission r
+      where
+        s = tEncode t
+    -- 3 = 2 bytes reserved for pad size + 1 for transmission count
+    bSize' = bSize - 3
+    addTransmission :: (SentRawTransmission, r) -> ([TransportBatch r], Int, Int, [ByteString], [r]) -> ([TransportBatch r], Int, Int, [ByteString], [r])
+    addTransmission (t, r) acc@(bs, len, n, ss, rs)
+      | len' <= bSize' && n < 255 = (bs, len', 1 + n, s : ss, r : rs)
+      | sLen <= bSize' = (addBatch acc, sLen, 1, [s], [r])
+      | otherwise = (TBLargeTransmission r : addBatch acc, 0, 0, [], [])
+      where
+        s = tEncodeForBatch t
+        sLen = B.length s
+        len' = len + sLen
+    addBatch :: ([TransportBatch r], Int, Int, [ByteString], [r]) -> [TransportBatch r]
+    addBatch (bs, _len, n, ss, rs) = if n == 0 then bs else TBTransmissions b n rs : bs
+      where
+        b = B.concat $ B.singleton (lenEncode n) : ss
 
 tEncode :: SentRawTransmission -> ByteString
 tEncode (sig, t) = smpEncode (C.signatureBytes sig) <> t
 {-# INLINE tEncode #-}
 
-tEncodeBatch :: Int -> ByteString -> ByteString
-tEncodeBatch n s = lenEncode n `B.cons` s
-{-# INLINE tEncodeBatch #-}
+tEncodeForBatch :: SentRawTransmission -> ByteString
+tEncodeForBatch = smpEncode . Large . tEncode
+{-# INLINE tEncodeForBatch #-}
+
+tEncodeBatch1 :: SentRawTransmission -> ByteString
+tEncodeBatch1 t = lenEncode 1 `B.cons` tEncodeForBatch t
+{-# INLINE tEncodeBatch1 #-}
 
 encodeTransmission :: ProtocolEncoding e c => Version -> ByteString -> Transmission c -> ByteString
 encodeTransmission v sessionId (CorrId corrId, queueId, command) =

src/Simplex/Messaging/Server.hs

@@ -203,9 +203,9 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
       liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
       ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, msgExpired, activeQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount} <- asks serverStats
       let interval = 1000000 * logInterval
-      withFile statsFilePath AppendMode $ \h -> liftIO $ do
-        hSetBuffering h LineBuffering
-        forever $ do
+      forever $ do
+        withFile statsFilePath AppendMode $ \h -> liftIO $ do
+          hSetBuffering h LineBuffering
           ts <- getCurrentTime
           fromTime' <- atomically $ swapTVar fromTime ts
           qCreated' <- atomically $ swapTVar qCreated 0
@@ -241,7 +241,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                 show msgCount',
                 show msgExpired'
               ]
-          threadDelay' interval
+        liftIO $ threadDelay' interval
 
     runClient :: Transport c => TProxy c -> c -> M ()
     runClient tp h = do
@@ -439,7 +439,7 @@ send h@THandle {thVersion = v} Client {sndQ, sessionId, sndActiveAt} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
   forever $ do
     ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
-    void . liftIO . tPut h Nothing $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
+    void . liftIO . tPut h $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
     atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
   where
     tOrder :: Transmission BrokerMsg -> Int

src/Simplex/Messaging/Server/CLI.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE ApplicativeDo #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
@@ -10,6 +11,7 @@
 module Simplex.Messaging.Server.CLI where
 
 import Control.Monad
+import Data.ASN1.Types (asn1CharacterToString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (fromRight)
@@ -17,6 +19,8 @@ import Data.Ini (Ini, lookupValue)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
+import qualified Data.X509 as X
+import qualified Data.X509.File as XF
 import Data.X509.Validation (Fingerprint (..))
 import Network.Socket (HostName, ServiceName)
 import Options.Applicative
@@ -27,12 +31,14 @@ import Simplex.Messaging.Transport.Server (loadFingerprint)
 import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (eitherToMaybe, whenM)
 import System.Directory (doesDirectoryExist, listDirectory, removeDirectoryRecursive, removePathForcibly)
+import System.Environment (lookupEnv)
 import System.Exit (exitFailure)
 import System.FilePath (combine)
 import System.IO (IOMode (..), hFlush, hGetLine, stdout, withFile)
 import System.Process (readCreateProcess, shell)
+import Text.Read (readMaybe)
 
-exitError :: String -> IO ()
+exitError :: String -> IO a
 exitError msg = putStrLn msg >> exitFailure
 
 confirmOrExit :: String -> IO ()
@@ -84,14 +90,18 @@ getCliCommand' cmdP version =
     versionOption = infoOption version (long "version" <> short 'v' <> help "Show version")
 
 createServerX509 :: FilePath -> X509Config -> IO ByteString
-createServerX509 cfgPath x509cfg = do
-  createOpensslCaConf
-  createOpensslServerConf
+createServerX509 = createServerX509_ True
+
+createServerX509_ :: Bool -> FilePath -> X509Config -> IO ByteString
+createServerX509_ createCA cfgPath x509cfg = do
   let alg = show $ signAlgorithm (x509cfg :: X509Config)
   -- CA certificate (identity/offline)
-  run $ "openssl genpkey -algorithm " <> alg <> " -out " <> c caKeyFile
-  run $ "openssl req -new -x509 -days 999999 -config " <> c opensslCaConfFile <> " -extensions v3 -key " <> c caKeyFile <> " -out " <> c caCrtFile
+  when createCA $ do
+    createOpensslCaConf
+    run $ "openssl genpkey -algorithm " <> alg <> " -out " <> c caKeyFile
+    run $ "openssl req -new -x509 -days 999999 -config " <> c opensslCaConfFile <> " -extensions v3 -key " <> c caKeyFile <> " -out " <> c caCrtFile
   -- server certificate (online)
+  createOpensslServerConf
   run $ "openssl genpkey -algorithm " <> alg <> " -out " <> c serverKeyFile
   run $ "openssl req -new -config " <> c opensslServerConfFile <> " -reqexts v3 -key " <> c serverKeyFile <> " -out " <> c serverCsrFile
   run $ "openssl x509 -req -days 999999 -extfile " <> c opensslServerConfFile <> " -extensions v3 -in " <> c serverCsrFile <> " -CA " <> c caCrtFile <> " -CAkey " <> c caKeyFile <> " -CAcreateserial -out " <> c serverCrtFile
@@ -131,6 +141,59 @@ createServerX509 cfgPath x509cfg = do
       withFile (c fingerprintFile) WriteMode (`B.hPutStrLn` strEncode fp)
       pure fp
 
+data CertOptions = CertOptions
+  { signAlgorithm_ :: Maybe SignAlgorithm,
+    commonName_ :: Maybe HostName
+  }
+  deriving (Show)
+
+certOptionsP :: Parser CertOptions
+certOptionsP = do
+  signAlgorithm_ <-
+    optional $
+      option
+        (maybeReader readMaybe)
+        ( long "sign-algorithm"
+            <> short 'a'
+            <> help "Set new signature algorithm used for TLS certificates: ED25519, ED448"
+            <> metavar "ALG"
+        )
+  commonName_ <-
+    optional $
+      strOption
+        ( long "cn"
+            <> help
+              "Set new Common Name for TLS online certificate"
+            <> metavar "FQDN"
+        )
+  pure CertOptions {signAlgorithm_, commonName_}
+
+genOnline :: FilePath -> CertOptions -> IO ()
+genOnline cfgPath CertOptions {signAlgorithm_, commonName_} = do
+  (signAlgorithm, commonName) <-
+    case (signAlgorithm_, commonName_) of
+      (Just alg, Just cn) -> pure (alg, cn)
+      _ ->
+        XF.readSignedObject certPath >>= \case
+          [old] -> either exitError pure . fromX509 . X.signedObject $ X.getSigned old
+          [] -> exitError $ "No certificate found at " <> certPath
+          _ -> exitError $ "Too many certificates at " <> certPath
+  let x509cfg = defaultX509Config {signAlgorithm, commonName}
+  void $ createServerX509_ False cfgPath x509cfg
+  putStrLn "Generated new server credentials"
+  warnCAPrivateKeyFile cfgPath x509cfg
+  where
+    certPath = combine cfgPath $ serverCrtFile defaultX509Config
+    fromX509 X.Certificate {certSignatureAlg, certSubjectDN} = (,) <$> maybe oldAlg Right signAlgorithm_ <*> maybe oldCN Right commonName_
+      where
+        oldAlg = case certSignatureAlg of
+          X.SignatureALG_IntrinsicHash X.PubKeyALG_Ed448 -> Right ED448
+          X.SignatureALG_IntrinsicHash X.PubKeyALG_Ed25519 -> Right ED25519
+          alg -> Left $ "Unexpected signature algorithm " <> show alg
+        oldCN = case X.getDnElement X.DnCommonName certSubjectDN of
+          Nothing -> Left "Certificate subject has no CN element"
+          Just cn -> maybe (Left "Certificate subject CN decoding failed") Right $ asn1CharacterToString cn
+
 warnCAPrivateKeyFile :: FilePath -> X509Config -> IO ()
 warnCAPrivateKeyFile cfgPath X509Config {caKeyFile} =
   putStrLn $
@@ -235,3 +298,6 @@ printServiceInfo serverVersion srv@(ProtoServerWithAuth ProtocolServer {keyHash}
 
 clearDirIfExists :: FilePath -> IO ()
 clearDirIfExists path = whenM (doesDirectoryExist path) $ listDirectory path >>= mapM_ (removePathForcibly . combine path)
+
+getEnvPath :: String -> FilePath -> IO FilePath
+getEnvPath name def = maybe def (\case "" -> def; f -> f) <$> lookupEnv name

src/Simplex/Messaging/Server/Main.hs

@@ -41,6 +41,10 @@ smpServerCLI cfgPath logPath =
       doesFileExist iniFile >>= \case
         True -> exitError $ "Error: server is already initialized (" <> iniFile <> " exists).\nRun `" <> executableName <> " start`."
         _ -> initializeServer opts
+    OnlineCert certOpts ->
+      doesFileExist iniFile >>= \case
+        True -> genOnline cfgPath certOpts
+        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
     Start ->
       doesFileExist iniFile >>= \case
         True -> readIniFile iniFile >>= either exitError runServer
@@ -56,8 +60,8 @@ smpServerCLI cfgPath logPath =
     defaultServerPort = "5223"
     executableName = "smp-server"
     storeLogFilePath = combine logPath "smp-server-store.log"
-    initializeServer opts
-      | scripted opts = initialize opts
+    initializeServer opts@InitOptions {ip, fqdn, scripted}
+      | scripted = initialize opts
       | otherwise = do
           putStrLn "Use `smp-server init -h` for available options."
           void $ withPrompt "SMP server will be initialized (press Enter)" getLine
@@ -65,9 +69,9 @@ smpServerCLI cfgPath logPath =
           logStats <- onOffPrompt "Enable logging daily statistics" False
           putStrLn "Require a password to create new messaging queues?"
           password <- withPrompt "'r' for random (default), 'n' - no password, or enter password: " serverPassword
-          let host = fromMaybe (ip opts) (fqdn opts)
+          let host = fromMaybe ip fqdn
           host' <- withPrompt ("Enter server FQDN or IP address for certificate (" <> host <> "): ") getLine
-          initialize opts {enableStoreLog, logStats, fqdn = if null host' then fqdn opts else Just host', password}
+          initialize opts {enableStoreLog, logStats, fqdn = if null host' then fqdn else Just host', password}
       where
         serverPassword =
           getLine >>= \case
@@ -78,7 +82,7 @@ smpServerCLI cfgPath logPath =
               case strDecode $ encodeUtf8 $ T.pack s of
                 Right auth -> pure . Just $ ServerPassword auth
                 _ -> putStrLn "Invalid password. Only latin letters, digits and symbols other than '@' and ':' are allowed" >> serverPassword
-        initialize InitOptions {enableStoreLog, logStats, signAlgorithm, ip, fqdn, password} = do
+        initialize InitOptions {enableStoreLog, logStats, signAlgorithm, password} = do
           clearDirIfExists cfgPath
           clearDirIfExists logPath
           createDirectoryIfMissing True cfgPath
@@ -210,6 +214,7 @@ smpServerCLI cfgPath logPath =
 
 data CliCommand
   = Init InitOptions
+  | OnlineCert CertOptions
   | Start
   | Delete
 
@@ -231,6 +236,7 @@ cliCommandP :: FilePath -> FilePath -> FilePath -> Parser CliCommand
 cliCommandP cfgPath logPath iniFile =
   hsubparser
     ( command "init" (info (Init <$> initP) (progDesc $ "Initialize server - creates " <> cfgPath <> " and " <> logPath <> " directories and configuration files"))
+        <> command "cert" (info (OnlineCert <$> certOptionsP) (progDesc $ "Generate new online TLS server credentials (configuration: " <> iniFile <> ")"))
         <> command "start" (info (pure Start) (progDesc $ "Start server (configuration: " <> iniFile <> ")"))
         <> command "delete" (info (pure Delete) (progDesc "Delete configuration and log files"))
     )
@@ -255,7 +261,7 @@ cliCommandP cfgPath logPath iniFile =
           ( long "sign-algorithm"
               <> short 'a'
               <> help "Signature algorithm used for TLS certificates: ED25519, ED448"
-              <> value ED448
+              <> value ED25519
               <> showDefault
               <> metavar "ALG"
           )
@@ -295,3 +301,4 @@ cliCommandP cfgPath logPath iniFile =
       pure InitOptions {enableStoreLog, logStats, signAlgorithm, ip, fqdn, password, scripted}
     parseBasicAuth :: ReadM ServerPassword
     parseBasicAuth = eitherReader $ fmap ServerPassword . strDecode . B.pack
+

src/Simplex/Messaging/ServiceScheme.hs

@@ -0,0 +1,35 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.ServiceScheme where
+
+import Control.Applicative ((<|>))
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import qualified Data.ByteString.Char8 as B
+import Data.Functor (($>))
+import Network.Socket (HostName, ServiceName)
+import Simplex.Messaging.Encoding.String (StrEncoding (..))
+
+data ServiceScheme = SSSimplex | SSAppServer SrvLoc
+  deriving (Eq, Show)
+
+instance StrEncoding ServiceScheme where
+  strEncode = \case
+    SSSimplex -> "simplex:"
+    SSAppServer srv -> "https://" <> strEncode srv
+  strP =
+    "simplex:" $> SSSimplex
+      <|> "https://" *> (SSAppServer <$> strP)
+
+data SrvLoc = SrvLoc HostName ServiceName
+  deriving (Eq, Ord, Show)
+
+instance StrEncoding SrvLoc where
+  strEncode (SrvLoc host port) = B.pack $ host <> if null port then "" else ':' : port
+  strP = SrvLoc <$> host <*> (port <|> pure "")
+    where
+      host = B.unpack <$> A.takeWhile1 (A.notInClass ":#,;/ ")
+      port = show <$> (A.char ':' *> (A.decimal :: A.Parser Int))
+
+simplexChat :: ServiceScheme
+simplexChat = SSAppServer $ SrvLoc "simplex.chat" ""

src/Simplex/Messaging/Transport.hs

@@ -65,11 +65,12 @@ import Control.Monad.Except
 import Control.Monad.Trans.Except (throwE)
 import qualified Data.Aeson.TH as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
+import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)
 import Data.Bitraversable (bimapM)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Lazy as BL
+import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Default (def)
 import Data.Functor (($>))
 import Data.Version (showVersion)
@@ -80,7 +81,7 @@ import qualified Network.TLS.Extra as TE
 import qualified Paths_simplexmq as SMQ
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
-import Simplex.Messaging.Parsers (dropPrefix, parse, parseRead1, sumTypeJSON)
+import Simplex.Messaging.Parsers (dropPrefix, parseRead1, sumTypeJSON)
 import Simplex.Messaging.Transport.Buffer
 import Simplex.Messaging.Util (bshow, catchAll, catchAll_)
 import Simplex.Messaging.Version
@@ -217,8 +218,8 @@ instance Transport TLS where
     getBuffered tlsBuffer n t_ (T.recvData tlsContext)
 
   cPut :: TLS -> ByteString -> IO ()
-  cPut TLS {tlsContext, tlsTransportConfig = TransportConfig {transportTimeout = t_}} s =
-    withTimedErr t_ . T.sendData tlsContext $ BL.fromStrict s
+  cPut TLS {tlsContext, tlsTransportConfig = TransportConfig {transportTimeout = t_}} =
+    withTimedErr t_ . T.sendData tlsContext . LB.fromStrict
 
   getLn :: TLS -> IO ByteString
   getLn TLS {tlsContext, tlsBuffer} = do
@@ -358,8 +359,9 @@ smpThHandle th v = (th :: THandle c) {thVersion = v, batch = v >= 4}
 sendHandshake :: (Transport c, Encoding smp) => THandle c -> smp -> ExceptT TransportError IO ()
 sendHandshake th = ExceptT . tPutBlock th . smpEncode
 
+-- ignores tail bytes to allow future extensions
 getHandshake :: (Transport c, Encoding smp) => THandle c -> ExceptT TransportError IO smp
-getHandshake th = ExceptT $ (parse smpP (TEHandshake PARSE) =<<) <$> tGetBlock th
+getHandshake th = ExceptT $ (first (\_ -> TEHandshake PARSE) . A.parseOnly smpP =<<) <$> tGetBlock th
 
 smpTHandle :: Transport c => c -> THandle c
 smpTHandle c = THandle {connection = c, sessionId = tlsUnique c, blockSize = smpBlockSize, thVersion = 0, batch = False}

src/Simplex/Messaging/Transport/Client.hs

@@ -21,6 +21,7 @@ module Simplex.Messaging.Transport.Client
 where
 
 import Control.Applicative (optional)
+import Control.Logger.Simple (logError)
 import Control.Monad.IO.Unlift
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Attoparsec.ByteString.Char8 as A
@@ -48,7 +49,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll, parseString)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.KeepAlive
-import Simplex.Messaging.Util (bshow, (<$?>))
+import Simplex.Messaging.Util (bshow, (<$?>), catchAll, tshow)
 import System.IO.Error
 import Text.Read (readMaybe)
 import UnliftIO.Exception (IOException)
@@ -135,7 +136,7 @@ runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy,
         _ -> connectTCPClient hostName
   c <- liftIO $ do
     sock <- connectTCP port
-    mapM_ (setSocketKeepAlive sock) tcpKeepAlive
+    mapM_ (setSocketKeepAlive sock) tcpKeepAlive `catchAll` \e -> logError ("Error setting TCP keep-alive" <> tshow e)
     let tCfg = clientTransportConfig cfg
     connectTLS (Just hostName) tCfg clientParams sock >>= getClientConnection tCfg
   client c `E.finally` liftIO (closeConnection c)

src/Simplex/Messaging/Transport/WebSockets.hs

@@ -7,7 +7,7 @@ module Simplex.Messaging.Transport.WebSockets (WS (..)) where
 import qualified Control.Exception as E
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Lazy as BL
+import qualified Data.ByteString.Lazy as LB
 import qualified Network.TLS as T
 import Network.WebSockets
 import Network.WebSockets.Stream (Stream)
@@ -101,5 +101,5 @@ makeTLSContextStream cxt =
       (Just <$> T.recvData cxt) `E.catch` \case
         T.Error_EOF -> pure Nothing
         e -> E.throwIO e
-    writeStream :: Maybe BL.ByteString -> IO ()
+    writeStream :: Maybe LB.ByteString -> IO ()
     writeStream = maybe (closeTLS cxt) (T.sendData cxt)

tests/AgentTests.hs

@@ -359,7 +359,6 @@ testServerConnectionAfterError t _ = do
     withAgent2 $ \alice -> do
       withServer $ do
         connect (bob, "bob") (alice, "alice")
-
       bob <#. ("", "", DOWN server ["alice"])
       alice <#. ("", "", DOWN server ["bob"])
       alice #: ("1", "bob", "SEND F 5\nhello") #> ("1", "bob", MID 4)
@@ -386,10 +385,10 @@ testServerConnectionAfterError t _ = do
   where
     server = SMPServer "localhost" testPort2 testKeyHash
     withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
-    withAgent1 = withAgent agentTestPort testDB
-    withAgent2 = withAgent agentTestPort2 testDB2
-    withAgent :: String -> FilePath -> (c -> IO a) -> IO a
-    withAgent agentPort agentDB = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) (pure ()) . const . testSMPAgentClientOn agentPort
+    withAgent1 = withAgent agentTestPort testDB 0 
+    withAgent2 = withAgent agentTestPort2 testDB2 10
+    withAgent :: String -> FilePath -> Int -> (c -> IO a) -> IO a
+    withAgent agentPort agentDB initClientId = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) initClientId (pure ()) . const . testSMPAgentClientOn agentPort
 
 testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO ()
 testMsgDeliveryAgentRestart t bob = do
@@ -424,7 +423,7 @@ testMsgDeliveryAgentRestart t bob = do
   removeFile testDB
   where
     withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
-    withAgent = withSmpAgentThreadOn_ (ATransport t) (agentTestPort, testPort, testDB) (pure ()) . const . testSMPAgentClientOn agentTestPort
+    withAgent = withSmpAgentThreadOn_ (ATransport t) (agentTestPort, testPort, testDB) 0 (pure ()) . const . testSMPAgentClientOn agentTestPort
 
 testConcurrentMsgDelivery :: Transport c => TProxy c -> c -> c -> IO ()
 testConcurrentMsgDelivery _ alice bob = do

tests/AgentTests/ConnectionRequestTests.hs

@@ -13,6 +13,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtocolServer (..), supportedSMPClientVRange)
+import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import Simplex.Messaging.Version
 import Test.Hspec
 
@@ -51,7 +52,7 @@ testDhKeyStrUri = urlEncode True testDhKeyStr
 connReqData :: ConnReqUriData
 connReqData =
   ConnReqUriData
-    { crScheme = CRSSimplex,
+    { crScheme = SSSimplex,
       crAgentVRange = mkVersionRange 1 1,
       crSmpQueues = [queueV1],
       crClientData = Nothing

tests/AgentTests/FunctionalAPITests.hs

@@ -36,6 +36,7 @@ import AgentTests.ConnectionRequestTests (connReqData, queueAddr, testE2ERatchet
 import Control.Concurrent (killThread, threadDelay)
 import Control.Monad
 import Control.Monad.Except
+import Control.Monad.Reader
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight)
@@ -46,13 +47,15 @@ import qualified Data.Set as S
 import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Type.Equality
+import qualified Database.SQLite.Simple as SQL
 import SMPAgentClient
 import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..))
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..), createAgentStore)
 import Simplex.Messaging.Agent.Protocol as Agent
-import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..))
+import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteStore (dbNew))
+import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultClientConfig)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
@@ -73,9 +76,12 @@ type AEntityTransmission e = (ACorrId, ConnId, ACommand 'Agent e)
 a ##> t = withTimeout a (`shouldBe` t)
 
 (=##>) :: (Show a, HasCallStack, MonadUnliftIO m) => m a -> (a -> Bool) -> m ()
-a =##> p = withTimeout a (`shouldSatisfy` p)
+a =##> p =
+  withTimeout a $ \r -> do
+    unless (p r) $ liftIO $ putStrLn $ "value failed predicate: " <> show r
+    r `shouldSatisfy` p
 
-withTimeout :: MonadUnliftIO m => m a -> (a -> Expectation) -> m ()
+withTimeout :: (HasCallStack, MonadUnliftIO m) => m a -> (a -> Expectation) -> m ()
 withTimeout a test =
   timeout 10_000000 a >>= \case
     Nothing -> error "operation timed out"
@@ -111,6 +117,9 @@ pGet c = do
 pattern Msg :: MsgBody -> ACommand 'Agent e
 pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} _ msgBody
 
+pattern MsgErr :: AgentMsgId -> MsgErrorType -> MsgBody -> ACommand 'Agent e
+pattern MsgErr msgId err msgBody <- MSG MsgMeta {recipient = (msgId, _), integrity = MsgError err} _ msgBody
+
 pattern Rcvd :: AgentMsgId -> ACommand 'Agent e
 pattern Rcvd agentMsgId <- RCVD MsgMeta {integrity = MsgOk} [MsgReceipt {agentMsgId, msgRcptStatus = MROk}]
 
@@ -160,15 +169,18 @@ runRight action =
     Left e -> error $ "Unexpected error: " <> show e
 
 getInAnyOrder :: HasCallStack => AgentClient -> [ATransmission 'Agent -> Bool] -> Expectation
-getInAnyOrder _ [] = pure ()
-getInAnyOrder c rs = do
-  r <- pGet c
+getInAnyOrder c = inAnyOrder (pGet c)
+
+inAnyOrder :: (Show a, MonadIO m, HasCallStack) => m a -> [a -> Bool] -> m ()
+inAnyOrder _ [] = pure ()
+inAnyOrder g rs = do
+  r <- g
   let rest = filter (not . expected r) rs
   if length rest < length rs
-    then getInAnyOrder c rest
+    then inAnyOrder g rest
     else error $ "unexpected event: " <> show r
   where
-    expected :: ATransmission 'Agent -> (ATransmission 'Agent -> Bool) -> Bool
+    expected :: a -> (a -> Bool) -> Bool
     expected r rp = rp r
 
 functionalAPITests :: ATransport -> Spec
@@ -212,6 +224,11 @@ functionalAPITests t = do
       testDuplicateMessage t
     it "should report error via msg integrity on skipped messages" $
       testSkippedMessages t
+    describe "message expiration" $ do
+      it "should expire one message" $ testExpireMessage t
+      it "should expire multiple messages" $ testExpireManyMessages t
+      it "should expire one message if quota is exceeded" $ testExpireMessageQuota t
+      it "should expire multiple messages if quota is exceeded" $ testExpireManyMessagesQuota t
     describe "Ratchet synchronization" $ do
       it "should report ratchet de-synchronization, synchronize ratchets" $
         testRatchetSync t
@@ -375,8 +392,8 @@ runTestCfg2 aCfg bCfg baseMsgId runTest =
 
 withAgentClientsCfg2 :: AgentConfig -> AgentConfig -> (AgentClient -> AgentClient -> IO ()) -> IO ()
 withAgentClientsCfg2 aCfg bCfg runTest = do
-  a <- getSMPAgentClient' aCfg initAgentServers testDB
-  b <- getSMPAgentClient' bCfg initAgentServers testDB2
+  a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 bCfg initAgentServers testDB2
   runTest a b
   disconnectAgentClient a
   disconnectAgentClient b
@@ -385,7 +402,7 @@ withAgentClients2 :: (AgentClient -> AgentClient -> IO ()) -> IO ()
 withAgentClients2 = withAgentClientsCfg2 agentCfg agentCfg
 
 runAgentClientTest :: HasCallStack => AgentClient -> AgentClient -> AgentMsgId -> IO ()
-runAgentClientTest alice bob baseId = do
+runAgentClientTest alice bob baseId =
   runRight_ $ do
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
     aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
@@ -421,9 +438,9 @@ runAgentClientTest alice bob baseId = do
 
 testAgentClient3 :: HasCallStack => IO ()
 testAgentClient3 = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  c <- getSMPAgentClient' agentCfg initAgentServers testDB3
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  c <- getSMPAgentClient' 3 agentCfg initAgentServers testDB3
   runRight_ $ do
     (aIdForB, bId) <- makeConnection a b
     (aIdForC, cId) <- makeConnection a c
@@ -446,7 +463,7 @@ testAgentClient3 = do
     ackMessage c aIdForC 5 Nothing
 
 runAgentClientContactTest :: HasCallStack => AgentClient -> AgentClient -> AgentMsgId -> IO ()
-runAgentClientContactTest alice bob baseId = do
+runAgentClientContactTest alice bob baseId =
   runRight_ $ do
     (_, qInfo) <- createConnection alice 1 True SCMContact Nothing SMSubscribe
     aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
@@ -496,7 +513,7 @@ testAsyncInitiatingOffline =
     (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
     disconnectAgentClient alice
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
-    alice' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB
+    alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB
     subscribeConnection alice' bobId
     ("", _, CONF confId _ "bob's connInfo") <- get alice'
     allowConnection alice' bobId confId "alice's connInfo"
@@ -513,7 +530,7 @@ testAsyncJoiningOfflineBeforeActivation =
     disconnectAgentClient bob
     ("", _, CONF confId _ "bob's connInfo") <- get alice
     allowConnection alice bobId confId "alice's connInfo"
-    bob' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB2
+    bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     subscribeConnection bob' aliceId
     get alice ##> ("", bobId, CON)
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -527,11 +544,11 @@ testAsyncBothOffline =
     disconnectAgentClient alice
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
     disconnectAgentClient bob
-    alice' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB
+    alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB
     subscribeConnection alice' bobId
     ("", _, CONF confId _ "bob's connInfo") <- get alice'
     allowConnection alice' bobId confId "alice's connInfo"
-    bob' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB2
+    bob' <- liftIO $ getSMPAgentClient' 4 agentCfg initAgentServers testDB2
     subscribeConnection bob' aliceId
     get alice' ##> ("", bobId, CON)
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -574,8 +591,8 @@ testAsyncHelloTimeout = do
 testAllowConnectionClientRestart :: HasCallStack => ATransport -> IO ()
 testAllowConnectionClientRestart t = do
   let initAgentServersSrv2 = initAgentServers {smp = userServers [noAuthSrv testSMPServer2]}
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServersSrv2 testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
   withSmpServerStoreLogOn t testPort $ \_ -> do
     (aliceId, bobId, confId) <-
       withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
@@ -589,13 +606,13 @@ testAllowConnectionClientRestart t = do
 
     runRight_ $ do
       allowConnectionAsync alice "1" bobId confId "alice's connInfo"
-      ("1", _, OK) <- get alice
+      get alice =##> \case ("1", _, OK) -> True; _ -> False
       pure ()
 
     threadDelay 100000 -- give time to enqueue confirmation (enqueueConfirmation)
     disconnectAgentClient alice
 
-    alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
 
     withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
       runRight $ do
@@ -613,8 +630,8 @@ testAllowConnectionClientRestart t = do
 
 testIncreaseConnAgentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersion t = do
-  alice <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ do
       (aliceId, bobId) <- makeConnection alice bob
@@ -626,7 +643,7 @@ testIncreaseConnAgentVersion t = do
     -- version doesn't increase if incompatible
 
     disconnectAgentClient alice
-    alice2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
 
     runRight_ $ do
       subscribeConnection alice2 bobId
@@ -637,7 +654,7 @@ testIncreaseConnAgentVersion t = do
     -- version increases if compatible
 
     disconnectAgentClient bob
-    bob2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
+    bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection bob2 aliceId
@@ -648,7 +665,7 @@ testIncreaseConnAgentVersion t = do
     -- version doesn't decrease, even if incompatible
 
     disconnectAgentClient alice2
-    alice3 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 2 2} initAgentServers testDB
+    alice3 <- getSMPAgentClient' 5 agentCfg {smpAgentVRange = mkVersionRange 2 2} initAgentServers testDB
 
     runRight_ $ do
       subscribeConnection alice3 bobId
@@ -657,7 +674,7 @@ testIncreaseConnAgentVersion t = do
       checkVersion bob2 aliceId 3
 
     disconnectAgentClient bob2
-    bob3 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 1} initAgentServers testDB2
+    bob3 <- getSMPAgentClient' 6 agentCfg {smpAgentVRange = mkVersionRange 1 1} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection bob3 aliceId
@@ -674,8 +691,8 @@ checkVersion c connId v = do
 
 testIncreaseConnAgentVersionMaxCompatible :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersionMaxCompatible t = do
-  alice <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ do
       (aliceId, bobId) <- makeConnection alice bob
@@ -687,9 +704,9 @@ testIncreaseConnAgentVersionMaxCompatible t = do
     -- version increases to max compatible
 
     disconnectAgentClient alice
-    alice2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
     disconnectAgentClient bob
-    bob2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
+    bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection alice2 bobId
@@ -702,8 +719,8 @@ testIncreaseConnAgentVersionMaxCompatible t = do
 
 testIncreaseConnAgentVersionStartDifferentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersionStartDifferentVersion t = do
-  alice <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ do
       (aliceId, bobId) <- makeConnection alice bob
@@ -715,7 +732,7 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
     -- version increases to max compatible
 
     disconnectAgentClient alice
-    alice2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
+    alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
 
     runRight_ $ do
       subscribeConnection alice2 bobId
@@ -727,8 +744,8 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
 
 testDeliverClientRestart :: HasCallStack => ATransport -> IO ()
 testDeliverClientRestart t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
 
   (aliceId, bobId) <- withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight $ do
@@ -743,7 +760,7 @@ testDeliverClientRestart t = do
 
   disconnectAgentClient bob
 
-  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  bob2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
 
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight_ $ do
@@ -758,8 +775,8 @@ testDeliverClientRestart t = do
 
 testDuplicateMessage :: HasCallStack => ATransport -> IO ()
 testDuplicateMessage t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId, bob1) <- withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ makeConnection alice bob
     runRight_ $ do
@@ -769,7 +786,7 @@ testDuplicateMessage t = do
     disconnectAgentClient bob
 
     -- if the agent user did not send ACK, the message will be delivered again
-    bob1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    bob1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     runRight_ $ do
       subscribeConnection bob1 aliceId
       get bob1 =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
@@ -785,13 +802,13 @@ testDuplicateMessage t = do
   -- commenting two lines below and uncommenting further two lines would also runRight_,
   -- it is the scenario tested above, when the message was not acknowledged by the user
   threadDelay 200000
-  Left (BROKER _ TIMEOUT) <- runExceptT $ ackMessage bob1 aliceId 5 Nothing
+  Left (BROKER _ NETWORK) <- runExceptT $ ackMessage bob1 aliceId 5 Nothing
 
   disconnectAgentClient alice
   disconnectAgentClient bob1
 
-  alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB
+  bob2 <- getSMPAgentClient' 5 agentCfg initAgentServers testDB2
 
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight_ $ do
@@ -808,8 +825,8 @@ testDuplicateMessage t = do
 
 testSkippedMessages :: HasCallStack => ATransport -> IO ()
 testSkippedMessages t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ makeConnection alice bob
     runRight_ $ do
@@ -835,8 +852,8 @@ testSkippedMessages t = do
 
   disconnectAgentClient alice
 
-  alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
+  bob2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
 
   withSmpServerStoreLogOn t testPort $ \_ -> do
     runRight_ $ do
@@ -855,6 +872,102 @@ testSkippedMessages t = do
   disconnectAgentClient alice2
   disconnectAgentClient bob2
 
+testExpireMessage :: HasCallStack => ATransport -> IO ()
+testExpireMessage t = do
+  a <- getSMPAgentClient' 1 agentCfg {messageTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ makeConnection a b
+  nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+  nGet b =##> \case ("", "", DOWN _ [c]) -> c == aId; _ -> False
+  4 <- runRight $ sendMessage a bId SMP.noMsgFlags "1"
+  threadDelay 1000000
+  5 <- runRight $ sendMessage a bId SMP.noMsgFlags "2" -- this won't expire
+  get a =##> \case ("", c, MERR 4 (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False
+  withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
+    withUP a bId $ \case ("", _, SENT 5) -> True; _ -> False
+    withUP b aId $ \case ("", _, MsgErr 4 (MsgSkipped 3 3) "2") -> True; _ -> False
+    ackMessage b aId 4 Nothing
+
+testExpireManyMessages :: HasCallStack => ATransport -> IO ()
+testExpireManyMessages t = do
+  a <- getSMPAgentClient' 1 agentCfg {messageTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ makeConnection a b
+  runRight_ $ do
+    nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+    nGet b =##> \case ("", "", DOWN _ [c]) -> c == aId; _ -> False
+    4 <- sendMessage a bId SMP.noMsgFlags "1"
+    5 <- sendMessage a bId SMP.noMsgFlags "2"
+    6 <- sendMessage a bId SMP.noMsgFlags "3"
+    liftIO $ threadDelay 1000000
+    7 <- sendMessage a bId SMP.noMsgFlags "4" -- this won't expire
+    get a =##> \case ("", c, MERR 4 (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False
+    get a =##> \case ("", c, MERRS [5, 6] (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False
+  withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
+    withUP a bId $ \case ("", _, SENT 7) -> True; _ -> False
+    withUP b aId $ \case ("", _, MsgErr 4 (MsgSkipped 3 5) "4") -> True; _ -> False
+    ackMessage b aId 4 Nothing
+
+withUP :: AgentClient -> ConnId -> (AEntityTransmission 'AEConn -> Bool) -> ExceptT AgentErrorType IO ()
+withUP a bId p =
+  liftIO $
+    getInAnyOrder
+      a
+      [ \case ("", "", APC SAENone (UP _ [c])) -> c == bId; _ -> False,
+        \case (corrId, c, APC SAEConn cmd) -> c == bId && p (corrId, c, cmd); _ -> False
+      ]
+
+testExpireMessageQuota :: HasCallStack => ATransport -> IO ()
+testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
+  a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  (aId, bId) <- runRight $ do
+    (aId, bId) <- makeConnection a b
+    liftIO $ threadDelay 500000
+    disconnectAgentClient b
+    4 <- sendMessage a bId SMP.noMsgFlags "1"
+    get a ##> ("", bId, SENT 4)
+    5 <- sendMessage a bId SMP.noMsgFlags "2"
+    liftIO $ threadDelay 1000000
+    6 <- sendMessage a bId SMP.noMsgFlags "3" -- this won't expire
+    get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False
+    pure (aId, bId)
+  b' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
+  runRight_ $ do
+    subscribeConnection b' aId
+    get b' =##> \case ("", c, Msg "1") -> c == aId; _ -> False
+    ackMessage b' aId 4 Nothing
+    get a ##> ("", bId, SENT 6)
+    get b' =##> \case ("", c, MsgErr 6 (MsgSkipped 4 4) "3") -> c == aId; _ -> False
+    ackMessage b' aId 6 Nothing
+
+testExpireManyMessagesQuota :: HasCallStack => ATransport -> IO ()
+testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
+  a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  (aId, bId) <- runRight $ do
+    (aId, bId) <- makeConnection a b
+    liftIO $ threadDelay 500000
+    disconnectAgentClient b
+    4 <- sendMessage a bId SMP.noMsgFlags "1"
+    get a ##> ("", bId, SENT 4)
+    5 <- sendMessage a bId SMP.noMsgFlags "2"
+    6 <- sendMessage a bId SMP.noMsgFlags "3"
+    7 <- sendMessage a bId SMP.noMsgFlags "4"
+    liftIO $ threadDelay 1000000
+    8 <- sendMessage a bId SMP.noMsgFlags "5" -- this won't expire
+    get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False
+    get a =##> \case ("", c, MERRS [6, 7] (SMP QUOTA)) -> bId == c; _ -> False
+    pure (aId, bId)
+  b' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
+  runRight_ $ do
+    subscribeConnection b' aId
+    get b' =##> \case ("", c, Msg "1") -> c == aId; _ -> False
+    ackMessage b' aId 4 Nothing
+    get a ##> ("", bId, SENT 8)
+    get b' =##> \case ("", c, MsgErr 6 (MsgSkipped 4 6) "5") -> c == aId; _ -> False
+    ackMessage b' aId 6 Nothing
+
 testRatchetSync :: HasCallStack => ATransport -> IO ()
 testRatchetSync t = withAgentClients2 $ \alice bob ->
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
@@ -899,7 +1012,7 @@ setupDesynchronizedRatchet alice bob = do
   -- importing database backup after progressing ratchet de-synchronizes ratchet
   liftIO $ renameFile (testDB2 <> ".bak") testDB2
 
-  bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  bob2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
 
   runRight_ $ do
     subscribeConnection bob2 aliceId
@@ -959,8 +1072,8 @@ serverUpP = \case
 
 testRatchetSyncClientRestart :: HasCallStack => ATransport -> IO ()
 testRatchetSyncClientRestart t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
     setupDesynchronizedRatchet alice bob
   ("", "", DOWN _ _) <- nGet alice
@@ -968,7 +1081,7 @@ testRatchetSyncClientRestart t = do
   ConnectionStats {ratchetSyncState} <- runRight $ synchronizeRatchet bob2 aliceId False
   liftIO $ ratchetSyncState `shouldBe` RSStarted
   disconnectAgentClient bob2
-  bob3 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  bob3 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight_ $ do
       ("", "", UP _ _) <- nGet alice
@@ -984,8 +1097,8 @@ testRatchetSyncClientRestart t = do
 
 testRatchetSyncSuspendForeground :: HasCallStack => ATransport -> IO ()
 testRatchetSyncSuspendForeground t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
     setupDesynchronizedRatchet alice bob
 
@@ -1018,8 +1131,8 @@ testRatchetSyncSuspendForeground t = do
 
 testRatchetSyncSimultaneous :: HasCallStack => ATransport -> IO ()
 testRatchetSyncSimultaneous t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
     setupDesynchronizedRatchet alice bob
 
@@ -1101,7 +1214,7 @@ testInactiveNoSubs :: ATransport -> IO ()
 testInactiveNoSubs t = do
   let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ -> do
-    alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+    alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     runRight_ . void $ createConnection alice 1 True SCMInvitation Nothing SMOnlyCreate -- do not subscribe to pass noSubscriptions check
     Just (_, _, APC SAENone (CONNECT _ _)) <- timeout 2000000 $ atomically (readTBQueue $ subQ alice)
     Just (_, _, APC SAENone (DISCONNECT _ _)) <- timeout 5000000 $ atomically (readTBQueue $ subQ alice)
@@ -1111,7 +1224,7 @@ testInactiveWithSubs :: ATransport -> IO ()
 testInactiveWithSubs t = do
   let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ -> do
-    alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+    alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     runRight_ . void $ createConnection alice 1 True SCMInvitation Nothing SMSubscribe
     Nothing <- 800000 `timeout` get alice
     liftIO $ threadDelay 1200000
@@ -1123,7 +1236,7 @@ testActiveClientNotDisconnected :: ATransport -> IO ()
 testActiveClientNotDisconnected t = do
   let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
   withSmpServerConfigOn t cfg' testPort $ \_ -> do
-    alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+    alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     ts <- getSystemTime
     runRight_ $ do
       (connId, _cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
@@ -1214,8 +1327,8 @@ testSuspendingAgentTimeout t = withAgentClients2 $ \a b -> do
 
 testBatchedSubscriptions :: Int -> Int -> ATransport -> IO ()
 testBatchedSubscriptions nCreate nDel t = do
-  a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers2 testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers2 testDB2
   conns <- runServers $ do
     conns <- replicateM (nCreate :: Int) $ makeConnection a b
     forM_ conns $ \(aId, bId) -> exchangeGreetings a bId b aId
@@ -1289,7 +1402,7 @@ testAsyncCommands =
     liftIO $ aliceId' `shouldBe` aliceId
     ("", _, CONF confId _ "bob's connInfo") <- get alice
     allowConnectionAsync alice "3" bobId confId "alice's connInfo"
-    ("3", _, OK) <- get alice
+    get alice =##> \case ("3", _, OK) -> True; _ -> False
     get alice ##> ("", bobId, CON)
     get bob ##> ("", aliceId, INFO "alice's connInfo")
     get bob ##> ("", aliceId, CON)
@@ -1300,20 +1413,26 @@ testAsyncCommands =
     get alice ##> ("", bobId, SENT $ baseId + 2)
     get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
     ackMessageAsync bob "4" aliceId (baseId + 1) Nothing
-    ("4", _, OK) <- get bob
-    get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
+    inAnyOrder
+      (get bob)
+      [ \case ("4", _, OK) -> True; _ -> False,
+        \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
+      ]
     ackMessageAsync bob "5" aliceId (baseId + 2) Nothing
-    ("5", _, OK) <- get bob
+    get bob =##> \case ("5", _, OK) -> True; _ -> False
     3 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "hello too"
     get bob ##> ("", aliceId, SENT $ baseId + 3)
     4 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 1"
     get bob ##> ("", aliceId, SENT $ baseId + 4)
     get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
     ackMessageAsync alice "6" bobId (baseId + 3) Nothing
-    ("6", _, OK) <- get alice
-    get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
+    inAnyOrder
+      (get alice)
+      [ \case ("6", _, OK) -> True; _ -> False,
+        \case ("", c, Msg "message 1") -> c == bobId; _ -> False
+      ]
     ackMessageAsync alice "7" bobId (baseId + 4) Nothing
-    ("7", _, OK) <- get alice
+    get alice =##> \case ("7", _, OK) -> True; _ -> False
     deleteConnectionAsync alice bobId
     get alice =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bobId; _ -> False
     get alice =##> \case ("", c, DEL_CONN) -> c == bobId; _ -> False
@@ -1324,15 +1443,15 @@ testAsyncCommands =
 
 testAsyncCommandsRestore :: ATransport -> IO ()
 testAsyncCommandsRestore t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bobId <- runRight $ createConnectionAsync alice 1 "1" True SCMInvitation SMSubscribe
   liftIO $ noMessages alice "alice doesn't receive INV because server is down"
   disconnectAgentClient alice
-  alice' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB
+  alice' <- liftIO $ getSMPAgentClient' 2 agentCfg initAgentServers testDB
   withSmpServerStoreLogOn t testPort $ \_ -> do
     runRight_ $ do
       subscribeConnection alice' bobId
-      ("1", _, INV _) <- get alice'
+      get alice' =##> \case ("1", _, INV _) -> True; _ -> False
       pure ()
   disconnectAgentClient alice'
 
@@ -1343,8 +1462,7 @@ testAcceptContactAsync =
     aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
     ("", _, REQ invId _ "bob's connInfo") <- get alice
     bobId <- acceptContactAsync alice "1" True invId "alice's connInfo" SMSubscribe
-    ("1", bobId', OK) <- get alice
-    liftIO $ bobId' `shouldBe` bobId
+    get alice =##> \case ("1", c, OK) -> c == bobId; _ -> False
     ("", _, CONF confId _ "alice's connInfo") <- get bob
     allowConnection bob aliceId confId "bob's connInfo"
     get alice ##> ("", bobId, INFO "bob's connInfo")
@@ -1378,7 +1496,7 @@ testAcceptContactAsync =
 
 testDeleteConnectionAsync :: ATransport -> IO ()
 testDeleteConnectionAsync t = do
-  a <- getSMPAgentClient' agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB
+  a <- getSMPAgentClient' 1 agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB
   connIds <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
     (bId1, _inv) <- createConnection a 1 True SCMInvitation Nothing SMSubscribe
     (bId2, _inv) <- createConnection a 1 True SCMInvitation Nothing SMSubscribe
@@ -1398,8 +1516,8 @@ testDeleteConnectionAsync t = do
 testJoinConnectionAsyncReplyError :: HasCallStack => ATransport -> IO ()
 testJoinConnectionAsyncReplyError t = do
   let initAgentServersSrv2 = initAgentServers {smp = userServers [noAuthSrv testSMPServer2]}
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServersSrv2 testDB2
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
   (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
     bId <- createConnectionAsync a 1 "1" True SCMInvitation SMSubscribe
     ("1", bId', INV (ACR _ qInfo)) <- get a
@@ -1410,8 +1528,7 @@ testJoinConnectionAsyncReplyError t = do
     pure (aId, bId)
   nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
   withSmpServerOn t testPort2 $ do
-    ("2", aId', OK) <- get b
-    liftIO $ aId' `shouldBe` aId
+    get b =##> \case ("2", c, OK) -> c == aId; _ -> False
     confId <- withSmpServerStoreLogOn t testPort $ \_ -> do
       pGet a >>= \case
         ("", "", APC _ (UP _ [_])) -> do
@@ -1491,8 +1608,8 @@ testUsersNoServer t = withAgentClientsCfg2 aCfg agentCfg $ \a b -> do
 
 testSwitchConnection :: InitialAgentServers -> IO ()
 testSwitchConnection servers = do
-  a <- getSMPAgentClient' agentCfg servers testDB
-  b <- getSMPAgentClient' agentCfg {initialClientId = 1} servers testDB2
+  a <- getSMPAgentClient' 1 agentCfg servers testDB
+  b <- getSMPAgentClient' 2 agentCfg servers testDB2
   runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
@@ -1575,12 +1692,12 @@ testSwitchAsync servers = do
     testFullSwitch a bId b aId 16
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
-withAgent :: AgentConfig -> InitialAgentServers -> FilePath -> (AgentClient -> IO a) -> IO a
-withAgent cfg' servers dbPath = bracket (getSMPAgentClient' cfg' servers dbPath) disconnectAgentClient
+withAgent :: Int -> AgentConfig -> InitialAgentServers -> FilePath -> (AgentClient -> IO a) -> IO a
+withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) disconnectAgentClient
 
 sessionSubscribe :: (forall a. (AgentClient -> IO a) -> IO a) -> [ConnId] -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO ()
 sessionSubscribe withC connIds a =
@@ -1593,8 +1710,8 @@ sessionSubscribe withC connIds a =
 
 testSwitchDelete :: InitialAgentServers -> IO ()
 testSwitchDelete servers = do
-  a <- getSMPAgentClient' agentCfg servers testDB
-  b <- getSMPAgentClient' agentCfg {initialClientId = 1} servers testDB2
+  a <- getSMPAgentClient' 1 agentCfg servers testDB
+  b <- getSMPAgentClient' 2 agentCfg servers testDB2
   runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
@@ -1656,9 +1773,9 @@ testAbortSwitchStarted servers = do
     testFullSwitch a bId b aId 18
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 testAbortSwitchStartedReinitiate :: HasCallStack => InitialAgentServers -> IO ()
 testAbortSwitchStartedReinitiate servers = do
@@ -1707,9 +1824,9 @@ testAbortSwitchStartedReinitiate servers = do
     testFullSwitch a bId b aId 18
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 switchPhaseRcvP :: ConnId -> SwitchPhase -> [Maybe RcvSwitchStatus] -> ATransmission 'Agent -> Bool
 switchPhaseRcvP cId sphase swchStatuses = switchPhaseP cId QDRcv sphase (\stats -> rcvSwchStatuses' stats == swchStatuses)
@@ -1761,9 +1878,9 @@ testCannotAbortSwitchSecured servers = do
     testFullSwitch a bId b aId 16
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 testSwitch2Connections :: HasCallStack => InitialAgentServers -> IO ()
 testSwitch2Connections servers = do
@@ -1819,9 +1936,9 @@ testSwitch2Connections servers = do
     testFullSwitch a bId2 b aId2 16
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 testSwitch2ConnectionsAbort1 :: HasCallStack => InitialAgentServers -> IO ()
 testSwitch2ConnectionsAbort1 servers = do
@@ -1872,14 +1989,14 @@ testSwitch2ConnectionsAbort1 servers = do
     testFullSwitch a bId2 b aId2 14
   where
     withA :: (AgentClient -> IO a) -> IO a
-    withA = withAgent agentCfg servers testDB
+    withA = withAgent 1 agentCfg servers testDB
     withB :: (AgentClient -> IO a) -> IO a
-    withB = withAgent agentCfg {initialClientId = 1} servers testDB2
+    withB = withAgent 2 agentCfg servers testDB2
 
 testCreateQueueAuth :: HasCallStack => (Maybe BasicAuth, Version) -> (Maybe BasicAuth, Version) -> IO Int
 testCreateQueueAuth clnt1 clnt2 = do
-  a <- getClient clnt1 testDB
-  b <- getClient clnt2 testDB2
+  a <- getClient 1 clnt1 testDB
+  b <- getClient 2 clnt2 testDB2
   r <- runRight $ do
     tryError (createConnection a 1 True SCMInvitation Nothing SMSubscribe) >>= \case
       Left (SMP AUTH) -> pure 0
@@ -1900,15 +2017,15 @@ testCreateQueueAuth clnt1 clnt2 = do
   disconnectAgentClient b
   pure r
   where
-    getClient (clntAuth, clntVersion) db =
+    getClient clientId (clntAuth, clntVersion) db =
       let servers = initAgentServers {smp = userServers [ProtoServerWithAuth testSMPServer clntAuth]}
           smpCfg = (defaultClientConfig :: ProtocolClientConfig) {serverVRange = mkVersionRange 4 clntVersion}
-       in getSMPAgentClient' agentCfg {smpCfg} servers db
+       in getSMPAgentClient' clientId agentCfg {smpCfg} servers db
 
 testSMPServerConnectionTest :: ATransport -> Maybe BasicAuth -> SMPServerWithAuth -> IO (Maybe ProtocolTestFailure)
 testSMPServerConnectionTest t newQueueBasicAuth srv =
   withSmpServerConfigOn t cfg {newQueueBasicAuth} testPort2 $ \_ -> do
-    a <- getSMPAgentClient' agentCfg initAgentServers testDB -- initially passed server is not running
+    a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB -- initially passed server is not running
     runRight $ testProtocolServer a 1 srv
 
 testRatchetAdHash :: HasCallStack => IO ()
@@ -1941,8 +2058,8 @@ testDeliveryReceipts =
 
 testDeliveryReceiptsVersion :: HasCallStack => ATransport -> IO ()
 testDeliveryReceiptsVersion t = do
-  a <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
+  a <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aId, bId) <- runRight $ do
       (aId, bId) <- makeConnection a b
@@ -1962,8 +2079,8 @@ testDeliveryReceiptsVersion t = do
 
     disconnectAgentClient a
     disconnectAgentClient b
-    a' <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB
-    b' <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
+    a' <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB
+    b' <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
 
     runRight_ $ do
       subscribeConnection a' bId
@@ -2103,10 +2220,12 @@ testTwoUsers = withAgentClients2 $ \a b -> do
     hasClients :: HasCallStack => AgentClient -> Int -> ExceptT AgentErrorType IO ()
     hasClients c n = liftIO $ M.size <$> readTVarIO (smpClients c) `shouldReturn` n
 
-getSMPAgentClient' :: AgentConfig -> InitialAgentServers -> FilePath -> IO AgentClient
-getSMPAgentClient' cfg' initServers dbPath = do
+getSMPAgentClient' :: Int -> AgentConfig -> InitialAgentServers -> FilePath -> IO AgentClient
+getSMPAgentClient' clientId cfg' initServers dbPath = do
   Right st <- liftIO $ createAgentStore dbPath "" False MCError
-  getSMPAgentClient cfg' initServers st False
+  c <- getSMPAgentClient_ clientId cfg' initServers st False
+  when (dbNew st) $ withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1)")
+  pure c
 
 testServerMultipleIdentities :: HasCallStack => IO ()
 testServerMultipleIdentities =
@@ -2122,7 +2241,7 @@ testServerMultipleIdentities =
     -- this saves queue with second server identity
     Left (BROKER _ NETWORK) <- runExceptT $ joinConnection bob 1 True secondIdentityCReq "bob's connInfo" SMSubscribe
     disconnectAgentClient bob
-    bob' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB2
+    bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     subscribeConnection bob' aliceId
     exchangeGreetingsMsgId 6 alice bobId bob' aliceId
   where

tests/AgentTests/NotificationTests.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -10,9 +11,10 @@ module AgentTests.NotificationTests where
 
 -- import Control.Logger.Simple (LogConfig (..), LogLevel (..), setLogLevel, withGlobalLogging)
 import AgentTests.FunctionalAPITests (exchangeGreetingsMsgId, get, getSMPAgentClient', makeConnection, nGet, runRight, runRight_, switchComplete, testServerMatrix2, (##>), (=##>), pattern Msg)
-import Control.Concurrent (killThread, threadDelay)
+import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Monad
 import Control.Monad.Except
+import Control.Monad.Reader (runReaderT)
 import Control.Monad.Trans.Except
 import qualified Data.Aeson as J
 import qualified Data.Aeson.Types as JT
@@ -21,17 +23,19 @@ import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
 import Data.Text.Encoding (encodeUtf8)
 import NtfClient
-import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2)
+import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testDB3, testNtfServer2)
 import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, xit')
 import Simplex.Messaging.Agent
-import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers)
+import Simplex.Messaging.Agent.Client (withStore')
+import Simplex.Messaging.Agent.Env.SQLite (InitialAgentServers)
 import Simplex.Messaging.Agent.Protocol
+import Simplex.Messaging.Agent.Store.SQLite (getSavedNtfToken)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Types (NtfToken (..))
-import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), SMPMsgMeta (..), SubscriptionMode (..))
+import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), ProtocolServer (..), SMPMsgMeta (..), SubscriptionMode (..))
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Transport (ATransport)
@@ -60,6 +64,12 @@ notificationTests t =
       it "should re-register token when notification server is restarted" $ \_ ->
         withAPNSMockServer $ \apns ->
           testNtfTokenServerRestart t apns
+      it "should work with multiple configured servers" $ \_ ->
+        withAPNSMockServer $ \apns ->
+          testNtfTokenMultipleServers t apns
+      it "should keep working with active token until replaced" $ \_ ->
+        withAPNSMockServer $ \apns ->
+          testNtfTokenChangeServers t apns
     describe "Managing notification subscriptions" $ do
       -- fails on Ubuntu CI?
       xit' "should create notification subscription for existing connection" $ \_ -> do
@@ -95,10 +105,20 @@ notificationTests t =
       testServerMatrix2 t $ \servers ->
         withAPNSMockServer $ \apns ->
           withNtfServer t $ testSwitchNotifications servers apns
+    it "should keep sending notifications for old token" $
+      withSmpServer t $
+        withAPNSMockServer $ \apns ->
+          withNtfServerOn t ntfTestPort $
+            testNotificationsOldToken apns
+    it "should update server from new token" $
+      withSmpServer t $
+        withAPNSMockServer $ \apns ->
+          withNtfServerOn t ntfTestPort2 . withNtfServerThreadOn t ntfTestPort $ \ntf ->
+            testNotificationsNewToken apns ntf
 
 testNotificationToken :: APNSMockServer -> IO ()
 testNotificationToken APNSMockServer {apnsQ} = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -126,7 +146,7 @@ testNtfTokenRepeatRegistration :: APNSMockServer -> IO ()
 testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do
   -- setLogLevel LogError -- LogDebug
   -- withGlobalLogging logCfg $ do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -150,8 +170,8 @@ testNtfTokenSecondRegistration :: APNSMockServer -> IO ()
 testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
   -- setLogLevel LogError -- LogDebug
   -- withGlobalLogging logCfg $ do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  a' <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  a' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   runRight_ $ do
     let tkn = DeviceToken PPApnsTest "abcd"
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -185,7 +205,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
 
 testNtfTokenServerRestart :: ATransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   let tkn = DeviceToken PPApnsTest "abcd"
   ntfData <- withNtfServer t . runRight $ do
     NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -196,7 +216,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
   -- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
   threadDelay 1000000
   disconnectAgentClient a
-  a' <- getSMPAgentClient' agentCfg initAgentServers testDB
+  a' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB
   -- server stopped before token is verified, so now the attempt to verify it will return AUTH error but re-register token,
   -- so that repeat verification happens without restarting the clients, when notification arrives
   withNtfServer t . runRight_ $ do
@@ -212,10 +232,72 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
     NTActive <- checkNtfToken a' tkn
     disconnectAgentClient a'
 
+getTestNtfTokenPort :: (MonadUnliftIO m, MonadError AgentErrorType m) => AgentClient -> m String
+getTestNtfTokenPort a =
+  runReaderT (withStore' a getSavedNtfToken) (agentEnv a) >>= \case
+    Just NtfToken {ntfServer = ProtocolServer {port}} -> pure port
+    Nothing -> error "no active NtfToken"
+
+testNtfTokenMultipleServers :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenMultipleServers t APNSMockServer {apnsQ} = do
+  let tkn = DeviceToken PPApnsTest "abcd"
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers2 testDB
+  withNtfServerThreadOn t ntfTestPort $ \ntf ->
+    withNtfServerThreadOn t ntfTestPort2 $ \ntf2 -> runRight_ $ do
+      -- register a new token, the agent picks a server and stores its choice
+      NTRegistered <- registerNtfToken a tkn NMPeriodic
+      APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
+        atomically $ readTBQueue apnsQ
+      verification <- ntfData .-> "verification"
+      nonce <- C.cbNonce <$> ntfData .-> "nonce"
+      liftIO $ sendApnsResponse APNSRespOk
+      verifyNtfToken a tkn nonce verification
+      NTActive <- checkNtfToken a tkn
+      -- shut down the "other" server
+      port <- getTestNtfTokenPort a
+      liftIO . killThread $ if port == ntfTestPort then ntf2 else ntf
+      -- still works
+      NTActive <- checkNtfToken a tkn
+      liftIO . killThread $ if port == ntfTestPort then ntf else ntf2
+      -- negative test, the correct server is now gone
+      Left _ <- tryError (checkNtfToken a tkn)
+      pure ()
+
+testNtfTokenChangeServers :: ATransport -> APNSMockServer -> IO ()
+testNtfTokenChangeServers t APNSMockServer {apnsQ} =
+  withNtfServerThreadOn t ntfTestPort $ \ntf -> do
+    tkn1 <- runRight $ do
+      a <- liftIO $ getSMPAgentClient' 1 agentCfg initAgentServers testDB
+      tkn <- registerTestToken a "abcd" NMInstant apnsQ
+      NTActive <- checkNtfToken a tkn
+      setNtfServers a [testNtfServer2]
+      NTActive <- checkNtfToken a tkn -- still works on old server
+      disconnectAgentClient a
+      pure tkn
+
+    threadDelay 1000000
+
+    a <- liftIO $ getSMPAgentClient' 2 agentCfg initAgentServers testDB
+    runRight_ $ do
+      getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort
+      NTActive <- checkNtfToken a tkn1
+      setNtfServers a [testNtfServer2] -- just change configured server list
+      getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort -- not yet changed
+      -- trigger token replace
+      tkn2 <- registerTestToken a "xyzw" NMInstant apnsQ
+      getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort -- not yet changed
+      deleteNtfToken a tkn2 -- force server switch
+      Left BROKER {brokerErr = NETWORK} <- tryError $ registerTestToken a "qwer" NMInstant apnsQ -- ok, it's down for now
+      getTestNtfTokenPort a >>= \port2 -> liftIO $ port2 `shouldBe` ntfTestPort2 -- but the token got updated
+    killThread ntf
+    withNtfServerOn t ntfTestPort2 $ runRight_ $ do
+      tkn <- registerTestToken a "qwer" NMInstant apnsQ
+      checkNtfToken a tkn >>= \r -> liftIO $ r `shouldBe` NTActive
+
 testNotificationSubscriptionExistingConnection :: APNSMockServer -> IO ()
 testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (bobId, aliceId, nonce, message) <- runRight $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
@@ -247,7 +329,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
   Left (CMD PROHIBITED) <- runExceptT $ getNotificationMessage alice nonce message
 
   -- aliceNtf client doesn't have subscription and is allowed to get notification message
-  aliceNtf <- getSMPAgentClient' agentCfg initAgentServers testDB
+  aliceNtf <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   runRight_ $ do
     (_, [SMPMsgMeta {msgFlags = MsgFlags True}]) <- getNotificationMessage aliceNtf nonce message
     pure ()
@@ -272,8 +354,8 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
 
 testNotificationSubscriptionNewConnection :: APNSMockServer -> IO ()
 testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   runRight_ $ do
     -- alice registers notification token
     DeviceToken {} <- registerTestToken alice "abcd" NMInstant apnsQ
@@ -319,8 +401,8 @@ registerTestToken :: AgentClient -> ByteString -> NotificationsMode -> TBQueue A
 registerTestToken a token mode apnsQ = do
   let tkn = DeviceToken PPApnsTest token
   NTRegistered <- registerNtfToken a tkn mode
-  APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-    atomically $ readTBQueue apnsQ
+  Just APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
+    timeout 1000000 . atomically $ readTBQueue apnsQ
   verification' <- ntfData' .-> "verification"
   nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
   liftIO $ sendApnsResponse' APNSRespOk
@@ -330,8 +412,8 @@ registerTestToken a token mode apnsQ = do
 
 testChangeNotificationsMode :: APNSMockServer -> IO ()
 testChangeNotificationsMode APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   runRight_ $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
@@ -396,8 +478,8 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
 
 testChangeToken :: APNSMockServer -> IO ()
 testChangeToken APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId) <- runRight $ do
     -- establish connection
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
@@ -419,7 +501,7 @@ testChangeToken APNSMockServer {apnsQ} = do
     pure (aliceId, bobId)
   disconnectAgentClient alice
 
-  alice1 <- getSMPAgentClient' agentCfg initAgentServers testDB
+  alice1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   runRight_ $ do
     subscribeConnection alice1 bobId
     -- change notification token
@@ -441,8 +523,8 @@ testChangeToken APNSMockServer {apnsQ} = do
 
 testNotificationsStoreLog :: ATransport -> APNSMockServer -> IO ()
 testNotificationsStoreLog t APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId) <- withNtfServerStoreLog t $ \threadId -> runRight $ do
     (aliceId, bobId) <- makeConnection alice bob
     _ <- registerTestToken alice "abcd" NMInstant apnsQ
@@ -469,8 +551,8 @@ testNotificationsStoreLog t APNSMockServer {apnsQ} = do
 
 testNotificationsSMPRestart :: ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \threadId -> runRight $ do
     (aliceId, bobId) <- makeConnection alice bob
     _ <- registerTestToken alice "abcd" NMInstant apnsQ
@@ -501,12 +583,13 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
 
 testNotificationsSMPRestartBatch :: Int -> ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
-  a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers2 testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers2 testDB2
+  threadDelay 1000000
   conns <- runServers $ do
     conns <- replicateM (n :: Int) $ makeConnection a b
     _ <- registerTestToken a "abcd" NMInstant apnsQ
-    liftIO $ threadDelay 1500000
+    liftIO $ threadDelay 5000000
     forM_ conns $ \(aliceId, bobId) -> do
       msgId <- sendMessage b aliceId (SMP.MsgFlags True) "hello"
       get b ##> ("", aliceId, SENT msgId)
@@ -549,8 +632,8 @@ testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
 
 testSwitchNotifications :: InitialAgentServers -> APNSMockServer -> IO ()
 testSwitchNotifications servers APNSMockServer {apnsQ} = do
-  a <- getSMPAgentClient' agentCfg servers testDB
-  b <- getSMPAgentClient' agentCfg {initialClientId = 1} servers testDB2
+  a <- getSMPAgentClient' 1 agentCfg servers testDB
+  b <- getSMPAgentClient' 2 agentCfg servers testDB2
   runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
@@ -570,9 +653,70 @@ testSwitchNotifications servers APNSMockServer {apnsQ} = do
   disconnectAgentClient a
   disconnectAgentClient b
 
-messageNotification :: TBQueue APNSMockRequest -> ExceptT AgentErrorType IO (C.CbNonce, ByteString)
+testNotificationsOldToken :: APNSMockServer -> IO ()
+testNotificationsOldToken APNSMockServer {apnsQ} = do
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  c <- getSMPAgentClient' 3 agentCfg initAgentServers testDB3
+  runRight_ $ do
+    (abId, baId) <- makeConnection a b
+    let testMessageAB = testMessage_ apnsQ a abId b baId
+    _ <- registerTestToken a "abcd" NMInstant apnsQ
+    liftIO $ threadDelay 250000
+    testMessageAB "hello"
+    -- change server
+    setNtfServers a [testNtfServer2] -- server 2 isn't running now, don't use
+    -- replacing token keeps server
+    _ <- registerTestToken a "xyzw" NMInstant apnsQ
+    getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort
+    testMessageAB "still there"
+    -- new connections keep server
+    (acId, caId) <- makeConnection a c
+    let testMessageAC = testMessage_ apnsQ a acId c caId
+    testMessageAC "greetings"
+  disconnectAgentClient a
+  disconnectAgentClient b
+  disconnectAgentClient c
+
+testNotificationsNewToken :: APNSMockServer -> ThreadId -> IO ()
+testNotificationsNewToken APNSMockServer {apnsQ} oldNtf = do
+  a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  c <- getSMPAgentClient' 3 agentCfg initAgentServers testDB3
+  runRight_ $ do
+    (abId, baId) <- makeConnection a b
+    let testMessageAB = testMessage_ apnsQ a abId b baId
+    tkn <- registerTestToken a "abcd" NMInstant apnsQ
+    getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort
+    liftIO $ threadDelay 250000
+    testMessageAB "hello"
+    -- switch
+    setNtfServers a [testNtfServer2]
+    deleteNtfToken a tkn
+    _ <- registerTestToken a "abcd" NMInstant apnsQ
+    getTestNtfTokenPort a >>= \port -> liftIO $ port `shouldBe` ntfTestPort2
+    liftIO $ threadDelay 250000
+    liftIO $ killThread oldNtf
+    -- -- back to work
+    testMessageAB "hello again"
+    (acId, caId) <- makeConnection a c
+    let testMessageAC = testMessage_ apnsQ a acId c caId
+    testMessageAC "greetings"
+  disconnectAgentClient a
+  disconnectAgentClient b
+  disconnectAgentClient c
+
+testMessage_ :: HasCallStack => TBQueue APNSMockRequest -> AgentClient -> ConnId -> AgentClient -> ConnId -> SMP.MsgBody -> ExceptT AgentErrorType IO ()
+testMessage_ apnsQ a aId b bId msg = do
+  msgId <- sendMessage b aId (SMP.MsgFlags True) msg
+  get b ##> ("", aId, SENT msgId)
+  void $ messageNotification apnsQ
+  get a =##> \case ("", c, Msg msg') -> c == bId && msg == msg'; _ -> False
+  ackMessage a bId msgId Nothing
+
+messageNotification :: HasCallStack => TBQueue APNSMockRequest -> ExceptT AgentErrorType IO (C.CbNonce, ByteString)
 messageNotification apnsQ = do
-  750000 `timeout` atomically (readTBQueue apnsQ) >>= \case
+  1000000 `timeout` atomically (readTBQueue apnsQ) >>= \case
     Nothing -> error "no notification"
     Just APNSMockRequest {notification = APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData}, sendApnsResponse} -> do
       nonce <- C.cbNonce <$> ntfData .-> "nonce"

tests/AgentTests/SQLiteTests.hs

@@ -34,6 +34,7 @@ import Simplex.Messaging.Agent.Client ()
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite
+import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import qualified Simplex.Messaging.Crypto as C
@@ -69,6 +70,7 @@ createEncryptedStore key keepKey = do
   -- IO operations on multiple similarly named files; error seems to be environment specific
   r <- randomIO :: IO Word32
   Right st <- createSQLiteStore (testDB <> show r) key keepKey Migrations.app MCError
+  withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1);")
   pure st
 
 removeStore :: SQLiteStore -> IO ()
@@ -656,7 +658,8 @@ rcvFileDescr1 =
               chunkSize = defaultChunkSize,
               replicas = [FileChunkReplica {server = xftpServer1, replicaId, replicaKey = testFileReplicaKey}]
             }
-        ]
+        ],
+      redirect = Nothing
     }
   where
     defaultChunkSize = FileSize $ mb 8
@@ -714,9 +717,9 @@ testGetNextSndFileToPrepare st = do
   withTransaction st $ \db -> do
     Right Nothing <- getNextSndFileToPrepare db 86400
 
-    Right _ <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce
+    Right _ <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce Nothing
     DB.execute_ db "UPDATE snd_files SET status = 'new', num_recipients = 'bad' WHERE snd_file_id = 1"
-    Right fId2 <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce
+    Right fId2 <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce Nothing
     DB.execute_ db "UPDATE snd_files SET status = 'new' WHERE snd_file_id = 2"
 
     Left e <- getNextSndFileToPrepare db 86400
@@ -742,12 +745,12 @@ testGetNextSndChunkToUpload st = do
     Right Nothing <- getNextSndChunkToUpload db xftpServer1 86400
 
     -- create file 1
-    Right _ <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce
+    Right _ <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce Nothing
     updateSndFileEncrypted db 1 (FileDigest "abc") [(XFTPChunkSpec "filepath" 1 1, FileDigest "ghi")]
     createSndFileReplica_ db 1 newSndChunkReplica1
     DB.execute_ db "UPDATE snd_files SET num_recipients = 'bad' WHERE snd_file_id = 1"
     -- create file 2
-    Right fId2 <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce
+    Right fId2 <- createSndFile db g 1 (CryptoFile "filepath" Nothing) 1 "filepath" testFileSbKey testFileCbNonce Nothing
     updateSndFileEncrypted db 2 (FileDigest "abc") [(XFTPChunkSpec "filepath" 1 1, FileDigest "ghi")]
     createSndFileReplica_ db 2 newSndChunkReplica1
 

tests/AgentTests/SchemaDump.hs

@@ -7,7 +7,10 @@ import Control.DeepSeq
 import Control.Monad (unless, void)
 import Data.List (dropWhileEnd)
 import Data.Maybe (fromJust, isJust)
+import Database.SQLite.Simple (Only (..))
+import qualified Database.SQLite.Simple as SQL
 import Simplex.Messaging.Agent.Store.SQLite
+import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration (..), MigrationsToRun (..), toDownMigration)
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import Simplex.Messaging.Util (ifM)
@@ -28,6 +31,8 @@ schemaDumpTest :: Spec
 schemaDumpTest = do
   it "verify and overwrite schema dump" testVerifySchemaDump
   it "verify schema down migrations" testSchemaMigrations
+  it "should NOT create user record for new database" testUsersMigrationNew
+  it "should create user record for old database" testUsersMigrationOld
 
 testVerifySchemaDump :: IO ()
 testVerifySchemaDump = do
@@ -61,6 +66,25 @@ testSchemaMigrations = do
       schema''' <- getSchema testDB testSchema
       schema''' `shouldBe` schema'
 
+testUsersMigrationNew :: IO ()
+testUsersMigrationNew = do
+  Right st <- createSQLiteStore testDB "" False Migrations.app MCError
+  withTransaction' st (`SQL.query_` "SELECT user_id FROM users;")
+    `shouldReturn` ([] :: [Only Int])
+  closeSQLiteStore st
+
+testUsersMigrationOld :: IO ()
+testUsersMigrationOld = do
+  let beforeUsers = takeWhile (("m20230110_users" /=) . name) Migrations.app
+  Right st <- createSQLiteStore testDB "" False beforeUsers MCError
+  withTransaction' st (`SQL.query_` "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'users';")
+    `shouldReturn` ([] :: [Only String])
+  closeSQLiteStore st
+  Right st' <- createSQLiteStore testDB "" False Migrations.app MCYesUp
+  withTransaction' st' (`SQL.query_` "SELECT user_id FROM users;")
+    `shouldReturn` ([Only (1 :: Int)])
+  closeSQLiteStore st'
+
 skipComparisonForDownMigrations :: [String]
 skipComparisonForDownMigrations =
   [ -- on down migration idx_messages_internal_snd_id_ts index moves down to the end of the file

tests/CLITests.hs

@@ -1,9 +1,12 @@
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
 
 module CLITests where
 
 import Data.Ini (lookupValue, readIniFile)
 import Data.List (isPrefixOf)
+import qualified Data.X509 as X
+import qualified Data.X509.File as XF
 import Simplex.FileTransfer.Server.Main (xftpServerCLI, xftpServerVersion)
 import Simplex.Messaging.Notifications.Server.Main
 import Simplex.Messaging.Server.Main
@@ -11,6 +14,7 @@ import Simplex.Messaging.Transport (simplexMQVersion)
 import Simplex.Messaging.Util (catchAll_)
 import System.Directory (doesFileExist)
 import System.Environment (withArgs)
+import System.FilePath ((</>))
 import System.IO.Silently (capture_)
 import System.Timeout (timeout)
 import Test.Hspec
@@ -51,6 +55,7 @@ cliTests = do
 
 smpServerTest :: Bool -> Bool -> IO ()
 smpServerTest storeLog basicAuth = do
+  -- init
   capture_ (withArgs (["init", "-y"] <> ["-l" | storeLog] <> ["--no-password" | not basicAuth]) $ smpServerCLI cfgPath logPath)
     >>= (`shouldSatisfy` (("Server initialized, you can modify configuration in " <> cfgPath <> "/smp-server.ini") `isPrefixOf`))
   Right ini <- readIniFile $ cfgPath <> "/smp-server.ini"
@@ -61,12 +66,30 @@ smpServerTest storeLog basicAuth = do
   lookupValue "AUTH" "new_queues" ini `shouldBe` Right "on"
   lookupValue "INACTIVE_CLIENTS" "disconnect" ini `shouldBe` Right "off"
   doesFileExist (cfgPath <> "/ca.key") `shouldReturn` True
+  -- start
   r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` smpServerCLI cfgPath logPath) `catchAll_` pure (Just ()))
   r `shouldContain` ["SMP server v" <> simplexMQVersion]
   r `shouldContain` (if storeLog then ["Store log: " <> logPath <> "/smp-server-store.log"] else ["Store log disabled."])
   r `shouldContain` ["Listening on port 5223 (TLS)..."]
   r `shouldContain` ["not expiring inactive clients"]
   r `shouldContain` (if basicAuth then ["creating new queues requires password"] else ["creating new queues allowed"])
+  -- cert
+  let certPath = cfgPath </> "server.crt"
+  oldCrt@X.Certificate {} <-
+    XF.readSignedObject certPath >>= \case
+      [cert] -> pure . X.signedObject $ X.getSigned cert
+      _ -> error "bad crt format"
+  r' <- lines <$> capture_ (withArgs ["cert"] $ (100000 `timeout` smpServerCLI cfgPath logPath) `catchAll_` pure (Just ()))
+  r' `shouldContain` ["Generated new server credentials"]
+  newCrt <-
+    XF.readSignedObject certPath >>= \case
+      [cert] -> pure . X.signedObject $ X.getSigned cert
+      _ -> error "bad crt format after cert"
+  X.certSignatureAlg oldCrt `shouldBe` X.certSignatureAlg newCrt
+  X.certSubjectDN oldCrt `shouldBe` X.certSubjectDN newCrt
+  X.certSerial oldCrt `shouldNotBe` X.certSerial newCrt
+  X.certPubKey oldCrt `shouldNotBe` X.certPubKey newCrt
+  -- delete
   capture_ (withStdin "Y" . withArgs ["delete"] $ smpServerCLI cfgPath logPath)
     >>= (`shouldSatisfy` ("WARNING: deleting the server will make all queues inaccessible" `isPrefixOf`))
   doesFileExist (cfgPath <> "/ca.key") `shouldReturn` False

tests/CoreTests/BatchingTests.hs

@@ -5,7 +5,7 @@ module CoreTests.BatchingTests (batchingTests) where
 import Control.Concurrent.STM
 import Control.Monad
 import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString as B
 import qualified Data.List.NonEmpty as L
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
@@ -20,7 +20,7 @@ batchingTests = do
     it "should batch with 90 subscriptions per batch" testBatchSubscriptions
     it "should break on message that does not fit" testBatchWithMessage
     it "should break on large message" testBatchWithLargeMessage
-  describe "batchClientTransmissions" $ do
+  describe "batchTransmissions'" $ do
     it "should batch with 90 subscriptions per batch" testClientBatchSubscriptions
     it "should break on message that does not fit" testClientBatchWithMessage
     it "should break on large message" testClientBatchWithLargeMessage
@@ -34,8 +34,8 @@ testBatchSubscriptions = do
   length batches1 `shouldBe` 200
   let batches = batchTransmissions True smpBlockSize $ L.fromList subs
   length batches `shouldBe` 3
-  [TBTransmissions n1 s1, TBTransmissions n2 s2, TBTransmissions n3 s3] <- pure batches
-  (n1, n2, n3) `shouldBe` (90, 90, 20)
+  [TBTransmissions s1 n1 _, TBTransmissions s2 n2 _, TBTransmissions s3 n3 _] <- pure batches
+  (n1, n2, n3) `shouldBe` (20, 90, 90)
   all lenOk [s1, s2, s3] `shouldBe` True
 
 testBatchWithMessage :: IO ()
@@ -50,8 +50,8 @@ testBatchWithMessage = do
   length batches1 `shouldBe` 101
   let batches = batchTransmissions True smpBlockSize $ L.fromList cmds
   length batches `shouldBe` 2
-  [TBTransmissions n1 s1, TBTransmissions n2 s2] <- pure batches
-  (n1, n2) `shouldBe` (60, 41)
+  [TBTransmissions s1 n1 _, TBTransmissions s2 n2 _] <- pure batches
+  (n1, n2) `shouldBe` (55, 46)
   all lenOk [s1, s2] `shouldBe` True
 
 testBatchWithLargeMessage :: IO ()
@@ -69,8 +69,8 @@ testBatchWithLargeMessage = do
   length batches1' `shouldBe` 160
   let batches = batchTransmissions True smpBlockSize $ L.fromList cmds
   length batches `shouldBe` 4
-  [TBTransmissions n1 s1, TBLargeTransmission, TBTransmissions n2 s2, TBTransmissions n3 s3] <- pure batches
-  (n1, n2, n3) `shouldBe` (60, 90, 10)
+  [TBTransmissions s1 n1 _, TBLargeTransmission _, TBTransmissions s2 n2 _, TBTransmissions s3 n3 _] <- pure batches
+  (n1, n2, n3) `shouldBe` (60, 10, 90)
   all lenOk [s1, s2, s3] `shouldBe` True
 
 testClientBatchSubscriptions :: IO ()
@@ -78,13 +78,13 @@ testClientBatchSubscriptions = do
   sessId <- atomically . C.randomBytes 32 =<< C.newRandom
   client <- atomically $ clientStub sessId
   subs <- replicateM 200 $ randomSUBCmd client
-  let batches1 = batchClientTransmissions False smpBlockSize $ L.fromList subs
-  all lenOk1' batches1 `shouldBe` True
-  let batches = batchClientTransmissions True smpBlockSize $ L.fromList subs
+  let batches1 = batchTransmissions' False smpBlockSize $ L.fromList subs
+  all lenOk1 batches1 `shouldBe` True
+  let batches = batchTransmissions' True smpBlockSize $ L.fromList subs
   length batches `shouldBe` 3
-  [CBTransmissions s1 n1 rs1, CBTransmissions s2 n2 rs2, CBTransmissions s3 n3 rs3] <- pure batches
-  (n1, n2, n3) `shouldBe` (90, 90, 20)
-  (length rs1, length rs2, length rs3) `shouldBe` (90, 90, 20)
+  [TBTransmissions s1 n1 rs1, TBTransmissions s2 n2 rs2, TBTransmissions s3 n3 rs3] <- pure batches
+  (n1, n2, n3) `shouldBe` (20, 90, 90)
+  (length rs1, length rs2, length rs3) `shouldBe` (20, 90, 90)
   all lenOk [s1, s2, s3] `shouldBe` True
 
 testClientBatchWithMessage :: IO ()
@@ -95,14 +95,14 @@ testClientBatchWithMessage = do
   send <- randomSENDCmd client 8000
   subs2 <- replicateM 40 $ randomSUBCmd client
   let cmds = subs1 <> [send] <> subs2
-      batches1 = batchClientTransmissions False smpBlockSize $ L.fromList cmds
-  all lenOk1' batches1 `shouldBe` True
+      batches1 = batchTransmissions' False smpBlockSize $ L.fromList cmds
+  all lenOk1 batches1 `shouldBe` True
   length batches1 `shouldBe` 101
-  let batches = batchClientTransmissions True smpBlockSize $ L.fromList cmds
+  let batches = batchTransmissions' True smpBlockSize $ L.fromList cmds
   length batches `shouldBe` 2
-  [CBTransmissions s1 n1 rs1, CBTransmissions s2 n2 rs2] <- pure batches
-  (n1, n2) `shouldBe` (60, 41)
-  (length rs1, length rs2) `shouldBe` (60, 41)
+  [TBTransmissions s1 n1 rs1, TBTransmissions s2 n2 rs2] <- pure batches
+  (n1, n2) `shouldBe` (55, 46)
+  (length rs1, length rs2) `shouldBe` (55, 46)
   all lenOk [s1, s2] `shouldBe` True
 
 testClientBatchWithLargeMessage :: IO ()
@@ -113,26 +113,26 @@ testClientBatchWithLargeMessage = do
   send <- randomSENDCmd client 17000
   subs2 <- replicateM 100 $ randomSUBCmd client
   let cmds = subs1 <> [send] <> subs2
-      batches1 = batchClientTransmissions False smpBlockSize $ L.fromList cmds
-  all lenOk1' batches1 `shouldBe` False
+      batches1 = batchTransmissions' False smpBlockSize $ L.fromList cmds
+  all lenOk1 batches1 `shouldBe` False
   length batches1 `shouldBe` 161
   let batches1' = take 60 batches1 <> drop 61 batches1
-  all lenOk1' batches1' `shouldBe` True
+  all lenOk1 batches1' `shouldBe` True
   length batches1' `shouldBe` 160
   --
-  let batches = batchClientTransmissions True smpBlockSize $ L.fromList cmds
+  let batches = batchTransmissions' True smpBlockSize $ L.fromList cmds
   length batches `shouldBe` 4
-  [CBTransmissions s1 n1 rs1, CBLargeTransmission _, CBTransmissions s2 n2 rs2, CBTransmissions s3 n3 rs3] <- pure batches
-  (n1, n2, n3) `shouldBe` (60, 90, 10)
-  (length rs1, length rs2, length rs3) `shouldBe` (60, 90, 10)
+  [TBTransmissions s1 n1 rs1, TBLargeTransmission _, TBTransmissions s2 n2 rs2, TBTransmissions s3 n3 rs3] <- pure batches
+  (n1, n2, n3) `shouldBe` (60, 10, 90)
+  (length rs1, length rs2, length rs3) `shouldBe` (60, 10, 90)
   all lenOk [s1, s2, s3] `shouldBe` True
   --
   let cmds' = [send] <> subs1 <> subs2
-  let batches' = batchClientTransmissions True smpBlockSize $ L.fromList cmds'
+  let batches' = batchTransmissions' True smpBlockSize $ L.fromList cmds'
   length batches' `shouldBe` 3
-  [CBLargeTransmission _, CBTransmissions s1' n1' rs1', CBTransmissions s2' n2' rs2'] <- pure batches'
-  (n1', n2') `shouldBe` (90, 70)
-  (length rs1', length rs2') `shouldBe` (90, 70)
+  [TBLargeTransmission _, TBTransmissions s1' n1' rs1', TBTransmissions s2' n2' rs2'] <- pure batches'
+  (n1', n2') `shouldBe` (70, 90)
+  (length rs1', length rs2') `shouldBe` (70, 90)
   all lenOk [s1', s2'] `shouldBe` True
 
 randomSUB :: ByteString -> IO (Maybe C.ASignature, ByteString)
@@ -172,12 +172,7 @@ randomSENDCmd c len = do
 lenOk :: ByteString -> Bool
 lenOk s = 0 < B.length s && B.length s <= smpBlockSize - 2
 
-lenOk1 :: TransportBatch -> Bool
+lenOk1 :: TransportBatch r -> Bool
 lenOk1 = \case
-  TBTransmission s -> lenOk s
-  _ -> False
-
-lenOk1' :: ClientBatch err msg -> Bool
-lenOk1' = \case
-  CBTransmission s _ -> lenOk s
+  TBTransmission s _ -> lenOk s
   _ -> False

tests/FileDescriptionTests.hs

@@ -13,16 +13,20 @@ import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String (StrEncoding (..))
+import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import System.Directory (removeFile)
 import Test.Hspec
 
 fileDescriptionTests :: Spec
-fileDescriptionTests =
+fileDescriptionTests = do
   describe "file description parsing / serializing" $ do
     it "parse YAML file description" testParseYAMLFileDescription
     it "serialize YAML file description" testSerializeYAMLFileDescription
     it "parse file description" testParseFileDescription
     it "serialize file description" testSerializeFileDescription
+    describe "file description URIs" $ do
+      it "round trip file description URI" testFileDescriptionURI
+      it "round trip file description URI with extra JSON" testFileDescriptionURIExtras
 
 fileDescPath :: FilePath
 fileDescPath = "tests/fixtures/file_description.yaml"
@@ -82,7 +86,8 @@ fileDesc =
                   FileChunkReplica {server = "xftp://abc=@example3.com", replicaId, replicaKey}
                 ]
             }
-        ]
+        ],
+      redirect = Nothing
     }
   where
     defaultChunkSize = FileSize $ mb 8
@@ -128,7 +133,8 @@ yamlFileDesc =
                   "3:YWJj:MC4CAQAwBQYDK2VwBCIEIDfEfevydXXfKajz3sRkcQ7RPvfWUPoq6pu1TYHV1DEe"
                 ]
             }
-        ]
+        ],
+      redirect = Nothing
     }
 
 testParseYAMLFileDescription :: IO ()
@@ -157,6 +163,18 @@ testSerializeFileDescription = withRemoveTmpFile $ do
   fdExp <- B.readFile fileDescPath
   fdSer `shouldBe` fdExp
 
+testFileDescriptionURI :: IO ()
+testFileDescriptionURI = do
+  vfd <- either fail pure $ validateFileDescription fileDesc
+  let descr = FileDescriptionURI SSSimplex vfd mempty
+  strDecode (strEncode descr) `shouldBe` Right descr
+
+testFileDescriptionURIExtras :: IO ()
+testFileDescriptionURIExtras = do
+  vfd <- either fail pure $ validateFileDescription fileDesc
+  let descr = FileDescriptionURI SSSimplex vfd $ Just "{\"something\":\"extra\",\"more\":true}"
+  strDecode (strEncode descr) `shouldBe` Right descr
+
 withRemoveTmpFile :: IO () -> IO ()
 withRemoveTmpFile =
   bracket_

tests/NtfClient.hs

@@ -57,6 +57,9 @@ testHost = "localhost"
 ntfTestPort :: ServiceName
 ntfTestPort = "6001"
 
+ntfTestPort2 :: ServiceName
+ntfTestPort2 = "6002"
+
 apnsTestPort :: ServiceName
 apnsTestPort = "6010"
 
@@ -77,7 +80,7 @@ testNtfClient client = do
 ntfServerCfg :: NtfServerConfig
 ntfServerCfg =
   NtfServerConfig
-    { transports = undefined,
+    { transports = [],
       subIdBytes = 24,
       regCodeBytes = 32,
       clientQSize = 1,
@@ -105,16 +108,19 @@ ntfServerCfg =
     }
 
 withNtfServerStoreLog :: ATransport -> (ThreadId -> IO a) -> IO a
-withNtfServerStoreLog t = withNtfServerCfg t ntfServerCfg {storeLogFile = Just ntfTestStoreLogFile}
+withNtfServerStoreLog t = withNtfServerCfg ntfServerCfg {storeLogFile = Just ntfTestStoreLogFile, transports = [(ntfTestPort, t)]}
 
 withNtfServerThreadOn :: ATransport -> ServiceName -> (ThreadId -> IO a) -> IO a
-withNtfServerThreadOn t port' = withNtfServerCfg t ntfServerCfg {transports = [(port', t)]}
+withNtfServerThreadOn t port' = withNtfServerCfg ntfServerCfg {transports = [(port', t)]}
 
-withNtfServerCfg :: ATransport -> NtfServerConfig -> (ThreadId -> IO a) -> IO a
-withNtfServerCfg t cfg =
-  serverBracket
-    (\started -> runNtfServerBlocking started cfg {transports = [(ntfTestPort, t)]})
-    (pure ())
+withNtfServerCfg :: HasCallStack => NtfServerConfig -> (ThreadId -> IO a) -> IO a
+withNtfServerCfg cfg@NtfServerConfig {transports} =
+  case transports of
+    [] -> error "no transports configured"
+    _ ->
+      serverBracket
+        (\started -> runNtfServerBlocking started cfg)
+        (pure ())
 
 withNtfServerOn :: ATransport -> ServiceName -> IO a -> IO a
 withNtfServerOn t port' = withNtfServerThreadOn t port' . const
@@ -136,7 +142,7 @@ ntfServerTest _ t = runNtfTest $ \h -> tPut' h t >> tGet' h
     tPut' :: THandle c -> (Maybe C.ASignature, ByteString, ByteString, smp) -> IO ()
     tPut' h@THandle {sessionId} (sig, corrId, queueId, smp) = do
       let t' = smpEncode (sessionId, corrId, queueId, smp)
-      [Right ()] <- tPut h Nothing [(sig, t')]
+      [Right ()] <- tPut h [(sig, t')]
       pure ()
     tGet' h = do
       [(Nothing, _, (CorrId corrId, qId, Right cmd))] <- tGet h

tests/SMPAgentClient.hs

@@ -1,6 +1,7 @@
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
@@ -9,12 +10,14 @@
 
 module SMPAgentClient where
 
+import Control.Monad
 import Control.Monad.IO.Unlift
 import Crypto.Random
 import qualified Data.ByteString.Char8 as B
 import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
+import qualified Database.SQLite.Simple as SQL
 import Network.Socket (ServiceName)
 import NtfClient (ntfTestPort)
 import SMPClient
@@ -30,10 +33,11 @@ import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Server (runSMPAgentBlocking)
-import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..))
+import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteStore (dbNew))
+import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import Simplex.Messaging.Client (ProtocolClientConfig (..), chooseTransportHost, defaultClientConfig, defaultNetworkConfig)
 import Simplex.Messaging.Parsers (parseAll)
-import Simplex.Messaging.Protocol (ProtoServerWithAuth)
+import Simplex.Messaging.Protocol (NtfServer, ProtoServerWithAuth)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
 import Test.Hspec
@@ -176,11 +180,17 @@ testSMPServer = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:50
 testSMPServer2 :: SMPServer
 testSMPServer2 = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5002"
 
+testNtfServer :: NtfServer
+testNtfServer = "ntf://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:6001"
+
+testNtfServer2 :: NtfServer
+testNtfServer2 = "ntf://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:6002"
+
 initAgentServers :: InitialAgentServers
 initAgentServers =
   InitialAgentServers
     { smp = userServers [noAuthSrv testSMPServer],
-      ntf = ["ntf://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:6001"],
+      ntf = [testNtfServer],
       xftp = userServers [noAuthSrv testXFTPServer],
       netCfg = defaultNetworkConfig {tcpTimeout = 500_000, tcpConnectTimeout = 500_000}
     }
@@ -194,27 +204,36 @@ agentCfg =
     { tcpPort = agentTestPort,
       tbqSize = 4,
       -- database = testDB,
-      smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS)},
-      ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS)},
-      reconnectInterval = defaultReconnectInterval {initialInterval = 50_000},
+      smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS), networkConfig},
+      ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS), networkConfig},
+      reconnectInterval = fastRetryInterval,
       xftpNotifyErrsOnRetry = False,
-      ntfWorkerDelay = 1000,
-      ntfSMPWorkerDelay = 1000,
+      ntfWorkerDelay = 100,
+      ntfSMPWorkerDelay = 100,
       caCertificateFile = "tests/fixtures/ca.crt",
       privateKeyFile = "tests/fixtures/server.key",
       certificateFile = "tests/fixtures/server.crt"
     }
+  where
+    networkConfig = defaultNetworkConfig {tcpConnectTimeout = 3_000_000, tcpTimeout = 2_000_000}
+
+fastRetryInterval :: RetryInterval
+fastRetryInterval = defaultReconnectInterval {initialInterval = 50_000}
+
+fastMessageRetryInterval :: RetryInterval2
+fastMessageRetryInterval = RetryInterval2 {riFast = fastRetryInterval, riSlow = fastRetryInterval}
 
 type AgentTestMonad m = (MonadUnliftIO m, MonadRandom m, MonadFail m)
 
-withSmpAgentThreadOn_ :: AgentTestMonad m => ATransport -> (ServiceName, ServiceName, FilePath) -> m () -> (ThreadId -> m a) -> m a
-withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =
+withSmpAgentThreadOn_ :: AgentTestMonad m => ATransport -> (ServiceName, ServiceName, FilePath) -> Int -> m () -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn_ t (port', smpPort', db') initClientId afterProcess =
   let cfg' = agentCfg {tcpPort = port'}
       initServers' = initAgentServers {smp = userServers [ProtoServerWithAuth (SMPServer "localhost" smpPort' testKeyHash) Nothing]}
    in serverBracket
         ( \started -> do
             Right st <- liftIO $ createAgentStore db' "" False MCError
-            runSMPAgentBlocking t cfg' initServers' st started
+            when (dbNew st) . liftIO $ withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1)")
+            runSMPAgentBlocking t cfg' initServers' st initClientId started
         )
         afterProcess
 
@@ -222,7 +241,7 @@ userServers :: NonEmpty (ProtoServerWithAuth p) -> Map UserId (NonEmpty (ProtoSe
 userServers srvs = M.fromList [(1, srvs)]
 
 withSmpAgentThreadOn :: AgentTestMonad m => ATransport -> (ServiceName, ServiceName, FilePath) -> (ThreadId -> m a) -> m a
-withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ removeFile db'
+withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a 0 $ removeFile db'
 
 withSmpAgentOn :: AgentTestMonad m => ATransport -> (ServiceName, ServiceName, FilePath) -> m a -> m a
 withSmpAgentOn t (port', smpPort', db') = withSmpAgentThreadOn t (port', smpPort', db') . const

tests/SMPClient.hs

@@ -130,7 +130,7 @@ serverBracket process afterProcess f = do
   E.bracket
     (forkIOWithUnmask ($ process started))
     (\t -> killThread t >> afterProcess >> waitFor started "stop")
-    (\t -> waitFor started "start" >> f t)
+    (\t -> waitFor started "start" >> f t >>= \r -> r <$ threadDelay 100000)
   where
     waitFor started s =
       5_000_000 `timeout` atomically (takeTMVar started) >>= \case
@@ -164,7 +164,7 @@ smpServerTest _ t = runSmpTest $ \h -> tPut' h t >> tGet' h
     tPut' :: THandle c -> (Maybe C.ASignature, ByteString, ByteString, smp) -> IO ()
     tPut' h@THandle {sessionId} (sig, corrId, queueId, smp) = do
       let t' = smpEncode (sessionId, corrId, queueId, smp)
-      [Right ()] <- tPut h Nothing [(sig, t')]
+      [Right ()] <- tPut h [(sig, t')]
       pure ()
     tGet' h = do
       [(Nothing, _, (CorrId corrId, qId, Right cmd))] <- tGet h

tests/ServerTests.hs

@@ -89,7 +89,7 @@ signSendRecv h@THandle {thVersion, sessionId} pk (corrId, qId, cmd) = do
 
 tPut1 :: Transport c => THandle c -> SentRawTransmission -> IO (Either TransportError ())
 tPut1 h t = do
-  [r] <- tPut h Nothing [t]
+  [r] <- tPut h [t]
   pure r
 
 tGet1 :: (ProtocolEncoding err cmd, Transport c, MonadIO m, MonadFail m) => THandle c -> m (SignedTransmission err cmd)

tests/XFTPAgent.hs

@@ -1,14 +1,15 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module XFTPAgent where
 
 import AgentTests.FunctionalAPITests (get, getSMPAgentClient', rfGet, runRight, runRight_, sfGet)
-import Control.Concurrent (threadDelay)
-import Control.Concurrent.STM
+
 import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
@@ -19,10 +20,10 @@ import Data.Int (Int64)
 import Data.List (find, isSuffixOf)
 import Data.Maybe (fromJust)
 import SMPAgentClient (agentCfg, initAgentServers, testDB, testDB2, testDB3)
-import Simplex.FileTransfer.Description
+import Simplex.FileTransfer.Description (FileDescription (..), FileDescriptionURI (..), ValidFileDescription, fileDescriptionURI, mb, qrSizeLimit, pattern ValidFileDescription)
 import Simplex.FileTransfer.Protocol (FileParty (..), XFTPErrorType (AUTH))
 import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..))
-import Simplex.Messaging.Agent (AgentClient, disconnectAgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendFile, xftpStartWorkers)
+import Simplex.Messaging.Agent (AgentClient, disconnectAgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendDescription, xftpSendFile, xftpStartWorkers)
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..))
 import Simplex.Messaging.Agent.Protocol (ACommand (..), AgentErrorType (..), BrokerErrorType (..), RcvFileId, SndFileId, noAuthSrv)
 import qualified Simplex.Messaging.Crypto as C
@@ -31,10 +32,12 @@ import qualified Simplex.Messaging.Crypto.File as CF
 import Simplex.Messaging.Encoding.String (StrEncoding (..))
 import Simplex.Messaging.Protocol (BasicAuth, ProtoServerWithAuth (..), ProtocolServer (..), XFTPServerWithAuth)
 import Simplex.Messaging.Server.Expiration (ExpirationConfig (..))
+import Simplex.Messaging.Util (tshow)
 import System.Directory (doesDirectoryExist, doesFileExist, getFileSize, listDirectory, removeFile)
 import System.FilePath ((</>))
-import System.Timeout (timeout)
 import Test.Hspec
+import UnliftIO
+import UnliftIO.Concurrent
 import XFTPCLI
 import XFTPClient
 
@@ -42,6 +45,8 @@ xftpAgentTests :: Spec
 xftpAgentTests = around_ testBracket . describe "agent XFTP API" $ do
   it "should send and receive file" testXFTPAgentSendReceive
   it "should send and receive with encrypted local files" testXFTPAgentSendReceiveEncrypted
+  it "should send and receive large file with a redirect" testXFTPAgentSendReceiveRedirect
+  it "should send and receive small file without a redirect" testXFTPAgentSendReceiveNoRedirect
   it "should resume receiving file after restart" testXFTPAgentReceiveRestore
   it "should cleanup rcv tmp path after permanent error" testXFTPAgentReceiveCleanup
   it "should resume sending file after restart" testXFTPAgentSendRestore
@@ -94,18 +99,18 @@ testXFTPAgentSendReceive :: HasCallStack => IO ()
 testXFTPAgentSendReceive = withXFTPServer $ do
   filePath <- createRandomFile
   -- send file, delete snd file internally
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   (rfd1, rfd2) <- runRight $ do
     (sfId, _, rfd1, rfd2) <- testSend sndr filePath
     xftpDeleteSndFileInternal sndr sfId
     pure (rfd1, rfd2)
 
   -- receive file, delete rcv file
-  testReceiveDelete rfd1 filePath
-  testReceiveDelete rfd2 filePath
+  testReceiveDelete 2 rfd1 filePath
+  testReceiveDelete 3 rfd2 filePath
   where
-    testReceiveDelete rfd originalFilePath = do
-      rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    testReceiveDelete clientId rfd originalFilePath = do
+      rcp <- getSMPAgentClient' clientId agentCfg initAgentServers testDB2
       runRight_ $ do
         rfId <- testReceive rcp rfd originalFilePath
         xftpDeleteRcvFile rcp rfId
@@ -118,31 +123,129 @@ testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do
   s <- LB.readFile filePath
   file <- atomically $ CryptoFile (senderFiles </> "encrypted_testfile") . Just <$> CF.randomArgs g
   runRight_ $ CF.writeFile file s
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   (rfd1, rfd2) <- runRight $ do
     (sfId, _, rfd1, rfd2) <- testSendCF sndr file
     xftpDeleteSndFileInternal sndr sfId
     pure (rfd1, rfd2)
   -- receive file, delete rcv file
-  testReceiveDelete rfd1 filePath g
-  testReceiveDelete rfd2 filePath g
+  testReceiveDelete 2 rfd1 filePath g
+  testReceiveDelete 3 rfd2 filePath g
   where
-    testReceiveDelete rfd originalFilePath g = do
-      rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    testReceiveDelete clientId rfd originalFilePath g = do
+      rcp <- getSMPAgentClient' clientId agentCfg initAgentServers testDB2
       cfArgs <- atomically $ Just <$> CF.randomArgs g
       runRight_ $ do
         rfId <- testReceiveCF rcp rfd cfArgs originalFilePath
         xftpDeleteRcvFile rcp rfId
       disconnectAgentClient rcp
 
+testXFTPAgentSendReceiveRedirect :: HasCallStack => IO ()
+testXFTPAgentSendReceiveRedirect = withXFTPServer $ do
+  --- sender
+  filePathIn <- createRandomFile
+  let fileSize = mb 17
+      totalSize = fileSize + mb 1
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  directFileId <- runRight $ xftpSendFile sndr 1 (CryptoFile filePathIn Nothing) 1
+  sfGet sndr `shouldReturn` ("", directFileId, SFPROG 4194304 totalSize)
+  sfGet sndr `shouldReturn` ("", directFileId, SFPROG 8388608 totalSize)
+  sfGet sndr `shouldReturn` ("", directFileId, SFPROG 12582912 totalSize)
+  sfGet sndr `shouldReturn` ("", directFileId, SFPROG 16777216 totalSize)
+  sfGet sndr `shouldReturn` ("", directFileId, SFPROG 17825792 totalSize)
+  sfGet sndr `shouldReturn` ("", directFileId, SFPROG totalSize totalSize)
+  vfdDirect <-
+    sfGet sndr >>= \case
+      (_, _, SFDONE _snd (vfd : _)) -> pure vfd
+      r -> error $ "Expected SFDONE, got " <> show r
+  redirectFileId <- runRight $ xftpSendDescription sndr 1 vfdDirect 1
+  logInfo $ "File sent, sending redirect: " <> tshow redirectFileId
+  sfGet sndr `shouldReturn` ("", redirectFileId, SFPROG 65536 65536)
+  vfdRedirect@(ValidFileDescription fdRedirect) <-
+    sfGet sndr >>= \case
+      (_, _, SFDONE _snd (vfd : _)) -> pure vfd
+      r -> error $ "Expected SFDONE, got " <> show r
+  case fdRedirect of
+    FileDescription {redirect = Just _} -> pure ()
+    _ -> error "missing RedirectFileInfo"
+  let uri = strEncode $ fileDescriptionURI vfdRedirect
+  case strDecode uri of
+    Left err -> fail err
+    Right ok -> ok `shouldBe` fileDescriptionURI vfdRedirect
+  disconnectAgentClient sndr
+  --- recipient
+  rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  FileDescriptionURI {description} <- either fail pure $ strDecode uri
+
+  rcvFileId <- runRight $ xftpReceiveFile rcp 1 description Nothing
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 65536 totalSize) -- extra RFPROG before switching to real file
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 4194304 totalSize)
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 8388608 totalSize)
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 12582912 totalSize)
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 16777216 totalSize)
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 17825792 totalSize)
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG totalSize totalSize)
+  out <-
+    rfGet rcp >>= \case
+      (_, _, RFDONE out) -> pure out
+      r -> error $ "Expected RFDONE, got " <> show r
+  disconnectAgentClient rcp
+
+  inBytes <- B.readFile filePathIn
+  B.readFile out `shouldReturn` inBytes
+
+testXFTPAgentSendReceiveNoRedirect :: HasCallStack => IO ()
+testXFTPAgentSendReceiveNoRedirect = withXFTPServer $ do
+  --- sender
+  let fileSize = mb 5
+  filePathIn <- createRandomFile_ fileSize "testfile"
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
+  directFileId <- runRight $ xftpSendFile sndr 1 (CryptoFile filePathIn Nothing) 1
+  let totalSize = fileSize + mb 1
+  sfGet sndr `shouldReturn` ("", directFileId, SFPROG 4194304 totalSize)
+  sfGet sndr `shouldReturn` ("", directFileId, SFPROG 5242880 totalSize)
+  sfGet sndr `shouldReturn` ("", directFileId, SFPROG totalSize totalSize)
+  vfdDirect <-
+    sfGet sndr >>= \case
+      (_, _, SFDONE _snd (vfd : _)) -> pure vfd
+      r -> error $ "Expected SFDONE, got " <> show r
+  B.putStrLn $ strEncode vfdDirect
+  let uri = strEncode $ fileDescriptionURI vfdDirect
+  B.length uri `shouldSatisfy` (< qrSizeLimit)
+  case strDecode uri of
+    Left err -> fail err
+    Right ok -> ok `shouldBe` fileDescriptionURI vfdDirect
+  disconnectAgentClient sndr
+  --- recipient
+  rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
+  FileDescriptionURI {description} <- either fail pure $ strDecode uri
+  let ValidFileDescription FileDescription {redirect} = description
+  redirect `shouldBe` Nothing
+  rcvFileId <- runRight $ xftpReceiveFile rcp 1 description Nothing
+  -- NO extra "RFPROG 65k 65k" before switching to real file
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 4194304 totalSize)
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG 5242880 totalSize)
+  rfGet rcp `shouldReturn` ("", rcvFileId, RFPROG totalSize totalSize)
+  out <-
+    rfGet rcp >>= \case
+      (_, _, RFDONE out) -> pure out
+      r -> error $ "Expected RFDONE, got " <> show r
+  disconnectAgentClient rcp
+
+  inBytes <- B.readFile filePathIn
+  B.readFile out `shouldReturn` inBytes
+
 createRandomFile :: HasCallStack => IO FilePath
 createRandomFile = createRandomFile' "testfile"
 
 createRandomFile' :: HasCallStack => FilePath -> IO FilePath
-createRandomFile' fileName = do
+createRandomFile' = createRandomFile_ (mb 17 :: Integer)
+
+createRandomFile_ :: (HasCallStack, Integral s, Show s) => s -> FilePath -> IO FilePath
+createRandomFile_ size fileName = do
   let filePath = senderFiles </> fileName
-  xftpCLI ["rand", filePath, "17mb"] `shouldReturn` ["File created: " <> filePath]
-  getFileSize filePath `shouldReturn` mb 17
+  xftpCLI ["rand", filePath, show size] `shouldReturn` ["File created: " <> filePath]
+  getFileSize filePath `shouldReturn` toInteger size
   pure filePath
 
 testSend :: HasCallStack => AgentClient -> FilePath -> ExceptT AgentErrorType IO (SndFileId, ValidFileDescription 'FSender, ValidFileDescription 'FRecipient, ValidFileDescription 'FRecipient)
@@ -188,13 +291,13 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
 
   rfd <- withXFTPServerStoreLogOn $ \_ -> do
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     runRight $ do
       (_, _, rfd, _) <- testSend sndr filePath
       pure rfd
 
   -- receive file - should not succeed with server down
-  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   rfId <- runRight $ do
     xftpStartWorkers rcp (Just recipientFiles)
     rfId <- xftpReceiveFile rcp 1 rfd Nothing
@@ -208,7 +311,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- receive file - should start downloading with server up
-    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     ("", rfId', RFPROG _ _) <- rfGet rcp'
     liftIO $ rfId' `shouldBe` rfId
@@ -218,7 +321,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- receive file - should continue downloading with server up
-    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp' <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     rfProgress rcp' $ mb 18
     ("", rfId', RFDONE path) <- rfGet rcp'
@@ -236,13 +339,13 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
 
   rfd <- withXFTPServerStoreLogOn $ \_ -> do
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     runRight $ do
       (_, _, rfd, _) <- testSend sndr filePath
       pure rfd
 
   -- receive file - should not succeed with server down
-  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   rfId <- runRight $ do
     xftpStartWorkers rcp (Just recipientFiles)
     rfId <- xftpReceiveFile rcp 1 rfd Nothing
@@ -256,7 +359,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerThreadOn $ \_ -> do
     -- receive file - should fail with AUTH error
-    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp'
     rfId' `shouldBe` rfId
@@ -269,7 +372,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
   filePath <- createRandomFile
 
   -- send file - should not succeed with server down
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   sfId <- runRight $ do
     xftpStartWorkers sndr (Just senderFiles)
     sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2
@@ -286,7 +389,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- send file - should start uploading with server up
-    sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB
     runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
     ("", sfId', SFPROG _ _) <- sfGet sndr'
     liftIO $ sfId' `shouldBe` sfId
@@ -296,7 +399,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- send file - should continue uploading with server up
-    sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
     runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
     sfProgress sndr' $ mb 18
     ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr'
@@ -308,7 +411,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
     doesFileExist encPath `shouldReturn` False
 
     -- receive file
-    rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp rfd1 filePath
 
@@ -318,7 +421,7 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
 
   sfId <- withXFTPServerStoreLogOn $ \_ -> do
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     sfId <- runRight $ do
       xftpStartWorkers sndr (Just senderFiles)
       sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2
@@ -339,7 +442,7 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerThreadOn $ \_ -> do
     -- send file - should fail with AUTH error
-    sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB
     runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
     ("", sfId', SFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- sfGet sndr'
     sfId' `shouldBe` sfId
@@ -354,11 +457,11 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
     filePath <- createRandomFile
 
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     (sfId, sndDescr, rfd1, rfd2) <- runRight $ testSend sndr filePath
 
     -- receive file
-    rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp1 <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp1 rfd1 filePath
 
@@ -376,7 +479,7 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
     length <$> listDirectory xftpServerFiles `shouldReturn` 0
 
     -- receive file - should fail with AUTH error
-    rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
     runRight $ do
       xftpStartWorkers rcp2 (Just recipientFiles)
       rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
@@ -389,11 +492,11 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
 
   (sfId, sndDescr, rfd2) <- withXFTPServerStoreLogOn $ \_ -> do
     -- send file
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     (sfId, sndDescr, rfd1, rfd2) <- runRight $ testSend sndr filePath
 
     -- receive file
-    rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp1 <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp1 rfd1 filePath
     disconnectAgentClient rcp1
@@ -401,7 +504,7 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
     pure (sfId, sndDescr, rfd2)
 
   -- delete file - should not succeed with server down
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   runRight $ do
     xftpStartWorkers sndr (Just senderFiles)
     xftpDeleteSndFileRemote sndr 1 sfId sndDescr
@@ -413,14 +516,14 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- delete file - should succeed with server up
-    sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr' <- getSMPAgentClient' 4 agentCfg initAgentServers testDB
     runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
 
     threadDelay 1000000
     length <$> listDirectory xftpServerFiles `shouldReturn` 0
 
     -- receive file - should fail with AUTH error
-    rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB3
+    rcp2 <- getSMPAgentClient' 5 agentCfg initAgentServers testDB3
     runRight $ do
       xftpStartWorkers rcp2 (Just recipientFiles)
       rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
@@ -433,11 +536,11 @@ testXFTPAgentDeleteOnServer = withGlobalLogging logCfgNoLogs $
     filePath1 <- createRandomFile' "testfile1"
 
     -- send file 1
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     (_, _, rfd1_1, rfd1_2) <- runRight $ testSend sndr filePath1
 
     -- receive file 1 successfully
-    rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp rfd1_1 filePath1
 
@@ -471,11 +574,11 @@ testXFTPAgentExpiredOnServer = withGlobalLogging logCfgNoLogs $ do
     filePath1 <- createRandomFile' "testfile1"
 
     -- send file 1
-    sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+    sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
     (_, _, rfd1_1, rfd1_2) <- runRight $ testSend sndr filePath1
 
     -- receive file 1 successfully
-    rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+    rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
     runRight_ . void $
       testReceive rcp rfd1_1 filePath1
 
@@ -509,7 +612,7 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
   filePath <- createRandomFile
 
   -- send file
-  sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
+  sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   rfds <- runRight $ do
     xftpStartWorkers sndr (Just senderFiles)
     sfId <- xftpSendFile sndr 1 (CF.plain filePath) 500
@@ -522,7 +625,7 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
 
   -- receive file using different descriptions
   -- ! revise number of recipients and indexes if xftpMaxRecipientsPerRequest is changed
-  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   runRight_ $ do
     void $ testReceive rcp (head rfds) filePath
     void $ testReceive rcp (rfds !! 99) filePath
@@ -532,5 +635,5 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
 testXFTPServerTest :: HasCallStack => Maybe BasicAuth -> XFTPServerWithAuth -> IO (Maybe ProtocolTestFailure)
 testXFTPServerTest newFileBasicAuth srv =
   withXFTPServerCfg testXFTPServerConfig {newFileBasicAuth, xftpPort = xftpTestPort2} $ \_ -> do
-    a <- getSMPAgentClient' agentCfg initAgentServers testDB -- initially passed server is not running
+    a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB -- initially passed server is not running
     runRight $ testProtocolServer a 1 srv

tests/XFTPCLI.hs

@@ -152,7 +152,7 @@ testPrepareChunkSizes = do
   prepareChunkSizes (mb 2 + 1) `shouldBe` [mb 1, mb 1, kb 256]
   prepareChunkSizes (3 * kb 256 + 1) `shouldBe` [mb 1]
   prepareChunkSizes (3 * kb 256) `shouldBe` r3 (kb 256)
-  prepareChunkSizes 1 `shouldBe` [kb 256]
+  prepareChunkSizes 1 `shouldBe` [kb 64]
   where
     r3 = replicate 3
 

tests/XFTPClient.hs

@@ -102,7 +102,7 @@ testXFTPServerConfig =
       storeLogFile = Nothing,
       filesPath = xftpServerFiles,
       fileSizeQuota = Nothing,
-      allowedChunkSizes = [kb 128, kb 256, mb 1, mb 4],
+      allowedChunkSizes = [kb 64, kb 128, kb 256, mb 1, mb 4],
       allowNewFiles = True,
       newFileBasicAuth = Nothing,
       fileExpiration = Just defaultFileExpiration,