WIP

src/Simplex/Chat.hs

@@ -3331,23 +3331,94 @@ agentSubscriber = do
 subscribeUserConnections :: VersionRangeChat -> Bool -> User -> CM ()
 subscribeUserConnections vr onlyNeeded user = do
   -- get user connections
-  conns <-
+  (ctConns, ucConns, mConns, sftConns, rftConns, pcConns) <-
     if onlyNeeded
-      then withStore' getConnectionsToSubscribe
+      then
+        -- XXX: can be streamed from DB without collecting everything, needs DB.fold wrapper in mq
+        foldl' addEntity ([], [], [], [], [], []) <$> withStore' (\db -> getConnectionsToSubscribe db vr user)
       else do
         withStore' unsetConnectionToSubscribe
-        ctConns <- mapMaybe (\ct -> if contactActive ct then contactConnId ct else Nothing) <$> withStore_ (`getUserContacts` vr)
-        ucConns <- map (aConnId . fst) <$> withStore_ (`getUserContactLinks` vr)
-        mConns <- concatMap (\(Group _ ms) -> mapMaybe memberConnId (filter (not . memberRemoved) ms)) <$> withStore_ (`getUserGroups` vr)
-        sftConns <- map sndFileTransferConnId <$> withStore_ getLiveSndFileTransfers
-        rftConns <- mapMaybe liveRcvFileTransferConnId <$> withStore_ getLiveRcvFileTransfers
-        pcConns <- map aConnId' <$> withStore_ getPendingContactConnections
-        pure $ concat [ctConns, ucConns, mConns, sftConns, rftConns, pcConns]
-  -- subscribe using batched commands
-  void $ withAgent (`Agent.subscribeConnections` conns)
+        ctConns <- getContactConns
+        ucConns <- getUserContactLinkConns
+        -- (gs, mConns, ms) <- getGroupMemberConns
+        mConns <- getGroupMemberConns
+        sftConns <- getSndFileTransferConns
+        rftConns <- getRcvFileTransferConns
+        pcConns <- getPendingContactConns
+        pure (ctConns, ucConns, mConns, sftConns, rftConns, pcConns)
+  let conns = concat [ctConns, ucConns, mConns, sftConns, rftConns, pcConns]
+  void . lift . forkIO . void . runExceptT $ do -- detach subscription and result processing
+    rs <- withAgent (`Agent.subscribeConnections` conns) -- subscribe using batched commands
+    let (errs, _oks) = M.mapEither id rs
+    -- api <- asks $ coreApi . config
+    -- refs <- withStore' $ \db -> getConnectionsContacts db (if api then M.keys errs else M.keys rs)
+    -- let connRefs = M.fromList $ map (\ref@ContactRef {agentConnId} -> (agentConnId, ref)) refs
+    ce <- asks $ subscriptionEvents . config
+    contactSubsToView errs ctConns ce
+    -- TODO: others
   where
+    addEntity (cts, ucs, ms, sfts, rfts, pcs) = \case
+      RcvDirectMsgConnection c (Just _ct) -> let cts' = aConnId c : cts in (cts', ucs, ms, sfts, rfts, pcs)
+      RcvDirectMsgConnection c Nothing -> let pcs' = aConnId c : pcs in (cts, ucs, ms, sfts, rfts, pcs')
+      RcvGroupMsgConnection c _g _m -> let ms' = aConnId c : ms in (cts, ucs, ms', sfts, rfts, pcs)
+      SndFileConnection c _sft -> let sfts' = aConnId c : sfts in (cts, ucs, ms, sfts', rfts, pcs)
+      RcvFileConnection c _rft -> let rfts' = aConnId c : rfts in (cts, ucs, ms, sfts, rfts', pcs)
+      UserContactConnection c _uc -> let ucs' = aConnId c : ucs in (cts, ucs', ms, sfts, rfts, pcs)
     withStore_ :: (DB.Connection -> User -> IO [a]) -> CM [a]
     withStore_ a = withStore' (`a` user) `catchChatError` \e -> toView (CRChatError (Just user) e) $> []
+    getContactConns :: CM [ConnId]
+    getContactConns = do
+      cts <- withStore_ (`getUserContacts` vr)
+      let cts' = mapMaybe (\ct -> (,ct) <$> contactConnId ct) $ filter contactActive cts
+      pure (map fst cts')
+    getUserContactLinkConns :: CM [ConnId]
+    getUserContactLinkConns = do
+      (cs, _ucs) <- unzip <$> withStore_ (`getUserContactLinks` vr)
+      let connIds = map aConnId cs
+      pure connIds
+    getGroupMemberConns :: CM [ConnId] -- ([Group], [ConnId], Map ConnId GroupMember)
+    getGroupMemberConns = do
+      gs <- withStore_ (`getUserGroups` vr)
+      let mPairs = concatMap (\(Group _ ms) -> mapMaybe (\m -> (,m) <$> memberConnId m) (filter (not . memberRemoved) ms)) gs
+      -- pure (gs, map fst mPairs, M.fromList mPairs)
+      pure $ map fst mPairs
+    getSndFileTransferConns :: CM [ConnId]
+    getSndFileTransferConns = do
+      sfts <- withStore_ getLiveSndFileTransfers
+      let connIds = map sndFileTransferConnId sfts
+      pure connIds
+    getRcvFileTransferConns :: CM [ConnId]
+    getRcvFileTransferConns = do
+      rfts <- withStore_ getLiveRcvFileTransfers
+      let rftPairs = mapMaybe (\ft -> (,ft) <$> liveRcvFileTransferConnId ft) rfts
+      pure (map fst rftPairs)
+    getPendingContactConns :: CM [ConnId]
+    getPendingContactConns = do
+      pcs <- withStore_ getPendingContactConnections
+      let connIds = map aConnId' pcs
+      pure connIds
+    contactSubsToView :: Map ConnId AgentErrorType -> [ConnId] -> Bool -> CM ()
+    contactSubsToView errs cts ce = do
+      -- chatModifyVar connNetworkStatuses $ M.union (M.fromList statuses) -- via UP
+      ifM (asks $ coreApi . config) notifyAPI notifyCLI
+      where
+        notifyCLI = do
+          let (okSubs, errSubs) = foldl' (\(os, es) acId -> if M.member acId errs then (os, es + 1) else (os + 1, es)) (0, 0) cts
+          toView $ CRConnectionSubSummary {user, okSubs, errSubs}
+          when (ce && errSubs > 0) $ toView $ error "TODO: CRContactSubError {user, contactName :: Text, chatError :: ChatError}"
+        notifyAPI = toView . CRNetworkStatuses (Just user) $ map (uncurry ConnNetworkStatus) statuses
+        statuses = foldr addStatus [] cts
+          where
+            addStatus :: ConnId -> [(AgentConnId, NetworkStatus)] -> [(AgentConnId, NetworkStatus)]
+            addStatus connId nss =
+              case M.lookup connId errs of
+                Nothing -> nss
+                Just err -> (AgentConnId connId, NSError $ errorNetworkStatus err) : nss
+            errorNetworkStatus :: AgentErrorType -> String
+            errorNetworkStatus = \case
+              BROKER _ NETWORK -> "network"
+              SMP _ SMP.AUTH -> "contact deleted"
+              e -> show e
 
 cleanupManager :: CM ()
 cleanupManager = do

src/Simplex/Chat/Controller.hs

@@ -668,8 +668,8 @@ data ChatResponse
   | CRSubscriptionEnd {user :: User, connectionEntity :: ConnectionEntity}
   | CRContactsDisconnected {server :: SMPServer, contactRefs :: [ContactRef]}
   | CRContactsSubscribed {server :: SMPServer, contactRefs :: [ContactRef]}
-  | CRContactSubError {user :: User, contact :: Contact, chatError :: ChatError}
-  | CRContactSubSummary {user :: User, contactSubscriptions :: [ContactSubStatus]}
+  | CRConnectionSubSummary {user :: User, okSubs :: Int, errSubs :: Int}
+  | CRContactSubError {user :: User, contactName :: ContactName, chatError :: ChatError}
   | CRUserContactSubSummary {user :: User, userContactSubscriptions :: [UserContactSubStatus]}
   | CRNetworkStatus {networkStatus :: NetworkStatus, connections :: [AgentConnId]}
   | CRNetworkStatuses {user_ :: Maybe User, networkStatuses :: [ConnNetworkStatus]}

src/Simplex/Chat/Store/Connections.hs

@@ -18,9 +18,10 @@ module Simplex.Chat.Store.Connections
 where
 
 import Control.Applicative ((<|>))
+import Control.Monad (forM)
 import Control.Monad.Except
 import Data.Int (Int64)
-import Data.Maybe (fromMaybe)
+import Data.Maybe (catMaybes, fromMaybe)
 import Database.SQLite.Simple (Only (..), (:.) (..))
 import Database.SQLite.Simple.QQ (sql)
 import Simplex.Chat.Protocol
@@ -210,10 +211,11 @@ 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 -> IO [ConnId]
-getConnectionsToSubscribe db = do
-  connIds <- map fromOnly <$> DB.query_ db "SELECT agent_conn_id FROM connections where to_subscribe = 1"
-  connIds <$ unsetConnectionToSubscribe db
+getConnectionsToSubscribe :: DB.Connection -> VersionRangeChat -> User -> IO [ConnectionEntity]
+getConnectionsToSubscribe db vr user@User {userId} = do
+  aConnIds <- map fromOnly <$> DB.query db "SELECT agent_conn_id FROM connections WHERE c.user_id = ? AND to_subscribe = 1" (Only userId)
+  unsetConnectionToSubscribe db
+  fmap catMaybes $ forM aConnIds $ \acId -> eitherToMaybe <$> runExceptT (getConnectionEntity db vr user acId)
 
 unsetConnectionToSubscribe :: DB.Connection -> IO ()
 unsetConnectionToSubscribe db = DB.execute_ db "UPDATE connections SET to_subscribe = 0 WHERE to_subscribe = 1"

src/Simplex/Chat/View.hs

@@ -229,15 +229,14 @@ responseToView hu@(currentRH, user_) ChatConfig {logLevel, showReactions, showRe
      in ttyUser u [sShow connId <> ": END"]
   CRContactsDisconnected srv cs -> [plain $ "server disconnected " <> showSMPServer srv <> " (" <> contactList cs <> ")"]
   CRContactsSubscribed srv cs -> [plain $ "server connected " <> showSMPServer srv <> " (" <> contactList cs <> ")"]
-  CRContactSubError u c e -> ttyUser u [ttyContact' c <> ": contact error " <> sShow e]
-  CRContactSubSummary u summary ->
-    ttyUser u $ [sShow (length subscribed) <> " contacts connected (use " <> highlight' "/cs" <> " for the list)" | not (null subscribed)] <> viewErrorsSummary errors " contact errors"
-    where
-      (errors, subscribed) = partition (isJust . contactError) summary
+  CRContactSubError u c e -> ttyUser u [ttyContact c <> ": contact error " <> sShow e]
+  CRConnectionSubSummary {user, okSubs, errSubs} ->
+  -- CRContactSubSummary u summary ->
+    ttyUser user $ [sShow okSubs <> " contacts connected (use " <> highlight' "/cs" <> " for the list)" | okSubs > 0] <> viewErrorsSummary errSubs " contact errors"
   CRUserContactSubSummary u summary ->
     ttyUser u $
       map addressSS addresses
-        <> ([sShow (length groupLinksSubscribed) <> " group links active" | not (null groupLinksSubscribed)] <> viewErrorsSummary groupLinkErrors " group link errors")
+        <> ([sShow (length groupLinksSubscribed) <> " group links active" | not (null groupLinksSubscribed)] <> viewErrorsSummary (length groupLinkErrors) " group link errors")
     where
       (addresses, groupLinks) = partition (\UserContactSubStatus {userContact} -> isNothing . userContactGroupId $ userContact) summary
       addressSS UserContactSubStatus {userContactError} = maybe ("Your address is active! To show: " <> highlight' "/sa") (\e -> "User address error: " <> sShow e <> ", to delete your address: " <> highlight' "/da") userContactError
@@ -276,7 +275,7 @@ responseToView hu@(currentRH, user_) ChatConfig {logLevel, showReactions, showRe
   CRNewMemberContactReceivedInv u ct g m -> ttyUser u [ttyGroup' g <> " " <> ttyMember m <> " is creating direct contact " <> ttyContact' ct <> " with you"]
   CRContactAndMemberAssociated u ct g m ct' -> ttyUser u $ viewContactAndMemberAssociated ct g m ct'
   CRMemberSubError u g m e -> ttyUser u [ttyGroup' g <> " member " <> ttyMember m <> " error: " <> sShow e]
-  CRMemberSubSummary u summary -> ttyUser u $ viewErrorsSummary (filter (isJust . memberError) summary) " group member errors"
+  CRMemberSubSummary u summary -> ttyUser u $ viewErrorsSummary (length $ filter (isJust . memberError) summary) " group member errors"
   CRGroupSubscribed u g -> ttyUser u $ viewGroupSubscribed g
   CRPendingSubSummary u _ -> ttyUser u []
   CRSndFileSubError u SndFileTransfer {fileId, fileName} e ->
@@ -444,8 +443,8 @@ responseToView hu@(currentRH, user_) ChatConfig {logLevel, showReactions, showRe
     testViewItem (CChatItem _ ci@ChatItem {meta = CIMeta {itemText}}) membership_ =
       let deleted_ = maybe "" (\t -> " [" <> t <> "]") (chatItemDeletedText ci membership_)
        in itemText <> deleted_
-    viewErrorsSummary :: [a] -> StyledString -> [StyledString]
-    viewErrorsSummary summary s = [ttyError (T.pack . show $ length summary) <> s <> " (run with -c option to show each error)" | not (null summary)]
+    viewErrorsSummary :: Int -> StyledString -> [StyledString]
+    viewErrorsSummary numErrors s = [ttyError (tshow numErrors) <> s <> " (run with -c option to show each error)" | numErrors > 0]
     contactList :: [ContactRef] -> String
     contactList cs = T.unpack . T.intercalate ", " $ map (\ContactRef {localDisplayName = n} -> "@" <> n) cs
     unmuted :: User -> ChatInfo c -> ChatItem c d -> [StyledString] -> [StyledString]