core: subscribe all queues (#6347)

* core: subscribe all queues * tests, plans, fixes * enable tests
2026-06-04 14:42:00 +00:00 · 2025-10-09 22:46:42 +01:00
parent f82b9a1a5a
commit 65e215509b
19 changed files with 232 additions and 418 deletions
@@ -155,7 +155,6 @@ data ChatConfig = ChatConfig
    cleanupManagerInterval :: NominalDiffTime,
    cleanupManagerStepDelay :: Int64,
    ciExpirationInterval :: Int64, -- microseconds
-    coreApi :: Bool,
    deliveryWorkerDelay :: Int64, -- microseconds
    deliveryBucketSize :: Int,
    highlyAvailable :: Bool,
@@ -827,12 +826,7 @@ data ChatEvent
  | CEvtContactSndReady {user :: User, contact :: Contact}
  | CEvtContactAnotherClient {user :: User, contact :: Contact}
  | CEvtSubscriptionEnd {user :: User, connectionEntity :: ConnectionEntity}
-  | CEvtContactsDisconnected {server :: SMPServer, contactRefs :: [ContactRef]}
-  | CEvtContactsSubscribed {server :: SMPServer, contactRefs :: [ContactRef]}
-  | CEvtConnSubError {user :: User, agentConnId :: AgentConnId, chatError :: ChatError}
-  | CEvtConnSubSummary {user :: User, connSubResults :: [ConnSubResult]}
-  | CEvtNetworkStatus {networkStatus :: NetworkStatus, connections :: [AgentConnId]}
-  | CEvtNetworkStatuses {user_ :: Maybe User, networkStatuses :: [ConnNetworkStatus]} -- there is the same command response
+  | CEvtNetworkStatus {server :: SMPServer, networkStatus :: NetworkStatus, connections :: [AgentConnId]}
  | CEvtHostConnected {protocol :: AProtocolType, transportHost :: TransportHost}
  | CEvtHostDisconnected {protocol :: AProtocolType, transportHost :: TransportHost}
  | CEvtReceivedGroupInvitation {user :: User, groupInfo :: GroupInfo, contact :: Contact, fromMemberRole :: GroupMemberRole, memberRole :: GroupMemberRole}
@@ -912,9 +906,7 @@ allowRemoteEvent = \case

 logEventToFile :: ChatEvent -> Bool
 logEventToFile = \case
-  CEvtContactsDisconnected {} -> True
-  CEvtContactsSubscribed {} -> True
-  CEvtConnSubError {} -> True
+  CEvtNetworkStatus {} -> True
  CEvtHostConnected {} -> True
  CEvtHostDisconnected {} -> True
  CEvtConnectionDisabled {} -> True
@@ -103,7 +103,6 @@ import Simplex.Messaging.Crypto.Ratchet (PQEncryption (..), PQSupport (..), patt
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (base64P)
 import Simplex.Messaging.Protocol (AProtoServerWithAuth (..), AProtocolType (..), MsgFlags (..), NtfServer, ProtoServerWithAuth (..), ProtocolServer, ProtocolType (..), ProtocolTypeI (..), SProtocolType (..), SubscriptionMode (..), UserProtocol, userProtocol)
-import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.ServiceScheme (ServiceScheme (..))
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport.Client (defaultSocksProxyWithAuth)
@@ -213,12 +212,8 @@ startChatController mainApp enableSndFiles = do

 subscribeUsers :: Bool -> [User] -> CM' ()
 subscribeUsers onlyNeeded users = do
-  let (us, us') = partition activeUser users
-  subscribe us
-  subscribe us'
-  where
-    subscribe :: [User] -> CM' ()
-    subscribe = mapM_ $ runExceptT . subscribeUserConnections onlyNeeded
+  let activeUserId_ = (\User {agentUserId = AgentUserId uId} -> uId) <$> find activeUser users
+  withAgent (\a -> subscribeAllConnections a onlyNeeded activeUserId_) `catchAllErrors'` eToView'

 startFilesToReceive :: [User] -> CM' ()
 startFilesToReceive users = do
@@ -4133,63 +4128,6 @@ agentSubscriber = do

 type AgentSubResult = Map ConnId (Either AgentErrorType (Maybe ClientServiceId))

-- TODO [certs rcv]
-subscribeUserConnections :: Bool -> User -> CM ()
-subscribeUserConnections onlyNeeded user = do
-  (ctConnIds, uclConnIds, memberConnIds, pendingConnIds) <-
-    withStore' $ \db -> do
-      ctConnIds <- getContactConnsToSub db user onlyNeeded
-      uclConnIds <- getUCLConnsToSub db user onlyNeeded
-      memberConnIds <- getMemberConnsToSub db user onlyNeeded
-      pendingConnIds <- getPendingConnsToSub db user onlyNeeded
-      unsetConnectionToSubscribe db user
-      pure (ctConnIds, uclConnIds, memberConnIds, pendingConnIds)
-  let allConnIds = ctConnIds <> uclConnIds <> memberConnIds <> pendingConnIds
-  rs <- withAgent $ \a -> subscribeConnections a allConnIds
-  processContactNetStatuses rs ctConnIds
-  unlessM (asks $ coreApi . config) $ notifyCLI rs allConnIds
-  where
-    processContactNetStatuses :: AgentSubResult -> [ConnId] -> CM ()
-    processContactNetStatuses rs ctConnIds = do
-      chatModifyVar connNetworkStatuses $ M.union (M.fromList statuses)
-      let networkStatuses = map (uncurry ConnNetworkStatus) statuses
-      toView $ CEvtNetworkStatuses (Just user) networkStatuses
-      where
-        statuses :: [(AgentConnId, NetworkStatus)]
-        statuses = foldr' addStatus [] ctConnIds
-          where
-            addStatus :: ConnId -> [(AgentConnId, NetworkStatus)] -> [(AgentConnId, NetworkStatus)]
-            addStatus connId nss =
-              let ns = (AgentConnId connId, netStatus $ subSuccessOrErr connId rs)
-               in ns : nss
-            netStatus :: Maybe ChatError -> NetworkStatus
-            netStatus = maybe NSConnected $ NSError . errorNetworkStatus
-            errorNetworkStatus :: ChatError -> String
-            errorNetworkStatus = \case
-              ChatErrorAgent (BROKER _ (NETWORK _)) _ -> "network"
-              ChatErrorAgent (SMP _ SMP.AUTH) _ -> "contact deleted"
-              e -> show e
-    notifyCLI :: AgentSubResult -> [ConnId] -> CM ()
-    notifyCLI rs connIds = do
-      let connSubResults = map (\(acId, err_) -> ConnSubResult (AgentConnId acId) err_) connIdsResults
-      toView $ CEvtConnSubSummary user connSubResults
-      whenM (asks $ subscriptionEvents . config) $ do
-        let connSubErrs = filterErrors connIdsResults
-        mapM_ (toView . uncurry (CEvtConnSubError user . AgentConnId)) connSubErrs
-      where
-        connIdsResults :: [(ConnId, Maybe ChatError)]
-        connIdsResults = foldr' addResult [] connIds
-          where
-            addResult :: ConnId -> [(ConnId, Maybe ChatError)] -> [(ConnId, Maybe ChatError)]
-            addResult connId idsResults = (connId, subSuccessOrErr connId rs) : idsResults
-        filterErrors :: [(ConnId, Maybe ChatError)] -> [(ConnId, ChatError)]
-        filterErrors = mapMaybe (\(a, e_) -> (a,) <$> e_)
-    subSuccessOrErr :: ConnId -> AgentSubResult -> Maybe ChatError
-    subSuccessOrErr connId rs = case M.lookup connId rs of
-      Just (Right _) -> Nothing -- success
-      Just (Left e) -> Just $ ChatErrorAgent e Nothing
-      Nothing -> Just . ChatError . CEAgentNoSubResult $ AgentConnId connId
-
 cleanupManager :: CM ()
 cleanupManager = do
  interval <- asks (cleanupManagerInterval . config)
@@ -131,23 +131,19 @@ processAgentMessageNoConn :: AEvent 'AENone -> CM ()
 processAgentMessageNoConn = \case
  CONNECT p h -> hostEvent $ CEvtHostConnected p h
  DISCONNECT p h -> hostEvent $ CEvtHostDisconnected p h
-  DOWN srv conns -> serverEvent srv conns NSDisconnected CEvtContactsDisconnected
-  UP srv conns -> serverEvent srv conns NSConnected CEvtContactsSubscribed
+  DOWN srv conns -> serverEvent srv NSDisconnected conns
+  UP srv conns -> serverEvent srv NSConnected conns
  SUSPENDED -> toView CEvtChatSuspended
  DEL_USER agentUserId -> toView $ CEvtAgentUserDeleted agentUserId
-  ERRS cErrs -> errsEvent cErrs
+  ERRS cErrs -> errsEvent $ L.toList cErrs
  where
    hostEvent :: ChatEvent -> CM ()
    hostEvent = whenM (asks $ hostEvents . config) . toView
-    serverEvent srv conns nsStatus event = do
+    serverEvent srv nsStatus conns = do
      chatModifyVar connNetworkStatuses $ \m -> foldl' (\m' cId -> M.insert cId nsStatus m') m connIds
-      ifM (asks $ coreApi . config) (notifyAPI connIds) notifyCLI
+      toView $ CEvtNetworkStatus srv nsStatus connIds
      where
        connIds = map AgentConnId conns
-        notifyAPI = toView . CEvtNetworkStatus nsStatus
-        notifyCLI = do
-          cs <- withStore' (`getConnectionsContacts` conns)
-          toView $ event srv cs
    errsEvent :: [(ConnId, AgentErrorType)] -> CM ()
    errsEvent cErrs = do
      vr <- chatVersionRange
@@ -277,7 +277,6 @@ defaultMobileConfig =
  defaultChatConfig
    { confirmMigrations = MCYesUp,
      logLevel = CLLError,
-      coreApi = True,
      deviceNameForRemote = "Mobile"
    }

@@ -378,14 +378,40 @@ USE TEMP B-TREE FOR DISTINCT
 Query: 
        SELECT DISTINCT c.conn_id, c.host, c.port, COALESCE(c.server_key_hash, s.key_hash)
        FROM commands c
+        JOIN connections cs ON c.conn_id = cs.conn_id
        LEFT JOIN servers s ON s.host = c.host AND s.port = c.port
-        ORDER BY c.conn_id
+        WHERE cs.deleted = 0
      
 Plan:
 SCAN c USING INDEX idx_commands_conn_id
+SEARCH cs USING PRIMARY KEY (conn_id=?)
 SEARCH s USING PRIMARY KEY (host=? AND port=?) LEFT-JOIN
 USE TEMP B-TREE FOR DISTINCT

+Query: 
+        SELECT DISTINCT c.user_id, q.host, q.port, COALESCE(q.server_key_hash, s.key_hash)
+        FROM rcv_queues q
+        JOIN servers s ON q.host = s.host AND q.port = s.port
+        JOIN connections c ON q.conn_id = c.conn_id
+       WHERE  c.deleted = 0 AND q.deleted = 0
+Plan:
+SCAN q USING INDEX idx_rcv_queues_link_id
+SEARCH c USING PRIMARY KEY (conn_id=?)
+SEARCH s USING PRIMARY KEY (host=? AND port=?)
+USE TEMP B-TREE FOR DISTINCT
+
+Query: 
+        SELECT DISTINCT c.user_id, q.host, q.port, COALESCE(q.server_key_hash, s.key_hash)
+        FROM rcv_queues q
+        JOIN servers s ON q.host = s.host AND q.port = s.port
+        JOIN connections c ON q.conn_id = c.conn_id
+       WHERE q.to_subscribe = 1 AND  c.deleted = 0 AND q.deleted = 0
+Plan:
+SEARCH q USING INDEX idx_rcv_queues_to_subscribe (to_subscribe=?)
+SEARCH s USING PRIMARY KEY (host=? AND port=?)
+SEARCH c USING PRIMARY KEY (conn_id=?)
+USE TEMP B-TREE FOR DISTINCT
+
 Query: 
        SELECT DISTINCT ntf_host, ntf_port, ntf_key_hash
        FROM ntf_tokens_to_delete
@@ -555,10 +581,10 @@ SEARCH messages USING COVERING INDEX idx_messages_conn_id_internal_rcv_id (conn_
 Query: 
      INSERT INTO rcv_queues
        ( host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret,
-          snd_id, queue_mode, status, rcv_queue_id, rcv_primary, replace_rcv_queue_id, smp_client_version, server_key_hash,
+          snd_id, queue_mode, status, to_subscribe, rcv_queue_id, rcv_primary, replace_rcv_queue_id, smp_client_version, server_key_hash,
          link_id, link_key, link_priv_sig_key, link_enc_fixed_data,
          ntf_public_key, ntf_private_key, ntf_id, rcv_ntf_dh_secret
-        ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?);
+        ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?);
    
 Plan:

@@ -580,16 +606,15 @@ SEARCH messages USING COVERING INDEX idx_messages_conn_id_internal_snd_id (conn_

 Query: 
      INSERT INTO snd_queues
-        (host, port, snd_id, queue_mode, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret,
+        (host, port, snd_id, queue_mode, conn_id, snd_private_key, e2e_pub_key, e2e_dh_secret,
         status, snd_queue_id, snd_primary, replace_snd_queue_id, smp_client_version, server_key_hash)
-      VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+      VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)
      ON CONFLICT (host, port, snd_id) DO UPDATE SET
        host=EXCLUDED.host,
        port=EXCLUDED.port,
        snd_id=EXCLUDED.snd_id,
        queue_mode=EXCLUDED.queue_mode,
        conn_id=EXCLUDED.conn_id,
-        snd_public_key=EXCLUDED.snd_public_key,
        snd_private_key=EXCLUDED.snd_private_key,
        e2e_pub_key=EXCLUDED.e2e_pub_key,
        e2e_dh_secret=EXCLUDED.e2e_dh_secret,
@@ -734,7 +759,28 @@ SEARCH snd_queues USING PRIMARY KEY (host=? AND port=? AND snd_id=?)
 Query: 
    SELECT
      c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.snd_id, q.queue_mode,
-      q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status,
+      q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status,
+      q.snd_queue_id, q.snd_primary, q.replace_snd_queue_id, q.switch_status, q.smp_client_version
+    FROM snd_queues q
+    JOIN servers s ON q.host = s.host AND q.port = s.port
+    JOIN connections c ON q.conn_id = c.conn_id
+   
+      JOIN (SELECT DISTINCT conn_id, snd_queue_id FROM snd_message_deliveries WHERE failed = 0) d
+      ON d.conn_id = q.conn_id AND d.snd_queue_id = q.snd_queue_id
+      WHERE c.deleted = 0
+    
+Plan:
+MATERIALIZE d
+SCAN snd_message_deliveries USING COVERING INDEX idx_snd_message_deliveries_expired
+SCAN d
+SEARCH q USING INDEX idx_snd_queue_id (conn_id=? AND snd_queue_id=?)
+SEARCH s USING PRIMARY KEY (host=? AND port=?)
+SEARCH c USING PRIMARY KEY (conn_id=?)
+
+Query: 
+    SELECT
+      c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.snd_id, q.queue_mode,
+      q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status,
      q.snd_queue_id, q.snd_primary, q.replace_snd_queue_id, q.switch_status, q.smp_client_version
    FROM snd_queues q
    JOIN servers s ON q.host = s.host AND q.port = s.port
@@ -747,7 +793,7 @@ SEARCH s USING PRIMARY KEY (host=? AND port=?)

 Query: 
    SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
-      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status,
+      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status, c.enable_ntfs,
      q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.switch_status, q.smp_client_version, q.delete_errors,
      q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret,
      q.link_id, q.link_key, q.link_priv_sig_key, q.link_enc_fixed_data
@@ -762,7 +808,7 @@ SEARCH s USING PRIMARY KEY (host=? AND port=?)

 Query: 
    SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
-      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status,
+      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status, c.enable_ntfs,
      q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.switch_status, q.smp_client_version, q.delete_errors,
      q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret,
      q.link_id, q.link_key, q.link_priv_sig_key, q.link_enc_fixed_data
@@ -777,7 +823,7 @@ SEARCH c USING PRIMARY KEY (conn_id=?)

 Query: 
    SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
-      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status,
+      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status, c.enable_ntfs,
      q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.switch_status, q.smp_client_version, q.delete_errors,
      q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret,
      q.link_id, q.link_key, q.link_priv_sig_key, q.link_enc_fixed_data
@@ -792,7 +838,7 @@ SEARCH c USING PRIMARY KEY (conn_id=?)

 Query: 
    SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
-      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status,
+      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status, c.enable_ntfs,
      q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.switch_status, q.smp_client_version, q.delete_errors,
      q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret,
      q.link_id, q.link_key, q.link_priv_sig_key, q.link_enc_fixed_data
@@ -807,7 +853,7 @@ SEARCH s USING PRIMARY KEY (host=? AND port=?)

 Query: 
    SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
-      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status,
+      q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.queue_mode, q.status, c.enable_ntfs,
      q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.switch_status, q.smp_client_version, q.delete_errors,
      q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret,
      q.link_id, q.link_key, q.link_priv_sig_key, q.link_enc_fixed_data
@@ -820,6 +866,30 @@ SEARCH q USING PRIMARY KEY (host=? AND port=? AND rcv_id=?)
 SEARCH s USING PRIMARY KEY (host=? AND port=?)
 SEARCH c USING PRIMARY KEY (conn_id=?)

+Query: 
+    SELECT c.user_id, q.conn_id, q.host, q.port, COALESCE(q.server_key_hash, s.key_hash), q.rcv_id, q.rcv_private_key, q.status, c.enable_ntfs,
+      q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id
+    FROM rcv_queues q
+    JOIN servers s ON q.host = s.host AND q.port = s.port
+    JOIN connections c ON q.conn_id = c.conn_id
+   WHERE  c.deleted = 0 AND q.deleted = 0 AND c.user_id = ? AND q.host = ? AND q.port = ?
+Plan:
+SEARCH s USING PRIMARY KEY (host=? AND port=?)
+SEARCH q USING PRIMARY KEY (host=? AND port=?)
+SEARCH c USING PRIMARY KEY (conn_id=?)
+
+Query: 
+    SELECT c.user_id, q.conn_id, q.host, q.port, COALESCE(q.server_key_hash, s.key_hash), q.rcv_id, q.rcv_private_key, q.status, c.enable_ntfs,
+      q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id
+    FROM rcv_queues q
+    JOIN servers s ON q.host = s.host AND q.port = s.port
+    JOIN connections c ON q.conn_id = c.conn_id
+   WHERE q.to_subscribe = 1 AND  c.deleted = 0 AND q.deleted = 0 AND c.user_id = ? AND q.host = ? AND q.port = ?
+Plan:
+SEARCH q USING INDEX idx_rcv_queues_to_subscribe (to_subscribe=? AND host=? AND port=?)
+SEARCH c USING PRIMARY KEY (conn_id=?)
+SEARCH s USING PRIMARY KEY (host=? AND port=?)
+
 Query: DELETE FROM commands WHERE command_id = ?
 Plan:
 SEARCH commands USING INTEGER PRIMARY KEY (rowid=?)
@@ -965,10 +1035,6 @@ Query: SELECT 1 FROM snd_message_deliveries WHERE conn_id = ? AND failed = 0 LIM
 Plan:
 SEARCH snd_message_deliveries USING COVERING INDEX idx_snd_message_deliveries_expired (conn_id=?)

-Query: SELECT DISTINCT conn_id FROM snd_message_deliveries WHERE failed = 0
-Plan:
-SCAN snd_message_deliveries USING COVERING INDEX idx_snd_message_deliveries_expired
-
 Query: SELECT conn_id FROM connections WHERE user_id = ?
 Plan:
 SEARCH connections USING COVERING INDEX idx_connections_user (user_id=?)
@@ -1129,6 +1195,10 @@ Query: UPDATE rcv_queues SET rcv_primary = ?, replace_rcv_queue_id = ? WHERE con
 Plan:
 SEARCH rcv_queues USING COVERING INDEX idx_rcv_queue_id (conn_id=? AND rcv_queue_id=?)

+Query: UPDATE rcv_queues SET to_subscribe = 0 WHERE to_subscribe = 1
+Plan:
+SEARCH rcv_queues USING COVERING INDEX idx_rcv_queues_to_subscribe (to_subscribe=?)
+
 Query: UPDATE servers_stats SET servers_stats = ?, updated_at = ? WHERE servers_stats_id = 1
 Plan:
 SEARCH servers_stats USING INTEGER PRIMARY KEY (rowid=?)
@@ -1355,20 +1355,6 @@ SEARCH cp USING COVERING INDEX idx_contact_profiles_contact_link (user_id=? AND
 SEARCH ct USING COVERING INDEX idx_contacts_contact_profile_id (contact_profile_id=?)
 SEARCH c USING COVERING INDEX idx_connections_contact_id (contact_id=?) LEFT-JOIN

-Query: 
-          SELECT ct.contact_id, c.connection_id, c.agent_conn_id, ct.local_display_name
-          FROM contacts ct
-          JOIN connections c ON c.contact_id = ct.contact_id
-          WHERE c.agent_conn_id IN (SELECT conn_id FROM temp_conn_ids)
-            AND c.conn_type = ?
-            AND ct.deleted = 0
-        
-Plan:
-SEARCH c USING INDEX sqlite_autoindex_connections_1 (agent_conn_id=?)
-LIST SUBQUERY 1
-SCAN temp_conn_ids
-SEARCH ct USING INTEGER PRIMARY KEY (rowid=?)
-
 Query: 
          SELECT g.group_id
          FROM groups g
@@ -3117,76 +3103,6 @@ Query:
 Plan:
 SEARCH chat_items USING COVERING INDEX idx_chat_items_notes (user_id=? AND note_folder_id=? AND item_status=?)

-Query: 
-        SELECT agent_conn_id
-        FROM connections
-        WHERE user_id = ?
-       
-        AND conn_type = ?
-        AND contact_id IS NULL
-        AND conn_status != ?
-      
-Plan:
-SEARCH connections USING INDEX idx_connections_contact_id (contact_id=?)
-
-Query: 
-        SELECT agent_conn_id
-        FROM connections
-        WHERE user_id = ?
-       AND to_subscribe = 1 
-        AND conn_type = ?
-        AND contact_id IS NULL
-        AND conn_status != ?
-      
-Plan:
-SEARCH connections USING INDEX idx_connections_contact_id (contact_id=?)
-
-Query: 
-        SELECT c.agent_conn_id
-        FROM connections c
-        JOIN contacts ct ON ct.contact_id = c.contact_id
-        WHERE c.user_id = ?
-       
-        AND c.conn_status != ?
-        AND ct.contact_status = ? AND ct.deleted = 0
-      
-Plan:
-SEARCH c USING INDEX idx_connections_to_subscribe (user_id=?)
-SEARCH ct USING INTEGER PRIMARY KEY (rowid=?)
-
-Query: 
-        SELECT c.agent_conn_id
-        FROM connections c
-        JOIN contacts ct ON ct.contact_id = c.contact_id
-        WHERE c.user_id = ?
-       AND c.to_subscribe = 1 
-        AND c.conn_status != ?
-        AND ct.contact_status = ? AND ct.deleted = 0
-      
-Plan:
-SEARCH c USING INDEX idx_connections_to_subscribe (user_id=? AND to_subscribe=?)
-SEARCH ct USING INTEGER PRIMARY KEY (rowid=?)
-
-Query: 
-        SELECT c.agent_conn_id
-        FROM connections c
-        JOIN user_contact_links ucl ON ucl.user_contact_link_id = c.user_contact_link_id
-        WHERE c.user_id = ?
-        AND c.conn_status != ?
-Plan:
-SEARCH c USING INDEX idx_connections_to_subscribe (user_id=?)
-SEARCH ucl USING INTEGER PRIMARY KEY (rowid=?)
-
-Query: 
-        SELECT c.agent_conn_id
-        FROM connections c
-        JOIN user_contact_links ucl ON ucl.user_contact_link_id = c.user_contact_link_id
-        WHERE c.user_id = ?
-       AND c.to_subscribe = 1  AND c.conn_status != ?
-Plan:
-SEARCH c USING INDEX idx_connections_to_subscribe (user_id=? AND to_subscribe=?)
-SEARCH ucl USING INTEGER PRIMARY KEY (rowid=?)
-
 Query: 
        SELECT c.connection_id, c.agent_conn_id, c.conn_level, c.via_contact, c.via_user_contact_link, c.via_group_link, c.group_link_id, c.xcontact_id, c.custom_user_profile_id,
          c.conn_status, c.conn_type, c.contact_conn_initiated, c.local_alias, c.contact_id, c.group_member_id, c.user_contact_link_id,
@@ -3837,56 +3753,6 @@ Query:
 Plan:
 SEARCH groups USING INTEGER PRIMARY KEY (rowid=?)

-Query: 
-        WITH user_groups AS MATERIALIZED (
-          SELECT g.group_id
-          FROM groups g
-          JOIN group_members mu ON mu.group_id = g.group_id
-          WHERE g.user_id = ?
-            AND mu.contact_id = ?
-            AND mu.member_status NOT IN (?,?,?)
-        )
-        SELECT c.agent_conn_id
-        FROM connections c
-        JOIN group_members m ON m.group_member_id = c.group_member_id
-        JOIN user_groups ug ON ug.group_id = m.group_id
-        WHERE c.user_id = ?
-          AND c.conn_status != ?
-          AND m.member_status NOT IN (?,?,?)
-      
-Plan:
-MATERIALIZE user_groups
-SEARCH mu USING INDEX idx_group_members_contact_id (contact_id=?)
-SEARCH g USING INTEGER PRIMARY KEY (rowid=?)
-SEARCH c USING INDEX idx_connections_to_subscribe (user_id=?)
-SEARCH m USING INTEGER PRIMARY KEY (rowid=?)
-SEARCH ug USING AUTOMATIC COVERING INDEX (group_id=?)
-
-Query: 
-        WITH user_groups AS MATERIALIZED (
-          SELECT g.group_id
-          FROM groups g
-          JOIN group_members mu ON mu.group_id = g.group_id
-          WHERE g.user_id = ?
-            AND mu.contact_id = ?
-            AND mu.member_status NOT IN (?,?,?)
-        )
-        SELECT c.agent_conn_id
-        FROM connections c
-        JOIN group_members m ON m.group_member_id = c.group_member_id
-        JOIN user_groups ug ON ug.group_id = m.group_id
-        WHERE c.user_id = ?
-          AND c.conn_status != ?
-          AND m.member_status NOT IN (?,?,?)
-       AND c.to_subscribe = 1
-Plan:
-MATERIALIZE user_groups
-SEARCH mu USING INDEX idx_group_members_contact_id (contact_id=?)
-SEARCH g USING INTEGER PRIMARY KEY (rowid=?)
-SEARCH c USING INDEX idx_connections_to_subscribe (user_id=? AND to_subscribe=?)
-SEARCH m USING INTEGER PRIMARY KEY (rowid=?)
-SEARCH ug USING AUTOMATIC COVERING INDEX (group_id=?)
-
 Query: 
      DELETE FROM chat_items
      WHERE group_scope_group_member_id = ?
@@ -5133,7 +4999,7 @@ Query:
    FROM group_members m
    JOIN contact_profiles p ON p.contact_profile_id = COALESCE(m.member_profile_id, m.contact_profile_id)
    LEFT JOIN connections c ON c.group_member_id = m.group_member_id
-  
+   
                WHERE m.group_id = ? AND m.user_id = ? AND (m.contact_id IS NULL OR m.contact_id != ?)
                  AND m.member_status IN (?, ?, ?, ?)
                  AND m.group_member_id NOT IN (
@@ -5575,9 +5441,6 @@ Query:  SELECT chat_item_id FROM chat_items WHERE  user_id = ? AND group_id = ?
 Plan:
 SEARCH chat_items USING COVERING INDEX idx_chat_items_groups_msg_content_tag_item_ts (user_id=? AND group_id=? AND msg_content_tag=?)

-Query: CREATE TABLE temp_conn_ids (conn_id BLOB)
-Error: SQLite3 returned ErrorError while attempting to perform prepare "explain query plan CREATE TABLE temp_conn_ids (conn_id BLOB)": table temp_conn_ids already exists
-
 Query: CREATE TABLE temp_delete_members (contact_profile_id INTEGER, member_profile_id INTEGER, local_display_name TEXT)
 Error: SQLite3 returned ErrorError while attempting to perform prepare "explain query plan CREATE TABLE temp_delete_members (contact_profile_id INTEGER, member_profile_id INTEGER, local_display_name TEXT)": table temp_delete_members already exists

@@ -5978,15 +5841,9 @@ SEARCH contacts USING COVERING INDEX sqlite_autoindex_contacts_2 (user_id=?)
 SEARCH display_names USING COVERING INDEX sqlite_autoindex_display_names_2 (user_id=?)
 SEARCH contact_profiles USING COVERING INDEX idx_contact_profiles_user_id (user_id=?)

-Query: DROP TABLE IF EXISTS temp_conn_ids
-Plan:
-
 Query: DROP TABLE IF EXISTS temp_delete_members
 Plan:

-Query: DROP TABLE temp_conn_ids
-Plan:
-
 Query: DROP TABLE temp_delete_members
 Plan:

@@ -6083,9 +5940,6 @@ Query: INSERT INTO snd_files (file_id, file_status, file_descr_id, group_member_
 Plan:
 SEARCH connections USING INTEGER PRIMARY KEY (rowid=?)

-Query: INSERT INTO temp_conn_ids (conn_id) VALUES (?)
-Plan:
-
 Query: INSERT INTO user_contact_links (user_id, conn_req_contact, short_link_contact, short_link_data_set, short_link_large_data_set, created_at, updated_at) VALUES (?,?,?,?,?,?,?)
 Plan:

@@ -6396,10 +6250,6 @@ Query: UPDATE connections SET security_code = ?, security_code_verified_at = ?,
 Plan:
 SEARCH connections USING INTEGER PRIMARY KEY (rowid=?)

-Query: UPDATE connections SET to_subscribe = 0 WHERE user_id = ? AND to_subscribe = 1
-Plan:
-SEARCH connections USING INDEX idx_connections_to_subscribe (user_id=? AND to_subscribe=?)
-
 Query: UPDATE contact_requests SET business_group_id = ? WHERE contact_request_id = ?
 Plan:
 SEARCH contact_requests USING INTEGER PRIMARY KEY (rowid=?)
@@ -33,11 +33,9 @@ import Simplex.Chat.Styled
 import Simplex.Chat.Terminal.Notification (Notification (..), initializeNotifications)
 import Simplex.Chat.Types
 import Simplex.Chat.View
-import Simplex.Messaging.Agent.Protocol
-import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (safeDecodeUtf8, tshow)
+import Simplex.Messaging.Util (tshow)
 import System.Console.ANSI.Types
 import System.IO (IOMode (..), hPutStrLn, withFile)
 import System.Mem.Weak (Weak)
@@ -196,8 +194,6 @@ chatEventNotification t@ChatTerminal {sendNotification} cc = \case
  CEvtContactAnotherClient u ct -> do
    whenCurrUser cc u $ unsetActiveContact t ct
    when (contactNtf u ct False) $ sendNtf (viewContactName ct <> "> ", "connected to another client")
-  CEvtContactsDisconnected srv _ -> serverNtf srv "disconnected"
-  CEvtContactsSubscribed srv _ -> serverNtf srv "connected"
  CEvtReceivedGroupInvitation u g ct _ _ ->
    when (contactNtf u ct False) $
      sendNtf ("#" <> viewGroupName g <> " " <> viewContactName ct <> "> ", "invited you to join the group")
@@ -215,7 +211,6 @@ chatEventNotification t@ChatTerminal {sendNotification} cc = \case
  _ -> pure ()
  where
    sendNtf = maybe (\_ -> pure ()) (. uncurry Notification) sendNotification
-    serverNtf (SMPServer host _ _) str = sendNtf ("server " <> str, safeDecodeUtf8 $ strEncode host)

 msgText :: MsgContent -> Maybe MarkdownList -> Text
 msgText (MCFile _) _ = "wants to send a file"
@@ -1813,7 +1813,7 @@ data NetworkStatus
 netStatusStr :: NetworkStatus -> String
 netStatusStr = \case
  NSUnknown -> "unknown"
-  NSConnected -> "connected"
+  NSConnected -> "subscribed"
  NSDisconnected -> "disconnected"
  NSError e -> "error: " <> e

@@ -22,7 +22,7 @@ import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Char (isSpace, toUpper)
 import Data.Function (on)
 import Data.Int (Int64)
-import Data.List (groupBy, intercalate, intersperse, partition, sortOn)
+import Data.List (groupBy, intercalate, intersperse, sortOn)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
@@ -384,7 +384,7 @@ contactList :: [ContactRef] -> String
 contactList cs = T.unpack . T.intercalate ", " $ map (\ContactRef {localDisplayName = n} -> "@" <> n) cs

 chatEventToView :: (Maybe RemoteHostId, Maybe User) -> ChatConfig -> Bool -> CurrentTime -> TimeZone -> Maybe RemoteHostId -> ChatEvent -> [StyledString]
-chatEventToView hu ChatConfig {logLevel, showReactions, showReceipts, testView, coreApi} liveItems ts tz outputRH = \case
+chatEventToView hu ChatConfig {logLevel, showReactions, showReceipts, testView} liveItems ts tz outputRH = \case
  CEvtChatSuspended -> ["chat suspended"]
  CEvtContactSwitch u ct progress -> ttyUser u $ viewContactSwitch ct progress
  CEvtGroupMemberSwitch u g m progress -> ttyUser u $ viewGroupMemberSwitch g m progress
@@ -452,15 +452,7 @@ chatEventToView hu ChatConfig {logLevel, showReactions, showReceipts, testView,
  CEvtSubscriptionEnd u acEntity ->
    let Connection {connId} = entityConnection acEntity
     in ttyUser u [sShow connId <> ": END"]
-  CEvtContactsDisconnected srv cs -> [plain $ "server disconnected " <> showSMPServer srv <> " (" <> contactList cs <> ")"]
-  CEvtContactsSubscribed srv cs -> [plain $ "server connected " <> showSMPServer srv <> " (" <> contactList cs <> ")"]
-  CEvtConnSubError u connId e -> ttyUser u ["conn ID " <> sShow connId <> ": subscription error " <> sShow e]
-  CEvtConnSubSummary u connSubResults ->
-    ttyUser u $ [sShow (length subscribed) <> " connections subscribed" | not (null subscribed)] <> viewSubErrorsSummary errs
-    where
-      (errs, subscribed) = partition (isJust . connSubError) connSubResults
-  CEvtNetworkStatus status conns -> if testView && coreApi then [plain $ show (length conns) <> " connections " <> netStatusStr status] else []
-  CEvtNetworkStatuses u statuses -> if testView && coreApi then ttyUser' u $ viewNetworkStatuses statuses else []
+  CEvtNetworkStatus srv status conns -> [plain $ netStatusStr status <> " " <> show (length conns) <> " connections on server " <> showSMPServer srv]
  CEvtReceivedGroupInvitation {user = u, groupInfo = g, contact = c, memberRole = r} -> ttyUser u $ viewReceivedGroupInvitation g c r
  CEvtUserJoinedGroup u g _ -> ttyUser u $ viewUserJoinedGroup g
  CEvtJoinedGroupMember u g m -> ttyUser u $ viewJoinedGroupMember g m