3.3.0 (#540 )

lints (#539 )
* lints * revert * add import * remove StrictData
2026-08-31 11:48:24 +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
41 changed files with 1493 additions and 509 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
@@ -51,6 +51,8 @@ library
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220608_v2
      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
@@ -80,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
@@ -113,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
@@ -124,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.*
@@ -174,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
@@ -185,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.*
@@ -236,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
@@ -247,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.*
@@ -298,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
@@ -309,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.*
@@ -377,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
@@ -390,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.*
@@ -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
@@ -88,8 +91,8 @@ import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 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
@@ -130,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
@@ -154,12 +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 :: TMap ConnId SMPServer,
+    subscrConns :: TVar (Set ConnId),
+    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,
@@ -205,12 +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 <- TM.empty
+  subscrConns <- newTVar S.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
@@ -222,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, 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)
@@ -261,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' (subscrConns 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

@@ -298,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 ()
@@ -404,15 +407,24 @@ 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 connMsgsQueued
  clear smpQueueMsgQueues
+  clear connCmdsQueued
+  clear asyncCmdQueues
  clear getMsgLocks
  where
-    clear :: (AgentClient -> TMap k a) -> IO ()
-    clear sel = atomically $ writeTVar (sel c) M.empty
+    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 =
@@ -426,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} =
@@ -508,17 +520,19 @@ newRcvQueue_ a c srv vRange = do
 subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()
 subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
  whenM (atomically . TM.member (server, rcvId) $ getMsgLocks c) . throwError $ CMD PROHIBITED
-  atomically $ addPendingSubscription c rq connId
+  atomically $ do
+    modifyTVar (subscrConns c) $ S.insert 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

@@ -538,7 +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_ $ atomically . uncurry (addPendingSubscription c) . swap
+  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)
@@ -551,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
@@ -560,38 +576,21 @@ subscribeQueues c srv qs = do

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

 hasActiveSubscription :: AgentClient -> ConnId -> STM Bool
-hasActiveSubscription c connId = TM.member connId (subscrConns 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@AgentClient {subscrConns} connId = do
-  server_ <- TM.lookupDelete connId subscrConns
-  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)
+removeSubscription c connId = do
+  modifyTVar (subscrConns c) $ S.delete connId
+  TM2.delete connId $ activeSubs c
+  TM2.delete connId $ pendingSubs c

 getSubscriptions :: AgentClient -> STM (Set ConnId)
-getSubscriptions AgentClient {subscrConns} = do
-  m <- readTVar subscrConns
-  pure $ M.keysSet m
+getSubscriptions = readTVar . subscrConns

 logServer :: MonadIO m => ByteString -> AgentClient -> ProtocolServer s -> QueueId -> ByteString -> m ()
 logServer dir AgentClient {clientId} srv qId cmdStr =
@@ -815,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)

@@ -828,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
@@ -151,10 +160,101 @@ deriving instance Show SomeConn
 data ConnData = ConnData
  { connId :: ConnId,
    connAgentVersion :: Version,
+    enableNtfs :: Bool,
    duplexHandshake :: Maybe Bool -- added in agent protocol v2
  }
  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
@@ -269,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.
@@ -292,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,11 +22,18 @@ module Simplex.Messaging.Agent.Store.SQLite
  ( SQLiteStore (..),
    createSQLiteStore,
    connectSQLiteStore,
+    closeSQLiteStore,
+    sqlString,
+    execSQL,

    -- * Queues and connections
+    createNewConn,
+    updateNewConnRcv,
+    updateNewConnSnd,
    createRcvConn,
    createSndConn,
    getConn,
+    getConnData,
    getRcvConn,
    deleteConn,
    upgradeRcvConnToDuplex,
@@ -46,6 +53,7 @@ module Simplex.Messaging.Agent.Store.SQLite
    createInvitation,
    getInvitation,
    acceptInvitation,
+    unacceptInvitation,
    deleteInvitation,
    -- Messages
    updateRcvIds,
@@ -64,6 +72,11 @@ module Simplex.Messaging.Agent.Store.SQLite
    getRatchet,
    getSkippedMsgKeys,
    updateRatchet,
+    -- Async commands
+    createCommand,
+    getPendingCommands,
+    getPendingCommand,
+    deleteCommand,
    -- Notification device token persistence
    createNtfToken,
    getSavedNtfToken,
@@ -84,6 +97,7 @@ module Simplex.Messaging.Agent.Store.SQLite
    getNextNtfSubSMPAction,
    getActiveNtfToken,
    getNtfRcvQueue,
+    setConnectionNtfs,

    -- * utilities
    withConnection,
@@ -102,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)
@@ -137,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

@@ -145,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
@@ -189,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
@@ -220,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)

@@ -255,18 +290,49 @@ 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 q@RcvQueue {server} cMode =
+createRcvConn db gVar cData@ConnData {connAgentVersion, enableNtfs, duplexHandshake} q@RcvQueue {server} cMode =
  createConn_ gVar cData $ \connId -> do
    upsertServer_ db server
-    DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, duplex_handshake) VALUES (?, ?, ?, ?)" (connId, cMode, connAgentVersion cData, duplexHandshake cData)
+    DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?, ?, ?, ?, ?)" (connId, cMode, connAgentVersion, enableNtfs, duplexHandshake)
    insertRcvQueue_ db connId q

 createSndConn :: DB.Connection -> TVar ChaChaDRG -> ConnData -> SndQueue -> IO (Either StoreError ConnId)
-createSndConn db gVar cData q@SndQueue {server} =
+createSndConn db gVar cData@ConnData {connAgentVersion, enableNtfs, duplexHandshake} q@SndQueue {server} =
  createConn_ gVar cData $ \connId -> do
    upsertServer_ db server
-    DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, duplex_handshake) VALUES (?, ?, ?, ?)" (connId, SCMInvitation, connAgentVersion cData, duplexHandshake cData)
+    DB.execute db "INSERT INTO connections (conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake) VALUES (?, ?, ?, ?, ?)" (connId, SCMInvitation, connAgentVersion, enableNtfs, duplexHandshake)
    insertSndQueue_ db connId q

 getRcvConn :: DB.Connection -> SMPServer -> SMP.RecipientId -> IO (Either StoreError SomeConn)
@@ -502,6 +568,10 @@ acceptInvitation db invitationId ownConnInfo =
      ":invitation_id" := invitationId
    ]

+unacceptInvitation :: DB.Connection -> InvitationId -> IO ()
+unacceptInvitation db invitationId =
+  DB.execute db "UPDATE conn_invitations SET accepted = 0, own_conn_info = NULL WHERE invitation_id = ?" (Only invitationId)
+
 deleteInvitation :: DB.Connection -> ConnId -> InvitationId -> IO (Either StoreError ())
 deleteInvitation db contactConnId invId =
  getConn db contactConnId $>>= \case
@@ -647,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
@@ -947,6 +1061,10 @@ getNtfRcvQueue db SMPQueueNtf {smpServer = (SMPServer host port _), notifierId}
    res (connId, Just rcvNtfDhSecret) = Right (connId, rcvNtfDhSecret)
    res _ = Left SEConnNotFound

+setConnectionNtfs :: DB.Connection -> ConnId -> Bool -> IO ()
+setConnectionNtfs db connId enableNtfs =
+  DB.execute db "UPDATE connections SET enable_ntfs = ? WHERE conn_id = ?" (enableNtfs, connId)
+
 -- * Auxiliary helpers

 instance ToField QueueStatus where toField = toField . serializeQueueStatus
@@ -1005,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
@@ -1102,7 +1228,7 @@ insertSndQueue_ dbConn connId SndQueue {..} = do

 getConn :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn)
 getConn dbConn connId =
-  getConnData_ dbConn connId >>= \case
+  getConnData dbConn connId >>= \case
    Nothing -> pure $ Left SEConnNotFound
    Just (connData, cMode) -> do
      rQ <- getRcvQueueByConnId_ dbConn connId
@@ -1112,14 +1238,15 @@ 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))
-getConnData_ dbConn connId' =
+getConnData :: DB.Connection -> ConnId -> IO (Maybe (ConnData, ConnectionMode))
+getConnData dbConn connId' =
  connData
-    <$> DB.query dbConn "SELECT conn_id, conn_mode, smp_agent_version, duplex_handshake FROM connections WHERE conn_id = ?;" (Only connId')
+    <$> DB.query dbConn "SELECT conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake FROM connections WHERE conn_id = ?;" (Only connId')
  where
-    connData [(connId, cMode, connAgentVersion, duplexHandshake)] = Just (ConnData {connId, connAgentVersion, duplexHandshake}, cMode)
+    connData [(connId, cMode, connAgentVersion, enableNtfs_, duplexHandshake)] = Just (ConnData {connId, connAgentVersion, enableNtfs = fromMaybe True enableNtfs_, duplexHandshake}, cMode)
    connData _ = Nothing

 getRcvQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe RcvQueue)
@@ -16,11 +16,10 @@ module Simplex.Messaging.Agent.Store.SQLite.Migrations
  )
 where

-import Control.Monad (forM_)
-import Data.List (intercalate, sortBy)
+import Control.Monad (forM_, when)
+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)
@@ -35,6 +34,8 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220322_notifications
 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)

@@ -48,12 +49,14 @@ schemaMigrations =
    ("20220322_notifications", m20220322_notifications),
    ("20220607_v2", m20220608_v2),
    ("m20220625_v2_ntf_mode", m20220625_v2_ntf_mode),
-    ("m20220811_onion_hosts", m20220811_onion_hosts)
+    ("m20220811_onion_hosts", m20220811_onion_hosts),
+    ("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}

@@ -63,16 +66,16 @@ get conn migrations =
    <$> DB.query_ conn "SELECT name FROM migrations ORDER BY name ASC;"

 run :: Connection -> [Migration] -> IO ()
-run conn ms = DB.withImmediateTransaction conn . forM_ ms $
-  \Migration {name, up} -> insert name >> execSQL up >> updateServers name
+run conn ms = forM_ ms $ \Migration {name, up} -> do
+  when (name == "m20220811_onion_hosts") updateServers
+  DB.withImmediateTransaction conn $ insert name >> execSQL up
  where
    insert name = DB.execute conn "INSERT INTO migrations (name, ts) VALUES (?, ?);" . (name,) =<< getCurrentTime
    execSQL = SQLite3.exec $ DB.connectionHandle conn
-    updateServers = \case
-      "m20220811_onion_hosts" -> forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
+    updateServers = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
+      DB.withImmediateTransaction conn $
        let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
         in DB.execute conn "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
-      _ -> pure ()

 initialize :: Connection -> IO ()
 initialize conn =
@@ -0,0 +1,12 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220817_connection_ntfs where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20220817_connection_ntfs :: Query
+m20220817_connection_ntfs =
+  [sql|
+ALTER TABLE connections ADD COLUMN enable_ntfs INTEGER;
+|]
@@ -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
+);
+|]
@@ -20,7 +20,8 @@ CREATE TABLE connections(
  last_snd_msg_hash BLOB NOT NULL DEFAULT x'',
  smp_agent_version INTEGER NOT NULL DEFAULT 1
  ,
-  duplex_handshake INTEGER NULL DEFAULT 0
+  duplex_handshake INTEGER NULL DEFAULT 0,
+  enable_ntfs INTEGER
 ) WITHOUT ROWID;
 CREATE TABLE rcv_queues(
  host TEXT NOT NULL,
@@ -193,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,60 +114,65 @@ 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 ()
 runAgentClientTest alice bob baseId = do
  Right () <- runExceptT $ do
-    (bobId, qInfo) <- createConnection alice SCMInvitation
-    aliceId <- joinConnection bob qInfo "bob's connInfo"
+    (bobId, qInfo) <- createConnection alice True SCMInvitation
+    aliceId <- joinConnection bob True qInfo "bob's connInfo"
    ("", _, CONF confId _ "bob's connInfo") <- get alice
    allowConnection alice bobId confId "alice's connInfo"
    get alice ##> ("", bobId, CON)
@@ -202,10 +207,10 @@ runAgentClientTest alice bob baseId = do
 runAgentClientContactTest :: AgentClient -> AgentClient -> AgentMsgId -> IO ()
 runAgentClientContactTest alice bob baseId = do
  Right () <- runExceptT $ do
-    (_, qInfo) <- createConnection alice SCMContact
-    aliceId <- joinConnection bob qInfo "bob's connInfo"
+    (_, qInfo) <- createConnection alice True SCMContact
+    aliceId <- joinConnection bob True qInfo "bob's connInfo"
    ("", _, REQ invId _ "bob's connInfo") <- get alice
-    bobId <- acceptContact alice invId "alice's connInfo"
+    bobId <- acceptContact alice True invId "alice's connInfo"
    ("", _, CONF confId _ "alice's connInfo") <- get bob
    allowConnection bob aliceId confId "bob's connInfo"
    get alice ##> ("", bobId, INFO "bob's connInfo")
@@ -248,11 +253,11 @@ 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 SCMInvitation
+    (bobId, cReq) <- createConnection alice True SCMInvitation
    disconnectAgentClient alice
-    aliceId <- joinConnection bob cReq "bob's connInfo"
+    aliceId <- joinConnection bob True cReq "bob's connInfo"
    alice' <- liftIO $ getSMPAgentClient agentCfg initAgentServers
    subscribeConnection alice' bobId
    ("", _, CONF confId _ "bob's connInfo") <- get 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 SCMInvitation
-    aliceId <- joinConnection bob qInfo "bob's connInfo"
+    (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,17 +289,17 @@ 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 SCMInvitation
+    (bobId, cReq) <- createConnection alice True SCMInvitation
    disconnectAgentClient alice
-    aliceId <- joinConnection bob cReq "bob's connInfo"
+    aliceId <- joinConnection bob True cReq "bob's connInfo"
    disconnectAgentClient bob
    alice' <- liftIO $ getSMPAgentClient agentCfg initAgentServers
    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,12 +310,12 @@ 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 SCMInvitation
+    runExceptT $ createConnection alice True SCMInvitation
  -- connection fails
-  Left (BROKER NETWORK) <- runExceptT $ joinConnection bob cReq "bob's connInfo"
+  Left (BROKER NETWORK) <- runExceptT $ joinConnection bob True cReq "bob's connInfo"
  ("", "", DOWN srv conns) <- get alice
  srv `shouldBe` testSMPServer
  conns `shouldBe` [bobId]
@@ -320,7 +325,7 @@ testAsyncServerOffline t = do
    liftIO $ do
      srv1 `shouldBe` testSMPServer
      conns1 `shouldBe` [bobId]
-    aliceId <- joinConnection bob cReq "bob's connInfo"
+    aliceId <- joinConnection bob True cReq "bob's connInfo"
    ("", _, CONF confId _ "bob's connInfo") <- get alice
    allowConnection alice bobId confId "alice's connInfo"
    get alice ##> ("", bobId, CON)
@@ -333,18 +338,18 @@ 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 SCMInvitation
+    (_, cReq) <- createConnection alice True SCMInvitation
    disconnectAgentClient alice
-    aliceId <- joinConnection bob cReq "bob's connInfo"
+    aliceId <- joinConnection bob True cReq "bob's connInfo"
    get bob ##> ("", aliceId, ERR $ CONN NOT_ACCEPTED)
  pure ()

 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
@@ -392,8 +397,8 @@ testDuplicateMessage t = do

 makeConnection :: AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnection alice bob = do
-  (bobId, qInfo) <- createConnection alice SCMInvitation
-  aliceId <- joinConnection bob qInfo "bob's connInfo"
+  (bobId, qInfo) <- createConnection alice True SCMInvitation
+  aliceId <- joinConnection bob True qInfo "bob's connInfo"
  ("", _, CONF confId _ "bob's connInfo") <- get alice
  allowConnection alice bobId confId "alice's connInfo"
  get alice ##> ("", bobId, CON)
@@ -407,7 +412,7 @@ testInactiveClientDisconnected t = do
  withSmpServerConfigOn t cfg' testPort $ \_ -> do
    alice <- getSMPAgentClient agentCfg initAgentServers
    Right () <- runExceptT $ do
-      (connId, _cReq) <- createConnection alice SCMInvitation
+      (connId, _cReq) <- createConnection alice True SCMInvitation
      get alice ##> ("", "", DOWN testSMPServer [connId])
    pure ()

@@ -418,13 +423,13 @@ testActiveClientNotDisconnected t = do
    alice <- getSMPAgentClient agentCfg initAgentServers
    ts <- getSystemTime
    Right () <- runExceptT $ do
-      (connId, _cReq) <- createConnection alice SCMInvitation
+      (connId, _cReq) <- createConnection alice True SCMInvitation
      keepSubscribing alice connId ts
    pure ()
  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,11 +207,11 @@ 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 SCMInvitation
-    aliceId <- joinConnection bob qInfo "bob's connInfo"
+    (bobId, qInfo) <- createConnection alice True SCMInvitation
+    aliceId <- joinConnection bob True qInfo "bob's connInfo"
    ("", _, CONF confId _ "bob's connInfo") <- get alice
    allowConnection alice bobId confId "alice's connInfo"
    get bob ##> ("", aliceId, INFO "alice's connInfo")
@@ -249,7 +249,8 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
    get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
    ackMessage alice bobId $ baseId + 1
    -- delete notification subscription
-    deleteNtfSub alice bobId
+    toggleConnectionNtfs alice bobId False
+    liftIO $ threadDelay 250000
    -- send message
    2 <- msgId <$> sendMessage bob aliceId (SMP.MsgFlags True) "hello again"
    get bob ##> ("", aliceId, SENT $ baseId + 2)
@@ -263,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
@@ -271,34 +272,29 @@ testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
    _ <- registerTestToken bob "bcde" NMInstant apnsQ
    -- establish connection
    liftIO $ threadDelay 50000
-    (bobId, qInfo) <- createConnection alice SCMInvitation
-    liftIO $ threadDelay 500000
-    aliceId <- joinConnection bob qInfo "bob's connInfo"
-    liftIO $ print 0
+    (bobId, qInfo) <- createConnection alice True SCMInvitation
+    liftIO $ threadDelay 1000000
+    aliceId <- joinConnection bob True qInfo "bob's connInfo"
+    liftIO $ threadDelay 750000
    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
@@ -325,11 +321,11 @@ 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 SCMInvitation
-    aliceId <- joinConnection bob qInfo "bob's connInfo"
+    (bobId, qInfo) <- createConnection alice True SCMInvitation
+    aliceId <- joinConnection bob True qInfo "bob's connInfo"
    ("", _, CONF confId _ "bob's connInfo") <- get alice
    allowConnection alice bobId confId "alice's connInfo"
    get bob ##> ("", aliceId, INFO "alice's connInfo")
@@ -347,7 +343,7 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
    -- set mode to NMPeriodic
    NTActive <- registerNtfToken alice tkn NMPeriodic
    -- send message, no notification
-    liftIO $ threadDelay 500000
+    liftIO $ threadDelay 750000
    2 <- msgId <$> sendMessage bob aliceId (SMP.MsgFlags True) "hello again"
    get bob ##> ("", aliceId, SENT $ baseId + 2)
    noNotification apnsQ
@@ -390,11 +386,11 @@ 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 SCMInvitation
-    aliceId <- joinConnection bob qInfo "bob's connInfo"
+    (bobId, qInfo) <- createConnection alice True SCMInvitation
+    aliceId <- joinConnection bob True qInfo "bob's connInfo"
    ("", _, CONF confId _ "bob's connInfo") <- get alice
    allowConnection alice bobId confId "alice's connInfo"
    get bob ##> ("", aliceId, INFO "alice's connInfo")
@@ -434,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
@@ -461,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
@@ -139,7 +139,7 @@ testForeignKeysEnabled =
      `shouldThrow` (\e -> DB.sqlError e == DB.ErrorConstraint)

 cData1 :: ConnData
-cData1 = ConnData {connId = "conn1", connAgentVersion = 1, duplexHandshake = Nothing}
+cData1 = ConnData {connId = "conn1", connAgentVersion = 1, enableNtfs = True, duplexHandshake = Nothing}

 testPrivateSignKey :: C.APrivateSignKey
 testPrivateSignKey = C.APrivateSignKey C.SEd25519 "MC4CAQAwBQYDK2VwBCIEIDfEfevydXXfKajz3sRkcQ7RPvfWUPoq6pu1TYHV1DEe"
@@ -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 <2769109+epoberezkin@users.noreply.github.com>	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 <8711996+jr-simplex@users.noreply.github.com> Co-authored-by: JRoberts <8711996+jr-simplex@users.noreply.github.com>	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 <2769109+epoberezkin@users.noreply.github.com>	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
Evgeny Poberezkin	5894f91927	fix connections passed to ntf supervisor to include pending, not only active (#506 ) * fix connections passed to ntf supervisor to include pending, not only active * fix * fix 2	2022-08-30 08:26:42 +01:00
Evgeny Poberezkin	1459a0613a	set invitation as not accepted in case of join error (#505 ) * set invitation as not accepted in case of join error * remove unused name	2022-08-25 12:17:55 +01:00
Evgeny Poberezkin	a7b39b710c	fix toggleConnectionNtfs (#503 )	2022-08-20 13:26:25 +01:00
Evgeny Poberezkin	212a9bc4ab	enable/disable connection notifications (#501 ) * enable/disable connection notifications * filter connections when intializing/disabling subscriptions * fix * fix 2 * add delay * increase test delay * increase test delay * fix * increase test delays * increase test delays	2022-08-19 15:23:57 +01:00
Evgeny Poberezkin	3fc9afbb35	move servers update before migration (so it re-tries in case of crash)	2022-08-18 08:39:22 +01:00
Evgeny Poberezkin	b2d4bfd0a8	split migration to multiple transactions (trying to prevent Android crash) (#502 )	2022-08-17 22:08:01 +01:00
Evgeny Poberezkin	cfab5a18d3	move server update to SQL migration (#500 )	2022-08-16 21:31:19 +01:00