postgres: db interfaces wip (sqlite passes) (#1419)

src/Simplex/FileTransfer/Agent.hs

@@ -66,8 +66,8 @@ import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Stats
-import Simplex.Messaging.Agent.Store.SQLite
-import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
+import Simplex.Messaging.Agent.Store.AgentStore
+import qualified Simplex.Messaging.Agent.Store.DB as DB
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.File (CryptoFile (..), CryptoFileArgs)
 import qualified Simplex.Messaging.Crypto.File as CF

src/Simplex/Messaging/Agent.hs

@@ -167,9 +167,11 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Stats
 import Simplex.Messaging.Agent.Store
-import Simplex.Messaging.Agent.Store.SQLite
-import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
-import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+import Simplex.Messaging.Agent.Store.AgentStore
+import Simplex.Messaging.Agent.Store.Common (DBStore)
+import qualified Simplex.Messaging.Agent.Store.DB as DB
+import qualified Simplex.Messaging.Agent.Store.Migrations as Migrations
+import Simplex.Messaging.Agent.Store.Shared (UpMigration (..), upMigration)
 import Simplex.Messaging.Client (SMPClientError, ServerTransmission (..), ServerTransmissionBatch, temporaryClientError, unexpectedResponse)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.File (CryptoFile, CryptoFileArgs)
@@ -200,11 +202,11 @@ import UnliftIO.STM
 type AE a = ExceptT AgentErrorType IO a
 
 -- | Creates an SMP agent client instance
-getSMPAgentClient :: AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> IO AgentClient
+getSMPAgentClient :: AgentConfig -> InitialAgentServers -> DBStore -> Bool -> IO AgentClient
 getSMPAgentClient = getSMPAgentClient_ 1
 {-# INLINE getSMPAgentClient #-}
 
-getSMPAgentClient_ :: Int -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> IO AgentClient
+getSMPAgentClient_ :: Int -> AgentConfig -> InitialAgentServers -> DBStore -> Bool -> IO AgentClient
 getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp} store backgroundMode =
   newSMPAgentEnv cfg store >>= runReaderT runAgent
   where
@@ -277,7 +279,7 @@ disposeAgentClient c@AgentClient {acThread, agentEnv = Env {store}} = do
   t_ <- atomically (swapTVar acThread Nothing) $>>= (liftIO . deRefWeak)
   disconnectAgentClient c
   mapM_ killThread t_
-  liftIO $ closeSQLiteStore store
+  liftIO $ closeStore store
 
 resumeAgentClient :: AgentClient -> IO ()
 resumeAgentClient c = atomically $ writeTVar (active c) True
@@ -2154,7 +2156,7 @@ execAgentStoreSQL :: AgentClient -> Text -> AE [Text]
 execAgentStoreSQL c sql = withAgentEnv c $ withStore' c (`execSQL` sql)
 
 getAgentMigrations :: AgentClient -> AE [UpMigration]
-getAgentMigrations c = withAgentEnv c $ map upMigration <$> withStore' c (Migrations.getCurrent . DB.conn)
+getAgentMigrations c = withAgentEnv c $ map upMigration <$> withStore' c Migrations.getCurrent
 
 debugAgentLocks :: AgentClient -> IO AgentLocks
 debugAgentLocks AgentClient {connLocks = cs, invLocks = is, deleteLock = d} = do

src/Simplex/Messaging/Agent/Client.hs

@@ -221,8 +221,8 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Stats
 import Simplex.Messaging.Agent.Store
-import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore (..), withTransaction)
-import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
+import Simplex.Messaging.Agent.Store.Common (DBStore, withTransaction)
+import qualified Simplex.Messaging.Agent.Store.DB as DB
 import Simplex.Messaging.Agent.TRcvQueues (TRcvQueues (getRcvQueues))
 import qualified Simplex.Messaging.Agent.TRcvQueues as RQ
 import Simplex.Messaging.Client
@@ -555,7 +555,7 @@ slowNetworkConfig cfg@NetworkConfig {tcpConnectTimeout, tcpTimeout, tcpTimeoutPe
     slow :: Integral a => a -> a
     slow t = (t * 3) `div` 2
 
-agentClientStore :: AgentClient -> SQLiteStore
+agentClientStore :: AgentClient -> DBStore
 agentClientStore AgentClient {agentEnv = Env {store}} = store
 {-# INLINE agentClientStore #-}
 
@@ -1649,7 +1649,7 @@ disableQueuesNtfs = sendTSessionBatches "NDEL" snd disableQueues_
 sendAck :: AgentClient -> RcvQueue -> MsgId -> AM ()
 sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId =
   withSMPClient c rq ("ACK:" <> logSecret' msgId) $ \smp ->
-    ackSMPMessage smp rcvPrivateKey rcvId msgId      
+    ackSMPMessage smp rcvPrivateKey rcvId msgId
 
 hasGetLock :: AgentClient -> RcvQueue -> IO Bool
 hasGetLock c RcvQueue {server, rcvId} =
@@ -2044,7 +2044,7 @@ pickServer = \case
 getNextServer ::
   (ProtocolTypeI p, UserProtocol p) =>
   AgentClient ->
-  UserId -> 
+  UserId ->
   (UserServers p -> NonEmpty (Maybe OperatorId, ProtoServerWithAuth p)) ->
   [ProtocolServer p] ->
   AM (ProtoServerWithAuth p)
@@ -2097,7 +2097,7 @@ withNextSrv ::
   UserId ->
   (UserServers p -> NonEmpty (Maybe OperatorId, ProtoServerWithAuth p)) ->
   TVar (Set TransportHost) ->
-  [ProtocolServer p] -> 
+  [ProtocolServer p] ->
   (ProtoServerWithAuth p -> AM a) ->
   AM a
 withNextSrv c userId srvsSel triedHosts usedSrvs action = do

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

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
@@ -52,7 +53,6 @@ import Control.Monad.Reader
 import Crypto.Random
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Aeson.TH as JQ
-import Data.ByteArray (ScrubbedBytes)
 import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
@@ -68,8 +68,9 @@ import Numeric.Natural
 import Simplex.FileTransfer.Client (XFTPClientConfig (..), defaultXFTPClientConfig)
 import Simplex.Messaging.Agent.Protocol
 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 (createStore)
+import Simplex.Messaging.Agent.Store.Common (DBStore)
+import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..), MigrationError (..))
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (VersionRangeE2E, supportedE2EEncryptVRange)
@@ -86,6 +87,10 @@ import Simplex.Messaging.Util (allFinally, catchAllErrors, catchAllErrors', tryA
 import System.Mem.Weak (Weak)
 import System.Random (StdGen, newStdGen)
 import UnliftIO.STM
+#if defined(dbPostgres)
+#else
+import Data.ByteArray (ScrubbedBytes)
+#endif
 
 type AM' a = ReaderT Env IO a
 
@@ -254,7 +259,7 @@ defaultAgentConfig =
 
 data Env = Env
   { config :: AgentConfig,
-    store :: SQLiteStore,
+    store :: DBStore,
     random :: TVar ChaChaDRG,
     randomServer :: TVar StdGen,
     ntfSupervisor :: NtfSupervisor,
@@ -262,7 +267,7 @@ data Env = Env
     multicastSubscribers :: TMVar Int
   }
 
-newSMPAgentEnv :: AgentConfig -> SQLiteStore -> IO Env
+newSMPAgentEnv :: AgentConfig -> DBStore -> IO Env
 newSMPAgentEnv config store = do
   random <- C.newRandom
   randomServer <- newTVarIO =<< liftIO newStdGen
@@ -271,8 +276,14 @@ newSMPAgentEnv config store = do
   multicastSubscribers <- newTMVarIO 0
   pure Env {config, store, random, randomServer, ntfSupervisor, xftpAgent, multicastSubscribers}
 
-createAgentStore :: FilePath -> ScrubbedBytes -> Bool -> MigrationConfirmation -> IO (Either MigrationError SQLiteStore)
-createAgentStore dbFilePath dbKey keepKey = createSQLiteStore dbFilePath dbKey keepKey Migrations.app
+#if defined(dbPostgres)
+-- TODO [postgres] pass db name / ConnectInfo?
+createAgentStore :: MigrationConfirmation -> IO (Either MigrationError DBStore)
+createAgentStore = createStore
+#else
+createAgentStore :: FilePath -> ScrubbedBytes -> Bool -> MigrationConfirmation -> IO (Either MigrationError DBStore)
+createAgentStore = createStore
+#endif
 
 data NtfSupervisor = NtfSupervisor
   { ntfTkn :: TVar (Maybe NtfToken),

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -43,8 +43,8 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Stats
 import Simplex.Messaging.Agent.Store
-import Simplex.Messaging.Agent.Store.SQLite
-import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
+import Simplex.Messaging.Agent.Store.AgentStore
+import qualified Simplex.Messaging.Agent.Store.DB as DB
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Types

src/Simplex/Messaging/Agent/Store.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
@@ -25,10 +26,15 @@ import Data.List (find)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Maybe (isJust)
+import Data.Text (Text)
 import Data.Time (UTCTime)
 import Data.Type.Equality
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval (RI2State)
+import Simplex.Messaging.Agent.Store.Common
+import qualified Simplex.Messaging.Agent.Store.DB as DB
+import qualified Simplex.Messaging.Agent.Store.Migrations as Migrations
+import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..), MigrationError (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (PQEncryption, PQSupport, RatchetX448)
 import Simplex.Messaging.Encoding.String
@@ -42,12 +48,32 @@ import Simplex.Messaging.Protocol
     RcvDhSecret,
     RcvNtfDhSecret,
     RcvPrivateAuthKey,
+    SenderCanSecure,
     SndPrivateAuthKey,
     SndPublicAuthKey,
-    SenderCanSecure,
     VersionSMPC,
   )
 import qualified Simplex.Messaging.Protocol as SMP
+#if defined(dbPostgres)
+import qualified Simplex.Messaging.Agent.Store.Postgres as StoreFunctions
+#else
+import qualified Simplex.Messaging.Agent.Store.SQLite as StoreFunctions
+import Data.ByteArray (ScrubbedBytes)
+#endif
+
+#if defined(dbPostgres)
+createStore :: MigrationConfirmation -> IO (Either MigrationError DBStore)
+createStore = StoreFunctions.createDBStore Migrations.app
+#else
+createStore :: FilePath -> ScrubbedBytes -> Bool -> MigrationConfirmation -> IO (Either MigrationError DBStore)
+createStore dbFilePath dbKey keepKey = StoreFunctions.createDBStore dbFilePath dbKey keepKey Migrations.app
+#endif
+
+closeStore :: DBStore -> IO ()
+closeStore = StoreFunctions.closeDBStore
+
+execSQL :: DB.Connection -> Text -> IO [Text]
+execSQL = StoreFunctions.execSQL
 
 -- * Queue types
 

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

@@ -0,0 +1,14 @@
+{-# LANGUAGE CPP #-}
+
+module Simplex.Messaging.Agent.Store.Common
+#if defined(dbPostgres)
+  ( module Simplex.Messaging.Agent.Store.Postgres.Common,
+  )
+  where
+import Simplex.Messaging.Agent.Store.Postgres.Common
+#else
+  ( module Simplex.Messaging.Agent.Store.SQLite.Common,
+  )
+  where
+import Simplex.Messaging.Agent.Store.SQLite.Common
+#endif

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

@@ -0,0 +1,15 @@
+{-# LANGUAGE CPP #-}
+
+module Simplex.Messaging.Agent.Store.DB
+#if defined(dbPostgres)
+  ( module Simplex.Messaging.Agent.Store.Postgres.DB,
+  )
+  where
+import Simplex.Messaging.Agent.Store.Postgres.DB
+#else
+  ( module Simplex.Messaging.Agent.Store.SQLite.DB,
+  )
+  where
+import Simplex.Messaging.Agent.Store.SQLite.DB
+#endif
+

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

@@ -0,0 +1,95 @@
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE LambdaCase #-}
+
+module Simplex.Messaging.Agent.Store.Migrations
+  ( Migration (..),
+    MigrationsToRun (..),
+    DownMigration (..),
+    Migrations.app,
+    Migrations.getCurrent,
+    get,
+    Migrations.initialize,
+    Migrations.run,
+    migrateSchema,
+    -- for tests
+    migrationsToRun,
+    toDownMigration,
+  )
+where
+
+import Control.Monad
+import Data.Char (toLower)
+import Data.Functor (($>))
+import Data.Maybe (isNothing, mapMaybe)
+import Simplex.Messaging.Agent.Store.Common
+import Simplex.Messaging.Agent.Store.Shared
+import System.Exit (exitFailure)
+import System.IO (hFlush, stdout)
+#if defined(dbPostgres)
+import qualified Simplex.Messaging.Agent.Store.Postgres.Migrations as Migrations
+#else
+import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+import System.Directory (copyFile)
+#endif
+
+get :: DBStore -> [Migration] -> IO (Either MTRError MigrationsToRun)
+get st migrations = migrationsToRun migrations <$> withTransaction st Migrations.getCurrent
+
+migrationsToRun :: [Migration] -> [Migration] -> Either MTRError MigrationsToRun
+migrationsToRun [] [] = Right MTRNone
+migrationsToRun appMs [] = Right $ MTRUp appMs
+migrationsToRun [] dbMs
+  | length dms == length dbMs = Right $ MTRDown dms
+  | otherwise = Left $ MTRENoDown $ mapMaybe nameNoDown dbMs
+  where
+    dms = mapMaybe toDownMigration dbMs
+    nameNoDown m = if isNothing (down m) then Just $ name m else Nothing
+migrationsToRun (a : as) (d : ds)
+  | name a == name d = migrationsToRun as ds
+  | otherwise = Left $ MTREDifferent (name a) (name d)
+
+migrateSchema :: DBStore -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError ())
+migrateSchema st migrations confirmMigrations = do
+  Migrations.initialize st
+  get st migrations >>= \case
+    Left e -> do
+      when (confirmMigrations == MCConsole) $ confirmOrExit ("Database state error: " <> mtrErrorDescription e)
+      pure . Left $ MigrationError e
+    Right MTRNone -> pure $ Right ()
+    Right ms@(MTRUp ums)
+      | dbNew st -> Migrations.run st ms $> Right ()
+      | otherwise -> case confirmMigrations of
+          MCYesUp -> runWithBackup st ms
+          MCYesUpDown -> runWithBackup st ms
+          MCConsole -> confirm err >> runWithBackup st ms
+          MCError -> pure $ Left err
+      where
+        err = MEUpgrade $ map upMigration ums -- "The app has a newer version than the database.\nConfirm to back up and upgrade using these migrations: " <> intercalate ", " (map name ums)
+    Right ms@(MTRDown dms) -> case confirmMigrations of
+      MCYesUpDown -> runWithBackup st ms
+      MCConsole -> confirm err >> runWithBackup st ms
+      MCYesUp -> pure $ Left err
+      MCError -> pure $ Left err
+      where
+        err = MEDowngrade $ map downName dms
+  where
+    confirm err = confirmOrExit $ migrationErrorDescription err
+
+runWithBackup :: DBStore -> MigrationsToRun -> IO (Either a ())
+#if defined(dbPostgres)
+runWithBackup st ms = Migrations.run st ms $> Right ()
+#else
+runWithBackup st ms = do
+  let f = dbFilePath st
+  copyFile f (f <> ".bak")
+  Migrations.run st ms
+  pure $ Right ()
+#endif
+
+confirmOrExit :: String -> IO ()
+confirmOrExit s = do
+  putStrLn s
+  putStr "Continue (y/N): "
+  hFlush stdout
+  ok <- getLine
+  when (map toLower ok /= "y") exitFailure

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

@@ -0,0 +1,42 @@
+module Simplex.Messaging.Agent.Store.Postgres
+  ( createDBStore,
+    closeDBStore,
+    execSQL,
+  )
+where
+
+import Data.Text (Text)
+import qualified Database.PostgreSQL.Simple as PSQL
+import Simplex.Messaging.Agent.Store.Postgres.Common
+import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationError (..))
+
+-- TODO [postgres] pass db name / ConnectInfo?
+createDBStore :: [Migration] -> MigrationConfirmation -> IO (Either MigrationError DBStore)
+createDBStore = undefined
+
+closeDBStore :: DBStore -> IO ()
+closeDBStore = undefined
+
+execSQL :: PSQL.Connection -> Text -> IO [Text]
+execSQL = undefined
+
+-- createDatabaseIfNotExists :: ConnectInfo -> String -> IO ()
+-- createDatabaseIfNotExists defaultConnectInfo targetDbName = do
+--     -- Connect to the default maintenance database (e.g., postgres)
+--     bracket (connect defaultConnectInfo) close $ \conn -> do
+--         -- Check if the database already exists
+--         [Only dbExists] <- query conn
+--             [sql|
+--                 SELECT EXISTS (
+--                     SELECT 1 FROM pg_catalog.pg_database
+--                     WHERE datname = ?
+--                 )
+--             |] (Only targetDbName)
+
+--         -- If it doesn't exist, create the database
+--         if not dbExists
+--             then do
+--                 putStrLn $ "Creating database: " ++ targetDbName
+--                 execute_ conn (Query $ "CREATE DATABASE " <> targetDbName)
+--                 putStrLn "Database created."
+--             else putStrLn "Database already exists."

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

@@ -0,0 +1,47 @@
+{-# LANGUAGE NamedFieldPuns #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Common
+  ( DBStore (..),
+    withConnection,
+    withConnection',
+    withTransaction,
+    withTransaction',
+    withTransactionPriority,
+  )
+where
+
+import qualified Database.PostgreSQL.Simple as PSQL
+import UnliftIO.MVar
+import UnliftIO.STM
+
+-- TODO [postgres] use log_min_duration_statement instead of custom slow queries (SQLite's Connection type)
+data DBStore = DBStore
+  { dbConnectInfo :: PSQL.ConnectInfo,
+    dbConnection :: MVar PSQL.Connection,
+    dbClosed :: TVar Bool,
+    dbNew :: Bool
+  }
+
+-- TODO [postgres] semaphore / connection pool?
+withConnectionPriority :: DBStore -> Bool -> (PSQL.Connection -> IO a) -> IO a
+withConnectionPriority DBStore {dbConnection} _priority action =
+  withMVar dbConnection action
+
+withConnection :: DBStore -> (PSQL.Connection -> IO a) -> IO a
+withConnection st = withConnectionPriority st False
+
+withConnection' :: DBStore -> (PSQL.Connection -> IO a) -> IO a
+withConnection' = withConnection
+
+withTransaction' :: DBStore -> (PSQL.Connection -> IO a) -> IO a
+withTransaction' = withTransaction
+
+withTransaction :: DBStore -> (PSQL.Connection -> IO a) -> IO a
+withTransaction st = withTransactionPriority st False
+{-# INLINE withTransaction #-}
+
+-- TODO [postgres] analogue for dbBusyLoop?
+withTransactionPriority :: DBStore -> Bool -> (PSQL.Connection -> IO a) -> IO a
+withTransactionPriority st priority action = withConnectionPriority st priority transaction
+  where
+    transaction conn = PSQL.withTransaction conn $ action conn

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

@@ -0,0 +1,26 @@
+module Simplex.Messaging.Agent.Store.Postgres.DB
+  ( PSQL.Connection,
+    PSQL.connect,
+    PSQL.close,
+    execute,
+    execute_,
+    executeMany,
+    PSQL.query,
+    PSQL.query_,
+  )
+where
+
+import Control.Monad (void)
+import qualified Database.PostgreSQL.Simple as PSQL
+
+execute :: PSQL.ToRow q => PSQL.Connection -> PSQL.Query -> q -> IO ()
+execute db q qs = void $ PSQL.execute db q qs
+{-# INLINE execute #-}
+
+execute_ :: PSQL.Connection -> PSQL.Query -> IO ()
+execute_ db q = void $ PSQL.execute_ db q
+{-# INLINE execute_ #-}
+
+executeMany :: PSQL.ToRow q => PSQL.Connection -> PSQL.Query -> [q] -> IO ()
+executeMany db q qs = void $ PSQL.executeMany db q qs
+{-# INLINE executeMany #-}

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

@@ -0,0 +1,42 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations
+  ( app,
+    initialize,
+    run,
+    getCurrent,
+  )
+where
+
+import Data.List (sortOn)
+import Data.Text (Text)
+import qualified Data.Text as T
+import qualified Database.PostgreSQL.Simple as PSQL
+import Simplex.Messaging.Agent.Store.Postgres.Common
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20241210_initial
+import Simplex.Messaging.Agent.Store.Shared
+
+schemaMigrations :: [(String, Text, Maybe Text)]
+schemaMigrations =
+  [ ("20241210_initial", m20241210_initial, Nothing)
+  ]
+
+-- | The list of migrations in ascending order by date
+app :: [Migration]
+app = sortOn name $ map migration schemaMigrations
+  where
+    migration (name, up, down) = Migration {name, up, down = down}
+
+-- TODO [postgres] initialize
+initialize :: DBStore -> IO ()
+initialize st = undefined
+
+-- TODO [postgres] run
+run :: DBStore -> MigrationsToRun -> IO ()
+run st = undefined
+
+getCurrent :: PSQL.Connection -> IO [Migration]
+getCurrent db = map toMigration <$> PSQL.query_ db "SELECT name, down FROM migrations ORDER BY name ASC;"
+  where
+    toMigration (name, down) = Migration {name, up = T.pack "", down}

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

@@ -2,26 +2,22 @@
 
 module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20241210_initial where
 
-import Database.PostgreSQL.Simple (Query)
-import Database.PostgreSQL.Simple.SqlQQ (sql)
+import Data.Text (Text)
+import qualified Data.Text as T
+import Text.RawString.QQ (r)
 
-m20241210_initial :: Query
+m20241210_initial :: Text
 m20241210_initial =
-  [sql|
+  T.pack
+    [r|
 REVOKE CREATE ON SCHEMA public FROM PUBLIC;
 
--- TODO remove
+-- TODO [postgres] remove
 DROP SCHEMA IF EXISTS agent_schema CASCADE;
 CREATE SCHEMA agent_schema;
 
 SET search_path TO agent_schema;
 
-CREATE TABLE migrations(
-  name TEXT NOT NULL,
-  ts TEXT NOT NULL,
-  down TEXT,
-  PRIMARY KEY(name)
-);
 CREATE TABLE users(
   user_id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
   deleted INTEGER NOT NULL DEFAULT 0

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

@@ -4,7 +4,7 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Simplex.Messaging.Agent.Store.SQLite.Common
-  ( SQLiteStore (..),
+  ( DBStore (..),
     withConnection,
     withConnection',
     withTransaction,
@@ -30,7 +30,7 @@ import UnliftIO.STM
 storeKey :: ScrubbedBytes -> Bool -> Maybe ScrubbedBytes
 storeKey key keepKey = if keepKey || BA.null key then Just key else Nothing
 
-data SQLiteStore = SQLiteStore
+data DBStore = DBStore
   { dbFilePath :: FilePath,
     dbKey :: TVar (Maybe ScrubbedBytes),
     dbSem :: TVar Int,
@@ -39,8 +39,8 @@ data SQLiteStore = SQLiteStore
     dbNew :: Bool
   }
 
-withConnectionPriority :: SQLiteStore -> Bool -> (DB.Connection -> IO a) -> IO a
-withConnectionPriority SQLiteStore {dbSem, dbConnection} priority action
+withConnectionPriority :: DBStore -> Bool -> (DB.Connection -> IO a) -> IO a
+withConnectionPriority DBStore {dbSem, dbConnection} priority action
   | priority = E.bracket_ signal release $ withMVar dbConnection action
   | otherwise = lowPriority
   where
@@ -50,20 +50,20 @@ withConnectionPriority SQLiteStore {dbSem, dbConnection} priority action
     wait = unlessM free $ atomically $ unlessM ((0 ==) <$> readTVar dbSem) retry
     free = (0 ==) <$> readTVarIO dbSem
 
-withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withConnection :: DBStore -> (DB.Connection -> IO a) -> IO a
 withConnection st = withConnectionPriority st False
 
-withConnection' :: SQLiteStore -> (SQL.Connection -> IO a) -> IO a
+withConnection' :: DBStore -> (SQL.Connection -> IO a) -> IO a
 withConnection' st action = withConnection st $ action . DB.conn
 
-withTransaction' :: SQLiteStore -> (SQL.Connection -> IO a) -> IO a
+withTransaction' :: DBStore -> (SQL.Connection -> IO a) -> IO a
 withTransaction' st action = withTransaction st $ action . DB.conn
 
-withTransaction :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransaction :: DBStore -> (DB.Connection -> IO a) -> IO a
 withTransaction st = withTransactionPriority st False
 {-# INLINE withTransaction #-}
 
-withTransactionPriority :: SQLiteStore -> Bool -> (DB.Connection -> IO a) -> IO a
+withTransactionPriority :: DBStore -> Bool -> (DB.Connection -> IO a) -> IO a
 withTransactionPriority st priority action = withConnectionPriority st priority $ dbBusyLoop . transaction
   where
     transaction db@DB.Connection {conn} = SQL.withImmediateTransaction conn $ action db

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

@@ -20,8 +20,8 @@ module Simplex.Messaging.Agent.Store.SQLite.DB
 where
 
 import Control.Concurrent.STM
-import Control.Monad (when)
 import Control.Exception
+import Control.Monad (when)
 import qualified Data.Aeson.TH as J
 import Data.Int (Int64)
 import Data.Map.Strict (Map)
@@ -51,9 +51,10 @@ data SlowQueryStats = SlowQueryStats
 timeIt :: TMap Query SlowQueryStats -> Query -> IO a -> IO a
 timeIt slow sql a = do
   t <- getCurrentTime
-  r <- a `catch` \e -> do
-    atomically $ TM.alter (Just . updateQueryErrors e) sql slow
-    throwIO e
+  r <-
+    a `catch` \e -> do
+      atomically $ TM.alter (Just . updateQueryErrors e) sql slow
+      throwIO e
   t' <- getCurrentTime
   let diff = diffToMilliseconds $ diffUTCTime t' t
   when (diff > 1) $ atomically $ TM.alter (updateQueryStats diff) sql slow
@@ -91,6 +92,7 @@ execute_ :: Connection -> Query -> IO ()
 execute_ Connection {conn, slow} sql = timeIt slow sql $ SQL.execute_ conn sql
 {-# INLINE execute_ #-}
 
+-- TODO [postgres] remove
 executeNamed :: Connection -> Query -> [NamedParam] -> IO ()
 executeNamed Connection {conn, slow} sql = timeIt slow sql . SQL.executeNamed conn sql
 {-# INLINE executeNamed #-}
@@ -107,6 +109,7 @@ query_ :: FromRow r => Connection -> Query -> IO [r]
 query_ Connection {conn, slow} sql = timeIt slow sql $ SQL.query_ conn sql
 {-# INLINE query_ #-}
 
+-- TODO [postgres] remove
 queryNamed :: FromRow r => Connection -> Query -> [NamedParam] -> IO [r]
 queryNamed Connection {conn, slow} sql = timeIt slow sql . SQL.queryNamed conn sql
 {-# INLINE queryNamed #-}

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

@@ -5,40 +5,29 @@
 {-# LANGUAGE QuasiQuotes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StrictData #-}
-{-# LANGUAGE TemplateHaskell #-}
 {-# LANGUAGE TupleSections #-}
 
 module Simplex.Messaging.Agent.Store.SQLite.Migrations
-  ( Migration (..),
-    MigrationsToRun (..),
-    MTRError (..),
-    DownMigration (..),
-    app,
+  ( app,
     initialize,
-    get,
     run,
     getCurrent,
-    mtrErrorDescription,
-    -- for unit tests
-    migrationsToRun,
-    toDownMigration,
   )
 where
 
 import Control.Monad (forM_, when)
-import qualified Data.Aeson.TH as J
-import Data.List (intercalate, sortOn)
+import Data.List (sortOn)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.Map.Strict as M
-import Data.Maybe (isNothing, mapMaybe)
 import Data.Text (Text)
 import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (getCurrentTime)
-import Database.SQLite.Simple (Connection, Only (..), Query (..))
-import qualified Database.SQLite.Simple as DB
+import Database.SQLite.Simple (Only (..), Query (..))
+import qualified Database.SQLite.Simple as SQL
 import Database.SQLite.Simple.QQ (sql)
 import qualified Database.SQLite3 as SQLite3
 import Simplex.Messaging.Agent.Protocol (extraSMPServerHosts)
+import qualified Simplex.Messaging.Agent.Store.DB as DB
 import Simplex.Messaging.Agent.Store.SQLite.Common
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
@@ -76,13 +65,10 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240624_snd_secure
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240702_servers_stats
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240930_ntf_tokens_to_delete
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241007_rcv_queues_last_broker_ts
+import Simplex.Messaging.Agent.Store.Shared
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
 
-data Migration = Migration {name :: String, up :: Text, down :: Maybe Text}
-  deriving (Eq, Show)
-
 schemaMigrations :: [(String, Query, Maybe Query)]
 schemaMigrations =
   [ ("20220101_initial", m20220101_initial, Nothing),
@@ -129,15 +115,12 @@ app = sortOn name $ map migration schemaMigrations
   where
     migration (name, up, down) = Migration {name, up = fromQuery up, down = fromQuery <$> down}
 
-get :: SQLiteStore -> [Migration] -> IO (Either MTRError MigrationsToRun)
-get st migrations = migrationsToRun migrations <$> withTransaction' st getCurrent
-
-getCurrent :: Connection -> IO [Migration]
-getCurrent db = map toMigration <$> DB.query_ db "SELECT name, down FROM migrations ORDER BY name ASC;"
+getCurrent :: DB.Connection -> IO [Migration]
+getCurrent DB.Connection {DB.conn} = map toMigration <$> SQL.query_ conn "SELECT name, down FROM migrations ORDER BY name ASC;"
   where
     toMigration (name, down) = Migration {name, up = "", down}
 
-run :: SQLiteStore -> MigrationsToRun -> IO ()
+run :: DBStore -> MigrationsToRun -> IO ()
 run st = \case
   MTRUp [] -> pure ()
   MTRUp ms -> mapM_ runUp ms >> withConnection' st (`execSQL` "VACUUM;")
@@ -148,27 +131,27 @@ run st = \case
       when (name == "m20220811_onion_hosts") $ updateServers db
       insert db >> execSQL db up'
       where
-        insert db = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
+        insert db = SQL.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
         up'
           | dbNew st && name == "m20230110_users" = fromQuery new_m20230110_users
           | otherwise = up
         updateServers db = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
           let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
-           in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
+           in SQL.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
     runDown DownMigration {downName, downQuery} = withTransaction' st $ \db -> do
       execSQL db downQuery
-      DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
-    execSQL db = SQLite3.exec $ DB.connectionHandle db
+      SQL.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
+    execSQL db = SQLite3.exec $ SQL.connectionHandle db
 
-initialize :: SQLiteStore -> IO ()
+initialize :: DBStore -> IO ()
 initialize st = withTransaction' st $ \db -> do
-  cs :: [Text] <- map fromOnly <$> DB.query_ db "SELECT name FROM pragma_table_info('migrations')"
+  cs :: [Text] <- map fromOnly <$> SQL.query_ db "SELECT name FROM pragma_table_info('migrations')"
   case cs of
     [] -> createMigrations db
-    _ -> when ("down" `notElem` cs) $ DB.execute_ db "ALTER TABLE migrations ADD COLUMN down TEXT"
+    _ -> when ("down" `notElem` cs) $ SQL.execute_ db "ALTER TABLE migrations ADD COLUMN down TEXT"
   where
     createMigrations db =
-      DB.execute_
+      SQL.execute_
         db
         [sql|
           CREATE TABLE IF NOT EXISTS migrations (
@@ -178,37 +161,3 @@ initialize st = withTransaction' st $ \db -> do
             PRIMARY KEY (name)
           );
         |]
-
-data DownMigration = DownMigration {downName :: String, downQuery :: Text}
-  deriving (Eq, Show)
-
-toDownMigration :: Migration -> Maybe DownMigration
-toDownMigration Migration {name, down} = DownMigration name <$> down
-
-data MigrationsToRun = MTRUp [Migration] | MTRDown [DownMigration] | MTRNone
-  deriving (Eq, Show)
-
-data MTRError
-  = MTRENoDown {dbMigrations :: [String]}
-  | MTREDifferent {appMigration :: String, dbMigration :: String}
-  deriving (Eq, Show)
-
-mtrErrorDescription :: MTRError -> String
-mtrErrorDescription = \case
-  MTRENoDown ms -> "database version is newer than the app, but no down migration for: " <> intercalate ", " ms
-  MTREDifferent a d -> "different migration in the app/database: " <> a <> " / " <> d
-
-migrationsToRun :: [Migration] -> [Migration] -> Either MTRError MigrationsToRun
-migrationsToRun [] [] = Right MTRNone
-migrationsToRun appMs [] = Right $ MTRUp appMs
-migrationsToRun [] dbMs
-  | length dms == length dbMs = Right $ MTRDown dms
-  | otherwise = Left $ MTRENoDown $ mapMaybe nameNoDown dbMs
-  where
-    dms = mapMaybe toDownMigration dbMs
-    nameNoDown m = if isNothing (down m) then Just $ name m else Nothing
-migrationsToRun (a : as) (d : ds)
-  | name a == name d = migrationsToRun as ds
-  | otherwise = Left $ MTREDifferent (name a) (name d)
-
-$(J.deriveJSON (sumTypeJSON $ dropPrefix "MTRE") ''MTRError)

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

@@ -0,0 +1,93 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TemplateHaskell #-}
+
+module Simplex.Messaging.Agent.Store.Shared
+  ( Migration (..),
+    MigrationsToRun (..),
+    DownMigration (..),
+    MTRError (..),
+    mtrErrorDescription,
+    MigrationConfirmation (..),
+    MigrationError (..),
+    UpMigration (..),
+    migrationErrorDescription,
+    -- for tests
+    toDownMigration,
+    upMigration,
+  )
+where
+
+import qualified Data.Aeson.TH as J
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.List (intercalate)
+import Data.Maybe (isJust)
+import Data.Text (Text)
+import Simplex.Messaging.Encoding.String
+import Simplex.Messaging.Parsers (defaultJSON, dropPrefix, sumTypeJSON)
+
+data Migration = Migration {name :: String, up :: Text, down :: Maybe Text}
+  deriving (Eq, Show)
+
+data DownMigration = DownMigration {downName :: String, downQuery :: Text}
+  deriving (Eq, Show)
+
+toDownMigration :: Migration -> Maybe DownMigration
+toDownMigration Migration {name, down} = DownMigration name <$> down
+
+data MigrationsToRun = MTRUp [Migration] | MTRDown [DownMigration] | MTRNone
+  deriving (Eq, Show)
+
+data MTRError
+  = MTRENoDown {dbMigrations :: [String]}
+  | MTREDifferent {appMigration :: String, dbMigration :: String}
+  deriving (Eq, Show)
+
+mtrErrorDescription :: MTRError -> String
+mtrErrorDescription = \case
+  MTRENoDown ms -> "database version is newer than the app, but no down migration for: " <> intercalate ", " ms
+  MTREDifferent a d -> "different migration in the app/database: " <> a <> " / " <> d
+
+data MigrationError
+  = MEUpgrade {upMigrations :: [UpMigration]}
+  | MEDowngrade {downMigrations :: [String]}
+  | MigrationError {mtrError :: MTRError}
+  deriving (Eq, Show)
+
+migrationErrorDescription :: MigrationError -> String
+migrationErrorDescription = \case
+  MEUpgrade ums ->
+    "The app has a newer version than the database.\nConfirm to back up and upgrade using these migrations: " <> intercalate ", " (map upName ums)
+  MEDowngrade dms ->
+    "Database version is newer than the app.\nConfirm to back up and downgrade using these migrations: " <> intercalate ", " dms
+  MigrationError err -> mtrErrorDescription err
+
+data UpMigration = UpMigration {upName :: String, withDown :: Bool}
+  deriving (Eq, Show)
+
+upMigration :: Migration -> UpMigration
+upMigration Migration {name, down} = UpMigration name $ isJust down
+
+data MigrationConfirmation = MCYesUp | MCYesUpDown | MCConsole | MCError
+  deriving (Eq, Show)
+
+instance StrEncoding MigrationConfirmation where
+  strEncode = \case
+    MCYesUp -> "yesUp"
+    MCYesUpDown -> "yesUpDown"
+    MCConsole -> "console"
+    MCError -> "error"
+  strP =
+    A.takeByteString >>= \case
+      "yesUp" -> pure MCYesUp
+      "yesUpDown" -> pure MCYesUpDown
+      "console" -> pure MCConsole
+      "error" -> pure MCError
+      _ -> fail "invalid MigrationConfirmation"
+
+$(J.deriveJSON (sumTypeJSON $ dropPrefix "MTRE") ''MTRError)
+
+$(J.deriveJSON defaultJSON ''UpMigration)
+
+$(J.deriveToJSON (sumTypeJSON $ dropPrefix "ME") ''MigrationError)