smp-server: remove journal store leftovers (#1946)

* smp-server: remove postgres queue cache

* smp-server: remove queue locks

* smp-server: simplify message store class

* smp-server: remove unused functions and exports

* tests: remove journal leftovers

* agent: remove unused lock exports

* fix

* fixes

---------

Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>

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,18 +2397,17 @@ 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
@@ -2425,7 +2415,6 @@ importMessages tty ms f old_ skipWarnings  = do
         | 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

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

@@ -15,6 +15,11 @@ module Simplex.Messaging.Server.MsgStore.STM
   ( STMMsgStore (..),
     STMStoreConfig (..),
     STMQueue,
+    getMsgQueue,
+    withPeekMsgQueue,
+    getQueueMessages_,
+    setOverQuota_,
+    tryDeleteMsg_,
   )
 where
 
@@ -24,7 +29,8 @@ import Control.Monad.Trans.Except
 import Data.Functor (($>))
 import Data.Int (Int64)
 import qualified Data.Map.Strict as M
-import Data.Text (Text)
+import Data.Maybe (catMaybes)
+import Data.Time.Clock.System (SystemTime (systemSeconds))
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
@@ -61,12 +67,8 @@ instance StoreQueueClass STMQueue where
   {-# INLINE recipientId #-}
   queueRec = queueRec'
   {-# INLINE queueRec #-}
-  withQueueLock _ _ = id
-  {-# INLINE withQueueLock #-}
 
 instance MsgStoreClass STMMsgStore where
-  type StoreMonad STMMsgStore = STM
-  type MsgQueue STMMsgStore = STMMsgQueue
   type QueueStore STMMsgStore = STMQueueStore STMQueue
   type StoreQueue STMMsgStore = STMQueue
   type MsgStoreConfig STMMsgStore = STMStoreConfig
@@ -78,24 +80,16 @@ instance MsgStoreClass STMMsgStore where
 
   closeMsgStore = closeQueueStore @STMQueue . queueStore_
   {-# INLINE closeMsgStore #-}
-  withActiveMsgQueues = withLoadedQueues . queueStore_
-  {-# INLINE withActiveMsgQueues #-}
-  unsafeWithAllMsgQueues _ = withLoadedQueues . queueStore_
-  {-# INLINE unsafeWithAllMsgQueues #-}
 
   expireOldMessages :: Bool -> STMMsgStore -> Int64 -> Int64 -> IO MessageStats
   expireOldMessages _tty ms now ttl =
-    withLoadedQueues (queueStore_ ms) $ atomically . expireQueueMsgs ms now (now - ttl)
+    withLoadedQueues (queueStore_ ms) $ atomically . expireQueueMsgs (now - ttl)
 
   foldRcvServiceMessages :: STMMsgStore -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO (Either ErrorType a)
   foldRcvServiceMessages ms serviceId f = fmap Right . foldRcvServiceQueues (queueStore_ ms) serviceId f'
     where
       f' a (q, qr) = runExceptT (tryPeekMsg ms q) >>= f a (recipientId q) . ((qr,) <$$>)
 
-  logQueueStates _ = pure ()
-  {-# INLINE logQueueStates #-}
-  logQueueState _ = pure ()
-  {-# INLINE logQueueState #-}
   queueStore = queueStore_
   {-# INLINE queueStore #-}
 
@@ -103,34 +97,11 @@ instance MsgStoreClass STMMsgStore where
   loadedQueueCounts STMMsgStore {queueStore_ = st} = do
     loadedQueueCount <- M.size <$> readTVarIO (queues st)
     loadedNotifierCount <- M.size <$> readTVarIO (notifiers st)
-    pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount, queueLockCount = 0, notifierLockCount = 0}
+    pure LoadedQueueCounts {loadedQueueCount, loadedNotifierCount}
 
-  mkQueue _ _ rId qr = STMQueue rId <$> newTVarIO (Just qr) <*> newTVarIO Nothing
+  mkQueue _ rId qr = STMQueue rId <$> newTVarIO (Just qr) <*> newTVarIO Nothing
   {-# INLINE mkQueue #-}
 
-  getMsgQueue :: STMMsgStore -> STMQueue -> Bool -> STM STMMsgQueue
-  getMsgQueue _ STMQueue {msgQueue'} _ = readTVar msgQueue' >>= maybe newQ pure
-    where
-      newQ = do
-        msgTQueue <- newTQueue
-        canWrite <- newTVar True
-        size <- newTVar 0
-        let q = STMMsgQueue {msgTQueue, canWrite, size}
-        writeTVar msgQueue' (Just q)
-        pure q
-
-  getPeekMsgQueue :: STMMsgStore -> STMQueue -> STM (Maybe (STMMsgQueue, Message))
-  getPeekMsgQueue _ q@STMQueue {msgQueue'} = readTVar msgQueue' $>>= \mq -> (mq,) <$$> tryPeekMsg_ q mq
-
-  -- does not create queue if it does not exist, does not delete it if it does (can't just close in-memory queue)
-  withIdleMsgQueue :: Int64 -> STMMsgStore -> STMQueue -> (STMMsgQueue -> STM a) -> STM (Maybe a, Int)
-  withIdleMsgQueue _ _ STMQueue {msgQueue'} action = readTVar msgQueue' >>= \case
-    Just q -> do
-      r <- action q
-      sz <- getQueueSize_ q
-      pure (Just r, sz)
-    Nothing -> pure (Nothing, 0)
-
   deleteQueue :: STMMsgStore -> STMQueue -> IO (Either ErrorType QueueRec)
   deleteQueue ms q = fst <$$> deleteQueue_ ms q
 
@@ -140,17 +111,9 @@ instance MsgStoreClass STMMsgStore where
     where
       getSize = maybe (pure 0) (\STMMsgQueue {size} -> readTVarIO size)
 
-  getQueueMessages_ :: Bool -> STMQueue -> STMMsgQueue -> STM [Message]
-  getQueueMessages_ drainMsgs _ = (if drainMsgs then flushTQueue else snapshotTQueue) . msgTQueue
-    where
-      snapshotTQueue q = do
-        msgs <- flushTQueue q
-        mapM_ (writeTQueue q) msgs
-        pure msgs
-
   writeMsg :: STMMsgStore -> STMQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
   writeMsg ms q' _logState msg = liftIO $ atomically $ do
-    STMMsgQueue {msgTQueue = q, canWrite, size} <- getMsgQueue ms q' True
+    STMMsgQueue {msgTQueue = q, canWrite, size} <- getMsgQueue q'
     canWrt <- readTVar canWrite
     empty <- isEmptyTQueue q
     if canWrt || empty
@@ -166,29 +129,96 @@ instance MsgStoreClass STMMsgStore where
       STMMsgStore {storeConfig = STMStoreConfig {quota}} = ms
       msgQuota = MessageQuota {msgId = messageId msg, msgTs = messageTs msg}
 
-  setOverQuota_ :: STMQueue -> IO ()
-  setOverQuota_ q = readTVarIO (msgQueue' q) >>= mapM_ (\mq -> atomically $ writeTVar (canWrite mq) False)
+  tryPeekMsg :: STMMsgStore -> STMQueue -> ExceptT ErrorType IO (Maybe Message)
+  tryPeekMsg _ q = snd <$$> withPeekMsgQueue q pure
+  {-# INLINE tryPeekMsg #-}
 
-  getQueueSize_ :: STMMsgQueue -> STM Int
-  getQueueSize_ STMMsgQueue {size} = readTVar size
+  tryPeekMsgs :: STMMsgStore -> [STMQueue] -> ExceptT ErrorType IO (M.Map RecipientId Message)
+  tryPeekMsgs st qs = M.fromList . catMaybes <$> mapM (\q -> (recipientId q,) <$$> tryPeekMsg st q) qs
 
-  tryPeekMsg_ :: STMQueue -> STMMsgQueue -> STM (Maybe Message)
-  tryPeekMsg_ _ = tryPeekTQueue . msgTQueue
-  {-# INLINE tryPeekMsg_ #-}
+  tryDelMsg :: STMMsgStore -> STMQueue -> MsgId -> ExceptT ErrorType IO (Maybe Message)
+  tryDelMsg _ q msgId' =
+    withPeekMsgQueue q $
+      maybe (pure Nothing) $ \(mq, msg) ->
+        if messageId msg == msgId'
+          then tryDeleteMsg_ mq $> Just msg
+          else pure Nothing
 
-  tryDeleteMsg_ :: STMQueue -> STMMsgQueue -> Bool -> STM ()
-  tryDeleteMsg_ _ STMMsgQueue {msgTQueue = q, size} _logState =
-    tryReadTQueue q >>= \case
-      Just _ -> modifyTVar' size (subtract 1)
-      _ -> pure ()
+  tryDelPeekMsg :: STMMsgStore -> STMQueue -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
+  tryDelPeekMsg _ q msgId' =
+    withPeekMsgQueue q $
+      maybe (pure (Nothing, Nothing)) $ \(mq, msg) ->
+        if messageId msg == msgId'
+          then (Just msg,) <$> (tryDeleteMsg_ mq >> tryPeekMsg_ mq)
+          else pure (Nothing, Just msg)
 
-  isolateQueue :: STMMsgStore -> STMQueue -> Text -> STM a -> ExceptT ErrorType IO a
-  isolateQueue _ _ _ = liftIO . atomically
-  {-# INLINE isolateQueue #-}
+  deleteExpiredMsgs :: STMMsgStore -> STMQueue -> Int64 -> ExceptT ErrorType IO Int
+  deleteExpiredMsgs _ q old = liftIO $ atomically $ getMsgQueue q >>= deleteExpireMsgs_ old
 
-  unsafeRunStore :: STMQueue -> Text -> STM a -> IO a
-  unsafeRunStore _ _ = atomically
-  {-# INLINE unsafeRunStore #-}
+  getQueueSize :: STMMsgStore -> STMQueue -> ExceptT ErrorType IO Int
+  getQueueSize _ q = withPeekMsgQueue q $ maybe (pure 0) (getQueueSize_ . fst)
+  {-# INLINE getQueueSize #-}
+
+-- can ONLY be used while restoring messages, not while server running
+setOverQuota_ :: STMQueue -> IO ()
+setOverQuota_ q = readTVarIO (msgQueue' q) >>= mapM_ (\mq -> atomically $ writeTVar (canWrite mq) False)
+
+getQueueSize_ :: STMMsgQueue -> STM Int
+getQueueSize_ STMMsgQueue {size} = readTVar size
+
+tryPeekMsg_ :: STMMsgQueue -> STM (Maybe Message)
+tryPeekMsg_ = tryPeekTQueue . msgTQueue
+{-# INLINE tryPeekMsg_ #-}
+
+tryDeleteMsg_ :: STMMsgQueue -> STM ()
+tryDeleteMsg_ STMMsgQueue {msgTQueue = q, size} =
+  tryReadTQueue q >>= \case
+    Just _ -> modifyTVar' size (subtract 1)
+    _ -> pure ()
+
+getMsgQueue :: STMQueue -> STM STMMsgQueue
+getMsgQueue STMQueue {msgQueue'} = readTVar msgQueue' >>= maybe newQ pure
+  where
+    newQ = do
+      msgTQueue <- newTQueue
+      canWrite <- newTVar True
+      size <- newTVar 0
+      let q = STMMsgQueue {msgTQueue, canWrite, size}
+      writeTVar msgQueue' (Just q)
+      pure q
+
+-- The action is called with Nothing when it is known that the queue is empty
+withPeekMsgQueue :: STMQueue -> (Maybe (STMMsgQueue, Message) -> STM a) -> ExceptT ErrorType IO a
+withPeekMsgQueue STMQueue {msgQueue'} a =
+  liftIO $ atomically $ (readTVar msgQueue' $>>= \mq -> (mq,) <$$> tryPeekMsg_ mq) >>= a
+
+getQueueMessages_ :: Bool -> STMMsgQueue -> STM [Message]
+getQueueMessages_ drainMsgs = (if drainMsgs then flushTQueue else snapshotTQueue) . msgTQueue
+  where
+    snapshotTQueue q = do
+      msgs <- flushTQueue q
+      mapM_ (writeTQueue q) msgs
+      pure msgs
+
+-- does not create queue if it does not exist
+expireQueueMsgs :: Int64 -> STMQueue -> STM MessageStats
+expireQueueMsgs old STMQueue {msgQueue'} =
+  readTVar msgQueue' >>= \case
+    Just mq -> do
+      expiredMsgsCount <- deleteExpireMsgs_ old mq
+      storedMsgsCount <- getQueueSize_ mq
+      pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1}
+    Nothing -> pure newMessageStats {storedQueues = 1}
+
+deleteExpireMsgs_ :: Int64 -> STMMsgQueue -> STM Int
+deleteExpireMsgs_ old mq = loop 0
+  where
+    loop dc =
+      tryPeekMsg_ mq >>= \case
+        Just Message {msgTs}
+          | systemSeconds msgTs < old ->
+              tryDeleteMsg_ mq >> loop (dc + 1)
+        _ -> pure dc
 
 deleteQueue_ :: STMMsgStore -> STMQueue -> IO (Either ErrorType (QueueRec, Maybe STMMsgQueue))
 deleteQueue_ ms q = deleteStoreQueue (queueStore_ ms) q >>= mapM remove

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

@@ -4,17 +4,12 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE MultiWayIf #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE TypeFamilyDependencies #-}
-{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}
-
-{-# HLINT ignore "Redundant multi-way if" #-}
-
 module Simplex.Messaging.Server.MsgStore.Types
   ( MsgStoreClass (..),
     MSType (..),
@@ -30,99 +25,43 @@ module Simplex.Messaging.Server.MsgStore.Types
     getQueues,
     getQueueRecs,
     readQueueRec,
-    withPeekMsgQueue,
-    expireQueueMsgs,
-    deleteExpireMsgs_,
   ) where
 
 import Control.Concurrent.STM
 import Control.Monad
 import Control.Monad.Trans.Except
-import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.Kind
 import Data.Map.Strict (Map)
-import qualified Data.Map.Strict as M
-import Data.Maybe (catMaybes, fromMaybe)
-import Data.Text (Text)
-import Data.Time.Clock.System (SystemTime (systemSeconds))
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.Types
-import Simplex.Messaging.Util ((<$$>), ($>>=))
+import Simplex.Messaging.Util (($>>=))
 
-class (Monad (StoreMonad s), QueueStoreClass (StoreQueue s) (QueueStore s)) => MsgStoreClass s where
-  type StoreMonad s = (m :: Type -> Type) | m -> s
+class QueueStoreClass (StoreQueue s) (QueueStore s) => MsgStoreClass s where
   type MsgStoreConfig s = c | c -> s
-  type MsgQueue s = q | q -> s
   type StoreQueue s = q | q -> s
   type QueueStore s = qs | qs -> s
   newMsgStore :: MsgStoreConfig s -> IO s
   closeMsgStore :: s -> IO ()
-  withActiveMsgQueues :: Monoid a => s -> (StoreQueue s -> IO a) -> IO a
-  -- This function can only be used in server CLI commands or before server is started.
-  -- tty, store
-  unsafeWithAllMsgQueues :: Monoid a => Bool -> s -> (StoreQueue s -> IO a) -> IO a
   -- tty, store, now, ttl
   expireOldMessages :: Bool -> s -> Int64 -> Int64 -> IO MessageStats
   foldRcvServiceMessages :: s -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO (Either ErrorType a)
-  logQueueStates :: s -> IO ()
-  logQueueState :: StoreQueue s -> StoreMonad s ()
   queueStore :: s -> QueueStore s
   loadedQueueCounts :: s -> IO LoadedQueueCounts
 
   -- message store methods
-  mkQueue :: s -> Bool -> RecipientId -> QueueRec -> IO (StoreQueue s)
-  getMsgQueue :: s -> StoreQueue s -> Bool -> StoreMonad s (MsgQueue s)
-  getPeekMsgQueue :: s -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue s, Message))
-
-  -- the journal queue will be closed after action if it was initially closed or idle longer than interval in config
-  withIdleMsgQueue :: Int64 -> s -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int)
+  mkQueue :: s -> RecipientId -> QueueRec -> IO (StoreQueue s)
   deleteQueue :: s -> StoreQueue s -> IO (Either ErrorType QueueRec)
   deleteQueueSize :: s -> StoreQueue s -> IO (Either ErrorType (QueueRec, Int))
-  getQueueMessages_ :: Bool -> StoreQueue s -> MsgQueue s -> StoreMonad s [Message]
   writeMsg :: s -> StoreQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
-  setOverQuota_ :: StoreQueue s -> IO () -- can ONLY be used while restoring messages, not while server running
-  getQueueSize_ :: MsgQueue s -> StoreMonad s Int
-  tryPeekMsg_ :: StoreQueue s -> MsgQueue s -> StoreMonad s (Maybe Message)
-  tryDeleteMsg_ :: StoreQueue s -> MsgQueue s -> Bool -> StoreMonad s ()
-  isolateQueue :: s -> StoreQueue s -> Text -> StoreMonad s a -> ExceptT ErrorType IO a
-  unsafeRunStore :: StoreQueue s -> Text -> StoreMonad s a -> IO a
-
-  -- default implementations are overridden for PostgreSQL storage of messages
   tryPeekMsg :: s -> StoreQueue s -> ExceptT ErrorType IO (Maybe Message)
-  tryPeekMsg st q = snd <$$> withPeekMsgQueue st q "tryPeekMsg" pure
-  {-# INLINE tryPeekMsg #-}
-  
   tryPeekMsgs :: s -> [StoreQueue s] -> ExceptT ErrorType IO (Map RecipientId Message)
-  tryPeekMsgs st qs = M.fromList . catMaybes <$> mapM (\q -> (recipientId q,) <$$> tryPeekMsg st q) qs
-
   tryDelMsg :: s -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
-  tryDelMsg st q msgId' =
-    withPeekMsgQueue st q "tryDelMsg" $
-      maybe (pure Nothing) $ \(mq, msg) ->
-        if
-          | messageId msg == msgId' ->
-              tryDeleteMsg_ q mq True $> Just msg
-          | otherwise -> pure Nothing
-
   -- atomic delete (== read) last and peek next message if available
   tryDelPeekMsg :: s -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
-  tryDelPeekMsg st q msgId' =
-    withPeekMsgQueue st q "tryDelPeekMsg" $
-      maybe (pure (Nothing, Nothing)) $ \(mq, msg) ->
-        if
-          | messageId msg == msgId' -> (Just msg,) <$> (tryDeleteMsg_ q mq True >> tryPeekMsg_ q mq)
-          | otherwise -> pure (Nothing, Just msg)
-
   deleteExpiredMsgs :: s -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
-  deleteExpiredMsgs st q old =
-    isolateQueue st q "deleteExpiredMsgs" $
-      getMsgQueue st q False >>= deleteExpireMsgs_ old q
-
   getQueueSize :: s -> StoreQueue s -> ExceptT ErrorType IO Int
-  getQueueSize st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst)
-  {-# INLINE getQueueSize #-}
 
 data MSType = MSMemory | MSPostgres
 
@@ -153,16 +92,14 @@ instance Semigroup MessageStats where
 
 data LoadedQueueCounts = LoadedQueueCounts
   { loadedQueueCount :: Int,
-    loadedNotifierCount :: Int,
-    queueLockCount :: Int,
-    notifierLockCount :: Int
+    loadedNotifierCount :: Int
   }
 
 newMessageStats :: MessageStats
 newMessageStats = MessageStats 0 0 0
 
 addQueue :: MsgStoreClass s => s -> RecipientId -> QueueRec -> IO (Either ErrorType (StoreQueue s))
-addQueue st = addQueue_ (queueStore st) (mkQueue st True)
+addQueue st = addQueue_ (queueStore st) (mkQueue st)
 {-# INLINE addQueue #-}
 
 getQueue :: (MsgStoreClass s, QueueParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s))
@@ -182,28 +119,3 @@ getQueueRecs st party qIds = getQueues st party qIds >>= mapM (fmap join . mapM
 readQueueRec :: StoreQueueClass q => q -> IO (Either ErrorType (q, QueueRec))
 readQueueRec q = maybe (Left AUTH) (Right . (q,)) <$> readTVarIO (queueRec q)
 {-# INLINE readQueueRec #-}
-
--- The action is called with Nothing when it is known that the queue is empty
-withPeekMsgQueue :: MsgStoreClass s => s -> StoreQueue s -> Text -> (Maybe (MsgQueue s, Message) -> StoreMonad s a) -> ExceptT ErrorType IO a
-withPeekMsgQueue st q op a = isolateQueue st q op $ getPeekMsgQueue st q >>= a
-{-# INLINE withPeekMsgQueue #-}
-
--- not used with PostgreSQL message store
-expireQueueMsgs :: MsgStoreClass s => s -> Int64 -> Int64 -> StoreQueue s -> StoreMonad s MessageStats
-expireQueueMsgs st now old q = do
-  (expired_, stored) <- withIdleMsgQueue now st q $ deleteExpireMsgs_ old q
-  pure MessageStats {storedMsgsCount = stored, expiredMsgsCount = fromMaybe 0 expired_, storedQueues = 1}
-
--- not used with PostgreSQL message store
-deleteExpireMsgs_ :: MsgStoreClass s => Int64 -> StoreQueue s -> MsgQueue s -> StoreMonad s Int
-deleteExpireMsgs_ old q mq = do
-  n <- loop 0
-  logQueueState q
-  pure n
-  where
-    loop dc =
-      tryPeekMsg_ q mq >>= \case
-        Just Message {msgTs}
-          | systemSeconds msgTs < old ->
-              tryDeleteMsg_ q mq False >> loop (dc + 1)
-        _ -> pure dc

src/Simplex/Messaging/Server/Prometheus.hs

@@ -592,15 +592,7 @@ prometheusMetrics sm rtm ts =
       \\n\
       \# HELP simplex_smp_loaded_queues_ntf_count Total loaded ntf credential references (all ntf credentials for memory storage)\n\
       \# TYPE simplex_smp_loaded_queues_ntf_count gauge\n\
-      \simplex_smp_loaded_queues_ntf_count " <> mshow (loadedNotifierCount loadedCounts) <> "\n# loadedCounts.loadedNotifierCount\n\
-      \\n\
-      \# HELP simplex_smp_loaded_queues_queue_lock_count Total queue locks (0 for memory storage)\n\
-      \# TYPE simplex_smp_loaded_queues_queue_lock_count gauge\n\
-      \simplex_smp_loaded_queues_queue_lock_count " <> mshow (queueLockCount loadedCounts) <> "\n# loadedCounts.queueLockCount\n\
-      \\n\
-      \# HELP simplex_smp_loaded_queues_ntf_lock_count Total notifier locks (0 for memory storage)\n\
-      \# TYPE simplex_smp_loaded_queues_ntf_lock_count gauge\n\
-      \simplex_smp_loaded_queues_ntf_lock_count " <> mshow (notifierLockCount loadedCounts) <> "\n# loadedCounts.notifierLockCount\n"
+      \simplex_smp_loaded_queues_ntf_count " <> mshow (loadedNotifierCount loadedCounts) <> "\n# loadedCounts.loadedNotifierCount\n"
 
     showTimeBuckets :: Text -> IM.IntMap Int -> Text
     showTimeBuckets metric = T.concat . snd . mapAccumL accumBucket (0, 0) . IM.assocs

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

@@ -24,9 +24,7 @@ module Simplex.Messaging.Server.QueueStore.Postgres
     batchInsertServices,
     batchInsertQueues,
     foldServiceRecs,
-    foldRcvServiceQueueRecs,
     foldQueueRecs,
-    foldRecentQueueRecs,
     handleDuplicate,
     rowToQueueRec,
     withLog_,
@@ -43,13 +41,12 @@ import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Class
 import Control.Monad.Trans.Except
-import Data.Bifunctor (first)
 import Data.ByteString.Builder (Builder)
 import qualified Data.ByteString.Builder as BB
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Lazy as LB
 import Data.Bitraversable (bimapM)
-import Data.Either (fromRight, lefts)
+import Data.Either (fromRight)
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (foldl', intersperse, partition)
@@ -91,7 +88,7 @@ import Simplex.Messaging.SystemTime
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (SMPServiceRole (..))
-import Simplex.Messaging.Util (eitherToMaybe, firstRow, ifM, maybeFirstRow, maybeFirstRow', tshow, (<$$>), ($>>=))
+import Simplex.Messaging.Util (eitherToMaybe, firstRow, maybeFirstRow, maybeFirstRow', tshow, (<$$>), ($>>=))
 import System.Exit (exitFailure)
 import System.IO (IOMode (..), hFlush, stdout)
 import UnliftIO.STM
@@ -103,35 +100,19 @@ import Simplex.Messaging.Encoding.String
 data PostgresQueueStore q = PostgresQueueStore
   { dbStore :: DBStore,
     dbStoreLog :: Maybe (StoreLog 'WriteMode),
-    -- this map caches all created and opened queues
-    queues :: TMap RecipientId q,
-    -- this map only cashes the queues that were attempted to send messages to,
-    senders :: TMap SenderId RecipientId,
-    links :: TMap LinkId RecipientId,
-    -- this map only cashes the queues that were attempted to be subscribed to,
-    notifiers :: TMap NotifierId RecipientId,
-    notifierLocks :: TMap NotifierId Lock,
     serviceLocks :: TMap CertFingerprint Lock,
-    deletedTTL :: Int64,
-    useCache :: Bool
+    deletedTTL :: Int64
   }
 
-type UseQueueCache = Bool
-
 instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
-  type QueueStoreCfg (PostgresQueueStore q) = (PostgresStoreCfg, UseQueueCache)
+  type QueueStoreCfg (PostgresQueueStore q) = PostgresStoreCfg
 
-  newQueueStore :: (PostgresStoreCfg, UseQueueCache)  -> IO (PostgresQueueStore q)
-  newQueueStore (PostgresStoreCfg {dbOpts, dbStoreLogPath, confirmMigrations, deletedTTL}, useCache) = do
+  newQueueStore :: PostgresStoreCfg -> IO (PostgresQueueStore q)
+  newQueueStore PostgresStoreCfg {dbOpts, dbStoreLogPath, confirmMigrations, deletedTTL} = do
     dbStore <- either err pure =<< createDBStore dbOpts serverMigrations (MigrationConfig confirmMigrations Nothing)
     dbStoreLog <- mapM (openWriteStoreLog True) dbStoreLogPath
-    queues <- TM.emptyIO
-    senders <- TM.emptyIO
-    links <- TM.emptyIO
-    notifiers <- TM.emptyIO
-    notifierLocks <- TM.emptyIO
     serviceLocks <- TM.emptyIO
-    pure PostgresQueueStore {dbStore, dbStoreLog, queues, senders, links, notifiers, notifierLocks, serviceLocks, deletedTTL, useCache}
+    pure PostgresQueueStore {dbStore, dbStoreLog, serviceLocks, deletedTTL}
     where
       err e = do
         logError $ "STORE: newQueueStore, error opening PostgreSQL database, " <> tshow e
@@ -142,9 +123,6 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
     closeDBStore dbStore
     mapM_ closeStoreLog dbStoreLog
 
-  loadedQueues = queues
-  {-# INLINE loadedQueues #-}
-
   compactQueues :: PostgresQueueStore q -> IO Int64
   compactQueues st@PostgresQueueStore {deletedTTL} = do
     old <- subtract deletedTTL . systemSeconds <$> liftIO getSystemTime
@@ -170,131 +148,44 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
           (SRMessaging, SRNotifier)
       pure EntityCounts {queueCount, notifierCount, rcvServiceCount, ntfServiceCount, rcvServiceQueuesCount, ntfServiceQueuesCount}
 
-  -- this implementation assumes that the lock is already taken by addQueue
-  -- and relies on unique constraints in the database to prevent duplicate IDs.
+  -- this implementation relies on unique constraints in the database to prevent duplicate IDs.
   addQueue_ :: PostgresQueueStore q -> (RecipientId -> QueueRec -> IO q) -> RecipientId -> QueueRec -> IO (Either ErrorType q)
   addQueue_ st mkQ rId qr = do
     sq <- mkQ rId qr
-    withQueueLock sq "addQueue_" $ E.uninterruptibleMask_ $ runExceptT $ do
+    E.uninterruptibleMask_ $ runExceptT $ do
       void $ withDB "addQueue_" st $ \db ->
         E.try (DB.execute db insertQueueQuery $ queueRecToRow (rId, qr))
           >>= bimapM handleDuplicate pure
-      when useCache $ do
-        atomically $ TM.insert rId sq queues
-        atomically $ TM.insert (senderId qr) rId senders
-        forM_ (notifier qr) $ \NtfCreds {notifierId = nId} -> atomically $ TM.insert nId rId notifiers
-        forM_ (queueData qr) $ \(lnkId, _) -> atomically $ TM.insert lnkId rId links
       withLog "addStoreQueue" st $ \s -> logCreateQueue s rId qr
       pure sq
-    where
-      PostgresQueueStore {queues, senders, links, notifiers, useCache} = st
-      -- Not doing duplicate checks in maps as the probability of duplicates is very low.
-      -- It needs to be reconsidered when IDs are supplied by the users.
-      -- hasId = anyM [TM.memberIO rId queues, TM.memberIO senderId senders, hasNotifier]
-      -- hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.memberIO notifierId notifiers) notifier
 
-  getQueue_ :: QueueParty p => PostgresQueueStore q -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q)
-  getQueue_ st mkQ party qId
-    | useCache = case party of
-        SRecipient -> getRcvQueue qId
-        SSender -> TM.lookupIO qId senders >>= maybe (mask loadSndQueue) getRcvQueue
-        SSenderLink -> TM.lookupIO qId links >>= maybe (mask loadLinkQueue) getRcvQueue
-        -- loaded queue is deleted from notifiers map to reduce cache size after queue was subscribed to by ntf server
-        SNotifier -> TM.lookupIO qId notifiers >>= maybe (mask loadNtfQueue) (getRcvQueue >=> (atomically (TM.delete qId notifiers) $>))
-    | otherwise = case party of
-        SRecipient -> loadQueueNoCache " WHERE recipient_id = ?"
-        SSender -> loadQueueNoCache " WHERE sender_id = ?"
-        SSenderLink -> loadQueueNoCache " WHERE link_id = ?"
-        SNotifier -> loadQueueNoCache " WHERE notifier_id = ?"
+  getQueue_ :: QueueParty p => PostgresQueueStore q -> (RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q)
+  getQueue_ st mkQ party qId = case party of
+    SRecipient -> loadQueue " WHERE recipient_id = ?"
+    SSender -> loadQueue " WHERE sender_id = ?"
+    SSenderLink -> loadQueue " WHERE link_id = ?"
+    SNotifier -> loadQueue " WHERE notifier_id = ?"
     where
-      PostgresQueueStore {queues, senders, links, notifiers, useCache} = st
-      getRcvQueue rId = TM.lookupIO rId queues >>= maybe (mask loadRcvQueue) (pure . Right)
-      loadRcvQueue = do
-        (rId, qRec) <- loadQueue " WHERE recipient_id = ?"
-        liftIO $ cacheQueue rId qRec $ \_ -> pure () -- recipient map already checked, not caching sender ref
-      loadSndQueue = loadSndQueue_ " WHERE sender_id = ?"
-      loadLinkQueue = loadSndQueue_ " WHERE link_id = ?"
-      loadNtfQueue = do
-        (rId, qRec) <- loadQueue " WHERE notifier_id = ?"
-        liftIO $
-          TM.lookupIO rId queues -- checking recipient map first, not creating lock in map, not caching queue
-            >>= maybe (mkQ False rId qRec) pure
-      loadSndQueue_ condition = do
-        (rId, qRec) <- loadQueue condition
-        liftIO $
-          TM.lookupIO rId queues -- checking recipient map first
-            >>= maybe (cacheQueue rId qRec cacheSender) (atomically (cacheSender rId) $>)
-      loadQueueNoCache cond = mask $ loadQueue cond >>= liftIO . uncurry (mkQ True)
+      loadQueue cond = mask $ loadQueueRec cond >>= liftIO . uncurry mkQ
       mask = E.uninterruptibleMask_ . runExceptT
-      cacheSender rId = TM.insert qId rId senders
-      loadQueue condition =
+      loadQueueRec condition =
         withDB "getQueue_" st $ \db -> firstRow rowToQueueRec AUTH $
           DB.query db (queueRecQuery <> condition <> " AND deleted_at IS NULL") (Only qId)
-      cacheQueue rId qRec insertRef = do
-        sq <- mkQ True rId qRec -- loaded queue
-        -- This lock prevents the scenario when the queue is added to cache,
-        -- while another thread is proccessing the same queue in withAllMsgQueues
-        -- without adding it to cache, possibly trying to open the same files twice.
-        -- Alse see comment in idleDeleteExpiredMsgs.
-        withQueueLock sq "getQueue_" $ atomically $
-          -- checking the cache again for concurrent reads,
-          -- use previously loaded queue if exists.
-          TM.lookup rId queues >>= \case
-            Just sq' -> pure sq'
-            Nothing -> do
-              insertRef rId
-              TM.insert rId sq queues
-              pure sq
 
-  getQueues_ :: forall p. BatchParty p => PostgresQueueStore q -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q]
+  getQueues_ :: forall p. BatchParty p => PostgresQueueStore q -> (RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q]
   getQueues_ st mkQ party qIds
     | null qIds = pure []
-    | useCache = case party of
-        SRecipient -> do
-          qs <- readTVarIO queues
-          let qs' = map (\qId -> get qs qId qId) qIds
-          E.uninterruptibleMask_ $ loadQueues qs' " WHERE recipient_id IN ?" cacheRcvQueue
-        SNotifier -> do
-          ns <- readTVarIO notifiers
-          qs <- readTVarIO queues
-          let qs' = map (\qId -> get ns qId qId >>= get qs qId) qIds
-          E.uninterruptibleMask_ $ loadQueues qs' " WHERE notifier_id IN ?" $ \(rId, qRec) ->
-            forM (notifier qRec) $ \NtfCreds {notifierId = nId} -> -- it is always Just with this query
-              (nId,) <$> maybe (mkQ False rId qRec) pure (M.lookup rId qs)
     | otherwise = E.uninterruptibleMask_ $ case party of
-        SRecipient -> loadQueuesNoCache " WHERE recipient_id IN ?" $ \(rId, qRec) ->
-          Just . (rId,) <$> mkQ False rId qRec
-        SNotifier -> loadQueuesNoCache " WHERE notifier_id IN ?" $ \(rId, qRec) ->
-          forM (notifier qRec) $ \NtfCreds {notifierId = nId} -> (nId,) <$> mkQ False rId qRec
+        SRecipient -> loadQueues " WHERE recipient_id IN ?" $ \(rId, qRec) ->
+          Just . (rId,) <$> mkQ rId qRec
+        SNotifier -> loadQueues " WHERE notifier_id IN ?" $ \(rId, qRec) ->
+          forM (notifier qRec) $ \NtfCreds {notifierId = nId} -> (nId,) <$> mkQ rId qRec
     where
-      PostgresQueueStore {queues, notifiers, useCache} = st
-      get :: M.Map QueueId a -> QueueId -> QueueId -> Either QueueId a
-      get m qId = maybe (Left qId) Right . (`M.lookup` m)
-      loadQueues :: [Either QueueId q] -> Query -> ((RecipientId, QueueRec) -> IO (Maybe (QueueId, q))) -> IO [Either ErrorType q]
-      loadQueues qs' cond mkCacheQueue = do
-        let qIds' = lefts qs'
-        if null qIds'
-          then pure $ map (first (const INTERNAL)) qs'
-          else do
-            qs_ <- dbLoadQueues qIds' cond mkCacheQueue
-            pure $ map (result qs_) qs'
-        where
-          result :: Either ErrorType (M.Map QueueId q) -> Either QueueId q -> Either ErrorType q
-          result _ (Right q) = Right q
-          result qs_ (Left qId) = maybe (Left AUTH) Right . M.lookup qId =<< qs_
       dbLoadQueues qIds' cond mkQueue' =
         runExceptT $ fmap M.fromList $
           withDB' "getQueues_" st (\db -> DB.query db (queueRecQuery <> cond <> " AND deleted_at IS NULL") (Only (In qIds')))
             >>= liftIO . fmap catMaybes . mapM (mkQueue' . rowToQueueRec)
-      cacheRcvQueue (rId, qRec) = do
-        sq <- mkQ True rId qRec
-        sq' <- withQueueLock sq "getQueue_" $ atomically $
-          -- checking the cache again for concurrent reads, use previously loaded queue if exists.
-          TM.lookup rId queues >>= \case
-            Just sq' -> pure sq'
-            Nothing -> sq <$ TM.insert rId sq queues
-        pure $ Just (rId, sq')
-      loadQueuesNoCache cond mkQueue' = do
+      loadQueues cond mkQueue' = do
         qs_ <- dbLoadQueues qIds cond mkQueue'
         pure $ map (result qs_) qIds
         where
@@ -312,7 +203,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
 
   addQueueLinkData :: PostgresQueueStore q -> q -> LinkId -> QueueLinkData -> IO (Either ErrorType ())
   addQueueLinkData st sq lnkId d =
-    withQueueRec sq "addQueueLinkData" $ \q -> case queueData q of
+    withQueueRec sq $ \q -> case queueData q of
       Nothing ->
         addLink q $ \db -> DB.execute db qry (d :. (lnkId, rId, QMContact))
       Just (lnkId', _) | lnkId' == lnkId ->
@@ -329,7 +220,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
 
   deleteQueueLinkData :: PostgresQueueStore q -> q -> IO (Either ErrorType ())
   deleteQueueLinkData st sq =
-    withQueueRec sq "deleteQueueLinkData" $ \q -> case queueData q of
+    withQueueRec sq $ \q -> case queueData q of
       Just _ -> do
         assertUpdated $ withDB' "deleteQueueLinkData" st $ \db ->
           DB.execute db "UPDATE msg_queues SET link_id = NULL, fixed_data = NULL, user_data = NULL WHERE recipient_id = ? AND deleted_at IS NULL" (Only rId)
@@ -341,7 +232,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
 
   secureQueue :: PostgresQueueStore q -> q -> SndPublicAuthKey -> IO (Either ErrorType ())
   secureQueue st sq sKey =
-    withQueueRec sq "secureQueue" $ \q -> do
+    withQueueRec sq $ \q -> do
       verify q
       assertUpdated $ withDB' "secureQueue" st $ \db ->
         DB.execute db "UPDATE msg_queues SET sender_key = ? WHERE recipient_id = ? AND deleted_at IS NULL AND (sender_key IS NULL OR sender_key = ?)" (sKey, rId, sKey)
@@ -355,7 +246,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
 
   updateKeys :: PostgresQueueStore q -> q -> NonEmpty RcvPublicAuthKey -> IO (Either ErrorType ())
   updateKeys st sq rKeys =
-    withQueueRec sq "updateKeys" $ \q -> do
+    withQueueRec sq $ \q -> do
       assertUpdated $ withDB' "updateKeys" st $ \db ->
         DB.execute db "UPDATE msg_queues SET recipient_keys = ? WHERE recipient_id = ? AND deleted_at IS NULL" (rKeys, rId)
       atomically $ writeTVar (queueRec sq) $ Just q {recipientKeys = rKeys}
@@ -365,24 +256,14 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
 
   addQueueNotifier :: PostgresQueueStore q -> q -> NtfCreds -> IO (Either ErrorType (Maybe NtfCreds))
   addQueueNotifier st sq ntfCreds@NtfCreds {notifierId = nId, notifierKey, rcvNtfDhSecret} =
-    withQueueRec sq "addQueueNotifier" $ \q ->
-      checkCachedNotifier $ do
-        assertUpdated $ withDB "addQueueNotifier" st $ \db ->
-          E.try (update db) >>= bimapM handleDuplicate pure
-        nc_ <- forM (notifier q) $ \nc@NtfCreds {notifierId} -> atomically (TM.delete notifierId notifiers) $> nc
-        let !q' = q {notifier = Just ntfCreds}
-        atomically $ writeTVar (queueRec sq) $ Just q'
-        when useCache $ do
-          atomically $ TM.insert nId rId notifiers
-        withLog "addQueueNotifier" st $ \s -> logAddNotifier s rId ntfCreds
-        pure nc_
+    withQueueRec sq $ \q -> do
+      assertUpdated $ withDB "addQueueNotifier" st $ \db ->
+        E.try (update db) >>= bimapM handleDuplicate pure
+      let !q' = q {notifier = Just ntfCreds}
+      atomically $ writeTVar (queueRec sq) $ Just q'
+      withLog "addQueueNotifier" st $ \s -> logAddNotifier s rId ntfCreds
+      pure $ notifier q
     where
-      checkCachedNotifier add
-        | useCache =
-            ExceptT $ withLockMap (notifierLocks st) nId "addQueueNotifier" $
-              ifM (TM.memberIO nId notifiers) (pure $ Left DUPLICATE_) $ runExceptT add
-        | otherwise = add
-      PostgresQueueStore {notifiers, useCache} = st
       rId = recipientId sq
       update db =
         DB.execute
@@ -396,18 +277,13 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
 
   deleteQueueNotifier :: PostgresQueueStore q -> q -> IO (Either ErrorType (Maybe NtfCreds))
   deleteQueueNotifier st sq =
-    withQueueRec sq "deleteQueueNotifier" $ \q ->
-      ExceptT $ fmap sequence $ forM (notifier q) $ \nc@NtfCreds {notifierId = nId} ->
-        withNotifierLock nId $ runExceptT $ do
-          assertUpdated $ withDB' "deleteQueueNotifier" st update
-          when (useCache st) $ atomically $ TM.delete nId $ notifiers st
-          atomically $ writeTVar (queueRec sq) $ Just q {notifier = Nothing}
-          withLog "deleteQueueNotifier" st (`logDeleteNotifier` rId)
-          pure nc
+    withQueueRec sq $ \q ->
+      forM (notifier q) $ \nc -> do
+        assertUpdated $ withDB' "deleteQueueNotifier" st update
+        atomically $ writeTVar (queueRec sq) $ Just q {notifier = Nothing}
+        withLog "deleteQueueNotifier" st (`logDeleteNotifier` rId)
+        pure nc
     where
-      withNotifierLock nId
-        | useCache st = withLockMap (notifierLocks st) nId "deleteQueueNotifier"
-        | otherwise = id
       rId = recipientId sq
       update db =
         DB.execute
@@ -436,7 +312,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
 
   updateQueueTime :: PostgresQueueStore q -> q -> SystemDate -> IO (Either ErrorType QueueRec)
   updateQueueTime st sq t =
-    withQueueRec sq "updateQueueTime" $ \q@QueueRec {updatedAt} ->
+    withQueueRec sq $ \q@QueueRec {updatedAt} ->
       if updatedAt == Just t
         then pure q
         else do
@@ -449,7 +325,6 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
     where
       rId = recipientId sq
 
-  -- this method is called from JournalMsgStore deleteQueue that already locks the queue
   deleteStoreQueue :: PostgresQueueStore q -> q -> IO (Either ErrorType QueueRec)
   deleteStoreQueue st sq = E.uninterruptibleMask_ $ runExceptT $ do
     q <- ExceptT $ readQueueRecIO qr
@@ -457,11 +332,6 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
     assertUpdated $ withDB' "deleteStoreQueue" st $ \db ->
       DB.execute db "UPDATE msg_queues SET deleted_at = ? WHERE recipient_id = ? AND deleted_at IS NULL" (ts, rId)
     atomically $ writeTVar qr Nothing
-    when (useCache st) $ do
-      atomically $ TM.delete (senderId q) $ senders st
-      forM_ (notifier q) $ \NtfCreds {notifierId} -> do
-        atomically $ TM.delete notifierId $ notifiers st
-        atomically $ TM.delete notifierId $ notifierLocks st
     withLog "deleteStoreQueue" st (`logDeleteQueue` rId)
     pure q
     where
@@ -484,7 +354,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
       pure serviceId
 
   setQueueService :: (PartyI p, ServiceParty p) => PostgresQueueStore q -> q -> SParty p -> Maybe ServiceId -> IO (Either ErrorType ())
-  setQueueService st sq party serviceId = withQueueRec sq "setQueueService" $ \q -> case party of
+  setQueueService st sq party serviceId = withQueueRec sq $ \q -> case party of
     SRecipientService
       | rcvServiceId q == serviceId -> pure ()
       | otherwise -> do
@@ -609,11 +479,6 @@ foldServiceRecs st f =
     DB.fold_ db "SELECT service_id, service_role, service_cert, service_cert_hash, created_at FROM services" mempty $
       \ !acc -> fmap (acc <>) . f . rowToServiceRec
 
-foldRcvServiceQueueRecs :: PostgresQueueStore q -> ServiceId -> (a -> (RecipientId, QueueRec) -> IO a) -> a -> IO (Either ErrorType a)
-foldRcvServiceQueueRecs st serviceId f acc =
-  runExceptT $ withDB' "foldRcvServiceQueueRecs" st $ \db ->
-    DB.fold db (queueRecQuery <> " WHERE rcv_service_id = ? AND deleted_at IS NULL") (Only serviceId) acc $ \a -> f a . rowToQueueRec
-
 foldQueueRecs :: Monoid a => Bool -> Bool -> PostgresQueueStore q -> ((RecipientId, QueueRec) -> IO a) -> IO a
 foldQueueRecs withData = foldQueueRecs_ foldRecs
   where
@@ -622,12 +487,6 @@ foldQueueRecs withData = foldQueueRecs_ foldRecs
       | otherwise = DB.fold_ db (queueRecQuery <> cond) acc $ \acc' -> f' acc' . rowToQueueRec
     cond = " WHERE deleted_at IS NULL ORDER BY recipient_id ASC"
 
-foldRecentQueueRecs :: Monoid a => Int64 -> Bool -> PostgresQueueStore q -> ((RecipientId, QueueRec) -> IO a) -> IO a
-foldRecentQueueRecs old = foldQueueRecs_ foldRecs
-  where
-    foldRecs db acc f' = DB.fold db (queueRecQuery <> cond) (Only old) acc $ \acc' -> f' acc' . rowToQueueRec
-    cond = " WHERE deleted_at IS NULL AND updated_at > ? ORDER BY recipient_id ASC"
-
 foldQueueRecs_ ::
   Monoid a =>
   (DB.Connection -> (Int, a) -> ((Int, a) -> (RecipientId, QueueRec) -> IO (Int, a)) -> IO (Int, a)) ->
@@ -751,15 +610,15 @@ rowToServiceRec (serviceId, serviceRole, serviceCert, Binary fp, serviceCreatedA
 
 setStatusDB :: StoreQueueClass q => Text -> PostgresQueueStore q -> q -> ServerEntityStatus -> ExceptT ErrorType IO () -> IO (Either ErrorType ())
 setStatusDB op st sq status writeLog =
-  withQueueRec sq op $ \q -> do
+  withQueueRec sq $ \q -> do
     assertUpdated $ withDB' op st $ \db ->
       DB.execute db "UPDATE msg_queues SET status = ? WHERE recipient_id = ? AND deleted_at IS NULL" (status, recipientId sq)
     atomically $ writeTVar (queueRec sq) $ Just q {status}
     writeLog
 
-withQueueRec :: StoreQueueClass q => q -> Text -> (QueueRec -> ExceptT ErrorType IO a) -> IO (Either ErrorType a)
-withQueueRec sq op action =
-  withQueueLock sq op $ E.uninterruptibleMask_ $ runExceptT $ ExceptT (readQueueRecIO $ queueRec sq) >>= action
+withQueueRec :: StoreQueueClass q => q -> (QueueRec -> ExceptT ErrorType IO a) -> IO (Either ErrorType a)
+withQueueRec sq action =
+  E.uninterruptibleMask_ $ runExceptT $ ExceptT (readQueueRecIO $ queueRec sq) >>= action
 
 assertUpdated :: ExceptT ErrorType IO Int64 -> ExceptT ErrorType IO ()
 assertUpdated = (>>= \n -> when (n == 0) (throwE AUTH))

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

@@ -18,6 +18,7 @@ module Simplex.Messaging.Server.QueueStore.STM
   ( STMQueueStore (..),
     STMService (..),
     foldRcvServiceQueues,
+    withLoadedQueues,
     setStoreLog,
     withLog',
     readQueueRecIO,
@@ -91,8 +92,6 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
     atomically $ TM.clear senders
     atomically $ TM.clear notifiers
 
-  loadedQueues = queues
-  {-# INLINE loadedQueues #-}
   compactQueues _ = pure 0
   {-# INLINE compactQueues #-}
 
@@ -134,7 +133,7 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
       hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.member notifierId notifiers) notifier
       hasLink = maybe (pure False) (\(lnkId, _) -> TM.member lnkId links) queueData
 
-  getQueue_ :: QueueParty p => STMQueueStore q -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q)
+  getQueue_ :: QueueParty p => STMQueueStore q -> (RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q)
   getQueue_ st _ party qId =
     maybe (Left AUTH) Right <$> case party of
       SRecipient -> TM.lookupIO qId queues
@@ -144,7 +143,7 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
     where
       STMQueueStore {queues, senders, notifiers, links} = st
 
-  getQueues_ :: BatchParty p => STMQueueStore q -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q]
+  getQueues_ :: BatchParty p => STMQueueStore q -> (RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q]
   getQueues_ st _ party qIds = case party of
     SRecipient -> do
       qs <- readTVarIO queues
@@ -379,6 +378,11 @@ foldRcvServiceQueues st serviceId f acc =
   where
     get rId = TM.lookupIO rId (queues st) $>>= \q -> (q,) <$$> readTVarIO (queueRec q)
 
+withLoadedQueues :: Monoid a => STMQueueStore q -> (q -> IO a) -> IO a
+withLoadedQueues st f = readTVarIO (queues st) >>= foldM run mempty
+  where
+    run !acc = fmap (acc <>) . f
+
 withQueueRec :: TVar (Maybe QueueRec) -> (QueueRec -> STM a) -> IO (Either ErrorType a)
 withQueueRec qr a = atomically $ readQueueRec qr >>= mapM a
 

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

@@ -1,5 +1,4 @@
 {-# LANGUAGE AllowAmbiguousTypes #-}
-{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE TypeFamilies #-}
@@ -9,35 +8,29 @@ module Simplex.Messaging.Server.QueueStore.Types
   ( StoreQueueClass (..),
     QueueStoreClass (..),
     EntityCounts (..),
-    withLoadedQueues,
   ) where
 
 import Control.Concurrent.STM
-import Control.Monad
 import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
-import Data.Text (Text)
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.SystemTime
-import Simplex.Messaging.TMap (TMap)
 
 class StoreQueueClass q where
   recipientId :: q -> RecipientId
   queueRec :: q -> TVar (Maybe QueueRec)
-  withQueueLock :: q -> Text -> IO a -> IO a
 
 class StoreQueueClass q => QueueStoreClass q s where
   type QueueStoreCfg s
   newQueueStore :: QueueStoreCfg s -> IO s
   closeQueueStore :: s -> IO ()
   getEntityCounts :: s -> IO EntityCounts
-  loadedQueues :: s -> TMap RecipientId q
   compactQueues :: s -> IO Int64
   addQueue_ :: s -> (RecipientId -> QueueRec -> IO q) -> RecipientId -> QueueRec -> IO (Either ErrorType q)
-  getQueue_ :: QueueParty p => s -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q)
-  getQueues_ :: BatchParty p => s -> (Bool -> RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q]
+  getQueue_ :: QueueParty p => s -> (RecipientId -> QueueRec -> IO q) -> SParty p -> QueueId -> IO (Either ErrorType q)
+  getQueues_ :: BatchParty p => s -> (RecipientId -> QueueRec -> IO q) -> SParty p -> [QueueId] -> IO [Either ErrorType q]
   getQueueLinkData :: s -> q -> LinkId -> IO (Either ErrorType QueueLinkData)
   addQueueLinkData :: s -> q -> LinkId -> QueueLinkData -> IO (Either ErrorType ())
   deleteQueueLinkData :: s -> q -> IO (Either ErrorType ())
@@ -64,8 +57,3 @@ data EntityCounts = EntityCounts
     rcvServiceQueuesCount :: Int,
     ntfServiceQueuesCount :: Int
   }
-
-withLoadedQueues :: (Monoid a, QueueStoreClass q s) => s -> (q -> IO a) -> IO a
-withLoadedQueues st f = readTVarIO (loadedQueues st) >>= foldM run mempty
-  where
-    run !acc = fmap (acc <>) . f

src/Simplex/Messaging/Server/StoreLog/ReadWrite.hs

@@ -23,7 +23,7 @@ import Data.Text.Encoding (decodeLatin1)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ASubscriberParty (..), ErrorType, RecipientId, SParty (..))
 import Simplex.Messaging.Server.QueueStore (QueueRec, ServiceRec (..))
-import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore (..), STMService (..))
+import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore (..), STMService (..), withLoadedQueues)
 import Simplex.Messaging.Server.QueueStore.Types
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.Util (tshow)
@@ -71,7 +71,7 @@ readQueueStore tty mkQ f st = readLogLines tty f $ \_ -> processLine
         withQueue qId op a = runExceptT go >>= qError qId op
           where
             go = do
-              q <- ExceptT $ getQueue_ st (\_ -> mkQ) SRecipient qId
+              q <- ExceptT $ getQueue_ st mkQ SRecipient qId
               liftIO (readTVarIO $ queueRec q) >>= \case
                 Nothing -> logWarn $ logPfx qId op <> "already deleted"
                 Just _ -> void $ ExceptT $ a q

tests/AgentTests/FunctionalAPITests.hs

@@ -2035,7 +2035,7 @@ testOldContactQueueShortLink ps@(_, msType) = withAgentClients2 $ \a b -> do
     ASSCfg _ _ (SSCMemory sp_) -> mapM_ (\StorePaths {storeLogFile} -> updateStoreLog storeLogFile) sp_
 #if defined(dbServerPostgres)
     ASSCfg _ _ (SSCDatabase storeCfg) -> do
-      st :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue (storeCfg, False)
+      st :: PostgresQueueStore PostgresQueue <- newQueueStore @PostgresQueue storeCfg
       updateDbStore st
       closeQueueStore @PostgresQueue st
 #endif

tests/CoreTests/MsgStoreTests.hs

@@ -9,9 +9,7 @@
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE StandaloneDeriving #-}
 {-# LANGUAGE TypeApplications #-}
-{-# OPTIONS_GHC -Wno-orphans #-}
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
 module CoreTests.MsgStoreTests where
@@ -65,7 +63,6 @@ msgStoreTests = do
     someMsgStoreTests cfg = do
       it "should get queue and store/read messages" $ withMsgStore cfg testGetQueue
       it "should write/ack messages" $ withMsgStore cfg testWriteAckMessages
-      it "should not fail on EOF when changing read journal" $ withMsgStore cfg testChangeReadJournal
       it "should not add link data to secured messaging queue" $ withMsgStore cfg testLinkDataSecuredQueue
 
 -- TODO constrain to STM stores?
@@ -193,25 +190,6 @@ testWriteAckMessages ms = do
     void $ ExceptT $ deleteQueue ms q1
     void $ ExceptT $ deleteQueue ms q2
 
-testChangeReadJournal :: MsgStoreClass s => s -> IO ()
-testChangeReadJournal ms = do
-  g <- C.newRandom
-  (rId, qr) <- testNewQueueRec g QMMessaging
-  runRight_ $ do
-    q <- ExceptT $ addQueue ms rId qr
-    let write s = writeMsg ms q True =<< mkMessage s
-    Just (Message {msgId = mId1}, True) <- write "message 1"
-    (Msg "message 1", Nothing) <- tryDelPeekMsg ms q mId1
-    Just (Message {msgId = mId2}, True) <- write "message 2"
-    (Msg "message 2", Nothing) <- tryDelPeekMsg ms q mId2
-    Just (Message {msgId = mId3}, True) <- write "message 3"
-    (Msg "message 3", Nothing) <- tryDelPeekMsg ms q mId3
-    Just (Message {msgId = mId4}, True) <- write "message 4"
-    (Msg "message 4", Nothing) <- tryDelPeekMsg ms q mId4
-    Just (Message {msgId = mId5}, True) <- write "message 5"
-    (Msg "message 5", Nothing) <- tryDelPeekMsg ms q mId5
-    void $ ExceptT $ deleteQueue ms q
-
 testLinkDataSecuredQueue :: MsgStoreClass s => s -> IO ()
 testLinkDataSecuredQueue ms = do
   g <- C.newRandom

tests/CoreTests/StoreLogTests.hs

@@ -189,7 +189,7 @@ testSMPStoreLog testSuite tests =
   where
     testReadWrite SLTC {compacted, state} = do
       st <- newMsgStore testSMTStoreConfig
-      l <- readWriteQueueStore True (mkQueue st True) testStoreLogFile $ queueStore st
+      l <- readWriteQueueStore True (mkQueue st) testStoreLogFile $ queueStore st
       storeState st `shouldReturn` state
       closeStoreLog l
       ([], compacted') <- partitionEithers . map strDecode . B.lines <$> B.readFile testStoreLogFile

tests/SMPProxyTests.hs

@@ -406,12 +406,8 @@ agentViaProxyRetryOffline = do
   where
     withServer :: (ThreadId -> IO a) -> IO a
     withServer = withServer_ testStoreLogFile testStoreMsgsFile testStoreNtfsFile testPort
-    -- TODO [postgres]
-    -- withServer = withServer_ testStoreDBOpts testStoreMsgsDir testStoreNtfsFile testPort
     withServer2 :: (ThreadId -> IO a) -> IO a
     withServer2 = withServer_ testStoreLogFile2 testStoreMsgsFile2 testStoreNtfsFile2 testPort2
-    -- TODO [postgres]
-    -- withServer2 = withServer_ testStoreDBOpts2 testStoreMsgsDir2 testStoreNtfsFile2 testPort2
     withServer_ storeLog storeMsgs storeNtfs =
       let cfg' = updateCfg (memoryCfg proxyCfg storeLog storeMsgs) $ \cfg_ -> cfg_ {storeNtfsFile = Just storeNtfs}
        in withSmpServerConfigOn (transport @TLS) cfg'

tests/ServerTests.hs

@@ -1226,7 +1226,7 @@ testRestoreExpireMessages =
           closeMsgStore ms'
         readWriteQueues = do
           ms <- newMsgStore testSMTStoreConfig
-          readWriteQueueStore True (mkQueue ms True) testStoreLogFile (queueStore ms) >>= closeStoreLog
+          readWriteQueueStore True (mkQueue ms) testStoreLogFile (queueStore ms) >>= closeStoreLog
           removeFileIfExists testStoreMsgsFile
           pure ms
 #else