xftp-server: support postgresql backend

simplexmq.cabal

@@ -241,6 +241,7 @@ library
           Simplex.FileTransfer.Server.Prometheus
           Simplex.FileTransfer.Server.Stats
           Simplex.FileTransfer.Server.Store
+          Simplex.FileTransfer.Server.Store.STM
           Simplex.FileTransfer.Server.StoreLog
           Simplex.Messaging.Server
           Simplex.Messaging.Server.CLI
@@ -281,6 +282,9 @@ library
               Simplex.Messaging.Notifications.Server.Store.Postgres
               Simplex.Messaging.Notifications.Server.Store.Types
               Simplex.Messaging.Notifications.Server.StoreLog
+              Simplex.FileTransfer.Server.Store.Postgres
+              Simplex.FileTransfer.Server.Store.Postgres.Config
+              Simplex.FileTransfer.Server.Store.Postgres.Migrations
               Simplex.Messaging.Server.MsgStore.Postgres
               Simplex.Messaging.Server.QueueStore.Postgres
               Simplex.Messaging.Server.QueueStore.Postgres.Migrations
@@ -523,6 +527,7 @@ test-suite simplexmq-test
   if flag(server_postgres)
       other-modules:
           AgentTests.NotificationTests
+          CoreTests.XFTPStoreTests
           NtfClient
           NtfServerTests
           PostgresSchemaDump

src/Simplex/FileTransfer/Server/Env.hs

@@ -1,37 +1,53 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE StrictData #-}
 
 module Simplex.FileTransfer.Server.Env
   ( XFTPServerConfig (..),
+    XFTPStoreConfig (..),
     XFTPEnv (..),
     XFTPRequest (..),
     defaultInactiveClientExpiration,
     defFileExpirationHours,
     defaultFileExpiration,
     newXFTPServerEnv,
-    countUsedStorage,
+    runWithStoreConfig,
+    checkFileStoreMode,
+    importToDatabase,
+    exportFromDatabase,
   ) where
 
 import Control.Logger.Simple
 import Control.Monad
 import Crypto.Random
 import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty)
-import qualified Data.Map.Strict as M
 import Data.Time.Clock (getCurrentTime)
 import Data.Word (Word32)
 import Data.X509.Validation (Fingerprint (..))
 import Network.Socket
 import qualified Network.TLS as T
 import Simplex.FileTransfer.Protocol (FileCmd, FileInfo (..), XFTPFileId)
 import Simplex.FileTransfer.Server.Stats
+import Data.Ini (Ini)
 import Simplex.FileTransfer.Server.Store
+import Simplex.FileTransfer.Server.Store.STM (STMFileStore (..))
+import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation)
+#if defined(dbServerPostgres)
+import Data.Functor (($>))
+import Simplex.FileTransfer.Server.Store.Postgres (PostgresFileStore, importFileStore, exportFileStore)
+import Simplex.FileTransfer.Server.Store.Postgres.Config (PostgresFileStoreCfg (..), defaultXFTPDBOpts)
+import Simplex.Messaging.Server.CLI (iniDBOptions, settingIsOn)
+import System.Directory (doesFileExist)
+import System.Exit (exitFailure)
+#endif
 import Simplex.FileTransfer.Server.StoreLog
 import Simplex.FileTransfer.Transport (VersionRangeXFTP)
 import qualified Simplex.Messaging.Crypto as C
@@ -88,9 +104,16 @@ defaultInactiveClientExpiration =
       checkInterval = 3600 -- seconds, 1 hours
     }
 
-data XFTPEnv = XFTPEnv
+data XFTPStoreConfig s where
+  XSCMemory :: Maybe FilePath -> XFTPStoreConfig STMFileStore
+#if defined(dbServerPostgres)
+  XSCDatabase :: PostgresFileStoreCfg -> XFTPStoreConfig PostgresFileStore
+#endif
+
+data XFTPEnv s = XFTPEnv
   { config :: XFTPServerConfig,
-    store :: FileStore,
+    store :: s,
+    usedStorage :: TVar Int64,
     storeLog :: Maybe (StoreLog 'WriteMode),
     random :: TVar ChaChaDRG,
     serverIdentity :: C.KeyHash,
@@ -109,26 +132,93 @@ defaultFileExpiration =
       checkInterval = 2 * 3600 -- seconds, 2 hours
     }
 
-newXFTPServerEnv :: XFTPServerConfig -> IO XFTPEnv
-newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, xftpCredentials, httpCredentials} = do
+newXFTPServerEnv :: FileStoreClass s => XFTPStoreConfig s -> XFTPServerConfig -> IO (XFTPEnv s)
+newXFTPServerEnv storeCfg config@XFTPServerConfig {fileSizeQuota, xftpCredentials, httpCredentials} = do
   random <- C.newRandom
-  store <- newFileStore
-  storeLog <- mapM (`readWriteFileStore` store) storeLogFile
-  used <- countUsedStorage <$> readTVarIO (files store)
-  atomically $ writeTVar (usedStorage store) used
+  (store, storeLog) <- case storeCfg of
+    XSCMemory storeLogPath -> do
+      st <- newFileStore ()
+      sl <- mapM (`readWriteFileStore` st) storeLogPath
+      pure (st, sl)
+#if defined(dbServerPostgres)
+    XSCDatabase dbCfg -> do
+      st <- newFileStore dbCfg
+      pure (st, Nothing)
+#endif
+  used <- getUsedStorage store
+  usedStorage <- newTVarIO used
   forM_ fileSizeQuota $ \quota -> do
     logNote $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used)
     when (quota < used) $ logWarn "WARNING: storage quota is less than used storage, no files can be uploaded!"
   tlsServerCreds <- loadServerCredential xftpCredentials
   httpServerCreds <- mapM loadServerCredential httpCredentials
   Fingerprint fp <- loadFingerprint xftpCredentials
   serverStats <- newFileServerStats =<< getCurrentTime
-  pure XFTPEnv {config, store, storeLog, random, tlsServerCreds, httpServerCreds, serverIdentity = C.KeyHash fp, serverStats}
-
-countUsedStorage :: M.Map k FileRec -> Int64
-countUsedStorage = M.foldl' (\acc FileRec {fileInfo = FileInfo {size}} -> acc + fromIntegral size) 0
+  pure XFTPEnv {config, store, usedStorage, storeLog, random, tlsServerCreds, httpServerCreds, serverIdentity = C.KeyHash fp, serverStats}
 
 data XFTPRequest
   = XFTPReqNew FileInfo (NonEmpty RcvPublicAuthKey) (Maybe BasicAuth)
   | XFTPReqCmd XFTPFileId FileRec FileCmd
   | XFTPReqPing
+
+-- | Select and run the store config based on INI settings.
+-- CPP guards for Postgres are handled here so Main.hs stays CPP-free.
+runWithStoreConfig ::
+  Ini ->
+  String ->
+  Maybe FilePath ->
+  FilePath ->
+  MigrationConfirmation ->
+  (forall s. FileStoreClass s => XFTPStoreConfig s -> IO ()) ->
+  IO ()
+runWithStoreConfig _ini storeType storeLogFile_ _storeLogFilePath _confirmMigrations run = case storeType of
+  "memory" -> run $ XSCMemory storeLogFile_
+#if defined(dbServerPostgres)
+  "database" -> run $ XSCDatabase dbCfg
+    where
+      enableDbStoreLog' = settingIsOn "STORE_LOG" "db_store_log" _ini
+      dbStoreLogPath = enableDbStoreLog' $> _storeLogFilePath
+      dbCfg = PostgresFileStoreCfg {dbOpts = iniDBOptions _ini defaultXFTPDBOpts, dbStoreLogPath, confirmMigrations = _confirmMigrations}
+#else
+  "database" -> error "Error: server binary is compiled without support for PostgreSQL database.\nPlease re-compile with `cabal build -fserver_postgres`."
+#endif
+  _ -> error $ "Invalid store_files value: " <> storeType
+
+-- | Validate startup config when store_files=database.
+checkFileStoreMode :: Ini -> String -> FilePath -> IO ()
+#if defined(dbServerPostgres)
+checkFileStoreMode ini storeType storeLogFilePath = case storeType of
+  "database" -> do
+    storeLogExists <- doesFileExist storeLogFilePath
+    let dbStoreLogOn = settingIsOn "STORE_LOG" "db_store_log" ini
+    when (storeLogExists && isNothing_ dbStoreLogOn) $ do
+      putStrLn $ "Error: store log file " <> storeLogFilePath <> " exists but store_files is `database`."
+      putStrLn "Use `file-server database import` to migrate, or set `db_store_log: on`."
+      exitFailure
+  _ -> pure ()
+  where
+    isNothing_ Nothing = True
+    isNothing_ _ = False
+#else
+checkFileStoreMode _ _ _ = pure ()
+#endif
+
+-- | Import StoreLog to PostgreSQL database.
+importToDatabase :: FilePath -> Ini -> MigrationConfirmation -> IO ()
+#if defined(dbServerPostgres)
+importToDatabase storeLogFilePath ini _confirmMigrations = do
+  let dbCfg = PostgresFileStoreCfg {dbOpts = iniDBOptions ini defaultXFTPDBOpts, dbStoreLogPath = Nothing, confirmMigrations = _confirmMigrations}
+  importFileStore storeLogFilePath dbCfg
+#else
+importToDatabase _ _ _ = error "Error: server binary is compiled without support for PostgreSQL database.\nPlease re-compile with `cabal build -fserver_postgres`."
+#endif
+
+-- | Export PostgreSQL database to StoreLog.
+exportFromDatabase :: FilePath -> Ini -> MigrationConfirmation -> IO ()
+#if defined(dbServerPostgres)
+exportFromDatabase storeLogFilePath ini _confirmMigrations = do
+  let dbCfg = PostgresFileStoreCfg {dbOpts = iniDBOptions ini defaultXFTPDBOpts, dbStoreLogPath = Nothing, confirmMigrations = _confirmMigrations}
+  exportFileStore storeLogFilePath dbCfg
+#else
+exportFromDatabase _ _ _ = error "Error: server binary is compiled without support for PostgreSQL database.\nPlease re-compile with `cabal build -fserver_postgres`."
+#endif

src/Simplex/FileTransfer/Server/Main.hs

@@ -12,7 +12,7 @@ module Simplex.FileTransfer.Server.Main
     xftpServerCLI_,
   ) where
 
-import Control.Monad (when)
+import Control.Monad (unless, when)
 import Data.Either (fromRight)
 import Data.Functor (($>))
 import Data.Ini (lookupValue, readIniFile)
@@ -28,11 +28,12 @@ import Options.Applicative
 import Simplex.FileTransfer.Chunks
 import Simplex.FileTransfer.Description (FileSize (..))
 import Simplex.FileTransfer.Server (runXFTPServer)
-import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..), defFileExpirationHours, defaultFileExpiration, defaultInactiveClientExpiration)
+import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..), defFileExpirationHours, defaultFileExpiration, defaultInactiveClientExpiration, runWithStoreConfig, checkFileStoreMode, importToDatabase, exportFromDatabase)
 import Simplex.FileTransfer.Transport (alpnSupportedXFTPhandshakes, supportedFileServerVRange)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), pattern XFTPServer)
+import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
 import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information (ServerPublicInfo (..))
@@ -66,9 +67,13 @@ xftpServerCLI_ generateSite serveStaticFiles cfgPath logPath = do
       doesFileExist iniFile >>= \case
         True -> genOnline cfgPath certOpts
         _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
-    Start ->
+    Start opts ->
       doesFileExist iniFile >>= \case
-        True -> readIniFile iniFile >>= either exitError runServer
+        True -> readIniFile iniFile >>= either exitError (runServer opts)
+        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
+    Database cmd ->
+      doesFileExist iniFile >>= \case
+        True -> readIniFile iniFile >>= either exitError (runDatabaseCmd cmd)
         _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
     Delete -> do
       confirmOrExit
@@ -84,6 +89,21 @@ xftpServerCLI_ generateSite serveStaticFiles cfgPath logPath = do
     executableName = "file-server"
     storeLogFilePath = combine logPath "file-server-store.log"
     defaultStaticPath = combine logPath "www"
+    runDatabaseCmd cmd ini = case cmd of
+      SCImport -> do
+        storeLogExists <- doesFileExist storeLogFilePath
+        unless storeLogExists $ exitError $ "Error: store log file " <> storeLogFilePath <> " does not exist."
+        confirmOrExit
+          ("Import store log " <> storeLogFilePath <> " to PostgreSQL database?")
+          "Import cancelled."
+        importToDatabase storeLogFilePath ini MCYesUp
+      SCExport -> do
+        storeLogExists <- doesFileExist storeLogFilePath
+        when storeLogExists $ exitError $ "Error: store log file " <> storeLogFilePath <> " already exists."
+        confirmOrExit
+          ("Export PostgreSQL database to store log " <> storeLogFilePath <> "?")
+          "Export cancelled."
+        exportFromDatabase storeLogFilePath ini MCConsole
     initializeServer InitOptions {enableStoreLog, signAlgorithm, ip, fqdn, filesPath, fileSizeQuota, webStaticPath = webStaticPath_} = do
       clearDirIfExists cfgPath
       clearDirIfExists logPath
@@ -126,6 +146,14 @@ xftpServerCLI_ generateSite serveStaticFiles cfgPath logPath = do
           \# and restoring it when the server is started.\n\
           \# Log is compacted on start (deleted objects are removed).\n"
             <> ("enable: " <> onOff enableStoreLog <> "\n\n")
+            <> "# File storage mode: `memory` or `database` (PostgreSQL).\n\
+               \store_files: memory\n\n\
+               \# Database connection settings for PostgreSQL database (`store_files: database`).\n\
+               \# db_connection: postgresql://xftp@/xftp_server_store\n\
+               \# db_schema: xftp_server\n\
+               \# db_pool_size: 10\n\n\
+               \# Write database changes to store log file\n\
+               \# db_store_log: off\n\n"
             <> "# Expire files after the specified number of hours.\n"
             <> ("expire_files_hours: " <> tshow defFileExpirationHours <> "\n\n")
             <> "log_stats: off\n\
@@ -173,7 +201,7 @@ xftpServerCLI_ generateSite serveStaticFiles cfgPath logPath = do
                \# TLS credentials for HTTPS web server on the same port as XFTP.\n\
                \# cert: " <> T.pack (cfgPath `combine` "web.crt") <> "\n\
                \# key: " <> T.pack (cfgPath `combine` "web.key") <> "\n"
-    runServer ini = do
+    runServer StartOptions {confirmMigrations} ini = do
       hSetBuffering stdout LineBuffering
       hSetBuffering stderr LineBuffering
       fp <- checkSavedFingerprint cfgPath defaultX509Config
@@ -194,7 +222,10 @@ xftpServerCLI_ generateSite serveStaticFiles cfgPath logPath = do
           when (isJust webHttpPort || isJust webHttpsParams') $
             serveStaticFiles EmbeddedWebParams {webStaticPath = path, webHttpPort, webHttpsParams = webHttpsParams'}
         Nothing -> pure ()
-      runXFTPServer serverConfig
+      let storeType = fromRight "memory" $ T.unpack <$> lookupValue "STORE_LOG" "store_files" ini
+      checkFileStoreMode ini storeType storeLogFilePath
+      runWithStoreConfig ini storeType (storeLogFile serverConfig) storeLogFilePath confirmMigrations $
+        \storeCfg -> runXFTPServer storeCfg serverConfig
       where
         isOnion = \case THOnionHost _ -> True; _ -> False
         enableStoreLog = settingIsOn "STORE_LOG" "enable" ini
@@ -289,9 +320,16 @@ xftpServerCLI_ generateSite serveStaticFiles cfgPath logPath = do
 data CliCommand
   = Init InitOptions
   | OnlineCert CertOptions
-  | Start
+  | Start StartOptions
+  | Database StoreCmd
   | Delete
 
+data StoreCmd = SCImport | SCExport
+
+newtype StartOptions = StartOptions
+  { confirmMigrations :: MigrationConfirmation
+  }
+
 data InitOptions = InitOptions
   { enableStoreLog :: Bool,
     signAlgorithm :: SignAlgorithm,
@@ -308,7 +346,8 @@ cliCommandP cfgPath logPath iniFile =
   hsubparser
     ( command "init" (info (Init <$> initP) (progDesc $ "Initialize server - creates " <> cfgPath <> " and " <> logPath <> " directories and configuration files"))
         <> command "cert" (info (OnlineCert <$> certOptionsP) (progDesc $ "Generate new online TLS server credentials (configuration: " <> iniFile <> ")"))
-        <> command "start" (info (pure Start) (progDesc $ "Start server (configuration: " <> iniFile <> ")"))
+        <> command "start" (info (Start <$> startOptsP) (progDesc $ "Start server (configuration: " <> iniFile <> ")"))
+        <> command "database" (info (Database <$> storeCmdP) (progDesc "Import/export file store to/from PostgreSQL database"))
         <> command "delete" (info (pure Delete) (progDesc "Delete configuration and log files"))
     )
   where
@@ -375,3 +414,26 @@ cliCommandP cfgPath logPath iniFile =
               <> metavar "PATH"
           )
       pure InitOptions {enableStoreLog, signAlgorithm, ip, fqdn, filesPath, fileSizeQuota, webStaticPath}
+    startOptsP :: Parser StartOptions
+    startOptsP = do
+      confirmMigrations <-
+        option
+          parseConfirmMigrations
+          ( long "confirm-migrations"
+              <> metavar "CONFIRM_MIGRATIONS"
+              <> help "Confirm PostgreSQL database migration: up, down (default is manual confirmation)"
+              <> value MCConsole
+          )
+      pure StartOptions {confirmMigrations}
+      where
+        parseConfirmMigrations :: ReadM MigrationConfirmation
+        parseConfirmMigrations = eitherReader $ \case
+          "up" -> Right MCYesUp
+          "down" -> Right MCYesUpDown
+          _ -> Left "invalid migration confirmation, pass 'up' or 'down'"
+    storeCmdP :: Parser StoreCmd
+    storeCmdP =
+      hsubparser
+        ( command "import" (info (pure SCImport) (progDesc "Import store log file into PostgreSQL database"))
+            <> command "export" (info (pure SCExport) (progDesc "Export PostgreSQL database to store log file"))
+        )