Sync broadcast subscribers list (#7578)

fix #7497

src/chat.rs

@@ -4941,12 +4941,20 @@ async fn set_contacts_by_fingerprints(
     context
         .sql
         .transaction(move |transaction| {
-            transaction.execute("DELETE FROM chats_contacts WHERE chat_id=?", (id,))?;
+            // For broadcast channels, we only add members,
+            // because we don't use the membership consistency algorithm,
+            // and are using sync messages as a basic way to ensure consistency between devices.
+            // For groups, we also remove members,
+            // because the sync message is used in order to sync unpromoted groups.
+            if chat.typ != Chattype::OutBroadcast {
+                transaction.execute("DELETE FROM chats_contacts WHERE chat_id=?", (id,))?;
+            }
 
             // We do not care about `add_timestamp` column
             // because timestamps are not used for broadcast channels.
-            let mut statement = transaction
+            let mut statement = transaction.prepare(
-                .prepare("INSERT INTO chats_contacts (chat_id, contact_id) VALUES (?, ?)")?;
+                "INSERT OR IGNORE INTO chats_contacts (chat_id, contact_id) VALUES (?, ?)",
+            )?;
             for contact_id in &contacts {
                 statement.execute((id, contact_id))?;
             }

src/chat/chat_tests.rs

@@ -2862,6 +2862,123 @@ async fn test_broadcast_multidev() -> Result<()> {
     Ok(())
 }
 
+/// Test that, if the broadcast channel owner has multiple devices
+/// and they have diverging views on the recipients,
+/// it is synced when sending a member-addition message.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_broadcast_recipients_sync1() -> Result<()> {
+    let mut tcm = TestContextManager::new();
+    let alice1 = &tcm.alice().await;
+    let alice2 = &tcm.alice().await;
+    let bob = &tcm.bob().await;
+    let charlie = &tcm.charlie().await;
+    for a in &[alice1, alice2] {
+        a.set_config_bool(Config::SyncMsgs, true).await?;
+    }
+
+    // Alice1 creates a broadcast and adds Bob, but for some reason
+    // (e.g. because alice2 runs an older version of DC),
+    // Alice2 doesn't get to know about it
+    let a1_broadcast_id = create_broadcast(alice1, "Channel".to_string()).await?;
+    alice1.send_sync_msg().await.unwrap();
+    alice1.pop_sent_msg().await;
+
+    let qr = get_securejoin_qr(alice1, Some(a1_broadcast_id))
+        .await
+        .unwrap();
+    tcm.exec_securejoin_qr(bob, alice1, &qr).await;
+
+    // The first sync message got lost, so, alice2 doesn't know about the channel now
+    sync(alice1, alice2).await;
+    let a2_chatlist = Chatlist::try_load(alice2, 0, Some("Channel"), None).await?;
+    assert!(a2_chatlist.is_empty());
+
+    // Alice1 adds Charlie to the broadcast channel,
+    // and now, Alice2 receives the messages
+    join_securejoin(charlie, &qr).await.unwrap();
+
+    let request = charlie.pop_sent_msg().await;
+    alice1.recv_msg_trash(&request).await;
+    alice2.recv_msg_trash(&request).await;
+
+    let auth_required = alice1.pop_sent_msg().await;
+    charlie.recv_msg_trash(&auth_required).await;
+    alice2.recv_msg_trash(&auth_required).await;
+
+    let request_with_auth = charlie.pop_sent_msg().await;
+    alice1.recv_msg_trash(&request_with_auth).await;
+    alice2.recv_msg_trash(&request_with_auth).await;
+
+    let member_added = alice1.pop_sent_msg().await;
+    let a2_member_added = alice2.recv_msg(&member_added).await;
+    let _c_member_added = charlie.recv_msg(&member_added).await;
+
+    // Alice1 will now sync the full member list to Alice2:
+    sync(alice1, alice2).await;
+    let a2_chatlist = Chatlist::try_load(alice2, 0, Some("Channel"), None).await?;
+    assert_eq!(a2_chatlist.get_msg_id(0)?.unwrap(), a2_member_added.id);
+
+    let a2_bob_contact = alice2.add_or_lookup_contact_id(bob).await;
+    let a2_charlie_contact = alice2.add_or_lookup_contact_id(charlie).await;
+
+    let a2_chat_members = get_chat_contacts(alice2, a2_member_added.chat_id).await?;
+    assert!(a2_chat_members.contains(&a2_bob_contact));
+    assert!(a2_chat_members.contains(&a2_charlie_contact));
+    assert_eq!(a2_chat_members.len(), 2);
+
+    Ok(())
+}
+
+/// Test that, if the broadcast channel owner has multiple devices
+/// and they have diverging views on the recipients,
+/// sync messages only add members but don't remove them.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_broadcast_recipients_sync2() -> Result<()> {
+    let mut tcm = TestContextManager::new();
+    let alice1 = &tcm.alice().await;
+    let alice2 = &tcm.alice().await;
+    let bob = &tcm.bob().await;
+    let charlie = &tcm.charlie().await;
+    for a in &[alice1, alice2] {
+        a.set_config_bool(Config::SyncMsgs, true).await?;
+    }
+
+    let a1_broadcast_id = create_broadcast(alice1, "Channel".to_string()).await?;
+    sync(alice1, alice2).await;
+
+    tcm.section("Alice1 adds Bob, but Alice2 misses it for some reason");
+    let qr = get_securejoin_qr(alice1, Some(a1_broadcast_id))
+        .await
+        .unwrap();
+    tcm.exec_securejoin_qr(bob, alice1, &qr).await;
+
+    tcm.section("Alice2 adds Charlie, but Alice1 misses it for some reason");
+    let a2_broadcast_id = Chatlist::try_load(alice2, 0, Some("Channel"), None)
+        .await?
+        .get_chat_id(0)
+        .unwrap();
+    let qr = get_securejoin_qr(alice2, Some(a2_broadcast_id))
+        .await
+        .unwrap();
+    tcm.exec_securejoin_qr(charlie, alice2, &qr).await;
+
+    tcm.section("The sync messages should correct the problem");
+    sync(alice1, alice2).await;
+    sync(alice2, alice1).await;
+
+    for (alice, broadcast_id) in [(alice1, a1_broadcast_id), (alice2, a2_broadcast_id)] {
+        let bob_contact = alice.add_or_lookup_contact_id(bob).await;
+        let charlie_contact = alice.add_or_lookup_contact_id(charlie).await;
+
+        let chat_members = get_chat_contacts(alice, broadcast_id).await?;
+        assert!(chat_members.contains(&bob_contact));
+        assert!(chat_members.contains(&charlie_contact));
+        assert_eq!(chat_members.len(), 2);
+    }
+
+    Ok(())
+}
+
 /// - Create a broadcast channel
 /// - Send a message into it in order to promote it
 /// - Add a contact

src/securejoin.rs

@@ -596,8 +596,15 @@ pub(crate) async fn handle_securejoin_handshake(
                 // Join group.
                 chat::add_contact_to_chat_ex(context, Nosync, joining_chat_id, contact_id, true)
                     .await?;
+
                 let chat = Chat::load_from_db(context, joining_chat_id).await?;
 
+                if chat.typ == Chattype::OutBroadcast {
+                    // We don't use the membership consistency algorithm for broadcast channels,
+                    // so, sync the memberlist when adding a contact
+                    chat.sync_contacts(context).await.log_err(context).ok();
+                }
+
                 inviter_progress(context, contact_id, joining_chat_id, chat.typ)?;
                 // IMAP-delete the message to avoid handling it by another device and adding the
                 // member twice. Another device will know the member's key from Autocrypt-Gossip.