WIP: stream db results for subscribeUserConnections

cabal.project

@@ -12,7 +12,7 @@ constraints: zip +disable-bzip2 +disable-zstd
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/simplexmq.git
-    tag: ee90ea6a69fe8283d37d9821cd83798fd0a76260
+    tag: 8f12555be5054a04cca88acf443f307af4ee84d8
 
 source-repository-package
     type: git

src/Simplex/Chat.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
@@ -3010,11 +3011,10 @@ subscribeUserConnections :: forall m. ChatMonad m => (PQSupport -> VersionRangeC
 subscribeUserConnections vr onlyNeeded agentBatchSubscribe user = do
   -- get user connections
   ce <- asks $ subscriptionEvents . config
-  (conns, cts, ucs, gs, ms, sfts, rfts, pcs) <-
+  (!conns, !cts, !ucs, !gs, !ms, !sfts, !rfts, !pcs) <-
     if onlyNeeded
       then do
-        (conns, entities) <- withStore' (`getConnectionsToSubscribe` vr)
-        let (cts, ucs, ms, sfts, rfts, pcs) = foldl' addEntity (M.empty, M.empty, M.empty, M.empty, M.empty, M.empty) entities
+        (conns, (cts, ucs, ms, sfts, rfts, pcs)) <- withStore' $ \db -> getConnectionsToSubscribe db vr initialEntities addEntity
         pure (conns, cts, ucs, [], ms, sfts, rfts, pcs)
       else do
         withStore' unsetConnectionToSubscribe
@@ -3037,7 +3037,8 @@ subscribeUserConnections vr onlyNeeded agentBatchSubscribe user = do
   rcvFileSubsToView rs rfts
   pendingConnSubsToView rs pcs
   where
-    addEntity (cts, ucs, ms, sfts, rfts, pcs) = \case
+    initialEntities = (M.empty, M.empty, M.empty, M.empty, M.empty, M.empty)
+    addEntity (!cts, !ucs, !ms, !sfts, !rfts, !pcs) = \case
       RcvDirectMsgConnection c (Just ct) -> let cts' = addConn c ct cts in (cts', ucs, ms, sfts, rfts, pcs)
       RcvDirectMsgConnection c Nothing -> let pcs' = addConn c (toPCC c) pcs in (cts, ucs, ms, sfts, rfts, pcs')
       RcvGroupMsgConnection c _g m -> let ms' = addConn c m ms in (cts, ucs, ms', sfts, rfts, pcs)

src/Simplex/Chat/Store/Connections.hs

@@ -189,16 +189,18 @@ getContactConnEntityByConnReqHash db vr user@User {userId} (cReqHash1, cReqHash2
         (userId, cReqHash1, cReqHash2, ConnDeleted)
   maybe (pure Nothing) (fmap eitherToMaybe . runExceptT . getConnectionEntity db vr user) connId_
 
-getConnectionsToSubscribe :: DB.Connection -> (PQSupport -> VersionRangeChat) -> IO ([ConnId], [ConnectionEntity])
-getConnectionsToSubscribe db vr = do
-  aConnIds <- map fromOnly <$> DB.query_ db "SELECT agent_conn_id FROM connections where to_subscribe = 1"
-  entities <- forM aConnIds $ \acId -> do
-    getUserByAConnId db acId >>= \case
-      Just user -> eitherToMaybe <$> runExceptT (getConnectionEntity db vr user acId)
-      Nothing -> pure Nothing
-  unsetConnectionToSubscribe db
-  let connIds = map (\(AgentConnId connId) -> connId) aConnIds
-  pure (connIds, catMaybes entities)
+getConnectionsToSubscribe :: DB.Connection -> (PQSupport -> VersionRangeChat) -> es -> (es -> ConnectionEntity -> es) -> IO ([ConnId], es)
+getConnectionsToSubscribe db vr initialES addEntity = do
+  r <- DB.fold_ db "SELECT agent_conn_id FROM connections where to_subscribe = 1" ([], initialES) collect
+  r <$ unsetConnectionToSubscribe db
+  where
+    collect (cids, es) (Only acId@(AgentConnId connId)) = do
+      es' <- getUserByAConnId db acId >>= \case
+        Just user -> runExceptT (getConnectionEntity db vr user acId) >>= \case
+          Right ce -> pure $! addEntity es ce
+          Left _err -> pure es
+        Nothing -> pure es
+      pure (connId : cids, es')
 
 unsetConnectionToSubscribe :: DB.Connection -> IO ()
 unsetConnectionToSubscribe db = DB.execute_ db "UPDATE connections SET to_subscribe = 0 WHERE to_subscribe = 1"