6.3.0.1

* servers: blocking records for content moderation

* update

* encode BLOCKED as AUTH in old versions

* update

* unblock queue command

* test, status command

CHANGELOG.md

@@ -1,3 +1,27 @@
+# 6.2.2
+
+SMP server:
+- add optional Prometheus metrics (#1411).
+
+Build:
+- remove three modules from client library.
+
+# 6.2.0
+
+Version 6.2.0.7
+
+Build:
+- client_library flag to build only used modules in the clients, remove package yaml
+
+SMP server:
+- journal storage for messages (BETA).
+- prevent race condition when deleting queue and to avoid "orphan" messages (#1395).
+
+SMP agent:
+- support SMP and XFTP server roles (storage/proxy) and operators (#1343).
+- treat blocked STM and other critical errors that offer restart as temporary for message delivery (#1405).
+- fix inconsistent state after app restart while accepting contact request (#1412).
+
 # 6.1.3
 
 SMP server: fix restoring notification credentials.

simplexmq.cabal

@@ -1,7 +1,7 @@
 cabal-version: 1.12
 
 name:           simplexmq
-version:        6.2.0.6
+version:        6.3.0.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -67,13 +67,17 @@ flag client_library
   manual: True
   default: False
 
+flag client_postgres
+  description: Build with PostgreSQL instead of SQLite.
+  manual: True
+  default: False
+
 library
   exposed-modules:
       Simplex.FileTransfer.Agent
       Simplex.FileTransfer.Chunks
       Simplex.FileTransfer.Client
       Simplex.FileTransfer.Client.Agent
-      Simplex.FileTransfer.Client.Main
       Simplex.FileTransfer.Client.Presets
       Simplex.FileTransfer.Crypto
       Simplex.FileTransfer.Description
@@ -91,46 +95,11 @@ library
       Simplex.Messaging.Agent.RetryInterval
       Simplex.Messaging.Agent.Stats
       Simplex.Messaging.Agent.Store
-      Simplex.Messaging.Agent.Store.SQLite
-      Simplex.Messaging.Agent.Store.SQLite.Common
-      Simplex.Messaging.Agent.Store.SQLite.DB
-      Simplex.Messaging.Agent.Store.SQLite.Migrations
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220322_notifications
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220608_v2
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220625_v2_ntf_mode
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220811_onion_hosts
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220817_connection_ntfs
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220915_connection_queues
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230110_users
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230117_fkey_indexes
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230120_delete_errors
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230217_server_key_hash
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230223_files
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230320_retry_state
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230720_delete_expired_messages
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230814_indexes
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230829_crypto_files
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231222_command_created_at
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231225_failed_work_items
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240121_message_delivery_indexes
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240124_file_redirect
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240223_connections_wait_delivery
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240225_ratchet_kem
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240417_rcv_files_approved_relays
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240624_snd_secure
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240702_servers_stats
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240930_ntf_tokens_to_delete
-      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241007_rcv_queues_last_broker_ts
+      Simplex.Messaging.Agent.Store.AgentStore
+      Simplex.Messaging.Agent.Store.Common
+      Simplex.Messaging.Agent.Store.DB
+      Simplex.Messaging.Agent.Store.Migrations
+      Simplex.Messaging.Agent.Store.Shared
       Simplex.Messaging.Agent.TRcvQueues
       Simplex.Messaging.Client
       Simplex.Messaging.Client.Agent
@@ -152,7 +121,6 @@ library
       Simplex.Messaging.Notifications.Types
       Simplex.Messaging.Parsers
       Simplex.Messaging.Protocol
-      Simplex.Messaging.Server.CLI
       Simplex.Messaging.Server.Expiration
       Simplex.Messaging.Server.QueueStore.QueueInfo
       Simplex.Messaging.ServiceScheme
@@ -168,7 +136,6 @@ library
       Simplex.Messaging.Transport.HTTP2.Server
       Simplex.Messaging.Transport.KeepAlive
       Simplex.Messaging.Transport.Server
-      Simplex.Messaging.Transport.WebSockets
       Simplex.Messaging.Util
       Simplex.Messaging.Version
       Simplex.Messaging.Version.Internal
@@ -177,8 +144,62 @@ library
       Simplex.RemoteControl.Discovery.Multicast
       Simplex.RemoteControl.Invitation
       Simplex.RemoteControl.Types
+  if flag(client_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.Migrations.M20241210_initial
+      if !flag(client_library)
+          exposed-modules:
+              Simplex.Messaging.Agent.Store.Postgres.Util
+  else
+      exposed-modules:
+          Simplex.Messaging.Agent.Store.SQLite
+          Simplex.Messaging.Agent.Store.SQLite.Common
+          Simplex.Messaging.Agent.Store.SQLite.DB
+          Simplex.Messaging.Agent.Store.SQLite.Migrations
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220322_notifications
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220608_v2
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220625_v2_ntf_mode
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220811_onion_hosts
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220817_connection_ntfs
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220905_commands
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220915_connection_queues
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230110_users
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230117_fkey_indexes
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230120_delete_errors
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230217_server_key_hash
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230223_files
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230320_retry_state
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230720_delete_expired_messages
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230814_indexes
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230829_crypto_files
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231222_command_created_at
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20231225_failed_work_items
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240121_message_delivery_indexes
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240124_file_redirect
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240223_connections_wait_delivery
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240225_ratchet_kem
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240417_rcv_files_approved_relays
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240624_snd_secure
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240702_servers_stats
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240930_ntf_tokens_to_delete
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241007_rcv_queues_last_broker_ts
+          Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241224_ratchet_e2e_snd_params
   if !flag(client_library)
       exposed-modules:
+          Simplex.FileTransfer.Client.Main
           Simplex.FileTransfer.Server
           Simplex.FileTransfer.Server.Control
           Simplex.FileTransfer.Server.Env
@@ -186,21 +207,6 @@ library
           Simplex.FileTransfer.Server.Stats
           Simplex.FileTransfer.Server.Store
           Simplex.FileTransfer.Server.StoreLog
-          Simplex.Messaging.Server
-          Simplex.Messaging.Server.Control
-          Simplex.Messaging.Server.Env.STM
-          Simplex.Messaging.Server.Information
-          Simplex.Messaging.Server.Main
-          Simplex.Messaging.Server.MsgStore
-          Simplex.Messaging.Server.MsgStore.Journal
-          Simplex.Messaging.Server.MsgStore.STM
-          Simplex.Messaging.Server.MsgStore.Types
-          Simplex.Messaging.Server.NtfStore
-          Simplex.Messaging.Server.QueueStore
-          Simplex.Messaging.Server.QueueStore.STM
-          Simplex.Messaging.Server.Stats
-          Simplex.Messaging.Server.StoreLog
-          Simplex.Messaging.Server.StoreLog.Types
           Simplex.Messaging.Notifications.Server
           Simplex.Messaging.Notifications.Server.Control
           Simplex.Messaging.Notifications.Server.Env
@@ -210,6 +216,24 @@ library
           Simplex.Messaging.Notifications.Server.Stats
           Simplex.Messaging.Notifications.Server.Store
           Simplex.Messaging.Notifications.Server.StoreLog
+          Simplex.Messaging.Server
+          Simplex.Messaging.Server.CLI
+          Simplex.Messaging.Server.Control
+          Simplex.Messaging.Server.Env.STM
+          Simplex.Messaging.Server.Information
+          Simplex.Messaging.Server.Main
+          Simplex.Messaging.Server.MsgStore
+          Simplex.Messaging.Server.MsgStore.Journal
+          Simplex.Messaging.Server.MsgStore.STM
+          Simplex.Messaging.Server.MsgStore.Types
+          Simplex.Messaging.Server.NtfStore
+          Simplex.Messaging.Server.Prometheus
+          Simplex.Messaging.Server.QueueStore
+          Simplex.Messaging.Server.QueueStore.STM
+          Simplex.Messaging.Server.Stats
+          Simplex.Messaging.Server.StoreLog
+          Simplex.Messaging.Server.StoreLog.Types
+          Simplex.Messaging.Transport.WebSockets
   other-modules:
       Paths_simplexmq
   hs-source-dirs:
@@ -241,7 +265,6 @@ library
     , crypton-x509-validation ==1.6.*
     , cryptostore ==0.3.*
     , data-default ==0.7.*
-    , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
     , hourglass ==0.2.*
@@ -261,7 +284,6 @@ library
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
     , socks ==0.6.*
-    , sqlcipher-simple ==0.4.*
     , stm ==2.5.*
     , temporary ==1.3.*
     , time ==1.12.*
@@ -270,7 +292,6 @@ library
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
-    , websockets ==0.12.*
     , yaml ==0.11.*
     , zstd ==0.1.3.*
   default-language: Haskell2010
@@ -280,6 +301,17 @@ library
     build-depends:
         case-insensitive ==1.2.*
       , hashable ==1.4.*
+      , websockets ==0.12.*
+  if flag(client_postgres)
+    build-depends:
+        postgresql-libpq >=0.10.0.0
+      , postgresql-simple ==0.7.*
+      , raw-strings-qq ==1.1.*
+    cpp-options: -DdbPostgres
+  else
+    build-depends:
+        direct-sqlcipher ==2.3.*
+      , sqlcipher-simple ==0.4.*
   if impl(ghc >= 9.6.2)
     build-depends:
         bytestring ==0.11.*
@@ -382,10 +414,7 @@ test-suite simplexmq-test
       AgentTests.EqInstances
       AgentTests.FunctionalAPITests
       AgentTests.MigrationTests
-      AgentTests.NotificationTests
-      AgentTests.SchemaDump
       AgentTests.ServerChoice
-      AgentTests.SQLiteTests
       CLITests
       CoreTests.BatchingTests
       CoreTests.CryptoFileTests
@@ -399,6 +428,7 @@ test-suite simplexmq-test
       CoreTests.UtilTests
       CoreTests.VersionRangeTests
       FileDescriptionTests
+      Fixtures
       NtfClient
       NtfServerTests
       RemoteControl
@@ -414,6 +444,11 @@ test-suite simplexmq-test
       Static
       Static.Embedded
       Paths_simplexmq
+  if !flag(client_postgres)
+      other-modules:
+          AgentTests.NotificationTests
+          AgentTests.SchemaDump
+          AgentTests.SQLiteTests
   hs-source-dirs:
       tests
       apps/smp-server/web
@@ -454,7 +489,6 @@ test-suite simplexmq-test
     , silently ==1.2.*
     , simple-logger
     , simplexmq
-    , sqlcipher-simple
     , stm
     , text
     , time
@@ -469,3 +503,12 @@ test-suite simplexmq-test
     , warp-tls
     , yaml
   default-language: Haskell2010
+  if flag(client_postgres)
+    build-depends:
+        postgresql-libpq >=0.10.0.0
+      , postgresql-simple ==0.7.*
+      , raw-strings-qq ==1.1.*
+    cpp-options: -DdbPostgres
+  else
+    build-depends:
+        sqlcipher-simple

src/Simplex/FileTransfer/Agent.hs

@@ -51,8 +51,7 @@ import Data.Text (Text)
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Format (defaultTimeLocale, formatTime)
 import Simplex.FileTransfer.Chunks (toKB)
-import Simplex.FileTransfer.Client (XFTPChunkSpec (..))
-import Simplex.FileTransfer.Client.Main
+import Simplex.FileTransfer.Client (XFTPChunkSpec (..), getChunkDigest, prepareChunkSizes, prepareChunkSpecs, singleChunkSize)
 import Simplex.FileTransfer.Crypto
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..), SFileParty (..))
@@ -66,8 +65,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/FileTransfer/Client.hs

@@ -20,14 +20,18 @@ import Data.Bifunctor (first)
 import Data.ByteString.Builder (Builder, byteString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Lazy as LB
 import Data.Int (Int64)
+import Data.List (foldl')
 import Data.List.NonEmpty (NonEmpty (..))
+import Data.Maybe (listToMaybe)
 import Data.Time.Clock (UTCTime)
 import Data.Word (Word32)
 import qualified Data.X509 as X
 import qualified Data.X509.Validation as XV
 import qualified Network.HTTP.Types as N
 import qualified Network.HTTP2.Client as H
+import Simplex.FileTransfer.Chunks
 import Simplex.FileTransfer.Protocol
 import Simplex.FileTransfer.Transport
 import Simplex.Messaging.Client
@@ -298,3 +302,41 @@ noFile HTTP2Body {bodyPart} a = case bodyPart of
 
 -- FACK :: FileCommand Recipient
 -- PING :: FileCommand Recipient
+
+singleChunkSize :: Int64 -> Maybe Word32
+singleChunkSize size' =
+  listToMaybe $ dropWhile (< chunkSize) serverChunkSizes
+  where
+    chunkSize = fromIntegral size'
+
+prepareChunkSizes :: Int64 -> [Word32]
+prepareChunkSizes size' = prepareSizes size'
+  where
+    (smallSize, bigSize)
+      | size' > size34 chunkSize3 = (chunkSize2, chunkSize3)
+      | size' > size34 chunkSize2 = (chunkSize1, chunkSize2)
+      | otherwise = (chunkSize0, chunkSize1)
+    size34 sz = (fromIntegral sz * 3) `div` 4
+    prepareSizes 0 = []
+    prepareSizes size
+      | size >= fromIntegral bigSize = replicate (fromIntegral n1) bigSize <> prepareSizes remSz
+      | size > size34 bigSize = [bigSize]
+      | otherwise = replicate (fromIntegral n2') smallSize
+      where
+        (n1, remSz) = size `divMod` fromIntegral bigSize
+        n2' = let (n2, remSz2) = (size `divMod` fromIntegral smallSize) in if remSz2 == 0 then n2 else n2 + 1
+
+prepareChunkSpecs :: FilePath -> [Word32] -> [XFTPChunkSpec]
+prepareChunkSpecs filePath chunkSizes = reverse . snd $ foldl' addSpec (0, []) chunkSizes
+  where
+    addSpec :: (Int64, [XFTPChunkSpec]) -> Word32 -> (Int64, [XFTPChunkSpec])
+    addSpec (chunkOffset, specs) sz =
+      let spec = XFTPChunkSpec {filePath, chunkOffset, chunkSize = sz}
+       in (chunkOffset + fromIntegral sz, spec : specs)
+
+getChunkDigest :: XFTPChunkSpec -> IO ByteString
+getChunkDigest XFTPChunkSpec {filePath = chunkPath, chunkOffset, chunkSize} =
+  withFile chunkPath ReadMode $ \h -> do
+    hSeek h AbsoluteSeek $ fromIntegral chunkOffset
+    chunk <- LB.hGet h (fromIntegral chunkSize)
+    pure $! LC.sha256Hash chunk

src/Simplex/FileTransfer/Client/Main.hs

@@ -19,11 +19,7 @@ module Simplex.FileTransfer.Client.Main
     singleChunkSize,
     prepareChunkSizes,
     prepareChunkSpecs,
-    maxFileSize,
-    maxFileSizeHard,
-    fileSizeLen,
     getChunkDigest,
-    SentRecipientReplica (..),
   )
 where
 
@@ -34,7 +30,6 @@ import Control.Monad.Trans.Except
 import Crypto.Random (ChaChaDRG)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)
-import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Char (toLower)
@@ -45,7 +40,7 @@ import Data.List.NonEmpty (NonEmpty (..), nonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
-import Data.Maybe (fromMaybe, listToMaybe)
+import Data.Maybe (fromMaybe)
 import qualified Data.Text as T
 import Data.Word (Word32)
 import GHC.Records (HasField (getField))
@@ -80,20 +75,6 @@ import UnliftIO.Directory
 xftpClientVersion :: String
 xftpClientVersion = "1.0.1"
 
--- | Soft limit for XFTP clients. Should be checked and reported to user.
-maxFileSize :: Int64
-maxFileSize = gb 1
-
-maxFileSizeStr :: String
-maxFileSizeStr = B.unpack . strEncode $ FileSize maxFileSize
-
--- | Hard internal limit for XFTP agent after which it refuses to prepare chunks.
-maxFileSizeHard :: Int64
-maxFileSizeHard = gb 5
-
-fileSizeLen :: Int64
-fileSizeLen = 8
-
 newtype CLIError = CLIError String
   deriving (Eq, Show, Exception)
 
@@ -231,16 +212,6 @@ data SentFileChunkReplica = SentFileChunkReplica
   }
   deriving (Show)
 
-data SentRecipientReplica = SentRecipientReplica
-  { chunkNo :: Int,
-    server :: XFTPServer,
-    rcvNo :: Int,
-    replicaId :: ChunkReplicaId,
-    replicaKey :: C.APrivateAuthKey,
-    digest :: FileDigest,
-    chunkSize :: FileSize Word32
-  }
-
 logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
 
@@ -414,13 +385,6 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
       B.writeFile fdSndPath $ strEncode fdSnd
       pure (fdRcvPaths, fdSndPath)
 
-getChunkDigest :: XFTPChunkSpec -> IO ByteString
-getChunkDigest XFTPChunkSpec {filePath = chunkPath, chunkOffset, chunkSize} =
-  withFile chunkPath ReadMode $ \h -> do
-    hSeek h AbsoluteSeek $ fromIntegral chunkOffset
-    chunk <- LB.hGet h (fromIntegral chunkSize)
-    pure $! LC.sha256Hash chunk
-
 cliReceiveFile :: ReceiveOptions -> ExceptT CLIError IO ()
 cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath, verbose, yes} =
   getFileDescription' fileDescription >>= receive
@@ -536,37 +500,6 @@ getFileDescription' path =
   getFileDescription path >>= \case
     AVFD fd -> either (throwE . CLIError) pure $ checkParty fd
 
-singleChunkSize :: Int64 -> Maybe Word32
-singleChunkSize size' =
-  listToMaybe $ dropWhile (< chunkSize) serverChunkSizes
-  where
-    chunkSize = fromIntegral size'
-
-prepareChunkSizes :: Int64 -> [Word32]
-prepareChunkSizes size' = prepareSizes size'
-  where
-    (smallSize, bigSize)
-      | size' > size34 chunkSize3 = (chunkSize2, chunkSize3)
-      | size' > size34 chunkSize2 = (chunkSize1, chunkSize2)
-      | otherwise = (chunkSize0, chunkSize1)
-    size34 sz = (fromIntegral sz * 3) `div` 4
-    prepareSizes 0 = []
-    prepareSizes size
-      | size >= fromIntegral bigSize = replicate (fromIntegral n1) bigSize <> prepareSizes remSz
-      | size > size34 bigSize = [bigSize]
-      | otherwise = replicate (fromIntegral n2') smallSize
-      where
-        (n1, remSz) = size `divMod` fromIntegral bigSize
-        n2' = let (n2, remSz2) = (size `divMod` fromIntegral smallSize) in if remSz2 == 0 then n2 else n2 + 1
-
-prepareChunkSpecs :: FilePath -> [Word32] -> [XFTPChunkSpec]
-prepareChunkSpecs filePath chunkSizes = reverse . snd $ foldl' addSpec (0, []) chunkSizes
-  where
-    addSpec :: (Int64, [XFTPChunkSpec]) -> Word32 -> (Int64, [XFTPChunkSpec])
-    addSpec (chunkOffset, specs) sz =
-      let spec = XFTPChunkSpec {filePath, chunkOffset, chunkSize = sz}
-       in (chunkOffset + fromIntegral sz, spec : specs)
-
 getEncPath :: MonadIO m => Maybe FilePath -> String -> m FilePath
 getEncPath path name = (`uniqueCombine` (name <> ".encrypted")) =<< maybe (liftIO getCanonicalTemporaryDirectory) pure path
 

src/Simplex/FileTransfer/Description.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DerivingStrategies #-}
@@ -9,6 +10,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StandaloneDeriving #-}
 {-# LANGUAGE TemplateHaskell #-}
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
@@ -37,6 +39,10 @@ module Simplex.FileTransfer.Description
     FileClientData,
     fileDescriptionURI,
     qrSizeLimit,
+    maxFileSize,
+    maxFileSizeStr,
+    maxFileSizeHard,
+    fileSizeLen,
   )
 where
 
@@ -62,17 +68,23 @@ import Data.Text (Text)
 import Data.Text.Encoding (encodeUtf8)
 import Data.Word (Word32)
 import qualified Data.Yaml as Y
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
 import Simplex.FileTransfer.Chunks
 import Simplex.FileTransfer.Protocol
 import Simplex.Messaging.Agent.QueryString
+import Simplex.Messaging.Agent.Store.DB (Binary (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (defaultJSON, parseAll)
 import Simplex.Messaging.Protocol (XFTPServer)
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import Simplex.Messaging.Util (bshow, safeDecodeUtf8, (<$?>))
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple.FromField (FromField (..))
+import Database.PostgreSQL.Simple.ToField (ToField (..))
+#else
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
+#endif
 
 data FileDescription (p :: FileParty) = FileDescription
   { party :: SFileParty p,
@@ -109,6 +121,9 @@ fdSeparator = "################################\n"
 
 newtype FileDigest = FileDigest {unFileDigest :: ByteString}
   deriving (Eq, Show)
+  deriving newtype (FromField)
+
+instance ToField FileDigest where toField (FileDigest s) = toField $ Binary s
 
 instance StrEncoding FileDigest where
   strEncode (FileDigest fd) = strEncode fd
@@ -122,10 +137,6 @@ instance ToJSON FileDigest where
   toJSON = strToJSON
   toEncoding = strToJEncoding
 
-instance FromField FileDigest where fromField f = FileDigest <$> fromField f
-
-instance ToField FileDigest where toField (FileDigest s) = toField s
-
 data FileChunk = FileChunk
   { chunkNo :: Int,
     chunkSize :: FileSize Word32,
@@ -266,6 +277,21 @@ instance StrEncoding FileDescriptionURI where
 qrSizeLimit :: Int
 qrSizeLimit = 1002 -- ~2 chunks in URLencoded YAML with some spare size for server hosts
 
+-- | Soft limit for XFTP clients. Should be checked and reported to user.
+maxFileSize :: Int64
+maxFileSize = gb 1
+
+maxFileSizeStr :: String
+maxFileSizeStr = B.unpack . strEncode $ FileSize maxFileSize
+
+-- | Hard internal limit for XFTP agent after which it refuses to prepare chunks.
+maxFileSizeHard :: Int64
+maxFileSizeHard = gb 5
+
+fileSizeLen :: Int64
+fileSizeLen = 8
+
+
 instance (Integral a, Show a) => StrEncoding (FileSize a) where
   strEncode (FileSize b)
     | b' /= 0 = bshow b
@@ -288,9 +314,9 @@ instance (Integral a, Show a) => StrEncoding (FileSize a) where
 instance (Integral a, Show a) => IsString (FileSize a) where
   fromString = either error id . strDecode . B.pack
 
-instance FromField a => FromField (FileSize a) where fromField f = FileSize <$> fromField f
+deriving newtype instance FromField a => FromField (FileSize a)
 
-instance ToField a => ToField (FileSize a) where toField (FileSize s) = toField s
+deriving newtype instance ToField a => ToField (FileSize a)
 
 groupReplicasByServer :: FileSize Word32 -> [FileChunk] -> [NonEmpty FileServerReplica]
 groupReplicasByServer defChunkSize =

src/Simplex/FileTransfer/Protocol.hs

@@ -25,7 +25,7 @@ import Data.List.NonEmpty (NonEmpty (..))
 import Data.Maybe (isNothing)
 import Data.Type.Equality
 import Data.Word (Word32)
-import Simplex.FileTransfer.Transport (XFTPErrorType (..), XFTPVersion, xftpClientHandshakeStub)
+import Simplex.FileTransfer.Transport (XFTPErrorType (..), XFTPVersion, blockedFilesXFTPVersion, xftpClientHandshakeStub)
 import Simplex.Messaging.Client (authTransmission)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
@@ -276,12 +276,14 @@ data FileResponse
 
 instance ProtocolEncoding XFTPVersion XFTPErrorType FileResponse where
   type Tag FileResponse = FileResponseTag
-  encodeProtocol _v = \case
+  encodeProtocol v = \case
     FRSndIds fId rIds -> e (FRSndIds_, ' ', fId, rIds)
     FRRcvIds rIds -> e (FRRcvIds_, ' ', rIds)
     FRFile rDhKey nonce -> e (FRFile_, ' ', rDhKey, nonce)
     FROk -> e FROk_
-    FRErr err -> e (FRErr_, ' ', err)
+    FRErr err -> case err of
+      BLOCKED _ | v < blockedFilesXFTPVersion -> e (FRErr_, ' ', AUTH)
+      _ -> e (FRErr_, ' ', err)
     FRPong -> e FRPong_
     where
       e :: Encoding a => a -> ByteString

src/Simplex/FileTransfer/Server.hs

@@ -53,11 +53,11 @@ import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (CorrId (..), EntityId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, TransmissionAuth, pattern NoEntity)
+import Simplex.Messaging.Protocol (CorrId (..), BlockingInfo, EntityId (..), RcvPublicAuthKey, RcvPublicDhKey, RecipientId, TransmissionAuth, pattern NoEntity)
 import Simplex.Messaging.Server (dummyVerifyCmd, verifyCmdAuthorization)
 import Simplex.Messaging.Server.Control (CPClientRole (..))
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Server.QueueStore (RoundedSystemTime, getRoundedSystemTime)
+import Simplex.Messaging.Server.QueueStore (RoundedSystemTime, ServerEntityStatus (..), getRoundedSystemTime)
 import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
@@ -287,11 +287,15 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
               CPDelete fileId -> withUserRole $ unliftIO u $ do
                 fs <- asks store
                 r <- runExceptT $ do
-                  let asSender = ExceptT . atomically $ getFile fs SFSender fileId
-                  let asRecipient = ExceptT . atomically $ getFile fs SFRecipient fileId
-                  (fr, _) <- asSender `catchError` const asRecipient
+                  (fr, _) <- ExceptT $ atomically $ getFile fs SFSender fileId
                   ExceptT $ deleteServerFile_ fr
                 liftIO . hPutStrLn h $ either (\e -> "error: " <> show e) (\() -> "ok") r
+              CPBlock fileId info -> withUserRole $ unliftIO u $ do
+                fs <- asks store
+                r <- runExceptT $ do
+                  (fr, _) <- ExceptT $ atomically $ getFile fs SFSender fileId
+                  ExceptT $ blockServerFile fr info
+                liftIO . hPutStrLn h $ either (\e -> "error: " <> show e) (\() -> "ok") r
               CPHelp -> hPutStrLn h "commands: stats-rts, delete, help, quit"
               CPQuit -> pure ()
               CPSkip -> pure ()
@@ -321,7 +325,7 @@ processRequest XFTPTransportRequest {thParams, reqBody = body@HTTP2Body {bodyHea
               let THandleParams {thAuth} = thParams
               verifyXFTPTransmission ((,C.cbNonce (bs corrId)) <$> thAuth) sig_ signed fId cmd >>= \case
                 VRVerified req -> uncurry send =<< processXFTPRequest body req
-                VRFailed -> send (FRErr AUTH) Nothing
+                VRFailed e -> send (FRErr e) Nothing
             Left e -> send (FRErr e) Nothing
           where
             send resp = sendXFTPResponse (corrId, fId, resp)
@@ -355,7 +359,7 @@ randomDelay = do
     threadDelay $ (d * (1000 + pc)) `div` 1000
 #endif
 
-data VerificationResult = VRVerified XFTPRequest | VRFailed
+data VerificationResult = VRVerified XFTPRequest | VRFailed XFTPErrorType
 
 verifyXFTPTransmission :: Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> XFTPFileId -> FileCmd -> M VerificationResult
 verifyXFTPTransmission auth_ tAuth authorized fId cmd =
@@ -367,13 +371,19 @@ verifyXFTPTransmission auth_ tAuth authorized fId cmd =
     verifyCmd :: SFileParty p -> M VerificationResult
     verifyCmd party = do
       st <- asks store
-      atomically $ verify <$> getFile st party fId
+      atomically $ verify =<< getFile st party fId
       where
         verify = \case
-          Right (fr, k) -> XFTPReqCmd fId fr cmd `verifyWith` k
-          _ -> maybe False (dummyVerifyCmd Nothing authorized) tAuth `seq` VRFailed
+          Right (fr, k) -> result <$> readTVar (fileStatus fr)
+            where
+              result = \case
+                EntityActive -> XFTPReqCmd fId fr cmd `verifyWith` k
+                EntityBlocked info -> VRFailed $ BLOCKED info
+                EntityOff -> noFileAuth
+          Left _ -> pure noFileAuth
+        noFileAuth = maybe False (dummyVerifyCmd Nothing authorized) tAuth `seq` VRFailed AUTH
     -- TODO verify with DH authorization
-    req `verifyWith` k = if verifyCmdAuthorization auth_ tAuth authorized k then VRVerified req else VRFailed
+    req `verifyWith` k = if verifyCmdAuthorization auth_ tAuth authorized k then VRVerified req else VRFailed AUTH
 
 processXFTPRequest :: HTTP2Body -> XFTPRequest -> M (FileResponse, Maybe ServerFile)
 processXFTPRequest HTTP2Body {bodyPart} = \case
@@ -390,7 +400,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
     FACK -> noFile =<< ackFileReception fId fr
     -- it should never get to the commands below, they are passed in other constructors of XFTPRequest
     FNEW {} -> noFile $ FRErr INTERNAL
-    PING -> noFile FRPong
+    PING -> noFile $ FRErr INTERNAL
   XFTPReqPing -> noFile FRPong
   where
     noFile resp = pure (resp, Nothing)
@@ -518,15 +528,24 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
       pure FROk
 
 deleteServerFile_ :: FileRec -> M (Either XFTPErrorType ())
-deleteServerFile_ FileRec {senderId, fileInfo, filePath} = do
+deleteServerFile_ fr@FileRec {senderId} = do
   withFileLog (`logDeleteFile` senderId)
-  runExceptT $ do
+  deleteOrBlockServerFile_ fr filesDeleted (`deleteFile` senderId)
+
+-- this also deletes the file from storage, but doesn't include it in delete statistics
+blockServerFile :: FileRec -> BlockingInfo -> M (Either XFTPErrorType ())
+blockServerFile fr@FileRec {senderId} info = do
+  withFileLog $ \sl -> logBlockFile sl senderId info
+  deleteOrBlockServerFile_ fr filesBlocked $ \st -> blockFile st senderId info True
+
+deleteOrBlockServerFile_ :: FileRec -> (FileServerStats -> IORef Int) -> (FileStore -> STM (Either XFTPErrorType ())) -> M (Either XFTPErrorType ())
+deleteOrBlockServerFile_ FileRec {filePath, fileInfo} stat storeAction = runExceptT $ do
     path <- readTVarIO filePath
     stats <- asks serverStats
     ExceptT $ first (\(_ :: SomeException) -> FILE_IO) <$> try (forM_ path $ \p -> whenM (doesFileExist p) (removeFile p >> deletedStats stats))
     st <- asks store
-    void $ atomically $ deleteFile st senderId
-    lift $ incFileStat filesDeleted
+    void $ atomically $ storeAction st
+    lift $ incFileStat stat
   where
     deletedStats stats = do
       liftIO $ atomicModifyIORef'_ (filesCount stats) (subtract 1)

src/Simplex/FileTransfer/Server/Control.hs

@@ -6,12 +6,13 @@ module Simplex.FileTransfer.Server.Control where
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Simplex.FileTransfer.Protocol (XFTPFileId)
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (BasicAuth)
+import Simplex.Messaging.Protocol (BasicAuth, BlockingInfo)
 
 data ControlProtocol
   = CPAuth BasicAuth
   | CPStatsRTS
   | CPDelete XFTPFileId
+  | CPBlock XFTPFileId BlockingInfo
   | CPHelp
   | CPQuit
   | CPSkip
@@ -21,6 +22,7 @@ instance StrEncoding ControlProtocol where
     CPAuth tok -> "auth " <> strEncode tok
     CPStatsRTS -> "stats-rts"
     CPDelete fId -> strEncode (Str "delete", fId)
+    CPBlock fId info -> strEncode (Str "block", fId, info)
     CPHelp -> "help"
     CPQuit -> "quit"
     CPSkip -> ""
@@ -29,6 +31,7 @@ instance StrEncoding ControlProtocol where
       "auth" -> CPAuth <$> _strP
       "stats-rts" -> pure CPStatsRTS
       "delete" -> CPDelete <$> _strP
+      "block" -> CPBlock <$> _strP <*> _strP
       "help" -> pure CPHelp
       "quit" -> pure CPQuit
       "" -> pure CPSkip

src/Simplex/FileTransfer/Server/Stats.hs

@@ -20,6 +20,7 @@ data FileServerStats = FileServerStats
     filesUploaded :: IORef Int,
     filesExpired :: IORef Int,
     filesDeleted :: IORef Int,
+    filesBlocked :: IORef Int,
     filesDownloaded :: PeriodStats,
     fileDownloads :: IORef Int,
     fileDownloadAcks :: IORef Int,
@@ -34,6 +35,7 @@ data FileServerStatsData = FileServerStatsData
     _filesUploaded :: Int,
     _filesExpired :: Int,
     _filesDeleted :: Int,
+    _filesBlocked :: Int,
     _filesDownloaded :: PeriodStatsData,
     _fileDownloads :: Int,
     _fileDownloadAcks :: Int,
@@ -50,12 +52,13 @@ newFileServerStats ts = do
   filesUploaded <- newIORef 0
   filesExpired <- newIORef 0
   filesDeleted <- newIORef 0
+  filesBlocked <- newIORef 0
   filesDownloaded <- newPeriodStats
   fileDownloads <- newIORef 0
   fileDownloadAcks <- newIORef 0
   filesCount <- newIORef 0
   filesSize <- newIORef 0
-  pure FileServerStats {fromTime, filesCreated, fileRecipients, filesUploaded, filesExpired, filesDeleted, filesDownloaded, fileDownloads, fileDownloadAcks, filesCount, filesSize}
+  pure FileServerStats {fromTime, filesCreated, fileRecipients, filesUploaded, filesExpired, filesDeleted, filesBlocked, filesDownloaded, fileDownloads, fileDownloadAcks, filesCount, filesSize}
 
 getFileServerStatsData :: FileServerStats -> IO FileServerStatsData
 getFileServerStatsData s = do
@@ -65,12 +68,13 @@ getFileServerStatsData s = do
   _filesUploaded <- readIORef $ filesUploaded s
   _filesExpired <- readIORef $ filesExpired s
   _filesDeleted <- readIORef $ filesDeleted s
+  _filesBlocked <- readIORef $ filesBlocked s
   _filesDownloaded <- getPeriodStatsData $ filesDownloaded s
   _fileDownloads <- readIORef $ fileDownloads s
   _fileDownloadAcks <- readIORef $ fileDownloadAcks s
   _filesCount <- readIORef $ filesCount s
   _filesSize <- readIORef $ filesSize s
-  pure FileServerStatsData {_fromTime, _filesCreated, _fileRecipients, _filesUploaded, _filesExpired, _filesDeleted, _filesDownloaded, _fileDownloads, _fileDownloadAcks, _filesCount, _filesSize}
+  pure FileServerStatsData {_fromTime, _filesCreated, _fileRecipients, _filesUploaded, _filesExpired, _filesDeleted, _filesBlocked, _filesDownloaded, _fileDownloads, _fileDownloadAcks, _filesCount, _filesSize}
 
 -- this function is not thread safe, it is used on server start only
 setFileServerStats :: FileServerStats -> FileServerStatsData -> IO ()
@@ -81,6 +85,7 @@ setFileServerStats s d = do
   writeIORef (filesUploaded s) $! _filesUploaded d
   writeIORef (filesExpired s) $! _filesExpired d
   writeIORef (filesDeleted s) $! _filesDeleted d
+  writeIORef (filesBlocked s) $! _filesBlocked d
   setPeriodStats (filesDownloaded s) $! _filesDownloaded d
   writeIORef (fileDownloads s) $! _fileDownloads d
   writeIORef (fileDownloadAcks s) $! _fileDownloadAcks d
@@ -88,7 +93,7 @@ setFileServerStats s d = do
   writeIORef (filesSize s) $! _filesSize d
 
 instance StrEncoding FileServerStatsData where
-  strEncode FileServerStatsData {_fromTime, _filesCreated, _fileRecipients, _filesUploaded, _filesExpired, _filesDeleted, _filesDownloaded, _fileDownloads, _fileDownloadAcks, _filesCount, _filesSize} =
+  strEncode FileServerStatsData {_fromTime, _filesCreated, _fileRecipients, _filesUploaded, _filesExpired, _filesDeleted, _filesBlocked, _filesDownloaded, _fileDownloads, _fileDownloadAcks, _filesCount, _filesSize} =
     B.unlines
       [ "fromTime=" <> strEncode _fromTime,
         "filesCreated=" <> strEncode _filesCreated,
@@ -96,6 +101,7 @@ instance StrEncoding FileServerStatsData where
         "filesUploaded=" <> strEncode _filesUploaded,
         "filesExpired=" <> strEncode _filesExpired,
         "filesDeleted=" <> strEncode _filesDeleted,
+        "filesBlocked=" <> strEncode _filesBlocked,
         "filesCount=" <> strEncode _filesCount,
         "filesSize=" <> strEncode _filesSize,
         "filesDownloaded:",
@@ -110,9 +116,10 @@ instance StrEncoding FileServerStatsData where
     _filesUploaded <- "filesUploaded=" *> strP <* A.endOfLine
     _filesExpired <- "filesExpired=" *> strP <* A.endOfLine <|> pure 0
     _filesDeleted <- "filesDeleted=" *> strP <* A.endOfLine
+    _filesBlocked <- "filesBlocked=" *> strP <* A.endOfLine
     _filesCount <- "filesCount=" *> strP <* A.endOfLine <|> pure 0
     _filesSize <- "filesSize=" *> strP <* A.endOfLine <|> pure 0
     _filesDownloaded <- "filesDownloaded:" *> A.endOfLine *> strP <* A.endOfLine
     _fileDownloads <- "fileDownloads=" *> strP <* A.endOfLine
     _fileDownloadAcks <- "fileDownloadAcks=" *> strP <* A.endOfLine
-    pure FileServerStatsData {_fromTime, _filesCreated, _fileRecipients, _filesUploaded, _filesExpired, _filesDeleted, _filesDownloaded, _fileDownloads, _fileDownloadAcks, _filesCount, _filesSize}
+    pure FileServerStatsData {_fromTime, _filesCreated, _fileRecipients, _filesUploaded, _filesExpired, _filesDeleted, _filesBlocked, _filesDownloaded, _fileDownloads, _fileDownloadAcks, _filesCount, _filesSize}

src/Simplex/FileTransfer/Server/Store.hs

@@ -13,6 +13,7 @@ module Simplex.FileTransfer.Server.Store
     setFilePath,
     addRecipient,
     deleteFile,
+    blockFile,
     deleteRecipient,
     expiredFilePath,
     getFile,
@@ -22,6 +23,7 @@ module Simplex.FileTransfer.Server.Store
 where
 
 import Control.Concurrent.STM
+import Control.Monad
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Int (Int64)
 import Data.Set (Set)
@@ -30,8 +32,8 @@ import Simplex.FileTransfer.Protocol (FileInfo (..), SFileParty (..), XFTPFileId
 import Simplex.FileTransfer.Transport (XFTPErrorType (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (RcvPublicAuthKey, RecipientId, SenderId)
-import Simplex.Messaging.Server.QueueStore (RoundedSystemTime (..))
+import Simplex.Messaging.Protocol (BlockingInfo, RcvPublicAuthKey, RecipientId, SenderId)
+import Simplex.Messaging.Server.QueueStore (RoundedSystemTime (..), ServerEntityStatus (..))
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util (ifM, ($>>=))
@@ -47,7 +49,8 @@ data FileRec = FileRec
     fileInfo :: FileInfo,
     filePath :: TVar (Maybe FilePath),
     recipientIds :: TVar (Set RecipientId),
-    createdAt :: RoundedSystemTime
+    createdAt :: RoundedSystemTime,
+    fileStatus :: TVar ServerEntityStatus
   }
 
 fileTimePrecision :: Int64
@@ -78,7 +81,8 @@ newFileRec :: SenderId -> FileInfo -> RoundedSystemTime -> STM FileRec
 newFileRec senderId fileInfo createdAt = do
   recipientIds <- newTVar S.empty
   filePath <- newTVar Nothing
-  pure FileRec {senderId, fileInfo, filePath, recipientIds, createdAt}
+  fileStatus <- newTVar EntityActive
+  pure FileRec {senderId, fileInfo, filePath, recipientIds, createdAt, fileStatus}
 
 setFilePath :: FileStore -> SenderId -> FilePath -> STM (Either XFTPErrorType ())
 setFilePath st sId fPath =
@@ -109,6 +113,14 @@ deleteFile FileStore {files, recipients, usedStorage} senderId = do
       pure $ Right ()
     _ -> pure $ Left AUTH
 
+-- this function must be called after the file is deleted from the file system
+blockFile :: FileStore -> SenderId -> BlockingInfo -> Bool -> STM (Either XFTPErrorType ())
+blockFile st@FileStore {usedStorage} senderId info deleted =
+  withFile st senderId $ \FileRec {fileInfo, fileStatus} -> do
+    when deleted $ modifyTVar' usedStorage $ subtract (fromIntegral $ size fileInfo)
+    writeTVar fileStatus $! EntityBlocked info
+    pure $ Right ()
+
 deleteRecipient :: FileStore -> RecipientId -> FileRec -> STM ()
 deleteRecipient FileStore {recipients} rId FileRec {recipientIds} = do
   TM.delete rId recipients

src/Simplex/FileTransfer/Server/StoreLog.hs

@@ -14,6 +14,7 @@ module Simplex.FileTransfer.Server.StoreLog
     logPutFile,
     logAddRecipients,
     logDeleteFile,
+    logBlockFile,
     logAckFile,
   )
 where
@@ -31,7 +32,7 @@ import qualified Data.Map.Strict as M
 import Simplex.FileTransfer.Protocol (FileInfo (..))
 import Simplex.FileTransfer.Server.Store
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (RcvPublicAuthKey, RecipientId, SenderId)
+import Simplex.Messaging.Protocol (BlockingInfo, RcvPublicAuthKey, RecipientId, SenderId)
 import Simplex.Messaging.Server.QueueStore (RoundedSystemTime)
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.Util (bshow)
@@ -42,7 +43,8 @@ data FileStoreLogRecord
   | PutFile SenderId FilePath
   | AddRecipients SenderId (NonEmpty FileRecipient)
   | DeleteFile SenderId
-  | AckFile RecipientId
+  | BlockFile SenderId BlockingInfo
+  | AckFile RecipientId -- TODO add senderId as well?
   deriving (Show)
 
 instance StrEncoding FileStoreLogRecord where
@@ -51,6 +53,7 @@ instance StrEncoding FileStoreLogRecord where
     PutFile sId path -> strEncode (Str "FPUT", sId, path)
     AddRecipients sId rcps -> strEncode (Str "FADD", sId, rcps)
     DeleteFile sId -> strEncode (Str "FDEL", sId)
+    BlockFile sId info -> strEncode (Str "FBLK", sId, info)
     AckFile rId -> strEncode (Str "FACK", rId)
   strP =
     A.choice
@@ -58,6 +61,7 @@ instance StrEncoding FileStoreLogRecord where
         "FPUT " *> (PutFile <$> strP_ <*> strP),
         "FADD " *> (AddRecipients <$> strP_ <*> strP),
         "FDEL " *> (DeleteFile <$> strP),
+        "FBLK " *> (BlockFile <$> strP_ <*> strP),
         "FACK " *> (AckFile <$> strP)
       ]
 
@@ -76,6 +80,9 @@ logAddRecipients s = logFileStoreRecord s .: AddRecipients
 logDeleteFile :: StoreLog 'WriteMode -> SenderId -> IO ()
 logDeleteFile s = logFileStoreRecord s . DeleteFile
 
+logBlockFile :: StoreLog 'WriteMode -> SenderId -> BlockingInfo -> IO ()
+logBlockFile s fId = logFileStoreRecord s . BlockFile fId
+
 logAckFile :: StoreLog 'WriteMode -> RecipientId -> IO ()
 logAckFile s = logFileStoreRecord s . AckFile
 
@@ -96,6 +103,7 @@ readFileStore f st = mapM_ (addFileLogRecord . LB.toStrict) . LB.lines =<< LB.re
       PutFile qId path -> setFilePath st qId path
       AddRecipients sId rcps -> runExceptT $ addRecipients sId rcps
       DeleteFile sId -> deleteFile st sId
+      BlockFile sId info -> blockFile st sId info True
       AckFile rId -> ackFile st rId
     addRecipients sId rcps = mapM_ (ExceptT . addRecipient st sId) rcps
 

src/Simplex/FileTransfer/Transport.hs

@@ -11,6 +11,7 @@
 module Simplex.FileTransfer.Transport
   ( supportedFileServerVRange,
     authCmdsXFTPVersion,
+    blockedFilesXFTPVersion,
     xftpClientHandshakeStub,
     supportedXFTPhandshakes,
     XFTPClientHandshake (..),
@@ -33,7 +34,6 @@ module Simplex.FileTransfer.Transport
   )
 where
 
-import Control.Applicative ((<|>))
 import qualified Control.Exception as E
 import Control.Logger.Simple
 import Control.Monad
@@ -57,7 +57,7 @@ import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
-import Simplex.Messaging.Protocol (CommandError)
+import Simplex.Messaging.Protocol (BlockingInfo, CommandError)
 import Simplex.Messaging.Transport (ALPN, SessionId, THandle (..), THandleParams (..), TransportError (..), TransportPeer (..))
 import Simplex.Messaging.Transport.HTTP2.File
 import Simplex.Messaging.Util (bshow, tshow)
@@ -92,8 +92,11 @@ initialXFTPVersion = VersionXFTP 1
 authCmdsXFTPVersion :: VersionXFTP
 authCmdsXFTPVersion = VersionXFTP 2
 
+blockedFilesXFTPVersion :: VersionXFTP
+blockedFilesXFTPVersion = VersionXFTP 3
+
 currentXFTPVersion :: VersionXFTP
-currentXFTPVersion = VersionXFTP 2
+currentXFTPVersion = VersionXFTP 3
 
 supportedFileServerVRange :: VersionRangeXFTP
 supportedFileServerVRange = mkVersionRange initialXFTPVersion currentXFTPVersion
@@ -211,6 +214,8 @@ data XFTPErrorType
     CMD {cmdErr :: CommandError}
   | -- | command authorization error - bad signature or non-existing SMP queue
     AUTH
+  | -- | command with the entity that was blocked
+    BLOCKED {blockInfo :: BlockingInfo}
   | -- | incorrent file size
     SIZE
   | -- | storage quota exceeded
@@ -231,15 +236,46 @@ data XFTPErrorType
     INTERNAL
   | -- | used internally, never returned by the server (to be removed)
     DUPLICATE_ -- not part of SMP protocol, used internally
-  deriving (Eq, Read, Show)
+  deriving (Eq, Show)
 
 instance StrEncoding XFTPErrorType where
   strEncode = \case
+    BLOCK -> "BLOCK"
+    SESSION -> "SESSION"
+    HANDSHAKE -> "HANDSHAKE"
     CMD e -> "CMD " <> bshow e
-    e -> bshow e
+    AUTH -> "AUTH"
+    BLOCKED info -> "BLOCKED " <> strEncode info
+    SIZE -> "SIZE"
+    QUOTA -> "QUOTA"
+    DIGEST -> "DIGEST"
+    CRYPTO -> "CRYPTO"
+    NO_FILE -> "NO_FILE"
+    HAS_FILE -> "HAS_FILE"
+    FILE_IO -> "FILE_IO"
+    TIMEOUT -> "TIMEOUT"
+    INTERNAL -> "INTERNAL"
+    DUPLICATE_ -> "DUPLICATE_"
+
   strP =
-    "CMD " *> (CMD <$> parseRead1)
-      <|> parseRead1
+    A.takeTill (== ' ') >>= \case
+      "BLOCK" -> pure BLOCK
+      "SESSION" -> pure SESSION
+      "HANDSHAKE" -> pure HANDSHAKE
+      "CMD" -> CMD <$> parseRead1
+      "AUTH" -> pure AUTH
+      "BLOCKED" -> BLOCKED <$> _strP
+      "SIZE" -> pure SIZE
+      "QUOTA" -> pure QUOTA
+      "DIGEST" -> pure DIGEST
+      "CRYPTO" -> pure CRYPTO
+      "NO_FILE" -> pure NO_FILE
+      "HAS_FILE" -> pure HAS_FILE
+      "FILE_IO" -> pure FILE_IO
+      "TIMEOUT" -> pure TIMEOUT
+      "INTERNAL" -> pure INTERNAL
+      "DUPLICATE_" -> pure DUPLICATE_
+      _ -> fail "bad error type"
 
 instance Encoding XFTPErrorType where
   smpEncode = \case
@@ -248,6 +284,7 @@ instance Encoding XFTPErrorType where
     HANDSHAKE -> "HANDSHAKE"
     CMD err -> "CMD " <> smpEncode err
     AUTH -> "AUTH"
+    BLOCKED info -> "BLOCKED " <> smpEncode info
     SIZE -> "SIZE"
     QUOTA -> "QUOTA"
     DIGEST -> "DIGEST"
@@ -266,6 +303,7 @@ instance Encoding XFTPErrorType where
       "HANDSHAKE" -> pure HANDSHAKE
       "CMD" -> CMD <$> _smpP
       "AUTH" -> pure AUTH
+      "BLOCKED" -> BLOCKED <$> _smpP
       "SIZE" -> pure SIZE
       "QUOTA" -> pure QUOTA
       "DIGEST" -> pure DIGEST

src/Simplex/FileTransfer/Types.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -13,8 +14,6 @@ import Data.Int (Int64)
 import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
 import Data.Word (Word32)
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..))
 import Simplex.FileTransfer.Description
 import qualified Simplex.Messaging.Crypto as C
@@ -24,6 +23,13 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
 import Simplex.Messaging.Protocol (XFTPServer)
 import System.FilePath ((</>))
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple.FromField (FromField (..))
+import Database.PostgreSQL.Simple.ToField (ToField (..))
+#else
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
+#endif
 
 type RcvFileId = ByteString -- Agent entity ID
 
@@ -246,6 +252,16 @@ data DeletedSndChunkReplica = DeletedSndChunkReplica
   }
   deriving (Show)
 
+data SentRecipientReplica = SentRecipientReplica
+  { chunkNo :: Int,
+    server :: XFTPServer,
+    rcvNo :: Int,
+    replicaId :: ChunkReplicaId,
+    replicaKey :: C.APrivateAuthKey,
+    digest :: FileDigest,
+    chunkSize :: FileSize Word32
+  }
+
 data FileErrorType
   = -- | cannot proceed with download from not approved relays without proxy
     NOT_APPROVED

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
@@ -687,9 +689,15 @@ allowConnectionAsync' c corrId connId confId ownConnInfo =
       enqueueCommand c corrId connId (Just server) $ AClientCommand $ LET confId ownConnInfo
     _ -> throwE $ CMD PROHIBITED "allowConnectionAsync"
 
+-- TODO
+-- Unlike `acceptContact` (synchronous version), `acceptContactAsync` uses `unacceptInvitation` in case of error,
+-- because we're not taking lock here. In practice it is less likely to fail because it doesn't involve network IO,
+-- and also it can't be triggered by user concurrently several times in a row. It could be improved similarly to
+-- `acceptContact` by creating a new map for invitation locks and taking lock here, and removing `unacceptInvitation`
+-- while marking invitation as accepted inside "lock level transaction" after successful `joinConnAsync`.
 acceptContactAsync' :: AgentClient -> ACorrId -> Bool -> InvitationId -> ConnInfo -> PQSupport -> SubscriptionMode -> AM ConnId
 acceptContactAsync' c corrId enableNtfs invId ownConnInfo pqSupport subMode = do
-  Invitation {contactConnId, connReq} <- withStore c (`getInvitation` invId)
+  Invitation {contactConnId, connReq} <- withStore c $ \db -> getInvitation db "acceptContactAsync'" invId
   withStore c (`getConn` contactConnId) >>= \case
     SomeConn _ (ContactConnection ConnData {userId} _) -> do
       withStore' c $ \db -> acceptInvitation db invId ownConnInfo
@@ -809,7 +817,7 @@ newConnToJoin c userId connId enableNtfs cReq pqSup = case cReq of
 
 newConnToAccept :: AgentClient -> ConnId -> Bool -> ConfirmationId -> PQSupport -> AM ConnId
 newConnToAccept c connId enableNtfs invId pqSup = do
-  Invitation {connReq, contactConnId} <- withStore c (`getInvitation` invId)
+  Invitation {connReq, contactConnId} <- withStore c $ \db -> getInvitation db "newConnToAccept" invId
   withStore c (`getConn` contactConnId) >>= \case
     SomeConn _ (ContactConnection ConnData {userId} _) ->
       newConnToJoin c userId connId enableNtfs connReq pqSup
@@ -828,26 +836,39 @@ joinConn c userId connId enableNtfs cReq cInfo pqSupport subMode = do
 startJoinInvitation :: AgentClient -> UserId -> ConnId -> Maybe SndQueue -> Bool -> ConnectionRequestUri 'CMInvitation -> PQSupport -> AM (ConnData, SndQueue, CR.SndE2ERatchetParams 'C.X448)
 startJoinInvitation c userId connId sq_ enableNtfs cReqUri pqSup =
   lift (compatibleInvitationUri cReqUri) >>= \case
-    Just (qInfo, Compatible e2eRcvParams@(CR.E2ERatchetParams v _ rcDHRr kem_), Compatible connAgentVersion) -> do
-      g <- asks random
+    Just (qInfo, Compatible e2eRcvParams@(CR.E2ERatchetParams v _ _ _), Compatible connAgentVersion) -> do
+      -- this case avoids re-generating queue keys and subsequent failure of SKEY that timed out
+      -- e2ePubKey is always present, it's Maybe historically
       let pqSupport = pqSup `CR.pqSupportAnd` versionPQSupport_ connAgentVersion (Just v)
+      (sq', e2eSndParams) <- case sq_ of
+        Just sq@SndQueue {e2ePubKey = Just _k} -> do
+          e2eSndParams <-
+            withStore' c (\db -> getSndRatchet db connId v) >>= \case
+              Right r -> pure $ snd r
+              Left e -> do
+                atomically $ writeTBQueue (subQ c) ("", connId, AEvt SAEConn (ERR $ INTERNAL $ "no snd ratchet " <> show e))
+                createRatchet_ pqSupport e2eRcvParams
+          pure (sq, e2eSndParams)
+        _ -> do
+          q <- lift $ fst <$> newSndQueue userId "" qInfo
+          e2eSndParams <- createRatchet_ pqSupport e2eRcvParams
+          withStore c $ \db -> runExceptT $ do
+            sq' <- maybe (ExceptT $ updateNewConnSnd db connId q) pure sq_
+            pure (sq', e2eSndParams)
+      let cData = ConnData {userId, connId, connAgentVersion, enableNtfs, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk, pqSupport}
+      pure (cData, sq', e2eSndParams)
+    Nothing -> throwE $ AGENT A_VERSION
+  where
+    createRatchet_ pqSupport e2eRcvParams@(CR.E2ERatchetParams v _ rcDHRr kem_) = do
+      g <- asks random
       (pk1, pk2, pKem, e2eSndParams) <- liftIO $ CR.generateSndE2EParams g v (CR.replyKEM_ v kem_ pqSupport)
       (_, rcDHRs) <- atomically $ C.generateKeyPair g
       rcParams <- liftEitherWith cryptoError $ CR.pqX3dhSnd pk1 pk2 pKem e2eRcvParams
       maxSupported <- asks $ maxVersion . e2eEncryptVRange . config
       let rcVs = CR.RatchetVersions {current = v, maxSupported}
           rc = CR.initSndRatchet rcVs rcDHRr rcDHRs rcParams
-      -- this case avoids re-generating queue keys and subsequent failure of SKEY that timed out
-      -- e2ePubKey is always present, it's Maybe historically
-      q <- case sq_ of
-        Just sq@SndQueue {e2ePubKey = Just _k} -> pure (sq :: SndQueue) {dbQueueId = DBNewQueue}
-        _ -> lift $ fst <$> newSndQueue userId "" qInfo
-      let cData = ConnData {userId, connId, connAgentVersion, enableNtfs, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk, pqSupport}
-      sq' <- withStore c $ \db -> runExceptT $ do
-        liftIO $ createRatchet db connId rc
-        maybe (ExceptT $ updateNewConnSnd db connId q) pure sq_
-      pure (cData, sq', e2eSndParams)
-    Nothing -> throwE $ AGENT A_VERSION
+      withStore' c $ \db -> createSndRatchet db connId rc e2eSndParams
+      pure e2eSndParams
 
 connRequestPQSupport :: AgentClient -> PQSupport -> ConnectionRequestUri c -> IO (Maybe (VersionSMPA, PQSupport))
 connRequestPQSupport c pqSup cReq = withAgentEnv' c $ case cReq of
@@ -886,6 +907,7 @@ joinConnSrv c userId connId enableNtfs inv@CRInvitationUri {} cInfo pqSup subMod
     case conn of
       NewConnection _ -> doJoin Nothing
       SndConnection _ sq -> doJoin $ Just sq
+      DuplexConnection _ (RcvQueue {status = New} :| _) (sq@SndQueue {status = New} :| _) -> doJoin $ Just sq
       _ -> throwE $ CMD PROHIBITED $ "joinConnSrv: bad connection " <> show cType
   where
     doJoin :: Maybe SndQueue -> AM SndQueueSecured
@@ -941,13 +963,12 @@ allowConnection' c connId confId ownConnInfo = withConnLock c connId "allowConne
 -- | Accept contact (ACPT command) in Reader monad
 acceptContact' :: AgentClient -> ConnId -> Bool -> InvitationId -> ConnInfo -> PQSupport -> SubscriptionMode -> AM SndQueueSecured
 acceptContact' c connId enableNtfs invId ownConnInfo pqSupport subMode = withConnLock c connId "acceptContact" $ do
-  Invitation {contactConnId, connReq} <- withStore c (`getInvitation` invId)
+  Invitation {contactConnId, connReq} <- withStore c $ \db -> getInvitation db "acceptContact'" invId
   withStore c (`getConn` contactConnId) >>= \case
     SomeConn _ (ContactConnection ConnData {userId} _) -> do
+      sqSecured <- joinConn c userId connId enableNtfs connReq ownConnInfo pqSupport subMode
       withStore' c $ \db -> acceptInvitation db invId ownConnInfo
-      joinConn c userId connId enableNtfs connReq ownConnInfo pqSupport subMode `catchAgentError` \err -> do
-        withStore' c (`unacceptInvitation` invId)
-        throwE err
+      pure sqSecured
     _ -> throwE $ CMD PROHIBITED "acceptContact"
 
 -- | Reject contact (RJCT command) in Reader monad
@@ -2149,7 +2170,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

@@ -1,5 +1,6 @@
 {-# LANGUAGE AllowAmbiguousTypes #-}
 {-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
@@ -205,7 +206,6 @@ import Data.Text.Encoding
 import Data.Time (UTCTime, addUTCTime, defaultTimeLocale, formatTime, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Word (Word16)
-import qualified Database.SQLite.Simple as SQL
 import Network.Socket (HostName)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientConfig (..), XFTPClientError)
 import qualified Simplex.FileTransfer.Client as X
@@ -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
@@ -282,6 +282,9 @@ import UnliftIO.Concurrent (forkIO, mkWeakThreadId)
 import UnliftIO.Directory (doesFileExist, getTemporaryDirectory, removeFile)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
+#if !defined(dbPostgres)
+import qualified Database.SQLite.Simple as SQL
+#endif
 
 type ClientVar msg = SessionVar (Either (AgentErrorType, Maybe UTCTime) (Client msg))
 
@@ -555,7 +558,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 +1652,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} =
@@ -1989,6 +1992,13 @@ withStore c action = do
   withExceptT storeError . ExceptT . liftIO . agentOperationBracket c AODatabase (\_ -> pure ()) $
     withTransaction st action `E.catches` handleDBErrors
   where
+#if defined(dbPostgres)
+    -- TODO [postgres] postgres specific error handling
+    handleDBErrors :: [E.Handler IO (Either StoreError a)]
+    handleDBErrors =
+      [ E.Handler $ \(E.SomeException e) -> pure . Left $ SEInternal $ bshow e
+      ]
+#else
     handleDBErrors :: [E.Handler IO (Either StoreError a)]
     handleDBErrors =
       [ E.Handler $ \(e :: SQL.SQLError) ->
@@ -1997,6 +2007,7 @@ withStore c action = do
            in pure . Left . (if busy then SEDatabaseBusy else SEInternal) $ bshow se,
         E.Handler $ \(E.SomeException e) -> pure . Left $ SEInternal $ bshow e
       ]
+#endif
 
 withStoreBatch :: Traversable t => AgentClient -> (DB.Connection -> t (IO (Either AgentErrorType a))) -> AM' (t (Either AgentErrorType a))
 withStoreBatch c actions = do
@@ -2020,7 +2031,7 @@ storeError = \case
   SEConnDuplicate -> CONN DUPLICATE
   SEBadConnType CRcv -> CONN SIMPLEX
   SEBadConnType CSnd -> CONN SIMPLEX
-  SEInvitationNotFound -> CMD PROHIBITED "SEInvitationNotFound"
+  SEInvitationNotFound cxt invId -> CMD PROHIBITED $ "SEInvitationNotFound " <> cxt <> ", invitationId = " <> show invId
   -- this error is never reported as store error,
   -- it is used to wrap agent operations when "transaction-like" store access is needed
   -- NOTE: network IO should NOT be used inside AgentStoreMonad
@@ -2044,7 +2055,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 +2108,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,11 @@ import Simplex.Messaging.Util (allFinally, catchAllErrors, catchAllErrors', tryA
 import System.Mem.Weak (Weak)
 import System.Random (StdGen, newStdGen)
 import UnliftIO.STM
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple (ConnectInfo (..))
+#else
+import Data.ByteArray (ScrubbedBytes)
+#endif
 
 type AM' a = ReaderT Env IO a
 
@@ -254,7 +260,7 @@ defaultAgentConfig =
 
 data Env = Env
   { config :: AgentConfig,
-    store :: SQLiteStore,
+    store :: DBStore,
     random :: TVar ChaChaDRG,
     randomServer :: TVar StdGen,
     ntfSupervisor :: NtfSupervisor,
@@ -262,7 +268,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 +277,13 @@ 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)
+createAgentStore :: ConnectInfo -> String -> MigrationConfirmation -> IO (Either MigrationError DBStore)
+createAgentStore = createStore
+#else
+createAgentStore :: FilePath -> ScrubbedBytes -> Bool -> MigrationConfirmation -> Bool -> 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/Protocol.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DuplicateRecordFields #-}
@@ -167,13 +168,12 @@ import Data.Time.Clock.System (SystemTime)
 import Data.Type.Equality
 import Data.Typeable ()
 import Data.Word (Word16, Word32)
-import Database.SQLite.Simple.FromField
-import Database.SQLite.Simple.ToField
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..))
 import Simplex.FileTransfer.Transport (XFTPErrorType)
 import Simplex.FileTransfer.Types (FileErrorType)
 import Simplex.Messaging.Agent.QueryString
+import Simplex.Messaging.Agent.Store.DB (Binary (..))
 import Simplex.Messaging.Client (ProxyClientError)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet
@@ -224,6 +224,13 @@ import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
 import Simplex.RemoteControl.Types
 import UnliftIO.Exception (Exception)
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple.FromField (FromField (..))
+import Database.PostgreSQL.Simple.ToField (ToField (..))
+#else
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
+#endif
 
 -- SMP agent protocol version history:
 -- 1 - binary protocol encoding (1/1/2022)
@@ -644,7 +651,7 @@ instance ToJSON NotificationsMode where
 instance FromJSON NotificationsMode where
   parseJSON = strParseJSON "NotificationsMode"
 
-instance ToField NotificationsMode where toField = toField . strEncode
+instance ToField NotificationsMode where toField = toField . Binary . strEncode
 
 instance FromField NotificationsMode where fromField = blobFieldDecoder $ parseAll strP
 

src/Simplex/Messaging/Agent/Stats.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -10,13 +11,18 @@ import qualified Data.Aeson.TH as J
 import Data.Int (Int64)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
 import Simplex.Messaging.Agent.Protocol (UserId)
 import Simplex.Messaging.Parsers (defaultJSON, fromTextField_)
-import Simplex.Messaging.Protocol (SMPServer, XFTPServer, NtfServer)
+import Simplex.Messaging.Protocol (NtfServer, SMPServer, XFTPServer)
 import Simplex.Messaging.Util (decodeJSON, encodeJSON)
 import UnliftIO.STM
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple.FromField (FromField (..))
+import Database.PostgreSQL.Simple.ToField (ToField (..))
+#else
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
+#endif
 
 data AgentSMPServerStats = AgentSMPServerStats
   { sentDirect :: TVar Int, -- successfully sent messages

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,33 @@ import Simplex.Messaging.Protocol
     RcvDhSecret,
     RcvNtfDhSecret,
     RcvPrivateAuthKey,
+    SenderCanSecure,
     SndPrivateAuthKey,
     SndPublicAuthKey,
-    SenderCanSecure,
     VersionSMPC,
   )
 import qualified Simplex.Messaging.Protocol as SMP
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple (ConnectInfo (..))
+import qualified Simplex.Messaging.Agent.Store.Postgres as Store
+#else
+import Data.ByteArray (ScrubbedBytes)
+import qualified Simplex.Messaging.Agent.Store.SQLite as Store
+#endif
+
+#if defined(dbPostgres)
+createStore :: ConnectInfo -> String -> MigrationConfirmation -> IO (Either MigrationError DBStore)
+createStore connectInfo schema = Store.createDBStore connectInfo schema Migrations.app
+#else
+createStore :: FilePath -> ScrubbedBytes -> Bool -> MigrationConfirmation -> Bool -> IO (Either MigrationError DBStore)
+createStore dbFilePath dbKey keepKey = Store.createDBStore dbFilePath dbKey keepKey Migrations.app
+#endif
+
+closeStore :: DBStore -> IO ()
+closeStore = Store.closeDBStore
+
+execSQL :: DB.Connection -> Text -> IO [Text]
+execSQL = Store.execSQL
 
 -- * Queue types
 
@@ -623,7 +650,7 @@ data StoreError
   | -- | Confirmation not found.
     SEConfirmationNotFound
   | -- | Invitation not found
-    SEInvitationNotFound
+    SEInvitationNotFound String InvitationId
   | -- | Message not found
     SEMsgNotFound
   | -- | Command not found

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 -> Bool -> IO (Either MigrationError ())
+migrateSchema st migrations confirmMigrations vacuum = 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 vacuum ms $> Right ()
+      | otherwise -> case confirmMigrations of
+          MCYesUp -> runWithBackup st vacuum ms
+          MCYesUpDown -> runWithBackup st vacuum ms
+          MCConsole -> confirm err >> runWithBackup st vacuum 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 vacuum ms
+      MCConsole -> confirm err >> runWithBackup st vacuum ms
+      MCYesUp -> pure $ Left err
+      MCError -> pure $ Left err
+      where
+        err = MEDowngrade $ map downName dms
+  where
+    confirm err = confirmOrExit $ migrationErrorDescription err
+
+runWithBackup :: DBStore -> Bool -> MigrationsToRun -> IO (Either a ())
+#if defined(dbPostgres)
+runWithBackup st vacuum ms = Migrations.run st vacuum ms $> Right ()
+#else
+runWithBackup st vacuum ms = do
+  let f = dbFilePath st
+  copyFile f (f <> ".bak")
+  Migrations.run st vacuum 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,86 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Agent.Store.Postgres
+  ( createDBStore,
+    closeDBStore,
+    execSQL
+  )
+where
+
+import Control.Exception (throwIO)
+import Control.Monad (unless, void)
+import Data.Functor (($>))
+import Data.String (fromString)
+import Data.Text (Text)
+import Database.PostgreSQL.Simple (ConnectInfo (..), Only (..), defaultConnectInfo)
+import qualified Database.PostgreSQL.Simple as PSQL
+import Database.PostgreSQL.Simple.SqlQQ (sql)
+import Simplex.Messaging.Agent.Store.Migrations (migrateSchema)
+import Simplex.Messaging.Agent.Store.Postgres.Common
+import qualified Simplex.Messaging.Agent.Store.Postgres.DB as DB
+import Simplex.Messaging.Agent.Store.Postgres.Util (createDBAndUserIfNotExists)
+import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationError (..))
+import Simplex.Messaging.Util (ifM)
+import UnliftIO.Exception (onException)
+import UnliftIO.MVar
+import UnliftIO.STM
+
+-- | Create a new Postgres DBStore with the given connection info, schema name and migrations.
+-- This function creates the user and/or database passed in connectInfo if they do not exist
+-- (expects the default 'postgres' user and 'postgres' db to exist).
+-- If passed schema does not exist in connectInfo database, it will be created.
+-- Applies necessary migrations to schema.
+-- TODO [postgres] authentication / user password, db encryption (?)
+createDBStore :: ConnectInfo -> String -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError DBStore)
+createDBStore connectInfo schema migrations confirmMigrations = do
+  createDBAndUserIfNotExists connectInfo
+  st <- connectPostgresStore connectInfo schema
+  r <- migrateSchema st migrations confirmMigrations True `onException` closeDBStore st
+  case r of
+    Right () -> pure $ Right st
+    Left e -> closeDBStore st $> Left e
+
+connectPostgresStore :: ConnectInfo -> String -> IO DBStore
+connectPostgresStore dbConnectInfo schema = do
+  (dbConn, dbNew) <- connectDB dbConnectInfo schema -- TODO [postgres] analogue for dbBusyLoop?
+  dbConnection <- newMVar dbConn
+  dbClosed <- newTVarIO False
+  pure DBStore {dbConnectInfo, dbConnection, dbNew, dbClosed}
+
+connectDB :: ConnectInfo -> String -> IO (DB.Connection, Bool)
+connectDB dbConnectInfo schema = do
+  db <- PSQL.connect dbConnectInfo
+  schemaExists <- prepare db `onException` PSQL.close db
+  let dbNew = not schemaExists
+  pure (db, dbNew)
+  where
+    prepare db = do
+      void $ PSQL.execute_ db "SET client_min_messages TO WARNING"
+      [Only schemaExists] <-
+        PSQL.query
+          db
+          [sql|
+            SELECT EXISTS (
+              SELECT 1 FROM pg_catalog.pg_namespace
+              WHERE nspname = ?
+            )
+          |]
+          (Only schema)
+      unless schemaExists $ void $ PSQL.execute_ db (fromString $ "CREATE SCHEMA " <> schema)
+      void $ PSQL.execute_ db (fromString $ "SET search_path TO " <> schema)
+      pure schemaExists
+
+-- can share with SQLite
+closeDBStore :: DBStore -> IO ()
+closeDBStore st@DBStore {dbClosed} =
+  ifM (readTVarIO dbClosed) (putStrLn "closeDBStore: already closed") $
+    withConnection st $ \conn -> do
+      DB.close conn
+      atomically $ writeTVar dbClosed True
+
+-- TODO [postgres] not necessary for postgres (used for ExecAgentStoreSQL, ExecChatStoreSQL)
+execSQL :: PSQL.Connection -> Text -> IO [Text]
+execSQL _db _query = throwIO (userError "not implemented")

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] 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,63 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.DB
+  ( BoolInt (..),
+    PSQL.Binary (..),
+    PSQL.Connection,
+    PSQL.connect,
+    PSQL.close,
+    execute,
+    execute_,
+    executeMany,
+    PSQL.query,
+    PSQL.query_,
+  )
+where
+
+import Control.Monad (void)
+import Data.Int (Int32, Int64)
+import Data.Word (Word16, Word32)
+import Database.PostgreSQL.Simple (ResultError (..))
+import qualified Database.PostgreSQL.Simple as PSQL
+import Database.PostgreSQL.Simple.FromField (FromField (..), returnError)
+import Database.PostgreSQL.Simple.ToField (ToField (..))
+
+newtype BoolInt = BI {unBI :: Bool}
+
+instance FromField BoolInt where
+  fromField field dat = BI . (/= (0 :: Int)) <$> fromField field dat
+  {-# INLINE fromField #-}
+
+instance ToField BoolInt where
+  toField (BI b) = toField ((if b then 1 else 0) :: Int)
+  {-# INLINE toField #-}
+
+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 #-}
+
+-- orphan instances
+
+-- used in FileSize
+instance FromField Word32 where
+  fromField field dat = do
+    i <- fromField field dat
+    if i >= (0 :: Int64)
+      then pure (fromIntegral i :: Word32)
+      else returnError ConversionFailed field "Negative value can't be converted to Word32"
+
+-- used in Version
+instance FromField Word16 where
+  fromField field dat = do
+    i <- fromField field dat
+    if i >= (0 :: Int32)
+      then pure (fromIntegral i :: Word16)
+      else returnError ConversionFailed field "Negative value can't be converted to Word16"

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

@@ -0,0 +1,78 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE TupleSections #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations
+  ( app,
+    initialize,
+    run,
+    getCurrent,
+  )
+where
+
+import Control.Monad (void)
+import Data.List (sortOn)
+import Data.Text (Text)
+import qualified Data.Text as T
+import qualified Data.Text.Encoding as TE
+import Data.Time.Clock (getCurrentTime)
+import qualified Database.PostgreSQL.LibPQ as LibPQ
+import Database.PostgreSQL.Simple (Only (..))
+import qualified Database.PostgreSQL.Simple as PSQL
+import Database.PostgreSQL.Simple.Internal (Connection (..))
+import Database.PostgreSQL.Simple.SqlQQ (sql)
+import Simplex.Messaging.Agent.Store.Postgres.Common
+import Simplex.Messaging.Agent.Store.Postgres.Migrations.M20241210_initial
+import Simplex.Messaging.Agent.Store.Shared
+import UnliftIO.MVar
+
+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}
+
+initialize :: DBStore -> IO ()
+initialize st = withTransaction' st $ \db ->
+  void $
+    PSQL.execute_
+      db
+      [sql|
+      CREATE TABLE IF NOT EXISTS migrations (
+        name TEXT NOT NULL,
+        ts TIMESTAMP NOT NULL,
+        down TEXT,
+        PRIMARY KEY (name)
+      )
+    |]
+
+run :: DBStore -> Bool -> MigrationsToRun -> IO ()
+run st _vacuum = \case
+  MTRUp [] -> pure ()
+  MTRUp ms -> mapM_ runUp ms
+  MTRDown ms -> mapM_ runDown $ reverse ms
+  MTRNone -> pure ()
+  where
+    runUp Migration {name, up, down} = withTransaction' st $ \db -> do
+      insert db
+      execSQL db up
+      where
+        insert db = void $ PSQL.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
+    runDown DownMigration {downName, downQuery} = withTransaction' st $ \db -> do
+      execSQL db downQuery
+      void $ PSQL.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
+    execSQL db query =
+      withMVar (connectionHandle db) $ \pqConn ->
+        void $ LibPQ.exec pqConn (TE.encodeUtf8 query)
+
+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

@@ -0,0 +1,545 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Migrations.M20241210_initial where
+
+import Data.Text (Text)
+import qualified Data.Text as T
+import Text.RawString.QQ (r)
+
+m20241210_initial :: Text
+m20241210_initial =
+  T.pack
+    [r|
+CREATE TABLE users(
+  user_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  deleted SMALLINT NOT NULL DEFAULT 0
+);
+CREATE TABLE servers(
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  key_hash BYTEA NOT NULL,
+  PRIMARY KEY(host, port)
+);
+CREATE TABLE connections(
+  conn_id BYTEA NOT NULL PRIMARY KEY,
+  conn_mode TEXT NOT NULL,
+  last_internal_msg_id BIGINT NOT NULL DEFAULT 0,
+  last_internal_rcv_msg_id BIGINT NOT NULL DEFAULT 0,
+  last_internal_snd_msg_id BIGINT NOT NULL DEFAULT 0,
+  last_external_snd_msg_id BIGINT NOT NULL DEFAULT 0,
+  last_rcv_msg_hash BYTEA NOT NULL DEFAULT ''::BYTEA,
+  last_snd_msg_hash BYTEA NOT NULL DEFAULT ''::BYTEA,
+  smp_agent_version INTEGER NOT NULL DEFAULT 1,
+  duplex_handshake SMALLINT NULL DEFAULT 0,
+  enable_ntfs SMALLINT,
+  deleted SMALLINT NOT NULL DEFAULT 0,
+  user_id BIGINT NOT NULL REFERENCES users ON DELETE CASCADE,
+  ratchet_sync_state TEXT NOT NULL DEFAULT 'ok',
+  deleted_at_wait_delivery TIMESTAMPTZ,
+  pq_support SMALLINT NOT NULL DEFAULT 0
+);
+CREATE TABLE rcv_queues(
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  rcv_id BYTEA NOT NULL,
+  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
+  rcv_private_key BYTEA NOT NULL,
+  rcv_dh_secret BYTEA NOT NULL,
+  e2e_priv_key BYTEA NOT NULL,
+  e2e_dh_secret BYTEA,
+  snd_id BYTEA NOT NULL,
+  snd_key BYTEA,
+  status TEXT NOT NULL,
+  smp_server_version INTEGER NOT NULL DEFAULT 1,
+  smp_client_version INTEGER,
+  ntf_public_key BYTEA,
+  ntf_private_key BYTEA,
+  ntf_id BYTEA,
+  rcv_ntf_dh_secret BYTEA,
+  rcv_queue_id BIGINT NOT NULL,
+  rcv_primary SMALLINT NOT NULL,
+  replace_rcv_queue_id BIGINT NULL,
+  delete_errors BIGINT NOT NULL DEFAULT 0,
+  server_key_hash BYTEA,
+  switch_status TEXT,
+  deleted SMALLINT NOT NULL DEFAULT 0,
+  snd_secure SMALLINT NOT NULL DEFAULT 0,
+  last_broker_ts TIMESTAMPTZ,
+  PRIMARY KEY(host, port, rcv_id),
+  FOREIGN KEY(host, port) REFERENCES servers
+  ON DELETE RESTRICT ON UPDATE CASCADE,
+  UNIQUE(host, port, snd_id)
+);
+CREATE TABLE snd_queues(
+  host TEXT NOT NULL,
+  port TEXT NOT NULL,
+  snd_id BYTEA NOT NULL,
+  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
+  snd_private_key BYTEA NOT NULL,
+  e2e_dh_secret BYTEA NOT NULL,
+  status TEXT NOT NULL,
+  smp_server_version INTEGER NOT NULL DEFAULT 1,
+  smp_client_version INTEGER NOT NULL DEFAULT 1,
+  snd_public_key BYTEA,
+  e2e_pub_key BYTEA,
+  snd_queue_id BIGINT NOT NULL,
+  snd_primary SMALLINT NOT NULL,
+  replace_snd_queue_id BIGINT NULL,
+  server_key_hash BYTEA,
+  switch_status TEXT,
+  snd_secure SMALLINT NOT NULL DEFAULT 0,
+  PRIMARY KEY(host, port, snd_id),
+  FOREIGN KEY(host, port) REFERENCES servers
+  ON DELETE RESTRICT ON UPDATE CASCADE
+);
+CREATE TABLE messages(
+  conn_id BYTEA NOT NULL REFERENCES connections(conn_id)
+  ON DELETE CASCADE,
+  internal_id BIGINT NOT NULL,
+  internal_ts TIMESTAMPTZ NOT NULL,
+  internal_rcv_id BIGINT,
+  internal_snd_id BIGINT,
+  msg_type BYTEA NOT NULL,
+  msg_body BYTEA NOT NULL DEFAULT ''::BYTEA,
+  msg_flags TEXT NULL,
+  pq_encryption SMALLINT NOT NULL DEFAULT 0,
+  PRIMARY KEY(conn_id, internal_id)
+);
+CREATE TABLE rcv_messages(
+  conn_id BYTEA NOT NULL,
+  internal_rcv_id BIGINT NOT NULL,
+  internal_id BIGINT NOT NULL,
+  external_snd_id BIGINT NOT NULL,
+  broker_id BYTEA NOT NULL,
+  broker_ts TIMESTAMPTZ NOT NULL,
+  internal_hash BYTEA NOT NULL,
+  external_prev_snd_hash BYTEA NOT NULL,
+  integrity BYTEA NOT NULL,
+  user_ack SMALLINT NULL DEFAULT 0,
+  rcv_queue_id BIGINT NOT NULL,
+  PRIMARY KEY(conn_id, internal_rcv_id),
+  FOREIGN KEY(conn_id, internal_id) REFERENCES messages
+  ON DELETE CASCADE
+);
+ALTER TABLE messages
+ADD CONSTRAINT fk_messages_rcv_messages
+  FOREIGN KEY (conn_id, internal_rcv_id) REFERENCES rcv_messages
+  ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED;
+CREATE TABLE snd_messages(
+  conn_id BYTEA NOT NULL,
+  internal_snd_id BIGINT NOT NULL,
+  internal_id BIGINT NOT NULL,
+  internal_hash BYTEA NOT NULL,
+  previous_msg_hash BYTEA NOT NULL DEFAULT ''::BYTEA,
+  retry_int_slow BIGINT,
+  retry_int_fast BIGINT,
+  rcpt_internal_id BIGINT,
+  rcpt_status TEXT,
+  PRIMARY KEY(conn_id, internal_snd_id),
+  FOREIGN KEY(conn_id, internal_id) REFERENCES messages
+  ON DELETE CASCADE
+);
+ALTER TABLE messages
+ADD CONSTRAINT fk_messages_snd_messages
+  FOREIGN KEY (conn_id, internal_snd_id) REFERENCES snd_messages
+  ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED;
+CREATE TABLE conn_confirmations(
+  confirmation_id BYTEA NOT NULL PRIMARY KEY,
+  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
+  e2e_snd_pub_key BYTEA NOT NULL,
+  sender_key BYTEA,
+  ratchet_state BYTEA NOT NULL,
+  sender_conn_info BYTEA NOT NULL,
+  accepted SMALLINT NOT NULL,
+  own_conn_info BYTEA,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  smp_reply_queues BYTEA NULL,
+  smp_client_version INTEGER
+);
+CREATE TABLE conn_invitations(
+  invitation_id BYTEA NOT NULL PRIMARY KEY,
+  contact_conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
+  cr_invitation BYTEA NOT NULL,
+  recipient_conn_info BYTEA NOT NULL,
+  accepted SMALLINT NOT NULL DEFAULT 0,
+  own_conn_info BYTEA,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+CREATE TABLE ratchets(
+  conn_id BYTEA NOT NULL PRIMARY KEY REFERENCES connections
+  ON DELETE CASCADE,
+  x3dh_priv_key_1 BYTEA,
+  x3dh_priv_key_2 BYTEA,
+  ratchet_state BYTEA,
+  e2e_version INTEGER NOT NULL DEFAULT 1,
+  x3dh_pub_key_1 BYTEA,
+  x3dh_pub_key_2 BYTEA,
+  pq_priv_kem BYTEA,
+  pq_pub_kem BYTEA
+);
+CREATE TABLE skipped_messages(
+  skipped_message_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  conn_id BYTEA NOT NULL REFERENCES ratchets
+  ON DELETE CASCADE,
+  header_key BYTEA NOT NULL,
+  msg_n BIGINT NOT NULL,
+  msg_key BYTEA NOT NULL
+);
+CREATE TABLE ntf_servers(
+  ntf_host TEXT NOT NULL,
+  ntf_port TEXT NOT NULL,
+  ntf_key_hash BYTEA NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  PRIMARY KEY(ntf_host, ntf_port)
+);
+CREATE TABLE ntf_tokens(
+  provider TEXT NOT NULL,
+  device_token TEXT NOT NULL,
+  ntf_host TEXT NOT NULL,
+  ntf_port TEXT NOT NULL,
+  tkn_id BYTEA,
+  tkn_pub_key BYTEA NOT NULL,
+  tkn_priv_key BYTEA NOT NULL,
+  tkn_pub_dh_key BYTEA NOT NULL,
+  tkn_priv_dh_key BYTEA NOT NULL,
+  tkn_dh_secret BYTEA,
+  tkn_status TEXT NOT NULL,
+  tkn_action BYTEA,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  ntf_mode TEXT NULL,
+  PRIMARY KEY(provider, device_token, ntf_host, ntf_port),
+  FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers
+  ON DELETE RESTRICT ON UPDATE CASCADE
+);
+CREATE TABLE ntf_subscriptions(
+  conn_id BYTEA NOT NULL,
+  smp_host TEXT NULL,
+  smp_port TEXT NULL,
+  smp_ntf_id BYTEA,
+  ntf_host TEXT NOT NULL,
+  ntf_port TEXT NOT NULL,
+  ntf_sub_id BYTEA,
+  ntf_sub_status TEXT NOT NULL,
+  ntf_sub_action TEXT,
+  ntf_sub_smp_action TEXT,
+  ntf_sub_action_ts TIMESTAMPTZ,
+  updated_by_supervisor SMALLINT NOT NULL DEFAULT 0,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  smp_server_key_hash BYTEA,
+  ntf_failed SMALLINT DEFAULT 0,
+  smp_failed SMALLINT DEFAULT 0,
+  PRIMARY KEY(conn_id),
+  FOREIGN KEY(smp_host, smp_port) REFERENCES servers(host, port)
+  ON DELETE SET NULL ON UPDATE CASCADE,
+  FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers
+  ON DELETE RESTRICT ON UPDATE CASCADE
+);
+CREATE TABLE commands(
+  command_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
+  host TEXT,
+  port TEXT,
+  corr_id BYTEA NOT NULL,
+  command_tag BYTEA NOT NULL,
+  command BYTEA NOT NULL,
+  agent_version INTEGER NOT NULL DEFAULT 1,
+  server_key_hash BYTEA,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT '1970-01-01 00:00:00',
+  failed SMALLINT DEFAULT 0,
+  FOREIGN KEY(host, port) REFERENCES servers
+  ON DELETE RESTRICT ON UPDATE CASCADE
+);
+CREATE TABLE snd_message_deliveries(
+  snd_message_delivery_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
+  snd_queue_id BIGINT NOT NULL,
+  internal_id BIGINT NOT NULL,
+  failed SMALLINT DEFAULT 0,
+  FOREIGN KEY(conn_id, internal_id) REFERENCES messages ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED
+);
+CREATE TABLE xftp_servers(
+  xftp_server_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  xftp_host TEXT NOT NULL,
+  xftp_port TEXT NOT NULL,
+  xftp_key_hash BYTEA NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  UNIQUE(xftp_host, xftp_port, xftp_key_hash)
+);
+CREATE TABLE rcv_files(
+  rcv_file_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  rcv_file_entity_id BYTEA NOT NULL,
+  user_id BIGINT NOT NULL REFERENCES users ON DELETE CASCADE,
+  size BIGINT NOT NULL,
+  digest BYTEA NOT NULL,
+  key BYTEA NOT NULL,
+  nonce BYTEA NOT NULL,
+  chunk_size BIGINT NOT NULL,
+  prefix_path TEXT NOT NULL,
+  tmp_path TEXT,
+  save_path TEXT NOT NULL,
+  status TEXT NOT NULL,
+  deleted SMALLINT NOT NULL DEFAULT 0,
+  error TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  save_file_key BYTEA,
+  save_file_nonce BYTEA,
+  failed SMALLINT DEFAULT 0,
+  redirect_id BIGINT REFERENCES rcv_files ON DELETE SET NULL,
+  redirect_entity_id BYTEA,
+  redirect_size BIGINT,
+  redirect_digest BYTEA,
+  approved_relays SMALLINT NOT NULL DEFAULT 0,
+  UNIQUE(rcv_file_entity_id)
+);
+CREATE TABLE rcv_file_chunks(
+  rcv_file_chunk_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  rcv_file_id BIGINT NOT NULL REFERENCES rcv_files ON DELETE CASCADE,
+  chunk_no BIGINT NOT NULL,
+  chunk_size BIGINT NOT NULL,
+  digest BYTEA NOT NULL,
+  tmp_path TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+CREATE TABLE rcv_file_chunk_replicas(
+  rcv_file_chunk_replica_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  rcv_file_chunk_id BIGINT NOT NULL REFERENCES rcv_file_chunks ON DELETE CASCADE,
+  replica_number BIGINT NOT NULL,
+  xftp_server_id BIGINT NOT NULL REFERENCES xftp_servers ON DELETE CASCADE,
+  replica_id BYTEA NOT NULL,
+  replica_key BYTEA NOT NULL,
+  received SMALLINT NOT NULL DEFAULT 0,
+  delay BIGINT,
+  retries BIGINT NOT NULL DEFAULT 0,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+CREATE TABLE snd_files(
+  snd_file_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  snd_file_entity_id BYTEA NOT NULL,
+  user_id BIGINT NOT NULL REFERENCES users ON DELETE CASCADE,
+  num_recipients BIGINT NOT NULL,
+  digest BYTEA,
+  key BYTEA NOT NUll,
+  nonce BYTEA NOT NUll,
+  path TEXT NOT NULL,
+  prefix_path TEXT,
+  status TEXT NOT NULL,
+  deleted SMALLINT NOT NULL DEFAULT 0,
+  error TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  src_file_key BYTEA,
+  src_file_nonce BYTEA,
+  failed SMALLINT DEFAULT 0,
+  redirect_size BIGINT,
+  redirect_digest BYTEA
+);
+CREATE TABLE snd_file_chunks(
+  snd_file_chunk_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  snd_file_id BIGINT NOT NULL REFERENCES snd_files ON DELETE CASCADE,
+  chunk_no BIGINT NOT NULL,
+  chunk_offset BIGINT NOT NULL,
+  chunk_size BIGINT NOT NULL,
+  digest BYTEA NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+CREATE TABLE snd_file_chunk_replicas(
+  snd_file_chunk_replica_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  snd_file_chunk_id BIGINT NOT NULL REFERENCES snd_file_chunks ON DELETE CASCADE,
+  replica_number BIGINT NOT NULL,
+  xftp_server_id BIGINT NOT NULL REFERENCES xftp_servers ON DELETE CASCADE,
+  replica_id BYTEA NOT NULL,
+  replica_key BYTEA NOT NULL,
+  replica_status TEXT NOT NULL,
+  delay BIGINT,
+  retries BIGINT NOT NULL DEFAULT 0,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+CREATE TABLE snd_file_chunk_replica_recipients(
+  snd_file_chunk_replica_recipient_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  snd_file_chunk_replica_id BIGINT NOT NULL REFERENCES snd_file_chunk_replicas ON DELETE CASCADE,
+  rcv_replica_id BYTEA NOT NULL,
+  rcv_replica_key BYTEA NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+CREATE TABLE deleted_snd_chunk_replicas(
+  deleted_snd_chunk_replica_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  user_id BIGINT NOT NULL REFERENCES users ON DELETE CASCADE,
+  xftp_server_id BIGINT NOT NULL REFERENCES xftp_servers ON DELETE CASCADE,
+  replica_id BYTEA NOT NULL,
+  replica_key BYTEA NOT NULL,
+  chunk_digest BYTEA NOT NULL,
+  delay BIGINT,
+  retries BIGINT NOT NULL DEFAULT 0,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  failed SMALLINT DEFAULT 0
+);
+CREATE TABLE encrypted_rcv_message_hashes(
+  encrypted_rcv_message_hash_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
+  hash BYTEA NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+CREATE TABLE processed_ratchet_key_hashes(
+  processed_ratchet_key_hash_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  conn_id BYTEA NOT NULL REFERENCES connections ON DELETE CASCADE,
+  hash BYTEA NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+CREATE TABLE servers_stats(
+  servers_stats_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  servers_stats TEXT,
+  started_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now()),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+INSERT INTO servers_stats DEFAULT VALUES;
+CREATE TABLE ntf_tokens_to_delete(
+  ntf_token_to_delete_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+  ntf_host TEXT NOT NULL,
+  ntf_port TEXT NOT NULL,
+  ntf_key_hash BYTEA NOT NULL,
+  tkn_id BYTEA NOT NULL,
+  tkn_priv_key BYTEA NOT NULL,
+  del_failed SMALLINT DEFAULT 0,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT (now())
+);
+CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id);
+CREATE UNIQUE INDEX idx_rcv_queue_id ON rcv_queues(conn_id, rcv_queue_id);
+CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(conn_id, snd_queue_id);
+CREATE INDEX idx_snd_message_deliveries ON snd_message_deliveries(
+  conn_id,
+  snd_queue_id
+);
+CREATE INDEX idx_connections_user ON connections(user_id);
+CREATE INDEX idx_commands_conn_id ON commands(conn_id);
+CREATE INDEX idx_commands_host_port ON commands(host, port);
+CREATE INDEX idx_conn_confirmations_conn_id ON conn_confirmations(conn_id);
+CREATE INDEX idx_conn_invitations_contact_conn_id ON conn_invitations(
+  contact_conn_id
+);
+CREATE INDEX idx_messages_conn_id_internal_snd_id ON messages(
+  conn_id,
+  internal_snd_id
+);
+CREATE INDEX idx_messages_conn_id_internal_rcv_id ON messages(
+  conn_id,
+  internal_rcv_id
+);
+CREATE INDEX idx_messages_conn_id ON messages(conn_id);
+CREATE INDEX idx_ntf_subscriptions_ntf_host_ntf_port ON ntf_subscriptions(
+  ntf_host,
+  ntf_port
+);
+CREATE INDEX idx_ntf_subscriptions_smp_host_smp_port ON ntf_subscriptions(
+  smp_host,
+  smp_port
+);
+CREATE INDEX idx_ntf_tokens_ntf_host_ntf_port ON ntf_tokens(
+  ntf_host,
+  ntf_port
+);
+CREATE INDEX idx_ratchets_conn_id ON ratchets(conn_id);
+CREATE INDEX idx_rcv_messages_conn_id_internal_id ON rcv_messages(
+  conn_id,
+  internal_id
+);
+CREATE INDEX idx_skipped_messages_conn_id ON skipped_messages(conn_id);
+CREATE INDEX idx_snd_message_deliveries_conn_id_internal_id ON snd_message_deliveries(
+  conn_id,
+  internal_id
+);
+CREATE INDEX idx_snd_messages_conn_id_internal_id ON snd_messages(
+  conn_id,
+  internal_id
+);
+CREATE INDEX idx_snd_queues_host_port ON snd_queues(host, port);
+CREATE INDEX idx_rcv_files_user_id ON rcv_files(user_id);
+CREATE INDEX idx_rcv_file_chunks_rcv_file_id ON rcv_file_chunks(rcv_file_id);
+CREATE INDEX idx_rcv_file_chunk_replicas_rcv_file_chunk_id ON rcv_file_chunk_replicas(
+  rcv_file_chunk_id
+);
+CREATE INDEX idx_rcv_file_chunk_replicas_xftp_server_id ON rcv_file_chunk_replicas(
+  xftp_server_id
+);
+CREATE INDEX idx_snd_files_user_id ON snd_files(user_id);
+CREATE INDEX idx_snd_file_chunks_snd_file_id ON snd_file_chunks(snd_file_id);
+CREATE INDEX idx_snd_file_chunk_replicas_snd_file_chunk_id ON snd_file_chunk_replicas(
+  snd_file_chunk_id
+);
+CREATE INDEX idx_snd_file_chunk_replicas_xftp_server_id ON snd_file_chunk_replicas(
+  xftp_server_id
+);
+CREATE INDEX idx_snd_file_chunk_replica_recipients_snd_file_chunk_replica_id ON snd_file_chunk_replica_recipients(
+  snd_file_chunk_replica_id
+);
+CREATE INDEX idx_deleted_snd_chunk_replicas_user_id ON deleted_snd_chunk_replicas(
+  user_id
+);
+CREATE INDEX idx_deleted_snd_chunk_replicas_xftp_server_id ON deleted_snd_chunk_replicas(
+  xftp_server_id
+);
+CREATE INDEX idx_rcv_file_chunk_replicas_pending ON rcv_file_chunk_replicas(
+  received,
+  replica_number
+);
+CREATE INDEX idx_snd_file_chunk_replicas_pending ON snd_file_chunk_replicas(
+  replica_status,
+  replica_number
+);
+CREATE INDEX idx_deleted_snd_chunk_replicas_pending ON deleted_snd_chunk_replicas(
+  created_at
+);
+CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON encrypted_rcv_message_hashes(
+  conn_id,
+  hash
+);
+CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(
+  conn_id,
+  hash
+);
+CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages(
+  conn_id,
+  rcpt_internal_id
+);
+CREATE INDEX idx_processed_ratchet_key_hashes_created_at ON processed_ratchet_key_hashes(
+  created_at
+);
+CREATE INDEX idx_encrypted_rcv_message_hashes_created_at ON encrypted_rcv_message_hashes(
+  created_at
+);
+CREATE INDEX idx_messages_internal_ts ON messages(internal_ts);
+CREATE INDEX idx_commands_server_commands ON commands(
+  host,
+  port,
+  created_at,
+  command_id
+);
+CREATE INDEX idx_rcv_files_status_created_at ON rcv_files(status, created_at);
+CREATE INDEX idx_snd_files_status_created_at ON snd_files(status, created_at);
+CREATE INDEX idx_snd_files_snd_file_entity_id ON snd_files(snd_file_entity_id);
+CREATE INDEX idx_messages_snd_expired ON messages(
+  conn_id,
+  internal_snd_id,
+  internal_ts
+);
+CREATE INDEX idx_snd_message_deliveries_expired ON snd_message_deliveries(
+  conn_id,
+  snd_queue_id,
+  failed,
+  internal_id
+);
+CREATE INDEX idx_rcv_files_redirect_id on rcv_files(redirect_id);
+|]

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

@@ -0,0 +1,101 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Agent.Store.Postgres.Util
+  ( createDBAndUserIfNotExists,
+    -- for tests
+    dropSchema,
+    dropAllSchemasExceptSystem,
+    dropDatabaseAndUser,
+  )
+where
+
+import Control.Exception (bracket)
+import Control.Monad (forM_, unless, void, when)
+import Data.String (fromString)
+import Database.PostgreSQL.Simple (ConnectInfo (..), Only (..), defaultConnectInfo)
+import qualified Database.PostgreSQL.Simple as PSQL
+import Database.PostgreSQL.Simple.SqlQQ (sql)
+
+createDBAndUserIfNotExists :: ConnectInfo -> IO ()
+createDBAndUserIfNotExists ConnectInfo {connectUser = user, connectDatabase = dbName} = do
+  -- connect to the default "postgres" maintenance database
+  bracket (PSQL.connect defaultConnectInfo {connectUser = "postgres", connectDatabase = "postgres"}) PSQL.close $
+    \postgresDB -> do
+      void $ PSQL.execute_ postgresDB "SET client_min_messages TO WARNING"
+      -- check if the user exists, create if not
+      [Only userExists] <-
+        PSQL.query
+          postgresDB
+          [sql|
+            SELECT EXISTS (
+              SELECT 1 FROM pg_catalog.pg_roles
+              WHERE rolname = ?
+            )
+          |]
+          (Only user)
+      unless userExists $ void $ PSQL.execute_ postgresDB (fromString $ "CREATE USER " <> user)
+      -- check if the database exists, create if not
+      dbExists <- checkDBExists postgresDB dbName
+      unless dbExists $ void $ PSQL.execute_ postgresDB (fromString $ "CREATE DATABASE " <> dbName <> " OWNER " <> user)
+
+checkDBExists :: PSQL.Connection -> String -> IO Bool
+checkDBExists postgresDB dbName = do
+  [Only dbExists] <-
+    PSQL.query
+      postgresDB
+      [sql|
+        SELECT EXISTS (
+          SELECT 1 FROM pg_catalog.pg_database
+          WHERE datname = ?
+        )
+      |]
+      (Only dbName)
+  pure dbExists
+
+dropSchema :: ConnectInfo -> String -> IO ()
+dropSchema connectInfo schema =
+  bracket (PSQL.connect connectInfo) PSQL.close $
+    \db -> do
+      void $ PSQL.execute_ db "SET client_min_messages TO WARNING"
+      void $ PSQL.execute_ db (fromString $ "DROP SCHEMA IF EXISTS " <> schema <> " CASCADE")
+
+dropAllSchemasExceptSystem :: ConnectInfo -> IO ()
+dropAllSchemasExceptSystem connectInfo =
+  bracket (PSQL.connect connectInfo) PSQL.close $
+    \db -> do
+      void $ PSQL.execute_ db "SET client_min_messages TO WARNING"
+      schemaNames :: [Only String] <-
+        PSQL.query_
+          db
+          [sql|
+            SELECT schema_name
+            FROM information_schema.schemata
+            WHERE schema_name NOT IN ('public', 'pg_catalog', 'information_schema')
+          |]
+      forM_ schemaNames $ \(Only schema) ->
+        PSQL.execute_ db (fromString $ "DROP SCHEMA " <> schema <> " CASCADE")
+
+dropDatabaseAndUser :: ConnectInfo -> IO ()
+dropDatabaseAndUser ConnectInfo {connectUser = user, connectDatabase = dbName} =
+  bracket (PSQL.connect defaultConnectInfo {connectUser = "postgres", connectDatabase = "postgres"}) PSQL.close $
+    \postgresDB -> do
+      void $ PSQL.execute_ postgresDB "SET client_min_messages TO WARNING"
+      dbExists <- checkDBExists postgresDB dbName
+      when dbExists $ do
+        void $ PSQL.execute_ postgresDB (fromString $ "ALTER DATABASE " <> dbName <> " WITH ALLOW_CONNECTIONS false")
+        -- terminate all connections to the database
+        _r :: [Only Bool] <-
+          PSQL.query
+            postgresDB
+            [sql|
+              SELECT pg_terminate_backend(pg_stat_activity.pid)
+              FROM pg_stat_activity
+              WHERE datname = ?
+                AND pid <> pg_backend_pid()
+            |]
+            (Only dbName)
+        void $ PSQL.execute_ postgresDB (fromString $ "DROP DATABASE " <> dbName)
+      void $ PSQL.execute_ postgresDB (fromString $ "DROP USER IF EXISTS " <> user)

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

@@ -1,40 +1,51 @@
 {-# LANGUAGE DeriveAnyClass #-}
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE StrictData #-}
 {-# LANGUAGE TemplateHaskell #-}
 
 module Simplex.Messaging.Agent.Store.SQLite.DB
-  ( Connection (..),
+  ( BoolInt (..),
+    Binary (..),
+    Connection (..),
     SlowQueryStats (..),
     open,
     close,
     execute,
     execute_,
-    executeNamed,
     executeMany,
     query,
     query_,
-    queryNamed,
   )
 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.ByteString (ByteString)
 import Data.Int (Int64)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Text (Text)
 import Data.Time (diffUTCTime, getCurrentTime)
-import Database.SQLite.Simple (FromRow, NamedParam, Query, ToRow)
+import Database.SQLite.Simple (FromRow, Query, ToRow)
 import qualified Database.SQLite.Simple as SQL
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
 import Simplex.Messaging.Parsers (defaultJSON)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util (diffToMilliseconds, tshow)
 
+newtype BoolInt = BI {unBI :: Bool}
+  deriving newtype (FromField, ToField)
+
+newtype Binary = Binary {fromBinary :: ByteString}
+  deriving newtype (FromField, ToField)
+
 data Connection = Connection
   { conn :: SQL.Connection,
     slow :: TMap Query SlowQueryStats
@@ -51,9 +62,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,10 +103,6 @@ execute_ :: Connection -> Query -> IO ()
 execute_ Connection {conn, slow} sql = timeIt slow sql $ SQL.execute_ conn sql
 {-# INLINE execute_ #-}
 
-executeNamed :: Connection -> Query -> [NamedParam] -> IO ()
-executeNamed Connection {conn, slow} sql = timeIt slow sql . SQL.executeNamed conn sql
-{-# INLINE executeNamed #-}
-
 executeMany :: ToRow q => Connection -> Query -> [q] -> IO ()
 executeMany Connection {conn, slow} sql = timeIt slow sql . SQL.executeMany conn sql
 {-# INLINE executeMany #-}
@@ -107,8 +115,4 @@ query_ :: FromRow r => Connection -> Query -> IO [r]
 query_ Connection {conn, slow} sql = timeIt slow sql $ SQL.query_ conn sql
 {-# INLINE query_ #-}
 
-queryNamed :: FromRow r => Connection -> Query -> [NamedParam] -> IO [r]
-queryNamed Connection {conn, slow} sql = timeIt slow sql . SQL.queryNamed conn sql
-{-# INLINE queryNamed #-}
-
 $(J.deriveJSON defaultJSON ''SlowQueryStats)

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,11 @@ 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.SQLite.Migrations.M20241224_ratchet_e2e_snd_params
+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),
@@ -120,7 +107,8 @@ schemaMigrations =
     ("m20240624_snd_secure", m20240624_snd_secure, Just down_m20240624_snd_secure),
     ("m20240702_servers_stats", m20240702_servers_stats, Just down_m20240702_servers_stats),
     ("m20240930_ntf_tokens_to_delete", m20240930_ntf_tokens_to_delete, Just down_m20240930_ntf_tokens_to_delete),
-    ("m20241007_rcv_queues_last_broker_ts", m20241007_rcv_queues_last_broker_ts, Just down_m20241007_rcv_queues_last_broker_ts)
+    ("m20241007_rcv_queues_last_broker_ts", m20241007_rcv_queues_last_broker_ts, Just down_m20241007_rcv_queues_last_broker_ts),
+    ("m20241224_ratchet_e2e_snd_params", m20241224_ratchet_e2e_snd_params, Just down_m20241224_ratchet_e2e_snd_params)
   ]
 
 -- | The list of migrations in ascending order by date
@@ -129,18 +117,17 @@ 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 st = \case
+run :: DBStore -> Bool -> MigrationsToRun -> IO ()
+run st vacuum = \case
   MTRUp [] -> pure ()
-  MTRUp ms -> mapM_ runUp ms >> withConnection' st (`execSQL` "VACUUM;")
+  MTRUp ms -> do
+    mapM_ runUp ms
+    when vacuum $ withConnection' st (`execSQL` "VACUUM;")
   MTRDown ms -> mapM_ runDown $ reverse ms
   MTRNone -> pure ()
   where
@@ -148,27 +135,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 +165,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/SQLite/Migrations/M20241224_ratchet_e2e_snd_params.hs

@@ -0,0 +1,18 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241224_ratchet_e2e_snd_params where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20241224_ratchet_e2e_snd_params :: Query
+m20241224_ratchet_e2e_snd_params =
+    [sql|
+ALTER TABLE ratchets ADD COLUMN pq_pub_kem BLOB;
+|]
+
+down_m20241224_ratchet_e2e_snd_params :: Query
+down_m20241224_ratchet_e2e_snd_params =
+    [sql|
+ALTER TABLE ratchets DROP COLUMN pq_pub_kem;
+|]

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -166,7 +166,8 @@ CREATE TABLE ratchets(
   ,
   x3dh_pub_key_1 BLOB,
   x3dh_pub_key_2 BLOB,
-  pq_priv_kem BLOB
+  pq_priv_kem BLOB,
+  pq_pub_kem BLOB
 ) WITHOUT ROWID;
 CREATE TABLE skipped_messages(
   skipped_message_id INTEGER PRIMARY KEY,

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)

src/Simplex/Messaging/Client.hs

@@ -145,7 +145,6 @@ import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (SocksAuth (..), SocksProxyWithAuth (..), TransportClientConfig (..), TransportHost (..), defaultSMPPort, defaultTcpConnectTimeout, runTransportClient)
 import Simplex.Messaging.Transport.KeepAlive
-import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (bshow, diffToMicroseconds, ifM, liftEitherWith, raceAny_, threadDelay', tshow, whenM)
 import Simplex.Messaging.Version
 import System.Mem.Weak (Weak, deRefWeak)
@@ -544,7 +543,6 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
       "" -> case protocolTypeI @(ProtoType msg) of
         SPSMP | smpWebPort -> ("443", transport @TLS)
         _ -> defaultTransport cfg
-      "80" -> ("80", transport @WS)
       p -> (p, transport @TLS)
 
     client :: forall c. Transport c => TProxy c -> PClient v err msg -> TMVar (Either (ProtocolClientError err) (ProtocolClient v err msg)) -> c -> IO ()

src/Simplex/Messaging/Crypto.hs

@@ -1,11 +1,14 @@
 {-# LANGUAGE AllowAmbiguousTypes #-}
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
+{-# LANGUAGE DerivingStrategies #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE GADTs #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -233,14 +236,20 @@ import Data.Typeable (Proxy (Proxy), Typeable)
 import Data.Word (Word32)
 import Data.X509
 import Data.X509.Validation (Fingerprint (..), getFingerprint)
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
 import GHC.TypeLits (ErrorMessage (..), KnownNat, Nat, TypeError, natVal, type (+))
 import Network.Transport.Internal (decodeWord16, encodeWord16)
+import Simplex.Messaging.Agent.Store.DB (Binary (..))
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (blobFieldDecoder, parseAll, parseString)
 import Simplex.Messaging.Util ((<$?>))
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple.FromField (FromField (..))
+import Database.PostgreSQL.Simple.ToField (ToField (..))
+#else
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
+#endif
 
 -- | Cryptographic algorithms.
 data Algorithm = Ed25519 | Ed448 | X25519 | X448
@@ -721,23 +730,23 @@ generateKeyPair_ = case sAlgorithm @a of
       let k = X448.toPublic pk
        in pure (PublicKeyX448 k, PrivateKeyX448 pk k)
 
-instance ToField APrivateSignKey where toField = toField . encodePrivKey
+instance ToField APrivateSignKey where toField = toField . Binary . encodePrivKey
 
-instance ToField APublicVerifyKey where toField = toField . encodePubKey
+instance ToField APublicVerifyKey where toField = toField . Binary . encodePubKey
 
-instance ToField APrivateAuthKey where toField = toField . encodePrivKey
+instance ToField APrivateAuthKey where toField = toField . Binary . encodePrivKey
 
-instance ToField APublicAuthKey where toField = toField . encodePubKey
+instance ToField APublicAuthKey where toField = toField . Binary . encodePubKey
 
-instance ToField APrivateDhKey where toField = toField . encodePrivKey
+instance ToField APrivateDhKey where toField = toField . Binary . encodePrivKey
 
-instance ToField APublicDhKey where toField = toField . encodePubKey
+instance ToField APublicDhKey where toField = toField . Binary . encodePubKey
 
-instance AlgorithmI a => ToField (PrivateKey a) where toField = toField . encodePrivKey
+instance AlgorithmI a => ToField (PrivateKey a) where toField = toField . Binary . encodePrivKey
 
-instance AlgorithmI a => ToField (PublicKey a) where toField = toField . encodePubKey
+instance AlgorithmI a => ToField (PublicKey a) where toField = toField . Binary . encodePubKey
 
-instance ToField (DhSecret a) where toField = toField . dhBytes'
+instance ToField (DhSecret a) where toField = toField . Binary . dhBytes'
 
 instance FromField APrivateSignKey where fromField = blobFieldDecoder decodePrivKey
 
@@ -888,10 +897,9 @@ validSignatureSize n =
 -- | AES key newtype.
 newtype Key = Key {unKey :: ByteString}
   deriving (Eq, Ord, Show)
+  deriving newtype (FromField)
 
-instance ToField Key where toField = toField . unKey
-
-instance FromField Key where fromField f = Key <$> fromField f
+instance ToField Key where toField (Key s) = toField $ Binary s
 
 instance ToJSON Key where
   toJSON = strToJSON . unKey
@@ -952,7 +960,7 @@ instance FromJSON KeyHash where
 instance IsString KeyHash where
   fromString = parseString $ parseAll strP
 
-instance ToField KeyHash where toField = toField . strEncode
+instance ToField KeyHash where toField = toField . Binary . strEncode
 
 instance FromField KeyHash where fromField = blobFieldDecoder $ parseAll strP
 
@@ -1162,10 +1170,14 @@ instance SignatureAlgorithmX509 pk => SignatureAlgorithmX509 (a, pk) where
 newtype SignedObject a = SignedObject {getSignedExact :: SignedExact a}
 
 instance (Typeable a, Eq a, Show a, ASN1Object a) => FromField (SignedObject a) where
+#if defined(dbPostgres)
+  fromField f dat = SignedObject <$> blobFieldDecoder decodeSignedObject f dat
+#else
   fromField = fmap SignedObject . blobFieldDecoder decodeSignedObject
+#endif
 
 instance (Eq a, Show a, ASN1Object a) => ToField (SignedObject a) where
-  toField (SignedObject s) = toField $ encodeSignedObject s
+  toField (SignedObject s) = toField . Binary $ encodeSignedObject s
 
 instance (Eq a, Show a, ASN1Object a) => Encoding (SignedObject a) where
   smpEncode (SignedObject exact) = smpEncode . Large $ encodeSignedObject exact
@@ -1265,6 +1277,9 @@ cbVerify k pk nonce (CbAuthenticator s) authorized = cbDecryptNoPad (dh' k pk) n
 
 newtype CbNonce = CryptoBoxNonce {unCbNonce :: ByteString}
   deriving (Eq, Show)
+  deriving newtype (FromField)
+
+instance ToField CbNonce where toField (CryptoBoxNonce s) = toField $ Binary s
 
 pattern CbNonce :: ByteString -> CbNonce
 pattern CbNonce s <- CryptoBoxNonce s
@@ -1282,10 +1297,6 @@ instance ToJSON CbNonce where
 instance FromJSON CbNonce where
   parseJSON = strParseJSON "CbNonce"
 
-instance FromField CbNonce where fromField f = CryptoBoxNonce <$> fromField f
-
-instance ToField CbNonce where toField (CryptoBoxNonce s) = toField s
-
 cbNonce :: ByteString -> CbNonce
 cbNonce s
   | len == 24 = CryptoBoxNonce s
@@ -1309,6 +1320,9 @@ instance Encoding CbNonce where
 
 newtype SbKey = SecretBoxKey {unSbKey :: ByteString}
   deriving (Eq, Show)
+  deriving newtype (FromField)
+
+instance ToField SbKey where toField (SecretBoxKey s) = toField $ Binary s
 
 pattern SbKey :: ByteString -> SbKey
 pattern SbKey s <- SecretBoxKey s
@@ -1326,10 +1340,6 @@ instance ToJSON SbKey where
 instance FromJSON SbKey where
   parseJSON = strParseJSON "SbKey"
 
-instance FromField SbKey where fromField f = SecretBoxKey <$> fromField f
-
-instance ToField SbKey where toField (SecretBoxKey s) = toField s
-
 sbKey :: ByteString -> Either String SbKey
 sbKey s
   | B.length s == 32 = Right $ SecretBoxKey s

src/Simplex/Messaging/Crypto/Ratchet.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DuplicateRecordFields #-}
@@ -109,9 +110,8 @@ import Data.Maybe (fromMaybe, isJust)
 import Data.Type.Equality
 import Data.Typeable (Typeable)
 import Data.Word (Word16, Word32)
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
 import Simplex.Messaging.Agent.QueryString
+import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..))
 import Simplex.Messaging.Crypto
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings
 import Simplex.Messaging.Encoding
@@ -121,6 +121,13 @@ import Simplex.Messaging.Util (($>>=), (<$?>))
 import Simplex.Messaging.Version
 import Simplex.Messaging.Version.Internal
 import UnliftIO.STM
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple.FromField (FromField (..))
+import Database.PostgreSQL.Simple.ToField (ToField (..))
+#else
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
+#endif
 
 -- e2e encryption headers version history:
 -- 1 - binary protocol encoding (1/1/2022)
@@ -206,6 +213,10 @@ instance Encoding ARKEMParams where
       'A' -> ARKP SRKSAccepted .: RKParamsAccepted <$> smpP <*> smpP
       _ -> fail "bad ratchet KEM params"
 
+instance ToField ARKEMParams where toField = toField . Binary . smpEncode
+
+instance FromField ARKEMParams where fromField = blobFieldDecoder smpDecode
+
 data E2ERatchetParams (s :: RatchetKEMState) (a :: Algorithm)
   = E2ERatchetParams VersionE2E (PublicKey a) (PublicKey a) (Maybe (RKEMParams s))
   deriving (Show)
@@ -359,7 +370,7 @@ instance Encoding APrivRKEMParams where
       'A' -> APRKP SRKSAccepted .:. PrivateRKParamsAccepted <$> smpP <*> smpP <*> smpP
       _ -> fail "bad APrivRKEMParams"
 
-instance RatchetKEMStateI s => ToField (PrivRKEMParams s) where toField = toField . smpEncode
+instance RatchetKEMStateI s => ToField (PrivRKEMParams s) where toField = toField . Binary . smpEncode
 
 instance (Typeable s, RatchetKEMStateI s) => FromField (PrivRKEMParams s) where fromField = blobFieldDecoder smpDecode
 
@@ -576,7 +587,7 @@ instance ToJSON RatchetKey where
 instance FromJSON RatchetKey where
   parseJSON = fmap RatchetKey . strParseJSON "Key"
 
-instance ToField MessageKey where toField = toField . smpEncode
+instance ToField MessageKey where toField = toField . Binary . smpEncode
 
 instance FromField MessageKey where fromField = blobFieldDecoder smpDecode
 
@@ -1120,14 +1131,24 @@ instance AlgorithmI a => ToJSON (Ratchet a) where
 instance AlgorithmI a => FromJSON (Ratchet a) where
   parseJSON = $(JQ.mkParseJSON defaultJSON ''Ratchet)
 
-instance AlgorithmI a => ToField (Ratchet a) where toField = toField . LB.toStrict . J.encode
+instance AlgorithmI a => ToField (Ratchet a) where toField = toField . Binary . LB.toStrict . J.encode
 
 instance (AlgorithmI a, Typeable a) => FromField (Ratchet a) where fromField = blobFieldDecoder J.eitherDecodeStrict'
 
-instance ToField PQEncryption where toField (PQEncryption pqEnc) = toField pqEnc
+instance ToField PQEncryption where toField (PQEncryption pqEnc) = toField (BI pqEnc)
 
-instance FromField PQEncryption where fromField f = PQEncryption <$> fromField f
+instance FromField PQEncryption where 
+#if defined(dbPostgres)
+  fromField f dat = PQEncryption . unBI <$> fromField f dat
+#else
+  fromField f = PQEncryption . unBI <$> fromField f
+#endif
 
-instance ToField PQSupport where toField (PQSupport pqEnc) = toField pqEnc
+instance ToField PQSupport where toField (PQSupport pqEnc) = toField (BI pqEnc)
 
-instance FromField PQSupport where fromField f = PQSupport <$> fromField f
+instance FromField PQSupport where
+#if defined(dbPostgres)
+  fromField f dat = PQSupport . unBI <$> fromField f dat
+#else
+  fromField f = PQSupport . unBI <$> fromField f
+#endif

src/Simplex/Messaging/Crypto/SNTRUP761/Bindings.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE TypeApplications #-}
 
 module Simplex.Messaging.Crypto.SNTRUP761.Bindings where
@@ -9,14 +10,19 @@ import Data.Bifunctor (bimap)
 import Data.ByteArray (ScrubbedBytes)
 import qualified Data.ByteArray as BA
 import Data.ByteString (ByteString)
-import Database.SQLite.Simple.FromField
-import Database.SQLite.Simple.ToField
 import Foreign (nullPtr)
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings.Defines
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings.FFI
 import Simplex.Messaging.Crypto.SNTRUP761.Bindings.RNG (withDRG)
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple.FromField
+import Database.PostgreSQL.Simple.ToField
+#else
+import Database.SQLite.Simple.FromField
+import Database.SQLite.Simple.ToField
+#endif
 
 newtype KEMPublicKey = KEMPublicKey ByteString
   deriving (Eq, Show)
@@ -121,7 +127,11 @@ instance ToField KEMSharedKey where
   toField (KEMSharedKey k) = toField (BA.convert k :: ByteString)
 
 instance FromField KEMSharedKey where
+#if defined(dbPostgres)
+  fromField f dat = KEMSharedKey . BA.convert @ByteString <$> fromField f dat
+#else
   fromField f = KEMSharedKey . BA.convert @ByteString <$> fromField f
+#endif
 
 instance ToJSON KEMSharedKey where
   toJSON = strToJSON

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
@@ -27,8 +28,6 @@ import Data.Text.Encoding (decodeLatin1, encodeUtf8)
 import Data.Time.Clock.System
 import Data.Type.Equality
 import Data.Word (Word16)
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
 import Simplex.Messaging.Agent.Protocol (updateSMPServerHosts)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
@@ -37,6 +36,13 @@ import Simplex.Messaging.Notifications.Transport (NTFVersion, ntfClientHandshake
 import Simplex.Messaging.Parsers (fromTextField_)
 import Simplex.Messaging.Protocol hiding (Command (..), CommandTag (..))
 import Simplex.Messaging.Util (eitherToMaybe, (<$?>))
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple.FromField (FromField (..))
+import Database.PostgreSQL.Simple.ToField (ToField (..))
+#else
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
+#endif
 
 data NtfEntity = Token | Subscription
   deriving (Show)

src/Simplex/Messaging/Notifications/Types.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE LambdaCase #-}
@@ -9,14 +10,20 @@ module Simplex.Messaging.Notifications.Types where
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
 import Data.Time (UTCTime)
-import Database.SQLite.Simple.FromField (FromField (..))
-import Database.SQLite.Simple.ToField (ToField (..))
 import Simplex.Messaging.Agent.Protocol (ConnId, NotificationsMode (..), UserId)
+import Simplex.Messaging.Agent.Store.DB (Binary (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Parsers (blobFieldDecoder, fromTextField_)
 import Simplex.Messaging.Protocol (NotifierId, NtfServer, SMPServer)
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple.FromField (FromField (..))
+import Database.PostgreSQL.Simple.ToField (ToField (..))
+#else
+import Database.SQLite.Simple.FromField (FromField (..))
+import Database.SQLite.Simple.ToField (ToField (..))
+#endif
 
 data NtfTknAction
   = NTARegister
@@ -41,7 +48,7 @@ instance Encoding NtfTknAction where
 
 instance FromField NtfTknAction where fromField = blobFieldDecoder smpDecode
 
-instance ToField NtfTknAction where toField = toField . smpEncode
+instance ToField NtfTknAction where toField = toField . Binary . smpEncode
 
 data NtfToken = NtfToken
   { deviceToken :: DeviceToken,
@@ -119,7 +126,7 @@ instance Encoding NtfSubNTFAction where
 
 instance FromField NtfSubNTFAction where fromField = blobFieldDecoder smpDecode
 
-instance ToField NtfSubNTFAction where toField = toField . smpEncode
+instance ToField NtfSubNTFAction where toField = toField . Binary . smpEncode
 
 data NtfSubSMPAction
   = NSASmpKey
@@ -138,7 +145,7 @@ instance Encoding NtfSubSMPAction where
 
 instance FromField NtfSubSMPAction where fromField = blobFieldDecoder smpDecode
 
-instance ToField NtfSubSMPAction where toField = toField . smpEncode
+instance ToField NtfSubSMPAction where toField = toField . Binary . smpEncode
 
 data NtfAgentSubStatus
   = -- | subscription started

src/Simplex/Messaging/Parsers.hs

@@ -1,5 +1,6 @@
 {-# LANGUAGE CPP #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 
@@ -20,12 +21,19 @@ import qualified Data.Text as T
 import Data.Time.Clock (UTCTime)
 import Data.Time.ISO8601 (parseISO8601)
 import Data.Typeable (Typeable)
+import Simplex.Messaging.Util (safeDecodeUtf8, (<$?>))
+import Text.Read (readMaybe)
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple (ResultError (..))
+import Database.PostgreSQL.Simple.FromField (FromField(..), FieldParser, returnError, Field (..))
+import Database.PostgreSQL.Simple.TypeInfo.Static (textOid, varcharOid)
+import qualified Data.Text.Encoding as TE
+#else
 import Database.SQLite.Simple (ResultError (..), SQLData (..))
 import Database.SQLite.Simple.FromField (FieldParser, returnError)
 import Database.SQLite.Simple.Internal (Field (..))
 import Database.SQLite.Simple.Ok (Ok (Ok))
-import Simplex.Messaging.Util (safeDecodeUtf8, (<$?>))
-import Text.Read (readMaybe)
+#endif
 
 base64P :: Parser ByteString
 base64P = decode <$?> paddedBase64 rawBase64P
@@ -70,6 +78,7 @@ parseRead2 = parseRead $ do
 
 wordEnd :: Char -> Bool
 wordEnd c = c == ' ' || c == '\n'
+{-# INLINE wordEnd #-}
 
 parseString :: (ByteString -> Either String a) -> (String -> a)
 parseString p = either error id . p . B.pack
@@ -77,6 +86,14 @@ parseString p = either error id . p . B.pack
 blobFieldParser :: Typeable k => Parser k -> FieldParser k
 blobFieldParser = blobFieldDecoder . parseAll
 
+#if defined(dbPostgres)
+blobFieldDecoder :: Typeable k => (ByteString -> Either String k) -> FieldParser k
+blobFieldDecoder dec f val = do
+  x <- fromField f val
+  case dec x of
+    Right k -> pure k
+    Left e -> returnError ConversionFailed f ("couldn't parse field: " ++ e)
+#else
 blobFieldDecoder :: Typeable k => (ByteString -> Either String k) -> FieldParser k
 blobFieldDecoder dec = \case
   f@(Field (SQLBlob b) _) ->
@@ -84,7 +101,20 @@ blobFieldDecoder dec = \case
       Right k -> Ok k
       Left e -> returnError ConversionFailed f ("couldn't parse field: " ++ e)
   f -> returnError ConversionFailed f "expecting SQLBlob column type"
+#endif
 
+-- TODO [postgres] review
+#if defined(dbPostgres)
+fromTextField_ :: Typeable a => (Text -> Maybe a) -> FieldParser a
+fromTextField_ fromText f val =
+  if typeOid f `elem` [textOid, varcharOid]
+    then case val of
+      Just t -> case fromText (TE.decodeUtf8 t) of
+        Just x -> pure x
+        _ -> returnError ConversionFailed f "invalid text value"
+      Nothing -> returnError UnexpectedNull f "NULL value found for non-NULL field"
+    else returnError Incompatible f "expecting TEXT or VARCHAR column type"
+#else
 fromTextField_ :: Typeable a => (Text -> Maybe a) -> Field -> Ok a
 fromTextField_ fromText = \case
   f@(Field (SQLText t) _) ->
@@ -92,6 +122,7 @@ fromTextField_ fromText = \case
       Just x -> Ok x
       _ -> returnError ConversionFailed f ("invalid text: " <> T.unpack t)
   f -> returnError ConversionFailed f "expecting SQLText column type"
+#endif
 
 fstToLower :: String -> String
 fstToLower "" = ""

src/Simplex/Messaging/Protocol.hs

@@ -70,6 +70,8 @@ module Simplex.Messaging.Protocol
     CommandError (..),
     ProxyError (..),
     BrokerErrorType (..),
+    BlockingInfo (..),
+    BlockingReason (..),
     Transmission,
     TransmissionAuth (..),
     SignedTransmission,
@@ -1194,6 +1196,8 @@ data ErrorType
     PROXY {proxyErr :: ProxyError}
   | -- | command authorization error - bad signature or non-existing SMP queue
     AUTH
+  | -- | command with the entity that was blocked
+    BLOCKED {blockInfo :: BlockingInfo}
   | -- | encryption/decryption error in proxy protocol
     CRYPTO
   | -- | SMP queue capacity is exceeded on the server
@@ -1210,17 +1214,41 @@ data ErrorType
     INTERNAL
   | -- | used internally, never returned by the server (to be removed)
     DUPLICATE_ -- not part of SMP protocol, used internally
-  deriving (Eq, Read, Show)
+  deriving (Eq, Show)
 
 instance StrEncoding ErrorType where
   strEncode = \case
+    BLOCK -> "BLOCK"
+    SESSION -> "SESSION"
     CMD e -> "CMD " <> bshow e
     PROXY e -> "PROXY " <> strEncode e
-    e -> bshow e
+    AUTH -> "AUTH"
+    BLOCKED info -> "BLOCKED " <> strEncode info
+    CRYPTO -> "CRYPTO"
+    QUOTA -> "QUOTA"
+    STORE e -> "STORE " <> encodeUtf8 (T.pack e)
+    NO_MSG -> "NO_MSG"
+    LARGE_MSG -> "LARGE_MSG"
+    EXPIRED -> "EXPIRED"
+    INTERNAL -> "INTERNAL"
+    DUPLICATE_ -> "DUPLICATE_"
   strP =
-    "CMD " *> (CMD <$> parseRead1)
-      <|> "PROXY " *> (PROXY <$> strP)
-      <|> parseRead1
+    A.choice
+      [ "BLOCK" $> BLOCK,
+        "SESSION" $> SESSION,
+        "CMD " *> (CMD <$> parseRead1),
+        "PROXY " *> (PROXY <$> strP),
+        "AUTH" $> AUTH,
+        "BLOCKED " *> strP,
+        "CRYPTO" $> CRYPTO,
+        "QUOTA" $> QUOTA,
+        "STORE " *> (STORE . T.unpack . safeDecodeUtf8 <$> A.takeByteString),
+        "NO_MSG" $> NO_MSG,
+        "LARGE_MSG" $> LARGE_MSG,
+        "EXPIRED" $> EXPIRED,
+        "INTERNAL" $> INTERNAL,
+        "DUPLICATE_" $> DUPLICATE_
+      ]
 
 -- | SMP command error type.
 data CommandError
@@ -1248,7 +1276,7 @@ data ProxyError
     BASIC_AUTH
   | -- no destination server error
     NO_SESSION
-  deriving (Eq, Read, Show)
+  deriving (Eq, Show)
 
 -- | SMP server errors.
 data BrokerErrorType
@@ -1266,6 +1294,37 @@ data BrokerErrorType
     TIMEOUT
   deriving (Eq, Read, Show, Exception)
 
+data BlockingInfo = BlockingInfo
+  { reason :: BlockingReason
+  }
+  deriving (Eq, Show)
+
+data BlockingReason = BRSpam | BRContent
+  deriving (Eq, Show)
+
+instance StrEncoding BlockingInfo where
+  strEncode BlockingInfo {reason} = "reason=" <> strEncode reason
+  strP = do
+    reason <- "reason=" *> strP
+    pure BlockingInfo {reason}
+
+instance Encoding BlockingInfo where
+  smpEncode = strEncode
+  smpP = strP
+
+instance StrEncoding BlockingReason where
+  strEncode = \case
+    BRSpam -> "spam"
+    BRContent -> "content"
+  strP = "spam" $> BRSpam <|> "content" $> BRContent
+
+instance ToJSON BlockingReason where
+  toJSON = strToJSON
+  toEncoding = strToJEncoding
+
+instance FromJSON BlockingReason where
+  parseJSON = strParseJSON "BlockingReason"
+
 -- | SMP transmission parser.
 transmissionP :: THandleParams v p -> Parser RawTransmission
 transmissionP THandleParams {sessionId, implySessId} = do
@@ -1435,7 +1494,9 @@ instance ProtocolEncoding SMPVersion ErrorType BrokerMsg where
       | otherwise -> e END_
     INFO info -> e (INFO_, ' ', info)
     OK -> e OK_
-    ERR err -> e (ERR_, ' ', err)
+    ERR err -> case err of
+      BLOCKED _ | v < blockedEntityErrorSMPVersion -> e (ERR_, ' ', AUTH)
+      _ -> e (ERR_, ' ', err)
     PONG -> e PONG_
     where
       e :: Encoding a => a -> ByteString
@@ -1513,6 +1574,7 @@ instance Encoding ErrorType where
     CMD err -> "CMD " <> smpEncode err
     PROXY err -> "PROXY " <> smpEncode err
     AUTH -> "AUTH"
+    BLOCKED info -> "BLOCKED " <> smpEncode info
     CRYPTO -> "CRYPTO"
     QUOTA -> "QUOTA"
     STORE err -> "STORE " <> smpEncode err
@@ -1529,6 +1591,7 @@ instance Encoding ErrorType where
       "CMD" -> CMD <$> _smpP
       "PROXY" -> PROXY <$> _smpP
       "AUTH" -> pure AUTH
+      "BLOCKED" -> BLOCKED <$> _smpP
       "CRYPTO" -> pure CRYPTO
       "QUOTA" -> pure QUOTA
       "STORE" -> STORE <$> _smpP
@@ -1759,9 +1822,11 @@ tDecodeParseValidate THandleParams {sessionId, thVersion = v, implySessId} = \ca
 
 $(J.deriveJSON defaultJSON ''MsgFlags)
 
-$(J.deriveJSON (sumTypeJSON id) ''CommandError)
+$(J.deriveJSON (taggedObjectJSON id) ''CommandError)
 
-$(J.deriveJSON (sumTypeJSON id) ''BrokerErrorType)
+$(J.deriveJSON (taggedObjectJSON id) ''BrokerErrorType)
+
+$(J.deriveJSON defaultJSON ''BlockingInfo)
 
 -- run deriveJSON in one TH splice to allow mutual instance
 $(concat <$> mapM @[] (J.deriveJSON (sumTypeJSON id)) [''ProxyError, ''ErrorType])

src/Simplex/Messaging/Server.hs

@@ -71,6 +71,7 @@ import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing)
 import Data.Semigroup (Sum (..))
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
+import qualified Data.Text.IO as T
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
@@ -98,6 +99,7 @@ import Simplex.Messaging.Server.MsgStore.Journal (JournalQueue, closeMsgQueue)
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.NtfStore
+import Simplex.Messaging.Server.Prometheus
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Server.QueueStore.STM
@@ -176,7 +178,11 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
         : receiveFromProxyAgent pa
         : expireNtfsThread cfg
         : sigIntHandlerThread
-        : map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
+        : map runServer transports
+            <> expireMessagesThread_ cfg
+            <> serverStatsThread_ cfg
+            <> prometheusMetricsThread_ cfg
+            <> controlPortThread_ cfg
     )
     `finally` stopServer s
   where
@@ -555,6 +561,50 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
         showProxyStats ProxyStatsData {_pRequests, _pSuccesses, _pErrorsConnect, _pErrorsCompat, _pErrorsOther} =
           [show _pRequests, show _pSuccesses, show _pErrorsConnect, show _pErrorsCompat, show _pErrorsOther]
 
+    prometheusMetricsThread_ :: ServerConfig -> [M ()]
+    prometheusMetricsThread_ ServerConfig {prometheusInterval = Just interval, prometheusMetricsFile} =
+      [savePrometheusMetrics interval prometheusMetricsFile]
+    prometheusMetricsThread_ _ = []
+
+    savePrometheusMetrics :: Int -> FilePath -> M ()
+    savePrometheusMetrics saveInterval metricsFile = do
+      labelMyThread "savePrometheusMetrics"
+      liftIO $ putStrLn $ "Prometheus metrics saved every " <> show saveInterval <> " seconds to " <> metricsFile
+      AMS _ st <- asks msgStore
+      ss <- asks serverStats
+      env <- ask
+      let interval = 1000000 * saveInterval
+      liftIO $ forever $ do
+        threadDelay interval
+        ts <- getCurrentTime
+        sm <- getServerMetrics st ss
+        rtm <- getRealTimeMetrics env
+        T.writeFile metricsFile $ prometheusMetrics sm rtm ts
+
+    getServerMetrics :: STMQueueStore s => s -> ServerStats -> IO ServerMetrics
+    getServerMetrics st ss = do
+      d <- getServerStatsData ss
+      let ps = periodStatDataCounts $ _activeQueues d
+          psNtf = periodStatDataCounts $ _activeQueuesNtf d
+      queueCount <- M.size <$> readTVarIO (activeMsgQueues st)
+      notifierCount <- M.size <$> readTVarIO (notifiers' st)
+      pure ServerMetrics {statsData = d, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount}
+
+    getRealTimeMetrics :: Env -> IO RealTimeMetrics
+    getRealTimeMetrics Env {clients, sockets, server = Server {subscribers, notifiers, subClients, ntfSubClients}} = do
+      socketStats <- mapM (traverse getSocketStats) =<< readTVarIO sockets
+#if MIN_VERSION_base(4,18,0)
+      threadsCount <- length <$> listThreads
+#else
+      let threadsCount = 0
+#endif
+      clientsCount <- IM.size <$> readTVarIO clients
+      smpSubsCount <- M.size <$> readTVarIO subscribers
+      smpSubClientsCount <- IM.size <$> readTVarIO subClients
+      ntfSubsCount <- M.size <$> readTVarIO notifiers
+      ntfSubClientsCount <- IM.size <$> readTVarIO ntfSubClients
+      pure RealTimeMetrics {socketStats, threadsCount, clientsCount, smpSubsCount, smpSubClientsCount, ntfSubsCount, ntfSubClientsCount}
+
     runClient :: Transport c => C.APrivateSignKey -> TProxy c -> c -> M ()
     runClient signKey tp h = do
       kh <- asks serverIdentity
@@ -695,13 +745,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
 #endif
               CPSockets -> withUserRole $ unliftIO u (asks sockets) >>= readTVarIO >>= mapM_ putSockets
                 where
-                  putSockets (tcpPort, (accepted', closed', active')) = do
-                    (accepted, closed, active) <- (,,) <$> readTVarIO accepted' <*> readTVarIO closed' <*> readTVarIO active'
+                  putSockets (tcpPort, socketsState) = do
+                    ss <- getSocketStats socketsState
                     hPutStrLn h $ "Sockets for port " <> tcpPort <> ":"
-                    hPutStrLn h $ "accepted: " <> show accepted
-                    hPutStrLn h $ "closed: " <> show closed
-                    hPutStrLn h $ "active: " <> show (IM.size active)
-                    hPutStrLn h $ "leaked: " <> show (accepted - closed - IM.size active)
+                    hPutStrLn h $ "accepted: " <> show (socketsAccepted ss)
+                    hPutStrLn h $ "closed: " <> show (socketsClosed ss)
+                    hPutStrLn h $ "active: " <> show (socketsActive ss)
+                    hPutStrLn h $ "leaked: " <> show (socketsLeaked ss)
               CPSocketThreads -> withAdminRole $ do
 #if MIN_VERSION_base(4,18,0)
                 unliftIO u (asks sockets) >>= readTVarIO >>= mapM_ putSocketThreads
@@ -799,16 +849,41 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
                               SubPending -> (c1, c2 + 1, c3, c4)
                               SubThread _ -> (c1, c2, c3 + 1, c4)
                           ProhibitSub -> pure (c1, c2, c3, c4 + 1)
-              CPDelete qId -> withUserRole $ unliftIO u $ do
+              CPDelete sId -> withUserRole $ unliftIO u $ do
                 AMS _ st <- asks msgStore
                 r <- liftIO $ runExceptT $ do
-                  (q, qr) <- ExceptT (getQueueRec st SSender qId) `catchE` \_ -> ExceptT (getQueueRec st SRecipient qId)
+                  (q, qr) <- ExceptT $ getQueueRec st SSender sId
                   ExceptT $ deleteQueueSize st (recipientId qr) q
                 case r of
                   Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
                   Right (qr, numDeleted) -> do
                     updateDeletedStats qr
                     liftIO $ hPutStrLn h $ "ok, " <> show numDeleted <> " messages deleted"
+              CPStatus sId -> withUserRole $ unliftIO u $ do
+                AMS _ st <- asks msgStore
+                q <- liftIO $ getQueueRec st SSender sId
+                liftIO $ hPutStrLn h $ case q of
+                  Left e -> "error: " <> show e
+                  Right (_, QueueRec {sndSecure, status, updatedAt}) ->
+                    "status: " <> show status <> ", updatedAt: " <> show updatedAt <> ", sndSecure: " <> show sndSecure
+              CPBlock sId info -> withUserRole $ unliftIO u $ do
+                AMS _ st <- asks msgStore
+                r <- liftIO $ runExceptT $ do
+                  q <- ExceptT $ getQueue st SSender sId
+                  ExceptT $ blockQueue st q info
+                case r of
+                  Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
+                  Right () -> do
+                    incStat . qBlocked =<< asks serverStats
+                    liftIO $ hPutStrLn h "ok"
+              CPUnblock sId -> withUserRole $ unliftIO u $ do
+                AMS _ st <- asks msgStore
+                r <- liftIO $ runExceptT $ do
+                  q <- ExceptT $ getQueue st SSender sId
+                  ExceptT $ unblockQueue st q
+                liftIO $ hPutStrLn h $ case r of
+                  Left e -> "error: " <> show e
+                  Right () -> "ok"
               CPSave -> withAdminRole $ withLock' (savingLock srv) "control" $ do
                 hPutStrLn h "saving server state..."
                 unliftIO u $ saveServer False
@@ -1175,7 +1250,7 @@ client
         SKEY sKey ->
           withQueue $ \q QueueRec {sndSecure} ->
             (corrId,entId,) <$> if sndSecure then secureQueue_ q sKey else pure $ ERR AUTH
-        SEND flags msgBody -> withQueue $ sendMessage flags msgBody
+        SEND flags msgBody -> withQueue_ False $ sendMessage flags msgBody
         PING -> pure (corrId, NoEntity, PONG)
         RFWD encBlock -> (corrId, NoEntity,) <$> processForwardedCommand encBlock
       Cmd SNotifier NSUB -> Just <$> subscribeNotifications
@@ -1197,7 +1272,7 @@ client
           NKEY nKey dhKey -> withQueue $ \q _ -> addQueueNotifier_ q nKey dhKey
           NDEL -> withQueue $ \q _ -> deleteQueueNotifier_ q
           OFF -> maybe (pure $ err INTERNAL) suspendQueue_ q_
-          DEL -> maybe (pure $ err INTERNAL) delQueueAndMsgs q_          
+          DEL -> maybe (pure $ err INTERNAL) delQueueAndMsgs q_
           QUE -> withQueue $ \q qr -> (corrId,entId,) <$> getQueueInfo q qr
       where
         createQueue :: RcvPublicAuthKey -> RcvPublicDhKey -> SubscriptionMode -> SenderCanSecure -> M (Transmission BrokerMsg)
@@ -1214,7 +1289,7 @@ client
                     rcvDhSecret,
                     senderKey = Nothing,
                     notifier = Nothing,
-                    status = QueueActive,
+                    status = EntityActive,
                     sndSecure,
                     updatedAt
                   }
@@ -1356,13 +1431,18 @@ client
                   Nothing -> incStat (msgGetNoMsg stats) $> ok
 
         withQueue :: (StoreQueue s -> QueueRec -> M (Transmission BrokerMsg)) -> M (Transmission BrokerMsg)
-        withQueue action = case q_ of
+        withQueue = withQueue_ True
+
+        withQueue_ :: Bool -> (StoreQueue s -> QueueRec -> M (Transmission BrokerMsg)) -> M (Transmission BrokerMsg)
+        withQueue_ queueNotBlocked action = case q_ of
           Nothing -> pure $ err INTERNAL
-          Just (q, qr@QueueRec {updatedAt}) -> do
-            t <- liftIO getSystemDate
-            if updatedAt == Just t
-              then action q qr
-              else liftIO (updateQueueTime ms q t) >>= either (pure . err) (action q)
+          Just (q, qr@QueueRec {status, updatedAt}) -> case status of
+            EntityBlocked info | queueNotBlocked -> pure $ err $ BLOCKED info
+            _ -> do
+              t <- liftIO getSystemDate
+              if updatedAt == Just t
+                then action q qr
+                else liftIO (updateQueueTime ms q t) >>= either (pure . err) (action q)
 
         subscribeNotifications :: M (Transmission BrokerMsg)
         subscribeNotifications = do
@@ -1433,10 +1513,13 @@ client
           | otherwise = do
               stats <- asks serverStats
               case status qr of
-                QueueOff -> do
+                EntityOff -> do
                   incStat $ msgSentAuth stats
                   pure $ err AUTH
-                QueueActive ->
+                EntityBlocked info -> do
+                  incStat $ msgSentBlock stats
+                  pure $ err $ BLOCKED info
+                EntityActive ->
                   case C.maxLenBS msgBody of
                     Left _ -> pure $ err LARGE_MSG
                     Right body -> do
@@ -1684,7 +1767,7 @@ updateDeletedStats q = do
   let delSel = if isNothing (senderKey q) then qDeletedNew else qDeletedSecured
   incStat $ delSel stats
   incStat $ qDeletedAll stats
-  incStat $ qCount stats
+  liftIO $ atomicModifyIORef'_ (qCount stats) (subtract 1)
 
 incStat :: MonadIO m => IORef Int -> m ()
 incStat r = liftIO $ atomicModifyIORef'_ r (+ 1)

src/Simplex/Messaging/Server/Control.hs

@@ -5,7 +5,7 @@ module Simplex.Messaging.Server.Control where
 
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (BasicAuth, SenderId)
+import Simplex.Messaging.Protocol (BasicAuth, BlockingInfo, SenderId)
 
 data CPClientRole = CPRNone | CPRUser | CPRAdmin
   deriving (Eq)
@@ -22,6 +22,9 @@ data ControlProtocol
   | CPSocketThreads
   | CPServerInfo
   | CPDelete SenderId
+  | CPStatus SenderId
+  | CPBlock SenderId BlockingInfo
+  | CPUnblock SenderId
   | CPSave
   | CPHelp
   | CPQuit
@@ -39,14 +42,17 @@ instance StrEncoding ControlProtocol where
     CPSockets -> "sockets"
     CPSocketThreads -> "socket-threads"
     CPServerInfo -> "server-info"
-    CPDelete bs -> "delete " <> strEncode bs
+    CPDelete sId -> "delete " <> strEncode sId
+    CPStatus sId -> "status " <> strEncode sId
+    CPBlock sId info -> "block " <> strEncode sId <> " " <> strEncode info
+    CPUnblock sId -> "unblock " <> strEncode sId
     CPSave -> "save"
     CPHelp -> "help"
     CPQuit -> "quit"
     CPSkip -> ""
   strP =
     A.takeTill (== ' ') >>= \case
-      "auth" -> CPAuth <$> (A.space *> strP)
+      "auth" -> CPAuth <$> _strP
       "suspend" -> pure CPSuspend
       "resume" -> pure CPResume
       "clients" -> pure CPClients
@@ -56,7 +62,10 @@ instance StrEncoding ControlProtocol where
       "sockets" -> pure CPSockets
       "socket-threads" -> pure CPSocketThreads
       "server-info" -> pure CPServerInfo
-      "delete" -> CPDelete <$> (A.space *> strP)
+      "delete" -> CPDelete <$> _strP
+      "status" -> CPStatus <$> _strP
+      "block" -> CPBlock <$> _strP <*> _strP
+      "unblock" -> CPUnblock <$> _strP
       "save" -> pure CPSave
       "help" -> pure CPHelp
       "quit" -> pure CPQuit

src/Simplex/Messaging/Server/Env/STM.hs

@@ -96,6 +96,9 @@ data ServerConfig = ServerConfig
     serverStatsLogFile :: FilePath,
     -- | file to save and restore stats
     serverStatsBackupFile :: Maybe FilePath,
+    -- | interval and file to save prometheus metrics
+    prometheusInterval :: Maybe Int,
+    prometheusMetricsFile :: FilePath,
     -- | notification delivery interval
     ntfDeliveryInterval :: Int,
     -- | interval between sending pending END events to unsubscribed clients, seconds

src/Simplex/Messaging/Server/Main.hs

@@ -47,7 +47,7 @@ import Simplex.Messaging.Server.Information
 import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..))
 import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore)
 import Simplex.Messaging.Server.QueueStore.STM (readQueueStore)
-import Simplex.Messaging.Transport (batchCmdsSMPVersion, sendingProxySMPVersion, simplexMQVersion, supportedServerSMPRelayVRange)
+import Simplex.Messaging.Transport (batchCmdsSMPVersion, currentServerSMPRelayVersion, simplexMQVersion, supportedServerSMPRelayVRange)
 import Simplex.Messaging.Transport.Client (SocksProxy, TransportHost (..), defaultSocksProxy)
 import Simplex.Messaging.Transport.Server (ServerCredentials (..), TransportServerConfig (..), defaultTransportServerConfig)
 import Simplex.Messaging.Util (eitherToMaybe, ifM, safeDecodeUtf8, tshow)
@@ -253,7 +253,9 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 <> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
                 <> "# Log daily server statistics to CSV file\n"
                 <> ("log_stats: " <> onOff logStats <> "\n\n")
-                <> "[AUTH]\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\
@@ -431,6 +433,8 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
               logStatsStartTime = 0, -- seconds from 00:00 UTC
               serverStatsLogFile = combine logPath "smp-server-stats.daily.log",
               serverStatsBackupFile = logStats $> combine logPath "smp-server-stats.log",
+              prometheusInterval = eitherToMaybe $ read . T.unpack <$> lookupValue "STORE_LOG" "prometheus_interval" ini,
+              prometheusMetricsFile = combine logPath "smp-server-metrics.txt",
               pendingENDInterval = 15000000, -- 15 seconds
               ntfDeliveryInterval = 3000000, -- 3 seconds
               smpServerVRange = supportedServerSMPRelayVRange,
@@ -443,7 +447,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 defaultSMPClientAgentConfig
                   { smpCfg =
                       (smpCfg defaultSMPClientAgentConfig)
-                        { serverVRange = mkVersionRange batchCmdsSMPVersion sendingProxySMPVersion,
+                        { serverVRange = mkVersionRange batchCmdsSMPVersion currentServerSMPRelayVersion,
                           agreeSecret = True,
                           networkConfig =
                             defaultNetworkConfig

src/Simplex/Messaging/Server/Prometheus.hs

@@ -0,0 +1,386 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeApplications #-}
+{-# OPTIONS_GHC -fno-warn-unrecognised-pragmas #-}
+
+module Simplex.Messaging.Server.Prometheus where
+
+import Data.Int (Int64)
+import Data.Text (Text)
+import qualified Data.Text as T
+import Data.Time.Clock (UTCTime (..), diffUTCTime)
+import Data.Time.Clock.System (systemEpochDay)
+import Data.Time.Format.ISO8601 (iso8601Show)
+import Network.Socket (ServiceName)
+import Simplex.Messaging.Server.Stats
+import Simplex.Messaging.Transport.Server (SocketStats (..))
+
+data ServerMetrics = ServerMetrics
+  { statsData :: ServerStatsData,
+    activeQueueCounts :: PeriodStatCounts,
+    activeNtfCounts :: PeriodStatCounts,
+    queueCount :: Int,
+    notifierCount :: Int
+  }
+
+data RealTimeMetrics = RealTimeMetrics
+  { socketStats :: [(ServiceName, SocketStats)],
+    threadsCount :: Int,
+    clientsCount :: Int,
+    smpSubsCount :: Int,
+    smpSubClientsCount :: Int,
+    ntfSubsCount :: Int,
+    ntfSubClientsCount :: Int
+  }
+
+{-# FOURMOLU_DISABLE\n#-}
+prometheusMetrics :: ServerMetrics -> RealTimeMetrics -> UTCTime -> Text
+prometheusMetrics sm rtm ts =
+  time <> queues <> subscriptions <> messages <> ntfMessages <> ntfs <> relays <> info
+  where
+    ServerMetrics {statsData, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount} = sm
+    RealTimeMetrics
+      { socketStats,
+        threadsCount,
+        clientsCount,
+        smpSubsCount,
+        smpSubClientsCount,
+        ntfSubsCount,
+        ntfSubClientsCount
+      } = rtm
+    ServerStatsData
+      { _fromTime,
+        _qCreated,
+        _qSecured,
+        _qDeletedAll,
+        _qDeletedAllB,
+        _qDeletedNew,
+        _qDeletedSecured,
+        _qBlocked,
+        _qSub,
+        _qSubAllB,
+        _qSubAuth,
+        _qSubDuplicate,
+        _qSubProhibited,
+        _qSubEnd,
+        _qSubEndB,
+        _ntfCreated,
+        _ntfDeleted,
+        _ntfDeletedB,
+        _ntfSub,
+        _ntfSubB,
+        _ntfSubAuth,
+        _ntfSubDuplicate,
+        _msgSent,
+        _msgSentAuth,
+        _msgSentQuota,
+        _msgSentLarge,
+        _msgSentBlock,
+        _msgRecv,
+        _msgRecvGet,
+        _msgGet,
+        _msgGetNoMsg,
+        _msgGetAuth,
+        _msgGetDuplicate,
+        _msgGetProhibited,
+        _msgExpired,
+        _activeQueues,
+        _msgSentNtf,
+        _msgRecvNtf,
+        _activeQueuesNtf,
+        _msgNtfs,
+        _msgNtfsB,
+        _msgNtfNoSub,
+        _msgNtfLost,
+        _msgNtfExpired,
+        _pRelays,
+        _pRelaysOwn,
+        _pMsgFwds,
+        _pMsgFwdsOwn,
+        _pMsgFwdsRecv,
+        _qCount,
+        _msgCount,
+        _ntfCount
+      } = statsData
+    time =
+      "# Recorded at: " <> T.pack (iso8601Show ts) <> "\n\
+      \# Stats from: " <> T.pack (iso8601Show _fromTime) <> "\n\
+      \\n"
+    queues =
+      "# Queues\n\
+      \# ------\n\
+      \\n\
+      \# HELP simplex_smp_queues_created Created queues\n\
+      \# TYPE simplex_smp_queues_created counter\n\
+      \simplex_smp_queues_created " <> mshow _qCreated <> "\n# qCreated\n\
+      \\n\
+      \# HELP simplex_smp_queues_secured Secured queues\n\
+      \# TYPE simplex_smp_queues_secured counter\n\
+      \simplex_smp_queues_secured " <> mshow _qSecured <> "\n# qSecured\n\
+      \\n\
+      \# HELP simplex_smp_queues_deleted Deleted queues\n\
+      \# TYPE simplex_smp_queues_deleted counter\n\
+      \simplex_smp_queues_deleted{type=\"all\"} " <> mshow _qDeletedAll <> "\n# qDeleted\n\
+      \simplex_smp_queues_deleted{type=\"new\"} " <> mshow _qDeletedNew <> "\n# qDeletedNew\n\
+      \simplex_smp_queues_deleted{type=\"secured\"} " <> mshow _qDeletedSecured <> "\n# qDeletedSecured\n\
+      \\n\
+      \# HELP simplex_smp_queues_deleted Deleted queues\n\
+      \# TYPE simplex_smp_queues_deleted counter\n\
+      \simplex_smp_queues_blocked " <> mshow _qBlocked <> "\n# qBlocked\n\
+      \\n\
+      \# HELP simplex_smp_queues_deleted_batch Batched requests to delete queues\n\
+      \# TYPE simplex_smp_queues_deleted_batch counter\n\
+      \simplex_smp_queues_deleted_batch " <> mshow _qDeletedAllB <> "\n# qDeletedAllB\n\
+      \\n\
+      \# HELP simplex_smp_queues_total1 Total number of stored queues (first type of count).\n\
+      \# TYPE simplex_smp_queues_total1 gauge\n\
+      \simplex_smp_queues_total1 " <> mshow _qCount <> "\n# qCount\n\
+      \\n\
+      \# HELP simplex_smp_queues_total2 Total number of stored queues (second type of count).\n\
+      \# TYPE simplex_smp_queues_total2 gauge\n\
+      \simplex_smp_queues_total2 " <> mshow queueCount <> "\n# qCount2\n\
+      \\n\
+      \# HELP simplex_smp_queues_daily Daily active queues.\n\
+      \# TYPE simplex_smp_queues_daily gauge\n\
+      \simplex_smp_queues_daily " <> mstr (dayCount ps) <> "\n# dayMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_weekly Weekly active queues.\n\
+      \# TYPE simplex_smp_queues_weekly gauge\n\
+      \simplex_smp_queues_weekly " <> mstr (weekCount ps) <> "\n# weekMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_monthly Monthly active queues.\n\
+      \# TYPE simplex_smp_queues_monthly gauge\n\
+      \simplex_smp_queues_monthly " <> mstr (monthCount ps) <> "\n# monthMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_daily Daily active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_daily gauge\n\
+      \simplex_smp_queues_notify_daily " <> mstr (dayCount psNtf) <> "\n# dayCountNtf\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_weekly Weekly active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_weekly gauge\n\
+      \simplex_smp_queues_notify_weekly " <> mstr (weekCount psNtf) <> "\n# weekCountNtf\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_monthly Monthly active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_monthly gauge\n\
+      \simplex_smp_queues_notify_monthly " <> mstr (monthCount psNtf) <> "\n# monthCountNtf\n\
+      \\n"
+    subscriptions =
+      "# Subscriptions\n\
+      \# -------------\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_successes Successful subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_successes counter\n\
+      \simplex_smp_subscribtion_successes " <> mshow _qSub <> "\n# qSub\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_successes_batch Batched successful subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_successes_batch counter\n\
+      \simplex_smp_subscribtion_successes_batch " <> mshow _qSubAllB <> "\n# qSubAllB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_end Ended subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_end counter\n\
+      \simplex_smp_subscribtion_end " <> mshow _qSubEnd <> "\n# qSubEnd\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_end_batch Batched ended subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_end_batch counter\n\
+      \simplex_smp_subscribtion_end_batch " <> mshow _qSubEndB <> "\n# qSubEndB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_errors Subscription errors.\n\
+      \# TYPE simplex_smp_subscribtion_errors counter\n\
+      \simplex_smp_subscribtion_errors{type=\"auth\"} " <> mshow _qSubAuth <> "\n# qSubAuth\n\
+      \simplex_smp_subscribtion_errors{type=\"duplicate\"} " <> mshow _qSubDuplicate <> "\n# qSubDuplicate\n\
+      \simplex_smp_subscribtion_errors{type=\"prohibited\"} " <> mshow _qSubProhibited <> "\n# qSubProhibited\n\
+      \\n"
+    messages =
+      "# Messages\n\
+      \# --------\n\
+      \\n\
+      \# HELP simplex_smp_messages_sent Sent messages.\n\
+      \# TYPE simplex_smp_messages_sent counter\n\
+      \simplex_smp_messages_sent " <> mshow _msgSent <> "\n# msgSent\n\
+      \\n\
+      \# HELP simplex_smp_messages_sent_errors Total number of messages errors by type.\n\
+      \# TYPE simplex_smp_messages_sent_errors counter\n\
+      \simplex_smp_messages_sent_errors{type=\"auth\"} " <> mshow _msgSentAuth <> "\n# msgSentAuth\n\
+      \simplex_smp_messages_sent_errors{type=\"quota\"} " <> mshow _msgSentQuota <> "\n# msgSentQuota\n\
+      \simplex_smp_messages_sent_errors{type=\"large\"} " <> mshow _msgSentLarge <> "\n# msgSentLarge\n\
+      \simplex_smp_messages_sent_errors{type=\"block\"} " <> mshow _msgSentBlock <> "\n# msgSentBlock\n\
+      \\n\
+      \# HELP simplex_smp_messages_received Received messages.\n\
+      \# TYPE simplex_smp_messages_received counter\n\
+      \simplex_smp_messages_received " <> mshow _msgRecv <> "\n# msgRecv\n\
+      \\n\
+      \# HELP simplex_smp_messages_expired Expired messages.\n\
+      \# TYPE simplex_smp_messages_expired counter\n\
+      \simplex_smp_messages_expired " <> mshow _msgExpired <> "\n# msgExpired\n\
+      \\n\
+      \# HELP simplex_smp_messages_total Total number of messages stored.\n\
+      \# TYPE simplex_smp_messages_total gauge\n\
+      \simplex_smp_messages_total " <> mshow _msgCount <> "\n# msgCount\n\
+      \\n"
+    ntfMessages =
+      "# Notification messages (client)\n\
+      \# ------------------------------\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_sent Sent messages with notification flag (cleint).\n\
+      \# TYPE simplex_smp_messages_notify_sent counter\n\
+      \simplex_smp_messages_notify_sent " <> mshow _msgSentNtf <> "\n# msgSentNtf\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_received Received messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_received counter\n\
+      \simplex_smp_messages_notify_received " <> mshow _msgRecvNtf <> "\n# msgRecvNtf\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_sent Requests to get messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_get_sent counter\n\
+      \simplex_smp_messages_notify_get_sent " <> mshow _msgGet <> "\n# msgGet\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_received Succesfully received get requests messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_get_received counter\n\
+      \simplex_smp_messages_notify_get_received " <> mshow _msgRecvGet <> "\n# msgRecvGet\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_errors Error events with messages with notification flag (client). \n\
+      \# TYPE simplex_smp_messages_notify_get_errors counter\n\
+      \simplex_smp_messages_notify_get_errors{type=\"nomsg\"} " <> mshow _msgGetNoMsg <> "\n# msgGetNoMsg\n\
+      \simplex_smp_messages_notify_get_errors{type=\"auth\"} " <> mshow _msgGetAuth <> "\n# msgGetAuth\n\
+      \simplex_smp_messages_notify_get_errors{type=\"duplicate\"} " <> mshow _msgGetDuplicate <> "\n# msgGetDuplicate\n\
+      \simplex_smp_messages_notify_get_errors{type=\"prohibited\"} " <> mshow _msgGetProhibited <> "\n# msgGetProhibited\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_created Created queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_created counter\n\
+      \simplex_smp_queues_notify_created " <> mshow _ntfCreated <> "\n# ntfCreated\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_deleted Deleted queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_deleted counter\n\
+      \simplex_smp_queues_notify_deleted " <> mshow _ntfDeleted <> "\n# ntfDeleted\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_deleted_batch Deleted batched queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_deleted_batch counter\n\
+      \simplex_smp_queues_notify_deleted_batch " <> mshow _ntfDeletedB <> "\n# ntfDeletedB\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_total1 Total number of stored queues with notification flag (first type of count).\n\
+      \# TYPE simplex_smp_queues_notify_total1 gauge\n\
+      \simplex_smp_queues_notify_total1 " <> mshow _ntfCount <> "\n# ntfCount1\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_total2 Total number of stored queues with notification flag (second type of count).\n\
+      \# TYPE simplex_smp_queues_notify_total2 gauge\n\
+      \simplex_smp_queues_notify_total2 " <> mshow notifierCount <> "\n# ntfCount2\n\
+      \\n"
+    ntfs =
+      "# Notifications (server)\n\
+      \# ----------------------\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_successes Successful events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_successes counter\n\
+      \simplex_smp_messages_ntf_successes " <> mshow _msgNtfs <> "\n# msgNtfs\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_successes_batch Successful batched events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_successes_batch counter\n\
+      \simplex_smp_messages_ntf_successes_batch " <> mshow _msgNtfsB <> "\n# msgNtfsB\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_errors Error events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_errors counter\n\
+      \simplex_smp_messages_ntf_errors{type=\"nosub\"} " <> mshow _msgNtfNoSub <> "\n# msgNtfNoSub\n\
+      \simplex_smp_messages_ntf_errors{type=\"lost\"} " <> mshow _msgNtfLost <> "\n# msgNtfLost\n\
+      \simplex_smp_messages_ntf_errors{type=\"expired\"} " <> mshow _msgNtfExpired <> "\n# msgNtfExpired\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_requests Subscription requests with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscription_ntf_requests counter\n\
+      \simplex_smp_subscription_ntf_requests " <> mshow _ntfSub <> "\n# ntfSub\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_requests_batch Batched subscription requests with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscription_ntf_requests_batch counter\n\
+      \simplex_smp_subscription_ntf_requests_batch " <> mshow _ntfSubB <> "\n# ntfSubB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_ntf_errors Subscription errors with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscribtion_ntf_errors counter\n\
+      \simplex_smp_subscribtion_ntf_errors{type=\"auth\"} " <> mshow _ntfSubAuth <> "\n# ntfSubAuth\n\
+      \simplex_smp_subscribtion_ntf_errors{type=\"duplicate\"} " <> mshow _ntfSubDuplicate <> "\n# ntfSubDuplicate\n\
+      \\n"
+    relays =
+      "# Relays\n\
+      \# ------\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_requests Session requests through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_requests counter\n\
+      \simplex_smp_relay_sessions_requests{source=\"all\"} " <> mshow (_pRequests _pRelays) <> "\n# pRelays_pRequests\n\
+      \simplex_smp_relay_sessions_requests{source=\"own\"} " <> mshow (_pRequests _pRelaysOwn) <> "\n# pRelaysOwn_pRequests\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_successes Successful session events through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_successes counter\n\
+      \simplex_smp_relay_sessions_successes{source=\"all\"} " <> mshow (_pSuccesses _pRelays) <> "\n# pRelays_pSuccesses\n\
+      \simplex_smp_relay_sessions_successes{source=\"own\"} " <> mshow (_pSuccesses _pRelaysOwn) <> "\n# pRelaysOwn_pSuccesses\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_errors Error session events through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_errors counter\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"connect\"} " <> mshow (_pErrorsConnect _pRelays) <> "\n# pRelays_pErrorsConnect\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"compat\"} " <> mshow (_pErrorsCompat _pRelays) <> "\n# pRelays_pErrorsCompat\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"other\"} " <> mshow (_pErrorsOther _pRelays) <> "\n# pRelays_pErrorsOther\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"connect\"} " <> mshow (_pErrorsConnect _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsConnect\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"compat\"} " <> mshow (_pErrorsCompat _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsCompat\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"other\"} " <> mshow (_pErrorsOther _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsOther\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_requests Message requests sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_requests counter\n\
+      \simplex_smp_relay_messages_requests{source=\"all\"} " <> mshow (_pRequests _pMsgFwds) <> "\n# pMsgFwds_pRequests\n\
+      \simplex_smp_relay_messages_requests{source=\"own\"} " <> mshow (_pRequests _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pRequests\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_successes Successful messages sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_successes counter\n\
+      \simplex_smp_relay_messages_successes{source=\"all\"} " <> mshow (_pSuccesses _pMsgFwds) <> "\n# pMsgFwds_pSuccesses\n\
+      \simplex_smp_relay_messages_successes{source=\"own\"} " <> mshow (_pSuccesses _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pSuccesses\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_errors Error events with messages sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_errors counter\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"connect\"} " <> mshow (_pErrorsConnect _pMsgFwds) <> "\n# pMsgFwds_pErrorsConnect\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"compat\"} " <> mshow (_pErrorsCompat _pMsgFwds) <> "\n# pMsgFwds_pErrorsCompat\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"other\"} " <> mshow (_pErrorsOther _pMsgFwds) <> "\n# pMsgFwds_pErrorsOther\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"connect\"} " <> mshow (_pErrorsConnect _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsConnect\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"compat\"} " <> mshow (_pErrorsCompat _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsCompat\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"other\"} " <> mshow (_pErrorsOther _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsOther\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_received Relay messages statistics.\n\
+      \# TYPE simplex_smp_relay_messages_received counter\n\
+      \simplex_smp_relay_messages_received " <> mshow _pMsgFwdsRecv <> "\n# pMsgFwdsRecv\n\
+      \\n"
+    info =
+      "# Info\n\
+      \# ----\n\
+      \\n"
+      <> socketsMetric socketsAccepted "simplex_smp_sockets_accepted" "Accepted sockets"
+      <> socketsMetric socketsClosed "simplex_smp_sockets_closed" "Closed sockets"
+      <> socketsMetric socketsActive "simplex_smp_sockets_active" "Active sockets"
+      <> socketsMetric socketsLeaked "simplex_smp_sockets_leaked" "Leaked sockets"
+      <> "# HELP simplex_smp_threads_total Threads\n\
+      \# TYPE simplex_smp_threads_total gauge\n\
+      \simplex_smp_threads_total " <> mshow threadsCount <> "\n\
+      \\n\
+      \# HELP simplex_smp_clients_total Clients\n\
+      \# TYPE simplex_smp_clients_total gauge\n\
+      \simplex_smp_clients_total " <> mshow clientsCount <> "\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_total Total subscriptions\n\
+      \# TYPE simplex_smp_subscribtion_total gauge\n\
+      \simplex_smp_subscribtion_total " <> mshow smpSubsCount <> "\n# smpSubs\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_clients_total Subscribed clients, first counting method\n\
+      \# TYPE simplex_smp_subscribtion_clients_total gauge\n\
+      \simplex_smp_subscribtion_clients_total " <> mshow smpSubClientsCount <> "\n# smpSubClients\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_total Total notification subscripbtions (from ntf server)\n\
+      \# TYPE simplex_smp_subscription_ntf_total gauge\n\
+      \simplex_smp_subscription_ntf_total " <> mshow ntfSubsCount <> "\n# ntfSubs\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_clients_total Total subscribed NTF servers, first counting method\n\
+      \# TYPE simplex_smp_subscription_ntf_clients_total gauge\n\
+      \simplex_smp_subscription_ntf_clients_total " <> mshow ntfSubClientsCount <> "\n# ntfSubClients\n"
+    socketsMetric :: (SocketStats -> Int) -> Text -> Text -> Text
+    socketsMetric sel metric descr =
+      "# HELP " <> metric <> " " <> descr <> "\n"
+        <> "# TYPE " <> metric <> " gauge\n"
+        <> T.concat (map (\(port, ss) -> metric <> "{port=\"" <> T.pack port <> "\"} " <> mshow (sel ss) <> "\n") socketStats)
+        <> "\n"
+    mstr a = T.pack a <> " " <> tsEpoch
+    mshow :: Show a => a -> Text
+    mshow = mstr . show
+    tsEpoch = T.pack $ show @Int64 $ floor @Double $ realToFrac (ts `diffUTCTime` epoch) * 1000
+    epoch = UTCTime systemEpochDay 0
+{-# FOURMOLU_ENABLE\n#-}

src/Simplex/Messaging/Server/QueueStore.hs

@@ -1,11 +1,15 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
 
 module Simplex.Messaging.Server.QueueStore where
 
+import Control.Applicative ((<|>))
+import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Simplex.Messaging.Encoding.String
@@ -19,7 +23,7 @@ data QueueRec = QueueRec
     senderKey :: !(Maybe SndPublicAuthKey),
     sndSecure :: !SenderCanSecure,
     notifier :: !(Maybe NtfCreds),
-    status :: !ServerQueueStatus,
+    status :: !ServerEntityStatus,
     updatedAt :: !(Maybe RoundedSystemTime)
   }
   deriving (Show)
@@ -37,7 +41,21 @@ instance StrEncoding NtfCreds where
     (notifierId, notifierKey, rcvNtfDhSecret) <- strP
     pure NtfCreds {notifierId, notifierKey, rcvNtfDhSecret}
 
-data ServerQueueStatus = QueueActive | QueueOff deriving (Eq, Show)
+data ServerEntityStatus
+  = EntityActive
+  | EntityBlocked BlockingInfo
+  | EntityOff
+  deriving (Eq, Show)
+
+instance StrEncoding ServerEntityStatus where
+  strEncode = \case
+    EntityActive -> "active"
+    EntityBlocked info -> "blocked," <> strEncode info
+    EntityOff -> "off"
+  strP =
+    "active" $> EntityActive
+      <|> "blocked," *> (EntityBlocked <$> strP)
+      <|> "off" $> EntityOff
 
 newtype RoundedSystemTime = RoundedSystemTime Int64
   deriving (Eq, Ord, Show)

src/Simplex/Messaging/Server/QueueStore/STM.hs

@@ -21,6 +21,8 @@ module Simplex.Messaging.Server.QueueStore.STM
     addQueueNotifier,
     deleteQueueNotifier,
     suspendQueue,
+    blockQueue,
+    unblockQueue,
     updateQueueTime,
     deleteQueue',
     readQueueStore,
@@ -118,7 +120,27 @@ suspendQueue st sq =
   where
     qr = queueRec' sq
     suspend q = do
-      writeTVar qr $! Just q {status = QueueOff}
+      writeTVar qr $! Just q {status = EntityOff}
+      pure $ recipientId q
+
+blockQueue :: STMQueueStore s => s -> StoreQueue s -> BlockingInfo -> IO (Either ErrorType ())
+blockQueue st sq info =
+  atomically (readQueueRec qr >>= mapM block)
+    $>>= \rId -> withLog "blockQueue" st (\sl -> logBlockQueue sl rId info)
+  where
+    qr = queueRec' sq
+    block q = do
+      writeTVar qr $ Just q {status = EntityBlocked info}
+      pure $ recipientId q
+
+unblockQueue :: STMQueueStore s => s -> StoreQueue s -> IO (Either ErrorType ())
+unblockQueue st sq =
+  atomically (readQueueRec qr >>= mapM unblock)
+    $>>= \rId -> withLog "unblockQueue" st (`logUnblockQueue` rId)
+  where
+    qr = queueRec' sq
+    unblock q = do
+      writeTVar qr $ Just q {status = EntityActive}
       pure $ recipientId q
 
 updateQueueTime :: STMQueueStore s => s -> StoreQueue s -> RoundedSystemTime -> IO (Either ErrorType QueueRec)
@@ -177,6 +199,8 @@ readQueueStore f st = withFile f ReadMode $ LB.hGetContents >=> mapM_ processLin
           SecureQueue qId sKey -> withQueue qId "SecureQueue" $ \q -> secureQueue st q sKey
           AddNotifier qId ntfCreds -> withQueue qId "AddNotifier" $ \q -> addQueueNotifier st q ntfCreds
           SuspendQueue qId -> withQueue qId "SuspendQueue" $ suspendQueue st
+          BlockQueue qId info -> withQueue qId "BlockQueue" $ \q -> blockQueue st q info
+          UnblockQueue qId -> withQueue qId "UnblockQueue" $ unblockQueue st
           DeleteQueue qId -> withQueue qId "DeleteQueue" $ deleteQueue st qId
           DeleteNotifier qId -> withQueue qId "DeleteNotifier" $ deleteQueueNotifier st
           UpdateTime qId t -> withQueue qId "UpdateTime" $ \q -> updateQueueTime st q t

src/Simplex/Messaging/Server/Stats.hs

@@ -34,6 +34,7 @@ data ServerStats = ServerStats
     qDeletedAllB :: IORef Int,
     qDeletedNew :: IORef Int,
     qDeletedSecured :: IORef Int,
+    qBlocked :: IORef Int,
     qSub :: IORef Int, -- only includes subscriptions when there were pending messages
     -- qSubNoMsg :: IORef Int, -- this stat creates too many STM transactions
     qSubAllB :: IORef Int, -- count of all subscription batches (with and without pending messages)
@@ -53,6 +54,7 @@ data ServerStats = ServerStats
     msgSentAuth :: IORef Int,
     msgSentQuota :: IORef Int,
     msgSentLarge :: IORef Int,
+    msgSentBlock :: IORef Int,
     msgRecv :: IORef Int,
     msgRecvGet :: IORef Int,
     msgGet :: IORef Int,
@@ -89,6 +91,7 @@ data ServerStatsData = ServerStatsData
     _qDeletedAllB :: Int,
     _qDeletedNew :: Int,
     _qDeletedSecured :: Int,
+    _qBlocked :: Int,
     _qSub :: Int,
     _qSubAllB :: Int,
     _qSubAuth :: Int,
@@ -107,6 +110,7 @@ data ServerStatsData = ServerStatsData
     _msgSentAuth :: Int,
     _msgSentQuota :: Int,
     _msgSentLarge :: Int,
+    _msgSentBlock :: Int,
     _msgRecv :: Int,
     _msgRecvGet :: Int,
     _msgGet :: Int,
@@ -144,6 +148,7 @@ newServerStats ts = do
   qDeletedAllB <- newIORef 0
   qDeletedNew <- newIORef 0
   qDeletedSecured <- newIORef 0
+  qBlocked <- newIORef 0
   qSub <- newIORef 0
   qSubAllB <- newIORef 0
   qSubAuth <- newIORef 0
@@ -162,6 +167,7 @@ newServerStats ts = do
   msgSentAuth <- newIORef 0
   msgSentQuota <- newIORef 0
   msgSentLarge <- newIORef 0
+  msgSentBlock <- newIORef 0
   msgRecv <- newIORef 0
   msgRecvGet <- newIORef 0
   msgGet <- newIORef 0
@@ -196,6 +202,7 @@ newServerStats ts = do
         qDeletedAllB,
         qDeletedNew,
         qDeletedSecured,
+        qBlocked,
         qSub,
         qSubAllB,
         qSubAuth,
@@ -214,6 +221,7 @@ newServerStats ts = do
         msgSentAuth,
         msgSentQuota,
         msgSentLarge,
+        msgSentBlock,
         msgRecv,
         msgRecvGet,
         msgGet,
@@ -250,6 +258,7 @@ getServerStatsData s = do
   _qDeletedAllB <- readIORef $ qDeletedAllB s
   _qDeletedNew <- readIORef $ qDeletedNew s
   _qDeletedSecured <- readIORef $ qDeletedSecured s
+  _qBlocked <- readIORef $ qBlocked s
   _qSub <- readIORef $ qSub s
   _qSubAllB <- readIORef $ qSubAllB s
   _qSubAuth <- readIORef $ qSubAuth s
@@ -268,6 +277,7 @@ getServerStatsData s = do
   _msgSentAuth <- readIORef $ msgSentAuth s
   _msgSentQuota <- readIORef $ msgSentQuota s
   _msgSentLarge <- readIORef $ msgSentLarge s
+  _msgSentBlock <- readIORef $ msgSentBlock s
   _msgRecv <- readIORef $ msgRecv s
   _msgRecvGet <- readIORef $ msgRecvGet s
   _msgGet <- readIORef $ msgGet s
@@ -302,6 +312,7 @@ getServerStatsData s = do
         _qDeletedAllB,
         _qDeletedNew,
         _qDeletedSecured,
+        _qBlocked,
         _qSub,
         _qSubAllB,
         _qSubAuth,
@@ -320,6 +331,7 @@ getServerStatsData s = do
         _msgSentAuth,
         _msgSentQuota,
         _msgSentLarge,
+        _msgSentBlock,
         _msgRecv,
         _msgRecvGet,
         _msgGet,
@@ -357,6 +369,7 @@ setServerStats s d = do
   writeIORef (qDeletedAllB s) $! _qDeletedAllB d
   writeIORef (qDeletedNew s) $! _qDeletedNew d
   writeIORef (qDeletedSecured s) $! _qDeletedSecured d
+  writeIORef (qBlocked s) $! _qBlocked d
   writeIORef (qSub s) $! _qSub d
   writeIORef (qSubAllB s) $! _qSubAllB d
   writeIORef (qSubAuth s) $! _qSubAuth d
@@ -375,6 +388,7 @@ setServerStats s d = do
   writeIORef (msgSentAuth s) $! _msgSentAuth d
   writeIORef (msgSentQuota s) $! _msgSentQuota d
   writeIORef (msgSentLarge s) $! _msgSentLarge d
+  writeIORef (msgSentBlock s) $! _msgSentBlock d
   writeIORef (msgRecv s) $! _msgRecv d
   writeIORef (msgRecvGet s) $! _msgRecvGet d
   writeIORef (msgGet s) $! _msgGet d
@@ -411,6 +425,7 @@ instance StrEncoding ServerStatsData where
         "qDeletedNew=" <> strEncode (_qDeletedNew d),
         "qDeletedSecured=" <> strEncode (_qDeletedSecured d),
         "qDeletedAllB=" <> strEncode (_qDeletedAllB d),
+        "qBlocked=" <> strEncode (_qBlocked d),
         "qCount=" <> strEncode (_qCount d),
         "qSub=" <> strEncode (_qSub d),
         "qSubAllB=" <> strEncode (_qSubAllB d),
@@ -430,6 +445,7 @@ instance StrEncoding ServerStatsData where
         "msgSentAuth=" <> strEncode (_msgSentAuth d),
         "msgSentQuota=" <> strEncode (_msgSentQuota d),
         "msgSentLarge=" <> strEncode (_msgSentLarge d),
+        "msgSentBlock=" <> strEncode (_msgSentBlock d),
         "msgRecv=" <> strEncode (_msgRecv d),
         "msgRecvGet=" <> strEncode (_msgRecvGet d),
         "msgGet=" <> strEncode (_msgGet d),
@@ -467,6 +483,7 @@ instance StrEncoding ServerStatsData where
       (,0,0) <$> ("qDeleted=" *> strP <* A.endOfLine)
         <|> ((,,) <$> ("qDeletedAll=" *> strP <* A.endOfLine) <*> ("qDeletedNew=" *> strP <* A.endOfLine) <*> ("qDeletedSecured=" *> strP <* A.endOfLine))
     _qDeletedAllB <- opt "qDeletedAllB="
+    _qBlocked <- opt "qBlocked="
     _qCount <- opt "qCount="
     _qSub <- opt "qSub="
     _qSubNoMsg <- skipInt "qSubNoMsg=" -- skipping it for backward compatibility
@@ -487,6 +504,7 @@ instance StrEncoding ServerStatsData where
     _msgSentAuth <- opt "msgSentAuth="
     _msgSentQuota <- opt "msgSentQuota="
     _msgSentLarge <- opt "msgSentLarge="
+    _msgSentBlock <- opt "msgSentBlock="
     _msgRecv <- "msgRecv=" *> strP <* A.endOfLine
     _msgRecvGet <- opt "msgRecvGet="
     _msgGet <- opt "msgGet="
@@ -532,6 +550,7 @@ instance StrEncoding ServerStatsData where
           _qDeletedAllB,
           _qDeletedNew,
           _qDeletedSecured,
+          _qBlocked,
           _qSub,
           _qSubAllB,
           _qSubAuth,
@@ -550,6 +569,7 @@ instance StrEncoding ServerStatsData where
           _msgSentAuth,
           _msgSentQuota,
           _msgSentLarge,
+          _msgSentBlock,
           _msgRecv,
           _msgRecvGet,
           _msgGet,
@@ -638,6 +658,14 @@ data PeriodStatCounts = PeriodStatCounts
     monthCount :: String
   }
 
+periodStatDataCounts :: PeriodStatsData -> PeriodStatCounts
+periodStatDataCounts PeriodStatsData {_day, _week, _month} =
+  PeriodStatCounts
+    { dayCount = show $ IS.size _day,
+      weekCount = show $ IS.size _week,
+      monthCount = show $ IS.size _month
+    }
+
 periodStatCounts :: PeriodStats -> UTCTime -> IO PeriodStatCounts
 periodStatCounts ps ts = do
   let d = utctDay ts

src/Simplex/Messaging/Server/StoreLog.hs

@@ -21,6 +21,8 @@ module Simplex.Messaging.Server.StoreLog
     logSecureQueue,
     logAddNotifier,
     logSuspendQueue,
+    logBlockQueue,
+    logUnblockQueue,
     logDeleteQueue,
     logDeleteNotifier,
     logUpdateQueueTime,
@@ -33,9 +35,9 @@ import Control.Applicative (optional, (<|>))
 import Control.Concurrent.STM
 import qualified Control.Exception as E
 import Control.Logger.Simple
-import Control.Monad
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import qualified Data.ByteString.Char8 as B
+import Data.Functor (($>))
 import qualified Data.Map.Strict as M
 import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
@@ -56,6 +58,8 @@ data StoreLogRecord
   | SecureQueue QueueId SndPublicAuthKey
   | AddNotifier QueueId NtfCreds
   | SuspendQueue QueueId
+  | BlockQueue QueueId BlockingInfo
+  | UnblockQueue QueueId
   | DeleteQueue QueueId
   | DeleteNotifier QueueId
   | UpdateTime QueueId RoundedSystemTime
@@ -66,12 +70,14 @@ data SLRTag
   | SecureQueue_
   | AddNotifier_
   | SuspendQueue_
+  | BlockQueue_
+  | UnblockQueue_
   | DeleteQueue_
   | DeleteNotifier_
   | UpdateTime_
 
 instance StrEncoding QueueRec where
-  strEncode QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, updatedAt} =
+  strEncode QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt} =
     B.unwords
       [ "rid=" <> strEncode recipientId,
         "rk=" <> strEncode recipientKey,
@@ -82,10 +88,14 @@ instance StrEncoding QueueRec where
       <> sndSecureStr
       <> maybe "" notifierStr notifier
       <> maybe "" updatedAtStr updatedAt
+      <> statusStr
     where
       sndSecureStr = if sndSecure then " sndSecure=" <> strEncode sndSecure else ""
       notifierStr ntfCreds = " notifier=" <> strEncode ntfCreds
       updatedAtStr t = " updated_at=" <> strEncode t
+      statusStr = case status of
+        EntityActive -> ""
+        _ -> " status=" <> strEncode status
 
   strP = do
     recipientId <- "rid=" *> strP_
@@ -96,7 +106,8 @@ instance StrEncoding QueueRec where
     sndSecure <- (" sndSecure=" *> strP) <|> pure False
     notifier <- optional $ " notifier=" *> strP
     updatedAt <- optional $ " updated_at=" *> strP
-    pure QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status = QueueActive, updatedAt}
+    status <- (" status=" *> strP) <|> pure EntityActive
+    pure QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt}
 
 instance StrEncoding SLRTag where
   strEncode = \case
@@ -104,20 +115,24 @@ instance StrEncoding SLRTag where
     SecureQueue_ -> "SECURE"
     AddNotifier_ -> "NOTIFIER"
     SuspendQueue_ -> "SUSPEND"
+    BlockQueue_ -> "BLOCK"
+    UnblockQueue_ -> "UNBLOCK"
     DeleteQueue_ -> "DELETE"
     DeleteNotifier_ -> "NDELETE"
     UpdateTime_ -> "TIME"
 
   strP =
-    A.takeTill (== ' ') >>= \case
-      "CREATE" -> pure CreateQueue_
-      "SECURE" -> pure SecureQueue_
-      "NOTIFIER" -> pure AddNotifier_
-      "SUSPEND" -> pure SuspendQueue_
-      "DELETE" -> pure DeleteQueue_
-      "NDELETE" -> pure DeleteNotifier_
-      "TIME" -> pure UpdateTime_
-      s -> fail $ "invalid log record tag: " <> B.unpack s
+    A.choice
+      [ "CREATE" $> CreateQueue_,
+        "SECURE" $> SecureQueue_,
+        "NOTIFIER" $> AddNotifier_,
+        "SUSPEND" $> SuspendQueue_,
+        "BLOCK" $> BlockQueue_,
+        "UNBLOCK" $> UnblockQueue_,
+        "DELETE" $> DeleteQueue_,
+        "NDELETE" $> DeleteNotifier_,
+        "TIME" $> UpdateTime_
+      ]
 
 instance StrEncoding StoreLogRecord where
   strEncode = \case
@@ -125,6 +140,8 @@ instance StrEncoding StoreLogRecord where
     SecureQueue rId sKey -> strEncode (SecureQueue_, rId, sKey)
     AddNotifier rId ntfCreds -> strEncode (AddNotifier_, rId, ntfCreds)
     SuspendQueue rId -> strEncode (SuspendQueue_, rId)
+    BlockQueue rId info -> strEncode (BlockQueue_, rId, info)
+    UnblockQueue rId -> strEncode (UnblockQueue_, rId)
     DeleteQueue rId -> strEncode (DeleteQueue_, rId)
     DeleteNotifier rId -> strEncode (DeleteNotifier_, rId)
     UpdateTime rId t ->  strEncode (UpdateTime_, rId, t)
@@ -135,6 +152,8 @@ instance StrEncoding StoreLogRecord where
       SecureQueue_ -> SecureQueue <$> strP_ <*> strP
       AddNotifier_ -> AddNotifier <$> strP_ <*> strP
       SuspendQueue_ -> SuspendQueue <$> strP
+      BlockQueue_ -> BlockQueue <$> strP_ <*> strP
+      UnblockQueue_ -> UnblockQueue <$> strP
       DeleteQueue_ -> DeleteQueue <$> strP
       DeleteNotifier_ -> DeleteNotifier <$> strP
       UpdateTime_ -> UpdateTime <$> strP_ <*> strP
@@ -179,6 +198,12 @@ logAddNotifier s qId ntfCreds = writeStoreLogRecord s $ AddNotifier qId ntfCreds
 logSuspendQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
 logSuspendQueue s = writeStoreLogRecord s . SuspendQueue
 
+logBlockQueue :: StoreLog 'WriteMode -> QueueId -> BlockingInfo -> IO ()
+logBlockQueue s qId info = writeStoreLogRecord s $ BlockQueue qId info
+
+logUnblockQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
+logUnblockQueue s = writeStoreLogRecord s . UnblockQueue
+
 logDeleteQueue :: StoreLog 'WriteMode -> QueueId -> IO ()
 logDeleteQueue s = writeStoreLogRecord s . DeleteQueue
 
@@ -228,6 +253,5 @@ writeQueueStore s st = readTVarIO (activeMsgQueues st) >>= mapM_ writeQueue . M.
   where
     writeQueue (rId, q) =
       readTVarIO (queueRec' q) >>= \case
-        Just q' -> when (active q') $ logCreateQueue s q' -- TODO we should log suspended queues when we use them
+        Just q' -> logCreateQueue s q'
         Nothing -> atomically $ TM.delete rId $ activeMsgQueues st
-    active QueueRec {status} = status == QueueActive

src/Simplex/Messaging/Transport.hs

@@ -49,6 +49,7 @@ module Simplex.Messaging.Transport
     sndAuthKeySMPVersion,
     deletedEventSMPVersion,
     encryptedBlockSMPVersion,
+    blockedEntityErrorSMPVersion,
     simplexMQVersion,
     smpBlockSize,
     TransportConfig (..),
@@ -142,6 +143,7 @@ smpBlockSize = 16384
 -- 9 - faster handshake: SKEY command for sender to secure queue
 -- 10 - DELD event to subscriber when queue is deleted via another connnection
 -- 11 - additional encryption of transport blocks with forward secrecy (9/14/2024)
+-- 12 - BLOCKED error for blocked queues (1/11/2025)
 
 data SMPVersion
 
@@ -178,14 +180,17 @@ deletedEventSMPVersion = VersionSMP 10
 encryptedBlockSMPVersion :: VersionSMP
 encryptedBlockSMPVersion = VersionSMP 11
 
+blockedEntityErrorSMPVersion :: VersionSMP
+blockedEntityErrorSMPVersion = VersionSMP 12
+
 currentClientSMPRelayVersion :: VersionSMP
-currentClientSMPRelayVersion = VersionSMP 11
+currentClientSMPRelayVersion = VersionSMP 12
 
 legacyServerSMPRelayVersion :: VersionSMP
 legacyServerSMPRelayVersion = VersionSMP 6
 
 currentServerSMPRelayVersion :: VersionSMP
-currentServerSMPRelayVersion = VersionSMP 11
+currentServerSMPRelayVersion = VersionSMP 12
 
 -- Max SMP protocol version to be used in e2e encrypted
 -- connection between client and server, as defined by SMP proxy.
@@ -193,7 +198,7 @@ currentServerSMPRelayVersion = VersionSMP 11
 -- to prevent client version fingerprinting by the
 -- destination relays when clients upgrade at different times.
 proxiedSMPRelayVersion :: VersionSMP
-proxiedSMPRelayVersion = VersionSMP 9
+proxiedSMPRelayVersion = VersionSMP 12
 
 -- minimal supported protocol version is 4
 -- TODO remove code that supports sending commands without batching

src/Simplex/Messaging/Transport/Server.hs

@@ -12,7 +12,9 @@ module Simplex.Messaging.Transport.Server
     runTransportServerState,
     runTransportServerState_,
     SocketState,
+    SocketStats (..),
     newSocketState,
+    getSocketStats,
     runTransportServer,
     runTransportServerSocket,
     runLocalTCPServer,
@@ -163,9 +165,24 @@ safeAccept sock =
 
 type SocketState = (TVar Int, TVar Int, TVar (IntMap (Weak ThreadId)))
 
+data SocketStats = SocketStats
+  { socketsAccepted :: Int,
+    socketsClosed :: Int,
+    socketsActive :: Int,
+    socketsLeaked :: Int
+  }
+
 newSocketState :: IO SocketState
 newSocketState = (,,) <$> newTVarIO 0 <*> newTVarIO 0 <*> newTVarIO mempty
 
+getSocketStats :: SocketState -> IO SocketStats
+getSocketStats (accepted, closed, active) = do
+  socketsAccepted <- readTVarIO accepted
+  socketsClosed <- readTVarIO closed
+  socketsActive <- IM.size <$> readTVarIO active
+  let socketsLeaked = socketsAccepted - socketsClosed - socketsActive
+  pure SocketStats {socketsAccepted, socketsClosed, socketsActive, socketsLeaked}
+
 closeServer :: TMVar Bool -> TVar (IntMap (Weak ThreadId)) -> Socket -> IO ()
 closeServer started clients sock = do
   close sock

tests/AgentTests.hs

@@ -1,12 +1,10 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE GADTs #-}
-{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE PostfixOperators #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TypeApplications #-}
 
 module AgentTests (agentTests) where
 
@@ -14,18 +12,30 @@ import AgentTests.ConnectionRequestTests
 import AgentTests.DoubleRatchetTests (doubleRatchetTests)
 import AgentTests.FunctionalAPITests (functionalAPITests)
 import AgentTests.MigrationTests (migrationTests)
-import AgentTests.NotificationTests (notificationTests)
-import AgentTests.SQLiteTests (storeTests)
 import AgentTests.ServerChoice (serverChoiceTests)
 import Simplex.Messaging.Transport (ATransport (..))
 import Test.Hspec
+#if defined(dbPostgres)
+import Fixtures
+import Simplex.Messaging.Agent.Store.Postgres.Util (dropAllSchemasExceptSystem)
+#else
+import AgentTests.NotificationTests (notificationTests)
+import AgentTests.SQLiteTests (storeTests)
+#endif
 
 agentTests :: ATransport -> Spec
 agentTests (ATransport t) = do
+  describe "Migration tests" migrationTests
   describe "Connection request" connectionRequestTests
   describe "Double ratchet tests" doubleRatchetTests
+#if defined(dbPostgres)
+  after_ (dropAllSchemasExceptSystem testDBConnectInfo) $ do
+    describe "Functional API" $ functionalAPITests (ATransport t)
+    describe "Chosen servers" serverChoiceTests
+#else
   describe "Functional API" $ functionalAPITests (ATransport t)
+  describe "Chosen servers" serverChoiceTests
+  -- notifications aren't tested with postgres, as we don't plan to use iOS client with it
   describe "Notification tests" $ notificationTests (ATransport t)
   describe "SQLite store" storeTests
-  describe "Chosen servers" serverChoiceTests
-  describe "Migration tests" migrationTests
+#endif

tests/AgentTests/FunctionalAPITests.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
@@ -74,7 +75,6 @@ import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Type.Equality (testEquality, (:~:) (Refl))
 import Data.Word (Word16)
-import qualified Database.SQLite.Simple as SQL
 import GHC.Stack (withFrozenCallStack)
 import SMPAgentClient
 import SMPClient (cfg, prevRange, prevVersion, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerProxy, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
@@ -84,8 +84,9 @@ import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestSte
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers (..), createAgentStore)
 import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, REQ, SENT)
 import qualified Simplex.Messaging.Agent.Protocol as A
-import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteStore (dbNew))
-import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
+import Simplex.Messaging.Agent.Store.Common (DBStore (..), withTransaction)
+import qualified Simplex.Messaging.Agent.Store.DB as DB
+import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..), MigrationError (..))
 import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), SMPProxyFallback (..), SMPProxyMode (..), TransportSessionMode (..), defaultClientConfig)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern IKPQOff, pattern IKPQOn, pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
@@ -108,6 +109,9 @@ import Test.Hspec
 import UnliftIO
 import Util
 import XFTPClient (testXFTPServer)
+#if defined(dbPostgres)
+import Fixtures
+#endif
 
 type AEntityTransmission e = (ACorrId, ConnId, AEvent e)
 
@@ -325,6 +329,8 @@ functionalAPITests t = do
       it "should expire multiple messages" $ testExpireManyMessages t
       it "should expire one message if quota is exceeded" $ testExpireMessageQuota t
       it "should expire multiple messages if quota is exceeded" $ testExpireManyMessagesQuota t
+#if !defined(dbPostgres)
+    -- TODO [postgres] restore from outdated db backup (we use copyFile/renameFile for sqlite)
     describe "Ratchet synchronization" $ do
       it "should report ratchet de-synchronization, synchronize ratchets" $
         testRatchetSync t
@@ -336,6 +342,7 @@ functionalAPITests t = do
         testRatchetSyncSuspendForeground t
       it "should synchronize ratchets when clients start synchronization simultaneously" $
         testRatchetSyncSimultaneous t
+#endif
     describe "Subscription mode OnlyCreate" $ do
       it "messages delivered only when polled (v8 - slow handshake)" $
         withSmpServer t testOnlyCreatePullSlowHandshake
@@ -2561,7 +2568,7 @@ testSwitchAsync servers = do
     withB :: (AgentClient -> IO a) -> IO a
     withB = withAgent 2 agentCfg servers testDB2
 
-withAgent :: HasCallStack => Int -> AgentConfig -> InitialAgentServers -> FilePath -> (HasCallStack => AgentClient -> IO a) -> IO a
+withAgent :: HasCallStack => Int -> AgentConfig -> InitialAgentServers -> String -> (HasCallStack => AgentClient -> IO a) -> IO a
 withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) (\a -> disposeAgentClient a >> threadDelay 100000)
 
 sessionSubscribe :: (forall a. (AgentClient -> IO a) -> IO a) -> [ConnId] -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO ()
@@ -3091,13 +3098,27 @@ testTwoUsers = withAgentClients2 $ \a b -> do
     hasClients :: HasCallStack => AgentClient -> Int -> ExceptT AgentErrorType IO ()
     hasClients c n = liftIO $ M.size <$> readTVarIO (smpClients c) `shouldReturn` n
 
-getSMPAgentClient' :: Int -> AgentConfig -> InitialAgentServers -> FilePath -> IO AgentClient
+getSMPAgentClient' :: Int -> AgentConfig -> InitialAgentServers -> String -> IO AgentClient
 getSMPAgentClient' clientId cfg' initServers dbPath = do
-  Right st <- liftIO $ createAgentStore dbPath "" False MCError
+  Right st <- liftIO $ createStore dbPath
   c <- getSMPAgentClient_ clientId cfg' initServers st False
-  when (dbNew st) $ withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1)")
+  when (dbNew st) $ insertUser st
   pure c
 
+#if defined(dbPostgres)
+createStore :: String -> IO (Either MigrationError DBStore)
+createStore schema = createAgentStore testDBConnectInfo schema MCError
+
+insertUser :: DBStore -> IO ()
+insertUser st = withTransaction st (`DB.execute_` "INSERT INTO users DEFAULT VALUES")
+#else
+createStore :: String -> IO (Either MigrationError DBStore)
+createStore dbPath = createAgentStore dbPath "" False MCError True
+
+insertUser :: DBStore -> IO ()
+insertUser st = withTransaction st (`DB.execute_` "INSERT INTO users (user_id) VALUES (1)")
+#endif
+
 testServerMultipleIdentities :: HasCallStack => IO ()
 testServerMultipleIdentities =
   withAgentClients2 $ \alice bob -> runRight_ $ do

tests/AgentTests/MigrationTests.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE OverloadedStrings #-}
 
 module AgentTests.MigrationTests (migrationTests) where
@@ -5,13 +6,23 @@ module AgentTests.MigrationTests (migrationTests) where
 import Control.Monad
 import Data.Maybe (fromJust)
 import Data.Word (Word32)
-import Database.SQLite.Simple (fromOnly)
-import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), MigrationError (MEDowngrade, MEUpgrade, MigrationError), SQLiteStore, closeSQLiteStore, createSQLiteStore, upMigration, withTransaction)
-import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
-import Simplex.Messaging.Agent.Store.SQLite.Migrations
-import System.Directory (removeFile)
+import Simplex.Messaging.Agent.Store.Common (DBStore, withTransaction)
+import Simplex.Messaging.Agent.Store.Migrations (migrationsToRun)
+import Simplex.Messaging.Agent.Store.Shared
 import System.Random (randomIO)
 import Test.Hspec
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple (fromOnly)
+import Fixtures
+import Simplex.Messaging.Agent.Store.Postgres (closeDBStore, createDBStore)
+import Simplex.Messaging.Agent.Store.Postgres.Util (dropSchema)
+import qualified Simplex.Messaging.Agent.Store.Postgres.DB as DB
+#else
+import Database.SQLite.Simple (fromOnly)
+import Simplex.Messaging.Agent.Store.SQLite (closeDBStore, createDBStore)
+import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
+import System.Directory (removeFile)
+#endif
 
 migrationTests :: Spec
 migrationTests = do
@@ -95,9 +106,6 @@ migrationTests = do
         ([m1, m2, m3, m4], [t1, t2, t3, t4])
         ([m1, m2, m4], [MCYesUp, MCYesUpDown, MCError], Left . MigrationError $ MTREDifferent (name m4) (name m3))
 
-testDB :: FilePath
-testDB = "tests/tmp/test_migrations.db"
-
 m1 :: Migration
 m1 = Migration "20230301-migration1" "create table test1 (id1 integer primary key);" Nothing
 
@@ -177,21 +185,46 @@ testMigration ::
   IO ()
 testMigration (initMs, initTables) (finalMs, confirmModes, tablesOrError) = forM_ confirmModes $ \confirmMode -> do
   r <- randomIO :: IO Word32
-  let dpPath = testDB <> show r
-  Right st <- createSQLiteStore dpPath "" False initMs MCError
+  Right st <- createStore r initMs MCError
   st `shouldHaveTables` initTables
-  closeSQLiteStore st
+  closeDBStore st
   case tablesOrError of
     Right tables -> do
-      Right st' <- createSQLiteStore dpPath "" False finalMs confirmMode
+      Right st' <- createStore r finalMs confirmMode
       st' `shouldHaveTables` tables
-      closeSQLiteStore st'
+      closeDBStore st'
     Left e -> do
-      Left e' <- createSQLiteStore dpPath "" False finalMs confirmMode
+      Left e' <- createStore r finalMs confirmMode
       e `shouldBe` e'
-  removeFile dpPath
-  where
-    shouldHaveTables :: SQLiteStore -> [String] -> IO ()
-    st `shouldHaveTables` expected = do
-      tables <- map fromOnly <$> withTransaction st (`DB.query_` "SELECT name FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY 1;")
-      tables `shouldBe` "migrations" : expected
+  cleanup r
+
+#if defined(dbPostgres)
+testSchema :: Word32 -> String
+testSchema randSuffix = "test_migrations_schema" <> show randSuffix
+
+createStore :: Word32 -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError DBStore)
+createStore randSuffix migrations confirmMigrations =
+  createDBStore testDBConnectInfo (testSchema randSuffix) migrations confirmMigrations
+
+cleanup :: Word32 -> IO ()
+cleanup randSuffix = dropSchema testDBConnectInfo (testSchema randSuffix)
+
+shouldHaveTables :: DBStore -> [String] -> IO ()
+st `shouldHaveTables` expected = do
+  tables <- map fromOnly <$> withTransaction st (`DB.query_` "SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema() AND table_type = 'BASE TABLE' ORDER BY 1")
+  tables `shouldBe` "migrations" : expected
+#else
+testDB :: Word32 -> FilePath
+testDB randSuffix = "tests/tmp/test_migrations.db" <> show randSuffix
+
+createStore :: Word32 -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError DBStore)
+createStore randSuffix migrations migrationConf = createDBStore (testDB randSuffix) "" False migrations migrationConf True
+
+cleanup :: Word32 -> IO ()
+cleanup randSuffix = removeFile (testDB randSuffix)
+
+shouldHaveTables :: DBStore -> [String] -> IO ()
+st `shouldHaveTables` expected = do
+  tables <- map fromOnly <$> withTransaction st (`DB.query_` "SELECT name FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY 1")
+  tables `shouldBe` "migrations" : expected
+#endif

tests/AgentTests/NotificationTests.hs

@@ -61,7 +61,9 @@ import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMes
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), withStore')
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, Env (..), InitialAgentServers)
 import Simplex.Messaging.Agent.Protocol hiding (CON, CONF, INFO, SENT)
-import Simplex.Messaging.Agent.Store.SQLite (closeSQLiteStore, getSavedNtfToken, reopenSQLiteStore, withTransaction)
+import Simplex.Messaging.Agent.Store.AgentStore (getSavedNtfToken)
+import Simplex.Messaging.Agent.Store.SQLite (closeDBStore, reopenSQLiteStore)
+import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction)
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
@@ -74,15 +76,9 @@ import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), NtfSer
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Transport (ATransport)
-import System.Directory (doesFileExist, removeFile)
 import Test.Hspec
 import UnliftIO
 
-removeFileIfExists :: FilePath -> IO ()
-removeFileIfExists filePath = do
-  fileExists <- doesFileExist filePath
-  when fileExists $ removeFile filePath
-
 notificationTests :: ATransport -> Spec
 notificationTests t = do
   describe "Managing notification tokens" $ do
@@ -500,7 +496,7 @@ testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {ag
 
   threadDelay 500000
   suspendAgent alice 0
-  closeSQLiteStore store
+  closeDBStore store
   threadDelay 1000000
   putStrLn "before opening the database from another agent"
 

tests/AgentTests/SQLiteTests.hs

@@ -41,10 +41,12 @@ import Simplex.FileTransfer.Types
 import Simplex.Messaging.Agent.Client ()
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store
+import Simplex.Messaging.Agent.Store.AgentStore
 import Simplex.Messaging.Agent.Store.SQLite
-import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
+import Simplex.Messaging.Agent.Store.SQLite.Common (DBStore (..), withTransaction')
 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.Shared (MigrationConfirmation (..))
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Crypto.File (CryptoFile (..))
 import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), pattern PQSupportOn)
@@ -59,36 +61,36 @@ import UnliftIO.Directory (removeFile)
 testDB :: String
 testDB = "tests/tmp/smp-agent.test.db"
 
-withStore :: SpecWith SQLiteStore -> Spec
-withStore = before createStore . after removeStore
+withStore :: SpecWith DBStore -> Spec
+withStore = before createStore' . after removeStore
 
-withStore2 :: SpecWith (SQLiteStore, SQLiteStore) -> Spec
+withStore2 :: SpecWith (DBStore, DBStore) -> Spec
 withStore2 = before connect2 . after (removeStore . fst)
   where
-    connect2 :: IO (SQLiteStore, SQLiteStore)
+    connect2 :: IO (DBStore, DBStore)
     connect2 = do
-      s1 <- createStore
+      s1 <- createStore'
       s2 <- connectSQLiteStore (dbFilePath s1) "" False
       pure (s1, s2)
 
-createStore :: IO SQLiteStore
-createStore = createEncryptedStore "" False
+createStore' :: IO DBStore
+createStore' = createEncryptedStore "" False
 
-createEncryptedStore :: ScrubbedBytes -> Bool -> IO SQLiteStore
+createEncryptedStore :: ScrubbedBytes -> Bool -> IO DBStore
 createEncryptedStore key keepKey = do
   -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
   -- IO operations on multiple similarly named files; error seems to be environment specific
   r <- randomIO :: IO Word32
-  Right st <- createSQLiteStore (testDB <> show r) key keepKey Migrations.app MCError
+  Right st <- createDBStore (testDB <> show r) key keepKey Migrations.app MCError True
   withTransaction' st (`SQL.execute_` "INSERT INTO users (user_id) VALUES (1);")
   pure st
 
-removeStore :: SQLiteStore -> IO ()
+removeStore :: DBStore -> IO ()
 removeStore db = do
   close db
   removeFile $ dbFilePath db
   where
-    close :: SQLiteStore -> IO ()
+    close :: DBStore -> IO ()
     close st = mapM_ DB.close =<< tryTakeMVar (dbConnection st)
 
 storeTests :: Spec
@@ -147,7 +149,7 @@ storeTests = do
     it "should close and re-open encrypted store" testCloseReopenEncryptedStore
     it "should close and re-open encrypted store (keep key)" testReopenEncryptedStoreKeepKey
 
-testConcurrentWrites :: SpecWith (SQLiteStore, SQLiteStore)
+testConcurrentWrites :: SpecWith (DBStore, DBStore)
 testConcurrentWrites =
   it "should complete multiple concurrent write transactions w/t sqlite busy errors" $ \(s1, s2) -> do
     g <- C.newRandom
@@ -156,22 +158,22 @@ testConcurrentWrites =
     let ConnData {connId} = cData1
     concurrently_ (runTest s1 connId rq) (runTest s2 connId rq)
   where
-    runTest :: SQLiteStore -> ConnId -> RcvQueue -> IO ()
+    runTest :: DBStore -> ConnId -> RcvQueue -> IO ()
     runTest st connId rq = replicateM_ 100 . withTransaction st $ \db -> do
       (internalId, internalRcvId, _, _) <- updateRcvIds db connId
       let rcvMsgData = mkRcvMsgData internalId internalRcvId 0 "0" "hash_dummy"
       createRcvMsg db connId rq rcvMsgData
 
-testCompiledThreadsafe :: SpecWith SQLiteStore
+testCompiledThreadsafe :: SpecWith DBStore
 testCompiledThreadsafe =
   it "compiled sqlite library should be threadsafe" . withStoreTransaction $ \db -> do
     compileOptions <- DB.query_ db "pragma COMPILE_OPTIONS;" :: IO [[T.Text]]
     compileOptions `shouldNotContain` [["THREADSAFE=0"]]
 
-withStoreTransaction :: (DB.Connection -> IO a) -> SQLiteStore -> IO a
+withStoreTransaction :: (DB.Connection -> IO a) -> DBStore -> IO a
 withStoreTransaction = flip withTransaction
 
-testForeignKeysEnabled :: SpecWith SQLiteStore
+testForeignKeysEnabled :: SpecWith DBStore
 testForeignKeysEnabled =
   it "foreign keys should be enabled" . withStoreTransaction $ \db -> do
     let inconsistentQuery =
@@ -261,7 +263,7 @@ createRcvConn db g cData rq cMode = runExceptT $ do
   rq' <- ExceptT $ updateNewConnRcv db connId rq
   pure (connId, rq')
 
-testCreateRcvConn :: SpecWith SQLiteStore
+testCreateRcvConn :: SpecWith DBStore
 testCreateRcvConn =
   it "should create RcvConnection and add SndQueue" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -275,7 +277,7 @@ testCreateRcvConn =
     getConn db "conn1"
       `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq] [sq]))
 
-testCreateRcvConnRandomId :: SpecWith SQLiteStore
+testCreateRcvConnRandomId :: SpecWith DBStore
 testCreateRcvConnRandomId =
   it "should create RcvConnection and add SndQueue with random ID" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -287,7 +289,7 @@ testCreateRcvConnRandomId =
     getConn db connId
       `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} [rq] [sq]))
 
-testCreateRcvConnDuplicate :: SpecWith SQLiteStore
+testCreateRcvConnDuplicate :: SpecWith DBStore
 testCreateRcvConnDuplicate =
   it "should throw error on attempt to create duplicate RcvConnection" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -295,7 +297,7 @@ testCreateRcvConnDuplicate =
     createRcvConn db g cData1 rcvQueue1 SCMInvitation
       `shouldReturn` Left SEConnDuplicate
 
-testCreateSndConn :: SpecWith SQLiteStore
+testCreateSndConn :: SpecWith DBStore
 testCreateSndConn =
   it "should create SndConnection and add RcvQueue" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -309,7 +311,7 @@ testCreateSndConn =
     getConn db "conn1"
       `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 [rq] [sq]))
 
-testCreateSndConnRandomID :: SpecWith SQLiteStore
+testCreateSndConnRandomID :: SpecWith DBStore
 testCreateSndConnRandomID =
   it "should create SndConnection and add RcvQueue with random ID" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -321,7 +323,7 @@ testCreateSndConnRandomID =
     getConn db connId
       `shouldReturn` Right (SomeConn SCDuplex (DuplexConnection cData1 {connId} [rq] [sq]))
 
-testCreateSndConnDuplicate :: SpecWith SQLiteStore
+testCreateSndConnDuplicate :: SpecWith DBStore
 testCreateSndConnDuplicate =
   it "should throw error on attempt to create duplicate SndConnection" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -329,7 +331,7 @@ testCreateSndConnDuplicate =
     createSndConn db g cData1 sndQueue1
       `shouldReturn` Left SEConnDuplicate
 
-testGetRcvConn :: SpecWith SQLiteStore
+testGetRcvConn :: SpecWith DBStore
 testGetRcvConn =
   it "should get connection using rcv queue id and server" . withStoreTransaction $ \db -> do
     let smpServer = SMPServer "smp.simplex.im" "5223" testKeyHash
@@ -339,7 +341,7 @@ testGetRcvConn =
     getRcvConn db smpServer recipientId
       `shouldReturn` Right (rq, SomeConn SCRcv (RcvConnection cData1 rq))
 
-testSetConnUserIdNewConn :: SpecWith SQLiteStore
+testSetConnUserIdNewConn :: SpecWith DBStore
 testSetConnUserIdNewConn =
   it "should set user id for new connection" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -352,9 +354,9 @@ testSetConnUserIdNewConn =
         let ConnData {userId} = connData
         userId `shouldBe` newUserId
       _ -> do
-         expectationFailure "Failed to get connection"
+        expectationFailure "Failed to get connection"
 
-testDeleteRcvConn :: SpecWith SQLiteStore
+testDeleteRcvConn :: SpecWith DBStore
 testDeleteRcvConn =
   it "should create RcvConnection and delete it" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -366,7 +368,7 @@ testDeleteRcvConn =
     getConn db "conn1"
       `shouldReturn` Left SEConnNotFound
 
-testDeleteSndConn :: SpecWith SQLiteStore
+testDeleteSndConn :: SpecWith DBStore
 testDeleteSndConn =
   it "should create SndConnection and delete it" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -378,7 +380,7 @@ testDeleteSndConn =
     getConn db "conn1"
       `shouldReturn` Left SEConnNotFound
 
-testDeleteDuplexConn :: SpecWith SQLiteStore
+testDeleteDuplexConn :: SpecWith DBStore
 testDeleteDuplexConn =
   it "should create DuplexConnection and delete it" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -391,7 +393,7 @@ testDeleteDuplexConn =
     getConn db "conn1"
       `shouldReturn` Left SEConnNotFound
 
-testUpgradeRcvConnToDuplex :: SpecWith SQLiteStore
+testUpgradeRcvConnToDuplex :: SpecWith DBStore
 testUpgradeRcvConnToDuplex =
   it "should throw error on attempt to add SndQueue to SndConnection or DuplexConnection" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -420,7 +422,7 @@ testUpgradeRcvConnToDuplex =
     upgradeRcvConnToDuplex db "conn1" anotherSndQueue
       `shouldReturn` Left (SEBadConnType CDuplex)
 
-testUpgradeSndConnToDuplex :: SpecWith SQLiteStore
+testUpgradeSndConnToDuplex :: SpecWith DBStore
 testUpgradeSndConnToDuplex =
   it "should throw error on attempt to add RcvQueue to RcvConnection or DuplexConnection" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -452,7 +454,7 @@ testUpgradeSndConnToDuplex =
     upgradeSndConnToDuplex db "conn1" anotherRcvQueue
       `shouldReturn` Left (SEBadConnType CDuplex)
 
-testSetRcvQueueStatus :: SpecWith SQLiteStore
+testSetRcvQueueStatus :: SpecWith DBStore
 testSetRcvQueueStatus =
   it "should update status of RcvQueue" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -464,7 +466,7 @@ testSetRcvQueueStatus =
     getConn db "conn1"
       `shouldReturn` Right (SomeConn SCRcv (RcvConnection cData1 rq {status = Confirmed}))
 
-testSetSndQueueStatus :: SpecWith SQLiteStore
+testSetSndQueueStatus :: SpecWith DBStore
 testSetSndQueueStatus =
   it "should update status of SndQueue" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -476,7 +478,7 @@ testSetSndQueueStatus =
     getConn db "conn1"
       `shouldReturn` Right (SomeConn SCSnd (SndConnection cData1 sq {status = Confirmed}))
 
-testSetQueueStatusDuplex :: SpecWith SQLiteStore
+testSetQueueStatusDuplex :: SpecWith DBStore
 testSetQueueStatusDuplex =
   it "should update statuses of RcvQueue and SndQueue in DuplexConnection" . withStoreTransaction $ \db -> do
     g <- C.newRandom
@@ -529,7 +531,7 @@ testCreateRcvMsg_ db expectedPrevSndId expectedPrevHash connId rq rcvMsgData@Rcv
   createRcvMsg db connId rq rcvMsgData
     `shouldReturn` ()
 
-testCreateRcvMsg :: SpecWith SQLiteStore
+testCreateRcvMsg :: SpecWith DBStore
 testCreateRcvMsg =
   it "should reserve internal ids and create a RcvMsg" $ \st -> do
     g <- C.newRandom
@@ -563,7 +565,7 @@ testCreateSndMsg_ db expectedPrevHash connId sq sndMsgData@SndMsgData {..} = do
   createSndMsgDelivery db connId sq internalId
     `shouldReturn` ()
 
-testCreateSndMsg :: SpecWith SQLiteStore
+testCreateSndMsg :: SpecWith DBStore
 testCreateSndMsg =
   it "should create a SndMsg and return InternalId and PrevSndMsgHash" $ \st -> do
     g <- C.newRandom
@@ -574,7 +576,7 @@ testCreateSndMsg =
       testCreateSndMsg_ db "" connId sq $ mkSndMsgData (InternalId 1) (InternalSndId 1) "hash_dummy"
       testCreateSndMsg_ db "hash_dummy" connId sq $ mkSndMsgData (InternalId 2) (InternalSndId 2) "new_hash_dummy"
 
-testCreateRcvAndSndMsgs :: SpecWith SQLiteStore
+testCreateRcvAndSndMsgs :: SpecWith DBStore
 testCreateRcvAndSndMsgs =
   it "should create multiple RcvMsg and SndMsg, correctly ordering internal Ids and returning previous state" $ \st -> do
     let ConnData {connId} = cData1
@@ -592,15 +594,15 @@ testCreateRcvAndSndMsgs =
 
 testCloseReopenStore :: IO ()
 testCloseReopenStore = do
-  st <- createStore
+  st <- createStore'
   hasMigrations st
-  closeSQLiteStore st
-  closeSQLiteStore st
+  closeDBStore st
+  closeDBStore st
   errorGettingMigrations st
   openSQLiteStore st "" False
   openSQLiteStore st "" False
   hasMigrations st
-  closeSQLiteStore st
+  closeDBStore st
   errorGettingMigrations st
   reopenSQLiteStore st
   hasMigrations st
@@ -610,14 +612,14 @@ testCloseReopenEncryptedStore = do
   let key = "test_key"
   st <- createEncryptedStore key False
   hasMigrations st
-  closeSQLiteStore st
-  closeSQLiteStore st
+  closeDBStore st
+  closeDBStore st
   errorGettingMigrations st
   reopenSQLiteStore st `shouldThrow` \(e :: SomeException) -> "reopenSQLiteStore: no key" `isInfixOf` show e
   openSQLiteStore st key True
   openSQLiteStore st key True
   hasMigrations st
-  closeSQLiteStore st
+  closeDBStore st
   errorGettingMigrations st
   reopenSQLiteStore st
   hasMigrations st
@@ -627,21 +629,21 @@ testReopenEncryptedStoreKeepKey = do
   let key = "test_key"
   st <- createEncryptedStore key True
   hasMigrations st
-  closeSQLiteStore st
+  closeDBStore st
   errorGettingMigrations st
   reopenSQLiteStore st
   hasMigrations st
 
-getMigrations :: SQLiteStore -> IO Bool
-getMigrations st = not . null <$> withTransaction st (Migrations.getCurrent . DB.conn)
+getMigrations :: DBStore -> IO Bool
+getMigrations st = not . null <$> withTransaction st Migrations.getCurrent
 
-hasMigrations :: SQLiteStore -> Expectation
+hasMigrations :: DBStore -> Expectation
 hasMigrations st = getMigrations st `shouldReturn` True
 
-errorGettingMigrations :: SQLiteStore -> Expectation
+errorGettingMigrations :: DBStore -> Expectation
 errorGettingMigrations st = getMigrations st `shouldThrow` \(e :: SomeException) -> "ErrorMisuse" `isInfixOf` show e
 
-testGetPendingQueueMsg :: SQLiteStore -> Expectation
+testGetPendingQueueMsg :: DBStore -> Expectation
 testGetPendingQueueMsg st = do
   g <- C.newRandom
   withTransaction st $ \db -> do
@@ -658,7 +660,7 @@ testGetPendingQueueMsg st = do
     Right (Just (Nothing, PendingMsgData {msgId})) <- getPendingQueueMsg db connId sq
     msgId `shouldBe` InternalId 2
 
-testGetPendingServerCommand :: SQLiteStore -> Expectation
+testGetPendingServerCommand :: DBStore -> Expectation
 testGetPendingServerCommand st = do
   g <- C.newRandom
   withTransaction st $ \db -> do
@@ -728,7 +730,7 @@ testFileCbNonce = either error id $ strDecode "dPSF-wrQpDiK_K6sYv0BDBZ9S4dg-jmu"
 testFileReplicaKey :: C.APrivateAuthKey
 testFileReplicaKey = C.APrivateAuthKey C.SEd25519 "MC4CAQAwBQYDK2VwBCIEIDfEfevydXXfKajz3sRkcQ7RPvfWUPoq6pu1TYHV1DEe"
 
-testGetNextRcvChunkToDownload :: SQLiteStore -> Expectation
+testGetNextRcvChunkToDownload :: DBStore -> Expectation
 testGetNextRcvChunkToDownload st = do
   g <- C.newRandom
   withTransaction st $ \db -> do
@@ -745,7 +747,7 @@ testGetNextRcvChunkToDownload st = do
     Right (Just (RcvFileChunk {rcvFileEntityId}, _, Nothing)) <- getNextRcvChunkToDownload db xftpServer1 86400
     rcvFileEntityId `shouldBe` fId2
 
-testGetNextRcvFileToDecrypt :: SQLiteStore -> Expectation
+testGetNextRcvFileToDecrypt :: DBStore -> Expectation
 testGetNextRcvFileToDecrypt st = do
   g <- C.newRandom
   withTransaction st $ \db -> do
@@ -764,7 +766,7 @@ testGetNextRcvFileToDecrypt st = do
     Right (Just RcvFile {rcvFileEntityId}) <- getNextRcvFileToDecrypt db 86400
     rcvFileEntityId `shouldBe` fId2
 
-testGetNextSndFileToPrepare :: SQLiteStore -> Expectation
+testGetNextSndFileToPrepare :: DBStore -> Expectation
 testGetNextSndFileToPrepare st = do
   g <- C.newRandom
   withTransaction st $ \db -> do
@@ -791,7 +793,7 @@ newSndChunkReplica1 =
       rcvIdsKeys = [(ChunkReplicaId $ EntityId "abc", testFileReplicaKey)]
     }
 
-testGetNextSndChunkToUpload :: SQLiteStore -> Expectation
+testGetNextSndChunkToUpload :: DBStore -> Expectation
 testGetNextSndChunkToUpload st = do
   g <- C.newRandom
   withTransaction st $ \db -> do
@@ -814,7 +816,7 @@ testGetNextSndChunkToUpload st = do
     Right (Just SndFileChunk {sndFileEntityId}) <- getNextSndChunkToUpload db xftpServer1 86400
     sndFileEntityId `shouldBe` fId2
 
-testGetNextDeletedSndChunkReplica :: SQLiteStore -> Expectation
+testGetNextDeletedSndChunkReplica :: DBStore -> Expectation
 testGetNextDeletedSndChunkReplica st = do
   withTransaction st $ \db -> do
     Right Nothing <- getNextDeletedSndChunkReplica db xftpServer1 86400
@@ -830,17 +832,17 @@ testGetNextDeletedSndChunkReplica st = do
     Right (Just DeletedSndChunkReplica {deletedSndChunkReplicaId}) <- getNextDeletedSndChunkReplica db xftpServer1 86400
     deletedSndChunkReplicaId `shouldBe` 2
 
-testMarkNtfSubActionNtfFailed :: SQLiteStore -> Expectation
+testMarkNtfSubActionNtfFailed :: DBStore -> Expectation
 testMarkNtfSubActionNtfFailed st = do
   withTransaction st $ \db -> do
     markNtfSubActionNtfFailed_ db "abc"
 
-testMarkNtfSubActionSMPFailed :: SQLiteStore -> Expectation
+testMarkNtfSubActionSMPFailed :: DBStore -> Expectation
 testMarkNtfSubActionSMPFailed st = do
   withTransaction st $ \db -> do
     markNtfSubActionSMPFailed_ db "abc"
 
-testMarkNtfTokenToDeleteFailed :: SQLiteStore -> Expectation
+testMarkNtfTokenToDeleteFailed :: DBStore -> Expectation
 testMarkNtfTokenToDeleteFailed st = do
   withTransaction st $ \db -> do
     markNtfTokenToDeleteFailed_ db 1

tests/AgentTests/SchemaDump.hs

@@ -12,8 +12,8 @@ import Database.SQLite.Simple (Only (..))
 import qualified Database.SQLite.Simple as SQL
 import Simplex.Messaging.Agent.Store.SQLite
 import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
-import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration (..), MigrationsToRun (..), toDownMigration)
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
+import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationsToRun (..), toDownMigration)
 import Simplex.Messaging.Util (ifM)
 import System.Directory (createDirectoryIfMissing, doesFileExist, removeDirectoryRecursive, removeFile)
 import System.Process (readCreateProcess, shell)
@@ -49,7 +49,7 @@ testVerifySchemaDump :: IO ()
 testVerifySchemaDump = do
   savedSchema <- ifM (doesFileExist appSchema) (readFile appSchema) (pure "")
   savedSchema `deepseq` pure ()
-  void $ createSQLiteStore testDB "" False Migrations.app MCConsole
+  void $ createDBStore testDB "" False Migrations.app MCConsole True
   getSchema testDB appSchema `shouldReturn` savedSchema
   removeFile testDB
 
@@ -57,7 +57,7 @@ testVerifyLintFKeyIndexes :: IO ()
 testVerifyLintFKeyIndexes = do
   savedLint <- ifM (doesFileExist appLint) (readFile appLint) (pure "")
   savedLint `deepseq` pure ()
-  void $ createSQLiteStore testDB "" False Migrations.app MCConsole
+  void $ createDBStore testDB "" False Migrations.app MCConsole True
   getLintFKeyIndexes testDB "tests/tmp/agent_lint.sql" `shouldReturn` savedLint
   removeFile testDB
 
@@ -70,9 +70,9 @@ withTmpFiles =
 testSchemaMigrations :: IO ()
 testSchemaMigrations = do
   let noDownMigrations = dropWhileEnd (\Migration {down} -> isJust down) Migrations.app
-  Right st <- createSQLiteStore testDB "" False noDownMigrations MCError
+  Right st <- createDBStore testDB "" False noDownMigrations MCError True
   mapM_ (testDownMigration st) $ drop (length noDownMigrations) Migrations.app
-  closeSQLiteStore st
+  closeDBStore st
   removeFile testDB
   removeFile testSchema
   where
@@ -80,35 +80,35 @@ testSchemaMigrations = do
       putStrLn $ "down migration " <> name m
       let downMigr = fromJust $ toDownMigration m
       schema <- getSchema testDB testSchema
-      Migrations.run st $ MTRUp [m]
+      Migrations.run st True $ MTRUp [m]
       schema' <- getSchema testDB testSchema
       schema' `shouldNotBe` schema
-      Migrations.run st $ MTRDown [downMigr]
+      Migrations.run st True $ MTRDown [downMigr]
       unless (name m `elem` skipComparisonForDownMigrations) $ do
         schema'' <- getSchema testDB testSchema
         schema'' `shouldBe` schema
-      Migrations.run st $ MTRUp [m]
+      Migrations.run st True $ MTRUp [m]
       schema''' <- getSchema testDB testSchema
       schema''' `shouldBe` schema'
 
 testUsersMigrationNew :: IO ()
 testUsersMigrationNew = do
-  Right st <- createSQLiteStore testDB "" False Migrations.app MCError
+  Right st <- createDBStore testDB "" False Migrations.app MCError True
   withTransaction' st (`SQL.query_` "SELECT user_id FROM users;")
     `shouldReturn` ([] :: [Only Int])
-  closeSQLiteStore st
+  closeDBStore st
 
 testUsersMigrationOld :: IO ()
 testUsersMigrationOld = do
   let beforeUsers = takeWhile (("m20230110_users" /=) . name) Migrations.app
-  Right st <- createSQLiteStore testDB "" False beforeUsers MCError
+  Right st <- createDBStore testDB "" False beforeUsers MCError True
   withTransaction' st (`SQL.query_` "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'users';")
     `shouldReturn` ([] :: [Only String])
-  closeSQLiteStore st
-  Right st' <- createSQLiteStore testDB "" False Migrations.app MCYesUp
+  closeDBStore st
+  Right st' <- createDBStore testDB "" False Migrations.app MCYesUp True
   withTransaction' st' (`SQL.query_` "SELECT user_id FROM users;")
     `shouldReturn` ([Only (1 :: Int)])
-  closeSQLiteStore st'
+  closeDBStore st'
 
 skipComparisonForDownMigrations :: [String]
 skipComparisonForDownMigrations =

tests/CoreTests/MsgStoreTests.hs

@@ -112,7 +112,7 @@ testNewQueueRec g sndSecure = do
             senderKey = Nothing,
             sndSecure,
             notifier = Nothing,
-            status = QueueActive,
+            status = EntityActive,
             updatedAt = Nothing
           }
   pure (rId, qr)

tests/CoreTests/StoreLogTests.hs

@@ -10,13 +10,12 @@ module CoreTests.StoreLogTests where
 
 import Control.Concurrent.STM
 import Control.Monad
+import CoreTests.MsgStoreTests
 import Crypto.Random (ChaChaDRG)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (partitionEithers)
 import qualified Data.Map.Strict as M
 import SMPClient
-import AgentTests.SQLiteTests
-import CoreTests.MsgStoreTests
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
@@ -27,6 +26,9 @@ import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.StoreLog
 import Test.Hspec
 
+testPublicAuthKey :: C.APublicAuthKey
+testPublicAuthKey = C.APublicAuthKey C.SEd25519 (C.publicKey "MC4CAQAwBQYDK2VwBCIEIDfEfevydXXfKajz3sRkcQ7RPvfWUPoq6pu1TYHV1DEe")
+
 testNtfCreds :: TVar ChaChaDRG -> IO NtfCreds
 testNtfCreds g = do
   (notifierKey, _) <- atomically $ C.generateAuthKeyPair C.SX25519 g
@@ -54,7 +56,8 @@ storeLogTests =
     ((rId, qr), ntfCreds, date) <- runIO $ do
       g <- C.newRandom
       (,,) <$> testNewQueueRec g sndSecure <*> testNtfCreds g <*> getSystemDate
-    testSMPStoreLog ("SMP server store log, sndSecure = " <> show sndSecure)
+    testSMPStoreLog
+      ("SMP server store log, sndSecure = " <> show sndSecure)
       [ SLTC
           { name = "create new queue",
             saved = [CreateQueue qr],
@@ -66,7 +69,7 @@ storeLogTests =
             saved = [CreateQueue qr, SecureQueue rId testPublicAuthKey],
             compacted = [CreateQueue qr {senderKey = Just testPublicAuthKey}],
             state = M.fromList [(rId, qr {senderKey = Just testPublicAuthKey})]
-          },          
+          },
         SLTC
           { name = "create and delete queue",
             saved = [CreateQueue qr, DeleteQueue rId],
@@ -90,7 +93,7 @@ storeLogTests =
             saved = [CreateQueue qr, UpdateTime rId date],
             compacted = [CreateQueue qr {updatedAt = Just date}],
             state = M.fromList [(rId, qr {updatedAt = Just date})]
-          }          
+          }
       ]
 
 testSMPStoreLog :: String -> [SMPStoreLogTestCase] -> Spec

tests/Fixtures.hs

@@ -0,0 +1,16 @@
+{-# LANGUAGE CPP #-}
+
+module Fixtures where
+
+#if defined(dbPostgres)
+import Database.PostgreSQL.Simple (ConnectInfo (..), defaultConnectInfo)
+#endif
+
+#if defined(dbPostgres)
+testDBConnectInfo :: ConnectInfo
+testDBConnectInfo =
+  defaultConnectInfo {
+    connectUser = "test_agent_user",
+    connectDatabase = "test_agent_db"
+  }
+#endif

tests/SMPAgentClient.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
@@ -25,6 +26,17 @@ import Simplex.Messaging.Protocol (NtfServer, ProtoServerWithAuth (..), Protocol
 import Simplex.Messaging.Transport
 import XFTPClient (testXFTPServer)
 
+-- name fixtures are reused, but they are used as schema name instead of database file path
+#if defined(dbPostgres)
+testDB :: String
+testDB = "smp_agent_test_protocol_schema"
+
+testDB2 :: String
+testDB2 = "smp_agent2_test_protocol_schema"
+
+testDB3 :: String
+testDB3 = "smp_agent3_test_protocol_schema"
+#else
 testDB :: FilePath
 testDB = "tests/tmp/smp-agent.test.protocol.db"
 
@@ -33,6 +45,7 @@ testDB2 = "tests/tmp/smp-agent2.test.protocol.db"
 
 testDB3 :: FilePath
 testDB3 = "tests/tmp/smp-agent3.test.protocol.db"
+#endif
 
 testSMPServer :: SMPServer
 testSMPServer = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"

tests/SMPClient.hs

@@ -79,6 +79,9 @@ testStoreNtfsFile = "tests/tmp/smp-server-ntfs.log"
 testStoreNtfsFile2 :: FilePath
 testStoreNtfsFile2 = "tests/tmp/smp-server-ntfs.log.2"
 
+testPrometheusMetricsFile :: FilePath
+testPrometheusMetricsFile = "tests/tmp/smp-server-metrics.txt"
+
 testServerStatsBackupFile :: FilePath
 testServerStatsBackupFile = "tests/tmp/smp-server-stats.log"
 
@@ -141,8 +144,10 @@ cfgMS msType =
       inactiveClientExpiration = Just defaultInactiveClientExpiration,
       logStatsInterval = Nothing,
       logStatsStartTime = 0,
-      serverStatsLogFile = "tests/smp-server-stats.daily.log",
+      serverStatsLogFile = "tests/tmp/smp-server-stats.daily.log",
       serverStatsBackupFile = Nothing,
+      prometheusInterval = Nothing,
+      prometheusMetricsFile = testPrometheusMetricsFile,
       pendingENDInterval = 500000,
       ntfDeliveryInterval = 200000,
       smpCredentials =

tests/SMPProxyTests.hs

@@ -1,4 +1,5 @@
 {-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
@@ -46,6 +47,10 @@ import System.Random (randomRIO)
 import Test.Hspec
 import UnliftIO
 import Util
+#if defined(dbPostgres)
+import Fixtures
+import Simplex.Messaging.Agent.Store.Postgres.Util (dropAllSchemasExceptSystem)
+#endif
 
 smpProxyTests :: Spec
 smpProxyTests = do
@@ -101,7 +106,11 @@ smpProxyTests = do
         it "100x100 N4 C16" . twoServersMoreConc $ withNumCapabilities 4 $ 100 `inParrallel` deliver 100
         it "100x100 N" . twoServersFirstProxy $ withNCPUCapabilities $ 100 `inParrallel` deliver 100
         it "500x20" . twoServersFirstProxy $ 500 `inParrallel` deliver 20
+#if defined(dbPostgres)
+    after_ (dropAllSchemasExceptSystem testDBConnectInfo) . describe "agent API" $ do
+#else
     describe "agent API" $ do
+#endif
       describe "one server" $ do
         it "always via proxy" . oneServer $
           agentDeliverMessageViaProxy ([srv1], SPMAlways, True) ([srv1], SPMAlways, True) C.SEd448 "hello 1" "hello 2" 1

tests/ServerTests.hs

@@ -15,13 +15,12 @@
 
 module ServerTests where
 
-import AgentTests.NotificationTests (removeFileIfExists)
-import CoreTests.MsgStoreTests (testJournalStoreCfg)
 import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (SomeException, try)
 import Control.Monad
 import Control.Monad.IO.Class
+import CoreTests.MsgStoreTests (testJournalStoreCfg)
 import Data.Bifunctor (first)
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
@@ -42,18 +41,20 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..))
 import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore)
 import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..))
-import Simplex.Messaging.Server.StoreLog (closeStoreLog)
+import Simplex.Messaging.Server.StoreLog (StoreLogRecord (..), closeStoreLog)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Util (whenM)
 import Simplex.Messaging.Version (mkVersionRange)
-import System.Directory (doesDirectoryExist, removeDirectoryRecursive, removeFile)
+import System.Directory (doesDirectoryExist, doesFileExist, removeDirectoryRecursive, removeFile)
+import System.IO (IOMode (..), withFile)
 import System.TimeIt (timeItT)
 import System.Timeout
 import Test.HUnit
 import Test.Hspec
+import Util (removeFileIfExists)
 
 serverTests :: SpecWith (ATransport, AMSType)
-serverTests  = do
+serverTests = do
   describe "SMP queues" $ do
     describe "NEW and KEY commands, SEND messages" testCreateSecure
     describe "NEW and SKEY commands" $ do
@@ -71,12 +72,14 @@ serverTests  = do
   describe "Store log" testWithStoreLog
   describe "Restore messages" testRestoreMessages
   describe "Restore messages (old / v2)" testRestoreExpireMessages
+  describe "Save prometheus metrics" testPrometheusMetrics
   describe "Timing of AUTH error" testTiming
   describe "Message notifications" testMessageNotifications
   describe "Message expiration" $ do
     testMsgExpireOnSend
     testMsgExpireOnInterval
     testMsgNOTExpireOnInterval
+  describe "Blocking queues" $ testBlockMessageQueue
 
 pattern Resp :: CorrId -> QueueId -> BrokerMsg -> SignedTransmission ErrorType BrokerMsg
 pattern Resp corrId queueId command <- (_, _, (corrId, queueId, Right command))
@@ -687,7 +690,7 @@ testRestoreMessages =
 
     logSize testStoreLogFile `shouldReturn` 2
     -- logSize testStoreMsgsFile `shouldReturn` 5
-    logSize testServerStatsBackupFile `shouldReturn` 74
+    logSize testServerStatsBackupFile `shouldReturn` 76
     Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats1 [rId] 5 1
 
@@ -705,7 +708,7 @@ testRestoreMessages =
     logSize testStoreLogFile `shouldReturn` 1
     -- the last message is not removed because it was not ACK'd
     -- logSize testStoreMsgsFile `shouldReturn` 3
-    logSize testServerStatsBackupFile `shouldReturn` 74
+    logSize testServerStatsBackupFile `shouldReturn` 76
     Right stats2 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats2 [rId] 5 3
 
@@ -723,7 +726,7 @@ testRestoreMessages =
       pure ()
     logSize testStoreLogFile `shouldReturn` 1
     -- logSize testStoreMsgsFile `shouldReturn` 0
-    logSize testServerStatsBackupFile `shouldReturn` 74
+    logSize testServerStatsBackupFile `shouldReturn` 76
     Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats3 [rId] 5 5
 
@@ -825,6 +828,13 @@ testRestoreExpireMessages =
     runClient :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> Expectation
     runClient _ test' = testSMPClient test' `shouldReturn` ()
 
+testPrometheusMetrics :: SpecWith (ATransport, AMSType)
+testPrometheusMetrics =
+  it "should save Prometheus metrics" $ \(at, msType) -> do
+    let cfg' = (cfgMS msType) {prometheusInterval = Just 1}
+    withSmpServerConfigOn at cfg' testPort $ \_ -> threadDelay 1000000
+    doesFileExist testPrometheusMetricsFile `shouldReturn` True
+
 createAndSecureQueue :: Transport c => THandleSMP c 'TClient -> SndPublicAuthKey -> IO (SenderId, RecipientId, RcvPrivateAuthKey, RcvDhSecret)
 createAndSecureQueue h sPub = do
   g <- C.newRandom
@@ -988,7 +998,7 @@ testMsgExpireOnInterval =
 
 testMsgNOTExpireOnInterval :: SpecWith (ATransport, AMSType)
 testMsgNOTExpireOnInterval =
-  it "should NOT expire messages that are not received before messageTTL if expiry interval is large" $ \(ATransport (t :: TProxy c), msType) -> do
+  it "should block and unblock message queues" $ \(ATransport (t :: TProxy c), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
@@ -1005,6 +1015,30 @@ testMsgNOTExpireOnInterval =
             Nothing -> return ()
             Just _ -> error "nothing else should be delivered"
 
+testBlockMessageQueue :: SpecWith (ATransport, AMSType)
+testBlockMessageQueue =
+  it "should return BLOCKED error when queue is blocked" $ \(at@(ATransport (t :: TProxy c)), msType) -> do
+    g <- C.newRandom
+    (rId, sId) <- withSmpServerStoreLogOnMS at msType testPort $ runTest t $ \h -> do
+      (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
+      (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
+      Resp "abcd" rId1 (Ids rId sId _srvDh) <- signSendRecv h rKey ("abcd", NoEntity, NEW rPub dhPub Nothing SMSubscribe True)
+      (rId1, NoEntity) #== "creates queue"
+      pure (rId, sId)
+
+    withFile testStoreLogFile AppendMode $ \h -> B.hPutStrLn h $ strEncode $ BlockQueue rId $ BlockingInfo BRContent
+
+    withSmpServerStoreLogOnMS at msType testPort $ runTest t $ \h -> do
+      (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
+      Resp "dabc" sId2 (ERR (BLOCKED (BlockingInfo BRContent))) <- signSendRecv h sKey ("dabc", sId, SKEY sPub)
+      (sId2, sId) #== "same queue ID in response"
+  where
+    runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO a) -> ThreadId -> IO a
+    runTest _ test' server = do
+      a <- testSMPClient test'
+      killThread server
+      pure a
+
 samplePubKey :: C.APublicVerifyKey
 samplePubKey = C.APublicVerifyKey C.SEd25519 "MCowBQYDK2VwAyEAfAOflyvbJv1fszgzkQ6buiZJVgSpQWsucXq7U6zjMgY="
 

tests/Test.hs

@@ -1,8 +1,8 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE TypeApplications #-}
 
 import AgentTests (agentTests)
-import AgentTests.SchemaDump (schemaDumpTest)
 import CLITests
 import Control.Concurrent (threadDelay)
 import qualified Control.Exception as E
@@ -34,6 +34,12 @@ import Test.Hspec
 import XFTPAgent
 import XFTPCLI
 import XFTPServerTests (xftpServerTests)
+#if defined(dbPostgres)
+import Fixtures
+import Simplex.Messaging.Agent.Store.Postgres.Util (createDBAndUserIfNotExists, dropDatabaseAndUser)
+#else
+import AgentTests.SchemaDump (schemaDumpTest)
+#endif
 
 logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
@@ -45,10 +51,17 @@ main = do
     setEnv "APNS_KEY_ID" "H82WD9K9AQ"
     setEnv "APNS_KEY_FILE" "./tests/fixtures/AuthKey_H82WD9K9AQ.p8"
     hspec
+#if defined(dbPostgres)
+      . beforeAll_ (dropDatabaseAndUser testDBConnectInfo >> createDBAndUserIfNotExists testDBConnectInfo)
+      . afterAll_ (dropDatabaseAndUser testDBConnectInfo)
+#endif
       . before_ (createDirectoryIfMissing False "tests/tmp")
       . after_ (eventuallyRemove "tests/tmp" 3)
       $ do
+-- TODO [postgres] schema dump for postgres
+#if !defined(dbPostgres)
         describe "Agent SQLite schema dump" schemaDumpTest
+#endif
         describe "Core tests" $ do
           describe "Batching tests" batchingTests
           describe "Encoding tests" encodingTests

tests/Util.hs

@@ -1,9 +1,10 @@
 module Util where
 
-import Control.Monad (replicateM)
+import Control.Monad (replicateM, when)
 import Data.Either (partitionEithers)
 import Data.List (tails)
 import GHC.Conc (getNumCapabilities, getNumProcessors, setNumCapabilities)
+import System.Directory (doesFileExist, removeFile)
 import Test.Hspec
 import UnliftIO
 
@@ -26,3 +27,8 @@ inParrallel n action = do
 combinations :: Int -> [a] -> [[a]]
 combinations 0 _ = [[]]
 combinations k xs = [y : ys | y : xs' <- tails xs, ys <- combinations (k - 1) xs']
+
+removeFileIfExists :: FilePath -> IO ()
+removeFileIfExists filePath = do
+  fileExists <- doesFileExist filePath
+  when fileExists $ removeFile filePath

tests/XFTPAgent.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
@@ -46,40 +47,49 @@ import UnliftIO
 import UnliftIO.Concurrent
 import XFTPCLI
 import XFTPClient
+#if defined(dbPostgres)
+import Fixtures
+import Simplex.Messaging.Agent.Store.Postgres.Util (dropAllSchemasExceptSystem)
+#endif
 
 xftpAgentTests :: Spec
-xftpAgentTests = around_ testBracket . describe "agent XFTP API" $ do
-  it "should send and receive file" $ withXFTPServer testXFTPAgentSendReceive
-  -- uncomment CPP option slow_servers and run hpack to run this test
-  xit "should send and receive file with slow server responses" $
-    withXFTPServerCfg testXFTPServerConfig {responseDelay = 500000} $
-      \_ -> testXFTPAgentSendReceive
-  it "should send and receive with encrypted local files" testXFTPAgentSendReceiveEncrypted
-  it "should send and receive large file with a redirect" testXFTPAgentSendReceiveRedirect
-  it "should send and receive small file without a redirect" testXFTPAgentSendReceiveNoRedirect
-  describe "sending and receiving with version negotiation" testXFTPAgentSendReceiveMatrix
-  it "should resume receiving file after restart" testXFTPAgentReceiveRestore
-  it "should cleanup rcv tmp path after permanent error" testXFTPAgentReceiveCleanup
-  it "should resume sending file after restart" testXFTPAgentSendRestore
-  xit'' "should cleanup snd prefix path after permanent error" testXFTPAgentSendCleanup
-  it "should delete sent file on server" testXFTPAgentDelete
-  it "should resume deleting file after restart" testXFTPAgentDeleteRestore
-  -- TODO when server is fixed to correctly send AUTH error, this test has to be modified to expect AUTH error
-  it "if file is deleted on server, should limit retries and continue receiving next file" testXFTPAgentDeleteOnServer
-  it "if file is expired on server, should report error and continue receiving next file" testXFTPAgentExpiredOnServer
-  it "should request additional recipient IDs when number of recipients exceeds maximum per request" testXFTPAgentRequestAdditionalRecipientIDs
-  describe "XFTP server test via agent API" $ do
-    it "should pass without basic auth" $ testXFTPServerTest Nothing (noAuthSrv testXFTPServer2) `shouldReturn` Nothing
-    let srv1 = testXFTPServer2 {keyHash = "1234"}
-    it "should fail with incorrect fingerprint" $ do
-      testXFTPServerTest Nothing (noAuthSrv srv1) `shouldReturn` Just (ProtocolTestFailure TSConnect $ BROKER (B.unpack $ strEncode srv1) NETWORK)
-    describe "server with password" $ do
-      let auth = Just "abcd"
-          srv = ProtoServerWithAuth testXFTPServer2
-          authErr = Just (ProtocolTestFailure TSCreateFile $ XFTP (B.unpack $ strEncode testXFTPServer2) AUTH)
-      it "should pass with correct password" $ testXFTPServerTest auth (srv auth) `shouldReturn` Nothing
-      it "should fail without password" $ testXFTPServerTest auth (srv Nothing) `shouldReturn` authErr
-      it "should fail with incorrect password" $ testXFTPServerTest auth (srv $ Just "wrong") `shouldReturn` authErr
+xftpAgentTests =
+  around_ testBracket
+#if defined(dbPostgres)
+    . after_ (dropAllSchemasExceptSystem testDBConnectInfo)
+#endif
+    . describe "agent XFTP API" $ do
+      it "should send and receive file" $ withXFTPServer testXFTPAgentSendReceive
+      -- uncomment CPP option slow_servers and run hpack to run this test
+      xit "should send and receive file with slow server responses" $
+        withXFTPServerCfg testXFTPServerConfig {responseDelay = 500000} $
+          \_ -> testXFTPAgentSendReceive
+      it "should send and receive with encrypted local files" testXFTPAgentSendReceiveEncrypted
+      it "should send and receive large file with a redirect" testXFTPAgentSendReceiveRedirect
+      it "should send and receive small file without a redirect" testXFTPAgentSendReceiveNoRedirect
+      describe "sending and receiving with version negotiation" testXFTPAgentSendReceiveMatrix
+      it "should resume receiving file after restart" testXFTPAgentReceiveRestore
+      it "should cleanup rcv tmp path after permanent error" testXFTPAgentReceiveCleanup
+      it "should resume sending file after restart" testXFTPAgentSendRestore
+      xit'' "should cleanup snd prefix path after permanent error" testXFTPAgentSendCleanup
+      it "should delete sent file on server" testXFTPAgentDelete
+      it "should resume deleting file after restart" testXFTPAgentDeleteRestore
+      -- TODO when server is fixed to correctly send AUTH error, this test has to be modified to expect AUTH error
+      it "if file is deleted on server, should limit retries and continue receiving next file" testXFTPAgentDeleteOnServer
+      it "if file is expired on server, should report error and continue receiving next file" testXFTPAgentExpiredOnServer
+      it "should request additional recipient IDs when number of recipients exceeds maximum per request" testXFTPAgentRequestAdditionalRecipientIDs
+      describe "XFTP server test via agent API" $ do
+        it "should pass without basic auth" $ testXFTPServerTest Nothing (noAuthSrv testXFTPServer2) `shouldReturn` Nothing
+        let srv1 = testXFTPServer2 {keyHash = "1234"}
+        it "should fail with incorrect fingerprint" $ do
+          testXFTPServerTest Nothing (noAuthSrv srv1) `shouldReturn` Just (ProtocolTestFailure TSConnect $ BROKER (B.unpack $ strEncode srv1) NETWORK)
+        describe "server with password" $ do
+          let auth = Just "abcd"
+              srv = ProtoServerWithAuth testXFTPServer2
+              authErr = Just (ProtocolTestFailure TSCreateFile $ XFTP (B.unpack $ strEncode testXFTPServer2) AUTH)
+          it "should pass with correct password" $ testXFTPServerTest auth (srv auth) `shouldReturn` Nothing
+          it "should fail without password" $ testXFTPServerTest auth (srv Nothing) `shouldReturn` authErr
+          it "should fail with incorrect password" $ testXFTPServerTest auth (srv $ Just "wrong") `shouldReturn` authErr
 
 rfProgress :: forall m. (HasCallStack, MonadIO m, MonadFail m) => AgentClient -> Int64 -> m ()
 rfProgress c expected = loop 0

tests/XFTPServerTests.hs

@@ -289,7 +289,7 @@ testFileLog = do
     download g c rpKey1 rId1 digest bytes
     download g c rpKey2 rId2 digest bytes
   logSize testXFTPLogFile `shouldReturn` 3
-  logSize testXFTPStatsBackupFile `shouldReturn` 14
+  logSize testXFTPStatsBackupFile `shouldReturn` 15
 
   threadDelay 100000
 
@@ -316,7 +316,7 @@ testFileLog = do
     -- recipient 2 can download
     download g c rpKey2 rId2 digest bytes
   logSize testXFTPLogFile `shouldReturn` 4
-  logSize testXFTPStatsBackupFile `shouldReturn` 14
+  logSize testXFTPStatsBackupFile `shouldReturn` 15
 
   threadDelay 100000
 
@@ -337,13 +337,13 @@ testFileLog = do
     -- sender can delete - +1 to log
     deleteXFTPChunk c spKey sId
   logSize testXFTPLogFile `shouldReturn` 4
-  logSize testXFTPStatsBackupFile `shouldReturn` 14
+  logSize testXFTPStatsBackupFile `shouldReturn` 15
 
   threadDelay 100000
 
   withXFTPServerStoreLogOn $ \_ -> pure () -- compacts on start
   logSize testXFTPLogFile `shouldReturn` 0
-  logSize testXFTPStatsBackupFile `shouldReturn` 14
+  logSize testXFTPStatsBackupFile `shouldReturn` 15
 
   threadDelay 100000