dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-04-26 17:27:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

smp server: split postgres support to a separate executable, to not require postgres library in the main binary (#1482)

Browse Source 
* smp server: split postgres support to a separate executable, to not require postgres library in the main binary

* comments

* enable server_postgres flag by default, add CPP option to test

* refactor

* change default for server_postgres to False

* diff
This commit is contained in:
Evgeny
2025-03-16 11:37:25 +00:00
committed by GitHub
parent 1b5a9f3b0c
commit fe64d42db1
16 changed files with 425 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/build.yml vendored
View File

@@ -58,7 +58,7 @@ jobs:
- name: Build binaries
shell: bash
run: docker exec -t -e apps="$apps" builder sh -c 'cabal build --enable-tests && mkdir /out && for i in $apps; do bin=$(find /project/dist-newstyle -name "$i" -type f -executable); strip "$bin"; chmod +x "$bin"; mv "$bin" /out/; done'
run: docker exec -t -e apps="$apps" builder sh -c 'cabal build --enable-tests -fserver_postgres && mkdir /out && for i in $apps; do bin=$(find /project/dist-newstyle -name "$i" -type f -executable); strip "$bin"; chmod +x "$bin"; mv "$bin" /out/; done'
- name: Copy binaries from container and prepare them
if: startsWith(github.ref, 'refs/tags/v')

58
simplexmq.cabal
View File

@@ -72,6 +72,11 @@ flag client_postgres
manual: True
default: False
flag server_postgres
description: Build server with support of PostgreSQL.
manual: True
default: False
library
exposed-modules:
Simplex.FileTransfer.Agent
@@ -101,6 +106,7 @@ library
Simplex.Messaging.Agent.Store.Interface
Simplex.Messaging.Agent.Store.Migrations
Simplex.Messaging.Agent.Store.Migrations.App
Simplex.Messaging.Agent.Store.Postgres.Options
Simplex.Messaging.Agent.Store.Shared
Simplex.Messaging.Agent.TRcvQueues
Simplex.Messaging.Client
@@ -124,6 +130,7 @@ library
Simplex.Messaging.Parsers
Simplex.Messaging.Protocol
Simplex.Messaging.Server.Expiration
Simplex.Messaging.Server.QueueStore.Postgres.Config
Simplex.Messaging.Server.QueueStore.QueueInfo
Simplex.Messaging.ServiceScheme
Simplex.Messaging.Session
@@ -206,11 +213,6 @@ library
Simplex.FileTransfer.Server.Stats
Simplex.FileTransfer.Server.Store
Simplex.FileTransfer.Server.StoreLog
Simplex.Messaging.Agent.Store.Postgres
Simplex.Messaging.Agent.Store.Postgres.Common
Simplex.Messaging.Agent.Store.Postgres.DB
Simplex.Messaging.Agent.Store.Postgres.Migrations
Simplex.Messaging.Agent.Store.Postgres.Util
Simplex.Messaging.Notifications.Server
Simplex.Messaging.Notifications.Server.Control
Simplex.Messaging.Notifications.Server.Env
@@ -226,6 +228,7 @@ library
Simplex.Messaging.Server.Env.STM
Simplex.Messaging.Server.Information
Simplex.Messaging.Server.Main
Simplex.Messaging.Server.Main.Init
Simplex.Messaging.Server.MsgStore
Simplex.Messaging.Server.MsgStore.Journal
Simplex.Messaging.Server.MsgStore.Journal.SharedLock
@@ -235,14 +238,24 @@ library
Simplex.Messaging.Server.Prometheus
Simplex.Messaging.Server.QueueStore
Simplex.Messaging.Server.QueueStore.STM
Simplex.Messaging.Server.QueueStore.Postgres
Simplex.Messaging.Server.QueueStore.Postgres.Migrations
Simplex.Messaging.Server.QueueStore.Types
Simplex.Messaging.Server.Stats
Simplex.Messaging.Server.StoreLog
Simplex.Messaging.Server.StoreLog.ReadWrite
Simplex.Messaging.Server.StoreLog.Types
Simplex.Messaging.Transport.WebSockets
if flag(client_postgres) || flag(server_postgres)
exposed-modules:
Simplex.Messaging.Agent.Store.Postgres
Simplex.Messaging.Agent.Store.Postgres.Common
Simplex.Messaging.Agent.Store.Postgres.DB
Simplex.Messaging.Agent.Store.Postgres.Migrations
Simplex.Messaging.Agent.Store.Postgres.Util
if flag(server_postgres)
exposed-modules:
Simplex.Messaging.Server.QueueStore.Postgres
Simplex.Messaging.Server.QueueStore.Postgres.Migrations
other-modules:
Paths_simplexmq
hs-source-dirs:
@@ -307,21 +320,23 @@ library
case-insensitive ==1.2.*
, hashable ==1.4.*
, ini ==0.4.1
, postgresql-simple ==0.7.*
, optparse-applicative >=0.15 && <0.17
, process ==1.6.*
, raw-strings-qq ==1.1.*
, temporary ==1.3.*
, websockets ==0.12.*
if flag(client_postgres) || !flag(client_library)
if flag(client_postgres) || flag(server_postgres)
build-depends:
postgresql-libpq >=0.10.0.0
, postgresql-simple ==0.7.*
, raw-strings-qq ==1.1.*
if flag(client_postgres)
cpp-options: -DdbPostgres
else
build-depends:
direct-sqlcipher ==2.3.*
, sqlcipher-simple ==0.4.*
if flag(server_postgres)
cpp-options: -DdbServerPostgres
if impl(ghc >= 9.6.2)
build-depends:
bytestring ==0.11.*
@@ -439,7 +454,6 @@ test-suite simplexmq-test
CoreTests.UtilTests
CoreTests.VersionRangeTests
FileDescriptionTests
Fixtures
NtfClient
NtfServerTests
RemoteControl
@@ -455,7 +469,10 @@ test-suite simplexmq-test
Static
Static.Embedded
Paths_simplexmq
if !flag(client_postgres)
if flag(client_postgres)
other-modules:
Fixtures
else
other-modules:
AgentTests.SchemaDump
AgentTests.SQLiteTests
@@ -477,7 +494,6 @@ test-suite simplexmq-test
, crypton-x509
, crypton-x509-store
, crypton-x509-validation
, deepseq ==1.4.*
, directory
, file-embed
, filepath
@@ -491,11 +507,8 @@ test-suite simplexmq-test
, ini
, iso8601-time
, main-tester ==0.2.*
, memory
, mtl
, network
, postgresql-simple ==0.7.*
, process
, QuickCheck ==2.14.*
, random
, silently ==1.2.*
@@ -516,10 +529,15 @@ test-suite simplexmq-test
, yaml
default-language: Haskell2010
if flag(client_postgres)
build-depends:
postgresql-libpq >=0.10.0.0
, raw-strings-qq ==1.1.*
cpp-options: -DdbPostgres
else
build-depends:
sqlcipher-simple
deepseq ==1.4.*
, memory
, process
, sqlcipher-simple
if flag(client_postgres) || flag(server_postgres)
build-depends:
postgresql-simple ==0.7.*
if flag(server_postgres)
cpp-options: -DdbServerPostgres

10
src/Simplex/Messaging/Agent/Store/Postgres/Common.hs
View File

@@ -20,7 +20,7 @@ import Control.Concurrent.STM
import Control.Exception (bracket)
import Data.ByteString (ByteString)
import qualified Database.PostgreSQL.Simple as PSQL
import Numeric.Natural
import Simplex.Messaging.Agent.Store.Postgres.Options
-- TODO [postgres] use log_min_duration_statement instead of custom slow queries (SQLite's Connection type)
data DBStore = DBStore
@@ -35,14 +35,6 @@ data DBStore = DBStore
dbNew :: Bool
}
data DBOpts = DBOpts
{ connstr :: ByteString,
schema :: ByteString,
poolSize :: Natural,
createSchema :: Bool
}
deriving (Show)
withConnectionPriority :: DBStore -> Bool -> (PSQL.Connection -> IO a) -> IO a
withConnectionPriority DBStore {dbPool, dbSem} _priority =
bracket

12
src/Simplex/Messaging/Agent/Store/Postgres/Options.hs Normal file
View File

@@ -0,0 +1,12 @@
module Simplex.Messaging.Agent.Store.Postgres.Options where
import Data.ByteString (ByteString)
import Numeric.Natural
data DBOpts = DBOpts
{ connstr :: ByteString,
schema :: ByteString,
poolSize :: Natural,
createSchema :: Bool
}
deriving (Show)

4
src/Simplex/Messaging/Server/CLI.hs
View File

@@ -27,11 +27,11 @@ import qualified Data.X509.File as XF
import Data.X509.Validation (Fingerprint (..))
import Network.Socket (HostName, ServiceName)
import Options.Applicative
import Simplex.Messaging.Agent.Store.Postgres.Common (DBOpts (..))
import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), ProtocolServer (..), ProtocolTypeI)
import Simplex.Messaging.Server.Env.STM (AServerStoreCfg (..), ServerStoreCfg (..), StorePaths (..))
import Simplex.Messaging.Server.QueueStore.Postgres (PostgresStoreCfg (..))
import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
import Simplex.Messaging.Transport (ATransport (..), TLS, Transport (..))
import Simplex.Messaging.Transport.Server (AddHTTP, loadFileFingerprint)
import Simplex.Messaging.Transport.WebSockets (WS)

16
src/Simplex/Messaging/Server/Env/STM.hs
View File

@@ -56,7 +56,7 @@ import Simplex.Messaging.Server.MsgStore.STM
import Simplex.Messaging.Server.MsgStore.Types
import Simplex.Messaging.Server.NtfStore
import Simplex.Messaging.Server.QueueStore
import Simplex.Messaging.Server.QueueStore.Postgres (PostgresStoreCfg (..))
import Simplex.Messaging.Server.QueueStore.Postgres.Config
import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore, setStoreLog)
import Simplex.Messaging.Server.QueueStore.Types
import Simplex.Messaging.Server.Stats
@@ -324,7 +324,7 @@ newProhibitedSub = do
return Sub {subThread = ProhibitSub, delivered}
newEnv :: ServerConfig -> IO Env
newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines, startOptions} = do
newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines} = do
serverActive <- newTVarIO True
server <- newServer
msgStore <- case serverStoreCfg of
@@ -339,12 +339,16 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
ms <- newMsgStore cfg
loadStoreLog (mkQueue ms) storeLogFile $ stmQueueStore ms
pure $ AMS qt mt ms
#if defined(dbServerPostgres)
ASSCfg qt mt SSCDatabaseJournal {storeCfg, storeMsgsPath'} -> do
let StartOptions {confirmMigrations} = startOptions
let StartOptions {confirmMigrations} = startOptions config
qsCfg = PQStoreCfg (storeCfg {confirmMigrations} :: PostgresStoreCfg)
cfg = mkJournalStoreConfig qsCfg storeMsgsPath' msgQueueQuota maxJournalMsgCount maxJournalStateLines idleQueueInterval
ms <- newMsgStore cfg
pure $ AMS qt mt ms
#else
ASSCfg _ _ SSCDatabaseJournal {} -> noPostgresExit
#endif
ntfStore <- NtfStore <$> TM.emptyIO
random <- C.newRandom
tlsServerCreds <- getCredentials "SMP" smpCredentials
@@ -404,6 +408,12 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
_ -> SPMQueues
_ -> SPMMessages
noPostgresExit :: IO a
noPostgresExit = do
putStrLn "Error: server binary is compiled without support for PostgreSQL database."
putStrLn "Please download `smp-server-postgres` or re-compile with `cabal build -fserver_postgres`."
exitFailure
mkJournalStoreConfig :: QStoreCfg s -> FilePath -> Int -> Int -> Int -> Int64 -> JournalStoreConfig s
mkJournalStoreConfig queueStoreCfg storePath msgQueueQuota maxJournalMsgCount maxJournalStateLines idleQueueInterval =
JournalStoreConfig

274
src/Simplex/Messaging/Server/Main.hs
View File

@@ -1,4 +1,5 @@
{-# LANGUAGE ApplicativeDo #-}
{-# LANGUAGE CPP #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE GADTs #-}
@@ -25,21 +26,16 @@ import Data.Char (isAlpha, isAscii, toUpper)
import Data.Either (fromRight)
import Data.Functor (($>))
import Data.Ini (Ini, lookupValue, readIniFile)
import Data.Int (Int64)
import Data.List (find, isPrefixOf)
import qualified Data.List.NonEmpty as L
import Data.Maybe (fromMaybe, isJust, isNothing)
import Data.Semigroup (Sum (..))
import Data.Text (Text)
import qualified Data.Text as T
import Data.Text.Encoding (decodeLatin1, encodeUtf8)
import qualified Data.Text.IO as T
import Network.Socket (HostName)
import Numeric.Natural (Natural)
import Options.Applicative
import Simplex.Messaging.Agent.Protocol (connReqUriP')
import Simplex.Messaging.Agent.Store.Postgres (checkSchemaExists)
import Simplex.Messaging.Agent.Store.Postgres.Common (DBOpts (..))
import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
import Simplex.Messaging.Client (HostMode (..), NetworkConfig (..), ProtocolClientConfig (..), SocksMode (..), defaultNetworkConfig, textToHostMode)
import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..), defaultSMPClientAgentConfig)
@@ -52,22 +48,33 @@ import Simplex.Messaging.Server.CLI
import Simplex.Messaging.Server.Env.STM
import Simplex.Messaging.Server.Expiration
import Simplex.Messaging.Server.Information
import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore (..), JournalQueue, QStoreCfg (..), postgresQueueStore, stmQueueStore)
import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), QSType (..), SQSType (..), SMSType (..), newMsgStore)
import Simplex.Messaging.Server.QueueStore.Postgres (PostgresStoreCfg (..), batchInsertQueues, foldQueueRecs)
import Simplex.Messaging.Server.QueueStore.Types
import Simplex.Messaging.Server.StoreLog (logCreateQueue, openWriteStoreLog)
import Simplex.Messaging.Server.Main.Init
import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore (..), QStoreCfg (..), stmQueueStore)
import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), SQSType (..), SMSType (..), newMsgStore)
import Simplex.Messaging.Server.QueueStore.Postgres.Config
import Simplex.Messaging.Server.StoreLog.ReadWrite (readQueueStore)
import Simplex.Messaging.Transport (simplexMQVersion, supportedProxyClientSMPRelayVRange, supportedServerSMPRelayVRange)
import Simplex.Messaging.Transport.Client (SocksProxy, TransportHost (..), defaultSocksProxy)
import Simplex.Messaging.Transport.Client (TransportHost (..), defaultSocksProxy)
import Simplex.Messaging.Transport.Server (ServerCredentials (..), TransportServerConfig (..), defaultTransportServerConfig)
import Simplex.Messaging.Util (eitherToMaybe, ifM, safeDecodeUtf8, tshow)
import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist, renameFile)
import Simplex.Messaging.Util (eitherToMaybe, ifM)
import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist)
import System.Exit (exitFailure)
import System.FilePath (combine)
import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
import Text.Read (readMaybe)
#if defined(dbServerPostgres)
import Data.Semigroup (Sum (..))
import Simplex.Messaging.Agent.Store.Postgres (checkSchemaExists)
import Simplex.Messaging.Server.MsgStore.Journal (JournalQueue)
import Simplex.Messaging.Server.MsgStore.Types (QSType (..))
import Simplex.Messaging.Server.MsgStore.Journal (postgresQueueStore)
import Simplex.Messaging.Server.QueueStore.Postgres (batchInsertQueues, foldQueueRecs)
import Simplex.Messaging.Server.QueueStore.Types
import Simplex.Messaging.Server.StoreLog (logCreateQueue, openWriteStoreLog)
import System.Directory (renameFile)
#endif
smpServerCLI :: FilePath -> FilePath -> IO ()
smpServerCLI = smpServerCLI_ (\_ _ _ -> pure ()) (\_ -> pure ()) (\_ -> error "attachStaticFiles not available")
@@ -129,14 +136,20 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
("WARNING: journal directory " <> storeMsgsJournalDir <> " will be exported to message log file " <> storeMsgsFilePath)
"Journal not exported"
ms <- newJournalMsgStore MQStoreCfg
-- TODO [postgres] in case postgres configured, queues must be read from database
readQueueStore True (mkQueue ms) storeLogFile $ stmQueueStore ms
exportMessages True ms storeMsgsFilePath False
putStrLn "Export completed"
putStrLn $ case readStoreType ini of
Right (ASType SQSMemory SMSMemory) -> "store_messages set to `memory`, start the server."
Right (ASType SQSMemory SMSJournal) -> "store_messages set to `journal`, update it to `memory` in INI file"
Right (ASType SQSPostgres SMSJournal) -> "store_messages set to `journal`, store_queues is set to `database`.\nExport queues to store log to use memory storage for messages (`smp-server database export`)."
Left e -> e <> ", configure storage correctly"
case readStoreType ini of
Right (ASType SQSMemory SMSMemory) -> putStrLn "store_messages set to `memory`, start the server."
Right (ASType SQSMemory SMSJournal) -> putStrLn "store_messages set to `journal`, update it to `memory` in INI file"
Right (ASType SQSPostgres SMSJournal) ->
#if defined(dbServerPostgres)
putStrLn "store_messages set to `journal`, store_queues is set to `database`.\nExport queues to store log to use memory storage for messages (`smp-server database export`)."
#else
noPostgresExit
#endif
Left e -> putStrLn $ e <> ", configure storage correctly"
SCDelete
| not msgsDirExists -> do
putStrLn $ storeMsgsJournalDir <> " directory does not exists."
@@ -147,6 +160,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
"Messages NOT deleted"
deleteDirIfExists storeMsgsJournalDir
putStrLn $ "Deleted all messages in journal " <> storeMsgsJournalDir
#if defined(dbServerPostgres)
Database cmd dbOpts@DBOpts {connstr, schema} -> withIniFile $ \ini -> do
schemaExists <- checkSchemaExists connstr schema
storeLogExists <- doesFileExist storeLogFilePath
@@ -205,6 +219,9 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
| otherwise -> do
putStrLn $ "Open database: psql " <> B.unpack connstr
putStrLn $ "Delete schema: DROP SCHEMA " <> B.unpack schema <> " CASCADE;"
#else
Database {} -> noPostgresExit
#endif
where
withIniFile a =
doesFileExist iniFile >>= \case
@@ -224,7 +241,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
in newMsgStore cfg
iniFile = combine cfgPath "smp-server.ini"
serverVersion = "SMP server v" <> simplexMQVersion
defaultServerPorts = "5223,443"
executableName = "smp-server"
storeLogFilePath = combine logPath "smp-server-store.log"
storeMsgsFilePath = combine logPath "smp-server-messages.log"
@@ -240,7 +256,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
where
iniStoreQueues = fromRight "memory" $ lookupValue "STORE_LOG" "store_queues" ini
iniStoreMessage = fromRight "memory" $ lookupValue "STORE_LOG" "store_messages" ini
iniDBOptions :: Ini -> DBOpts
iniDBOptions ini =
DBOpts
{ connstr = either (const defaultDBConnStr) encodeUtf8 $ lookupValue "STORE_LOG" "db_connection" ini,
@@ -248,20 +263,14 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
poolSize = readIniDefault defaultDBPoolSize "STORE_LOG" "db_pool_size" ini,
createSchema = False
}
dbOptsIniContent :: DBOpts -> Text
dbOptsIniContent DBOpts {connstr, schema, poolSize} =
(optDisabled' (connstr == defaultDBConnStr) <> "db_connection: " <> safeDecodeUtf8 connstr <> "\n")
<> (optDisabled' (schema == defaultDBSchema) <> "db_schema: " <> safeDecodeUtf8 schema <> "\n")
<> (optDisabled' (poolSize == defaultDBPoolSize) <> "db_pool_size: " <> tshow poolSize <> "\n\n")
iniDeletedTTL ini = readIniDefault (86400 * defaultDeletedTTL) "STORE_LOG" "db_deleted_ttl" ini
httpsCertFile = combine cfgPath "web.crt"
httpsKeyFile = combine cfgPath "web.key"
defaultStaticPath = combine logPath "www"
enableStoreLog' = settingIsOn "STORE_LOG" "enable"
enableDbStoreLog' = settingIsOn "STORE_LOG" "db_store_log"
initializeServer opts@InitOptions {ip, fqdn, sourceCode = src', webStaticPath = sp', disableWeb = noWeb', scripted}
| scripted = initialize opts
initializeServer opts
| scripted opts = initialize opts
| otherwise = do
let InitOptions {ip, fqdn, sourceCode = src', webStaticPath = sp', disableWeb = noWeb'} = opts
putStrLn "Use `smp-server init -h` for available options."
checkInitOptions opts
void $ withPrompt "SMP server will be initialized (press Enter)" getLine
@@ -303,7 +312,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
Just "Error: passing --hosting-country requires passing --hosting"
| otherwise = Nothing
forM_ err_ $ \err -> putStrLn err >> exitFailure
initialize opts'@InitOptions {enableStoreLog, dbOptions, logStats, signAlgorithm, password, controlPort, socksProxy, ownDomains, sourceCode, webStaticPath, disableWeb} = do
initialize opts'@InitOptions {ip, fqdn, signAlgorithm, password, controlPort, sourceCode} = do
checkInitOptions opts'
clearDirIfExists cfgPath
clearDirIfExists logPath
@@ -315,7 +324,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
controlPortPwds <- forM controlPort $ \_ -> let pwd = decodeLatin1 <$> randomBase64 18 in (,) <$> pwd <*> pwd
let host = fromMaybe (if ip == "127.0.0.1" then "<hostnames>" else ip) fqdn
srv = ProtoServerWithAuth (SMPServer [THDomainName host] "" (C.KeyHash fp)) basicAuth
T.writeFile iniFile $ iniFileContent host basicAuth controlPortPwds
T.writeFile iniFile $ iniFileContent cfgPath logPath opts' host basicAuth controlPortPwds
putStrLn $ "Server initialized, please provide additional server information in " <> iniFile <> "."
putStrLn $ "Run `" <> executableName <> " start` to start server."
warnCAPrivateKeyFile cfgPath x509cfg
@@ -326,103 +335,6 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
ServerPassword s -> pure s
SPRandom -> BasicAuth <$> randomBase64 32
randomBase64 n = strEncode <$> (atomically . C.randomBytes n =<< C.newRandom)
iniFileContent host basicAuth controlPortPwds =
informationIniContent opts'
<> "[STORE_LOG]\n\
\# The server uses memory or PostgreSQL database for persisting queue records.\n\
\# Use `enable: on` to use append-only log to preserve and restore queue records on restart.\n\
\# Log is compacted on start (deleted objects are removed).\n"
<> ("enable: " <> onOff enableStoreLog <> "\n\n")
<> "# Queue storage mode: `memory` or `database` (to store queue records in PostgreSQL database).\n\
\# `memory` - in-memory persistence, with optional append-only log (`enable: on`).\n\
\# `database`- PostgreSQL databass (requires `store_messages: journal`).\n\
\store_queues: memory\n\n\
\# Database connection settings for PostgreSQL database (`store_queues: database`).\n"
<> dbOptsIniContent dbOptions
<> "# Write database changes to store log file\n\
\# db_store_log: off\n\n\
\# Time to retain deleted queues in the database, days.\n"
<> ("db_deleted_ttl: " <> tshow defaultDeletedTTL <> "\n\n")
<> "# Message storage mode: `memory` or `journal`.\n\
\store_messages: memory\n\n\
\# When store_messages is `memory`, undelivered messages are optionally saved and restored\n\
\# when the server restarts, they are preserved in the .bak file until the next restart.\n"
<> ("restore_messages: " <> onOff enableStoreLog <> "\n\n")
<> "# Messages and notifications expiration periods.\n"
<> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n")
<> "expire_messages_on_start: on\n"
<> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
<> "# Log daily server statistics to CSV file\n"
<> ("log_stats: " <> onOff logStats <> "\n\n")
<> "# Log interval for real-time Prometheus metrics\n\
\# prometheus_interval: 300\n\n\
\[AUTH]\n\
\# Set new_queues option to off to completely prohibit creating new messaging queues.\n\
\# This can be useful when you want to decommission the server, but not all connections are switched yet.\n\
\new_queues: on\n\n\
\# Use create_password option to enable basic auth to create new messaging queues.\n\
\# The password should be used as part of server address in client configuration:\n\
\# smp://fingerprint:password@host1,host2\n\
\# The password will not be shared with the connecting contacts, you must share it only\n\
\# with the users who you want to allow creating messaging queues on your server.\n"
<> ( let noPassword = "password to create new queues and forward messages (any printable ASCII characters without whitespace, '@', ':' and '/')"
in optDisabled basicAuth <> "create_password: " <> maybe noPassword (safeDecodeUtf8 . strEncode) basicAuth
)
<> "\n\n"
<> (optDisabled controlPortPwds <> "control_port_admin_password: " <> maybe "" fst controlPortPwds <> "\n")
<> (optDisabled controlPortPwds <> "control_port_user_password: " <> maybe "" snd controlPortPwds <> "\n")
<> "\n\
\[TRANSPORT]\n\
\# Host is only used to print server address on start.\n\
\# You can specify multiple server ports.\n"
<> ("host: " <> T.pack host <> "\n")
<> ("port: " <> T.pack defaultServerPorts <> "\n")
<> "log_tls_errors: off\n\n\
\# Use `websockets: 443` to run websockets server in addition to plain TLS.\n\
\# This option is deprecated and should be used for testing only.\n\
\# , port 443 should be specified in port above\n\
\websockets: off\n"
<> (optDisabled controlPort <> "control_port: " <> tshow (fromMaybe defaultControlPort controlPort))
<> "\n\n\
\[PROXY]\n\
\# Network configuration for SMP proxy client.\n\
\# `host_mode` can be 'public' (default) or 'onion'.\n\
\# It defines prefferred hostname for destination servers with multiple hostnames.\n\
\# host_mode: public\n\
\# required_host_mode: off\n\n\
\# The domain suffixes of the relays you operate (space-separated) to count as separate proxy statistics.\n"
<> (optDisabled ownDomains <> "own_server_domains: " <> maybe "" (safeDecodeUtf8 . strEncode) ownDomains)
<> "\n\n\
\# SOCKS proxy port for forwarding messages to destination servers.\n\
\# You may need a separate instance of SOCKS proxy for incoming single-hop requests.\n"
<> (optDisabled socksProxy <> "socks_proxy: " <> maybe "localhost:9050" (safeDecodeUtf8 . strEncode) socksProxy)
<> "\n\n\
\# `socks_mode` can be 'onion' for SOCKS proxy to be used for .onion destination hosts only (default)\n\
\# or 'always' to be used for all destination hosts (can be used if it is an .onion server).\n\
\# socks_mode: onion\n\n\
\# Limit number of threads a client can spawn to process proxy commands in parrallel.\n"
<> ("# client_concurrency: " <> tshow defaultProxyClientConcurrency)
<> "\n\n\
\[INACTIVE_CLIENTS]\n\
\# TTL and interval to check inactive clients\n\
\disconnect: on\n"
<> ("ttl: " <> tshow (ttl defaultInactiveClientExpiration) <> "\n")
<> ("check_interval: " <> tshow (checkInterval defaultInactiveClientExpiration))
<> "\n\n\
\[WEB]\n\
\# Set path to generate static mini-site for server information and qr codes/links\n"
<> ("static_path: " <> T.pack (fromMaybe defaultStaticPath webStaticPath) <> "\n\n")
<> "# Run an embedded server on this port\n\
\# Onion sites can use any port and register it in the hidden service config.\n\
\# Running on a port 80 may require setting process capabilities.\n\
\# http: 8000\n\n\
\# You can run an embedded TLS web server too if you provide port and cert and key files.\n\
\# Not required for running relay on onion address.\n"
<> (webDisabled <> "https: 443\n")
<> (webDisabled <> "cert: " <> T.pack httpsCertFile <> "\n")
<> (webDisabled <> "key: " <> T.pack httpsKeyFile <> "\n")
where
webDisabled = if disableWeb then "# " else ""
runServer startOptions ini = do
hSetBuffering stdout LineBuffering
hSetBuffering stderr LineBuffering
@@ -607,6 +519,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
| otherwise -> case qs of
SQSMemory ->
unless (storeLogExists) $ putStrLn $ "store_queues is `memory`, " <> storeLogFilePath <> " file will be created."
#if defined(dbServerPostgres)
SQSPostgres -> do
let DBOpts {connstr, schema} = iniDBOptions ini
schemaExists <- checkSchemaExists connstr schema
@@ -629,6 +542,9 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
noDatabaseSchema connstr schema = do
putStrLn $ "Error: store_queues is `database`, create schema " <> B.unpack schema <> " in PostgreSQL database " <> B.unpack connstr
exitFailure
#else
SQSPostgres -> noPostgresExit
#endif
ASType SQSMemory SMSMemory
| msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
| msgsDirExists -> do
@@ -642,10 +558,12 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
putStrLn "Configure memory storage."
exitFailure
#if defined(dbServerPostgres)
exitConfigureQueueStore connstr schema = do
putStrLn $ "Error: both " <> storeLogFilePath <> " file and " <> B.unpack schema <> " schema are present (database: " <> B.unpack connstr <> ")."
putStrLn "Configure queue storage."
exitFailure
#endif
data EmbeddedWebParams = EmbeddedWebParams
{ webStaticPath :: FilePath,
@@ -669,72 +587,6 @@ getServerSourceCode =
simplexmqSource :: String
simplexmqSource = "https://github.com/simplex-chat/simplexmq"
defaultDBConnStr :: ByteString
defaultDBConnStr = "postgresql://smp@/smp_server_store"
defaultDBSchema :: ByteString
defaultDBSchema = "smp_server"
defaultDBPoolSize :: Natural
defaultDBPoolSize = 10
-- time to retain deleted queues in the database (days), for debugging
defaultDeletedTTL :: Int64
defaultDeletedTTL = 21
defaultControlPort :: Int
defaultControlPort = 5224
informationIniContent :: InitOptions -> Text
informationIniContent InitOptions {sourceCode, serverInfo} =
"[INFORMATION]\n\
\# AGPLv3 license requires that you make any source code modifications\n\
\# available to the end users of the server.\n\
\# LICENSE: https://github.com/simplex-chat/simplexmq/blob/stable/LICENSE\n\
\# Include correct source code URI in case the server source code is modified in any way.\n\
\# If any other information fields are present, source code property also MUST be present.\n\n"
<> (optDisabled sourceCode <> "source_code: " <> fromMaybe "URI" sourceCode)
<> "\n\n\
\# Declaring all below information is optional, any of these fields can be omitted.\n\
\\n\
\# Server usage conditions and amendments.\n\
\# It is recommended to use standard conditions with any amendments in a separate document.\n\
\# usage_conditions: https://github.com/simplex-chat/simplex-chat/blob/stable/PRIVACY.md\n\
\# condition_amendments: link\n\
\\n\
\# Server location and operator.\n"
<> countryStr "server" serverCountry
<> enitiyStrs "operator" operator
<> (optDisabled website <> "website: " <> fromMaybe "" website)
<> "\n\n\
\# Administrative contacts.\n\
\# admin_simplex: SimpleX address\n\
\# admin_email:\n\
\# admin_pgp:\n\
\# admin_pgp_fingerprint:\n\
\\n\
\# Contacts for complaints and feedback.\n\
\# complaints_simplex: SimpleX address\n\
\# complaints_email:\n\
\# complaints_pgp:\n\
\# complaints_pgp_fingerprint:\n\
\\n\
\# Hosting provider.\n"
<> enitiyStrs "hosting" hosting
<> "\n\
\# Hosting type can be `virtual`, `dedicated`, `colocation`, `owned`\n"
<> ("hosting_type: " <> maybe "virtual" (decodeLatin1 . strEncode) hostingType <> "\n\n")
where
ServerPublicInfo {operator, website, hosting, hostingType, serverCountry} = serverInfo
countryStr optName country = optDisabled country <> optName <> "_country: " <> fromMaybe "ISO-3166 2-letter code" country <> "\n"
enitiyStrs optName entity =
optDisabled entity
<> optName
<> ": "
<> maybe "entity (organization or person name)" name entity
<> "\n"
<> countryStr optName (country =<< entity)
serverPublicInfo :: Ini -> Maybe ServerPublicInfo
serverPublicInfo ini = serverInfo <$!> infoValue "source_code"
where
@@ -767,14 +619,6 @@ serverPublicInfo ini = serverInfo <$!> infoValue "source_code"
(Nothing, Nothing, _, Nothing) -> Nothing
(_, _, pkURI, pkFingerprint) -> Just ServerContactAddress {simplex, email, pgp = PGPKey <$> pkURI <*> pkFingerprint}
optDisabled :: Maybe a -> Text
optDisabled = optDisabled' . isNothing
{-# INLINE optDisabled #-}
optDisabled' :: Bool -> Text
optDisabled' cond = if cond then "# " else ""
{-# INLINE optDisabled' #-}
validCountryValue :: String -> String -> Either String Text
validCountryValue field s
| length s == 2 && all (\c -> isAscii c && isAlpha c) s = Right $ T.pack $ map toUpper s
@@ -797,30 +641,6 @@ data CliCommand
data StoreCmd = SCImport | SCExport | SCDelete
data InitOptions = InitOptions
{ enableStoreLog :: Bool,
dbOptions :: DBOpts,
logStats :: Bool,
signAlgorithm :: SignAlgorithm,
ip :: HostName,
fqdn :: Maybe HostName,
password :: Maybe ServerPassword,
controlPort :: Maybe Int,
socksProxy :: Maybe SocksProxy,
ownDomains :: Maybe (L.NonEmpty TransportHost),
sourceCode :: Maybe Text,
serverInfo :: ServerPublicInfo,
operatorCountry :: Maybe Text,
hostingCountry :: Maybe Text,
webStaticPath :: Maybe FilePath,
disableWeb :: Bool,
scripted :: Bool
}
deriving (Show)
data ServerPassword = ServerPassword BasicAuth | SPRandom
deriving (Show)
cliCommandP :: FilePath -> FilePath -> FilePath -> Parser CliCommand
cliCommandP cfgPath logPath iniFile =
hsubparser

230
src/Simplex/Messaging/Server/Main/Init.hs Normal file
View File

@@ -0,0 +1,230 @@
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
module Simplex.Messaging.Server.Main.Init where
import Data.ByteString.Char8 (ByteString)
import Data.Int (Int64)
import qualified Data.List.NonEmpty as L
import Data.Maybe (fromMaybe, isNothing)
import Numeric.Natural (Natural)
import Data.Text (Text)
import qualified Data.Text as T
import Data.Text.Encoding (decodeLatin1)
import Network.Socket (HostName)
import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol (BasicAuth)
import Simplex.Messaging.Server.CLI (SignAlgorithm, onOff)
import Simplex.Messaging.Server.Env.STM
import Simplex.Messaging.Server.Expiration (ExpirationConfig (..))
import Simplex.Messaging.Server.Information (Entity (..), ServerPublicInfo (..))
import Simplex.Messaging.Transport.Client (SocksProxy, TransportHost)
import Simplex.Messaging.Util (safeDecodeUtf8, tshow)
import System.FilePath ((</>))
defaultControlPort :: Int
defaultControlPort = 5224
defaultDBConnStr :: ByteString
defaultDBConnStr = "postgresql://smp@/smp_server_store"
defaultDBSchema :: ByteString
defaultDBSchema = "smp_server"
defaultDBPoolSize :: Natural
defaultDBPoolSize = 10
-- time to retain deleted queues in the database (days), for debugging
defaultDeletedTTL :: Int64
defaultDeletedTTL = 21
data InitOptions = InitOptions
{ enableStoreLog :: Bool,
dbOptions :: DBOpts,
logStats :: Bool,
signAlgorithm :: SignAlgorithm,
ip :: HostName,
fqdn :: Maybe HostName,
password :: Maybe ServerPassword,
controlPort :: Maybe Int,
socksProxy :: Maybe SocksProxy,
ownDomains :: Maybe (L.NonEmpty TransportHost),
sourceCode :: Maybe Text,
serverInfo :: ServerPublicInfo,
operatorCountry :: Maybe Text,
hostingCountry :: Maybe Text,
webStaticPath :: Maybe FilePath,
disableWeb :: Bool,
scripted :: Bool
}
deriving (Show)
data ServerPassword = ServerPassword BasicAuth | SPRandom
deriving (Show)
iniFileContent :: FilePath -> FilePath -> InitOptions -> HostName -> Maybe BasicAuth -> Maybe (Text, Text) -> Text
iniFileContent cfgPath logPath opts host basicAuth controlPortPwds =
informationIniContent opts
<> "[STORE_LOG]\n\
\# The server uses memory or PostgreSQL database for persisting queue records.\n\
\# Use `enable: on` to use append-only log to preserve and restore queue records on restart.\n\
\# Log is compacted on start (deleted objects are removed).\n"
<> ("enable: " <> onOff enableStoreLog <> "\n\n")
<> "# Queue storage mode: `memory` or `database` (to store queue records in PostgreSQL database).\n\
\# `memory` - in-memory persistence, with optional append-only log (`enable: on`).\n\
\# `database`- PostgreSQL databass (requires `store_messages: journal`).\n\
\store_queues: memory\n\n\
\# Database connection settings for PostgreSQL database (`store_queues: database`).\n"
<> (optDisabled' (connstr == defaultDBConnStr) <> "db_connection: " <> safeDecodeUtf8 connstr <> "\n")
<> (optDisabled' (schema == defaultDBSchema) <> "db_schema: " <> safeDecodeUtf8 schema <> "\n")
<> (optDisabled' (poolSize == defaultDBPoolSize) <> "db_pool_size: " <> tshow poolSize <> "\n\n")
<> "# Write database changes to store log file\n\
\# db_store_log: off\n\n\
\# Time to retain deleted queues in the database, days.\n"
<> ("db_deleted_ttl: " <> tshow defaultDeletedTTL <> "\n\n")
<> "# Message storage mode: `memory` or `journal`.\n\
\store_messages: memory\n\n\
\# When store_messages is `memory`, undelivered messages are optionally saved and restored\n\
\# when the server restarts, they are preserved in the .bak file until the next restart.\n"
<> ("restore_messages: " <> onOff enableStoreLog <> "\n\n")
<> "# Messages and notifications expiration periods.\n"
<> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n")
<> "expire_messages_on_start: on\n"
<> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
<> "# Log daily server statistics to CSV file\n"
<> ("log_stats: " <> onOff logStats <> "\n\n")
<> "# Log interval for real-time Prometheus metrics\n\
\# prometheus_interval: 300\n\n\
\[AUTH]\n\
\# Set new_queues option to off to completely prohibit creating new messaging queues.\n\
\# This can be useful when you want to decommission the server, but not all connections are switched yet.\n\
\new_queues: on\n\n\
\# Use create_password option to enable basic auth to create new messaging queues.\n\
\# The password should be used as part of server address in client configuration:\n\
\# smp://fingerprint:password@host1,host2\n\
\# The password will not be shared with the connecting contacts, you must share it only\n\
\# with the users who you want to allow creating messaging queues on your server.\n"
<> ( let noPassword = "password to create new queues and forward messages (any printable ASCII characters without whitespace, '@', ':' and '/')"
in optDisabled basicAuth <> "create_password: " <> maybe noPassword (safeDecodeUtf8 . strEncode) basicAuth
)
<> "\n\n"
<> (optDisabled controlPortPwds <> "control_port_admin_password: " <> maybe "" fst controlPortPwds <> "\n")
<> (optDisabled controlPortPwds <> "control_port_user_password: " <> maybe "" snd controlPortPwds <> "\n")
<> "\n\
\[TRANSPORT]\n\
\# Host is only used to print server address on start.\n\
\# You can specify multiple server ports.\n"
<> ("host: " <> T.pack host <> "\n")
<> ("port: " <> defaultServerPorts <> "\n")
<> "log_tls_errors: off\n\n\
\# Use `websockets: 443` to run websockets server in addition to plain TLS.\n\
\# This option is deprecated and should be used for testing only.\n\
\# , port 443 should be specified in port above\n\
\websockets: off\n"
<> (optDisabled controlPort <> "control_port: " <> tshow (fromMaybe defaultControlPort controlPort))
<> "\n\n\
\[PROXY]\n\
\# Network configuration for SMP proxy client.\n\
\# `host_mode` can be 'public' (default) or 'onion'.\n\
\# It defines prefferred hostname for destination servers with multiple hostnames.\n\
\# host_mode: public\n\
\# required_host_mode: off\n\n\
\# The domain suffixes of the relays you operate (space-separated) to count as separate proxy statistics.\n"
<> (optDisabled ownDomains <> "own_server_domains: " <> maybe "" (safeDecodeUtf8 . strEncode) ownDomains)
<> "\n\n\
\# SOCKS proxy port for forwarding messages to destination servers.\n\
\# You may need a separate instance of SOCKS proxy for incoming single-hop requests.\n"
<> (optDisabled socksProxy <> "socks_proxy: " <> maybe "localhost:9050" (safeDecodeUtf8 . strEncode) socksProxy)
<> "\n\n\
\# `socks_mode` can be 'onion' for SOCKS proxy to be used for .onion destination hosts only (default)\n\
\# or 'always' to be used for all destination hosts (can be used if it is an .onion server).\n\
\# socks_mode: onion\n\n\
\# Limit number of threads a client can spawn to process proxy commands in parrallel.\n"
<> ("# client_concurrency: " <> tshow defaultProxyClientConcurrency)
<> "\n\n\
\[INACTIVE_CLIENTS]\n\
\# TTL and interval to check inactive clients\n\
\disconnect: on\n"
<> ("ttl: " <> tshow (ttl defaultInactiveClientExpiration) <> "\n")
<> ("check_interval: " <> tshow (checkInterval defaultInactiveClientExpiration))
<> "\n\n\
\[WEB]\n\
\# Set path to generate static mini-site for server information and qr codes/links\n"
<> ("static_path: " <> T.pack (fromMaybe defaultStaticPath webStaticPath) <> "\n\n")
<> "# Run an embedded server on this port\n\
\# Onion sites can use any port and register it in the hidden service config.\n\
\# Running on a port 80 may require setting process capabilities.\n\
\# http: 8000\n\n\
\# You can run an embedded TLS web server too if you provide port and cert and key files.\n\
\# Not required for running relay on onion address.\n"
<> (webDisabled <> "https: 443\n")
<> (webDisabled <> "cert: " <> T.pack httpsCertFile <> "\n")
<> (webDisabled <> "key: " <> T.pack httpsKeyFile <> "\n")
where
InitOptions {enableStoreLog, dbOptions, socksProxy, ownDomains, controlPort, webStaticPath, disableWeb, logStats} = opts
DBOpts {connstr, schema, poolSize} = dbOptions
defaultServerPorts = "5223,443"
defaultStaticPath = logPath </> "www"
httpsCertFile = cfgPath </> "web.crt"
httpsKeyFile = cfgPath </> "web.key"
webDisabled = if disableWeb then "# " else ""
informationIniContent :: InitOptions -> Text
informationIniContent InitOptions {sourceCode, serverInfo} =
"[INFORMATION]\n\
\# AGPLv3 license requires that you make any source code modifications\n\
\# available to the end users of the server.\n\
\# LICENSE: https://github.com/simplex-chat/simplexmq/blob/stable/LICENSE\n\
\# Include correct source code URI in case the server source code is modified in any way.\n\
\# If any other information fields are present, source code property also MUST be present.\n\n"
<> (optDisabled sourceCode <> "source_code: " <> fromMaybe "URI" sourceCode)
<> "\n\n\
\# Declaring all below information is optional, any of these fields can be omitted.\n\
\\n\
\# Server usage conditions and amendments.\n\
\# It is recommended to use standard conditions with any amendments in a separate document.\n\
\# usage_conditions: https://github.com/simplex-chat/simplex-chat/blob/stable/PRIVACY.md\n\
\# condition_amendments: link\n\
\\n\
\# Server location and operator.\n"
<> countryStr "server" serverCountry
<> enitiyStrs "operator" operator
<> (optDisabled website <> "website: " <> fromMaybe "" website)
<> "\n\n\
\# Administrative contacts.\n\
\# admin_simplex: SimpleX address\n\
\# admin_email:\n\
\# admin_pgp:\n\
\# admin_pgp_fingerprint:\n\
\\n\
\# Contacts for complaints and feedback.\n\
\# complaints_simplex: SimpleX address\n\
\# complaints_email:\n\
\# complaints_pgp:\n\
\# complaints_pgp_fingerprint:\n\
\\n\
\# Hosting provider.\n"
<> enitiyStrs "hosting" hosting
<> "\n\
\# Hosting type can be `virtual`, `dedicated`, `colocation`, `owned`\n"
<> ("hosting_type: " <> maybe "virtual" (decodeLatin1 . strEncode) hostingType <> "\n\n")
where
ServerPublicInfo {operator, website, hosting, hostingType, serverCountry} = serverInfo
countryStr optName country = optDisabled country <> optName <> "_country: " <> fromMaybe "ISO-3166 2-letter code" country <> "\n"
enitiyStrs optName entity =
optDisabled entity
<> optName
<> ": "
<> maybe "entity (organization or person name)" name entity
<> "\n"
<> countryStr optName (country =<< entity)
optDisabled :: Maybe a -> Text
optDisabled = optDisabled' . isNothing
{-# INLINE optDisabled #-}
optDisabled' :: Bool -> Text
optDisabled' cond = if cond then "# " else ""
{-# INLINE optDisabled' #-}

28
src/Simplex/Messaging/Server/MsgStore/Journal.hs
View File

@@ -1,4 +1,5 @@
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE CPP #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE DuplicateRecordFields #-}
@@ -43,7 +44,9 @@ module Simplex.Messaging.Server.MsgStore.Journal
journalFilePath,
logFileExt,
stmQueueStore,
#if defined(dbServerPostgres)
postgresQueueStore,
#endif
)
where
@@ -73,7 +76,9 @@ import Simplex.Messaging.Protocol
import Simplex.Messaging.Server.MsgStore.Journal.SharedLock
import Simplex.Messaging.Server.MsgStore.Types
import Simplex.Messaging.Server.QueueStore
#if defined(dbServerPostgres)
import Simplex.Messaging.Server.QueueStore.Postgres
#endif
import Simplex.Messaging.Server.QueueStore.STM
import Simplex.Messaging.Server.QueueStore.Types
import Simplex.Messaging.TMap (TMap)
@@ -96,25 +101,33 @@ data JournalMsgStore s = JournalMsgStore
data QStore (s :: QSType) where
MQStore :: QStoreType 'QSMemory -> QStore 'QSMemory
#if defined(dbServerPostgres)
PQStore :: QStoreType 'QSPostgres -> QStore 'QSPostgres
#endif
type family QStoreType s where
QStoreType 'QSMemory = STMQueueStore (JournalQueue 'QSMemory)
#if defined(dbServerPostgres)
QStoreType 'QSPostgres = PostgresQueueStore (JournalQueue 'QSPostgres)
#endif
withQS :: (QueueStoreClass (JournalQueue s) (QStoreType s) => QStoreType s -> r) -> QStore s -> r
withQS f = \case
MQStore st -> f st
#if defined(dbServerPostgres)
PQStore st -> f st
#endif
{-# INLINE withQS #-}
stmQueueStore :: JournalMsgStore 'QSMemory -> STMQueueStore (JournalQueue 'QSMemory)
stmQueueStore st = case queueStore_ st of
MQStore st' -> st'
#if defined(dbServerPostgres)
postgresQueueStore :: JournalMsgStore 'QSPostgres -> PostgresQueueStore (JournalQueue 'QSPostgres)
postgresQueueStore st = case queueStore_ st of
PQStore st' -> st'
#endif
data JournalStoreConfig s = JournalStoreConfig
{ storePath :: FilePath,
@@ -136,7 +149,9 @@ data JournalStoreConfig s = JournalStoreConfig
data QStoreCfg s where
MQStoreCfg :: QStoreCfg 'QSMemory
#if defined(dbServerPostgres)
PQStoreCfg :: PostgresStoreCfg -> QStoreCfg 'QSPostgres
#endif
data JournalQueue (s :: QSType) = JournalQueue
{ recipientId' :: RecipientId,
@@ -290,7 +305,9 @@ instance QueueStoreClass (JournalQueue s) (QStore s) where
newQueueStore :: QStoreCfg s -> IO (QStore s)
newQueueStore = \case
MQStoreCfg -> MQStore <$> newQueueStore @(JournalQueue s) ()
#if defined(dbServerPostgres)
PQStoreCfg cfg -> PQStore <$> newQueueStore @(JournalQueue s) cfg
#endif
closeQueueStore = withQS (closeQueueStore @(JournalQueue s))
{-# INLINE closeQueueStore #-}
@@ -321,9 +338,11 @@ instance QueueStoreClass (JournalQueue s) (QStore s) where
deleteStoreQueue = withQS deleteStoreQueue
{-# INLINE deleteStoreQueue #-}
#if defined(dbServerPostgres)
mkTempQueue :: JournalMsgStore s -> RecipientId -> QueueRec -> IO (JournalQueue s)
mkTempQueue ms rId qr = createLockIO >>= makeQueue_ ms rId qr
{-# INLINE mkTempQueue #-}
#endif
makeQueue_ :: JournalMsgStore s -> RecipientId -> QueueRec -> Lock -> IO (JournalQueue s)
makeQueue_ JournalMsgStore {sharedLock} rId qr queueLock = do
@@ -373,7 +392,9 @@ instance MsgStoreClass (JournalMsgStore s) where
unsafeWithAllMsgQueues :: Monoid a => Bool -> JournalMsgStore s -> (JournalQueue s -> IO a) -> IO a
unsafeWithAllMsgQueues tty ms action = case queueStore_ ms of
MQStore st -> withLoadedQueues st run
#if defined(dbServerPostgres)
PQStore st -> foldQueueRecs tty st $ uncurry (mkTempQueue ms) >=> run
#endif
where
run q = do
r <- action q
@@ -382,15 +403,18 @@ instance MsgStoreClass (JournalMsgStore s) where
-- This function is concurrency safe, it is used to expire queues.
withAllMsgQueues :: forall a. Monoid a => Bool -> String -> JournalMsgStore s -> (JournalQueue s -> StoreIO s a) -> IO a
withAllMsgQueues tty op ms@JournalMsgStore {queueLocks, sharedLock} action = case queueStore_ ms of
withAllMsgQueues tty op ms action = case queueStore_ ms of
MQStore st ->
withLoadedQueues st $ \q ->
run $ isolateQueue q op $ action q
PQStore st ->
#if defined(dbServerPostgres)
PQStore st -> do
let JournalMsgStore {queueLocks, sharedLock} = ms
foldQueueRecs tty st $ \(rId, qr) -> do
q <- mkTempQueue ms rId qr
withSharedWaitLock rId queueLocks sharedLock $
run $ tryStore' op rId $ unStoreIO $ action q
#endif
where
run :: ExceptT ErrorType IO a -> IO a
run = fmap (fromRight mempty) . runExceptT

13
src/Simplex/Messaging/Server/QueueStore.hs
View File

@@ -1,3 +1,4 @@
{-# LANGUAGE CPP #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DerivingStrategies #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
@@ -12,14 +13,16 @@ module Simplex.Messaging.Server.QueueStore where
import Control.Applicative ((<|>))
import Data.Functor (($>))
import Data.Int (Int64)
import Data.Text.Encoding (decodeLatin1, encodeUtf8)
import Data.Time.Clock.System (SystemTime (..), getSystemTime)
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol
#if defined(dbServerPostgres)
import Data.Text.Encoding (decodeLatin1, encodeUtf8)
import Database.PostgreSQL.Simple.FromField (FromField (..))
import Database.PostgreSQL.Simple.ToField (ToField (..))
import Simplex.Messaging.Agent.Store.Postgres.DB (fromTextField_)
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Protocol
import Simplex.Messaging.Util (eitherToMaybe)
#endif
data QueueRec = QueueRec
{ recipientKey :: !RcvPublicAuthKey,
@@ -62,13 +65,17 @@ instance StrEncoding ServerEntityStatus where
<|> "blocked," *> (EntityBlocked <$> strP)
<|> "off" $> EntityOff
#if defined(dbServerPostgres)
instance FromField ServerEntityStatus where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8
instance ToField ServerEntityStatus where toField = toField . decodeLatin1 . strEncode
#endif
newtype RoundedSystemTime = RoundedSystemTime Int64
deriving (Eq, Ord, Show)
#if defined(dbServerPostgres)
deriving newtype (FromField, ToField)
#endif
instance StrEncoding RoundedSystemTime where
strEncode (RoundedSystemTime t) = strEncode t

10
src/Simplex/Messaging/Server/QueueStore/Postgres.hs
View File

@@ -51,9 +51,9 @@ import Simplex.Messaging.Agent.Client (withLockMap)
import Simplex.Messaging.Agent.Lock (Lock)
import Simplex.Messaging.Agent.Store.Postgres (createDBStore, closeDBStore)
import Simplex.Messaging.Agent.Store.Postgres.Common
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation)
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server.QueueStore
import Simplex.Messaging.Server.QueueStore.Postgres.Config
import Simplex.Messaging.Server.QueueStore.Postgres.Migrations (serverMigrations)
import Simplex.Messaging.Server.QueueStore.STM (readQueueRecIO)
import Simplex.Messaging.Server.QueueStore.Types
@@ -64,6 +64,7 @@ import Simplex.Messaging.Util (firstRow, ifM, tshow, (<$$>))
import System.Exit (exitFailure)
import System.IO (IOMode (..), hFlush, stdout)
import UnliftIO.STM
#if !defined(dbPostgres)
import Simplex.Messaging.Agent.Store.Postgres.DB (blobFieldDecoder)
import qualified Simplex.Messaging.Crypto as C
@@ -83,13 +84,6 @@ data PostgresQueueStore q = PostgresQueueStore
deletedTTL :: Int64
}
data PostgresStoreCfg = PostgresStoreCfg
{ dbOpts :: DBOpts,
dbStoreLogPath :: Maybe FilePath,
confirmMigrations :: MigrationConfirmation,
deletedTTL :: Int64
}
instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
type QueueStoreCfg (PostgresQueueStore q) = PostgresStoreCfg

12
src/Simplex/Messaging/Server/QueueStore/Postgres/Config.hs Normal file
View File

@@ -0,0 +1,12 @@
module Simplex.Messaging.Server.QueueStore.Postgres.Config where
import Data.Int (Int64)
import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts)
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation)
data PostgresStoreCfg = PostgresStoreCfg
{ dbOpts :: DBOpts,
dbStoreLogPath :: Maybe FilePath,
confirmMigrations :: MigrationConfirmation,
deletedTTL :: Int64
}

5
tests/Fixtures.hs
View File

@@ -1,14 +1,10 @@
{-# LANGUAGE CPP #-}
{-# LANGUAGE OverloadedStrings #-}
module Fixtures where
#if defined(dbPostgres)
import Data.ByteString (ByteString)
import Database.PostgreSQL.Simple (ConnectInfo (..), defaultConnectInfo)
#endif
#if defined(dbPostgres)
testDBConnstr :: ByteString
testDBConnstr = "postgresql://test_agent_user@/test_agent_db"
@@ -18,4 +14,3 @@ testDBConnectInfo =
connectUser = "test_agent_user",
connectDatabase = "test_agent_db"
}
#endif

15
tests/SMPClient.hs
View File

@@ -1,3 +1,4 @@
{-# LANGUAGE CPP #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DuplicateRecordFields #-}
{-# LANGUAGE FlexibleContexts #-}
@@ -16,9 +17,8 @@ module SMPClient where
import Control.Monad.Except (runExceptT)
import Data.ByteString.Char8 (ByteString)
import Data.List.NonEmpty (NonEmpty)
import Database.PostgreSQL.Simple (ConnectInfo (..), defaultConnectInfo)
import Network.Socket
import Simplex.Messaging.Agent.Store.Postgres.Common (DBOpts (..))
import Simplex.Messaging.Agent.Store.Postgres.Options (DBOpts (..))
import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
import Simplex.Messaging.Client (ProtocolClientConfig (..), chooseTransportHost, defaultNetworkConfig)
import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..), defaultSMPClientAgentConfig)
@@ -28,7 +28,7 @@ import Simplex.Messaging.Protocol
import Simplex.Messaging.Server (runSMPServerBlocking)
import Simplex.Messaging.Server.Env.STM
import Simplex.Messaging.Server.MsgStore.Types (SMSType (..), SQSType (..))
import Simplex.Messaging.Server.QueueStore.Postgres (PostgresStoreCfg (..))
import Simplex.Messaging.Server.QueueStore.Postgres.Config (PostgresStoreCfg (..))
import Simplex.Messaging.Transport
import Simplex.Messaging.Transport.Client
import qualified Simplex.Messaging.Transport.Client as Client
@@ -44,6 +44,10 @@ import UnliftIO.STM (TMVar, atomically, newEmptyTMVarIO, putTMVar, takeTMVar)
import UnliftIO.Timeout (timeout)
import Util
#if defined(dbServerPostgres)
import Database.PostgreSQL.Simple (ConnectInfo (..), defaultConnectInfo)
#endif
testHost :: NonEmpty TransportHost
testHost = "localhost"
@@ -80,12 +84,14 @@ testStoreDBOpts2 = testStoreDBOpts {schema = "smp_server2"}
testServerDBConnstr :: ByteString
testServerDBConnstr = "postgresql://test_server_user@/test_server_db"
#if defined(dbServerPostgres)
testServerDBConnectInfo :: ConnectInfo
testServerDBConnectInfo =
defaultConnectInfo {
connectUser = "test_server_user",
connectDatabase = "test_server_db"
}
#endif
testStoreMsgsFile :: FilePath
testStoreMsgsFile = "tests/tmp/smp-server-messages.log"
@@ -146,9 +152,6 @@ testSMPClient_ host port vr client = do
cfg :: ServerConfig
cfg = cfgMS (ASType SQSMemory SMSJournal)
cfgDB :: ServerConfig
cfgDB = cfgMS (ASType SQSPostgres SMSJournal)
cfgJ2 :: ServerConfig
cfgJ2 = journalCfg cfg testStoreLogFile2 testStoreMsgsDir2

25
tests/Test.hs
View File

@@ -23,7 +23,6 @@ import GHC.IO.Exception (IOException (..))
import qualified GHC.IO.Exception as IOException
import NtfServerTests (ntfServerTests)
import RemoteControl (remoteControlTests)
import SMPClient (testServerDBConnectInfo)
import SMPProxyTests (smpProxyTests)
import ServerTests
import Simplex.Messaging.Server.Env.STM (AStoreType (..))
@@ -33,16 +32,25 @@ import Simplex.Messaging.Transport (TLS, Transport (..))
import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive)
import System.Environment (setEnv)
import Test.Hspec
import Util (postgressBracket)
import XFTPAgent
import XFTPCLI
import XFTPServerTests (xftpServerTests)
#if defined(dbPostgres)
import Fixtures
#else
import AgentTests.SchemaDump (schemaDumpTest)
#endif
#if defined(dbServerPostgres)
import SMPClient (testServerDBConnectInfo)
#endif
#if defined(dbPostgres) || defined(dbServerPostgres)
import Database.PostgreSQL.Simple (ConnectInfo (..))
import Simplex.Messaging.Agent.Store.Postgres.Util (createDBAndUserIfNotExists, dropDatabaseAndUser)
#endif
logCfg :: LogConfig
logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
@@ -75,11 +83,12 @@ main = do
describe "Store log tests" storeLogTests
describe "TRcvQueues tests" tRcvQueuesTests
describe "Util tests" utilTests
#if defined(dbServerPostgres)
aroundAll_ (postgressBracket testServerDBConnectInfo)
-- TODO [postgres] fix store log tests
$ describe "SMP server via TLS, postgres+jornal message store" $ do
describe "SMP syntax" $ serverSyntaxTests (transport @TLS)
before (pure (transport @TLS, ASType SQSPostgres SMSJournal)) serverTests
#endif
describe "SMP server via TLS, jornal message store" $ do
describe "SMP syntax" $ serverSyntaxTests (transport @TLS)
before (pure (transport @TLS, ASType SQSMemory SMSJournal)) serverTests
@@ -89,10 +98,12 @@ main = do
-- describe "SMP syntax" $ serverSyntaxTests (transport @WS)
-- before (pure (transport @WS, ASType SQSMemory SMSJournal)) serverTests
describe "Notifications server" $ ntfServerTests (transport @TLS)
#if defined(dbServerPostgres)
aroundAll_ (postgressBracket testServerDBConnectInfo) $ do
describe "SMP client agent, postgres+jornal message store" $ agentTests (transport @TLS, ASType SQSPostgres SMSJournal)
describe "SMP proxy, postgres+jornal message store" $
before (pure $ ASType SQSPostgres SMSJournal) smpProxyTests
#endif
describe "SMP client agent, jornal message store" $ agentTests (transport @TLS, ASType SQSMemory SMSJournal)
describe "SMP proxy, jornal message store" $
before (pure $ ASType SQSMemory SMSJournal) smpProxyTests
@@ -113,3 +124,11 @@ eventuallyRemove path retries = case retries of
_ -> E.throwIO ioe
where
action = removeDirectoryRecursive path
#if defined(dbPostgres) || defined(dbServerPostgres)
postgressBracket :: ConnectInfo -> IO a -> IO a
postgressBracket connInfo =
E.bracket_
(dropDatabaseAndUser connInfo >> createDBAndUserIfNotExists connInfo)
(dropDatabaseAndUser connInfo)
#endif

9
tests/Util.hs
View File

@@ -1,12 +1,9 @@
module Util where
import qualified Control.Exception as E
import Control.Monad (replicateM, when)
import Data.Either (partitionEithers)
import Data.List (tails)
import Database.PostgreSQL.Simple (ConnectInfo (..))
import GHC.Conc (getNumCapabilities, getNumProcessors, setNumCapabilities)
import Simplex.Messaging.Agent.Store.Postgres.Util (createDBAndUserIfNotExists, dropDatabaseAndUser)
import System.Directory (doesFileExist, removeFile)
import Test.Hspec
import UnliftIO
@@ -35,9 +32,3 @@ removeFileIfExists :: FilePath -> IO ()
removeFileIfExists filePath = do
fileExists <- doesFileExist filePath
when fileExists $ removeFile filePath
postgressBracket :: ConnectInfo -> IO a -> IO a
postgressBracket connInfo =
E.bracket_
(dropDatabaseAndUser connInfo >> createDBAndUserIfNotExists connInfo)
(dropDatabaseAndUser connInfo)