Merge branch 'queue-journals' into ep/qj-rename

rfcs/2024-11-25-queue-blobs-2.md

@@ -43,9 +43,9 @@ Additional suggestion to reduce probability of queue_state.log and queue_rec.log
 - check the last byte of the file and log EOL if it is not EOL. Probably cleanest approach, but with a small performance cost.
 
 If queue folder is a reference to the queue, it may have one of these files:
-- notifier.id
-- sender.id
-- link.id
+- notifier.ref
+- sender.ref
+- link.ref
 
 These files would contain a one line with the recipient ID of the queue. These files would never change, they can only be deleted when queue is deleted or when notifier/link is deleted.
 

simplexmq.cabal

@@ -302,6 +302,7 @@ library
     build-depends:
         case-insensitive ==1.2.*
       , hashable ==1.4.*
+      , unix ==2.8.*
       , websockets ==0.12.*
   if flag(client_postgres)
     build-depends:

src/Simplex/Messaging/Server.hs

@@ -95,14 +95,13 @@ import Simplex.Messaging.Server.Control
 import Simplex.Messaging.Server.Env.STM as Env
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore
-import Simplex.Messaging.Server.MsgStore.Journal (JournalQueue, closeMsgQueue)
+import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore, JournalQueue, closeMsgQueue)
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.NtfStore
 import Simplex.Messaging.Server.Prometheus
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.QueueInfo
-import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
@@ -391,14 +390,16 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
         threadDelay' interval
         old <- expireBeforeEpoch expCfg
         now <- systemSeconds <$> getSystemTime
+        -- TODO [queues] this should iterate all queues, there are more queues than active queues in journal mode
+        -- TODO [queues] it should also compact journals (see 2024-11-25-journal-expiration.md)
         msgStats@MessageStats {storedMsgsCount = stored, expiredMsgsCount = expired} <-
           withActiveMsgQueues ms $ expireQueueMsgs now ms old
         atomicWriteIORef (msgCount stats) stored
         atomicModifyIORef'_ (msgExpired stats) (+ expired)
         printMessageStats "STORE: messages" msgStats
       where
-        expireQueueMsgs now ms old rId q = fmap (fromRight newMessageStats) . runExceptT $ do
-          (expired_, stored) <- idleDeleteExpiredMsgs now ms rId q old
+        expireQueueMsgs now ms old q = fmap (fromRight newMessageStats) . runExceptT $ do
+          (expired_, stored) <- idleDeleteExpiredMsgs now ms q old
           pure MessageStats {storedMsgsCount = stored, expiredMsgsCount = fromMaybe 0 expired_, storedQueues = 1}
 
     expireNtfsThread :: ServerConfig -> M ()
@@ -429,9 +430,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
       ss@ServerStats {fromTime, qCreated, qSecured, qDeletedAll, qDeletedAllB, qDeletedNew, qDeletedSecured, qSub, qSubAllB, qSubAuth, qSubDuplicate, qSubProhibited, qSubEnd, qSubEndB, ntfCreated, ntfDeleted, ntfDeletedB, ntfSub, ntfSubB, ntfSubAuth, ntfSubDuplicate, msgSent, msgSentAuth, msgSentQuota, msgSentLarge, msgRecv, msgRecvGet, msgGet, msgGetNoMsg, msgGetAuth, msgGetDuplicate, msgGetProhibited, msgExpired, activeQueues, msgSentNtf, msgRecvNtf, activeQueuesNtf, qCount, msgCount, ntfCount, pRelays, pRelaysOwn, pMsgFwds, pMsgFwdsOwn, pMsgFwdsRecv}
         <- asks serverStats
       AMS _ st <- asks msgStore
-      let queues = activeMsgQueues st
-          notifiers = notifiers' st
-          interval = 1000000 * logInterval
+      QueueCounts {queueCount, notifierCount} <- liftIO $ queueCounts st
+      let interval = 1000000 * logInterval
       forever $ do
         withFile statsFilePath AppendMode $ \h -> liftIO $ do
           hSetBuffering h LineBuffering
@@ -484,8 +484,6 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
           pMsgFwdsOwn' <- getResetProxyStatsData pMsgFwdsOwn
           pMsgFwdsRecv' <- atomicSwapIORef pMsgFwdsRecv 0
           qCount' <- readIORef qCount
-          qCount'' <- M.size <$> readTVarIO queues
-          notifierCount' <- M.size <$> readTVarIO notifiers
           msgCount' <- readIORef msgCount
           ntfCount' <- readIORef ntfCount
           hPutStrLn h $
@@ -538,13 +536,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
                        "0", -- dayCount psSub; psSub is removed to reduce memory usage
                        "0", -- weekCount psSub
                        "0", -- monthCount psSub
-                       show qCount'',
+                       show queueCount,
                        show ntfCreated',
                        show ntfDeleted',
                        show ntfSub',
                        show ntfSubAuth',
                        show ntfSubDuplicate',
-                       show notifierCount',
+                       show notifierCount,
                        show qDeletedAllB',
                        show qSubAllB',
                        show qSubEnd',
@@ -581,13 +579,12 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
         rtm <- getRealTimeMetrics env
         T.writeFile metricsFile $ prometheusMetrics sm rtm ts
 
-    getServerMetrics :: STMQueueStore s => s -> ServerStats -> IO ServerMetrics
+    getServerMetrics :: MsgStoreClass s => s -> ServerStats -> IO ServerMetrics
     getServerMetrics st ss = do
       d <- getServerStatsData ss
       let ps = periodStatDataCounts $ _activeQueues d
           psNtf = periodStatDataCounts $ _activeQueuesNtf d
-      queueCount <- M.size <$> readTVarIO (activeMsgQueues st)
-      notifierCount <- M.size <$> readTVarIO (notifiers' st)
+      QueueCounts {queueCount, notifierCount} <- queueCounts st
       pure ServerMetrics {statsData = d, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount}
 
     getRealTimeMetrics :: Env -> IO RealTimeMetrics
@@ -675,9 +672,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
               CPStats -> withUserRole $ do
                 ss <- unliftIO u $ asks serverStats
                 AMS _ st <- unliftIO u $ asks msgStore
-                let queues = activeMsgQueues st
-                    notifiers = notifiers' st
-                    getStat :: (ServerStats -> IORef a) -> IO a
+                QueueCounts {queueCount, notifierCount} <- queueCounts st
+                let getStat :: (ServerStats -> IORef a) -> IO a
                     getStat var = readIORef (var ss)
                     putStat :: Show a => String -> (ServerStats -> IORef a) -> IO ()
                     putStat label var = getStat var >>= \v -> hPutStrLn h $ label <> ": " <> show v
@@ -714,9 +710,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
                 putStat "msgNtfsB" msgNtfsB
                 putStat "msgNtfExpired" msgNtfExpired
                 putStat "qCount" qCount
-                qCount2 <- M.size <$> readTVarIO queues
-                hPutStrLn h $ "qCount 2: " <> show qCount2
-                notifierCount <- M.size <$> readTVarIO notifiers
+                hPutStrLn h $ "qCount 2: " <> show queueCount
                 hPutStrLn h $ "notifiers: " <> show notifierCount
                 putStat "msgCount" msgCount
                 putStat "ntfCount" ntfCount
@@ -852,8 +846,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
               CPDelete sId -> withUserRole $ unliftIO u $ do
                 AMS _ st <- asks msgStore
                 r <- liftIO $ runExceptT $ do
-                  (q, qr) <- ExceptT $ getQueueRec st SSender sId
-                  ExceptT $ deleteQueueSize st (recipientId qr) q
+                  q <- ExceptT $ getQueue st SSender sId
+                  ExceptT $ deleteQueueSize st q
                 case r of
                   Left e -> liftIO $ hPutStrLn h $ "error: " <> show e
                   Right (qr, numDeleted) -> do
@@ -916,7 +910,7 @@ runClientTransport h@THandle {params = thParams@THandleParams {thVersion, sessio
   c <- liftIO $ newClient msType clientId q thVersion sessionId ts
   runClientThreads msType ms active c clientId `finally` clientDisconnected c
   where
-    runClientThreads :: STMQueueStore (MsgStore s) => SMSType s -> MsgStore s -> TVar (IM.IntMap (Maybe AClient)) -> Client (MsgStore s) -> IS.Key -> M ()
+    runClientThreads :: MsgStoreClass (MsgStore s) => SMSType s -> MsgStore s -> TVar (IM.IntMap (Maybe AClient)) -> Client (MsgStore s) -> IS.Key -> M ()
     runClientThreads msType ms active c clientId = do
       atomically $ modifyTVar' active $ IM.insert clientId $ Just (AClient msType c)
       s <- asks server
@@ -972,7 +966,7 @@ cancelSub s = case subThread s of
       _ -> pure ()
   ProhibitSub -> pure ()
 
-receive :: forall c s. (Transport c, STMQueueStore s) => THandleSMP c 'TServer -> s -> Client s -> M ()
+receive :: forall c s. (Transport c, MsgStoreClass s) => THandleSMP c 'TServer -> s -> Client s -> M ()
 receive h@THandle {params = THandleParams {thAuth}} ms Client {rcvQ, sndQ, rcvActiveAt, sessionId} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " receive"
   sa <- asks serverActive
@@ -1072,7 +1066,7 @@ data VerificationResult s = VRVerified (Maybe (StoreQueue s, QueueRec)) | VRFail
 -- - the queue or party key do not exist.
 -- In all cases, the time of the verification should depend only on the provided authorization type,
 -- a dummy key is used to run verification in the last two cases, and failure is returned irrespective of the result.
-verifyTransmission :: forall s. STMQueueStore s => s -> Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> QueueId -> Cmd -> M (VerificationResult s)
+verifyTransmission :: forall s. MsgStoreClass s => s -> Maybe (THandleAuth 'TServer, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> QueueId -> Cmd -> M (VerificationResult s)
 verifyTransmission ms auth_ tAuth authorized queueId cmd =
   case cmd of
     Cmd SRecipient (NEW k _ _ _ _) -> pure $ Nothing `verifiedWith` k
@@ -1149,7 +1143,7 @@ forkClient Client {endThreads, endThreadSeq} label action = do
     action `finally` atomically (modifyTVar' endThreads $ IM.delete tId)
   mkWeakThreadId t >>= atomically . modifyTVar' endThreads . IM.insert tId
 
-client :: forall s. STMQueueStore s => THandleParams SMPVersion 'TServer -> Server -> s -> Client s -> M ()
+client :: forall s. MsgStoreClass s => THandleParams SMPVersion 'TServer -> Server -> s -> Client s -> M ()
 client
   thParams'
   Server {subscribedQ, ntfSubscribedQ, subscribers}
@@ -1282,10 +1276,9 @@ client
           updatedAt <- Just <$> liftIO getSystemDate
           let rcvDhSecret = C.dh' dhKey privDhKey
               qik (rcvId, sndId) = QIK {rcvId, sndId, rcvPublicDhKey, sndSecure}
-              qRec (recipientId, senderId) =
+              qRec senderId =
                 QueueRec
-                  { recipientId,
-                    senderId,
+                  { senderId,
                     recipientKey,
                     rcvDhSecret,
                     senderKey = Nothing,
@@ -1297,12 +1290,12 @@ client
           (corrId,entId,) <$> addQueueRetry 3 qik qRec
           where
             addQueueRetry ::
-              Int -> ((RecipientId, SenderId) -> QueueIdsKeys) -> ((RecipientId, SenderId) -> QueueRec) -> M BrokerMsg
+              Int -> ((RecipientId, SenderId) -> QueueIdsKeys) -> (SenderId -> QueueRec) -> M BrokerMsg
             addQueueRetry 0 _ _ = pure $ ERR INTERNAL
             addQueueRetry n qik qRec = do
-              ids <- getIds
-              let qr = qRec ids
-              liftIO (addQueue ms qr) >>= \case
+              ids@(rId, sId) <- getIds
+              let qr = qRec sId
+              liftIO (addQueue ms rId qr) >>= \case
                 Left DUPLICATE_ -> addQueueRetry (n - 1) qik qRec
                 Left e -> pure $ ERR e
                 Right q -> do
@@ -1379,7 +1372,7 @@ client
                   incStat $ qSubDuplicate stats
                   atomically (tryTakeTMVar $ delivered s) >> deliver False s
           where
-            rId = recipientId qr
+            rId = recipientId' q
             newSub :: M Sub
             newSub = time "SUB newSub" . atomically $ do
               writeTQueue subscribedQ (rId, clientId, True)
@@ -1390,7 +1383,7 @@ client
             deliver inc sub = do
               stats <- asks serverStats
               fmap (either (\e -> (corrId, rId, ERR e)) id) $ liftIO $ runExceptT $ do
-                msg_ <- tryPeekMsg ms rId q
+                msg_ <- tryPeekMsg ms q
                 liftIO $ when (inc && isJust msg_) $ incStat (qSub stats)
                 liftIO $ deliverMessage "SUB" qr rId sub msg_
 
@@ -1424,7 +1417,7 @@ client
             getMessage_ s delivered_ = do
               stats <- asks serverStats
               fmap (either err id) $ liftIO $ runExceptT $
-                tryPeekMsg ms (recipientId qr) q >>= \case
+                tryPeekMsg ms q >>= \case
                   Just msg -> do
                     let encMsg = encryptMsg qr msg
                     incStat $ (if isJust delivered_ then msgGetDuplicate else msgGet) stats
@@ -1471,11 +1464,11 @@ client
                   fmap (either err id) $ liftIO $ runExceptT $ do
                     case st of
                       ProhibitSub -> do
-                        deletedMsg_ <- tryDelMsg ms (recipientId qr) q msgId
+                        deletedMsg_ <- tryDelMsg ms q msgId
                         liftIO $ mapM_ (updateStats stats True) deletedMsg_
                         pure ok
                       _ -> do
-                        (deletedMsg_, msg_) <- tryDelPeekMsg ms (recipientId qr) q msgId
+                        (deletedMsg_, msg_) <- tryDelPeekMsg ms q msgId
                         liftIO $ mapM_ (updateStats stats False) deletedMsg_
                         liftIO $ deliverMessage "ACK" qr entId sub msg_
                 _ -> pure $ err NO_MSG
@@ -1529,7 +1522,7 @@ client
                       msg_ <- liftIO $ time "SEND" $ runExceptT $ do
                         expireMessages messageExpiration stats
                         msg <- liftIO $ mkMessage msgId body
-                        writeMsg ms (recipientId qr) q True msg
+                        writeMsg ms q True msg
                       case msg_ of
                         Left e -> pure $ err e
                         Right Nothing -> do
@@ -1540,10 +1533,10 @@ client
                           when (notification msgFlags) $ do
                             mapM_ (`enqueueNotification` msg) (notifier qr)
                             incStat $ msgSentNtf stats
-                            liftIO $ updatePeriodStats (activeQueuesNtf stats) (recipientId qr)
+                            liftIO $ updatePeriodStats (activeQueuesNtf stats) (recipientId' q)
                           incStat $ msgSent stats
                           incStat $ msgCount stats
-                          liftIO $ updatePeriodStats (activeQueues stats) (recipientId qr)
+                          liftIO $ updatePeriodStats (activeQueues stats) (recipientId' q)
                           pure ok
           where
             mkMessage :: MsgId -> C.MaxLenBS MaxMessageLen -> IO Message
@@ -1553,7 +1546,7 @@ client
 
             expireMessages :: Maybe ExpirationConfig -> ServerStats -> ExceptT ErrorType IO ()
             expireMessages msgExp stats = do
-              deleted <- maybe (pure 0) (deleteExpiredMsgs ms (recipientId qr) q <=< liftIO . expireBeforeEpoch) msgExp
+              deleted <- maybe (pure 0) (deleteExpiredMsgs ms q <=< liftIO . expireBeforeEpoch) msgExp
               liftIO $ when (deleted > 0) $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
 
             -- The condition for delivery of the message is:
@@ -1571,7 +1564,7 @@ client
               whenM (TM.memberIO rId subscribers) $
                 atomically deliverToSub >>= mapM_ forkDeliver
               where
-                rId = recipientId qr
+                rId = recipientId' q
                 deliverToSub =
                   -- lookup has ot be in the same transaction,
                   -- so that if subscription ends, it re-evalutates
@@ -1715,7 +1708,7 @@ client
 
         delQueueAndMsgs :: (StoreQueue s, QueueRec) -> M (Transmission BrokerMsg)
         delQueueAndMsgs (q, _) = do
-          liftIO (deleteQueue ms entId q) >>= \case
+          liftIO (deleteQueue ms q) >>= \case
             Right qr -> do
               -- Possibly, the same should be done if the queue is suspended, but currently we do not use it
               atomically $ do
@@ -1735,11 +1728,11 @@ client
             Left e -> pure $ err e
 
         getQueueInfo :: StoreQueue s -> QueueRec -> M BrokerMsg
-        getQueueInfo q QueueRec {recipientId = rId, senderKey, notifier} = do
+        getQueueInfo q QueueRec {senderKey, notifier} = do
           fmap (either ERR id) $ liftIO $ runExceptT $ do
             qiSub <- liftIO $ TM.lookupIO entId subscriptions >>= mapM mkQSub
-            qiSize <- getQueueSize ms rId q
-            qiMsg <- toMsgInfo <$$> tryPeekMsg ms rId q
+            qiSize <- getQueueSize ms q
+            qiMsg <- toMsgInfo <$$> tryPeekMsg ms q
             let info = QueueInfo {qiSnd = isJust senderKey, qiNtf = isJust notifier, qiSub, qiSize, qiMsg}
             pure $ INFO info
           where
@@ -1797,7 +1790,7 @@ saveServerMessages drainMsgs = \case
   AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
     Just f -> exportMessages False ms f drainMsgs
     Nothing -> logInfo "undelivered messages are not saved"
-  AMS SMSJournal _ -> logInfo "closed journal message storage"
+  AMS _ _ -> logInfo "closed journal message storage"
 
 exportMessages :: MsgStoreClass s => Bool -> s -> FilePath -> Bool -> IO ()
 exportMessages tty ms f drainMsgs = do
@@ -1809,8 +1802,9 @@ exportMessages tty ms f drainMsgs = do
         logError $ "error exporting messages: " <> tshow e
         exitFailure
   where
-    saveQueueMsgs h rId q =
-      runExceptT (getQueueMessages drainMsgs ms rId q) >>= \case
+    saveQueueMsgs h q = do
+      let rId = recipientId' q
+      runExceptT (getQueueMessages drainMsgs ms q) >>= \case
         Right msgs -> Sum (length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs)
         Left e -> do
           logError $ "STORE: saveQueueMsgs, error exporting messages from queue " <> decodeLatin1 (strEncode rId) <> ", " <> tshow e
@@ -1828,17 +1822,21 @@ processServerMessages = do
         AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
           Just f -> ifM (doesFileExist f) (Just <$> importMessages False ms f old_) (pure Nothing)
           Nothing -> pure Nothing
-        AMS SMSJournal ms
-          | expire -> Just <$> case old_ of
-              Just old -> do
-                logInfo "expiring journal store messages..."
-                withAllMsgQueues False ms $ processExpireQueue old
-              Nothing -> do
-                logInfo "validating journal store messages..."
-                withAllMsgQueues False ms $ processValidateQueue
-          | otherwise -> logWarn "skipping message expiration" $> Nothing
-          where
-            processExpireQueue old rId q =
+        AMS SMSHybrid ms -> processJournalMessages old_ expire ms
+        AMS SMSJournal ms -> processJournalMessages old_ expire ms
+      processJournalMessages :: forall s. JournalStoreType s => Maybe Int64 -> Bool -> JournalMsgStore s -> IO (Maybe MessageStats)
+      processJournalMessages old_ expire ms
+        | expire = Just <$> case old_ of
+            Just old -> do
+              logInfo "expiring journal store messages..."
+              withAllMsgQueues False ms $ processExpireQueue old
+            Nothing -> do
+              logInfo "validating journal store messages..."
+              withAllMsgQueues False ms $ processValidateQueue
+        | otherwise = logWarn "skipping message expiration" $> Nothing
+        where
+            processExpireQueue :: Int64 -> JournalQueue s -> IO MessageStats
+            processExpireQueue old q =
               runExceptT expireQueue >>= \case
                 Right (storedMsgsCount, expiredMsgsCount) ->
                   pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1}
@@ -1847,20 +1845,19 @@ processServerMessages = do
                   exitFailure
               where
                 expireQueue = do
-                  expired'' <- deleteExpiredMsgs ms rId q old
-                  stored'' <- getQueueSize ms rId q
+                  expired'' <- deleteExpiredMsgs ms q old
+                  stored'' <- getQueueSize ms q
                   liftIO $ closeMsgQueue q
                   pure (stored'', expired'')
-            processValidateQueue :: RecipientId -> JournalQueue -> IO MessageStats
-            processValidateQueue rId q =
-              runExceptT (getQueueSize ms rId q) >>= \case
+            processValidateQueue :: JournalQueue s -> IO MessageStats
+            processValidateQueue q =
+              runExceptT (getQueueSize ms q) >>= \case
                 Right storedMsgsCount -> pure newMessageStats {storedMsgsCount, storedQueues = 1}
                 Left e -> do
                   logError $ "STORE: processValidateQueue, failed opening message queue, " <> tshow e
                   exitFailure
 
--- TODO this function should be called after importing queues from store log
-importMessages :: forall s. STMQueueStore s => Bool -> s -> FilePath -> Maybe Int64 -> IO MessageStats
+importMessages :: forall s. MsgStoreClass s => Bool -> s -> FilePath -> Maybe Int64 -> IO MessageStats
 importMessages tty ms f old_ = do
   logInfo $ "restoring messages from file " <> T.pack f
   LB.readFile f >>= runExceptT . foldM restoreMsg (0, Nothing, (0, 0, M.empty)) . LB.lines >>= \case
@@ -1873,8 +1870,8 @@ importMessages tty ms f old_ = do
       renameFile f $ f <> ".bak"
       mapM_ setOverQuota_ overQuota
       logQueueStates ms
-      storedQueues <- M.size <$> readTVarIO (activeMsgQueues ms)
-      pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues}
+      QueueCounts {queueCount} <- queueCounts ms
+      pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = queueCount}
   where
     progress i = "Processed " <> show i <> " lines"
     restoreMsg :: (Int, Maybe (RecipientId, StoreQueue s), (Int, Int, M.Map RecipientId (StoreQueue s))) -> LB.ByteString -> ExceptT String IO (Int, Maybe (RecipientId, StoreQueue s), (Int, Int, M.Map RecipientId (StoreQueue s)))
@@ -1892,7 +1889,7 @@ importMessages tty ms f old_ = do
           (i + 1,Just (rId, q),) <$> case msg of
             Message {msgTs}
               | maybe True (systemSeconds msgTs >=) old_ -> do
-                  writeMsg ms rId q False msg >>= \case
+                  writeMsg ms q False msg >>= \case
                     Just _ -> pure (stored + 1, expired, overQuota)
                     Nothing -> do
                       logError $ decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (messageId msg)
@@ -1901,11 +1898,11 @@ importMessages tty ms f old_ = do
             MessageQuota {} ->
               -- queue was over quota at some point,
               -- it will be set as over quota once fully imported
-              mergeQuotaMsgs >> writeMsg ms rId q False msg $> (stored, expired, M.insert rId q overQuota)
+              mergeQuotaMsgs >> writeMsg ms q False msg $> (stored, expired, M.insert rId q overQuota)
               where
                 -- if the first message in queue head is "quota", remove it.
                 mergeQuotaMsgs =
-                  withPeekMsgQueue ms rId q "mergeQuotaMsgs" $ maybe (pure ()) $ \case
+                  withPeekMsgQueue ms q "mergeQuotaMsgs" $ maybe (pure ()) $ \case
                     (mq, MessageQuota {}) -> tryDeleteMsg_ q mq False
                     _ -> pure ()
         msgErr :: Show e => String -> e -> String
@@ -1982,7 +1979,7 @@ restoreServerStats msgStats_ ntfStats = asks (serverStatsBackupFile . config) >>
         Right d@ServerStatsData {_qCount = statsQCount, _msgCount = statsMsgCount, _ntfCount = statsNtfCount} -> do
           s <- asks serverStats
           AMS _ st <- asks msgStore
-          _qCount <- M.size <$> readTVarIO (activeMsgQueues st)
+          QueueCounts {queueCount = _qCount} <- liftIO $ queueCounts st
           let _msgCount = maybe statsMsgCount storedMsgsCount msgStats_
               _ntfCount = storedMsgsCount ntfStats
               _msgExpired' = _msgExpired d + maybe 0 expiredMsgsCount msgStats_

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

@@ -18,6 +18,7 @@ import Control.Monad
 import qualified Crypto.PubKey.RSA as RSA
 import Crypto.Random
 import Data.ByteString.Char8 (ByteString)
+import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.IntMap.Strict (IntMap)
 import qualified Data.IntMap.Strict as IM
@@ -187,9 +188,10 @@ data Env = Env
 
 type family MsgStore s where
   MsgStore 'MSMemory = STMMsgStore
-  MsgStore 'MSJournal = JournalMsgStore
+  MsgStore 'MSHybrid = JournalMsgStore 'MSHybrid
+  MsgStore 'MSJournal = JournalMsgStore 'MSJournal
 
-data AMsgStore = forall s. (STMQueueStore (MsgStore s), MsgStoreClass (MsgStore s)) => AMS (SMSType s) (MsgStore s)
+data AMsgStore = forall s. MsgStoreClass (MsgStore s) => AMS (SMSType s) (MsgStore s)
 
 data AStoreQueue = forall s. MsgStoreClass (MsgStore s) => ASQ (SMSType s) (StoreQueue (MsgStore s))
 
@@ -197,7 +199,7 @@ data AMsgStoreCfg = forall s. MsgStoreClass (MsgStore s) => AMSC (SMSType s) (Ms
 
 msgPersistence :: AMsgStoreCfg -> Bool
 msgPersistence (AMSC SMSMemory (STMStoreConfig {storePath})) = isJust storePath
-msgPersistence (AMSC SMSJournal _) = True
+msgPersistence _ = True
 
 type Subscribed = Bool
 
@@ -294,19 +296,9 @@ newEnv :: ServerConfig -> IO Env
 newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgStoreType, storeMsgsFile, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines} = do
   serverActive <- newTVarIO True
   server <- newServer
-  msgStore@(AMS _ store) <- case msgStoreType of
-    AMSType SMSMemory -> AMS SMSMemory <$> newMsgStore STMStoreConfig {storePath = storeMsgsFile, quota = msgQueueQuota}
-    AMSType SMSJournal -> case storeMsgsFile of
-      Just storePath -> 
-        let cfg = JournalStoreConfig {storePath, quota = msgQueueQuota, pathParts = journalMsgStoreDepth, maxMsgCount = maxJournalMsgCount, maxStateLines = maxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval = idleQueueInterval}
-         in AMS SMSJournal <$> newMsgStore cfg
-      Nothing -> putStrLn "Error: journal msg store require path in [STORE_LOG], restore_messages" >> exitFailure
+  msgStore <- createMsgStore
   ntfStore <- NtfStore <$> TM.emptyIO
   random <- C.newRandom
-  forM_ storeLogFile $ \f -> do
-    logInfo $ "restoring queues from file " <> T.pack f
-    sl <- readWriteQueueStore f store
-    setStoreLog store sl
   tlsServerCreds <- getCredentials "SMP" smpCredentials
   httpServerCreds <- mapM (getCredentials "HTTPS") httpCredentials
   mapM_ checkHTTPSCredentials httpServerCreds
@@ -319,6 +311,26 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgSt
   proxyAgent <- newSMPProxyAgent smpAgentCfg random
   pure Env {serverActive, config, serverInfo, server, serverIdentity, msgStore, ntfStore, random, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
   where
+    createMsgStore :: IO AMsgStore
+    createMsgStore = case (msgStoreType, storeMsgsFile) of
+      (AMSType SMSMemory, _) -> do
+        st <- newMsgStore STMStoreConfig {storePath = storeLogFile, quota = msgQueueQuota}
+        loadStoreLog st $> AMS SMSMemory st
+      (AMSType SMSHybrid, Just storePath) -> do
+        st <- newMsgStore $ storeCfg SMSHybrid storePath
+        loadStoreLog st $> AMS SMSHybrid st
+      (AMSType SMSJournal, Just storePath) ->
+        AMS SMSJournal <$> newMsgStore (storeCfg SMSJournal storePath)
+      (_, Nothing) -> putStrLn "Error: journal msg store requires that restore_messages is enabled in [STORE_LOG]" >> exitFailure
+      where
+        storeCfg :: SMSType s -> FilePath -> JournalStoreConfig s
+        storeCfg queueStoreType storePath =
+          JournalStoreConfig {storePath, quota = msgQueueQuota, pathParts = journalMsgStoreDepth, queueStoreType, maxMsgCount = maxJournalMsgCount, maxStateLines = maxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval = idleQueueInterval}
+        loadStoreLog :: STMStoreClass s => s -> IO ()
+        loadStoreLog st = forM_ storeLogFile $ \f -> do
+          logInfo $ "restoring queues from file " <> T.pack f
+          sl <- readWriteQueueStore f st
+          setStoreLog (stmQueueStore st) sl
     getCredentials protocol creds = do
       files <- missingCreds
       unless (null files) $ do
@@ -362,5 +374,5 @@ newSMPProxyAgent smpAgentCfg random = do
   smpAgent <- newSMPClientAgent smpAgentCfg random
   pure ProxyAgent {smpAgent}
 
-readWriteQueueStore :: STMQueueStore s => FilePath -> s -> IO (StoreLog 'WriteMode)
+readWriteQueueStore :: STMStoreClass s => FilePath -> s -> IO (StoreLog 'WriteMode)
 readWriteQueueStore = readWriteStoreLog readQueueStore writeQueueStore

src/Simplex/Messaging/Server/Main.hs

@@ -19,12 +19,12 @@ import Control.Logger.Simple
 import Control.Monad
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Char (isAlpha, isAscii, toUpper)
-import Data.Either (fromRight)
+import Data.Char (isAlpha, isAscii, toLower, toUpper)
 import Data.Functor (($>))
 import Data.Ini (Ini, lookupValue, readIniFile)
 import Data.List (find, isPrefixOf)
 import qualified Data.List.NonEmpty as L
+import qualified Data.Map.Strict as M
 import Data.Maybe (fromMaybe, isJust, isNothing)
 import Data.Text (Text)
 import qualified Data.Text as T
@@ -44,14 +44,16 @@ import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
-import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..))
-import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore)
+import Simplex.Messaging.Server.MsgStore.Journal (JournalMsgStore (..), JournalStoreConfig (..))
+import Simplex.Messaging.Server.MsgStore.STM (STMStoreConfig (..))
+import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore.STM (readQueueStore)
+import Simplex.Messaging.Server.StoreLog (openWriteStoreLog, logCreateQueue)
 import Simplex.Messaging.Transport (simplexMQVersion, supportedProxyClientSMPRelayVRange, supportedServerSMPRelayVRange)
 import Simplex.Messaging.Transport.Client (SocksProxy, TransportHost (..), defaultSocksProxy)
 import Simplex.Messaging.Transport.Server (ServerCredentials (..), TransportServerConfig (..), defaultTransportServerConfig)
-import Simplex.Messaging.Util (eitherToMaybe, ifM, safeDecodeUtf8, tshow)
-import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist)
+import Simplex.Messaging.Util (eitherToMaybe, safeDecodeUtf8, tshow)
+import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist, renameFile)
 import System.Exit (exitFailure)
 import System.FilePath (combine)
 import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
@@ -84,54 +86,138 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
       putStrLn "Deleted configuration and log files"
     Journal cmd -> withIniFile $ \ini -> do
       msgsDirExists <- doesDirectoryExist storeMsgsJournalDir
+      qsFileExists <- doesFileExist storeLogFilePath
       msgsFileExists <- doesFileExist storeMsgsFilePath
       let enableStoreLog = settingIsOn "STORE_LOG" "enable" ini
-      storeLogFile <- case enableStoreLog $> storeLogFilePath of
-        Just storeLogFile -> do
-          ifM
-            (doesFileExist storeLogFile)
-            (pure storeLogFile)
-            (putStrLn ("Store log file " <> storeLogFile <> " not found") >> exitFailure)
+      case enableStoreLog $> storeLogFilePath of
+        Just storeLogFile ->
+          unless qsFileExists $
+            putStrLn ("Store log file " <> storeLogFile <> " not found") >> exitFailure
         Nothing -> putStrLn "Store log disabled, see `[STORE_LOG] enable`" >> exitFailure
       case cmd of
-        JCImport
+        JCImport sCmd
           | msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
-          | msgsDirExists -> do
-              putStrLn $ storeMsgsJournalDir <> " directory already exists."
-              exitFailure
-          | not msgsFileExists -> do
-              putStrLn $ storeMsgsFilePath <> " file does not exists."
-              exitFailure
-          | otherwise -> do
-              confirmOrExit
-                ("WARNING: message log file " <> storeMsgsFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir)
-                "Messages not imported"
-              ms <- newJournalMsgStore
-              readQueueStore storeLogFile ms
-              msgStats <- importMessages True ms storeMsgsFilePath Nothing -- no expiration
-              putStrLn "Import completed"
-              printMessageStats "Messages" msgStats
-              putStrLn $ case readMsgStoreType ini of
-                Right (AMSType SMSMemory) -> "store_messages set to `memory`, update it to `journal` in INI file"
-                Right (AMSType SMSJournal) -> "store_messages set to `journal`"
-                Left e -> e <> ", update it to `journal` in INI file"
-        JCExport
+          | otherwise -> case sCmd of
+              Just JSCMessages -- TODO deprecated, remove in v6.3
+                | msgsDirExists -> putStrLn ("Directory already exists: " <> storeMsgsJournalDir) >> exitFailure
+                | not msgsFileExists -> putStrLn ("File does not exists: " <> storeMsgsFilePath) >> exitFailure
+                | otherwise -> do
+                    confirmOrExit
+                      ("WARNING: this command is deprecated.\nMessage log file " <> storeMsgsFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir)
+                      "Messages not imported"
+                    ms <- newJournalMsgStore SMSHybrid
+                    readQueueStore storeLogFilePath ms
+                    msgStats <- importMessages True ms storeMsgsFilePath Nothing -- no expiration
+                    putStrLn "Import of messages completed"
+                    printMessageStats "Messages" msgStats
+                    putStrLn $ case readMsgStoreType ini of
+                      Right (AMSType SMSMemory) -> "store_messages set to `memory`, update it to `journal` in INI file"
+                      Right (AMSType SMSHybrid) -> "store_messages set to `journal`, all correct"
+                      Right (AMSType SMSJournal) -> "store_messages and store_queues set to `journal`, it is incorrect as only messages were imported"
+                      Left e -> e <> ", update it to `journal` in INI file"
+              Just JSCQueues -- TODO deprecated, remove in v6.3
+                | not msgsDirExists -> putStrLn ("Directory must exists to use this command: " <> storeMsgsJournalDir <> "\nUse `journal import` instead.") >> exitFailure
+                | otherwise -> do
+                    confirmOrExit
+                      ("WARNING: message queues log file " <> storeLogFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir)
+                      "Queues not imported"
+                    void importQueueStoreLog
+                    putStrLn "Import of queues completed"
+                    putStrLn importStoreSettings
+              Nothing
+                | msgsDirExists -> putStrLn ("Directory already exists: " <> storeMsgsJournalDir) >> exitFailure
+                | not msgsFileExists -> putStrLn ("File does not exists: " <> storeMsgsFilePath) >> exitFailure
+                | otherwise -> do
+                    confirmOrExit
+                      ("WARNING: message queues log file " <> storeLogFilePath <> " and message log file " <> storeMsgsFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir)
+                      "Queues and messages not imported"
+                    ms <- importQueueStoreLog
+                    msgStats <- importMessages True ms storeMsgsFilePath Nothing -- no expiration
+                    putStrLn "Import of queues and messages completed"
+                    printMessageStats "Messages" msgStats
+                    putStrLn importStoreSettings
+          where
+            importQueueStoreLog = do
+              putStrLn $ "restoring queues from file " <> storeLogFilePath
+              st <- newMsgStore STMStoreConfig {storePath = Just storeLogFilePath, quota = defaultMsgQueueQuota}
+              readQueueStore storeLogFilePath st
+              ms <- newJournalMsgStore SMSJournal
+              writeJournalQueues st ms
+              pure ms
+              where
+                writeJournalQueues st ms = do
+                  putStrLn $ "saving queues to journal directory " <> storeMsgsJournalDir
+                  let qs = queues $ stmQueueStore st
+                  readTVarIO qs >>= mapM_ (writeQueue ms) . M.assocs
+                  renameFile storeLogFilePath (storeLogFilePath <> ".bak")
+                writeQueue ms (rId, q) =
+                  readTVarIO (queueRec' q) >>= \case
+                    Just q' ->
+                      addQueue ms rId q' >>= \case
+                        Right _ -> pure ()
+                        Left e -> do
+                          putStrLn $ "error saving queue " <> B.unpack (strEncode rId) <> ": " <> show e
+                          exitFailure
+                    _ -> putStrLn $ "skipping suspended queue " <> B.unpack (strEncode rId)
+            importStoreSettings = case readMsgStoreType ini of
+              Right (AMSType SMSMemory) -> "store_messages set to `memory`, set store_messages and store_queues to `journal` in INI file"
+              Right (AMSType SMSHybrid) -> "store_messages set to `journal`, set store_queues to `journal` in INI file"
+              Right (AMSType SMSJournal) -> "store_messages and store_queues set to `journal`, all correct"
+              Left e -> e <> ", update it to `journal` in INI file"
+        JCExport sCmd
           | msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
-          | msgsFileExists -> do
-              putStrLn $ storeMsgsFilePath <> " file already exists."
-              exitFailure
-          | otherwise -> do
-              confirmOrExit
-                ("WARNING: journal directory " <> storeMsgsJournalDir <> " will be exported to message log file " <> storeMsgsFilePath)
-                "Journal not exported"
-              ms <- newJournalMsgStore
-              readQueueStore storeLogFile ms
-              exportMessages True ms storeMsgsFilePath False
-              putStrLn "Export completed"
-              putStrLn $ case readMsgStoreType ini of
-                Right (AMSType SMSMemory) -> "store_messages set to `memory`"
-                Right (AMSType SMSJournal) -> "store_messages set to `journal`, update it to `memory` in INI file"
-                Left e -> e <> ", update it to `memory` in INI file"
+          | not msgsDirExists -> putStrLn ("Directory does not exist: " <> storeMsgsJournalDir) >> exitFailure
+          | otherwise -> case sCmd of
+              Just JSCMessages -- TODO deprecated, remove in v6.3
+                | msgsFileExists -> putStrLn ("File already exists: " <> storeMsgsFilePath) >> exitFailure
+                | otherwise -> do
+                    confirmOrExit
+                      ("WARNING: messages from journal directory " <> storeMsgsJournalDir <> " will be exported to message log file " <> storeMsgsFilePath)
+                      "Journal messages not exported"
+                    ms <- newJournalMsgStore SMSHybrid
+                    readQueueStore storeLogFilePath ms
+                    exportMessages True ms storeMsgsFilePath False
+                    putStrLn "Export of messages completed"
+                    putStrLn exportStoreSettings
+              Just JSCQueues -- TODO deprecated, remove in v6.3
+                | qsFileExists -> putStrLn ("File already exists: " <> storeLogFilePath) >> exitFailure
+                | otherwise -> do
+                    confirmOrExit
+                      ("WARNING: queues from journal directory " <> storeMsgsJournalDir <> " will be exported to queue store log file " <> storeLogFilePath)
+                      "Journal queues not exported"
+                    ms <- newJournalMsgStore SMSJournal
+                    exportQueueStoreLog ms
+                    putStrLn "Export of queues completed"
+                    putStrLn $ case readMsgStoreType ini of
+                      Right (AMSType SMSMemory) -> "store_messages set to `memory`, update it to `journal` in INI file"
+                      Right (AMSType SMSHybrid) -> "store_messages set to `journal`, all correct"
+                      Right (AMSType SMSJournal) -> "store_queues set to `journal`, update it to `memory` in INI file"
+                      Left e -> e <> ", set store_messages to `journal` and `store_queues` to `memory` in INI file"
+              Nothing
+                | qsFileExists -> putStrLn (storeLogFilePath <> " file already exists.") >> exitFailure
+                | msgsFileExists -> putStrLn (storeMsgsFilePath <> " file already exists.") >> exitFailure
+                | otherwise -> do
+                    confirmOrExit
+                      ("WARNING: queues and messages from journal directory " <> storeMsgsJournalDir <> " will be exported to queue store log file " <> storeLogFilePath <> " and to message log file " <> storeMsgsFilePath)
+                      "Journal queues and messages not exported"
+                    ms <- newJournalMsgStore SMSJournal
+                    exportQueueStoreLog ms
+                    exportMessages True ms storeMsgsFilePath False
+                    putStrLn "Export of queues and messages completed"
+                    putStrLn exportStoreSettings
+          where
+            exportQueueStoreLog ms = do
+              s <- openWriteStoreLog storeLogFilePath
+              withAllMsgQueues True ms $ \q -> do
+                let rId = recipientId' q
+                readTVarIO (queueRec' q) >>= \case
+                  Just q' -> logCreateQueue s rId q' -- TODO we should log suspended queues when we use them
+                  Nothing -> putStrLn $ "WARN: deleted queue " <> B.unpack (strEncode rId) <> ", verify the journal folder"
+            exportStoreSettings = case readMsgStoreType ini of
+              Right (AMSType SMSMemory) -> "store_messages set to `memory`, all correct"
+              Right (AMSType SMSHybrid) -> "store_messages set to `journal`, update it to `memory` in INI file"
+              Right (AMSType SMSJournal) -> "store_messages and store_queues set to `journal`, update it to `memory` in INI file"
+              Left e -> e <> ", update it to `memory` in INI file"
         JCDelete
           | not msgsDirExists -> do
               putStrLn $ storeMsgsJournalDir <> " directory does not exists."
@@ -147,7 +233,8 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
       doesFileExist iniFile >>= \case
         True -> readIniFile iniFile >>= either exitError a
         _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
-    newJournalMsgStore = newMsgStore JournalStoreConfig {storePath = storeMsgsJournalDir, pathParts = journalMsgStoreDepth, quota = defaultMsgQueueQuota, maxMsgCount = defaultMaxJournalMsgCount, maxStateLines = defaultMaxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval = checkInterval defaultMessageExpiration}
+    newJournalMsgStore :: JournalStoreType s => SMSType s -> IO (JournalMsgStore s)
+    newJournalMsgStore queueStoreType = newMsgStore JournalStoreConfig {storePath = storeMsgsJournalDir, pathParts = journalMsgStoreDepth, queueStoreType, quota = defaultMsgQueueQuota, maxMsgCount = defaultMaxJournalMsgCount, maxStateLines = defaultMaxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval = checkInterval defaultMessageExpiration}
     iniFile = combine cfgPath "smp-server.ini"
     serverVersion = "SMP server v" <> simplexMQVersion
     defaultServerPorts = "5223,443"
@@ -157,48 +244,92 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
     storeMsgsJournalDir = combine logPath "messages"
     storeNtfsFilePath = combine logPath "smp-server-ntfs.log"
     readMsgStoreType :: Ini -> Either String AMSType
-    readMsgStoreType = textToMsgStoreType . fromRight "memory" . lookupValue "STORE_LOG" "store_messages"
-    textToMsgStoreType = \case
-      "memory" -> Right $ AMSType SMSMemory
-      "journal" -> Right $ AMSType SMSJournal
-      s -> Left $ "invalid store_messages: " <> T.unpack s
+    readMsgStoreType ini = do
+      queues <- journalStore "store_queues" False ini
+      messages <- journalStore "store_messages" queues ini
+      case (queues, messages) of
+        (False, False) -> Right $ AMSType SMSMemory
+        (False, True) -> Right $ AMSType SMSHybrid
+        (True, True) -> Right $ AMSType SMSJournal
+        (True, False) -> Left "`store_queues: journal` requires `store_messages: journal`"
+    journalStore param def = either (const $ Right def) isJournal . lookupValue "STORE_LOG" param
+      where
+        isJournal = \case
+          "memory" -> Right False
+          "journal" -> Right True
+          s -> Left $ "invalid " <> T.unpack (param <> ": " <> s)
+    encodeMsgStoreType :: Maybe Bool -> Text
+    encodeMsgStoreType = \case
+      Just True -> "journal"
+      _ -> "memory"
+    encodeEnablePersistence :: Maybe Bool -> Text
+    encodeEnablePersistence = \case
+      Just _ -> "on"
+      Nothing -> "off"
     httpsCertFile = combine cfgPath "web.crt"
     httpsKeyFile = combine cfgPath "web.key"
     defaultStaticPath = combine logPath "www"
-    initializeServer opts@InitOptions {ip, fqdn, sourceCode = src', webStaticPath = sp', disableWeb = noWeb', scripted}
+    initializeServer opts@InitOptions {journalPersistence, logStats, ip, fqdn, password, sourceCode = src', webStaticPath = sp', disableWeb = noWeb', scripted}
       | scripted = initialize opts
       | otherwise = do
           putStrLn "Use `smp-server init -h` for available options."
           checkInitOptions opts
           void $ withPrompt "SMP server will be initialized (press Enter)" getLine
-          enableStoreLog <- onOffPrompt "Enable store log to restore queues and messages on server restart" True
-          logStats <- onOffPrompt "Enable logging daily statistics" False
-          putStrLn "Require a password to create new messaging queues?"
-          password <- withPrompt "'r' for random (default), 'n' - no password, or enter password: " serverPassword
+          journalPersistence' <- getPersistenceMode
+          logStats' <- onOffPrompt "Enable logging daily statistics" logStats
+          password' <- getServerPassword
           let host = fromMaybe ip fqdn
           host' <- withPrompt ("Enter server FQDN or IP address for certificate (" <> host <> "): ") getLine
           sourceCode' <- withPrompt ("Enter server source code URI (" <> maybe simplexmqSource T.unpack src' <> "): ") getServerSourceCode
           staticPath' <- withPrompt ("Enter path to store generated static site with server information (" <> fromMaybe defaultStaticPath sp' <> "): ") getLine
           initialize
             opts
-              { enableStoreLog,
-                logStats,
+              { journalPersistence = journalPersistence',
+                logStats = logStats',
                 fqdn = if null host' then fqdn else Just host',
-                password,
+                password = password',
                 sourceCode = (T.pack <$> sourceCode') <|> src' <|> Just (T.pack simplexmqSource),
                 webStaticPath = if null staticPath' then sp' else Just staticPath',
                 disableWeb = noWeb'
               }
       where
-        serverPassword =
-          getLine >>= \case
-            "" -> pure $ Just SPRandom
-            "r" -> pure $ Just SPRandom
-            "n" -> pure Nothing
-            s ->
-              case strDecode $ encodeUtf8 $ T.pack s of
-                Right auth -> pure . Just $ ServerPassword auth
-                _ -> putStrLn "Invalid password. Only latin letters, digits and symbols other than '@' and ':' are allowed" >> serverPassword
+        getPersistenceMode = do
+          putStrLn "Server persistence mode:"
+          putStrLn "  'm' - in-memory store with append-inly log (default, dump and restore messages on restart)"
+          putStrLn "  'j' - journal (BETA, durable memory-efficient persistence for queues and messages)"
+          putStrLn "  'd' - disable persistence (not recommended, all data will be lost on restart)"
+          let options = case journalPersistence of
+                Just False -> "Mjd"
+                Just True -> "mJd"
+                Nothing -> "mjD"
+          withPrompt ("Choose mode (" <> options <> "): ") get
+          where
+            get =
+              (map toLower <$> getLine) >>= \case
+                "" -> pure journalPersistence
+                "m" -> pure $ Just False
+                "j" -> pure $ Just True
+                "d" -> pure Nothing
+                _ -> withPrompt "Invalid mode, please enter 'm', 'j' or 'd'" get
+        getServerPassword = do
+          putStrLn "Require a password to create new messaging queues and to use server as proxy?"
+          let options = case password of
+                Just SPRandom -> "'r' - random (default), 'n' - no password"
+                Just (ServerPassword _) -> "'r' - random, 'n' - no password, Enter - to confirm password in options"
+                Nothing -> "'r' - random, 'n' - no password (default)"
+          withPrompt (options <> ", or enter password: ") get
+          where
+            get =
+              getLine >>= \case
+                "" -> pure password
+                "r" -> pure $ Just SPRandom
+                "R" -> pure $ Just SPRandom
+                "n" -> pure Nothing
+                "N" -> pure Nothing
+                s ->
+                  case strDecode $ encodeUtf8 $ T.pack s of
+                    Right auth -> pure . Just $ ServerPassword auth
+                    _ -> putStrLn "Invalid password. Only latin letters, digits and symbols other than '@' and ':' are allowed" >> get
         checkInitOptions InitOptions {sourceCode, serverInfo, operatorCountry, hostingCountry} = do
           let err_
                 | isNothing sourceCode && hasServerInfo serverInfo =
@@ -209,7 +340,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                     Just "Error: passing --hosting-country requires passing --hosting"
                 | otherwise = Nothing
           forM_ err_ $ \err -> putStrLn err >> exitFailure
-        initialize opts'@InitOptions {enableStoreLog, logStats, signAlgorithm, password, controlPort, socksProxy, ownDomains, sourceCode, webStaticPath, disableWeb} = do
+        initialize opts'@InitOptions {signAlgorithm, controlPort, socksProxy, ownDomains, sourceCode, webStaticPath, disableWeb} = do
           checkInitOptions opts'
           clearDirIfExists cfgPath
           clearDirIfExists logPath
@@ -240,12 +371,15 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                    \# This option enables saving memory to append only log,\n\
                    \# and restoring it when the server is started.\n\
                    \# Log is compacted on start (deleted objects are removed).\n"
-                <> ("enable: " <> onOff enableStoreLog <> "\n\n")
+                <> ("enable: " <> encodeEnablePersistence journalPersistence <> "\n\n")
+                <> "# Queue storage mode: `memory` or `journal`.\n"
+                <> ("store_queues: " <> encodeMsgStoreType journalPersistence <> "\n\n")
                 <> "# Message storage mode: `memory` or `journal`.\n\
-                   \store_messages: memory\n\n\
-                   \# When store_messages is `memory`, undelivered messages are optionally saved and restored\n\
+                   \This option is deprecated and will be removed, do NOT use `journal` here if `store_queues` is memory.\n"
+                <> ("store_messages: " <> encodeMsgStoreType journalPersistence <> "\n\n")
+                <> "# When store_messages is `memory`, undelivered messages are optionally saved and restored\n\
                    \# when the server restarts, they are preserved in the .bak file until the next restart.\n"
-                <> ("restore_messages: " <> onOff enableStoreLog <> "\n\n")
+                <> ("restore_messages: " <> encodeEnablePersistence journalPersistence <> "\n\n")
                 <> "# Messages and notifications expiration periods.\n"
                 <> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n")
                 <> "expire_messages_on_start: on\n"
@@ -404,7 +538,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
               storeLogFile = enableStoreLog $> storeLogFilePath,
               storeMsgsFile = case iniMsgStoreType of
                 AMSType SMSMemory -> restoreMessagesFile storeMsgsFilePath
-                AMSType SMSJournal -> Just storeMsgsJournalDir,
+                AMSType _ -> Just storeMsgsJournalDir,
               storeNtfsFile = restoreMessagesFile storeNtfsFilePath,
               -- allow creating new queues by default
               allowNewQueues = fromMaybe True $ iniOnOff "AUTH" "new_queues" ini,
@@ -490,7 +624,8 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
       msgsFileExists <- doesFileExist storeMsgsFilePath
       case mode of
         _ | msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
-        AMSType SMSJournal
+        AMSType SMSJournal -> undefined -- TODO [queues]
+        AMSType SMSHybrid
           | msgsFileExists -> do
               putStrLn $ "Error: store_messages is `journal` with " <> storeMsgsFilePath <> " file present."
               putStrLn "Set store_messages to `memory` or use `smp-server journal export` to migrate."
@@ -505,7 +640,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
         _ -> pure ()
 
     exitConfigureMsgStorage = do
-      putStrLn $ "Error: both " <> storeMsgsFilePath <> " file and " <> storeMsgsJournalDir <> " directory are present."
+      putStrLn $ "Error: file " <> storeMsgsFilePath <> " and directory " <> storeMsgsJournalDir <> " are present."
       putStrLn "Configure memory storage."
       exitFailure
 
@@ -638,10 +773,12 @@ data CliCommand
   | Delete
   | Journal JournalCmd
 
-data JournalCmd = JCImport | JCExport | JCDelete
+data JournalCmd = JCImport (Maybe JournalSubCmd) | JCExport (Maybe JournalSubCmd) | JCDelete
+
+data JournalSubCmd = JSCQueues | JSCMessages
 
 data InitOptions = InitOptions
-  { enableStoreLog :: Bool,
+  { journalPersistence :: Maybe Bool,
     logStats :: Bool,
     signAlgorithm :: SignAlgorithm,
     ip :: HostName,
@@ -675,12 +812,26 @@ cliCommandP cfgPath logPath iniFile =
   where
     initP :: Parser InitOptions
     initP = do
-      enableStoreLog <-
-        switch
+      journalPersistence <-
+        flag' (Just False)
           ( long "store-log"
               <> short 'l'
-              <> help "Enable store log for persistence"
+              <> long "memory"
+              <> short 'm'
+              <> help "In-memory store with append-only log (default, dump and restore messages on restart)"
           )
+          <|>
+            flag' (Just True)
+              ( long "journal"
+                  <> short 'j'
+                  <> help "Journal (BETA, durable memory-efficient persistence for queues and messages)"
+              )
+          <|>
+            flag' Nothing
+              ( long "disable-store"
+                  <> help "Disable persistence (not recommended, all data will be lost on restart)"
+              )
+          <|> pure (Just False)
       logStats <-
         switch
           ( long "daily-stats"
@@ -782,7 +933,7 @@ cliCommandP cfgPath logPath iniFile =
           )
       pure
         InitOptions
-          { enableStoreLog,
+          { journalPersistence,
             logStats,
             signAlgorithm,
             ip,
@@ -811,11 +962,19 @@ cliCommandP cfgPath logPath iniFile =
             scripted
           }
     journalCmdP =
-      hsubparser
-        ( command "import" (info (pure JCImport) (progDesc "Import message log file into a new journal storage"))
-            <> command "export" (info (pure JCExport) (progDesc "Export journal storage to message log file"))
-            <> command "delete" (info (pure JCDelete) (progDesc "Delete journal storage"))
-        )
+      hsubparser $
+        command "import" (info (JCImport <$> optional (journalSubCmdP True)) (progDesc "Import log files into a new journal storage"))
+          <> command "export" (info (JCExport <$> optional (journalSubCmdP False)) (progDesc "Export journal storage to log files"))
+          <> command "delete" (info (pure JCDelete) (progDesc "Delete journal storage"))
+    journalSubCmdP importing
+      | importing =
+          hsubparser $
+            command "queues" (info (pure JSCQueues) (progDesc "Import queues from store log file"))
+              <> command "messages" (info (pure JSCMessages) (progDesc "Import messages from message log log"))
+      | otherwise =
+          hsubparser $
+            command "queues" (info (pure JSCQueues) (progDesc "Export queues to store log file"))
+              <> command "messages" (info (pure JSCMessages) (progDesc "Export messages to message log file"))
 
     parseBasicAuth :: ReadM ServerPassword
     parseBasicAuth = eitherReader $ fmap ServerPassword . strDecode . B.pack

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

@@ -2,6 +2,7 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DerivingStrategies #-}
 {-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE InstanceSigs #-}
@@ -13,12 +14,13 @@
 {-# LANGUAGE StandaloneDeriving #-}
 {-# LANGUAGE TypeFamilies #-}
 {-# LANGUAGE TupleSections #-}
+{-# LANGUAGE UndecidableInstances #-}
 
 module Simplex.Messaging.Server.MsgStore.Journal
-  ( JournalMsgStore (queues, senders, notifiers, random),
-    JournalQueue,
-    JournalMsgQueue (queue, state),
-    JMQueue (queueDirectory, statePath),
+  ( JournalMsgStore (queueStore, random),
+    QueueStore (..),
+    JournalQueue (queueDirectory),
+    JournalMsgQueue (state),
     JournalStoreConfig (..),
     closeMsgQueue,
     closeMsgQueueHandles,
@@ -44,12 +46,14 @@ import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Trans.Except
 import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (intercalate)
-import Data.Maybe (catMaybes, fromMaybe, isNothing)
+import qualified Data.Map.Strict as M
+import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing)
 import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
@@ -65,7 +69,7 @@ import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Util (ifM, tshow, ($>>=), (<$$>))
+import Simplex.Messaging.Util (anyM, ifM, tshow, whenM, ($>>=), (<$$>))
 import System.Directory
 import System.Exit
 import System.FilePath ((</>))
@@ -73,19 +77,27 @@ import System.IO (BufferMode (..), Handle, IOMode (..), SeekMode (..), stdout)
 import qualified System.IO as IO
 import System.Random (StdGen, genByteString, newStdGen)
 
-data JournalMsgStore = JournalMsgStore
-  { config :: JournalStoreConfig,
+data JournalMsgStore s = JournalMsgStore
+  { config :: JournalStoreConfig s,
     random :: TVar StdGen,
     queueLocks :: TMap RecipientId Lock,
-    queues :: TMap RecipientId JournalQueue,
-    senders :: TMap SenderId RecipientId,
-    notifiers :: TMap NotifierId RecipientId,
-    storeLog :: TVar (Maybe (StoreLog 'WriteMode))
+    queueStore :: QueueStore s
   }
 
-data JournalStoreConfig = JournalStoreConfig
+data QueueStore (s :: MSType) where
+  MQStore :: STMQueueStore (JournalQueue 'MSHybrid) -> QueueStore 'MSHybrid
+  -- maps store cached queues
+  -- Nothing in map indicates that the queue doesn't exist
+  JQStore ::
+    { queues_ :: TMap RecipientId (Maybe (JournalQueue 'MSJournal)),
+      senders_ :: TMap SenderId (Maybe RecipientId),
+      notifiers_ :: TMap NotifierId (Maybe RecipientId)
+    } -> QueueStore 'MSJournal
+
+data JournalStoreConfig s = JournalStoreConfig
   { storePath :: FilePath,
     pathParts :: Int,
+    queueStoreType :: SMSType s,
     quota :: Int,
     -- Max number of messages per journal file - ignored in STM store.
     -- When this limit is reached, the file will be changed.
@@ -97,26 +109,22 @@ data JournalStoreConfig = JournalStoreConfig
     idleInterval :: Int64
   }
 
-data JournalQueue = JournalQueue
-  { queueLock :: Lock,
+data JournalQueue (s :: MSType) = JournalQueue
+  { recipientId :: RecipientId,
+    queueLock :: Lock,
     -- To avoid race conditions and errors when restoring queues,
     -- Nothing is written to TVar when queue is deleted.
     queueRec :: TVar (Maybe QueueRec),
-    msgQueue_ :: TVar (Maybe JournalMsgQueue),
+    msgQueue_ :: TVar (Maybe (JournalMsgQueue s)),
     -- system time in seconds since epoch
     activeAt :: TVar Int64,
     -- Just True - empty, Just False - non-empty, Nothing - unknown
-    isEmpty :: TVar (Maybe Bool)
+    isEmpty :: TVar (Maybe Bool),
+    queueDirectory :: FilePath
   }
 
-data JMQueue = JMQueue
-  { queueDirectory :: FilePath,
-    statePath :: FilePath
-  }
-
-data JournalMsgQueue = JournalMsgQueue
-  { queue :: JMQueue,
-    state :: TVar MsgQueueState,
+data JournalMsgQueue (s :: MSType) = JournalMsgQueue
+  { state :: TVar MsgQueueState,
     -- tipMsg contains last message and length incl. newline
     -- Nothing - unknown, Just Nothing - empty queue.
     -- It  prevents reading each message twice,
@@ -209,85 +217,136 @@ instance JournalTypeI t => StrEncoding (JournalState t) where
 queueLogFileName :: String
 queueLogFileName = "queue_state"
 
+queueRecFileName :: String
+queueRecFileName = "queue_rec"
+
 msgLogFileName :: String
 msgLogFileName = "messages"
 
 logFileExt :: String
 logFileExt = ".log"
 
-newtype StoreIO a = StoreIO {unStoreIO :: IO a}
+data QueueRef = QRSender | QRNotifier
+
+queueRefFileName :: QueueRef -> String
+queueRefFileName = \case
+  QRSender -> "sender"
+  QRNotifier -> "notifier"
+
+queueRefFileExt :: String
+queueRefFileExt = ".ref"
+
+newtype StoreIO (s :: MSType) a = StoreIO {unStoreIO :: IO a}
   deriving newtype (Functor, Applicative, Monad)
 
-instance STMQueueStore JournalMsgStore where
-  queues' = queues
-  senders' = senders
-  notifiers' = notifiers
-  storeLog' = storeLog
-  mkQueue st qr = do
-    lock <- getMapLock (queueLocks st) $ recipientId qr
-    q <- newTVar $ Just qr
-    mq <- newTVar Nothing
-    activeAt <- newTVar 0
-    isEmpty <- newTVar Nothing
-    pure $ JournalQueue lock q mq activeAt isEmpty
-  msgQueue_' = msgQueue_
+instance STMStoreClass (JournalMsgStore 'MSHybrid) where
+  stmQueueStore JournalMsgStore {queueStore = MQStore st} = st
+  mkQueue st rId qr = do
+    lock <- atomically $ getMapLock (queueLocks st) rId
+    let dir = msgQueueDirectory st rId
+    makeQueue dir lock rId qr
 
-instance MsgStoreClass JournalMsgStore where
-  type StoreMonad JournalMsgStore = StoreIO
-  type StoreQueue JournalMsgStore = JournalQueue
-  type MsgQueue JournalMsgStore = JournalMsgQueue
-  type MsgStoreConfig JournalMsgStore = JournalStoreConfig
+makeQueue :: FilePath -> Lock -> RecipientId -> QueueRec -> IO (JournalQueue s)
+makeQueue queueDirectory queueLock rId qr = do
+  queueRec <- newTVarIO $ Just qr
+  msgQueue_ <- newTVarIO Nothing
+  activeAt <- newTVarIO 0
+  isEmpty <- newTVarIO Nothing
+  pure
+    JournalQueue
+      { recipientId = rId,
+        queueLock,
+        queueRec,
+        msgQueue_,
+        activeAt,
+        isEmpty,
+        queueDirectory
+      }
 
-  newMsgStore :: JournalStoreConfig -> IO JournalMsgStore
+instance JournalStoreType s => MsgStoreClass (JournalMsgStore s) where
+  type StoreMonad (JournalMsgStore s) = StoreIO s
+  type StoreQueue (JournalMsgStore s) = JournalQueue s
+  type MsgQueue (JournalMsgStore s) = JournalMsgQueue s
+  type MsgStoreConfig (JournalMsgStore s) = (JournalStoreConfig s)
+
+  newMsgStore :: JournalStoreConfig s -> IO (JournalMsgStore s)
   newMsgStore config = do
     random <- newTVarIO =<< newStdGen
-    queueLocks <- TM.emptyIO
-    queues <- TM.emptyIO
-    senders <- TM.emptyIO
-    notifiers <- TM.emptyIO
-    storeLog <- newTVarIO Nothing
-    pure JournalMsgStore {config, random, queueLocks, queues, senders, notifiers, storeLog}
+    queueLocks :: TMap RecipientId Lock <- TM.emptyIO
+    case queueStoreType config of
+      SMSHybrid -> do
+        queueStore <- MQStore <$> newQueueStore
+        pure JournalMsgStore {config, random, queueLocks, queueStore}
+      SMSJournal -> do
+        queues_ <- TM.emptyIO
+        senders_ <- TM.emptyIO
+        notifiers_ <- TM.emptyIO
+        let queueStore = JQStore {queues_, senders_, notifiers_}
+        pure JournalMsgStore {config, random, queueLocks, queueStore}
 
-  setStoreLog :: JournalMsgStore -> StoreLog 'WriteMode -> IO ()
-  setStoreLog st sl = atomically $ writeTVar (storeLog st) (Just sl)
+  closeMsgStore ms = case queueStore ms of
+    MQStore st -> do
+      readTVarIO (storeLog st) >>= mapM_ closeStoreLog
+      readTVarIO (queues st) >>= mapM_ closeMsgQueue
+    st@JQStore {} ->
+      readTVarIO (queues_ st) >>= mapM_ (mapM closeMsgQueue)
 
-  closeMsgStore st = do
-    readTVarIO (storeLog st) >>= mapM_ closeStoreLog
-    readTVarIO (queues st) >>= mapM_ closeMsgQueue
-
-  activeMsgQueues = queues
-  {-# INLINE activeMsgQueues #-}
+  withActiveMsgQueues :: Monoid a => JournalMsgStore s -> (JournalQueue s -> IO a) -> IO a
+  withActiveMsgQueues ms f = case queueStore ms of
+    MQStore st -> withQueues st f
+    st@JQStore {} -> readTVarIO (queues_ st) >>= foldM run mempty
+      where
+        run !acc = maybe (pure acc) (fmap (acc <>) . f)
 
   -- This function is a "foldr" that opens and closes all queues, processes them as defined by action and accumulates the result.
   -- It is used to export storage to a single file and also to expire messages and validate all queues when server is started.
   -- TODO this function requires case-sensitive file system, because it uses queue directory as recipient ID.
   -- It can be made to support case-insensite FS by supporting more than one queue per directory, by getting recipient ID from state file name.
-  withAllMsgQueues :: forall a. Monoid a => Bool -> JournalMsgStore -> (RecipientId -> JournalQueue -> IO a) -> IO a
+  withAllMsgQueues :: forall a. Monoid a => Bool -> JournalMsgStore s -> (JournalQueue s -> IO a) -> IO a
   withAllMsgQueues tty ms@JournalMsgStore {config} action = ifM (doesDirectoryExist storePath) processStore (pure mempty)
     where
       processStore = do
         (!count, !res) <- foldQueues 0 processQueue (0, mempty) ("", storePath)
         putStrLn $ progress count
         pure res
-      JournalStoreConfig {storePath, pathParts} = config
+      JournalStoreConfig {queueStoreType, storePath, pathParts} = config
       processQueue :: (Int, a) -> (String, FilePath) -> IO (Int, a)
       processQueue (!i, !r) (queueId, dir) = do
         when (tty && i `mod` 100 == 0) $ putStr (progress i <> "\r") >> IO.hFlush stdout
         r' <- case strDecode $ B.pack queueId of
           Right rId ->
-            getQueue ms SRecipient rId >>= \case
-              Right q -> unStoreIO (getMsgQueue ms rId q) *> action rId q <* closeMsgQueue q
-              Left AUTH -> do
-                logWarn $ "STORE: processQueue, queue " <> T.pack queueId <> " was removed, removing " <> T.pack dir
-                removeQueueDirectory_ dir
+            validRecipientDir dir rId >>= \case
+              Just True ->
+                getQueue ms SRecipient rId >>= \case
+                  Right q -> unStoreIO (getMsgQueue ms q) *> action q <* closeMsgQueue q
+                  Left AUTH -> case queueStoreType of
+                    SMSJournal -> do
+                      logError $ "STORE: processQueue, queue " <> T.pack queueId <> " failed to open, directory: " <> T.pack dir
+                      exitFailure
+                    SMSHybrid -> do
+                      logWarn $ "STORE: processQueue, queue " <> T.pack queueId <> " was removed, removing " <> T.pack dir
+                      removeQueueDirectory_ dir
+                      pure mempty
+                  Left e -> do
+                    logError $ "STORE: processQueue, error getting queue " <> T.pack queueId <> ", " <> tshow e
+                    exitFailure
+              Just False -> pure mempty
+              Nothing -> do
+                logWarn $ "STORE: processQueue, skipping unknown entity " <> T.pack queueId <> ", directory: " <> T.pack dir
                 pure mempty
-              Left e -> do
-                logError $ "STORE: processQueue, error getting queue " <> T.pack queueId <> ", " <> tshow e
-                exitFailure
           Left e -> do
             logError $ "STORE: processQueue, message queue directory " <> T.pack dir <> " is invalid, " <> tshow e
             exitFailure
         pure (i + 1, r <> r')
+      validRecipientDir dir qId = do
+        ifM
+          (anyExists [queueRecPath, msgQueueStatePath])
+          (pure $ Just True)
+          (ifM (anyExists [queueRefPath QRSender, queueRefPath QRNotifier]) (pure $ Just False) (pure Nothing))
+        where
+          anyExists fs =
+            let paths = map (\f -> f dir qId) fs
+             in anyM $ map doesFileExist $ paths <> map (<> ".bak") paths
       progress i = "Processed: " <> show i <> " queues"
       foldQueues depth f acc (queueId, path) = do
         let f' = if depth == pathParts - 1 then f else foldQueues (depth + 1) f
@@ -302,40 +361,190 @@ instance MsgStoreClass JournalMsgStore where
               (pure $ Just (queueId', path'))
               (Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))
 
-  logQueueStates :: JournalMsgStore -> IO ()
-  logQueueStates ms = withActiveMsgQueues ms $ \_ -> unStoreIO . logQueueState
+  logQueueStates :: JournalMsgStore s -> IO ()
+  logQueueStates ms = withActiveMsgQueues ms $ unStoreIO . logQueueState
 
-  logQueueState :: JournalQueue -> StoreIO ()
+  logQueueState :: JournalQueue s -> StoreIO s ()
   logQueueState q =
     StoreIO . void $
       readTVarIO (msgQueue_ q)
         $>>= \mq -> readTVarIO (handles mq)
         $>>= (\hs -> (readTVarIO (state mq) >>= appendState (stateHandle hs)) $> Just ())
 
+  recipientId' = recipientId
+  {-# INLINE recipientId' #-}
+
   queueRec' = queueRec
   {-# INLINE queueRec' #-}
 
-  getMsgQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> StoreIO JournalMsgQueue
-  getMsgQueue ms@JournalMsgStore {random} rId JournalQueue {msgQueue_} =
+  msgQueue_' = msgQueue_
+  {-# INLINE msgQueue_' #-}
+
+  queueCounts :: JournalMsgStore s -> IO QueueCounts
+  queueCounts ms = case queueStore ms of
+    MQStore st -> do
+      queueCount <- M.size <$> readTVarIO (queues st)
+      notifierCount <- M.size <$> readTVarIO (notifiers st)
+      pure QueueCounts {queueCount, notifierCount}
+    st@JQStore {} -> do
+      queueCount <- M.size <$> readTVarIO (queues_ st)
+      notifierCount <- M.size <$> readTVarIO (notifiers_ st)
+      pure QueueCounts {queueCount, notifierCount}
+
+  addQueue :: JournalMsgStore s -> RecipientId -> QueueRec -> IO (Either ErrorType (JournalQueue s))
+  addQueue ms@JournalMsgStore {queueLocks = ls} rId qr@QueueRec {senderId = sId, notifier} = case queueStore ms of
+    MQStore {} -> addQueue' ms rId qr
+    JQStore {queues_, senders_, notifiers_} -> do
+      lock <- atomically $ getMapLock ls rId
+      tryStore "addQueue" rId $
+        withLock' lock "addQueue" $ withLockMap ls sId "addQueueS" $ withNotifierLock $
+          ifM hasAnyId (pure $ Left DUPLICATE_) $ E.uninterruptibleMask_ $ do
+            let dir = msgQueueDirectory ms rId
+            q <- makeQueue dir lock rId qr
+            storeNewQueue q qr
+            atomically $ TM.insert rId (Just q) queues_
+            saveQueueRef ms QRSender sId rId senders_
+            forM_ notifier $ \NtfCreds {notifierId} -> saveQueueRef ms QRNotifier notifierId rId notifiers_
+            pure $ Right q
+      where
+        hasAnyId = anyM [hasId rId queues_, hasId sId senders_, withNotifier (`hasId` notifiers_), hasDir rId, hasDir sId, withNotifier hasDir]
+        withNotifier p = maybe (pure False) (\NtfCreds {notifierId} -> p notifierId) notifier
+        withNotifierLock a = maybe a (\NtfCreds {notifierId} -> withLockMap ls notifierId "addQueueN" a) notifier
+        hasId :: EntityId -> TMap EntityId (Maybe a) -> IO Bool
+        hasId qId m = maybe False isJust <$> atomically (TM.lookup qId m)
+        hasDir qId = doesDirectoryExist $ msgQueueDirectory ms qId
+
+  getQueue :: DirectParty p => JournalMsgStore s -> SParty p -> QueueId -> IO (Either ErrorType (JournalQueue s))
+  getQueue ms party qId = case queueStore ms of
+    MQStore st -> getQueue' st party qId
+    st@JQStore {queues_} ->
+      isolateQueueId "getQueue" ms qId $
+        case party of
+          SRecipient -> getQueue_ qId
+          SSender -> getQueueRef QRSender (senders_ st) $>>= isolateGetQueue
+          SNotifier -> getQueueRef QRNotifier (notifiers_ st) $>>= isolateGetQueue
+      where
+        isolateGetQueue rId = isolateQueueId "getQueueR" ms rId $ getQueue_ rId
+        getQueue_ rId = TM.lookupIO rId queues_ >>= maybe loadQueue (pure . maybe (Left AUTH) Right) 
+          where
+            loadQueue = do
+              let dir = msgQueueDirectory ms rId
+                  f = queueRecPath dir rId
+              ifM (doesFileExist f) (load dir f) $ do
+                atomically $ TM.insert rId Nothing queues_
+                pure $ Left AUTH
+            load dir f = do
+              -- TODO [queues] read backup if exists, remove old timed backups
+              qr_ <- first STORE . strDecode  <$> B.readFile f
+              forM qr_ $ \qr -> do
+                lock <- atomically $ getMapLock (queueLocks ms) rId
+                q <- makeQueue dir lock rId qr
+                atomically $ TM.insert rId (Just q) queues_
+                pure q
+        getQueueRef :: QueueRef -> TMap EntityId (Maybe RecipientId) -> IO (Either ErrorType RecipientId)
+        getQueueRef qRef m = TM.lookupIO qId m >>= maybe loadQueueRef (pure . maybe (Left AUTH) Right)
+          where
+            loadQueueRef = do
+              let dir = msgQueueDirectory ms qId
+                  f = queueRefPath qRef dir qId
+              ifM (doesFileExist f) (loadRef f) $ do
+                atomically $ TM.insert qId Nothing m
+                pure $ Left AUTH
+            loadRef f = do
+              -- TODO [queues] read backup if exists, remove old timed backups
+              rId_ <- first STORE . strDecode <$> B.readFile f
+              forM rId_ $ \rId -> do
+                atomically $ TM.insert qId (Just rId) m
+                pure rId
+
+  secureQueue :: JournalMsgStore s -> JournalQueue s -> SndPublicAuthKey -> IO (Either ErrorType ())
+  secureQueue st sq sKey = case queueStore st of
+    MQStore st' -> secureQueue' st' sq sKey
+    JQStore {} ->
+      isolateQueueRec sq "secureQueue" $ \q -> case senderKey q of
+        Just k -> pure $ if sKey == k then Right () else Left AUTH
+        Nothing -> storeQueue sq q {senderKey = Just sKey} $> Right ()
+
+  addQueueNotifier :: JournalMsgStore s -> JournalQueue s -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
+  addQueueNotifier st sq ntfCreds@NtfCreds {notifierId = nId} = case queueStore st of
+    MQStore st' -> addQueueNotifier' st' sq ntfCreds
+    JQStore {notifiers_} ->
+      isolateQueueRec sq "addQueueNotifier" $ \q ->
+        withLockMap (queueLocks st) nId "addQueueNotifierN" $
+          ifM hasNotifierId (pure $ Left DUPLICATE_) $ E.uninterruptibleMask_ $ do
+            nId_ <- forM (notifier q) $ \NtfCreds {notifierId = nId'} -> 
+              withLockMap (queueLocks st) nId' "addQueueNotifierD" $
+                deleteQueueRef st QRNotifier nId' notifiers_ $> nId'
+            storeQueue sq q {notifier = Just ntfCreds}
+            saveQueueRef st QRNotifier nId (recipientId sq) notifiers_
+            pure $ Right nId_
+      where
+        hasNotifierId = anyM [hasId, hasDir]
+        hasId = maybe False isJust <$> atomically (TM.lookup nId notifiers_)
+        hasDir = doesDirectoryExist $ msgQueueDirectory st nId
+
+  deleteQueueNotifier :: JournalMsgStore s -> JournalQueue s -> IO (Either ErrorType (Maybe NotifierId))
+  deleteQueueNotifier ms sq = case queueStore ms of
+    MQStore st -> deleteQueueNotifier' st sq
+    st@JQStore {} ->
+      isolateQueueRec sq "deleteQueueNotifier" $ \q ->
+        fmap Right $ forM (notifier q) $ \NtfCreds {notifierId = nId} ->
+          withLockMap (queueLocks ms) nId "deleteQueueNotifierN" $ do
+            deleteQueueRef ms QRNotifier nId (notifiers_ st)
+            storeQueue sq q {notifier = Nothing}
+            pure nId
+
+  suspendQueue :: JournalMsgStore s -> JournalQueue s -> IO (Either ErrorType ())
+  suspendQueue ms sq = case queueStore ms of
+    MQStore st -> suspendQueue' st sq
+    JQStore {} ->
+      isolateQueueRec sq "suspendQueue" $ \q ->
+        fmap Right $ storeQueue sq q {status = EntityOff}
+
+  blockQueue :: JournalMsgStore s -> JournalQueue s -> BlockingInfo -> IO (Either ErrorType ())
+  blockQueue ms sq info = case queueStore ms of
+    MQStore st -> blockQueue' st sq info
+    JQStore {} ->
+      isolateQueueRec sq "blockQueue" $ \q ->
+        fmap Right $ storeQueue sq q {status = EntityBlocked info}
+
+  unblockQueue :: JournalMsgStore s -> JournalQueue s -> IO (Either ErrorType ())
+  unblockQueue ms sq = case queueStore ms of
+    MQStore st -> unblockQueue' st sq
+    JQStore {} ->
+      isolateQueueRec sq "unblockQueue" $ \q ->
+        fmap Right $ storeQueue sq q {status = EntityActive}
+
+  updateQueueTime :: JournalMsgStore s -> JournalQueue s -> RoundedSystemTime -> IO (Either ErrorType QueueRec)
+  updateQueueTime ms sq t = case queueStore ms of
+    MQStore st -> updateQueueTime' st sq t
+    JQStore {} -> isolateQueueRec sq "updateQueueTime" $ fmap Right . update
+      where
+        update q@QueueRec {updatedAt}
+          | updatedAt == Just t = pure q
+          | otherwise =
+              let !q' = q {updatedAt = Just t}
+               in storeQueue sq q' $> q'
+
+  getMsgQueue :: JournalMsgStore s -> JournalQueue s -> StoreIO s (JournalMsgQueue s)
+  getMsgQueue ms@JournalMsgStore {random} sq@JournalQueue {recipientId, msgQueue_, queueDirectory} =
     StoreIO $ readTVarIO msgQueue_ >>= maybe newQ pure
     where
       newQ = do
-        let dir = msgQueueDirectory ms rId
-            statePath = msgQueueStatePath dir $ B.unpack (strEncode rId)
-            queue = JMQueue {queueDirectory = dir, statePath}
-        q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue) (createQ queue)
+        let statePath = msgQueueStatePath queueDirectory recipientId
+        q <- ifM (doesFileExist statePath) (openMsgQueue ms sq statePath) createQ
         atomically $ writeTVar msgQueue_ $ Just q
         pure q
         where
-          createQ :: JMQueue -> IO JournalMsgQueue
-          createQ queue = do
+          createQ :: IO (JournalMsgQueue s)
+          createQ = do
             -- folder and files are not created here,
             -- to avoid file IO for queues without messages during subscription
             journalId <- newJournalId random
-            mkJournalQueue queue (newMsgQueueState journalId) Nothing
+            mkJournalQueue (newMsgQueueState journalId) Nothing
 
-  getPeekMsgQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> StoreIO (Maybe (JournalMsgQueue, Message))
-  getPeekMsgQueue ms rId q@JournalQueue {isEmpty} =
+  getPeekMsgQueue :: JournalMsgStore s -> JournalQueue s -> StoreIO s (Maybe (JournalMsgQueue s, Message))
+  getPeekMsgQueue ms q@JournalQueue {isEmpty} =
     StoreIO (readTVarIO isEmpty) >>= \case
       Just True -> pure Nothing
       Just False -> peek
@@ -350,16 +559,16 @@ instance MsgStoreClass JournalMsgStore where
         pure r
     where
       peek = do
-        mq <- getMsgQueue ms rId q
+        mq <- getMsgQueue ms q
         (mq,) <$$> tryPeekMsg_ q mq
 
   -- only runs action if queue is not empty
-  withIdleMsgQueue :: Int64 -> JournalMsgStore -> RecipientId -> JournalQueue -> (JournalMsgQueue -> StoreIO a) -> StoreIO (Maybe a, Int)
-  withIdleMsgQueue now ms@JournalMsgStore {config} rId q action =
+  withIdleMsgQueue :: Int64 -> JournalMsgStore s -> JournalQueue s -> (JournalMsgQueue s -> StoreIO s a) -> StoreIO s (Maybe a, Int)
+  withIdleMsgQueue now ms@JournalMsgStore {config} q action =
     StoreIO $ readTVarIO (msgQueue_ q) >>= \case
       Nothing ->
         E.bracket
-          (unStoreIO $ getPeekMsgQueue ms rId q)
+          (unStoreIO $ getPeekMsgQueue ms q)
           (mapM_ $ \_ -> closeMsgQueue q)
           (maybe (pure (Nothing, 0)) (unStoreIO . run))
         where
@@ -375,31 +584,30 @@ instance MsgStoreClass JournalMsgStore where
         sz <- unStoreIO $ getQueueSize_ mq
         pure (r, sz)
 
-  deleteQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType QueueRec)
-  deleteQueue ms rId q =
-    fst <$$> deleteQueue_ ms rId q
+  deleteQueue :: JournalMsgStore s -> JournalQueue s -> IO (Either ErrorType QueueRec)
+  deleteQueue ms q = fst <$$> deleteQueue_ ms q
 
-  deleteQueueSize :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType (QueueRec, Int))
-  deleteQueueSize ms rId q =
-    deleteQueue_ ms rId q >>= mapM (traverse getSize)
+  deleteQueueSize :: JournalMsgStore s -> JournalQueue s -> IO (Either ErrorType (QueueRec, Int))
+  deleteQueueSize ms q =
+    deleteQueue_ ms q >>= mapM (traverse getSize)
     -- traverse operates on the second tuple element
     where
       getSize = maybe (pure (-1)) (fmap size . readTVarIO . state)
 
-  getQueueMessages_ :: Bool -> JournalMsgQueue -> StoreIO [Message]
-  getQueueMessages_ drainMsgs q = StoreIO (run [])
+  getQueueMessages_ :: Bool -> JournalQueue s -> JournalMsgQueue s -> StoreIO s [Message]
+  getQueueMessages_ drainMsgs sq q = StoreIO (run [])
     where
       run msgs = readTVarIO (handles q) >>= maybe (pure []) (getMsg msgs)
-      getMsg msgs hs = chooseReadJournal q drainMsgs hs >>= maybe (pure msgs) readMsg
+      getMsg msgs hs = chooseReadJournal sq q drainMsgs hs >>= maybe (pure msgs) readMsg
         where
           readMsg (rs, h) = do
             (msg, len) <- hGetMsgAt h $ bytePos rs
             updateReadPos q drainMsgs len hs
             (msg :) <$> run msgs
 
-  writeMsg :: JournalMsgStore -> RecipientId -> JournalQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
-  writeMsg ms rId q' logState msg = isolateQueue rId q' "writeMsg" $ do
-    q <- getMsgQueue ms rId q'
+  writeMsg :: JournalMsgStore s -> JournalQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
+  writeMsg ms q'@JournalQueue {recipientId, queueDirectory = dir} logState msg = isolateQueue q' "writeMsg" $ do
+    q <- getMsgQueue ms q'
     StoreIO $ (`E.finally` updateActiveAt q') $ do
       st@MsgQueueState {canWrite, size} <- readTVarIO (state q)
       let empty = size == 0
@@ -414,9 +622,10 @@ instance MsgStoreClass JournalMsgStore where
     where
       JournalStoreConfig {quota, maxMsgCount} = config ms
       msgQuota = MessageQuota {msgId = messageId msg, msgTs = messageTs msg}
-      writeToJournal q st@MsgQueueState {writeState, readState = rs, size} canWrt' !msg' = do
+      writeToJournal q@JournalMsgQueue {handles} st@MsgQueueState {writeState, readState = rs, size} canWrt' !msg' = do
         let msgStr = strEncode msg' `B.snoc` '\n'
             msgLen = fromIntegral $ B.length msgStr
+        -- TODO [queues] this should just work, if queue was not opened it will be created, and folder will be used if exists
         hs <- maybe createQueueDir pure =<< readTVarIO handles
         (ws, wh) <- case writeHandle hs of
           Nothing | msgCount writeState >= maxMsgCount -> switchWriteJournal hs
@@ -430,33 +639,34 @@ instance MsgStoreClass JournalMsgStore where
         updateQueueState q logState hs st' $
           when (size == 0) $ writeTVar (tipMsg q) $ Just (Just (msg, msgLen))
         where
-          JournalMsgQueue {queue = JMQueue {queueDirectory, statePath}, handles} = q
           createQueueDir = do
-            createDirectoryIfMissing True queueDirectory
+            createDirectoryIfMissing True dir
+            let statePath = msgQueueStatePath dir recipientId
             sh <- openFile statePath AppendMode
             B.hPutStr sh ""
-            rh <- createNewJournal queueDirectory $ journalId rs
+            rh <- createNewJournal dir $ journalId rs
             let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = Nothing}
             atomically $ writeTVar handles $ Just hs
             pure hs
           switchWriteJournal hs = do
             journalId <- newJournalId $ random ms
-            wh <- createNewJournal queueDirectory journalId
+            wh <- createNewJournal dir journalId
             atomically $ writeTVar handles $ Just $ hs {writeHandle = Just wh}
             pure (newJournalState journalId, wh)
 
   -- can ONLY be used while restoring messages, not while server running
-  setOverQuota_ :: JournalQueue -> IO ()
+  setOverQuota_ :: JournalQueue s -> IO ()
   setOverQuota_ q =
     readTVarIO (msgQueue_ q)
       >>= mapM_ (\JournalMsgQueue {state} -> atomically $ modifyTVar' state $ \st -> st {canWrite = False})
 
-  getQueueSize_ :: JournalMsgQueue -> StoreIO Int
+  getQueueSize_ :: JournalMsgQueue s -> StoreIO s Int
   getQueueSize_ JournalMsgQueue {state} = StoreIO $ size <$> readTVarIO state
+  {-# INLINE getQueueSize_ #-}
 
-  tryPeekMsg_ :: JournalQueue -> JournalMsgQueue -> StoreIO (Maybe Message)
+  tryPeekMsg_ :: JournalQueue s -> JournalMsgQueue s -> StoreIO s (Maybe Message)
   tryPeekMsg_ q mq@JournalMsgQueue {tipMsg, handles} =
-    StoreIO $ (readTVarIO handles $>>= chooseReadJournal mq True $>>= peekMsg) >>= setEmpty
+    StoreIO $ (readTVarIO handles $>>= chooseReadJournal q mq True $>>= peekMsg) >>= setEmpty
     where
       peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst)
         where
@@ -468,7 +678,7 @@ instance MsgStoreClass JournalMsgStore where
         atomically $ writeTVar (isEmpty q) (Just $ isNothing msg)
         pure msg
 
-  tryDeleteMsg_ :: JournalQueue -> JournalMsgQueue -> Bool -> StoreIO ()
+  tryDeleteMsg_ :: JournalQueue s -> JournalMsgQueue s -> Bool -> StoreIO s ()
   tryDeleteMsg_ q mq@JournalMsgQueue {tipMsg, handles} logState = StoreIO $ (`E.finally` when logState (updateActiveAt q)) $
     void $
       readTVarIO tipMsg -- if there is no cached tipMsg, do nothing
@@ -476,52 +686,101 @@ instance MsgStoreClass JournalMsgStore where
         $>>= \len -> readTVarIO handles
         $>>= \hs -> updateReadPos mq logState len hs $> Just ()
 
-  isolateQueue :: RecipientId -> JournalQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
-  isolateQueue rId JournalQueue {queueLock} op =
-    tryStore' op rId . withLock' queueLock op . unStoreIO
+  isolateQueue :: JournalQueue s -> String -> StoreIO s a -> ExceptT ErrorType IO a
+  isolateQueue q op = ExceptT . isolateQueue_ q op . fmap Right . unStoreIO
+  {-# INLINE isolateQueue #-}
 
-updateActiveAt :: JournalQueue -> IO ()
+updateActiveAt :: JournalQueue s -> IO ()
 updateActiveAt q = atomically . writeTVar (activeAt q) . systemSeconds =<< getSystemTime
 
-tryStore' :: String -> RecipientId -> IO a -> ExceptT ErrorType IO a
-tryStore' op rId = tryStore op rId . fmap Right
-
-tryStore :: forall a. String -> RecipientId -> IO (Either ErrorType a) -> ExceptT ErrorType IO a
-tryStore op rId a = ExceptT $ E.mask_ $ E.try a >>= either storeErr pure
+tryStore :: forall a. String -> RecipientId -> IO (Either ErrorType a) -> IO (Either ErrorType a)
+tryStore op rId a = E.mask_ $ E.try a >>= either storeErr pure
   where
     storeErr :: E.SomeException -> IO (Either ErrorType a)
     storeErr e =
       let e' = intercalate ", " [op, B.unpack $ strEncode rId, show e]
        in logError ("STORE: " <> T.pack e') $> Left (STORE e')
 
-isolateQueueId :: String -> JournalMsgStore -> RecipientId -> IO (Either ErrorType a) -> ExceptT ErrorType IO a
+isolateQueueRec :: JournalQueue s -> String -> (QueueRec -> IO (Either ErrorType a)) -> IO (Either ErrorType a)
+isolateQueueRec sq op a = isolateQueue_ sq op (atomically (readQueueRec qr) $>>= a)
+  where
+    qr = queueRec sq
+
+isolateQueue_ :: JournalQueue s -> String -> IO (Either ErrorType a) -> IO (Either ErrorType a)
+isolateQueue_ JournalQueue {recipientId, queueLock} op = tryStore op recipientId . withLock' queueLock op
+
+isolateQueueId :: String -> JournalMsgStore s -> RecipientId -> IO (Either ErrorType a) -> IO (Either ErrorType a)
 isolateQueueId op ms rId = tryStore op rId . withLockMap (queueLocks ms) rId op
 
-openMsgQueue :: JournalMsgStore -> JMQueue -> IO JournalMsgQueue
-openMsgQueue ms q@JMQueue {queueDirectory = dir, statePath} = do
+storeNewQueue :: JournalQueue s -> QueueRec -> IO ()
+storeNewQueue sq@JournalQueue {queueDirectory} q = do
+  createDirectoryIfMissing True queueDirectory
+  storeQueue_ sq q
+
+storeQueue :: JournalQueue s -> QueueRec -> IO ()
+storeQueue sq@JournalQueue {queueRec} q = do
+  storeQueue_ sq q
+  atomically $ writeTVar queueRec $ Just q
+
+saveQueueRef :: JournalMsgStore 'MSJournal -> QueueRef -> QueueId -> RecipientId -> TMap QueueId (Maybe RecipientId) -> IO ()
+saveQueueRef st qRef qId rId m = do
+  let dir = msgQueueDirectory st qId
+      f = queueRefPath qRef dir qId
+  createDirectoryIfMissing True dir
+  safeReplaceFile f $ strEncode rId
+  atomically $ TM.insert qId (Just rId) m
+
+deleteQueueRef :: JournalMsgStore 'MSJournal -> QueueRef -> QueueId -> TMap QueueId (Maybe RecipientId) -> IO ()
+deleteQueueRef st qRef qId m = do
+  let dir = msgQueueDirectory st qId
+      f = queueRefPath qRef dir qId
+  whenM (doesFileExist f) $ removeFile f
+  -- TODO [queues] remove folder if it's empty or has only timed backups
+  -- TODO [queues] remove empty parent folders up to storage depth
+  atomically $ TM.delete qId m
+
+storeQueue_ :: JournalQueue s -> QueueRec -> IO ()
+storeQueue_ JournalQueue {recipientId, queueDirectory} q = do
+  let f = queueRecPath queueDirectory recipientId
+  safeReplaceFile f $ strEncode q
+
+safeReplaceFile :: FilePath -> ByteString -> IO ()
+safeReplaceFile f s = ifM (doesFileExist f) replace (B.writeFile f s)
+  where
+    temp = f <> ".bak"
+    replace = do
+      renameFile f temp
+      B.writeFile f s
+      renameFile temp =<< timedBackupName f
+
+timedBackupName :: FilePath -> IO FilePath
+timedBackupName f = do
+  ts <- getCurrentTime
+  pure $ f <> "." <> iso8601Show ts <> ".bak"
+
+openMsgQueue :: JournalMsgStore s -> JournalQueue s -> FilePath -> IO (JournalMsgQueue s)
+openMsgQueue ms JournalQueue {queueDirectory = dir} statePath = do
   (st, sh) <- readWriteQueueState ms statePath
   (st', rh, wh_) <- closeOnException sh $ openJournals ms dir st sh
   let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = wh_}
-  mkJournalQueue q st' (Just hs)
+  mkJournalQueue st' (Just hs)
 
-mkJournalQueue :: JMQueue -> MsgQueueState -> Maybe MsgQueueHandles -> IO JournalMsgQueue
-mkJournalQueue queue st hs_ = do
+mkJournalQueue :: MsgQueueState -> Maybe MsgQueueHandles -> IO (JournalMsgQueue s)
+mkJournalQueue st hs_ = do
   state <- newTVarIO st
   tipMsg <- newTVarIO Nothing
   handles <- newTVarIO hs_
-  -- using the same queue lock which is currently locked,
-  -- to avoid map lookup on queue operations
-  pure JournalMsgQueue {queue, state, tipMsg, handles}
+  pure JournalMsgQueue {state, tipMsg, handles}
 
-chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle))
-chooseReadJournal q log' hs = do
+chooseReadJournal :: JournalQueue s -> JournalMsgQueue s -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle))
+chooseReadJournal sq q log' hs = do
   st@MsgQueueState {writeState = ws, readState = rs} <- readTVarIO (state q)
   case writeHandle hs of
     Just wh | msgPos rs >= msgCount rs && journalId rs /= journalId ws -> do
       -- switching to write journal
       atomically $ writeTVar (handles q) $ Just hs {readHandle = wh, writeHandle = Nothing}
       hClose $ readHandle hs
-      when log' $ removeJournal (queueDirectory $ queue q) rs
+      when log' $ removeJournal (queueDirectory sq) rs
       let !rs' = (newJournalState $ journalId ws) {msgCount = msgCount ws, byteCount = byteCount ws}
           !st' = st {readState = rs'}
       updateQueueState q log' hs st' $ pure ()
@@ -529,7 +788,7 @@ chooseReadJournal q log' hs = do
     _ | msgPos rs >= msgCount rs && journalId rs == journalId ws -> pure Nothing
     _ -> pure $ Just (rs, readHandle hs)
 
-updateQueueState :: JournalMsgQueue -> Bool -> MsgQueueHandles -> MsgQueueState -> STM () -> IO ()
+updateQueueState :: JournalMsgQueue s -> Bool -> MsgQueueHandles -> MsgQueueState -> STM () -> IO ()
 updateQueueState q log' hs st a = do
   unless (validQueueState st) $ E.throwIO $ userError $ "updateQueueState invalid state: " <> show st
   when log' $ appendState (stateHandle hs) st
@@ -538,7 +797,7 @@ updateQueueState q log' hs st a = do
 appendState :: Handle -> MsgQueueState -> IO ()
 appendState h st = E.uninterruptibleMask_ $ B.hPutStr h $ strEncode st `B.snoc` '\n'
 
-updateReadPos :: JournalMsgQueue -> Bool -> Int64 -> MsgQueueHandles -> IO ()
+updateReadPos :: JournalMsgQueue s -> Bool -> Int64 -> MsgQueueHandles -> IO ()
 updateReadPos q log' len hs = do
   st@MsgQueueState {readState = rs, size} <- readTVarIO (state q)
   let JournalState {msgPos, bytePos} = rs
@@ -547,9 +806,9 @@ updateReadPos q log' len hs = do
       st' = st {readState = rs', size = size - 1}
   updateQueueState q log' hs st' $ writeTVar (tipMsg q) Nothing
 
-msgQueueDirectory :: JournalMsgStore -> RecipientId -> FilePath
-msgQueueDirectory JournalMsgStore {config = JournalStoreConfig {storePath, pathParts}} rId =
-  storePath </> B.unpack (B.intercalate "/" $ splitSegments pathParts $ strEncode rId)
+msgQueueDirectory :: JournalMsgStore s -> RecipientId -> FilePath
+msgQueueDirectory JournalMsgStore {config = JournalStoreConfig {storePath, pathParts}} qId =
+  storePath </> B.unpack (B.intercalate "/" $ splitSegments pathParts $ strEncode qId)
   where
     splitSegments _ "" = []
     splitSegments 1 s = [s]
@@ -557,12 +816,18 @@ msgQueueDirectory JournalMsgStore {config = JournalStoreConfig {storePath, pathP
       let (seg, s') = B.splitAt 2 s
        in seg : splitSegments (n - 1) s'
 
-msgQueueStatePath :: FilePath -> String -> FilePath
-msgQueueStatePath dir queueId = dir </> (queueLogFileName <> "." <> queueId <> logFileExt)
+queueRecPath :: FilePath -> RecipientId -> FilePath
+queueRecPath dir rId = dir </> (queueRecFileName <> "." <> B.unpack (strEncode rId) <> logFileExt)
+
+queueRefPath :: QueueRef -> FilePath -> QueueId -> FilePath
+queueRefPath qRef dir qId = dir </> (queueRefFileName qRef <> "." <> B.unpack (strEncode qId) <> queueRefFileExt)
+
+msgQueueStatePath :: FilePath -> RecipientId -> FilePath
+msgQueueStatePath dir rId = dir </> (queueLogFileName <> "." <> B.unpack (strEncode rId) <> logFileExt)
 
 createNewJournal :: FilePath -> ByteString -> IO Handle
 createNewJournal dir journalId = do
-  let path = journalFilePath dir journalId -- TODO retry if file exists
+  let path = journalFilePath dir journalId -- TODO [queues] retry if file exists
   h <- openFile path ReadWriteMode
   B.hPutStr h ""
   pure h
@@ -570,7 +835,7 @@ createNewJournal dir journalId = do
 newJournalId :: TVar StdGen -> IO ByteString
 newJournalId g = strEncode <$> atomically (stateTVar g $ genByteString 12)
 
-openJournals :: JournalMsgStore -> FilePath -> MsgQueueState -> Handle -> IO (MsgQueueState, Handle, Maybe Handle)
+openJournals :: JournalMsgStore s -> FilePath -> MsgQueueState -> Handle -> IO (MsgQueueState, Handle, Maybe Handle)
 openJournals ms dir st@MsgQueueState {readState = rs, writeState = ws} sh = do
   let rjId = journalId rs
       wjId = journalId ws
@@ -643,7 +908,8 @@ removeJournal dir JournalState {journalId} = do
 
 -- This function is supposed to be resilient to crashes while updating state files,
 -- and also resilient to crashes during its execution.
-readWriteQueueState :: JournalMsgStore -> FilePath -> IO (MsgQueueState, Handle)
+-- TODO [queues] remove old timed backups
+readWriteQueueState :: JournalMsgStore s -> FilePath -> IO (MsgQueueState, Handle)
 readWriteQueueState JournalMsgStore {random, config} statePath =
   ifM
     (doesFileExist tempBackup)
@@ -686,8 +952,7 @@ readWriteQueueState JournalMsgStore {random, config} statePath =
       -- Temporary backup file will be used when it is present.
       renameFile statePath tempBackup -- 1) temp backup
       r <- writeQueueState st -- 2) save state
-      ts <- getCurrentTime
-      renameFile tempBackup (statePath <> "." <> iso8601Show ts <> ".bak") -- 3) timed backup
+      renameFile tempBackup =<< timedBackupName statePath -- 3) timed backup
       pure r
     writeQueueState st = do
       sh <- openFile statePath AppendMode
@@ -721,20 +986,33 @@ validQueueState MsgQueueState {readState = rs, writeState = ws, size}
         && msgPos ws == msgCount ws
         && bytePos ws == byteCount ws
 
-deleteQueue_ :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType (QueueRec, Maybe JournalMsgQueue))
-deleteQueue_ ms rId q =
-  runExceptT $ isolateQueueId "deleteQueue_" ms rId $
-    deleteQueue' ms rId q >>= mapM remove
+deleteQueue_ :: forall s. JournalMsgStore s -> JournalQueue s -> IO (Either ErrorType (QueueRec, Maybe (JournalMsgQueue s)))
+deleteQueue_ ms sq =
+  isolateQueueId "deleteQueue_" ms rId $ E.uninterruptibleMask_ $
+    delete >>= mapM (traverse remove)
   where
-    remove r@(_, mq_) = do
-      mapM_ closeMsgQueueHandles mq_
+    rId = recipientId sq
+    qr = queueRec sq
+    delete :: IO (Either ErrorType (QueueRec, Maybe (JournalMsgQueue s)))
+    delete = case queueStore ms of
+      MQStore st -> deleteQueue' st sq
+      st@JQStore {} -> atomically (readQueueRec qr) >>= mapM jqDelete
+        where
+          jqDelete q = E.uninterruptibleMask_ $ do
+            deleteQueueRef ms QRSender (senderId q) (senders_ st)
+            forM_ (notifier q) $ \NtfCreds {notifierId} -> deleteQueueRef ms QRNotifier notifierId (notifiers_ st)
+            atomically $ writeTVar qr Nothing
+            (q,) <$> atomically (swapTVar (msgQueue_' sq) Nothing)
+    remove :: Maybe (JournalMsgQueue s) -> IO (Maybe (JournalMsgQueue s))
+    remove mq = do
+      mapM_ closeMsgQueueHandles mq
       removeQueueDirectory ms rId
-      pure r
+      pure mq
 
-closeMsgQueue :: JournalQueue -> IO ()
+closeMsgQueue :: JournalQueue s -> IO ()
 closeMsgQueue JournalQueue {msgQueue_} = atomically (swapTVar msgQueue_ Nothing) >>= mapM_ closeMsgQueueHandles
 
-closeMsgQueueHandles :: JournalMsgQueue -> IO ()
+closeMsgQueueHandles :: JournalMsgQueue s -> IO ()
 closeMsgQueueHandles q = readTVarIO (handles q) >>= mapM_ closeHandles
   where
     closeHandles (MsgQueueHandles sh rh wh_) = do
@@ -742,7 +1020,7 @@ closeMsgQueueHandles q = readTVarIO (handles q) >>= mapM_ closeHandles
       hClose rh
       mapM_ hClose wh_
 
-removeQueueDirectory :: JournalMsgStore -> RecipientId -> IO ()
+removeQueueDirectory :: JournalMsgStore s -> RecipientId -> IO ()
 removeQueueDirectory st = removeQueueDirectory_ . msgQueueDirectory st
 
 removeQueueDirectory_ :: FilePath -> IO ()
@@ -773,7 +1051,8 @@ openFile f mode = do
   pure h
 
 hClose :: Handle -> IO ()
-hClose h =
+hClose h = do
+  IO.hFlush h
   IO.hClose h `catchAny` \e -> do
     name <- IO.hShow h
     logError $ "STORE: hClose, " <> T.pack name <> ", " <> tshow e

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

@@ -21,27 +21,23 @@ import Control.Monad.IO.Class
 import Control.Monad.Trans.Except
 import Data.Functor (($>))
 import Data.Int (Int64)
+import qualified Data.Map.Strict as M
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.TMap (TMap)
-import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util ((<$$>), ($>>=))
-import System.IO (IOMode (..))
 
 data STMMsgStore = STMMsgStore
   { storeConfig :: STMStoreConfig,
-    queues :: TMap RecipientId STMQueue,
-    senders :: TMap SenderId RecipientId,
-    notifiers :: TMap NotifierId RecipientId,
-    storeLog :: TVar (Maybe (StoreLog 'WriteMode))
+    queueStore :: STMQueueStore STMQueue
   }
 
 data STMQueue = STMQueue
   { -- To avoid race conditions and errors when restoring queues,
     -- Nothing is written to TVar when queue is deleted.
+    recipientId :: RecipientId,
     queueRec :: TVar (Maybe QueueRec),
     msgQueue_ :: TVar (Maybe STMMsgQueue)
   }
@@ -57,13 +53,9 @@ data STMStoreConfig = STMStoreConfig
     quota :: Int
   }
 
-instance STMQueueStore STMMsgStore where
-  queues' = queues
-  senders' = senders
-  notifiers' = notifiers
-  storeLog' = storeLog
-  mkQueue _ qr = STMQueue <$> newTVar (Just qr) <*> newTVar Nothing
-  msgQueue_' = msgQueue_
+instance STMStoreClass STMMsgStore where
+  stmQueueStore = queueStore
+  mkQueue _ rId qr = STMQueue rId <$> newTVarIO (Just qr) <*> newTVarIO Nothing
 
 instance MsgStoreClass STMMsgStore where
   type StoreMonad STMMsgStore = STM
@@ -72,33 +64,64 @@ instance MsgStoreClass STMMsgStore where
   type MsgStoreConfig STMMsgStore = STMStoreConfig
 
   newMsgStore :: STMStoreConfig -> IO STMMsgStore
-  newMsgStore storeConfig = do
-    queues <- TM.emptyIO
-    senders <- TM.emptyIO
-    notifiers <- TM.emptyIO
-    storeLog <- newTVarIO Nothing
-    pure STMMsgStore {storeConfig, queues, senders, notifiers, storeLog}
+  newMsgStore storeConfig = STMMsgStore storeConfig <$> newQueueStore
 
-  setStoreLog :: STMMsgStore -> StoreLog 'WriteMode -> IO ()
-  setStoreLog st sl = atomically $ writeTVar (storeLog st) (Just sl)
+  closeMsgStore st = readTVarIO (storeLog $ queueStore st) >>= mapM_ closeStoreLog
 
-  closeMsgStore st = readTVarIO (storeLog st) >>= mapM_ closeStoreLog
+  withActiveMsgQueues = withQueues . queueStore
+  {-# INLINE withActiveMsgQueues #-}
 
-  activeMsgQueues = queues
-  {-# INLINE activeMsgQueues #-}
-
-  withAllMsgQueues _ = withActiveMsgQueues
+  withAllMsgQueues _ = withQueues . queueStore
   {-# INLINE withAllMsgQueues #-}
 
   logQueueStates _ = pure ()
 
   logQueueState _ = pure ()
 
+  recipientId' = recipientId
+  {-# INLINE recipientId' #-}
+
   queueRec' = queueRec
   {-# INLINE queueRec' #-}
 
-  getMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM STMMsgQueue
-  getMsgQueue _ _ STMQueue {msgQueue_} = readTVar msgQueue_ >>= maybe newQ pure
+  msgQueue_' = msgQueue_
+  {-# INLINE msgQueue_' #-}
+
+  queueCounts :: STMMsgStore -> IO QueueCounts
+  queueCounts STMMsgStore {queueStore} = do
+    queueCount <- M.size <$> readTVarIO (queues queueStore)
+    notifierCount <- M.size <$> readTVarIO (notifiers queueStore)
+    pure QueueCounts {queueCount, notifierCount}
+
+  addQueue = addQueue'
+  {-# INLINE addQueue #-}
+
+  getQueue = getQueue' . queueStore
+  {-# INLINE getQueue #-}
+
+  secureQueue = secureQueue' . queueStore
+  {-# INLINE secureQueue #-}
+
+  addQueueNotifier = addQueueNotifier' . queueStore
+  {-# INLINE addQueueNotifier #-}
+
+  deleteQueueNotifier = deleteQueueNotifier' . queueStore
+  {-# INLINE deleteQueueNotifier #-}
+
+  suspendQueue = suspendQueue' . queueStore
+  {-# INLINE suspendQueue #-}
+
+  blockQueue = blockQueue' . queueStore
+  {-# INLINE blockQueue #-}
+
+  unblockQueue = unblockQueue' . queueStore
+  {-# INLINE unblockQueue #-}
+
+  updateQueueTime = updateQueueTime' . queueStore
+  {-# INLINE updateQueueTime #-}
+
+  getMsgQueue :: STMMsgStore -> STMQueue -> STM STMMsgQueue
+  getMsgQueue _ STMQueue {msgQueue_} = readTVar msgQueue_ >>= maybe newQ pure
     where
       newQ = do
         msgQueue <- newTQueue
@@ -108,38 +131,38 @@ instance MsgStoreClass STMMsgStore where
         writeTVar msgQueue_ (Just q)
         pure q
 
-  getPeekMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM (Maybe (STMMsgQueue, Message))
-  getPeekMsgQueue _ _ q@STMQueue {msgQueue_} = readTVar msgQueue_ $>>= \mq -> (mq,) <$$> tryPeekMsg_ q mq
+  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 -> RecipientId -> STMQueue -> (STMMsgQueue -> STM a) -> STM (Maybe a, Int)
-  withIdleMsgQueue _ _ _ STMQueue {msgQueue_} action = readTVar msgQueue_ >>= \case
+  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 -> RecipientId -> STMQueue -> IO (Either ErrorType QueueRec)
-  deleteQueue ms rId q = fst <$$> deleteQueue' ms rId q
+  deleteQueue :: STMMsgStore -> STMQueue -> IO (Either ErrorType QueueRec)
+  deleteQueue ms q = fst <$$> deleteQueue' (queueStore ms) q
 
-  deleteQueueSize :: STMMsgStore -> RecipientId -> STMQueue -> IO (Either ErrorType (QueueRec, Int))
-  deleteQueueSize ms rId q = deleteQueue' ms rId q >>= mapM (traverse getSize)
+  deleteQueueSize :: STMMsgStore -> STMQueue -> IO (Either ErrorType (QueueRec, Int))
+  deleteQueueSize ms q = deleteQueue' (queueStore ms) q >>= mapM (traverse getSize)
     -- traverse operates on the second tuple element
     where
       getSize = maybe (pure 0) (\STMMsgQueue {size} -> readTVarIO size)
 
-  getQueueMessages_ :: Bool -> STMMsgQueue -> STM [Message]
-  getQueueMessages_ drainMsgs = (if drainMsgs then flushTQueue else snapshotTQueue) . msgQueue
+  getQueueMessages_ :: Bool -> STMQueue -> STMMsgQueue -> STM [Message]
+  getQueueMessages_ drainMsgs _ = (if drainMsgs then flushTQueue else snapshotTQueue) . msgQueue
     where
       snapshotTQueue q = do
         msgs <- flushTQueue q
         mapM_ (writeTQueue q) msgs
         pure msgs
 
-  writeMsg :: STMMsgStore -> RecipientId -> STMQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
-  writeMsg ms rId q' _logState msg = liftIO $ atomically $ do
-    STMMsgQueue {msgQueue = q, canWrite, size} <- getMsgQueue ms rId q'
+  writeMsg :: STMMsgStore -> STMQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
+  writeMsg ms q' _logState msg = liftIO $ atomically $ do
+    STMMsgQueue {msgQueue = q, canWrite, size} <- getMsgQueue ms q'
     canWrt <- readTVar canWrite
     empty <- isEmptyTQueue q
     if canWrt || empty
@@ -160,6 +183,7 @@ instance MsgStoreClass STMMsgStore where
 
   getQueueSize_ :: STMMsgQueue -> STM Int
   getQueueSize_ STMMsgQueue {size} = readTVar size
+  {-# INLINE getQueueSize_ #-}
 
   tryPeekMsg_ :: STMQueue -> STMMsgQueue -> STM (Maybe Message)
   tryPeekMsg_ _ = tryPeekTQueue . msgQueue
@@ -171,5 +195,6 @@ instance MsgStoreClass STMMsgStore where
       Just _ -> modifyTVar' size (subtract 1)
       _ -> pure ()
 
-  isolateQueue :: RecipientId -> STMQueue -> String -> STM a -> ExceptT ErrorType IO a
-  isolateQueue _ _ _ = liftIO . atomically
+  isolateQueue :: STMQueue -> String -> STM a -> ExceptT ErrorType IO a
+  isolateQueue _ _ = liftIO . atomically
+  {-# INLINE isolateQueue #-}

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

@@ -8,34 +8,39 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeFamilyDependencies #-}
+{-# LANGUAGE TypeOperators #-}
 {-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}
+{-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
 
 {-# HLINT ignore "Redundant multi-way if" #-}
 
 module Simplex.Messaging.Server.MsgStore.Types where
 
 import Control.Concurrent.STM
-import Control.Monad (foldM)
 import Control.Monad.Trans.Except
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.Kind
-import qualified Data.Map.Strict as M
 import Data.Time.Clock.System (SystemTime (systemSeconds))
+import GHC.TypeLits (ErrorMessage (..), TypeError)
+import qualified GHC.TypeLits as Type
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.StoreLog.Types
 import Simplex.Messaging.TMap (TMap)
-import Simplex.Messaging.Util ((<$$>))
+import Simplex.Messaging.Util ((<$$>), ($>>=))
 import System.IO (IOMode (..))
 
-class MsgStoreClass s => STMQueueStore s where
-  queues' :: s -> TMap RecipientId (StoreQueue s)
-  senders' :: s -> TMap SenderId RecipientId
-  notifiers' :: s -> TMap NotifierId RecipientId
-  storeLog' :: s -> TVar (Maybe (StoreLog 'WriteMode))
-  mkQueue :: s -> QueueRec -> STM (StoreQueue s)
-  msgQueue_' :: StoreQueue s -> TVar (Maybe (MsgQueue s))
+data STMQueueStore q = STMQueueStore
+  { queues :: TMap RecipientId q,
+    senders :: TMap SenderId RecipientId,
+    notifiers :: TMap NotifierId RecipientId,
+    storeLog :: TVar (Maybe (StoreLog 'WriteMode))
+  }
+
+class MsgStoreClass s => STMStoreClass s where
+  stmQueueStore :: s -> STMQueueStore (StoreQueue s)
+  mkQueue :: s -> RecipientId -> QueueRec -> IO (StoreQueue s)
 
 class Monad (StoreMonad s) => MsgStoreClass s where
   type StoreMonad s = (m :: Type -> Type) | m -> s
@@ -43,58 +48,82 @@ class Monad (StoreMonad s) => MsgStoreClass s where
   type StoreQueue s = q | q -> s
   type MsgQueue s = q | q -> s
   newMsgStore :: MsgStoreConfig s -> IO s
-  setStoreLog :: s -> StoreLog 'WriteMode -> IO ()
   closeMsgStore :: s -> IO ()
-  activeMsgQueues :: s -> TMap RecipientId (StoreQueue s)
-  withAllMsgQueues :: Monoid a => Bool -> s -> (RecipientId -> StoreQueue s -> IO a) -> IO a
+  withActiveMsgQueues :: Monoid a => s -> (StoreQueue s -> IO a) -> IO a
+  withAllMsgQueues :: Monoid a => Bool -> s -> (StoreQueue s -> IO a) -> IO a
   logQueueStates :: s -> IO ()
   logQueueState :: StoreQueue s -> StoreMonad s ()
+  recipientId' :: StoreQueue s -> RecipientId
   queueRec' :: StoreQueue s -> TVar (Maybe QueueRec)
-  getPeekMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue s, Message))
-  getMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (MsgQueue s)
+  msgQueue_' :: StoreQueue s -> TVar (Maybe (MsgQueue s))
+  queueCounts :: s -> IO QueueCounts
+
+  addQueue :: s -> RecipientId -> QueueRec -> IO (Either ErrorType (StoreQueue s))
+  getQueue :: DirectParty p => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s))
+  secureQueue :: s -> StoreQueue s -> SndPublicAuthKey -> IO (Either ErrorType ())
+  addQueueNotifier :: s -> StoreQueue s -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
+  deleteQueueNotifier :: s -> StoreQueue s -> IO (Either ErrorType (Maybe NotifierId))
+  suspendQueue :: s -> StoreQueue s -> IO (Either ErrorType ())
+  blockQueue :: s -> StoreQueue s -> BlockingInfo -> IO (Either ErrorType ())
+  unblockQueue :: s -> StoreQueue s -> IO (Either ErrorType ())
+  updateQueueTime :: s -> StoreQueue s -> RoundedSystemTime -> IO (Either ErrorType QueueRec)
+
+  getPeekMsgQueue :: s -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue s, Message))
+  getMsgQueue :: s -> StoreQueue s -> StoreMonad s (MsgQueue s)
 
   -- the journal queue will be closed after action if it was initially closed or idle longer than interval in config
-  withIdleMsgQueue :: Int64 -> s -> RecipientId -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int)
-  deleteQueue :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType QueueRec)
-  deleteQueueSize :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType (QueueRec, Int))
-  getQueueMessages_ :: Bool -> MsgQueue s -> StoreMonad s [Message]
-  writeMsg :: s -> RecipientId -> StoreQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
+  withIdleMsgQueue :: Int64 -> s -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int)
+  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 :: RecipientId -> StoreQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
+  isolateQueue :: StoreQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
 
-data MSType = MSMemory | MSJournal
+data QueueCounts = QueueCounts
+  { queueCount :: Int,
+    notifierCount :: Int
+  }
+
+data MSType = MSMemory | MSHybrid | MSJournal
+  deriving (Show)
 
 data SMSType :: MSType -> Type where
   SMSMemory :: SMSType 'MSMemory
+  SMSHybrid :: SMSType 'MSHybrid
   SMSJournal :: SMSType 'MSJournal
 
+type family JournalStoreType (s :: MSType) :: Constraint where
+  JournalStoreType 'MSHybrid = ()
+  JournalStoreType 'MSJournal = ()
+  JournalStoreType p =
+    (Int ~ Bool, TypeError ('Type.Text "Store " :<>: 'ShowType p :<>: 'Type.Text " is not journal"))
+
 data AMSType = forall s. AMSType (SMSType s)
 
-withActiveMsgQueues :: (MsgStoreClass s, Monoid a) => s -> (RecipientId -> StoreQueue s -> IO a) -> IO a
-withActiveMsgQueues st f = readTVarIO (activeMsgQueues st) >>= foldM run mempty . M.assocs
-  where
-    run !acc (k, v) = do
-      r <- f k v
-      pure $! acc <> r
+getQueueRec :: (MsgStoreClass s, DirectParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s, QueueRec))
+getQueueRec st party qId =
+  getQueue st party qId
+    $>>= (\q -> maybe (Left AUTH) (Right . (q,)) <$> readTVarIO (queueRec' q))
 
-getQueueMessages :: MsgStoreClass s => Bool -> s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO [Message]
-getQueueMessages drainMsgs st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure []) (getQueueMessages_ drainMsgs . fst)
+getQueueMessages :: MsgStoreClass s => Bool -> s -> StoreQueue s -> ExceptT ErrorType IO [Message]
+getQueueMessages drainMsgs st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure []) (getQueueMessages_ drainMsgs q . fst)
 {-# INLINE getQueueMessages #-}
 
-getQueueSize :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO Int
-getQueueSize st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst)
+getQueueSize :: MsgStoreClass s => s -> StoreQueue s -> ExceptT ErrorType IO Int
+getQueueSize st q = withPeekMsgQueue st q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst)
 {-# INLINE getQueueSize #-}
 
-tryPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO (Maybe Message)
-tryPeekMsg st rId q = snd <$$> withPeekMsgQueue st rId q "tryPeekMsg" pure
+tryPeekMsg :: MsgStoreClass s => s -> StoreQueue s -> ExceptT ErrorType IO (Maybe Message)
+tryPeekMsg st q = snd <$$> withPeekMsgQueue st q "tryPeekMsg" pure
 {-# INLINE tryPeekMsg #-}
 
-tryDelMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
-tryDelMsg st rId q msgId' =
-  withPeekMsgQueue st rId q "tryDelMsg" $
+tryDelMsg :: MsgStoreClass s => 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' ->
@@ -102,30 +131,30 @@ tryDelMsg st rId q msgId' =
         | otherwise -> pure Nothing
 
 -- atomic delete (== read) last and peek next message if available
-tryDelPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
-tryDelPeekMsg st rId q msgId' =
-  withPeekMsgQueue st rId q "tryDelPeekMsg" $
+tryDelPeekMsg :: MsgStoreClass s => 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)
 
 -- The action is called with Nothing when it is known that the queue is empty
-withPeekMsgQueue :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> String -> (Maybe (MsgQueue s, Message) -> StoreMonad s a) -> ExceptT ErrorType IO a
-withPeekMsgQueue st rId q op a = isolateQueue rId q op $ getPeekMsgQueue st rId q >>= a
+withPeekMsgQueue :: MsgStoreClass s => s -> StoreQueue s -> String -> (Maybe (MsgQueue s, Message) -> StoreMonad s a) -> ExceptT ErrorType IO a
+withPeekMsgQueue st q op a = isolateQueue q op $ getPeekMsgQueue st q >>= a
 {-# INLINE withPeekMsgQueue #-}
 
-deleteExpiredMsgs :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
-deleteExpiredMsgs st rId q old =
-  isolateQueue rId q "deleteExpiredMsgs" $
-    getMsgQueue st rId q >>= deleteExpireMsgs_ old q
+deleteExpiredMsgs :: MsgStoreClass s => s -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
+deleteExpiredMsgs st q old =
+  isolateQueue q "deleteExpiredMsgs" $
+    getMsgQueue st q >>= deleteExpireMsgs_ old q
 
 -- closed and idle queues will be closed after expiration
 -- returns (expired count, queue size after expiration)
-idleDeleteExpiredMsgs :: MsgStoreClass s => Int64 -> s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO (Maybe Int, Int)
-idleDeleteExpiredMsgs now st rId q old =
-  isolateQueue rId q "idleDeleteExpiredMsgs" $
-    withIdleMsgQueue now st rId q (deleteExpireMsgs_ old q)
+idleDeleteExpiredMsgs :: MsgStoreClass s => Int64 -> s -> StoreQueue s -> Int64 -> ExceptT ErrorType IO (Maybe Int, Int)
+idleDeleteExpiredMsgs now st q old =
+  isolateQueue q "idleDeleteExpiredMsgs" $
+    withIdleMsgQueue now st q (deleteExpireMsgs_ old q)
 
 deleteExpireMsgs_ :: MsgStoreClass s => Int64 -> StoreQueue s -> MsgQueue s -> StoreMonad s Int
 deleteExpireMsgs_ old q mq = do

src/Simplex/Messaging/Server/QueueStore.hs

@@ -16,8 +16,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 
 data QueueRec = QueueRec
-  { recipientId :: !RecipientId,
-    recipientKey :: !RcvPublicAuthKey,
+  { recipientKey :: !RcvPublicAuthKey,
     rcvDhSecret :: !RcvDhSecret,
     senderId :: !SenderId,
     senderKey :: !(Maybe SndPublicAuthKey),

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

@@ -14,18 +14,22 @@
 {-# LANGUAGE UndecidableInstances #-}
 
 module Simplex.Messaging.Server.QueueStore.STM
-  ( addQueue,
-    getQueue,
-    getQueueRec,
-    secureQueue,
-    addQueueNotifier,
-    deleteQueueNotifier,
-    suspendQueue,
-    blockQueue,
-    unblockQueue,
-    updateQueueTime,
+  ( STMQueueStore (..),
+    newQueueStore,
+    setStoreLog,
+    withQueues,
+    addQueue',
+    getQueue',
+    secureQueue',
+    addQueueNotifier',
+    deleteQueueNotifier',
+    suspendQueue',
+    blockQueue',
+    unblockQueue',
+    updateQueueTime',
     deleteQueue',
     readQueueStore,
+    readQueueRec,
     withLog',
   )
 where
@@ -47,104 +51,110 @@ import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.StoreLog
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (ifM, tshow, ($>>=), (<$$))
+import Simplex.Messaging.Util (anyM, ifM, tshow, ($>>=), (<$$))
 import System.IO
 import UnliftIO.STM
 
-addQueue :: STMQueueStore s => s -> QueueRec -> IO (Either ErrorType (StoreQueue s))
-addQueue st qr@QueueRec {recipientId = rId, senderId = sId, notifier}=
-  atomically add
-    $>>= \q -> q <$$ withLog "addQueue" st (`logCreateQueue` qr)
+newQueueStore :: IO (STMQueueStore q)
+newQueueStore = do
+  queues <- TM.emptyIO
+  senders <- TM.emptyIO
+  notifiers <- TM.emptyIO
+  storeLog <- newTVarIO Nothing
+  pure STMQueueStore {queues, senders, notifiers, storeLog}
+
+setStoreLog :: STMQueueStore q -> StoreLog 'WriteMode -> IO ()
+setStoreLog st sl = atomically $ writeTVar (storeLog st) (Just sl)
+
+withQueues :: Monoid a => STMQueueStore (StoreQueue s) -> (StoreQueue s -> IO a) -> IO a
+withQueues st f = readTVarIO (queues st) >>= foldM run mempty
   where
-    add = ifM hasId (pure $ Left DUPLICATE_) $ do
-      q <- mkQueue st qr -- STMQueue lock <$> (newTVar $! Just qr) <*> newTVar Nothing
-      TM.insert rId q $ queues' st
-      TM.insert sId rId $ senders' st
-      forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId $ notifiers' st
+    run !acc = fmap (acc <>) . f
+
+addQueue' :: STMStoreClass s => s -> RecipientId -> QueueRec -> IO (Either ErrorType (StoreQueue s))
+addQueue' ms rId qr@QueueRec {senderId = sId, notifier} =
+  (mkQueue ms rId qr >>= atomically . add)
+    $>>= \q -> q <$$ withLog "addQueue" st (\s -> logCreateQueue s rId qr)
+  where
+    st = stmQueueStore ms
+    add q = ifM hasId (pure $ Left DUPLICATE_) $ do
+      TM.insert rId q $ queues st
+      TM.insert sId rId $ senders st
+      forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId $ notifiers st
       pure $ Right q
-    hasId = or <$> sequence [TM.member rId $ queues' st, TM.member sId $ senders' st, hasNotifier]
-    hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.member notifierId (notifiers' st)) notifier
+    hasId = anyM [TM.member rId $ queues st, TM.member sId $ senders st, hasNotifier]
+    hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.member notifierId (notifiers st)) notifier
 
-getQueue :: (STMQueueStore s, DirectParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s))
-getQueue st party qId =
+getQueue' :: DirectParty p => STMQueueStore q -> SParty p -> QueueId -> IO (Either ErrorType q)
+getQueue' st party qId =
   maybe (Left AUTH) Right <$> case party of
-    SRecipient -> TM.lookupIO qId $ queues' st
-    SSender -> TM.lookupIO qId (senders' st) $>>= (`TM.lookupIO` queues' st)
-    SNotifier -> TM.lookupIO qId (notifiers' st) $>>= (`TM.lookupIO` queues' st)
+    SRecipient -> TM.lookupIO qId $ queues st
+    SSender -> TM.lookupIO qId (senders st) $>>= (`TM.lookupIO` queues st)
+    SNotifier -> TM.lookupIO qId (notifiers st) $>>= (`TM.lookupIO` queues st)
 
-getQueueRec :: (STMQueueStore s, DirectParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s, QueueRec))
-getQueueRec st party qId =
-  getQueue st party qId
-    $>>= (\q -> maybe (Left AUTH) (Right . (q,)) <$> readTVarIO (queueRec' q))
-
-secureQueue :: STMQueueStore s => s -> StoreQueue s -> SndPublicAuthKey -> IO (Either ErrorType ())
-secureQueue st sq sKey =
+secureQueue' :: MsgStoreClass s => STMQueueStore (StoreQueue s) -> StoreQueue s -> SndPublicAuthKey -> IO (Either ErrorType ())
+secureQueue' st sq sKey =
   atomically (readQueueRec qr $>>= secure)
-    $>>= \rId -> withLog "secureQueue" st $ \s -> logSecureQueue s rId sKey
+    $>>= \_ -> withLog "secureQueue" st $ \s -> logSecureQueue s (recipientId' sq) sKey
   where
     qr = queueRec' sq
-    secure q@QueueRec {recipientId = rId} = case senderKey q of
-      Just k -> pure $ if sKey == k then Right rId else Left AUTH
+    secure q = case senderKey q of
+      Just k -> pure $ if sKey == k then Right () else Left AUTH
       Nothing -> do
         writeTVar qr $ Just q {senderKey = Just sKey}
-        pure $ Right rId
+        pure $ Right ()
 
-addQueueNotifier :: STMQueueStore s => s -> StoreQueue s -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
-addQueueNotifier st sq ntfCreds@NtfCreds {notifierId = nId} =
+addQueueNotifier' :: MsgStoreClass s => STMQueueStore (StoreQueue s) -> StoreQueue s -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
+addQueueNotifier' st sq ntfCreds@NtfCreds {notifierId = nId} =
   atomically (readQueueRec qr $>>= add)
-    $>>= \(rId, nId_) -> nId_ <$$ withLog "addQueueNotifier" st (\s -> logAddNotifier s rId ntfCreds)
+    $>>= \nId_ -> nId_ <$$ withLog "addQueueNotifier" st (\s -> logAddNotifier s rId ntfCreds)
   where
     qr = queueRec' sq
-    add q@QueueRec {recipientId = rId} = ifM (TM.member nId (notifiers' st)) (pure $ Left DUPLICATE_) $ do
-      nId_ <- forM (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId (notifiers' st) $> notifierId
+    rId = recipientId' sq
+    add q = ifM (TM.member nId (notifiers st)) (pure $ Left DUPLICATE_) $ do
+      nId_ <- forM (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId (notifiers st) $> notifierId
       let !q' = q {notifier = Just ntfCreds}
       writeTVar qr $ Just q'
-      TM.insert nId rId $ notifiers' st
-      pure $ Right (rId, nId_)
+      TM.insert nId rId $ notifiers st
+      pure $ Right nId_
 
-deleteQueueNotifier :: STMQueueStore s => s -> StoreQueue s -> IO (Either ErrorType (Maybe NotifierId))
-deleteQueueNotifier st sq =
+deleteQueueNotifier' :: MsgStoreClass s => STMQueueStore (StoreQueue s) -> StoreQueue s -> IO (Either ErrorType (Maybe NotifierId))
+deleteQueueNotifier' st sq =
   atomically (readQueueRec qr >>= mapM delete)
-    $>>= \(rId, nId_) -> nId_ <$$ withLog "deleteQueueNotifier" st (`logDeleteNotifier` rId)
+    $>>= \nId_ -> nId_ <$$ withLog "deleteQueueNotifier" st (`logDeleteNotifier` recipientId' sq)
   where
     qr = queueRec' sq
-    delete q = fmap (recipientId q,) $ forM (notifier q) $ \NtfCreds {notifierId} -> do
-      TM.delete notifierId $ notifiers' st
+    delete q = forM (notifier q) $ \NtfCreds {notifierId} -> do
+      TM.delete notifierId $ notifiers st
       writeTVar qr $! Just q {notifier = Nothing}
       pure notifierId
 
-suspendQueue :: STMQueueStore s => s -> StoreQueue s -> IO (Either ErrorType ())
-suspendQueue st sq =
+suspendQueue' :: MsgStoreClass s => STMQueueStore (StoreQueue s) -> StoreQueue s -> IO (Either ErrorType ())
+suspendQueue' st sq =
   atomically (readQueueRec qr >>= mapM suspend)
-    $>>= \rId -> withLog "suspendQueue" st (`logSuspendQueue` rId)
+    $>>= \_ -> withLog "suspendQueue" st (`logSuspendQueue` recipientId' sq)
   where
     qr = queueRec' sq
-    suspend q = do
-      writeTVar qr $! Just q {status = EntityOff}
-      pure $ recipientId q
+    suspend q = writeTVar qr $! Just q {status = EntityOff}
 
-blockQueue :: STMQueueStore s => s -> StoreQueue s -> BlockingInfo -> IO (Either ErrorType ())
-blockQueue st sq info =
+blockQueue' :: MsgStoreClass s => STMQueueStore (StoreQueue s) -> StoreQueue s -> BlockingInfo -> IO (Either ErrorType ())
+blockQueue' st sq info =
   atomically (readQueueRec qr >>= mapM block)
-    $>>= \rId -> withLog "blockQueue" st (\sl -> logBlockQueue sl rId info)
+    $>>= \_ -> withLog "blockQueue" st (\sl -> logBlockQueue sl (recipientId' sq) info)
   where
     qr = queueRec' sq
-    block q = do
-      writeTVar qr $ Just q {status = EntityBlocked info}
-      pure $ recipientId q
+    block q = writeTVar qr $ Just q {status = EntityBlocked info}
 
-unblockQueue :: STMQueueStore s => s -> StoreQueue s -> IO (Either ErrorType ())
-unblockQueue st sq =
+unblockQueue' :: MsgStoreClass s => STMQueueStore (StoreQueue s) -> StoreQueue s -> IO (Either ErrorType ())
+unblockQueue' st sq =
   atomically (readQueueRec qr >>= mapM unblock)
-    $>>= \rId -> withLog "unblockQueue" st (`logUnblockQueue` rId)
+    $>>= \_ -> withLog "unblockQueue" st (`logUnblockQueue` recipientId' sq)
   where
     qr = queueRec' sq
-    unblock q = do
-      writeTVar qr $ Just q {status = EntityActive}
-      pure $ recipientId q
+    unblock q = writeTVar qr $ Just q {status = EntityActive}
 
-updateQueueTime :: STMQueueStore s => s -> StoreQueue s -> RoundedSystemTime -> IO (Either ErrorType QueueRec)
-updateQueueTime st sq t = atomically (readQueueRec qr >>= mapM update) $>>= log'
+updateQueueTime' :: MsgStoreClass s => STMQueueStore (StoreQueue s) -> StoreQueue s -> RoundedSystemTime -> IO (Either ErrorType QueueRec)
+updateQueueTime' st sq t = atomically (readQueueRec qr >>= mapM update) $>>= log'
   where
     qr = queueRec' sq
     update q@QueueRec {updatedAt}
@@ -153,20 +163,20 @@ updateQueueTime st sq t = atomically (readQueueRec qr >>= mapM update) $>>= log'
           let !q' = q {updatedAt = Just t}
            in (writeTVar qr $! Just q') $> (q', True)
     log' (q, changed)
-      | changed = q <$$ withLog "updateQueueTime" st (\sl -> logUpdateQueueTime sl (recipientId q) t)
+      | changed = q <$$ withLog "updateQueueTime" st (\sl -> logUpdateQueueTime sl (recipientId' sq) t)
       | otherwise = pure $ Right q
 
-deleteQueue' :: STMQueueStore s => s -> RecipientId -> StoreQueue s -> IO (Either ErrorType (QueueRec, Maybe (MsgQueue s)))
-deleteQueue' st rId sq =
+deleteQueue' :: MsgStoreClass s => STMQueueStore (StoreQueue s) -> StoreQueue s -> IO (Either ErrorType (QueueRec, Maybe (MsgQueue s)))
+deleteQueue' st sq =
   atomically (readQueueRec qr >>= mapM delete)
-    $>>= \q -> withLog "deleteQueue" st (`logDeleteQueue` rId)
+    $>>= \q -> withLog "deleteQueue" st (`logDeleteQueue` recipientId' sq)
     >>= bimapM pure (\_ -> (q,) <$> atomically (swapTVar (msgQueue_' sq) Nothing))
   where
     qr = queueRec' sq
     delete q = do
       writeTVar qr Nothing
-      TM.delete (senderId q) $ senders' st
-      forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId $ notifiers' st
+      TM.delete (senderId q) $ senders st
+      forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId $ notifiers st
       pure q
 
 readQueueRec :: TVar (Maybe QueueRec) -> STM (Either ErrorType QueueRec)
@@ -183,34 +193,35 @@ withLog' name sl action =
       where
         err = name <> ", withLog, " <> show e
 
-withLog :: STMQueueStore s => String -> s -> (StoreLog 'WriteMode -> IO ()) -> IO (Either ErrorType ())
-withLog name = withLog' name . storeLog'
+withLog :: String -> STMQueueStore q -> (StoreLog 'WriteMode -> IO ()) -> IO (Either ErrorType ())
+withLog name = withLog' name . storeLog
 
-readQueueStore :: forall s. STMQueueStore s => FilePath -> s -> IO ()
-readQueueStore f st = withFile f ReadMode $ LB.hGetContents >=> mapM_ processLine . LB.lines
+readQueueStore :: forall s. STMStoreClass s => FilePath -> s -> IO ()
+readQueueStore f ms = withFile f ReadMode $ LB.hGetContents >=> mapM_ processLine . LB.lines
   where
+    st = stmQueueStore ms
     processLine :: LB.ByteString -> IO ()
     processLine s' = either printError procLogRecord (strDecode s)
       where
         s = LB.toStrict s'
         procLogRecord :: StoreLogRecord -> IO ()
         procLogRecord = \case
-          CreateQueue q -> addQueue st q >>= qError (recipientId q) "CreateQueue"
-          SecureQueue qId sKey -> withQueue qId "SecureQueue" $ \q -> secureQueue st q sKey
-          AddNotifier qId ntfCreds -> withQueue qId "AddNotifier" $ \q -> addQueueNotifier st q ntfCreds
-          SuspendQueue qId -> withQueue qId "SuspendQueue" $ suspendQueue st
-          BlockQueue qId info -> withQueue qId "BlockQueue" $ \q -> blockQueue st q info
-          UnblockQueue qId -> withQueue qId "UnblockQueue" $ unblockQueue st
-          DeleteQueue qId -> withQueue qId "DeleteQueue" $ deleteQueue st qId
-          DeleteNotifier qId -> withQueue qId "DeleteNotifier" $ deleteQueueNotifier st
-          UpdateTime qId t -> withQueue qId "UpdateTime" $ \q -> updateQueueTime st q t
+          CreateQueue rId q -> addQueue ms rId q >>= qError rId "CreateQueue"
+          SecureQueue qId sKey -> withQueue qId "SecureQueue" $ \q -> secureQueue' st q sKey
+          AddNotifier qId ntfCreds -> withQueue qId "AddNotifier" $ \q -> addQueueNotifier' st q ntfCreds
+          SuspendQueue qId -> withQueue qId "SuspendQueue" $ suspendQueue' st
+          BlockQueue qId info -> withQueue qId "BlockQueue" $ \q -> blockQueue' st q info
+          UnblockQueue qId -> withQueue qId "UnblockQueue" $ unblockQueue' st
+          DeleteQueue qId -> withQueue qId "DeleteQueue" $ deleteQueue' st
+          DeleteNotifier qId -> withQueue qId "DeleteNotifier" $ deleteQueueNotifier' st
+          UpdateTime qId t -> withQueue qId "UpdateTime" $ \q -> updateQueueTime' st q t
         printError :: String -> IO ()
         printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
         withQueue :: forall a. RecipientId -> T.Text -> (StoreQueue s -> IO (Either ErrorType a)) -> IO ()
         withQueue qId op a = runExceptT go >>= qError qId op
           where
             go = do
-              q <- ExceptT $ getQueue st SRecipient qId
+              q <- ExceptT $ getQueue' st SRecipient qId
               liftIO (readTVarIO $ queueRec' q) >>= \case
                 Nothing -> logWarn $ logPfx qId op <> "already deleted"
                 Just _ -> void $ ExceptT $ a q

src/Simplex/Messaging/Server/StoreLog.hs

@@ -54,7 +54,7 @@ import System.Directory (doesFileExist, renameFile)
 import System.IO
 
 data StoreLogRecord
-  = CreateQueue QueueRec
+  = CreateQueue RecipientId QueueRec
   | SecureQueue QueueId SndPublicAuthKey
   | AddNotifier QueueId NtfCreds
   | SuspendQueue QueueId
@@ -77,10 +77,9 @@ data SLRTag
   | UpdateTime_
 
 instance StrEncoding QueueRec where
-  strEncode QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt} =
+  strEncode QueueRec {recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt} =
     B.unwords
-      [ "rid=" <> strEncode recipientId,
-        "rk=" <> strEncode recipientKey,
+      [ "rk=" <> strEncode recipientKey,
         "rdh=" <> strEncode rcvDhSecret,
         "sid=" <> strEncode senderId,
         "sk=" <> strEncode senderKey
@@ -98,7 +97,6 @@ instance StrEncoding QueueRec where
         _ -> " status=" <> strEncode status
 
   strP = do
-    recipientId <- "rid=" *> strP_
     recipientKey <- "rk=" *> strP_
     rcvDhSecret <- "rdh=" *> strP_
     senderId <- "sid=" *> strP_
@@ -107,7 +105,7 @@ instance StrEncoding QueueRec where
     notifier <- optional $ " notifier=" *> strP
     updatedAt <- optional $ " updated_at=" *> strP
     status <- (" status=" *> strP) <|> pure EntityActive
-    pure QueueRec {recipientId, recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt}
+    pure QueueRec {recipientKey, rcvDhSecret, senderId, senderKey, sndSecure, notifier, status, updatedAt}
 
 instance StrEncoding SLRTag where
   strEncode = \case
@@ -136,7 +134,7 @@ instance StrEncoding SLRTag where
 
 instance StrEncoding StoreLogRecord where
   strEncode = \case
-    CreateQueue q -> strEncode (CreateQueue_, q)
+    CreateQueue rId q -> strEncode (CreateQueue_, Str $ "rid=" <> strEncode rId, q)
     SecureQueue rId sKey -> strEncode (SecureQueue_, rId, sKey)
     AddNotifier rId ntfCreds -> strEncode (AddNotifier_, rId, ntfCreds)
     SuspendQueue rId -> strEncode (SuspendQueue_, rId)
@@ -148,7 +146,7 @@ instance StrEncoding StoreLogRecord where
 
   strP =
     strP_ >>= \case
-      CreateQueue_ -> CreateQueue <$> strP
+      CreateQueue_ -> CreateQueue <$> ("rid=" *> strP_) <*> strP
       SecureQueue_ -> SecureQueue <$> strP_ <*> strP
       AddNotifier_ -> AddNotifier <$> strP_ <*> strP
       SuspendQueue_ -> SuspendQueue <$> strP
@@ -186,8 +184,8 @@ writeStoreLogRecord (WriteStoreLog _ h) r = E.uninterruptibleMask_ $ do
   B.hPut h $ strEncode r `B.snoc` '\n' -- hPutStrLn makes write non-atomic for length > 1024
   hFlush h
 
-logCreateQueue :: StoreLog 'WriteMode -> QueueRec -> IO ()
-logCreateQueue s = writeStoreLogRecord s . CreateQueue
+logCreateQueue :: StoreLog 'WriteMode -> RecipientId -> QueueRec -> IO ()
+logCreateQueue s rId q = writeStoreLogRecord s $ CreateQueue rId q
 
 logSecureQueue :: StoreLog 'WriteMode -> QueueId -> SndPublicAuthKey -> IO ()
 logSecureQueue s qId sKey = writeStoreLogRecord s $ SecureQueue qId sKey
@@ -248,10 +246,11 @@ readWriteStoreLog readStore writeStore f st =
       renameFile tempBackup timedBackup
       logInfo $ "original state preserved as " <> T.pack timedBackup
 
-writeQueueStore :: STMQueueStore s => StoreLog 'WriteMode -> s -> IO ()
-writeQueueStore s st = readTVarIO (activeMsgQueues st) >>= mapM_ writeQueue . M.assocs
+writeQueueStore :: STMStoreClass s => StoreLog 'WriteMode -> s -> IO ()
+writeQueueStore s st = readTVarIO qs >>= mapM_ writeQueue . M.assocs
   where
+    qs = queues $ stmQueueStore st
     writeQueue (rId, q) =
       readTVarIO (queueRec' q) >>= \case
-        Just q' -> logCreateQueue s q'
-        Nothing -> atomically $ TM.delete rId $ activeMsgQueues st
+        Just q' -> logCreateQueue s rId q'
+        Nothing -> atomically $ TM.delete rId qs

src/Simplex/Messaging/Util.hs

@@ -85,6 +85,10 @@ unlessM :: Monad m => m Bool -> m () -> m ()
 unlessM b = ifM b $ pure ()
 {-# INLINE unlessM #-}
 
+anyM :: Monad m => [m Bool] -> m Bool
+anyM = foldM (\r a -> if r then pure r else (r ||) <$!> a) False
+{-# INLINE anyM #-}
+
 ($>>=) :: (Monad m, Monad f, Traversable f) => m (f a) -> (a -> m (f b)) -> m (f b)
 f $>>= g = f >>= fmap join . mapM g
 

tests/AgentTests/FunctionalAPITests.hs

@@ -132,7 +132,7 @@ a =##> p =
 
 withTimeout :: (HasCallStack, MonadUnliftIO m) => m a -> (HasCallStack => a -> Expectation) -> m ()
 withTimeout a test =
-  timeout 10_000000 a >>= \case
+  timeout 100_000000 a >>= \case
     Nothing -> error "operation timed out"
     Just t -> liftIO $ test t
 

tests/CoreTests/MsgStoreTests.hs

@@ -23,7 +23,6 @@ import Control.Monad.Trans.Except
 import Crypto.Random (ChaChaDRG)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Base64.URL as B64
 import Data.Maybe (fromJust)
 import Data.Time.Clock.System (getSystemTime)
 import Simplex.Messaging.Crypto (pattern MaxLenBS)
@@ -35,7 +34,6 @@ import Simplex.Messaging.Server.MsgStore.Journal
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
-import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.StoreLog (closeStoreLog, logCreateQueue)
 import SMPClient (testStoreLogFile, testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
 import System.Directory (copyFile, createDirectoryIfMissing, listDirectory, removeFile, renameFile)
@@ -46,34 +44,41 @@ import Test.Hspec
 msgStoreTests :: Spec
 msgStoreTests = do
   around (withMsgStore testSMTStoreConfig) $ describe "STM message store" someMsgStoreTests
-  around (withMsgStore testJournalStoreCfg) $ describe "Journal message store" $ do
-    someMsgStoreTests
-    it "should export and import journal store" testExportImportStore
-    describe "queue state" $ do
-      it "should restore queue state from the last line" testQueueState
-      it "should recover when message is written and state is not" testMessageState
-    describe "missing files" $ do
-      it "should create read file when missing" testReadFileMissing
-      it "should switch to write file when read file missing" testReadFileMissingSwitch
-      it "should create write file when missing" testWriteFileMissing
-      it "should create read file when read and write files are missing" testReadAndWriteFilesMissing
+  around (withMsgStore $ testJournalStoreCfg SMSHybrid) $
+    describe "Hybrid message store" $ do
+      journalMsgStoreTests
+      it "should export and import journal store" testExportImportStore
+  around (withMsgStore $ testJournalStoreCfg SMSJournal) $
+    describe "Journal message store" journalMsgStoreTests
   where
-    someMsgStoreTests :: STMQueueStore s => SpecWith s
+    journalMsgStoreTests :: JournalStoreType s => SpecWith (JournalMsgStore s)
+    journalMsgStoreTests = do
+      someMsgStoreTests
+      describe "queue state" $ do
+        it "should restore queue state from the last line" testQueueState
+        it "should recover when message is written and state is not" testMessageState
+      describe "missing files" $ do
+        it "should create read file when missing" testReadFileMissing
+        it "should switch to write file when read file missing" testReadFileMissingSwitch
+        it "should create write file when missing" testWriteFileMissing
+        it "should create read file when read and write files are missing" testReadAndWriteFilesMissing
+    someMsgStoreTests :: MsgStoreClass s => SpecWith s
     someMsgStoreTests = do
       it "should get queue and store/read messages" testGetQueue
       it "should not fail on EOF when changing read journal" testChangeReadJournal
 
-withMsgStore :: STMQueueStore s => MsgStoreConfig s -> (s -> IO ()) -> IO ()
+withMsgStore :: MsgStoreClass s => MsgStoreConfig s -> (s -> IO ()) -> IO ()
 withMsgStore cfg = bracket (newMsgStore cfg) closeMsgStore
 
 testSMTStoreConfig :: STMStoreConfig
 testSMTStoreConfig = STMStoreConfig {storePath = Nothing, quota = 3}
 
-testJournalStoreCfg :: JournalStoreConfig
-testJournalStoreCfg =
+testJournalStoreCfg :: SMSType s -> JournalStoreConfig s
+testJournalStoreCfg queueStoreType =
   JournalStoreConfig
     { storePath = testStoreMsgsDir,
       pathParts = journalMsgStoreDepth,
+      queueStoreType,
       quota = 3,
       maxMsgCount = 4,
       maxStateLines = 2,
@@ -105,8 +110,7 @@ testNewQueueRec g sndSecure = do
   (k, pk) <- atomically $ C.generateKeyPair @'C.X25519 g
   let qr =
         QueueRec
-          { recipientId = rId,
-            recipientKey,
+          { recipientKey,
             rcvDhSecret = C.dh' k pk,
             senderId,
             senderKey = Nothing,
@@ -117,89 +121,89 @@ testNewQueueRec g sndSecure = do
           }
   pure (rId, qr)
 
-testGetQueue :: STMQueueStore s => s -> IO ()
+testGetQueue :: MsgStoreClass s => s -> IO ()
 testGetQueue ms = do
   g <- C.newRandom
   (rId, qr) <- testNewQueueRec g True
   runRight_ $ do
-    q <- ExceptT $ addQueue ms qr
-    let write s = writeMsg ms rId q True =<< mkMessage s
+    q <- ExceptT $ addQueue ms rId qr
+    let write s = writeMsg ms q True =<< mkMessage s
     Just (Message {msgId = mId1}, True) <- write "message 1"
     Just (Message {msgId = mId2}, False) <- write "message 2"
     Just (Message {msgId = mId3}, False) <- write "message 3"
-    Msg "message 1" <- tryPeekMsg ms rId q
-    Msg "message 1" <- tryPeekMsg ms rId q
-    Nothing <- tryDelMsg ms rId q mId2
-    Msg "message 1" <- tryDelMsg ms rId q mId1
-    Nothing <- tryDelMsg ms rId q mId1
-    Msg "message 2" <- tryPeekMsg ms rId q
-    Nothing <- tryDelMsg ms rId q mId1
-    (Nothing, Msg "message 2") <- tryDelPeekMsg ms rId q mId1
-    (Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms rId q mId2
-    (Nothing, Msg "message 3") <- tryDelPeekMsg ms rId q mId2
-    Msg "message 3" <- tryPeekMsg ms rId q
-    (Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
-    Nothing <- tryDelMsg ms rId q mId2
-    Nothing <- tryDelMsg ms rId q mId3
-    Nothing <- tryPeekMsg ms rId q
+    Msg "message 1" <- tryPeekMsg ms q
+    Msg "message 1" <- tryPeekMsg ms q
+    Nothing <- tryDelMsg ms q mId2
+    Msg "message 1" <- tryDelMsg ms q mId1
+    Nothing <- tryDelMsg ms q mId1
+    Msg "message 2" <- tryPeekMsg ms q
+    Nothing <- tryDelMsg ms q mId1
+    (Nothing, Msg "message 2") <- tryDelPeekMsg ms q mId1
+    (Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms q mId2
+    (Nothing, Msg "message 3") <- tryDelPeekMsg ms q mId2
+    Msg "message 3" <- tryPeekMsg ms q
+    (Msg "message 3", Nothing) <- tryDelPeekMsg ms q mId3
+    Nothing <- tryDelMsg ms q mId2
+    Nothing <- tryDelMsg ms q mId3
+    Nothing <- tryPeekMsg ms q
     Just (Message {msgId = mId4}, True) <- write "message 4"
-    Msg "message 4" <- tryPeekMsg ms rId q
+    Msg "message 4" <- tryPeekMsg ms q
     Just (Message {msgId = mId5}, False) <- write "message 5"
-    (Nothing, Msg "message 4") <- tryDelPeekMsg ms rId q mId3
-    (Msg "message 4", Msg "message 5") <- tryDelPeekMsg ms rId q mId4
+    (Nothing, Msg "message 4") <- tryDelPeekMsg ms q mId3
+    (Msg "message 4", Msg "message 5") <- tryDelPeekMsg ms q mId4
     Just (Message {msgId = mId6}, False) <- write "message 6"
     Just (Message {msgId = mId7}, False) <- write "message 7"
     Nothing <- write "message 8"
-    Msg "message 5" <- tryPeekMsg ms rId q
-    (Nothing, Msg "message 5") <- tryDelPeekMsg ms rId q mId4
-    (Msg "message 5", Msg "message 6") <- tryDelPeekMsg ms rId q mId5
-    (Msg "message 6", Msg "message 7") <- tryDelPeekMsg ms rId q mId6
-    (Msg "message 7", Just MessageQuota {msgId = mId8}) <- tryDelPeekMsg ms rId q mId7
-    (Just MessageQuota {}, Nothing) <- tryDelPeekMsg ms rId q mId8
-    (Nothing, Nothing) <- tryDelPeekMsg ms rId q mId8
-    void $ ExceptT $ deleteQueue ms rId q
+    Msg "message 5" <- tryPeekMsg ms q
+    (Nothing, Msg "message 5") <- tryDelPeekMsg ms q mId4
+    (Msg "message 5", Msg "message 6") <- tryDelPeekMsg ms q mId5
+    (Msg "message 6", Msg "message 7") <- tryDelPeekMsg ms q mId6
+    (Msg "message 7", Just MessageQuota {msgId = mId8}) <- tryDelPeekMsg ms q mId7
+    (Just MessageQuota {}, Nothing) <- tryDelPeekMsg ms q mId8
+    (Nothing, Nothing) <- tryDelPeekMsg ms q mId8
+    void $ ExceptT $ deleteQueue ms q
 
-testChangeReadJournal :: STMQueueStore s => s -> IO ()
+testChangeReadJournal :: MsgStoreClass s => s -> IO ()
 testChangeReadJournal ms = do
   g <- C.newRandom
   (rId, qr) <- testNewQueueRec g True
   runRight_ $ do
-    q <- ExceptT $ addQueue ms qr
-    let write s = writeMsg ms rId q True =<< mkMessage s
+    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 rId q mId1
+    (Msg "message 1", Nothing) <- tryDelPeekMsg ms q mId1
     Just (Message {msgId = mId2}, True) <- write "message 2"
-    (Msg "message 2", Nothing) <- tryDelPeekMsg ms rId q mId2
+    (Msg "message 2", Nothing) <- tryDelPeekMsg ms q mId2
     Just (Message {msgId = mId3}, True) <- write "message 3"
-    (Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
+    (Msg "message 3", Nothing) <- tryDelPeekMsg ms q mId3
     Just (Message {msgId = mId4}, True) <- write "message 4"
-    (Msg "message 4", Nothing) <- tryDelPeekMsg ms rId q mId4
+    (Msg "message 4", Nothing) <- tryDelPeekMsg ms q mId4
     Just (Message {msgId = mId5}, True) <- write "message 5"
-    (Msg "message 5", Nothing) <- tryDelPeekMsg ms rId q mId5
-    void $ ExceptT $ deleteQueue ms rId q
+    (Msg "message 5", Nothing) <- tryDelPeekMsg ms q mId5
+    void $ ExceptT $ deleteQueue ms q
 
-testExportImportStore :: JournalMsgStore -> IO ()
+testExportImportStore :: JournalMsgStore 'MSHybrid -> IO ()
 testExportImportStore ms = do
   g <- C.newRandom
   (rId1, qr1) <- testNewQueueRec g True
   (rId2, qr2) <- testNewQueueRec g True
   sl <- readWriteQueueStore testStoreLogFile ms
   runRight_ $ do
-    let write rId q s = writeMsg ms rId q True =<< mkMessage s
-    q1 <- ExceptT $ addQueue ms qr1
-    liftIO $ logCreateQueue sl qr1
-    Just (Message {}, True) <- write rId1 q1 "message 1"
-    Just (Message {}, False) <- write rId1 q1 "message 2"
-    q2 <- ExceptT $ addQueue ms qr2
-    liftIO $ logCreateQueue sl qr2
-    Just (Message {msgId = mId3}, True) <- write rId2 q2 "message 3"
-    Just (Message {msgId = mId4}, False) <- write rId2 q2 "message 4"
-    (Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms rId2 q2 mId3
-    (Msg "message 4", Nothing) <- tryDelPeekMsg ms rId2 q2 mId4
-    Just (Message {}, True) <- write rId2 q2 "message 5"
-    Just (Message {}, False) <- write rId2 q2 "message 6"
-    Just (Message {}, False) <- write rId2 q2 "message 7"
-    Nothing <- write rId2 q2 "message 8"
+    let write q s = writeMsg ms q True =<< mkMessage s
+    q1 <- ExceptT $ addQueue ms rId1 qr1
+    liftIO $ logCreateQueue sl rId1 qr1
+    Just (Message {}, True) <- write q1 "message 1"
+    Just (Message {}, False) <- write q1 "message 2"
+    q2 <- ExceptT $ addQueue ms rId2 qr2
+    liftIO $ logCreateQueue sl rId2 qr2
+    Just (Message {msgId = mId3}, True) <- write q2 "message 3"
+    Just (Message {msgId = mId4}, False) <- write q2 "message 4"
+    (Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms q2 mId3
+    (Msg "message 4", Nothing) <- tryDelPeekMsg ms q2 mId4
+    Just (Message {}, True) <- write q2 "message 5"
+    Just (Message {}, False) <- write q2 "message 6"
+    Just (Message {}, False) <- write q2 "message 7"
+    Nothing <- write q2 "message 8"
     pure ()
   length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
   length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 3
@@ -209,7 +213,7 @@ testExportImportStore ms = do
   closeStoreLog sl
   exportMessages False ms testStoreMsgsFile False
   (B.readFile testStoreMsgsFile `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".copy")
-  let cfg = (testJournalStoreCfg :: JournalStoreConfig) {storePath = testStoreMsgsDir2}
+  let cfg = (testJournalStoreCfg SMSHybrid :: JournalStoreConfig 'MSHybrid) {storePath = testStoreMsgsDir2}
   ms' <- newMsgStore cfg
   readWriteQueueStore testStoreLogFile ms' >>= closeStoreLog
   stats@MessageStats {storedMsgsCount = 5, expiredMsgsCount = 0, storedQueues = 2} <-
@@ -226,12 +230,12 @@ testExportImportStore ms = do
   exportMessages False stmStore testStoreMsgsFile False
   (B.sort <$> B.readFile testStoreMsgsFile `shouldReturn`) =<< (B.sort <$> B.readFile (testStoreMsgsFile2 <> ".bak"))
 
-testQueueState :: JournalMsgStore -> IO ()
+testQueueState :: JournalMsgStore s -> IO ()
 testQueueState ms = do
   g <- C.newRandom
   rId <- EntityId <$> atomically (C.randomBytes 24 g)
   let dir = msgQueueDirectory ms rId
-      statePath = msgQueueStatePath dir $ B.unpack (B64.encode $ unEntityId rId)
+      statePath = msgQueueStatePath dir rId
   createDirectoryIfMissing True dir
   state <- newMsgQueueState <$> newJournalId (random ms)
   withFile statePath WriteMode (`appendState` state)
@@ -291,16 +295,16 @@ testQueueState ms = do
         let f = dir </> name
          in unless (f == keep) $ removeFile f
 
-testMessageState :: JournalMsgStore -> IO ()
+testMessageState :: JournalStoreType s => JournalMsgStore s -> IO ()
 testMessageState ms = do
   g <- C.newRandom
   (rId, qr) <- testNewQueueRec g True
   let dir = msgQueueDirectory ms rId
-      statePath = msgQueueStatePath dir $ B.unpack (B64.encode $ unEntityId rId)
-      write q s = writeMsg ms rId q True =<< mkMessage s
+      statePath = msgQueueStatePath dir rId
+      write q s = writeMsg ms q True =<< mkMessage s
 
   mId1 <- runRight $ do
-    q <- ExceptT $ addQueue ms qr
+    q <- ExceptT $ addQueue ms rId qr
     Just (Message {msgId = mId1}, True) <- write q "message 1"
     Just (Message {}, False) <- write q "message 2"
     liftIO $ closeMsgQueue q
@@ -312,35 +316,35 @@ testMessageState ms = do
   runRight_ $ do
     q <- ExceptT $ getQueue ms SRecipient rId
     Just (Message {msgId = mId3}, False) <- write q "message 3"
-    (Msg "message 1", Msg "message 3") <- tryDelPeekMsg ms rId q mId1
-    (Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
+    (Msg "message 1", Msg "message 3") <- tryDelPeekMsg ms q mId1
+    (Msg "message 3", Nothing) <- tryDelPeekMsg ms q mId3
     liftIO $ closeMsgQueue q
 
-testReadFileMissing :: JournalMsgStore -> IO ()
+testReadFileMissing :: JournalStoreType s => JournalMsgStore s -> IO ()
 testReadFileMissing ms = do
   g <- C.newRandom
   (rId, qr) <- testNewQueueRec g True
-  let write q s = writeMsg ms rId q True =<< mkMessage s
+  let write q s = writeMsg ms q True =<< mkMessage s
   q <- runRight $ do
-    q <- ExceptT $ addQueue ms qr
+    q <- ExceptT $ addQueue ms rId qr
     Just (Message {}, True) <- write q "message 1"
-    Msg "message 1" <- tryPeekMsg ms rId q
+    Msg "message 1" <- tryPeekMsg ms q
     pure q
 
   mq <- fromJust <$> readTVarIO (msgQueue_' q)
   MsgQueueState {readState = rs} <- readTVarIO $ state mq
   closeMsgStore ms
-  let path = journalFilePath (queueDirectory $ queue mq) $ journalId rs
+  let path = journalFilePath (queueDirectory q) $ journalId rs
   removeFile path
 
   runRight_ $ do
     q' <- ExceptT $ getQueue ms SRecipient rId
-    Nothing <- tryPeekMsg ms rId q'
+    Nothing <- tryPeekMsg ms q'
     Just (Message {}, True) <- write q' "message 2"
-    Msg "message 2" <- tryPeekMsg ms rId q'
+    Msg "message 2" <- tryPeekMsg ms q'
     pure ()
 
-testReadFileMissingSwitch :: JournalMsgStore -> IO ()
+testReadFileMissingSwitch :: JournalStoreType s => JournalMsgStore s -> IO ()
 testReadFileMissingSwitch ms = do
   g <- C.newRandom
   (rId, qr) <- testNewQueueRec g True
@@ -349,16 +353,16 @@ testReadFileMissingSwitch ms = do
   mq <- fromJust <$> readTVarIO (msgQueue_' q)
   MsgQueueState {readState = rs} <- readTVarIO $ state mq
   closeMsgStore ms
-  let path = journalFilePath (queueDirectory $ queue mq) $ journalId rs
+  let path = journalFilePath (queueDirectory q) $ journalId rs
   removeFile path
 
   runRight_ $ do
     q' <- ExceptT $ getQueue ms SRecipient rId
-    Just (Message {}, False) <- writeMsg ms rId q' True =<< mkMessage "message 6"
-    Msg "message 5" <- tryPeekMsg ms rId q'
+    Just (Message {}, False) <- writeMsg ms q' True =<< mkMessage "message 6"
+    Msg "message 5" <- tryPeekMsg ms q'
     pure ()
 
-testWriteFileMissing :: JournalMsgStore -> IO ()
+testWriteFileMissing :: JournalStoreType s => JournalMsgStore s -> IO ()
 testWriteFileMissing ms = do
   g <- C.newRandom
   (rId, qr) <- testNewQueueRec g True
@@ -367,21 +371,21 @@ testWriteFileMissing ms = do
   mq <- fromJust <$> readTVarIO (msgQueue_' q)
   MsgQueueState {writeState = ws} <- readTVarIO $ state mq
   closeMsgStore ms
-  let path = journalFilePath (queueDirectory $ queue mq) $ journalId ws
+  let path = journalFilePath (queueDirectory q) $ journalId ws
   print path
   removeFile path
 
   runRight_ $ do
     q' <- ExceptT $ getQueue ms SRecipient rId
-    Just Message {msgId = mId3} <- tryPeekMsg ms rId q'
-    (Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms rId q' mId3
-    Just Message {msgId = mId4} <- tryPeekMsg ms rId q'
-    (Msg "message 4", Nothing) <- tryDelPeekMsg ms rId q' mId4
-    Just (Message {}, True) <- writeMsg ms rId q' True =<< mkMessage "message 6"
-    Msg "message 6" <- tryPeekMsg ms rId q'
+    Just Message {msgId = mId3} <- tryPeekMsg ms q'
+    (Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms q' mId3
+    Just Message {msgId = mId4} <- tryPeekMsg ms q'
+    (Msg "message 4", Nothing) <- tryDelPeekMsg ms q' mId4
+    Just (Message {}, True) <- writeMsg ms q' True =<< mkMessage "message 6"
+    Msg "message 6" <- tryPeekMsg ms q'
     pure ()
 
-testReadAndWriteFilesMissing :: JournalMsgStore -> IO ()
+testReadAndWriteFilesMissing :: JournalStoreType s => JournalMsgStore s -> IO ()
 testReadAndWriteFilesMissing ms = do
   g <- C.newRandom
   (rId, qr) <- testNewQueueRec g True
@@ -390,25 +394,25 @@ testReadAndWriteFilesMissing ms = do
   mq <- fromJust <$> readTVarIO (msgQueue_' q)
   MsgQueueState {readState = rs, writeState = ws} <- readTVarIO $ state mq
   closeMsgStore ms
-  removeFile $ journalFilePath (queueDirectory $ queue mq) $ journalId rs
-  removeFile $ journalFilePath (queueDirectory $ queue mq) $ journalId ws
+  removeFile $ journalFilePath (queueDirectory q) $ journalId rs
+  removeFile $ journalFilePath (queueDirectory q) $ journalId ws
 
   runRight_ $ do
     q' <- ExceptT $ getQueue ms SRecipient rId
-    Nothing <- tryPeekMsg ms rId q'
-    Just (Message {}, True) <- writeMsg ms rId q' True =<< mkMessage "message 6"
-    Msg "message 6" <- tryPeekMsg ms rId q'
+    Nothing <- tryPeekMsg ms q'
+    Just (Message {}, True) <- writeMsg ms q' True =<< mkMessage "message 6"
+    Msg "message 6" <- tryPeekMsg ms q'
     pure ()
 
-writeMessages :: JournalMsgStore -> RecipientId -> QueueRec -> IO JournalQueue
+writeMessages :: JournalStoreType s => JournalMsgStore s -> RecipientId -> QueueRec -> IO (JournalQueue s)
 writeMessages ms rId qr = runRight $ do
-  q <- ExceptT $ addQueue ms qr
-  let write s = writeMsg ms rId q True =<< mkMessage s
+  q <- ExceptT $ addQueue ms rId qr
+  let write s = writeMsg ms q True =<< mkMessage s
   Just (Message {msgId = mId1}, True) <- write "message 1"
   Just (Message {msgId = mId2}, False) <- write "message 2"
   Just (Message {}, False) <- write "message 3"
-  (Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms rId q mId1
-  (Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms rId q mId2
+  (Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms q mId1
+  (Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms q mId2
   Just (Message {}, False) <- write "message 4"
   Just (Message {}, False) <- write "message 5"
   pure q

tests/CoreTests/StoreLogTests.hs

@@ -60,38 +60,38 @@ storeLogTests =
       ("SMP server store log, sndSecure = " <> show sndSecure)
       [ SLTC
           { name = "create new queue",
-            saved = [CreateQueue qr],
-            compacted = [CreateQueue qr],
+            saved = [CreateQueue rId qr],
+            compacted = [CreateQueue rId qr],
             state = M.fromList [(rId, qr)]
           },
         SLTC
           { name = "secure queue",
-            saved = [CreateQueue qr, SecureQueue rId testPublicAuthKey],
-            compacted = [CreateQueue qr {senderKey = Just testPublicAuthKey}],
+            saved = [CreateQueue rId qr, SecureQueue rId testPublicAuthKey],
+            compacted = [CreateQueue rId qr {senderKey = Just testPublicAuthKey}],
             state = M.fromList [(rId, qr {senderKey = Just testPublicAuthKey})]
           },
         SLTC
           { name = "create and delete queue",
-            saved = [CreateQueue qr, DeleteQueue rId],
+            saved = [CreateQueue rId qr, DeleteQueue rId],
             compacted = [],
             state = M.fromList []
           },
         SLTC
           { name = "create queue and add notifier",
-            saved = [CreateQueue qr, AddNotifier rId ntfCreds],
-            compacted = [CreateQueue $ qr {notifier = Just ntfCreds}],
+            saved = [CreateQueue rId qr, AddNotifier rId ntfCreds],
+            compacted = [CreateQueue rId qr {notifier = Just ntfCreds}],
             state = M.fromList [(rId, qr {notifier = Just ntfCreds})]
           },
         SLTC
           { name = "delete notifier",
-            saved = [CreateQueue qr, AddNotifier rId ntfCreds, DeleteNotifier rId],
-            compacted = [CreateQueue qr],
+            saved = [CreateQueue rId qr, AddNotifier rId ntfCreds, DeleteNotifier rId],
+            compacted = [CreateQueue rId qr],
             state = M.fromList [(rId, qr)]
           },
         SLTC
           { name = "update time",
-            saved = [CreateQueue qr, UpdateTime rId date],
-            compacted = [CreateQueue qr {updatedAt = Just date}],
+            saved = [CreateQueue rId qr, UpdateTime rId date],
+            compacted = [CreateQueue rId qr {updatedAt = Just date}],
             state = M.fromList [(rId, qr {updatedAt = Just date})]
           }
       ]
@@ -105,11 +105,11 @@ testSMPStoreLog testSuite tests =
     replicateM_ 3 $ testReadWrite t
   where
     testReadWrite SLTC {compacted, state} = do
-      st <- newMsgStore testJournalStoreCfg
+      st <- newMsgStore $ testJournalStoreCfg SMSHybrid
       l <- readWriteQueueStore testStoreLogFile st
       storeState st `shouldReturn` state
       closeStoreLog l
       ([], compacted') <- partitionEithers . map strDecode . B.lines <$> B.readFile testStoreLogFile
       compacted' `shouldBe` compacted
-    storeState :: JournalMsgStore -> IO (M.Map RecipientId QueueRec)
-    storeState st = M.mapMaybe id <$> (readTVarIO (queues st) >>= mapM (readTVarIO . queueRec'))
+    storeState :: JournalMsgStore 'MSHybrid -> IO (M.Map RecipientId QueueRec)
+    storeState st = M.mapMaybe id <$> (readTVarIO (queues $ stmQueueStore st) >>= mapM (readTVarIO . queueRec'))

tests/SMPClient.hs

@@ -130,8 +130,9 @@ cfgMS msType =
       msgIdBytes = 24,
       storeLogFile = Just testStoreLogFile,
       storeMsgsFile = Just $ case msType of
-        AMSType SMSJournal -> testStoreMsgsDir
-        AMSType SMSMemory -> testStoreMsgsFile,
+        AMSType SMSMemory -> testStoreMsgsFile
+        AMSType SMSHybrid -> testStoreMsgsDir
+        AMSType SMSJournal -> testStoreMsgsDir,
       storeNtfsFile = Nothing,
       allowNewQueues = True,
       newQueueBasicAuth = Nothing,

tests/ServerTests.hs

@@ -17,7 +17,7 @@ module ServerTests where
 
 import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
-import Control.Exception (SomeException, try)
+import Control.Exception (SomeException, throwIO, try)
 import Control.Monad
 import Control.Monad.IO.Class
 import CoreTests.MsgStoreTests (testJournalStoreCfg)
@@ -575,7 +575,6 @@ testWithStoreLog =
     senderId1 <- newTVarIO NoEntity
     senderId2 <- newTVarIO NoEntity
     notifierId <- newTVarIO NoEntity
-
     withSmpServerStoreLogOnMS at msType testPort . runTest t $ \h -> runClient t $ \h1 -> do
       (sId1, rId1, rKey1, dhShared) <- createAndSecureQueue h sPub1
       (rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
@@ -606,16 +605,14 @@ testWithStoreLog =
 
       Resp "dabc" _ OK <- signSendRecv h rKey2 ("dabc", rId2, DEL)
       pure ()
-
-    logSize testStoreLogFile `shouldReturn` 6
-
+    withHybridStore msType $
+      logSize testStoreLogFile `shouldReturn` 6
     let cfg' = (cfgMS msType) {msgStoreType = AMSType SMSMemory, storeLogFile = Nothing, storeMsgsFile = Nothing}
     withSmpServerConfigOn at cfg' testPort . runTest t $ \h -> do
       sId1 <- readTVarIO senderId1
       -- fails if store log is disabled
       Resp "bcda" _ (ERR AUTH) <- signSendRecv h sKey1 ("bcda", sId1, _SEND "hello")
       pure ()
-
     withSmpServerStoreLogOnMS at msType testPort . runTest t $ \h -> runClient t $ \h1 -> do
       -- this queue is restored
       rId1 <- readTVarIO recipientId1
@@ -632,9 +629,9 @@ testWithStoreLog =
       sId2 <- readTVarIO senderId2
       Resp "cdab" _ (ERR AUTH) <- signSendRecv h sKey2 ("cdab", sId2, _SEND "hello too")
       pure ()
-
-    logSize testStoreLogFile `shouldReturn` 1
-    removeFile testStoreLogFile
+    withHybridStore msType $ do
+      logSize testStoreLogFile `shouldReturn` 1
+      removeFile testStoreLogFile
   where
     runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
     runTest _ test' server = do
@@ -644,11 +641,20 @@ testWithStoreLog =
     runClient :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> Expectation
     runClient _ test' = testSMPClient test' `shouldReturn` ()
 
+withHybridStore :: AMSType -> IO () -> IO ()
+withHybridStore msType a = case msType of
+  AMSType SMSHybrid -> a
+  _ -> pure ()
+
 logSize :: FilePath -> IO Int
-logSize f =
-  try (length . B.lines <$> B.readFile f) >>= \case
-    Right l -> pure l
-    Left (_ :: SomeException) -> logSize f
+logSize f = go (3 :: Int)
+  where
+    go n =
+      try (length . B.lines <$> B.readFile f) >>= \case
+        Right l -> pure l
+        Left (e :: SomeException)
+          | n == 0 -> throwIO e
+          | otherwise -> threadDelay 100000 >> go (n - 1)
 
 testRestoreMessages :: SpecWith (ATransport, AMSType)
 testRestoreMessages =
@@ -688,7 +694,8 @@ testRestoreMessages =
 
     rId <- readTVarIO recipientId
 
-    logSize testStoreLogFile `shouldReturn` 2
+    withHybridStore msType $
+      logSize testStoreLogFile `shouldReturn` 2
     -- logSize testStoreMsgsFile `shouldReturn` 5
     logSize testServerStatsBackupFile `shouldReturn` 76
     Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile
@@ -705,7 +712,7 @@ testRestoreMessages =
       Resp "4" _ (Msg mId4 msg4) <- signSendRecv h rKey ("4", rId, ACK mId3)
       (dec mId4 msg4, Right "hello 4") #== "restored message delivered"
 
-    logSize testStoreLogFile `shouldReturn` 1
+    withHybridStore msType $ logSize testStoreLogFile `shouldReturn` 1
     -- the last message is not removed because it was not ACK'd
     -- logSize testStoreMsgsFile `shouldReturn` 3
     logSize testServerStatsBackupFile `shouldReturn` 76
@@ -724,13 +731,13 @@ testRestoreMessages =
       (dec mId6 msg6, Left "ClientRcvMsgQuota") #== "restored message delivered"
       Resp "7" _ OK <- signSendRecv h rKey ("7", rId, ACK mId6)
       pure ()
-    logSize testStoreLogFile `shouldReturn` 1
+    withHybridStore msType $ logSize testStoreLogFile `shouldReturn` 1
     -- logSize testStoreMsgsFile `shouldReturn` 0
     logSize testServerStatsBackupFile `shouldReturn` 76
     Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats3 [rId] 5 5
 
-    removeFile testStoreLogFile
+    withHybridStore msType $ removeFile testStoreLogFile
     removeFileIfExists testStoreMsgsFile
     whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
     removeFile testServerStatsBackupFile
@@ -785,7 +792,8 @@ testRestoreExpireMessages =
       Resp "4" _ OK <- signSendRecv h sKey ("4", sId, _SEND "hello 4")
       pure ()
 
-    logSize testStoreLogFile `shouldReturn` 2
+    withHybridStore msType $
+      logSize testStoreLogFile `shouldReturn` 2
     exportStoreMessages msType
     msgs <- B.readFile testStoreMsgsFile
     length (B.lines msgs) `shouldBe` 4
@@ -794,7 +802,8 @@ testRestoreExpireMessages =
         cfg1 = (cfgMS msType) {messageExpiration = expCfg1, serverStatsBackupFile = Just testServerStatsBackupFile}
     withSmpServerConfigOn at cfg1 testPort . runTest t $ \_ -> pure ()
 
-    logSize testStoreLogFile `shouldReturn` 1
+    withHybridStore msType $
+      logSize testStoreLogFile `shouldReturn` 1
     exportStoreMessages msType
     msgs' <- B.readFile testStoreMsgsFile
     msgs' `shouldBe` msgs
@@ -803,7 +812,8 @@ testRestoreExpireMessages =
         cfg2 = (cfgMS msType) {messageExpiration = expCfg2, serverStatsBackupFile = Just testServerStatsBackupFile}
     withSmpServerConfigOn at cfg2 testPort . runTest t $ \_ -> pure ()
 
-    logSize testStoreLogFile `shouldReturn` 1
+    withHybridStore msType $
+      logSize testStoreLogFile `shouldReturn` 1
     -- two messages expired
     exportStoreMessages msType
     msgs'' <- B.readFile testStoreMsgsFile
@@ -813,9 +823,13 @@ testRestoreExpireMessages =
     _msgExpired `shouldBe` 2
   where
     exportStoreMessages :: AMSType -> IO ()
-    exportStoreMessages = \case
+    exportStoreMessages msType = case msType of
       AMSType SMSJournal -> do
-        ms <- newMsgStore testJournalStoreCfg {quota = 4}
+        ms <- newMsgStore $ (testJournalStoreCfg SMSJournal) {quota = 4}
+        removeFileIfExists testStoreMsgsFile
+        exportMessages False ms testStoreMsgsFile False
+      AMSType SMSHybrid -> do
+        ms <- newMsgStore $ (testJournalStoreCfg SMSHybrid) {quota = 4}
         readWriteQueueStore testStoreLogFile ms >>= closeStoreLog
         removeFileIfExists testStoreMsgsFile
         exportMessages False ms testStoreMsgsFile False

tests/Test.hs

@@ -74,14 +74,16 @@ main = do
           describe "Store log tests" storeLogTests
           describe "TRcvQueues tests" tRcvQueuesTests
           describe "Util tests" utilTests
-        describe "SMP server via TLS, jornal message store" $ do
+        describe "SMP server via TLS, hybrid store" $ do
           describe "SMP syntax" $ serverSyntaxTests (transport @TLS)
+          before (pure (transport @TLS, AMSType SMSHybrid)) serverTests
+        describe "SMP server via TLS, journal message store" $ do
           before (pure (transport @TLS, AMSType SMSJournal)) serverTests
         describe "SMP server via TLS, memory message store" $
           before (pure (transport @TLS, AMSType SMSMemory)) serverTests
         -- xdescribe "SMP server via WebSockets" $ do
         --   describe "SMP syntax" $ serverSyntaxTests (transport @WS)
-        --   before (pure (transport @WS, AMSType SMSJournal)) serverTests
+        --   before (pure (transport @WS, AMSType SMSHybrid)) serverTests
         describe "Notifications server" $ ntfServerTests (transport @TLS)
         describe "SMP client agent" $ agentTests (transport @TLS)
         describe "SMP proxy" smpProxyTests