wip

* use aeson fork with the option to encode/decode nullary constructors as empty objects

* update dependencies

.github/changelog_conf.json

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

.github/workflows/build.yml

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

apps/smp-agent/Main.hs

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

apps/smp-server/Main.hs

@@ -22,12 +22,13 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Server (runSMPServer)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.StoreLog (StoreLog, openReadStoreLog, storeLogFilePath)
-import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..), loadFingerprint, simplexMQVersion)
+import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..), simplexMQVersion)
+import Simplex.Messaging.Transport.Server (loadFingerprint)
 import Simplex.Messaging.Transport.WebSockets (WS)
 import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist, removeDirectoryRecursive)
 import System.Exit (exitFailure)
 import System.FilePath (combine)
-import System.IO (BufferMode (..), IOMode (..), hGetLine, withFile, hSetBuffering, stderr, stdout)
+import System.IO (BufferMode (..), IOMode (..), hGetLine, hSetBuffering, stderr, stdout, withFile)
 import System.Process (readCreateProcess, shell)
 import Text.Read (readMaybe)
 

cabal.project

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

package.yaml

@@ -22,12 +22,12 @@ extra-source-files:
   - CHANGELOG.md
 
 dependencies:
-  - aeson == 1.5.*
+  - aeson == 2.0.*
   - ansi-terminal >= 0.10 && < 0.12
   - asn1-encoding == 0.9.*
   - asn1-types == 0.3.*
   - async == 2.2.*
-  - attoparsec == 0.13.*
+  - attoparsec == 0.14.*
   - base >= 4.7 && < 5
   - base64-bytestring >= 1.0 && < 1.3
   - bytestring == 0.10.*
@@ -47,6 +47,7 @@ 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-linode.sh

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

simplexmq.cabal

@@ -31,11 +31,16 @@ 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
@@ -54,6 +59,8 @@ library
       Simplex.Messaging.Server.QueueStore.STM
       Simplex.Messaging.Server.StoreLog
       Simplex.Messaging.Transport
+      Simplex.Messaging.Transport.Client
+      Simplex.Messaging.Transport.Server
       Simplex.Messaging.Transport.WebSockets
       Simplex.Messaging.Util
       Simplex.Messaging.Version
@@ -64,12 +71,12 @@ library
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -89,6 +96,7 @@ 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.*
@@ -115,12 +123,12 @@ executable smp-agent
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -140,6 +148,7 @@ 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
@@ -167,12 +176,12 @@ executable smp-server
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -194,6 +203,7 @@ 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.*
@@ -235,12 +245,12 @@ test-suite smp-server-test
   build-depends:
       HUnit ==1.6.*
     , QuickCheck ==2.14.*
-    , aeson ==1.5.*
+    , aeson ==2.0.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
-    , attoparsec ==0.13.*
+    , attoparsec ==0.14.*
     , base >=4.7 && <5
     , base64-bytestring >=1.0 && <1.3
     , bytestring ==0.10.*
@@ -262,6 +272,7 @@ 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

@@ -26,11 +26,7 @@
 --
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md
 module Simplex.Messaging.Agent
-  ( -- * SMP agent over TCP
-    runSMPAgent,
-    runSMPAgentBlocking,
-
-    -- * queue-based SMP agent
+  ( -- * queue-based SMP agent
     getAgentClient,
     runAgentClient,
 
@@ -51,6 +47,7 @@ module Simplex.Messaging.Agent
     ackMessage,
     suspendConnection,
     deleteConnection,
+    logConnection,
   )
 where
 
@@ -62,7 +59,6 @@ import Control.Monad.Reader
 import Crypto.Random (MonadRandom)
 import Data.Bifunctor (first, second)
 import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
 import Data.Composition ((.:), (.:.))
 import Data.Functor (($>))
 import Data.List.NonEmpty (NonEmpty (..))
@@ -70,16 +66,15 @@ import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
 import Data.Maybe (isJust)
 import qualified Data.Text as T
-import Data.Text.Encoding (decodeUtf8)
 import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
 import Database.SQLite.Simple (SQLError)
 import Simplex.Messaging.Agent.Client
-import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Agent.Env.Postgres
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
-import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
+import Simplex.Messaging.Agent.Store.Postgres (PostgresStore)
 import Simplex.Messaging.Client (SMPServerTransmission)
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
@@ -87,7 +82,6 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (parse)
 import Simplex.Messaging.Protocol (MsgBody)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), loadTLSServerParams, runTransportServer, simplexMQVersion)
 import Simplex.Messaging.Util (bshow, liftError, tryError, unlessM)
 import Simplex.Messaging.Version
 import System.Random (randomR)
@@ -95,33 +89,6 @@ import UnliftIO.Async (async, race_)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
--- | Runs an SMP agent as a TCP service using passed configuration.
---
--- See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs
-runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> m ()
-runSMPAgent t cfg = do
-  started <- newEmptyTMVarIO
-  runSMPAgentBlocking t started cfg
-
--- | Runs an SMP agent as a TCP service using passed configuration with signalling.
---
--- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
--- and when it is disconnected from the TCP socket once the server thread is killed (False).
-runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> TMVar Bool -> AgentConfig -> m ()
-runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} = do
-  runReaderT (smpAgent t) =<< newSMPAgentEnv cfg
-  where
-    smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()
-    smpAgent _ = do
-      -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
-      tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
-      runTransportServer started tcpPort tlsServerParams $ \(h :: c) -> do
-        liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
-        c <- getAgentClient
-        logConnection c True
-        race_ (connectClient h c) (runAgentClient c)
-          `E.finally` disconnectAgentClient c
-
 -- | Creates an SMP agent client instance
 getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> m AgentClient
 getSMPAgentClient cfg = newSMPAgentEnv cfg >>= runReaderT runAgent
@@ -186,9 +153,6 @@ withAgentEnv c = (`runReaderT` agentEnv c)
 getAgentClient :: (MonadUnliftIO m, MonadReader Env m) => m AgentClient
 getAgentClient = ask >>= atomically . newAgentClient
 
-connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m ()
-connectClient h c = race_ (send h c) (receive h c)
-
 logConnection :: MonadUnliftIO m => AgentClient -> Bool -> m ()
 logConnection c connected =
   let event = if connected then "connected to" else "disconnected from"
@@ -198,28 +162,6 @@ logConnection c connected =
 runAgentClient :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
 runAgentClient c = race_ (subscriber c) (client c)
 
-receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
-receive h c@AgentClient {rcvQ, subQ} = forever $ do
-  (corrId, connId, cmdOrErr) <- tGet SClient h
-  case cmdOrErr of
-    Right cmd -> write rcvQ (corrId, connId, cmd)
-    Left e -> write subQ (corrId, connId, ERR e)
-  where
-    write :: TBQueue (ATransmission p) -> ATransmission p -> m ()
-    write q t = do
-      logClient c "-->" t
-      atomically $ writeTBQueue q t
-
-send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
-send h c@AgentClient {subQ} = forever $ do
-  t <- atomically $ readTBQueue subQ
-  tPut h t
-  logClient c "<--" t
-
-logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()
-logClient AgentClient {clientId} dir (corrId, connId, cmd) = do
-  logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]
-
 client :: forall m. (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
 client c@AgentClient {rcvQ, subQ} = forever $ do
   (corrId, connId, cmd) <- atomically $ readTBQueue rcvQ
@@ -230,18 +172,22 @@ client c@AgentClient {rcvQ, subQ} = forever $ do
 
 withStore ::
   AgentMonad m =>
-  (forall m'. (MonadUnliftIO m', MonadError StoreError m') => SQLiteStore -> m' a) ->
+  (forall m'. (MonadUnliftIO m', MonadError StoreError m') => PostgresStore -> m' a) ->
   m a
 withStore action = do
   st <- asks store
   runExceptT (action st `E.catch` handleInternal) >>= \case
     Right c -> return c
-    Left e -> throwError $ storeError e
+    Left e -> do
+      liftIO $ print e
+      throwError $ storeError e
   where
     -- TODO when parsing exception happens in store, the agent hangs;
     -- changing SQLError to SomeException does not help
-    handleInternal :: (MonadError StoreError m') => SQLError -> m' a
-    handleInternal e = throwError . SEInternal $ bshow e
+    handleInternal :: (MonadUnliftIO m', MonadError StoreError m') => SQLError -> m' a
+    handleInternal e = do
+      liftIO $ print e
+      throwError . SEInternal $ bshow e
     storeError :: StoreError -> AgentErrorType
     storeError = \case
       SEConnNotFound -> CONN NOT_FOUND
@@ -296,8 +242,11 @@ joinConn c connId (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2
       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
@@ -679,9 +628,12 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
                 Nothing -> notify . ERR $ AGENT A_VERSION
                 Just qInfo' -> do
                   (sq, smpConf) <- newSndQueue qInfo' ownConnInfo
+                  liftIO $ print "before: upgradeRcvConnToDuplex st connId sq"
                   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
             _ -> prohibited
 

src/Simplex/Messaging/Agent/Client.hs

@@ -53,7 +53,7 @@ import Data.Maybe (isNothing)
 import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text.Encoding
-import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Agent.Env.Postgres
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store

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

@@ -0,0 +1,88 @@
+{-# 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/Protocol.hs

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

src/Simplex/Messaging/Agent/Server.hs

@@ -0,0 +1,81 @@
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Agent.Server
+  ( -- * SMP agent over TCP
+    runSMPAgent,
+    runSMPAgentBlocking,
+  )
+where
+
+import Control.Logger.Simple (logInfo)
+import Control.Monad.Except
+import Control.Monad.IO.Unlift (MonadUnliftIO)
+import Control.Monad.Reader
+import Crypto.Random (MonadRandom)
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import Data.Text.Encoding (decodeUtf8)
+import Simplex.Messaging.Agent
+import Simplex.Messaging.Agent.Env.Postgres
+import Simplex.Messaging.Agent.Protocol
+import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), simplexMQVersion)
+import Simplex.Messaging.Transport.Server (loadTLSServerParams, runTransportServer)
+import Simplex.Messaging.Util (bshow)
+import UnliftIO.Async (race_)
+import qualified UnliftIO.Exception as E
+import UnliftIO.STM
+
+-- | Runs an SMP agent as a TCP service using passed configuration.
+--
+-- See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs
+runSMPAgent :: (MonadRandom m, MonadUnliftIO m) => ATransport -> AgentConfig -> m ()
+runSMPAgent t cfg = do
+  started <- newEmptyTMVarIO
+  runSMPAgentBlocking t started cfg
+
+-- | Runs an SMP agent as a TCP service using passed configuration with signalling.
+--
+-- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
+-- and when it is disconnected from the TCP socket once the server thread is killed (False).
+runSMPAgentBlocking :: (MonadRandom m, MonadUnliftIO m) => ATransport -> TMVar Bool -> AgentConfig -> m ()
+runSMPAgentBlocking (ATransport t) started cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} = do
+  runReaderT (smpAgent t) =<< newSMPAgentEnv cfg
+  where
+    smpAgent :: forall c m'. (Transport c, MonadUnliftIO m', MonadReader Env m') => TProxy c -> m' ()
+    smpAgent _ = do
+      -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
+      tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
+      runTransportServer started tcpPort tlsServerParams $ \(h :: c) -> do
+        liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
+        c <- getAgentClient
+        logConnection c True
+        race_ (connectClient h c) (runAgentClient c)
+          `E.finally` disconnectAgentClient c
+
+connectClient :: Transport c => MonadUnliftIO m => c -> AgentClient -> m ()
+connectClient h c = race_ (send h c) (receive h c)
+
+receive :: forall c m. (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
+receive h c@AgentClient {rcvQ, subQ} = forever $ do
+  (corrId, connId, cmdOrErr) <- tGet SClient h
+  case cmdOrErr of
+    Right cmd -> write rcvQ (corrId, connId, cmd)
+    Left e -> write subQ (corrId, connId, ERR e)
+  where
+    write :: TBQueue (ATransmission p) -> ATransmission p -> m ()
+    write q t = do
+      logClient c "-->" t
+      atomically $ writeTBQueue q t
+
+send :: (Transport c, MonadUnliftIO m) => c -> AgentClient -> m ()
+send h c@AgentClient {subQ} = forever $ do
+  t <- atomically $ readTBQueue subQ
+  tPut h t
+  logClient c "<--" t
+
+logClient :: MonadUnliftIO m => AgentClient -> ByteString -> ATransmission a -> m ()
+logClient AgentClient {clientId} dir (corrId, connId, cmd) = do
+  logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]

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

@@ -0,0 +1,957 @@
+{-# 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

@@ -0,0 +1,73 @@
+{-# 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

@@ -0,0 +1,158 @@
+{-# 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

@@ -0,0 +1,9 @@
+# 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

@@ -569,9 +569,9 @@ instance ToField AMsgType where toField = toField . smpEncode
 
 instance FromField AMsgType where fromField = blobFieldParser smpP
 
-instance ToField MsgIntegrity where toField = toField . serializeMsgIntegrity
+instance ToField MsgIntegrity where toField = toField . strEncode
 
-instance FromField MsgIntegrity where fromField = blobFieldParser msgIntegrityP
+instance FromField MsgIntegrity where fromField = blobFieldParser strP
 
 instance ToField SMPQueueUri where toField = toField . strEncode
 

src/Simplex/Messaging/Client.hs

@@ -63,7 +63,8 @@ import Network.Socket (ServiceName)
 import Numeric.Natural
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake, runTransportClient)
+import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake)
+import Simplex.Messaging.Transport.Client (runTransportClient)
 import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (bshow, liftError, raceAny_)
 import System.Timeout (timeout)

src/Simplex/Messaging/Crypto.hs

@@ -149,14 +149,20 @@ import Data.String
 import Data.Type.Equality
 import Data.Typeable (Typeable)
 import Data.X509
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
+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 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
@@ -540,33 +546,62 @@ generateKeyPair' = case sAlgorithm @a of
       let k = X448.toPublic pk
        in pure (PublicKeyX448 k, PrivateKeyX448 pk k)
 
-instance ToField APrivateSignKey where toField = toField . encodePrivKey
+instance ST.ToField APrivateSignKey where toField = ST.toField . encodePrivKey
 
-instance ToField APublicVerifyKey where toField = toField . encodePubKey
+instance ST.ToField APublicVerifyKey where toField = ST.toField . encodePubKey
 
-instance ToField APrivateDhKey where toField = toField . encodePrivKey
+instance ST.ToField APrivateDhKey where toField = ST.toField . encodePrivKey
 
-instance ToField APublicDhKey where toField = toField . encodePubKey
+instance ST.ToField APublicDhKey where toField = ST.toField . encodePubKey
 
-instance AlgorithmI a => ToField (PrivateKey a) where toField = toField . encodePrivKey
+instance AlgorithmI a => ST.ToField (PrivateKey a) where toField = ST.toField . encodePrivKey
 
-instance AlgorithmI a => ToField (PublicKey a) where toField = toField . encodePubKey
+instance AlgorithmI a => ST.ToField (PublicKey a) where toField = ST.toField . encodePubKey
 
-instance ToField (DhSecret a) where toField = toField . dhBytes'
+instance ST.ToField (DhSecret a) where toField = ST.toField . dhBytes'
 
-instance FromField APrivateSignKey where fromField = blobFieldDecoder decodePrivKey
+instance SF.FromField APrivateSignKey where fromField = blobFieldDecoder decodePrivKey
 
-instance FromField APublicVerifyKey where fromField = blobFieldDecoder decodePubKey
+instance SF.FromField APublicVerifyKey where fromField = blobFieldDecoder decodePubKey
 
-instance FromField APrivateDhKey where fromField = blobFieldDecoder decodePrivKey
+instance SF.FromField APrivateDhKey where fromField = blobFieldDecoder decodePrivKey
 
-instance FromField APublicDhKey where fromField = blobFieldDecoder decodePubKey
+instance SF.FromField APublicDhKey where fromField = blobFieldDecoder decodePubKey
 
-instance (Typeable a, AlgorithmI a) => FromField (PrivateKey a) where fromField = blobFieldDecoder decodePrivKey
+instance (Typeable a, AlgorithmI a) => SF.FromField (PrivateKey a) where fromField = blobFieldDecoder decodePrivKey
 
-instance (Typeable a, AlgorithmI a) => FromField (PublicKey a) where fromField = blobFieldDecoder decodePubKey
+instance (Typeable a, AlgorithmI a) => SF.FromField (PublicKey a) where fromField = blobFieldDecoder decodePubKey
 
-instance (Typeable a, AlgorithmI a) => FromField (DhSecret a) where fromField = blobFieldDecoder strDecode
+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 IsString (Maybe ASignature) where
   fromString = parseString $ decode >=> decodeSignature
@@ -690,9 +725,13 @@ validSignatureSize n =
 newtype Key = Key {unKey :: ByteString}
   deriving (Eq, Ord, Show)
 
-instance ToField Key where toField = toField . unKey
+instance ST.ToField Key where toField = ST.toField . unKey
 
-instance FromField Key where fromField f = Key <$> fromField f
+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 ToJSON Key where
   toJSON = strToJSON . unKey
@@ -730,9 +769,27 @@ instance StrEncoding KeyHash where
 instance IsString KeyHash where
   fromString = parseString $ parseAll strP
 
-instance ToField KeyHash where toField = toField . strEncode
+instance ST.ToField KeyHash where toField = ST.toField . strEncode
 
-instance FromField KeyHash where fromField = blobFieldDecoder $ parseAll strP
+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)
 
 -- | SHA256 digest.
 sha256Hash :: ByteString -> ByteString

src/Simplex/Messaging/Crypto/Ratchet.hs

@@ -30,8 +30,12 @@ import qualified Data.Map.Strict as M
 import Data.Maybe (fromMaybe)
 import Data.Typeable (Typeable)
 import Data.Word (Word32)
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
+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 GHC.Generics
 import Simplex.Messaging.Agent.QueryString
 import Simplex.Messaging.Crypto
@@ -197,13 +201,32 @@ instance ToJSON RatchetKey where
 instance FromJSON RatchetKey where
   parseJSON = fmap RatchetKey . strParseJSON "Key"
 
-instance AlgorithmI a => ToField (Ratchet a) where toField = toField . LB.toStrict . J.encode
+instance AlgorithmI a => ST.ToField (Ratchet a) where toField = ST.toField . LB.toStrict . J.encode
 
-instance (AlgorithmI a, Typeable a) => FromField (Ratchet a) where fromField = blobFieldDecoder J.eitherDecodeStrict'
+instance AlgorithmI a => PT.ToField (Ratchet a) where toField = PT.toField . LB.toStrict . J.encode
 
-instance ToField MessageKey where toField = toField . smpEncode
+instance (AlgorithmI a, Typeable a) => PF.FromField (Ratchet a) where fromField = fromByteStringField J.eitherDecodeStrict'
 
-instance FromField MessageKey where fromField = blobFieldDecoder smpDecode
+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)
 
 -- | Sending ratchet initialization, equivalent to RatchetInitAliceHE in double ratchet spec
 --

src/Simplex/Messaging/Parsers.hs

@@ -4,20 +4,24 @@
 module Simplex.Messaging.Parsers where
 
 import Control.Monad.Trans.Except
+import qualified Data.Aeson as J
 import Data.Attoparsec.ByteString.Char8 (Parser)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Char (isAlphaNum)
+import Data.Char (isAlphaNum, toLower)
 import Data.Time.Clock (UTCTime)
 import Data.Time.ISO8601 (parseISO8601)
 import Data.Typeable (Typeable)
+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 Database.SQLite.Simple.FromField (FieldParser, returnError)
-import Database.SQLite.Simple.Internal (Field (..))
-import Database.SQLite.Simple.Ok (Ok (Ok))
+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 Simplex.Messaging.Util ((<$?>))
 import Text.Read (readMaybe)
 
@@ -68,13 +72,58 @@ 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 -> FieldParser k
+blobFieldParser :: Typeable k => Parser k -> SF.FieldParser k
 blobFieldParser = blobFieldDecoder . parseAll
 
-blobFieldDecoder :: Typeable k => (ByteString -> Either String k) -> FieldParser k
+blobFieldDecoder :: Typeable k => (ByteString -> Either String k) -> SF.FieldParser k
 blobFieldDecoder dec = \case
-  f@(Field (SQLBlob b) _) ->
+  f@(SI.Field (SQLBlob b) _) ->
     case dec b of
-      Right k -> Ok k
-      Left e -> returnError ConversionFailed f ("couldn't parse field: " ++ e)
-  f -> returnError ConversionFailed f "expecting SQLBlob column type"
+      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"
+
+fstToLower :: String -> String
+fstToLower "" = ""
+fstToLower (h : t) = toLower h : t
+
+dropPrefix :: String -> String -> String
+dropPrefix pfx s =
+  let (p, rest) = splitAt (length pfx) s
+   in fstToLower $ if p == pfx then rest else s
+
+enumJSON :: (String -> String) -> J.Options
+enumJSON tagModifier =
+  J.defaultOptions
+    { J.constructorTagModifier = tagModifier,
+      J.allNullaryToStringTag = True
+    }
+
+sumTypeJSON :: (String -> String) -> J.Options
+sumTypeJSON = singleFieldJSON
+
+taggedObjectJSON :: (String -> String) -> J.Options
+taggedObjectJSON tagModifier =
+  J.defaultOptions
+    { J.sumEncoding = J.TaggedObject "type" "data",
+      J.constructorTagModifier = tagModifier,
+      J.nullaryToObject = True,
+      J.omitNothingFields = True
+    }
+
+singleFieldJSON :: (String -> String) -> J.Options
+singleFieldJSON tagModifier =
+  J.defaultOptions
+    { J.sumEncoding = J.ObjectWithSingleField,
+      J.constructorTagModifier = tagModifier,
+      J.nullaryToObject = True,
+      J.omitNothingFields = True
+    }

src/Simplex/Messaging/Protocol.hs

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

src/Simplex/Messaging/Server.hs

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

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

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

src/Simplex/Messaging/Transport.hs

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

src/Simplex/Messaging/Transport/Client.hs

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

src/Simplex/Messaging/Transport/Server.hs

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

stack.yaml

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

tests/AgentTests.hs

@@ -13,6 +13,7 @@ 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)
@@ -36,6 +37,7 @@ 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" $
@@ -329,7 +331,7 @@ testMsgDeliveryAgentRestart t bob = do
       bob #: ("12", "alice", "ACK 5") #> ("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
@@ -391,10 +393,10 @@ sendMessage (h1, name1) (h2, name2) msg = do
   ("m1", name2', Right (MID mId)) <- h1 #: ("m1", name2, "SEND :" <> msg)
   name2' `shouldBe` name2
   h1 <#= \case ("", n, SENT m) -> n == name2 && m == mId; _ -> False
-  ("", name1', Right (MSG MsgMeta {recipient = (msgId, _)} msg')) <- (h2 <#:)
+  ("", name1', Right (MSG MsgMeta {recipient = (msgId', _)} msg')) <- (h2 <#:)
   name1' `shouldBe` name1
   msg' `shouldBe` msg
-  h2 #: ("m2", name1, "ACK " <> bshow msgId) =#> \case ("m2", n, OK) -> n == name1; _ -> False
+  h2 #: ("m2", name1, "ACK " <> bshow msgId') =#> \case ("m2", n, OK) -> n == name1; _ -> False
 
 -- connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)
 -- connect' h1 h2 = do
@@ -422,6 +424,7 @@ 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" $

tests/AgentTests/FunctionalAPITests.hs

@@ -12,7 +12,7 @@ import Control.Monad.IO.Unlift
 import SMPAgentClient
 import SMPClient (withSmpServer)
 import Simplex.Messaging.Agent
-import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..))
+import Simplex.Messaging.Agent.Env.Postgres (AgentConfig (..))
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)
 import Simplex.Messaging.Transport (ATransport (..))
@@ -50,7 +50,7 @@ functionalAPITests t = do
 testAgentClient :: IO ()
 testAgentClient = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbFile = testDB2}
+  bob <- getSMPAgentClient cfg {dbConnInfo = testDB2}
   Right () <- runExceptT $ do
     (bobId, qInfo) <- createConnection alice SCMInvitation
     aliceId <- joinConnection bob qInfo "bob's connInfo"
@@ -94,7 +94,7 @@ testAgentClient = do
 testAsyncInitiatingOffline :: IO ()
 testAsyncInitiatingOffline = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbFile = testDB2}
+  bob <- getSMPAgentClient cfg {dbConnInfo = testDB2}
   Right () <- runExceptT $ do
     (bobId, cReq) <- createConnection alice SCMInvitation
     disconnectAgentClient alice
@@ -112,14 +112,14 @@ testAsyncInitiatingOffline = do
 testAsyncJoiningOfflineBeforeActivation :: IO ()
 testAsyncJoiningOfflineBeforeActivation = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbFile = testDB2}
+  bob <- getSMPAgentClient cfg {dbConnInfo = 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 {dbFile = testDB2}
+    bob' <- liftIO $ getSMPAgentClient cfg {dbConnInfo = testDB2}
     subscribeConnection bob' aliceId
     get alice ##> ("", bobId, CON)
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -130,7 +130,7 @@ testAsyncJoiningOfflineBeforeActivation = do
 testAsyncBothOffline :: IO ()
 testAsyncBothOffline = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbFile = testDB2}
+  bob <- getSMPAgentClient cfg {dbConnInfo = testDB2}
   Right () <- runExceptT $ do
     (bobId, cReq) <- createConnection alice SCMInvitation
     disconnectAgentClient alice
@@ -140,7 +140,7 @@ testAsyncBothOffline = do
     subscribeConnection alice' bobId
     ("", _, CONF confId "bob's connInfo") <- get alice'
     allowConnection alice' bobId confId "alice's connInfo"
-    bob' <- liftIO $ getSMPAgentClient cfg {dbFile = testDB2}
+    bob' <- liftIO $ getSMPAgentClient cfg {dbConnInfo = testDB2}
     subscribeConnection bob' aliceId
     get alice' ##> ("", bobId, CON)
     get bob' ##> ("", aliceId, INFO "alice's connInfo")
@@ -151,7 +151,7 @@ testAsyncBothOffline = do
 testAsyncHelloTimeout :: IO ()
 testAsyncHelloTimeout = do
   alice <- getSMPAgentClient cfg
-  bob <- getSMPAgentClient cfg {dbFile = testDB2, helloTimeout = 1}
+  bob <- getSMPAgentClient cfg {dbConnInfo = testDB2, helloTimeout = 1}
   Right () <- runExceptT $ do
     (_, cReq) <- createConnection alice SCMInvitation
     disconnectAgentClient alice

tests/AgentTests/PostgresTests.hs

@@ -0,0 +1,136 @@
+{-# 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/CoreTests/ProtocolErrorTests.hs

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

tests/SMPAgentClient.hs

@@ -10,6 +10,7 @@ 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,
@@ -20,12 +21,13 @@ import SMPClient
     withSmpServerOn,
     withSmpServerThreadOn,
   )
-import Simplex.Messaging.Agent (runSMPAgentBlocking)
-import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Agent.Env.Postgres
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
+import Simplex.Messaging.Agent.Server (runSMPAgentBlocking)
 import Simplex.Messaging.Client (SMPClientConfig (..), smpDefaultConfig)
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Client
 import Test.Hspec
 import UnliftIO.Concurrent
 import UnliftIO.Directory
@@ -42,14 +44,23 @@ 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"
 
-testDB2 :: String
-testDB2 = "tests/tmp/smp-agent2.test.protocol.db"
+testDB :: ConnectInfo
+testDB = defaultConnectInfo {connectDatabase = "agent_poc_1"}
 
-testDB3 :: String
-testDB3 = "tests/tmp/smp-agent3.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"}
 
 smpAgentTest :: forall c. Transport c => TProxy c -> ARawTransmission -> IO ARawTransmission
 smpAgentTest _ cmd = runSmpAgentTest $ \(h :: c) -> tPutRaw h cmd >> tGetRaw h
@@ -70,10 +81,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, String)] -> ([c] -> m a) -> m a
+runSmpAgentTestN :: forall c m a. (Transport c, MonadUnliftIO m, MonadRandom m) => [(ServiceName, ServiceName, ConnectInfo)] -> ([c] -> m a) -> m a
 runSmpAgentTestN agents test = withSmpServer t $ run agents []
   where
-    run :: [(ServiceName, ServiceName, String)] -> [c] -> m a
+    run :: [(ServiceName, ServiceName, ConnectInfo)] -> [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
@@ -86,7 +97,7 @@ runSmpAgentTestN_1 nClients test = withSmpServer t . withSmpAgent t $ run nClien
     run n hs = testSMPAgentClient $ \h -> run (n - 1) (h : hs)
     t = transport @c
 
-smpAgentTestN :: Transport c => [(ServiceName, ServiceName, String)] -> ([c] -> IO ()) -> Expectation
+smpAgentTestN :: Transport c => [(ServiceName, ServiceName, ConnectInfo)] -> ([c] -> IO ()) -> Expectation
 smpAgentTestN agents test' = runSmpAgentTestN agents test' `shouldReturn` ()
 
 smpAgentTestN_1 :: Transport c => Int -> ([c] -> IO ()) -> Expectation
@@ -158,7 +169,7 @@ cfg =
     { tcpPort = agentTestPort,
       smpServers = L.fromList ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"],
       tbqSize = 1,
-      dbFile = testDB,
+      dbConnInfo = testDB,
       smpCfg =
         smpDefaultConfig
           { qSize = 1,
@@ -171,17 +182,17 @@ cfg =
       certificateFile = "tests/fixtures/server.crt"
     }
 
-withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m () -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, ConnectInfo) -> m () -> (ThreadId -> m a) -> m a
 withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =
-  let cfg' = cfg {tcpPort = port', dbFile = db', smpServers = L.fromList [SMPServer "localhost" smpPort' testKeyHash]}
+  let cfg' = cfg {tcpPort = port', dbConnInfo = db', smpServers = L.fromList [SMPServer "localhost" smpPort' testKeyHash]}
    in serverBracket
         (\started -> runSMPAgentBlocking t started cfg')
         afterProcess
 
-withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a
-withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ removeFile db'
+withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, ConnectInfo) -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ pure () -- $ removeFile db'
 
-withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m a -> m a
+withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, ConnectInfo) -> 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

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