core: sync connections (#6345)

* core: sync subsbriptions

* sha map

* sync event

* should delete flag

* wip

* schema

* wip

* delete

* update simplexmq

* plans

* rename, adapt

* update

* plans

* option to show ids

* plans

* fix, test

* more tests

* postgres

* plans

* cleanup

* plans

---------

Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>

bots/src/API/Docs/Commands.hs

@@ -440,6 +440,7 @@ undocumentedCommands =
     "ReconnectAllServers",
     "ReconnectServer",
     "ResetAgentServersStats",
+    "ShowConnectionsDiff",
     "ResubscribeAllConnections",
     "SetAllContactReceipts",
     "SetChatItemTTL",

bots/src/API/Docs/Events.hs

@@ -199,6 +199,7 @@ undocumentedEvents =
     "CEvtSndFileRedirectStartXFTP",
     "CEvtSndFileStart", -- legacy SMP files
     "CEvtSndStandaloneFileComplete",
+    "CEvtConnectionsDiff",
     "CEvtSubscriptionEnd",
     "CEvtTerminalEvent",
     "CEvtTimedAction",

bots/src/API/Docs/Responses.hs

@@ -130,6 +130,7 @@ undocumentedResponses =
     "CRChats",
     "CRChatStarted",
     "CRChatStopped",
+    "CRConnectionsDiff",
     "CRChatTags",
     "CRConnectionAliasUpdated",
     "CRConnectionIncognitoUpdated",

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: 80aa56cbcce92f4b61cda06a965eef3d0f640df1
+    tag: 1dbc15b2e6225c0e254564747bc8412970273e85
 
 source-repository-package
     type: git

scripts/nix/sha256map.nix

@@ -1,5 +1,5 @@
 {
-  "https://github.com/simplex-chat/simplexmq.git"."a3d1a72eb06df0dc6e1f2a8d72cab8535870fb03" = "16pmac5l4fwjhi6msf3y9dljc51mbfhvql4kv1sqn6gkbf9lnmpx";
+  "https://github.com/simplex-chat/simplexmq.git"."1dbc15b2e6225c0e254564747bc8412970273e85" = "03hmlynixssyp0720h984slw4lkrzn3kr63k3mah50lbyxzsmnrs";
   "https://github.com/simplex-chat/hs-socks.git"."a30cc7a79a08d8108316094f8f2f82a0c5e1ac51" = "0yasvnr7g91k76mjkamvzab2kvlb1g5pspjyjn2fr6v83swjhj38";
   "https://github.com/simplex-chat/direct-sqlcipher.git"."f814ee68b16a9447fbb467ccc8f29bdd3546bfd9" = "1ql13f4kfwkbaq7nygkxgw84213i0zm7c1a8hwvramayxl38dq5d";
   "https://github.com/simplex-chat/sqlcipher-simple.git"."a46bd361a19376c5211f1058908fc0ae6bf42446" = "1z0r78d8f0812kxbgsm735qf6xx8lvaz27k1a0b4a2m0sshpd5gl";

simplex-chat.cabal

@@ -120,6 +120,7 @@ library
           Simplex.Chat.Store.Postgres.Migrations.M20250813_delivery_tasks
           Simplex.Chat.Store.Postgres.Migrations.M20250919_group_summary
           Simplex.Chat.Store.Postgres.Migrations.M20250922_remove_unused_connections
+          Simplex.Chat.Store.Postgres.Migrations.M20251007_connections_sync
   else
       exposed-modules:
           Simplex.Chat.Archive
@@ -263,6 +264,7 @@ library
           Simplex.Chat.Store.SQLite.Migrations.M20250813_delivery_tasks
           Simplex.Chat.Store.SQLite.Migrations.M20250919_group_summary
           Simplex.Chat.Store.SQLite.Migrations.M20250922_remove_unused_connections
+          Simplex.Chat.Store.SQLite.Migrations.M20251007_connections_sync
   other-modules:
       Paths_simplex_chat
   hs-source-dirs:

src/Simplex/Chat/Controller.hs

@@ -69,7 +69,7 @@ import Simplex.Chat.Types.Shared
 import Simplex.Chat.Types.UITheme
 import Simplex.Chat.Util (liftIOEither)
 import Simplex.FileTransfer.Description (FileDescriptionURI)
-import Simplex.Messaging.Agent (AgentClient, SubscriptionsInfo)
+import Simplex.Messaging.Agent (AgentClient, DatabaseDiff, SubscriptionsInfo)
 import Simplex.Messaging.Agent.Client (AgentLocks, AgentQueuesInfo (..), AgentWorkersDetails (..), AgentWorkersSummary (..), ProtocolTestFailure, SMPServerSubs, ServerQueueInfo, UserNetworkInfo)
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, NetworkConfig, ServerCfg, Worker)
 import Simplex.Messaging.Agent.Lock
@@ -292,6 +292,7 @@ data ChatCommand
   | APIStopChat
   | APIActivateChat {restoreChat :: Bool}
   | APISuspendChat {suspendTimeout :: Int}
+  | ShowConnectionsDiff Bool
   | ResubscribeAllConnections
   | SetTempFolder FilePath
   | SetFilesFolder FilePath
@@ -633,6 +634,7 @@ data ChatResponse
   | CRChatStarted
   | CRChatRunning
   | CRChatStopped
+  | CRConnectionsDiff {showIds :: Bool, userIds :: DatabaseDiff AgentUserId, connIds :: DatabaseDiff AgentConnId}
   | CRApiChats {user :: User, chats :: [AChat]}
   | CRChats {chats :: [AChat]}
   | CRApiChat {user :: User, chat :: AChat, navInfo :: Maybe NavigationInfo}
@@ -825,6 +827,7 @@ data ChatEvent
   | CEvtContactConnected {user :: User, contact :: Contact, userCustomProfile :: Maybe Profile}
   | CEvtContactSndReady {user :: User, contact :: Contact}
   | CEvtContactAnotherClient {user :: User, contact :: Contact}
+  | CEvtConnectionsDiff {userIds :: DatabaseDiff AgentUserId, connIds :: DatabaseDiff AgentConnId}
   | CEvtSubscriptionEnd {user :: User, connectionEntity :: ConnectionEntity}
   | CEvtNetworkStatus {server :: SMPServer, networkStatus :: NetworkStatus, connections :: [AgentConnId]}
   | CEvtHostConnected {protocol :: AProtocolType, transportHost :: TransportHost}

src/Simplex/Chat/Library/Commands.hs

@@ -164,10 +164,20 @@ startChatController mainApp enableSndFiles = do
   asks smpAgent >>= liftIO . resumeAgentClient
   unless mainApp $ chatWriteVar' subscriptionMode SMOnlyCreate
   users <- fromRight [] <$> runExceptT (withFastStore' getUsers)
+  runExceptT (syncConnections' users) >>= \case
+    Left e -> liftIO $ putStrLn $ "Error synchronizing connections: " <> show e
+    Right _ -> pure ()
   restoreCalls
   s <- asks agentAsync
   readTVarIO s >>= maybe (start s users) (pure . fst)
   where
+    syncConnections' users =
+      whenM (withFastStore' shouldSyncConnections) $ do
+        let aUserIds = map aUserId users
+        connIds <- concat <$> forM users getConnsToSub
+        (userDiff, connDiff) <- withAgent (\a -> syncConnections a aUserIds connIds)
+        withFastStore' setConnectionsSyncTs
+        toView $ CEvtConnectionsDiff (AgentUserId <$> userDiff) (AgentConnId <$> connDiff)
     start s users = do
       a1 <- async agentSubscriber
       a2 <-
@@ -210,6 +220,15 @@ startChatController mainApp enableSndFiles = do
             ttlCount <- getChatTTLCount db user
             pure $ ttl > 0 || ttlCount > 0
 
+getConnsToSub :: User -> CM [ConnId]
+getConnsToSub user =
+  withFastStore' $ \db -> do
+    ctConnIds <- getContactConnsToSub db user False
+    uclConnIds <- getUCLConnsToSub db user False
+    memberConnIds <- getMemberConnsToSub db user False
+    pendingConnIds <- getPendingConnsToSub db user False
+    pure $ ctConnIds <> uclConnIds <> memberConnIds <> pendingConnIds
+
 subscribeUsers :: Bool -> [User] -> CM' ()
 subscribeUsers onlyNeeded users = do
   let activeUserId_ = (\User {agentUserId = AgentUserId uId} -> uId) <$> find activeUser users
@@ -443,6 +462,12 @@ processChatCommand vr nm = \case
     stopRemoteCtrl
     lift $ withAgent' (`suspendAgent` t)
     ok_
+  ShowConnectionsDiff showIds -> do
+    users <- withFastStore' getUsers
+    let aUserIds = map aUserId users
+    connIds <- concat <$> forM users getConnsToSub
+    (userDiff, connDiff) <- withAgent (\a -> compareConnections a aUserIds connIds)
+    pure $ CRConnectionsDiff showIds (AgentUserId <$> userDiff) (AgentConnId <$> connDiff)
   ResubscribeAllConnections -> withStore' getUsers >>= lift . subscribeUsers False >> ok_
   -- has to be called before StartChat
   SetTempFolder tf -> do
@@ -4291,6 +4316,7 @@ chatCommandP =
       "/_app activate restore=" *> (APIActivateChat <$> onOffP),
       "/_app activate" $> APIActivateChat True,
       "/_app suspend " *> (APISuspendChat <$> A.decimal),
+      "/_connections diff" *> (ShowConnectionsDiff <$> (" show_ids=" *> onOffP <|> pure False)),
       "/_resubscribe all" $> ResubscribeAllConnections,
       -- deprecated, use /set file paths
       "/_temp_folder " *> (SetTempFolder <$> filePath),

src/Simplex/Chat/Store/Connections.hs

@@ -19,7 +19,8 @@ module Simplex.Chat.Store.Connections
     getUCLConnsToSub,
     getMemberConnsToSub,
     getPendingConnsToSub,
-    unsetConnectionToSubscribe,
+    shouldSyncConnections,
+    setConnectionsSyncTs,
   )
 where
 
@@ -28,13 +29,14 @@ import Control.Monad.IO.Class
 import Data.Bitraversable (bitraverse)
 import Data.Int (Int64)
 import Data.Maybe (fromMaybe)
+import Data.Time.Clock (getCurrentTime)
 import Simplex.Chat.Protocol
 import Simplex.Chat.Store.Direct
 import Simplex.Chat.Store.Groups
 import Simplex.Chat.Store.Shared
 import Simplex.Chat.Types
 import Simplex.Messaging.Agent.Protocol (ConnId)
-import Simplex.Messaging.Agent.Store.AgentStore (firstRow, firstRow', maybeFirstRow)
+import Simplex.Messaging.Agent.Store.AgentStore (firstRow, firstRow', fromOnlyBI, maybeFirstRow)
 import Simplex.Messaging.Agent.Store.DB (BoolInt (..))
 import qualified Simplex.Messaging.Agent.Store.DB as DB
 import Simplex.Messaging.Util (eitherToMaybe)
@@ -321,9 +323,21 @@ getPendingConnsToSub db User {userId} filterToSubscribe =
         AND conn_status != ?
       |]
 
-unsetConnectionToSubscribe :: DB.Connection -> User -> IO ()
-unsetConnectionToSubscribe db User {userId} =
+shouldSyncConnections :: DB.Connection -> IO Bool
+shouldSyncConnections db =
+  fromOnlyBI . head
+    <$> DB.query_
+      db
+      "SELECT should_sync FROM connections_sync WHERE connections_sync_id = 1"
+
+setConnectionsSyncTs :: DB.Connection -> IO ()
+setConnectionsSyncTs db = do
+  currentTs <- getCurrentTime
   DB.execute
     db
-    "UPDATE connections SET to_subscribe = 0 WHERE user_id = ? AND to_subscribe = 1"
-    (Only userId)
+    [sql|
+      UPDATE connections_sync
+      SET should_sync = 0, last_sync_ts = ?
+      WHERE connections_sync_id = 1
+    |]
+    (Only currentTs)

src/Simplex/Chat/Store/Postgres/Migrations.hs

@@ -19,6 +19,7 @@ import Simplex.Chat.Store.Postgres.Migrations.M20250802_chat_peer_type
 import Simplex.Chat.Store.Postgres.Migrations.M20250813_delivery_tasks
 import Simplex.Chat.Store.Postgres.Migrations.M20250919_group_summary
 import Simplex.Chat.Store.Postgres.Migrations.M20250922_remove_unused_connections
+import Simplex.Chat.Store.Postgres.Migrations.M20251007_connections_sync
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Text, Maybe Text)]
@@ -37,7 +38,8 @@ schemaMigrations =
     ("20250802_chat_peer_type", m20250802_chat_peer_type, Just down_m20250802_chat_peer_type),
     ("20250813_delivery_tasks", m20250813_delivery_tasks, Just down_m20250813_delivery_tasks),
     ("20250919_group_summary", m20250919_group_summary, Just down_m20250919_group_summary),
-    ("20250922_remove_unused_connections", m20250922_remove_unused_connections, Just down_m20250922_remove_unused_connections)
+    ("20250922_remove_unused_connections", m20250922_remove_unused_connections, Just down_m20250922_remove_unused_connections),
+    ("20251007_connections_sync", m20251007_connections_sync, Just down_m20251007_connections_sync)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Chat/Store/Postgres/Migrations/M20251007_connections_sync.hs

@@ -0,0 +1,27 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Chat.Store.Postgres.Migrations.M20251007_connections_sync where
+
+import Data.Text (Text)
+import qualified Data.Text as T
+import Text.RawString.QQ (r)
+
+m20251007_connections_sync :: Text
+m20251007_connections_sync =
+  T.pack
+    [r|
+CREATE TABLE connections_sync(
+  connections_sync_id BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY,
+  should_sync SMALLINT NOT NULL DEFAULT 0,
+  last_sync_ts TIMESTAMPTZ
+);
+
+INSERT INTO connections_sync (connections_sync_id, should_sync, last_sync_ts) VALUES (1,0,NULL);
+|]
+
+down_m20251007_connections_sync :: Text
+down_m20251007_connections_sync =
+  T.pack
+    [r|
+DROP TABLE connections_sync;
+|]

src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql

@@ -390,6 +390,25 @@ ALTER TABLE test_chat_schema.connections ALTER COLUMN connection_id ADD GENERATE
 
 
 
+CREATE TABLE test_chat_schema.connections_sync (
+    connections_sync_id bigint NOT NULL,
+    should_sync smallint DEFAULT 0 NOT NULL,
+    last_sync_ts timestamp with time zone
+);
+
+
+
+ALTER TABLE test_chat_schema.connections_sync ALTER COLUMN connections_sync_id ADD GENERATED BY DEFAULT AS IDENTITY (
+    SEQUENCE NAME test_chat_schema.connections_sync_connections_sync_id_seq
+    START WITH 1
+    INCREMENT BY 1
+    NO MINVALUE
+    NO MAXVALUE
+    CACHE 1
+);
+
+
+
 CREATE TABLE test_chat_schema.contact_profiles (
     contact_profile_id bigint NOT NULL,
     display_name text NOT NULL,
@@ -1358,6 +1377,11 @@ ALTER TABLE ONLY test_chat_schema.connections
 
 
 
+ALTER TABLE ONLY test_chat_schema.connections_sync
+    ADD CONSTRAINT connections_sync_pkey PRIMARY KEY (connections_sync_id);
+
+
+
 ALTER TABLE ONLY test_chat_schema.contact_profiles
     ADD CONSTRAINT contact_profiles_pkey PRIMARY KEY (contact_profile_id);
 

src/Simplex/Chat/Store/SQLite/Migrations.hs

@@ -142,6 +142,7 @@ import Simplex.Chat.Store.SQLite.Migrations.M20250802_chat_peer_type
 import Simplex.Chat.Store.SQLite.Migrations.M20250813_delivery_tasks
 import Simplex.Chat.Store.SQLite.Migrations.M20250919_group_summary
 import Simplex.Chat.Store.SQLite.Migrations.M20250922_remove_unused_connections
+import Simplex.Chat.Store.SQLite.Migrations.M20251007_connections_sync
 import Simplex.Messaging.Agent.Store.Shared (Migration (..))
 
 schemaMigrations :: [(String, Query, Maybe Query)]
@@ -283,7 +284,8 @@ schemaMigrations =
     ("20250802_chat_peer_type", m20250802_chat_peer_type, Just down_m20250802_chat_peer_type),
     ("20250813_delivery_tasks", m20250813_delivery_tasks, Just down_m20250813_delivery_tasks),
     ("20250919_group_summary", m20250919_group_summary, Just down_m20250919_group_summary),
-    ("20250922_remove_unused_connections", m20250922_remove_unused_connections, Just down_m20250922_remove_unused_connections)
+    ("20250922_remove_unused_connections", m20250922_remove_unused_connections, Just down_m20250922_remove_unused_connections),
+    ("20251007_connections_sync", m20251007_connections_sync, Just down_m20251007_connections_sync)
   ]
 
 -- | The list of migrations in ascending order by date

src/Simplex/Chat/Store/SQLite/Migrations/M20251007_connections_sync.hs

@@ -0,0 +1,25 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Chat.Store.SQLite.Migrations.M20251007_connections_sync where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+-- should_sync should be set manually when required
+m20251007_connections_sync :: Query
+m20251007_connections_sync =
+  [sql|
+CREATE TABLE connections_sync(
+  connections_sync_id INTEGER PRIMARY KEY AUTOINCREMENT,
+  should_sync INTEGER NOT NULL DEFAULT 0,
+  last_sync_ts TEXT
+);
+
+INSERT INTO connections_sync (connections_sync_id, should_sync, last_sync_ts) VALUES (1,0,NULL);
+|]
+
+down_m20251007_connections_sync :: Query
+down_m20251007_connections_sync =
+  [sql|
+DROP TABLE connections_sync;
+|]

src/Simplex/Chat/Store/SQLite/Migrations/agent_query_plans.txt

@@ -912,6 +912,20 @@ SEARCH messages USING COVERING INDEX idx_messages_conn_id (conn_id=?)
 SEARCH snd_queues USING COVERING INDEX idx_snd_queue_id (conn_id=?)
 SEARCH rcv_queues USING COVERING INDEX idx_rcv_queue_id (conn_id=?)
 
+Query: DELETE FROM connections WHERE user_id = 2
+Plan:
+SEARCH connections USING COVERING INDEX idx_connections_user (user_id=?)
+SEARCH processed_ratchet_key_hashes USING COVERING INDEX idx_processed_ratchet_key_hashes_hash (conn_id=?)
+SEARCH encrypted_rcv_message_hashes USING COVERING INDEX idx_encrypted_rcv_message_hashes_hash (conn_id=?)
+SEARCH snd_message_deliveries USING COVERING INDEX idx_snd_message_deliveries_conn_id_internal_id (conn_id=?)
+SEARCH commands USING COVERING INDEX idx_commands_conn_id (conn_id=?)
+SEARCH ratchets USING PRIMARY KEY (conn_id=?)
+SEARCH conn_invitations USING COVERING INDEX idx_conn_invitations_contact_conn_id (contact_conn_id=?)
+SEARCH conn_confirmations USING COVERING INDEX idx_conn_confirmations_conn_id (conn_id=?)
+SEARCH messages USING COVERING INDEX idx_messages_conn_id (conn_id=?)
+SEARCH snd_queues USING COVERING INDEX idx_snd_queue_id (conn_id=?)
+SEARCH rcv_queues USING COVERING INDEX idx_rcv_queue_id (conn_id=?)
+
 Query: DELETE FROM deleted_snd_chunk_replicas WHERE deleted_snd_chunk_replica_id = ?
 Plan:
 SEARCH deleted_snd_chunk_replicas USING INTEGER PRIMARY KEY (rowid=?)
@@ -963,6 +977,14 @@ Query: DELETE FROM snd_queues WHERE conn_id = ? AND snd_queue_id = ?
 Plan:
 SEARCH snd_queues USING COVERING INDEX idx_snd_queue_id (conn_id=? AND snd_queue_id=?)
 
+Query: DELETE FROM users WHERE user_id = 2
+Plan:
+SEARCH users USING INTEGER PRIMARY KEY (rowid=?)
+SEARCH deleted_snd_chunk_replicas USING COVERING INDEX idx_deleted_snd_chunk_replicas_user_id (user_id=?)
+SEARCH snd_files USING COVERING INDEX idx_snd_files_user_id (user_id=?)
+SEARCH rcv_files USING COVERING INDEX idx_rcv_files_user_id (user_id=?)
+SEARCH connections USING COVERING INDEX idx_connections_user (user_id=?)
+
 Query: DELETE FROM users WHERE user_id = ?
 Plan:
 SEARCH users USING INTEGER PRIMARY KEY (rowid=?)
@@ -1035,14 +1057,26 @@ 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 conn_id FROM connections WHERE deleted = 0
+Plan:
+SCAN connections
+
 Query: SELECT conn_id FROM connections WHERE user_id = ?
 Plan:
 SEARCH connections USING COVERING INDEX idx_connections_user (user_id=?)
 
+Query: SELECT count(1) FROM connections
+Plan:
+SCAN connections USING COVERING INDEX idx_connections_user
+
 Query: SELECT count(1) FROM snd_message_bodies
 Plan:
 SCAN snd_message_bodies
 
+Query: SELECT count(1) FROM users
+Plan:
+SCAN users
+
 Query: SELECT deleted FROM snd_files WHERE snd_file_id = ?
 Plan:
 SEARCH snd_files USING INTEGER PRIMARY KEY (rowid=?)
@@ -1103,6 +1137,10 @@ Query: SELECT started_at, servers_stats FROM servers_stats WHERE servers_stats_i
 Plan:
 SEARCH servers_stats USING INTEGER PRIMARY KEY (rowid=?)
 
+Query: SELECT user_id FROM users WHERE deleted = 0
+Plan:
+SCAN users
+
 Query: SELECT user_id FROM users WHERE user_id = ? AND deleted = ?
 Plan:
 SEARCH users USING INTEGER PRIMARY KEY (rowid=?)

src/Simplex/Chat/Store/SQLite/Migrations/chat_query_plans.txt

@@ -3103,6 +3103,41 @@ 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 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 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.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,
@@ -3753,6 +3788,31 @@ 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: 
       DELETE FROM chat_items
       WHERE group_scope_group_member_id = ?
@@ -4581,6 +4641,14 @@ Query:
 Plan:
 SEARCH connections USING INTEGER PRIMARY KEY (rowid=?)
 
+Query: 
+      UPDATE connections_sync
+      SET should_sync = 0, last_sync_ts = ?
+      WHERE connections_sync_id = 1
+    
+Plan:
+SEARCH connections_sync USING INTEGER PRIMARY KEY (rowid=?)
+
 Query: 
       UPDATE contact_profiles
       SET contact_link = ?, updated_at = ?
@@ -5571,6 +5639,17 @@ Query: DELETE FROM commands WHERE user_id = ? AND command_id = ?
 Plan:
 SEARCH commands USING INTEGER PRIMARY KEY (rowid=?)
 
+Query: DELETE FROM connections WHERE contact_id = (SELECT contact_id FROM contacts WHERE local_display_name = 'cath')
+Plan:
+SEARCH connections USING INDEX idx_connections_contact_id (contact_id=?)
+SCALAR SUBQUERY 1
+SCAN contacts USING COVERING INDEX sqlite_autoindex_contacts_1
+SEARCH msg_deliveries USING COVERING INDEX idx_msg_deliveries_agent_msg_id (connection_id=?)
+SEARCH commands USING COVERING INDEX idx_commands_connection_id (connection_id=?)
+SEARCH messages USING COVERING INDEX idx_messages_connection_id (connection_id=?)
+SEARCH snd_files USING COVERING INDEX idx_snd_files_connection_id (connection_id=?)
+SEARCH contacts USING COVERING INDEX idx_contacts_grp_direct_inv_from_member_conn_id (grp_direct_inv_from_member_conn_id=?)
+
 Query: DELETE FROM connections WHERE user_id = ? AND connection_id = ?
 Plan:
 SEARCH connections USING INTEGER PRIMARY KEY (rowid=?)
@@ -6178,6 +6257,10 @@ Query: SELECT sent_inv_queue_info FROM group_members WHERE group_member_id = ? A
 Plan:
 SEARCH group_members USING INTEGER PRIMARY KEY (rowid=?)
 
+Query: SELECT should_sync FROM connections_sync WHERE connections_sync_id = 1
+Plan:
+SEARCH connections_sync USING INTEGER PRIMARY KEY (rowid=?)
+
 Query: SELECT user_contact_link_id FROM contact_requests WHERE contact_request_id = ?
 Plan:
 SEARCH contact_requests USING INTEGER PRIMARY KEY (rowid=?)
@@ -6250,6 +6333,10 @@ Query: UPDATE connections SET security_code = ?, security_code_verified_at = ?,
 Plan:
 SEARCH connections USING INTEGER PRIMARY KEY (rowid=?)
 
+Query: UPDATE connections_sync SET should_sync = 1 WHERE connections_sync_id = 1
+Plan:
+SEARCH connections_sync USING INTEGER PRIMARY KEY (rowid=?)
+
 Query: UPDATE contact_requests SET business_group_id = ? WHERE contact_request_id = ?
 Plan:
 SEARCH contact_requests USING INTEGER PRIMARY KEY (rowid=?)

src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql

@@ -716,6 +716,11 @@ CREATE TABLE group_member_status_predicates(
   member_status TEXT NOT NULL PRIMARY KEY,
   current_member INTEGER NOT NULL DEFAULT 0
 );
+CREATE TABLE connections_sync(
+  connections_sync_id INTEGER PRIMARY KEY AUTOINCREMENT,
+  should_sync INTEGER NOT NULL DEFAULT 0,
+  last_sync_ts TEXT
+);
 CREATE INDEX contact_profiles_index ON contact_profiles(
   display_name,
   full_name

src/Simplex/Chat/View.hs

@@ -57,6 +57,7 @@ import Simplex.Chat.Types.Preferences
 import Simplex.Chat.Types.Shared
 import Simplex.Chat.Types.UITheme
 import qualified Simplex.FileTransfer.Transport as XFTP
+import Simplex.Messaging.Agent (DatabaseDiff (..))
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), SubscriptionsInfo (..))
 import Simplex.Messaging.Agent.Env.SQLite (NetworkConfig (..), ServerRoles (..))
 import Simplex.Messaging.Agent.Protocol
@@ -116,6 +117,9 @@ chatResponseToView hu cfg@ChatConfig {logLevel, showReactions, testView} liveIte
   CRChatStarted -> ["chat started"]
   CRChatRunning -> ["chat is running"]
   CRChatStopped -> ["chat stopped"]
+  CRConnectionsDiff showIds userDiff connDiff
+    | showIds -> viewConnDiffIds userDiff connDiff
+    | otherwise -> viewConnDiffSummary userDiff connDiff
   CRApiChats u chats -> ttyUser u $ if testView then testViewChats chats else [viewJSON chats]
   CRChats chats -> viewChats ts tz chats
   CRApiChat u chat _ -> ttyUser u $ if testView then testViewChat chat else [viewJSON chat]
@@ -449,6 +453,7 @@ chatEventToView hu ChatConfig {logLevel, showReactions, showReceipts, testView}
   CEvtContactConnected u ct userCustomProfile -> ttyUser u $ viewContactConnected ct userCustomProfile testView
   CEvtContactSndReady u ct -> ttyUser u [ttyFullContact ct <> ": you can send messages to contact"]
   CEvtContactAnotherClient u c -> ttyUser u [ttyContact' c <> ": contact is connected to another client"]
+  CEvtConnectionsDiff userDiff connDiff -> viewConnDiffSync userDiff connDiff
   CEvtSubscriptionEnd u acEntity ->
     let Connection {connId} = entityConnection acEntity
      in ttyUser u [sShow connId <> ": END"]
@@ -1423,6 +1428,42 @@ viewUserPrivacy User {userId} User {userId = userId', localDisplayName = n', sho
     "profile is " <> if isJust viewPwdHash then "hidden" else "visible"
   ]
 
+viewConnDiffSync :: DatabaseDiff AgentUserId -> DatabaseDiff AgentConnId -> [StyledString]
+viewConnDiffSync userDiff connDiff =
+  viewConnDiffSummary userDiff connDiff
+    <> ["removed extra users in agent" | not (null $ extraIds userDiff)]
+    <> ["removed extra connections in agent" | not (null $ extraIds connDiff)]
+
+viewConnDiffSummary :: DatabaseDiff AgentUserId -> DatabaseDiff AgentConnId -> [StyledString]
+viewConnDiffSummary userDiff connDiff
+  | noDiff userDiff && noDiff connDiff =
+      ["no difference between agent and chat connections"]
+  | otherwise =
+      ["connections difference summary:"]
+        <> showDatabaseDiff "users" userDiff
+        <> showDatabaseDiff "connections" connDiff
+  where
+    noDiff DatabaseDiff {missingIds, extraIds} = null missingIds && null extraIds
+    showDatabaseDiff name DatabaseDiff {missingIds, extraIds} =
+      ["number of missing " <> name <> " in agent: " <> sShow (length missingIds) | not (null missingIds)]
+        <> ["number of extra " <> name <> " in agent: " <> sShow (length extraIds) | not (null extraIds)]
+
+viewConnDiffIds :: DatabaseDiff AgentUserId -> DatabaseDiff AgentConnId -> [StyledString]
+viewConnDiffIds userDiff connDiff
+  | noDiff userDiff && noDiff connDiff =
+      ["no difference between agent and chat connections"]
+  | otherwise =
+      ["connections difference:"]
+        <> showDatabaseDiff "users" (\(AgentUserId uId) -> uId) userDiff
+        <> showDatabaseDiff "connections" (\(AgentConnId cId) -> cId) connDiff
+  where
+    noDiff DatabaseDiff {missingIds, extraIds} = null missingIds && null extraIds
+    showDatabaseDiff name unwrapId DatabaseDiff {missingIds, extraIds} =
+      ["missing " <> name <> " in agent (agent IDs): " <> showIds missingIds | not (null missingIds)]
+        <> ["extra " <> name <> " in agent (agent IDs): " <> showIds extraIds | not (null extraIds)]
+      where
+        showIds = plain . T.intercalate ", " . map (tshow . unwrapId)
+
 viewUserServers :: UserOperatorServers -> [StyledString]
 viewUserServers (UserOperatorServers _ [] []) = []
 viewUserServers UserOperatorServers {operator, smpServers, xftpServers} =

tests/ChatTests/Direct.hs

@@ -16,7 +16,7 @@ import ChatTests.DBUtils
 import ChatTests.Utils
 import Control.Concurrent (threadDelay)
 import Control.Concurrent.Async (concurrently_)
-import Control.Monad (forM_)
+import Control.Monad (forM_, void)
 import Data.Aeson (ToJSON)
 import qualified Data.Aeson as J
 import qualified Data.ByteString.Char8 as B
@@ -118,6 +118,11 @@ chatDirectTests = do
     it "export/import chat with files" testMaintenanceModeWithFiles
     it "encrypt/decrypt database" testDatabaseEncryption
 #endif
+  describe "connections synchronization" $ do
+    it "should report users missing in agent" testConnSyncMissingAgentUsers
+    it "should remove and report extra users in agent" testConnSyncExtraAgentUsers
+    it "should report connections missing in agent" testConnSyncMissingAgentConns
+    it "should remove and report extra connections in agent" testConnSyncExtraAgentConns
   describe "coordination between app and NSE" $ do
     it "should not subscribe in NSE and subscribe in the app" testSubscribeAppNSE
   describe "mute/unmute messages" $ do
@@ -1523,6 +1528,168 @@ testDatabaseEncryption ps = do
       testChatWorking alice bob
 #endif
 
+testConnSyncMissingAgentUsers :: HasCallStack => TestParams -> IO ()
+testConnSyncMissingAgentUsers ps = do
+  withNewTestChat ps "bob" bobProfile $ \bob -> do
+    withNewTestChat ps "alice" aliceProfile $ \alice -> do
+      connectUsers alice bob
+
+      alice ##> "/create user alisa"
+      showActiveUser alice "alisa"
+      alice ##> "/user alice"
+      showActiveUser alice "alice (Alice)"
+
+      alice ##> "/_connections diff"
+      alice <## "no difference between agent and chat connections"
+
+      void $ withCCAgentTransaction alice $ \db ->
+        DB.execute_ db "DELETE FROM users WHERE user_id = 2"
+
+      alice ##> "/_connections diff"
+      alice <## "connections difference summary:"
+      alice <## "number of missing users in agent: 1"
+
+      alice ##> "/_connections diff show_ids=on"
+      alice <## "connections difference:"
+      alice <## "missing users in agent (agent IDs): 2"
+
+      void $ withCCTransaction alice $ \db ->
+        DB.execute_ db "UPDATE connections_sync SET should_sync = 1 WHERE connections_sync_id = 1"
+
+    withTestChat ps "alice" $ \alice -> do
+      alice <## "connections difference summary:"
+      alice <## "number of missing users in agent: 1"
+
+      alice <## "subscribed 1 connections on server localhost"
+
+      alice <##> bob
+
+testConnSyncExtraAgentUsers :: HasCallStack => TestParams -> IO ()
+testConnSyncExtraAgentUsers ps = do
+  withNewTestChat ps "bob" bobProfile $ \bob -> do
+    withNewTestChat ps "alice" aliceProfile $ \alice -> do
+      connectUsers alice bob
+
+      alice ##> "/_connections diff"
+      alice <## "no difference between agent and chat connections"
+
+      void $ withCCAgentTransaction alice $ \db ->
+        DB.execute_ db "INSERT INTO users DEFAULT VALUES"
+      agentUserCount <- withCCAgentTransaction alice $ \db ->
+        DB.query_ db "SELECT count(1) FROM users" :: IO [[Int]]
+      agentUserCount `shouldBe` [[2]]
+
+      alice ##> "/_connections diff"
+      alice <## "connections difference summary:"
+      alice <## "number of extra users in agent: 1"
+
+      alice ##> "/_connections diff show_ids=on"
+      alice <## "connections difference:"
+      alice <## "extra users in agent (agent IDs): 2"
+
+      void $ withCCTransaction alice $ \db ->
+        DB.execute_ db "UPDATE connections_sync SET should_sync = 1 WHERE connections_sync_id = 1"
+
+    withTestChat ps "alice" $ \alice -> do
+      alice <## "connections difference summary:"
+      alice <## "number of extra users in agent: 1"
+      alice <## "removed extra users in agent"
+
+      alice <## "subscribed 1 connections on server localhost"
+
+      threadDelay 100000
+      agentUserCount <- withCCAgentTransaction alice $ \db ->
+        DB.query_ db "SELECT count(1) FROM users" :: IO [[Int]]
+      agentUserCount `shouldBe` [[1]]
+
+      alice <##> bob
+
+testConnSyncMissingAgentConns :: HasCallStack => TestParams -> IO ()
+testConnSyncMissingAgentConns ps = do
+  withNewTestChat ps "bob" bobProfile $ \bob -> do
+    withNewTestChat ps "cath" cathProfile $ \cath -> do
+      withNewTestChat ps "alice" aliceProfile $ \alice -> do
+        connectUsers alice bob
+
+        alice ##> "/create user alisa"
+        showActiveUser alice "alisa"
+
+        -- connection with cath is in user 2, below we delete connection by user_id
+        -- because it's one of the simplest ways to differentiate them in agent db
+        connectUsers alice cath
+
+        alice ##> "/user alice"
+        showActiveUser alice "alice (Alice)"
+
+        alice ##> "/_connections diff"
+        alice <## "no difference between agent and chat connections"
+
+        void $ withCCAgentTransaction alice $ \db ->
+          DB.execute_ db "DELETE FROM connections WHERE user_id = 2"
+
+        alice ##> "/_connections diff"
+        alice <## "connections difference summary:"
+        alice <## "number of missing connections in agent: 1"
+
+        alice ##> "/_connections diff show_ids=on"
+        alice <## "connections difference:"
+        alice <##. "missing connections in agent (agent IDs):"
+
+        void $ withCCTransaction alice $ \db ->
+          DB.execute_ db "UPDATE connections_sync SET should_sync = 1 WHERE connections_sync_id = 1"
+
+      withTestChat ps "alice" $ \alice -> do
+        alice <## "connections difference summary:"
+        alice <## "number of missing connections in agent: 1"
+
+        alice <## "subscribed 1 connections on server localhost"
+        -- alice <## "[user: alisa] 1 subscription errors (run with -c option to show each error)"
+
+        alice <##> bob
+
+testConnSyncExtraAgentConns :: HasCallStack => TestParams -> IO ()
+testConnSyncExtraAgentConns ps = do
+  withNewTestChat ps "bob" bobProfile $ \bob -> do
+    withNewTestChat ps "cath" cathProfile $ \cath -> do
+      withNewTestChat ps "alice" aliceProfile $ \alice -> do
+        connectUsers alice bob
+        connectUsers alice cath
+
+        alice ##> "/_connections diff"
+        alice <## "no difference between agent and chat connections"
+
+        -- deleting connection record in chat db
+        void $ withCCTransaction alice $ \db ->
+          DB.execute_ db "DELETE FROM connections WHERE contact_id = (SELECT contact_id FROM contacts WHERE local_display_name = 'cath')"
+        agentConnCount <- withCCAgentTransaction alice $ \db ->
+          DB.query_ db "SELECT count(1) FROM connections" :: IO [[Int]]
+        agentConnCount `shouldBe` [[2]]
+
+        alice ##> "/_connections diff"
+        alice <## "connections difference summary:"
+        alice <## "number of extra connections in agent: 1"
+
+        alice ##> "/_connections diff show_ids=on"
+        alice <## "connections difference:"
+        alice <##. "extra connections in agent (agent IDs):"
+
+        void $ withCCTransaction alice $ \db ->
+          DB.execute_ db "UPDATE connections_sync SET should_sync = 1 WHERE connections_sync_id = 1"
+
+      withTestChat ps "alice" $ \alice -> do
+        alice <## "connections difference summary:"
+        alice <## "number of extra connections in agent: 1"
+        alice <## "removed extra connections in agent"
+
+        alice <## "subscribed 1 connections on server localhost"
+
+        threadDelay 100000
+        agentConnCount <- withCCAgentTransaction alice $ \db ->
+          DB.query_ db "SELECT count(1) FROM connections" :: IO [[Int]]
+        agentConnCount `shouldBe` [[1]]
+
+        alice <##> bob
+
 testSubscribeAppNSE :: HasCallStack => TestParams -> IO ()
 testSubscribeAppNSE ps =
   withNewTestChat ps "bob" bobProfile $ \bob -> do