migration

* make sending confirmation asynchronous

* confirm first queue synchronously (on JOIN, and delete connection on failure), second queue asynchronously (from REPLY msg)

* test to retry JOIN when the first attempt fails

* process confirmation delivery errors

CHANGELOG.md

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

apps/smp-agent/Main.hs

@@ -6,12 +6,12 @@ module Main where
 
 import Control.Logger.Simple
 import qualified Data.List.NonEmpty as L
-import Simplex.Messaging.Agent.Env.Postgres
+import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Server (runSMPAgent)
 import Simplex.Messaging.Transport (TLS, Transport (..))
 
 cfg :: AgentConfig
-cfg = defaultAgentConfig {smpServers = L.fromList ["smp://bU0K-bRg24xWW__lS0umO1Zdw_SXqpJNtm1_RrPLViE=@localhost:5223"]}
+cfg = defaultAgentConfig {initialSMPServers = L.fromList ["smp://bU0K-bRg24xWW__lS0umO1Zdw_SXqpJNtm1_RrPLViE=@localhost:5223"]}
 
 logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}

apps/smp-server/Main.hs

@@ -277,8 +277,8 @@ runServer IniOptions {enableStoreLog, port, enableWebsockets} = do
       ServerConfig
         { transports = (port, transport @TLS) : [("80", transport @WS) | enableWebsockets],
           tbqSize = 16,
-          serverTbqSize = 128,
-          msgQueueQuota = 256,
+          serverTbqSize = 64,
+          msgQueueQuota = 128,
           queueIdBytes = 24,
           msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
           caCertificateFile = caCrtFile,

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 1.0.2
+version: 1.0.3
 synopsis: SimpleXMQ message broker
 description:  |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -47,7 +47,6 @@ dependencies:
   - mtl == 2.2.*
   - network == 3.1.*
   - network-transport == 0.5.*
-  - postgresql-simple == 0.6.*
   - QuickCheck == 2.14.*
   - random >= 1.1 && < 1.3
   - simple-logger == 0.1.*

scripts/smp-server-digitalocean-droplet/README.md

@@ -16,3 +16,8 @@ brew install hashicorp/tap/packer
 cd ./scripts/smp-server-digitalocean-droplet
 DIGITALOCEAN_TOKEN=$YOUR_TOKEN packer build -on-error=ask -color=false ./marketplace-image.json
 ```
+
+**TODO** (see Linode script)
+
+- Increase file descriptors limit
+- Configure Restart for systemd service

scripts/smp-server-linode.sh

@@ -44,6 +44,12 @@ ufw allow ssh
 ufw allow https
 ufw allow 5223
 
+# Increase file descriptors limit
+echo 'fs.file-max = 1000000' >> /etc/sysctl.conf
+echo 'fs.inode-max = 1000000' >> /etc/sysctl.conf
+echo 'root soft nofile unlimited' >> /etc/security/limits.conf
+echo 'root hard nofile unlimited' >> /etc/security/limits.conf
+
 # Download latest release
 bin_dir="/opt/simplex/bin"
 binary="$bin_dir/smp-server"
@@ -151,6 +157,10 @@ Description=SMP server
 [Service]
 Type=simple
 ExecStart=/bin/sh -c "exec $binary start >> /var/opt/simplex/smp-server.log 2>&1"
+Restart=always
+RestartSec=10
+LimitNOFILE=1000000
+LimitNOFILESoft=1000000
 
 [Install]
 WantedBy=multi-user.target

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        1.0.2
+version:        1.0.3
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -31,19 +31,17 @@ library
   exposed-modules:
       Simplex.Messaging.Agent
       Simplex.Messaging.Agent.Client
-      Simplex.Messaging.Agent.Env.Postgres
       Simplex.Messaging.Agent.Env.SQLite
       Simplex.Messaging.Agent.Protocol
       Simplex.Messaging.Agent.QueryString
       Simplex.Messaging.Agent.RetryInterval
       Simplex.Messaging.Agent.Server
       Simplex.Messaging.Agent.Store
-      Simplex.Messaging.Agent.Store.Postgres
-      Simplex.Messaging.Agent.Store.Postgres.Migrations
-      Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial
       Simplex.Messaging.Agent.Store.SQLite
       Simplex.Messaging.Agent.Store.SQLite.Migrations
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220320_server_ips
       Simplex.Messaging.Client
       Simplex.Messaging.Crypto
       Simplex.Messaging.Crypto.Ratchet
@@ -96,7 +94,6 @@ library
     , mtl ==2.2.*
     , network ==3.1.*
     , network-transport ==0.5.*
-    , postgresql-simple ==0.6.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , sqlite-simple ==0.4.*
@@ -148,7 +145,6 @@ executable smp-agent
     , mtl ==2.2.*
     , network ==3.1.*
     , network-transport ==0.5.*
-    , postgresql-simple ==0.6.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , simplexmq
@@ -203,7 +199,6 @@ executable smp-server
     , network ==3.1.*
     , network-transport ==0.5.*
     , optparse-applicative >=0.15 && <0.17
-    , postgresql-simple ==0.6.*
     , process ==1.6.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
@@ -272,7 +267,6 @@ test-suite smp-server-test
     , mtl ==2.2.*
     , network ==3.1.*
     , network-transport ==0.5.*
-    , postgresql-simple ==0.6.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , simplexmq

src/Simplex/Messaging/Agent.hs

@@ -47,6 +47,7 @@ module Simplex.Messaging.Agent
     ackMessage,
     suspendConnection,
     deleteConnection,
+    setSMPServers,
     logConnection,
   )
 where
@@ -70,11 +71,11 @@ import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
 import Database.SQLite.Simple (SQLError)
 import Simplex.Messaging.Agent.Client
-import Simplex.Messaging.Agent.Env.Postgres
+import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
-import Simplex.Messaging.Agent.Store.Postgres (PostgresStore)
+import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
 import Simplex.Messaging.Client (SMPServerTransmission)
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
@@ -143,6 +144,10 @@ suspendConnection c = withAgentEnv c . suspendConnection' c
 deleteConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ()
 deleteConnection c = withAgentEnv c . deleteConnection' c
 
+-- | Change servers to be used for creating new queues
+setSMPServers :: AgentErrorMonad m => AgentClient -> NonEmpty SMPServer -> m ()
+setSMPServers c = withAgentEnv c . setSMPServers' c
+
 withAgentEnv :: AgentClient -> ReaderT Env m a -> m a
 withAgentEnv c = (`runReaderT` agentEnv c)
 
@@ -172,22 +177,18 @@ client c@AgentClient {rcvQ, subQ} = forever $ do
 
 withStore ::
   AgentMonad m =>
-  (forall m'. (MonadUnliftIO m', MonadError StoreError m') => PostgresStore -> m' a) ->
+  (forall m'. (MonadUnliftIO m', MonadError StoreError m') => SQLiteStore -> m' a) ->
   m a
 withStore action = do
   st <- asks store
   runExceptT (action st `E.catch` handleInternal) >>= \case
     Right c -> return c
-    Left e -> do
-      liftIO $ print e
-      throwError $ storeError e
+    Left e -> throwError $ storeError e
   where
     -- TODO when parsing exception happens in store, the agent hangs;
     -- changing SQLError to SomeException does not help
-    handleInternal :: (MonadUnliftIO m', MonadError StoreError m') => SQLError -> m' a
-    handleInternal e = do
-      liftIO $ print e
-      throwError . SEInternal $ bshow e
+    handleInternal :: (MonadError StoreError m') => SQLError -> m' a
+    handleInternal e = throwError . SEInternal $ bshow e
     storeError :: StoreError -> AgentErrorType
     storeError = \case
       SEConnNotFound -> CONN NOT_FOUND
@@ -213,7 +214,7 @@ processCommand c (connId, cmd) = case cmd of
 
 newConn :: AgentMonad m => AgentClient -> ConnId -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
 newConn c connId cMode = do
-  srv <- getSMPServer
+  srv <- getSMPServer c
   (rq, qUri) <- newRcvQueue c srv
   g <- asks idsDrg
   let cData = ConnData {connId}
@@ -238,19 +239,20 @@ joinConn c connId (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2
       (pk1, pk2, e2eSndParams) <- liftIO . CR.generateE2EParams $ version e2eRcvParams
       (_, rcDHRs) <- liftIO C.generateKeyPair'
       let rc = CR.initSndRatchet rcDHRr rcDHRs $ CR.x3dhSnd pk1 pk2 e2eRcvParams
-      (sq, smpConf) <- newSndQueue qInfo cInfo
+      sq <- newSndQueue qInfo
       g <- asks idsDrg
       let cData = ConnData {connId}
       connId' <- withStore $ \st -> do
-        liftIO $ print "before: createSndConn st g cData sq"
         connId' <- createSndConn st g cData sq
-        liftIO $ print "before: createRatchet st connId' rc"
         createRatchet st connId' rc
-        liftIO $ print "after: createRatchet st connId' rc"
         pure connId'
-      confirmQueue c connId' sq smpConf $ Just e2eSndParams
-      void $ enqueueMessage c connId' sq HELLO
-      pure connId'
+      tryError (confirmQueue c connId' sq cInfo $ Just e2eSndParams) >>= \case
+        Right _ -> do
+          void $ enqueueMessage c connId' sq HELLO
+          pure connId'
+        Left e -> do
+          withStore (`deleteConn` connId')
+          throwError e
     _ -> throwError $ AGENT A_VERSION
 joinConn c connId (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInfo =
   case ( qUri `compatibleVersion` SMP.smpClientVRange,
@@ -265,7 +267,7 @@ joinConn c connId (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInf
 
 createReplyQueue :: AgentMonad m => AgentClient -> ConnId -> SndQueue -> m ()
 createReplyQueue c connId sq = do
-  srv <- getSMPServer
+  srv <- getSMPServer c
   (rq, qUri) <- newRcvQueue c srv
   -- TODO reply queue version should be the same as send queue, ignoring it in v1
   let qInfo = toVersionT qUri SMP.smpClientVersion
@@ -312,16 +314,6 @@ subscribeConnection' c connId =
     SomeConn _ (DuplexConnection _ rq sq) -> do
       resumeMsgDelivery c connId sq
       subscribeQueue c rq connId
-      case status (sq :: SndQueue) of
-        Confirmed -> do
-          -- TODO if there is no confirmation saved, just update the status without securing the queue
-          AcceptedConfirmation {senderConf = SMPConfirmation {senderKey}} <-
-            withStore (`getAcceptedConfirmation` connId)
-          secureQueue c rq senderKey
-          withStore $ \st -> setRcvQueueStatus st rq Secured
-        Secured -> pure ()
-        Active -> pure ()
-        _ -> throwError $ INTERNAL "unexpected queue status"
     SomeConn _ (SndConnection _ sq) -> do
       resumeMsgDelivery c connId sq
       case status (sq :: SndQueue) of
@@ -354,12 +346,12 @@ enqueueMessage c connId sq aMessage = do
       internalTs <- liftIO getCurrentTime
       (internalId, internalSndId, prevMsgHash) <- withStore (`updateSndIds` connId)
       let privHeader = APrivHeader (unSndId internalSndId) prevMsgHash
-          agentMessage = smpEncode $ AgentMessage privHeader aMessage
-          internalHash = C.sha256Hash agentMessage
-
-      encAgentMessage <- agentRatchetEncrypt connId agentMessage e2eEncUserMsgLength
+          agentMsg = AgentMessage privHeader aMessage
+          agentMsgStr = smpEncode agentMsg
+          internalHash = C.sha256Hash agentMsgStr
+      encAgentMessage <- agentRatchetEncrypt connId agentMsgStr e2eEncUserMsgLength
       let msgBody = smpEncode $ AgentMsgEnvelope {agentVersion = smpAgentVersion, encAgentMessage}
-          msgType = aMessageType aMessage
+          msgType = agentMessageType agentMsg
           msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, internalHash, prevMsgHash}
       withStore $ \st -> createSndMsg st connId msgData
       pure internalId
@@ -409,30 +401,38 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} connId sq = do
         notify $ MERR mId (INTERNAL $ show e)
       Right (rq_, (msgType, msgBody, internalTs)) ->
         withRetryInterval ri $ \loop ->
-          tryError (sendAgentMessage c sq msgBody) >>= \case
+          tryError (send msgType c sq msgBody) >>= \case
             Left e -> do
+              let err = if msgType == AM_CONN_INFO then ERR e else MERR mId e
               case e of
-                SMP SMP.QUOTA -> loop
+                SMP SMP.QUOTA -> case msgType of
+                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
+                  _ -> loop
                 SMP SMP.AUTH -> case msgType of
-                  HELLO_ -> do
+                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
+                  AM_HELLO_ -> do
                     helloTimeout <- asks $ helloTimeout . config
                     currentTime <- liftIO getCurrentTime
                     if diffUTCTime currentTime internalTs > helloTimeout
                       then case rq_ of
                         -- party initiating connection
-                        Just _ -> notifyDel msgId . ERR $ CONN NOT_AVAILABLE
+                        Just _ -> connError msgId NOT_AVAILABLE
                         -- party joining connection
-                        _ -> notifyDel msgId . ERR $ CONN NOT_ACCEPTED
+                        _ -> connError msgId NOT_ACCEPTED
                       else loop
-                  REPLY_ -> notifyDel msgId $ ERR e
-                  A_MSG_ -> notifyDel msgId $ MERR mId e
-                SMP (SMP.CMD _) -> notifyDel msgId $ MERR mId e
-                SMP SMP.LARGE_MSG -> notifyDel msgId $ MERR mId e
-                SMP {} -> notify (MERR mId e) >> loop
+                  AM_REPLY_ -> notifyDel msgId $ ERR e
+                  AM_A_MSG_ -> notifyDel msgId $ MERR mId e
+                SMP (SMP.CMD _) -> notifyDel msgId err
+                SMP SMP.LARGE_MSG -> notifyDel msgId err
+                SMP {} -> notify err >> loop
                 _ -> loop
             Right () -> do
               case msgType of
-                HELLO_ -> do
+                AM_CONN_INFO -> do
+                  withStore $ \st -> setSndQueueStatus st sq Confirmed
+                  when (isJust rq_) $ withStore (`removeConfirmations` connId)
+                  void $ enqueueMessage c connId sq HELLO
+                AM_HELLO_ -> do
                   withStore $ \st -> setSndQueueStatus st sq Active
                   case rq_ of
                     -- party initiating connection
@@ -441,16 +441,20 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} connId sq = do
                       notify CON
                     -- party joining connection
                     _ -> createReplyQueue c connId sq
-                A_MSG_ -> notify $ SENT mId
+                AM_A_MSG_ -> notify $ SENT mId
                 _ -> pure ()
               delMsg msgId
   where
+    send = \case
+      AM_CONN_INFO -> sendConfirmation
+      _ -> sendAgentMessage
     delMsg :: InternalId -> m ()
     delMsg msgId = withStore $ \st -> deleteMsg st connId msgId
     notify :: ACommand 'Agent -> m ()
     notify cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
     notifyDel :: InternalId -> ACommand 'Agent -> m ()
     notifyDel msgId cmd = notify cmd >> delMsg msgId
+    connError msgId = notifyDel msgId . ERR . CONN
 
 ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
 ackMessage' c connId msgId = do
@@ -489,9 +493,15 @@ deleteConnection' c connId =
       removeSubscription c connId
       withStore (`deleteConn` connId)
 
-getSMPServer :: AgentMonad m => m SMPServer
-getSMPServer =
-  asks (smpServers . config) >>= \case
+-- | Change servers to be used for creating new queues, in Reader monad
+setSMPServers' :: forall m. AgentMonad m => AgentClient -> NonEmpty SMPServer -> m ()
+setSMPServers' c servers = do
+  atomically $ writeTVar (smpServers c) servers
+
+getSMPServer :: AgentMonad m => AgentClient -> m SMPServer
+getSMPServer c = do
+  smpServers <- readTVarIO $ smpServers c
+  case smpServers of
     srv :| [] -> pure srv
     servers -> do
       gen <- asks randomServer
@@ -536,8 +546,9 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
                 (SMP.PHEmpty, AgentMsgEnvelope _ encAgentMsg) -> do
                   agentMsgBody <- agentRatchetDecrypt connId encAgentMsg
                   parseMessage agentMsgBody >>= \case
-                    AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage -> do
-                      (msgId, msgMeta) <- agentClientMsg prevMsgHash sndMsgId (srvMsgId, systemToUTCTime srvTs) agentMsgBody aMessage
+                    agentMsg@(AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage) -> do
+                      let msgType = agentMessageType agentMsg
+                      (msgId, msgMeta) <- agentClientMsg prevMsgHash sndMsgId (srvMsgId, systemToUTCTime srvTs) agentMsgBody msgType
                       case aMessage of
                         HELLO -> helloMsg >> ack >> withStore (\st -> deleteMsg st connId msgId)
                         REPLY cReq -> replyMsg cReq >> ack >> withStore (\st -> deleteMsg st connId msgId)
@@ -627,18 +638,13 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
               case qInfo `proveCompatible` SMP.smpClientVRange of
                 Nothing -> notify . ERR $ AGENT A_VERSION
                 Just qInfo' -> do
-                  (sq, smpConf) <- newSndQueue qInfo' ownConnInfo
-                  liftIO $ print "before: upgradeRcvConnToDuplex st connId sq"
+                  sq <- newSndQueue qInfo'
                   withStore $ \st -> upgradeRcvConnToDuplex st connId sq
-                  confirmQueue c connId sq smpConf Nothing
-                  liftIO $ print "before: `removeConfirmations` connId"
-                  withStore (`removeConfirmations` connId)
-                  liftIO $ print "after: `removeConfirmations` connId"
-                  void $ enqueueMessage c connId sq HELLO
+                  enqueueConfirmation c connId sq ownConnInfo Nothing
             _ -> prohibited
 
-        agentClientMsg :: PrevRcvMsgHash -> ExternalSndId -> (BrokerId, BrokerTs) -> MsgBody -> AMessage -> m (InternalId, MsgMeta)
-        agentClientMsg externalPrevSndHash sndMsgId broker msgBody aMessage = do
+        agentClientMsg :: PrevRcvMsgHash -> ExternalSndId -> (BrokerId, BrokerTs) -> MsgBody -> AgentMessageType -> m (InternalId, MsgMeta)
+        agentClientMsg externalPrevSndHash sndMsgId broker msgBody msgType = do
           logServer "<--" c srv rId "MSG <MSG>"
           let internalHash = C.sha256Hash msgBody
           internalTs <- liftIO getCurrentTime
@@ -646,7 +652,6 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
           let integrity = checkMsgIntegrity prevExtSndId sndMsgId prevRcvMsgHash externalPrevSndHash
               recipient = (unId internalId, internalTs)
               msgMeta = MsgMeta {integrity, recipient, broker, sndMsgId}
-              msgType = aMessageType aMessage
               rcvMsg = RcvMsgData {msgMeta, msgType, msgBody, internalRcvId, internalHash, externalPrevSndHash}
           withStore $ \st -> createRcvMsg st connId rcvMsg
           pure (internalId, msgMeta)
@@ -671,8 +676,9 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
           | internalPrevMsgHash /= receivedPrevMsgHash = MsgError MsgBadHash
           | otherwise = MsgError MsgDuplicate -- this case is not possible
 
-confirmQueue :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> SMPConfirmation -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
-confirmQueue c connId sq SMPConfirmation {senderKey, e2ePubKey, connInfo} e2eEncryption = do
+confirmQueue :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
+confirmQueue c connId sq connInfo e2eEncryption = do
+  _ <- withStore (`updateSndIds` connId)
   msg <- mkConfirmation
   sendConfirmation c sq msg
   withStore $ \st -> setSndQueueStatus st sq Confirmed
@@ -680,9 +686,27 @@ confirmQueue c connId sq SMPConfirmation {senderKey, e2ePubKey, connInfo} e2eEnc
     mkConfirmation :: m MsgBody
     mkConfirmation = do
       encConnInfo <- agentRatchetEncrypt connId (smpEncode $ AgentConnInfo connInfo) e2eEncConnInfoLength
-      let agentEnvelope = AgentConfirmation {agentVersion = smpAgentVersion, e2eEncryption, encConnInfo}
-      agentCbEncrypt sq (Just e2ePubKey) . smpEncode $
-        SMP.ClientMessage (SMP.PHConfirmation senderKey) $ smpEncode agentEnvelope
+      pure . smpEncode $ AgentConfirmation {agentVersion = smpAgentVersion, e2eEncryption, encConnInfo}
+
+enqueueConfirmation :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
+enqueueConfirmation c connId sq connInfo e2eEncryption = do
+  resumeMsgDelivery c connId sq
+  msgId <- storeConfirmation
+  queuePendingMsgs c connId sq [msgId]
+  where
+    storeConfirmation :: m InternalId
+    storeConfirmation = do
+      internalTs <- liftIO getCurrentTime
+      (internalId, internalSndId, prevMsgHash) <- withStore (`updateSndIds` connId)
+      let agentMsg = AgentConnInfo connInfo
+          agentMsgStr = smpEncode agentMsg
+          internalHash = C.sha256Hash agentMsgStr
+      encConnInfo <- agentRatchetEncrypt connId agentMsgStr e2eEncConnInfoLength
+      let msgBody = smpEncode $ AgentConfirmation {agentVersion = smpAgentVersion, e2eEncryption, encConnInfo}
+          msgType = agentMessageType agentMsg
+          msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, internalHash, prevMsgHash}
+      withStore $ \st -> createSndMsg st connId msgData
+      pure internalId
 
 -- encoded AgentMessage -> encoded EncAgentMessage
 agentRatchetEncrypt :: AgentMonad m => ConnId -> ByteString -> Int -> m ByteString
@@ -704,27 +728,27 @@ agentRatchetDecrypt connId encAgentMsg = do
 notifyConnected :: AgentMonad m => AgentClient -> ConnId -> m ()
 notifyConnected c connId = atomically $ writeTBQueue (subQ c) ("", connId, CON)
 
-newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> ConnInfo -> m (SndQueue, SMPConfirmation)
-newSndQueue qInfo cInfo =
+newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> m SndQueue
+newSndQueue qInfo =
   asks (cmdSignAlg . config) >>= \case
-    C.SignAlg a -> newSndQueue_ a qInfo cInfo
+    C.SignAlg a -> newSndQueue_ a qInfo
 
 newSndQueue_ ::
   (C.SignatureAlgorithm a, C.AlgorithmI a, MonadUnliftIO m) =>
   C.SAlgorithm a ->
   Compatible SMPQueueInfo ->
-  ConnInfo ->
-  m (SndQueue, SMPConfirmation)
-newSndQueue_ a (Compatible (SMPQueueInfo _clientVersion smpServer senderId rcvE2ePubDhKey)) cInfo = do
+  m SndQueue
+newSndQueue_ a (Compatible (SMPQueueInfo _clientVersion smpServer senderId rcvE2ePubDhKey)) = do
   -- this function assumes clientVersion is compatible - it was tested before
-  (senderKey, sndPrivateKey) <- liftIO $ C.generateSignatureKeyPair a
+  (sndPublicKey, sndPrivateKey) <- liftIO $ C.generateSignatureKeyPair a
   (e2ePubKey, e2ePrivKey) <- liftIO C.generateKeyPair'
-  let sndQueue =
-        SndQueue
-          { server = smpServer,
-            sndId = senderId,
-            sndPrivateKey,
-            e2eDhSecret = C.dh' rcvE2ePubDhKey e2ePrivKey,
-            status = New
-          }
-  pure (sndQueue, SMPConfirmation senderKey e2ePubKey cInfo)
+  pure
+    SndQueue
+      { server = smpServer,
+        sndId = senderId,
+        sndPublicKey = Just sndPublicKey,
+        sndPrivateKey,
+        e2eDhSecret = C.dh' rcvE2ePubDhKey e2ePrivKey,
+        e2ePubKey = Just e2ePubKey,
+        status = New
+      }

src/Simplex/Messaging/Agent/Client.hs

@@ -37,7 +37,7 @@ module Simplex.Messaging.Agent.Client
 where
 
 import Control.Concurrent (forkIO)
-import Control.Concurrent.Async (Async, async, uninterruptibleCancel)
+import Control.Concurrent.Async (Async, uninterruptibleCancel)
 import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple
 import Control.Monad.Except
@@ -47,13 +47,14 @@ import Data.Bifunctor (first)
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Maybe (isNothing)
 import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text.Encoding
-import Simplex.Messaging.Agent.Env.Postgres
+import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
@@ -62,8 +63,10 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol (QueueId, QueueIdsKeys (..), SndPublicVerifyKey)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Util (bshow, liftEitherError, liftError, liftIOEither, tryError)
+import Simplex.Messaging.Util (bshow, liftEitherError, liftError, tryError)
 import Simplex.Messaging.Version
+import System.Timeout (timeout)
+import UnliftIO (async)
 import UnliftIO.Exception (Exception, IOException)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -74,13 +77,16 @@ data AgentClient = AgentClient
   { rcvQ :: TBQueue (ATransmission 'Client),
     subQ :: TBQueue (ATransmission 'Agent),
     msgQ :: TBQueue SMPServerTransmission,
+    smpServers :: TVar (NonEmpty SMPServer),
     smpClients :: TVar (Map SMPServer SMPClientVar),
     subscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
+    pendingSubscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
     subscrConns :: TVar (Map ConnId SMPServer),
     connMsgsQueued :: TVar (Map ConnId Bool),
     smpQueueMsgQueues :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId)),
     smpQueueMsgDeliveries :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (Async ())),
     reconnections :: TVar [Async ()],
+    asyncClients :: TVar [Async ()],
     clientId :: Int,
     agentEnv :: Env,
     smpSubscriber :: Async (),
@@ -93,16 +99,19 @@ newAgentClient agentEnv = do
   rcvQ <- newTBQueue qSize
   subQ <- newTBQueue qSize
   msgQ <- newTBQueue qSize
+  smpServers <- newTVar $ initialSMPServers (config agentEnv)
   smpClients <- newTVar M.empty
   subscrSrvrs <- newTVar M.empty
+  pendingSubscrSrvrs <- newTVar M.empty
   subscrConns <- newTVar M.empty
   connMsgsQueued <- newTVar M.empty
   smpQueueMsgQueues <- newTVar M.empty
   smpQueueMsgDeliveries <- newTVar M.empty
   reconnections <- newTVar []
+  asyncClients <- newTVar []
   clientId <- stateTVar (clientCounter agentEnv) $ \i -> (i + 1, i + 1)
   lock <- newTMVar ()
-  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock}
+  return AgentClient {rcvQ, subQ, msgQ, smpServers, smpClients, subscrSrvrs, pendingSubscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, asyncClients, clientId, agentEnv, smpSubscriber = undefined, lock}
 
 -- | Agent monad with MonadReader Env and MonadError AgentErrorType
 type AgentMonad m = (MonadUnliftIO m, MonadReader Env m, MonadError AgentErrorType m)
@@ -133,20 +142,39 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
       pure smpVar
 
     waitForSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
-    waitForSMPClient = liftIOEither . atomically . readTMVar
+    waitForSMPClient smpVar = do
+      SMPClientConfig {tcpTimeout} <- asks $ smpCfg . config
+      smpClient_ <- liftIO $ tcpTimeout `timeout` atomically (readTMVar smpVar)
+      liftEither $ case smpClient_ of
+        Just (Right smpClient) -> Right smpClient
+        Just (Left e) -> Left e
+        Nothing -> Left $ BROKER TIMEOUT
 
     newSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
-    newSMPClient smpVar =
-      tryError connectClient >>= \r -> case r of
-        Right smp -> do
-          logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
-          atomically $ putTMVar smpVar r
-          pure smp
-        Left e -> do
-          atomically $ do
-            putTMVar smpVar r
-            modifyTVar smpClients $ M.delete srv
-          throwError e
+    newSMPClient smpVar = tryConnectClient pure tryConnectAsync
+      where
+        tryConnectClient :: (SMPClient -> m a) -> m () -> m a
+        tryConnectClient successAction retryAction =
+          tryError connectClient >>= \r -> case r of
+            Right smp -> do
+              logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
+              atomically $ putTMVar smpVar r
+              successAction smp
+            Left e -> do
+              if e == BROKER NETWORK || e == BROKER TIMEOUT
+                then retryAction
+                else atomically $ do
+                  putTMVar smpVar (Left e)
+                  modifyTVar smpClients $ M.delete srv
+              throwError e
+        tryConnectAsync :: m ()
+        tryConnectAsync = do
+          a <- async connectAsync
+          atomically $ modifyTVar (asyncClients c) (a :)
+        connectAsync :: m ()
+        connectAsync = do
+          ri <- asks $ reconnectInterval . config
+          withRetryInterval ri $ \loop -> void $ tryConnectClient (const reconnectClient) loop
 
     connectClient :: m SMPClient
     connectClient = do
@@ -169,33 +197,43 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
       cs <- M.lookup srv <$> readTVar (subscrSrvrs c)
       modifyTVar (subscrSrvrs c) $ M.delete srv
       modifyTVar (subscrConns c) $ maybe id (deleteKeys . M.keysSet) cs
+      mapM_ (modifyTVar (pendingSubscrSrvrs c) . addPendingSubs) cs
       return cs
       where
+        addPendingSubs :: Map ConnId RcvQueue -> Map SMPServer (Map ConnId RcvQueue) -> Map SMPServer (Map ConnId RcvQueue)
+        addPendingSubs cs = M.alter (Just . addSubs cs) srv
+        addSubs cs = maybe cs (M.union cs)
         deleteKeys :: Ord k => Set k -> Map k a -> Map k a
         deleteKeys ks m = S.foldr' M.delete m ks
 
     serverDown :: UnliftIO m -> Map ConnId RcvQueue -> IO ()
     serverDown u cs = unless (M.null cs) $ do
       mapM_ (notifySub DOWN) $ M.keysSet cs
-      a <- async . unliftIO u $ tryReconnectClient cs
+      unliftIO u reconnectServer
+
+    reconnectServer :: m ()
+    reconnectServer = do
+      a <- async tryReconnectClient
       atomically $ modifyTVar (reconnections c) (a :)
 
-    tryReconnectClient :: Map ConnId RcvQueue -> m ()
-    tryReconnectClient cs = do
+    tryReconnectClient :: m ()
+    tryReconnectClient = do
       ri <- asks $ reconnectInterval . config
       withRetryInterval ri $ \loop ->
-        reconnectClient cs `catchError` const loop
+        reconnectClient `catchError` const loop
 
-    reconnectClient :: Map ConnId RcvQueue -> m ()
-    reconnectClient cs = do
+    reconnectClient :: m ()
+    reconnectClient = do
       withAgentLock c . withSMP c srv $ \smp -> do
         subs <- readTVarIO $ subscrConns c
-        forM_ (M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->
+        cs <- M.lookup srv <$> readTVarIO (pendingSubscrSrvrs c)
+        forM_ (maybe [] M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->
           when (isNothing $ M.lookup connId subs) $ do
             subscribeSMPQueue smp rcvPrivateKey rcvId
               `catchError` \case
-                SMPServerError e -> liftIO $ notifySub (ERR $ SMP e) connId
-                e -> throwError e
+                e@SMPResponseTimeout -> throwError e
+                e@SMPNetworkError -> throwError e
+                e -> liftIO $ notifySub (ERR $ smpClientError e) connId
             addSubscription c rq connId
             liftIO $ notifySub UP connId
 
@@ -206,6 +244,7 @@ closeAgentClient :: MonadUnliftIO m => AgentClient -> m ()
 closeAgentClient c = liftIO $ do
   closeSMPServerClients c
   cancelActions $ reconnections c
+  cancelActions $ asyncClients c
   cancelActions $ smpQueueMsgDeliveries c
 
 closeSMPServerClients :: AgentClient -> IO ()
@@ -294,26 +333,40 @@ newRcvQueue_ a c srv = do
 
 subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()
 subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
-  withLogSMP c server rcvId "SUB" $ \smp ->
-    subscribeSMPQueue smp rcvPrivateKey rcvId
-  addSubscription c rq connId
+  addPendingSubscription c rq connId
+  withLogSMP c server rcvId "SUB" $ \smp -> do
+    liftIO (runExceptT $ subscribeSMPQueue smp rcvPrivateKey rcvId) >>= \case
+      Left e -> do
+        atomically . when (e /= SMPNetworkError && e /= SMPResponseTimeout) $
+          removePendingSubscription c server connId
+        throwError e
+      Right _ -> addSubscription c rq connId
 
 addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
 addSubscription c rq@RcvQueue {server} connId = atomically $ do
   modifyTVar (subscrConns c) $ M.insert connId server
-  modifyTVar (subscrSrvrs c) $ M.alter (Just . addSub) server
-  where
-    addSub :: Maybe (Map ConnId RcvQueue) -> Map ConnId RcvQueue
-    addSub (Just cs) = M.insert connId rq cs
-    addSub _ = M.singleton connId rq
+  addSubs_ (subscrSrvrs c) rq connId
+  removePendingSubscription c server connId
 
-removeSubscription :: AgentMonad m => AgentClient -> ConnId -> m ()
-removeSubscription AgentClient {subscrConns, subscrSrvrs} connId = atomically $ do
-  cs <- readTVar subscrConns
-  writeTVar subscrConns $ M.delete connId cs
-  mapM_
-    (modifyTVar subscrSrvrs . M.alter (>>= delSub))
-    (M.lookup connId cs)
+addPendingSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
+addPendingSubscription c rq connId =
+  atomically $ addSubs_ (pendingSubscrSrvrs c) rq connId
+
+addSubs_ :: TVar (Map SMPServer (Map ConnId RcvQueue)) -> RcvQueue -> ConnId -> STM ()
+addSubs_ ss rq@RcvQueue {server} connId = modifyTVar ss $ M.alter (Just . addSub) server
+  where
+    addSub = maybe (M.singleton connId rq) (M.insert connId rq)
+
+removeSubscription :: MonadUnliftIO m => AgentClient -> ConnId -> m ()
+removeSubscription c@AgentClient {subscrConns} connId = atomically $ do
+  server_ <- stateTVar subscrConns $ \cs -> (M.lookup connId cs, M.delete connId cs)
+  mapM_ (\server -> removeSubs_ (subscrSrvrs c) server connId) server_
+
+removePendingSubscription :: AgentClient -> SMPServer -> ConnId -> STM ()
+removePendingSubscription c = removeSubs_ (pendingSubscrSrvrs c)
+
+removeSubs_ :: TVar (Map SMPServer (Map ConnId RcvQueue)) -> SMPServer -> ConnId -> STM ()
+removeSubs_ ss server connId = modifyTVar ss $ M.update delSub server
   where
     delSub :: Map ConnId RcvQueue -> Maybe (Map ConnId RcvQueue)
     delSub cs =
@@ -331,11 +384,13 @@ showServer SMPServer {host, port} =
 logSecret :: ByteString -> ByteString
 logSecret bs = encode $ B.take 3 bs
 
--- TODO maybe package E2ERatchetParams into SMPConfirmation
 sendConfirmation :: forall m. AgentMonad m => AgentClient -> SndQueue -> ByteString -> m ()
-sendConfirmation c SndQueue {server, sndId} encConfirmation =
-  withLogSMP_ c server sndId "SEND <CONF>" $ \smp ->
-    liftSMP $ sendSMPMessage smp Nothing sndId encConfirmation
+sendConfirmation c sq@SndQueue {server, sndId, sndPublicKey = Just sndPublicKey, e2ePubKey = e2ePubKey@Just {}} agentConfirmation =
+  withLogSMP_ c server sndId "SEND <CONF>" $ \smp -> do
+    let clientMsg = SMP.ClientMessage (SMP.PHConfirmation sndPublicKey) agentConfirmation
+    msg <- agentCbEncrypt sq e2ePubKey $ smpEncode clientMsg
+    liftSMP $ sendSMPMessage smp Nothing sndId msg
+sendConfirmation _ _ _ = throwError $ INTERNAL "sendConfirmation called without snd_queue public key(s) in the database"
 
 sendInvitation :: forall m. AgentMonad m => AgentClient -> Compatible SMPQueueInfo -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> m ()
 sendInvitation c (Compatible SMPQueueInfo {smpServer, senderId, dhPublicKey}) connReq connInfo =
@@ -370,7 +425,6 @@ deleteQueue c RcvQueue {server, rcvId, rcvPrivateKey} =
   withLogSMP c server rcvId "DEL" $ \smp ->
     deleteSMPQueue smp rcvPrivateKey rcvId
 
--- TODO this is just wrong
 sendAgentMessage :: forall m. AgentMonad m => AgentClient -> SndQueue -> ByteString -> m ()
 sendAgentMessage c sq@SndQueue {server, sndId, sndPrivateKey} agentMsg =
   withLogSMP_ c server sndId "SEND <MSG>" $ \smp -> do

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

@@ -1,88 +0,0 @@
-{-# LANGUAGE DataKinds #-}
-{-# LANGUAGE DuplicateRecordFields #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE NumericUnderscores #-}
-{-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
-
-module Simplex.Messaging.Agent.Env.Postgres
-  ( AgentConfig (..),
-    defaultAgentConfig,
-    Env (..),
-    newSMPAgentEnv,
-  )
-where
-
-import Control.Monad.IO.Unlift
-import Crypto.Random
-import Data.List.NonEmpty (NonEmpty)
-import Data.Time.Clock (NominalDiffTime, nominalDay)
-import Database.PostgreSQL.Simple (ConnectInfo (..), defaultConnectInfo)
-import Network.Socket
-import Numeric.Natural
-import Simplex.Messaging.Agent.Protocol (SMPServer)
-import Simplex.Messaging.Agent.RetryInterval
-import Simplex.Messaging.Agent.Store.Postgres
-import qualified Simplex.Messaging.Agent.Store.Postgres.Migrations as Migrations
-import Simplex.Messaging.Client
-import qualified Simplex.Messaging.Crypto as C
-import System.Random (StdGen, newStdGen)
-import UnliftIO.STM
-
-data AgentConfig = AgentConfig
-  { tcpPort :: ServiceName,
-    smpServers :: NonEmpty SMPServer,
-    cmdSignAlg :: C.SignAlg,
-    connIdBytes :: Int,
-    tbqSize :: Natural,
-    dbConnInfo :: ConnectInfo,
-    dbPoolSize :: Int,
-    smpCfg :: SMPClientConfig,
-    reconnectInterval :: RetryInterval,
-    helloTimeout :: NominalDiffTime,
-    caCertificateFile :: FilePath,
-    privateKeyFile :: FilePath,
-    certificateFile :: FilePath
-  }
-
-defaultAgentConfig :: AgentConfig
-defaultAgentConfig =
-  AgentConfig
-    { tcpPort = "5224",
-      smpServers = undefined, -- TODO move it elsewhere?
-      cmdSignAlg = C.SignAlg C.SEd448,
-      connIdBytes = 12,
-      tbqSize = 16,
-      dbConnInfo = defaultConnectInfo {connectDatabase = "agent_poc_1"},
-      dbPoolSize = 4,
-      smpCfg = smpDefaultConfig,
-      reconnectInterval =
-        RetryInterval
-          { initialInterval = second,
-            increaseAfter = 10 * second,
-            maxInterval = 10 * second
-          },
-      helloTimeout = 7 * nominalDay,
-      -- CA certificate private key is not needed for initialization
-      -- ! we do not generate these
-      caCertificateFile = "/etc/opt/simplex-agent/ca.crt",
-      privateKeyFile = "/etc/opt/simplex-agent/agent.key",
-      certificateFile = "/etc/opt/simplex-agent/agent.crt"
-    }
-  where
-    second = 1_000_000
-
-data Env = Env
-  { config :: AgentConfig,
-    store :: PostgresStore,
-    idsDrg :: TVar ChaChaDRG,
-    clientCounter :: TVar Int,
-    randomServer :: TVar StdGen
-  }
-
-newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env
-newSMPAgentEnv cfg@AgentConfig {dbConnInfo, dbPoolSize} = do
-  idsDrg <- newTVarIO =<< drgNew
-  store <- liftIO $ createPostgresStore dbConnInfo dbPoolSize Migrations.app
-  clientCounter <- newTVarIO 0
-  randomServer <- newTVarIO =<< liftIO newStdGen
-  return Env {config = cfg, store, idsDrg, clientCounter, randomServer}

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

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

src/Simplex/Messaging/Agent/Protocol.hs

@@ -46,9 +46,9 @@ module Simplex.Messaging.Agent.Protocol
     SMPConfirmation (..),
     AgentMsgEnvelope (..),
     AgentMessage (..),
+    AgentMessageType (..),
     APrivHeader (..),
     AMessage (..),
-    AMsgType (..),
     SMPServer (..),
     SrvLoc (..),
     SMPQueueUri (..),
@@ -89,7 +89,7 @@ module Simplex.Messaging.Agent.Protocol
     connModeT,
     serializeQueueStatus,
     queueStatusT,
-    aMessageType,
+    agentMessageType,
 
     -- * TCP transport functions
     tPut,
@@ -343,6 +343,31 @@ instance Encoding AgentMessage where
       'M' -> AgentMessage <$> smpP <*> smpP
       _ -> fail "bad AgentMessage"
 
+data AgentMessageType = AM_CONN_INFO | AM_HELLO_ | AM_REPLY_ | AM_A_MSG_
+  deriving (Eq, Show)
+
+instance Encoding AgentMessageType where
+  smpEncode = \case
+    AM_CONN_INFO -> "C"
+    AM_HELLO_ -> "H"
+    AM_REPLY_ -> "R"
+    AM_A_MSG_ -> "M"
+  smpP =
+    A.anyChar >>= \case
+      'C' -> pure AM_CONN_INFO
+      'H' -> pure AM_HELLO_
+      'R' -> pure AM_REPLY_
+      'M' -> pure AM_A_MSG_
+      _ -> fail "bad AgentMessageType"
+
+agentMessageType :: AgentMessage -> AgentMessageType
+agentMessageType = \case
+  AgentConnInfo _ -> AM_CONN_INFO
+  AgentMessage _ aMsg -> case aMsg of
+    HELLO -> AM_HELLO_
+    REPLY _ -> AM_REPLY_
+    A_MSG _ -> AM_A_MSG_
+
 data APrivHeader = APrivHeader
   { -- | sequential ID assigned by the sending agent
     sndMsgId :: AgentMsgId,
@@ -371,12 +396,6 @@ instance Encoding AMsgType where
       'M' -> pure A_MSG_
       _ -> fail "bad AMsgType"
 
-aMessageType :: AMessage -> AMsgType
-aMessageType = \case
-  HELLO -> HELLO_
-  REPLY _ -> REPLY_
-  A_MSG _ -> A_MSG_
-
 -- | Messages sent between SMP agents once SMP queue is secured.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md#messages-between-smp-agents
@@ -705,7 +724,7 @@ data ConnectionErrorType
     SIMPLEX
   | -- | connection not accepted on join HELLO after timeout
     NOT_ACCEPTED
-  | -- | connection not available on reply HELLO after timeout
+  | -- | connection not available on reply confirmation/HELLO after timeout
     NOT_AVAILABLE
   deriving (Eq, Generic, Read, Show, Exception)
 

src/Simplex/Messaging/Agent/Server.hs

@@ -19,7 +19,7 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Text.Encoding (decodeUtf8)
 import Simplex.Messaging.Agent
-import Simplex.Messaging.Agent.Env.Postgres
+import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), simplexMQVersion)
 import Simplex.Messaging.Transport.Server (loadTLSServerParams, runTransportServer)

src/Simplex/Messaging/Agent/Store.hs

@@ -62,7 +62,7 @@ class Monad m => MonadAgentStore s m where
   createRcvMsg :: s -> ConnId -> RcvMsgData -> m ()
   updateSndIds :: s -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)
   createSndMsg :: s -> ConnId -> SndMsgData -> m ()
-  getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))
+  getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AgentMessageType, MsgBody, InternalTs))
   getPendingMsgs :: s -> ConnId -> m [InternalId]
   checkRcvMsg :: s -> ConnId -> InternalId -> m ()
   deleteMsg :: s -> ConnId -> InternalId -> m ()
@@ -102,8 +102,11 @@ data SndQueue = SndQueue
   { server :: SMPServer,
     -- | sender queue ID
     sndId :: SMP.SenderId,
-    -- | key used by the sender to sign transmissions
+    -- | key pair used by the sender to sign transmissions
+    sndPublicKey :: Maybe C.APublicVerifyKey,
     sndPrivateKey :: SndPrivateSignKey,
+    -- | DH public key used to negotiate per-queue e2e encryption
+    e2ePubKey :: Maybe C.PublicKeyX25519,
     -- | shared DH secret agreed for simple per-queue e2e encryption
     e2eDhSecret :: C.DhSecretX25519,
     -- | queue status
@@ -221,7 +224,7 @@ type PrevSndMsgHash = MsgHash
 
 data RcvMsgData = RcvMsgData
   { msgMeta :: MsgMeta,
-    msgType :: AMsgType,
+    msgType :: AgentMessageType,
     msgBody :: MsgBody,
     internalRcvId :: InternalRcvId,
     internalHash :: MsgHash,
@@ -232,7 +235,7 @@ data SndMsgData = SndMsgData
   { internalId :: InternalId,
     internalSndId :: InternalSndId,
     internalTs :: InternalTs,
-    msgType :: AMsgType,
+    msgType :: AgentMessageType,
     msgBody :: MsgBody,
     internalHash :: MsgHash,
     prevMsgHash :: MsgHash

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

@@ -1,957 +0,0 @@
-{-# LANGUAGE DataKinds #-}
-{-# LANGUAGE DuplicateRecordFields #-}
-{-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE FlexibleInstances #-}
-{-# LANGUAGE GADTs #-}
-{-# LANGUAGE InstanceSigs #-}
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE MultiParamTypeClasses #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE NumericUnderscores #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE QuasiQuotes #-}
-{-# LANGUAGE RecordWildCards #-}
-{-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TupleSections #-}
-{-# LANGUAGE UndecidableInstances #-}
-{-# OPTIONS_GHC -fno-warn-orphans #-}
-
-module Simplex.Messaging.Agent.Store.Postgres
-  ( PostgresStore (..),
-    createPostgresStore,
-    connectPostgresStore,
-    withConnection,
-    withTransaction,
-    fromTextField_,
-    firstRow,
-  )
-where
-
-import Control.Concurrent (threadDelay)
-import Control.Concurrent.STM
-import Control.Exception (bracket)
-import Control.Monad (void)
-import Control.Monad.Except
-import Control.Monad.IO.Unlift (MonadUnliftIO)
-import Crypto.Random (ChaChaDRG, randomBytesGenerate)
-import Data.Bifunctor (second)
-import Data.ByteString (ByteString)
-import qualified Data.ByteString.Base64.URL as U
-import qualified Data.ByteString.Char8 as B
-import Data.Char (toLower)
-import Data.Functor (($>))
-import Data.List (find, foldl')
-import qualified Data.Map.Strict as M
-import Data.Text (Text)
-import qualified Data.Text as T
-import Data.Text.Encoding (decodeLatin1)
-import Database.PostgreSQL.Simple (FromRow, Only (..), Query, SqlError, ToRow, withSavepoint)
-import qualified Database.PostgreSQL.Simple as DB
-import Database.PostgreSQL.Simple.Errors (constraintViolation)
-import Database.PostgreSQL.Simple.FromField
-import Database.PostgreSQL.Simple.Internal (Conversion (..), Field (..))
-import Database.PostgreSQL.Simple.SqlQQ (sql)
-import Database.PostgreSQL.Simple.ToField (ToField (..))
-import qualified Database.PostgreSQL.Simple.TypeInfo
-import Database.PostgreSQL.Simple.TypeInfo.Static (bytea, text)
-import qualified Database.PostgreSQL.Simple.TypeInfo.Static
-import GHC.Word (Word32)
-import Simplex.Messaging.Agent.Protocol
-import Simplex.Messaging.Agent.Store
-import Simplex.Messaging.Agent.Store.Postgres.Migrations (Migration)
-import qualified Simplex.Messaging.Agent.Store.Postgres.Migrations as Migrations
-import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Crypto.Ratchet (RatchetX448, SkippedMsgDiff (..), SkippedMsgKeys)
-import Simplex.Messaging.Encoding
-import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Parsers (blobFieldParser, parseAll)
-import Simplex.Messaging.Protocol (MsgBody)
-import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Util (bshow, liftIOEither)
-import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
-import System.Exit (exitFailure)
-import System.FilePath (takeDirectory)
-import System.IO (hFlush, stdout)
-import qualified UnliftIO.Exception as E
-import Network.Socket (HostName, ServiceName)
-import Simplex.Messaging.Crypto (KeyHash)
-
--- * Postgres Store implementation
-
-data PostgresStore = PostgresStore
-  { dbConnInfo :: DB.ConnectInfo,
-    dbConnPool :: TBQueue DB.Connection,
-    dbNew :: Bool
-  }
-
-createPostgresStore :: DB.ConnectInfo -> Int -> [Migration] -> IO PostgresStore
-createPostgresStore dbConnInfo poolSize migrations = do
-  st <- connectPostgresStore dbConnInfo poolSize
-  migrateSchema st migrations
-  pure st
-
-migrateSchema :: PostgresStore -> [Migration] -> IO ()
-migrateSchema st migrations = withConnection st $ \db -> do
-  Migrations.initialize db
-  Migrations.get db migrations >>= \case
-    Left e -> confirmOrExit $ "Database error: " <> e
-    Right [] -> pure ()
-    Right ms -> do
-      unless (dbNew st) $ do
-        confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
-      -- TODO backup
-      -- let f = dbFilePath st
-      -- copyFile f (f <> ".bak")
-      Migrations.run db ms
-
-confirmOrExit :: String -> IO ()
-confirmOrExit s = do
-  putStrLn s
-  putStr "Continue (y/N): "
-  hFlush stdout
-  ok <- getLine
-  when (map toLower ok /= "y") exitFailure
-
-connectPostgresStore :: DB.ConnectInfo -> Int -> IO PostgresStore
-connectPostgresStore dbConnInfo poolSize = do
-  let dbNew = True -- TODO scan migrations
-  dbConnPool <- newTBQueueIO $ toEnum poolSize
-  replicateM_ poolSize $
-    connectDB dbConnInfo >>= atomically . writeTBQueue dbConnPool
-  pure PostgresStore {dbConnInfo, dbConnPool, dbNew}
-
-connectDB :: DB.ConnectInfo -> IO DB.Connection
-connectDB = DB.connect
-
-checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)
-checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err)
-
-handleSQLError :: StoreError -> SqlError -> StoreError
-handleSQLError err e = case constraintViolation e of
-  Just _ -> err
-  Nothing -> SEInternal $ bshow e
-
-withConnection :: PostgresStore -> (DB.Connection -> IO a) -> IO a
-withConnection PostgresStore {dbConnPool} =
-  bracket
-    (atomically $ readTBQueue dbConnPool)
-    (atomically . writeTBQueue dbConnPool)
-
-execute :: ToRow q => DB.Connection -> Query -> q -> IO ()
-execute db query q = void $ DB.execute db query q
-
--- TODO not sure this logic is needed with Postgres, also no such error
--- withTransaction :: forall a. PostgresStore -> (DB.Connection -> IO a) -> IO a
--- withTransaction st action = withConnection st $ loop 100 100_000
---   where
---     loop :: Int -> Int -> DB.Connection -> IO a
---     loop t tLim db =
---       DB.withTransaction db (action db) `E.catch` \(e :: SQLError) ->
---         if tLim > t && DB.sqlError e == DB.ErrorBusy
---           then do
---             threadDelay t
---             loop (t * 9 `div` 8) (tLim - t) db
---           else E.throwIO e
-
-withTransaction :: forall a. PostgresStore -> (DB.Connection -> IO a) -> IO a
-withTransaction st action = withConnection st inTransaction
-  where
-    inTransaction :: DB.Connection -> IO a
-    inTransaction db = DB.withTransaction db (action db)
-
-createConn_ ::
-  (MonadUnliftIO m, MonadError StoreError m) =>
-  PostgresStore ->
-  TVar ChaChaDRG ->
-  ConnData ->
-  (DB.Connection -> ByteString -> IO ()) ->
-  m ByteString
-createConn_ st gVar cData create = do
-  connId <- liftIOEither . checkConstraint SEConnDuplicate . withTransaction st $ \db ->
-    case cData of
-      ConnData {connId = ""} -> createWithRandomId gVar $ create db
-      ConnData {connId} -> create db connId $> Right connId
-  liftIO $ print "before: getConn_ db connId"
-  conn <- liftIO $ withTransaction st $ \db -> getConn_ db connId
-  liftIO $ print conn
-  pure connId
-
-instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore PostgresStore m where
-  createRcvConn :: PostgresStore -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> m ConnId
-  createRcvConn st gVar cData q@RcvQueue {server} cMode =
-    createConn_ st gVar cData $ \db connId -> do
-      upsertServer_ db server
-      execute db "INSERT INTO connections (conn_id, conn_mode) VALUES (?, ?)" (connId, cMode)
-      insertRcvQueue_ db connId q
-
-  createSndConn :: PostgresStore -> TVar ChaChaDRG -> ConnData -> SndQueue -> m ConnId
-  createSndConn st gVar cData q@SndQueue {server} =
-    createConn_ st gVar cData $ \db connId -> do
-      upsertServer_ db server
-      execute db "INSERT INTO connections (conn_id, conn_mode) VALUES (?, ?)" (connId, SCMInvitation)
-      insertSndQueue_ db connId q
-
-  getConn :: PostgresStore -> ConnId -> m SomeConn
-  getConn st connId =
-    liftIOEither . withTransaction st $ \db ->
-      getConn_ db connId
-
-  getRcvConn :: PostgresStore -> SMPServer -> SMP.RecipientId -> m SomeConn
-  getRcvConn st SMPServer {host, port} rcvId =
-    liftIOEither . withTransaction st $ \db ->
-      DB.query
-        db
-        [sql|
-          SELECT q.conn_id
-          FROM rcv_queues q
-          WHERE q.host = ? AND q.port = ? AND q.rcv_id = ?;
-        |]
-        (host, port, rcvId)
-        >>= \case
-          [Only connId] -> getConn_ db connId
-          _ -> pure $ Left SEConnNotFound
-
-  deleteConn :: PostgresStore -> ConnId -> m ()
-  deleteConn st connId =
-    liftIO . withTransaction st $ \db ->
-      execute
-        db
-        "DELETE FROM connections WHERE conn_id = ?;"
-        (Only connId)
-
-  upgradeRcvConnToDuplex :: PostgresStore -> ConnId -> SndQueue -> m ()
-  upgradeRcvConnToDuplex st connId sq@SndQueue {server} =
-    liftIOEither . withTransaction st $ \db ->
-      getConn_ db connId >>= \case
-        Right (SomeConn _ RcvConnection {}) -> do
-          upsertServer_ db server
-          insertSndQueue_ db connId sq
-          pure $ Right ()
-        Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
-        _ -> pure $ Left SEConnNotFound
-
-  upgradeSndConnToDuplex :: PostgresStore -> ConnId -> RcvQueue -> m ()
-  upgradeSndConnToDuplex st connId rq@RcvQueue {server} =
-    liftIOEither . withTransaction st $ \db ->
-      getConn_ db connId >>= \case
-        Right (SomeConn _ SndConnection {}) -> do
-          upsertServer_ db server
-          insertRcvQueue_ db connId rq
-          pure $ Right ()
-        Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c
-        _ -> pure $ Left SEConnNotFound
-
-  setRcvQueueStatus :: PostgresStore -> RcvQueue -> QueueStatus -> m ()
-  setRcvQueueStatus st RcvQueue {rcvId, server = SMPServer {host, port}} status =
-    -- ? throw error if queue does not exist?
-    liftIO . withTransaction st $ \db ->
-      execute
-        db
-        [sql|
-          UPDATE rcv_queues
-          SET status = ?
-          WHERE host = ? AND port = ? AND rcv_id = ?;
-        |]
-        (status, host, port, rcvId)
-
-  setRcvQueueConfirmedE2E :: PostgresStore -> RcvQueue -> C.DhSecretX25519 -> m ()
-  setRcvQueueConfirmedE2E st RcvQueue {rcvId, server = SMPServer {host, port}} e2eDhSecret =
-    liftIO . withTransaction st $ \db ->
-      execute
-        db
-        [sql|
-          UPDATE rcv_queues
-          SET e2e_dh_secret = ?,
-              status = ?
-          WHERE host = ? AND port = ? AND rcv_id = ?
-        |]
-        (Confirmed, e2eDhSecret, host, port, rcvId)
-
-  setSndQueueStatus :: PostgresStore -> SndQueue -> QueueStatus -> m ()
-  setSndQueueStatus st SndQueue {sndId, server = SMPServer {host, port}} status =
-    -- ? throw error if queue does not exist?
-    liftIO . withTransaction st $ \db ->
-      execute
-        db
-        [sql|
-          UPDATE snd_queues
-          SET status = ?
-          WHERE host = ? AND port = ? AND snd_id = ?;
-        |]
-        (status, host, port, sndId)
-
-  createConfirmation :: PostgresStore -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId
-  createConfirmation st gVar NewConfirmation {connId, senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo}, ratchetState} =
-    liftIOEither . withTransaction st $ \db ->
-      createWithRandomId gVar $ \confirmationId ->
-        execute
-          db
-          [sql|
-            INSERT INTO conn_confirmations
-            (confirmation_id, conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, accepted) VALUES (?, ?, ?, ?, ?, ?, 0);
-          |]
-          (confirmationId, connId, senderKey, e2ePubKey, ratchetState, connInfo)
-
-  acceptConfirmation :: PostgresStore -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation
-  acceptConfirmation st confirmationId ownConnInfo =
-    liftIOEither . withTransaction st $ \db -> do
-      execute
-        db
-        [sql|
-          UPDATE conn_confirmations
-          SET accepted = 1,
-              own_conn_info = ?
-          WHERE confirmation_id = ?;
-        |]
-        (ownConnInfo, confirmationId)
-      firstRow confirmation SEConfirmationNotFound $
-        DB.query
-          db
-          [sql|
-            SELECT conn_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info
-            FROM conn_confirmations
-            WHERE confirmation_id = ?;
-          |]
-          (Only confirmationId)
-    where
-      confirmation (connId, senderKey, e2ePubKey, ratchetState, connInfo) =
-        AcceptedConfirmation
-          { confirmationId,
-            connId,
-            senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo},
-            ratchetState,
-            ownConnInfo
-          }
-
-  getAcceptedConfirmation :: PostgresStore -> ConnId -> m AcceptedConfirmation
-  getAcceptedConfirmation st connId =
-    liftIOEither . withTransaction st $ \db ->
-      firstRow confirmation SEConfirmationNotFound $
-        DB.query
-          db
-          [sql|
-            SELECT confirmation_id, sender_key, e2e_snd_pub_key, ratchet_state, sender_conn_info, own_conn_info
-            FROM conn_confirmations
-            WHERE conn_id = ? AND accepted = 1;
-          |]
-          (Only connId)
-    where
-      confirmation (confirmationId, senderKey, e2ePubKey, ratchetState, connInfo, ownConnInfo) =
-        AcceptedConfirmation
-          { confirmationId,
-            connId,
-            senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo},
-            ratchetState,
-            ownConnInfo
-          }
-
-  removeConfirmations :: PostgresStore -> ConnId -> m ()
-  removeConfirmations st connId =
-    liftIO . withTransaction st $ \db ->
-      execute
-        db
-        [sql|
-          DELETE FROM conn_confirmations
-          WHERE conn_id = ?;
-        |]
-        (Only connId)
-
-  createInvitation :: PostgresStore -> TVar ChaChaDRG -> NewInvitation -> m InvitationId
-  createInvitation st gVar NewInvitation {contactConnId, connReq, recipientConnInfo} =
-    liftIOEither . withTransaction st $ \db ->
-      createWithRandomId gVar $ \invitationId ->
-        execute
-          db
-          [sql|
-            INSERT INTO conn_invitations
-            (invitation_id,  contact_conn_id, cr_invitation, recipient_conn_info, accepted) VALUES (?, ?, ?, ?, 0);
-          |]
-          (invitationId, contactConnId, connReq, recipientConnInfo)
-
-  getInvitation :: PostgresStore -> InvitationId -> m Invitation
-  getInvitation st invitationId =
-    liftIOEither . withTransaction st $ \db ->
-      firstRow invitation SEInvitationNotFound $
-        DB.query
-          db
-          [sql|
-            SELECT contact_conn_id, cr_invitation, recipient_conn_info, own_conn_info, accepted
-            FROM conn_invitations
-            WHERE invitation_id = ?
-              AND accepted = 0
-          |]
-          (Only invitationId)
-    where
-      invitation (contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted) =
-        Invitation {invitationId, contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted}
-
-  acceptInvitation :: PostgresStore -> InvitationId -> ConnInfo -> m ()
-  acceptInvitation st invitationId ownConnInfo =
-    liftIO . withTransaction st $ \db -> do
-      execute
-        db
-        [sql|
-          UPDATE conn_invitations
-          SET accepted = 1,
-              own_conn_info = ?
-          WHERE invitation_id = ?
-        |]
-        (ownConnInfo, invitationId)
-
-  deleteInvitation :: PostgresStore -> ConnId -> InvitationId -> m ()
-  deleteInvitation st contactConnId invId =
-    liftIOEither . withTransaction st $ \db ->
-      runExceptT $
-        ExceptT (getConn_ db contactConnId) >>= \case
-          SomeConn SCContact _ ->
-            liftIO $ execute db "DELETE FROM conn_invitations WHERE contact_conn_id = ? AND invitation_id = ?" (contactConnId, invId)
-          _ -> throwError SEConnNotFound
-
-  updateRcvIds :: PostgresStore -> ConnId -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)
-  updateRcvIds st connId =
-    liftIO . withTransaction st $ \db -> do
-      (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash) <- retrieveLastIdsAndHashRcv_ db connId
-      let internalId = InternalId $ unId lastInternalId + 1
-          internalRcvId = InternalRcvId $ unRcvId lastInternalRcvId + 1
-      updateLastIdsRcv_ db connId internalId internalRcvId
-      pure (internalId, internalRcvId, lastExternalSndId, lastRcvHash)
-
-  createRcvMsg :: PostgresStore -> ConnId -> RcvMsgData -> m ()
-  createRcvMsg st connId rcvMsgData =
-    liftIO . withTransaction st $ \db -> do
-      insertRcvMsgBase_ db connId rcvMsgData
-      insertRcvMsgDetails_ db connId rcvMsgData
-      updateHashRcv_ db connId rcvMsgData
-
-  updateSndIds :: PostgresStore -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)
-  updateSndIds st connId =
-    liftIO . withTransaction st $ \db -> do
-      (lastInternalId, lastInternalSndId, prevSndHash) <- retrieveLastIdsAndHashSnd_ db connId
-      let internalId = InternalId $ unId lastInternalId + 1
-          internalSndId = InternalSndId $ unSndId lastInternalSndId + 1
-      updateLastIdsSnd_ db connId internalId internalSndId
-      pure (internalId, internalSndId, prevSndHash)
-
-  createSndMsg :: PostgresStore -> ConnId -> SndMsgData -> m ()
-  createSndMsg st connId sndMsgData =
-    liftIO . withTransaction st $ \db -> do
-      insertSndMsgBase_ db connId sndMsgData
-      insertSndMsgDetails_ db connId sndMsgData
-      updateHashSnd_ db connId sndMsgData
-
-  getPendingMsgData :: PostgresStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))
-  getPendingMsgData st connId msgId =
-    liftIOEither . withTransaction st $ \db -> runExceptT $ do
-      rq_ <- liftIO $ getRcvQueueByConnId_ db connId
-      msgData <-
-        ExceptT . firstRow id SEMsgNotFound $
-          DB.query
-            db
-            [sql|
-                SELECT m.msg_type, m.msg_body, m.internal_ts
-                FROM messages m
-                JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id
-                WHERE m.conn_id = ? AND m.internal_id = ?
-              |]
-            (connId, msgId)
-      pure (rq_, msgData)
-
-  getPendingMsgs :: PostgresStore -> ConnId -> m [InternalId]
-  getPendingMsgs st connId =
-    liftIO . withTransaction st $ \db ->
-      map fromOnly
-        <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_id = ?" (Only connId)
-
-  checkRcvMsg :: PostgresStore -> ConnId -> InternalId -> m ()
-  checkRcvMsg st connId msgId =
-    liftIOEither . withTransaction st $ \db ->
-      hasMsg
-        <$> DB.query
-          db
-          [sql|
-            SELECT conn_id, internal_id
-            FROM rcv_messages
-            WHERE conn_id = ? AND internal_id = ?
-          |]
-          (connId, msgId)
-    where
-      hasMsg :: [(ConnId, InternalId)] -> Either StoreError ()
-      hasMsg r = if null r then Left SEMsgNotFound else Right ()
-
-  deleteMsg :: PostgresStore -> ConnId -> InternalId -> m ()
-  deleteMsg st connId msgId =
-    liftIO . withTransaction st $ \db ->
-      execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)
-
-  createRatchetX3dhKeys :: PostgresStore -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> m ()
-  createRatchetX3dhKeys st connId x3dhPrivKey1 x3dhPrivKey2 =
-    liftIO . withTransaction st $ \db ->
-      execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2)
-
-  getRatchetX3dhKeys :: PostgresStore -> ConnId -> m (C.PrivateKeyX448, C.PrivateKeyX448)
-  getRatchetX3dhKeys st connId =
-    liftIOEither . withTransaction st $ \db ->
-      fmap hasKeys $
-        firstRow id SEX3dhKeysNotFound $
-          DB.query db "SELECT x3dh_priv_key_1, x3dh_priv_key_2 FROM ratchets WHERE conn_id = ?" (Only connId)
-    where
-      hasKeys = \case
-        Right (Just k1, Just k2) -> Right (k1, k2)
-        _ -> Left SEX3dhKeysNotFound
-
-  createRatchet :: PostgresStore -> ConnId -> RatchetX448 -> m ()
-  createRatchet st connId rc =
-    liftIO . withTransaction st $ \db -> do
-      execute
-        db
-        [sql|
-          INSERT INTO ratchets (conn_id, ratchet_state)
-          VALUES (?, ?)
-          ON CONFLICT (conn_id) DO UPDATE SET
-            ratchet_state = ?,
-            x3dh_priv_key_1 = NULL,
-            x3dh_priv_key_2 = NULL
-        |]
-        (connId, rc, rc)
-
-  getRatchet :: PostgresStore -> ConnId -> m RatchetX448
-  getRatchet st connId =
-    liftIOEither . withTransaction st $ \db ->
-      ratchet
-        <$> DB.query db "SELECT ratchet_state FROM ratchets WHERE conn_id = ?" (Only connId)
-    where
-      ratchet (Only (Just rc) : _) = Right rc
-      ratchet _ = Left SERatchetNotFound
-
-  getSkippedMsgKeys :: PostgresStore -> ConnId -> m SkippedMsgKeys
-  getSkippedMsgKeys st connId =
-    liftIO . withTransaction st $ \db ->
-      skipped <$> DB.query db "SELECT header_key, msg_n, msg_key FROM skipped_messages WHERE conn_id = ?" (Only connId)
-    where
-      skipped ms = foldl' addSkippedKey M.empty ms
-      addSkippedKey smks (hk, msgN, mk) = M.alter (Just . addMsgKey) hk smks
-        where
-          addMsgKey = maybe (M.singleton msgN mk) (M.insert msgN mk)
-
-  updateRatchet :: PostgresStore -> ConnId -> RatchetX448 -> SkippedMsgDiff -> m ()
-  updateRatchet st connId rc skipped =
-    liftIO . withTransaction st $ \db -> do
-      execute db "UPDATE ratchets SET ratchet_state = ? WHERE conn_id = ?" (rc, connId)
-      case skipped of
-        SMDNoChange -> pure ()
-        SMDRemove hk msgN ->
-          execute db "DELETE FROM skipped_messages WHERE conn_id = ? AND header_key = ? AND msg_n = ?" (connId, hk, msgN)
-        SMDAdd smks ->
-          forM_ (M.assocs smks) $ \(hk, mks) ->
-            forM_ (M.assocs mks) $ \(msgN, mk) ->
-              execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, hk, msgN, mk)
-
--- -- * Auxiliary helpers
-
-instance ToField QueueStatus where toField = toField . serializeQueueStatus
-
-instance FromField QueueStatus where fromField = fromTextField_ queueStatusT
-
-instance ToField InternalRcvId where toField (InternalRcvId x) = toField x
-
-instance FromField InternalRcvId where fromField x = fromField x
-
-instance ToField InternalSndId where toField (InternalSndId x) = toField x
-
-instance FromField InternalSndId where fromField x = fromField x
-
-instance ToField InternalId where toField (InternalId x) = toField x
-
-instance FromField InternalId where fromField x = fromField x
-
-instance ToField AMsgType where toField = toField . smpEncode
-
-instance FromField AMsgType where fromField = fromByteStringField $ parseAll smpP
-
-instance ToField MsgIntegrity where toField = toField . strEncode
-
-instance FromField MsgIntegrity where fromField = fromByteStringField $ parseAll strP
-
-instance ToField SMPQueueUri where toField = toField . strEncode
-
-instance FromField SMPQueueUri where fromField = fromByteStringField $ parseAll strP
-
-instance ToField AConnectionRequestUri where toField = toField . strEncode
-
-instance FromField AConnectionRequestUri where fromField = fromByteStringField $ parseAll strP
-
-instance ConnectionModeI c => ToField (ConnectionRequestUri c) where toField = toField . strEncode
-
-instance (E.Typeable c, ConnectionModeI c) => FromField (ConnectionRequestUri c) where fromField = fromByteStringField $ parseAll strP
-
-instance ToField ConnectionMode where toField = toField . decodeLatin1 . strEncode
-
-instance FromField ConnectionMode where fromField = fromTextField_ connModeT
-
-instance ToField (SConnectionMode c) where toField = toField . connMode
-
-instance FromField AConnectionMode where fromField = fromTextField_ $ fmap connMode' . connModeT
-
-instance FromField Word32 where fromField x = fromField x
-
-fromTextField_ :: E.Typeable a => (Text -> Maybe a) -> Field -> Maybe ByteString -> Conversion a
-fromTextField_ fromText f mdata =
-  if typeOid f /= typoid text
-    then returnError Incompatible f ""
-    else case mdata of
-      Nothing -> returnError UnexpectedNull f ""
-      Just dat ->
-        case fromText ((T.pack . B.unpack) dat) of
-          Just x -> return x
-          _ -> returnError ConversionFailed f (B.unpack dat)
-
--- TODO same as in Crypto
-fromByteStringField :: E.Typeable a => (ByteString -> Either String a) -> Field -> Maybe ByteString -> Conversion a
-fromByteStringField dec f mdata =
-  if typeOid f /= typoid bytea
-    then returnError Incompatible f ""
-    else case mdata of
-      Nothing -> returnError UnexpectedNull f ""
-      Just dat ->
-        case dec dat of
-          Right x -> return x
-          _ -> returnError ConversionFailed f (B.unpack dat)
-
-listToEither :: e -> [a] -> Either e a
-listToEither _ (x : _) = Right x
-listToEither e _ = Left e
-
-firstRow :: (a -> b) -> e -> IO [a] -> IO (Either e b)
-firstRow f e a = second f . listToEither e <$> a
-
--- {- ORMOLU_DISABLE -}
--- -- SQLite.Simple only has these up to 10 fields, which is insufficient for some of our queries
--- instance (FromField a, FromField b, FromField c, FromField d, FromField e,
---           FromField f, FromField g, FromField h, FromField i, FromField j,
---           FromField k) =>
---   FromRow (a,b,c,d,e,f,g,h,i,j,k) where
---   fromRow = (,,,,,,,,,,) <$> field <*> field <*> field <*> field <*> field
---                          <*> field <*> field <*> field <*> field <*> field
---                          <*> field
-
--- instance (FromField a, FromField b, FromField c, FromField d, FromField e,
---           FromField f, FromField g, FromField h, FromField i, FromField j,
---           FromField k, FromField l) =>
---   FromRow (a,b,c,d,e,f,g,h,i,j,k,l) where
---   fromRow = (,,,,,,,,,,,) <$> field <*> field <*> field <*> field <*> field
---                           <*> field <*> field <*> field <*> field <*> field
---                           <*> field <*> field
-
--- instance (ToField a, ToField b, ToField c, ToField d, ToField e, ToField f,
---           ToField g, ToField h, ToField i, ToField j, ToField k, ToField l) =>
---   ToRow (a,b,c,d,e,f,g,h,i,j,k,l) where
---   toRow (a,b,c,d,e,f,g,h,i,j,k,l) =
---     [ toField a, toField b, toField c, toField d, toField e, toField f,
---       toField g, toField h, toField i, toField j, toField k, toField l
---     ]
-
--- {- ORMOLU_ENABLE -}
-
--- * Server upsert helper
-
-upsertServer_ :: DB.Connection -> SMPServer -> IO ()
-upsertServer_ dbConn SMPServer {host, port, keyHash} = do
-  execute
-    dbConn
-    [sql|
-      INSERT INTO servers (host, port, key_hash) VALUES (?,?,?)
-      ON CONFLICT (host, port) DO UPDATE SET
-        host=excluded.host,
-        port=excluded.port,
-        key_hash=excluded.key_hash;
-    |]
-    (host, port, keyHash)
-
--- * createRcvConn helpers
-
-insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO ()
-insertRcvQueue_ dbConn connId RcvQueue {..} = do
-  execute
-    dbConn
-    [sql|
-      INSERT INTO rcv_queues
-        ( host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status)
-      VALUES
-        (?,?,?,?,?,?,?,?,?,?);
-    |]
-    (host server, port server, rcvId, connId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status)
-
--- * createSndConn helpers
-
-insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO ()
-insertSndQueue_ dbConn connId SndQueue {..} = do
-  execute
-    dbConn
-    [sql|
-      INSERT INTO snd_queues
-        ( host, port, snd_id, conn_id, snd_private_key, e2e_dh_secret, status)
-      VALUES
-        (?,?,?,?,?,?,?);
-    |]
-    (host server, port server, DB.Binary sndId, connId, sndPrivateKey, e2eDhSecret, status)
-
--- * getConn helpers
-
-getConn_ :: DB.Connection -> ConnId -> IO (Either StoreError SomeConn)
-getConn_ dbConn connId =
-  getConnData_ dbConn connId >>= \case
-    Nothing -> pure $ Left SEConnNotFound
-    Just (connData, cMode) -> do
-      liftIO $ print "before: getRcvQueueByConnId_ dbConn connId"
-      rQ <- getRcvQueueByConnId_ dbConn connId
-      liftIO $ print $ "rQ: " <> show rQ
-      liftIO $ print "before: getSndQueueByConnId_ dbConn connId"
-      sQ <- getSndQueueByConnId_ dbConn connId
-      liftIO $ print $ "sQ: " <> show sQ
-      liftIO $ print "after: getSndQueueByConnId_ dbConn connId"
-      pure $ case (rQ, sQ, cMode) of
-        (Just rcvQ, Just sndQ, CMInvitation) -> Right $ SomeConn SCDuplex (DuplexConnection connData rcvQ sndQ)
-        (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)
-        _ -> Left SEConnNotFound
-
-getConnData_ :: DB.Connection -> ConnId -> IO (Maybe (ConnData, ConnectionMode))
-getConnData_ dbConn connId' =
-  connData
-    <$> DB.query dbConn "SELECT conn_id, conn_mode FROM connections WHERE conn_id = ?;" (Only connId')
-  where
-    connData [(connId, cMode)] = Just (ConnData {connId}, cMode)
-    connData _ = Nothing
-
-getRcvQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe RcvQueue)
-getRcvQueueByConnId_ dbConn connId =
-  rcvQueue
-    <$> DB.query
-      dbConn
-      [sql|
-        SELECT s.key_hash, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
-          q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status
-        FROM rcv_queues q
-        INNER JOIN servers s ON q.host = s.host AND q.port = s.port
-        WHERE q.conn_id = ?;
-      |]
-      (Only connId)
-  where
-    rcvQueue [(keyHash, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status)] =
-      let server = SMPServer host port keyHash
-       in Just RcvQueue {server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status}
-    rcvQueue _ = Nothing
-
-getSndQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe SndQueue)
-getSndQueueByConnId_ dbConn connId = do
-  -- sndQueue
-  --   <$> DB.query
-  --     dbConn
-  --     -- [sql|
-  --     --   SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_private_key, q.e2e_dh_secret, q.status
-  --     --   FROM snd_queues q
-  --     --   INNER JOIN servers s ON q.host = s.host AND q.port = s.port
-  --     --   WHERE q.conn_id = ?;
-  --     -- |]
-  --     [sql|
-  --       SELECT s.key_hash, q.host, q.port, q.snd_private_key, q.status
-  --       FROM snd_queues q
-  --       INNER JOIN servers s ON q.host = s.host AND q.port = s.port
-  --       WHERE q.conn_id = ?;
-  --     |]
-  --     (Only connId)
-  print "inside: getSndQueueByConnId_"
-  -- r1 <- (DB.query
-  --     dbConn
-  --     [sql|
-  --       SELECT host, port, key_hash
-  --       FROM servers
-  --       WHERE host = ?
-  --     |]
-  --     (DB.Only ("localhost" :: HostName))) :: (IO [(HostName, ServiceName, KeyHash)])
-  -- putStrLn $ show r1
-  r <- DB.query
-      dbConn
-      [sql|
-        SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_private_key, q.e2e_dh_secret, q.status
-        FROM snd_queues q
-        INNER JOIN servers s ON q.host = s.host AND q.port = s.port
-        WHERE q.conn_id = ?;
-      |]
-      -- [sql|
-      --   SELECT q.host, q.port, q.status
-      --   FROM snd_queues q
-      --   INNER JOIN servers s ON q.host = s.host AND q.port = s.port
-      --   WHERE q.conn_id = ?;
-      -- |]
-      (DB.Only connId)
-  print $ "r: " <> show r
-  let q = sndQueue r
-  print $ "q: " <> show q
-  pure q
-  where
-    sndQueue [(keyHash, host, port, DB.Binary sndId, sndPrivateKey, e2eDhSecret, status)] =
-      let server = SMPServer host port keyHash
-       in Just SndQueue {server, sndId, sndPrivateKey, e2eDhSecret, status}
-    sndQueue _ = Nothing
-    -- sndQueue [(host, port, status)] = do
-    --   let server = SMPServer host port "abcd"
-    --    in Just SndQueue {server, sndId="3456", sndPrivateKey=(C.APrivateSignKey C.SEd25519 "MC4CAQAwBQYDK2VwBCIEIDfEfevydXXfKajz3sRkcQ7RPvfWUPoq6pu1TYHV1DEe"), e2eDhSecret="MCowBQYDK2VuAyEAjiswwI3O_NlS8Fk3HJUW870EY2bAwmttMBsvRB9eV3o=", status}
-    -- sndQueue _ = Nothing
-
--- * updateRcvIds helpers
-
-retrieveLastIdsAndHashRcv_ :: DB.Connection -> ConnId -> IO (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)
-retrieveLastIdsAndHashRcv_ dbConn connId = do
-  [(lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash)] <-
-    DB.query
-      dbConn
-      [sql|
-        SELECT last_internal_msg_id, last_internal_rcv_msg_id, last_external_snd_msg_id, last_rcv_msg_hash
-        FROM connections
-        WHERE conn_id = ?;
-      |]
-      (Only connId)
-  return (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash)
-
-updateLastIdsRcv_ :: DB.Connection -> ConnId -> InternalId -> InternalRcvId -> IO ()
-updateLastIdsRcv_ dbConn connId newInternalId newInternalRcvId =
-  execute
-    dbConn
-    [sql|
-      UPDATE connections
-      SET last_internal_msg_id = :last_internal_msg_id,
-          last_internal_rcv_msg_id = :last_internal_rcv_msg_id
-      WHERE conn_id = :conn_id;
-    |]
-    (newInternalId, newInternalRcvId, connId)
-
--- * createRcvMsg helpers
-
-insertRcvMsgBase_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
-insertRcvMsgBase_ dbConn connId RcvMsgData {msgMeta, msgType, msgBody, internalRcvId} = do
-  let MsgMeta {recipient = (internalId, internalTs)} = msgMeta
-  execute
-    dbConn
-    [sql|
-      INSERT INTO messages
-        ( conn_id, internal_id, internal_ts, internal_rcv_id, internal_snd_id, msg_type, msg_body)
-      VALUES
-        (?,?,?,?,NULL,?,?);
-    |]
-    (connId, internalId, internalTs, internalRcvId, msgType, msgBody)
-
-insertRcvMsgDetails_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
-insertRcvMsgDetails_ dbConn connId RcvMsgData {msgMeta, internalRcvId, internalHash, externalPrevSndHash} = do
-  let MsgMeta {integrity, recipient, broker, sndMsgId} = msgMeta
-  execute
-    dbConn
-    [sql|
-      INSERT INTO rcv_messages
-        ( conn_id, internal_rcv_id, internal_id, external_snd_id,
-          broker_id, broker_ts,
-          internal_hash, external_prev_snd_hash, integrity)
-      VALUES
-        (?,?,?,?,
-         ?,?,
-         ?,?,?);
-    |]
-    (connId, internalRcvId, fst recipient, sndMsgId, fst broker, snd broker, internalHash, externalPrevSndHash, integrity)
-
-updateHashRcv_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
-updateHashRcv_ dbConn connId RcvMsgData {msgMeta, internalHash, internalRcvId} =
-  execute
-    dbConn
-    -- last_internal_rcv_msg_id equality check prevents race condition in case next id was reserved
-    [sql|
-      UPDATE connections
-      SET last_external_snd_msg_id = ?,
-          last_rcv_msg_hash = ?
-      WHERE conn_id = ?
-        AND last_internal_rcv_msg_id = ?;
-    |]
-    (sndMsgId (msgMeta :: MsgMeta), internalHash, connId, internalRcvId)
-
--- * updateSndIds helpers
-
-retrieveLastIdsAndHashSnd_ :: DB.Connection -> ConnId -> IO (InternalId, InternalSndId, PrevSndMsgHash)
-retrieveLastIdsAndHashSnd_ dbConn connId = do
-  [(lastInternalId, lastInternalSndId, lastSndHash)] <-
-    DB.query
-      dbConn
-      [sql|
-        SELECT last_internal_msg_id, last_internal_snd_msg_id, last_snd_msg_hash
-        FROM connections
-        WHERE conn_id = ?;
-      |]
-      (Only connId)
-  return (lastInternalId, lastInternalSndId, lastSndHash)
-
-updateLastIdsSnd_ :: DB.Connection -> ConnId -> InternalId -> InternalSndId -> IO ()
-updateLastIdsSnd_ dbConn connId newInternalId newInternalSndId =
-  execute
-    dbConn
-    [sql|
-      UPDATE connections
-      SET last_internal_msg_id = ?,
-          last_internal_snd_msg_id = ?
-      WHERE conn_id = ?;
-    |]
-    (newInternalId, newInternalSndId, connId)
-
--- * createSndMsg helpers
-
-insertSndMsgBase_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
-insertSndMsgBase_ dbConn connId SndMsgData {..} = do
-  execute
-    dbConn
-    [sql|
-      INSERT INTO messages
-        ( conn_id, internal_id, internal_ts, internal_rcv_id, internal_snd_id, msg_type, msg_body)
-      VALUES
-        (?,?,?,NULL,?,?, ?);
-    |]
-    (connId, internalId, internalTs, internalSndId, msgType, msgBody)
-
-insertSndMsgDetails_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
-insertSndMsgDetails_ dbConn connId SndMsgData {..} =
-  execute
-    dbConn
-    [sql|
-      INSERT INTO snd_messages
-        ( conn_id, internal_snd_id, internal_id, internal_hash, previous_msg_hash)
-      VALUES
-        (?,?,?,?,?);
-    |]
-    (connId, internalSndId, internalId, internalHash, prevMsgHash)
-
-updateHashSnd_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
-updateHashSnd_ dbConn connId SndMsgData {..} =
-  execute
-    dbConn
-    -- last_internal_snd_msg_id equality check prevents race condition in case next id was reserved
-    [sql|
-      UPDATE connections
-      SET last_snd_msg_hash = ?
-      WHERE conn_id = ?
-        AND last_internal_snd_msg_id = ?;
-    |]
-    (internalHash, connId, internalSndId)
-
--- create record with a random ID
-createWithRandomId :: TVar ChaChaDRG -> (ByteString -> IO ()) -> IO (Either StoreError ByteString)
-createWithRandomId gVar create = tryCreate 3
-  where
-    tryCreate :: Int -> IO (Either StoreError ByteString)
-    tryCreate 0 = pure $ Left SEUniqueID
-    tryCreate n = do
-      id' <- randomId gVar 12
-      E.try (create id') >>= \case
-        Right _ -> pure $ Right id'
-        Left e -> case constraintViolation e of
-          Just _ -> tryCreate (n - 1)
-          Nothing -> pure . Left . SEInternal $ bshow e
-
-randomId :: TVar ChaChaDRG -> Int -> IO ByteString
-randomId gVar n = U.encode <$> (atomically . stateTVar gVar $ randomBytesGenerate n)

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

@@ -1,73 +0,0 @@
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE QuasiQuotes #-}
-{-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TemplateHaskell #-}
-{-# LANGUAGE TupleSections #-}
-
-module Simplex.Messaging.Agent.Store.Postgres.Migrations
-  ( Migration (..),
-    app,
-    initialize,
-    get,
-    run,
-  )
-where
-
-import Control.Monad (forM_, void)
-import Data.Function (on)
-import Data.List (intercalate, sortBy)
-import Data.Time.Clock (getCurrentTime)
-import Database.PostgreSQL.Simple (Connection, Only (..))
-import qualified Database.PostgreSQL.Simple as DB
-import Database.PostgreSQL.Simple.Internal (exec)
-import Database.PostgreSQL.Simple.SqlQQ (sql)
-import Database.PostgreSQL.Simple.Transaction (withTransaction)
-import Database.PostgreSQL.Simple.Types (Query (..))
-import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial (m20220202_initial)
-
-data Migration = Migration {name :: String, up :: Query}
-  deriving (Show)
-
-schemaMigrations :: [(String, Query)]
-schemaMigrations =
-  [ ("20220101_initial", m20220202_initial)
-  ]
-
--- | The list of migrations in ascending order by date
-app :: [Migration]
-app = sortBy (compare `on` name) $ map migration schemaMigrations
-  where
-    migration (name, query) = Migration {name, up = query}
-
-get :: Connection -> [Migration] -> IO (Either String [Migration])
-get conn migrations =
-  migrationsToRun migrations . map fromOnly
-    <$> DB.query_ conn "SELECT name FROM migrations ORDER BY name ASC;"
-
-run :: Connection -> [Migration] -> IO ()
-run conn ms = withTransaction conn . forM_ ms $
-  \Migration {name, up} -> insert name >> exec conn (fromQuery up)
-  where
-    insert name = DB.execute conn "INSERT INTO migrations (name, ts) VALUES (?, ?);" . (name,) =<< getCurrentTime
-
-initialize :: Connection -> IO ()
-initialize conn =
-  void $
-    DB.execute_
-      conn
-      [sql|
-    CREATE TABLE IF NOT EXISTS migrations (
-      name TEXT NOT NULL,
-      ts TEXT NOT NULL,
-      PRIMARY KEY (name)
-    );
-  |]
-
-migrationsToRun :: [Migration] -> [String] -> Either String [Migration]
-migrationsToRun appMs [] = Right appMs
-migrationsToRun [] dbMs = Left $ "database version is newer than the app: " <> intercalate ", " dbMs
-migrationsToRun (a : as) (d : ds)
-  | name a == d = migrationsToRun as ds
-  | otherwise = Left $ "different migration in the app/database: " <> name a <> " / " <> d

src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20220202_initial.hs

@@ -1,158 +0,0 @@
-{-# LANGUAGE QuasiQuotes #-}
-
-module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial where
-
-import Database.PostgreSQL.Simple (Query)
-import Database.PostgreSQL.Simple.SqlQQ (sql)
-
-m20220202_initial :: Query
-m20220202_initial =
-  [sql|
--- for easy testing
-DROP SCHEMA public CASCADE;
-CREATE SCHEMA public;
-
-CREATE TABLE servers (
-  host TEXT NOT NULL,
-  port TEXT NOT NULL,
-  key_hash BYTEA NOT NULL,
-  PRIMARY KEY (host, port)
-);
-
-CREATE TABLE connections (
-  conn_id BYTEA NOT NULL PRIMARY KEY,
-  conn_mode TEXT NOT NULL,
-  last_internal_msg_id INTEGER NOT NULL DEFAULT 0,
-  last_internal_rcv_msg_id INTEGER NOT NULL DEFAULT 0,
-  last_internal_snd_msg_id INTEGER NOT NULL DEFAULT 0,
-  last_external_snd_msg_id INTEGER NOT NULL DEFAULT 0,
-  last_rcv_msg_hash BYTEA NOT NULL DEFAULT '',
-  last_snd_msg_hash BYTEA NOT NULL DEFAULT '',
-  smp_agent_version INTEGER NOT NULL DEFAULT 1
-);
-
-CREATE TABLE rcv_queues (
-  host TEXT NOT NULL,
-  port TEXT NOT NULL,
-  rcv_id BYTEA NOT NULL,
-  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
-  rcv_private_key BYTEA NOT NULL,
-  rcv_dh_secret BYTEA NOT NULL,
-  e2e_priv_key BYTEA NOT NULL,
-  e2e_dh_secret BYTEA,
-  snd_id BYTEA NOT NULL,
-  snd_key BYTEA,
-  status TEXT NOT NULL,
-  smp_server_version INTEGER NOT NULL DEFAULT 1,
-  smp_client_version INTEGER,
-  PRIMARY KEY (host, port, rcv_id),
-  FOREIGN KEY (host, port) REFERENCES servers
-    ON DELETE RESTRICT ON UPDATE CASCADE,
-  UNIQUE (host, port, snd_id)
-);
-
-CREATE TABLE snd_queues (
-  host TEXT NOT NULL,
-  port TEXT NOT NULL,
-  snd_id BYTEA NOT NULL,
-  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
-  snd_private_key BYTEA NOT NULL,
-  e2e_dh_secret BYTEA NOT NULL,
-  status TEXT NOT NULL,
-  smp_server_version INTEGER NOT NULL DEFAULT 1,
-  smp_client_version INTEGER NOT NULL DEFAULT 1,
-  PRIMARY KEY (host, port, snd_id),
-  FOREIGN KEY (host, port) REFERENCES servers
-    ON DELETE RESTRICT ON UPDATE CASCADE
-);
-
-CREATE TABLE messages (
-  conn_id BYTEA NOT NULL REFERENCES connections (conn_id)
-    ON DELETE CASCADE,
-  internal_id INTEGER NOT NULL,
-  internal_ts TIMESTAMP NOT NULL,
-  internal_rcv_id INTEGER,
-  internal_snd_id INTEGER,
-  msg_type BYTEA NOT NULL, -- (H)ELLO, (R)EPLY, (D)ELETE. Should SMP confirmation be saved too?
-  msg_body BYTEA NOT NULL DEFAULT '',
-  PRIMARY KEY (conn_id, internal_id)
-);
-
-CREATE TABLE rcv_messages (
-  conn_id BYTEA NOT NULL,
-  internal_rcv_id INTEGER NOT NULL,
-  internal_id INTEGER NOT NULL,
-  external_snd_id INTEGER NOT NULL,
-  broker_id BYTEA NOT NULL,
-  broker_ts TIMESTAMP NOT NULL,
-  internal_hash BYTEA NOT NULL,
-  external_prev_snd_hash BYTEA NOT NULL,
-  integrity BYTEA NOT NULL, -- in the list of keywords
-  PRIMARY KEY (conn_id, internal_rcv_id),
-  FOREIGN KEY (conn_id, internal_id) REFERENCES messages
-    ON DELETE CASCADE
-);
-
-ALTER TABLE messages
-ADD CONSTRAINT fk_messages_rcv_messages
-  FOREIGN KEY (conn_id, internal_rcv_id) REFERENCES rcv_messages
-  ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED;
-
-CREATE TABLE snd_messages (
-  conn_id BYTEA NOT NULL,
-  internal_snd_id INTEGER NOT NULL,
-  internal_id INTEGER NOT NULL,
-  internal_hash BYTEA NOT NULL,
-  previous_msg_hash BYTEA NOT NULL DEFAULT '',
-  PRIMARY KEY (conn_id, internal_snd_id),
-  FOREIGN KEY (conn_id, internal_id) REFERENCES messages
-    ON DELETE CASCADE
-);
-
-ALTER TABLE messages
-ADD CONSTRAINT fk_messages_snd_messages
-  FOREIGN KEY (conn_id, internal_snd_id) REFERENCES snd_messages
-  ON DELETE CASCADE DEFERRABLE INITIALLY deferred;
-
-CREATE TABLE conn_confirmations (
-  confirmation_id BYTEA NOT NULL PRIMARY KEY,
-  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
-  e2e_snd_pub_key BYTEA NOT NULL, -- TODO per-queue key. Split?
-  sender_key BYTEA NOT NULL, -- TODO per-queue key. Split?
-  ratchet_state BYTEA NOT NULL,
-  sender_conn_info BYTEA NOT NULL,
-  accepted INTEGER NOT NULL,
-  own_conn_info BYTEA,
-  created_at TIMESTAMP NOT NULL DEFAULT (now())
-);
-
-CREATE TABLE conn_invitations (
-  invitation_id BYTEA NOT NULL PRIMARY KEY,
-  contact_conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
-  cr_invitation BYTEA NOT NULL,
-  recipient_conn_info BYTEA NOT NULL,
-  accepted INTEGER NOT NULL DEFAULT 0,
-  own_conn_info BYTEA,
-  created_at TIMESTAMP NOT NULL DEFAULT (now())
-);
-
-CREATE TABLE ratchets (
-  conn_id BYTEA NOT NULL PRIMARY KEY REFERENCES connections
-    ON DELETE CASCADE,
-  -- x3dh keys are not saved on the sending side (the side accepting the connection)
-  x3dh_priv_key_1 BYTEA,
-  x3dh_priv_key_2 BYTEA,
-  -- ratchet is initially empty on the receiving side (the side offering the connection)
-  ratchet_state BYTEA,
-  e2e_version INTEGER NOT NULL DEFAULT 1
-);
-
-CREATE TABLE skipped_messages (
-  skipped_message_id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
-  conn_id BYTEA NOT NULL REFERENCES ratchets
-    ON DELETE CASCADE,
-  header_key BYTEA NOT NULL,
-  msg_n INTEGER NOT NULL,
-  msg_key BYTEA NOT NULL
-);
-|]

src/Simplex/Messaging/Agent/Store/Postgres/README.md

@@ -1,9 +0,0 @@
-# Postgres setup
-
-Create three databases - `agent_poc_1`, `agent_poc_2`, `agent_poc_3` - and have Postgres server running.
-
-~~`brew install postgresql` - required by postgresql-simple.~~
-
-~~You may run into compilation errors, then you might also need to `brew install libpq --build-from-source`, see [this Stack Overflow answer](https://stackoverflow.com/a/70012033).~~
-
-In the end I managed to build using cabal.

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

@@ -76,13 +76,13 @@ data SQLiteStore = SQLiteStore
     dbNew :: Bool
   }
 
-createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore
-createSQLiteStore dbFilePath poolSize migrations = do
+createSQLiteStore :: FilePath -> Int -> [Migration] -> Bool -> IO SQLiteStore
+createSQLiteStore dbFilePath poolSize migrations yesToMigrations = do
   let dbDir = takeDirectory dbFilePath
   createDirectoryIfMissing False dbDir
   st <- connectSQLiteStore dbFilePath poolSize
   checkThreadsafe st
-  migrateSchema st migrations
+  migrateSchema st migrations yesToMigrations
   pure st
 
 checkThreadsafe :: SQLiteStore -> IO ()
@@ -94,15 +94,16 @@ checkThreadsafe st = withConnection st $ \db -> do
     Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found"
     _ -> return ()
 
-migrateSchema :: SQLiteStore -> [Migration] -> IO ()
-migrateSchema st migrations = withConnection st $ \db -> do
+migrateSchema :: SQLiteStore -> [Migration] -> Bool -> IO ()
+migrateSchema st migrations yesToMigrations = withConnection st $ \db -> do
   Migrations.initialize db
   Migrations.get db migrations >>= \case
     Left e -> confirmOrExit $ "Database error: " <> e
     Right [] -> pure ()
     Right ms -> do
       unless (dbNew st) $ do
-        confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
+        unless yesToMigrations $
+          confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
         let f = dbFilePath st
         copyFile f (f <> ".bak")
       Migrations.run db ms
@@ -440,7 +441,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
       insertSndMsgDetails_ db connId sndMsgData
       updateHashSnd_ db connId sndMsgData
 
-  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))
+  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AgentMessageType, MsgBody, InternalTs))
   getPendingMsgData st connId msgId =
     liftIOEither . withTransaction st $ \db -> runExceptT $ do
       rq_ <- liftIO $ getRcvQueueByConnId_ db connId
@@ -565,9 +566,9 @@ instance ToField InternalId where toField (InternalId x) = toField x
 
 instance FromField InternalId where fromField x = InternalId <$> fromField x
 
-instance ToField AMsgType where toField = toField . smpEncode
+instance ToField AgentMessageType where toField = toField . smpEncode
 
-instance FromField AMsgType where fromField = blobFieldParser smpP
+instance FromField AgentMessageType where fromField = blobFieldParser smpP
 
 instance ToField MsgIntegrity where toField = toField . strEncode
 
@@ -655,46 +656,25 @@ upsertServer_ dbConn SMPServer {host, port, keyHash} = do
 
 insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO ()
 insertRcvQueue_ dbConn connId RcvQueue {..} = do
-  DB.executeNamed
+  DB.execute
     dbConn
     [sql|
       INSERT INTO rcv_queues
-        ( host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status)
-      VALUES
-        (:host,:port,:rcv_id,:conn_id,:rcv_private_key,:rcv_dh_secret,:e2e_priv_key,:e2e_dh_secret,:snd_id,:status);
+        ( host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status) VALUES (?,?,?,?,?,?,?,?,?,?);
     |]
-    [ ":host" := host server,
-      ":port" := port server,
-      ":rcv_id" := rcvId,
-      ":conn_id" := connId,
-      ":rcv_private_key" := rcvPrivateKey,
-      ":rcv_dh_secret" := rcvDhSecret,
-      ":e2e_priv_key" := e2ePrivKey,
-      ":e2e_dh_secret" := e2eDhSecret,
-      ":snd_id" := sndId,
-      ":status" := status
-    ]
+    (host server, port server, rcvId, connId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status)
 
 -- * createSndConn helpers
 
 insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO ()
 insertSndQueue_ dbConn connId SndQueue {..} = do
-  DB.executeNamed
+  DB.execute
     dbConn
     [sql|
       INSERT INTO snd_queues
-        ( host, port, snd_id, conn_id, snd_private_key, e2e_dh_secret, status)
-      VALUES
-        (:host,:port,:snd_id,:conn_id,:snd_private_key,:e2e_dh_secret,:status);
+        (host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status) VALUES (?,?,?,?,?, ?,?, ?,?);
     |]
-    [ ":host" := host server,
-      ":port" := port server,
-      ":snd_id" := sndId,
-      ":conn_id" := connId,
-      ":snd_private_key" := sndPrivateKey,
-      ":e2e_dh_secret" := e2eDhSecret,
-      ":status" := status
-    ]
+    (host server, port server, sndId, connId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status)
 
 -- * getConn helpers
 
@@ -745,16 +725,16 @@ getSndQueueByConnId_ dbConn connId =
     <$> DB.query
       dbConn
       [sql|
-        SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_private_key, q.e2e_dh_secret, q.status
+        SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status
         FROM snd_queues q
         INNER JOIN servers s ON q.host = s.host AND q.port = s.port
         WHERE q.conn_id = ?;
       |]
       (Only connId)
   where
-    sndQueue [(keyHash, host, port, sndId, sndPrivateKey, e2eDhSecret, status)] =
+    sndQueue [(keyHash, host, port, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status)] =
       let server = SMPServer host port keyHash
-       in Just SndQueue {server, sndId, sndPrivateKey, e2eDhSecret, status}
+       in Just SndQueue {server, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status}
     sndQueue _ = Nothing
 
 -- * updateRcvIds helpers

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

@@ -25,13 +25,17 @@ import qualified Database.SQLite.Simple as DB
 import Database.SQLite.Simple.QQ (sql)
 import qualified Database.SQLite3 as SQLite3
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220320_server_ips
 
 data Migration = Migration {name :: String, up :: Text}
   deriving (Show)
 
 schemaMigrations :: [(String, Query)]
 schemaMigrations =
-  [ ("20220101_initial", m20220101_initial)
+  [ ("20220101_initial", m20220101_initial),
+    ("20220301_snd_queue_keys", m20220301_snd_queue_keys),
+    ("20220320_server_ips", m20220320_server_ips)
   ]
 
 -- | The list of migrations in ascending order by date

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

@@ -0,0 +1,13 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20220301_snd_queue_keys :: Query
+m20220301_snd_queue_keys =
+  [sql|
+ALTER TABLE snd_queues ADD COLUMN snd_public_key BLOB;
+ALTER TABLE snd_queues ADD COLUMN e2e_pub_key BLOB;
+|]

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

@@ -0,0 +1,14 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220320_server_ips where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20220320_server_ips :: Query
+m20220320_server_ips =
+  [sql|
+UPDATE servers SET host = '178.79.168.175' WHERE host = 'smp8.simplex.im';
+UPDATE servers SET host = '178.79.169.107' WHERE host = 'smp9.simplex.im';
+UPDATE servers SET host = '45.33.54.229' WHERE host = 'smp10.simplex.im';
+|]

src/Simplex/Messaging/Client.hs

@@ -107,7 +107,7 @@ data SMPClientConfig = SMPClientConfig
 smpDefaultConfig :: SMPClientConfig
 smpDefaultConfig =
   SMPClientConfig
-    { qSize = 16,
+    { qSize = 64,
       defaultTransport = ("5223", transport @TLS),
       tcpTimeout = 4_000_000,
       smpPing = 30_000_000

src/Simplex/Messaging/Crypto.hs

@@ -149,20 +149,14 @@ import Data.String
 import Data.Type.Equality
 import Data.Typeable (Typeable)
 import Data.X509
-import qualified Database.PostgreSQL.Simple as PDB
-import qualified Database.PostgreSQL.Simple.FromField as PF
-import qualified Database.PostgreSQL.Simple.ToField as PT
-import qualified Database.PostgreSQL.Simple.TypeInfo as PTI
-import qualified Database.PostgreSQL.Simple.TypeInfo.Static as PTIS
-import qualified Database.SQLite.Simple.FromField as SF
-import qualified Database.SQLite.Simple.ToField as ST
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
 import GHC.TypeLits (ErrorMessage (..), TypeError)
 import Network.Transport.Internal (decodeWord16, encodeWord16)
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (blobFieldDecoder, parseAll, parseString)
 import Simplex.Messaging.Util ((<$?>))
-import qualified Database.PostgreSQL.Simple as PDB
 
 -- | Cryptographic algorithms.
 data Algorithm = Ed25519 | Ed448 | X25519 | X448
@@ -546,62 +540,33 @@ generateKeyPair' = case sAlgorithm @a of
       let k = X448.toPublic pk
        in pure (PublicKeyX448 k, PrivateKeyX448 pk k)
 
-instance ST.ToField APrivateSignKey where toField = ST.toField . encodePrivKey
+instance ToField APrivateSignKey where toField = toField . encodePrivKey
 
-instance ST.ToField APublicVerifyKey where toField = ST.toField . encodePubKey
+instance ToField APublicVerifyKey where toField = toField . encodePubKey
 
-instance ST.ToField APrivateDhKey where toField = ST.toField . encodePrivKey
+instance ToField APrivateDhKey where toField = toField . encodePrivKey
 
-instance ST.ToField APublicDhKey where toField = ST.toField . encodePubKey
+instance ToField APublicDhKey where toField = toField . encodePubKey
 
-instance AlgorithmI a => ST.ToField (PrivateKey a) where toField = ST.toField . encodePrivKey
+instance AlgorithmI a => ToField (PrivateKey a) where toField = toField . encodePrivKey
 
-instance AlgorithmI a => ST.ToField (PublicKey a) where toField = ST.toField . encodePubKey
+instance AlgorithmI a => ToField (PublicKey a) where toField = toField . encodePubKey
 
-instance ST.ToField (DhSecret a) where toField = ST.toField . dhBytes'
+instance ToField (DhSecret a) where toField = toField . dhBytes'
 
-instance SF.FromField APrivateSignKey where fromField = blobFieldDecoder decodePrivKey
+instance FromField APrivateSignKey where fromField = blobFieldDecoder decodePrivKey
 
-instance SF.FromField APublicVerifyKey where fromField = blobFieldDecoder decodePubKey
+instance FromField APublicVerifyKey where fromField = blobFieldDecoder decodePubKey
 
-instance SF.FromField APrivateDhKey where fromField = blobFieldDecoder decodePrivKey
+instance FromField APrivateDhKey where fromField = blobFieldDecoder decodePrivKey
 
-instance SF.FromField APublicDhKey where fromField = blobFieldDecoder decodePubKey
+instance FromField APublicDhKey where fromField = blobFieldDecoder decodePubKey
 
-instance (Typeable a, AlgorithmI a) => SF.FromField (PrivateKey a) where fromField = blobFieldDecoder decodePrivKey
+instance (Typeable a, AlgorithmI a) => FromField (PrivateKey a) where fromField = blobFieldDecoder decodePrivKey
 
-instance (Typeable a, AlgorithmI a) => SF.FromField (PublicKey a) where fromField = blobFieldDecoder decodePubKey
+instance (Typeable a, AlgorithmI a) => FromField (PublicKey a) where fromField = blobFieldDecoder decodePubKey
 
-instance (Typeable a, AlgorithmI a) => SF.FromField (DhSecret a) where fromField = blobFieldDecoder strDecode
-
-instance PT.ToField APrivateSignKey where toField = PT.toField . encodePrivKey
-
-instance PT.ToField APublicVerifyKey where toField = PT.toField . encodePubKey
-
-instance PT.ToField APrivateDhKey where toField = PT.toField . encodePrivKey
-
-instance PT.ToField APublicDhKey where toField = PT.toField . encodePubKey
-
-instance AlgorithmI a => PT.ToField (PrivateKey a) where toField = PT.toField . encodePrivKey
-
-instance AlgorithmI a => PT.ToField (PublicKey a) where toField = PT.toField . encodePubKey
-
-instance PT.ToField (DhSecret a) where toField = PT.toField . PDB.Binary . dhBytes'
-
-instance PF.FromField APrivateSignKey where fromField = fromByteStringField decodePrivKey
-
-instance PF.FromField APublicVerifyKey where fromField = fromByteStringField decodePubKey
-
-instance PF.FromField APrivateDhKey where fromField = fromByteStringField decodePrivKey
-
-instance PF.FromField APublicDhKey where fromField = fromByteStringField decodePubKey
-
-instance (Typeable a, AlgorithmI a) => PF.FromField (PrivateKey a) where fromField = fromByteStringField decodePrivKey
-
-instance (Typeable a, AlgorithmI a) => PF.FromField (PublicKey a) where fromField = fromByteStringField decodePubKey
-
--- instance (Typeable a, AlgorithmI a) => PF.FromField (DhSecret a) where fromField = fromByteStringField strDecode
-instance (Typeable a, AlgorithmI a) => PF.FromField (DhSecret a) where fromField x = fromByteStringField strDecode x
+instance (Typeable a, AlgorithmI a) => FromField (DhSecret a) where fromField = blobFieldDecoder strDecode
 
 instance IsString (Maybe ASignature) where
   fromString = parseString $ decode >=> decodeSignature
@@ -725,13 +690,9 @@ validSignatureSize n =
 newtype Key = Key {unKey :: ByteString}
   deriving (Eq, Ord, Show)
 
-instance ST.ToField Key where toField = ST.toField . unKey
+instance ToField Key where toField = toField . unKey
 
-instance PT.ToField Key where toField = PT.toField . unKey
-
-instance SF.FromField Key where fromField f = Key <$> SF.fromField f
-
-instance PF.FromField Key where fromField f = PF.fromField f
+instance FromField Key where fromField f = Key <$> fromField f
 
 instance ToJSON Key where
   toJSON = strToJSON . unKey
@@ -769,27 +730,9 @@ instance StrEncoding KeyHash where
 instance IsString KeyHash where
   fromString = parseString $ parseAll strP
 
-instance ST.ToField KeyHash where toField = ST.toField . strEncode
+instance ToField KeyHash where toField = toField . strEncode
 
-instance SF.FromField KeyHash where fromField = blobFieldDecoder $ parseAll strP
-
-instance PT.ToField KeyHash where toField = PT.toField . strEncode
-
--- TODO
--- instance PF.FromField KeyHash where fromField = blobFieldDecoderPostgres $ parseAll strP
-
-instance PF.FromField KeyHash where fromField = fromByteStringField $ parseAll strP
-
-fromByteStringField :: Typeable a => (ByteString -> Either String a) -> PF.Field -> Maybe ByteString -> PF.Conversion a
-fromByteStringField dec f mdata =
-  if PF.typeOid f /= PTI.typoid PTIS.bytea
-    then PF.returnError PF.Incompatible f ""
-    else case mdata of
-      Nothing -> PF.returnError PF.UnexpectedNull f ""
-      Just dat ->
-        case dec dat of
-          Right x -> return x
-          _ -> PF.returnError PF.ConversionFailed f (B.unpack dat)
+instance FromField KeyHash where fromField = blobFieldDecoder $ parseAll strP
 
 -- | SHA256 digest.
 sha256Hash :: ByteString -> ByteString

src/Simplex/Messaging/Crypto/Ratchet.hs

@@ -30,12 +30,8 @@ import qualified Data.Map.Strict as M
 import Data.Maybe (fromMaybe)
 import Data.Typeable (Typeable)
 import Data.Word (Word32)
-import qualified Database.PostgreSQL.Simple.FromField as PF
-import qualified Database.PostgreSQL.Simple.ToField as PT
-import qualified Database.PostgreSQL.Simple.TypeInfo as PTI
-import qualified Database.PostgreSQL.Simple.TypeInfo.Static as PTIS
-import qualified Database.SQLite.Simple.FromField as SF
-import qualified Database.SQLite.Simple.ToField as ST
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
 import GHC.Generics
 import Simplex.Messaging.Agent.QueryString
 import Simplex.Messaging.Crypto
@@ -201,32 +197,13 @@ instance ToJSON RatchetKey where
 instance FromJSON RatchetKey where
   parseJSON = fmap RatchetKey . strParseJSON "Key"
 
-instance AlgorithmI a => ST.ToField (Ratchet a) where toField = ST.toField . LB.toStrict . J.encode
+instance AlgorithmI a => ToField (Ratchet a) where toField = toField . LB.toStrict . J.encode
 
-instance AlgorithmI a => PT.ToField (Ratchet a) where toField = PT.toField . LB.toStrict . J.encode
+instance (AlgorithmI a, Typeable a) => FromField (Ratchet a) where fromField = blobFieldDecoder J.eitherDecodeStrict'
 
-instance (AlgorithmI a, Typeable a) => PF.FromField (Ratchet a) where fromField = fromByteStringField J.eitherDecodeStrict'
+instance ToField MessageKey where toField = toField . smpEncode
 
-instance (AlgorithmI a, Typeable a) => SF.FromField (Ratchet a) where fromField = blobFieldDecoder J.eitherDecodeStrict'
-
-instance ST.ToField MessageKey where toField = ST.toField . smpEncode
-
-instance PT.ToField MessageKey where toField = PT.toField . smpEncode
-
-instance SF.FromField MessageKey where fromField = blobFieldDecoder smpDecode
-
-instance PF.FromField MessageKey where fromField = fromByteStringField smpDecode
-
-fromByteStringField :: Typeable a => (ByteString -> Either String a) -> PF.Field -> Maybe ByteString -> PF.Conversion a
-fromByteStringField dec f mdata =
-  if PF.typeOid f /= PTI.typoid PTIS.bytea
-    then PF.returnError PF.Incompatible f ""
-    else case mdata of
-      Nothing -> PF.returnError PF.UnexpectedNull f ""
-      Just dat ->
-        case dec dat of
-          Right x -> return x
-          _ -> PF.returnError PF.ConversionFailed f (B.unpack dat)
+instance FromField MessageKey where fromField = blobFieldDecoder smpDecode
 
 -- | Sending ratchet initialization, equivalent to RatchetInitAliceHE in double ratchet spec
 --

src/Simplex/Messaging/Parsers.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
 
@@ -15,13 +16,10 @@ import Data.Char (isAlphaNum, toLower)
 import Data.Time.Clock (UTCTime)
 import Data.Time.ISO8601 (parseISO8601)
 import Data.Typeable (Typeable)
-import qualified Database.PostgreSQL.Simple.FromField as PF
-import qualified Database.PostgreSQL.Simple.Internal as PI
-import qualified Database.PostgreSQL.Simple.Ok as PO
 import Database.SQLite.Simple (ResultError (..), SQLData (..))
-import qualified Database.SQLite.Simple.FromField as SF
-import qualified Database.SQLite.Simple.Internal as SI
-import qualified Database.SQLite.Simple.Ok as SO
+import Database.SQLite.Simple.FromField (FieldParser, returnError)
+import Database.SQLite.Simple.Internal (Field (..))
+import Database.SQLite.Simple.Ok (Ok (Ok))
 import Simplex.Messaging.Util ((<$?>))
 import Text.Read (readMaybe)
 
@@ -72,24 +70,16 @@ wordEnd c = c == ' ' || c == '\n'
 parseString :: (ByteString -> Either String a) -> (String -> a)
 parseString p = either error id . p . B.pack
 
-blobFieldParser :: Typeable k => Parser k -> SF.FieldParser k
+blobFieldParser :: Typeable k => Parser k -> FieldParser k
 blobFieldParser = blobFieldDecoder . parseAll
 
-blobFieldDecoder :: Typeable k => (ByteString -> Either String k) -> SF.FieldParser k
+blobFieldDecoder :: Typeable k => (ByteString -> Either String k) -> FieldParser k
 blobFieldDecoder dec = \case
-  f@(SI.Field (SQLBlob b) _) ->
+  f@(Field (SQLBlob b) _) ->
     case dec b of
-      Right k -> SO.Ok k
-      Left e -> SF.returnError SF.ConversionFailed f ("couldn't parse field: " ++ e)
-  f -> SF.returnError SF.ConversionFailed f "expecting SQLBlob column type"
-
--- blobFieldDecoderPostgres :: Typeable k => (ByteString -> Either String k) -> PF.FieldParser k
--- blobFieldDecoderPostgres dec = \case
---   f@(PI.Field b _ _) ->
---     case dec b of
---       Right k -> PO.Ok k
---       Left e -> PF.returnError PF.ConversionFailed f ("couldn't parse field: " ++ e)
---   f -> PF.returnError PF.ConversionFailed f "expecting SQLBlob column type"
+      Right k -> Ok k
+      Left e -> returnError ConversionFailed f ("couldn't parse field: " ++ e)
+  f -> returnError ConversionFailed f "expecting SQLBlob column type"
 
 fstToLower :: String -> String
 fstToLower "" = ""
@@ -108,13 +98,18 @@ enumJSON tagModifier =
     }
 
 sumTypeJSON :: (String -> String) -> J.Options
+#if defined(darwin_HOST_OS)
 sumTypeJSON = singleFieldJSON
+#else
+sumTypeJSON = taggedObjectJSON
+#endif
 
 taggedObjectJSON :: (String -> String) -> J.Options
 taggedObjectJSON tagModifier =
   J.defaultOptions
     { J.sumEncoding = J.TaggedObject "type" "data",
       J.constructorTagModifier = tagModifier,
+      J.allNullaryToStringTag = False,
       J.nullaryToObject = True,
       J.omitNothingFields = True
     }
@@ -124,6 +119,7 @@ singleFieldJSON tagModifier =
   J.defaultOptions
     { J.sumEncoding = J.ObjectWithSingleField,
       J.constructorTagModifier = tagModifier,
+      J.allNullaryToStringTag = False,
       J.nullaryToObject = True,
       J.omitNothingFields = True
     }

src/Simplex/Messaging/Protocol.hs

@@ -399,6 +399,10 @@ instance StrEncoding SMPServer where
     SrvLoc host port <- strP
     pure SMPServer {host, port, keyHash}
 
+instance ToJSON SMPServer where
+  toJSON = strToJSON
+  toEncoding = strToJEncoding
+
 data SrvLoc = SrvLoc HostName ServiceName
   deriving (Eq, Ord, Show)
 

src/Simplex/Messaging/Transport.hs

@@ -96,7 +96,7 @@ supportedSMPVersions :: VersionRange
 supportedSMPVersions = mkVersionRange 1 1
 
 simplexMQVersion :: String
-simplexMQVersion = "1.0.2"
+simplexMQVersion = "1.0.3"
 
 -- * Transport connection class
 

src/Simplex/Messaging/Transport/Client.hs

@@ -75,7 +75,7 @@ validateCertificateChain (C.KeyHash kh) host port cc@(X.CertificateChain sc@[_,
     x509validate = XV.validate X.HashSHA256 hooks checks certStore cache serviceID cc
       where
         hooks = XV.defaultHooks
-        checks = XV.defaultChecks
+        checks = XV.defaultChecks {XV.checkFQHN = False}
         certStore = XS.makeCertificateStore sc
         cache = XV.exceptionValidationCache [] -- we manually check fingerprint only of the identity certificate (ca.crt)
         serviceID = (host, port)

tests/AgentTests.hs

@@ -13,7 +13,6 @@ import AgentTests.ConnectionRequestTests
 import AgentTests.DoubleRatchetTests (doubleRatchetTests)
 import AgentTests.FunctionalAPITests (functionalAPITests)
 import AgentTests.SQLiteTests (storeTests)
-import AgentTests.PostgresTests (postgresStoreTests)
 import Control.Concurrent
 import Control.Monad (forM_)
 import Data.ByteString.Char8 (ByteString)
@@ -37,7 +36,6 @@ agentTests (ATransport t) = do
   describe "Double ratchet tests" doubleRatchetTests
   describe "Functional API" $ functionalAPITests (ATransport t)
   describe "SQLite store" storeTests
-  describe "Postgres store" postgresStoreTests
   describe "SMP agent protocol syntax" $ syntaxTests t
   describe "Establishing duplex connection" $ do
     it "should connect via one server and one agent" $
@@ -64,9 +62,11 @@ agentTests (ATransport t) = do
       smpAgentTest3_1_1 $ testSubscription t
     it "should send notifications to client when server disconnects" $
       smpAgentServerTest $ testSubscrNotification t
-  describe "Message delivery" $ do
+  describe "Message delivery and server reconnection" $ do
     it "should deliver messages after losing server connection and re-connecting" $
       smpAgentTest2_2_2_needs_server $ testMsgDeliveryServerRestart t
+    it "should connect to the server when server goes up if it initially was down" $
+      smpAgentTestN [] $ testServerConnectionAfterError t
     it "should deliver pending messages after agent restarting" $
       smpAgentTest1_1_1 $ testMsgDeliveryAgentRestart t
     it "should concurrently deliver messages to connections without blocking" $
@@ -128,25 +128,25 @@ testDuplexConnection _ alice bob = do
   bob <# ("", "alice", CON)
   alice <# ("", "bob", CON)
   -- message IDs 1 to 3 get assigned to control messages, so first MSG is assigned ID 4
-  alice #: ("3", "bob", "SEND :hello") #> ("3", "bob", MID 4)
-  alice <# ("", "bob", SENT 4)
-  bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
-  bob #: ("12", "alice", "ACK 4") #> ("12", "alice", OK)
-  alice #: ("4", "bob", "SEND :how are you?") #> ("4", "bob", MID 5)
+  alice #: ("3", "bob", "SEND :hello") #> ("3", "bob", MID 5)
   alice <# ("", "bob", SENT 5)
+  bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
+  bob #: ("12", "alice", "ACK 5") #> ("12", "alice", OK)
+  alice #: ("4", "bob", "SEND :how are you?") #> ("4", "bob", MID 6)
+  alice <# ("", "bob", SENT 6)
   bob <#= \case ("", "alice", Msg "how are you?") -> True; _ -> False
-  bob #: ("13", "alice", "ACK 5") #> ("13", "alice", OK)
-  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 6)
-  bob <# ("", "alice", SENT 6)
-  alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
-  alice #: ("3a", "bob", "ACK 6") #> ("3a", "bob", OK)
-  bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", MID 7)
+  bob #: ("13", "alice", "ACK 6") #> ("13", "alice", OK)
+  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 7)
   bob <# ("", "alice", SENT 7)
+  alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
+  alice #: ("3a", "bob", "ACK 7") #> ("3a", "bob", OK)
+  bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", MID 8)
+  bob <# ("", "alice", SENT 8)
   alice <#= \case ("", "bob", Msg "message 1") -> True; _ -> False
-  alice #: ("4a", "bob", "ACK 7") #> ("4a", "bob", OK)
+  alice #: ("4a", "bob", "ACK 8") #> ("4a", "bob", OK)
   alice #: ("5", "bob", "OFF") #> ("5", "bob", OK)
-  bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", MID 8)
-  bob <# ("", "alice", MERR 8 (SMP AUTH))
+  bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", MID 9)
+  bob <# ("", "alice", MERR 9 (SMP AUTH))
   alice #: ("6", "bob", "DEL") #> ("6", "bob", OK)
   alice #:# "nothing else should be delivered to alice"
 
@@ -161,25 +161,25 @@ testDuplexConnRandomIds _ alice bob = do
   bob <# ("", aliceConn, INFO "alice's connInfo")
   bob <# ("", aliceConn, CON)
   alice <# ("", bobConn, CON)
-  alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, MID 4)
-  alice <# ("", bobConn, SENT 4)
-  bob <#= \case ("", c, Msg "hello") -> c == aliceConn; _ -> False
-  bob #: ("12", aliceConn, "ACK 4") #> ("12", aliceConn, OK)
-  alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, MID 5)
+  alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, MID 5)
   alice <# ("", bobConn, SENT 5)
+  bob <#= \case ("", c, Msg "hello") -> c == aliceConn; _ -> False
+  bob #: ("12", aliceConn, "ACK 5") #> ("12", aliceConn, OK)
+  alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, MID 6)
+  alice <# ("", bobConn, SENT 6)
   bob <#= \case ("", c, Msg "how are you?") -> c == aliceConn; _ -> False
-  bob #: ("13", aliceConn, "ACK 5") #> ("13", aliceConn, OK)
-  bob #: ("14", aliceConn, "SEND 9\nhello too") #> ("14", aliceConn, MID 6)
-  bob <# ("", aliceConn, SENT 6)
-  alice <#= \case ("", c, Msg "hello too") -> c == bobConn; _ -> False
-  alice #: ("3a", bobConn, "ACK 6") #> ("3a", bobConn, OK)
-  bob #: ("15", aliceConn, "SEND 9\nmessage 1") #> ("15", aliceConn, MID 7)
+  bob #: ("13", aliceConn, "ACK 6") #> ("13", aliceConn, OK)
+  bob #: ("14", aliceConn, "SEND 9\nhello too") #> ("14", aliceConn, MID 7)
   bob <# ("", aliceConn, SENT 7)
+  alice <#= \case ("", c, Msg "hello too") -> c == bobConn; _ -> False
+  alice #: ("3a", bobConn, "ACK 7") #> ("3a", bobConn, OK)
+  bob #: ("15", aliceConn, "SEND 9\nmessage 1") #> ("15", aliceConn, MID 8)
+  bob <# ("", aliceConn, SENT 8)
   alice <#= \case ("", c, Msg "message 1") -> c == bobConn; _ -> False
-  alice #: ("4a", bobConn, "ACK 7") #> ("4a", bobConn, OK)
+  alice #: ("4a", bobConn, "ACK 8") #> ("4a", bobConn, OK)
   alice #: ("5", bobConn, "OFF") #> ("5", bobConn, OK)
-  bob #: ("17", aliceConn, "SEND 9\nmessage 3") #> ("17", aliceConn, MID 8)
-  bob <# ("", aliceConn, MERR 8 (SMP AUTH))
+  bob #: ("17", aliceConn, "SEND 9\nmessage 3") #> ("17", aliceConn, MID 9)
+  bob <# ("", aliceConn, MERR 9 (SMP AUTH))
   alice #: ("6", bobConn, "DEL") #> ("6", bobConn, OK)
   alice #:# "nothing else should be delivered to alice"
 
@@ -196,10 +196,10 @@ testContactConnection _ alice bob tom = do
   alice <# ("", "bob", INFO "bob's connInfo 2")
   alice <# ("", "bob", CON)
   bob <# ("", "alice", CON)
-  alice #: ("3", "bob", "SEND :hi") #> ("3", "bob", MID 4)
-  alice <# ("", "bob", SENT 4)
+  alice #: ("3", "bob", "SEND :hi") #> ("3", "bob", MID 5)
+  alice <# ("", "bob", SENT 5)
   bob <#= \case ("", "alice", Msg "hi") -> True; _ -> False
-  bob #: ("13", "alice", "ACK 4") #> ("13", "alice", OK)
+  bob #: ("13", "alice", "ACK 5") #> ("13", "alice", OK)
 
   tom #: ("21", "alice", "JOIN " <> cReq' <> " 14\ntom's connInfo") #> ("21", "alice", OK)
   ("", "alice_contact", Right (REQ aInvId' "tom's connInfo")) <- (alice <#:)
@@ -209,10 +209,10 @@ testContactConnection _ alice bob tom = do
   alice <# ("", "tom", INFO "tom's connInfo 2")
   alice <# ("", "tom", CON)
   tom <# ("", "alice", CON)
-  alice #: ("5", "tom", "SEND :hi there") #> ("5", "tom", MID 4)
-  alice <# ("", "tom", SENT 4)
+  alice #: ("5", "tom", "SEND :hi there") #> ("5", "tom", MID 5)
+  alice <# ("", "tom", SENT 5)
   tom <#= \case ("", "alice", Msg "hi there") -> True; _ -> False
-  tom #: ("23", "alice", "ACK 4") #> ("23", "alice", OK)
+  tom #: ("23", "alice", "ACK 5") #> ("23", "alice", OK)
 
 testContactConnRandomIds :: Transport c => TProxy c -> c -> c -> IO ()
 testContactConnRandomIds _ alice bob = do
@@ -232,10 +232,10 @@ testContactConnRandomIds _ alice bob = do
   alice <# ("", bobConn, CON)
   bob <# ("", aliceConn, CON)
 
-  alice #: ("3", bobConn, "SEND :hi") #> ("3", bobConn, MID 4)
-  alice <# ("", bobConn, SENT 4)
+  alice #: ("3", bobConn, "SEND :hi") #> ("3", bobConn, MID 5)
+  alice <# ("", bobConn, SENT 5)
   bob <#= \case ("", c, Msg "hi") -> c == aliceConn; _ -> False
-  bob #: ("13", aliceConn, "ACK 4") #> ("13", aliceConn, OK)
+  bob #: ("13", aliceConn, "ACK 5") #> ("13", aliceConn, OK)
 
 testRejectContactRequest :: Transport c => TProxy c -> c -> c -> IO ()
 testRejectContactRequest _ alice bob = do
@@ -252,20 +252,20 @@ testRejectContactRequest _ alice bob = do
 testSubscription :: Transport c => TProxy c -> c -> c -> c -> IO ()
 testSubscription _ alice1 alice2 bob = do
   (alice1, "alice") `connect` (bob, "bob")
-  bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", MID 4)
-  bob <# ("", "alice", SENT 4)
-  alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False
-  alice1 #: ("1", "bob", "ACK 4") #> ("1", "bob", OK)
-  bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", MID 5)
+  bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", MID 5)
   bob <# ("", "alice", SENT 5)
+  alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False
+  alice1 #: ("1", "bob", "ACK 5") #> ("1", "bob", OK)
+  bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", MID 6)
+  bob <# ("", "alice", SENT 6)
   alice1 <#= \case ("", "bob", Msg "hello again") -> True; _ -> False
-  alice1 #: ("2", "bob", "ACK 5") #> ("2", "bob", OK)
+  alice1 #: ("2", "bob", "ACK 6") #> ("2", "bob", OK)
   alice2 #: ("21", "bob", "SUB") #> ("21", "bob", OK)
   alice1 <# ("", "bob", END)
-  bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", MID 6)
-  bob <# ("", "alice", SENT 6)
+  bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", MID 7)
+  bob <# ("", "alice", SENT 7)
   alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False
-  alice2 #: ("22", "bob", "ACK 6") #> ("22", "bob", OK)
+  alice2 #: ("22", "bob", "ACK 7") #> ("22", "bob", OK)
   alice1 #:# "nothing else should be delivered to alice1"
 
 testSubscrNotification :: Transport c => TProxy c -> (ThreadId, ThreadId) -> c -> IO ()
@@ -281,40 +281,77 @@ testMsgDeliveryServerRestart :: Transport c => TProxy c -> c -> c -> IO ()
 testMsgDeliveryServerRestart t alice bob = do
   withServer $ do
     connect (alice, "alice") (bob, "bob")
-    bob #: ("1", "alice", "SEND 2\nhi") #> ("1", "alice", MID 4)
-    bob <# ("", "alice", SENT 4)
+    bob #: ("1", "alice", "SEND 2\nhi") #> ("1", "alice", MID 5)
+    bob <# ("", "alice", SENT 5)
     alice <#= \case ("", "bob", Msg "hi") -> True; _ -> False
-    alice #: ("11", "bob", "ACK 4") #> ("11", "bob", OK)
+    alice #: ("11", "bob", "ACK 5") #> ("11", "bob", OK)
     alice #:# "nothing else delivered before the server is killed"
 
   alice <# ("", "bob", DOWN)
-  bob #: ("2", "alice", "SEND 11\nhello again") #> ("2", "alice", MID 5)
+  bob #: ("2", "alice", "SEND 11\nhello again") #> ("2", "alice", MID 6)
   bob #:# "nothing else delivered before the server is restarted"
   alice #:# "nothing else delivered before the server is restarted"
 
   withServer $ do
-    bob <# ("", "alice", SENT 5)
+    bob <# ("", "alice", SENT 6)
     alice <# ("", "bob", UP)
     alice <#= \case ("", "bob", Msg "hello again") -> True; _ -> False
-    alice #: ("12", "bob", "ACK 5") #> ("12", "bob", OK)
+    alice #: ("12", "bob", "ACK 6") #> ("12", "bob", OK)
 
   removeFile testStoreLogFile
   where
     withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
 
+testServerConnectionAfterError :: forall c. Transport c => TProxy c -> [c] -> IO ()
+testServerConnectionAfterError t _ = do
+  withAgent1 $ \bob -> do
+    withAgent2 $ \alice -> do
+      withServer $ do
+        connect (bob, "bob") (alice, "alice")
+
+      bob <# ("", "alice", DOWN)
+      alice <# ("", "bob", DOWN)
+      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 5)
+      alice #:# "nothing else delivered before the server is restarted"
+      bob #:# "nothing else delivered before the server is restarted"
+
+  withAgent1 $ \bob -> do
+    withAgent2 $ \alice -> do
+      bob #: ("1", "alice", "SUB") #> ("1", "alice", ERR (BROKER NETWORK))
+      alice #: ("1", "bob", "SUB") #> ("1", "bob", ERR (BROKER NETWORK))
+      withServer $ do
+        alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 5; _ -> False
+        alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 5; _ -> False
+        bob <# ("", "alice", UP)
+        bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
+        bob #: ("2", "alice", "ACK 5") #> ("2", "alice", OK)
+        alice #: ("1", "bob", "SEND 11\nhello again") #> ("1", "bob", MID 6)
+        alice <# ("", "bob", SENT 6)
+        bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False
+
+  removeFile testStoreLogFile
+  removeFile testDB
+  removeFile testDB2
+  where
+    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
+    withAgent1 = withAgent agentTestPort testDB
+    withAgent2 = withAgent agentTestPort2 testDB2
+    withAgent :: String -> String -> (c -> IO a) -> IO a
+    withAgent agentPort agentDB = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) (pure ()) . const . testSMPAgentClientOn agentPort
+
 testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO ()
 testMsgDeliveryAgentRestart t bob = do
   withAgent $ \alice -> do
     withServer $ do
       connect (bob, "bob") (alice, "alice")
-      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 4)
-      alice <# ("", "bob", SENT 4)
+      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 5)
+      alice <# ("", "bob", SENT 5)
       bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
-      bob #: ("11", "alice", "ACK 4") #> ("11", "alice", OK)
+      bob #: ("11", "alice", "ACK 5") #> ("11", "alice", OK)
       bob #:# "nothing else delivered before the server is down"
 
     bob <# ("", "alice", DOWN)
-    alice #: ("2", "bob", "SEND 11\nhello again") #> ("2", "bob", MID 5)
+    alice #: ("2", "bob", "SEND 11\nhello again") #> ("2", "bob", MID 6)
     alice #:# "nothing else delivered before the server is restarted"
     bob #:# "nothing else delivered before the server is restarted"
 
@@ -324,14 +361,14 @@ testMsgDeliveryAgentRestart t bob = do
       alice <#= \case
         (corrId, "bob", cmd) ->
           (corrId == "3" && cmd == OK)
-            || (corrId == "" && cmd == SENT 5)
+            || (corrId == "" && cmd == SENT 6)
         _ -> False
       bob <# ("", "alice", UP)
       bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False
-      bob #: ("12", "alice", "ACK 5") #> ("12", "alice", OK)
+      bob #: ("12", "alice", "ACK 6") #> ("12", "alice", OK)
 
   removeFile testStoreLogFile
-  -- removeFile testDB
+  removeFile testDB
   where
     withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
     withAgent = withSmpAgentThreadOn_ (ATransport t) (agentTestPort, testPort, testDB) (pure ()) . const . testSMPAgentClientOn agentTestPort
@@ -356,11 +393,11 @@ testConcurrentMsgDelivery _ alice bob = do
   -- alice <# ("", "bob", SENT 1)
   -- bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
   -- bob #: ("12", "alice", "ACK 1") #> ("12", "alice", OK)
-  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 5)
-  bob <# ("", "alice", SENT 5)
+  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 6)
+  bob <# ("", "alice", SENT 6)
   -- if delivery is blocked it won't go further
   alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
-  alice #: ("3", "bob", "ACK 5") #> ("3", "bob", OK)
+  alice #: ("3", "bob", "ACK 6") #> ("3", "bob", OK)
 
 testMsgDeliveryQuotaExceeded :: Transport c => TProxy c -> c -> c -> IO ()
 testMsgDeliveryQuotaExceeded _ alice bob = do
@@ -373,9 +410,9 @@ testMsgDeliveryQuotaExceeded _ alice bob = do
     alice <#= \case ("", "bob", SENT m) -> m == mId; _ -> False
   (_, "bob", Right (MID _)) <- alice #: ("5", "bob", "SEND :over quota")
 
-  alice #: ("1", "bob2", "SEND :hello") #> ("1", "bob2", MID 4)
+  alice #: ("1", "bob2", "SEND :hello") #> ("1", "bob2", MID 5)
   -- if delivery is blocked it won't go further
-  alice <# ("", "bob2", SENT 4)
+  alice <# ("", "bob2", SENT 5)
 
 connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO ()
 connect (h1, name1) (h2, name2) = do
@@ -424,7 +461,6 @@ syntaxTests t = do
       -- TODO: add tests with defined connection id
       it "with incorrect parameter" $ ("222", "", "NEW hi") >#> ("222", "", "ERR CMD SYNTAX")
 
-  -- focus this test to test postgres
   describe "JOIN" $ do
     describe "valid" $ do
       it "using same server as in invitation" $
@@ -432,7 +468,7 @@ syntaxTests t = do
           "a",
           "JOIN https://simpex.chat/invitation#/?smp=smp%3A%2F%2F"
             <> urlEncode True "LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI="
-            <> "%40localhost%3A5001%2F3456-w%3D%3D%23"
+            <> "%40127.0.0.1%3A5001%2F3456-w%3D%3D%23"
             <> urlEncode True sampleDhKey
             <> "&v=1"
             <> "&e2e=v%3D1%26x3dh%3DMEIwBQYDK2VvAzkAmKuSYeQ_m0SixPDS8Wq8VBaTS1cW-Lp0n0h4Diu-kUpR-qXx4SDJ32YGEFoGFGSbGPry5Ychr6U%3D%2CMEIwBQYDK2VvAzkAmKuSYeQ_m0SixPDS8Wq8VBaTS1cW-Lp0n0h4Diu-kUpR-qXx4SDJ32YGEFoGFGSbGPry5Ychr6U%3D"

tests/AgentTests/FunctionalAPITests.hs

@@ -10,9 +10,9 @@ module AgentTests.FunctionalAPITests (functionalAPITests) where
 import Control.Monad.Except (ExceptT, runExceptT)
 import Control.Monad.IO.Unlift
 import SMPAgentClient
-import SMPClient (withSmpServer)
+import SMPClient (testPort, withSmpServer, withSmpServerStoreLogOn)
 import Simplex.Messaging.Agent
-import Simplex.Messaging.Agent.Env.Postgres (AgentConfig (..))
+import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..))
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)
 import Simplex.Messaging.Transport (ATransport (..))
@@ -44,13 +44,15 @@ functionalAPITests t = do
       withSmpServer t testAsyncJoiningOfflineBeforeActivation
     it "should connect with both clients going offline" $
       withSmpServer t testAsyncBothOffline
+    it "should connect on the second attempt if server was offline" $
+      testAsyncServerOffline t
     it "should notify after HELLO timeout" $
       withSmpServer t testAsyncHelloTimeout
 
 testAgentClient :: IO ()
 testAgentClient = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbConnInfo = testDB2}
+  bob <- getSMPAgentClient cfg {dbFile = testDB2}
   Right () <- runExceptT $ do
     (bobId, qInfo) <- createConnection alice SCMInvitation
     aliceId <- joinConnection bob qInfo "bob's connInfo"
@@ -59,26 +61,26 @@ testAgentClient = do
     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
-    4 <- sendMessage alice bobId "hello"
-    get alice ##> ("", bobId, SENT 4)
-    5 <- sendMessage alice bobId "how are you?"
+    -- message IDs 1 to 4 get assigned to control messages, so first MSG is assigned ID 5
+    5 <- sendMessage alice bobId "hello"
     get alice ##> ("", bobId, SENT 5)
+    6 <- sendMessage alice bobId "how are you?"
+    get alice ##> ("", bobId, SENT 6)
     get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
-    ackMessage bob aliceId 4
-    get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
     ackMessage bob aliceId 5
-    6 <- sendMessage bob aliceId "hello too"
-    get bob ##> ("", aliceId, SENT 6)
-    7 <- sendMessage bob aliceId "message 1"
+    get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
+    ackMessage bob aliceId 6
+    7 <- sendMessage bob aliceId "hello too"
     get bob ##> ("", aliceId, SENT 7)
+    8 <- sendMessage bob aliceId "message 1"
+    get bob ##> ("", aliceId, SENT 8)
     get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
-    ackMessage alice bobId 6
-    get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
     ackMessage alice bobId 7
+    get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
+    ackMessage alice bobId 8
     suspendConnection alice bobId
-    8 <- sendMessage bob aliceId "message 2"
-    get bob ##> ("", aliceId, MERR 8 (SMP AUTH))
+    9 <- sendMessage bob aliceId "message 2"
+    get bob ##> ("", aliceId, MERR 9 (SMP AUTH))
     deleteConnection alice bobId
     liftIO $ noMessages alice "nothing else should be delivered to alice"
   pure ()
@@ -94,7 +96,7 @@ testAgentClient = do
 testAsyncInitiatingOffline :: IO ()
 testAsyncInitiatingOffline = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbConnInfo = testDB2}
+  bob <- getSMPAgentClient cfg {dbFile = testDB2}
   Right () <- runExceptT $ do
     (bobId, cReq) <- createConnection alice SCMInvitation
     disconnectAgentClient alice
@@ -112,14 +114,14 @@ testAsyncInitiatingOffline = do
 testAsyncJoiningOfflineBeforeActivation :: IO ()
 testAsyncJoiningOfflineBeforeActivation = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbConnInfo = testDB2}
+  bob <- getSMPAgentClient cfg {dbFile = testDB2}
   Right () <- runExceptT $ do
     (bobId, qInfo) <- createConnection alice SCMInvitation
     aliceId <- joinConnection bob qInfo "bob's connInfo"
     disconnectAgentClient bob
     ("", _, CONF confId "bob's connInfo") <- get alice
     allowConnection alice bobId confId "alice's connInfo"
-    bob' <- liftIO $ getSMPAgentClient cfg {dbConnInfo = testDB2}
+    bob' <- liftIO $ getSMPAgentClient cfg {dbFile = testDB2}
     subscribeConnection bob' aliceId
     get alice ##> ("", bobId, CON)
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -130,7 +132,7 @@ testAsyncJoiningOfflineBeforeActivation = do
 testAsyncBothOffline :: IO ()
 testAsyncBothOffline = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbConnInfo = testDB2}
+  bob <- getSMPAgentClient cfg {dbFile = testDB2}
   Right () <- runExceptT $ do
     (bobId, cReq) <- createConnection alice SCMInvitation
     disconnectAgentClient alice
@@ -140,7 +142,7 @@ testAsyncBothOffline = do
     subscribeConnection alice' bobId
     ("", _, CONF confId "bob's connInfo") <- get alice'
     allowConnection alice' bobId confId "alice's connInfo"
-    bob' <- liftIO $ getSMPAgentClient cfg {dbConnInfo = testDB2}
+    bob' <- liftIO $ getSMPAgentClient cfg {dbFile = testDB2}
     subscribeConnection bob' aliceId
     get alice' ##> ("", bobId, CON)
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -148,10 +150,34 @@ testAsyncBothOffline = do
     exchangeGreetings alice' bobId bob' aliceId
   pure ()
 
+testAsyncServerOffline :: ATransport -> IO ()
+testAsyncServerOffline t = do
+  alice <- getSMPAgentClient cfg
+  bob <- getSMPAgentClient cfg {dbFile = testDB2}
+  -- create connection and shutdown the server
+  Right (bobId, cReq) <- withSmpServerStoreLogOn t testPort $ \_ ->
+    runExceptT $ createConnection alice SCMInvitation
+  -- connection fails
+  Left (BROKER NETWORK) <- runExceptT $ joinConnection bob cReq "bob's connInfo"
+  ("", bobId1, DOWN) <- get alice
+  bobId1 `shouldBe` bobId
+  -- connection succeeds after server start
+  Right () <- withSmpServerStoreLogOn t testPort $ \_ -> runExceptT $ do
+    ("", bobId2, UP) <- get alice
+    liftIO $ bobId2 `shouldBe` bobId
+    aliceId <- joinConnection bob cReq "bob's connInfo"
+    ("", _, CONF confId "bob's connInfo") <- get alice
+    allowConnection alice bobId confId "alice's connInfo"
+    get alice ##> ("", bobId, CON)
+    get bob ##> ("", aliceId, INFO "alice's connInfo")
+    get bob ##> ("", aliceId, CON)
+    exchangeGreetings alice bobId bob aliceId
+  pure ()
+
 testAsyncHelloTimeout :: IO ()
 testAsyncHelloTimeout = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbConnInfo = testDB2, helloTimeout = 1}
+  bob <- getSMPAgentClient cfg {dbFile = testDB2, helloTimeout = 1}
   Right () <- runExceptT $ do
     (_, cReq) <- createConnection alice SCMInvitation
     disconnectAgentClient alice
@@ -161,11 +187,11 @@ testAsyncHelloTimeout = do
 
 exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
 exchangeGreetings alice bobId bob aliceId = do
-  4 <- sendMessage alice bobId "hello"
-  get alice ##> ("", bobId, SENT 4)
+  5 <- sendMessage alice bobId "hello"
+  get alice ##> ("", bobId, SENT 5)
   get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
-  ackMessage bob aliceId 4
-  5 <- sendMessage bob aliceId "hello too"
-  get bob ##> ("", aliceId, SENT 5)
+  ackMessage bob aliceId 5
+  6 <- sendMessage bob aliceId "hello too"
+  get bob ##> ("", aliceId, SENT 6)
   get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
-  ackMessage alice bobId 5
+  ackMessage alice bobId 6

tests/AgentTests/PostgresTests.hs

@@ -1,136 +0,0 @@
-{-# LANGUAGE DataKinds #-}
-{-# LANGUAGE DuplicateRecordFields #-}
-{-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE QuasiQuotes #-}
-{-# LANGUAGE RecordWildCards #-}
-
-module AgentTests.PostgresTests (postgresStoreTests) where
-
-import Control.Concurrent.Async (concurrently_)
-import Control.Concurrent.STM
-import Control.Monad (replicateM_)
-import Control.Monad.Except (ExceptT, runExceptT)
-import Crypto.Random (drgNew)
-import Data.ByteString.Char8 (ByteString)
-import qualified Data.Text as T
-import Data.Text.Encoding (encodeUtf8)
-import Data.Time
-import Data.Word (Word32)
-import Database.PostgreSQL.Simple (ConnectInfo (..), defaultConnectInfo)
-import qualified Database.PostgreSQL.Simple as DB
-import SMPClient (testKeyHash)
-import Simplex.Messaging.Agent.Client ()
-import Simplex.Messaging.Agent.Protocol
-import Simplex.Messaging.Agent.Store
-import Simplex.Messaging.Agent.Store.Postgres
-import qualified Simplex.Messaging.Agent.Store.Postgres.Migrations as Migrations
-import qualified Simplex.Messaging.Crypto as C
-import System.Random
-import Test.Hspec
-import UnliftIO.Directory (removeFile)
-
-withStore :: SpecWith PostgresStore -> Spec
-withStore = before createStore
-
-createStore :: IO PostgresStore
-createStore = do
-  let dbConnInfo = defaultConnectInfo {connectDatabase = "agent_poc_1"}
-  createPostgresStore dbConnInfo 1 Migrations.app
-
-returnsResult :: (Eq a, Eq e, Show a, Show e) => ExceptT e IO a -> a -> Expectation
-action `returnsResult` r = runExceptT action `shouldReturn` Right r
-
-throwsError :: (Eq a, Eq e, Show a, Show e) => ExceptT e IO a -> e -> Expectation
-action `throwsError` e = runExceptT action `shouldReturn` Left e
-
--- TODO add null port tests
-postgresStoreTests :: Spec
-postgresStoreTests = do
-  -- withStore2 $ do
-  --   describe "stress test" testConcurrentWrites
-  withStore $ do
-    -- describe "store setup" $ do
-    --   testCompiledThreadsafe
-    --   testForeignKeysEnabled
-    describe "store methods" $ do
-      describe "Queue and Connection management" $ do
-        -- describe "createRcvConn" $ do
-        --   testCreateRcvConn
-        --   testCreateRcvConnRandomId
-        --   testCreateRcvConnDuplicate
-        fdescribe "createSndConn" $ do
-          testCreateSndConn
-
---     testCreateSndConnRandomID
---     testCreateSndConnDuplicate
---   describe "getRcvConn" testGetRcvConn
---   describe "deleteConn" $ do
---     testDeleteRcvConn
---     testDeleteSndConn
---     testDeleteDuplexConn
---   describe "upgradeRcvConnToDuplex" $ do
---     testUpgradeRcvConnToDuplex
---   describe "upgradeSndConnToDuplex" $ do
---     testUpgradeSndConnToDuplex
---   describe "set Queue status" $ do
---     describe "setRcvQueueStatus" $ do
---       testSetRcvQueueStatus
---     describe "setSndQueueStatus" $ do
---       testSetSndQueueStatus
---     testSetQueueStatusDuplex
--- describe "Msg management" $ do
---   describe "create Msg" $ do
---     testCreateRcvMsg
---     testCreateSndMsg
---     testCreateRcvAndSndMsgs
-
-cData1 :: ConnData
-cData1 = ConnData {connId = "conn1"}
-
-testPrivateSignKey :: C.APrivateSignKey
-testPrivateSignKey = C.APrivateSignKey C.SEd25519 "MC4CAQAwBQYDK2VwBCIEIDfEfevydXXfKajz3sRkcQ7RPvfWUPoq6pu1TYHV1DEe"
-
-testPrivDhKey :: C.PrivateKeyX25519
-testPrivDhKey = "MC4CAQAwBQYDK2VuBCIEINCzbVFaCiYHoYncxNY8tSIfn0pXcIAhLBfFc0m+gOpk"
-
-testDhSecret :: C.DhSecretX25519
-testDhSecret = "01234567890123456789012345678901"
-
-rcvQueue1 :: RcvQueue
-rcvQueue1 =
-  RcvQueue
-    { server = SMPServer "smp.simplex.im" "5223" testKeyHash,
-      rcvId = "1234",
-      rcvPrivateKey = testPrivateSignKey,
-      rcvDhSecret = testDhSecret,
-      e2ePrivKey = testPrivDhKey,
-      e2eDhSecret = Nothing,
-      sndId = Just "2345",
-      status = New
-    }
-
-sndQueue1 :: SndQueue
-sndQueue1 =
-  SndQueue
-    { server = SMPServer "smp.simplex.im" "5223" testKeyHash,
-      sndId = "3456",
-      sndPrivateKey = testPrivateSignKey,
-      e2eDhSecret = testDhSecret,
-      status = New
-    }
-
-testCreateSndConn :: SpecWith PostgresStore
-testCreateSndConn =
-  it "should create SndConnection and add RcvQueue" $ \store -> do
-    g <- newTVarIO =<< drgNew
-    createSndConn store g cData1 sndQueue1
-      `returnsResult` "conn1"
-    getConn store "conn1"
-      `returnsResult` SomeConn SCSnd (SndConnection cData1 sndQueue1)
-
--- upgradeSndConnToDuplex store "conn1" rcvQueue1
---   `returnsResult` ()
--- getConn store "conn1"
---   `returnsResult` SomeConn SCDuplex (DuplexConnection cData1 rcvQueue1 sndQueue1)

tests/AgentTests/SQLiteTests.hs

@@ -51,7 +51,7 @@ createStore = do
   -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
   -- IO operations on multiple similarly named files; error seems to be environment specific
   r <- randomIO :: IO Word32
-  createSQLiteStore (testDB <> show r) 4 Migrations.app
+  createSQLiteStore (testDB <> show r) 4 Migrations.app True
 
 removeStore :: SQLiteStore -> IO ()
 removeStore store = do
@@ -173,7 +173,9 @@ sndQueue1 =
   SndQueue
     { server = SMPServer "smp.simplex.im" "5223" testKeyHash,
       sndId = "3456",
+      sndPublicKey = Nothing,
       sndPrivateKey = testPrivateSignKey,
+      e2ePubKey = Nothing,
       e2eDhSecret = testDhSecret,
       status = New
     }
@@ -303,7 +305,9 @@ testUpgradeRcvConnToDuplex =
           SndQueue
             { server = SMPServer "smp.simplex.im" "5223" testKeyHash,
               sndId = "2345",
+              sndPublicKey = Nothing,
               sndPrivateKey = testPrivateSignKey,
+              e2ePubKey = Nothing,
               e2eDhSecret = testDhSecret,
               status = New
             }
@@ -393,7 +397,7 @@ mkRcvMsgData internalId internalRcvId externalSndId brokerId internalHash =
             sndMsgId = externalSndId,
             broker = (brokerId, ts)
           },
-      msgType = A_MSG_,
+      msgType = AM_A_MSG_,
       msgBody = hw,
       internalHash,
       externalPrevSndHash = "hash_from_sender"
@@ -422,7 +426,7 @@ mkSndMsgData internalId internalSndId internalHash =
     { internalId,
       internalSndId,
       internalTs = ts,
-      msgType = A_MSG_,
+      msgType = AM_A_MSG_,
       msgBody = hw,
       internalHash,
       prevMsgHash = internalHash

tests/SMPAgentClient.hs

@@ -10,7 +10,6 @@ import Control.Monad.IO.Unlift
 import Crypto.Random
 import qualified Data.ByteString.Char8 as B
 import qualified Data.List.NonEmpty as L
-import Database.PostgreSQL.Simple (ConnectInfo (..), defaultConnectInfo)
 import Network.Socket (HostName, ServiceName)
 import SMPClient
   ( serverBracket,
@@ -21,7 +20,7 @@ import SMPClient
     withSmpServerOn,
     withSmpServerThreadOn,
   )
-import Simplex.Messaging.Agent.Env.Postgres
+import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Server (runSMPAgentBlocking)
@@ -44,23 +43,14 @@ agentTestPort2 = "5011"
 agentTestPort3 :: ServiceName
 agentTestPort3 = "5012"
 
--- testDB :: String
--- testDB = "tests/tmp/smp-agent.test.protocol.db"
+testDB :: String
+testDB = "tests/tmp/smp-agent.test.protocol.db"
 
-testDB :: ConnectInfo
-testDB = defaultConnectInfo {connectDatabase = "agent_poc_1"}
+testDB2 :: String
+testDB2 = "tests/tmp/smp-agent2.test.protocol.db"
 
--- testDB2 :: String
--- testDB2 = "tests/tmp/smp-agent2.test.protocol.db"
-
-testDB2 :: ConnectInfo
-testDB2 = defaultConnectInfo {connectDatabase = "agent_poc_2"}
-
--- testDB3 :: String
--- testDB3 = "tests/tmp/smp-agent3.test.protocol.db"
-
-testDB3 :: ConnectInfo
-testDB3 = defaultConnectInfo {connectDatabase = "agent_poc_3"}
+testDB3 :: String
+testDB3 = "tests/tmp/smp-agent3.test.protocol.db"
 
 smpAgentTest :: forall c. Transport c => TProxy c -> ARawTransmission -> IO ARawTransmission
 smpAgentTest _ cmd = runSmpAgentTest $ \(h :: c) -> tPutRaw h cmd >> tGetRaw h
@@ -81,10 +71,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) => [(ServiceName, ServiceName, ConnectInfo)] -> ([c] -> m a) -> m a
+runSmpAgentTestN :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => [(ServiceName, ServiceName, String)] -> ([c] -> m a) -> m a
 runSmpAgentTestN agents test = withSmpServer t $ run agents []
   where
-    run :: [(ServiceName, ServiceName, ConnectInfo)] -> [c] -> m a
+    run :: [(ServiceName, ServiceName, String)] -> [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
@@ -97,7 +87,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, ConnectInfo)] -> ([c] -> IO ()) -> Expectation
+smpAgentTestN :: Transport c => [(ServiceName, ServiceName, String)] -> ([c] -> IO ()) -> Expectation
 smpAgentTestN agents test' = runSmpAgentTestN agents test' `shouldReturn` ()
 
 smpAgentTestN_1 :: Transport c => Int -> ([c] -> IO ()) -> Expectation
@@ -167,9 +157,9 @@ cfg :: AgentConfig
 cfg =
   defaultAgentConfig
     { tcpPort = agentTestPort,
-      smpServers = L.fromList ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"],
+      initialSMPServers = L.fromList ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@127.0.0.1:5001"],
       tbqSize = 1,
-      dbConnInfo = testDB,
+      dbFile = testDB,
       smpCfg =
         smpDefaultConfig
           { qSize = 1,
@@ -182,17 +172,17 @@ cfg =
       certificateFile = "tests/fixtures/server.crt"
     }
 
-withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, ConnectInfo) -> m () -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m () -> (ThreadId -> m a) -> m a
 withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =
-  let cfg' = cfg {tcpPort = port', dbConnInfo = db', smpServers = L.fromList [SMPServer "localhost" smpPort' testKeyHash]}
+  let cfg' = cfg {tcpPort = port', dbFile = db', initialSMPServers = L.fromList [SMPServer "127.0.0.1" smpPort' testKeyHash]}
    in serverBracket
         (\started -> runSMPAgentBlocking t started cfg')
         afterProcess
 
-withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, ConnectInfo) -> (ThreadId -> m a) -> m a
-withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ pure () -- $ removeFile db'
+withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ removeFile db'
 
-withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, ConnectInfo) -> m a -> m a
+withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> 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

tests/SMPClient.hs

@@ -29,7 +29,7 @@ import UnliftIO.STM (TMVar, atomically, newEmptyTMVarIO, takeTMVar)
 import UnliftIO.Timeout (timeout)
 
 testHost :: HostName
-testHost = "localhost"
+testHost = "127.0.0.1"
 
 testPort :: ServiceName
 testPort = "5001"