smp-server: remove journal store leftovers

src/Simplex/Messaging/Agent/Client.hs

@@ -34,7 +34,6 @@ module Simplex.Messaging.Agent.Client
     withInvLock,
     withLockMap,
     withLocksMap,
-    getMapLock,
     ipAddressProtected,
     closeAgentClient,
     closeProtocolServerClients,

src/Simplex/Messaging/Agent/Lock.hs

@@ -6,7 +6,6 @@ module Simplex.Messaging.Agent.Lock
     withLock',
     withGetLock,
     withGetLocks,
-    getPutLock,
   )
 where
 

src/Simplex/Messaging/Server.hs

@@ -33,9 +33,6 @@ module Simplex.Messaging.Server
   ( runSMPServer,
     runSMPServerBlocking,
     controlPortAuth,
-    importMessages,
-    exportMessages,
-    printMessageStats,
     disconnectTransport,
     verifyCmdAuthorization,
     dummyVerifyCmd,
@@ -115,6 +112,7 @@ 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 (withLoadedQueues)
 import Simplex.Messaging.Server.QueueStore.Types
 import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.Server.StoreLog (foldLogLines)
@@ -144,7 +142,7 @@ import GHC.Conc.Sync (threadLabel)
 #endif
 
 #if defined(dbServerPostgres)
-import Simplex.Messaging.Server.MsgStore.Postgres (exportDbMessages, getDbMessageStats)
+import Simplex.Messaging.Server.MsgStore.Postgres (getDbMessageStats)
 #endif
 
 -- | Runs an SMP server using passed configuration.
@@ -2346,34 +2344,27 @@ randomId = fmap EntityId . randomId'
 
 saveServerMessages :: Bool -> MsgStore s -> IO ()
 saveServerMessages drainMsgs ms = case ms of
-  StoreMemory STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
-    Just f -> exportMessages False ms f drainMsgs
+  StoreMemory ms'@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
+    Just f -> exportMessages ms' f drainMsgs
     Nothing -> logNote "undelivered messages are not saved"
 #if defined(dbServerPostgres)
   StoreDatabase _ -> logNote "closed postgres message storage"
 #endif
 
-exportMessages :: forall s. MsgStoreClass s => Bool -> MsgStore s -> FilePath -> Bool -> IO ()
-exportMessages tty st f drainMsgs = do
+exportMessages :: STMMsgStore -> FilePath -> Bool -> IO ()
+exportMessages ms f drainMsgs = do
   logNote $ "saving messages to file " <> T.pack f
-  run $ case st of
-    StoreMemory ms -> exportMessages_ ms $ getMsgs ms
-#if defined(dbServerPostgres)
-    StoreDatabase ms -> exportDbMessages tty ms
-#endif
+  run $ fmap (\(Sum n) -> n) . withLoadedQueues (queueStore ms) . saveQueueMsgs
   where
-    exportMessages_ ms get = fmap (\(Sum n) -> n) . unsafeWithAllMsgQueues tty ms . saveQueueMsgs get
     run :: (Handle -> IO Int) -> IO ()
     run a = liftIO $ withFile f WriteMode $ tryAny . a >=> \case
       Right n -> logNote $ "messages saved: " <> tshow n
       Left e -> do
         logError $ "error exporting messages: " <> tshow e
         exitFailure
-    getMsgs :: MsgStoreClass s' => s' -> StoreQueue s' -> IO [Message]
-    getMsgs ms q = unsafeRunStore q "saveQueueMsgs" $ getQueueMessages_ drainMsgs q =<< getMsgQueue ms q False
-    saveQueueMsgs :: (StoreQueue s -> IO [Message]) -> Handle -> StoreQueue s -> IO (Sum Int)
-    saveQueueMsgs get h q = do
-      msgs <- get q
+    saveQueueMsgs :: Handle -> STMQueue -> IO (Sum Int)
+    saveQueueMsgs h q = do
+      msgs <- atomically $ getQueueMessages_ drainMsgs =<< getMsgQueue q
       unless (null msgs) $ BLD.hPutBuilder h $ encodeMessages (recipientId q) msgs
       pure $ Sum $ length msgs
     encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')
@@ -2391,7 +2382,7 @@ processServerMessages StartOptions {skipWarnings} = do
       processMessages old_ _ = \case
 #endif
         StoreMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
-          Just f -> ifM (doesFileExist f) (Just <$> importMessages False ms f old_ skipWarnings) (pure Nothing)
+          Just f -> ifM (doesFileExist f) (Just <$> importMessages ms f old_ skipWarnings) (pure Nothing)
           Nothing -> pure Nothing
 #if defined(dbServerPostgres)
         StoreDatabase ms -> processDbMessages old_ expire ms
@@ -2406,26 +2397,24 @@ processServerMessages StartOptions {skipWarnings} = do
         | otherwise = logWarn "skipping message expiration" $> Nothing
 #endif
 
-importMessages :: forall s. MsgStoreClass s => Bool -> s -> FilePath -> Maybe Int64 -> Bool -> IO MessageStats
-importMessages tty ms f old_ skipWarnings  = do
+importMessages :: STMMsgStore -> FilePath -> Maybe Int64 -> Bool -> IO MessageStats
+importMessages ms f old_ skipWarnings  = do
   logNote $ "restoring messages from file " <> T.pack f
   (_, (storedMsgsCount, expiredMsgsCount, overQuota)) <-
-    foldLogLines tty f restoreMsg (Nothing, (0, 0, M.empty))
+    foldLogLines False f restoreMsg (Nothing, (0, 0, M.empty))
   renameFile f $ f <> ".bak"
   mapM_ setOverQuota_ overQuota
-  logQueueStates ms
-  EntityCounts {queueCount} <- liftIO $ getEntityCounts @(StoreQueue s) $ queueStore ms
+  EntityCounts {queueCount} <- liftIO $ getEntityCounts @STMQueue $ queueStore ms
   pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = queueCount}
   where
-    restoreMsg :: (Maybe (RecipientId, StoreQueue s), (Int, Int, Map RecipientId (StoreQueue s))) -> Bool -> ByteString -> IO (Maybe (RecipientId, StoreQueue s), (Int, Int, Map RecipientId (StoreQueue s)))
+    restoreMsg :: (Maybe (RecipientId, STMQueue), (Int, Int, Map RecipientId STMQueue)) -> Bool -> ByteString -> IO (Maybe (RecipientId, STMQueue), (Int, Int, Map RecipientId STMQueue))
     restoreMsg (q_, counts@(!stored, !expired, !overQuota)) eof s = case strDecode s of
       Right (MLRv3 rId msg) -> runExceptT (addToMsgQueue rId msg) >>= either (exitErr . tshow) pure
       Left e
         | eof -> warnOrExit (parsingErr e) $> (q_, counts)
         | otherwise -> exitErr $ parsingErr e
       where
         exitErr e = do
-          when tty $ putStrLn ""
           logError $ "error restoring messages: " <> e
           liftIO exitFailure
         parsingErr :: String -> Text
@@ -2438,7 +2427,6 @@ importMessages tty ms f old_ skipWarnings  = do
           case qOrErr of
             Right q -> addToQueue_ q rId msg
             Left AUTH -> liftIO $ do
-              when tty $ putStrLn ""
               warnOrExit $ "queue " <> safeDecodeUtf8 (encode $ unEntityId rId) <> " does not exist"
               pure (Nothing, counts)
             Left e -> throwE e
@@ -2449,7 +2437,6 @@ importMessages tty ms f old_ skipWarnings  = do
                   writeMsg ms q False msg >>= \case
                     Just _ -> pure (stored + 1, expired, overQuota)
                     Nothing -> liftIO $ do
-                      when tty $ putStrLn ""
                       logError $ decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (messageId msg)
                       pure counts
               | otherwise -> pure (stored, expired + 1, overQuota)
@@ -2460,8 +2447,8 @@ importMessages tty ms f old_ skipWarnings  = do
               where
                 -- if the first message in queue head is "quota", remove it.
                 mergeQuotaMsgs =
-                  withPeekMsgQueue ms q "mergeQuotaMsgs" $ maybe (pure ()) $ \case
-                    (mq, MessageQuota {}) -> tryDeleteMsg_ q mq False
+                  withPeekMsgQueue q $ maybe (pure ()) $ \case
+                    (mq, MessageQuota {}) -> tryDeleteMsg_ mq
                     _ -> pure ()
         warnOrExit e
           | skipWarnings = logWarn e'

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

@@ -555,7 +555,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
     SSCMemory storePaths_ -> do
       let storePath = storeMsgsFile =<< storePaths_
       ms <- newMsgStore STMStoreConfig {storePath, quota = msgQueueQuota}
-      forM_ storePaths_ $ \StorePaths {storeLogFile = f} -> loadStoreLog (mkQueue ms True) f $ queueStore ms
+      forM_ storePaths_ $ \StorePaths {storeLogFile = f} -> loadStoreLog (mkQueue ms) f $ queueStore ms
       pure $ StoreMemory ms
 #if defined(dbServerPostgres)
     SSCDatabase storeCfg -> do
@@ -614,8 +614,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smp
       Just f -> do
         logNote $ "compacting queues in file " <> T.pack f
         st <- newMsgStore STMStoreConfig {storePath = Nothing, quota = msgQueueQuota}
-        -- we don't need to have locks in the map
-        sl <- readWriteQueueStore False (mkQueue st False) f (queueStore st)
+        sl <- readWriteQueueStore False (mkQueue st) f (queueStore st)
         setStoreLog (queueStore st) sl
         closeMsgStore st
       Nothing -> do

src/Simplex/Messaging/Server/Main.hs

@@ -28,7 +28,6 @@ module Simplex.Messaging.Server.Main
     importMessagesToDatabase,
     exportDatabaseToStoreLog,
 #endif
-    storeMsgsJournalDir',
     getServerSourceCode,
     simplexmqSource,
     serverPublicInfo,
@@ -95,7 +94,7 @@ import Data.Semigroup (Sum (..))
 import Simplex.Messaging.Agent.Store.Postgres (checkSchemaExists)
 import Simplex.Messaging.Agent.Store.Shared (MigrationConfirmation (..))
 import Simplex.Messaging.Server.MsgStore.Postgres
-import Simplex.Messaging.Server.MsgStore.STM (STMQueue, STMStoreConfig (..))
+import Simplex.Messaging.Server.MsgStore.STM (STMStoreConfig (..))
 import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..), newMsgStore)
 import Simplex.Messaging.Server.QueueStore.Postgres.Config
 import Simplex.Messaging.Server.QueueStore.Postgres (PostgresQueueStore, batchInsertQueues, batchInsertServices, foldQueueRecs, foldServiceRecs)
@@ -617,14 +616,14 @@ importStoreLogToDatabase :: FilePath -> DBOpts -> IO (Int64, Int64)
 importStoreLogToDatabase storeLogFile dbOpts = do
   ms <- newMsgStore STMStoreConfig {storePath = Nothing, quota = defaultMsgQueueQuota}
   let st = queueStore ms
-  sl <- readWriteQueueStore True (mkQueue ms False) storeLogFile st
+  sl <- readWriteQueueStore True (mkQueue ms) storeLogFile st
   closeStoreLog sl
-  queues <- readTVarIO $ loadedQueues st
+  qs <- readTVarIO $ queues st
   services' <- M.elems <$> readTVarIO (services st)
   let storeCfg = PostgresStoreCfg {dbOpts = dbOpts {createSchema = True}, dbStoreLogPath = Nothing, confirmMigrations = MCConsole, deletedTTL = 86400 * defaultDeletedTTL}
-  ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue (storeCfg, False)
+  ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue storeCfg
   sCnt <- batchInsertServices services' ps
-  qCnt <- batchInsertQueues @STMQueue True queues ps
+  qCnt <- batchInsertQueues True qs ps
   closeQueueStore @PostgresQueue ps
   renameFile storeLogFile $ storeLogFile <> ".bak"
   pure (sCnt, qCnt)
@@ -647,7 +646,7 @@ importMessagesToDatabase msgsLogFile dbOpts = do
 exportDatabaseToStoreLog :: DBOpts -> FilePath -> IO (Int, Int)
 exportDatabaseToStoreLog dbOpts storeLogFilePath = do
   let storeCfg = PostgresStoreCfg {dbOpts, dbStoreLogPath = Nothing, confirmMigrations = MCConsole, deletedTTL = 86400 * defaultDeletedTTL}
-  ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue (storeCfg, False)
+  ps :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue storeCfg
   sl <- openWriteStoreLog False storeLogFilePath
   Sum sCnt <- foldServiceRecs ps $ \sr -> logNewService sl sr $> Sum (1 :: Int)
   Sum qCnt <- foldQueueRecs True True ps $ \(rId, qr) -> logCreateQueue sl rId qr $> Sum (1 :: Int)

src/Simplex/Messaging/Server/MsgStore/Postgres.hs

@@ -29,7 +29,6 @@ where
 import Control.Concurrent.STM
 import qualified Control.Exception as E
 import Control.Monad
-import Control.Monad.Reader
 import Control.Monad.Trans.Except
 import qualified Data.ByteString as B
 import qualified Data.ByteString.Builder as BB
@@ -39,7 +38,6 @@ import Data.IORef
 import Data.Int (Int64)
 import Data.List (intersperse)
 import qualified Data.Map.Strict as M
-import Data.Text (Text)
 import Data.Time.Clock.System (SystemTime (..))
 import Database.PostgreSQL.Simple (Binary (..), In (..), Only (..), (:.) (..))
 import qualified Database.PostgreSQL.Simple as DB
@@ -56,7 +54,7 @@ import Simplex.Messaging.Server.QueueStore.Postgres
 import Simplex.Messaging.Server.QueueStore.Types
 import Simplex.Messaging.Server.StoreLog (foldLogLines)
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Util (maybeFirstRow, maybeFirstRow', (<$$>))
+import Simplex.Messaging.Util (maybeFirstRow, maybeFirstRow')
 import System.IO (Handle, hFlush, stdout)
 
 data PostgresMsgStore = PostgresMsgStore
@@ -81,32 +79,20 @@ instance StoreQueueClass PostgresQueue where
   {-# INLINE recipientId #-}
   queueRec = queueRec'
   {-# INLINE queueRec #-}
-  withQueueLock PostgresQueue {} _ = id -- TODO [messages] maybe it's just transaction?
-  {-# INLINE withQueueLock #-}
-
-newtype DBTransaction = DBTransaction {dbConn :: DB.Connection}
-
-type DBStoreIO a = ReaderT DBTransaction IO a
 
 instance MsgStoreClass PostgresMsgStore where
-  type StoreMonad PostgresMsgStore = ReaderT DBTransaction IO
-  type MsgQueue PostgresMsgStore = ()
   type QueueStore PostgresMsgStore = PostgresQueueStore'
   type StoreQueue PostgresMsgStore = PostgresQueue
   type MsgStoreConfig PostgresMsgStore = PostgresMsgStoreCfg
 
   newMsgStore :: PostgresMsgStoreCfg -> IO PostgresMsgStore
   newMsgStore config = do
-    queueStore_ <- newQueueStore @PostgresQueue (queueStoreCfg config, False)
+    queueStore_ <- newQueueStore @PostgresQueue (queueStoreCfg config)
     pure PostgresMsgStore {config, queueStore_}
 
   closeMsgStore :: PostgresMsgStore -> IO ()
   closeMsgStore = closeQueueStore @PostgresQueue . queueStore_
 
-  withActiveMsgQueues _ _ = error "withActiveMsgQueues not used"
-
-  unsafeWithAllMsgQueues _ _ _ = error "unsafeWithAllMsgQueues not used"
-
   expireOldMessages :: Bool -> PostgresMsgStore -> Int64 -> Int64 -> IO MessageStats
   expireOldMessages _tty ms now ttl =
     maybeFirstRow' newMessageStats toMessageStats $ withConnection st $ \db ->
@@ -150,35 +136,16 @@ instance MsgStoreClass PostgresMsgStore where
             msg_ = toMaybeMessage mRow
          in f a rId $ Right ((qr,) <$> msg_)
 
-  logQueueStates _ = error "logQueueStates not used"
-
-  logQueueState _ = error "logQueueState not used"
-
   queueStore = queueStore_
   {-# INLINE queueStore #-}
 
   loadedQueueCounts :: PostgresMsgStore -> IO LoadedQueueCounts
-  loadedQueueCounts ms = do
-    loadedQueueCount <- M.size <$> readTVarIO queues
-    loadedNotifierCount <- M.size <$> readTVarIO notifiers
-    notifierLockCount <- M.size <$> readTVarIO notifierLocks
-    pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, queueLockCount = 0, notifierLockCount}
-    where
-      PostgresQueueStore {queues, notifiers, notifierLocks} = queueStore_ ms
+  loadedQueueCounts _ = pure LoadedQueueCounts {loadedQueueCount = 0, loadedNotifierCount = 0}
 
-  mkQueue :: PostgresMsgStore -> Bool -> RecipientId -> QueueRec -> IO PostgresQueue
-  mkQueue _ _keepLock rId qr = PostgresQueue rId <$> newTVarIO (Just qr)
+  mkQueue :: PostgresMsgStore -> RecipientId -> QueueRec -> IO PostgresQueue
+  mkQueue _ rId qr = PostgresQueue rId <$> newTVarIO (Just qr)
   {-# INLINE mkQueue #-}
 
-  getMsgQueue _ _ _ = pure ()
-  {-# INLINE getMsgQueue #-}
-
-  getPeekMsgQueue :: PostgresMsgStore -> PostgresQueue -> DBStoreIO (Maybe ((), Message))
-  getPeekMsgQueue _ q = ((),) <$$> tryPeekMsg_ q ()
-
-  withIdleMsgQueue :: Int64 -> PostgresMsgStore -> PostgresQueue -> (() -> DBStoreIO a) -> DBStoreIO (Maybe a, Int)
-  withIdleMsgQueue _ _ _ _ = error "withIdleMsgQueue not used"
-
   deleteQueue :: PostgresMsgStore -> PostgresQueue -> IO (Either ErrorType QueueRec)
   deleteQueue ms q = deleteStoreQueue (queueStore_ ms) q
   {-# INLINE deleteQueue #-}
@@ -189,8 +156,6 @@ instance MsgStoreClass PostgresMsgStore where
     qr <- ExceptT $ deleteStoreQueue (queueStore_ ms) q
     pure (qr, size)
 
-  getQueueMessages_ _ _ _ = error "getQueueMessages_ not used"
-
   writeMsg :: PostgresMsgStore -> PostgresQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
   writeMsg ms q _ msg =
     uninterruptibleMask_ $
@@ -209,43 +174,25 @@ instance MsgStoreClass PostgresMsgStore where
         [] -> Nothing
       PostgresMsgStore {config = PostgresMsgStoreCfg {quota}} = ms
 
-  setOverQuota_ :: PostgresQueue -> IO () -- can ONLY be used while restoring messages, not while server running
-  setOverQuota_ _ = error "TODO setOverQuota_" -- TODO [messages]
-
-  getQueueSize_ :: () -> DBStoreIO Int
-  getQueueSize_ _ = error "getQueueSize_ not used"
-
   getQueueSize :: PostgresMsgStore -> PostgresQueue -> ExceptT ErrorType IO Int
   getQueueSize ms q =
     withDB' "getQueueSize" (queueStore_ ms) $ \db ->
       maybeFirstRow' 0 fromOnly $
         DB.query db "SELECT msg_queue_size FROM msg_queues WHERE recipient_id = ? AND deleted_at IS NULL" (Only (recipientId' q))
 
-  tryPeekMsg_ :: PostgresQueue -> () -> DBStoreIO (Maybe Message)
-  tryPeekMsg_ q _ = do
-    db <- asks dbConn
-    liftIO $ maybeFirstRow toMessage $
-      DB.query
-        db
-        [sql|
-          SELECT msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body
-          FROM messages
-          WHERE recipient_id = ?
-          ORDER BY message_id ASC LIMIT 1
-        |]
-        (Only (recipientId' q))
-
-  tryDeleteMsg_ :: PostgresQueue -> () -> Bool -> DBStoreIO ()
-  tryDeleteMsg_ _q _ _ = error "tryDeleteMsg_ not used" -- do
-
-  isolateQueue :: PostgresMsgStore -> PostgresQueue -> Text -> DBStoreIO a -> ExceptT ErrorType IO a
-  isolateQueue ms _q op a = uninterruptibleMask_ $ withDB' op (queueStore_ ms) $ runReaderT a . DBTransaction
-
-  unsafeRunStore _ _ _ = error "unsafeRunStore not used"
-
   tryPeekMsg :: PostgresMsgStore -> PostgresQueue -> ExceptT ErrorType IO (Maybe Message)
-  tryPeekMsg ms q = isolateQueue ms q "tryPeekMsg" $ tryPeekMsg_ q ()
-  {-# INLINE tryPeekMsg #-}
+  tryPeekMsg ms q =
+    uninterruptibleMask_ $ withDB' "tryPeekMsg" (queueStore_ ms) $ \db ->
+      maybeFirstRow toMessage $
+        DB.query
+          db
+          [sql|
+            SELECT msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body
+            FROM messages
+            WHERE recipient_id = ?
+            ORDER BY message_id ASC LIMIT 1
+          |]
+          (Only (recipientId' q))
 
   tryPeekMsgs :: PostgresMsgStore -> [PostgresQueue] -> ExceptT ErrorType IO (M.Map RecipientId Message)
   tryPeekMsgs _ms [] = pure M.empty