asynchronously subscribe to user connections (#310)

* asynchronously subscribe to user connections

* send subscription status summaries to view/api

* refactor

* add help messages in summaries

* update simplexmq

* rename config field

Co-authored-by: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com>

cabal.project

@@ -3,7 +3,7 @@ packages: .
 source-repository-package
     type: git
     location: git://github.com/simplex-chat/simplexmq.git
-    tag: dff5cad1bef67376e82c3dc15cccdb5ba9e675ab
+    tag: 7d1fdadef0541e0587d4966bc95c2930bf0f95ff
 
 source-repository-package
     type: git

sha256map.nix

@@ -1,5 +1,5 @@
 {
-  "git://github.com/simplex-chat/simplexmq.git"."dff5cad1bef67376e82c3dc15cccdb5ba9e675ab" = "06291v6vw7i00r0j13qx5apkz794jak68n1yr875gi32dxx5lhnp";
+  "git://github.com/simplex-chat/simplexmq.git"."7d1fdadef0541e0587d4966bc95c2930bf0f95ff" = "1hzyswxjpilkdalyr9i5swi2djlv3wf8nwfv7k06m5ajmi1zb4i3";
   "git://github.com/simplex-chat/aeson.git"."3eb66f9a68f103b5f1489382aad89f5712a64db7" = "0kilkx59fl6c3qy3kjczqvm8c3f4n3p0bdk9biyflf51ljnzp4yp";
   "git://github.com/simplex-chat/haskell-terminal.git"."f708b00009b54890172068f168bf98508ffcd495" = "0zmq7lmfsk8m340g47g5963yba7i88n4afa6z93sg9px5jv1mijj";
   "git://github.com/zw3rk/android-support.git"."3c3a5ab0b8b137a072c98d3d0937cbdc96918ddb" = "1r6jyxbim3dsvrmakqfyxbd6ms6miaghpbwyl0sr6dzwpgaprz97";

src/Simplex/Chat.hs

@@ -58,7 +58,7 @@ import System.Exit (exitFailure, exitSuccess)
 import System.FilePath (combine, splitExtensions, takeFileName)
 import System.IO (Handle, IOMode (..), SeekMode (..), hFlush, openFile, stdout)
 import Text.Read (readMaybe)
-import UnliftIO.Async (Async, async, race_)
+import UnliftIO.Async
 import UnliftIO.Concurrent (forkIO, threadDelay)
 import UnliftIO.Directory (doesDirectoryExist, doesFileExist, getFileSize, getHomeDirectory, getTemporaryDirectory)
 import qualified UnliftIO.Exception as E
@@ -78,8 +78,9 @@ defaultChatConfig =
           },
       dbPoolSize = 1,
       yesToMigrations = False,
-      tbqSize = 16,
+      tbqSize = 64,
       fileChunkSize = 15780,
+      subscriptionEvents = False,
       testView = False
     }
 
@@ -87,12 +88,13 @@ logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
 
 newChatController :: SQLiteStore -> Maybe User -> ChatConfig -> ChatOpts -> (Notification -> IO ()) -> IO ChatController
-newChatController chatStore user config@ChatConfig {agentConfig = cfg, tbqSize} ChatOpts {dbFilePrefix, smpServers} sendNotification = do
+newChatController chatStore user cfg@ChatConfig {agentConfig = aCfg, tbqSize} ChatOpts {dbFilePrefix, smpServers, logConnections} sendNotification = do
   let f = chatStoreFile dbFilePrefix
+  let config = cfg {subscriptionEvents = logConnections}
   activeTo <- newTVarIO ActiveNone
   firstTime <- not <$> doesFileExist f
   currentUser <- newTVarIO user
-  smpAgent <- getSMPAgentClient cfg {dbFile = dbFilePrefix <> "_agent.db", smpServers}
+  smpAgent <- getSMPAgentClient aCfg {dbFile = dbFilePrefix <> "_agent.db", smpServers}
   agentAsync <- newTVarIO Nothing
   idsDrg <- newTVarIO =<< drgNew
   inputQ <- newTBQueueIO tbqSize
@@ -462,36 +464,48 @@ agentSubscriber user = do
       processAgentMessage u connId msg `catchError` (toView . CRChatError)
 
 subscribeUserConnections :: (MonadUnliftIO m, MonadReader ChatController m) => User -> m ()
-subscribeUserConnections user@User {userId} = void . runExceptT $ do
-  subscribeContacts
-  subscribeGroups
-  subscribeFiles
-  subscribePendingConnections
-  subscribeUserContactLink
+subscribeUserConnections user@User {userId} = do
+  ce <- asks $ subscriptionEvents . config
+  void . runExceptT . (mapConcurrently_ id) $
+    [ subscribeContacts ce,
+      subscribeGroups ce,
+      subscribeFiles,
+      subscribePendingConnections,
+      subscribeUserContactLink
+    ]
   where
-    subscribeContacts = do
+    subscribeContacts ce = do
       contacts <- withStore (`getUserContacts` user)
-      forM_ contacts $ \ct ->
-        (subscribe (contactConnId ct) >> toView (CRContactSubscribed ct)) `catchError` (toView . CRContactSubError ct)
-    subscribeGroups = do
+      toView . CRContactSubSummary =<< forConcurrently contacts (\ct -> ContactSubStatus ct <$> subscribeContact ce ct)
+    subscribeContact ce ct =
+      (subscribe (contactConnId ct) >> when ce (toView $ CRContactSubscribed ct) $> Nothing)
+        `catchError` (\e -> when ce (toView $ CRContactSubError ct e) $> Just e)
+    subscribeGroups ce = do
       groups <- withStore (`getUserGroups` user)
-      forM_ groups $ \(Group g@GroupInfo {membership} members) -> do
-        let connectedMembers = mapMaybe (\m -> (m,) <$> memberConnId m) members
-        if memberStatus membership == GSMemInvited
-          then toView $ CRGroupInvitation g
-          else
-            if null connectedMembers
-              then
-                if memberActive membership
-                  then toView $ CRGroupEmpty g
-                  else toView $ CRGroupRemoved g
-              else do
-                forM_ connectedMembers $ \(GroupMember {localDisplayName = c}, cId) ->
-                  subscribe cId `catchError` (toView . CRMemberSubError g c)
-                toView $ CRGroupSubscribed g
+      toView . CRMemberSubErrors . mconcat =<< forConcurrently groups (subscribeGroup ce)
+    subscribeGroup ce (Group g@GroupInfo {membership} members) = do
+      let connectedMembers = mapMaybe (\m -> (m,) <$> memberConnId m) members
+      if memberStatus membership == GSMemInvited
+        then do
+          toView $ CRGroupInvitation g
+          pure []
+        else
+          if null connectedMembers
+            then do
+              if memberActive membership
+                then toView $ CRGroupEmpty g
+                else toView $ CRGroupRemoved g
+              pure []
+            else do
+              ms <- forConcurrently connectedMembers $ \(m@GroupMember {localDisplayName = c}, cId) ->
+                (m,) <$> ((subscribe cId $> Nothing) `catchError` (\e -> when ce (toView $ CRMemberSubError g c e) $> Just e))
+              toView $ CRGroupSubscribed g
+              pure $ mapMaybe (\(m, e) -> maybe Nothing (Just . MemberSubError m) e) ms
     subscribeFiles = do
-      withStore (`getLiveSndFileTransfers` user) >>= mapM_ subscribeSndFile
-      withStore (`getLiveRcvFileTransfers` user) >>= mapM_ subscribeRcvFile
+      sndFileTransfers <- withStore (`getLiveSndFileTransfers` user)
+      forConcurrently_ sndFileTransfers $ \sft -> async $ subscribeSndFile sft
+      rcvFileTransfers <- withStore (`getLiveRcvFileTransfers` user)
+      forConcurrently_ rcvFileTransfers $ \rft -> async $ subscribeRcvFile rft
       where
         subscribeSndFile ft@SndFileTransfer {fileId, fileStatus, agentConnId = AgentConnId cId} = do
           subscribe cId `catchError` (toView . CRSndFileSubError ft)
@@ -520,7 +534,7 @@ subscribeUserConnections user@User {userId} = void . runExceptT $ do
     subscribe cId = withAgent (`subscribeConnection` cId)
     subscribeConns conns =
       withAgent $ \a ->
-        forM_ conns $ subscribeConnection a . aConnId
+        forConcurrently_ conns $ \c -> subscribeConnection a (aConnId c)
 
 processAgentMessage :: forall m. ChatMonad m => Maybe User -> ConnId -> ACommand 'Agent -> m ()
 processAgentMessage Nothing _ _ = throwChatError CENoActiveUser

src/Simplex/Chat/Controller.hs

@@ -50,6 +50,7 @@ data ChatConfig = ChatConfig
     yesToMigrations :: Bool,
     tbqSize :: Natural,
     fileChunkSize :: Integer,
+    subscriptionEvents :: Bool,
     testView :: Bool
   }
 
@@ -186,6 +187,7 @@ data ChatResponse
   | CRContactDisconnected {contact :: Contact}
   | CRContactSubscribed {contact :: Contact}
   | CRContactSubError {contact :: Contact, chatError :: ChatError}
+  | CRContactSubSummary {contactSubscriptions :: [ContactSubStatus]}
   | CRGroupInvitation {groupInfo :: GroupInfo}
   | CRReceivedGroupInvitation {groupInfo :: GroupInfo, contact :: Contact, memberRole :: GroupMemberRole}
   | CRUserJoinedGroup {groupInfo :: GroupInfo}
@@ -199,6 +201,7 @@ data ChatResponse
   | CRGroupRemoved {groupInfo :: GroupInfo}
   | CRGroupDeleted {groupInfo :: GroupInfo, member :: GroupMember}
   | CRMemberSubError {groupInfo :: GroupInfo, contactName :: ContactName, chatError :: ChatError} -- TODO Contact?  or GroupMember?
+  | CRMemberSubErrors {memberSubErrors :: [MemberSubError]}
   | CRGroupSubscribed {groupInfo :: GroupInfo}
   | CRSndFileSubError {sndFileTransfer :: SndFileTransfer, chatError :: ChatError}
   | CRRcvFileSubError {rcvFileTransfer :: RcvFileTransfer, chatError :: ChatError}
@@ -213,6 +216,25 @@ instance ToJSON ChatResponse where
   toJSON = J.genericToJSON . sumTypeJSON $ dropPrefix "CR"
   toEncoding = J.genericToEncoding . sumTypeJSON $ dropPrefix "CR"
 
+data ContactSubStatus = ContactSubStatus
+  { contact :: Contact,
+    contactError :: Maybe ChatError
+  }
+  deriving (Show, Generic)
+
+instance ToJSON ContactSubStatus where
+  toJSON = J.genericToJSON J.defaultOptions {J.omitNothingFields = True}
+  toEncoding = J.genericToEncoding J.defaultOptions {J.omitNothingFields = True}
+
+data MemberSubError = MemberSubError
+  { member :: GroupMember,
+    memberError :: ChatError
+  }
+  deriving (Show, Generic)
+
+instance ToJSON MemberSubError where
+  toEncoding = J.genericToEncoding J.defaultOptions
+
 data ChatError
   = ChatError {errorType :: ChatErrorType}
   | ChatErrorAgent {agentError :: AgentErrorType}

src/Simplex/Chat/Mobile.hs

@@ -49,7 +49,8 @@ mobileChatOpts =
   ChatOpts
     { dbFilePrefix = "simplex_v1", -- two database files will be created: simplex_v1_chat.db and simplex_v1_agent.db
       smpServers = defaultSMPServers,
-      logging = False
+      logConnections = False,
+      logAgent = False
     }
 
 defaultMobileConfig :: ChatConfig

src/Simplex/Chat/Options.hs

@@ -21,7 +21,8 @@ import System.FilePath (combine)
 data ChatOpts = ChatOpts
   { dbFilePrefix :: String,
     smpServers :: NonEmpty SMPServer,
-    logging :: Bool
+    logConnections :: Bool,
+    logAgent :: Bool
   }
 
 defaultSMPServers :: NonEmpty SMPServer
@@ -55,9 +56,14 @@ chatOpts appDir =
           <> value defaultSMPServers
       )
     <*> switch
-      ( long "log"
+      ( long "connections"
+          <> short 'c'
+          <> help "Log every contact and group connection on start"
+      )
+    <*> switch
+      ( long "log-agent"
           <> short 'l'
-          <> help "Enable logging"
+          <> help "Enable logs from SMP agent"
       )
   where
     defaultDbFilePath = combine appDir "simplex_v1"

src/Simplex/Chat/Terminal.hs

@@ -20,7 +20,7 @@ import UnliftIO (async, waitEither_)
 
 simplexChat :: WithTerminal t => ChatConfig -> ChatOpts -> t -> IO ()
 simplexChat cfg@ChatConfig {dbPoolSize, yesToMigrations} opts t
-  | logging opts = do
+  | logAgent opts = do
     setLogLevel LogInfo -- LogError
     withGlobalLogging logCfg initRun
   | otherwise = initRun

src/Simplex/Chat/View.hs

@@ -10,7 +10,8 @@ module Simplex.Chat.View where
 import qualified Data.Aeson as J
 import Data.Function (on)
 import Data.Int (Int64)
-import Data.List (groupBy, intersperse, sortOn)
+import Data.List (groupBy, intersperse, partition, sortOn)
+import Data.Maybe (isJust)
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock (DiffTime)
@@ -101,6 +102,10 @@ responseToView testView = \case
   CRContactDisconnected c -> [ttyContact' c <> ": disconnected from server (messages will be queued)"]
   CRContactSubscribed c -> [ttyContact' c <> ": connected to server"]
   CRContactSubError c e -> [ttyContact' c <> ": contact error " <> sShow e]
+  CRContactSubSummary summary ->
+    (if null connected then [] else [sShow (length connected) <> " contacts connected (use " <> highlight' "/cs" <> " for the list)"]) <> viewErrorsSummary errors " contact errors"
+    where
+      (errors, connected) = partition (isJust . contactError) summary
   CRGroupInvitation GroupInfo {localDisplayName = ldn, groupProfile = GroupProfile {fullName}} ->
     [groupInvitation ldn fullName]
   CRReceivedGroupInvitation g c role -> viewReceivedGroupInvitation g c role
@@ -115,6 +120,7 @@ responseToView testView = \case
   CRGroupRemoved g -> [ttyFullGroup g <> ": you are no longer a member or group deleted"]
   CRGroupDeleted g m -> [ttyGroup' g <> ": " <> ttyMember m <> " deleted the group", "use " <> highlight ("/d #" <> groupName' g) <> " to delete the local copy of the group"]
   CRMemberSubError g c e -> [ttyGroup' g <> " member " <> ttyContact c <> " error: " <> sShow e]
+  CRMemberSubErrors summary -> viewErrorsSummary summary " group member errors"
   CRGroupSubscribed g -> [ttyFullGroup g <> ": connected to server(s)"]
   CRSndFileSubError SndFileTransfer {fileId, fileName} e ->
     ["sent file " <> sShow fileId <> " (" <> plain fileName <> ") error: " <> sShow e]
@@ -140,6 +146,8 @@ responseToView testView = \case
       where
         toChatView :: CChatItem c -> (Int, Text)
         toChatView (CChatItem dir ChatItem {meta}) = (msgDirectionInt $ toMsgDirection dir, itemText meta)
+    viewErrorsSummary :: [a] -> StyledString -> [StyledString]
+    viewErrorsSummary summary s = if null summary then [] else [styled (colored Red) (T.pack . show $ length summary) <> s <> " (run with -c option to show each error)"]
 
 viewChatItem :: ChatInfo c -> ChatItem c d -> [StyledString]
 viewChatItem chat (ChatItem cd meta content _) = case (chat, cd) of

stack.yaml

@@ -48,7 +48,7 @@ extra-deps:
   # - simplexmq-1.0.0@sha256:34b2004728ae396e3ae449cd090ba7410781e2b3cefc59259915f4ca5daa9ea8,8561
   # - ../simplexmq
   - github: simplex-chat/simplexmq
-    commit: dff5cad1bef67376e82c3dc15cccdb5ba9e675ab
+    commit: 7d1fdadef0541e0587d4966bc95c2930bf0f95ff
   # - terminal-0.2.0.0@sha256:de6770ecaae3197c66ac1f0db5a80cf5a5b1d3b64a66a05b50f442de5ad39570,2977
   - github: simplex-chat/aeson
     commit: 3eb66f9a68f103b5f1489382aad89f5712a64db7

tests/ChatClient.hs

@@ -43,7 +43,8 @@ opts =
   ChatOpts
     { dbFilePrefix = undefined,
       smpServers = ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"],
-      logging = False
+      logConnections = False,
+      logAgent = False
     }
 
 termSettings :: VirtualTerminalSettings