wip

apps/smp-agent/Main.hs

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

package.yaml

@@ -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.*

simplexmq.cabal

@@ -96,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.*
@@ -147,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
@@ -201,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.*
@@ -269,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

@@ -70,11 +70,11 @@ import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
 import Database.SQLite.Simple (SQLError)
 import Simplex.Messaging.Agent.Client
-import Simplex.Messaging.Agent.Env.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
@@ -172,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
@@ -238,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
@@ -621,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

@@ -16,12 +16,13 @@ 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.SQLite
-import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+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)
@@ -33,7 +34,7 @@ data AgentConfig = AgentConfig
     cmdSignAlg :: C.SignAlg,
     connIdBytes :: Int,
     tbqSize :: Natural,
-    dbFile :: FilePath,
+    dbConnInfo :: ConnectInfo,
     dbPoolSize :: Int,
     smpCfg :: SMPClientConfig,
     reconnectInterval :: RetryInterval,
@@ -51,7 +52,7 @@ defaultAgentConfig =
       cmdSignAlg = C.SignAlg C.SEd448,
       connIdBytes = 12,
       tbqSize = 16,
-      dbFile = "smp-agent.db",
+      dbConnInfo = defaultConnectInfo {connectDatabase = "agent_poc_1"},
       dbPoolSize = 4,
       smpCfg = smpDefaultConfig,
       reconnectInterval =
@@ -72,16 +73,16 @@ defaultAgentConfig =
 
 data Env = Env
   { config :: AgentConfig,
-    store :: SQLiteStore,
+    store :: PostgresStore,
     idsDrg :: TVar ChaChaDRG,
     clientCounter :: TVar Int,
     randomServer :: TVar StdGen
   }
 
 newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env
-newSMPAgentEnv cfg = do
+newSMPAgentEnv cfg@AgentConfig {dbConnInfo, dbPoolSize} = do
   idsDrg <- newTVarIO =<< drgNew
-  store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app
+  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/Server.hs

@@ -19,7 +19,7 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Text.Encoding (decodeUtf8)
 import Simplex.Messaging.Agent
-import Simplex.Messaging.Agent.Env.SQLite
+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)

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

@@ -15,18 +15,19 @@ module Simplex.Messaging.Agent.Store.Postgres.Migrations
   )
 where
 
-import Control.Monad (forM_)
+import Control.Monad (forM_, void)
 import Data.Function (on)
 import Data.List (intercalate, sortBy)
-import Data.Text (Text)
 import Data.Time.Clock (getCurrentTime)
-import Database.SQLite.Simple (Connection, Only (..), Query (..))
-import qualified Database.SQLite.Simple as DB
-import Database.SQLite.Simple.QQ (sql)
-import qualified Database.SQLite3 as SQLite3
-import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial
+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 :: Text}
+data Migration = Migration {name :: String, up :: Query}
   deriving (Show)
 
 schemaMigrations :: [(String, Query)]
@@ -38,7 +39,7 @@ schemaMigrations =
 app :: [Migration]
 app = sortBy (compare `on` name) $ map migration schemaMigrations
   where
-    migration (name, query) = Migration {name = name, up = fromQuery query}
+    migration (name, query) = Migration {name, up = query}
 
 get :: Connection -> [Migration] -> IO (Either String [Migration])
 get conn migrations =
@@ -46,23 +47,23 @@ get conn migrations =
     <$> DB.query_ conn "SELECT name FROM migrations ORDER BY name ASC;"
 
 run :: Connection -> [Migration] -> IO ()
-run conn ms = DB.withImmediateTransaction conn . forM_ ms $
-  \Migration {name, up} -> insert name >> execSQL up
+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
-    execSQL = SQLite3.exec $ DB.connectionHandle conn
 
 initialize :: Connection -> IO ()
 initialize conn =
-  DB.execute_
-    conn
-    [sql|
-      CREATE TABLE IF NOT EXISTS migrations (
-        name TEXT NOT NULL,
-        ts TEXT NOT NULL,
-        PRIMARY KEY (name)
-      );
-    |]
+  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

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

@@ -2,42 +2,46 @@
 
 module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20220202_initial where
 
-import Database.SQLite.Simple (Query)
-import Database.SQLite.Simple.QQ (sql)
+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 BLOB NOT NULL,
+  key_hash BYTEA NOT NULL,
   PRIMARY KEY (host, port)
-) WITHOUT ROWID;
+);
 
 CREATE TABLE connections (
-  conn_id BLOB NOT NULL PRIMARY KEY,
+  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 BLOB NOT NULL DEFAULT x'',
-  last_snd_msg_hash BLOB NOT NULL DEFAULT x'',
+  last_rcv_msg_hash BYTEA NOT NULL DEFAULT '',
+  last_snd_msg_hash BYTEA NOT NULL DEFAULT '',
   smp_agent_version INTEGER NOT NULL DEFAULT 1
-) WITHOUT ROWID;
+);
 
 CREATE TABLE rcv_queues (
   host TEXT NOT NULL,
   port TEXT NOT NULL,
-  rcv_id BLOB NOT NULL,
-  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
-  rcv_private_key BLOB NOT NULL,
-  rcv_dh_secret BLOB NOT NULL,
-  e2e_priv_key BLOB NOT NULL,
-  e2e_dh_secret BLOB,
-  snd_id BLOB NOT NULL,
-  snd_key BLOB,
+  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,
@@ -45,104 +49,110 @@ CREATE TABLE rcv_queues (
   FOREIGN KEY (host, port) REFERENCES servers
     ON DELETE RESTRICT ON UPDATE CASCADE,
   UNIQUE (host, port, snd_id)
-) WITHOUT ROWID;
+);
 
 CREATE TABLE snd_queues (
   host TEXT NOT NULL,
   port TEXT NOT NULL,
-  snd_id BLOB NOT NULL,
-  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
-  snd_private_key BLOB NOT NULL,
-  e2e_dh_secret BLOB 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
-) WITHOUT ROWID;
+);
 
 CREATE TABLE messages (
-  conn_id BLOB NOT NULL REFERENCES connections (conn_id)
+  conn_id BYTEA NOT NULL REFERENCES connections (conn_id)
     ON DELETE CASCADE,
   internal_id INTEGER NOT NULL,
-  internal_ts TEXT NOT NULL,
+  internal_ts TIMESTAMP NOT NULL,
   internal_rcv_id INTEGER,
   internal_snd_id INTEGER,
-  msg_type BLOB NOT NULL, -- (H)ELLO, (R)EPLY, (D)ELETE. Should SMP confirmation be saved too?
-  msg_body BLOB NOT NULL DEFAULT x'',
-  PRIMARY KEY (conn_id, internal_id),
-  FOREIGN KEY (conn_id, internal_rcv_id) REFERENCES rcv_messages
-    ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
-  FOREIGN KEY (conn_id, internal_snd_id) REFERENCES snd_messages
-    ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED
-) WITHOUT ROWID;
+  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 BLOB NOT NULL,
+  conn_id BYTEA NOT NULL,
   internal_rcv_id INTEGER NOT NULL,
   internal_id INTEGER NOT NULL,
   external_snd_id INTEGER NOT NULL,
-  broker_id BLOB NOT NULL,
-  broker_ts TEXT NOT NULL,
-  internal_hash BLOB NOT NULL,
-  external_prev_snd_hash BLOB NOT NULL,
-  integrity BLOB 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
-) WITHOUT ROWID;
+);
+
+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 BLOB NOT NULL,
+  conn_id BYTEA NOT NULL,
   internal_snd_id INTEGER NOT NULL,
   internal_id INTEGER NOT NULL,
-  internal_hash BLOB NOT NULL,
-  previous_msg_hash BLOB NOT NULL DEFAULT x'',
+  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
-) WITHOUT ROWID;
+);
+
+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 BLOB NOT NULL PRIMARY KEY,
-  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
-  e2e_snd_pub_key BLOB NOT NULL, -- TODO per-queue key. Split?
-  sender_key BLOB NOT NULL, -- TODO per-queue key. Split?
-  ratchet_state BLOB NOT NULL,
-  sender_conn_info BLOB NOT NULL,
+  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 BLOB,
-  created_at TEXT NOT NULL DEFAULT (datetime('now'))
-) WITHOUT ROWID;
+  own_conn_info BYTEA,
+  created_at TIMESTAMP NOT NULL DEFAULT (now())
+);
 
 CREATE TABLE conn_invitations (
-  invitation_id BLOB NOT NULL PRIMARY KEY,
-  contact_conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
-  cr_invitation BLOB NOT NULL,
-  recipient_conn_info BLOB NOT NULL,
+  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 BLOB,
-  created_at TEXT NOT NULL DEFAULT (datetime('now'))
-) WITHOUT ROWID;
+  own_conn_info BYTEA,
+  created_at TIMESTAMP NOT NULL DEFAULT (now())
+);
 
 CREATE TABLE ratchets (
-  conn_id BLOB NOT NULL PRIMARY KEY REFERENCES connections
+  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 BLOB,
-  x3dh_priv_key_2 BLOB,
+  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 BLOB,
+  ratchet_state BYTEA,
   e2e_version INTEGER NOT NULL DEFAULT 1
-) WITHOUT ROWID;
+);
 
 CREATE TABLE skipped_messages (
-  skipped_message_id INTEGER PRIMARY KEY,
-  conn_id BLOB NOT NULL REFERENCES ratchets
+  skipped_message_id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  conn_id BYTEA NOT NULL REFERENCES ratchets
     ON DELETE CASCADE,
-  header_key BLOB NOT NULL,
+  header_key BYTEA NOT NULL,
   msg_n INTEGER NOT NULL,
-  msg_key BLOB 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/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

@@ -15,10 +15,13 @@ 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)
 
@@ -69,16 +72,24 @@ 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 "" = ""

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
@@ -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/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,7 +21,7 @@ import SMPClient
     withSmpServerOn,
     withSmpServerThreadOn,
   )
-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)
@@ -43,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
@@ -71,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
@@ -87,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
@@ -159,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,
@@ -172,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