Merge pull request #1524 from simplex-chat/ntf-storage

ntf server: PostgreSQL database storage (feature branch)

Author: epoberezkin

apps/ntf-server/Main.hs

@@ -15,7 +15,6 @@ logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
 
 main :: IO ()
 main = do
-  setLogLevel LogInfo
   cfgPath <- getEnvPath "NTF_SERVER_CFG_PATH" defaultCfgPath
   logPath <- getEnvPath "NTF_SERVER_LOG_PATH" defaultLogPath
   withGlobalLogging logCfg $ ntfServerCLI cfgPath logPath

simplexmq.cabal

@@ -216,15 +216,6 @@ library
           Simplex.FileTransfer.Server.Stats
           Simplex.FileTransfer.Server.Store
           Simplex.FileTransfer.Server.StoreLog
-          Simplex.Messaging.Notifications.Server
-          Simplex.Messaging.Notifications.Server.Control
-          Simplex.Messaging.Notifications.Server.Env
-          Simplex.Messaging.Notifications.Server.Main
-          Simplex.Messaging.Notifications.Server.Push.APNS
-          Simplex.Messaging.Notifications.Server.Push.APNS.Internal
-          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
@@ -257,6 +248,19 @@ library
 
       if flag(server_postgres)
           exposed-modules:
+              Simplex.Messaging.Notifications.Server
+              Simplex.Messaging.Notifications.Server.Control
+              Simplex.Messaging.Notifications.Server.Env
+              Simplex.Messaging.Notifications.Server.Main
+              Simplex.Messaging.Notifications.Server.Prometheus
+              Simplex.Messaging.Notifications.Server.Push.APNS
+              Simplex.Messaging.Notifications.Server.Push.APNS.Internal
+              Simplex.Messaging.Notifications.Server.Stats
+              Simplex.Messaging.Notifications.Server.Store
+              Simplex.Messaging.Notifications.Server.Store.Migrations
+              Simplex.Messaging.Notifications.Server.Store.Postgres
+              Simplex.Messaging.Notifications.Server.Store.Types
+              Simplex.Messaging.Notifications.Server.StoreLog
               Simplex.Messaging.Server.QueueStore.Postgres
               Simplex.Messaging.Server.QueueStore.Postgres.Migrations
   other-modules:
@@ -340,6 +344,8 @@ library
       , sqlcipher-simple ==0.4.*
   if flag(server_postgres)
     cpp-options: -DdbServerPostgres
+    build-depends:
+        hex-text ==0.1.*
   if impl(ghc >= 9.6.2)
     build-depends:
         bytestring ==0.11.*
@@ -352,6 +358,10 @@ library
 executable ntf-server
   if flag(client_library)
       buildable: False
+  if flag(server_postgres)
+    cpp-options: -DdbServerPostgres
+  else
+    buildable: False
   main-is: Main.hs
   other-modules:
       Paths_simplexmq
@@ -444,7 +454,6 @@ test-suite simplexmq-test
       AgentTests.EqInstances
       AgentTests.FunctionalAPITests
       AgentTests.MigrationTests
-      AgentTests.NotificationTests
       AgentTests.ServerChoice
       AgentTests.ShortLinkTests
       CLITests
@@ -460,8 +469,6 @@ test-suite simplexmq-test
       CoreTests.UtilTests
       CoreTests.VersionRangeTests
       FileDescriptionTests
-      NtfClient
-      NtfServerTests
       RemoteControl
       ServerTests
       SMPAgentClient
@@ -484,7 +491,10 @@ test-suite simplexmq-test
           AgentTests.SQLiteTests
   if flag(server_postgres)
       other-modules:
-          ServerTests.SchemaDump  
+          AgentTests.NotificationTests
+          NtfClient
+          NtfServerTests
+          PostgresSchemaDump
   hs-source-dirs:
       tests
       apps/smp-server/web
@@ -537,6 +547,8 @@ test-suite simplexmq-test
     , warp-tls
     , yaml
   default-language: Haskell2010
+  if flag(server_postgres)
+    cpp-options: -DdbServerPostgres
   if flag(client_postgres)
     cpp-options: -DdbPostgres
   else
@@ -550,5 +562,3 @@ test-suite simplexmq-test
   if flag(client_postgres) || flag(server_postgres)
     build-depends:
         postgresql-simple ==0.7.*
-  if flag(server_postgres)
-    cpp-options: -DdbServerPostgres

src/Simplex/FileTransfer/Client/Main.hs

@@ -280,7 +280,7 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
       let chunkSpecs = prepareChunkSpecs encPath chunkSizes
           fdRcv = FileDescription {party = SFRecipient, size = FileSize encSize, digest = FileDigest digest, key, nonce, chunkSize = FileSize defChunkSize, chunks = [], redirect = Nothing}
           fdSnd = FileDescription {party = SFSender, size = FileSize encSize, digest = FileDigest digest, key, nonce, chunkSize = FileSize defChunkSize, chunks = [], redirect = Nothing}
-      logInfo $ "encrypted file to " <> tshow encPath
+      logDebug $ "encrypted file to " <> tshow encPath
       pure (encPath, fdRcv, fdSnd, chunkSpecs, encSize)
     uploadFile :: TVar ChaChaDRG -> [XFTPChunkSpec] -> TVar [Int64] -> Int64 -> ExceptT CLIError IO [SentFileChunk]
     uploadFile g chunks uploadedChunks encSize = do
@@ -293,22 +293,22 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
       -- TODO shuffle/unshuffle chunks
       -- the reason we don't do pooled downloads here within one server is that http2 library doesn't handle cleint concurrency, even though
       -- upload doesn't allow other requests within the same client until complete (but download does allow).
-      logInfo $ "uploading " <> tshow (length chunks) <> " chunks..."
+      logDebug $ "uploading " <> tshow (length chunks) <> " chunks..."
       (errs, rs) <- partitionEithers . concat <$> liftIO (pooledForConcurrentlyN 16 chunks' . mapM $ runExceptT . uploadFileChunk a)
       mapM_ throwE errs
       pure $ map snd (sortOn fst rs)
       where
         uploadFileChunk :: XFTPClientAgent -> (Int, XFTPChunkSpec, XFTPServerWithAuth) -> ExceptT CLIError IO (Int, SentFileChunk)
         uploadFileChunk a (chunkNo, chunkSpec@XFTPChunkSpec {chunkSize}, ProtoServerWithAuth xftpServer auth) = do
-          logInfo $ "uploading chunk " <> tshow chunkNo <> " to " <> showServer xftpServer <> "..."
+          logDebug $ "uploading chunk " <> tshow chunkNo <> " to " <> showServer xftpServer <> "..."
           (sndKey, spKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
           rKeys <- atomically $ L.fromList <$> replicateM numRecipients (C.generateAuthKeyPair C.SEd25519 g)
           digest <- liftIO $ getChunkDigest chunkSpec
           let ch = FileInfo {sndKey, size = chunkSize, digest}
           c <- withRetry retryCount $ getXFTPServerClient a xftpServer
           (sndId, rIds) <- withRetry retryCount $ createXFTPChunk c spKey ch (L.map fst rKeys) auth
           withReconnect a xftpServer retryCount $ \c' -> uploadXFTPChunk c' spKey sndId chunkSpec
-          logInfo $ "uploaded chunk " <> tshow chunkNo
+          logDebug $ "uploaded chunk " <> tshow chunkNo
           uploaded <- atomically . stateTVar uploadedChunks $ \cs ->
             let cs' = fromIntegral chunkSize : cs in (sum cs', cs')
           liftIO $ do
@@ -418,11 +418,11 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath,
     downloadFileChunk :: TVar ChaChaDRG -> XFTPClientAgent -> FilePath -> FileSize Int64 -> TVar [Int64] -> FileChunk -> ExceptT CLIError IO (Int, FilePath)
     downloadFileChunk g a encPath (FileSize encSize) downloadedChunks FileChunk {chunkNo, chunkSize, digest, replicas = replica : _} = do
       let FileChunkReplica {server, replicaId, replicaKey} = replica
-      logInfo $ "downloading chunk " <> tshow chunkNo <> " from " <> showServer server <> "..."
+      logDebug $ "downloading chunk " <> tshow chunkNo <> " from " <> showServer server <> "..."
       chunkPath <- uniqueCombine encPath $ show chunkNo
       let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest)
       withReconnect a server retryCount $ \c -> downloadXFTPChunk g c replicaKey (unChunkReplicaId replicaId) chunkSpec
-      logInfo $ "downloaded chunk " <> tshow chunkNo <> " to " <> T.pack chunkPath
+      logDebug $ "downloaded chunk " <> tshow chunkNo <> " to " <> T.pack chunkPath
       downloaded <- atomically . stateTVar downloadedChunks $ \cs ->
         let cs' = fromIntegral (unFileSize chunkSize) : cs in (sum cs', cs')
       liftIO $ do
@@ -467,7 +467,7 @@ cliDeleteFile DeleteOptions {fileDescription, retryCount, yes} = do
     deleteFileChunk a FileChunk {chunkNo, replicas = replica : _} = do
       let FileChunkReplica {server, replicaId, replicaKey} = replica
       withReconnect a server retryCount $ \c -> deleteXFTPChunk c replicaKey (unChunkReplicaId replicaId)
-      logInfo $ "deleted chunk " <> tshow chunkNo <> " from " <> showServer server
+      logDebug $ "deleted chunk " <> tshow chunkNo <> " from " <> showServer server
     deleteFileChunk _ _ = throwE $ CLIError "chunk has no replicas"
 
 cliFileDescrInfo :: InfoOptions -> ExceptT CLIError IO ()

src/Simplex/FileTransfer/Server.hs

@@ -181,7 +181,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
     stopServer = do
       withFileLog closeStoreLog
       saveServerStats
-      logInfo "Server stopped"
+      logNote "Server stopped"
 
     expireFilesThread_ :: XFTPServerConfig -> [M ()]
     expireFilesThread_ XFTPServerConfig {fileExpiration = Just fileExp} = [expireFiles fileExp]
@@ -560,13 +560,13 @@ expireServerFiles itemDelay expCfg = do
   usedStart <- readTVarIO $ usedStorage st
   old <- liftIO $ expireBeforeEpoch expCfg
   files' <- readTVarIO (files st)
-  logInfo $ "Expiration check: " <> tshow (M.size files') <> " files"
+  logNote $ "Expiration check: " <> tshow (M.size files') <> " files"
   forM_ (M.keys files') $ \sId -> do
     mapM_ threadDelay itemDelay
     atomically (expiredFilePath st sId old)
       >>= mapM_ (maybeRemove $ delete st sId)
   usedEnd <- readTVarIO $ usedStorage st
-  logInfo $ "Used " <> mbs usedStart <> " -> " <> mbs usedEnd <> ", " <> mbs (usedStart - usedEnd) <> " reclaimed."
+  logNote $ "Used " <> mbs usedStart <> " -> " <> mbs usedEnd <> ", " <> mbs (usedStart - usedEnd) <> " reclaimed."
   where
     mbs bs = tshow (bs `div` 1048576) <> "mb"
     maybeRemove del = maybe del (remove del)
@@ -600,15 +600,15 @@ saveServerStats =
     >>= mapM_ (\f -> asks serverStats >>= liftIO . getFileServerStatsData >>= liftIO . saveStats f)
   where
     saveStats f stats = do
-      logInfo $ "saving server stats to file " <> T.pack f
+      logNote $ "saving server stats to file " <> T.pack f
       B.writeFile f $ strEncode stats
-      logInfo "server stats saved"
+      logNote "server stats saved"
 
 restoreServerStats :: M ()
 restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats
   where
     restoreStats f = whenM (doesFileExist f) $ do
-      logInfo $ "restoring server stats from file " <> T.pack f
+      logNote $ "restoring server stats from file " <> T.pack f
       liftIO (strDecode <$> B.readFile f) >>= \case
         Right d@FileServerStatsData {_filesCount = statsFilesCount, _filesSize = statsFilesSize} -> do
           s <- asks serverStats
@@ -617,10 +617,10 @@ restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStat
           _filesSize <- readTVarIO usedStorage
           liftIO $ setFileServerStats s d {_filesCount, _filesSize}
           renameFile f $ f <> ".bak"
-          logInfo "server stats restored"
+          logNote "server stats restored"
           when (statsFilesCount /= _filesCount) $ logWarn $ "Files count differs: stats: " <> tshow statsFilesCount <> ", store: " <> tshow _filesCount
           when (statsFilesSize /= _filesSize) $ logWarn $ "Files size differs: stats: " <> tshow statsFilesSize <> ", store: " <> tshow _filesSize
-          logInfo $ "Restored " <> tshow (_filesSize `div` 1048576) <> " MBs in " <> tshow _filesCount <> " files"
+          logNote $ "Restored " <> tshow (_filesSize `div` 1048576) <> " MBs in " <> tshow _filesCount <> " files"
         Left e -> do
-          logInfo $ "error restoring server stats: " <> T.pack e
+          logNote $ "error restoring server stats: " <> T.pack e
           liftIO exitFailure

src/Simplex/FileTransfer/Server/Env.hs

@@ -103,8 +103,8 @@ newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, xftpCrede
   used <- countUsedStorage <$> readTVarIO (files store)
   atomically $ writeTVar (usedStorage store) used
   forM_ fileSizeQuota $ \quota -> do
-    logInfo $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used)
-    when (quota < used) $ logInfo "WARNING: storage quota is less than used storage, no files can be uploaded!"
+    logNote $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used)
+    when (quota < used) $ logWarn "WARNING: storage quota is less than used storage, no files can be uploaded!"
   tlsServerCreds <- loadServerCredential xftpCredentials
   Fingerprint fp <- loadFingerprint xftpCredentials
   serverStats <- newFileServerStats =<< getCurrentTime

src/Simplex/Messaging/Agent.hs

@@ -2200,10 +2200,9 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
               atomically $ nsUpdateToken ns tkn'
               agentNtfCheckToken c tknId tkn' >>= \case
                 NTActive -> do
-                  cron <- asks $ ntfCron . config
-                  agentNtfEnableCron c tknId tkn cron
                   when (suppliedNtfMode == NMInstant) $ initializeNtfSubs c
                   when (suppliedNtfMode == NMPeriodic && savedNtfMode == NMInstant) $ deleteNtfSubs c NSCSmpDelete
+                  lift $ setCronInterval c tknId tkn
                   t tkn' (NTActive, Just NTACheck) $ pure ()
                 status -> t tkn' (status, Nothing) $ pure ()
           | otherwise -> replaceToken tknId
@@ -2264,11 +2263,15 @@ verifyNtfToken' c deviceToken nonce code =
         withToken c tkn (Just (NTConfirmed, NTAVerify code')) (NTActive, Just NTACheck) $
           agentNtfVerifyToken c tknId tkn code'
       when (toStatus == NTActive) $ do
-        cron <- asks $ ntfCron . config
-        agentNtfEnableCron c tknId tkn cron
+        lift $ setCronInterval c tknId tkn
         when (ntfMode == NMInstant) $ initializeNtfSubs c
     _ -> throwE $ CMD PROHIBITED "verifyNtfToken: no token"
 
+setCronInterval :: AgentClient -> NtfTokenId -> NtfToken -> AM' ()
+setCronInterval c tknId tkn = do
+  cron <- asks $ ntfCron . config
+  void $ forkIO $ void $ runExceptT $ agentNtfSetCronInterval c tknId tkn cron
+
 checkNtfToken' :: AgentClient -> DeviceToken -> AM NtfTknStatus
 checkNtfToken' c deviceToken =
   withStore' c getSavedNtfToken >>= \case

src/Simplex/Messaging/Agent/Client.hs

@@ -77,7 +77,7 @@ module Simplex.Messaging.Agent.Client
     agentNtfCheckToken,
     agentNtfReplaceToken,
     agentNtfDeleteToken,
-    agentNtfEnableCron,
+    agentNtfSetCronInterval,
     agentNtfCreateSubscription,
     agentNtfCreateSubscriptions,
     agentNtfCheckSubscription,
@@ -1817,9 +1817,10 @@ agentNtfDeleteToken :: AgentClient -> NtfServer -> C.APrivateAuthKey -> NtfToken
 agentNtfDeleteToken c ntfServer ntfPrivKey tknId =
   withNtfClient c ntfServer tknId "TDEL" $ \ntf -> ntfDeleteToken ntf ntfPrivKey tknId
 
-agentNtfEnableCron :: AgentClient -> NtfTokenId -> NtfToken -> Word16 -> AM ()
-agentNtfEnableCron c tknId NtfToken {ntfServer, ntfPrivKey} interval =
-  withNtfClient c ntfServer tknId "TCRN" $ \ntf -> ntfEnableCron ntf ntfPrivKey tknId interval
+-- set to 0 to disable
+agentNtfSetCronInterval :: AgentClient -> NtfTokenId -> NtfToken -> Word16 -> AM ()
+agentNtfSetCronInterval c tknId NtfToken {ntfServer, ntfPrivKey} interval =
+  withNtfClient c ntfServer tknId "TCRN" $ \ntf -> ntfSetCronInterval ntf ntfPrivKey tknId interval
 
 agentNtfCreateSubscription :: AgentClient -> NtfTokenId -> NtfToken -> SMPQueueNtf -> SMP.NtfPrivateAuthKey -> AM NtfSubscriptionId
 agentNtfCreateSubscription c tknId NtfToken {ntfServer, ntfPrivKey} smpQueue nKey =

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

@@ -33,7 +33,6 @@ import qualified Simplex.Messaging.Agent.Store.Postgres.DB as DB
 import Simplex.Messaging.Agent.Store.Shared (Migration (..), MigrationConfirmation (..), MigrationError (..))
 import Simplex.Messaging.Util (ifM, safeDecodeUtf8)
 import System.Exit (exitFailure)
-import UnliftIO.MVar
 
 -- | Create a new Postgres DBStore with the given connection string, schema name and migrations.
 -- If passed schema does not exist in connectInfo database, it will be created.
@@ -54,23 +53,26 @@ createDBStore opts migrations confirmMigrations = do
 
 connectPostgresStore :: DBOpts -> IO DBStore
 connectPostgresStore DBOpts {connstr, schema, poolSize, createSchema} = do
-  dbSem <- newMVar ()
-  dbPool <- newTBQueueIO poolSize
+  dbPriorityPool <- newDBStorePool poolSize
+  dbPool <- newDBStorePool poolSize
   dbClosed <- newTVarIO True
-  let st = DBStore {dbConnstr = connstr, dbSchema = schema, dbPoolSize = fromIntegral poolSize, dbPool, dbSem, dbNew = False, dbClosed}
-  dbNew <- connectPool st createSchema
+  let st = DBStore {dbConnstr = connstr, dbSchema = schema, dbPoolSize = fromIntegral poolSize, dbPriorityPool, dbPool, dbNew = False, dbClosed}
+  dbNew <- connectStore st createSchema
   pure st {dbNew}
 
 -- uninterruptibleMask_ here and below is used here so that it is not interrupted half-way,
 -- it relies on the assumption that when dbClosed = True, the queue is empty,
 -- and when it is False, the queue is full (or will have connections returned to it by the threads that use them).
-connectPool :: DBStore -> Bool -> IO Bool
-connectPool DBStore {dbConnstr, dbSchema, dbPoolSize, dbPool, dbClosed} createSchema = uninterruptibleMask_ $ do
+connectStore :: DBStore -> Bool -> IO Bool
+connectStore DBStore {dbConnstr, dbSchema, dbPoolSize, dbPriorityPool, dbPool, dbClosed} createSchema = uninterruptibleMask_ $ do
   (conn, dbNew) <- connectDB dbConnstr dbSchema createSchema -- TODO [postgres] analogue for dbBusyLoop?
-  conns <- replicateM (dbPoolSize - 1) $ fst <$> connectDB dbConnstr dbSchema False
-  mapM_ (atomically . writeTBQueue dbPool) (conn : conns)
+  writeConns dbPriorityPool . (conn :) =<< mkConns (dbPoolSize - 1)
+  writeConns dbPool =<< mkConns dbPoolSize
   atomically $ writeTVar dbClosed False
   pure dbNew
+  where
+    writeConns pool conns = mapM_ (atomically . writeTBQueue (dbPoolConns pool)) conns
+    mkConns n = replicateM n $ fst <$> connectDB dbConnstr dbSchema False
 
 connectDB :: ByteString -> ByteString -> Bool -> IO (DB.Connection, Bool)
 connectDB connstr schema createSchema = do
@@ -111,16 +113,19 @@ doesSchemaExist db schema = do
   pure schemaExists
 
 closeDBStore :: DBStore -> IO ()
-closeDBStore DBStore {dbPool, dbPoolSize, dbClosed} =
+closeDBStore DBStore {dbPoolSize, dbPriorityPool, dbPool, dbClosed} =
   ifM (readTVarIO dbClosed) (putStrLn "closeDBStore: already closed") $ uninterruptibleMask_ $ do
-    replicateM_ dbPoolSize $ atomically (readTBQueue dbPool) >>= DB.close
+    closePool dbPriorityPool
+    closePool dbPool
     atomically $ writeTVar dbClosed True
+  where
+    closePool pool = replicateM_ dbPoolSize $ atomically (readTBQueue $ dbPoolConns pool) >>= DB.close
 
 reopenDBStore :: DBStore -> IO ()
 reopenDBStore st =
   ifM
     (readTVarIO $ dbClosed st)
-    (void $ connectPool st False)
+    (void $ connectStore st False)
     (putStrLn "reopenDBStore: already opened")
 
 -- not used with postgres client (used for ExecAgentStoreSQL, ExecChatStoreSQL)