feat: resend pinned state in broadcast channels

src/chat.rs

@@ -4019,44 +4019,54 @@ pub(crate) async fn add_contact_to_chat_ex(
         chat.sync_contacts(context).await.log_err(context).ok();
     }
     if chat.typ == Chattype::OutBroadcast {
-        resend_last_msgs(context, chat.id, &contact)
+        let msgs = get_broadcast_msgs_to_resend(context, chat_id).await?;
+        resend_msgs_ex(context, &msgs, contact.fingerprint())
             .await
             .log_err(context)
             .ok();
     }
     Ok(true)
 }
 
-async fn resend_last_msgs(context: &Context, chat_id: ChatId, to_contact: &Contact) -> Result<()> {
-    let msgs: Vec<MsgId> = context
+/// Get the messages to resend to a newly joined broadcast member.
+///
+/// These are the most recent messages plus some of the latest pinned messages.
+///
+/// Regarding webxdcs: It is not trivial to resend only the own status updates,
+/// and it is not trivial to resend them only to the newly-joined member,
+/// so that for now, webxdcs are not resend at all.
+async fn get_broadcast_msgs_to_resend(context: &Context, chat_id: ChatId) -> Result<Vec<MsgId>> {
+    let msgs = context
         .sql
         .query_map_vec(
             "
-SELECT id
-FROM msgs
-WHERE chat_id=?
-    AND hidden=0
-    AND NOT ( -- Exclude info and system messages
-        param GLOB '*\nS=*' OR param GLOB 'S=*'
-        OR from_id=?
-        OR to_id=?
+SELECT id, timestamp FROM msgs WHERE id IN
+    (
+        SELECT id FROM msgs WHERE chat_id=?1 -- UNION requires simple SELECT statements without LIMIT; therefore the sub-SELECT
+            AND pinned=1 AND hidden=0 AND type!=?2
+            ORDER BY timestamp DESC, id DESC LIMIT ?3
+    )
+UNION SELECT id, timestamp FROM msgs WHERE id IN
+    (
+        SELECT id FROM msgs WHERE chat_id=?1
+            AND hidden=0 AND type!=?2
+            AND NOT (param GLOB '*\nS=*' OR param GLOB 'S=*' OR from_id=?4 OR to_id=?4) -- Exclude info and system messages
+            ORDER BY timestamp DESC, id DESC LIMIT ?3
     )
-    AND type!=?
-ORDER BY timestamp DESC, id DESC LIMIT ?",
+ORDER BY timestamp DESC, id DESC -- final ORDER BY is needed as UNION does not guarantee ordering",
             (
                 chat_id,
-                ContactId::INFO,
-                ContactId::INFO,
                 Viewtype::Webxdc,
                 constants::N_MSGS_TO_NEW_BROADCAST_MEMBER,
+                ContactId::INFO,
             ),
             |row: &rusqlite::Row| Ok(row.get::<_, MsgId>(0)?),
         )
         .await?
         .into_iter()
         .rev()
         .collect();
-    resend_msgs_ex(context, &msgs, to_contact.fingerprint()).await
+    Ok(msgs)
 }
 
 /// Returns true if an avatar should be attached in the given chat.
@@ -4734,10 +4744,7 @@ pub async fn resend_msgs(context: &Context, msg_ids: &[MsgId]) -> Result<()> {
 /// Resends given messages to a contact with fingerprint `to_fingerprint` or, if it's `None`, to
 /// members of the corresponding chats.
 ///
-/// NB: Actually `to_fingerprint` is only passed for `OutBroadcast` chats when a new member is
-/// added. Regarding webxdcs: It is not trivial to resend only the own status updates,
-/// and it is not trivial to resend them only to the newly-joined member,
-/// so that for now, [`resend_last_msgs`] does not automatically resend webxdcs at all.
+/// `to_fingerprint` is only passed for `OutBroadcast` chats when a new member is added.
 pub(crate) async fn resend_msgs_ex(
     context: &Context,
     msg_ids: &[MsgId],

src/chat/chat_tests.rs

@@ -13,6 +13,7 @@ use crate::headerdef::HeaderDef;
 use crate::imex::{ImexMode, has_backup, imex};
 use crate::message::{Message, MessengerMessage, delete_msgs};
 use crate::mimeparser::{self, MimeMessage};
+use crate::pinned_messages::{get_pinned_messages, set_pinned_state};
 use crate::qr::{Qr, check_qr};
 use crate::receive_imf::receive_imf;
 use crate::securejoin::{get_securejoin_qr, join_securejoin};
@@ -3088,6 +3089,84 @@ async fn test_broadcast_muted() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_get_broadcast_msgs_to_resend() -> Result<()> {
+    let mut tcm = TestContextManager::new();
+
+    // Alice creates a channel
+    let alice = &tcm.alice().await;
+    let chat_id = create_broadcast(alice, "test channel".to_string()).await?;
+    assert_eq!(get_pinned_messages(alice, chat_id).await?.len(), 0);
+    let to_resend = get_broadcast_msgs_to_resend(alice, chat_id).await?;
+    assert_eq!(to_resend.len(), 0);
+
+    // Alice sends 5 messsage to the channel, all of them will be resent
+    let mut msg_ids = Vec::new(); // oldest is first
+    for i in 0..5 {
+        let msg_id = send_text_msg(alice, chat_id, format!("message {i}")).await?;
+        msg_ids.push(msg_id);
+    }
+    let to_resend = get_broadcast_msgs_to_resend(alice, chat_id).await?;
+    assert_eq!(to_resend.len(), 5);
+    for msg_id in &msg_ids[0..5] {
+        assert!(to_resend.contains(msg_id));
+    }
+
+    // If Alice has 50 messags in the channel, only the 10 newest will be resent
+    for i in 5..50 {
+        let msg_id = send_text_msg(alice, chat_id, format!("message {i}")).await?;
+        msg_ids.push(msg_id);
+    }
+    let to_resend = get_broadcast_msgs_to_resend(alice, chat_id).await?;
+    assert_eq!(to_resend.len(), N_MSGS_TO_NEW_BROADCAST_MEMBER);
+    for msg_id in &msg_ids[50 - N_MSGS_TO_NEW_BROADCAST_MEMBER..50] {
+        assert!(to_resend.contains(msg_id));
+    }
+
+    // Alice pins the 2 newest messages, they are included in the most recent ones
+    set_pinned_state(alice, msg_ids[50 - 1], true).await?;
+    set_pinned_state(alice, msg_ids[50 - 2], true).await?;
+    assert_eq!(get_pinned_messages(alice, chat_id).await?.len(), 2);
+    let to_resend = get_broadcast_msgs_to_resend(alice, chat_id).await?;
+    assert_eq!(to_resend.len(), N_MSGS_TO_NEW_BROADCAST_MEMBER);
+    assert!(to_resend.contains(&msg_ids[50 - 1]));
+    assert!(to_resend.contains(&msg_ids[50 - 2]));
+    for msg_id in &msg_ids[50 - N_MSGS_TO_NEW_BROADCAST_MEMBER..50] {
+        assert!(to_resend.contains(msg_id));
+    }
+
+    // Alice pins the 2 oldest messages, they will be resent additionally to the recent messages
+    set_pinned_state(alice, msg_ids[50 - 1], false).await?;
+    set_pinned_state(alice, msg_ids[50 - 2], false).await?;
+    set_pinned_state(alice, msg_ids[0], true).await?;
+    set_pinned_state(alice, msg_ids[1], true).await?;
+    assert_eq!(get_pinned_messages(alice, chat_id).await?.len(), 2);
+    let to_resend = get_broadcast_msgs_to_resend(alice, chat_id).await?;
+    assert_eq!(to_resend.len(), N_MSGS_TO_NEW_BROADCAST_MEMBER + 2);
+    assert!(to_resend.contains(&msg_ids[0]));
+    assert!(to_resend.contains(&msg_ids[1]));
+    for msg_id in &msg_ids[50 - N_MSGS_TO_NEW_BROADCAST_MEMBER..50] {
+        assert!(to_resend.contains(msg_id));
+    }
+
+    // If alice pins 23 old messages, only 10 recently pinned gets resend.
+    // plus 10 normal ones.
+    for msg_id in &msg_ids[0..23] {
+        set_pinned_state(alice, *msg_id, true).await?;
+    }
+    assert_eq!(get_pinned_messages(alice, chat_id).await?.len(), 23);
+    let to_resend = get_broadcast_msgs_to_resend(alice, chat_id).await?;
+    assert_eq!(to_resend.len(), N_MSGS_TO_NEW_BROADCAST_MEMBER * 2);
+    for msg_id in &msg_ids[23 - N_MSGS_TO_NEW_BROADCAST_MEMBER..23] {
+        assert!(to_resend.contains(msg_id));
+    }
+    for msg_id in &msg_ids[50 - N_MSGS_TO_NEW_BROADCAST_MEMBER..50] {
+        assert!(to_resend.contains(msg_id));
+    }
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_broadcast_resend_to_new_member() -> Result<()> {
     let mut tcm = TestContextManager::new();

src/constants.rs

@@ -211,7 +211,8 @@ Here is what to do:
 
 If you have any questions, please send an email to delta@merlinux.eu or ask at https://support.delta.chat/."#;
 
-/// How many recent messages should be re-sent to a new broadcast member.
+/// Number of recent messages that should be resent to a new broadcast member.
+/// Additionally, up to this amount of pinned messages will be resent.
 pub(crate) const N_MSGS_TO_NEW_BROADCAST_MEMBER: usize = 10;
 
 #[cfg(test)]

src/reaction/broadcast_reactions.rs

@@ -16,11 +16,12 @@ use crate::context::Context;
 use crate::log::warn;
 use crate::message::{Message, MsgId, rfc724_mid_exists};
 use crate::param::Param;
+use crate::pinned_messages::handle_pinned_state_from_wire;
 use crate::reaction::{Reaction, ReactionFrequency, get_msg_reactions, sort_frequencies};
 use crate::tools::time;
 use crate::{EventType, chatlist_events};
 
-/// Wire format for accumulated broadcast reactions
+/// Wire format for accumulated broadcast states
 /// (sent as JSON from broadcast channel owner to subscriber in `Chat-Broadcast-States:` header)
 #[derive(Debug, Serialize, Deserialize)]
 struct WirePayload {
@@ -33,14 +34,18 @@ struct WireMessage {
 
     /// Array of reaction entries.
     reactions: Vec<WireEntry>,
+
+    /// Pinned state.
+    #[serde(default)]
+    pinned: bool,
 }
 #[derive(Debug, Serialize, Deserialize)]
 struct WireEntry {
     emoji: String,
     count: usize,
 }
 
-/// Renders one or more message's reactions as a JSON string, ready to be sent in `Chat-Broadcast-States:` header.
+/// Renders one or more message's states as a JSON string, ready to be sent in `Chat-Broadcast-States:` header.
 ///
 /// The returned reaction array for a message may be empty,
 /// allowing to broadcast reaction removal.
@@ -62,6 +67,7 @@ pub(crate) async fn render_json(context: &Context, msg_ids: &[MsgId]) -> Result<
         messages.push(WireMessage {
             id: msg.rfc724_mid,
             reactions: entries, // can be empty if all reactions were removed
+            pinned: msg.pinned,
         });
     }
     if messages.is_empty() {
@@ -199,6 +205,7 @@ pub(crate) async fn receive_broadcast_reactions(context: &Context, json: &str) -
             })
             .collect();
         save_broadcast_reactions(context, msg_id, &frequencies).await?;
+        handle_pinned_state_from_wire(context, &msg, message.pinned).await?;
 
         context.emit_event(EventType::ReactionsChanged {
             // the event is for the subscriber, ReactionsIncoming is not needed
@@ -366,18 +373,20 @@ mod tests {
                             count: 2,
                         },
                     ],
+                    pinned: false,
                 },
                 WireMessage {
                     id: "23456789@bar".to_string(),
                     reactions: vec![],
+                    pinned: true,
                 },
             ],
         };
 
         let json = serde_json::to_string(&payload).unwrap();
         assert_eq!(
             json,
-            r#"{"messages":[{"id":"12345678@foo","reactions":[{"emoji":"😎","count":4},{"emoji":"🕺","count":2}]},{"id":"23456789@bar","reactions":[]}]}"#
+            r#"{"messages":[{"id":"12345678@foo","reactions":[{"emoji":"😎","count":4},{"emoji":"🕺","count":2}],"pinned":false},{"id":"23456789@bar","reactions":[],"pinned":true}]}"#
         );
 
         let payload: WirePayload = serde_json::from_str(&json).unwrap();