WIP

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

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
 

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 ()

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

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 ->

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
 

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,