service delivery WIP

Author: epoberezkin

src/Simplex/Messaging/Server.hs

@@ -1357,7 +1357,6 @@ forkClient Client {endThreads, endThreadSeq} label action = do
 
 client :: forall s. MsgStoreClass s => Server s -> s -> Client s -> M s ()
 client
-  -- TODO [certs rcv] rcv subscriptions
   Server {subscribers, ntfSubscribers}
   ms
   clnt@Client {clientId, rcvQ, sndQ, msgQ, clientTHParams = thParams'@THandleParams {sessionId}, procThreads} = do
@@ -1805,23 +1804,27 @@ client
               pure $ SOKS count idsHash
           where
             deliverServiceMessages expectedCnt = do
-              (qCnt, _msgCnt, _dupCnt, _errCnt) <- foldRcvServiceMessages ms serviceId deliverQueueMsg (0, 0, 0, 0)
-              atomically $ writeTBQueue msgQ [(NoCorrId, NoEntity, ALLS)]
-              -- TODO [certs rcv] compare with expected
-              logNote $ "Service subscriptions for " <> tshow serviceId <> " (" <> tshow qCnt <> " queues)"
-            deliverQueueMsg :: (Int, Int, Int, Int) -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO (Int, Int, Int, Int)
-            deliverQueueMsg (!qCnt, !msgCnt, !dupCnt, !errCnt) rId = \case
-              Left e -> pure (qCnt + 1, msgCnt, dupCnt, errCnt + 1) -- TODO [certs rcv] deliver subscription error
+              foldRcvServiceMessages ms serviceId deliverQueueMsg (0, 0, 0, []) >>= \case
+                Right (qCnt, _msgCnt, _dupCnt, _errCnt) -> do
+                  atomically $ writeTBQueue msgQ [(NoCorrId, NoEntity, ALLS)]
+                  -- TODO [certs rcv] compare with expected
+                  logNote $ "Service subscriptions for " <> tshow serviceId <> " (" <> tshow qCnt <> " queues)"
+                Left e -> do
+                  -- TODO [certs rcv] deliver SMP error
+                  logError $ "Service subscriptions for " <> tshow serviceId <> " error: " <> tshow e
+            deliverQueueMsg :: (Int, Int, Int, [ErrorType]) -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO (Int, Int, Int, [ErrorType])
+            deliverQueueMsg (!qCnt, !msgCnt, !dupCnt, !errs) rId = \case
+              Left e -> pure (qCnt + 1, msgCnt, dupCnt, e : errs) -- TODO [certs rcv] deliver subscription error
               Right qMsg_ -> case qMsg_ of
-                Nothing -> pure (qCnt + 1, msgCnt, dupCnt, errCnt)
+                Nothing -> pure (qCnt + 1, msgCnt, dupCnt, errs)
                 Just (qr, msg) ->
                   atomically (getSubscription rId) >>= \case
-                    Nothing -> pure (qCnt + 1, msgCnt, dupCnt + 1, errCnt)
+                    Nothing -> pure (qCnt + 1, msgCnt, dupCnt + 1, errs)
                     Just sub -> do
                       ts <- getSystemSeconds
                       atomically $ setDelivered sub msg ts
                       atomically $ writeTBQueue msgQ [(NoCorrId, rId, MSG (encryptMsg qr msg))]
-                      pure (qCnt + 1, msgCnt + 1, dupCnt, errCnt)
+                      pure (qCnt + 1, msgCnt + 1, dupCnt, errs)
             getSubscription rId =
               TM.lookup rId (subscriptions clnt) >>= \case
                 -- If delivery subscription already exists, then there is no need to deliver message.

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

@@ -444,9 +444,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 -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO a
+  foldRcvServiceMessages :: JournalMsgStore s -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO (Either ErrorType a)
   foldRcvServiceMessages ms serviceId f acc = case queueStore_ ms of
-    MQStore st -> foldRcvServiceQueues st serviceId f' acc
+    MQStore st -> fmap Right $ foldRcvServiceQueues st serviceId f' acc
       where
         f' a (q, qr) = runExceptT (tryPeekMsg ms q) >>= f a (recipientId q) . ((qr,) <$$>)
 #if defined(dbServerPostgres)

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

@@ -119,9 +119,9 @@ instance MsgStoreClass PostgresMsgStore where
       toMessageStats (expiredMsgsCount, storedMsgsCount, storedQueues) =
         MessageStats {expiredMsgsCount, storedMsgsCount, storedQueues}
 
-  foldRcvServiceMessages :: PostgresMsgStore -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO a
+  foldRcvServiceMessages :: PostgresMsgStore -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO (Either ErrorType a)
   foldRcvServiceMessages ms serviceId f acc =
-    withTransaction (dbStore $ queueStore_ ms) $ \db ->
+    runExceptT $ withDB' "foldRcvServiceMessages" (queueStore_ ms) $ \db ->
       DB.fold
         db
         [sql|

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

@@ -87,10 +87,10 @@ instance MsgStoreClass STMMsgStore where
   expireOldMessages _tty ms now ttl =
     withLoadedQueues (queueStore_ ms) $ atomically . expireQueueMsgs ms now (now - ttl)
 
-  foldRcvServiceMessages :: STMMsgStore -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO a
-  foldRcvServiceMessages ms serviceId f=
-    foldRcvServiceQueues (queueStore_ ms) serviceId $ \a (q, qr) ->
-      runExceptT (tryPeekMsg ms q) >>= f a (recipientId q) . ((qr,) <$$>)
+  foldRcvServiceMessages :: STMMsgStore -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO (Either ErrorType a)
+  foldRcvServiceMessages ms serviceId f = fmap Right . foldRcvServiceQueues (queueStore_ ms) serviceId f'
+    where
+      f' a (q, qr) = runExceptT (tryPeekMsg ms q) >>= f a (recipientId q) . ((qr,) <$$>)
 
   logQueueStates _ = pure ()
   {-# INLINE logQueueStates #-}

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

@@ -45,7 +45,7 @@ class (Monad (StoreMonad s), QueueStoreClass (StoreQueue s) (QueueStore s)) => M
   unsafeWithAllMsgQueues :: Monoid a => Bool -> s -> (StoreQueue s -> IO a) -> IO a
   -- tty, store, now, ttl
   expireOldMessages :: Bool -> s -> Int64 -> Int64 -> IO MessageStats
-  foldRcvServiceMessages :: s -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO a
+  foldRcvServiceMessages :: s -> ServiceId -> (a -> RecipientId -> Either ErrorType (Maybe (QueueRec, Message)) -> IO a) -> a -> IO (Either ErrorType a)
   logQueueStates :: s -> IO ()
   logQueueState :: StoreQueue s -> StoreMonad s ()
   queueStore :: s -> QueueStore s

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

@@ -581,9 +581,9 @@ foldServiceRecs st f =
     DB.fold_ db "SELECT service_id, service_role, service_cert, service_cert_hash, created_at FROM services" mempty $
       \ !acc -> fmap (acc <>) . f . rowToServiceRec
 
-foldRcvServiceQueueRecs :: PostgresQueueStore q -> ServiceId -> (a -> (RecipientId, QueueRec) -> IO a) -> a -> IO a
+foldRcvServiceQueueRecs :: PostgresQueueStore q -> ServiceId -> (a -> (RecipientId, QueueRec) -> IO a) -> a -> IO (Either ErrorType a)
 foldRcvServiceQueueRecs st serviceId f acc =
-  withTransaction (dbStore st) $ \db ->
+  runExceptT $ withDB' "foldRcvServiceQueueRecs" st $ \db ->
     DB.fold db (queueRecQuery <> " WHERE rcv_service_id = ? AND deleted_at IS NULL") (Only serviceId) acc $ \a -> f a . rowToQueueRec
 
 foldQueueRecs :: Monoid a => Bool -> Bool -> PostgresQueueStore q -> ((RecipientId, QueueRec) -> IO a) -> IO a