1.0.3

* increase queue sizes

* remove TODO

.github/changelog_conf.json

@@ -1,4 +1,4 @@
 {
   "template": "${{UNCATEGORIZED}}",
-  "pr_template": "- ${{TITLE}}\n"
+  "pr_template": "- ${{TITLE}}"
 }

.github/workflows/build.yml

@@ -66,7 +66,8 @@ jobs:
         with:
           body: |
             See full changelog [here](https://github.com/simplex-chat/simplexmq/blob/master/CHANGELOG.md).
-            Commits, chronological:
+
+            Commits:
             ${{ steps.build_changelog.outputs.changelog }}
           prerelease: ${{ steps.extract_release_candidate.outputs.release_candidate }}
           files: |

CHANGELOG.md

@@ -1,3 +1,15 @@
+# 1.0.3
+
+SMP server:
+- Reduce server message queue quota to 128 messages.
+
+SMP agent:
+- Add "yes to migrations" option.
+- Make new SMP client attempt to reconnect on network error.
+- Reduce connection handshake expiration to 2 days.
+
+JSON encoding of types used in simplex-chat, some other minor adjustments.
+
 # 1.0.2
 
 General:

apps/smp-agent/Main.hs

@@ -6,8 +6,8 @@ module Main where
 
 import Control.Logger.Simple
 import qualified Data.List.NonEmpty as L
-import Simplex.Messaging.Agent (runSMPAgent)
 import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Agent.Server (runSMPAgent)
 import Simplex.Messaging.Transport (TLS, Transport (..))
 
 cfg :: AgentConfig

apps/smp-server/Main.hs

@@ -22,12 +22,13 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Server (runSMPServer)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.StoreLog (StoreLog, openReadStoreLog, storeLogFilePath)
-import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..), loadFingerprint, simplexMQVersion)
+import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..), simplexMQVersion)
+import Simplex.Messaging.Transport.Server (loadFingerprint)
 import Simplex.Messaging.Transport.WebSockets (WS)
 import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist, removeDirectoryRecursive)
 import System.Exit (exitFailure)
 import System.FilePath (combine)
-import System.IO (BufferMode (..), IOMode (..), hGetLine, withFile, hSetBuffering, stderr, stdout)
+import System.IO (BufferMode (..), IOMode (..), hGetLine, hSetBuffering, stderr, stdout, withFile)
 import System.Process (readCreateProcess, shell)
 import Text.Read (readMaybe)
 
@@ -276,8 +277,8 @@ runServer IniOptions {enableStoreLog, port, enableWebsockets} = do
       ServerConfig
         { transports = (port, transport @TLS) : [("80", transport @WS) | enableWebsockets],
           tbqSize = 16,
-          serverTbqSize = 128,
-          msgQueueQuota = 256,
+          serverTbqSize = 64,
+          msgQueueQuota = 128,
           queueIdBytes = 24,
           msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
           caCertificateFile = caCrtFile,

cabal.project

@@ -1,7 +1,6 @@
 packages: .
 
--- source-repository-package
---   type: git
---   location: git://github.com/simplex-chat/hs-tls.git
---   tag: f6cc753611f80af300401cfae63846e9d7c40d9e
---   subdir: core
+source-repository-package
+    type: git
+    location: git://github.com/simplex-chat/aeson.git
+    tag: 3eb66f9a68f103b5f1489382aad89f5712a64db7

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 1.0.2
+version: 1.0.3
 synopsis: SimpleXMQ message broker
 description:  |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -22,12 +22,12 @@ extra-source-files:
   - CHANGELOG.md
 
 dependencies:
-  - aeson == 1.5.*
+  - aeson == 2.0.*
   - ansi-terminal >= 0.10 && < 0.12
   - asn1-encoding == 0.9.*
   - asn1-types == 0.3.*
   - async == 2.2.*
-  - attoparsec == 0.13.*
+  - attoparsec == 0.14.*
   - base >= 4.7 && < 5
   - base64-bytestring >= 1.0 && < 1.3
   - bytestring == 0.10.*

scripts/smp-server-linode.sh

@@ -52,12 +52,12 @@ curl -L -o $binary https://github.com/simplex-chat/simplexmq/releases/latest/dow
 chmod +x $binary
 
 # / Add to PATH
-cat <<EOT >> /etc/profile.d/simplex.sh
+cat > /etc/profile.d/simplex.sh << EOF
 #!/bin/bash
 
 export PATH="$PATH:$bin_dir"
 
-EOT
+EOF
 # Add to PATH /
 
 # Source and test PATH
@@ -94,13 +94,13 @@ fi
 on_login_script="/opt/simplex/on_login.sh"
 
 # / Welcome script
-cat <<EOT >> $on_login_script
+cat > $on_login_script << EOF
 #!/bin/bash
 
 fingerprint=\$1
 server_address=\$2
 
-cat <<EOF
+cat << EOF2
 ********************************************************************************
 
 SMP server address: smp://\$fingerprint@\$server_address
@@ -111,9 +111,9 @@ All ports are BLOCKED except 22 (SSH), 443 (HTTPS), 5223 (SMP server).
 
 ********************************************************************************
 To stop seeing this message delete line - bash /opt/simplex/on_login.sh - from /root/.bashrc
-EOF
+EOF2
 
-EOT
+EOF
 # Welcome script /
 
 chmod +x $on_login_script
@@ -139,23 +139,23 @@ if [[ -n "$API_TOKEN" ]]; then
   curl \
     -s -H "Content-Type: application/json" \
     -H "Authorization: Bearer $API_TOKEN" \
-    -X PUT -d "{\"tags\":[\"$server_address\",\"#$fingerprint\",\"$version\"]}" \
+    -X PUT -d "{\"tags\":[\"$server_address\",\"$fingerprint\",\"$version\"]}" \
     https://api.linode.com/v4/linode/instances/$LINODE_ID
 fi
 
 # / Create systemd service
-cat <<EOT >> /etc/systemd/system/smp-server.service
+cat > /etc/systemd/system/smp-server.service << EOF
 [Unit]
-Description=SMP server systemd service
+Description=SMP server
 
 [Service]
 Type=simple
-ExecStart=/bin/sh -c "$binary start"
+ExecStart=/bin/sh -c "exec $binary start >> /var/opt/simplex/smp-server.log 2>&1"
 
 [Install]
 WantedBy=multi-user.target
 
-EOT
+EOF
 # Create systemd service /
 
 # Start systemd service

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        1.0.2
+version:        1.0.3
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -35,6 +35,7 @@ library
       Simplex.Messaging.Agent.Protocol
       Simplex.Messaging.Agent.QueryString
       Simplex.Messaging.Agent.RetryInterval
+      Simplex.Messaging.Agent.Server
       Simplex.Messaging.Agent.Store
       Simplex.Messaging.Agent.Store.SQLite
       Simplex.Messaging.Agent.Store.SQLite.Migrations
@@ -54,6 +55,8 @@ library
       Simplex.Messaging.Server.QueueStore.STM
       Simplex.Messaging.Server.StoreLog
       Simplex.Messaging.Transport
+      Simplex.Messaging.Transport.Client
+      Simplex.Messaging.Transport.Server
       Simplex.Messaging.Transport.WebSockets
       Simplex.Messaging.Util
       Simplex.Messaging.Version
@@ -64,12 +67,12 @@ library
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -115,12 +118,12 @@ executable smp-agent
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -167,12 +170,12 @@ executable smp-server
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -235,12 +238,12 @@ test-suite smp-server-test
   build-depends:
       HUnit ==1.6.*
     , QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*

src/Simplex/Messaging/Agent.hs

@@ -26,11 +26,7 @@
 --
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md
 module Simplex.Messaging.Agent
-  ( -- * SMP agent over TCP
-    runSMPAgent,
-    runSMPAgentBlocking,
-
-    -- * queue-based SMP agent
+  ( -- * queue-based SMP agent
     getAgentClient,
     runAgentClient,
 
@@ -51,6 +47,7 @@ module Simplex.Messaging.Agent
     ackMessage,
     suspendConnection,
     deleteConnection,
+    logConnection,
   )
 where
 
@@ -62,7 +59,6 @@ import Control.Monad.Reader
 import Crypto.Random (MonadRandom)
 import Data.Bifunctor (first, second)
 import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
 import Data.Composition ((.:), (.:.))
 import Data.Functor (($>))
 import Data.List.NonEmpty (NonEmpty (..))
@@ -70,7 +66,6 @@ import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
 import Data.Maybe (isJust)
 import qualified Data.Text as T
-import Data.Text.Encoding (decodeUtf8)
 import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
 import Database.SQLite.Simple (SQLError)
@@ -87,7 +82,6 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (parse)
 import Simplex.Messaging.Protocol (MsgBody)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), loadTLSServerParams, runTransportServer, simplexMQVersion)
 import Simplex.Messaging.Util (bshow, liftError, tryError, unlessM)
 import Simplex.Messaging.Version
 import System.Random (randomR)
@@ -95,33 +89,6 @@ import UnliftIO.Async (async, race_)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
--- | Runs an SMP agent as a TCP service using passed configuration.
---
--- 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 -> m ()
-runSMPAgent t cfg = do
-  started <- newEmptyTMVarIO
-  runSMPAgentBlocking t started cfg
-
--- | 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 -> TMVar Bool -> AgentConfig -> m ()
-runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} = do
-  runReaderT (smpAgent t) =<< newSMPAgentEnv cfg
-  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
-      runTransportServer started tcpPort tlsServerParams $ \(h :: c) -> do
-        liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
-        c <- getAgentClient
-        logConnection c True
-        race_ (connectClient h c) (runAgentClient c)
-          `E.finally` disconnectAgentClient c
-
 -- | Creates an SMP agent client instance
 getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> m AgentClient
 getSMPAgentClient cfg = newSMPAgentEnv cfg >>= runReaderT runAgent
@@ -186,9 +153,6 @@ withAgentEnv c = (`runReaderT` agentEnv c)
 getAgentClient :: (MonadUnliftIO m, MonadReader Env m) => m AgentClient
 getAgentClient = ask >>= atomically . newAgentClient
 
-connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m ()
-connectClient h c = race_ (send h c) (receive h c)
-
 logConnection :: MonadUnliftIO m => AgentClient -> Bool -> m ()
 logConnection c connected =
   let event = if connected then "connected to" else "disconnected from"
@@ -198,28 +162,6 @@ logConnection c connected =
 runAgentClient :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
 runAgentClient c = race_ (subscriber c) (client c)
 
-receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
-receive h c@AgentClient {rcvQ, subQ} = forever $ do
-  (corrId, connId, cmdOrErr) <- tGet SClient h
-  case cmdOrErr of
-    Right cmd -> write rcvQ (corrId, connId, cmd)
-    Left e -> write subQ (corrId, connId, ERR e)
-  where
-    write :: TBQueue (ATransmission p) -> ATransmission p -> m ()
-    write q t = do
-      logClient c "-->" t
-      atomically $ writeTBQueue q t
-
-send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
-send h c@AgentClient {subQ} = forever $ do
-  t <- atomically $ readTBQueue subQ
-  tPut h t
-  logClient c "<--" t
-
-logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()
-logClient AgentClient {clientId} dir (corrId, connId, cmd) = do
-  logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]
-
 client :: forall m. (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
 client c@AgentClient {rcvQ, subQ} = forever $ do
   (corrId, connId, cmd) <- atomically $ readTBQueue rcvQ
@@ -363,16 +305,6 @@ subscribeConnection' c connId =
     SomeConn _ (DuplexConnection _ rq sq) -> do
       resumeMsgDelivery c connId sq
       subscribeQueue c rq connId
-      case status (sq :: SndQueue) of
-        Confirmed -> do
-          -- TODO if there is no confirmation saved, just update the status without securing the queue
-          AcceptedConfirmation {senderConf = SMPConfirmation {senderKey}} <-
-            withStore (`getAcceptedConfirmation` connId)
-          secureQueue c rq senderKey
-          withStore $ \st -> setRcvQueueStatus st rq Secured
-        Secured -> pure ()
-        Active -> pure ()
-        _ -> throwError $ INTERNAL "unexpected queue status"
     SomeConn _ (SndConnection _ sq) -> do
       resumeMsgDelivery c connId sq
       case status (sq :: SndQueue) of

src/Simplex/Messaging/Agent/Client.hs

@@ -37,7 +37,7 @@ module Simplex.Messaging.Agent.Client
 where
 
 import Control.Concurrent (forkIO)
-import Control.Concurrent.Async (Async, async, uninterruptibleCancel)
+import Control.Concurrent.Async (Async, uninterruptibleCancel)
 import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple
 import Control.Monad.Except
@@ -62,8 +62,10 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol (QueueId, QueueIdsKeys (..), SndPublicVerifyKey)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Util (bshow, liftEitherError, liftError, liftIOEither, tryError)
+import Simplex.Messaging.Util (bshow, liftEitherError, liftError, tryError)
 import Simplex.Messaging.Version
+import System.Timeout (timeout)
+import UnliftIO (async)
 import UnliftIO.Exception (Exception, IOException)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -76,11 +78,13 @@ data AgentClient = AgentClient
     msgQ :: TBQueue SMPServerTransmission,
     smpClients :: TVar (Map SMPServer SMPClientVar),
     subscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
+    pendingSubscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
     subscrConns :: TVar (Map ConnId SMPServer),
     connMsgsQueued :: TVar (Map ConnId Bool),
     smpQueueMsgQueues :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId)),
     smpQueueMsgDeliveries :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (Async ())),
     reconnections :: TVar [Async ()],
+    asyncClients :: TVar [Async ()],
     clientId :: Int,
     agentEnv :: Env,
     smpSubscriber :: Async (),
@@ -95,14 +99,16 @@ newAgentClient agentEnv = do
   msgQ <- newTBQueue qSize
   smpClients <- newTVar M.empty
   subscrSrvrs <- newTVar M.empty
+  pendingSubscrSrvrs <- newTVar M.empty
   subscrConns <- newTVar M.empty
   connMsgsQueued <- newTVar M.empty
   smpQueueMsgQueues <- newTVar M.empty
   smpQueueMsgDeliveries <- newTVar M.empty
   reconnections <- newTVar []
+  asyncClients <- newTVar []
   clientId <- stateTVar (clientCounter agentEnv) $ \i -> (i + 1, i + 1)
   lock <- newTMVar ()
-  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock}
+  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, pendingSubscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, asyncClients, clientId, agentEnv, smpSubscriber = undefined, lock}
 
 -- | Agent monad with MonadReader Env and MonadError AgentErrorType
 type AgentMonad m = (MonadUnliftIO m, MonadReader Env m, MonadError AgentErrorType m)
@@ -133,20 +139,39 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
       pure smpVar
 
     waitForSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
-    waitForSMPClient = liftIOEither . atomically . readTMVar
+    waitForSMPClient smpVar = do
+      SMPClientConfig {tcpTimeout} <- asks $ smpCfg . config
+      smpClient_ <- liftIO $ tcpTimeout `timeout` atomically (readTMVar smpVar)
+      liftEither $ case smpClient_ of
+        Just (Right smpClient) -> Right smpClient
+        Just (Left e) -> Left e
+        Nothing -> Left $ BROKER TIMEOUT
 
     newSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
-    newSMPClient smpVar =
-      tryError connectClient >>= \r -> case r of
-        Right smp -> do
-          logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
-          atomically $ putTMVar smpVar r
-          pure smp
-        Left e -> do
-          atomically $ do
-            putTMVar smpVar r
-            modifyTVar smpClients $ M.delete srv
-          throwError e
+    newSMPClient smpVar = tryConnectClient pure tryConnectAsync
+      where
+        tryConnectClient :: (SMPClient -> m a) -> m () -> m a
+        tryConnectClient successAction retryAction =
+          tryError connectClient >>= \r -> case r of
+            Right smp -> do
+              logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
+              atomically $ putTMVar smpVar r
+              successAction smp
+            Left e -> do
+              if e == BROKER NETWORK || e == BROKER TIMEOUT
+                then retryAction
+                else atomically $ do
+                  putTMVar smpVar (Left e)
+                  modifyTVar smpClients $ M.delete srv
+              throwError e
+        tryConnectAsync :: m ()
+        tryConnectAsync = do
+          a <- async connectAsync
+          atomically $ modifyTVar (asyncClients c) (a :)
+        connectAsync :: m ()
+        connectAsync = do
+          ri <- asks $ reconnectInterval . config
+          withRetryInterval ri $ \loop -> void $ tryConnectClient (const reconnectClient) loop
 
     connectClient :: m SMPClient
     connectClient = do
@@ -169,33 +194,43 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
       cs <- M.lookup srv <$> readTVar (subscrSrvrs c)
       modifyTVar (subscrSrvrs c) $ M.delete srv
       modifyTVar (subscrConns c) $ maybe id (deleteKeys . M.keysSet) cs
+      mapM_ (modifyTVar (pendingSubscrSrvrs c) . addPendingSubs) cs
       return cs
       where
+        addPendingSubs :: Map ConnId RcvQueue -> Map SMPServer (Map ConnId RcvQueue) -> Map SMPServer (Map ConnId RcvQueue)
+        addPendingSubs cs = M.alter (Just . addSubs cs) srv
+        addSubs cs = maybe cs (M.union cs)
         deleteKeys :: Ord k => Set k -> Map k a -> Map k a
         deleteKeys ks m = S.foldr' M.delete m ks
 
     serverDown :: UnliftIO m -> Map ConnId RcvQueue -> IO ()
     serverDown u cs = unless (M.null cs) $ do
       mapM_ (notifySub DOWN) $ M.keysSet cs
-      a <- async . unliftIO u $ tryReconnectClient cs
+      unliftIO u reconnectServer
+
+    reconnectServer :: m ()
+    reconnectServer = do
+      a <- async tryReconnectClient
       atomically $ modifyTVar (reconnections c) (a :)
 
-    tryReconnectClient :: Map ConnId RcvQueue -> m ()
-    tryReconnectClient cs = do
+    tryReconnectClient :: m ()
+    tryReconnectClient = do
       ri <- asks $ reconnectInterval . config
       withRetryInterval ri $ \loop ->
-        reconnectClient cs `catchError` const loop
+        reconnectClient `catchError` const loop
 
-    reconnectClient :: Map ConnId RcvQueue -> m ()
-    reconnectClient cs = do
+    reconnectClient :: m ()
+    reconnectClient = do
       withAgentLock c . withSMP c srv $ \smp -> do
         subs <- readTVarIO $ subscrConns c
-        forM_ (M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->
+        cs <- M.lookup srv <$> readTVarIO (pendingSubscrSrvrs c)
+        forM_ (maybe [] M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->
           when (isNothing $ M.lookup connId subs) $ do
             subscribeSMPQueue smp rcvPrivateKey rcvId
               `catchError` \case
-                SMPServerError e -> liftIO $ notifySub (ERR $ SMP e) connId
-                e -> throwError e
+                e@SMPResponseTimeout -> throwError e
+                e@SMPNetworkError -> throwError e
+                e -> liftIO $ notifySub (ERR $ smpClientError e) connId
             addSubscription c rq connId
             liftIO $ notifySub UP connId
 
@@ -206,6 +241,7 @@ closeAgentClient :: MonadUnliftIO m => AgentClient -> m ()
 closeAgentClient c = liftIO $ do
   closeSMPServerClients c
   cancelActions $ reconnections c
+  cancelActions $ asyncClients c
   cancelActions $ smpQueueMsgDeliveries c
 
 closeSMPServerClients :: AgentClient -> IO ()
@@ -294,26 +330,40 @@ newRcvQueue_ a c srv = do
 
 subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()
 subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
-  withLogSMP c server rcvId "SUB" $ \smp ->
-    subscribeSMPQueue smp rcvPrivateKey rcvId
-  addSubscription c rq connId
+  addPendingSubscription c rq connId
+  withLogSMP c server rcvId "SUB" $ \smp -> do
+    liftIO (runExceptT $ subscribeSMPQueue smp rcvPrivateKey rcvId) >>= \case
+      Left e -> do
+        atomically . when (e /= SMPNetworkError && e /= SMPResponseTimeout) $
+          removePendingSubscription c server connId
+        throwError e
+      Right _ -> addSubscription c rq connId
 
 addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
 addSubscription c rq@RcvQueue {server} connId = atomically $ do
   modifyTVar (subscrConns c) $ M.insert connId server
-  modifyTVar (subscrSrvrs c) $ M.alter (Just . addSub) server
-  where
-    addSub :: Maybe (Map ConnId RcvQueue) -> Map ConnId RcvQueue
-    addSub (Just cs) = M.insert connId rq cs
-    addSub _ = M.singleton connId rq
+  addSubs_ (subscrSrvrs c) rq connId
+  removePendingSubscription c server connId
 
-removeSubscription :: AgentMonad m => AgentClient -> ConnId -> m ()
-removeSubscription AgentClient {subscrConns, subscrSrvrs} connId = atomically $ do
-  cs <- readTVar subscrConns
-  writeTVar subscrConns $ M.delete connId cs
-  mapM_
-    (modifyTVar subscrSrvrs . M.alter (>>= delSub))
-    (M.lookup connId cs)
+addPendingSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
+addPendingSubscription c rq connId =
+  atomically $ addSubs_ (pendingSubscrSrvrs c) rq connId
+
+addSubs_ :: TVar (Map SMPServer (Map ConnId RcvQueue)) -> RcvQueue -> ConnId -> STM ()
+addSubs_ ss rq@RcvQueue {server} connId = modifyTVar ss $ M.alter (Just . addSub) server
+  where
+    addSub = maybe (M.singleton connId rq) (M.insert connId rq)
+
+removeSubscription :: MonadUnliftIO m => AgentClient -> ConnId -> m ()
+removeSubscription c@AgentClient {subscrConns} connId = atomically $ do
+  server_ <- stateTVar subscrConns $ \cs -> (M.lookup connId cs, M.delete connId cs)
+  mapM_ (\server -> removeSubs_ (subscrSrvrs c) server connId) server_
+
+removePendingSubscription :: AgentClient -> SMPServer -> ConnId -> STM ()
+removePendingSubscription c = removeSubs_ (pendingSubscrSrvrs c)
+
+removeSubs_ :: TVar (Map SMPServer (Map ConnId RcvQueue)) -> SMPServer -> ConnId -> STM ()
+removeSubs_ ss server connId = modifyTVar ss $ M.update delSub server
   where
     delSub :: Map ConnId RcvQueue -> Maybe (Map ConnId RcvQueue)
     delSub cs =
@@ -370,7 +420,6 @@ deleteQueue c RcvQueue {server, rcvId, rcvPrivateKey} =
   withLogSMP c server rcvId "DEL" $ \smp ->
     deleteSMPQueue smp rcvPrivateKey rcvId
 
--- TODO this is just wrong
 sendAgentMessage :: forall m. AgentMonad m => AgentClient -> SndQueue -> ByteString -> m ()
 sendAgentMessage c sq@SndQueue {server, sndId, sndPrivateKey} agentMsg =
   withLogSMP_ c server sndId "SEND <MSG>" $ \smp -> do

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

@@ -35,6 +35,7 @@ data AgentConfig = AgentConfig
     tbqSize :: Natural,
     dbFile :: FilePath,
     dbPoolSize :: Int,
+    yesToMigrations :: Bool,
     smpCfg :: SMPClientConfig,
     reconnectInterval :: RetryInterval,
     helloTimeout :: NominalDiffTime,
@@ -50,9 +51,10 @@ defaultAgentConfig =
       smpServers = undefined, -- TODO move it elsewhere?
       cmdSignAlg = C.SignAlg C.SEd448,
       connIdBytes = 12,
-      tbqSize = 16,
+      tbqSize = 64,
       dbFile = "smp-agent.db",
       dbPoolSize = 4,
+      yesToMigrations = False,
       smpCfg = smpDefaultConfig,
       reconnectInterval =
         RetryInterval
@@ -60,7 +62,7 @@ defaultAgentConfig =
             increaseAfter = 10 * second,
             maxInterval = 10 * second
           },
-      helloTimeout = 7 * nominalDay,
+      helloTimeout = 2 * nominalDay,
       -- CA certificate private key is not needed for initialization
       -- ! we do not generate these
       caCertificateFile = "/etc/opt/simplex-agent/ca.crt",
@@ -79,9 +81,9 @@ data Env = Env
   }
 
 newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env
-newSMPAgentEnv cfg = do
+newSMPAgentEnv config@AgentConfig {dbFile, dbPoolSize, yesToMigrations} = do
   idsDrg <- newTVarIO =<< drgNew
-  store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app
+  store <- liftIO $ createSQLiteStore dbFile dbPoolSize Migrations.app yesToMigrations
   clientCounter <- newTVarIO 0
   randomServer <- newTVarIO =<< liftIO newStdGen
-  return Env {config = cfg, store, idsDrg, clientCounter, randomServer}
+  return Env {config, store, idsDrg, clientCounter, randomServer}

src/Simplex/Messaging/Agent/Protocol.hs

@@ -83,16 +83,10 @@ module Simplex.Messaging.Agent.Protocol
 
     -- * Encode/decode
     serializeCommand,
-    serializeMsgIntegrity,
     connMode,
     connMode',
-    serializeAgentError,
-    serializeSmpErrorType,
     commandP,
     connModeT,
-    msgIntegrityP,
-    agentErrorTypeP,
-    smpErrorTypeP,
     serializeQueueStatus,
     queueStatusT,
     aMessageType,
@@ -108,6 +102,7 @@ where
 import Control.Applicative (optional, (<|>))
 import Control.Monad.IO.Class
 import Data.Aeson (FromJSON (..), ToJSON (..))
+import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Base64
@@ -131,7 +126,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (E2ERatchetParams, E2ERatchetParamsUri)
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Parsers (base64P, parse, parseRead, parseRead1, parseRead2, tsISO8601P)
+import Simplex.Messaging.Parsers
 import Simplex.Messaging.Protocol
   ( ErrorType,
     MsgBody,
@@ -305,7 +300,7 @@ data AgentMsgEnvelope
       }
   | AgentInvitation -- the connInfo in contactInvite is only encrypted with per-queue E2E, not with double ratchet,
       { agentVersion :: Version,
-        connReq :: (ConnectionRequestUri 'CMInvitation),
+        connReq :: ConnectionRequestUri 'CMInvitation,
         connInfo :: ByteString -- this message is only encrypted with per-queue E2E, not with double ratchet,
       }
   deriving (Show)
@@ -618,29 +613,70 @@ queueStatusT = \case
 type AgentMsgId = Int64
 
 -- | Result of received message integrity validation.
-data MsgIntegrity = MsgOk | MsgError MsgErrorType
-  deriving (Eq, Show)
+data MsgIntegrity = MsgOk | MsgError {errorInfo :: MsgErrorType}
+  deriving (Eq, Show, Generic)
+
+instance StrEncoding MsgIntegrity where
+  strP = "OK" $> MsgOk <|> "ERR " *> (MsgError <$> strP)
+  strEncode = \case
+    MsgOk -> "OK"
+    MsgError e -> "ERR" <> strEncode e
+
+instance ToJSON MsgIntegrity where
+  toJSON = J.genericToJSON $ sumTypeJSON fstToLower
+  toEncoding = J.genericToEncoding $ sumTypeJSON fstToLower
+
+instance FromJSON MsgIntegrity where
+  parseJSON = J.genericParseJSON $ sumTypeJSON fstToLower
 
 -- | Error of message integrity validation.
-data MsgErrorType = MsgSkipped AgentMsgId AgentMsgId | MsgBadId AgentMsgId | MsgBadHash | MsgDuplicate
-  deriving (Eq, Show)
+data MsgErrorType
+  = MsgSkipped {fromMsgId :: AgentMsgId, toMsgId :: AgentMsgId}
+  | MsgBadId {msgId :: AgentMsgId}
+  | MsgBadHash
+  | MsgDuplicate
+  deriving (Eq, Show, Generic)
+
+instance StrEncoding MsgErrorType where
+  strP =
+    "ID " *> (MsgBadId <$> A.decimal)
+      <|> "IDS " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)
+      <|> "HASH" $> MsgBadHash
+      <|> "DUPLICATE" $> MsgDuplicate
+  strEncode = \case
+    MsgSkipped fromMsgId toMsgId ->
+      B.unwords ["NO_ID", bshow fromMsgId, bshow toMsgId]
+    MsgBadId aMsgId -> "ID " <> bshow aMsgId
+    MsgBadHash -> "HASH"
+    MsgDuplicate -> "DUPLICATE"
+
+instance ToJSON MsgErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON fstToLower
+  toEncoding = J.genericToEncoding $ sumTypeJSON fstToLower
+
+instance FromJSON MsgErrorType where
+  parseJSON = J.genericParseJSON $ sumTypeJSON fstToLower
 
 -- | Error type used in errors sent to agent clients.
 data AgentErrorType
   = -- | command or response error
-    CMD CommandErrorType
+    CMD {cmdErr :: CommandErrorType}
   | -- | connection errors
-    CONN ConnectionErrorType
+    CONN {connErr :: ConnectionErrorType}
   | -- | SMP protocol errors forwarded to agent clients
-    SMP ErrorType
+    SMP {smpErr :: ErrorType}
   | -- | SMP server errors
-    BROKER BrokerErrorType
+    BROKER {brokerErr :: BrokerErrorType}
   | -- | errors of other agents
-    AGENT SMPAgentError
+    AGENT {agentErr :: SMPAgentError}
   | -- | agent implementation or dependency errors
-    INTERNAL String
+    INTERNAL {internalErr :: String}
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON AgentErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 -- | SMP agent protocol command or response error.
 data CommandErrorType
   = -- | command is prohibited in this context
@@ -655,6 +691,10 @@ data CommandErrorType
     LARGE
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON CommandErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 -- | Connection error.
 data ConnectionErrorType
   = -- | connection is not in the database
@@ -669,20 +709,28 @@ data ConnectionErrorType
     NOT_AVAILABLE
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON ConnectionErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 -- | SMP server errors.
 data BrokerErrorType
   = -- | invalid server response (failed to parse)
-    RESPONSE ErrorType
+    RESPONSE {smpErr :: ErrorType}
   | -- | unexpected response
     UNEXPECTED
   | -- | network error
     NETWORK
   | -- | handshake or other transport error
-    TRANSPORT TransportError
+    TRANSPORT {transportErr :: TransportError}
   | -- | command response timeout
     TIMEOUT
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON BrokerErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 -- | Errors of another SMP agent.
 -- TODO encode/decode without A prefix
 data SMPAgentError
@@ -696,6 +744,30 @@ data SMPAgentError
     A_ENCRYPTION
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON SMPAgentError where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
+instance StrEncoding AgentErrorType where
+  strP =
+    "CMD " *> (CMD <$> parseRead1)
+      <|> "CONN " *> (CONN <$> parseRead1)
+      <|> "SMP " *> (SMP <$> strP)
+      <|> "BROKER RESPONSE " *> (BROKER . RESPONSE <$> strP)
+      <|> "BROKER TRANSPORT " *> (BROKER . TRANSPORT <$> transportErrorP)
+      <|> "BROKER " *> (BROKER <$> parseRead1)
+      <|> "AGENT " *> (AGENT <$> parseRead1)
+      <|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)
+  strEncode = \case
+    CMD e -> "CMD " <> bshow e
+    CONN e -> "CONN " <> bshow e
+    SMP e -> "SMP " <> strEncode e
+    BROKER (RESPONSE e) -> "BROKER RESPONSE " <> strEncode e
+    BROKER (TRANSPORT e) -> "BROKER TRANSPORT " <> serializeTransportError e
+    BROKER e -> "BROKER " <> bshow e
+    AGENT e -> "AGENT " <> bshow e
+    INTERNAL e -> "INTERNAL " <> bshow e
+
 instance Arbitrary AgentErrorType where arbitrary = genericArbitraryU
 
 instance Arbitrary CommandErrorType where arbitrary = genericArbitraryU
@@ -746,27 +818,17 @@ commandP =
     sendCmd = ACmd SClient . SEND <$> A.takeByteString
     msgIdResp = ACmd SAgent . MID <$> A.decimal
     sentResp = ACmd SAgent . SENT <$> A.decimal
-    msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> agentErrorTypeP
+    msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> strP
     message = ACmd SAgent .: MSG <$> msgMetaP <* A.space <*> A.takeByteString
     ackCmd = ACmd SClient . ACK <$> A.decimal
     msgMetaP = do
-      integrity <- msgIntegrityP
+      integrity <- strP
       recipient <- " R=" *> partyMeta A.decimal
       broker <- " B=" *> partyMeta base64P
       sndMsgId <- " S=" *> A.decimal
       pure MsgMeta {integrity, recipient, broker, sndMsgId}
     partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P
-    agentError = ACmd SAgent . ERR <$> agentErrorTypeP
-
--- | Message integrity validation result parser.
-msgIntegrityP :: Parser MsgIntegrity
-msgIntegrityP = "OK" $> MsgOk <|> "ERR " *> (MsgError <$> msgErrorType)
-  where
-    msgErrorType =
-      "ID " *> (MsgBadId <$> A.decimal)
-        <|> "IDS " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)
-        <|> "HASH" $> MsgBadHash
-        <|> "DUPLICATE" $> MsgDuplicate
+    agentError = ACmd SAgent . ERR <$> strP
 
 parseCommand :: ByteString -> Either AgentErrorType ACmd
 parseCommand = parse commandP $ CMD SYNTAX
@@ -790,13 +852,13 @@ serializeCommand = \case
   SEND msgBody -> "SEND " <> serializeBinary msgBody
   MID mId -> "MID " <> bshow mId
   SENT mId -> "SENT " <> bshow mId
-  MERR mId e -> B.unwords ["MERR", bshow mId, serializeAgentError e]
+  MERR mId e -> B.unwords ["MERR", bshow mId, strEncode e]
   MSG msgMeta msgBody -> B.unwords ["MSG", serializeMsgMeta msgMeta, serializeBinary msgBody]
   ACK mId -> "ACK " <> bshow mId
   OFF -> "OFF"
   DEL -> "DEL"
   CON -> "CON"
-  ERR e -> "ERR " <> serializeAgentError e
+  ERR e -> "ERR " <> strEncode e
   OK -> "OK"
   where
     showTs :: UTCTime -> ByteString
@@ -804,49 +866,12 @@ serializeCommand = \case
     serializeMsgMeta :: MsgMeta -> ByteString
     serializeMsgMeta MsgMeta {integrity, recipient = (rmId, rTs), broker = (bmId, bTs), sndMsgId} =
       B.unwords
-        [ serializeMsgIntegrity integrity,
+        [ strEncode integrity,
           "R=" <> bshow rmId <> "," <> showTs rTs,
           "B=" <> encode bmId <> "," <> showTs bTs,
           "S=" <> bshow sndMsgId
         ]
 
--- | Serialize message integrity validation result.
-serializeMsgIntegrity :: MsgIntegrity -> ByteString
-serializeMsgIntegrity = \case
-  MsgOk -> "OK"
-  MsgError e ->
-    "ERR " <> case e of
-      MsgSkipped fromMsgId toMsgId ->
-        B.unwords ["NO_ID", bshow fromMsgId, bshow toMsgId]
-      MsgBadId aMsgId -> "ID " <> bshow aMsgId
-      MsgBadHash -> "HASH"
-      MsgDuplicate -> "DUPLICATE"
-
--- | SMP agent protocol error parser.
-agentErrorTypeP :: Parser AgentErrorType
-agentErrorTypeP =
-  "SMP " *> (SMP <$> smpErrorTypeP)
-    <|> "BROKER RESPONSE " *> (BROKER . RESPONSE <$> smpErrorTypeP)
-    <|> "BROKER TRANSPORT " *> (BROKER . TRANSPORT <$> transportErrorP)
-    <|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)
-    <|> parseRead2
-
--- | Serialize SMP agent protocol error.
-serializeAgentError :: AgentErrorType -> ByteString
-serializeAgentError = \case
-  SMP e -> "SMP " <> serializeSmpErrorType e
-  BROKER (RESPONSE e) -> "BROKER RESPONSE " <> serializeSmpErrorType e
-  BROKER (TRANSPORT e) -> "BROKER TRANSPORT " <> serializeTransportError e
-  e -> bshow e
-
--- | SMP error parser.
-smpErrorTypeP :: Parser ErrorType
-smpErrorTypeP = "CMD " *> (SMP.CMD <$> parseRead1) <|> parseRead1
-
--- | Serialize SMP error.
-serializeSmpErrorType :: ErrorType -> ByteString
-serializeSmpErrorType = bshow
-
 serializeBinary :: ByteString -> ByteString
 serializeBinary body = bshow (B.length body) <> "\n" <> body
 

src/Simplex/Messaging/Agent/Server.hs

@@ -0,0 +1,81 @@
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Agent.Server
+  ( -- * SMP agent over TCP
+    runSMPAgent,
+    runSMPAgentBlocking,
+  )
+where
+
+import Control.Logger.Simple (logInfo)
+import Control.Monad.Except
+import Control.Monad.IO.Unlift (MonadUnliftIO)
+import Control.Monad.Reader
+import Crypto.Random (MonadRandom)
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import Data.Text.Encoding (decodeUtf8)
+import Simplex.Messaging.Agent
+import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Agent.Protocol
+import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), simplexMQVersion)
+import Simplex.Messaging.Transport.Server (loadTLSServerParams, runTransportServer)
+import Simplex.Messaging.Util (bshow)
+import UnliftIO.Async (race_)
+import qualified UnliftIO.Exception as E
+import UnliftIO.STM
+
+-- | Runs an SMP agent as a TCP service using passed configuration.
+--
+-- 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 -> m ()
+runSMPAgent t cfg = do
+  started <- newEmptyTMVarIO
+  runSMPAgentBlocking t started cfg
+
+-- | 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 -> TMVar Bool -> AgentConfig -> m ()
+runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} = do
+  runReaderT (smpAgent t) =<< newSMPAgentEnv cfg
+  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
+      runTransportServer started tcpPort tlsServerParams $ \(h :: c) -> do
+        liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
+        c <- getAgentClient
+        logConnection c True
+        race_ (connectClient h c) (runAgentClient c)
+          `E.finally` disconnectAgentClient c
+
+connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m ()
+connectClient h c = race_ (send h c) (receive h c)
+
+receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
+receive h c@AgentClient {rcvQ, subQ} = forever $ do
+  (corrId, connId, cmdOrErr) <- tGet SClient h
+  case cmdOrErr of
+    Right cmd -> write rcvQ (corrId, connId, cmd)
+    Left e -> write subQ (corrId, connId, ERR e)
+  where
+    write :: TBQueue (ATransmission p) -> ATransmission p -> m ()
+    write q t = do
+      logClient c "-->" t
+      atomically $ writeTBQueue q t
+
+send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
+send h c@AgentClient {subQ} = forever $ do
+  t <- atomically $ readTBQueue subQ
+  tPut h t
+  logClient c "<--" t
+
+logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()
+logClient AgentClient {clientId} dir (corrId, connId, cmd) = do
+  logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]

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

@@ -76,13 +76,13 @@ data SQLiteStore = SQLiteStore
     dbNew :: Bool
   }
 
-createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore
-createSQLiteStore dbFilePath poolSize migrations = do
+createSQLiteStore :: FilePath -> Int -> [Migration] -> Bool -> IO SQLiteStore
+createSQLiteStore dbFilePath poolSize migrations yesToMigrations = do
   let dbDir = takeDirectory dbFilePath
   createDirectoryIfMissing False dbDir
   st <- connectSQLiteStore dbFilePath poolSize
   checkThreadsafe st
-  migrateSchema st migrations
+  migrateSchema st migrations yesToMigrations
   pure st
 
 checkThreadsafe :: SQLiteStore -> IO ()
@@ -94,15 +94,16 @@ checkThreadsafe st = withConnection st $ \db -> do
     Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found"
     _ -> return ()
 
-migrateSchema :: SQLiteStore -> [Migration] -> IO ()
-migrateSchema st migrations = withConnection st $ \db -> do
+migrateSchema :: SQLiteStore -> [Migration] -> Bool -> IO ()
+migrateSchema st migrations yesToMigrations = withConnection st $ \db -> do
   Migrations.initialize db
   Migrations.get db migrations >>= \case
     Left e -> confirmOrExit $ "Database error: " <> e
     Right [] -> pure ()
     Right ms -> do
       unless (dbNew st) $ do
-        confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
+        unless yesToMigrations $
+          confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
         let f = dbFilePath st
         copyFile f (f <> ".bak")
       Migrations.run db ms
@@ -569,9 +570,9 @@ instance ToField AMsgType where toField = toField . smpEncode
 
 instance FromField AMsgType where fromField = blobFieldParser smpP
 
-instance ToField MsgIntegrity where toField = toField . serializeMsgIntegrity
+instance ToField MsgIntegrity where toField = toField . strEncode
 
-instance FromField MsgIntegrity where fromField = blobFieldParser msgIntegrityP
+instance FromField MsgIntegrity where fromField = blobFieldParser strP
 
 instance ToField SMPQueueUri where toField = toField . strEncode
 

src/Simplex/Messaging/Client.hs

@@ -63,7 +63,8 @@ import Network.Socket (ServiceName)
 import Numeric.Natural
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake, runTransportClient)
+import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake)
+import Simplex.Messaging.Transport.Client (runTransportClient)
 import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (bshow, liftError, raceAny_)
 import System.Timeout (timeout)
@@ -106,7 +107,7 @@ data SMPClientConfig = SMPClientConfig
 smpDefaultConfig :: SMPClientConfig
 smpDefaultConfig =
   SMPClientConfig
-    { qSize = 16,
+    { qSize = 64,
       defaultTransport = ("5223", transport @TLS),
       tcpTimeout = 4_000_000,
       smpPing = 30_000_000

src/Simplex/Messaging/Parsers.hs

@@ -1,16 +1,18 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
 
 module Simplex.Messaging.Parsers where
 
 import Control.Monad.Trans.Except
+import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Char (isAlphaNum)
+import Data.Char (isAlphaNum, toLower)
 import Data.Time.Clock (UTCTime)
 import Data.Time.ISO8601 (parseISO8601)
 import Data.Typeable (Typeable)
@@ -78,3 +80,46 @@ blobFieldDecoder dec = \case
       Right k -> Ok k
       Left e -> returnError ConversionFailed f ("couldn't parse field: " ++ e)
   f -> returnError ConversionFailed f "expecting SQLBlob column type"
+
+fstToLower :: String -> String
+fstToLower "" = ""
+fstToLower (h : t) = toLower h : t
+
+dropPrefix :: String -> String -> String
+dropPrefix pfx s =
+  let (p, rest) = splitAt (length pfx) s
+   in fstToLower $ if p == pfx then rest else s
+
+enumJSON :: (String -> String) -> J.Options
+enumJSON tagModifier =
+  J.defaultOptions
+    { J.constructorTagModifier = tagModifier,
+      J.allNullaryToStringTag = True
+    }
+
+sumTypeJSON :: (String -> String) -> J.Options
+#if defined(darwin_HOST_OS)
+sumTypeJSON = singleFieldJSON
+#else
+sumTypeJSON = taggedObjectJSON
+#endif
+
+taggedObjectJSON :: (String -> String) -> J.Options
+taggedObjectJSON tagModifier =
+  J.defaultOptions
+    { J.sumEncoding = J.TaggedObject "type" "data",
+      J.constructorTagModifier = tagModifier,
+      J.allNullaryToStringTag = False,
+      J.nullaryToObject = True,
+      J.omitNothingFields = True
+    }
+
+singleFieldJSON :: (String -> String) -> J.Options
+singleFieldJSON tagModifier =
+  J.defaultOptions
+    { J.sumEncoding = J.ObjectWithSingleField,
+      J.constructorTagModifier = tagModifier,
+      J.allNullaryToStringTag = False,
+      J.nullaryToObject = True,
+      J.omitNothingFields = True
+    }

src/Simplex/Messaging/Protocol.hs

@@ -90,6 +90,8 @@ where
 
 import Control.Applicative (optional, (<|>))
 import Control.Monad.Except
+import Data.Aeson (ToJSON (..))
+import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Char8 (ByteString)
@@ -107,7 +109,7 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
 import Simplex.Messaging.Transport (THandle (..), Transport, TransportError (..), tGetBlock, tPutBlock)
-import Simplex.Messaging.Util ((<$?>))
+import Simplex.Messaging.Util (bshow, (<$?>))
 import Simplex.Messaging.Version
 import Test.QuickCheck (Arbitrary (..))
 
@@ -413,6 +415,15 @@ newtype CorrId = CorrId {bs :: ByteString} deriving (Eq, Ord, Show)
 instance IsString CorrId where
   fromString = CorrId . fromString
 
+instance StrEncoding CorrId where
+  strEncode (CorrId cId) = strEncode cId
+  strDecode s = CorrId <$> strDecode s
+  strP = CorrId <$> strP
+
+instance ToJSON CorrId where
+  toJSON = strToJSON
+  toEncoding = strToJEncoding
+
 -- | Queue IDs and keys
 data QueueIdsKeys = QIK
   { rcvId :: RecipientId,
@@ -462,7 +473,7 @@ data ErrorType
   | -- | incorrect SMP session ID (TLS Finished message / tls-unique binding RFC5929)
     SESSION
   | -- | SMP command is unknown or has invalid syntax
-    CMD CommandError
+    CMD {cmdErr :: CommandError}
   | -- | command authorization error - bad signature or non-existing SMP queue
     AUTH
   | -- | SMP queue capacity is exceeded on the server
@@ -477,6 +488,16 @@ data ErrorType
     DUPLICATE_ -- TODO remove, not part of SMP protocol
   deriving (Eq, Generic, Read, Show)
 
+instance ToJSON ErrorType where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
+instance StrEncoding ErrorType where
+  strEncode = \case
+    CMD e -> "CMD " <> bshow e
+    e -> bshow e
+  strP = "CMD " *> (CMD <$> parseRead1) <|> parseRead1
+
 -- | SMP command error type.
 data CommandError
   = -- | unknown command
@@ -491,6 +512,10 @@ data CommandError
     NO_QUEUE
   deriving (Eq, Generic, Read, Show)
 
+instance ToJSON CommandError where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 instance Arbitrary ErrorType where arbitrary = genericArbitraryU
 
 instance Arbitrary CommandError where arbitrary = genericArbitraryU

src/Simplex/Messaging/Server.hs

@@ -48,6 +48,7 @@ import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.STM (QueueStore)
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Server
 import Simplex.Messaging.Util
 import UnliftIO.Concurrent
 import UnliftIO.Exception

src/Simplex/Messaging/Server/Env/STM.hs

@@ -21,7 +21,8 @@ import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.QueueStore (QueueRec (..))
 import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Transport (ATransport, loadFingerprint, loadTLSServerParams)
+import Simplex.Messaging.Transport (ATransport)
+import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams)
 import System.IO (IOMode (..))
 import UnliftIO.STM
 

src/Simplex/Messaging/Transport.hs

@@ -36,15 +36,11 @@ module Simplex.Messaging.Transport
     ATransport (..),
     TransportPeer (..),
 
-    -- * Transport over TLS
-    runTransportServer,
-    runTransportClient,
-    loadTLSServerParams,
-    loadFingerprint,
-
     -- * TLS Transport
     TLS (..),
+    connectTLS,
     closeTLS,
+    supportedParameters,
     withTlsUnique,
 
     -- * SMP transport
@@ -64,9 +60,9 @@ where
 
 import Control.Applicative ((<|>))
 import Control.Monad.Except
-import Control.Monad.IO.Unlift
 import Control.Monad.Trans.Except (throwE)
-import qualified Crypto.Store.X509 as SX
+import Data.Aeson (ToJSON)
+import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import Data.Bifunctor (first)
 import Data.Bitraversable (bimapM)
@@ -75,14 +71,7 @@ import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy as BL
 import Data.Default (def)
 import Data.Functor (($>))
-import Data.Set (Set)
-import qualified Data.Set as S
-import qualified Data.X509 as X
-import qualified Data.X509.CertificateStore as XS
-import Data.X509.Validation (Fingerprint (..))
-import qualified Data.X509.Validation as XV
 import GHC.Generics (Generic)
-import GHC.IO.Exception (IOErrorType (..))
 import GHC.IO.Handle.Internals (ioe_EOF)
 import Generic.Random (genericArbitraryU)
 import Network.Socket
@@ -90,14 +79,11 @@ import qualified Network.TLS as T
 import qualified Network.TLS.Extra as TE
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
-import Simplex.Messaging.Parsers (parse, parseRead1)
+import Simplex.Messaging.Parsers (dropPrefix, parse, parseRead1, sumTypeJSON)
 import Simplex.Messaging.Util (bshow)
 import Simplex.Messaging.Version
-import System.Exit (exitFailure)
-import System.IO.Error
 import Test.QuickCheck (Arbitrary (..))
-import UnliftIO.Concurrent
-import UnliftIO.Exception (Exception, IOException)
+import UnliftIO.Exception (Exception)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
@@ -110,7 +96,7 @@ supportedSMPVersions :: VersionRange
 supportedSMPVersions = mkVersionRange 1 1
 
 simplexMQVersion :: String
-simplexMQVersion = "1.0.2"
+simplexMQVersion = "1.0.3"
 
 -- * Transport connection class
 
@@ -154,104 +140,6 @@ data TProxy c = TProxy
 
 data ATransport = forall c. Transport c => ATransport (TProxy c)
 
--- * Transport over TLS
-
--- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.
---
--- All accepted connections are passed to the passed function.
-runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> T.ServerParams -> (c -> m ()) -> m ()
-runTransportServer started port serverParams server = do
-  u <- askUnliftIO
-  liftIO $ do
-    clients <- newTVarIO S.empty
-    E.bracket
-      (startTCPServer started port)
-      (closeServer clients)
-      $ \sock -> forever $ do
-        (connSock, _) <- accept sock
-        tid <- forkIO $ connectClient u connSock `E.catch` \(_ :: E.SomeException) -> pure ()
-        atomically . modifyTVar clients $ S.insert tid
-  where
-    connectClient :: UnliftIO m -> Socket -> IO ()
-    connectClient u connSock =
-      E.bracket
-        (connectTLS serverParams connSock >>= getServerConnection)
-        closeConnection
-        (unliftIO u . server)
-    closeServer :: TVar (Set ThreadId) -> Socket -> IO ()
-    closeServer clients sock = do
-      readTVarIO clients >>= mapM_ killThread
-      close sock
-      void . atomically $ tryPutTMVar started False
-
-startTCPServer :: TMVar Bool -> ServiceName -> IO Socket
-startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
-  where
-    resolve =
-      let hints = defaultHints {addrFlags = [AI_PASSIVE], addrSocketType = Stream}
-       in head <$> getAddrInfo (Just hints) Nothing (Just port)
-    open addr = do
-      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
-      setSocketOption sock ReuseAddr 1
-      withFdSocket sock setCloseOnExecIfNeeded
-      bind sock $ addrAddress addr
-      listen sock 1024
-      return sock
-    setStarted sock = atomically (tryPutTMVar started True) >> pure sock
-
--- | Connect to passed TCP host:port and pass handle to the client.
-runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> C.KeyHash -> (c -> m a) -> m a
-runTransportClient host port keyHash client = do
-  let clientParams = mkTLSClientParams host port keyHash
-  c <- liftIO $ startTCPClient host port clientParams
-  client c `E.finally` liftIO (closeConnection c)
-
-startTCPClient :: forall c. Transport c => HostName -> ServiceName -> T.ClientParams -> IO c
-startTCPClient host port clientParams = withSocketsDo $ resolve >>= tryOpen err
-  where
-    err :: IOException
-    err = mkIOError NoSuchThing "no address" Nothing Nothing
-
-    resolve :: IO [AddrInfo]
-    resolve =
-      let hints = defaultHints {addrSocketType = Stream}
-       in getAddrInfo (Just hints) (Just host) (Just port)
-
-    tryOpen :: IOException -> [AddrInfo] -> IO c
-    tryOpen e [] = E.throwIO e
-    tryOpen _ (addr : as) =
-      E.try (open addr) >>= either (`tryOpen` as) pure
-
-    open :: AddrInfo -> IO c
-    open addr = do
-      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
-      connect sock $ addrAddress addr
-      ctx <- connectTLS clientParams sock
-      getClientConnection ctx
-
-loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams
-loadTLSServerParams caCertificateFile certificateFile privateKeyFile =
-  fromCredential <$> loadServerCredential
-  where
-    loadServerCredential :: IO T.Credential
-    loadServerCredential =
-      T.credentialLoadX509Chain certificateFile [caCertificateFile] privateKeyFile >>= \case
-        Right credential -> pure credential
-        Left _ -> putStrLn "invalid credential" >> exitFailure
-    fromCredential :: T.Credential -> T.ServerParams
-    fromCredential credential =
-      def
-        { T.serverWantClientCert = False,
-          T.serverShared = def {T.sharedCredentials = T.Credentials [credential]},
-          T.serverHooks = def,
-          T.serverSupported = supportedParameters
-        }
-
-loadFingerprint :: FilePath -> IO Fingerprint
-loadFingerprint certificateFile = do
-  (cert : _) <- SX.readSignedObject certificateFile
-  pure $ XV.getFingerprint (cert :: X.SignedExact X.Certificate) X.HashSHA256
-
 -- * TLS Transport
 
 data TLS = TLS
@@ -290,33 +178,6 @@ closeTLS ctx =
   (T.bye ctx >> T.contextClose ctx) -- sometimes socket was closed before 'TLS.bye'
     `E.catch` (\(_ :: E.SomeException) -> pure ()) -- so we catch the 'Broken pipe' error here
 
-mkTLSClientParams :: HostName -> ServiceName -> C.KeyHash -> T.ClientParams
-mkTLSClientParams host port keyHash = do
-  let p = B.pack port
-  (T.defaultParamsClient host p)
-    { T.clientShared = def,
-      T.clientHooks = def {T.onServerCertificate = \_ _ _ -> validateCertificateChain keyHash host p},
-      T.clientSupported = supportedParameters
-    }
-
-validateCertificateChain :: C.KeyHash -> HostName -> ByteString -> X.CertificateChain -> IO [XV.FailedReason]
-validateCertificateChain _ _ _ (X.CertificateChain []) = pure [XV.EmptyChain]
-validateCertificateChain _ _ _ (X.CertificateChain [_]) = pure [XV.EmptyChain]
-validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_, caCert]) =
-  if Fingerprint kh == XV.getFingerprint caCert X.HashSHA256
-    then x509validate
-    else pure [XV.UnknownCA]
-  where
-    x509validate :: IO [XV.FailedReason]
-    x509validate = XV.validate X.HashSHA256 hooks checks certStore cache serviceID cc
-      where
-        hooks = XV.defaultHooks
-        checks = XV.defaultChecks
-        certStore = XS.makeCertificateStore sc
-        cache = XV.exceptionValidationCache [] -- we manually check fingerprint only of the identity certificate (ca.crt)
-        serviceID = (host, port)
-validateCertificateChain _ _ _ _ = pure [XV.AuthorityTooDeep]
-
 supportedParameters :: T.Supported
 supportedParameters =
   def
@@ -428,9 +289,13 @@ data TransportError
   | -- | incorrect session ID
     TEBadSession
   | -- | transport handshake error
-    TEHandshake HandshakeError
+    TEHandshake {handshakeErr :: HandshakeError}
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON TransportError where
+  toJSON = J.genericToJSON . sumTypeJSON $ dropPrefix "TE"
+  toEncoding = J.genericToEncoding . sumTypeJSON $ dropPrefix "TE"
+
 -- | Transport handshake error.
 data HandshakeError
   = -- | parsing error
@@ -441,6 +306,10 @@ data HandshakeError
     IDENTITY
   deriving (Eq, Generic, Read, Show, Exception)
 
+instance ToJSON HandshakeError where
+  toJSON = J.genericToJSON $ sumTypeJSON id
+  toEncoding = J.genericToEncoding $ sumTypeJSON id
+
 instance Arbitrary TransportError where arbitrary = genericArbitraryU
 
 instance Arbitrary HandshakeError where arbitrary = genericArbitraryU

src/Simplex/Messaging/Transport/Client.hs

@@ -0,0 +1,82 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Transport.Client
+  ( runTransportClient,
+    clientHandshake,
+  )
+where
+
+import Control.Monad.Except
+import Control.Monad.IO.Unlift
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import Data.Default (def)
+import qualified Data.X509 as X
+import qualified Data.X509.CertificateStore as XS
+import Data.X509.Validation (Fingerprint (..))
+import qualified Data.X509.Validation as XV
+import GHC.IO.Exception (IOErrorType (..))
+import Network.Socket
+import qualified Network.TLS as T
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Transport
+import System.IO.Error
+import UnliftIO.Exception (IOException)
+import qualified UnliftIO.Exception as E
+
+-- | Connect to passed TCP host:port and pass handle to the client.
+runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> C.KeyHash -> (c -> m a) -> m a
+runTransportClient host port keyHash client = do
+  let clientParams = mkTLSClientParams host port keyHash
+  c <- liftIO $ startTCPClient host port clientParams
+  client c `E.finally` liftIO (closeConnection c)
+
+startTCPClient :: forall c. Transport c => HostName -> ServiceName -> T.ClientParams -> IO c
+startTCPClient host port clientParams = withSocketsDo $ resolve >>= tryOpen err
+  where
+    err :: IOException
+    err = mkIOError NoSuchThing "no address" Nothing Nothing
+
+    resolve :: IO [AddrInfo]
+    resolve =
+      let hints = defaultHints {addrSocketType = Stream}
+       in getAddrInfo (Just hints) (Just host) (Just port)
+
+    tryOpen :: IOException -> [AddrInfo] -> IO c
+    tryOpen e [] = E.throwIO e
+    tryOpen _ (addr : as) =
+      E.try (open addr) >>= either (`tryOpen` as) pure
+
+    open :: AddrInfo -> IO c
+    open addr = do
+      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
+      connect sock $ addrAddress addr
+      ctx <- connectTLS clientParams sock
+      getClientConnection ctx
+
+mkTLSClientParams :: HostName -> ServiceName -> C.KeyHash -> T.ClientParams
+mkTLSClientParams host port keyHash = do
+  let p = B.pack port
+  (T.defaultParamsClient host p)
+    { T.clientShared = def,
+      T.clientHooks = def {T.onServerCertificate = \_ _ _ -> validateCertificateChain keyHash host p},
+      T.clientSupported = supportedParameters
+    }
+
+validateCertificateChain :: C.KeyHash -> HostName -> ByteString -> X.CertificateChain -> IO [XV.FailedReason]
+validateCertificateChain _ _ _ (X.CertificateChain []) = pure [XV.EmptyChain]
+validateCertificateChain _ _ _ (X.CertificateChain [_]) = pure [XV.EmptyChain]
+validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_, caCert]) =
+  if Fingerprint kh == XV.getFingerprint caCert X.HashSHA256
+    then x509validate
+    else pure [XV.UnknownCA]
+  where
+    x509validate :: IO [XV.FailedReason]
+    x509validate = XV.validate X.HashSHA256 hooks checks certStore cache serviceID cc
+      where
+        hooks = XV.defaultHooks
+        checks = XV.defaultChecks
+        certStore = XS.makeCertificateStore sc
+        cache = XV.exceptionValidationCache [] -- we manually check fingerprint only of the identity certificate (ca.crt)
+        serviceID = (host, port)
+validateCertificateChain _ _ _ _ = pure [XV.AuthorityTooDeep]

src/Simplex/Messaging/Transport/Server.hs

@@ -0,0 +1,94 @@
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Transport.Server
+  ( runTransportServer,
+    loadTLSServerParams,
+    loadFingerprint,
+    serverHandshake,
+  )
+where
+
+import Control.Monad.Except
+import Control.Monad.IO.Unlift
+import qualified Crypto.Store.X509 as SX
+import Data.Default (def)
+import Data.Set (Set)
+import qualified Data.Set as S
+import qualified Data.X509 as X
+import Data.X509.Validation (Fingerprint (..))
+import qualified Data.X509.Validation as XV
+import Network.Socket
+import qualified Network.TLS as T
+import Simplex.Messaging.Transport
+import System.Exit (exitFailure)
+import UnliftIO.Concurrent
+import qualified UnliftIO.Exception as E
+import UnliftIO.STM
+
+-- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.
+--
+-- All accepted connections are passed to the passed function.
+runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> T.ServerParams -> (c -> m ()) -> m ()
+runTransportServer started port serverParams server = do
+  u <- askUnliftIO
+  liftIO $ do
+    clients <- newTVarIO S.empty
+    E.bracket
+      (startTCPServer started port)
+      (closeServer clients)
+      $ \sock -> forever $ do
+        (connSock, _) <- accept sock
+        tid <- forkIO $ connectClient u connSock `E.catch` \(_ :: E.SomeException) -> pure ()
+        atomically . modifyTVar clients $ S.insert tid
+  where
+    connectClient :: UnliftIO m -> Socket -> IO ()
+    connectClient u connSock =
+      E.bracket
+        (connectTLS serverParams connSock >>= getServerConnection)
+        closeConnection
+        (unliftIO u . server)
+    closeServer :: TVar (Set ThreadId) -> Socket -> IO ()
+    closeServer clients sock = do
+      readTVarIO clients >>= mapM_ killThread
+      close sock
+      void . atomically $ tryPutTMVar started False
+
+startTCPServer :: TMVar Bool -> ServiceName -> IO Socket
+startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
+  where
+    resolve =
+      let hints = defaultHints {addrFlags = [AI_PASSIVE], addrSocketType = Stream}
+       in head <$> getAddrInfo (Just hints) Nothing (Just port)
+    open addr = do
+      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
+      setSocketOption sock ReuseAddr 1
+      withFdSocket sock setCloseOnExecIfNeeded
+      bind sock $ addrAddress addr
+      listen sock 1024
+      return sock
+    setStarted sock = atomically (tryPutTMVar started True) >> pure sock
+
+loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams
+loadTLSServerParams caCertificateFile certificateFile privateKeyFile =
+  fromCredential <$> loadServerCredential
+  where
+    loadServerCredential :: IO T.Credential
+    loadServerCredential =
+      T.credentialLoadX509Chain certificateFile [caCertificateFile] privateKeyFile >>= \case
+        Right credential -> pure credential
+        Left _ -> putStrLn "invalid credential" >> exitFailure
+    fromCredential :: T.Credential -> T.ServerParams
+    fromCredential credential =
+      def
+        { T.serverWantClientCert = False,
+          T.serverShared = def {T.sharedCredentials = T.Credentials [credential]},
+          T.serverHooks = def,
+          T.serverSupported = supportedParameters
+        }
+
+loadFingerprint :: FilePath -> IO Fingerprint
+loadFingerprint certificateFile = do
+  (cert : _) <- SX.readSignedObject certificateFile
+  pure $ XV.getFingerprint (cert :: X.SignedExact X.Certificate) X.HashSHA256

stack.yaml

@@ -38,6 +38,15 @@ extra-deps:
   - cryptostore-0.2.1.0@sha256:9896e2984f36a1c8790f057fd5ce3da4cbcaf8aa73eb2d9277916886978c5b19,3881
   - simple-logger-0.1.0@sha256:be8ede4bd251a9cac776533bae7fb643369ebd826eb948a9a18df1a8dd252ff8,1079
   - tls-1.5.7@sha256:1cc30253a9696b65a9cafc0317fbf09f7dcea15e3a145ed6c9c0e28c632fa23a,6991
+# below dependancies are to update Aeson to 2.0.3
+  - OneTuple-0.3.1@sha256:a848c096c9d29e82ffdd30a9998aa2931cbccb3a1bc137539d80f6174d31603e,2262
+  - attoparsec-0.14.4@sha256:79584bdada8b730cb5138fca8c35c76fbef75fc1d1e01e6b1d815a5ee9843191,5810
+  - hashable-1.4.0.2@sha256:0cddd0229d1aac305ea0404409c0bbfab81f075817bd74b8b2929eff58333e55,5005
+  - semialign-1.2.0.1@sha256:0e179b4d3a8eff79001d374d6c91917c6221696b9620f0a4d86852fc6a9b9501,2836
+  - text-short-0.1.5@sha256:962c6228555debdc46f758d0317dea16e5240d01419b42966674b08a5c3d8fa6,3498
+  - time-compat-1.9.6.1@sha256:42d8f2e08e965e1718917d54ad69e1d06bd4b87d66c41dc7410f59313dba4ed1,5033
+  - github: simplex-chat/aeson
+    commit: 3eb66f9a68f103b5f1489382aad89f5712a64db7
   # - ../hs-tls/core
   # - github: simplex-chat/hs-tls
   #   commit: f6cc753611f80af300401cfae63846e9d7c40d9e

tests/AgentTests.hs

@@ -62,9 +62,11 @@ agentTests (ATransport t) = do
       smpAgentTest3_1_1 $ testSubscription t
     it "should send notifications to client when server disconnects" $
       smpAgentServerTest $ testSubscrNotification t
-  describe "Message delivery" $ do
+  describe "Message delivery and server reconnection" $ do
     it "should deliver messages after losing server connection and re-connecting" $
       smpAgentTest2_2_2_needs_server $ testMsgDeliveryServerRestart t
+    it "should connect to the server when server goes up if it initially was down" $
+      smpAgentTestN [] $ testServerConnectionAfterError t
     it "should deliver pending messages after agent restarting" $
       smpAgentTest1_1_1 $ testMsgDeliveryAgentRestart t
     it "should concurrently deliver messages to connections without blocking" $
@@ -300,6 +302,43 @@ testMsgDeliveryServerRestart t alice bob = do
   where
     withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
 
+testServerConnectionAfterError :: forall c. Transport c => TProxy c -> [c] -> IO ()
+testServerConnectionAfterError t _ = do
+  withAgent1 $ \bob -> do
+    withAgent2 $ \alice -> do
+      withServer $ do
+        connect (bob, "bob") (alice, "alice")
+
+      bob <# ("", "alice", DOWN)
+      alice <# ("", "bob", DOWN)
+      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 4)
+      alice #:# "nothing else delivered before the server is restarted"
+      bob #:# "nothing else delivered before the server is restarted"
+
+  withAgent1 $ \bob -> do
+    withAgent2 $ \alice -> do
+      bob #: ("1", "alice", "SUB") #> ("1", "alice", ERR (BROKER NETWORK))
+      alice #: ("1", "bob", "SUB") #> ("1", "bob", ERR (BROKER NETWORK))
+      withServer $ do
+        alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 4; _ -> False
+        alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 4; _ -> False
+        bob <# ("", "alice", UP)
+        bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
+        bob #: ("2", "alice", "ACK 4") #> ("2", "alice", OK)
+        alice #: ("1", "bob", "SEND 11\nhello again") #> ("1", "bob", MID 5)
+        alice <# ("", "bob", SENT 5)
+        bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False
+
+  removeFile testStoreLogFile
+  removeFile testDB
+  removeFile testDB2
+  where
+    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
+    withAgent1 = withAgent agentTestPort testDB
+    withAgent2 = withAgent agentTestPort2 testDB2
+    withAgent :: String -> String -> (c -> IO a) -> IO a
+    withAgent agentPort agentDB = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) (pure ()) . const . testSMPAgentClientOn agentPort
+
 testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO ()
 testMsgDeliveryAgentRestart t bob = do
   withAgent $ \alice -> do
@@ -391,10 +430,10 @@ sendMessage (h1, name1) (h2, name2) msg = do
   ("m1", name2', Right (MID mId)) <- h1 #: ("m1", name2, "SEND :" <> msg)
   name2' `shouldBe` name2
   h1 <#= \case ("", n, SENT m) -> n == name2 && m == mId; _ -> False
-  ("", name1', Right (MSG MsgMeta {recipient = (msgId, _)} msg')) <- (h2 <#:)
+  ("", name1', Right (MSG MsgMeta {recipient = (msgId', _)} msg')) <- (h2 <#:)
   name1' `shouldBe` name1
   msg' `shouldBe` msg
-  h2 #: ("m2", name1, "ACK " <> bshow msgId) =#> \case ("m2", n, OK) -> n == name1; _ -> False
+  h2 #: ("m2", name1, "ACK " <> bshow msgId') =#> \case ("m2", n, OK) -> n == name1; _ -> False
 
 -- connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)
 -- connect' h1 h2 = do

tests/AgentTests/SQLiteTests.hs

@@ -51,7 +51,7 @@ createStore = do
   -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
   -- IO operations on multiple similarly named files; error seems to be environment specific
   r <- randomIO :: IO Word32
-  createSQLiteStore (testDB <> show r) 4 Migrations.app
+  createSQLiteStore (testDB <> show r) 4 Migrations.app True
 
 removeStore :: SQLiteStore -> IO ()
 removeStore store = do

tests/CoreTests/ProtocolErrorTests.hs

@@ -1,8 +1,9 @@
 module CoreTests.ProtocolErrorTests where
 
-import Simplex.Messaging.Agent.Protocol (AgentErrorType, agentErrorTypeP, serializeAgentError, serializeSmpErrorType, smpErrorTypeP)
+import Simplex.Messaging.Agent.Protocol (AgentErrorType)
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol (ErrorType)
+import Simplex.Messaging.Encoding.String
 import Test.Hspec
 import Test.Hspec.QuickCheck (modifyMaxSuccess)
 import Test.QuickCheck
@@ -11,8 +12,8 @@ protocolErrorTests :: Spec
 protocolErrorTests = modifyMaxSuccess (const 1000) $ do
   describe "errors parsing / serializing" $ do
     it "should parse SMP protocol errors" . property $ \err ->
-      parseAll smpErrorTypeP (serializeSmpErrorType err)
+      parseAll strP (strEncode err)
         == Right (err :: ErrorType)
     it "should parse SMP agent errors" . property $ \err ->
-      parseAll agentErrorTypeP (serializeAgentError err)
+      parseAll strP (strEncode err)
         == Right (err :: AgentErrorType)

tests/SMPAgentClient.hs

@@ -20,12 +20,13 @@ import SMPClient
     withSmpServerOn,
     withSmpServerThreadOn,
   )
-import Simplex.Messaging.Agent (runSMPAgentBlocking)
 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.Client (SMPClientConfig (..), smpDefaultConfig)
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Client
 import Test.Hspec
 import UnliftIO.Concurrent
 import UnliftIO.Directory

tests/SMPClient.hs

@@ -21,6 +21,7 @@ import Simplex.Messaging.Server (runSMPServerBlocking)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.StoreLog (openReadStoreLog)
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Client
 import Test.Hspec
 import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E