Merge remote-tracking branch 'origin/master' into ab/tls-2

smp server: journal message store

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 6.1.3
+
+SMP server: fix restoring notification credentials.
+
 # 6.1.2
 
 Servers: more reliable restoring of state.

cabal.project

@@ -4,7 +4,15 @@ packages: .
 -- packages: . ../http2
 -- packages: . ../network-transport
 
-index-state: 2023-12-12T00:00:00Z
+-- package *
+   -- coverage: True
+   -- library-coverage: True
+
+-- package attoparsec
+   -- coverage: False
+   -- library-coverage: False
+
+index-state: 2024-06-01T00:00:00Z
 
 package cryptostore
     flags: +use_crypton

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 6.1.2.0
+version: 6.1.3.0
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -70,7 +70,7 @@ dependencies:
   - temporary == 1.3.*
   - time == 1.12.*
   - time-manager == 0.0.*
-  - tls >= 1.9.0 && < 1.10
+  - tls >= 2.0.6 && < 2.1
   - transformers == 0.6.*
   - unliftio == 0.2.*
   - unliftio-core == 0.2.*
@@ -188,6 +188,7 @@ tests:
       - -rtsopts
       - -with-rtsopts=-A64M
       - -with-rtsopts=-N1
+      # - -fhpc
 
 ghc-options:
   # - -haddock

rfcs/2024-09-01-smp-message-storage.md

@@ -96,7 +96,7 @@ else if write_msg = max_queue_messages:
     add quota_exceeded message to write_file
     update queue state: write_msg += 1
     append updated queue state to queue.log
-else ]
+else
     // It is required that `max_queue_messages < max_file_messages`,
     // so that we never need more than one additional write file.
     if write_msg >= max_file_messages: // queue file rotation

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        6.1.2.0
+version:        6.1.3.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -175,7 +175,9 @@ library
       Simplex.Messaging.Server.Information
       Simplex.Messaging.Server.Main
       Simplex.Messaging.Server.MsgStore
+      Simplex.Messaging.Server.MsgStore.Journal
       Simplex.Messaging.Server.MsgStore.STM
+      Simplex.Messaging.Server.MsgStore.Types
       Simplex.Messaging.Server.NtfStore
       Simplex.Messaging.Server.QueueStore
       Simplex.Messaging.Server.QueueStore.QueueInfo
@@ -263,7 +265,7 @@ library
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.9.0 && <1.10
+    , tls >=2.0.6 && <2.1
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -339,7 +341,7 @@ executable ntf-server
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.9.0 && <1.10
+    , tls >=2.0.6 && <2.1
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -419,7 +421,7 @@ executable smp-server
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.9.0 && <1.10
+    , tls >=2.0.6 && <2.1
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -499,7 +501,7 @@ executable xftp
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.9.0 && <1.10
+    , tls >=2.0.6 && <2.1
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -575,7 +577,7 @@ executable xftp-server
     , temporary ==1.3.*
     , time ==1.12.*
     , time-manager ==0.0.*
-    , tls >=1.9.0 && <1.10
+    , tls >=2.0.6 && <2.1
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
@@ -614,6 +616,7 @@ test-suite simplexmq-test
       CoreTests.CryptoFileTests
       CoreTests.CryptoTests
       CoreTests.EncodingTests
+      CoreTests.MsgStoreTests
       CoreTests.RetryIntervalTests
       CoreTests.SOCKSSettings
       CoreTests.StoreLogTests
@@ -700,7 +703,7 @@ test-suite simplexmq-test
     , time ==1.12.*
     , time-manager ==0.0.*
     , timeit ==2.0.*
-    , tls >=1.9.0 && <1.10
+    , tls >=2.0.6 && <2.1
     , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*

src/Simplex/FileTransfer/Server/Main.hs

@@ -51,7 +51,9 @@ xftpServerCLI cfgPath logPath = do
         True -> readIniFile iniFile >>= either exitError runServer
         _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
     Delete -> do
-      confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+      confirmOrExit
+        "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+        "Server NOT deleted"
       deleteDirIfExists cfgPath
       deleteDirIfExists logPath
       putStrLn "Deleted configuration and log files"

src/Simplex/Messaging/Agent/Client.hs

@@ -30,6 +30,7 @@ module Simplex.Messaging.Agent.Client
     withConnLocks,
     withInvLock,
     withLockMap,
+    getMapLock,
     ipAddressProtected,
     closeAgentClient,
     closeProtocolServerClients,
@@ -500,7 +501,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a
   getMsgLocks <- TM.emptyIO
   connLocks <- TM.emptyIO
   invLocks <- TM.emptyIO
-  deleteLock <- atomically createLock
+  deleteLock <- createLockIO
   smpSubWorkers <- TM.emptyIO
   smpServersStats <- TM.emptyIO
   xftpServersStats <- TM.emptyIO

src/Simplex/Messaging/Agent/Lock.hs

@@ -1,6 +1,7 @@
 module Simplex.Messaging.Agent.Lock
   ( Lock,
     createLock,
+    createLockIO,
     withLock,
     withLock',
     withGetLock,
@@ -24,6 +25,10 @@ createLock :: STM Lock
 createLock = newEmptyTMVar
 {-# INLINE createLock #-}
 
+createLockIO :: IO Lock
+createLockIO = newEmptyTMVarIO
+{-# INLINE createLockIO #-}
+
 withLock :: MonadUnliftIO m => Lock -> String -> ExceptT e m a -> ExceptT e m a
 withLock lock name = ExceptT . withLock' lock name . runExceptT
 {-# INLINE withLock #-}

src/Simplex/Messaging/Notifications/Server/Main.hs

@@ -53,7 +53,9 @@ ntfServerCLI cfgPath logPath =
         True -> readIniFile iniFile >>= either exitError runServer
         _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
     Delete -> do
-      confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+      confirmOrExit
+        "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+        "Server NOT deleted"
       deleteDirIfExists cfgPath
       deleteDirIfExists logPath
       putStrLn "Deleted configuration and log files"

src/Simplex/Messaging/Protocol.hs

@@ -399,8 +399,6 @@ data Command (p :: Party) where
   NKEY :: NtfPublicAuthKey -> RcvNtfPublicDhKey -> Command Recipient
   NDEL :: Command Recipient
   GET :: Command Recipient
-  -- ACK v1 has to be supported for encoding/decoding
-  -- ACK :: Command Recipient
   ACK :: MsgId -> Command Recipient
   OFF :: Command Recipient
   DEL :: Command Recipient

src/Simplex/Messaging/Server.hs

@@ -30,16 +30,19 @@
 module Simplex.Messaging.Server
   ( runSMPServer,
     runSMPServerBlocking,
+    importMessages,
+    exportMessages,
+    printMessageStats,
     disconnectTransport,
     verifyCmdAuthorization,
     dummyVerifyCmd,
     randomId,
     AttachHTTP,
+    MessageStats (..),
   )
 where
 
 import Control.Concurrent.STM (throwSTM)
-import Control.Concurrent.STM.TQueue (flushTQueue)
 import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
@@ -65,6 +68,7 @@ import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
 import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing)
+import Data.Semigroup (Sum (..))
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
@@ -90,7 +94,9 @@ 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 (JournalMsgQueue (..), JMQueue (..), closeMsgQueueHandles)
 import Simplex.Messaging.Server.MsgStore.STM
+import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.NtfStore
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.QueueInfo
@@ -137,13 +143,32 @@ runSMPServerBlocking started cfg attachHTTP_ = newEnv cfg >>= runReaderT (smpSer
 type M a = ReaderT Env IO a
 type AttachHTTP = Socket -> TLS.Context -> IO ()
 
+data MessageStats = MessageStats
+  { storedMsgsCount :: Int,
+    expiredMsgsCount :: Int,
+    storedQueues :: Int
+  }
+
+instance Monoid MessageStats where
+  mempty = MessageStats 0 0 0
+  {-# INLINE mempty #-}
+
+instance Semigroup MessageStats where
+  MessageStats a b c <> MessageStats x y z = MessageStats (a + x) (b + y) (c + z)
+  {-# INLINE (<>) #-}
+
+newMessageStats :: MessageStats
+newMessageStats = MessageStats 0 0 0
+
 smpServer :: TMVar Bool -> ServerConfig -> Maybe AttachHTTP -> M ()
 smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHTTP_ = do
   s <- asks server
   pa <- asks proxyAgent
-  expiredMsgs <- restoreServerMessages
-  expiredNtfs <- restoreServerNtfs
-  restoreServerStats expiredMsgs expiredNtfs
+  msgStats <- processServerMessages
+  ntfStats <- restoreServerNtfs
+  liftIO $ printMessageStats "messages" msgStats
+  liftIO $ printMessageStats "notifications" ntfStats
+  restoreServerStats msgStats ntfStats
   raceAny_
     ( serverThread s "server subscribedQ" subscribedQ subscribers subClients pendingSubEvents subscriptions cancelSub
         : serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers ntfSubClients pendingNtfSubEvents ntfSubscriptions (\_ -> pure ())
@@ -193,11 +218,15 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
     stopServer s = do
       asks serverActive >>= atomically . (`writeTVar` False)
       logInfo "Saving server state..."
-      withLock' (savingLock s) "final" $ saveServer False >> closeServer
+      withLock' (savingLock s) "final" $ saveServer True >> closeServer
       logInfo "Server stopped"
 
     saveServer :: Bool -> M ()
-    saveServer keepMsgs = withLog closeStoreLog >> saveServerMessages keepMsgs >> saveServerNtfs >> saveServerStats
+    saveServer drainMsgs = do
+      withLog closeStoreLog
+      saveServerMessages drainMsgs
+      saveServerNtfs
+      saveServerStats
 
     closeServer :: M ()
     closeServer = asks (smpAgent . proxyAgent) >>= liftIO . closeSMPClientAgent
@@ -351,19 +380,20 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
 
     expireMessagesThread :: ExpirationConfig -> M ()
     expireMessagesThread expCfg = do
-      ms <- asks msgStore
-      quota <- asks $ msgQueueQuota . config
+      AMS _ ms <- asks msgStore
       let interval = checkInterval expCfg * 1000000
       stats <- asks serverStats
       labelMyThread "expireMessagesThread"
-      forever $ do
-        liftIO $ threadDelay' interval
-        old <- liftIO $ expireBeforeEpoch expCfg
-        rIds <- M.keysSet <$> readTVarIO ms
-        forM_ rIds $ \rId -> do
-          q <- liftIO $ getMsgQueue ms rId quota
-          deleted <- liftIO $ deleteExpiredMsgs q old
-          liftIO $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
+      liftIO $ forever $ do
+        threadDelay' interval
+        old <- expireBeforeEpoch expCfg
+        Sum deleted <- withActiveMsgQueues ms $ \_ -> expireQueueMsgs stats old
+        logInfo $ "STORE: expireMessagesThread, expired " <> tshow deleted <> " messages"
+      where
+        expireQueueMsgs stats old q =
+          runExceptT (deleteExpiredMsgs q True old) >>= \case
+            Right deleted -> Sum deleted <$ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
+            Left _ -> pure 0
 
     expireNtfsThread :: ServerConfig -> M ()
     expireNtfsThread ServerConfig {notificationExpiration = expCfg} = do
@@ -767,7 +797,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
                           ProhibitSub -> pure (c1, c2, c3, c4 + 1)
               CPDelete queueId' -> withUserRole $ unliftIO u $ do
                 st <- asks queueStore
-                ms <- asks msgStore
+                AMS _ ms <- asks msgStore
                 queueId <- liftIO (getQueue st SSender queueId') >>= \case
                   Left _ -> pure queueId' -- fallback to using as recipientId directly
                   Right QueueRec {recipientId} -> pure recipientId
@@ -782,7 +812,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
                     liftIO . hPutStrLn h $ "ok, " <> show numDeleted <> " messages deleted"
               CPSave -> withAdminRole $ withLock' (savingLock srv) "control" $ do
                 hPutStrLn h "saving server state..."
-                unliftIO u $ saveServer True
+                unliftIO u $ saveServer False
                 hPutStrLn h "server state saved!"
               CPHelp -> hPutStrLn h "commands: stats, stats-rts, clients, sockets, socket-threads, threads, server-info, delete, save, help, quit"
               CPQuit -> pure ()
@@ -1166,7 +1196,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
           NDEL -> deleteQueueNotifier_ st
           OFF -> suspendQueue_ st
           DEL -> delQueueAndMsgs st
-          QUE -> withQueue getQueueInfo
+          QUE -> withQueue $ fmap (corrId,entId,) . getQueueInfo
       where
         createQueue :: QueueStore -> RcvPublicAuthKey -> RcvPublicDhKey -> SubscriptionMode -> SenderCanSecure -> M (Transmission BrokerMsg)
         createQueue st recipientKey dhKey subMode sndSecure = time "NEW" $ do
@@ -1262,7 +1292,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
           okResp <$> liftIO (suspendQueue st entId)
 
         subscribeQueue :: QueueRec -> RecipientId -> M (Transmission BrokerMsg)
-        subscribeQueue qr rId = do
+        subscribeQueue qr rId =
           atomically (TM.lookup rId subscriptions) >>= \case
             Nothing -> newSub >>= deliver True
             Just s@Sub {subThread} -> do
@@ -1284,11 +1314,13 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
               pure sub
             deliver :: Bool -> Sub -> M (Transmission BrokerMsg)
             deliver inc sub = do
-              q <- getStoreMsgQueue "SUB" rId
-              msg_ <- liftIO $ tryPeekMsgIO q
-              when (inc && isJust msg_) $
-                incStat . qSub =<< asks serverStats
-              deliverMessage "SUB" qr rId sub msg_
+              AMS _ ms <- asks msgStore
+              stats <- asks serverStats
+              fmap (either (\e -> (corrId, rId, ERR e)) id) $ liftIO $ runExceptT $ do
+                q <- getMsgQueue ms rId
+                msg_ <- tryPeekMsg q
+                liftIO $ when (inc && isJust msg_) $ incStat (qSub stats)
+                liftIO $ deliverMessage "SUB" qr rId sub msg_
 
         -- clients that use GET are not added to server subscribers
         getMessage :: QueueRec -> M (Transmission BrokerMsg)
@@ -1305,7 +1337,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                 _ -> do
                   stats <- asks serverStats
                   incStat $ msgGetProhibited stats
-                  pure (corrId, entId, ERR $ CMD PROHIBITED)
+                  pure $ err $ CMD PROHIBITED
           where
             newSub :: STM Sub
             newSub = do
@@ -1318,19 +1350,16 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
               pure s
             getMessage_ :: Sub -> Maybe MsgId -> M (Transmission BrokerMsg)
             getMessage_ s delivered_ = do
-              q <- getStoreMsgQueue "GET" entId
+              AMS _ ms <- asks msgStore
               stats <- asks serverStats
-              (statCnt, r) <-
-                -- TODO split STM, use tryPeekMsgIO
-                atomically $
-                  tryPeekMsg q >>= \case
-                    Just msg ->
-                      let encMsg = encryptMsg qr msg
-                          cnt = if isJust delivered_ then msgGetDuplicate else msgGet
-                       in setDelivered s msg $> (cnt, (corrId, entId, MSG encMsg))
-                    _ -> pure (msgGetNoMsg, (corrId, entId, OK))
-              incStat $ statCnt stats
-              pure r
+              fmap (either err id) $ liftIO $ runExceptT $ do
+                q <- getMsgQueue ms entId
+                tryPeekMsg q >>= \case
+                  Just msg -> do
+                    let encMsg = encryptMsg qr msg
+                    incStat $ (if isJust delivered_ then msgGetDuplicate else msgGet) stats
+                    atomically $ setDelivered s msg $> (corrId, entId, MSG encMsg)
+                  Nothing -> incStat (msgGetNoMsg stats) $> ok
 
         withQueue :: (QueueRec -> M (Transmission BrokerMsg)) -> M (Transmission BrokerMsg)
         withQueue action = case qr_ of
@@ -1367,16 +1396,19 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
             Just sub ->
               atomically (getDelivered sub) >>= \case
                 Just st -> do
-                  q <- getStoreMsgQueue "ACK" entId
-                  case st of
-                    ProhibitSub -> do
-                      deletedMsg_ <- liftIO $ tryDelMsg q msgId
-                      mapM_ (updateStats True) deletedMsg_
-                      pure ok
-                    _ -> do
-                      (deletedMsg_, msg_) <- liftIO $ tryDelPeekMsg q msgId
-                      mapM_ (updateStats False) deletedMsg_
-                      deliverMessage "ACK" qr entId sub msg_
+                  AMS _ ms <- asks msgStore
+                  stats <- asks serverStats
+                  fmap (either err id) $ liftIO $ runExceptT $ do
+                    q <- getMsgQueue ms entId
+                    case st of
+                      ProhibitSub -> do
+                        deletedMsg_ <- tryDelMsg q msgId
+                        liftIO $ mapM_ (updateStats stats True) deletedMsg_
+                        pure ok
+                      _ -> do
+                        (deletedMsg_, msg_) <- tryDelPeekMsg q msgId
+                        liftIO $ mapM_ (updateStats stats False) deletedMsg_
+                        liftIO $ deliverMessage "ACK" qr entId sub msg_
                 _ -> pure $ err NO_MSG
           where
             getDelivered :: Sub -> STM (Maybe ServerSub)
@@ -1385,11 +1417,10 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                 if msgId == msgId' || B.null msgId
                   then pure $ Just subThread
                   else putTMVar delivered msgId' $> Nothing
-            updateStats :: Bool -> Message -> M ()
-            updateStats isGet = \case
+            updateStats :: ServerStats -> Bool -> Message -> IO ()
+            updateStats stats isGet = \case
               MessageQuota {} -> pure ()
               Message {msgFlags} -> do
-                stats <- asks serverStats
                 incStat $ msgRecv stats
                 if isGet
                   then incStat $ msgRecvGet stats
@@ -1399,11 +1430,11 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                   --   ns <- asks ntfStore
                   --   atomically $ TM.lookup nId ns >>=
                   --     mapM_ (\MsgNtf {ntfMsgId} -> when (msgId == msgId') $ TM.delete nId ns)
-                liftIO $ atomicModifyIORef'_ (msgCount stats) (subtract 1)
-                liftIO $ updatePeriodStats (activeQueues stats) entId
+                atomicModifyIORef'_ (msgCount stats) (subtract 1)
+                updatePeriodStats (activeQueues stats) entId
                 when (notification msgFlags) $ do
                   incStat $ msgRecvNtf stats
-                  liftIO $ updatePeriodStats (activeQueuesNtf stats) entId
+                  updatePeriodStats (activeQueuesNtf stats) entId
 
         sendMessage :: QueueRec -> MsgFlags -> MsgBody -> M (Transmission BrokerMsg)
         sendMessage qr msgFlags msgBody
@@ -1421,15 +1452,20 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                   case C.maxLenBS msgBody of
                     Left _ -> pure $ err LARGE_MSG
                     Right body -> do
-                      msg_ <- time "SEND" $ do
-                        q <- getStoreMsgQueue "SEND" $ recipientId qr
-                        expireMessages q
-                        liftIO . writeMsg q =<< mkMessage body
+                      AMS _ ms <- asks msgStore
+                      ServerConfig {messageExpiration, msgIdBytes} <- asks config
+                      msgId <- randomId' msgIdBytes
+                      msg_ <- liftIO $ time "SEND" $ runExceptT $ do
+                        q <- getMsgQueue ms $ recipientId qr
+                        expireMessages q messageExpiration stats
+                        msg <- liftIO $ mkMessage msgId body
+                        writeMsg ms q True msg
                       case msg_ of
-                        Nothing -> do
+                        Left e -> pure $ err e
+                        Right Nothing -> do
                           incStat $ msgSentQuota stats
                           pure $ err QUOTA
-                        Just (msg, wasEmpty) -> time "SEND ok" $ do
+                        Right (Just (msg, wasEmpty)) -> time "SEND ok" $ do
                           when wasEmpty $ liftIO $ tryDeliverMessage msg
                           when (notification msgFlags) $ do
                             mapM_ (`enqueueNotification` msg) (notifier qr)
@@ -1441,20 +1477,15 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                           pure ok
           where
             THandleParams {thVersion} = thParams'
-            mkMessage :: C.MaxLenBS MaxMessageLen -> M Message
-            mkMessage body = do
-              msgId <- randomId' =<< asks (msgIdBytes . config)
-              msgTs <- liftIO getSystemTime
+            mkMessage :: MsgId -> C.MaxLenBS MaxMessageLen -> IO Message
+            mkMessage msgId body = do
+              msgTs <- getSystemTime
               pure $! Message msgId msgTs msgFlags body
 
-            expireMessages :: MsgQueue -> M ()
-            expireMessages q = do
-              msgExp <- asks $ messageExpiration . config
-              old <- liftIO $ mapM expireBeforeEpoch msgExp
-              deleted <- liftIO $ sum <$> mapM (deleteExpiredMsgs q) old
-              when (deleted > 0) $ do
-                stats <- asks serverStats
-                liftIO $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
+            expireMessages :: MsgStoreClass s => MsgQueue s -> Maybe ExpirationConfig -> ServerStats -> ExceptT ErrorType IO ()
+            expireMessages q msgExp stats = do
+              deleted <- maybe (pure 0) (deleteExpiredMsgs q True <=< liftIO . expireBeforeEpoch) msgExp
+              liftIO $ when (deleted > 0) $ atomicModifyIORef'_ (msgExpired stats) (+ deleted)
 
             -- The condition for delivery of the message is:
             -- - the queue was empty when the message was sent,
@@ -1472,7 +1503,6 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                 atomically deliverToSub >>= mapM_ forkDeliver
               where
                 rId = recipientId qr
-                -- remove tryPeekMsg
                 deliverToSub =
                   -- lookup has ot be in the same transaction,
                   -- so that if subscription ends, it re-evalutates
@@ -1586,7 +1616,7 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
                     verified = \case
                       VRVerified qr -> Right (qr, (corrId', entId', cmd'))
                       VRFailed -> Left (corrId', entId', ERR AUTH)
-        deliverMessage :: T.Text -> QueueRec -> RecipientId -> Sub -> Maybe Message -> M (Transmission BrokerMsg)
+        deliverMessage :: T.Text -> QueueRec -> RecipientId -> Sub -> Maybe Message -> IO (Transmission BrokerMsg)
         deliverMessage name qr rId s@Sub {subThread} msg_ = time (name <> " deliver") . atomically $
           case subThread of
             ProhibitSub -> pure resp
@@ -1614,16 +1644,10 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
         setDelivered :: Sub -> Message -> STM Bool
         setDelivered s msg = tryPutTMVar (delivered s) $! messageId msg
 
-        getStoreMsgQueue :: T.Text -> RecipientId -> M MsgQueue
-        getStoreMsgQueue name rId = time (name <> " getMsgQueue") $ do
-          ms <- asks msgStore
-          quota <- asks $ msgQueueQuota . config
-          liftIO $ getMsgQueue ms rId quota
-
         delQueueAndMsgs :: QueueStore -> M (Transmission BrokerMsg)
         delQueueAndMsgs st = do
           withLog (`logDeleteQueue` entId)
-          ms <- asks msgStore
+          AMS _ ms <- asks msgStore
           liftIO (deleteQueue st entId $>>= \q -> delMsgQueue ms entId $> Right q) >>= \case
             Right q -> do
               -- Possibly, the same should be done if the queue is suspended, but currently we do not use it
@@ -1643,14 +1667,16 @@ client thParams' clnt@Client {clientId, subscriptions, ntfSubscriptions, rcvQ, s
               pure ok
             Left e -> pure $ err e
 
-        getQueueInfo :: QueueRec -> M (Transmission BrokerMsg)
+        getQueueInfo :: QueueRec -> M BrokerMsg
         getQueueInfo QueueRec {senderKey, notifier} = do
-          q <- getStoreMsgQueue "getQueueInfo" entId
-          qiSub <- liftIO $ TM.lookupIO entId subscriptions >>= mapM mkQSub
-          qiSize <- liftIO $ getQueueSize q
-          qiMsg <- liftIO $ toMsgInfo <$$> tryPeekMsgIO q
-          let info = QueueInfo {qiSnd = isJust senderKey, qiNtf = isJust notifier, qiSub, qiSize, qiMsg}
-          pure (corrId, entId, INFO info)
+          AMS _ ms <- asks msgStore
+          fmap (either ERR id) $ liftIO $ runExceptT $ do
+            q <- getMsgQueue ms entId
+            qiSub <- liftIO $ TM.lookupIO entId subscriptions >>= mapM mkQSub
+            qiSize <- liftIO $ getQueueSize q
+            qiMsg <- toMsgInfo <$$> tryPeekMsg q
+            let info = QueueInfo {qiSnd = isJust senderKey, qiNtf = isJust notifier, qiSub, qiSize, qiMsg}
+            pure $ INFO info
           where
             mkQSub Sub {subThread, delivered} = do
               qSubThread <- case subThread of
@@ -1710,59 +1736,117 @@ randomId = fmap EntityId . randomId'
 {-# INLINE randomId #-}
 
 saveServerMessages :: Bool -> M ()
-saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessages
-  where
-    saveMessages f = do
-      logInfo $ "saving messages to file " <> T.pack f
-      ms <- asks msgStore
-      liftIO . withFile f WriteMode $ \h ->
-        readTVarIO ms >>= mapM_ (saveQueueMsgs h) . M.assocs
-      logInfo "messages saved"
-      where
-        saveQueueMsgs h (rId, q) = BLD.hPutBuilder h . encodeMessages rId =<< atomically (getMessages $ msgQueue q)
-        getMessages = if keepMsgs then snapshotTQueue else flushTQueue
-        snapshotTQueue q = do
-          msgs <- flushTQueue q
-          mapM_ (writeTQueue q) msgs
-          pure msgs
-        encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')
+saveServerMessages drainMsgs =
+  asks msgStore >>= \case
+    AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
+      Just f -> liftIO $ exportMessages False ms f drainMsgs
+      Nothing -> logInfo "undelivered messages are not saved"
+    AMS SMSJournal ms -> do
+      liftIO $ closeMsgStore ms
+      logInfo "closed journal message storage"
 
-restoreServerMessages :: M Int
-restoreServerMessages =
-  asks (storeMsgsFile . config) >>= \case
-    Just f -> ifM (doesFileExist f) (restoreMessages f) (pure 0)
-    Nothing -> pure 0
+exportMessages :: MsgStoreClass s => Bool -> s -> FilePath -> Bool -> IO ()
+exportMessages tty ms f drainMsgs = do
+  logInfo $ "saving messages to file " <> T.pack f
+  liftIO $ withFile f WriteMode $ \h ->
+    tryAny (withAllMsgQueues tty ms $ saveQueueMsgs h) >>= \case
+      Right (Sum total) -> logInfo $ "messages saved: " <> tshow total
+      Left e -> do
+        logError $ "error exporting messages: " <> tshow e
+        exitFailure
   where
-    restoreMessages f = do
-      logInfo $ "restoring messages from file " <> T.pack f
-      ms <- asks msgStore
-      quota <- asks $ msgQueueQuota . config
-      old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch)
-      runExceptT (liftIO (LB.readFile f) >>= foldM (\expired -> restoreMsg expired ms quota old_) 0 . LB.lines) >>= \case
-        Left e -> do
-          logError . T.pack $ "error restoring messages: " <> e
-          liftIO exitFailure
-        Right expired -> do
-          renameFile f $ f <> ".bak"
-          logInfo "messages restored"
-          pure expired
+    saveQueueMsgs h rId q = getQueueMessages drainMsgs q >>= \msgs -> Sum (length msgs) <$ BLD.hPutBuilder h (encodeMessages rId msgs)
+    encodeMessages rId = mconcat . map (\msg -> BLD.byteString (strEncode $ MLRv3 rId msg) <> BLD.char8 '\n')
+
+processServerMessages :: M MessageStats
+processServerMessages = do
+  old_ <- asks (messageExpiration . config) $>>= (liftIO . fmap Just . expireBeforeEpoch)
+  asks msgStore >>= liftIO . processMessages old_
+    where
+      processMessages :: Maybe Int64 -> AMsgStore -> IO MessageStats
+      processMessages old_ = \case
+        AMS SMSMemory ms@STMMsgStore {storeConfig = STMStoreConfig {storePath}} -> case storePath of
+          Just f -> ifM (doesFileExist f) (importMessages False ms f old_) (pure newMessageStats)
+          Nothing -> pure newMessageStats
+        AMS SMSJournal ms -> 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
+        where
+          processExpireQueue old q =
+            runExceptT expireQueue >>= \case
+              Right (storedMsgsCount, expiredMsgsCount) ->
+                pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues = 1}
+              Left e -> do
+                logError $ "failed expiring messages in queue " <> T.pack (queueDirectory $ queue q) <> ": " <> tshow e
+                exitFailure
+            where
+              expireQueue = do
+                expired'' <- deleteExpiredMsgs q False old
+                stored'' <- liftIO $ getQueueSize q
+                liftIO $ logQueueState q
+                liftIO $ closeMsgQueueHandles q
+                pure (stored'', expired'')
+          processValidateQueue q =
+            getQueueSize q >>= \storedMsgsCount -> pure mempty {storedMsgsCount, storedQueues = 1}
+
+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)) . zip [0..] . LB.lines >>= \case
+    Left e -> do
+      putStrLn ""
+      logError . T.pack $ "error restoring messages: " <> e
+      liftIO exitFailure
+    Right (lineCount, _, (storedMsgsCount, expiredMsgsCount, overQuota)) -> do
+      putStrLn $ progress lineCount
+      renameFile f $ f <> ".bak"
+      mapM_ setOverQuota_ overQuota
+      logQueueStates ms
+      storedQueues <- M.size <$> readTVarIO (activeMsgQueues ms)
+      pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues}
+  where
+    progress i = "Processed " <> show i <> " lines"
+    restoreMsg :: (Int, Maybe (RecipientId, MsgQueue s), (Int, Int, M.Map RecipientId (MsgQueue s))) -> (Int, LB.ByteString) -> ExceptT String IO (Int, Maybe (RecipientId, MsgQueue s), (Int, Int, M.Map RecipientId (MsgQueue s)))
+    restoreMsg (!lineCount, q_, (!stored, !expired, !overQuota)) (i, s') = do
+      when (tty && i `mod` 1000 == 0) $ liftIO $ putStr (progress i <> "\r")
+      MLRv3 rId msg <- liftEither . first (msgErr "parsing") $ strDecode s
+      liftError show $ addToMsgQueue rId msg
       where
-        restoreMsg !expired ms quota old_ s' = do
-          MLRv3 rId msg <- liftEither . first (msgErr "parsing") $ strDecode s
-          addToMsgQueue rId msg
-          where
-            s = LB.toStrict s'
-            addToMsgQueue rId msg = do
-              q <- liftIO $ getMsgQueue ms rId quota
-              (isExpired, logFull) <- liftIO $ case msg of
-                Message {msgTs}
-                  | maybe True (systemSeconds msgTs >=) old_ -> (False,) . isNothing <$> writeMsg q msg
-                  | otherwise -> pure (True, False)
-                MessageQuota {} -> writeMsg q msg $> (False, False)
-              when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (messageId msg)
-              pure $ if isExpired then expired + 1 else expired
-            msgErr :: Show e => String -> e -> String
-            msgErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
+        s = LB.toStrict s'
+        addToMsgQueue rId msg = do
+          q <- case q_ of
+            -- to avoid lookup when restoring the next message to the same queue
+            Just (rId', q') | rId' == rId -> pure q'
+            _ -> getMsgQueue ms rId
+          (lineCount + 1,Just (rId, q),) <$> case msg of
+            Message {msgTs}
+              | maybe True (systemSeconds msgTs >=) old_ -> do
+                  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)
+                      pure (stored, expired, overQuota)
+              | otherwise -> pure (stored, expired + 1, overQuota)
+            MessageQuota {} ->
+              -- queue was over quota at some point,
+              -- it will be set as over quota once fully imported
+              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 = isolateQueue q "mergeQuotaMsgs" $
+                  tryPeekMsg_ q >>= \case
+                    Just MessageQuota {} -> tryDeleteMsg_ q False
+                    _ -> pure ()
+        msgErr :: Show e => String -> e -> String
+        msgErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
+
+printMessageStats :: T.Text -> MessageStats -> IO ()
+printMessageStats name MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues} =
+  logInfo $ name <> " stored: " <> tshow storedMsgsCount <> ", expired: " <> tshow expiredMsgsCount <> ", queues: " <> tshow storedQueues
 
 saveServerNtfs :: M ()
 saveServerNtfs = asks (storeNtfsFile . config) >>= mapM_ saveNtfs
@@ -1778,33 +1862,37 @@ saveServerNtfs = asks (storeNtfsFile . config) >>= mapM_ saveNtfs
         saveQueueNtfs h (nId, v) = BLD.hPutBuilder h . encodeNtfs nId . reverse =<< readTVarIO v
         encodeNtfs nId = mconcat . map (\ntf -> BLD.byteString (strEncode $ NLRv1 nId ntf) <> BLD.char8 '\n')
 
-restoreServerNtfs :: M Int
+restoreServerNtfs :: M MessageStats
 restoreServerNtfs =
   asks (storeNtfsFile . config) >>= \case
-    Just f -> ifM (doesFileExist f) (restoreNtfs f) (pure 0)
-    Nothing -> pure 0
+    Just f -> ifM (doesFileExist f) (restoreNtfs f) (pure newMessageStats)
+    Nothing -> pure newMessageStats
   where
     restoreNtfs f = do
       logInfo $ "restoring notifications from file " <> T.pack f
       ns <- asks ntfStore
       old <- asks (notificationExpiration . config) >>= liftIO . expireBeforeEpoch
-      runExceptT (liftIO (LB.readFile f) >>= foldM (restoreNtf ns old) 0 . LB.lines) >>= \case
-        Left e -> do
-          logError . T.pack $ "error restoring notifications: " <> e
-          liftIO exitFailure
-        Right expired -> do
-          renameFile f $ f <> ".bak"
-          logInfo "notifications restored"
-          pure expired
+      liftIO $
+        LB.readFile f >>= runExceptT . foldM (restoreNtf ns old) (0, 0, 0) . LB.lines >>= \case
+          Left e -> do
+            logError . T.pack $ "error restoring notifications: " <> e
+            liftIO exitFailure
+          Right (lineCount, storedMsgsCount, expiredMsgsCount) -> do
+            renameFile f $ f <> ".bak"
+            let NtfStore ns' = ns
+            storedQueues <- M.size <$> readTVarIO ns'
+            logInfo $ "notifications restored, " <> tshow lineCount <> " lines processed" 
+            pure MessageStats {storedMsgsCount, expiredMsgsCount, storedQueues}
       where
-        restoreNtf ns old !expired s' = do
+        restoreNtf :: NtfStore -> Int64 -> (Int, Int, Int) -> LB.ByteString -> ExceptT String IO (Int, Int, Int)
+        restoreNtf ns old (!lineCount, !stored, !expired) s' = do
           NLRv1 nId ntf <- liftEither . first (ntfErr "parsing") $ strDecode s
           liftIO $ addToNtfs nId ntf
           where
             s = LB.toStrict s'
             addToNtfs nId ntf@MsgNtf {ntfTs}
-              | systemSeconds ntfTs < old = pure (expired + 1)
-              | otherwise = storeNtf ns nId ntf $> expired
+              | systemSeconds ntfTs < old = pure (lineCount + 1, stored, expired + 1)
+              | otherwise = storeNtf ns nId ntf $> (lineCount + 1, stored + 1, expired)
             ntfErr :: Show e => String -> e -> String
             ntfErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
 
@@ -1818,23 +1906,25 @@ saveServerStats =
       B.writeFile f $ strEncode stats
       logInfo "server stats saved"
 
-restoreServerStats :: Int -> Int -> M ()
-restoreServerStats expiredMsgs expiredNtfs = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats
+restoreServerStats :: MessageStats -> MessageStats -> M ()
+restoreServerStats msgStats ntfStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStats
   where
     restoreStats f = whenM (doesFileExist f) $ do
       logInfo $ "restoring server stats from file " <> T.pack f
       liftIO (strDecode <$> B.readFile f) >>= \case
-        Right d@ServerStatsData {_qCount = statsQCount} -> do
+        Right d@ServerStatsData {_qCount = statsQCount, _msgCount = statsMsgCount, _ntfCount = statsNtfCount} -> do
           s <- asks serverStats
           _qCount <- fmap M.size . readTVarIO . queues =<< asks queueStore
-          _msgCount <- liftIO . foldM (\(!n) q -> (n +) <$> getQueueSize q) 0 =<< readTVarIO =<< asks msgStore
-          NtfStore ns <- asks ntfStore
-          _ntfCount <- liftIO . foldM (\(!n) q -> (n +) . length <$> readTVarIO q) 0 =<< readTVarIO ns
-          liftIO $ setServerStats s d {_qCount, _msgCount, _ntfCount, _msgExpired = _msgExpired d + expiredMsgs, _msgNtfExpired = _msgNtfExpired d + expiredNtfs}
+          let _msgCount = storedMsgsCount msgStats
+              _ntfCount = storedMsgsCount ntfStats
+          liftIO $ setServerStats s d {_qCount, _msgCount, _ntfCount, _msgExpired = _msgExpired d + expiredMsgsCount msgStats, _msgNtfExpired = _msgNtfExpired d + expiredMsgsCount ntfStats}
           renameFile f $ f <> ".bak"
           logInfo "server stats restored"
-          when (_qCount /= statsQCount) $ logWarn $ "Queue count differs: stats: " <> tshow statsQCount <> ", store: " <> tshow _qCount
-          logInfo $ "Restored " <> tshow _msgCount <> " messages in " <> tshow _qCount <> " queues"
+          compareCounts "Queue" statsQCount _qCount
+          compareCounts "Message" statsMsgCount _msgCount
+          compareCounts "Notification" statsNtfCount _ntfCount
         Left e -> do
           logInfo $ "error restoring server stats: " <> T.pack e
           liftIO exitFailure
+    compareCounts name statsCnt storeCnt =
+      when (statsCnt /= storeCnt) $ logWarn $ name <> " count differs: stats: " <> tshow statsCnt <> ", store: " <> tshow storeCnt

src/Simplex/Messaging/Server/CLI.hs

@@ -44,11 +44,11 @@ import Text.Read (readMaybe)
 exitError :: String -> IO a
 exitError msg = putStrLn msg >> exitFailure
 
-confirmOrExit :: String -> IO ()
-confirmOrExit s =
+confirmOrExit :: String -> String -> IO ()
+confirmOrExit s no =
   withPrompt (s <> "\nContinue (Y/n): ") $ do
     ok <- getLine
-    when (ok /= "Y") $ putStrLn "Server NOT deleted" >> exitFailure
+    when (ok /= "Y") $ putStrLn no >> exitFailure
 
 data SignAlgorithm = ED448 | ED25519
   deriving (Read, Show)

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

@@ -1,10 +1,14 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StrictData #-}
+{-# LANGUAGE TypeFamilies #-}
 
 module Simplex.Messaging.Server.Env.STM where
 
@@ -35,7 +39,9 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
+import Simplex.Messaging.Server.MsgStore.Journal
 import Simplex.Messaging.Server.MsgStore.STM
+import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.NtfStore
 import Simplex.Messaging.Server.QueueStore (QueueRec (..))
 import Simplex.Messaging.Server.QueueStore.STM
@@ -55,7 +61,10 @@ data ServerConfig = ServerConfig
   { transports :: [(ServiceName, ATransport, AddHTTP)],
     smpHandshakeTimeout :: Int,
     tbqSize :: Natural,
+    msgStoreType :: AMSType,
     msgQueueQuota :: Int,
+    maxJournalMsgCount :: Int,
+    maxJournalStateLines :: Int,
     queueIdBytes :: Int,
     msgIdBytes :: Int,
     storeLogFile :: Maybe FilePath,
@@ -134,6 +143,18 @@ defaultInactiveClientExpiration =
 defaultProxyClientConcurrency :: Int
 defaultProxyClientConcurrency = 32
 
+journalMsgStoreDepth :: Int
+journalMsgStoreDepth = 5
+
+defaultMaxJournalStateLines :: Int
+defaultMaxJournalStateLines = 16
+
+defaultMaxJournalMsgCount :: Int
+defaultMaxJournalMsgCount = 256
+
+defaultMsgQueueQuota :: Int
+defaultMsgQueueQuota = 128
+
 data Env = Env
   { config :: ServerConfig,
     serverActive :: TVar Bool,
@@ -141,7 +162,7 @@ data Env = Env
     server :: Server,
     serverIdentity :: KeyHash,
     queueStore :: QueueStore,
-    msgStore :: STMMsgStore,
+    msgStore :: AMsgStore,
     ntfStore :: NtfStore,
     random :: TVar ChaChaDRG,
     storeLog :: Maybe (StoreLog 'WriteMode),
@@ -154,6 +175,20 @@ data Env = Env
     proxyAgent :: ProxyAgent -- senders served on this proxy
   }
 
+type family MsgStore s where
+  MsgStore 'MSMemory = STMMsgStore
+  MsgStore 'MSJournal = JournalMsgStore
+
+data AMsgStore = forall s. MsgStoreClass (MsgStore s) => AMS (SMSType s) (MsgStore s)
+
+data AMsgQueue = forall s. MsgStoreClass (MsgStore s) => AMQ (SMSType s) (MsgQueue (MsgStore s))
+
+data AMsgStoreCfg = forall s. MsgStoreClass (MsgStore s) => AMSC (SMSType s) (MsgStoreConfig (MsgStore s))
+
+msgPersistence :: AMsgStoreCfg -> Bool
+msgPersistence (AMSC SMSMemory (STMStoreConfig {storePath})) = isJust storePath
+msgPersistence (AMSC SMSJournal _) = True
+
 type Subscribed = Bool
 
 data Server = Server
@@ -211,7 +246,7 @@ newServer = do
   ntfSubClients <- newTVarIO IM.empty
   pendingSubEvents <- newTVarIO IM.empty
   pendingNtfSubEvents <- newTVarIO IM.empty
-  savingLock <- atomically createLock
+  savingLock <- createLockIO
   return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, subClients, ntfSubClients, pendingSubEvents, pendingNtfSubEvents, savingLock}
 
 newClient :: ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> IO Client
@@ -241,11 +276,17 @@ newProhibitedSub = do
   return Sub {subThread = ProhibitSub, delivered}
 
 newEnv :: ServerConfig -> IO Env
-newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAgentCfg, information, messageExpiration} = do
+newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgStoreType, storeMsgsFile, smpAgentCfg, information, messageExpiration, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines} = do
   serverActive <- newTVarIO True
   server <- newServer
   queueStore <- newQueueStore
-  msgStore <- newMsgStore
+  msgStore <- 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}
+         in AMS SMSJournal <$> newMsgStore cfg
+      Nothing -> putStrLn "Error: journal msg store require path in [STORE_LOG], restore_messages" >> exitFailure
   ntfStore <- NtfStore <$> TM.emptyIO
   random <- C.newRandom
   storeLog <-
@@ -299,7 +340,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAg
       where
         persistence
           | isNothing storeLogFile = SPMMemoryOnly
-          | isJust (storeMsgsFile config) = SPMMessages
+          | isJust storeMsgsFile = SPMMessages
           | otherwise = SPMQueues
 
 newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO ProxyAgent

src/Simplex/Messaging/Server/Main.hs

@@ -1,6 +1,6 @@
 {-# LANGUAGE ApplicativeDo #-}
-{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedLists #-}
@@ -20,6 +20,7 @@ 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.Functor (($>))
 import Data.Ini (Ini, lookupValue, readIniFile)
 import Data.List (find, isPrefixOf)
@@ -27,7 +28,7 @@ import qualified Data.List.NonEmpty as L
 import Data.Maybe (fromMaybe, isJust, isNothing)
 import Data.Text (Text)
 import qualified Data.Text as T
-import Data.Text.Encoding (encodeUtf8, decodeLatin1)
+import Data.Text.Encoding (decodeLatin1, encodeUtf8)
 import qualified Data.Text.IO as T
 import Network.Socket (HostName)
 import Options.Applicative
@@ -38,17 +39,19 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol (BasicAuth (..), ProtoServerWithAuth (ProtoServerWithAuth), pattern SMPServer)
-import Simplex.Messaging.Server (AttachHTTP, runSMPServer)
+import Simplex.Messaging.Server (AttachHTTP, exportMessages, importMessages, printMessageStats, runSMPServer)
 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.Transport (batchCmdsSMPVersion, sendingProxySMPVersion, simplexMQVersion, supportedServerSMPRelayVRange)
 import Simplex.Messaging.Transport.Client (SocksProxy, TransportHost (..), defaultSocksProxy)
 import Simplex.Messaging.Transport.Server (ServerCredentials (..), TransportServerConfig (..), defaultTransportServerConfig)
 import Simplex.Messaging.Util (eitherToMaybe, safeDecodeUtf8, tshow)
 import Simplex.Messaging.Version (mkVersionRange)
-import System.Directory (createDirectoryIfMissing, doesFileExist)
+import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist)
 import System.Exit (exitFailure)
 import System.FilePath (combine)
 import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
@@ -70,25 +73,85 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
       doesFileExist iniFile >>= \case
         True -> exitError $ "Error: server is already initialized (" <> iniFile <> " exists).\nRun `" <> executableName <> " start`."
         _ -> initializeServer opts
-    OnlineCert certOpts ->
-      doesFileExist iniFile >>= \case
-        True -> genOnline cfgPath certOpts
-        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
-    Start ->
-      doesFileExist iniFile >>= \case
-        True -> readIniFile iniFile >>= either exitError runServer
-        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
+    OnlineCert certOpts -> withIniFile $ \_ -> genOnline cfgPath certOpts
+    Start -> withIniFile runServer
     Delete -> do
-      confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+      confirmOrExit
+        "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+        "Server NOT deleted"
       deleteDirIfExists cfgPath
       deleteDirIfExists logPath
       putStrLn "Deleted configuration and log files"
+    Journal cmd -> withIniFile $ \ini -> do
+      msgsDirExists <- doesDirectoryExist storeMsgsJournalDir
+      msgsFileExists <- doesFileExist storeMsgsFilePath
+      case cmd of
+        JCImport
+          | 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
+              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
+          | 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
+              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"
+        JCDelete
+          | not msgsDirExists -> do
+              putStrLn $ storeMsgsJournalDir <> " directory does not exists."
+              exitFailure
+          | otherwise -> do
+              confirmOrExit
+                ("WARNING: journal directory " <> storeMsgsJournalDir <> " will be permanently deleted.\nTHIS CANNOT BE UNDONE!")
+                "Messages NOT deleted"
+              deleteDirIfExists storeMsgsJournalDir
+              putStrLn $ "Deleted all messages in journal " <> storeMsgsJournalDir
   where
+    withIniFile a =
+      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}
     iniFile = combine cfgPath "smp-server.ini"
     serverVersion = "SMP server v" <> simplexMQVersion
     defaultServerPorts = "5223,443"
     executableName = "smp-server"
     storeLogFilePath = combine logPath "smp-server-store.log"
+    storeMsgsFilePath = combine logPath "smp-server-messages.log"
+    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
     httpsCertFile = combine cfgPath "web.crt"
     httpsKeyFile = combine cfgPath "web.key"
     defaultStaticPath = combine logPath "www"
@@ -128,7 +191,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 _ -> putStrLn "Invalid password. Only latin letters, digits and symbols other than '@' and ':' are allowed" >> serverPassword
         checkInitOptions InitOptions {sourceCode, serverInfo, operatorCountry, hostingCountry} = do
           let err_
-                | isNothing sourceCode && hasServerInfo serverInfo = 
+                | isNothing sourceCode && hasServerInfo serverInfo =
                     Just "Error: passing any server information requires passing --source-code"
                 | isNothing (operator serverInfo) && isJust operatorCountry =
                     Just "Error: passing --operator-country requires passing --operator"
@@ -168,9 +231,12 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                    \# and restoring it when the server is started.\n\
                    \# Log is compacted on start (deleted objects are removed).\n"
                 <> ("enable: " <> onOff enableStoreLog <> "\n\n")
-                <> "# Undelivered messages are optionally saved and restored when the server restarts,\n\
-                   \# they are preserved in the .bak file until the next restart.\n"
-                <> ("restore_messages: " <> onOff enableStoreLog <> "\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\
+                   \# when the server restarts, they are preserved in the .bak file until the next restart.\n"
+                <> ("restore_messages: " <> onOff enableStoreLog <> "\n\n")
+                <> "# Messages and notifications expiration periods.\n"
                 <> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n")
                 <> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
                 <> "# Log daily server statistics to CSV file\n"
@@ -248,12 +314,13 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
       fp <- checkSavedFingerprint cfgPath defaultX509Config
       let host = either (const "<hostnames>") T.unpack $ lookupValue "TRANSPORT" "host" ini
           port = T.unpack $ strictIni "TRANSPORT" "port" ini
-          cfg@ServerConfig {information, storeLogFile, newQueueBasicAuth, messageExpiration, inactiveClientExpiration} = serverConfig
+          cfg@ServerConfig {information, storeLogFile, msgStoreType, newQueueBasicAuth, messageExpiration, inactiveClientExpiration} = serverConfig
           sourceCode' = (\ServerPublicInfo {sourceCode} -> sourceCode) <$> information
           srv = ProtoServerWithAuth (SMPServer [THDomainName host] (if port == "5223" then "" else port) (C.KeyHash fp)) newQueueBasicAuth
       printServiceInfo serverVersion srv
       printSourceCode sourceCode'
       printServerConfig transports storeLogFile
+      checkMsgStoreMode msgStoreType
       putStrLn $ case messageExpiration of
         Just ExpirationConfig {ttl} -> "expiring messages after " <> showTTL ttl
         _ -> "not expiring messages"
@@ -302,12 +369,16 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
           _ -> enableStoreLog $> path
         transports = iniTransports ini
         sharedHTTP = any (\(_, _, addHTTP) -> addHTTP) transports
+        iniMsgStoreType = either error id $! readMsgStoreType ini
         serverConfig =
           ServerConfig
             { transports,
               smpHandshakeTimeout = 120000000,
               tbqSize = 128,
-              msgQueueQuota = 128,
+              msgStoreType = iniMsgStoreType,
+              msgQueueQuota = defaultMsgQueueQuota,
+              maxJournalMsgCount = defaultMaxJournalMsgCount,
+              maxJournalStateLines = defaultMaxJournalStateLines,
               queueIdBytes = 24,
               msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
               smpCredentials =
@@ -318,8 +389,10 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                   },
               httpCredentials = (\WebHttpsParams {key, cert} -> ServerCredentials {caCertificateFile = Nothing, privateKeyFile = key, certificateFile = cert}) <$> webHttpsParams',
               storeLogFile = enableStoreLog $> storeLogFilePath,
-              storeMsgsFile = restoreMessagesFile $ combine logPath "smp-server-messages.log",
-              storeNtfsFile = restoreMessagesFile $ combine logPath "smp-server-ntfs.log",
+              storeMsgsFile = case iniMsgStoreType of
+                AMSType SMSMemory -> restoreMessagesFile storeMsgsFilePath
+                AMSType SMSJournal -> Just storeMsgsJournalDir,
+              storeNtfsFile = restoreMessagesFile storeNtfsFilePath,
               -- allow creating new queues by default
               allowNewQueues = fromMaybe True $ iniOnOff "AUTH" "new_queues" ini,
               newQueueBasicAuth = either error id <$!> strDecodeIni "AUTH" "create_password" ini,
@@ -393,6 +466,31 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
             pure WebHttpsParams {port, cert, key}
         webStaticPath' = eitherToMaybe $ T.unpack <$> lookupValue "WEB" "static_path" ini
 
+    checkMsgStoreMode :: AMSType -> IO ()
+    checkMsgStoreMode mode = do
+      msgsDirExists <- doesDirectoryExist storeMsgsJournalDir
+      msgsFileExists <- doesFileExist storeMsgsFilePath
+      case mode of
+        _ | msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
+        AMSType SMSJournal
+          | 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."
+              exitFailure
+          | not msgsDirExists ->
+              putStrLn $ "store_messages is `journal`, " <> storeMsgsJournalDir <> " directory will be created."
+        AMSType SMSMemory
+          | msgsDirExists -> do
+              putStrLn $ "Error: store_messages is `memory` with " <> storeMsgsJournalDir <> " directory present."
+              putStrLn "Set store_messages to `journal` or use `smp-server journal import` to migrate."
+              exitFailure
+        _ -> pure ()
+
+    exitConfigureMsgStorage = do
+      putStrLn $ "Error: both " <> storeMsgsFilePath <> " file and " <> storeMsgsJournalDir <> " directory are present."
+      putStrLn "Configure memory storage."
+      exitFailure
+
 data EmbeddedWebParams = EmbeddedWebParams
   { webStaticPath :: FilePath,
     webHttpPort :: Maybe Int,
@@ -458,12 +556,13 @@ informationIniContent InitOptions {sourceCode, serverInfo} =
        \# Hosting type can be `virtual`, `dedicated`, `colocation`, `owned`\n"
     <> ("hosting_type: " <> maybe "virtual" (decodeLatin1 . strEncode) hostingType <> "\n\n")
   where
-    ServerPublicInfo {operator, website, hosting, hostingType, serverCountry}= serverInfo
+    ServerPublicInfo {operator, website, hosting, hostingType, serverCountry} = serverInfo
     countryStr optName country = optDisabled country <> optName <> "_country: " <> fromMaybe "ISO-3166 2-letter code" country <> "\n"
     enitiyStrs optName entity =
       optDisabled entity
-        <> optName <> ": "
-        <> maybe ("entity (organization or person name)") name entity
+        <> optName
+        <> ": "
+        <> maybe "entity (organization or person name)" name entity
         <> "\n"
         <> countryStr optName (country =<< entity)
 
@@ -519,6 +618,9 @@ data CliCommand
   | OnlineCert CertOptions
   | Start
   | Delete
+  | Journal JournalCmd
+
+data JournalCmd = JCImport | JCExport | JCDelete
 
 data InitOptions = InitOptions
   { enableStoreLog :: Bool,
@@ -550,6 +652,7 @@ cliCommandP cfgPath logPath iniFile =
         <> command "cert" (info (OnlineCert <$> certOptionsP) (progDesc $ "Generate new online TLS server credentials (configuration: " <> iniFile <> ")"))
         <> command "start" (info (pure Start) (progDesc $ "Start server (configuration: " <> iniFile <> ")"))
         <> command "delete" (info (pure Delete) (progDesc "Delete configuration and log files"))
+        <> command "journal" (info (Journal <$> journalCmdP) (progDesc "Import/export messages to/from journal storage"))
     )
   where
     initP :: Parser InitOptions
@@ -607,22 +710,21 @@ cliCommandP cfgPath logPath iniFile =
           <|> option strParse (long "control-port" <> help "Enable control port" <> metavar "PORT" <> value Nothing)
       socksProxy <-
         flag' (Just defaultSocksProxy) (long "socks-proxy" <> help "Outgoing SOCKS proxy on port 9050")
-          <|>
-            option
-              strParse
-              ( long "socks-proxy"
-                  <> metavar "PROXY"
-                  <> help "Outgoing SOCKS proxy to forward messages to onion-only servers"
-                  <> value Nothing
-              )
+          <|> option
+            strParse
+            ( long "socks-proxy"
+                <> metavar "PROXY"
+                <> help "Outgoing SOCKS proxy to forward messages to onion-only servers"
+                <> value Nothing
+            )
       ownDomains :: Maybe (L.NonEmpty TransportHost) <-
         option
           strParse
-            ( long "own-domains"
-                <> metavar "DOMAINS"
-                <> help "Own server domain names (comma-separated)"
-                <> value Nothing
-            )
+          ( long "own-domains"
+              <> metavar "DOMAINS"
+              <> help "Own server domain names (comma-separated)"
+              <> value Nothing
+          )
       sourceCode <-
         flag' (Just simplexmqSource) (long "source-code" <> help ("Server source code (default: " <> simplexmqSource <> ")"))
           <|> (optional . strOption) (long "source-code" <> metavar "URI" <> help "Server source code")
@@ -690,6 +792,13 @@ cliCommandP cfgPath logPath iniFile =
             disableWeb,
             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"))
+        )
+
     parseBasicAuth :: ReadM ServerPassword
     parseBasicAuth = eitherReader $ fmap ServerPassword . strDecode . B.pack
     entityP :: String -> String -> String -> Parser (Maybe Entity, Maybe Text)

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

@@ -0,0 +1,638 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE InstanceSigs #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiWayIf #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StandaloneDeriving #-}
+{-# LANGUAGE TypeFamilies #-}
+
+module Simplex.Messaging.Server.MsgStore.Journal
+  ( JournalMsgStore (msgQueues, random),
+    JournalMsgQueue (queue),
+    JMQueue (queueDirectory, statePath),
+    JournalStoreConfig (..),
+    getQueueMessages,
+    closeMsgQueue,
+    closeMsgQueueHandles,
+    -- below are exported for tests
+    MsgQueueState (..),
+    JournalState (..),
+    SJournalType (..),
+    msgQueueDirectory,
+    msgQueueStatePath,
+    readWriteQueueState,
+    newMsgQueueState,
+    newJournalId,
+    appendState,
+    queueLogFileName,
+    logFileExt,
+  )
+where
+
+import Control.Concurrent.STM
+import qualified Control.Exception as E
+import Control.Logger.Simple
+import Control.Monad
+import Control.Monad.Trans.Except
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.Bitraversable (bimapM)
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Lazy.Char8 as LB
+import Data.Functor (($>))
+import Data.Int (Int64)
+import Data.List (intercalate)
+import qualified Data.Map.Strict as M
+import Data.Maybe (catMaybes, fromMaybe)
+import qualified Data.Text as T
+import Data.Time.Clock (getCurrentTime)
+import Data.Time.Format.ISO8601 (iso8601Show)
+import GHC.IO (catchAny)
+import Simplex.Messaging.Agent.Client (getMapLock, withLockMap)
+import Simplex.Messaging.Agent.Lock
+import Simplex.Messaging.Encoding.String
+import Simplex.Messaging.Protocol (ErrorType (..), Message (..), RecipientId)
+import Simplex.Messaging.Server.MsgStore.Types
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Util (ifM, tshow, ($>>=))
+import System.Directory
+import System.Exit
+import System.FilePath ((</>))
+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,
+    random :: TVar StdGen,
+    queueLocks :: TMap RecipientId Lock,
+    msgQueues :: TMap RecipientId JournalMsgQueue
+  }
+
+data JournalStoreConfig = JournalStoreConfig
+  { storePath :: FilePath,
+    pathParts :: Int,
+    quota :: Int,
+    -- Max number of messages per journal file - ignored in STM store.
+    -- When this limit is reached, the file will be changed.
+    -- This number should be set bigger than queue quota.
+    maxMsgCount :: Int,
+    maxStateLines :: Int
+  }
+
+data JMQueue = JMQueue
+  { queueDirectory :: FilePath,
+    queueLock :: Lock,
+    statePath :: FilePath
+  }
+
+data JournalMsgQueue = JournalMsgQueue
+  { queue :: JMQueue,
+    state :: TVar MsgQueueState,
+    -- tipMsg contains last message and length incl. newline
+    -- Nothing - unknown, Just Nothing - empty queue.
+    -- It  prevents reading each message twice,
+    -- and reading it after it was just written.
+    tipMsg :: TVar (Maybe (Maybe (Message, Int64))),
+    handles :: TVar (Maybe MsgQueueHandles)
+  }
+
+data MsgQueueState = MsgQueueState
+  { readState :: JournalState 'JTRead,
+    writeState :: JournalState 'JTWrite,
+    canWrite :: Bool,
+    size :: Int
+  }
+  deriving (Show)
+
+data MsgQueueHandles = MsgQueueHandles
+  { stateHandle :: Handle, -- handle to queue state log file, rotates and removes old backups when server is restarted
+    readHandle :: Handle,
+    writeHandle :: Maybe Handle -- optional, used when write file is different from read file
+  }
+
+data JournalState t = JournalState
+  { journalType :: SJournalType t,
+    journalId :: ByteString,
+    msgPos :: Int,
+    msgCount :: Int,
+    bytePos :: Int64,
+    byteCount :: Int64
+  }
+  deriving (Show)
+
+data JournalType = JTRead | JTWrite
+
+data SJournalType (t :: JournalType) where
+  SJTRead :: SJournalType 'JTRead
+  SJTWrite :: SJournalType 'JTWrite
+
+class JournalTypeI t where sJournalType :: SJournalType t
+
+instance JournalTypeI 'JTRead where sJournalType = SJTRead
+
+instance JournalTypeI 'JTWrite where sJournalType = SJTWrite
+
+deriving instance Show (SJournalType t)
+
+newMsgQueueState :: ByteString -> MsgQueueState
+newMsgQueueState journalId =
+  MsgQueueState
+    { writeState = newJournalState journalId,
+      readState = newJournalState journalId,
+      canWrite = True,
+      size = 0
+    }
+
+newJournalState :: JournalTypeI t => ByteString -> JournalState t
+newJournalState journalId = JournalState sJournalType journalId 0 0 0 0
+
+journalFilePath :: FilePath -> ByteString -> FilePath
+journalFilePath dir journalId = dir </> (msgLogFileName <> "." <> B.unpack journalId <> logFileExt)
+
+instance StrEncoding MsgQueueState where
+  strEncode MsgQueueState {writeState, readState, canWrite, size} =
+    B.unwords
+      [ "write=" <> strEncode writeState,
+        "read=" <> strEncode readState,
+        "canWrite=" <> strEncode canWrite,
+        "size=" <> strEncode size
+      ]
+  strP = do
+    writeState <- "write=" *> strP
+    readState <- " read=" *> strP
+    canWrite <- " canWrite=" *> strP
+    size <- " size=" *> strP
+    pure MsgQueueState {writeState, readState, canWrite, size}
+
+instance JournalTypeI t => StrEncoding (JournalState t) where
+  strEncode JournalState {journalId, msgPos, msgCount, bytePos, byteCount} =
+    B.intercalate "," [journalId, e msgPos, e msgCount, e bytePos, e byteCount]
+    where
+      e :: StrEncoding a => a -> ByteString
+      e = strEncode
+  strP = do
+    journalId <- A.takeTill (== ',')
+    JournalState sJournalType journalId <$> i <*> i <*> i <*> i
+    where
+      i :: Integral a => A.Parser a
+      i = A.char ',' *> A.decimal
+
+queueLogFileName :: String
+queueLogFileName = "queue_state"
+
+msgLogFileName :: String
+msgLogFileName = "messages"
+
+logFileExt :: String
+logFileExt = ".log"
+
+newtype StoreIO a = StoreIO {unStoreIO :: IO a}
+  deriving newtype (Functor, Applicative, Monad)
+
+instance MsgStoreClass JournalMsgStore where
+  type StoreMonad JournalMsgStore = StoreIO
+  type MsgQueue JournalMsgStore = JournalMsgQueue
+  type MsgStoreConfig JournalMsgStore = JournalStoreConfig
+
+  newMsgStore :: JournalStoreConfig -> IO JournalMsgStore
+  newMsgStore config = do
+    random <- newTVarIO =<< newStdGen
+    queueLocks <- TM.emptyIO
+    msgQueues <- TM.emptyIO
+    pure JournalMsgStore {config, random, queueLocks, msgQueues}
+
+  closeMsgStore st = atomically (swapTVar (msgQueues st) M.empty) >>= mapM_ closeMsgQueueHandles
+
+  activeMsgQueues = msgQueues
+  {-# INLINE activeMsgQueues #-}
+
+  -- 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 -> JournalMsgQueue -> IO a) -> IO a
+  withAllMsgQueues tty ms@JournalMsgStore {config} action = ifM (doesDirectoryExist storePath) processStore (pure mempty)
+    where
+      processStore = do
+        closeMsgStore ms
+        lock <- createLockIO -- the same lock is used for all queues
+        (!count, !res) <- foldQueues 0 (processQueue lock) (0, mempty) ("", storePath)
+        putStrLn $ progress count
+        pure res
+      JournalStoreConfig {storePath, pathParts} = config
+      processQueue :: Lock -> (Int, a) -> (String, FilePath) -> IO (Int, a)
+      processQueue queueLock (!i, !r) (queueId, dir) = do
+        when (tty && i `mod` 100 == 0) $ putStr (progress i <> "\r") >> IO.hFlush stdout
+        let statePath = msgQueueStatePath dir queueId
+        q <- openMsgQueue ms JMQueue {queueDirectory = dir, queueLock, statePath}
+        r' <- case strDecode $ B.pack queueId of
+          Right rId -> action rId q
+          Left e -> do
+            putStrLn ("Error: message queue directory " <> dir <> " is invalid: " <> e)
+            exitFailure
+        closeMsgQueueHandles q
+        pure (i + 1, r <> r')
+      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
+        listDirs >>= foldM f' acc
+        where
+          listDirs = fmap catMaybes . mapM queuePath =<< listDirectory path
+          queuePath dir = do
+            let !path' = path </> dir
+                !queueId' = queueId <> dir
+            ifM
+              (doesDirectoryExist path')
+              (pure $ Just (queueId', path'))
+              (Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))
+
+  logQueueStates :: JournalMsgStore -> IO ()
+  logQueueStates ms = withActiveMsgQueues ms $ \_ -> logQueueState
+
+  logQueueState :: JournalMsgQueue -> IO ()
+  logQueueState q = 
+    readTVarIO (handles q)
+      >>= maybe (pure ()) (\hs -> readTVarIO (state q) >>= appendState (stateHandle hs))
+
+  getMsgQueue :: JournalMsgStore -> RecipientId -> ExceptT ErrorType IO JournalMsgQueue
+  getMsgQueue ms@JournalMsgStore {queueLocks, msgQueues, random} rId =
+    tryStore "getMsgQueue" (B.unpack $ strEncode rId) $ withLockMap queueLocks rId "getMsgQueue" $
+      TM.lookupIO rId msgQueues >>= maybe newQ pure
+    where
+      newQ = do
+        queueLock <- atomically $ getMapLock queueLocks rId
+        let dir = msgQueueDirectory ms rId
+            statePath = msgQueueStatePath dir $ B.unpack (strEncode rId)
+            queue = JMQueue {queueDirectory = dir, queueLock, statePath}
+        q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue) (createQ queue)
+        atomically $ TM.insert rId q msgQueues
+        pure q
+        where
+          createQ :: JMQueue -> IO JournalMsgQueue
+          createQ queue = 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)
+
+  delMsgQueue :: JournalMsgStore -> RecipientId -> IO ()
+  delMsgQueue ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do
+    closeMsgQueue ms rId
+    removeQueueDirectory ms rId
+
+  delMsgQueueSize :: JournalMsgStore -> RecipientId -> IO Int
+  delMsgQueueSize ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do
+    st_ <-
+      atomically (TM.lookupDelete rId (msgQueues ms))
+        >>= mapM (\q -> closeMsgQueueHandles q >> readTVarIO (state q))
+    removeQueueDirectory ms rId
+    pure $ maybe (-1) size st_
+
+  getQueueMessages :: Bool -> JournalMsgQueue -> IO [Message]
+  getQueueMessages drainMsgs q = run []
+    where
+      run msgs = readTVarIO (handles q) >>= maybe (pure []) (getMsg msgs)
+      getMsg msgs hs = chooseReadJournal 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 -> JournalMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
+  writeMsg ms q@JournalMsgQueue {queue = JMQueue {queueDirectory, statePath}, handles} logState msg =
+    isolateQueue q "writeMsg" $ StoreIO $ do
+      st@MsgQueueState {canWrite, size} <- readTVarIO (state q)
+      let empty = size == 0
+      if canWrite || empty
+        then do
+          let canWrt' = quota > size
+          if canWrt'
+            then writeToJournal st canWrt' msg $> Just (msg, empty)
+            else writeToJournal st canWrt' msgQuota $> Nothing
+        else pure Nothing
+    where
+      JournalStoreConfig {quota, maxMsgCount} = config ms
+      msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
+      writeToJournal st@MsgQueueState {writeState, readState = rs, size} canWrt' !msg' = do
+        let msgStr = strEncode msg' `B.snoc` '\n'
+            msgLen = fromIntegral $ B.length msgStr
+        hs <- maybe createQueueDir pure =<< readTVarIO handles
+        (ws, wh) <- case writeHandle hs of
+          Nothing | msgCount writeState >= maxMsgCount -> switchWriteJournal hs
+          wh_ -> pure (writeState, fromMaybe (readHandle hs) wh_)
+        let msgPos' = msgPos ws + 1
+            bytePos' = bytePos ws + msgLen
+            ws' = ws {msgPos = msgPos', msgCount = msgPos', bytePos = bytePos', byteCount = bytePos'}
+            rs' = if journalId ws == journalId rs then rs {msgCount = msgPos', byteCount = bytePos'} else rs
+            !st' = st {writeState = ws', readState = rs', canWrite = canWrt', size = size + 1}
+        hAppend wh (bytePos ws) msgStr
+        updateQueueState q logState hs st' $
+          when (size == 0) $ writeTVar (tipMsg q) $ Just (Just (msg, msgLen))
+        where
+          createQueueDir = do
+            createDirectoryIfMissing True queueDirectory
+            sh <- openFile statePath AppendMode
+            B.hPutStr sh ""
+            rh <- createNewJournal queueDirectory $ 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
+            atomically $ writeTVar handles $ Just $ hs {writeHandle = Just wh}
+            pure (newJournalState journalId, wh)
+
+  -- can ONLY be used while restoring messages, not while server running
+  setOverQuota_ :: JournalMsgQueue -> IO ()
+  setOverQuota_ JournalMsgQueue {state} = atomically $ modifyTVar' state $ \st -> st {canWrite = False}
+
+  getQueueSize :: JournalMsgQueue -> IO Int
+  getQueueSize JournalMsgQueue {state} = size <$> readTVarIO state
+
+  tryPeekMsg_ :: JournalMsgQueue -> StoreIO (Maybe Message)
+  tryPeekMsg_ q@JournalMsgQueue {tipMsg, handles} =
+    StoreIO $ readTVarIO handles $>>= chooseReadJournal q True $>>= peekMsg
+    where
+      peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst)
+        where
+          readMsg = do
+            ml@(msg, _) <- hGetMsgAt h $ bytePos rs
+            atomically $ writeTVar tipMsg $ Just (Just ml)
+            pure $ Just msg
+
+  tryDeleteMsg_ :: JournalMsgQueue -> Bool -> StoreIO ()
+  tryDeleteMsg_ q@JournalMsgQueue {tipMsg, handles} logState = StoreIO $
+    void $
+      readTVarIO tipMsg -- if there is no cached tipMsg, do nothing
+        $>>= (pure . fmap snd)
+        $>>= \len -> readTVarIO handles
+        $>>= \hs -> updateReadPos q logState len hs $> Just ()
+
+  isolateQueue :: JournalMsgQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
+  isolateQueue JournalMsgQueue {queue = q} op =
+    tryStore op (queueDirectory q) . withLock' (queueLock q) op . unStoreIO
+
+tryStore :: String -> String -> IO a -> ExceptT ErrorType IO a
+tryStore op qId a = ExceptT $ E.mask_ $ E.try a >>= bimapM storeErr pure
+  where
+    storeErr :: E.SomeException -> IO ErrorType
+    storeErr e =
+      let e' = intercalate ", " [op, qId, show e]
+       in logError ("STORE: " <> T.pack e') $> STORE e'
+
+openMsgQueue :: JournalMsgStore -> JMQueue -> IO JournalMsgQueue
+openMsgQueue ms q@JMQueue {queueDirectory = dir, statePath} = do
+  (st, sh) <- readWriteQueueState ms statePath
+  (st', rh, wh_) <- closeOnException sh $ openJournals dir st sh
+  let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = wh_}
+  mkJournalQueue q (st', Just hs)
+
+mkJournalQueue :: JMQueue -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue
+mkJournalQueue queue (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}
+
+chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle))
+chooseReadJournal 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
+      let !rs' = (newJournalState $ journalId ws) {msgCount = msgCount ws, byteCount = byteCount ws}
+          !st' = st {readState = rs'}
+      updateQueueState q log' hs st' $ pure ()
+      pure $ Just (rs', wh)
+    _ | msgPos rs >= msgCount rs && journalId rs == journalId ws -> pure Nothing
+    _ -> pure $ Just (rs, readHandle hs)
+
+updateQueueState :: JournalMsgQueue -> 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
+  atomically $ writeTVar (state q) st >> a
+
+appendState :: Handle -> MsgQueueState -> IO ()
+appendState h st = E.uninterruptibleMask_ $ B.hPutStr h $ strEncode st `B.snoc` '\n'
+
+updateReadPos :: JournalMsgQueue -> Bool -> Int64 -> MsgQueueHandles -> IO ()
+updateReadPos q log' len hs = do
+  st@MsgQueueState {readState = rs, size} <- readTVarIO (state q)
+  let JournalState {msgPos, bytePos} = rs
+  let msgPos' = msgPos + 1
+      rs' = rs {msgPos = msgPos', bytePos = bytePos + len}
+      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)
+  where
+    splitSegments _ "" = []
+    splitSegments 1 s = [s]
+    splitSegments n s =
+      let (seg, s') = B.splitAt 2 s
+       in seg : splitSegments (n - 1) s'
+
+msgQueueStatePath :: FilePath -> String -> FilePath
+msgQueueStatePath dir queueId = dir </> (queueLogFileName <> "." <> queueId <> logFileExt)
+
+createNewJournal :: FilePath -> ByteString -> IO Handle
+createNewJournal dir journalId = do
+  let path = journalFilePath dir journalId -- TODO retry if file exists
+  h <- openFile path ReadWriteMode
+  B.hPutStr h ""
+  pure h
+
+newJournalId :: TVar StdGen -> IO ByteString
+newJournalId g = strEncode <$> atomically (stateTVar g $ genByteString 12)
+
+openJournals :: FilePath -> MsgQueueState -> Handle -> IO (MsgQueueState, Handle, Maybe Handle)
+openJournals dir st@MsgQueueState {readState = rs, writeState = ws} sh = do
+  let rjId = journalId rs
+      wjId = journalId ws
+  openJournal rs >>= \case
+    Left path -> do
+      logError $ "STORE: openJournals, no read file - creating new file, " <> T.pack path
+      rh <- createNewJournal dir rjId
+      let st' = newMsgQueueState rjId
+      closeOnException rh $ appendState sh st'
+      pure (st', rh, Nothing)
+    Right rh
+      | rjId == wjId -> do
+          closeOnException rh $ fixFileSize rh $ bytePos ws
+          pure (st, rh, Nothing)
+      | otherwise -> closeOnException rh $ do
+          fixFileSize rh $ byteCount rs
+          openJournal ws >>= \case
+            Left path -> do
+              logError $ "STORE: openJournals, no write file - creating new file, " <> T.pack path
+              wh <- createNewJournal dir wjId
+              let size' = msgCount rs - msgPos rs
+                  st' = st {writeState = newJournalState wjId, size = size'} -- we don't amend canWrite to trigger QCONT
+              closeOnException wh $ appendState sh st'
+              pure (st', rh, Just wh)
+            Right wh -> do
+              closeOnException wh $ fixFileSize wh $ bytePos ws
+              pure (st, rh, Just wh)
+  where
+    openJournal :: JournalState t -> IO (Either FilePath Handle)
+    openJournal JournalState {journalId} =
+      let path = journalFilePath dir journalId
+       in ifM (doesFileExist path) (Right <$> openFile path ReadWriteMode) (pure $ Left path)
+    -- do that for all append operations
+
+fixFileSize :: Handle -> Int64 -> IO ()
+fixFileSize h pos = do
+  let pos' = fromIntegral pos
+  size <- IO.hFileSize h
+  if
+    | size > pos' -> do
+        name <- IO.hShow h
+        logWarn $ "STORE: fixFileSize, size " <> tshow size <> " > pos " <> tshow pos <> " - truncating, " <> T.pack name
+        IO.hSetFileSize h pos'
+    | size < pos' -> do
+        -- From code logic this can't happen.
+        name <- IO.hShow h
+        E.throwIO $ userError $ "fixFileSize size " <> show size <> " < pos " <> show pos <> " - aborting: " <> name
+    | otherwise -> pure ()
+
+removeJournal :: FilePath -> JournalState t -> IO ()
+removeJournal dir JournalState {journalId} = do
+  let path = journalFilePath dir journalId
+  removeFile path `catchAny` (\e -> logError $ "STORE: removeJournal, " <> T.pack path <> ", " <> tshow e)
+
+-- 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)
+readWriteQueueState JournalMsgStore {random, config} statePath =
+  ifM
+    (doesFileExist tempBackup)
+    (renameFile tempBackup statePath >> readQueueState)
+    (ifM (doesFileExist statePath) readQueueState writeNewQueueState)
+  where
+    tempBackup = statePath <> ".bak"
+    readQueueState = do
+      ls <- LB.lines <$> LB.readFile statePath
+      case ls of
+        [] -> writeNewQueueState
+        _ -> do
+          r@(st, _) <- useLastLine (length ls) True ls
+          unless (validQueueState st) $ E.throwIO $ userError $ "readWriteQueueState inconsistent state: " <> show st
+          pure r
+    writeNewQueueState = do
+      logWarn $ "STORE: readWriteQueueState, empty queue state - initialized, " <> T.pack statePath
+      st <- newMsgQueueState <$> newJournalId random
+      writeQueueState st
+    useLastLine len isLastLine ls = case strDecode $ LB.toStrict $ last ls of
+      Right st
+        | len > maxStateLines config || not isLastLine ->
+            backupWriteQueueState st
+        | otherwise -> do
+            -- when state file has fewer than maxStateLines, we don't compact it
+            sh <- openFile statePath AppendMode
+            pure (st, sh)
+      Left e -- if the last line failed to parse
+        | isLastLine -> case init ls of -- or use the previous line
+            [] -> do
+              logWarn $ "STORE: readWriteQueueState, invalid 1-line queue state - initialized, " <> T.pack statePath
+              st <- newMsgQueueState <$> newJournalId random
+              backupWriteQueueState st
+            ls' -> do
+              logWarn $ "STORE: readWriteQueueState, invalid last line in queue state - using the previous line, " <> T.pack statePath
+              useLastLine len False ls'
+        | otherwise -> E.throwIO $ userError $ "readWriteQueueState invalid state " <> statePath <> ": " <> show e
+    backupWriteQueueState st = do
+      -- State backup is made in two steps to mitigate the crash during the backup.
+      -- 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
+      pure r
+    writeQueueState st = do
+      sh <- openFile statePath AppendMode
+      closeOnException sh $ appendState sh st
+      pure (st, sh)
+
+validQueueState :: MsgQueueState -> Bool
+validQueueState MsgQueueState {readState = rs, writeState = ws, size}
+  | journalId rs == journalId ws =
+      alwaysValid
+        && msgPos rs <= msgPos ws
+        && msgCount rs == msgCount ws
+        && bytePos rs <= bytePos ws
+        && byteCount rs == byteCount ws
+        && size == msgCount rs - msgPos rs
+  | otherwise =
+      alwaysValid
+        && size == msgCount ws + msgCount rs - msgPos rs
+  where
+    alwaysValid =
+      msgPos rs <= msgCount rs
+        && bytePos rs <= byteCount rs
+        && msgPos ws == msgCount ws
+        && bytePos ws == byteCount ws
+
+closeMsgQueue :: JournalMsgStore -> RecipientId -> IO ()
+closeMsgQueue ms rId =
+  atomically (TM.lookupDelete rId (msgQueues ms))
+    >>= mapM_ closeMsgQueueHandles
+
+closeMsgQueueHandles :: JournalMsgQueue -> IO ()
+closeMsgQueueHandles q = readTVarIO (handles q) >>= mapM_ closeHandles
+  where
+    closeHandles (MsgQueueHandles sh rh wh_) = do
+      hClose sh
+      hClose rh
+      mapM_ hClose wh_
+
+removeQueueDirectory :: JournalMsgStore -> RecipientId -> IO ()
+removeQueueDirectory st rId =
+  let dir = msgQueueDirectory st rId
+   in removePathForcibly dir `catchAny` (\e -> logError $ "STORE: removeQueueDirectory, " <> T.pack dir <> ", " <> tshow e)
+
+hAppend :: Handle -> Int64 -> ByteString -> IO ()
+hAppend h pos s = do
+  fixFileSize h pos
+  IO.hSeek h SeekFromEnd 0
+  B.hPutStr h s
+
+hGetMsgAt :: Handle -> Int64 -> IO (Message, Int64)
+hGetMsgAt h pos = do
+  IO.hSeek h AbsoluteSeek $ fromIntegral pos
+  s <- B.hGetLine h
+  case strDecode s of
+    Right !msg ->
+      let !len = fromIntegral (B.length s) + 1
+       in pure (msg, len)
+    Left e -> E.throwIO $ userError $ "hGetMsgAt invalid message: " <> e
+
+openFile :: FilePath -> IOMode -> IO Handle
+openFile f mode = do
+  h <- IO.openFile f mode
+  IO.hSetBuffering h LineBuffering
+  pure h
+
+hClose :: Handle -> IO ()
+hClose h = IO.hClose h `catchAny` (\e -> logError $ "STORE: hClose, error closing file, " <> tshow e)
+
+closeOnException :: Handle -> IO a -> IO a
+closeOnException h a = a `E.onException` hClose h

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

@@ -1,129 +1,131 @@
 {-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE ConstraintKinds #-}
+{-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE InstanceSigs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE TupleSections #-}
+{-# LANGUAGE TypeFamilies #-}
 
 module Simplex.Messaging.Server.MsgStore.STM
-  ( STMMsgStore,
-    MsgQueue (msgQueue),
-    newMsgStore,
-    getMsgQueue,
-    delMsgQueue,
-    delMsgQueueSize,
-    writeMsg,
-    tryPeekMsg,
-    tryPeekMsgIO,
-    tryDelMsg,
-    tryDelPeekMsg,
-    deleteExpiredMsgs,
-    getQueueSize,
+  ( STMMsgStore (..),
+    STMMsgQueue (msgQueue),
+    STMStoreConfig (..),
   )
 where
 
-import qualified Data.ByteString.Char8 as B
+import Control.Concurrent.STM
+import Control.Monad.IO.Class
+import Control.Monad.Trans.Except
 import Data.Functor (($>))
-import Data.Int (Int64)
-import Data.Time.Clock.System (SystemTime (systemSeconds))
-import Simplex.Messaging.Protocol (Message (..), MsgId, RecipientId)
+import Simplex.Messaging.Protocol (ErrorType, Message (..), RecipientId)
+import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import UnliftIO.STM
 
-data MsgQueue = MsgQueue
+data STMMsgQueue = STMMsgQueue
   { msgQueue :: TQueue Message,
     quota :: Int,
     canWrite :: TVar Bool,
     size :: TVar Int
   }
 
-type STMMsgStore = TMap RecipientId MsgQueue
+data STMMsgStore = STMMsgStore
+  { storeConfig :: STMStoreConfig,
+    msgQueues :: TMap RecipientId STMMsgQueue
+  }
 
-newMsgStore :: IO STMMsgStore
-newMsgStore = TM.emptyIO
+data STMStoreConfig = STMStoreConfig
+  { storePath :: Maybe FilePath,
+    quota :: Int
+  }
 
--- The reason for double lookup is that majority of messaging queues exist,
--- because multiple messages are sent to the same queue,
--- so the first lookup without STM transaction will return the queue faster.
--- In case the queue does not exist, it needs to be looked-up again inside transaction.
-getMsgQueue :: STMMsgStore -> RecipientId -> Int -> IO MsgQueue
-getMsgQueue st rId quota = TM.lookupIO rId st >>= maybe (atomically maybeNewQ) pure
-  where
-    maybeNewQ = TM.lookup rId st >>= maybe newQ pure
-    newQ = do
-      msgQueue <- newTQueue
-      canWrite <- newTVar True
-      size <- newTVar 0
-      let q = MsgQueue {msgQueue, quota, canWrite, size}
-      TM.insert rId q st
-      pure q
+instance MsgStoreClass STMMsgStore where
+  type StoreMonad STMMsgStore = STM
+  type MsgQueue STMMsgStore = STMMsgQueue
+  type MsgStoreConfig STMMsgStore = STMStoreConfig
 
-delMsgQueue :: STMMsgStore -> RecipientId -> IO ()
-delMsgQueue st rId = atomically $ TM.delete rId st
+  newMsgStore :: STMStoreConfig -> IO STMMsgStore
+  newMsgStore storeConfig = do
+    msgQueues <- TM.emptyIO
+    pure STMMsgStore {storeConfig, msgQueues}
 
-delMsgQueueSize :: STMMsgStore -> RecipientId -> IO Int
-delMsgQueueSize st rId = atomically (TM.lookupDelete rId st) >>= maybe (pure 0) (\MsgQueue {size} -> readTVarIO size)
+  closeMsgStore _ = pure ()
 
-writeMsg :: MsgQueue -> Message -> IO (Maybe (Message, Bool))
-writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} !msg = atomically $ do
-  canWrt <- readTVar canWrite
-  empty <- isEmptyTQueue q
-  if canWrt || empty
-    then do
-      canWrt' <- (quota >) <$> readTVar size
-      writeTVar canWrite $! canWrt'
-      modifyTVar' size (+ 1)
-      if canWrt'
-        then writeTQueue q msg $> Just (msg, empty)
-        else (writeTQueue q $! msgQuota) $> Nothing
-    else pure Nothing
-  where
-    msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
+  activeMsgQueues = msgQueues
+  {-# INLINE activeMsgQueues #-}
 
-tryPeekMsgIO :: MsgQueue -> IO (Maybe Message)
-tryPeekMsgIO = atomically . tryPeekTQueue . msgQueue
-{-# INLINE tryPeekMsgIO #-}
+  withAllMsgQueues _ = withActiveMsgQueues
+  {-# INLINE withAllMsgQueues #-}
 
--- TODO remove once deliverToSub is split
-tryPeekMsg :: MsgQueue -> STM (Maybe Message)
-tryPeekMsg = tryPeekTQueue . msgQueue
-{-# INLINE tryPeekMsg #-}
+  logQueueStates _ = pure ()
 
-tryDelMsg :: MsgQueue -> MsgId -> IO (Maybe Message)
-tryDelMsg mq msgId' = atomically $
-  tryPeekMsg mq >>= \case
-    msg_@(Just msg)
-      | msgId msg == msgId' || B.null msgId' -> tryDeleteMsg_ mq >> pure msg_
-      | otherwise -> pure Nothing
-    _ -> pure Nothing
+  logQueueState _ = pure ()
 
--- atomic delete (== read) last and peek next message if available
-tryDelPeekMsg :: MsgQueue -> MsgId -> IO (Maybe Message, Maybe Message)
-tryDelPeekMsg mq msgId' = atomically $
-  tryPeekMsg mq >>= \case
-    msg_@(Just msg)
-      | msgId msg == msgId' || B.null msgId' -> (msg_,) <$> (tryDeleteMsg_ mq >> tryPeekMsg mq)
-      | otherwise -> pure (Nothing, msg_)
-    _ -> pure (Nothing, Nothing)
+  -- The reason for double lookup is that majority of messaging queues exist,
+  -- because multiple messages are sent to the same queue,
+  -- so the first lookup without STM transaction will return the queue faster.
+  -- In case the queue does not exist, it needs to be looked-up again inside transaction.
+  getMsgQueue :: STMMsgStore -> RecipientId -> ExceptT ErrorType IO STMMsgQueue
+  getMsgQueue STMMsgStore {msgQueues = qs, storeConfig = STMStoreConfig {quota}} rId =
+    liftIO $ TM.lookupIO rId qs >>= maybe (atomically maybeNewQ) pure
+    where
+      maybeNewQ = TM.lookup rId qs >>= maybe newQ pure
+      newQ = do
+        msgQueue <- newTQueue
+        canWrite <- newTVar True
+        size <- newTVar 0
+        let q = STMMsgQueue {msgQueue, quota, canWrite, size}
+        TM.insert rId q qs
+        pure q
 
-deleteExpiredMsgs :: MsgQueue -> Int64 -> IO Int
-deleteExpiredMsgs mq old = atomically $ loop 0
-  where
-    loop dc =
-      tryPeekMsg mq >>= \case
-        Just Message {msgTs}
-          | systemSeconds msgTs < old ->
-              tryDeleteMsg_ mq >> loop (dc + 1)
-        _ -> pure dc
+  delMsgQueue :: STMMsgStore -> RecipientId -> IO ()
+  delMsgQueue st rId = atomically $ TM.delete rId $ msgQueues st
 
-tryDeleteMsg_ :: MsgQueue -> STM ()
-tryDeleteMsg_ MsgQueue {msgQueue = q, size} =
-  tryReadTQueue q >>= \case
-    Just _ -> modifyTVar' size (subtract 1)
-    _ -> pure ()
+  delMsgQueueSize :: STMMsgStore -> RecipientId -> IO Int
+  delMsgQueueSize st rId = atomically (TM.lookupDelete rId $ msgQueues st) >>= maybe (pure 0) (\STMMsgQueue {size} -> readTVarIO size)
 
-getQueueSize :: MsgQueue -> IO Int
-getQueueSize MsgQueue {size} = readTVarIO size
+  getQueueMessages :: Bool -> STMMsgQueue -> IO [Message]
+  getQueueMessages drainMsgs = atomically . (if drainMsgs then flushTQueue else snapshotTQueue) . msgQueue
+    where
+      snapshotTQueue q = do
+        msgs <- flushTQueue q
+        mapM_ (writeTQueue q) msgs
+        pure msgs
+
+  writeMsg :: STMMsgStore -> STMMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
+  writeMsg _ STMMsgQueue {msgQueue = q, quota, canWrite, size} _logState msg = liftIO $ atomically $ do
+    canWrt <- readTVar canWrite
+    empty <- isEmptyTQueue q
+    if canWrt || empty
+      then do
+        canWrt' <- (quota >) <$> readTVar size
+        writeTVar canWrite $! canWrt'
+        modifyTVar' size (+ 1)
+        if canWrt'
+          then (writeTQueue q $! msg) $> Just (msg, empty)
+          else (writeTQueue q $! msgQuota) $> Nothing
+      else pure Nothing
+    where
+      msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
+
+  setOverQuota_ :: STMMsgQueue -> IO ()
+  setOverQuota_ q = atomically $ writeTVar (canWrite q) False
+
+  getQueueSize :: STMMsgQueue -> IO Int
+  getQueueSize STMMsgQueue {size} = readTVarIO size
+
+  tryPeekMsg_ :: STMMsgQueue -> STM (Maybe Message)
+  tryPeekMsg_ = tryPeekTQueue . msgQueue
+  {-# INLINE tryPeekMsg_ #-}
+
+  tryDeleteMsg_ :: STMMsgQueue -> Bool -> STM ()
+  tryDeleteMsg_ STMMsgQueue {msgQueue = q, size} _logState =
+    tryReadTQueue q >>= \case
+      Just _ -> modifyTVar' size (subtract 1)
+      _ -> pure ()
+
+  isolateQueue :: STMMsgQueue -> String -> STM a -> ExceptT ErrorType IO a
+  isolateQueue _ _ = liftIO . atomically

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

@@ -0,0 +1,89 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE TupleSections #-}
+{-# LANGUAGE TypeFamilyDependencies #-}
+
+module Simplex.Messaging.Server.MsgStore.Types where
+
+import Control.Concurrent.STM
+import Control.Monad (foldM)
+import Control.Monad.Trans.Except
+import Data.Int (Int64)
+import Data.Kind
+import qualified Data.Map.Strict as M
+import Data.Time.Clock.System (SystemTime (systemSeconds))
+import Simplex.Messaging.Protocol (ErrorType, Message (..), MsgId, RecipientId)
+import Simplex.Messaging.TMap (TMap)
+
+class Monad (StoreMonad s) => MsgStoreClass s where
+  type StoreMonad s = (m :: Type -> Type) | m -> s
+  type MsgStoreConfig s = c | c -> s
+  type MsgQueue s = q | q -> s
+  newMsgStore :: MsgStoreConfig s -> IO s
+  closeMsgStore :: s -> IO ()
+  activeMsgQueues :: s -> TMap RecipientId (MsgQueue s)
+  withAllMsgQueues :: Monoid a => Bool -> s -> (RecipientId -> MsgQueue s -> IO a) -> IO a
+  logQueueStates :: s -> IO ()
+  logQueueState :: MsgQueue s -> IO ()
+  getMsgQueue :: s -> RecipientId -> ExceptT ErrorType IO (MsgQueue s)
+  delMsgQueue :: s -> RecipientId -> IO ()
+  delMsgQueueSize :: s -> RecipientId -> IO Int
+  getQueueMessages :: Bool -> MsgQueue s -> IO [Message]
+  writeMsg :: s -> MsgQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
+  setOverQuota_ :: MsgQueue s -> IO () -- can ONLY be used while restoring messages, not while server running
+  getQueueSize :: MsgQueue s -> IO Int
+  tryPeekMsg_ :: MsgQueue s -> StoreMonad s (Maybe Message)
+  tryDeleteMsg_ :: MsgQueue s -> Bool -> StoreMonad s ()
+  isolateQueue :: MsgQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
+
+data MSType = MSMemory | MSJournal
+
+data SMSType :: MSType -> Type where
+  SMSMemory :: SMSType 'MSMemory
+  SMSJournal :: SMSType 'MSJournal
+
+data AMSType = forall s. AMSType (SMSType s)
+
+withActiveMsgQueues :: (MsgStoreClass s, Monoid a) => s -> (RecipientId -> MsgQueue 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
+
+tryPeekMsg :: MsgStoreClass s => MsgQueue s -> ExceptT ErrorType IO (Maybe Message)
+tryPeekMsg mq = isolateQueue mq "tryPeekMsg" $ tryPeekMsg_ mq
+{-# INLINE tryPeekMsg #-}
+
+tryDelMsg :: MsgStoreClass s => MsgQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
+tryDelMsg mq msgId' =
+  isolateQueue mq "tryDelMsg" $
+    tryPeekMsg_ mq >>= \case
+      msg_@(Just msg)
+        | msgId msg == msgId' ->
+            tryDeleteMsg_ mq True >> pure msg_
+      _ -> pure Nothing
+
+-- atomic delete (== read) last and peek next message if available
+tryDelPeekMsg :: MsgStoreClass s => MsgQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
+tryDelPeekMsg mq msgId' =
+  isolateQueue mq "tryDelPeekMsg" $
+    tryPeekMsg_ mq >>= \case
+      msg_@(Just msg)
+        | msgId msg == msgId' -> (msg_,) <$> (tryDeleteMsg_ mq True >> tryPeekMsg_ mq)
+        | otherwise -> pure (Nothing, msg_)
+      _ -> pure (Nothing, Nothing)
+
+deleteExpiredMsgs :: MsgStoreClass s => MsgQueue s -> Bool -> Int64 -> ExceptT ErrorType IO Int
+deleteExpiredMsgs mq logState old = isolateQueue mq "deleteExpiredMsgs" $ loop 0
+  where
+    loop dc =
+      tryPeekMsg_ mq >>= \case
+        Just Message {msgTs}
+          | systemSeconds msgTs < old ->
+              tryDeleteMsg_ mq logState >> loop (dc + 1)
+        _ -> pure dc

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

@@ -47,10 +47,11 @@ newQueueStore = do
   pure QueueStore {queues, senders, notifiers}
 
 addQueue :: QueueStore -> QueueRec -> IO (Either ErrorType ())
-addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = atomically $ do
+addQueue QueueStore {queues, senders, notifiers} q@QueueRec {recipientId = rId, senderId = sId, notifier} = atomically $ do
   ifM hasId (pure $ Left DUPLICATE_) $ do
     TM.insertM rId (newTVar q) queues
     TM.insert sId rId senders
+    forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId notifiers
     pure $ Right ()
   where
     hasId = (||) <$> TM.member rId queues <*> TM.member sId senders

src/Simplex/Messaging/Server/StoreLog.hs

@@ -241,16 +241,16 @@ readQueues f st = withFile f ReadMode $ LB.hGetContents >=> mapM_ processLine .
         s = LB.toStrict s'
         procLogRecord :: StoreLogRecord -> IO ()
         procLogRecord = \case
-          CreateQueue q -> addQueue st q >>= qError (recipientId q)
-          SecureQueue qId sKey -> secureQueue st qId sKey >>= qError qId
-          AddNotifier qId ntfCreds -> addQueueNotifier st qId ntfCreds >>= qError qId
-          SuspendQueue qId -> suspendQueue st qId >>= qError qId
-          DeleteQueue qId -> deleteQueue st qId >>= qError qId
-          DeleteNotifier qId -> deleteQueueNotifier st qId >>= qError qId
+          CreateQueue q -> addQueue st q >>= qError "create" (recipientId q)
+          SecureQueue qId sKey -> secureQueue st qId sKey >>= qError "secure" qId
+          AddNotifier qId ntfCreds -> addQueueNotifier st qId ntfCreds >>= qError "addNotifier" qId
+          SuspendQueue qId -> suspendQueue st qId >>= qError "suspend" qId
+          DeleteQueue qId -> deleteQueue st qId >>= qError "delete" qId
+          DeleteNotifier qId -> deleteQueueNotifier st qId >>= qError "deleteNotifier" qId
           UpdateTime qId t -> updateQueueTime st qId t
         printError :: String -> IO ()
         printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
-        qError :: RecipientId -> Either ErrorType a -> IO ()
-        qError (EntityId qId) = \case
-          Left e -> B.putStrLn $ "stored queue " <> B64.encode qId <> " error: " <> bshow e
+        qError :: B.ByteString -> RecipientId -> Either ErrorType a -> IO ()
+        qError op (EntityId qId) = \case
+          Left e -> B.putStrLn $ op <> " stored queue " <> B64.encode qId <> " error: " <> bshow e
           Right _ -> pure ()

src/Simplex/Messaging/Util.hs

@@ -14,11 +14,13 @@ import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
-import Data.Int (Int64)
 import Data.IORef
+import Data.Int (Int64)
 import Data.List (groupBy, sortOn)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
+import Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With, encodeUtf8)
@@ -185,3 +187,7 @@ encodeJSON = safeDecodeUtf8 . LB.toStrict . J.encode
 
 decodeJSON :: FromJSON a => Text -> Maybe a
 decodeJSON = J.decode . LB.fromStrict . encodeUtf8
+
+traverseWithKey_ :: Monad m => (k -> v -> m ()) -> Map k v -> m ()
+traverseWithKey_ f = M.foldrWithKey (\k v -> (f k v >>)) (pure ())
+{-# INLINE traverseWithKey_ #-}

tests/AgentTests/FunctionalAPITests.hs

@@ -77,7 +77,7 @@ import Data.Word (Word16)
 import qualified Database.SQLite.Simple as SQL
 import GHC.Stack (withFrozenCallStack)
 import SMPAgentClient
-import SMPClient (cfg, prevRange, prevVersion, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerOn, withSmpServerProxy, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
+import SMPClient (cfg, prevRange, prevVersion, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerProxy, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
 import qualified Simplex.Messaging.Agent as A
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), ServerQueueInfo (..), UserNetworkInfo (..), UserNetworkType (..), waitForUserNetwork)
@@ -96,6 +96,7 @@ import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolS
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
 import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion, sndAuthKeySMPVersion, supportedSMPHandshakes)
 import Simplex.Messaging.Util (bshow, diffToMicroseconds)
@@ -530,7 +531,7 @@ testRatchetMatrix2 t runTest = do
 testServerMatrix2 :: HasCallStack => ATransport -> (InitialAgentServers -> IO ()) -> Spec
 testServerMatrix2 t runTest = do
   it "1 server" $ withSmpServer t $ runTest initAgentServers
-  it "2 servers" $ withSmpServer t . withSmpServerOn t testPort2 $ runTest initAgentServers2
+  it "2 servers" $ withSmpServer t $ withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> runTest initAgentServers2
 
 testPQMatrix2 :: HasCallStack => ATransport -> (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) -> Spec
 testPQMatrix2 = pqMatrix2_ True
@@ -1041,7 +1042,7 @@ testAllowConnectionClientRestart t = do
   bob <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
   withSmpServerStoreLogOn t testPort $ \_ -> do
     (aliceId, bobId, confId) <-
-      withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
+      withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
         runRight $ do
           (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
           (aliceId, sqSecured) <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
@@ -1063,8 +1064,7 @@ testAllowConnectionClientRestart t = do
     alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
     runRight_ $ subscribeConnection alice2 bobId
     threadDelay 500000
-
-    withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
+    withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
       runRight $ do
         ("", "", UP _ _) <- nGet bob
         get alice2 ##> ("", bobId, CON)
@@ -1282,7 +1282,7 @@ testSkippedMessages :: HasCallStack => ATransport -> IO ()
 testSkippedMessages t = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
-  (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \_ -> do
+  (aliceId, bobId) <- withSmpServerConfigOn t cfg' testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ makeConnection alice bob
     runRight_ $ do
       2 <- sendMessage alice bobId SMP.noMsgFlags "hello"
@@ -1310,7 +1310,7 @@ testSkippedMessages t = do
   alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   bob2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
 
-  withSmpServerStoreLogOn t testPort $ \_ -> do
+  withSmpServerConfigOn t cfg' testPort $ \_ -> do
     runRight_ $ do
       subscribeConnection bob2 aliceId
       subscribeConnection alice2 bobId
@@ -1326,6 +1326,8 @@ testSkippedMessages t = do
       ackMessage bob2 aliceId 4 Nothing
   disposeAgentClient alice2
   disposeAgentClient bob2
+  where
+    cfg' = cfg {msgStoreType = AMSType SMSMemory, storeMsgsFile = Nothing}
 
 testDeliveryAfterSubscriptionError :: HasCallStack => ATransport -> IO ()
 testDeliveryAfterSubscriptionError t = do
@@ -1429,7 +1431,7 @@ withUP a bId p =
       ]
 
 testExpireMessageQuota :: HasCallStack => ATransport -> IO ()
-testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
+testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1, maxJournalMsgCount = 2} testPort $ \_ -> do
   a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
   b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aId, bId) <- runRight $ do
@@ -1455,7 +1457,7 @@ testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testP
   disposeAgentClient a
 
 testExpireManyMessagesQuota :: ATransport -> IO ()
-testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
+testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1, maxJournalMsgCount = 2} testPort $ \_ -> do
   a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 2, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
   b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aId, bId) <- runRight $ do
@@ -1944,7 +1946,7 @@ testBatchedSubscriptions nCreate nDel t =
     runServers :: ExceptT AgentErrorType IO a -> IO a
     runServers a = do
       withSmpServerStoreLogOn t testPort $ \t1 -> do
-        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 ->
+        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \t2 ->
           runRight a `finally` killThread t2
         killThread t1
         pure res
@@ -2356,7 +2358,7 @@ testJoinConnectionAsyncReplyErrorV8 t = do
         ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId
         pure (aId, bId)
       nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
-      withSmpServerOn t testPort2 $ do
+      withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
         get b =##> \case ("2", c, JOINED sqSecured) -> c == aId && not sqSecured; _ -> False
         confId <- withSmpServerStoreLogOn t testPort $ \_ -> do
           pGet a >>= \case
@@ -2395,7 +2397,7 @@ testJoinConnectionAsyncReplyError t = do
         ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId
         pure (aId, bId)
       nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
-      withSmpServerOn t testPort2 $ do
+      withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
         confId <- withSmpServerStoreLogOn t testPort $ \_ -> do
           -- both servers need to be online for connection to progress because of SKEY
           get b =##> \case ("2", c, JOINED sqSecured) -> c == aId && sqSecured; _ -> False
@@ -2974,7 +2976,7 @@ testDeliveryReceiptsVersion t = do
 
 testDeliveryReceiptsConcurrent :: HasCallStack => ATransport -> IO ()
 testDeliveryReceiptsConcurrent t =
-  withSmpServerConfigOn t cfg {msgQueueQuota = 256} testPort $ \_ -> do
+  withSmpServerConfigOn t cfg {msgQueueQuota = 256, maxJournalMsgCount = 512} testPort $ \_ -> do
     withAgentClients2 $ \a b -> do
       (aId, bId) <- runRight $ makeConnection a b
       t1 <- liftIO getCurrentTime

tests/AgentTests/NotificationTests.hs

@@ -56,7 +56,7 @@ import Data.Text.Encoding (encodeUtf8)
 import Database.SQLite.Simple.QQ (sql)
 import NtfClient
 import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testNtfServer, testNtfServer2)
-import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
+import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), withStore')
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, Env (..), InitialAgentServers)
@@ -790,7 +790,7 @@ testNotificationsSMPRestartBatch n t apns =
     runServers :: ExceptT AgentErrorType IO a -> IO a
     runServers a = do
       withSmpServerStoreLogOn t testPort $ \t1 -> do
-        res <- withSmpServerConfigOn t (cfg :: ServerConfig) {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 ->
+        res <- withSmpServerConfigOn t (cfg :: ServerConfig) {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \t2 ->
           runRight a `finally` killThread t2
         killThread t1
         pure res

tests/CoreTests/MsgStoreTests.hs

@@ -0,0 +1,278 @@
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE PatternSynonyms #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StandaloneDeriving #-}
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+{-# OPTIONS_GHC -Wno-orphans #-}
+
+module CoreTests.MsgStoreTests where
+
+import AgentTests.FunctionalAPITests (runRight, runRight_)
+import Control.Concurrent.STM
+import Control.Exception (bracket)
+import Control.Monad
+import Control.Monad.IO.Class
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Base64.URL as B64
+import Data.Time.Clock.System (getSystemTime)
+import Simplex.Messaging.Crypto (pattern MaxLenBS)
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Protocol (EntityId (..), Message (..), noMsgFlags)
+import Simplex.Messaging.Server (MessageStats (..), exportMessages, importMessages, printMessageStats)
+import Simplex.Messaging.Server.Env.STM (journalMsgStoreDepth)
+import Simplex.Messaging.Server.MsgStore.Journal
+import Simplex.Messaging.Server.MsgStore.STM
+import Simplex.Messaging.Server.MsgStore.Types
+import SMPClient (testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
+import System.Directory (copyFile, createDirectoryIfMissing, listDirectory, removeFile, renameFile)
+import System.FilePath ((</>))
+import System.IO (IOMode (..), hClose, withFile)
+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
+  where
+    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 :: MsgStoreClass s => MsgStoreConfig s -> (s -> IO ()) -> IO ()
+withMsgStore cfg = bracket (newMsgStore cfg) closeMsgStore
+
+testSMTStoreConfig :: STMStoreConfig
+testSMTStoreConfig = STMStoreConfig {storePath = Nothing, quota = 3}
+
+testJournalStoreCfg :: JournalStoreConfig
+testJournalStoreCfg =
+  JournalStoreConfig
+    { storePath = testStoreMsgsDir,
+      pathParts = journalMsgStoreDepth,
+      quota = 3,
+      maxMsgCount = 4,
+      maxStateLines = 2
+    }
+
+mkMessage :: MonadIO m => ByteString -> m Message
+mkMessage body = liftIO $ do
+  g <- C.newRandom
+  msgTs <- getSystemTime
+  msgId <- atomically $ C.randomBytes 24 g
+  pure Message {msgId, msgTs, msgFlags = noMsgFlags, msgBody = C.unsafeMaxLenBS body}
+
+pattern Msg :: ByteString -> Maybe Message
+pattern Msg s <- Just Message {msgBody = MaxLenBS s}
+
+deriving instance Eq MsgQueueState
+
+deriving instance Eq (JournalState t)
+
+deriving instance Eq (SJournalType t)
+
+testGetQueue :: MsgStoreClass s => s -> IO ()
+testGetQueue ms = do
+  g <- C.newRandom
+  rId <- EntityId <$> atomically (C.randomBytes 24 g)
+  runRight_ $ do
+    q <- getMsgQueue ms rId
+    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 q
+    Msg "message 1" <- tryPeekMsg q
+    Nothing <- tryDelMsg q mId2
+    Msg "message 1" <- tryDelMsg q mId1
+    Nothing <- tryDelMsg q mId1
+    Msg "message 2" <- tryPeekMsg q
+    Nothing <- tryDelMsg q mId1
+    (Nothing, Msg "message 2") <- tryDelPeekMsg q mId1
+    (Msg "message 2", Msg "message 3") <- tryDelPeekMsg q mId2
+    (Nothing, Msg "message 3") <- tryDelPeekMsg q mId2
+    Msg "message 3" <- tryPeekMsg q
+    (Msg "message 3", Nothing) <- tryDelPeekMsg q mId3
+    Nothing <- tryDelMsg q mId2
+    Nothing <- tryDelMsg q mId3
+    Nothing <- tryPeekMsg q
+    Just (Message {msgId = mId4}, True) <- write "message 4"
+    Msg "message 4" <- tryPeekMsg q
+    Just (Message {msgId = mId5}, False) <- write "message 5"
+    (Nothing, Msg "message 4") <- tryDelPeekMsg q mId3
+    (Msg "message 4", Msg "message 5") <- tryDelPeekMsg 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 q
+    (Nothing, Msg "message 5") <- tryDelPeekMsg q mId4
+    (Msg "message 5", Msg "message 6") <- tryDelPeekMsg q mId5
+    (Msg "message 6", Msg "message 7") <- tryDelPeekMsg q mId6
+    (Msg "message 7", Just MessageQuota {msgId = mId8}) <- tryDelPeekMsg q mId7
+    (Just MessageQuota {}, Nothing) <- tryDelPeekMsg q mId8
+    (Nothing, Nothing) <- tryDelPeekMsg q mId8
+    pure ()
+  delMsgQueue ms rId
+
+testChangeReadJournal :: MsgStoreClass s => s -> IO ()
+testChangeReadJournal ms = do
+  g <- C.newRandom
+  rId <- EntityId <$> atomically (C.randomBytes 24 g)
+  runRight_ $ do
+    q <- getMsgQueue ms rId
+    let write s = writeMsg ms q True =<< mkMessage s
+    Just (Message {msgId = mId1}, True) <- write "message 1"
+    (Msg "message 1", Nothing) <- tryDelPeekMsg q mId1
+    Just (Message {msgId = mId2}, True) <- write "message 2"
+    (Msg "message 2", Nothing) <- tryDelPeekMsg q mId2
+    Just (Message {msgId = mId3}, True) <- write "message 3"
+    (Msg "message 3", Nothing) <- tryDelPeekMsg q mId3
+    Just (Message {msgId = mId4}, True) <- write "message 4"
+    (Msg "message 4", Nothing) <- tryDelPeekMsg q mId4
+    Just (Message {msgId = mId5}, True) <- write "message 5"
+    (Msg "message 5", Nothing) <- tryDelPeekMsg q mId5
+    pure ()
+  delMsgQueue ms rId
+
+testExportImportStore :: JournalMsgStore -> IO ()
+testExportImportStore ms = do
+  g <- C.newRandom
+  rId1 <- EntityId <$> atomically (C.randomBytes 24 g)
+  rId2 <- EntityId <$> atomically (C.randomBytes 24 g)
+  runRight_ $ do
+    let write q s = writeMsg ms q True =<< mkMessage s
+    q1 <- getMsgQueue ms rId1
+    Just (Message {}, True) <- write q1 "message 1"
+    Just (Message {}, False) <- write q1 "message 2"
+    q2 <- getMsgQueue ms rId2
+    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 q2 mId3
+    (Msg "message 4", Nothing) <- tryDelPeekMsg 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
+  exportMessages False ms testStoreMsgsFile False
+  renameFile testStoreMsgsFile (testStoreMsgsFile <> ".copy")
+  closeMsgStore ms
+  exportMessages False ms testStoreMsgsFile False
+  (B.readFile testStoreMsgsFile `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".copy")
+  let cfg = (testJournalStoreCfg :: JournalStoreConfig) {storePath = testStoreMsgsDir2}
+  ms' <- newMsgStore cfg
+  stats@MessageStats {storedMsgsCount = 5, expiredMsgsCount = 0, storedQueues = 2} <-
+    importMessages False ms' testStoreMsgsFile Nothing
+  printMessageStats "Messages" stats
+  length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
+  length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 4 -- state file is backed up, 2 message files
+  exportMessages False ms' testStoreMsgsFile2 False
+  (B.readFile testStoreMsgsFile2 `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".bak")
+  stmStore <- newMsgStore testSMTStoreConfig
+  MessageStats {storedMsgsCount = 5, expiredMsgsCount = 0, storedQueues = 2} <-
+    importMessages False stmStore testStoreMsgsFile2 Nothing
+  exportMessages False stmStore testStoreMsgsFile False
+  (B.sort <$> B.readFile testStoreMsgsFile `shouldReturn`) =<< (B.sort <$> B.readFile (testStoreMsgsFile2 <> ".bak"))
+
+testQueueState :: JournalMsgStore -> 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)
+  createDirectoryIfMissing True dir
+  state <- newMsgQueueState <$> newJournalId (random ms)
+  withFile statePath WriteMode (`appendState` state)
+  length . lines <$> readFile statePath `shouldReturn` 1
+  readQueueState statePath `shouldReturn` state
+  length <$> listDirectory dir `shouldReturn` 1 -- no backup
+
+  let state1 =
+        state
+          { size = 1,
+            readState = (readState state) {msgCount = 1, byteCount = 100},
+            writeState = (writeState state) {msgPos = 1, msgCount = 1, bytePos = 100, byteCount = 100}
+          }
+  withFile statePath AppendMode (`appendState` state1)
+  length . lines <$> readFile statePath `shouldReturn` 2
+  readQueueState statePath `shouldReturn` state1
+  length <$> listDirectory dir `shouldReturn` 1 -- no backup
+
+  let state2 =
+        state
+          { size = 2,
+            readState = (readState state) {msgCount = 2, byteCount = 200},
+            writeState = (writeState state) {msgPos = 2, msgCount = 2, bytePos = 200, byteCount = 200}
+          }
+  withFile statePath AppendMode (`appendState` state2)
+  length . lines <$> readFile statePath `shouldReturn` 3
+  copyFile statePath (statePath <> ".2")
+  readQueueState statePath `shouldReturn` state2
+  length <$> listDirectory dir `shouldReturn` 3 -- new state, copy + backup
+  length . lines <$> readFile statePath `shouldReturn` 1
+
+  -- corrupt the only line
+  corruptFile statePath
+  newState <- readQueueState statePath
+  newState `shouldBe` newMsgQueueState (journalId $ writeState newState)
+
+  -- corrupt the last line
+  renameFile (statePath <> ".2") statePath
+  removeOtherFiles dir statePath
+  length . lines <$> readFile statePath `shouldReturn` 3
+  corruptFile statePath
+  readQueueState statePath `shouldReturn` state1
+  length <$> listDirectory dir `shouldReturn` 2
+  length . lines <$> readFile statePath `shouldReturn` 1
+  where
+    readQueueState statePath = do
+      (state, h) <- readWriteQueueState ms statePath
+      hClose h
+      pure state
+    corruptFile f = do
+      s <- readFile f
+      removeFile f
+      writeFile f $ take (length s - 4) s
+    removeOtherFiles dir keep = do
+      names <- listDirectory dir
+      forM_ names $ \name ->
+        let f = dir </> name
+         in unless (f == keep) $ removeFile f
+
+testMessageState :: JournalMsgStore -> IO ()
+testMessageState 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)
+      write q s = writeMsg ms q True =<< mkMessage s
+
+  mId1 <- runRight $ do
+    q <- getMsgQueue ms rId
+    Just (Message {msgId = mId1}, True) <- write q "message 1"
+    Just (Message {}, False) <- write q "message 2"
+    liftIO $ closeMsgQueue ms rId
+    pure mId1
+
+  ls <- B.lines <$> B.readFile statePath
+  B.writeFile statePath $ B.unlines $ take (length ls - 1) ls
+
+  runRight_ $ do
+    q <- getMsgQueue ms rId
+    Just (Message {msgId = mId3}, False) <- write q "message 3"
+    (Msg "message 1", Msg "message 3") <- tryDelPeekMsg q mId1
+    (Msg "message 3", Nothing) <- tryDelPeekMsg q mId3
+    liftIO $ closeMsgQueueHandles q

tests/NtfServerTests.hs

@@ -104,7 +104,7 @@ testNotificationSubscription (ATransport t) =
     (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
     let tkn = DeviceToken PPApnsTest "abcd"
     withAPNSMockServer $ \apns ->
-      smpTest2 t $ \rh sh ->
+      smpTest2' t $ \rh sh ->
         ntfTest t $ \nh -> do
           -- create queue
           (sId, rId, rKey, rcvDhSecret) <- createAndSecureQueue rh sPub

tests/SMPClient.hs

@@ -1,6 +1,7 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
@@ -23,6 +24,7 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server (runSMPServerBlocking)
 import Simplex.Messaging.Server.Env.STM
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
 import qualified Simplex.Messaging.Transport.Client as Client
@@ -62,6 +64,12 @@ testStoreMsgsFile = "tests/tmp/smp-server-messages.log"
 testStoreMsgsFile2 :: FilePath
 testStoreMsgsFile2 = "tests/tmp/smp-server-messages.log.2"
 
+testStoreMsgsDir :: FilePath
+testStoreMsgsDir = "tests/tmp/messages"
+
+testStoreMsgsDir2 :: FilePath
+testStoreMsgsDir2 = "tests/tmp/messages.2"
+
 testStoreNtfsFile :: FilePath
 testStoreNtfsFile = "tests/tmp/smp-server-ntfs.log"
 
@@ -100,16 +108,24 @@ testSMPClient_ host port vr client = do
       | otherwise = Nothing
 
 cfg :: ServerConfig
-cfg =
+cfg = cfgMS (AMSType SMSJournal)
+
+cfgMS :: AMSType -> ServerConfig
+cfgMS msType =
   ServerConfig
     { transports = [],
       smpHandshakeTimeout = 60000000,
       tbqSize = 1,
+      msgStoreType = msType,
       msgQueueQuota = 4,
+      maxJournalMsgCount = 5,
+      maxJournalStateLines = 2,
       queueIdBytes = 24,
       msgIdBytes = 24,
-      storeLogFile = Nothing,
-      storeMsgsFile = Nothing,
+      storeLogFile = Just testStoreLogFile,
+      storeMsgsFile = Just $ case msType of
+        AMSType SMSJournal -> testStoreMsgsDir
+        AMSType SMSMemory -> testStoreMsgsFile,
       storeNtfsFile = Nothing,
       allowNewQueues = True,
       newQueueBasicAuth = Nothing,
@@ -168,18 +184,17 @@ proxyVRangeV8 :: VersionRangeSMP
 proxyVRangeV8 = mkVersionRange batchCmdsSMPVersion sendingProxySMPVersion
 
 withSmpServerStoreMsgLogOn :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
-withSmpServerStoreMsgLogOn t =
-  withSmpServerConfigOn
-    t
-    cfg
-      { storeLogFile = Just testStoreLogFile,
-        storeMsgsFile = Just testStoreMsgsFile,
-        storeNtfsFile = Just testStoreNtfsFile,
-        serverStatsBackupFile = Just testServerStatsBackupFile
-      }
+withSmpServerStoreMsgLogOn = (`withSmpServerStoreMsgLogOnMS` AMSType SMSJournal)
+
+withSmpServerStoreMsgLogOnMS :: HasCallStack => ATransport -> AMSType -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
+withSmpServerStoreMsgLogOnMS t msType =
+  withSmpServerConfigOn t (cfgMS msType) {storeNtfsFile = Just testStoreNtfsFile, serverStatsBackupFile = Just testServerStatsBackupFile}
 
 withSmpServerStoreLogOn :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
-withSmpServerStoreLogOn t = withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile, serverStatsBackupFile = Just testServerStatsBackupFile}
+withSmpServerStoreLogOn = (`withSmpServerStoreLogOnMS` AMSType SMSJournal)
+
+withSmpServerStoreLogOnMS :: HasCallStack => ATransport -> AMSType -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
+withSmpServerStoreLogOnMS t msType = withSmpServerConfigOn t (cfgMS msType) {serverStatsBackupFile = Just testServerStatsBackupFile}
 
 withSmpServerConfigOn :: HasCallStack => ATransport -> ServerConfig -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withSmpServerConfigOn t cfg' port' =
@@ -212,11 +227,11 @@ withSmpServer t = withSmpServerOn t testPort
 withSmpServerProxy :: HasCallStack => ATransport -> IO a -> IO a
 withSmpServerProxy t = withSmpServerConfigOn t proxyCfg testPort . const
 
-runSmpTest :: forall c a. (HasCallStack, Transport c) => (HasCallStack => THandleSMP c 'TClient -> IO a) -> IO a
-runSmpTest test = withSmpServer (transport @c) $ testSMPClient test
+runSmpTest :: forall c a. (HasCallStack, Transport c) => AMSType -> (HasCallStack => THandleSMP c 'TClient -> IO a) -> IO a
+runSmpTest msType test = withSmpServerConfigOn (transport @c) (cfgMS msType) testPort $ \_ -> testSMPClient test
 
-runSmpTestN :: forall c a. (HasCallStack, Transport c) => Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO a) -> IO a
-runSmpTestN = runSmpTestNCfg cfg supportedClientSMPRelayVRange
+runSmpTestN :: forall c a. (HasCallStack, Transport c) => AMSType -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO a) -> IO a
+runSmpTestN msType = runSmpTestNCfg (cfgMS msType) supportedClientSMPRelayVRange
 
 runSmpTestNCfg :: forall c a. (HasCallStack, Transport c) => ServerConfig -> VersionRangeSMP -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO a) -> IO a
 runSmpTestNCfg srvCfg clntVR nClients test = withSmpServerConfigOn (transport @c) srvCfg testPort $ \_ -> run nClients []
@@ -231,7 +246,7 @@ smpServerTest ::
   TProxy c ->
   (Maybe TransmissionAuth, ByteString, ByteString, smp) ->
   IO (Maybe TransmissionAuth, ByteString, ByteString, BrokerMsg)
-smpServerTest _ t = runSmpTest $ \h -> tPut' h t >> tGet' h
+smpServerTest _ t = runSmpTest (AMSType SMSJournal) $ \h -> tPut' h t >> tGet' h
   where
     tPut' :: THandleSMP c 'TClient -> (Maybe TransmissionAuth, ByteString, ByteString, smp) -> IO ()
     tPut' h@THandle {params = THandleParams {sessionId, implySessId}} (sig, corrId, queueId, smp) = do
@@ -242,14 +257,17 @@ smpServerTest _ t = runSmpTest $ \h -> tPut' h t >> tGet' h
       [(Nothing, _, (CorrId corrId, EntityId qId, Right cmd))] <- tGet h
       pure (Nothing, corrId, qId, cmd)
 
-smpTest :: (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> IO ()) -> Expectation
-smpTest _ test' = runSmpTest test' `shouldReturn` ()
+smpTest :: (HasCallStack, Transport c) => TProxy c -> AMSType -> (HasCallStack => THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest _ msType test' = runSmpTest msType test' `shouldReturn` ()
 
-smpTestN :: (HasCallStack, Transport c) => Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO ()) -> Expectation
-smpTestN n test' = runSmpTestN n test' `shouldReturn` ()
+smpTestN :: (HasCallStack, Transport c) => AMSType -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO ()) -> Expectation
+smpTestN msType n test' = runSmpTestN msType n test' `shouldReturn` ()
 
-smpTest2 :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
-smpTest2 = smpTest2Cfg cfg supportedClientSMPRelayVRange
+smpTest2' :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest2' = (`smpTest2` AMSType SMSJournal)
+
+smpTest2 :: forall c. (HasCallStack, Transport c) => TProxy c -> AMSType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest2 t msType = smpTest2Cfg (cfgMS msType) supportedClientSMPRelayVRange t
 
 smpTest2Cfg :: forall c. (HasCallStack, Transport c) => ServerConfig -> VersionRangeSMP -> TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
 smpTest2Cfg srvCfg clntVR _ test' = runSmpTestNCfg srvCfg clntVR 2 _test `shouldReturn` ()
@@ -258,15 +276,15 @@ smpTest2Cfg srvCfg clntVR _ test' = runSmpTestNCfg srvCfg clntVR 2 _test `should
     _test [h1, h2] = test' h1 h2
     _test _ = error "expected 2 handles"
 
-smpTest3 :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
-smpTest3 _ test' = smpTestN 3 _test
+smpTest3 :: forall c. (HasCallStack, Transport c) => TProxy c -> AMSType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest3 _ msType test' = smpTestN msType 3 _test
   where
     _test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()
     _test [h1, h2, h3] = test' h1 h2 h3
     _test _ = error "expected 3 handles"
 
-smpTest4 :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
-smpTest4 _ test' = smpTestN 4 _test
+smpTest4 :: forall c. (HasCallStack, Transport c) => TProxy c -> AMSType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest4 _ msType test' = smpTestN msType 4 _test
   where
     _test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()
     _test [h1, h2, h3, h4] = test' h1 h2 h3 h4

tests/SMPProxyTests.hs

@@ -135,14 +135,15 @@ smpProxyTests = do
         let deliver nAgents nMsgs = agentDeliverMessagesViaProxyConc (replicate nAgents [srv1]) (map bshow [1 :: Int .. nMsgs])
         it "25 agents, 300 pairs, 17 messages" . oneServer . withNumCapabilities 4 $ deliver 25 17
   where
-    oneServer = withSmpServerConfigOn (transport @TLS) proxyCfg {msgQueueQuota = 128} testPort . const
+    oneServer = withSmpServerConfigOn (transport @TLS) proxyCfg {msgQueueQuota = 128, maxJournalMsgCount = 256} testPort . const
     twoServers = twoServers_ proxyCfg proxyCfg
-    twoServersFirstProxy = twoServers_ proxyCfg cfgV8 {msgQueueQuota = 128}
-    twoServersMoreConc = twoServers_ proxyCfg {serverClientConcurrency = 128} cfgV8 {msgQueueQuota = 128}
-    twoServersNoConc = twoServers_ proxyCfg {serverClientConcurrency = 1} cfgV8 {msgQueueQuota = 128}
+    twoServersFirstProxy = twoServers_ proxyCfg cfgV8 {msgQueueQuota = 128, maxJournalMsgCount = 256}
+    twoServersMoreConc = twoServers_ proxyCfg {serverClientConcurrency = 128} cfgV8 {msgQueueQuota = 128, maxJournalMsgCount = 256}
+    twoServersNoConc = twoServers_ proxyCfg {serverClientConcurrency = 1} cfgV8 {msgQueueQuota = 128, maxJournalMsgCount = 256}
     twoServers_ cfg1 cfg2 runTest =
       withSmpServerConfigOn (transport @TLS) cfg1 testPort $ \_ ->
-        withSmpServerConfigOn (transport @TLS) cfg2 testPort2 $ const runTest
+        let cfg2' = cfg2 {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2}
+         in withSmpServerConfigOn (transport @TLS) cfg2' testPort2 $ const runTest
 
 deliverMessageViaProxy :: (C.AlgorithmI a, C.AuthAlgorithm a) => SMPServer -> SMPServer -> C.SAlgorithm a -> ByteString -> ByteString -> IO ()
 deliverMessageViaProxy proxyServ relayServ alg msg msg' = deliverMessagesViaProxy proxyServ relayServ alg [msg] [msg']
@@ -379,11 +380,11 @@ agentViaProxyRetryOffline = do
             ackMessage alice bobId (baseId + 4) Nothing
   where
     withServer :: (ThreadId -> IO a) -> IO a
-    withServer = withServer_ testStoreLogFile testStoreMsgsFile testStoreNtfsFile testPort
+    withServer = withServer_ testStoreLogFile testStoreMsgsDir testStoreNtfsFile testPort
     withServer2 :: (ThreadId -> IO a) -> IO a
-    withServer2 = withServer_ testStoreLogFile2 testStoreMsgsFile2 testStoreNtfsFile2 testPort2
-    withServer_ storeLog storeMsgs storeNtfs port =
-      withSmpServerConfigOn (transport @TLS) proxyCfg {storeLogFile = Just storeLog, storeMsgsFile = Just storeMsgs, storeNtfsFile = Just storeNtfs} port
+    withServer2 = withServer_ testStoreLogFile2 testStoreMsgsDir2 testStoreNtfsFile2 testPort2
+    withServer_ storeLog storeMsgs storeNtfs =
+      withSmpServerConfigOn (transport @TLS) proxyCfg {storeLogFile = Just storeLog, storeMsgsFile = Just storeMsgs, storeNtfsFile = Just storeNtfs}
     a `up` cId = nGet a =##> \case ("", "", UP _ [c]) -> c == cId; _ -> False
     a `down` cId = nGet a =##> \case ("", "", DOWN _ [c]) -> c == cId; _ -> False
     aCfg = agentCfg {messageRetryInterval = fastMessageRetryInterval}

tests/ServerTests.hs

@@ -16,6 +16,7 @@
 module ServerTests where
 
 import AgentTests.NotificationTests (removeFileIfExists)
+import CoreTests.MsgStoreTests (testJournalStoreCfg)
 import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (SomeException, try)
@@ -35,43 +36,46 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server (exportMessages)
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
+import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..))
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore)
 import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..))
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Util (whenM)
 import Simplex.Messaging.Version (mkVersionRange)
-import System.Directory (removeFile)
+import System.Directory (doesDirectoryExist, removeDirectoryRecursive, removeFile)
 import System.TimeIt (timeItT)
 import System.Timeout
 import Test.HUnit
 import Test.Hspec
 
-serverTests :: ATransport -> Spec
-serverTests t@(ATransport t') = do
-  describe "SMP syntax" $ syntaxTests t
+serverTests :: SpecWith (ATransport, AMSType)
+serverTests  = do
   describe "SMP queues" $ do
-    describe "NEW and KEY commands, SEND messages" $ testCreateSecure t
+    describe "NEW and KEY commands, SEND messages" testCreateSecure
     describe "NEW and SKEY commands" $ do
-      testCreateSndSecure t
-      testSndSecureProhibited t
-    describe "NEW, OFF and DEL commands, SEND messages" $ testCreateDelete t
-    describe "Stress test" $ stressTest t
-    describe "allowNewQueues setting" $ testAllowNewQueues t'
+      testCreateSndSecure
+      testSndSecureProhibited
+    describe "NEW, OFF and DEL commands, SEND messages" testCreateDelete
+    describe "Stress test" stressTest
+    describe "allowNewQueues setting" testAllowNewQueues
   describe "SMP messages" $ do
-    describe "duplex communication over 2 SMP connections" $ testDuplex t
-    describe "switch subscription to another TCP connection" $ testSwitchSub t
-    describe "GET command" $ testGetCommand t'
-    describe "GET & SUB commands" $ testGetSubCommands t'
-    describe "Exceeding queue quota" $ testExceedQueueQuota t'
-  describe "Store log" $ testWithStoreLog t
-  describe "Restore messages" $ testRestoreMessages t
-  describe "Restore messages (old / v2)" $ testRestoreExpireMessages t
-  describe "Timing of AUTH error" $ testTiming t
-  describe "Message notifications" $ testMessageNotifications t
+    describe "duplex communication over 2 SMP connections" testDuplex
+    describe "switch subscription to another TCP connection" testSwitchSub
+    describe "GET command" testGetCommand
+    describe "GET & SUB commands" testGetSubCommands
+    describe "Exceeding queue quota" testExceedQueueQuota
+  describe "Store log" testWithStoreLog
+  describe "Restore messages" testRestoreMessages
+  describe "Restore messages (old / v2)" testRestoreExpireMessages
+  describe "Timing of AUTH error" testTiming
+  describe "Message notifications" testMessageNotifications
   describe "Message expiration" $ do
-    testMsgExpireOnSend t'
-    testMsgExpireOnInterval t'
-    testMsgNOTExpireOnInterval t'
+    testMsgExpireOnSend
+    testMsgExpireOnInterval
+    testMsgNOTExpireOnInterval
 
 pattern Resp :: CorrId -> QueueId -> BrokerMsg -> SignedTransmission ErrorType BrokerMsg
 pattern Resp corrId queueId command <- (_, _, (corrId, queueId, Right command))
@@ -131,10 +135,10 @@ decryptMsgV3 dhShared nonce body =
     Right ClientRcvMsgQuota {} -> Left "ClientRcvMsgQuota"
     Left e -> Left e
 
-testCreateSecure :: ATransport -> Spec
-testCreateSecure (ATransport t) =
-  it "should create (NEW) and secure (KEY) queue" $
-    smpTest2 t $ \r s -> do
+testCreateSecure :: SpecWith (ATransport, AMSType)
+testCreateSecure =
+  it "should create (NEW) and secure (KEY) queue" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \r s -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -196,10 +200,10 @@ testCreateSecure (ATransport t) =
       Resp "bcda" _ (ERR LARGE_MSG) <- signSendRecv s sKey ("bcda", sId, _SEND biggerMessage)
       pure ()
 
-testCreateSndSecure :: ATransport -> Spec
-testCreateSndSecure (ATransport t) =
-  it "should create (NEW) and secure (SKEY) queue by sender" $
-    smpTest2 t $ \r s -> do
+testCreateSndSecure :: SpecWith (ATransport, AMSType)
+testCreateSndSecure =
+  it "should create (NEW) and secure (SKEY) queue by sender" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \r s -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -243,10 +247,10 @@ testCreateSndSecure (ATransport t) =
       Resp "bcda" _ (ERR LARGE_MSG) <- signSendRecv s sKey ("bcda", sId, _SEND biggerMessage)
       pure ()
 
-testSndSecureProhibited :: ATransport -> Spec
-testSndSecureProhibited (ATransport t) =
-  it "should create (NEW) without allowing sndSecure and fail to and secure queue by sender (SKEY)" $
-    smpTest2 t $ \r s -> do
+testSndSecureProhibited :: SpecWith (ATransport, AMSType)
+testSndSecureProhibited =
+  it "should create (NEW) without allowing sndSecure and fail to and secure queue by sender (SKEY)" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \r s -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -258,10 +262,10 @@ testSndSecureProhibited (ATransport t) =
       (sId2, sId) #== "secures queue, same queue ID in response"
       (err, ERR AUTH) #== "rejects SKEY when not allowed in NEW command"
 
-testCreateDelete :: ATransport -> Spec
-testCreateDelete (ATransport t) =
-  it "should create (NEW), suspend (OFF) and delete (DEL) queue" $
-    smpTest2 t $ \rh sh -> do
+testCreateDelete :: SpecWith (ATransport, AMSType)
+testCreateDelete =
+  it "should create (NEW), suspend (OFF) and delete (DEL) queue" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \rh sh -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -329,10 +333,10 @@ testCreateDelete (ATransport t) =
       Resp "cdab" _ err10 <- signSendRecv rh rKey ("cdab", rId, SUB)
       (err10, ERR AUTH) #== "rejects SUB when deleted"
 
-stressTest :: ATransport -> Spec
-stressTest (ATransport t) =
-  it "should create many queues, disconnect and re-connect" $
-    smpTest3 t $ \h1 h2 h3 -> do
+stressTest :: SpecWith (ATransport, AMSType)
+stressTest =
+  it "should create many queues, disconnect and re-connect" $ \(ATransport t, msType) ->
+    smpTest3 t msType $ \h1 h2 h3 -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
       (dhPub, _ :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -347,10 +351,10 @@ stressTest (ATransport t) =
       closeConnection $ connection h2
       subscribeQueues h3
 
-testAllowNewQueues :: forall c. Transport c => TProxy c -> Spec
-testAllowNewQueues t =
-  it "should prohibit creating new queues with allowNewQueues = False" $ do
-    withSmpServerConfigOn (ATransport t) cfg {allowNewQueues = False} testPort $ \_ ->
+testAllowNewQueues :: SpecWith (ATransport, AMSType)
+testAllowNewQueues =
+  it "should prohibit creating new queues with allowNewQueues = False" $ \(ATransport (t :: TProxy c), msType) ->
+    withSmpServerConfigOn (ATransport t) (cfgMS msType) {allowNewQueues = False} testPort $ \_ ->
       testSMPClient @c $ \h -> do
         g <- C.newRandom
         (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
@@ -358,10 +362,10 @@ testAllowNewQueues t =
         Resp "abcd" NoEntity (ERR AUTH) <- signSendRecv h rKey ("abcd", NoEntity, NEW rPub dhPub Nothing SMSubscribe False)
         pure ()
 
-testDuplex :: ATransport -> Spec
-testDuplex (ATransport t) =
-  it "should create 2 simplex connections and exchange messages" $
-    smpTest2 t $ \alice bob -> do
+testDuplex :: SpecWith (ATransport, AMSType)
+testDuplex =
+  it "should create 2 simplex connections and exchange messages" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \alice bob -> do
       g <- C.newRandom
       (arPub, arKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (aDhPub, aDhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -413,10 +417,10 @@ testDuplex (ATransport t) =
       Resp "bcda" _ OK <- signSendRecv bob brKey ("bcda", bRcv, ACK mId5)
       (bDec mId5 msg5, Right "how are you bob") #== "message received from alice"
 
-testSwitchSub :: ATransport -> Spec
-testSwitchSub (ATransport t) =
-  it "should create simplex connections and switch subscription to another TCP connection" $
-    smpTest3 t $ \rh1 rh2 sh -> do
+testSwitchSub :: SpecWith (ATransport, AMSType)
+testSwitchSub =
+  it "should create simplex connections and switch subscription to another TCP connection" $ \(ATransport t, msType) ->
+    smpTest3 t msType $ \rh1 rh2 sh -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -458,12 +462,12 @@ testSwitchSub (ATransport t) =
         Nothing -> return ()
         Just _ -> error "nothing else is delivered to the 1st TCP connection"
 
-testGetCommand :: forall c. Transport c => TProxy c -> Spec
-testGetCommand t =
-  it "should retrieve messages from the queue using GET command" $ do
+testGetCommand :: SpecWith (ATransport, AMSType)
+testGetCommand =
+  it "should retrieve messages from the queue using GET command" $ \(ATransport (t :: TProxy c), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    smpTest t $ \sh -> do
+    smpTest t msType $ \sh -> do
       queue <- newEmptyTMVarIO
       testSMPClient @c $ \rh ->
         atomically . putTMVar queue =<< createAndSecureQueue rh sPub
@@ -477,12 +481,12 @@ testGetCommand t =
         Resp "4" _ OK <- signSendRecv rh rKey ("4", rId, GET)
         pure ()
 
-testGetSubCommands :: forall c. Transport c => TProxy c -> Spec
-testGetSubCommands t =
-  it "should retrieve messages with GET and receive with SUB, only one ACK would work" $ do
+testGetSubCommands :: SpecWith (ATransport, AMSType)
+testGetSubCommands =
+  it "should retrieve messages with GET and receive with SUB, only one ACK would work" $ \(ATransport t, msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    smpTest3 t $ \rh1 rh2 sh -> do
+    smpTest3 t msType $ \rh1 rh2 sh -> do
       (sId, rId, rKey, dhShared) <- createAndSecureQueue rh1 sPub
       let dec = decryptMsgV3 dhShared
       Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, _SEND "hello 1")
@@ -527,10 +531,10 @@ testGetSubCommands t =
       Resp "12" _ OK <- signSendRecv rh2 rKey ("12", rId, GET)
       pure ()
 
-testExceedQueueQuota :: forall c. Transport c => TProxy c -> Spec
-testExceedQueueQuota t =
-  it "should reply with ERR QUOTA to sender and send QUOTA message to the recipient" $ do
-    withSmpServerConfigOn (ATransport t) cfg {msgQueueQuota = 2} testPort $ \_ ->
+testExceedQueueQuota :: SpecWith (ATransport, AMSType)
+testExceedQueueQuota =
+  it "should reply with ERR QUOTA to sender and send QUOTA message to the recipient" $ \(ATransport (t :: TProxy c), msType) -> do
+    withSmpServerConfigOn (ATransport t) (cfgMS msType) {msgQueueQuota = 2} testPort $ \_ ->
       testSMPClient @c $ \sh -> testSMPClient @c $ \rh -> do
         g <- C.newRandom
         (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
@@ -554,9 +558,9 @@ testExceedQueueQuota t =
         Resp "10" _ OK <- signSendRecv rh rKey ("10", rId, ACK mId4)
         pure ()
 
-testWithStoreLog :: ATransport -> Spec
-testWithStoreLog at@(ATransport t) =
-  it "should store simplex queues to log and restore them after server restart" $ do
+testWithStoreLog :: SpecWith (ATransport, AMSType)
+testWithStoreLog =
+  it "should store simplex queues to log and restore them after server restart" $ \(at@(ATransport t), msType) -> do
     g <- C.newRandom
     (sPub1, sKey1) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     (sPub2, sKey2) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
@@ -568,7 +572,7 @@ testWithStoreLog at@(ATransport t) =
     senderId2 <- newTVarIO NoEntity
     notifierId <- newTVarIO NoEntity
 
-    withSmpServerStoreLogOn at testPort . runTest t $ \h -> runClient t $ \h1 -> do
+    withSmpServerStoreLogOnMS at msType testPort . runTest t $ \h -> runClient t $ \h1 -> do
       (sId1, rId1, rKey1, dhShared) <- createAndSecureQueue h sPub1
       (rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
       Resp "abcd" _ (NID nId _) <- signSendRecv h rKey1 ("abcd", rId1, NKEY nPub rcvNtfPubDhKey)
@@ -601,13 +605,14 @@ testWithStoreLog at@(ATransport t) =
 
     logSize testStoreLogFile `shouldReturn` 6
 
-    withSmpServerThreadOn at testPort . runTest t $ \h -> do
+    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 ()
 
-    withSmpServerStoreLogOn at testPort . runTest t $ \h -> runClient t $ \h1 -> do
+    withSmpServerStoreLogOnMS at msType testPort . runTest t $ \h -> runClient t $ \h1 -> do
       -- this queue is restored
       rId1 <- readTVarIO recipientId1
       Just rKey1 <- readTVarIO recipientKey1
@@ -641,11 +646,12 @@ logSize f =
     Right l -> pure l
     Left (_ :: SomeException) -> logSize f
 
-testRestoreMessages :: ATransport -> Spec
-testRestoreMessages at@(ATransport t) =
-  it "should store messages on exit and restore on start" $ do
+testRestoreMessages :: SpecWith (ATransport, AMSType)
+testRestoreMessages =
+  it "should store messages on exit and restore on start" $ \(at@(ATransport t), msType) -> do
     removeFileIfExists testStoreLogFile
     removeFileIfExists testStoreMsgsFile
+    whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
     removeFileIfExists testServerStatsBackupFile
 
     g <- C.newRandom
@@ -655,7 +661,7 @@ testRestoreMessages at@(ATransport t) =
     dhShared <- newTVarIO Nothing
     senderId <- newTVarIO NoEntity
 
-    withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
+    withSmpServerStoreMsgLogOnMS at msType testPort . runTest t $ \h -> do
       runClient t $ \h1 -> do
         (sId, rId, rKey, dh) <- createAndSecureQueue h1 sPub
         atomically $ do
@@ -679,12 +685,12 @@ testRestoreMessages at@(ATransport t) =
     rId <- readTVarIO recipientId
 
     logSize testStoreLogFile `shouldReturn` 2
-    logSize testStoreMsgsFile `shouldReturn` 5
+    -- logSize testStoreMsgsFile `shouldReturn` 5
     logSize testServerStatsBackupFile `shouldReturn` 74
     Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats1 [rId] 5 1
 
-    withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
+    withSmpServerStoreMsgLogOnMS at msType testPort . runTest t $ \h -> do
       Just rKey <- readTVarIO recipientKey
       Just dh <- readTVarIO dhShared
       let dec = decryptMsgV3 dh
@@ -697,12 +703,12 @@ testRestoreMessages at@(ATransport t) =
 
     logSize testStoreLogFile `shouldReturn` 1
     -- the last message is not removed because it was not ACK'd
-    logSize testStoreMsgsFile `shouldReturn` 3
+    -- logSize testStoreMsgsFile `shouldReturn` 3
     logSize testServerStatsBackupFile `shouldReturn` 74
     Right stats2 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats2 [rId] 5 3
 
-    withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
+    withSmpServerStoreMsgLogOnMS at msType testPort . runTest t $ \h -> do
       Just rKey <- readTVarIO recipientKey
       Just dh <- readTVarIO dhShared
       let dec = decryptMsgV3 dh
@@ -714,15 +720,15 @@ testRestoreMessages at@(ATransport t) =
       (dec mId6 msg6, Left "ClientRcvMsgQuota") #== "restored message delivered"
       Resp "7" _ OK <- signSendRecv h rKey ("7", rId, ACK mId6)
       pure ()
-
     logSize testStoreLogFile `shouldReturn` 1
-    logSize testStoreMsgsFile `shouldReturn` 0
+    -- logSize testStoreMsgsFile `shouldReturn` 0
     logSize testServerStatsBackupFile `shouldReturn` 74
     Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats3 [rId] 5 5
 
     removeFile testStoreLogFile
-    removeFile testStoreMsgsFile
+    removeFileIfExists testStoreMsgsFile
+    whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
     removeFile testServerStatsBackupFile
   where
     runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
@@ -749,9 +755,9 @@ checkStats s qs sent received = do
   IS.toList _week `shouldBe` map (hash . unEntityId) qs
   IS.toList _month `shouldBe` map (hash . unEntityId) qs
 
-testRestoreExpireMessages :: ATransport -> Spec
-testRestoreExpireMessages at@(ATransport t) =
-  it "should store messages on exit and restore on start" $ do
+testRestoreExpireMessages :: SpecWith (ATransport, AMSType)
+testRestoreExpireMessages =
+  it "should store messages on exit and restore on start" $ \(at@(ATransport t), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     recipientId <- newTVarIO NoEntity
@@ -759,7 +765,7 @@ testRestoreExpireMessages at@(ATransport t) =
     dhShared <- newTVarIO Nothing
     senderId <- newTVarIO NoEntity
 
-    withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
+    withSmpServerStoreMsgLogOnMS at msType testPort . runTest t $ \h -> do
       runClient t $ \h1 -> do
         (sId, rId, rKey, dh) <- createAndSecureQueue h1 sPub
         atomically $ do
@@ -776,29 +782,39 @@ testRestoreExpireMessages at@(ATransport t) =
       pure ()
 
     logSize testStoreLogFile `shouldReturn` 2
+    exportStoreMessages msType
     msgs <- B.readFile testStoreMsgsFile
     length (B.lines msgs) `shouldBe` 4
 
     let expCfg1 = Just ExpirationConfig {ttl = 86400, checkInterval = 43200}
-        cfg1 = cfg {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg1, serverStatsBackupFile = Just testServerStatsBackupFile}
+        cfg1 = (cfgMS msType) {messageExpiration = expCfg1, serverStatsBackupFile = Just testServerStatsBackupFile}
     withSmpServerConfigOn at cfg1 testPort . runTest t $ \_ -> pure ()
 
     logSize testStoreLogFile `shouldReturn` 1
+    exportStoreMessages msType
     msgs' <- B.readFile testStoreMsgsFile
     msgs' `shouldBe` msgs
 
     let expCfg2 = Just ExpirationConfig {ttl = 2, checkInterval = 43200}
-        cfg2 = cfg {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg2, serverStatsBackupFile = Just testServerStatsBackupFile}
+        cfg2 = (cfgMS msType) {messageExpiration = expCfg2, serverStatsBackupFile = Just testServerStatsBackupFile}
     withSmpServerConfigOn at cfg2 testPort . runTest t $ \_ -> pure ()
 
     logSize testStoreLogFile `shouldReturn` 1
     -- two messages expired
+    exportStoreMessages msType
     msgs'' <- B.readFile testStoreMsgsFile
     length (B.lines msgs'') `shouldBe` 2
     B.lines msgs'' `shouldBe` drop 2 (B.lines msgs)
     Right ServerStatsData {_msgExpired} <- strDecode <$> B.readFile testServerStatsBackupFile
     _msgExpired `shouldBe` 2
   where
+    exportStoreMessages :: AMSType -> IO ()
+    exportStoreMessages = \case
+      AMSType SMSJournal -> do
+        ms <- newMsgStore testJournalStoreCfg {quota = 4}
+        removeFileIfExists testStoreMsgsFile
+        exportMessages False ms testStoreMsgsFile False
+      AMSType SMSMemory -> pure ()
     runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
     runTest _ test' server = do
       testSMPClient test' `shouldReturn` ()
@@ -818,12 +834,12 @@ createAndSecureQueue h sPub = do
   (rId', rId) #== "same queue ID"
   pure (sId, rId, rKey, dhShared)
 
-testTiming :: ATransport -> Spec
-testTiming (ATransport t) =
+testTiming :: SpecWith (ATransport, AMSType)
+testTiming =
   describe "should have similar time for auth error, whether queue exists or not, for all key types" $
     forM_ timingTests $ \tst ->
-      it (testName tst) $
-        smpTest2Cfg cfg (mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion) t $ \rh sh ->
+      it (testName tst) $ \(ATransport t, msType) ->
+        smpTest2Cfg (cfgMS msType) (mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion) t $ \rh sh ->
           testSameTiming rh sh tst
   where
     testName :: (C.AuthAlg, C.AuthAlg, Int) -> String
@@ -888,13 +904,13 @@ testTiming (ATransport t) =
             ]
           ok `shouldBe` True
 
-testMessageNotifications :: ATransport -> Spec
-testMessageNotifications (ATransport t) =
-  it "should create simplex connection, subscribe notifier and deliver notifications" $ do
+testMessageNotifications :: SpecWith (ATransport, AMSType)
+testMessageNotifications =
+  it "should create simplex connection, subscribe notifier and deliver notifications" $ \(ATransport t, msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     (nPub, nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    smpTest4 t $ \rh sh nh1 nh2 -> do
+    smpTest4 t msType $ \rh sh nh1 nh2 -> do
       (sId, rId, rKey, dhShared) <- createAndSecureQueue rh sPub
       let dec = decryptMsgV3 dhShared
       (rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
@@ -928,12 +944,12 @@ testMessageNotifications (ATransport t) =
         Nothing -> pure ()
         Just _ -> error "nothing else should be delivered to the 2nd notifier's TCP connection"
 
-testMsgExpireOnSend :: forall c. Transport c => TProxy c -> Spec
-testMsgExpireOnSend t =
-  it "should expire messages that are not received before messageTTL on SEND" $ do
+testMsgExpireOnSend :: SpecWith (ATransport, AMSType)
+testMsgExpireOnSend =
+  it "should expire messages that are not received before messageTTL on SEND" $ \(ATransport (t :: TProxy c), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    let cfg' = cfg {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
+    let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
     withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
       testSMPClient @c $ \sh -> do
         (sId, rId, rKey, dhShared) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
@@ -948,13 +964,13 @@ testMsgExpireOnSend t =
             Nothing -> return ()
             Just _ -> error "nothing else should be delivered"
 
-testMsgExpireOnInterval :: forall c. Transport c => TProxy c -> Spec
-testMsgExpireOnInterval t =
+testMsgExpireOnInterval :: SpecWith (ATransport, AMSType)
+testMsgExpireOnInterval =
   -- fails on ubuntu
-  xit' "should expire messages that are not received before messageTTL after expiry interval" $ do
+  xit' "should expire messages that are not received before messageTTL after expiry interval" $ \(ATransport (t :: TProxy c), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    let cfg' = cfg {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
+    let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
     withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
       testSMPClient @c $ \sh -> do
         (sId, rId, rKey, _) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
@@ -968,12 +984,12 @@ testMsgExpireOnInterval t =
             Nothing -> return ()
             Just _ -> error "nothing should be delivered"
 
-testMsgNOTExpireOnInterval :: forall c. Transport c => TProxy c -> Spec
-testMsgNOTExpireOnInterval t =
-  it "should NOT expire messages that are not received before messageTTL if expiry interval is large" $ do
+testMsgNOTExpireOnInterval :: SpecWith (ATransport, AMSType)
+testMsgNOTExpireOnInterval =
+  it "should NOT expire messages that are not received before messageTTL if expiry interval is large" $ \(ATransport (t :: TProxy c), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    let cfg' = cfg {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
+    let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
     withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
       testSMPClient @c $ \sh -> do
         (sId, rId, rKey, dhShared) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
@@ -1006,8 +1022,8 @@ instance Eq C.ASignature where
     Just Refl -> s == s'
     _ -> False
 
-syntaxTests :: ATransport -> Spec
-syntaxTests (ATransport t) = do
+serverSyntaxTests :: ATransport -> Spec
+serverSyntaxTests (ATransport t) = do
   it "unknown command" $ ("", "abcd", "1234", ('H', 'E', 'L', 'L', 'O')) >#> ("", "abcd", "1234", ERR $ CMD UNKNOWN)
   describe "NEW" $ do
     it "no parameters" $ (sampleSig, "bcda", "", NEW_) >#> ("", "bcda", "", ERR $ CMD SYNTAX)

tests/Test.hs

@@ -11,6 +11,7 @@ import CoreTests.BatchingTests
 import CoreTests.CryptoFileTests
 import CoreTests.CryptoTests
 import CoreTests.EncodingTests
+import CoreTests.MsgStoreTests
 import CoreTests.RetryIntervalTests
 import CoreTests.SOCKSSettings
 import CoreTests.StoreLogTests
@@ -24,8 +25,9 @@ import NtfServerTests (ntfServerTests)
 import RemoteControl (remoteControlTests)
 import SMPProxyTests (smpProxyTests)
 import ServerTests
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
 import Simplex.Messaging.Transport (TLS, Transport (..))
-import Simplex.Messaging.Transport.WebSockets (WS)
+-- import Simplex.Messaging.Transport.WebSockets (WS)
 import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive)
 import System.Environment (setEnv)
 import Test.Hspec
@@ -53,13 +55,20 @@ main = do
           describe "Version range" versionRangeTests
           describe "Encryption tests" cryptoTests
           describe "Encrypted files tests" cryptoFileTests
+          describe "Message store tests" msgStoreTests
           describe "Retry interval tests" retryIntervalTests
           describe "SOCKS settings tests" socksSettingsTests
           describe "Store log tests" storeLogTests
           describe "TRcvQueues tests" tRcvQueuesTests
           describe "Util tests" utilTests
-        describe "SMP server via TLS" $ serverTests (transport @TLS)
-        describe "SMP server via WebSockets" $ serverTests (transport @WS)
+        describe "SMP server via TLS, jornal message store" $ do
+          describe "SMP syntax" $ serverSyntaxTests (transport @TLS)
+          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
         describe "Notifications server" $ ntfServerTests (transport @TLS)
         describe "SMP client agent" $ agentTests (transport @TLS)
         describe "SMP proxy" smpProxyTests