From 4980db932db0b8a2838a22dc98418631c4dc3bf7 Mon Sep 17 00:00:00 2001 From: Efim Poberezkin <8711996+efim-poberezkin@users.noreply.github.com> Date: Thu, 3 Feb 2022 13:44:42 +0400 Subject: [PATCH] use posgres fork --- apps/smp-agent/Main.hs | 2 +- src/Simplex/Messaging/Agent.hs | 6 +- src/Simplex/Messaging/Agent/Client.hs | 2 +- src/Simplex/Messaging/Agent/Env/Postgres.hs | 8 +- src/Simplex/Messaging/Agent/Server.hs | 2 +- src/Simplex/Messaging/Agent/Store/Postgres.hs | 102 ++++++------- .../Postgres/Migrations/M20220202_initial.hs | 138 +++++++++--------- tests/AgentTests/FunctionalAPITests.hs | 2 +- tests/SMPAgentClient.hs | 2 +- 9 files changed, 135 insertions(+), 129 deletions(-) diff --git a/apps/smp-agent/Main.hs b/apps/smp-agent/Main.hs index 1087530f9..845d3d866 100644 --- a/apps/smp-agent/Main.hs +++ b/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 (..)) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 41d9a073f..f93be443e 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/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,7 +172,7 @@ 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 diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index e596602e0..ad87fd1ec 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/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 diff --git a/src/Simplex/Messaging/Agent/Env/Postgres.hs b/src/Simplex/Messaging/Agent/Env/Postgres.hs index 39b6ed60a..15068e91c 100644 --- a/src/Simplex/Messaging/Agent/Env/Postgres.hs +++ b/src/Simplex/Messaging/Agent/Env/Postgres.hs @@ -20,8 +20,8 @@ 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) @@ -72,7 +72,7 @@ defaultAgentConfig = data Env = Env { config :: AgentConfig, - store :: SQLiteStore, + store :: PostgresStore, idsDrg :: TVar ChaChaDRG, clientCounter :: TVar Int, randomServer :: TVar StdGen @@ -81,7 +81,7 @@ data Env = Env newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env newSMPAgentEnv cfg = do idsDrg <- newTVarIO =<< drgNew - store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app + store <- liftIO $ createPostgresStore (dbFile cfg) (dbPoolSize cfg) Migrations.app clientCounter <- newTVarIO 0 randomServer <- newTVarIO =<< liftIO newStdGen return Env {config = cfg, store, idsDrg, clientCounter, randomServer} diff --git a/src/Simplex/Messaging/Agent/Server.hs b/src/Simplex/Messaging/Agent/Server.hs index b3c70c9ac..ae0580b21 100644 --- a/src/Simplex/Messaging/Agent/Server.hs +++ b/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) diff --git a/src/Simplex/Messaging/Agent/Store/Postgres.hs b/src/Simplex/Messaging/Agent/Store/Postgres.hs index a3bc213ef..8d6512713 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres.hs +++ b/src/Simplex/Messaging/Agent/Store/Postgres.hs @@ -17,9 +17,9 @@ {-# OPTIONS_GHC -fno-warn-orphans #-} module Simplex.Messaging.Agent.Store.Postgres - ( SQLiteStore (..), - createSQLiteStore, - connectSQLiteStore, + ( PostgresStore (..), + createPostgresStore, + connectPostgresStore, withConnection, withTransaction, fromTextField_, @@ -52,8 +52,8 @@ import Database.SQLite.Simple.QQ (sql) import Database.SQLite.Simple.ToField (ToField (..)) import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.Store -import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration) -import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations +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 @@ -70,22 +70,22 @@ import qualified UnliftIO.Exception as E -- * SQLite Store implementation -data SQLiteStore = SQLiteStore +data PostgresStore = PostgresStore { dbFilePath :: FilePath, dbConnPool :: TBQueue DB.Connection, dbNew :: Bool } -createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore -createSQLiteStore dbFilePath poolSize migrations = do +createPostgresStore :: FilePath -> Int -> [Migration] -> IO PostgresStore +createPostgresStore dbFilePath poolSize migrations = do let dbDir = takeDirectory dbFilePath createDirectoryIfMissing False dbDir - st <- connectSQLiteStore dbFilePath poolSize + st <- connectPostgresStore dbFilePath poolSize checkThreadsafe st migrateSchema st migrations pure st -checkThreadsafe :: SQLiteStore -> IO () +checkThreadsafe :: PostgresStore -> IO () checkThreadsafe st = withConnection st $ \db -> do compileOptions <- DB.query_ db "pragma COMPILE_OPTIONS;" :: IO [[Text]] let threadsafeOption = find (T.isPrefixOf "THREADSAFE=") (concat compileOptions) @@ -94,7 +94,7 @@ checkThreadsafe st = withConnection st $ \db -> do Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found" _ -> return () -migrateSchema :: SQLiteStore -> [Migration] -> IO () +migrateSchema :: PostgresStore -> [Migration] -> IO () migrateSchema st migrations = withConnection st $ \db -> do Migrations.initialize db Migrations.get db migrations >>= \case @@ -115,13 +115,13 @@ confirmOrExit s = do ok <- getLine when (map toLower ok /= "y") exitFailure -connectSQLiteStore :: FilePath -> Int -> IO SQLiteStore -connectSQLiteStore dbFilePath poolSize = do +connectPostgresStore :: FilePath -> Int -> IO PostgresStore +connectPostgresStore dbFilePath poolSize = do dbNew <- not <$> doesFileExist dbFilePath dbConnPool <- newTBQueueIO $ toEnum poolSize replicateM_ poolSize $ connectDB dbFilePath >>= atomically . writeTBQueue dbConnPool - pure SQLiteStore {dbFilePath, dbConnPool, dbNew} + pure PostgresStore {dbFilePath, dbConnPool, dbNew} connectDB :: FilePath -> IO DB.Connection connectDB path = do @@ -137,13 +137,13 @@ handleSQLError err e | DB.sqlError e == DB.ErrorConstraint = err | otherwise = SEInternal $ bshow e -withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a -withConnection SQLiteStore {dbConnPool} = +withConnection :: PostgresStore -> (DB.Connection -> IO a) -> IO a +withConnection PostgresStore {dbConnPool} = bracket (atomically $ readTBQueue dbConnPool) (atomically . writeTBQueue dbConnPool) -withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a +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 @@ -157,7 +157,7 @@ withTransaction st action = withConnection st $ loop 100 100_000 createConn_ :: (MonadUnliftIO m, MonadError StoreError m) => - SQLiteStore -> + PostgresStore -> TVar ChaChaDRG -> ConnData -> (DB.Connection -> ByteString -> IO ()) -> @@ -168,27 +168,27 @@ createConn_ st gVar cData create = ConnData {connId = ""} -> createWithRandomId gVar $ create db ConnData {connId} -> create db connId $> Right connId -instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteStore m where - createRcvConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> RcvQueue -> SConnectionMode c -> m 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 DB.execute db "INSERT INTO connections (conn_id, conn_mode) VALUES (?, ?)" (connId, cMode) insertRcvQueue_ db connId q - createSndConn :: SQLiteStore -> TVar ChaChaDRG -> ConnData -> SndQueue -> m ConnId + 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 DB.execute db "INSERT INTO connections (conn_id, conn_mode) VALUES (?, ?)" (connId, SCMInvitation) insertSndQueue_ db connId q - getConn :: SQLiteStore -> ConnId -> m SomeConn + getConn :: PostgresStore -> ConnId -> m SomeConn getConn st connId = liftIOEither . withTransaction st $ \db -> getConn_ db connId - getRcvConn :: SQLiteStore -> SMPServer -> SMP.RecipientId -> m SomeConn + getRcvConn :: PostgresStore -> SMPServer -> SMP.RecipientId -> m SomeConn getRcvConn st SMPServer {host, port} rcvId = liftIOEither . withTransaction st $ \db -> DB.queryNamed @@ -203,7 +203,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto [Only connId] -> getConn_ db connId _ -> pure $ Left SEConnNotFound - deleteConn :: SQLiteStore -> ConnId -> m () + deleteConn :: PostgresStore -> ConnId -> m () deleteConn st connId = liftIO . withTransaction st $ \db -> DB.executeNamed @@ -211,7 +211,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto "DELETE FROM connections WHERE conn_id = :conn_id;" [":conn_id" := connId] - upgradeRcvConnToDuplex :: SQLiteStore -> ConnId -> SndQueue -> m () + upgradeRcvConnToDuplex :: PostgresStore -> ConnId -> SndQueue -> m () upgradeRcvConnToDuplex st connId sq@SndQueue {server} = liftIOEither . withTransaction st $ \db -> getConn_ db connId >>= \case @@ -222,7 +222,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c _ -> pure $ Left SEConnNotFound - upgradeSndConnToDuplex :: SQLiteStore -> ConnId -> RcvQueue -> m () + upgradeSndConnToDuplex :: PostgresStore -> ConnId -> RcvQueue -> m () upgradeSndConnToDuplex st connId rq@RcvQueue {server} = liftIOEither . withTransaction st $ \db -> getConn_ db connId >>= \case @@ -233,7 +233,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto Right (SomeConn c _) -> pure . Left . SEBadConnType $ connType c _ -> pure $ Left SEConnNotFound - setRcvQueueStatus :: SQLiteStore -> RcvQueue -> QueueStatus -> m () + 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 -> @@ -246,7 +246,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto |] [":status" := status, ":host" := host, ":port" := port, ":rcv_id" := rcvId] - setRcvQueueConfirmedE2E :: SQLiteStore -> RcvQueue -> C.DhSecretX25519 -> m () + setRcvQueueConfirmedE2E :: PostgresStore -> RcvQueue -> C.DhSecretX25519 -> m () setRcvQueueConfirmedE2E st RcvQueue {rcvId, server = SMPServer {host, port}} e2eDhSecret = liftIO . withTransaction st $ \db -> DB.executeNamed @@ -264,7 +264,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto ":rcv_id" := rcvId ] - setSndQueueStatus :: SQLiteStore -> SndQueue -> QueueStatus -> m () + 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 -> @@ -277,7 +277,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto |] [":status" := status, ":host" := host, ":port" := port, ":snd_id" := sndId] - createConfirmation :: SQLiteStore -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId + 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 -> @@ -289,7 +289,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto |] (confirmationId, connId, senderKey, e2ePubKey, ratchetState, connInfo) - acceptConfirmation :: SQLiteStore -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation + acceptConfirmation :: PostgresStore -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation acceptConfirmation st confirmationId ownConnInfo = liftIOEither . withTransaction st $ \db -> do DB.executeNamed @@ -322,7 +322,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto ownConnInfo } - getAcceptedConfirmation :: SQLiteStore -> ConnId -> m AcceptedConfirmation + getAcceptedConfirmation :: PostgresStore -> ConnId -> m AcceptedConfirmation getAcceptedConfirmation st connId = liftIOEither . withTransaction st $ \db -> firstRow confirmation SEConfirmationNotFound $ @@ -344,7 +344,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto ownConnInfo } - removeConfirmations :: SQLiteStore -> ConnId -> m () + removeConfirmations :: PostgresStore -> ConnId -> m () removeConfirmations st connId = liftIO . withTransaction st $ \db -> DB.executeNamed @@ -355,7 +355,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto |] [":conn_id" := connId] - createInvitation :: SQLiteStore -> TVar ChaChaDRG -> NewInvitation -> m InvitationId + createInvitation :: PostgresStore -> TVar ChaChaDRG -> NewInvitation -> m InvitationId createInvitation st gVar NewInvitation {contactConnId, connReq, recipientConnInfo} = liftIOEither . withTransaction st $ \db -> createWithRandomId gVar $ \invitationId -> @@ -367,7 +367,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto |] (invitationId, contactConnId, connReq, recipientConnInfo) - getInvitation :: SQLiteStore -> InvitationId -> m Invitation + getInvitation :: PostgresStore -> InvitationId -> m Invitation getInvitation st invitationId = liftIOEither . withTransaction st $ \db -> firstRow invitation SEInvitationNotFound $ @@ -384,7 +384,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto invitation (contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted) = Invitation {invitationId, contactConnId, connReq, recipientConnInfo, ownConnInfo, accepted} - acceptInvitation :: SQLiteStore -> InvitationId -> ConnInfo -> m () + acceptInvitation :: PostgresStore -> InvitationId -> ConnInfo -> m () acceptInvitation st invitationId ownConnInfo = liftIO . withTransaction st $ \db -> do DB.executeNamed @@ -399,7 +399,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto ":invitation_id" := invitationId ] - deleteInvitation :: SQLiteStore -> ConnId -> InvitationId -> m () + deleteInvitation :: PostgresStore -> ConnId -> InvitationId -> m () deleteInvitation st contactConnId invId = liftIOEither . withTransaction st $ \db -> runExceptT $ @@ -408,7 +408,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto liftIO $ DB.execute db "DELETE FROM conn_invitations WHERE contact_conn_id = ? AND invitation_id = ?" (contactConnId, invId) _ -> throwError SEConnNotFound - updateRcvIds :: SQLiteStore -> ConnId -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash) + updateRcvIds :: PostgresStore -> ConnId -> m (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash) updateRcvIds st connId = liftIO . withTransaction st $ \db -> do (lastInternalId, lastInternalRcvId, lastExternalSndId, lastRcvHash) <- retrieveLastIdsAndHashRcv_ db connId @@ -417,14 +417,14 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto updateLastIdsRcv_ db connId internalId internalRcvId pure (internalId, internalRcvId, lastExternalSndId, lastRcvHash) - createRcvMsg :: SQLiteStore -> ConnId -> RcvMsgData -> m () + 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 :: SQLiteStore -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash) + updateSndIds :: PostgresStore -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash) updateSndIds st connId = liftIO . withTransaction st $ \db -> do (lastInternalId, lastInternalSndId, prevSndHash) <- retrieveLastIdsAndHashSnd_ db connId @@ -433,14 +433,14 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto updateLastIdsSnd_ db connId internalId internalSndId pure (internalId, internalSndId, prevSndHash) - createSndMsg :: SQLiteStore -> ConnId -> SndMsgData -> m () + 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 :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs)) + 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 @@ -457,13 +457,13 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto (connId, msgId) pure (rq_, msgData) - getPendingMsgs :: SQLiteStore -> ConnId -> m [InternalId] + 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 :: SQLiteStore -> ConnId -> InternalId -> m () + checkRcvMsg :: PostgresStore -> ConnId -> InternalId -> m () checkRcvMsg st connId msgId = liftIOEither . withTransaction st $ \db -> hasMsg @@ -479,17 +479,17 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto hasMsg :: [(ConnId, InternalId)] -> Either StoreError () hasMsg r = if null r then Left SEMsgNotFound else Right () - deleteMsg :: SQLiteStore -> ConnId -> InternalId -> m () + deleteMsg :: PostgresStore -> ConnId -> InternalId -> m () deleteMsg st connId msgId = liftIO . withTransaction st $ \db -> DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId) - createRatchetX3dhKeys :: SQLiteStore -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> m () + createRatchetX3dhKeys :: PostgresStore -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> m () createRatchetX3dhKeys st connId x3dhPrivKey1 x3dhPrivKey2 = liftIO . withTransaction st $ \db -> DB.execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2) - getRatchetX3dhKeys :: SQLiteStore -> ConnId -> m (C.PrivateKeyX448, C.PrivateKeyX448) + getRatchetX3dhKeys :: PostgresStore -> ConnId -> m (C.PrivateKeyX448, C.PrivateKeyX448) getRatchetX3dhKeys st connId = liftIOEither . withTransaction st $ \db -> fmap hasKeys $ @@ -500,7 +500,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto Right (Just k1, Just k2) -> Right (k1, k2) _ -> Left SEX3dhKeysNotFound - createRatchet :: SQLiteStore -> ConnId -> RatchetX448 -> m () + createRatchet :: PostgresStore -> ConnId -> RatchetX448 -> m () createRatchet st connId rc = liftIO . withTransaction st $ \db -> do DB.executeNamed @@ -515,7 +515,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto |] [":conn_id" := connId, ":ratchet_state" := rc] - getRatchet :: SQLiteStore -> ConnId -> m RatchetX448 + getRatchet :: PostgresStore -> ConnId -> m RatchetX448 getRatchet st connId = liftIOEither . withTransaction st $ \db -> ratchet @@ -524,7 +524,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto ratchet (Only (Just rc) : _) = Right rc ratchet _ = Left SERatchetNotFound - getSkippedMsgKeys :: SQLiteStore -> ConnId -> m SkippedMsgKeys + 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) @@ -534,7 +534,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto where addMsgKey = maybe (M.singleton msgN mk) (M.insert msgN mk) - updateRatchet :: SQLiteStore -> ConnId -> RatchetX448 -> SkippedMsgDiff -> m () + updateRatchet :: PostgresStore -> ConnId -> RatchetX448 -> SkippedMsgDiff -> m () updateRatchet st connId rc skipped = liftIO . withTransaction st $ \db -> do DB.execute db "UPDATE ratchets SET ratchet_state = ? WHERE conn_id = ?" (rc, connId) diff --git a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20220202_initial.hs b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20220202_initial.hs index 4a706a49d..def3e4409 100644 --- a/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20220202_initial.hs +++ b/src/Simplex/Messaging/Agent/Store/Postgres/Migrations/M20220202_initial.hs @@ -11,33 +11,33 @@ m20220202_initial = 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 +45,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 ); |] diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 116a38b55..81bf2b097 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/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 (..)) diff --git a/tests/SMPAgentClient.hs b/tests/SMPAgentClient.hs index 4cf4f308a..a207e1787 100644 --- a/tests/SMPAgentClient.hs +++ b/tests/SMPAgentClient.hs @@ -20,7 +20,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)