feat: resend pinned state in broadcast channels (#8549)

this PR adds the "message pinned state" to the "reaction state" that is
already sent together with resent channel messages since #8496.
this change is done in the first commit, and in fact only changes few
lines (one can consider to rename "broadcast_reactions.rs" and related
stuff to "broadcast_state.rs" in another PR: i did not to that in this
PR to keep diff and review simple).

moreover, in the second commit ,the "selection of messages to resent" is
changed to keep an additional focus on the pinned messages, so that the
most recent pinned messages are resent as well.

successor of #8496 and #8546

src/chat.rs

@@ -4019,7 +4019,8 @@ 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();
@@ -4027,28 +4028,37 @@ pub(crate) async fn add_contact_to_chat_ex(
     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
     )
-    AND type!=?
-ORDER BY timestamp DESC, id DESC LIMIT ?",
+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
+    )
+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)?),
         )
@@ -4056,7 +4066,7 @@ ORDER BY timestamp DESC, id DESC LIMIT ?",
         .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,6 +34,10 @@ struct WireMessage {
 
     /// Array of reaction entries.
     reactions: Vec<WireEntry>,
+
+    /// Pinned state.
+    #[serde(default)]
+    pinned: bool,
 }
 #[derive(Debug, Serialize, Deserialize)]
 struct WireEntry {
@@ -40,7 +45,7 @@ struct WireEntry {
     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,10 +373,12 @@ mod tests {
                             count: 2,
                         },
                     ],
+                    pinned: false,
                 },
                 WireMessage {
                     id: "23456789@bar".to_string(),
                     reactions: vec![],
+                    pinned: true,
                 },
             ],
         };
@@ -377,7 +386,7 @@ mod tests {
         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();