From 94da3fc51e3efb0ebec1e0a4e48cbdaa4bc3a6b5 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Mon, 16 Jun 2025 08:42:43 +0100 Subject: [PATCH] WIP --- src/Simplex/Messaging/Server.hs | 31 +++++++++++++++++-- .../Messaging/Server/MsgStore/Journal.hs | 5 +++ src/Simplex/Messaging/Server/MsgStore/STM.hs | 3 ++ .../Messaging/Server/MsgStore/Types.hs | 1 + .../Messaging/Server/QueueStore/Postgres.hs | 9 ++++++ .../Messaging/Server/QueueStore/STM.hs | 12 ++++++- .../Messaging/Server/QueueStore/Types.hs | 1 + 7 files changed, 58 insertions(+), 4 deletions(-) diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index f2da2c550..c1bbad6e4 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -426,7 +426,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt if sent then updateEndStats else -- if queue is full it can block - forkClient c ("sendPendingEvtsThread.queueEvts") $ + forkClient c "sendPendingEvtsThread.queueEvts" $ atomically (writeTBQueue sndQ (ts, [])) >> updateEndStats where ts = L.map (\(entId, evt) -> (NoCorrId, entId, evt)) evts @@ -1307,7 +1307,7 @@ client -- TODO [certs rcv] rcv subscriptions Server {subscribers, ntfSubscribers} ms - clnt@Client {clientId, serviceSubsCount = _todo', rcvQ, sndQ, clientTHParams = thParams'@THandleParams {sessionId}, procThreads} = do + clnt@Client {clientId, rcvQ, sndQ, msgQ, clientTHParams = thParams'@THandleParams {sessionId}, procThreads} = do labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands" let THandleParams {thVersion} = thParams' clntServiceId = (\THClientService {serviceId} -> serviceId) <$> (peerClientService =<< thAuth thParams') @@ -1751,8 +1751,33 @@ client sharedSubscribeService SRecipientService serviceId subscribers serviceSubsCount >>= \case Left e -> pure $ ERR e Right (hasSub, count) -> do - unless hasSub $ pure () -- TODO [certs rcv] if hasSubs is False, deliver all pending messages on a separate thread + unless hasSub $ forkClient clnt "deliverServiceMessages" $ liftIO $ deliverServiceMessages count pure $ SOKS count + where + deliverServiceMessages expectedCnt = do + (qCnt, _msgCnt, _dupCnt, _errCnt) <- foldRcvServiceQueues (queueStore ms) serviceId deliverQueueMsg (0, 0, 0, 0) + -- TODO [cert rcv] compare with expected + logNote $ "Service subscriptions for " <> tshow serviceId <> " (" <> tshow qCnt <> " queues)" + deliverQueueMsg :: (Int, Int, Int, Int) -> (StoreQueue s, QueueRec) -> IO (Int, Int, Int, Int) + deliverQueueMsg acc@(!qCnt, !msgCnt, !dupCnt, !errCnt) (q, qr) = + runExceptT (tryPeekMsg ms q) >>= \case + Left e -> pure (qCnt + 1, msgCnt, dupCnt, errCnt + 1) -- TODO deliver subscription error + Right Nothing -> pure acc + Right (Just msg) -> do + let rId = recipientId q + atomically (getNewSub rId) >>= \case + Nothing -> pure (qCnt + 1, msgCnt, dupCnt + 1, errCnt) + Just sub -> do + void $ atomically $ setDelivered sub msg + atomically $ writeTBQueue msgQ [(rId, encryptMsg qr msg)] + pure (qCnt + 1, msgCnt + 1, dupCnt, errCnt) + getNewSub rId = + TM.lookup rId (subscriptions clnt) >>= \case + Just _sub -> pure Nothing -- if delivery subscription already exists, then there is no need to deliver message + Nothing -> do + sub <- newSubscription NoSub + TM.insert rId sub $ subscriptions clnt + pure $ Just sub sharedSubscribeService :: (PartyI p, ServiceParty p) => SParty p -> ServiceId -> ServerSubscribers s -> (Client s -> TVar Int64) -> M s (Either ErrorType (Bool, Int64)) sharedSubscribeService party serviceId srvSubscribers clientServiceSubs = do diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index 78f9c1393..dc73ea25b 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Journal.hs @@ -358,6 +358,8 @@ instance QueueStoreClass (JournalQueue s) (QStore s) where {-# INLINE getQueueNtfServices #-} getServiceQueueCount = withQS (getServiceQueueCount @(JournalQueue s)) {-# INLINE getServiceQueueCount #-} + foldRcvServiceQueues = withQS foldRcvServiceQueues + {-# INLINE foldRcvServiceQueues #-} makeQueue_ :: JournalMsgStore s -> RecipientId -> QueueRec -> Lock -> IO (JournalQueue s) makeQueue_ JournalMsgStore {sharedLock} rId qr queueLock = do @@ -444,6 +446,9 @@ instance MsgStoreClass (JournalMsgStore s) where getLoadedQueue :: JournalQueue s -> IO (JournalQueue s) getLoadedQueue q = fromMaybe q <$> TM.lookupIO (recipientId q) (loadedQueues $ queueStore_ ms) + foldRcvServiceMessages :: JournalMsgStore s -> Service -> (a -> RecipientId -> Message -> IO a) -> IO a + foldRcvServiceMessages = undefined + logQueueStates :: JournalMsgStore s -> IO () logQueueStates ms = withActiveMsgQueues ms $ unStoreIO . logQueueState diff --git a/src/Simplex/Messaging/Server/MsgStore/STM.hs b/src/Simplex/Messaging/Server/MsgStore/STM.hs index ed24e85a4..63b39425b 100644 --- a/src/Simplex/Messaging/Server/MsgStore/STM.hs +++ b/src/Simplex/Messaging/Server/MsgStore/STM.hs @@ -89,6 +89,9 @@ instance MsgStoreClass STMMsgStore where expireOldMessages _tty ms now ttl = withLoadedQueues (queueStore_ ms) $ atomically . expireQueueMsgs ms now (now - ttl) + foldRcvServiceMessages :: STMMsgStore -> Service -> (a -> RecipientId -> Message -> IO a) -> IO a + foldRcvServiceMessages = undefined + logQueueStates _ = pure () {-# INLINE logQueueStates #-} logQueueState _ = pure () diff --git a/src/Simplex/Messaging/Server/MsgStore/Types.hs b/src/Simplex/Messaging/Server/MsgStore/Types.hs index e2d139ffb..f66bccd80 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Types.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Types.hs @@ -44,6 +44,7 @@ class (Monad (StoreMonad s), QueueStoreClass (StoreQueue s) (QueueStore s)) => M unsafeWithAllMsgQueues :: Monoid a => Bool -> Bool -> s -> (StoreQueue s -> IO a) -> IO a -- tty, store, now, ttl expireOldMessages :: Bool -> s -> Int64 -> Int64 -> IO MessageStats + foldRcvServiceMessages :: s -> Service -> (a -> RecipientId -> Message -> IO a) -> IO a logQueueStates :: s -> IO () logQueueState :: StoreQueue s -> StoreMonad s () queueStore :: s -> QueueStore s diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs index 7258d9148..6cb3f73df 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs @@ -495,6 +495,15 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where SRecipientService -> "SELECT count(1) FROM msg_queues WHERE rcv_service_id = ? AND deleted_at IS NULL" SNotifierService -> "SELECT count(1) FROM msg_queues WHERE ntf_service_id = ? AND deleted_at IS NULL" + foldRcvServiceQueues :: PostgresQueueStore q -> (RecipientId -> QueueRec -> IO q) -> ServiceId -> (a -> (q, QueueRec) -> IO a) -> a -> IO a + foldRcvServiceQueues st mkQ serviceId f acc = + withConnection (dbStore st) $ \db -> + DB.fold db (queueRecQuery <> " WHERE rcv_service_id = ? AND deleted_at IS NULL") (Only serviceId) acc $ \acc' -> do + + f' acc' . rowToQueueRec + DB.fold db "SELECT service_id, service_role, service_cert, service_cert_hash, created_at FROM services" mempty $ + \ !acc -> fmap (acc <>) . f . rowToServiceRec + batchInsertServices :: [STMService] -> PostgresQueueStore q -> IO Int64 batchInsertServices services' toStore = withConnection (dbStore toStore) $ \db -> diff --git a/src/Simplex/Messaging/Server/QueueStore/STM.hs b/src/Simplex/Messaging/Server/QueueStore/STM.hs index 4dd6240a8..f33eace90 100644 --- a/src/Simplex/Messaging/Server/QueueStore/STM.hs +++ b/src/Simplex/Messaging/Server/QueueStore/STM.hs @@ -44,7 +44,7 @@ import Simplex.Messaging.Server.StoreLog import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport (SMPServiceRole (..)) -import Simplex.Messaging.Util (anyM, ifM, tshow, ($>>), ($>>=), (<$$)) +import Simplex.Messaging.Util (anyM, ifM, tshow, ($>>), ($>>=), (<$$), (<$$>)) import System.IO import UnliftIO.STM @@ -356,6 +356,16 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where SRecipientService -> serviceRcvQueues SNotifierService -> serviceNtfQueues + foldRcvServiceQueues :: STMQueueStore q -> ServiceId -> (a -> (q, QueueRec) -> IO a) -> a -> IO a + foldRcvServiceQueues st serviceId f acc = + TM.lookupIO serviceId (services st) >>= \case + Nothing -> pure acc + Just s -> + readTVarIO (serviceRcvQueues s) + >>= foldM (\a -> get >=> maybe (pure a) (f a)) acc + where + get rId = TM.lookupIO rId (queues st) $>>= \q -> (q,) <$$> readTVarIO (queueRec q) + withQueueRec :: TVar (Maybe QueueRec) -> (QueueRec -> STM a) -> IO (Either ErrorType a) withQueueRec qr a = atomically $ readQueueRec qr >>= mapM a diff --git a/src/Simplex/Messaging/Server/QueueStore/Types.hs b/src/Simplex/Messaging/Server/QueueStore/Types.hs index 55be4d21d..71a56e8ab 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Types.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Types.hs @@ -49,6 +49,7 @@ class StoreQueueClass q => QueueStoreClass q s where setQueueService :: (PartyI p, ServiceParty p) => s -> q -> SParty p -> Maybe ServiceId -> IO (Either ErrorType ()) getQueueNtfServices :: s -> [(NotifierId, a)] -> IO (Either ErrorType ([(Maybe ServiceId, [(NotifierId, a)])], [(NotifierId, a)])) getServiceQueueCount :: (PartyI p, ServiceParty p) => s -> SParty p -> ServiceId -> IO (Either ErrorType Int64) + foldRcvServiceQueues :: s -> ServiceId -> (a -> (q, QueueRec) -> IO a) -> a -> IO a data EntityCounts = EntityCounts { queueCount :: Int,