3.3.0 (#540 )

lints (#539 )
* lints * revert * add import * remove StrictData
2026-10-12 02:22:45 +00:00 · 2022-10-01 12:08:54 +01:00 · 2022-10-01 11:51:52 +01:00 · 2022-09-30 12:36:21 +01:00 · 2022-09-29 21:11:36 +04:00 · 2022-09-23 18:45:00 +01:00
40 changed files with 1352 additions and 429 deletions
@@ -5,4 +5,5 @@
 tests/tmp
 dist-newstyle/

-src/tor
+cabal.project.local
+cabal.project.local~
@@ -1,3 +1,15 @@
+# 3.3.0
+
+SMP server:
+
+- allow repeated KEY command with the same key (to avoid failures on retries)
+
+SMP agent:
+
+- enable/disable connection notifications
+- asynchronous commands that retry on network error
+- use SQLCipher
+
 # 3.2.0

 SMP agent:
@@ -108,6 +108,8 @@ You can run your SMP server as a Linux process, optionally using a service manag

 - Run `smp-server start` to start SMP server, or you can configure a service manager to run it as a service.

+- Optionally, `smp-server` can be setup for having an onion address in `tor` network. See: [`scripts/tor`](./scripts/tor/). In this case, the server address can have both public and onion hostname pointing to the same server, to allow two people connect when only one of them is using Tor. The server address would be: `smp://<fingerprint>@<public_hostname>,<onion_hostname>`
+
 See [this section](#smp-server) for more information. Run `smp-server -h` and `smp-server init -h` for explanation of commands and options.

 <img alt="Docker" src="./img/docker.svg" align="right" width="200">
@@ -1,6 +1,17 @@
 packages: .
+-- packages: . ../direct-sqlcipher ../sqlcipher-simple

 source-repository-package
    type: git
    location: https://github.com/simplex-chat/aeson.git
    tag: 3eb66f9a68f103b5f1489382aad89f5712a64db7
+
+source-repository-package
+    type: git
+    location: https://github.com/simplex-chat/direct-sqlcipher.git
+    tag: 34309410eb2069b029b8fc1872deb1e0db123294
+
+source-repository-package
+    type: git
+    location: https://github.com/simplex-chat/sqlcipher-simple.git
+    tag: 5e154a2aeccc33ead6c243ec07195ab673137221
@@ -1,5 +1,5 @@
 name: simplexmq
-version: 3.2.0
+version: 3.3.0
 synopsis: SimpleXMQ message broker
 description: |
  This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -38,7 +38,7 @@ dependencies:
  - cryptonite >= 0.27 && < 0.30
  - cryptostore == 0.2.*
  - data-default == 0.7.*
-  - direct-sqlite == 2.3.*
+  - direct-sqlcipher == 2.3.*
  - directory == 1.3.*
  - filepath == 1.4.*
  - http-types == 0.12.*
@@ -49,14 +49,14 @@ dependencies:
  - memory == 0.15.*
  - mtl == 2.2.*
  - network >= 3.1.2.7 && < 3.2
-  - network-transport == 0.5.*
+  - network-transport == 0.5.4
  - optparse-applicative >= 0.15 && < 0.17
  - QuickCheck == 2.14.*
  - process == 1.6.*
  - random >= 1.1 && < 1.3
  - simple-logger == 0.1.*
  - socks == 0.6.*
-  - sqlite-simple == 0.4.*
+  - sqlcipher-simple == 0.4.*
  - stm == 2.5.*
  - template-haskell == 2.16.*
  - text == 1.2.*
@@ -0,0 +1,37 @@
+1. Install `tor` following [official guide](https://community.torproject.org/onion-services/setup/install/).
+
+2. Modify `/etc/tor/torrc` configuration file:
+
+```sh
+...
+# Disable anonymous mode for better connectivity
+## Needed for HiddenServiceNonAnonymousMode
+SOCKSPort 0
+## Needed for HiddenServiceSingleHopMode
+HiddenServiceNonAnonymousMode 1
+## Flag to disable anonymous mode
+## This option reduces the latency of server connection, but it makes server itself not anonymous,
+## it only protects the anonymity of the users connecting to the server.
+## In case your server address has both public and onion hostnames it is not anonymous anyway,
+## so this is what you want.
+HiddenServiceSingleHopMode 1
+
+# Specify folder for smp-tor
+## Folder for keys, address, etc.
+HiddenServiceDir /var/lib/tor/simplex-smp/
+## Map smp port (5223) to tor
+HiddenServicePort 5223 localhost:5223
+...
+```
+
+3. Restart `tor` system service:
+
+```
+sudo systemctl restart tor
+```
+
+4. Onion address can be obtained from following file:
+
+```sh
+sudo cat /var/lib/tor/simplex-smp/hostname
+```
@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack

 name:           simplexmq
-version:        3.2.0
+version:        3.3.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                <./docs/Simplex-Messaging-Client.html client> and
@@ -52,6 +52,7 @@ library
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220625_v2_ntf_mode
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220811_onion_hosts
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220817_connection_ntfs
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands
      Simplex.Messaging.Client
      Simplex.Messaging.Client.Agent
      Simplex.Messaging.Crypto
@@ -81,6 +82,7 @@ library
      Simplex.Messaging.Server.Stats
      Simplex.Messaging.Server.StoreLog
      Simplex.Messaging.TMap
+      Simplex.Messaging.TMap2
      Simplex.Messaging.Transport
      Simplex.Messaging.Transport.Client
      Simplex.Messaging.Transport.HTTP2
@@ -114,7 +116,7 @@ library
    , cryptonite >=0.27 && <0.30
    , cryptostore ==0.2.*
    , data-default ==0.7.*
-    , direct-sqlite ==2.3.*
+    , direct-sqlcipher ==2.3.*
    , directory ==1.3.*
    , filepath ==1.4.*
    , generic-random >=1.3 && <1.5
@@ -125,13 +127,13 @@ library
    , memory ==0.15.*
    , mtl ==2.2.*
    , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.*
+    , network-transport ==0.5.4
    , optparse-applicative >=0.15 && <0.17
    , process ==1.6.*
    , random >=1.1 && <1.3
    , simple-logger ==0.1.*
    , socks ==0.6.*
-    , sqlite-simple ==0.4.*
+    , sqlcipher-simple ==0.4.*
    , stm ==2.5.*
    , template-haskell ==2.16.*
    , text ==1.2.*
@@ -175,7 +177,7 @@ executable ntf-server
    , cryptonite >=0.27 && <0.30
    , cryptostore ==0.2.*
    , data-default ==0.7.*
-    , direct-sqlite ==2.3.*
+    , direct-sqlcipher ==2.3.*
    , directory ==1.3.*
    , filepath ==1.4.*
    , generic-random >=1.3 && <1.5
@@ -186,14 +188,14 @@ executable ntf-server
    , memory ==0.15.*
    , mtl ==2.2.*
    , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.*
+    , network-transport ==0.5.4
    , optparse-applicative >=0.15 && <0.17
    , process ==1.6.*
    , random >=1.1 && <1.3
    , simple-logger ==0.1.*
    , simplexmq
    , socks ==0.6.*
-    , sqlite-simple ==0.4.*
+    , sqlcipher-simple ==0.4.*
    , stm ==2.5.*
    , template-haskell ==2.16.*
    , text ==1.2.*
@@ -237,7 +239,7 @@ executable smp-agent
    , cryptonite >=0.27 && <0.30
    , cryptostore ==0.2.*
    , data-default ==0.7.*
-    , direct-sqlite ==2.3.*
+    , direct-sqlcipher ==2.3.*
    , directory ==1.3.*
    , filepath ==1.4.*
    , generic-random >=1.3 && <1.5
@@ -248,14 +250,14 @@ executable smp-agent
    , memory ==0.15.*
    , mtl ==2.2.*
    , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.*
+    , network-transport ==0.5.4
    , optparse-applicative >=0.15 && <0.17
    , process ==1.6.*
    , random >=1.1 && <1.3
    , simple-logger ==0.1.*
    , simplexmq
    , socks ==0.6.*
-    , sqlite-simple ==0.4.*
+    , sqlcipher-simple ==0.4.*
    , stm ==2.5.*
    , template-haskell ==2.16.*
    , text ==1.2.*
@@ -299,7 +301,7 @@ executable smp-server
    , cryptonite >=0.27 && <0.30
    , cryptostore ==0.2.*
    , data-default ==0.7.*
-    , direct-sqlite ==2.3.*
+    , direct-sqlcipher ==2.3.*
    , directory ==1.3.*
    , filepath ==1.4.*
    , generic-random >=1.3 && <1.5
@@ -310,14 +312,14 @@ executable smp-server
    , memory ==0.15.*
    , mtl ==2.2.*
    , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.*
+    , network-transport ==0.5.4
    , optparse-applicative >=0.15 && <0.17
    , process ==1.6.*
    , random >=1.1 && <1.3
    , simple-logger ==0.1.*
    , simplexmq
    , socks ==0.6.*
-    , sqlite-simple ==0.4.*
+    , sqlcipher-simple ==0.4.*
    , stm ==2.5.*
    , template-haskell ==2.16.*
    , text ==1.2.*
@@ -378,7 +380,7 @@ test-suite smp-server-test
    , cryptonite >=0.27 && <0.30
    , cryptostore ==0.2.*
    , data-default ==0.7.*
-    , direct-sqlite ==2.3.*
+    , direct-sqlcipher ==2.3.*
    , directory ==1.3.*
    , filepath ==1.4.*
    , generic-random >=1.3 && <1.5
@@ -391,14 +393,14 @@ test-suite smp-server-test
    , memory ==0.15.*
    , mtl ==2.2.*
    , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.*
+    , network-transport ==0.5.4
    , optparse-applicative >=0.15 && <0.17
    , process ==1.6.*
    , random >=1.1 && <1.3
    , simple-logger ==0.1.*
    , simplexmq
    , socks ==0.6.*
-    , sqlite-simple ==0.4.*
+    , sqlcipher-simple ==0.4.*
    , stm ==2.5.*
    , template-haskell ==2.16.*
    , text ==1.2.*
@@ -12,7 +12,6 @@
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
-{-# LANGUAGE TypeApplications #-}

 -- |
 -- Module      : Simplex.Messaging.Agent
@@ -39,6 +38,11 @@ module Simplex.Messaging.Agent
    disconnectAgentClient,
    resumeAgentClient,
    withAgentLock,
+    createConnectionAsync,
+    joinConnectionAsync,
+    allowConnectionAsync,
+    ackMessageAsync,
+    deleteConnectionAsync,
    createConnection,
    joinConnection,
    allowConnection,
@@ -68,6 +72,7 @@ module Simplex.Messaging.Agent
    toggleConnectionNtfs,
    activateAgent,
    suspendAgent,
+    execAgentStoreSQL,
    logConnection,
  )
 where
@@ -80,13 +85,15 @@ import Control.Monad.Reader
 import Crypto.Random (MonadRandom)
 import Data.Bifunctor (bimap, first, second)
 import Data.ByteString.Char8 (ByteString)
-import Data.Composition ((.:), (.:.))
+import Data.Composition ((.:), (.:.), (.::))
 import Data.Functor (($>))
+import Data.List (deleteFirstsBy)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Maybe (isJust)
+import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
@@ -141,13 +148,33 @@ resumeAgentClient c = atomically $ writeTVar (active c) True
 -- |
 type AgentErrorMonad m = (MonadUnliftIO m, MonadError AgentErrorType m)

+-- | Create SMP agent connection (NEW command) asynchronously, synchronous response is new connection id
+createConnectionAsync :: forall m c. (AgentErrorMonad m, ConnectionModeI c) => AgentClient -> ACorrId -> Bool -> SConnectionMode c -> m ConnId
+createConnectionAsync c corrId enableNtfs cMode = withAgentEnv c $ newConnAsync c corrId enableNtfs cMode
+
+-- | Join SMP agent connection (JOIN command) asynchronously, synchronous response is new connection id
+joinConnectionAsync :: AgentErrorMonad m => AgentClient -> ACorrId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
+joinConnectionAsync c corrId enableNtfs = withAgentEnv c .: joinConnAsync c corrId enableNtfs
+
+-- | Allow connection to continue after CONF notification (LET command), no synchronous response
+allowConnectionAsync :: AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> ConfirmationId -> ConnInfo -> m ()
+allowConnectionAsync c = withAgentEnv c .:: allowConnectionAsync' c
+
+-- | Acknowledge message (ACK command) asynchronously, no synchronous response
+ackMessageAsync :: forall m. AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> m ()
+ackMessageAsync c = withAgentEnv c .:. ackMessageAsync' c
+
+-- | Delete SMP agent connection (DEL command) asynchronously, no synchronous response
+deleteConnectionAsync :: AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> m ()
+deleteConnectionAsync c = withAgentEnv c .: deleteConnectionAsync' c
+
 -- | Create SMP agent connection (NEW command)
 createConnection :: AgentErrorMonad m => AgentClient -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
-createConnection c enableNtfs cMode = withAgentEnv c $ newConn c "" enableNtfs cMode
+createConnection c enableNtfs cMode = withAgentEnv c $ newConn c "" False enableNtfs cMode

 -- | Join SMP agent connection (JOIN command)
 joinConnection :: AgentErrorMonad m => AgentClient -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
-joinConnection c enableNtfs = withAgentEnv c .: joinConn c "" enableNtfs
+joinConnection c enableNtfs = withAgentEnv c .: joinConn c "" False enableNtfs

 -- | Allow connection to continue after CONF notification (LET command)
 allowConnection :: AgentErrorMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
@@ -253,6 +280,9 @@ activateAgent c = withAgentEnv c $ activateAgent' c
 suspendAgent :: AgentErrorMonad m => AgentClient -> Int -> m ()
 suspendAgent c = withAgentEnv c . suspendAgent' c

+execAgentStoreSQL :: AgentErrorMonad m => AgentClient -> Text -> m [Text]
+execAgentStoreSQL c = withAgentEnv c . execAgentStoreSQL' c
+
 withAgentEnv :: AgentClient -> ReaderT Env m a -> m a
 withAgentEnv c = (`runReaderT` agentEnv c)

@@ -283,8 +313,8 @@ client c@AgentClient {rcvQ, subQ} = forever $ do
 -- | execute any SMP agent command
 processCommand :: forall m. AgentMonad m => AgentClient -> (ConnId, ACommand 'Client) -> m (ConnId, ACommand 'Agent)
 processCommand c (connId, cmd) = case cmd of
-  NEW (ACM cMode) -> second (INV . ACR cMode) <$> newConn c connId True cMode
-  JOIN (ACR _ cReq) connInfo -> (,OK) <$> joinConn c connId True cReq connInfo
+  NEW enableNtfs (ACM cMode) -> second (INV . ACR cMode) <$> newConn c connId False enableNtfs cMode
+  JOIN enableNtfs (ACR _ cReq) connInfo -> (,OK) <$> joinConn c connId False enableNtfs cReq connInfo
  LET confId ownCInfo -> allowConnection' c connId confId ownCInfo $> (connId, OK)
  ACPT invId ownCInfo -> (,OK) <$> acceptContact' c connId True invId ownCInfo
  RJCT invId -> rejectContact' c connId invId $> (connId, OK)
@@ -295,15 +325,74 @@ processCommand c (connId, cmd) = case cmd of
  DEL -> deleteConnection' c connId $> (connId, OK)
  CHK -> (connId,) . STAT <$> getConnectionServers' c connId

-newConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
-newConn c connId enableNtfs cMode = do
-  srv <- getSMPServer c
-  clientVRange <- asks $ smpClientVRange . config
-  (rq, qUri) <- newRcvQueue c srv clientVRange
+newConnAsync :: forall m c. (AgentMonad m, ConnectionModeI c) => AgentClient -> ACorrId -> Bool -> SConnectionMode c -> m ConnId
+newConnAsync c corrId enableNtfs cMode = do
  g <- asks idsDrg
  connAgentVersion <- asks $ maxVersion . smpAgentVRange . config
-  let cData = ConnData {connId, connAgentVersion, enableNtfs, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent
-  connId' <- withStore c $ \db -> createRcvConn db g cData rq cMode
+  let cData = ConnData {connId = "", connAgentVersion, enableNtfs, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent
+  connId <- withStore c $ \db -> createNewConn db g cData cMode
+  enqueueCommand c corrId connId Nothing $ AClientCommand $ NEW enableNtfs (ACM cMode)
+  pure connId
+
+joinConnAsync :: AgentMonad m => AgentClient -> ACorrId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
+joinConnAsync c corrId enableNtfs cReqUri@(CRInvitationUri (ConnReqUriData _ agentVRange _) _) cInfo = do
+  aVRange <- asks $ smpAgentVRange . config
+  case agentVRange `compatibleVersion` aVRange of
+    Just (Compatible connAgentVersion) -> do
+      g <- asks idsDrg
+      let duplexHS = connAgentVersion /= 1
+          cData = ConnData {connId = "", connAgentVersion, enableNtfs, duplexHandshake = Just duplexHS}
+      connId <- withStore c $ \db -> createNewConn db g cData SCMInvitation
+      enqueueCommand c corrId connId Nothing $ AClientCommand $ JOIN enableNtfs (ACR sConnectionMode cReqUri) cInfo
+      pure connId
+    _ -> throwError $ AGENT A_VERSION
+joinConnAsync _c _corrId _enableNtfs (CRContactUri _) _cInfo =
+  throwError $ CMD PROHIBITED
+
+allowConnectionAsync' :: AgentMonad m => AgentClient -> ACorrId -> ConnId -> ConfirmationId -> ConnInfo -> m ()
+allowConnectionAsync' c corrId connId confId ownConnInfo =
+  withStore c (`getConn` connId) >>= \case
+    SomeConn _ (RcvConnection _ RcvQueue {server}) ->
+      enqueueCommand c corrId connId (Just server) $ AClientCommand $ LET confId ownConnInfo
+    _ -> throwError $ CMD PROHIBITED
+
+ackMessageAsync' :: forall m. AgentMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> m ()
+ackMessageAsync' c corrId connId msgId =
+  withStore c (`getConn` connId) >>= \case
+    SomeConn _ (DuplexConnection _ rq _) -> enqueueAck rq
+    SomeConn _ (RcvConnection _ rq) -> enqueueAck rq
+    SomeConn _ (SndConnection _ _) -> throwError $ CONN SIMPLEX
+    SomeConn _ (ContactConnection _ _) -> throwError $ CMD PROHIBITED
+    SomeConn _ (NewConnection _) -> throwError $ CMD PROHIBITED
+  where
+    enqueueAck :: RcvQueue -> m ()
+    enqueueAck RcvQueue {server} =
+      enqueueCommand c corrId connId (Just server) $ AClientCommand $ ACK msgId
+
+deleteConnectionAsync' :: forall m. AgentMonad m => AgentClient -> ACorrId -> ConnId -> m ()
+deleteConnectionAsync' c@AgentClient {subQ} corrId connId =
+  withStore c (`getConn` connId) >>= \case
+    SomeConn _ (DuplexConnection _ rq _) -> enqueueDelete rq
+    SomeConn _ (RcvConnection _ rq) -> enqueueDelete rq
+    SomeConn _ (ContactConnection _ rq) -> enqueueDelete rq
+    SomeConn _ (SndConnection _ _) -> withStore' c (`deleteConn` connId) >> notifyDeleted
+    SomeConn _ (NewConnection _) -> withStore' c (`deleteConn` connId) >> notifyDeleted
+  where
+    enqueueDelete :: RcvQueue -> m ()
+    enqueueDelete RcvQueue {server} =
+      enqueueCommand c corrId connId (Just server) $ AClientCommand DEL
+    notifyDeleted :: m ()
+    notifyDeleted = atomically $ writeTBQueue subQ (corrId, connId, OK)
+
+newConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
+newConn c connId asyncMode enableNtfs cMode =
+  getSMPServer c >>= newConnSrv c connId asyncMode enableNtfs cMode
+
+newConnSrv :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> SConnectionMode c -> SMPServer -> m (ConnId, ConnectionRequestUri c)
+newConnSrv c connId asyncMode enableNtfs cMode srv = do
+  clientVRange <- asks $ smpClientVRange . config
+  (rq, qUri) <- newRcvQueue c srv clientVRange
+  connId' <- setUpConn asyncMode rq
  addSubscription c rq connId'
  when enableNtfs $ do
    ns <- asks ntfSupervisor
@@ -316,9 +405,26 @@ newConn c connId enableNtfs cMode = do
      (pk1, pk2, e2eRcvParams) <- liftIO $ CR.generateE2EParams CR.e2eEncryptVersion
      withStore' c $ \db -> createRatchetX3dhKeys db connId' pk1 pk2
      pure (connId', CRInvitationUri crData $ toVersionRangeT e2eRcvParams CR.e2eEncryptVRange)
+  where
+    setUpConn True rq = do
+      withStore c $ \db -> updateNewConnRcv db connId rq
+      pure connId
+    setUpConn False rq = do
+      g <- asks idsDrg
+      connAgentVersion <- asks $ maxVersion . smpAgentVRange . config
+      let cData = ConnData {connId, connAgentVersion, enableNtfs, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent
+      withStore c $ \db -> createRcvConn db g cData rq cMode

-joinConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
-joinConn c connId enableNtfs (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2eRcvParamsUri) cInfo = do
+joinConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
+joinConn c connId asyncMode enableNtfs cReq cInfo = do
+  srv <- case cReq of
+    CRInvitationUri ConnReqUriData {crSmpQueues = SMPQueueUri {queueAddress} :| _} _ ->
+      getNextSMPServer c [smpServer (queueAddress :: SMPQueueAddress)]
+    _ -> getSMPServer c
+  joinConnSrv c connId asyncMode enableNtfs cReq cInfo srv
+
+joinConnSrv :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> SMPServer -> m ConnId
+joinConnSrv c connId asyncMode enableNtfs (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2eRcvParamsUri) cInfo srv = do
  aVRange <- asks $ smpAgentVRange . config
  clientVRange <- asks $ smpClientVRange . config
  case ( qUri `compatibleVersion` clientVRange,
@@ -330,38 +436,47 @@ joinConn c connId enableNtfs (CRInvitationUri (ConnReqUriData _ agentVRange (qUr
      (_, rcDHRs) <- liftIO C.generateKeyPair'
      let rc = CR.initSndRatchet rcDHRr rcDHRs $ CR.x3dhSnd pk1 pk2 e2eRcvParams
      sq <- newSndQueue qInfo
-      g <- asks idsDrg
      let duplexHS = connAgentVersion /= 1
          cData = ConnData {connId, connAgentVersion, enableNtfs, duplexHandshake = Just duplexHS}
-      connId' <- withStore c $ \db -> runExceptT $ do
-        connId' <- ExceptT $ createSndConn db g cData sq
-        liftIO $ createRatchet db connId' rc
-        pure connId'
+      connId' <- setUpConn asyncMode cData sq rc
      let cData' = (cData :: ConnData) {connId = connId'}
-      tryError (confirmQueue aVersion c cData' sq cInfo $ Just e2eSndParams) >>= \case
+      tryError (confirmQueue aVersion c cData' sq srv cInfo $ Just e2eSndParams) >>= \case
        Right _ -> do
          unless duplexHS . void $ enqueueMessage c cData' sq SMP.noMsgFlags HELLO
          pure connId'
        Left e -> do
          -- TODO recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md
-          withStore' c (`deleteConn` connId')
+          unless asyncMode $ withStore' c (`deleteConn` connId')
          throwError e
+      where
+        setUpConn True _ sq rc =
+          withStore c $ \db -> runExceptT $ do
+            ExceptT $ updateNewConnSnd db connId sq
+            liftIO $ createRatchet db connId rc
+            pure connId
+        setUpConn False cData sq rc = do
+          g <- asks idsDrg
+          withStore c $ \db -> runExceptT $ do
+            connId' <- ExceptT $ createSndConn db g cData sq
+            liftIO $ createRatchet db connId' rc
+            pure connId'
    _ -> throwError $ AGENT A_VERSION
-joinConn c connId enableNtfs (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInfo = do
+joinConnSrv c connId False enableNtfs (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInfo srv = do
  aVRange <- asks $ smpAgentVRange . config
  clientVRange <- asks $ smpClientVRange . config
  case ( qUri `compatibleVersion` clientVRange,
         agentVRange `compatibleVersion` aVRange
       ) of
    (Just qInfo, Just vrsn) -> do
-      (connId', cReq) <- newConn c connId enableNtfs SCMInvitation
+      (connId', cReq) <- newConnSrv c connId False enableNtfs SCMInvitation srv
      sendInvitation c qInfo vrsn cReq cInfo
      pure connId'
    _ -> throwError $ AGENT A_VERSION
+joinConnSrv _c _connId True _enableNtfs (CRContactUri _) _cInfo _srv = do
+  throwError $ CMD PROHIBITED

-createReplyQueue :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> m SMPQueueInfo
-createReplyQueue c ConnData {connId, enableNtfs} SndQueue {smpClientVersion} = do
-  srv <- getSMPServer c
+createReplyQueue :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> SMPServer -> m SMPQueueInfo
+createReplyQueue c ConnData {connId, enableNtfs} SndQueue {smpClientVersion} srv = do
  (rq, qUri) <- newRcvQueue c srv $ versionToRange smpClientVersion
  let qInfo = toVersionT qUri smpClientVersion
  addSubscription c rq connId
@@ -375,13 +490,14 @@ createReplyQueue c ConnData {connId, enableNtfs} SndQueue {smpClientVersion} = d
 allowConnection' :: AgentMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
 allowConnection' c connId confId ownConnInfo =
  withStore c (`getConn` connId) >>= \case
-    SomeConn _ (RcvConnection cData rq) -> do
-      AcceptedConfirmation {senderConf} <- withStore c $ \db -> runExceptT $ do
-        conf <- ExceptT $ acceptConfirmation db confId ownConnInfo
-        liftIO $ createRatchet db connId $ ratchetState (conf :: AcceptedConfirmation)
-        pure conf
-      processConfirmation c rq senderConf
-      mapM_ (connectReplyQueues c cData ownConnInfo) (L.nonEmpty $ smpReplyQueues senderConf)
+    SomeConn _ (RcvConnection _ rq@RcvQueue {server, rcvId, e2ePrivKey, smpClientVersion = v}) -> do
+      senderKey <- withStore c $ \db -> runExceptT $ do
+        AcceptedConfirmation {ratchetState, senderConf = SMPConfirmation {senderKey, e2ePubKey, smpClientVersion = v'}} <- ExceptT $ acceptConfirmation db confId ownConnInfo
+        liftIO $ createRatchet db connId ratchetState
+        let dhSecret = C.dh' e2ePubKey e2ePrivKey
+        liftIO $ setRcvQueueConfirmedE2E db rq dhSecret $ min v v'
+        pure senderKey
+      enqueueCommand c "" connId (Just server) . AInternalCommand $ ICAllowSecure rcvId senderKey
    _ -> throwError $ CMD PROHIBITED

 -- | Accept contact (ACPT command) in Reader monad
@@ -391,7 +507,7 @@ acceptContact' c connId enableNtfs invId ownConnInfo = do
  withStore c (`getConn` contactConnId) >>= \case
    SomeConn _ ContactConnection {} -> do
      withStore' c $ \db -> acceptInvitation db invId ownConnInfo
-      joinConn c connId enableNtfs connReq ownConnInfo `catchError` \err -> do
+      joinConn c connId False enableNtfs connReq ownConnInfo `catchError` \err -> do
        withStore' c (`unacceptInvitation` invId)
        throwError err
    _ -> throwError $ CMD PROHIBITED
@@ -401,28 +517,24 @@ rejectContact' :: AgentMonad m => AgentClient -> ConnId -> InvitationId -> m ()
 rejectContact' c contactConnId invId =
  withStore c $ \db -> deleteInvitation db contactConnId invId

-processConfirmation :: AgentMonad m => AgentClient -> RcvQueue -> SMPConfirmation -> m ()
-processConfirmation c rq@RcvQueue {e2ePrivKey, smpClientVersion = v} SMPConfirmation {senderKey, e2ePubKey, smpClientVersion = v'} = do
-  let dhSecret = C.dh' e2ePubKey e2ePrivKey
-  withStore' c $ \db -> setRcvQueueConfirmedE2E db rq dhSecret $ min v v'
-  secureQueue c rq senderKey
-  withStore' c $ \db -> setRcvQueueStatus db rq Secured
-
 -- | Subscribe to receive connection messages (SUB command) in Reader monad
 subscribeConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
 subscribeConnection' c connId =
-  withStore c (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection cData rq sq) -> do
-      resumeMsgDelivery c cData sq
-      subscribe rq
-    SomeConn _ (SndConnection cData sq) -> do
-      resumeMsgDelivery c cData sq
-      case status (sq :: SndQueue) of
-        Confirmed -> pure ()
-        Active -> throwError $ CONN SIMPLEX
-        _ -> throwError $ INTERNAL "unexpected queue status"
-    SomeConn _ (RcvConnection _ rq) -> subscribe rq
-    SomeConn _ (ContactConnection _ rq) -> subscribe rq
+  withStore c (`getConn` connId) >>= \conn -> do
+    resumeConnCmds c connId
+    case conn of
+      SomeConn _ (DuplexConnection cData rq sq) -> do
+        resumeMsgDelivery c cData sq
+        subscribe rq
+      SomeConn _ (SndConnection cData sq) -> do
+        resumeMsgDelivery c cData sq
+        case status (sq :: SndQueue) of
+          Confirmed -> pure ()
+          Active -> throwError $ CONN SIMPLEX
+          _ -> throwError $ INTERNAL "unexpected queue status"
+      SomeConn _ (RcvConnection _ rq) -> subscribe rq
+      SomeConn _ (ContactConnection _ rq) -> subscribe rq
+      SomeConn _ (NewConnection _) -> pure ()
  where
    subscribe :: RcvQueue -> m ()
    subscribe rq = do
@@ -436,33 +548,34 @@ subscribeConnections' c connIds = do
  conns :: Map ConnId (Either StoreError SomeConn) <- M.fromList . zip connIds <$> withStore' c (forM connIds . getConn)
  let (errs, cs) = M.mapEither id conns
      errs' = M.map (Left . storeError) errs
-      (sndQs, rcvQs) = M.mapEither rcvOrSndQueue cs
-      sndRs = M.map (sndSubResult . fst) sndQs
-      srvRcvQs :: Map SMPServer (Map ConnId (RcvQueue, ConnData)) = M.foldlWithKey' addRcvQueue M.empty rcvQs
+      (subRs, rcvQs) = M.mapEither rcvQueueOrResult cs
+      srvRcvQs :: Map SMPServer (Map ConnId RcvQueue) = M.foldlWithKey' addRcvQueue M.empty rcvQs
  mapM_ (mapM_ (uncurry $ resumeMsgDelivery c) . sndQueue) cs
+  mapM_ (resumeConnCmds c) $ M.keys cs
  rcvRs <- mapConcurrently subscribe (M.assocs srvRcvQs)
  ns <- asks ntfSupervisor
  tkn <- readTVarIO (ntfTkn ns)
  when (instantNotifications tkn) . void . forkIO $ sendNtfCreate ns rcvRs
-  let rs = M.unions $ errs' : sndRs : rcvRs
+  let rs = M.unions $ errs' : subRs : rcvRs
  notifyResultError rs
  pure rs
  where
-    rcvOrSndQueue :: SomeConn -> Either (SndQueue, ConnData) (RcvQueue, ConnData)
-    rcvOrSndQueue = \case
-      SomeConn _ (DuplexConnection cData rq _) -> Right (rq, cData)
-      SomeConn _ (SndConnection cData sq) -> Left (sq, cData)
-      SomeConn _ (RcvConnection cData rq) -> Right (rq, cData)
-      SomeConn _ (ContactConnection cData rq) -> Right (rq, cData)
+    rcvQueueOrResult :: SomeConn -> Either (Either AgentErrorType ()) RcvQueue
+    rcvQueueOrResult = \case
+      SomeConn _ (DuplexConnection _ rq _) -> Right rq
+      SomeConn _ (SndConnection _ sq) -> Left $ sndSubResult sq
+      SomeConn _ (RcvConnection _ rq) -> Right rq
+      SomeConn _ (ContactConnection _ rq) -> Right rq
+      SomeConn _ (NewConnection _) -> Left (Right ())
    sndSubResult :: SndQueue -> Either AgentErrorType ()
    sndSubResult sq = case status (sq :: SndQueue) of
      Confirmed -> Right ()
      Active -> Left $ CONN SIMPLEX
      _ -> Left $ INTERNAL "unexpected queue status"
-    addRcvQueue :: Map SMPServer (Map ConnId (RcvQueue, ConnData)) -> ConnId -> (RcvQueue, ConnData) -> Map SMPServer (Map ConnId (RcvQueue, ConnData))
-    addRcvQueue m connId rq@(RcvQueue {server}, _) = M.alter (Just . maybe (M.singleton connId rq) (M.insert connId rq)) server m
-    subscribe :: (SMPServer, Map ConnId (RcvQueue, ConnData)) -> m (Map ConnId (Either AgentErrorType ()))
-    subscribe (srv, qs) = snd <$> subscribeQueues c srv (M.map fst qs)
+    addRcvQueue :: Map SMPServer (Map ConnId RcvQueue) -> ConnId -> RcvQueue -> Map SMPServer (Map ConnId RcvQueue)
+    addRcvQueue m connId rq@RcvQueue {server} = M.alter (Just . maybe (M.singleton connId rq) (M.insert connId rq)) server m
+    subscribe :: (SMPServer, Map ConnId RcvQueue) -> m (Map ConnId (Either AgentErrorType ()))
+    subscribe (srv, qs) = snd <$> subscribeQueues c srv qs
    sendNtfCreate :: NtfSupervisor -> [Map ConnId (Either AgentErrorType ())] -> m ()
    sendNtfCreate ns rcvRs =
      forM_ (concatMap M.assocs rcvRs) $ \case
@@ -502,6 +615,7 @@ getConnectionMessage' c connId = do
    SomeConn _ (RcvConnection _ rq) -> getQueueMessage c rq
    SomeConn _ (ContactConnection _ rq) -> getQueueMessage c rq
    SomeConn _ SndConnection {} -> throwError $ CONN SIMPLEX
+    SomeConn _ NewConnection {} -> throwError $ CMD PROHIBITED

 getNotificationMessage' :: forall m. AgentMonad m => AgentClient -> C.CbNonce -> ByteString -> m (NotificationInfo, [SMPMsgMeta])
 getNotificationMessage' c nonce encNtfInfo = do
@@ -541,11 +655,133 @@ sendMessage' c connId msgFlags msg =
    enqueueMsg :: ConnData -> SndQueue -> m AgentMsgId
    enqueueMsg cData sq = enqueueMessage c cData sq msgFlags $ A_MSG msg

+-- / async command processing v v v
+
+enqueueCommand :: AgentMonad m => AgentClient -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> m ()
+enqueueCommand c corrId connId server aCommand = do
+  resumeSrvCmds c server
+  commandId <- withStore' c $ \db -> createCommand db corrId connId server aCommand
+  queuePendingCommands c server [commandId]
+
+resumeSrvCmds :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m ()
+resumeSrvCmds c server =
+  unlessM (cmdProcessExists c server) $
+    async (runCommandProcessing c server)
+      >>= \a -> atomically (TM.insert server a $ asyncCmdProcesses c)
+
+resumeConnCmds :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
+resumeConnCmds c connId =
+  unlessM connQueued $
+    withStore' c (`getPendingCommands` connId)
+      >>= mapM_ (uncurry enqueueConnCmds)
+  where
+    enqueueConnCmds srv cmdIds = do
+      resumeSrvCmds c srv
+      queuePendingCommands c srv cmdIds
+    connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connCmdsQueued c)
+
+cmdProcessExists :: AgentMonad m => AgentClient -> Maybe SMPServer -> m Bool
+cmdProcessExists c srv = atomically $ TM.member srv (asyncCmdProcesses c)
+
+queuePendingCommands :: AgentMonad m => AgentClient -> Maybe SMPServer -> [AsyncCmdId] -> m ()
+queuePendingCommands c server cmdIds = atomically $ do
+  q <- getPendingCommandQ c server
+  mapM_ (writeTQueue q) cmdIds
+
+getPendingCommandQ :: AgentClient -> Maybe SMPServer -> STM (TQueue AsyncCmdId)
+getPendingCommandQ c server = do
+  maybe newMsgQueue pure =<< TM.lookup server (asyncCmdQueues c)
+  where
+    newMsgQueue = do
+      cq <- newTQueue
+      TM.insert server cq $ asyncCmdQueues c
+      pure cq
+
+runCommandProcessing :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m ()
+runCommandProcessing c@AgentClient {subQ} server = do
+  cq <- atomically $ getPendingCommandQ c server
+  ri <- asks $ messageRetryInterval . config -- different retry interval?
+  forever $ do
+    atomically $ endAgentOperation c AOSndNetwork
+    atomically $ throwWhenInactive c
+    cmdId <- atomically $ readTQueue cq
+    atomically $ beginAgentOperation c AOSndNetwork
+    E.try (withStore c $ \db -> getPendingCommand db cmdId) >>= \case
+      Left (e :: E.SomeException) -> atomically $ writeTBQueue subQ ("", "", ERR . INTERNAL $ show e)
+      Right (corrId, connId, cmd) -> processCmd ri corrId connId cmdId cmd
+  where
+    processCmd :: RetryInterval -> ACorrId -> ConnId -> AsyncCmdId -> AgentCommand -> m ()
+    processCmd ri corrId connId cmdId = \case
+      AClientCommand cmd -> case cmd of
+        NEW enableNtfs (ACM cMode) -> do
+          usedSrvs <- newTVarIO ([] :: [SMPServer])
+          tryCommand . withNextSrv usedSrvs [] $ \srv -> do
+            (_, cReq) <- newConnSrv c connId True enableNtfs cMode srv
+            notify $ INV (ACR cMode cReq)
+        JOIN enableNtfs (ACR _ cReq@(CRInvitationUri ConnReqUriData {crSmpQueues = SMPQueueUri {queueAddress} :| _} _)) connInfo -> do
+          let initUsed = [smpServer (queueAddress :: SMPQueueAddress)]
+          usedSrvs <- newTVarIO initUsed
+          tryCommand . withNextSrv usedSrvs initUsed $ \srv -> do
+            void $ joinConnSrv c connId True enableNtfs cReq connInfo srv
+            notify OK
+        LET confId ownCInfo -> tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK
+        ACK msgId -> tryCommand $ ackMessage' c connId msgId >> notify OK
+        DEL -> tryCommand $ deleteConnection' c connId >> notify OK
+        _ -> notify $ ERR $ INTERNAL $ "unsupported async command " <> show (aCommandTag cmd)
+      AInternalCommand cmd -> case server of
+        Just _srv -> case cmd of
+          ICAckDel _rId srvMsgId msgId -> tryCommand $ ack _rId srvMsgId >> withStore' c (\db -> deleteMsg db connId msgId)
+          ICAck _rId srvMsgId -> tryCommand $ ack _rId srvMsgId
+          ICAllowSecure _rId senderKey -> tryCommand $ do
+            (SomeConn _ conn, AcceptedConfirmation {senderConf, ownConnInfo}) <-
+              withStore c $ \db -> runExceptT $ (,) <$> ExceptT (getConn db connId) <*> ExceptT (getAcceptedConfirmation db connId)
+            case conn of
+              RcvConnection cData rq -> do
+                secure rq senderKey
+                mapM_ (connectReplyQueues c cData ownConnInfo) (L.nonEmpty $ smpReplyQueues senderConf)
+              _ -> throwError $ INTERNAL $ "incorrect connection type " <> show (internalCmdTag cmd)
+          ICDuplexSecure _rId senderKey -> tryCommand $ do
+            SomeConn _ conn <- withStore c (`getConn` connId)
+            case conn of
+              DuplexConnection cData rq sq -> do
+                secure rq senderKey
+                when (duplexHandshake cData == Just True) . void $
+                  enqueueMessage c cData sq SMP.MsgFlags {notification = True} HELLO
+              _ -> throwError $ INTERNAL $ "incorrect connection type " <> show (internalCmdTag cmd)
+        _ -> throwError $ INTERNAL $ "command requires server " <> show (internalCmdTag cmd)
+        where
+          ack _rId srvMsgId = do
+            -- TODO get particular queue
+            rq <- withStore c (`getRcvQueue` connId)
+            ackQueueMessage c rq srvMsgId
+          secure :: RcvQueue -> SMP.SndPublicVerifyKey -> m ()
+          secure rq senderKey = do
+            secureQueue c rq senderKey
+            withStore' c $ \db -> setRcvQueueStatus db rq Secured
+      where
+        tryCommand action = withRetryInterval ri $ \loop ->
+          tryError action >>= \case
+            Left e
+              | temporaryAgentError e || e == BROKER HOST -> retrySndOp c loop
+              | otherwise -> notify (ERR e) >> withStore' c (`deleteCommand` cmdId)
+            Right () -> withStore' c (`deleteCommand` cmdId)
+        notify cmd = atomically $ writeTBQueue subQ (corrId, connId, cmd)
+    withNextSrv :: TVar [SMPServer] -> [SMPServer] -> (SMPServer -> m ()) -> m ()
+    withNextSrv usedSrvs initUsed action = do
+      used <- readTVarIO usedSrvs
+      srv <- getNextSMPServer c used
+      atomically $ do
+        srvs <- readTVar $ smpServers c
+        let used' = if length used + 1 >= L.length srvs then initUsed else srv : used
+        writeTVar usedSrvs used'
+      action srv
+-- ^ ^ ^ async command processing /
+
 enqueueMessage :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> MsgFlags -> AMessage -> m AgentMsgId
 enqueueMessage c cData@ConnData {connId, connAgentVersion} sq msgFlags aMessage = do
  resumeMsgDelivery c cData sq
  msgId <- storeSentMsg
-  queuePendingMsgs c connId sq [msgId]
+  queuePendingMsgs c sq [msgId]
  pure $ unId msgId
  where
    storeSentMsg :: m InternalId
@@ -565,28 +801,28 @@ enqueueMessage c cData@ConnData {connId, connAgentVersion} sq msgFlags aMessage

 resumeMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> m ()
 resumeMsgDelivery c cData@ConnData {connId} sq@SndQueue {server, sndId} = do
-  let qKey = (connId, server, sndId)
+  let qKey = (server, sndId)
  unlessM (queueDelivering qKey) $
    async (runSmpQueueMsgDelivery c cData sq)
      >>= \a -> atomically (TM.insert qKey a $ smpQueueMsgDeliveries c)
  unlessM connQueued $
    withStore' c (`getPendingMsgs` connId)
-      >>= queuePendingMsgs c connId sq
+      >>= queuePendingMsgs c sq
  where
    queueDelivering qKey = atomically $ TM.member qKey (smpQueueMsgDeliveries c)
    connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connMsgsQueued c)

-queuePendingMsgs :: AgentMonad m => AgentClient -> ConnId -> SndQueue -> [InternalId] -> m ()
-queuePendingMsgs c connId sq msgIds = atomically $ do
+queuePendingMsgs :: AgentMonad m => AgentClient -> SndQueue -> [InternalId] -> m ()
+queuePendingMsgs c sq msgIds = atomically $ do
  modifyTVar' (msgDeliveryOp c) $ \s -> s {opsInProgress = opsInProgress s + length msgIds}
  -- s <- readTVar (msgDeliveryOp c)
  -- unsafeIOToSTM $ putStrLn $ "msgDeliveryOp: " <> show (opsInProgress s)
-  q <- getPendingMsgQ c connId sq
+  q <- getPendingMsgQ c sq
  mapM_ (writeTQueue q) msgIds

-getPendingMsgQ :: AgentClient -> ConnId -> SndQueue -> STM (TQueue InternalId)
-getPendingMsgQ c connId SndQueue {server, sndId} = do
-  let qKey = (connId, server, sndId)
+getPendingMsgQ :: AgentClient -> SndQueue -> STM (TQueue InternalId)
+getPendingMsgQ c SndQueue {server, sndId} = do
+  let qKey = (server, sndId)
  maybe (newMsgQueue qKey) pure =<< TM.lookup qKey (smpQueueMsgQueues c)
  where
    newMsgQueue qKey = do
@@ -596,14 +832,14 @@ getPendingMsgQ c connId SndQueue {server, sndId} = do

 runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> m ()
 runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandshake} sq = do
-  mq <- atomically $ getPendingMsgQ c connId sq
+  mq <- atomically $ getPendingMsgQ c sq
  ri <- asks $ messageRetryInterval . config
  forever $ do
    atomically $ endAgentOperation c AOSndNetwork
+    atomically $ throwWhenInactive c
    msgId <- atomically $ readTQueue mq
-    atomically $ do
-      beginAgentOperation c AOSndNetwork
-      endAgentOperation c AOMsgDelivery
+    atomically $ beginAgentOperation c AOSndNetwork
+    atomically $ endAgentOperation c AOMsgDelivery
    let mId = unId msgId
    E.try (withStore c $ \db -> getPendingMsgData db connId msgId) >>= \case
      Left (e :: E.SomeException) ->
@@ -620,7 +856,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                SMP SMP.QUOTA -> case msgType of
                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                  AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
-                  _ -> retrySending loop
+                  _ -> retrySndOp c loop
                SMP SMP.AUTH -> case msgType of
                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                  AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
@@ -629,7 +865,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                    -- 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 (retrySending loop)
+                      ifM (msgExpired helloTimeout) connErr (retrySndOp c loop)
                    where
                      connErr = case rq_ of
                        -- party initiating connection
@@ -643,7 +879,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                  -- the message sending would be retried
                  | temporaryAgentError e || e == BROKER HOST -> do
                    let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
-                    ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySending loop)
+                    ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndOp c loop)
                  | otherwise -> notifyDel msgId err
              where
                msgExpired timeoutSel = do
@@ -670,14 +906,15 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                      -- If initiating party were to send CON to the user without waiting for reply HELLO (to reduce handshake time),
                      -- it would lead to the non-deterministic internal ID of the first sent message, at to some other race conditions,
                      -- because it can be sent before HELLO is received
-                      -- With `status == Aclive` condition, CON is sent here only by the accepting party, that previously received HELLO
+                      -- With `status == Active` condition, CON is sent here only by the accepting party, that previously received HELLO
                      when (status == Active) $ notify CON
                    -- Party joining connection sends REPLY after HELLO in v1,
                    -- it is an error to send REPLY in duplexHandshake mode (v2),
                    -- and this branch should never be reached as receive is created before the confirmation,
                    -- so the condition is not necessary here, strictly speaking.
                    _ -> unless (duplexHandshake == Just True) $ do
-                      qInfo <- createReplyQueue c cData sq
+                      srv <- getSMPServer c
+                      qInfo <- createReplyQueue c cData sq srv
                      void . enqueueMessage c cData sq SMP.noMsgFlags $ REPLY [qInfo]
                AM_A_MSG_ -> notify $ SENT mId
                _ -> pure ()
@@ -690,11 +927,14 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
    notifyDel :: InternalId -> ACommand 'Agent -> m ()
    notifyDel msgId cmd = notify cmd >> delMsg msgId
    connError msgId = notifyDel msgId . ERR . CONN
-    retrySending loop = do
-      -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
-      atomically $ endAgentOperation c AOSndNetwork
-      atomically $ beginAgentOperation c AOSndNetwork
-      loop
+
+retrySndOp :: AgentMonad m => AgentClient -> m () -> m ()
+retrySndOp c loop = do
+  -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
+  atomically $ endAgentOperation c AOSndNetwork
+  atomically $ throwWhenInactive c
+  atomically $ beginAgentOperation c AOSndNetwork
+  loop

 ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
 ackMessage' c connId msgId = do
@@ -703,16 +943,21 @@ ackMessage' c connId msgId = do
    SomeConn _ (RcvConnection _ rq) -> ack rq
    SomeConn _ (SndConnection _ _) -> throwError $ CONN SIMPLEX
    SomeConn _ (ContactConnection _ _) -> throwError $ CMD PROHIBITED
+    SomeConn _ (NewConnection _) -> throwError $ CMD PROHIBITED
  where
    ack :: RcvQueue -> m ()
    ack rq = do
      let mId = InternalId msgId
      srvMsgId <- withStore c $ \db -> setMsgUserAck db connId mId
-      sendAck c rq srvMsgId `catchError` \case
-        SMP SMP.NO_MSG -> pure ()
-        e -> throwError e
+      ackQueueMessage c rq srvMsgId
      withStore' c $ \db -> deleteMsg db connId mId

+ackQueueMessage :: AgentMonad m => AgentClient -> RcvQueue -> SMP.MsgId -> m ()
+ackQueueMessage c rq srvMsgId =
+  sendAck c rq srvMsgId `catchError` \case
+    SMP SMP.NO_MSG -> pure ()
+    e -> throwError e
+
 -- | Suspend SMP agent connection (OFF command) in Reader monad
 suspendConnection' :: AgentMonad m => AgentClient -> ConnId -> m ()
 suspendConnection' c connId =
@@ -721,6 +966,7 @@ suspendConnection' c connId =
    SomeConn _ (RcvConnection _ rq) -> suspendQueue c rq
    SomeConn _ (ContactConnection _ rq) -> suspendQueue c rq
    SomeConn _ (SndConnection _ _) -> throwError $ CONN SIMPLEX
+    SomeConn _ (NewConnection _) -> throwError $ CMD PROHIBITED

 -- | Delete SMP agent connection (DEL command) in Reader monad
 deleteConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
@@ -730,6 +976,7 @@ deleteConnection' c connId =
    SomeConn _ (RcvConnection _ rq) -> delete rq
    SomeConn _ (ContactConnection _ rq) -> delete rq
    SomeConn _ (SndConnection _ _) -> withStore' c (`deleteConn` connId)
+    SomeConn _ (NewConnection _) -> withStore' c (`deleteConn` connId)
  where
    delete :: RcvQueue -> m ()
    delete rq = do
@@ -748,6 +995,7 @@ getConnectionServers' c connId = connServers <$> withStore c (`getConn` connId)
      SomeConn _ (SndConnection _ SndQueue {server}) -> ConnectionStats {rcvServers = [], sndServers = [server]}
      SomeConn _ (DuplexConnection _ RcvQueue {server = s1} SndQueue {server = s2}) -> ConnectionStats {rcvServers = [s1], sndServers = [s2]}
      SomeConn _ (ContactConnection _ RcvQueue {server}) -> ConnectionStats {rcvServers = [server], sndServers = []}
+      SomeConn _ (NewConnection _) -> ConnectionStats {rcvServers = [], sndServers = []}

 -- | Change servers to be used for creating new queues, in Reader monad
 setSMPServers' :: AgentMonad m => AgentClient -> NonEmpty SMPServer -> m ()
@@ -765,11 +1013,11 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
        (Just tknId, Nothing)
          | savedDeviceToken == suppliedDeviceToken ->
            when (ntfTknStatus == NTRegistered) (registerToken tkn) $> NTRegistered
-          | otherwise -> replaceToken tknId $> NTRegistered
+          | otherwise -> replaceToken tknId
        (Just tknId, Just (NTAVerify code))
          | savedDeviceToken == suppliedDeviceToken ->
            t tkn (NTActive, Just NTACheck) $ agentNtfVerifyToken c tknId tkn code
-          | otherwise -> replaceToken tknId $> NTRegistered
+          | otherwise -> replaceToken tknId
        (Just tknId, Just NTACheck)
          | savedDeviceToken == suppliedDeviceToken -> do
            ns <- asks ntfSupervisor
@@ -781,7 +1029,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
              when (suppliedNtfMode == NMPeriodic && savedNtfMode == NMInstant) $ deleteNtfSubs c NSCDelete
            pure ntfTknStatus -- TODO
            -- agentNtfCheckToken c tknId tkn >>= \case
-          | otherwise -> replaceToken tknId $> NTRegistered
+          | otherwise -> replaceToken tknId
        (Just tknId, Just NTADelete) -> do
          agentNtfDeleteToken c tknId tkn
          withStore' c (`removeNtfToken` tkn)
@@ -792,13 +1040,27 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
      withStore' c $ \db -> updateNtfMode db tkn suppliedNtfMode
      pure status
      where
-        replaceToken :: NtfTokenId -> m ()
+        replaceToken :: NtfTokenId -> m NtfTknStatus
        replaceToken tknId = do
-          agentNtfReplaceToken c tknId tkn suppliedDeviceToken
-          withStore' c $ \db -> updateDeviceToken db tkn suppliedDeviceToken
          ns <- asks ntfSupervisor
-          atomically $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}
-    _ ->
+          tryReplace ns `catchError` \e ->
+            if temporaryAgentError e || e == BROKER HOST
+              then throwError e
+              else do
+                withStore' c $ \db -> removeNtfToken db tkn
+                atomically $ nsRemoveNtfToken ns
+                createToken
+          where
+            tryReplace ns = do
+              agentNtfReplaceToken c tknId tkn suppliedDeviceToken
+              withStore' c $ \db -> updateDeviceToken db tkn suppliedDeviceToken
+              atomically $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}
+              pure NTRegistered
+    _ -> createToken
+  where
+    t tkn = withToken c tkn Nothing
+    createToken :: m NtfTknStatus
+    createToken =
      getNtfServer c >>= \case
        Just ntfServer ->
          asks (cmdSignAlg . config) >>= \case
@@ -810,8 +1072,6 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
              registerToken tkn
              pure NTRegistered
        _ -> throwError $ CMD PROHIBITED
-  where
-    t tkn = withToken c tkn Nothing
    registerToken :: NtfToken -> m ()
    registerToken tkn@NtfToken {ntfPubKey, ntfDhKeys = (pubDhKey, privDhKey)} = do
      (tknId, srvPubDhKey) <- agentNtfRegisterToken c tkn ntfPubKey pubDhKey
@@ -928,7 +1188,7 @@ sendNtfConnCommands c cmd = do
  ns <- asks ntfSupervisor
  connIds <- atomically $ getSubscriptions c
  forM_ connIds $ \connId -> do
-    withStore' c (\db -> getConnData db connId) >>= \case
+    withStore' c (`getConnData` connId) >>= \case
      Just (ConnData {enableNtfs}, _) ->
        when enableNtfs . atomically $ writeTBQueue (ntfSubQ ns) (connId, cmd)
      _ ->
@@ -971,20 +1231,32 @@ suspendAgent' c@AgentClient {agentState = as} maxDelay = do
      -- unsafeIOToSTM $ putStrLn $ "in timeout: suspendSendingAndDatabase"
      suspendSendingAndDatabase c

+execAgentStoreSQL' :: AgentMonad m => AgentClient -> Text -> m [Text]
+execAgentStoreSQL' c sql = withStore' c (`execSQL` sql)
+
 getSMPServer :: AgentMonad m => AgentClient -> m SMPServer
-getSMPServer c = do
-  smpServers <- readTVarIO $ smpServers c
-  case smpServers of
-    srv :| [] -> pure srv
-    servers -> do
-      gen <- asks randomServer
-      atomically . stateTVar gen $
-        first (servers L.!!) . randomR (0, L.length servers - 1)
+getSMPServer c = readTVarIO (smpServers c) >>= pickServer
+
+pickServer :: AgentMonad m => NonEmpty SMPServer -> m SMPServer
+pickServer = \case
+  srv :| [] -> pure srv
+  servers -> do
+    gen <- asks randomServer
+    atomically $ (servers L.!!) <$> stateTVar gen (randomR (0, L.length servers - 1))
+
+getNextSMPServer :: AgentMonad m => AgentClient -> [SMPServer] -> m SMPServer
+getNextSMPServer c usedSrvs = do
+  srvs <- readTVarIO $ smpServers c
+  case L.nonEmpty $ deleteFirstsBy sameAddr (L.toList srvs) usedSrvs of
+    Just srvs' -> pickServer srvs'
+    _ -> pickServer srvs
+  where
+    sameAddr (SMPServer host port _) (SMPServer host' port' _) = host == host' && port == port'

 subscriber :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
 subscriber c@AgentClient {msgQ} = forever $ do
  t <- atomically $ readTBQueue msgQ
-  agentOperationBracket c AORcvNetwork $
+  agentOperationBracket c AORcvNetwork waitUntilActive $
    withAgentLock c (runExceptT $ processSMPTransmission c t) >>= \case
      Left e -> liftIO $ print e
      Right _ -> return ()
@@ -1020,8 +1292,8 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                (SMP.PHEmpty, AgentMsgEnvelope _ encAgentMsg) ->
                  tryError agentClientMsg >>= \case
                    Right (Just (msgId, msgMeta, aMessage)) -> case aMessage of
-                      HELLO -> helloMsg >> ack >> withStore' c (\db -> deleteMsg db connId msgId)
-                      REPLY cReq -> replyMsg cReq >> ack >> withStore' c (\db -> deleteMsg db connId msgId)
+                      HELLO -> helloMsg >> ackDel msgId
+                      REPLY cReq -> replyMsg cReq >> ackDel msgId
                      -- note that there is no ACK sent for A_MSG, it is sent with agent's user ACK command
                      A_MSG body -> do
                        logServer "<--" c srv rId "MSG <MSG>"
@@ -1030,9 +1302,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                    Left e@(AGENT A_DUPLICATE) -> do
                      withStore' c (\db -> getLastMsg db connId srvMsgId) >>= \case
                        Just RcvMsg {internalId, msgMeta, msgBody = agentMsgBody, userAck}
-                          | userAck -> do
-                            ack
-                            withStore' c $ \db -> deleteMsg db connId internalId
+                          | userAck -> ackDel internalId
                          | otherwise -> do
                            liftEither (parse smpP (AGENT A_MESSAGE) agentMsgBody) >>= \case
                              AgentMessage _ (A_MSG body) -> do
@@ -1063,10 +1333,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
            _ -> prohibited >> ack
          where
            ack :: m ()
-            ack =
-              sendAck c rq srvMsgId `catchError` \case
-                SMP SMP.NO_MSG -> pure ()
-                e -> throwError e
+            ack = enqueueCmd $ ICAck rId srvMsgId
+            ackDel :: InternalId -> m ()
+            ackDel = enqueueCmd . ICAckDel rId srvMsgId
            handleNotifyAck :: m () -> m ()
            handleNotifyAck m = m `catchError` \e -> notify (ERR e) >> ack
        SMP.END ->
@@ -1092,6 +1361,9 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
        prohibited :: m ()
        prohibited = notify . ERR $ AGENT A_PROHIBITED

+        enqueueCmd :: InternalCommand -> m ()
+        enqueueCmd = enqueueCommand c "" connId (Just srv) . AInternalCommand
+
        decryptClientMessage :: C.DhSecretX25519 -> SMP.ClientMsgEnvelope -> m (SMP.PrivHeader, AgentMsgEnvelope)
        decryptClientMessage e2eDh SMP.ClientMsgEnvelope {cmNonce, cmEncBody} = do
          clientMsg <- agentCbDecrypt e2eDh cmNonce cmEncBody
@@ -1119,7 +1391,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
            New -> case (conn, e2eEncryption) of
              -- party initiating connection
              (RcvConnection {}, Just e2eSndParams) -> do
-                (pk1, rcDHRs) <- withStore c $ (`getRatchetX3dhKeys` connId)
+                (pk1, rcDHRs) <- withStore c (`getRatchetX3dhKeys` connId)
                let rc = CR.initRcvRatchet rcDHRs $ CR.x3dhRcv pk1 rcDHRs e2eSndParams
                (agentMsgBody_, rc', skipped) <- liftError cryptoError $ CR.rcDecrypt rc M.empty encConnInfo
                case (agentMsgBody_, skipped) of
@@ -1142,12 +1414,13 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                      queueServer (SMPQueueInfo _ SMPQueueAddress {smpServer}) = smpServer
                  _ -> prohibited
              -- party accepting connection
-              (DuplexConnection _ _ sq, Nothing) -> do
+              (DuplexConnection _ RcvQueue {smpClientVersion = v'} _, Nothing) -> do
                withStore c (\db -> runExceptT $ agentRatchetDecrypt db connId encConnInfo) >>= parseMessage >>= \case
                  AgentConnInfo connInfo -> do
                    notify $ INFO connInfo
-                    processConfirmation c rq $ SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = [], smpClientVersion}
-                    when (duplexHandshake == Just True) $ enqueueDuplexHello sq
+                    let dhSecret = C.dh' e2ePubKey e2ePrivKey
+                    withStore' c $ \db -> setRcvQueueConfirmedE2E db rq dhSecret $ min v' smpClientVersion
+                    enqueueCmd $ ICDuplexSecure rId senderKey
                  _ -> prohibited
              _ -> prohibited
            _ -> prohibited
@@ -1217,8 +1490,8 @@ connectReplyQueues c cData@ConnData {connId} ownConnInfo (qInfo :| _) = do
      withStore c $ \db -> upgradeRcvConnToDuplex db connId sq
      enqueueConfirmation c cData sq ownConnInfo Nothing

-confirmQueue :: forall m. AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
-confirmQueue (Compatible agentVersion) c cData@ConnData {connId} sq connInfo e2eEncryption = do
+confirmQueue :: forall m. AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> SMPServer -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
+confirmQueue (Compatible agentVersion) c cData@ConnData {connId} sq srv connInfo e2eEncryption = do
  aMessage <- mkAgentMessage agentVersion
  msg <- mkConfirmation aMessage
  sendConfirmation c sq msg
@@ -1232,14 +1505,14 @@ confirmQueue (Compatible agentVersion) c cData@ConnData {connId} sq connInfo e2e
    mkAgentMessage :: Version -> m AgentMessage
    mkAgentMessage 1 = pure $ AgentConnInfo connInfo
    mkAgentMessage _ = do
-      qInfo <- createReplyQueue c cData sq
+      qInfo <- createReplyQueue c cData sq srv
      pure $ AgentConnInfoReply (qInfo :| []) connInfo

 enqueueConfirmation :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
 enqueueConfirmation c cData@ConnData {connId, connAgentVersion} sq connInfo e2eEncryption = do
  resumeMsgDelivery c cData sq
  msgId <- storeConfirmation
-  queuePendingMsgs c connId sq [msgId]
+  queuePendingMsgs c sq [msgId]
  where
    storeConfirmation :: m InternalId
    storeConfirmation = withStore c $ \db -> runExceptT $ do
@@ -52,12 +52,14 @@ module Simplex.Messaging.Agent.Client
    logServer,
    removeSubscription,
    hasActiveSubscription,
-    agentDbPath,
+    agentClientStore,
    AgentOperation (..),
    AgentOpState (..),
    AgentState (..),
    agentOperations,
    agentOperationBracket,
+    waitUntilActive,
+    throwWhenInactive,
    beginAgentOperation,
    endAgentOperation,
    suspendSendingAndDatabase,
@@ -72,7 +74,8 @@ where

 import Control.Concurrent (forkIO)
 import Control.Concurrent.Async (Async, uninterruptibleCancel)
-import Control.Concurrent.STM (retry, stateTVar)
+import Control.Concurrent.STM (retry, stateTVar, throwSTM)
+import Control.Exception (AsyncException (..))
 import Control.Logger.Simple
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
@@ -90,7 +93,6 @@ import Data.Maybe (listToMaybe)
 import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text.Encoding
-import Data.Tuple (swap)
 import Data.Word (Word16)
 import qualified Database.SQLite.Simple as DB
 import Simplex.Messaging.Agent.Env.SQLite
@@ -131,6 +133,8 @@ import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.TMap2 (TMap2)
+import qualified Simplex.Messaging.TMap2 as TM2
 import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util
 import Simplex.Messaging.Version
@@ -155,13 +159,15 @@ data AgentClient = AgentClient
    ntfServers :: TVar [NtfServer],
    ntfClients :: TMap NtfServer NtfClientVar,
    useNetworkConfig :: TVar NetworkConfig,
-    subscrSrvrs :: TMap SMPServer (TMap ConnId RcvQueue),
-    pendingSubscrSrvrs :: TMap SMPServer (TMap ConnId RcvQueue),
    subscrConns :: TVar (Set ConnId),
-    activeSubscrConns :: TMap ConnId SMPServer,
+    activeSubs :: TMap2 SMPServer ConnId RcvQueue,
+    pendingSubs :: TMap2 SMPServer ConnId RcvQueue,
    connMsgsQueued :: TMap ConnId Bool,
-    smpQueueMsgQueues :: TMap (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId),
-    smpQueueMsgDeliveries :: TMap (ConnId, SMPServer, SMP.SenderId) (Async ()),
+    smpQueueMsgQueues :: TMap (SMPServer, SMP.SenderId) (TQueue InternalId),
+    smpQueueMsgDeliveries :: TMap (SMPServer, SMP.SenderId) (Async ()),
+    connCmdsQueued :: TMap ConnId Bool,
+    asyncCmdQueues :: TMap (Maybe SMPServer) (TQueue AsyncCmdId),
+    asyncCmdProcesses :: TMap (Maybe SMPServer) (Async ()),
    ntfNetworkOp :: TVar AgentOpState,
    rcvNetworkOp :: TVar AgentOpState,
    msgDeliveryOp :: TVar AgentOpState,
@@ -207,13 +213,15 @@ newAgentClient InitialAgentServers {smp, ntf, netCfg} agentEnv = do
  ntfServers <- newTVar ntf
  ntfClients <- TM.empty
  useNetworkConfig <- newTVar netCfg
-  subscrSrvrs <- TM.empty
-  pendingSubscrSrvrs <- TM.empty
  subscrConns <- newTVar S.empty
-  activeSubscrConns <- TM.empty
+  activeSubs <- TM2.empty
+  pendingSubs <- TM2.empty
  connMsgsQueued <- TM.empty
  smpQueueMsgQueues <- TM.empty
  smpQueueMsgDeliveries <- TM.empty
+  connCmdsQueued <- TM.empty
+  asyncCmdQueues <- TM.empty
+  asyncCmdProcesses <- TM.empty
  ntfNetworkOp <- newTVar $ AgentOpState False 0
  rcvNetworkOp <- newTVar $ AgentOpState False 0
  msgDeliveryOp <- newTVar $ AgentOpState False 0
@@ -225,10 +233,10 @@ newAgentClient InitialAgentServers {smp, ntf, netCfg} agentEnv = do
  asyncClients <- newTVar []
  clientId <- stateTVar (clientCounter agentEnv) $ \i -> let i' = i + 1 in (i', i')
  lock <- newTMVar ()
-  return AgentClient {active, rcvQ, subQ, msgQ, smpServers, smpClients, ntfServers, ntfClients, useNetworkConfig, subscrSrvrs, pendingSubscrSrvrs, subscrConns, activeSubscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, ntfNetworkOp, rcvNetworkOp, msgDeliveryOp, sndNetworkOp, databaseOp, agentState, getMsgLocks, reconnections, asyncClients, clientId, agentEnv, lock}
+  return AgentClient {active, rcvQ, subQ, msgQ, smpServers, smpClients, ntfServers, ntfClients, useNetworkConfig, subscrConns, activeSubs, pendingSubs, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, connCmdsQueued, asyncCmdQueues, asyncCmdProcesses, ntfNetworkOp, rcvNetworkOp, msgDeliveryOp, sndNetworkOp, databaseOp, agentState, getMsgLocks, reconnections, asyncClients, clientId, agentEnv, lock}

-agentDbPath :: AgentClient -> FilePath
-agentDbPath AgentClient {agentEnv = Env {store = SQLiteStore {dbFilePath}}} = dbFilePath
+agentClientStore :: AgentClient -> SQLiteStore
+agentClientStore AgentClient {agentEnv = Env {store}} = store

 class ProtocolServerClient msg where
  getProtocolServerClient :: AgentMonad m => AgentClient -> ProtoServer msg -> m (ProtocolClient msg)
@@ -264,26 +272,18 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} srv = do
        removeClientAndSubs :: IO (Maybe (Map ConnId RcvQueue))
        removeClientAndSubs = atomically $ do
          TM.delete srv smpClients
-          TM.lookupDelete srv (subscrSrvrs c) >>= mapM updateSubs
+          TM2.lookupDelete1 srv (activeSubs c) >>= mapM updateSubs
          where
            updateSubs cVar = do
-              cs <- readTVar cVar
-              modifyTVar' (activeSubscrConns c) (`M.withoutKeys` M.keysSet cs)
-              addPendingSubs cVar cs
-              pure cs
-
-            addPendingSubs cVar cs = do
-              let ps = pendingSubscrSrvrs c
-              TM.lookup srv ps >>= \case
-                Just v -> TM.union cs v
-                _ -> TM.insert srv cVar ps
+              TM2.insert1 srv cVar $ pendingSubs c
+              readTVar cVar

        serverDown :: Map ConnId RcvQueue -> IO ()
-        serverDown cs = unless (M.null cs) $
-          whenM (readTVarIO active) $ do
-            let conns = M.keys cs
-            notifySub "" $ hostEvent DISCONNECT client
-            unless (null conns) . notifySub "" $ DOWN srv conns
+        serverDown cs = whenM (readTVarIO active) $ do
+          notifySub "" $ hostEvent DISCONNECT client
+          let conns = M.keys cs
+          unless (null conns) $ do
+            notifySub "" $ DOWN srv conns
            atomically $ mapM_ (releaseGetLock c) cs
            unliftIO u reconnectServer

@@ -301,7 +301,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} srv = do
    reconnectClient :: m ()
    reconnectClient =
      withAgentLock c $
-        atomically (TM.lookup srv (pendingSubscrSrvrs c) >>= mapM readTVar)
+        atomically (TM2.lookup1 srv (pendingSubs c) >>= mapM readTVar)
          >>= mapM_ resubscribe
      where
        resubscribe :: Map ConnId RcvQueue -> m ()
@@ -407,17 +407,25 @@ closeAgentClient c = liftIO $ do
  cancelActions $ reconnections c
  cancelActions $ asyncClients c
  cancelActions $ smpQueueMsgDeliveries c
-  clear subscrSrvrs
-  clear pendingSubscrSrvrs
+  cancelActions $ asyncCmdProcesses c
+  atomically . TM2.clear $ activeSubs c
+  atomically . TM2.clear $ pendingSubs c
  clear subscrConns
-  clear activeSubscrConns
  clear connMsgsQueued
  clear smpQueueMsgQueues
+  clear connCmdsQueued
+  clear asyncCmdQueues
  clear getMsgLocks
  where
    clear :: Monoid m => (AgentClient -> TVar m) -> IO ()
    clear sel = atomically $ writeTVar (sel c) mempty

+waitUntilActive :: AgentClient -> STM ()
+waitUntilActive c = unlessM (readTVar $ active c) retry
+
+throwWhenInactive :: AgentClient -> STM ()
+throwWhenInactive c = unlessM (readTVar $ active c) $ throwSTM ThreadKilled
+
 closeProtocolServerClients :: AgentClient -> (AgentClient -> TMap (ProtoServer msg) (ClientVar msg)) -> IO ()
 closeProtocolServerClients c clientsSel =
  readTVarIO cs >>= mapM_ (forkIO . closeClient) >> atomically (writeTVar cs M.empty)
@@ -430,7 +438,7 @@ closeProtocolServerClients c clientsSel =
        _ -> pure ()

 cancelActions :: (Foldable f, Monoid (f (Async ()))) => TVar (f (Async ())) -> IO ()
-cancelActions as = readTVarIO as >>= mapM_ uninterruptibleCancel >> atomically (writeTVar as mempty)
+cancelActions as = readTVarIO as >>= mapM_ (forkIO . uninterruptibleCancel) >> atomically (writeTVar as mempty)

 withAgentLock :: MonadUnliftIO m => AgentClient -> m a -> m a
 withAgentLock AgentClient {lock} =
@@ -514,17 +522,17 @@ subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
  whenM (atomically . TM.member (server, rcvId) $ getMsgLocks c) . throwError $ CMD PROHIBITED
  atomically $ do
    modifyTVar (subscrConns c) $ S.insert connId
-    addPendingSubscription c rq connId
+    TM2.insert server connId rq $ pendingSubs c
  withLogClient c server rcvId "SUB" $ \smp ->
    liftIO (runExceptT (subscribeSMPQueue smp rcvPrivateKey rcvId) >>= processSubResult c rq connId)
      >>= either throwError pure

 processSubResult :: AgentClient -> RcvQueue -> ConnId -> Either ProtocolClientError () -> IO (Either ProtocolClientError ())
-processSubResult c rq@RcvQueue {server} connId r = do
+processSubResult c rq connId r = do
  case r of
    Left e ->
      atomically . unless (temporaryClientError e) $
-        removePendingSubscription c server connId
+        TM2.delete connId (pendingSubs c)
    _ -> addSubscription c rq connId
  pure r

@@ -544,9 +552,9 @@ temporaryAgentError = \case
 subscribeQueues :: AgentMonad m => AgentClient -> SMPServer -> Map ConnId RcvQueue -> m (Maybe SMPClient, Map ConnId (Either AgentErrorType ()))
 subscribeQueues c srv qs = do
  (errs, qs_) <- partitionEithers <$> mapM checkQueue (M.assocs qs)
-  forM_ qs_ $ \q -> atomically $ do
-    modifyTVar (subscrConns c) . S.insert $ fst q
-    uncurry (addPendingSubscription c) $ swap q
+  forM_ qs_ $ \(connId, rq@RcvQueue {server}) -> atomically $ do
+    modifyTVar (subscrConns c) $ S.insert connId
+    TM2.insert server connId rq $ pendingSubs c
  case L.nonEmpty qs_ of
    Just qs' -> do
      smp_ <- tryError (getSMPServerClient c srv)
@@ -559,7 +567,7 @@ subscribeQueues c srv qs = do
            liftIO $ zip qs_ . L.toList <$> subscribeSMPQueues smp qs2
          forM_ rs' $ \((connId, rq), r) -> liftIO $ processSubResult c rq connId r
          pure $ map (bimap fst (first $ protocolClientError SMP)) rs'
-    _ -> pure $ (Nothing, M.fromList errs)
+    _ -> pure (Nothing, M.fromList errs)
  where
    checkQueue rq@(connId, RcvQueue {rcvId, server}) = do
      prohibited <- atomically . TM.member (server, rcvId) $ getMsgLocks c
@@ -568,35 +576,18 @@ subscribeQueues c srv qs = do

 addSubscription :: MonadIO m => AgentClient -> RcvQueue -> ConnId -> m ()
 addSubscription c rq@RcvQueue {server} connId = atomically $ do
-  TM.insert connId server $ activeSubscrConns c
  modifyTVar (subscrConns c) $ S.insert connId
-  addSubs_ (subscrSrvrs c) rq connId
-  removePendingSubscription c server connId
+  TM2.insert server connId rq $ activeSubs c
+  TM2.delete connId $ pendingSubs c

 hasActiveSubscription :: AgentClient -> ConnId -> STM Bool
-hasActiveSubscription c connId = TM.member connId (activeSubscrConns c)
-
-addPendingSubscription :: AgentClient -> RcvQueue -> ConnId -> STM ()
-addPendingSubscription = addSubs_ . pendingSubscrSrvrs
-
-addSubs_ :: TMap SMPServer (TMap ConnId RcvQueue) -> RcvQueue -> ConnId -> STM ()
-addSubs_ ss rq@RcvQueue {server} connId =
-  TM.lookup server ss >>= \case
-    Just m -> TM.insert connId rq m
-    _ -> TM.singleton connId rq >>= \m -> TM.insert server m ss
+hasActiveSubscription c connId = TM2.member connId $ activeSubs c

 removeSubscription :: AgentClient -> ConnId -> STM ()
 removeSubscription c connId = do
  modifyTVar (subscrConns c) $ S.delete connId
-  server_ <- TM.lookupDelete connId $ activeSubscrConns c
-  mapM_ (\server -> removeSubs_ (subscrSrvrs c) server connId) server_
-
-removePendingSubscription :: AgentClient -> SMPServer -> ConnId -> STM ()
-removePendingSubscription = removeSubs_ . pendingSubscrSrvrs
-
-removeSubs_ :: TMap SMPServer (TMap ConnId RcvQueue) -> SMPServer -> ConnId -> STM ()
-removeSubs_ ss server connId =
-  TM.lookup server ss >>= mapM_ (TM.delete connId)
+  TM2.delete connId $ activeSubs c
+  TM2.delete connId $ pendingSubs c

 getSubscriptions :: AgentClient -> STM (Set ConnId)
 getSubscriptions = readTVar . subscrConns
@@ -823,10 +814,10 @@ beginAgentOperation c op = do
  -- unsafeIOToSTM $ putStrLn $ "beginOperation! " <> show op <> " " <> show (opsInProgress s + 1)
  writeTVar opVar $! s {opsInProgress = opsInProgress s + 1}

-agentOperationBracket :: MonadUnliftIO m => AgentClient -> AgentOperation -> m a -> m a
-agentOperationBracket c op action =
+agentOperationBracket :: MonadUnliftIO m => AgentClient -> AgentOperation -> (AgentClient -> STM ()) -> m a -> m a
+agentOperationBracket c op check action =
  E.bracket
-    (atomically $ beginAgentOperation c op)
+    (atomically (check c) >> atomically (beginAgentOperation c op))
    (\_ -> atomically $ endAgentOperation c op)
    (const action)

@@ -836,7 +827,7 @@ withStore' c action = withStore c $ fmap Right . action
 withStore :: AgentMonad m => AgentClient -> (DB.Connection -> IO (Either StoreError a)) -> m a
 withStore c action = do
  st <- asks store
-  liftEitherError storeError . agentOperationBracket c AODatabase $
+  liftEitherError storeError . agentOperationBracket c AODatabase (\_ -> pure ()) $
    withTransaction st action `E.catch` handleInternal
  where
    handleInternal :: E.SomeException -> IO (Either StoreError a)
@@ -3,6 +3,7 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE RankNTypes #-}
@@ -12,12 +13,15 @@
 module Simplex.Messaging.Agent.Env.SQLite
  ( AgentMonad,
    AgentConfig (..),
+    AgentDatabase (..),
+    databaseFile,
    InitialAgentServers (..),
    NetworkConfig (..),
    defaultAgentConfig,
    defaultReconnectInterval,
    Env (..),
    newSMPAgentEnv,
+    createAgentStore,
    NtfSupervisor (..),
    NtfSupervisorCommand (..),
  )
@@ -59,12 +63,21 @@ data InitialAgentServers = InitialAgentServers
    netCfg :: NetworkConfig
  }

+data AgentDatabase
+  = AgentDB SQLiteStore
+  | AgentDBFile {dbFile :: FilePath, dbKey :: String}
+
+databaseFile :: AgentDatabase -> FilePath
+databaseFile = \case
+  AgentDB SQLiteStore {dbFilePath} -> dbFilePath
+  AgentDBFile {dbFile} -> dbFile
+
 data AgentConfig = AgentConfig
  { tcpPort :: ServiceName,
    cmdSignAlg :: C.SignAlg,
    connIdBytes :: Int,
    tbqSize :: Natural,
-    dbFile :: FilePath,
+    database :: AgentDatabase,
    yesToMigrations :: Bool,
    smpCfg :: ProtocolClientConfig,
    ntfCfg :: ProtocolClientConfig,
@@ -107,7 +120,7 @@ defaultAgentConfig =
      cmdSignAlg = C.SignAlg C.SEd448,
      connIdBytes = 12,
      tbqSize = 64,
-      dbFile = "smp-agent.db",
+      database = AgentDBFile {dbFile = "smp-agent.db", dbKey = ""},
      yesToMigrations = False,
      smpCfg = defaultClientConfig {defaultTransport = (show defaultSMPPort, transport @TLS)},
      ntfCfg = defaultClientConfig {defaultTransport = ("443", transport @TLS)},
@@ -139,14 +152,19 @@ data Env = Env
  }

 newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env
-newSMPAgentEnv config@AgentConfig {dbFile, yesToMigrations} = do
+newSMPAgentEnv config@AgentConfig {database, yesToMigrations} = do
  idsDrg <- newTVarIO =<< drgNew
-  store <- liftIO $ createSQLiteStore dbFile Migrations.app yesToMigrations
+  store <- case database of
+    AgentDB st -> pure st
+    AgentDBFile {dbFile, dbKey} -> liftIO $ createAgentStore dbFile dbKey yesToMigrations
  clientCounter <- newTVarIO 0
  randomServer <- newTVarIO =<< liftIO newStdGen
  ntfSupervisor <- atomically . newNtfSubSupervisor $ tbqSize config
  return Env {config, store, idsDrg, clientCounter, randomServer, ntfSupervisor}

+createAgentStore :: FilePath -> String -> Bool -> IO SQLiteStore
+createAgentStore dbFilePath dbKey = createSQLiteStore dbFilePath dbKey Migrations.app
+
 data NtfSupervisor = NtfSupervisor
  { ntfTkn :: TVar (Maybe NtfToken),
    ntfSubQ :: TBQueue (ConnId, NtfSupervisorCommand),
@@ -55,7 +55,7 @@ runNtfSupervisor c = do
  ns <- asks ntfSupervisor
  forever $ do
    cmd@(connId, _) <- atomically . readTBQueue $ ntfSubQ ns
-    handleError connId . agentOperationBracket c AONtfNetwork $
+    handleError connId . agentOperationBracket c AONtfNetwork waitUntilActive $
      runExceptT (processNtfSub c cmd) >>= \case
        Left e -> notifyErr connId e
        Right _ -> return ()
@@ -162,7 +162,7 @@ runNtfWorker c srv doWork = do
  delay <- asks $ ntfWorkerDelay . config
  forever $ do
    void . atomically $ readTMVar doWork
-    agentOperationBracket c AONtfNetwork runNtfOperation
+    agentOperationBracket c AONtfNetwork throwWhenInactive runNtfOperation
    threadDelay delay
  where
    runNtfOperation :: m ()
@@ -238,7 +238,7 @@ runNtfSMPWorker c srv doWork = do
  delay <- asks $ ntfSMPWorkerDelay . config
  forever $ do
    void . atomically $ readTMVar doWork
-    agentOperationBracket c AONtfNetwork runNtfSMPOperation
+    agentOperationBracket c AONtfNetwork throwWhenInactive runNtfSMPOperation
    threadDelay delay
  where
    runNtfSMPOperation = do
@@ -305,6 +305,7 @@ retryOnError c name loop done e = do
  where
    retryLoop = do
      atomically $ endAgentOperation c AONtfNetwork
+      atomically $ throwWhenInactive c
      atomically $ beginAgentOperation c AONtfNetwork
      loop

@@ -40,9 +40,13 @@ module Simplex.Messaging.Agent.Protocol
    -- * SMP agent protocol types
    ConnInfo,
    ACommand (..),
+    ACommandTag (..),
+    aCommandTag,
    ACmd (..),
+    ACmdTag (..),
    AParty (..),
    SAParty (..),
+    APartyI (..),
    MsgHash,
    MsgMeta (..),
    ConnectionStats (..),
@@ -92,6 +96,8 @@ module Simplex.Messaging.Agent.Protocol
    serializeCommand,
    connMode,
    connMode',
+    networkCommandP,
+    dbCommandP,
    commandP,
    connModeT,
    serializeQueueStatus,
@@ -99,6 +105,7 @@ module Simplex.Messaging.Agent.Protocol
    agentMessageType,
    extraSMPServerHosts,
    updateSMPServerHosts,
+    checkParty,

    -- * TCP transport functions
    tPut,
@@ -117,7 +124,6 @@ import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Composition ((.:), (.:.))
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.Kind (Type)
@@ -211,7 +217,13 @@ instance TestEquality SAParty where
  testEquality SClient SClient = Just Refl
  testEquality _ _ = Nothing

-data ACmd = forall p. ACmd (SAParty p) (ACommand p)
+class APartyI (p :: AParty) where sAParty :: SAParty p
+
+instance APartyI Agent where sAParty = SAgent
+
+instance APartyI Client where sAParty = SClient
+
+data ACmd = forall p. APartyI p => ACmd (SAParty p) (ACommand p)

 deriving instance Show ACmd

@@ -219,9 +231,9 @@ type ConnInfo = ByteString

 -- | Parameterized type for SMP agent protocol commands and responses from all participants.
 data ACommand (p :: AParty) where
-  NEW :: AConnectionMode -> ACommand Client -- response INV
+  NEW :: Bool -> AConnectionMode -> ACommand Client -- response INV
  INV :: AConnectionRequestUri -> ACommand Agent
-  JOIN :: AConnectionRequestUri -> ConnInfo -> ACommand Client -- response OK
+  JOIN :: Bool -> AConnectionRequestUri -> ConnInfo -> ACommand Client -- response OK
  CONF :: ConfirmationId -> [SMPServer] -> ConnInfo -> ACommand Agent -- ConnInfo is from sender, [SMPServer] will be empty only in v1 handshake
  LET :: ConfirmationId -> ConnInfo -> ACommand Client -- ConnInfo is from client
  REQ :: InvitationId -> L.NonEmpty SMPServer -> ConnInfo -> ACommand Agent -- ConnInfo is from sender
@@ -253,6 +265,75 @@ deriving instance Eq (ACommand p)

 deriving instance Show (ACommand p)

+data ACmdTag = forall p. APartyI p => ACmdTag (SAParty p) (ACommandTag p)
+
+data ACommandTag (p :: AParty) where
+  NEW_ :: ACommandTag Client
+  INV_ :: ACommandTag Agent
+  JOIN_ :: ACommandTag Client
+  CONF_ :: ACommandTag Agent
+  LET_ :: ACommandTag Client
+  REQ_ :: ACommandTag Agent
+  ACPT_ :: ACommandTag Client
+  RJCT_ :: ACommandTag Client
+  INFO_ :: ACommandTag Agent
+  CON_ :: ACommandTag Agent
+  SUB_ :: ACommandTag Client
+  END_ :: ACommandTag Agent
+  CONNECT_ :: ACommandTag Agent
+  DISCONNECT_ :: ACommandTag Agent
+  DOWN_ :: ACommandTag Agent
+  UP_ :: ACommandTag Agent
+  SEND_ :: ACommandTag Client
+  MID_ :: ACommandTag Agent
+  SENT_ :: ACommandTag Agent
+  MERR_ :: ACommandTag Agent
+  MSG_ :: ACommandTag Agent
+  ACK_ :: ACommandTag Client
+  OFF_ :: ACommandTag Client
+  DEL_ :: ACommandTag Client
+  CHK_ :: ACommandTag Client
+  STAT_ :: ACommandTag Agent
+  OK_ :: ACommandTag Agent
+  ERR_ :: ACommandTag Agent
+  SUSPENDED_ :: ACommandTag Agent
+
+deriving instance Eq (ACommandTag p)
+
+deriving instance Show (ACommandTag p)
+
+aCommandTag :: ACommand p -> ACommandTag p
+aCommandTag = \case
+  NEW {} -> NEW_
+  INV _ -> INV_
+  JOIN {} -> JOIN_
+  CONF {} -> CONF_
+  LET {} -> LET_
+  REQ {} -> REQ_
+  ACPT {} -> ACPT_
+  RJCT _ -> RJCT_
+  INFO _ -> INFO_
+  CON -> CON_
+  SUB -> SUB_
+  END -> END_
+  CONNECT {} -> CONNECT_
+  DISCONNECT {} -> DISCONNECT_
+  DOWN {} -> DOWN_
+  UP {} -> UP_
+  SEND {} -> SEND_
+  MID _ -> MID_
+  SENT _ -> SENT_
+  MERR {} -> MERR_
+  MSG {} -> MSG_
+  ACK _ -> ACK_
+  OFF -> OFF_
+  DEL -> DEL_
+  CHK -> CHK_
+  STAT _ -> STAT_
+  OK -> OK_
+  ERR _ -> ERR_
+  SUSPENDED -> SUSPENDED_
+
 data ConnectionStats = ConnectionStats
  { rcvServers :: [SMPServer],
    sndServers :: [SMPServer]
@@ -654,7 +735,7 @@ instance StrEncoding SMPQueueUri where
    where
      query = strEncode . QSP QEscape
      queryParams = [("v", strEncode vr), ("dh", strEncode dhPublicKey)]
-      srvParam = [("srv", strEncode $ TransportHosts_ hs) | length hs > 0]
+      srvParam = [("srv", strEncode $ TransportHosts_ hs) | not (null hs)]
      hs = L.tail $ host srv
  strP = do
    srv@ProtocolServer {host = h :| host} <- strP <* A.char '/'
@@ -920,58 +1001,129 @@ instance Arbitrary BrokerErrorType where arbitrary = genericArbitraryU

 instance Arbitrary SMPAgentError where arbitrary = genericArbitraryU

+-- | SMP agent command and response parser for commands passed via network (only parses binary length)
+networkCommandP :: Parser ACmd
+networkCommandP = commandP A.takeByteString
+
+-- | SMP agent command and response parser for commands stored in db (fully parses binary bodies)
+dbCommandP :: Parser ACmd
+dbCommandP = commandP $ A.take =<< (A.decimal <* "\n")
+
+instance StrEncoding ACmdTag where
+  strEncode (ACmdTag _ cmd) = strEncode cmd
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "NEW" -> pure $ ACmdTag SClient NEW_
+      "INV" -> pure $ ACmdTag SAgent INV_
+      "JOIN" -> pure $ ACmdTag SClient JOIN_
+      "CONF" -> pure $ ACmdTag SAgent CONF_
+      "LET" -> pure $ ACmdTag SClient LET_
+      "REQ" -> pure $ ACmdTag SAgent REQ_
+      "ACPT" -> pure $ ACmdTag SClient ACPT_
+      "RJCT" -> pure $ ACmdTag SClient RJCT_
+      "INFO" -> pure $ ACmdTag SAgent INFO_
+      "CON" -> pure $ ACmdTag SAgent CON_
+      "SUB" -> pure $ ACmdTag SClient SUB_
+      "END" -> pure $ ACmdTag SAgent END_
+      "CONNECT" -> pure $ ACmdTag SAgent CONNECT_
+      "DISCONNECT" -> pure $ ACmdTag SAgent DISCONNECT_
+      "DOWN" -> pure $ ACmdTag SAgent DOWN_
+      "UP" -> pure $ ACmdTag SAgent UP_
+      "SEND" -> pure $ ACmdTag SClient SEND_
+      "MID" -> pure $ ACmdTag SAgent MID_
+      "SENT" -> pure $ ACmdTag SAgent SENT_
+      "MERR" -> pure $ ACmdTag SAgent MERR_
+      "MSG" -> pure $ ACmdTag SAgent MSG_
+      "ACK" -> pure $ ACmdTag SClient ACK_
+      "OFF" -> pure $ ACmdTag SClient OFF_
+      "DEL" -> pure $ ACmdTag SClient DEL_
+      "CHK" -> pure $ ACmdTag SClient CHK_
+      "STAT" -> pure $ ACmdTag SAgent STAT_
+      "OK" -> pure $ ACmdTag SAgent OK_
+      "ERR" -> pure $ ACmdTag SAgent ERR_
+      "SUSPENDED" -> pure $ ACmdTag SAgent SUSPENDED_
+      _ -> fail "bad ACmdTag"
+
+instance APartyI p => StrEncoding (ACommandTag p) where
+  strEncode = \case
+    NEW_ -> "NEW"
+    INV_ -> "INV"
+    JOIN_ -> "JOIN"
+    CONF_ -> "CONF"
+    LET_ -> "LET"
+    REQ_ -> "REQ"
+    ACPT_ -> "ACPT"
+    RJCT_ -> "RJCT"
+    INFO_ -> "INFO"
+    CON_ -> "CON"
+    SUB_ -> "SUB"
+    END_ -> "END"
+    CONNECT_ -> "CONNECT"
+    DISCONNECT_ -> "DISCONNECT"
+    DOWN_ -> "DOWN"
+    UP_ -> "UP"
+    SEND_ -> "SEND"
+    MID_ -> "MID"
+    SENT_ -> "SENT"
+    MERR_ -> "MERR"
+    MSG_ -> "MSG"
+    ACK_ -> "ACK"
+    OFF_ -> "OFF"
+    DEL_ -> "DEL"
+    CHK_ -> "CHK"
+    STAT_ -> "STAT"
+    OK_ -> "OK"
+    ERR_ -> "ERR"
+    SUSPENDED_ -> "SUSPENDED"
+  strP = (\(ACmdTag _ t) -> checkParty t) <$?> strP
+
+checkParty :: forall t p p'. (APartyI p, APartyI p') => t p' -> Either String (t p)
+checkParty x = case testEquality (sAParty @p) (sAParty @p') of
+  Just Refl -> Right x
+  Nothing -> Left "bad party"
+
 -- | SMP agent command and response parser
-commandP :: Parser ACmd
-commandP =
-  "NEW " *> newCmd
-    <|> "INV " *> invResp
-    <|> "JOIN " *> joinCmd
-    <|> "CONF " *> confMsg
-    <|> "LET " *> letCmd
-    <|> "REQ " *> reqMsg
-    <|> "ACPT " *> acptCmd
-    <|> "RJCT " *> rjctCmd
-    <|> "INFO " *> infoCmd
-    <|> "SUB" $> ACmd SClient SUB
-    <|> "END" $> ACmd SAgent END
-    <|> "CONNECT " *> connectResp
-    <|> "DISCONNECT " *> disconnectResp
-    <|> "DOWN " *> downResp
-    <|> "UP " *> upResp
-    <|> "SEND " *> sendCmd
-    <|> "MID " *> msgIdResp
-    <|> "SENT " *> sentResp
-    <|> "MERR " *> msgErrResp
-    <|> "MSG " *> message
-    <|> "ACK " *> ackCmd
-    <|> "OFF" $> ACmd SClient OFF
-    <|> "DEL" $> ACmd SClient DEL
-    <|> "CHK" $> ACmd SClient CHK
-    <|> "STAT " *> statResp
-    <|> "ERR " *> agentError
-    <|> "CON" $> ACmd SAgent CON
-    <|> "OK" $> ACmd SAgent OK
+commandP :: Parser ByteString -> Parser ACmd
+commandP binaryP =
+  strP
+    >>= \case
+      ACmdTag SClient cmd ->
+        ACmd SClient <$> case cmd of
+          NEW_ -> s (NEW <$> strP_ <*> strP)
+          JOIN_ -> s (JOIN <$> strP_ <*> strP_ <*> binaryP)
+          LET_ -> s (LET <$> A.takeTill (== ' ') <* A.space <*> binaryP)
+          ACPT_ -> s (ACPT <$> A.takeTill (== ' ') <* A.space <*> binaryP)
+          RJCT_ -> s (RJCT <$> A.takeByteString)
+          SUB_ -> pure SUB
+          SEND_ -> s (SEND <$> smpP <* A.space <*> binaryP)
+          ACK_ -> s (ACK <$> A.decimal)
+          OFF_ -> pure OFF
+          DEL_ -> pure DEL
+          CHK_ -> pure CHK
+      ACmdTag SAgent cmd ->
+        ACmd SAgent <$> case cmd of
+          INV_ -> s (INV <$> strP)
+          CONF_ -> s (CONF <$> A.takeTill (== ' ') <* A.space <*> strListP <* A.space <*> binaryP)
+          REQ_ -> s (REQ <$> A.takeTill (== ' ') <* A.space <*> strP_ <*> binaryP)
+          INFO_ -> s (INFO <$> binaryP)
+          CON_ -> pure CON
+          END_ -> pure END
+          CONNECT_ -> s (CONNECT <$> strP_ <*> strP)
+          DISCONNECT_ -> s (DISCONNECT <$> strP_ <*> strP)
+          DOWN_ -> s (DOWN <$> strP_ <*> connections)
+          UP_ -> s (UP <$> strP_ <*> connections)
+          MID_ -> s (MID <$> A.decimal)
+          SENT_ -> s (SENT <$> A.decimal)
+          MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP)
+          MSG_ -> s (MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> binaryP)
+          STAT_ -> s (STAT <$> strP)
+          OK_ -> pure OK
+          ERR_ -> s (ERR <$> strP)
+          SUSPENDED_ -> pure SUSPENDED
  where
-    newCmd = ACmd SClient . NEW <$> strP
-    invResp = ACmd SAgent . INV <$> strP
-    joinCmd = ACmd SClient .: JOIN <$> strP_ <*> A.takeByteString
-    confMsg = ACmd SAgent .:. CONF <$> A.takeTill (== ' ') <* A.space <*> strListP <* A.space <*> A.takeByteString
-    letCmd = ACmd SClient .: LET <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString
-    reqMsg = ACmd SAgent .:. REQ <$> A.takeTill (== ' ') <* A.space <*> strP_ <*> A.takeByteString
-    acptCmd = ACmd SClient .: ACPT <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString
-    rjctCmd = ACmd SClient . RJCT <$> A.takeByteString
-    infoCmd = ACmd SAgent . INFO <$> A.takeByteString
-    connectResp = ACmd SAgent .: CONNECT <$> strP_ <*> strP
-    disconnectResp = ACmd SAgent .: DISCONNECT <$> strP_ <*> strP
-    downResp = ACmd SAgent .: DOWN <$> strP_ <*> connections
-    upResp = ACmd SAgent .: UP <$> strP_ <*> connections
-    sendCmd = ACmd SClient .: SEND <$> smpP <* A.space <*> A.takeByteString
-    msgIdResp = ACmd SAgent . MID <$> A.decimal
-    sentResp = ACmd SAgent . SENT <$> A.decimal
-    msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> strP
-    message = ACmd SAgent .:. MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> A.takeByteString
-    ackCmd = ACmd SClient . ACK <$> A.decimal
-    statResp = ACmd SAgent . STAT <$> strP
+    s :: Parser a -> Parser a
+    s p = A.space *> p
+    connections :: Parser [ConnId]
    connections = strP `A.sepBy'` A.char ','
    msgMetaP = do
      integrity <- strP
@@ -980,17 +1132,16 @@ commandP =
      sndMsgId <- " S=" *> A.decimal
      pure MsgMeta {integrity, recipient, broker, sndMsgId}
    partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P
-    agentError = ACmd SAgent . ERR <$> strP

 parseCommand :: ByteString -> Either AgentErrorType ACmd
-parseCommand = parse commandP $ CMD SYNTAX
+parseCommand = parse (commandP A.takeByteString) $ CMD SYNTAX

 -- | Serialize SMP agent command.
 serializeCommand :: ACommand p -> ByteString
 serializeCommand = \case
-  NEW cMode -> "NEW " <> strEncode cMode
+  NEW ntfs cMode -> B.unwords ["NEW", strEncode ntfs, strEncode cMode]
  INV cReq -> "INV " <> strEncode cReq
-  JOIN cReq cInfo -> B.unwords ["JOIN", strEncode cReq, serializeBinary cInfo]
+  JOIN ntfs cReq cInfo -> B.unwords ["JOIN", strEncode ntfs, strEncode cReq, serializeBinary cInfo]
  CONF confId srvs cInfo -> B.unwords ["CONF", confId, strEncodeList srvs, serializeBinary cInfo]
  LET confId cInfo -> B.unwords ["LET", confId, serializeBinary cInfo]
  REQ invId srvs cInfo -> B.unwords ["REQ", invId, strEncode srvs, serializeBinary cInfo]
@@ -1068,7 +1219,7 @@ tGet party h = liftIO (tGetRaw h) >>= tParseLoadBody
    tConnId :: ARawTransmission -> ACommand p -> Either AgentErrorType (ACommand p)
    tConnId (_, connId, _) cmd = case cmd of
      -- NEW, JOIN and ACPT have optional connId
-      NEW _ -> Right cmd
+      NEW _ _ -> Right cmd
      JOIN {} -> Right cmd
      ACPT {} -> Right cmd
      -- ERROR response does not always have connId
@@ -1086,7 +1237,7 @@ tGet party h = liftIO (tGetRaw h) >>= tParseLoadBody
    cmdWithMsgBody = \case
      SEND msgFlags body -> SEND msgFlags <$$> getBody body
      MSG msgMeta msgFlags body -> MSG msgMeta msgFlags <$$> getBody body
-      JOIN qUri cInfo -> JOIN qUri <$$> getBody cInfo
+      JOIN ntfs qUri cInfo -> JOIN ntfs qUri <$$> getBody cInfo
      CONF confId srvs cInfo -> CONF confId srvs <$$> getBody cInfo
      LET confId cInfo -> LET confId <$$> getBody cInfo
      REQ invId srvs cInfo -> REQ invId srvs <$$> getBody cInfo
@@ -3,13 +3,17 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
 {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}

 module Simplex.Messaging.Agent.Store where

 import Control.Exception (Exception)
+import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Char8 (ByteString)
 import Data.Int (Int64)
 import Data.Kind (Type)
@@ -18,6 +22,7 @@ import Data.Type.Equality
 import Simplex.Messaging.Agent.Protocol
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (RatchetX448)
+import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
  ( MsgBody,
    MsgFlags,
@@ -31,6 +36,7 @@ import Simplex.Messaging.Protocol
    SndPrivateSignKey,
  )
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Util ((<$?>))
 import Simplex.Messaging.Version

 -- * Queue types
@@ -92,7 +98,7 @@ data SndQueue = SndQueue
 -- * Connection types

 -- | Type of a connection.
-data ConnType = CRcv | CSnd | CDuplex | CContact deriving (Eq, Show)
+data ConnType = CNew | CRcv | CSnd | CDuplex | CContact deriving (Eq, Show)

 -- | Connection of a specific type.
 --
@@ -105,6 +111,7 @@ data ConnType = CRcv | CSnd | CDuplex | CContact deriving (Eq, Show)
 -- - DuplexConnection is a connection that has both receive and send queues set up,
 --   typically created by upgrading a receive or a send connection with a missing queue.
 data Connection (d :: ConnType) where
+  NewConnection :: ConnData -> Connection CNew
  RcvConnection :: ConnData -> RcvQueue -> Connection CRcv
  SndConnection :: ConnData -> SndQueue -> Connection CSnd
  DuplexConnection :: ConnData -> RcvQueue -> SndQueue -> Connection CDuplex
@@ -115,12 +122,14 @@ deriving instance Eq (Connection d)
 deriving instance Show (Connection d)

 data SConnType :: ConnType -> Type where
+  SCNew :: SConnType CNew
  SCRcv :: SConnType CRcv
  SCSnd :: SConnType CSnd
  SCDuplex :: SConnType CDuplex
  SCContact :: SConnType CContact

 connType :: SConnType c -> ConnType
+connType SCNew = CNew
 connType SCRcv = CRcv
 connType SCSnd = CSnd
 connType SCDuplex = CDuplex
@@ -156,6 +165,96 @@ data ConnData = ConnData
  }
  deriving (Eq, Show)

+data AgentCmdType = ACClient | ACInternal
+
+instance StrEncoding AgentCmdType where
+  strEncode = \case
+    ACClient -> "CLIENT"
+    ACInternal -> "INTERNAL"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "CLIENT" -> pure ACClient
+      "INTERNAL" -> pure ACInternal
+      _ -> fail "bad AgentCmdType"
+
+data AgentCommand
+  = AClientCommand (ACommand 'Client)
+  | AInternalCommand InternalCommand
+
+instance StrEncoding AgentCommand where
+  strEncode = \case
+    AClientCommand cmd -> strEncode (ACClient, Str $ serializeCommand cmd)
+    AInternalCommand cmd -> strEncode (ACInternal, cmd)
+  strP =
+    strP_ >>= \case
+      ACClient -> AClientCommand <$> ((\(ACmd _ cmd) -> checkParty cmd) <$?> dbCommandP)
+      ACInternal -> AInternalCommand <$> strP
+
+data AgentCommandTag
+  = AClientCommandTag (ACommandTag 'Client)
+  | AInternalCommandTag InternalCommandTag
+
+instance StrEncoding AgentCommandTag where
+  strEncode = \case
+    AClientCommandTag t -> strEncode (ACClient, t)
+    AInternalCommandTag t -> strEncode (ACInternal, t)
+  strP =
+    strP_ >>= \case
+      ACClient -> AClientCommandTag <$> strP
+      ACInternal -> AInternalCommandTag <$> strP
+
+data InternalCommand
+  = ICAck SMP.RecipientId MsgId
+  | ICAckDel SMP.RecipientId MsgId InternalId
+  | ICAllowSecure SMP.RecipientId SMP.SndPublicVerifyKey
+  | ICDuplexSecure SMP.RecipientId SMP.SndPublicVerifyKey
+
+data InternalCommandTag
+  = ICAck_
+  | ICAckDel_
+  | ICAllowSecure_
+  | ICDuplexSecure_
+  deriving (Show)
+
+instance StrEncoding InternalCommand where
+  strEncode = \case
+    ICAck rId srvMsgId -> strEncode (ICAck_, rId, srvMsgId)
+    ICAckDel rId srvMsgId mId -> strEncode (ICAckDel_, rId, srvMsgId, mId)
+    ICAllowSecure rId sndKey -> strEncode (ICAllowSecure_, rId, sndKey)
+    ICDuplexSecure rId sndKey -> strEncode (ICDuplexSecure_, rId, sndKey)
+  strP =
+    strP_ >>= \case
+      ICAck_ -> ICAck <$> strP_ <*> strP
+      ICAckDel_ -> ICAckDel <$> strP_ <*> strP_ <*> strP
+      ICAllowSecure_ -> ICAllowSecure <$> strP_ <*> strP
+      ICDuplexSecure_ -> ICDuplexSecure <$> strP_ <*> strP
+
+instance StrEncoding InternalCommandTag where
+  strEncode = \case
+    ICAck_ -> "ACK"
+    ICAckDel_ -> "ACK_DEL"
+    ICAllowSecure_ -> "ALLOW_SECURE"
+    ICDuplexSecure_ -> "DUPLEX_SECURE"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "ACK" -> pure ICAck_
+      "ACK_DEL" -> pure ICAckDel_
+      "ALLOW_SECURE" -> pure ICAllowSecure_
+      "DUPLEX_SECURE" -> pure ICDuplexSecure_
+      _ -> fail "bad InternalCommandTag"
+
+agentCommandTag :: AgentCommand -> AgentCommandTag
+agentCommandTag = \case
+  AClientCommand cmd -> AClientCommandTag $ aCommandTag cmd
+  AInternalCommand cmd -> AInternalCommandTag $ internalCmdTag cmd
+
+internalCmdTag :: InternalCommand -> InternalCommandTag
+internalCmdTag = \case
+  ICAck {} -> ICAck_
+  ICAckDel {} -> ICAckDel_
+  ICAllowSecure {} -> ICAllowSecure_
+  ICDuplexSecure {} -> ICDuplexSecure_
+
 -- * Confirmation types

 data NewConfirmation = NewConfirmation
@@ -270,8 +369,14 @@ data MsgBase = MsgBase

 newtype InternalId = InternalId {unId :: Int64} deriving (Eq, Show)

+instance StrEncoding InternalId where
+  strEncode = strEncode . unId
+  strP = InternalId <$> strP
+
 type InternalTs = UTCTime

+type AsyncCmdId = Int64
+
 -- * Store errors

 -- | Agent store error.
@@ -293,6 +398,8 @@ data StoreError
    SEInvitationNotFound
  | -- | Message not found
    SEMsgNotFound
+  | -- | Command not found
+    SECmdNotFound
  | -- | Currently not used. The intention was to pass current expected queue status in methods,
    -- as we always know what it should be at any stage of the protocol,
    -- and in case it does not match use this error.
@@ -22,8 +22,14 @@ module Simplex.Messaging.Agent.Store.SQLite
  ( SQLiteStore (..),
    createSQLiteStore,
    connectSQLiteStore,
+    closeSQLiteStore,
+    sqlString,
+    execSQL,

    -- * Queues and connections
+    createNewConn,
+    updateNewConnRcv,
+    updateNewConnSnd,
    createRcvConn,
    createSndConn,
    getConn,
@@ -66,6 +72,11 @@ module Simplex.Messaging.Agent.Store.SQLite
    getRatchet,
    getSkippedMsgKeys,
    updateRatchet,
+    -- Async commands
+    createCommand,
+    getPendingCommands,
+    getPendingCommand,
+    deleteCommand,
    -- Notification device token persistence
    createNtfToken,
    getSavedNtfToken,
@@ -105,8 +116,11 @@ import Data.Bifunctor (second)
 import Data.ByteString (ByteString)
 import qualified Data.ByteString.Base64.URL as U
 import Data.Char (toLower)
+import Data.Function (on)
 import Data.Functor (($>))
-import Data.List (find, foldl')
+import Data.IORef
+import Data.Int (Int64)
+import Data.List (foldl', groupBy)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.Map.Strict as M
 import Data.Maybe (fromMaybe, listToMaybe)
@@ -140,7 +154,7 @@ import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.Exit (exitFailure)
 import System.FilePath (takeDirectory)
 import System.IO (hFlush, stdout)
-import UnliftIO.Exception (bracket)
+import UnliftIO.Exception (bracket, onException)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM

@@ -148,28 +162,19 @@ import UnliftIO.STM

 data SQLiteStore = SQLiteStore
  { dbFilePath :: FilePath,
+    dbEncrypted :: TVar Bool,
    dbConnection :: TMVar DB.Connection,
    dbNew :: Bool
  }

-createSQLiteStore :: FilePath -> [Migration] -> Bool -> IO SQLiteStore
-createSQLiteStore dbFilePath migrations yesToMigrations = do
+createSQLiteStore :: FilePath -> String -> [Migration] -> Bool -> IO SQLiteStore
+createSQLiteStore dbFilePath dbKey migrations yesToMigrations = do
  let dbDir = takeDirectory dbFilePath
  createDirectoryIfMissing False dbDir
-  st <- connectSQLiteStore dbFilePath
-  checkThreadsafe st
-  migrateSchema st migrations yesToMigrations
+  st <- connectSQLiteStore dbFilePath dbKey
+  migrateSchema st migrations yesToMigrations `onException` closeSQLiteStore st
  pure st

-checkThreadsafe :: SQLiteStore -> IO ()
-checkThreadsafe st = withConnection st $ \db -> do
-  compileOptions <- DB.query_ db "pragma COMPILE_OPTIONS;" :: IO [[Text]]
-  let threadsafeOption = find (T.isPrefixOf "THREADSAFE=") (concat compileOptions)
-  case threadsafeOption of
-    Just "THREADSAFE=0" -> confirmOrExit "SQLite compiled with non-threadsafe code."
-    Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found"
-    _ -> return ()
-
 migrateSchema :: SQLiteStore -> [Migration] -> Bool -> IO ()
 migrateSchema st migrations yesToMigrations = withConnection st $ \db -> do
  Migrations.initialize db
@@ -192,24 +197,38 @@ confirmOrExit s = do
  ok <- getLine
  when (map toLower ok /= "y") exitFailure

-connectSQLiteStore :: FilePath -> IO SQLiteStore
-connectSQLiteStore dbFilePath = do
+connectSQLiteStore :: FilePath -> String -> IO SQLiteStore
+connectSQLiteStore dbFilePath dbKey = do
  dbNew <- not <$> doesFileExist dbFilePath
-  dbConnection <- newTMVarIO =<< connectDB dbFilePath
-  pure SQLiteStore {dbFilePath, dbConnection, dbNew}
+  dbConnection <- newTMVarIO =<< connectDB dbFilePath dbKey
+  dbEncrypted <- newTVarIO . not $ null dbKey
+  pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection, dbNew}

-connectDB :: FilePath -> IO DB.Connection
-connectDB path = do
-  dbConn <- DB.open path
-  SQLite3.exec (DB.connectionHandle dbConn) . fromQuery $
-    [sql|
-      PRAGMA foreign_keys = ON;
-      -- PRAGMA trusted_schema = OFF;
-      PRAGMA secure_delete = ON;
-      PRAGMA auto_vacuum = FULL;
-    |]
-  -- _printPragmas dbConn path
-  pure dbConn
+connectDB :: FilePath -> String -> IO DB.Connection
+connectDB path key = do
+  db <- DB.open path
+  prepare db `onException` DB.close db
+  -- _printPragmas db path
+  pure db
+  where
+    prepare db = do
+      let exec = SQLite3.exec $ DB.connectionHandle db
+      unless (null key) . exec $ "PRAGMA key = " <> sqlString key <> ";"
+      exec . fromQuery $
+        [sql|
+          PRAGMA foreign_keys = ON;
+          -- PRAGMA trusted_schema = OFF;
+          PRAGMA secure_delete = ON;
+          PRAGMA auto_vacuum = FULL;
+        |]
+
+closeSQLiteStore :: SQLiteStore -> IO ()
+closeSQLiteStore st = atomically (takeTMVar $ dbConnection st) >>= DB.close
+
+sqlString :: String -> Text
+sqlString s = quote <> T.replace quote "''" (T.pack s) <> quote
+  where
+    quote = "'"

 -- _printPragmas :: DB.Connection -> FilePath -> IO ()
 -- _printPragmas db path = do
@@ -223,6 +242,19 @@ connectDB path = do
 --   auto_vacuum <- DB.query_ db "PRAGMA auto_vacuum;" :: IO [[Int]]
 --   print $ path <> " auto_vacuum: " <> show auto_vacuum

+execSQL :: DB.Connection -> Text -> IO [Text]
+execSQL db query = do
+  rs <- newIORef []
+  SQLite3.execWithCallback (DB.connectionHandle db) query (addSQLResultRow rs)
+  reverse <$> readIORef rs
+
+addSQLResultRow :: IORef [Text] -> SQLite3.ColumnIndex -> [Text] -> [Maybe Text] -> IO ()
+addSQLResultRow rs _count names values = modifyIORef' rs $ \case
+  [] -> [showValues values, T.intercalate "|" names]
+  rs' -> showValues values : rs'
+  where
+    showValues = T.intercalate "|" . map (fromMaybe "")
+
 checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)
 checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err)

@@ -258,6 +290,37 @@ createConn_ gVar cData create = checkConstraint SEConnDuplicate $ case cData of
  ConnData {connId = ""} -> createWithRandomId gVar create
  ConnData {connId} -> create connId $> Right connId

+createNewConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> SConnectionMode c -> IO (Either StoreError ConnId)
+createNewConn db gVar cData@ConnData {connAgentVersion, enableNtfs, duplexHandshake} cMode =
+  createConn_ gVar cData $ \connId -> do
+    DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?, ?, ?, ?, ?)" (connId, cMode, connAgentVersion, enableNtfs, duplexHandshake)
+
+updateNewConnRcv :: DB.Connection -> ConnId -> RcvQueue -> IO (Either StoreError ())
+updateNewConnRcv db connId rq@RcvQueue {server} =
+  getConn db connId $>>= \case
+    (SomeConn _ NewConnection {}) -> updateConn
+    (SomeConn _ RcvConnection {}) -> updateConn -- to allow retries
+    (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+  where
+    updateConn :: IO (Either StoreError ())
+    updateConn = do
+      upsertServer_ db server
+      insertRcvQueue_ db connId rq
+      pure $ Right ()
+
+updateNewConnSnd :: DB.Connection -> ConnId -> SndQueue -> IO (Either StoreError ())
+updateNewConnSnd db connId sq@SndQueue {server} =
+  getConn db connId $>>= \case
+    (SomeConn _ NewConnection {}) -> updateConn
+    (SomeConn _ SndConnection {}) -> updateConn -- to allow retries
+    (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
+  where
+    updateConn :: IO (Either StoreError ())
+    updateConn = do
+      upsertServer_ db server
+      insertSndQueue_ db connId sq
+      pure $ Right ()
+
 createRcvConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> IO (Either StoreError ConnId)
 createRcvConn db gVar cData@ConnData {connAgentVersion, enableNtfs, duplexHandshake} q@RcvQueue {server} cMode =
  createConn_ gVar cData $ \connId -> do
@@ -654,6 +717,50 @@ updateRatchet db connId rc skipped = do
        forM_ (M.assocs mks) $ \(msgN, mk) ->
          DB.execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, hk, msgN, mk)

+createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> IO AsyncCmdId
+createCommand db corrId connId srv cmd = do
+  DB.execute
+    db
+    "INSERT INTO commands (host, port, corr_id, conn_id, command_tag, command) VALUES (?,?,?,?,?,?)"
+    (host_, port_, corrId, connId, agentCommandTag cmd, cmd)
+  insertedRowId db
+  where
+    (host_, port_) =
+      case srv of
+        Just (SMPServer host port _) -> (Just host, Just port)
+        _ -> (Nothing, Nothing)
+
+insertedRowId :: DB.Connection -> IO Int64
+insertedRowId db = fromOnly . head <$> DB.query_ db "SELECT last_insert_rowid()"
+
+getPendingCommands :: DB.Connection -> ConnId -> IO [(Maybe SMPServer, [AsyncCmdId])]
+getPendingCommands db connId = do
+  map (\ids -> (fst $ head ids, map snd ids)) . groupBy ((==) `on` fst) . map srvCmdId
+    <$> DB.query
+      db
+      [sql|
+        SELECT c.host, c.port, s.key_hash, c.command_id
+        FROM commands c
+        LEFT JOIN servers s ON s.host = c.host AND s.port = c.port
+        WHERE conn_id = ?
+        ORDER BY c.host, c.port, c.command_id ASC
+      |]
+      (Only connId)
+  where
+    srvCmdId (host, port, keyHash, cmdId) = (SMPServer <$> host <*> port <*> keyHash, cmdId)
+
+getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ACorrId, ConnId, AgentCommand))
+getPendingCommand db msgId = do
+  firstRow id SECmdNotFound $
+    DB.query
+      db
+      "SELECT corr_id, conn_id, command FROM commands WHERE command_id = ?"
+      (Only msgId)
+
+deleteCommand :: DB.Connection -> AsyncCmdId -> IO ()
+deleteCommand db cmdId =
+  DB.execute db "DELETE FROM commands WHERE command_id = ?" (Only cmdId)
+
 createNtfToken :: DB.Connection -> NtfToken -> IO ()
 createNtfToken db NtfToken {deviceToken = DeviceToken provider token, ntfServer = srv@ProtocolServer {host, port}, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys = (ntfDhPubKey, ntfDhPrivKey), ntfDhSecret, ntfTknStatus, ntfTknAction, ntfMode} = do
  upsertNtfServer_ db srv
@@ -1016,6 +1123,14 @@ instance ToField (NonEmpty TransportHost) where toField = toField . decodeLatin1

 instance FromField (NonEmpty TransportHost) where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8

+instance ToField AgentCommand where toField = toField . strEncode
+
+instance FromField AgentCommand where fromField = blobFieldParser strP
+
+instance ToField AgentCommandTag where toField = toField . strEncode
+
+instance FromField AgentCommandTag where fromField = blobFieldParser strP
+
 listToEither :: e -> [a] -> Either e a
 listToEither _ (x : _) = Right x
 listToEither e _ = Left e
@@ -1123,6 +1238,7 @@ getConn dbConn connId =
        (Just rcvQ, Nothing, CMInvitation) -> Right $ SomeConn SCRcv (RcvConnection connData rcvQ)
        (Nothing, Just sndQ, CMInvitation) -> Right $ SomeConn SCSnd (SndConnection connData sndQ)
        (Just rcvQ, Nothing, CMContact) -> Right $ SomeConn SCContact (ContactConnection connData rcvQ)
+        (Nothing, Nothing, _) -> Right $ SomeConn SCNew (NewConnection connData)
        _ -> Left SEConnNotFound

 getConnData :: DB.Connection -> ConnId -> IO (Maybe (ConnData, ConnectionMode))
@@ -17,10 +17,9 @@ module Simplex.Messaging.Agent.Store.SQLite.Migrations
 where

 import Control.Monad (forM_, when)
-import Data.List (intercalate, sortBy)
+import Data.List (intercalate, sortOn)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map as M
-import Data.Ord (comparing)
 import Data.Text (Text)
 import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (getCurrentTime)
@@ -36,6 +35,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220608_v2
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220625_v2_ntf_mode
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220811_onion_hosts
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220817_connection_ntfs
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Transport.Client (TransportHost)

@@ -50,12 +50,13 @@ schemaMigrations =
    ("20220607_v2", m20220608_v2),
    ("m20220625_v2_ntf_mode", m20220625_v2_ntf_mode),
    ("m20220811_onion_hosts", m20220811_onion_hosts),
-    ("m20220817_connection_ntfs", m20220817_connection_ntfs)
+    ("m20220817_connection_ntfs", m20220817_connection_ntfs),
+    ("m20220905_commands", m20220905_commands)
  ]

 -- | The list of migrations in ascending order by date
 app :: [Migration]
-app = sortBy (comparing name) $ map migration schemaMigrations
+app = sortOn name $ map migration schemaMigrations
  where
    migration (name, query) = Migration {name = name, up = fromQuery query}

@@ -0,0 +1,23 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20220905_commands :: Query
+m20220905_commands =
+  [sql|
+CREATE TABLE commands (
+  command_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  host TEXT,
+  port TEXT,
+  corr_id BLOB NOT NULL,
+  command_tag BLOB NOT NULL,
+  command BLOB NOT NULL,
+  agent_version INTEGER NOT NULL DEFAULT 1,
+  FOREIGN KEY (host, port) REFERENCES servers
+    ON DELETE RESTRICT ON UPDATE CASCADE
+);
+|]
@@ -194,3 +194,15 @@ CREATE TABLE ntf_subscriptions(
  FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers
  ON DELETE RESTRICT ON UPDATE CASCADE
 ) WITHOUT ROWID;
+CREATE TABLE commands(
+  command_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  host TEXT,
+  port TEXT,
+  corr_id BLOB NOT NULL,
+  command_tag BLOB NOT NULL,
+  command BLOB NOT NULL,
+  agent_version INTEGER NOT NULL DEFAULT 1,
+  FOREIGN KEY(host, port) REFERENCES servers
+  ON DELETE RESTRICT ON UPDATE CASCADE
+);
@@ -105,7 +105,7 @@ data ProtocolClient msg = ProtocolClient
    tcpTimeout :: Int,
    clientCorrId :: TVar Natural,
    sentCommands :: TMap CorrId (Request msg),
-    sndQ :: TBQueue (NonEmpty (SentRawTransmission)),
+    sndQ :: TBQueue (NonEmpty SentRawTransmission),
    rcvQ :: TBQueue (NonEmpty (SignedTransmission msg)),
    msgQ :: Maybe (TBQueue (ServerTransmission msg))
  }
@@ -265,7 +265,7 @@ subscribeQueue ca srv sub = do

 showServer :: SMPServer -> ByteString
 showServer ProtocolServer {host, port} =
-  strEncode host <> (B.pack $ if null port then "" else ':' : port)
+  strEncode host <> B.pack (if null port then "" else ':' : port)

 smpSubscribe :: SMPClient -> (SMPSub, C.APrivateSignKey) -> ExceptT ProtocolClientError IO ()
 smpSubscribe smp ((party, queueId), privKey) = subscribe_ smp privKey queueId
@@ -387,7 +387,7 @@ instance StrEncoding DeviceToken where
  strP = DeviceToken <$> strP <* A.space <*> hexStringP
    where
      hexStringP =
-        A.takeWhile (\c -> (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f')) >>= \s ->
+        A.takeWhile (\c -> A.isDigit c || (c >= 'a' && c <= 'f')) >>= \s ->
          if even (B.length s) then pure s else fail "odd number of hex characters"

 instance ToJSON DeviceToken where
@@ -7,7 +7,6 @@
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}

@@ -106,6 +106,7 @@ removeInactiveTokenRegistrations st NtfTknData {ntfTknId = tId, token} =
      forM_ tIds $ \(regKey, tId') -> do
        TM.delete regKey tknRegs
        TM.delete tId' $ tokens st
+        -- TODO remove token subscriptions as in deleteNtfToken
      pure $ map snd tIds

 removeTokenRegistration :: NtfStore -> NtfTknData -> STM ()
@@ -130,6 +131,7 @@ deleteNtfToken st tknId = do
              )
      )

+  -- TODO refactor
  qs <-
    TM.lookupDelete tknId (tokenSubscriptions st)
      >>= mapM
@@ -3,7 +3,6 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE PatternSynonyms #-}

 module Simplex.Messaging.Notifications.Types where

@@ -927,7 +927,7 @@ instance ProtocolEncoding Cmd where
    CT SSender tag ->
      Cmd SSender <$> case tag of
        SEND_
-          | v == 1 -> SEND <$> pure noMsgFlags <*> (unTail <$> _smpP)
+          | v == 1 -> SEND noMsgFlags <$> (unTail <$> _smpP)
          | otherwise -> SEND <$> _smpP <*> (unTail <$> _smpP)
        PING_ -> pure PING
    CT SNotifier NSUB_ -> pure $ Cmd SNotifier NSUB
@@ -1049,7 +1049,7 @@ instance Encoding CommandError where
      _ -> fail "bad command error type"

 -- | Send signed SMP transmission to TCP transport.
-tPut :: Transport c => THandle c -> NonEmpty (SentRawTransmission) -> IO (NonEmpty (Either TransportError ()))
+tPut :: Transport c => THandle c -> NonEmpty SentRawTransmission -> IO (NonEmpty (Either TransportError ()))
 tPut th trs
  | batch th = tPutBatch [] $ L.map tEncode trs
  | otherwise = forM trs $ tPutBlock th . tEncode
@@ -8,7 +8,6 @@
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
@@ -429,7 +428,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv

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

        subscribeQueue :: QueueRec -> RecipientId -> m (Transmission BrokerMsg)
@@ -15,7 +15,7 @@ data QueueRec = QueueRec
    senderId :: SenderId,
    senderKey :: Maybe SndPublicVerifyKey,
    notifier :: Maybe NtfCreds,
-    status :: QueueStatus
+    status :: ServerQueueStatus
  }
  deriving (Eq, Show)

@@ -32,7 +32,7 @@ instance StrEncoding NtfCreds where
    (notifierId, notifierKey, rcvNtfDhSecret) <- strP
    pure NtfCreds {notifierId, notifierKey, rcvNtfDhSecret}

-data QueueStatus = QueueActive | QueueOff deriving (Eq, Show)
+data ServerQueueStatus = QueueActive | QueueOff deriving (Eq, Show)

 class MonadQueueStore s m where
  addQueue :: s -> QueueRec -> m (Either ErrorType ())
@@ -58,8 +58,10 @@ instance MonadQueueStore QueueStore STM where
  secureQueue QueueStore {queues} rId sKey =
    withQueue rId queues $ \qVar ->
      readTVar qVar >>= \q -> case senderKey q of
-        Just _ -> pure Nothing
-        _ -> writeTVar qVar q {senderKey = Just sKey} $> Just q
+        Just k -> pure $ if sKey == k then Just q else Nothing
+        _ ->
+          let q' = q {senderKey = Just sKey}
+          in writeTVar qVar q' $> Just q'

  addQueueNotifier :: QueueStore -> RecipientId -> NtfCreds -> STM (Either ErrorType QueueRec)
  addQueueNotifier QueueStore {queues, notifiers} rId ntfCreds@NtfCreds {notifierId = nId} = do
@@ -17,6 +17,7 @@ module Simplex.Messaging.Server.StoreLog
    logCreateQueue,
    logSecureQueue,
    logAddNotifier,
+    logSuspendQueue,
    logDeleteQueue,
    logDeleteNotifier,
    readWriteStoreLog,
@@ -36,7 +37,7 @@ import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..), QueueStatus (..))
+import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..), ServerQueueStatus (..))
 import Simplex.Messaging.Transport (trimCR)
 import System.Directory (doesFileExist)
 import System.IO
@@ -51,6 +52,7 @@ data StoreLogRecord
  = CreateQueue QueueRec
  | SecureQueue QueueId SndPublicVerifyKey
  | AddNotifier QueueId NtfCreds
+  | SuspendQueue QueueId
  | DeleteQueue QueueId
  | DeleteNotifier QueueId

@@ -81,6 +83,7 @@ instance StrEncoding StoreLogRecord where
    CreateQueue q -> strEncode (Str "CREATE", q)
    SecureQueue rId sKey -> strEncode (Str "SECURE", rId, sKey)
    AddNotifier rId ntfCreds -> strEncode (Str "NOTIFIER", rId, ntfCreds)
+    SuspendQueue rId -> strEncode (Str "SUSPEND", rId)
    DeleteQueue rId -> strEncode (Str "DELETE", rId)
    DeleteNotifier rId -> strEncode (Str "NDELETE", rId)

@@ -88,6 +91,7 @@ instance StrEncoding StoreLogRecord where
    "CREATE " *> (CreateQueue <$> strP)
      <|> "SECURE " *> (SecureQueue <$> strP_ <*> strP)
      <|> "NOTIFIER " *> (AddNotifier <$> strP_ <*> strP)
+      <|> "SUSPEND " *> (SuspendQueue <$> strP)
      <|> "DELETE " *> (DeleteQueue <$> strP)
      <|> "NDELETE " *> (DeleteNotifier <$> strP)

@@ -126,6 +130,9 @@ logSecureQueue s qId sKey = writeStoreLogRecord s $ SecureQueue qId sKey
 logAddNotifier :: StoreLog 'WriteMode -> QueueId -> NtfCreds -> IO ()
 logAddNotifier s qId ntfCreds = writeStoreLogRecord s $ AddNotifier qId ntfCreds

+logSuspendQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
+logSuspendQueue s = writeStoreLogRecord s . SuspendQueue
+
 logDeleteQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
 logDeleteQueue s = writeStoreLogRecord s . DeleteQueue

@@ -161,6 +168,7 @@ readQueues (ReadStoreLog _ h) = LB.hGetContents h >>= returnResult . procStoreLo
      CreateQueue q -> M.insert (recipientId q) q m
      SecureQueue qId sKey -> M.adjust (\q -> q {senderKey = Just sKey}) qId m
      AddNotifier qId ntfCreds -> M.adjust (\q -> q {notifier = Just ntfCreds}) qId m
+      SuspendQueue qId -> M.adjust (\q -> q {status = QueueOff}) qId m
      DeleteQueue qId -> M.delete qId m
      DeleteNotifier qId -> M.adjust (\q -> q {notifier = Nothing}) qId m
    printError :: LogParsingError -> IO ()
@@ -2,6 +2,7 @@ module Simplex.Messaging.TMap
  ( TMap,
    empty,
    singleton,
+    clear,
    Simplex.Messaging.TMap.null,
    Simplex.Messaging.TMap.lookup,
    member,
@@ -31,6 +32,10 @@ singleton :: k -> a -> STM (TMap k a)
 singleton k v = newTVar $ M.singleton k v
 {-# INLINE singleton #-}

+clear :: TMap k a -> STM ()
+clear m = writeTVar m M.empty
+{-# INLINE clear #-}
+
 null :: TMap k a -> STM Bool
 null m = M.null <$> readTVar m
 {-# INLINE null #-}
@@ -0,0 +1,82 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+
+module Simplex.Messaging.TMap2
+  ( TMap2,
+    empty,
+    clear,
+    Simplex.Messaging.TMap2.lookup,
+    lookup1,
+    member,
+    insert,
+    insert1,
+    delete,
+    lookupDelete1,
+  )
+where
+
+import Control.Concurrent.STM
+import Control.Monad (forM_, (>=>))
+import qualified Data.Map.Strict as M
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Util (whenM, ($>>=))
+
+-- | this type is designed for k2 being unique in the whole data, and k1 grouping multiple values with k2 keys.
+-- It allows direct access via k1 to a group of k2 values and via k2 to one value
+data TMap2 k1 k2 a = TMap2
+  { _m1 :: TMap k1 (TMap k2 a),
+    _m2 :: TMap k2 k1
+  }
+
+empty :: STM (TMap2 k1 k2 a)
+empty = TMap2 <$> TM.empty <*> TM.empty
+
+clear :: TMap2 k1 k2 a -> STM ()
+clear TMap2 {_m1, _m2} = TM.clear _m1 >> TM.clear _m2
+
+lookup :: (Ord k1, Ord k2) => k2 -> TMap2 k1 k2 a -> STM (Maybe a)
+lookup k2 TMap2 {_m1, _m2} = do
+  TM.lookup k2 _m2 $>>= (`TM.lookup` _m1) $>>= TM.lookup k2
+
+lookup1 :: Ord k1 => k1 -> TMap2 k1 k2 a -> STM (Maybe (TMap k2 a))
+lookup1 k1 TMap2 {_m1} = TM.lookup k1 _m1
+{-# INLINE lookup1 #-}
+
+member :: Ord k2 => k2 -> TMap2 k1 k2 a -> STM Bool
+member k2 TMap2 {_m2} = TM.member k2 _m2
+{-# INLINE member #-}
+
+insert :: (Ord k1, Ord k2) => k1 -> k2 -> a -> TMap2 k1 k2 a -> STM ()
+insert k1 k2 v TMap2 {_m1, _m2} =
+  TM.lookup k2 _m2 >>= \case
+    Just k1'
+      | k1 == k1' -> _insert1
+      | otherwise -> _delete1 k1' k2 _m1 >> _insert2
+    _ -> _insert2
+  where
+    _insert1 =
+      TM.lookup k1 _m1 >>= \case
+        Just m -> TM.insert k2 v m
+        _ -> TM.singleton k2 v >>= \m -> TM.insert k1 m _m1
+    _insert2 = TM.insert k2 k1 _m2 >> _insert1
+
+insert1 :: (Ord k1, Ord k2) => k1 -> TMap k2 a -> TMap2 k1 k2 a -> STM ()
+insert1 k1 m' TMap2 {_m1, _m2} =
+  TM.lookup k1 _m1 >>= \case
+    Just m -> readTVar m' >>= (`TM.union` m)
+    _ -> TM.insert k1 m' _m1
+
+delete :: (Ord k1, Ord k2) => k2 -> TMap2 k1 k2 a -> STM ()
+delete k2 TMap2 {_m1, _m2} = TM.lookupDelete k2 _m2 >>= mapM_ (\k1 -> _delete1 k1 k2 _m1)
+
+_delete1 :: (Ord k1, Ord k2) => k1 -> k2 -> TMap k1 (TMap k2 a) -> STM ()
+_delete1 k1 k2 m1 =
+  TM.lookup k1 m1
+    >>= mapM_ (\m -> TM.delete k2 m >> whenM (TM.null m) (TM.delete k1 m1))
+
+lookupDelete1 :: (Ord k1, Ord k2) => k1 -> TMap2 k1 k2 a -> STM (Maybe (TMap k2 a))
+lookupDelete1 k1 TMap2 {_m1, _m2} = do
+  m_ <- TM.lookupDelete k1 _m1
+  forM_ m_ $ readTVar >=> modifyTVar' _m2 . flip M.withoutKeys . M.keysSet
+  pure m_
@@ -99,7 +99,7 @@ supportedSMPServerVRange :: VersionRange
 supportedSMPServerVRange = mkVersionRange 1 4

 simplexMQVersion :: String
-simplexMQVersion = "3.2.0"
+simplexMQVersion = "3.3.0"

 -- * Transport connection class

@@ -1,4 +1,3 @@
-{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}

@@ -48,6 +48,12 @@ extra-deps:
  - time-compat-1.9.6.1@sha256:42d8f2e08e965e1718917d54ad69e1d06bd4b87d66c41dc7410f59313dba4ed1,5033
  - github: simplex-chat/aeson
    commit: 3eb66f9a68f103b5f1489382aad89f5712a64db7
+  # - ../direct-sqlcipher
+  - github: simplex-chat/direct-sqlcipher
+    commit: 34309410eb2069b029b8fc1872deb1e0db123294
+  # - ../sqlcipher-simple
+  - github: simplex-chat/sqlcipher-simple
+    commit: 5e154a2aeccc33ead6c243ec07195ab673137221
  # - ../hs-tls/core
  # - github: simplex-chat/hs-tls
  #   commit: f6cc753611f80af300401cfae63846e9d7c40d9e
@@ -22,6 +22,7 @@ import qualified Data.ByteString.Char8 as B
 import Network.HTTP.Types (urlEncode)
 import SMPAgentClient
 import SMPClient (testKeyHash, testPort, testPort2, testStoreLogFile, withSmpServer, withSmpServerStoreLogOn)
+import Simplex.Messaging.Agent.Env.SQLite (AgentDatabase, databaseFile)
 import Simplex.Messaging.Agent.Protocol
 import qualified Simplex.Messaging.Agent.Protocol as A
 import Simplex.Messaging.Encoding.String
@@ -131,9 +132,9 @@ pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} _ msgBody

 testDuplexConnection :: Transport c => TProxy c -> c -> c -> IO ()
 testDuplexConnection _ alice bob = do
-  ("1", "bob", Right (INV cReq)) <- alice #: ("1", "bob", "NEW INV")
+  ("1", "bob", Right (INV cReq)) <- alice #: ("1", "bob", "NEW T INV")
  let cReq' = strEncode cReq
-  bob #: ("11", "alice", "JOIN " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice", OK)
+  bob #: ("11", "alice", "JOIN T " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice", OK)
  ("", "bob", Right (CONF confId _ "bob's connInfo")) <- (alice <#:)
  alice #: ("2", "bob", "LET " <> confId <> " 16\nalice's connInfo") #> ("2", "bob", OK)
  bob <# ("", "alice", INFO "alice's connInfo")
@@ -164,9 +165,9 @@ testDuplexConnection _ alice bob = do

 testDuplexConnRandomIds :: Transport c => TProxy c -> c -> c -> IO ()
 testDuplexConnRandomIds _ alice bob = do
-  ("1", bobConn, Right (INV cReq)) <- alice #: ("1", "", "NEW INV")
+  ("1", bobConn, Right (INV cReq)) <- alice #: ("1", "", "NEW T INV")
  let cReq' = strEncode cReq
-  ("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN " <> cReq' <> " 14\nbob's connInfo")
+  ("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN T " <> cReq' <> " 14\nbob's connInfo")
  ("", bobConn', Right (CONF confId _ "bob's connInfo")) <- (alice <#:)
  bobConn' `shouldBe` bobConn
  alice #: ("2", bobConn, "LET " <> confId <> " 16\nalice's connInfo") =#> \case ("2", c, OK) -> c == bobConn; _ -> False
@@ -197,10 +198,10 @@ testDuplexConnRandomIds _ alice bob = do

 testContactConnection :: Transport c => TProxy c -> c -> c -> c -> IO ()
 testContactConnection _ alice bob tom = do
-  ("1", "alice_contact", Right (INV cReq)) <- alice #: ("1", "alice_contact", "NEW CON")
+  ("1", "alice_contact", Right (INV cReq)) <- alice #: ("1", "alice_contact", "NEW T CON")
  let cReq' = strEncode cReq

-  bob #: ("11", "alice", "JOIN " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice", OK)
+  bob #: ("11", "alice", "JOIN T " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice", OK)
  ("", "alice_contact", Right (REQ aInvId _ "bob's connInfo")) <- (alice <#:)
  alice #: ("2", "bob", "ACPT " <> aInvId <> " 16\nalice's connInfo") #> ("2", "bob", OK)
  ("", "alice", Right (CONF bConfId _ "alice's connInfo")) <- (bob <#:)
@@ -213,7 +214,7 @@ testContactConnection _ alice bob tom = do
  bob <#= \case ("", "alice", Msg "hi") -> True; _ -> False
  bob #: ("13", "alice", "ACK 4") #> ("13", "alice", OK)

-  tom #: ("21", "alice", "JOIN " <> cReq' <> " 14\ntom's connInfo") #> ("21", "alice", OK)
+  tom #: ("21", "alice", "JOIN T " <> cReq' <> " 14\ntom's connInfo") #> ("21", "alice", OK)
  ("", "alice_contact", Right (REQ aInvId' _ "tom's connInfo")) <- (alice <#:)
  alice #: ("4", "tom", "ACPT " <> aInvId' <> " 16\nalice's connInfo") #> ("4", "tom", OK)
  ("", "alice", Right (CONF tConfId _ "alice's connInfo")) <- (tom <#:)
@@ -228,10 +229,10 @@ testContactConnection _ alice bob tom = do

 testContactConnRandomIds :: Transport c => TProxy c -> c -> c -> IO ()
 testContactConnRandomIds _ alice bob = do
-  ("1", aliceContact, Right (INV cReq)) <- alice #: ("1", "", "NEW CON")
+  ("1", aliceContact, Right (INV cReq)) <- alice #: ("1", "", "NEW T CON")
  let cReq' = strEncode cReq

-  ("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN " <> cReq' <> " 14\nbob's connInfo")
+  ("11", aliceConn, Right OK) <- bob #: ("11", "", "JOIN T " <> cReq' <> " 14\nbob's connInfo")
  ("", aliceContact', Right (REQ aInvId _ "bob's connInfo")) <- (alice <#:)
  aliceContact' `shouldBe` aliceContact

@@ -251,9 +252,9 @@ testContactConnRandomIds _ alice bob = do

 testRejectContactRequest :: Transport c => TProxy c -> c -> c -> IO ()
 testRejectContactRequest _ alice bob = do
-  ("1", "a_contact", Right (INV cReq)) <- alice #: ("1", "a_contact", "NEW CON")
+  ("1", "a_contact", Right (INV cReq)) <- alice #: ("1", "a_contact", "NEW T CON")
  let cReq' = strEncode cReq
-  bob #: ("11", "alice", "JOIN " <> cReq' <> " 10\nbob's info") #> ("11", "alice", OK)
+  bob #: ("11", "alice", "JOIN T " <> cReq' <> " 10\nbob's info") #> ("11", "alice", OK)
  ("", "a_contact", Right (REQ aInvId _ "bob's info")) <- (alice <#:)
  -- RJCT must use correct contact connection
  alice #: ("2a", "bob", "RJCT " <> aInvId) #> ("2a", "bob", ERR $ CONN NOT_FOUND)
@@ -282,7 +283,7 @@ testSubscription _ alice1 alice2 bob = do

 testSubscrNotification :: Transport c => TProxy c -> (ThreadId, ThreadId) -> c -> IO ()
 testSubscrNotification t (server, _) client = do
-  client #: ("1", "conn1", "NEW INV") =#> \case ("1", "conn1", INV {}) -> True; _ -> False
+  client #: ("1", "conn1", "NEW T INV") =#> \case ("1", "conn1", INV {}) -> True; _ -> False
  client #:# "nothing should be delivered to client before the server is killed"
  killThread server
  client <# ("", "", DOWN testSMPServer ["conn1"])
@@ -330,8 +331,8 @@ testServerConnectionAfterError t _ = do

  withAgent1 $ \bob -> do
    withAgent2 $ \alice -> do
-      bob #: ("1", "alice", "SUB") #> ("1", "alice", ERR (BROKER NETWORK))
-      alice #: ("1", "bob", "SUB") #> ("1", "bob", ERR (BROKER NETWORK))
+      bob #: ("1", "alice", "SUB") =#> \("1", "alice", ERR (BROKER e)) -> e == NETWORK || e == TIMEOUT
+      alice #: ("1", "bob", "SUB") =#> \("1", "bob", ERR (BROKER e)) -> e == NETWORK || e == TIMEOUT
      withServer $ do
        alice <#= \case ("", "bob", SENT 4) -> True; ("", "", UP s ["bob"]) -> s == server; _ -> False
        alice <#= \case ("", "bob", SENT 4) -> True; ("", "", UP s ["bob"]) -> s == server; _ -> False
@@ -343,14 +344,14 @@ testServerConnectionAfterError t _ = do
        bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False

  removeFile testStoreLogFile
-  removeFile testDB
-  removeFile testDB2
+  removeFile $ databaseFile testDB
+  removeFile $ databaseFile testDB2
  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 -> String -> (c -> IO a) -> IO a
+    withAgent :: String -> AgentDatabase -> (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 ()
@@ -383,7 +384,7 @@ testMsgDeliveryAgentRestart t bob = do
      bob #: ("12", "alice", "ACK 5") #> ("12", "alice", OK)

  removeFile testStoreLogFile
-  removeFile testDB
+  removeFile $ databaseFile testDB
  where
    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
    withAgent = withSmpAgentThreadOn_ (ATransport t) (agentTestPort, testPort, testDB) (pure ()) . const . testSMPAgentClientOn agentTestPort
@@ -392,9 +393,9 @@ testConcurrentMsgDelivery :: Transport c => TProxy c -> c -> c -> IO ()
 testConcurrentMsgDelivery _ alice bob = do
  connect (alice, "alice") (bob, "bob")

-  ("1", "bob2", Right (INV cReq)) <- alice #: ("1", "bob2", "NEW INV")
+  ("1", "bob2", Right (INV cReq)) <- alice #: ("1", "bob2", "NEW T INV")
  let cReq' = strEncode cReq
-  bob #: ("11", "alice2", "JOIN " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice2", OK)
+  bob #: ("11", "alice2", "JOIN T " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice2", OK)
  ("", "bob2", Right (CONF _confId _ "bob's connInfo")) <- (alice <#:)
  -- below commands would be needed to accept bob's connection, but alice does not
  -- alice #: ("2", "bob", "LET " <> _confId <> " 16\nalice's connInfo") #> ("2", "bob", OK)
@@ -431,9 +432,9 @@ testMsgDeliveryQuotaExceeded _ alice bob = do

 connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO ()
 connect (h1, name1) (h2, name2) = do
-  ("c1", _, Right (INV cReq)) <- h1 #: ("c1", name2, "NEW INV")
+  ("c1", _, Right (INV cReq)) <- h1 #: ("c1", name2, "NEW T INV")
  let cReq' = strEncode cReq
-  h2 #: ("c2", name1, "JOIN " <> cReq' <> " 5\ninfo2") #> ("c2", name1, OK)
+  h2 #: ("c2", name1, "JOIN T " <> cReq' <> " 5\ninfo2") #> ("c2", name1, OK)
  ("", _, Right (CONF connId _ "info2")) <- (h1 <#:)
  h1 #: ("c3", name2, "LET " <> connId <> " 5\ninfo1") #> ("c3", name2, OK)
  h2 <# ("", name1, INFO "info1")
@@ -452,9 +453,9 @@ sendMessage (h1, name1) (h2, name2) msg = do

 -- connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)
 -- connect' h1 h2 = do
--   ("c1", conn2, Right (INV cReq)) <- h1 #: ("c1", "", "NEW INV")
+--   ("c1", conn2, Right (INV cReq)) <- h1 #: ("c1", "", "NEW T INV")
 --   let cReq' = strEncode cReq
--   ("c2", conn1, Right OK) <- h2 #: ("c2", "", "JOIN " <> cReq' <> " 5\ninfo2")
+--   ("c2", conn1, Right OK) <- h2 #: ("c2", "", "JOIN T " <> cReq' <> " 5\ninfo2")
 --   ("", _, Right (REQ connId _ "info2")) <- (h1 <#:)
 --   h1 #: ("c3", conn2, "ACPT " <> connId <> " 5\ninfo1") =#> \case ("c3", c, OK) -> c == conn2; _ -> False
 --   h2 <# ("", conn1, INFO "info1")
@@ -471,17 +472,17 @@ syntaxTests t = do
  describe "NEW" $ do
    describe "valid" $ do
      -- TODO: add tests with defined connection id
-      it "with correct parameter" $ ("211", "", "NEW INV") >#>= \case ("211", _, "INV" : _) -> True; _ -> False
+      it "with correct parameter" $ ("211", "", "NEW T INV") >#>= \case ("211", _, "INV" : _) -> True; _ -> False
    describe "invalid" $ do
      -- TODO: add tests with defined connection id
-      it "with incorrect parameter" $ ("222", "", "NEW hi") >#> ("222", "", "ERR CMD SYNTAX")
+      it "with incorrect parameter" $ ("222", "", "NEW T hi") >#> ("222", "", "ERR CMD SYNTAX")

  describe "JOIN" $ do
    describe "valid" $ do
      it "using same server as in invitation" $
        ( "311",
          "a",
-          "JOIN https://simpex.chat/invitation#/?smp=smp%3A%2F%2F"
+          "JOIN T https://simpex.chat/invitation#/?smp=smp%3A%2F%2F"
            <> urlEncode True "LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI="
            <> "%40localhost%3A5001%2F3456-w%3D%3D%23"
            <> urlEncode True sampleDhKey
@@ -114,53 +114,58 @@ functionalAPITests t = do
  describe "Batching SMP commands" $ do
    it "should subscribe to multiple subscriptions with batching" $
      testBatchedSubscriptions t
+  describe "Async agent commands" $ do
+    it "should connect using async agent commands" $
+      withSmpServer t testAsyncCommands
+    it "should restore and complete async commands on restart" $
+      testAsyncCommandsRestore t

 testAgentClient :: IO ()
 testAgentClient = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  runAgentClientTest alice bob 3

 testAgentClientV1toV1 :: IO ()
 testAgentClientV1toV1 = do
  alice <- getSMPAgentClient agentCfgV1 initAgentServers
-  bob <- getSMPAgentClient agentCfgV1 {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfgV1 {database = testDB2} initAgentServers
  runAgentClientTest alice bob 4

 testAgentClientV1toV2 :: IO ()
 testAgentClientV1toV2 = do
  alice <- getSMPAgentClient agentCfgV1 initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  runAgentClientTest alice bob 4

 testAgentClientV2toV1 :: IO ()
 testAgentClientV2toV1 = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfgV1 {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfgV1 {database = testDB2} initAgentServers
  runAgentClientTest alice bob 4

 testAgentClientContact :: IO ()
 testAgentClientContact = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  runAgentClientContactTest alice bob 3

 testAgentClientContactV1toV1 :: IO ()
 testAgentClientContactV1toV1 = do
  alice <- getSMPAgentClient agentCfgV1 initAgentServers
-  bob <- getSMPAgentClient agentCfgV1 {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfgV1 {database = testDB2} initAgentServers
  runAgentClientContactTest alice bob 4

 testAgentClientContactV1toV2 :: IO ()
 testAgentClientContactV1toV2 = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfgV1 {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfgV1 {database = testDB2} initAgentServers
  runAgentClientContactTest alice bob 4

 testAgentClientContactV2toV1 :: IO ()
 testAgentClientContactV2toV1 = do
  alice <- getSMPAgentClient agentCfgV1 initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  runAgentClientContactTest alice bob 4

 runAgentClientTest :: AgentClient -> AgentClient -> AgentMsgId -> IO ()
@@ -248,7 +253,7 @@ noMessages c err = tryGet `shouldReturn` ()
 testAsyncInitiatingOffline :: IO ()
 testAsyncInitiatingOffline = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right () <- runExceptT $ do
    (bobId, cReq) <- createConnection alice True SCMInvitation
    disconnectAgentClient alice
@@ -266,14 +271,14 @@ testAsyncInitiatingOffline = do
 testAsyncJoiningOfflineBeforeActivation :: IO ()
 testAsyncJoiningOfflineBeforeActivation = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right () <- runExceptT $ do
    (bobId, qInfo) <- createConnection alice True SCMInvitation
    aliceId <- joinConnection bob True qInfo "bob's connInfo"
    disconnectAgentClient bob
    ("", _, CONF confId _ "bob's connInfo") <- get alice
    allowConnection alice bobId confId "alice's connInfo"
-    bob' <- liftIO $ getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+    bob' <- liftIO $ getSMPAgentClient agentCfg {database = testDB2} initAgentServers
    subscribeConnection bob' aliceId
    get alice ##> ("", bobId, CON)
    get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -284,7 +289,7 @@ testAsyncJoiningOfflineBeforeActivation = do
 testAsyncBothOffline :: IO ()
 testAsyncBothOffline = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right () <- runExceptT $ do
    (bobId, cReq) <- createConnection alice True SCMInvitation
    disconnectAgentClient alice
@@ -294,7 +299,7 @@ testAsyncBothOffline = do
    subscribeConnection alice' bobId
    ("", _, CONF confId _ "bob's connInfo") <- get alice'
    allowConnection alice' bobId confId "alice's connInfo"
-    bob' <- liftIO $ getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+    bob' <- liftIO $ getSMPAgentClient agentCfg {database = testDB2} initAgentServers
    subscribeConnection bob' aliceId
    get alice' ##> ("", bobId, CON)
    get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -305,7 +310,7 @@ testAsyncBothOffline = do
 testAsyncServerOffline :: ATransport -> IO ()
 testAsyncServerOffline t = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  -- create connection and shutdown the server
  Right (bobId, cReq) <- withSmpServerStoreLogOn t testPort $ \_ ->
    runExceptT $ createConnection alice True SCMInvitation
@@ -333,7 +338,7 @@ testAsyncHelloTimeout :: IO ()
 testAsyncHelloTimeout = do
  -- this test would only work if any of the agent is v1, there is no HELLO timeout in v2
  alice <- getSMPAgentClient agentCfgV1 initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2, helloTimeout = 1} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2, helloTimeout = 1} initAgentServers
  Right () <- runExceptT $ do
    (_, cReq) <- createConnection alice True SCMInvitation
    disconnectAgentClient alice
@@ -344,7 +349,7 @@ testAsyncHelloTimeout = do
 testDuplicateMessage :: ATransport -> IO ()
 testDuplicateMessage t = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  (aliceId, bobId, bob1) <- withSmpServerStoreMsgLogOn t testPort $ \_ -> do
    Right (aliceId, bobId) <- runExceptT $ makeConnection alice bob
    Right () <- runExceptT $ do
@@ -354,7 +359,7 @@ testDuplicateMessage t = do
    disconnectAgentClient bob

    -- if the agent user did not send ACK, the message will be delivered again
-    bob1 <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+    bob1 <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
    Right () <- runExceptT $ do
      subscribeConnection bob1 aliceId
      get bob1 =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
@@ -376,7 +381,7 @@ testDuplicateMessage t = do
  disconnectAgentClient bob1

  alice2 <- getSMPAgentClient agentCfg initAgentServers
-  bob2 <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob2 <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers

  withSmpServerStoreMsgLogOn t testPort $ \_ -> do
    Right () <- runExceptT $ do
@@ -424,7 +429,7 @@ testActiveClientNotDisconnected t = do
  where
    keepSubscribing :: AgentClient -> ConnId -> SystemTime -> ExceptT AgentErrorType IO ()
    keepSubscribing alice connId ts = do
-      ts' <- liftIO $ getSystemTime
+      ts' <- liftIO getSystemTime
      if milliseconds ts' - milliseconds ts < 2200
        then do
          -- keep sending SUB for 2.2 seconds
@@ -442,7 +447,7 @@ testActiveClientNotDisconnected t = do
 testSuspendingAgent :: IO ()
 testSuspendingAgent = do
  a <- getSMPAgentClient agentCfg initAgentServers
-  b <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  b <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right () <- runExceptT $ do
    (aId, bId) <- makeConnection a b
    4 <- sendMessage a bId SMP.noMsgFlags "hello"
@@ -461,7 +466,7 @@ testSuspendingAgent = do
 testSuspendingAgentCompleteSending :: ATransport -> IO ()
 testSuspendingAgentCompleteSending t = do
  a <- getSMPAgentClient agentCfg initAgentServers
-  b <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  b <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runExceptT $ do
    (aId, bId) <- makeConnection a b
    4 <- sendMessage a bId SMP.noMsgFlags "hello"
@@ -495,7 +500,7 @@ testSuspendingAgentCompleteSending t = do
 testSuspendingAgentTimeout :: ATransport -> IO ()
 testSuspendingAgentTimeout t = do
  a <- getSMPAgentClient agentCfg initAgentServers
-  b <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  b <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right (aId, _) <- withSmpServer t . runExceptT $ do
    (aId, bId) <- makeConnection a b
    4 <- sendMessage a bId SMP.noMsgFlags "hello"
@@ -518,7 +523,7 @@ testSuspendingAgentTimeout t = do
 testBatchedSubscriptions :: ATransport -> IO ()
 testBatchedSubscriptions t = do
  a <- getSMPAgentClient agentCfg initAgentServers2
-  b <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers2
+  b <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers2
  Right conns <- runServers $ do
    conns <- forM [1 .. 200 :: Int] . const $ makeConnection a b
    forM_ conns $ \(aId, bId) -> exchangeGreetings a bId b aId
@@ -560,6 +565,66 @@ testBatchedSubscriptions t = do
        killThread t1
        pure res

+testAsyncCommands :: IO ()
+testAsyncCommands = do
+  alice <- getSMPAgentClient agentCfg initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
+  Right () <- runExceptT $ do
+    bobId <- createConnectionAsync alice "1" True SCMInvitation
+    ("1", bobId', INV (ACR _ qInfo)) <- get alice
+    liftIO $ bobId' `shouldBe` bobId
+    aliceId <- joinConnectionAsync bob "2" True qInfo "bob's connInfo"
+    ("2", aliceId', OK) <- get bob
+    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 ##> ("", bobId, CON)
+    get bob ##> ("", aliceId, INFO "alice's connInfo")
+    get bob ##> ("", aliceId, CON)
+    -- message IDs 1 to 3 get assigned to control messages, so first MSG is assigned ID 4
+    1 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "hello"
+    get alice ##> ("", bobId, SENT $ baseId + 1)
+    2 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
+    get alice ##> ("", bobId, SENT $ baseId + 2)
+    get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
+    ackMessageAsync bob "4" aliceId $ baseId + 1
+    ("4", _, OK) <- get bob
+    get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
+    ackMessageAsync bob "5" aliceId $ baseId + 2
+    ("5", _, OK) <- get bob
+    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
+    ("6", _, OK) <- get alice
+    get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
+    ackMessageAsync alice "7" bobId $ baseId + 4
+    ("7", _, OK) <- get alice
+    deleteConnectionAsync alice "8" bobId
+    ("8", _, OK) <- get alice
+    liftIO $ noMessages alice "nothing else should be delivered to alice"
+  pure ()
+  where
+    baseId = 3
+    msgId = subtract baseId
+
+testAsyncCommandsRestore :: ATransport -> IO ()
+testAsyncCommandsRestore t = do
+  alice <- getSMPAgentClient agentCfg initAgentServers
+  Right bobId <- runExceptT $ createConnectionAsync alice "1" True SCMInvitation
+  liftIO $ noMessages alice "alice doesn't receive INV because server is down"
+  disconnectAgentClient alice
+  alice' <- liftIO $ getSMPAgentClient agentCfg initAgentServers
+  withSmpServerStoreLogOn t testPort $ \_ -> do
+    Right () <- runExceptT $ do
+      subscribeConnection alice' bobId
+      ("1", _, INV _) <- get alice'
+      pure ()
+    pure ()
+
 exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
 exchangeGreetings = exchangeGreetingsMsgId 4

@@ -22,7 +22,7 @@ import NtfClient
 import SMPAgentClient (agentCfg, initAgentServers, testDB, testDB2)
 import SMPClient (testPort, withSmpServer, withSmpServerStoreLogOn)
 import Simplex.Messaging.Agent
-import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..))
+import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), databaseFile)
 import Simplex.Messaging.Agent.Protocol
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
@@ -44,7 +44,7 @@ removeFileIfExists filePath = do

 notificationTests :: ATransport -> Spec
 notificationTests t =
-  after_ (removeFile testDB >> removeFileIfExists testDB2) $ do
+  after_ (removeFile (databaseFile testDB) >> removeFileIfExists (databaseFile testDB2)) $ do
    describe "Managing notification tokens" $ do
      it "should register and verify notification token" $
        withAPNSMockServer $ \apns ->
@@ -142,7 +142,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
  -- setLogLevel LogError -- LogDebug
  -- withGlobalLogging logCfg $ do
  a <- getSMPAgentClient agentCfg initAgentServers
-  a' <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  a' <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right () <- runExceptT $ do
    let tkn = DeviceToken PPApnsTest "abcd"
    NTRegistered <- registerNtfToken a tkn NMPeriodic
@@ -207,7 +207,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
 testNotificationSubscriptionExistingConnection :: APNSMockServer -> IO ()
 testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right (bobId, aliceId, nonce, message) <- runExceptT $ do
    -- establish connection
    (bobId, qInfo) <- createConnection alice True SCMInvitation
@@ -264,7 +264,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
 testNotificationSubscriptionNewConnection :: APNSMockServer -> IO ()
 testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right () <- runExceptT $ do
    -- alice registers notification token
    _ <- registerTestToken alice "abcd" NMInstant apnsQ
@@ -276,31 +276,25 @@ testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
    liftIO $ threadDelay 1000000
    aliceId <- joinConnection bob True qInfo "bob's connInfo"
    liftIO $ threadDelay 750000
-    liftIO $ print 0
    void $ messageNotification apnsQ
    ("", _, CONF confId _ "bob's connInfo") <- get alice
    liftIO $ threadDelay 500000
    allowConnection alice bobId confId "alice's connInfo"
-    liftIO $ print 1
    void $ messageNotification apnsQ
    get bob ##> ("", aliceId, INFO "alice's connInfo")
-    liftIO $ print 2
    void $ messageNotification apnsQ
    get alice ##> ("", bobId, CON)
-    liftIO $ print 3
    void $ messageNotification apnsQ
    get bob ##> ("", aliceId, CON)
    -- bob sends message
    1 <- msgId <$> sendMessage bob aliceId (SMP.MsgFlags True) "hello"
    get bob ##> ("", aliceId, SENT $ baseId + 1)
-    liftIO $ print 4
    void $ messageNotification apnsQ
    get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
    ackMessage alice bobId $ baseId + 1
    -- alice sends message
    2 <- msgId <$> sendMessage alice bobId (SMP.MsgFlags True) "hey there"
    get alice ##> ("", bobId, SENT $ baseId + 2)
-    liftIO $ print 5
    void $ messageNotification apnsQ
    get bob =##> \case ("", c, Msg "hey there") -> c == aliceId; _ -> False
    ackMessage bob aliceId $ baseId + 2
@@ -327,7 +321,7 @@ registerTestToken a token mode apnsQ = do
 testChangeNotificationsMode :: APNSMockServer -> IO ()
 testChangeNotificationsMode APNSMockServer {apnsQ} = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right () <- runExceptT $ do
    -- establish connection
    (bobId, qInfo) <- createConnection alice True SCMInvitation
@@ -392,7 +386,7 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
 testChangeToken :: APNSMockServer -> IO ()
 testChangeToken APNSMockServer {apnsQ} = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right (aliceId, bobId) <- runExceptT $ do
    -- establish connection
    (bobId, qInfo) <- createConnection alice True SCMInvitation
@@ -436,7 +430,7 @@ testChangeToken APNSMockServer {apnsQ} = do
 testNotificationsStoreLog :: ATransport -> APNSMockServer -> IO ()
 testNotificationsStoreLog t APNSMockServer {apnsQ} = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right (aliceId, bobId) <- withNtfServerStoreLog t $ \threadId -> runExceptT $ do
    (aliceId, bobId) <- makeConnection alice bob
    _ <- registerTestToken alice "abcd" NMInstant apnsQ
@@ -463,7 +457,7 @@ testNotificationsStoreLog t APNSMockServer {apnsQ} = do
 testNotificationsSMPRestart :: ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
  alice <- getSMPAgentClient agentCfg initAgentServers
-  bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
+  bob <- getSMPAgentClient agentCfg {database = testDB2} initAgentServers
  Right (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \threadId -> runExceptT $ do
    (aliceId, bobId) <- makeConnection alice bob
    _ <- registerTestToken alice "abcd" NMInstant apnsQ
@@ -43,7 +43,7 @@ withStore2 = before connect2 . after (removeStore . fst)
    connect2 :: IO (SQLiteStore, SQLiteStore)
    connect2 = do
      s1 <- createStore
-      s2 <- connectSQLiteStore (dbFilePath s1)
+      s2 <- connectSQLiteStore (dbFilePath s1) ""
      pure (s1, s2)

 createStore :: IO SQLiteStore
@@ -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) Migrations.app True
+  createSQLiteStore (testDB <> show r) "" Migrations.app True

 removeStore :: SQLiteStore -> IO ()
 removeStore db = do
@@ -20,7 +20,7 @@ schemaDumpTest =

 testVerifySchemaDump :: IO ()
 testVerifySchemaDump = do
-  void $ createSQLiteStore testDB Migrations.app False
+  void $ createSQLiteStore testDB "" Migrations.app False
  void $ readCreateProcess (shell $ "touch " <> schema) ""
  savedSchema <- readFile schema
  savedSchema `seq` pure ()
@@ -48,21 +48,21 @@ agentTestPort2 = "5011"
 agentTestPort3 :: ServiceName
 agentTestPort3 = "5012"

-testDB :: String
-testDB = "tests/tmp/smp-agent.test.protocol.db"
+testDB :: AgentDatabase
+testDB = AgentDBFile {dbFile = "tests/tmp/smp-agent.test.protocol.db", dbKey = ""}

-testDB2 :: String
-testDB2 = "tests/tmp/smp-agent2.test.protocol.db"
+testDB2 :: AgentDatabase
+testDB2 = AgentDBFile {dbFile = "tests/tmp/smp-agent2.test.protocol.db", dbKey = ""}

-testDB3 :: String
-testDB3 = "tests/tmp/smp-agent3.test.protocol.db"
+testDB3 :: AgentDatabase
+testDB3 = AgentDBFile {dbFile = "tests/tmp/smp-agent3.test.protocol.db", dbKey = ""}

 smpAgentTest :: forall c. Transport c => TProxy c -> ARawTransmission -> IO ARawTransmission
 smpAgentTest _ cmd = runSmpAgentTest $ \(h :: c) -> tPutRaw h cmd >> get h
  where
    get h = do
      t@(_, _, cmdStr) <- tGetRaw h
-      case parseAll commandP cmdStr of
+      case parseAll networkCommandP cmdStr of
        Right (ACmd SAgent CONNECT {}) -> get h
        Right (ACmd SAgent DISCONNECT {}) -> get h
        _ -> pure t
@@ -83,10 +83,10 @@ runSmpAgentServerTest test =
 smpAgentServerTest :: Transport c => ((ThreadId, ThreadId) -> c -> IO ()) -> Expectation
 smpAgentServerTest test' = runSmpAgentServerTest test' `shouldReturn` ()

-runSmpAgentTestN :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m, MonadFail m) => [(ServiceName, ServiceName, String)] -> ([c] -> m a) -> m a
+runSmpAgentTestN :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m, MonadFail m) => [(ServiceName, ServiceName, AgentDatabase)] -> ([c] -> m a) -> m a
 runSmpAgentTestN agents test = withSmpServer t $ run agents []
  where
-    run :: [(ServiceName, ServiceName, String)] -> [c] -> m a
+    run :: [(ServiceName, ServiceName, AgentDatabase)] -> [c] -> m a
    run [] hs = test hs
    run (a@(p, _, _) : as) hs = withSmpAgentOn t a $ testSMPAgentClientOn p $ \h -> run as (h : hs)
    t = transport @c
@@ -99,7 +99,7 @@ runSmpAgentTestN_1 nClients test = withSmpServer t . withSmpAgent t $ run nClien
    run n hs = testSMPAgentClient $ \h -> run (n - 1) (h : hs)
    t = transport @c

-smpAgentTestN :: Transport c => [(ServiceName, ServiceName, String)] -> ([c] -> IO ()) -> Expectation
+smpAgentTestN :: Transport c => [(ServiceName, ServiceName, AgentDatabase)] -> ([c] -> IO ()) -> Expectation
 smpAgentTestN agents test' = runSmpAgentTestN agents test' `shouldReturn` ()

 smpAgentTestN_1 :: Transport c => Int -> ([c] -> IO ()) -> Expectation
@@ -187,7 +187,7 @@ agentCfg =
  defaultAgentConfig
    { tcpPort = agentTestPort,
      tbqSize = 4,
-      dbFile = testDB,
+      database = testDB,
      smpCfg =
        defaultClientConfig
          { qSize = 1,
@@ -207,18 +207,18 @@ agentCfg =
      certificateFile = "tests/fixtures/server.crt"
    }

-withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m () -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, AgentDatabase) -> m () -> (ThreadId -> m a) -> m a
 withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =
-  let cfg' = agentCfg {tcpPort = port', dbFile = db'}
+  let cfg' = agentCfg {tcpPort = port', database = db'}
      initServers' = initAgentServers {smp = L.fromList [SMPServer "localhost" smpPort' testKeyHash]}
   in serverBracket
        (\started -> runSMPAgentBlocking t started cfg' initServers')
        afterProcess

-withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a
-withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ removeFile db'
+withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, AgentDatabase) -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ removeFile (dbFile db')

-withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m a -> m a
+withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, AgentDatabase) -> m a -> m a
 withSmpAgentOn t (port', smpPort', db') = withSmpAgentThreadOn t (port', smpPort', db') . const

 withSmpAgent :: (MonadUnliftIO m, MonadRandom m) => ATransport -> m a -> m a
@@ -146,8 +146,10 @@ testCreateSecureV2 _ =
      (ok2, OK) #== "secures queue"
      (rId2, rId) #== "same queue ID in response 3"

-      Resp "abcd" _ err4 <- signSendRecv h rKey ("abcd", rId, KEY sPub)
-      (err4, ERR AUTH) #== "rejects KEY if already secured"
+      Resp "abcd" _ OK <- signSendRecv h rKey ("abcd", rId, KEY sPub)
+      (sPub', _) <- C.generateSignatureKeyPair C.SEd448
+      Resp "abcd" _ err4 <- signSendRecv h rKey ("abcd", rId, KEY sPub')
+      (err4, ERR AUTH) #== "rejects if secured with different key"

      Resp "bcda" _ ok3 <- signSendRecv h sKey ("bcda", sId, _SEND "hello again")
      (ok3, OK) #== "accepts signed SEND"
@@ -208,8 +210,10 @@ testCreateSecure (ATransport t) =
      (ok2, OK) #== "secures queue"
      (rId2, rId) #== "same queue ID in response 3"

-      Resp "abcd" _ err4 <- signSendRecv h rKey ("abcd", rId, KEY sPub)
-      (err4, ERR AUTH) #== "rejects KEY if already secured"
+      Resp "abcd" _ OK <- signSendRecv h rKey ("abcd", rId, KEY sPub)
+      (sPub', _) <- C.generateSignatureKeyPair C.SEd448
+      Resp "abcd" _ err4 <- signSendRecv h rKey ("abcd", rId, KEY sPub')
+      (err4, ERR AUTH) #== "rejects if secured with different key"

      Resp "bcda" _ ok3 <- signSendRecv h sKey ("bcda", sId, _SEND "hello again")
      (ok3, OK) #== "accepts signed SEND"
Author	SHA1	Message	Date
Evgeny Poberezkin	b686bc0c17	3.3.0 (#540 )	2022-10-01 12:08:54 +01:00
Evgeny Poberezkin	29332a5e9f	lints (#539 ) * lints * revert * add import * remove StrictData	2022-10-01 11:51:52 +01:00
Evgeny Poberezkin	9cd5f96fe4	import	2022-09-30 12:36:21 +01:00
JRoberts	a2eea4f18f	add deleteConnectionAsync functional api (#538 )	2022-09-29 21:11:36 +04:00
Evgeny Poberezkin	413aad5139	allow passing agent store instead of path and key (#535 )	2022-09-23 18:45:00 +01:00
Evgeny Poberezkin	0a9b6e4ab4	allow repeated secure with the same key (KEY command), store suspended queues in store log (#534 ) * allow repeated secure with the same key (KEY command), store suspended queues in store log * fix test	2022-09-22 13:09:39 +01:00
Evgeny Poberezkin	f8505d4add	make function processing sql rows module-level (#533 )	2022-09-22 12:12:24 +01:00
Evgeny Poberezkin	41d3c14157	secure queue asynchronously (#532 ) * secure queue asynchronously * delete commands with internal logic errors	2022-09-20 15:11:35 +01:00
Evgeny Poberezkin	d4962daf11	internal async commands (#530 ) * internal async commands * rename * remove GADT from AgentCommand	2022-09-18 14:02:20 +01:00
Evgeny Poberezkin	085b785ee8	readme typo	2022-09-18 13:40:20 +01:00
shandEvgeny Poberezkin	84ce7a90d0	readme: add tor configuration (#524 ) * readme: add tor configuration * update docs Co-authored-by: Evgeny Poberezkin <[email protected]>	2022-09-18 13:36:02 +01:00
Evgeny Poberezkin	abb3f20ed5	replace TVar with IORef in execSQL (#531 )	2022-09-18 13:25:06 +01:00
Evgeny Poberezkin	a3f58fdc6b	api to execute any sql query (#529 ) * api to execute any sql query * agent api	2022-09-17 15:08:58 +01:00
Evgeny Poberezkin	e3e05d474d	wrap agent command for async command processing, to allow extension with internal commands (#528 ) * wrap agent command for async command processing, to allow extension with internal commands * change encoding for constructor	2022-09-17 14:08:49 +01:00
Evgeny Poberezkin	13a9eee0cf	cancel threads asynchronously (#527 )	2022-09-17 00:18:58 +01:00
Evgeny Poberezkin	959ae34f16	fix agent not stopping and race conditions (#526 ) * fix agent not stopping and race conditions * simplify	2022-09-16 13:42:13 +01:00
Evgeny Poberezkin	8c1a9c529d	fix async commands (#525 )	2022-09-16 08:36:11 +01:00
Evgeny Poberezkin	f08d81252d	Merge pull request #509 from simplex-chat/sqlcipher switch to SQLCipher	2022-09-14 21:42:24 +01:00
Evgeny Poberezkin	afecefc3ad	Merge branch 'master' into sqlcipher	2022-09-14 18:22:46 +01:00
JRoberts	e328ae5d06	closeAgentClient async commands clean up (#523 )	2022-09-12 18:15:32 +01:00
Evgeny Poberezkin	605e6c64c6	simplify subscribeConnections (#522 )	2022-09-12 13:31:41 +01:00
JRoberts	4bf9f2b535	Eq for ACommandTag (#520 )	2022-09-10 22:03:51 +04:00
Evgeny Poberezkin	bab6a1577b	correlation IDs and command tags for async commands (#519 ) * agent protocol command tags * store/send async command correlation IDs * fix, refactor * delete failed command	2022-09-10 16:33:15 +01:00
Evgeny Poberezkin	ca6164167e	fix selecting different SMP server (#518 ) * fix selecting different SMP server * use different server with JOIN	2022-09-10 13:38:48 +01:00
Evgeny Poberezkin	42a96d6d00	refactor agent subscriptions with TMap2 (#517 ) * refactor agent subscriptions with TMap2 * refactor * refactor * comment	2022-09-09 16:31:57 +01:00
Evgeny PoberezkinandJRoberts	6adbc56021	try async commands without servers on different servers (#516 ) * refactor * retry commands with different servers * refactor * remove comment Co-authored-by: JRoberts <[email protected]> Co-authored-by: JRoberts <[email protected]>	2022-09-09 12:30:27 +01:00
JRobertsandEvgeny Poberezkin	882d2b123b	agent async commands (#513 ) * wip * wip * wip * wip * don't delete conn on join failure in async mode * test restart * remove unused pragma * simplify restore * rename variable * refactor * refactor * split * remove error * prohibit async join via contact request * refactor subscribeConnections * simplify * resum command processing on subscriptions * unfocus tests * schema * remove do * rename variable * corrections * move code * refactor getPendingCommands Co-authored-by: Evgeny Poberezkin <[email protected]>	2022-09-08 20:27:56 +04:00
Evgeny Poberezkin	da5058a0c4	remove connId from message delivery queue keys (#515 )	2022-09-06 14:39:50 +01:00
Evgeny Poberezkin	50c210c5c0	remove database key from SQLiteStore (#514 )	2022-09-05 12:55:14 +01:00
Evgeny Poberezkin	7d53504bb3	Merge branch 'master' into sqlcipher	2022-09-05 10:39:40 +01:00
Evgeny Poberezkin	e4b47825b5	functions to create and close store (#512 )	2022-09-02 15:42:37 +01:00
JRoberts	b215bd954d	remove token if token replace fails with permanent error (#511 )	2022-08-31 21:12:44 +04:00
Evgeny Poberezkin	26d149d17c	interpolate sql strings (#510 )	2022-08-31 17:57:38 +01:00
Evgeny Poberezkin	c66a7e371f	Merge branch 'master' into sqlcipher	2022-08-30 15:45:44 +01:00
Evgeny Poberezkin	f2c1455a27	fix network-transport at 0.5.4	2022-08-30 15:45:15 +01:00
Evgeny Poberezkin	f872c25f09	update dependencies (to not use SQL encryption unless flag is set)	2022-08-30 14:23:36 +01:00
Evgeny Poberezkin	e4b77ed9e6	use SQLCipher (#507 ) * use SQLCipher * pass database key via options, use local direct-sqlcipher and sqlcipher-simple * update stack.yaml * use dependencies in git * update sqlcipher dependencies	2022-08-30 12:31:41 +01:00