diff --git a/deltachat-jsonrpc/src/api/types/reactions.rs b/deltachat-jsonrpc/src/api/types/reactions.rs index 271cf8b9d6..4d40775acd 100644 --- a/deltachat-jsonrpc/src/api/types/reactions.rs +++ b/deltachat-jsonrpc/src/api/types/reactions.rs @@ -1,6 +1,5 @@ use std::collections::BTreeMap; -use deltachat::contact::ContactId; use deltachat::reaction::Reactions; use serde::Serialize; use typescript_type_def::TypeDef; @@ -34,30 +33,24 @@ pub struct JsonrpcReactions { impl From for JsonrpcReactions { fn from(reactions: Reactions) -> Self { let reactions_by_contact: BTreeMap> = reactions + .by_contact .iter() .map(|(key, value)| (key.to_u32(), vec![value.as_str().to_string()])) .collect(); - let self_reaction = reactions_by_contact.get(&ContactId::SELF.to_u32()); - - let mut reactions_v = Vec::new(); - for (emoji, count) in reactions.emoji_sorted_by_frequency() { - let is_from_self = if let Some(self_reaction) = self_reaction { - self_reaction.contains(&emoji) - } else { - false - }; - - let reaction = JsonrpcReaction { - emoji, - count, - is_from_self, - }; - reactions_v.push(reaction) - } + + let reactions = reactions + .frequencies + .into_iter() + .map(|entry| JsonrpcReaction { + emoji: entry.reaction.as_str().to_string(), + count: entry.count, + is_from_self: entry.is_from_self, + }) + .collect(); JsonrpcReactions { reactions_by_contact, - reactions: reactions_v, + reactions, } } } diff --git a/src/chat.rs b/src/chat.rs index c7203be6ba..b1562b314a 100644 --- a/src/chat.rs +++ b/src/chat.rs @@ -2685,10 +2685,11 @@ async fn prepare_send_msg( // from the chat. CantSendReason::NotAMember => msg.param.get_cmd() == SystemMessage::MemberRemovedFromGroup, CantSendReason::InBroadcast => { - matches!( - msg.param.get_cmd(), - SystemMessage::MemberRemovedFromGroup | SystemMessage::SecurejoinMessage - ) + msg.param.get_int(Param::Reaction).unwrap_or_default() != 0 + || matches!( + msg.param.get_cmd(), + SystemMessage::MemberRemovedFromGroup | SystemMessage::SecurejoinMessage + ) } CantSendReason::MissingKey => msg .param @@ -3677,14 +3678,15 @@ pub(crate) async fn create_out_broadcast_ex( t.execute( "INSERT INTO chats - (type, name, name_normalized, grpid, created_timestamp, param) - VALUES(?, ?, ?, ?, ?, ?)", + (type, name, name_normalized, grpid, created_timestamp, muted_until, param) + VALUES(?, ?, ?, ?, ?, ?, ?)", ( Chattype::OutBroadcast, &chat_name, normalize_text(&chat_name), &grpid, timestamp, + MuteDuration::Forever, params.to_string(), ), )?; diff --git a/src/chat/chat_tests.rs b/src/chat/chat_tests.rs index c76fe7655e..a43ad866ff 100644 --- a/src/chat/chat_tests.rs +++ b/src/chat/chat_tests.rs @@ -3055,6 +3055,29 @@ async fn test_broadcast_change_name() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn test_broadcast_muted() -> Result<()> { + let mut tcm = TestContextManager::new(); + let alice = &tcm.alice().await; + let bob = &tcm.bob().await; + + // Alice's new outgoing broadcast channel is muted after creation: + // Channel owners can only get reaction notifications; they are usually not of much interest. + let alice_chat_id = create_broadcast(alice, "Channel".to_string()).await?; + let qr = get_securejoin_qr(alice, Some(alice_chat_id)).await?; + let alice_chat = Chat::load_from_db(alice, alice_chat_id).await?; + assert!(alice_chat.is_muted()); + + // Bob joins the channel, for him, it is not muted: + // For channel subscribers, new messages to newly subscribed channels are often interesting. + let bob_chat_id = tcm.exec_securejoin_qr(bob, alice, &qr).await; + bob_chat_id.accept(bob).await?; + let bob_chat = Chat::load_from_db(bob, bob_chat_id).await?; + assert!(!bob_chat.is_muted()); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_broadcast_resend_to_new_member() -> Result<()> { let mut tcm = TestContextManager::new(); diff --git a/src/config.rs b/src/config.rs index 8c84a9de8c..57fde82f8c 100644 --- a/src/config.rs +++ b/src/config.rs @@ -346,6 +346,9 @@ pub enum Config { /// Timestamp of the last time housekeeping was run LastHousekeeping, + /// Timestamp of the last time accumulated broadcast channel reactions were sent + LastReactionsBroadcast, + /// Timestamp of the last `CantDecryptOutgoingMsgs` notification. LastCantDecryptOutgoingMsgs, diff --git a/src/context.rs b/src/context.rs index ed822448b8..29a0ea9191 100644 --- a/src/context.rs +++ b/src/context.rs @@ -961,6 +961,12 @@ impl Context { .await? .to_string(), ); + res.insert( + "last_reactions_broadcast", + self.get_config_int(Config::LastReactionsBroadcast) + .await? + .to_string(), + ); res.insert( "last_cant_decrypt_outgoing_msgs", self.get_config_int(Config::LastCantDecryptOutgoingMsgs) diff --git a/src/events/payload.rs b/src/events/payload.rs index 8300f02b1c..5be1657438 100644 --- a/src/events/payload.rs +++ b/src/events/payload.rs @@ -92,7 +92,7 @@ pub enum EventType { /// ID of the message for which reactions were changed. msg_id: MsgId, - /// ID of the contact whose reaction set is changed. + /// ID of the contact whose reaction set is changed. May be 0 eg. in case of broadcasted reactions. contact_id: ContactId, }, diff --git a/src/headerdef.rs b/src/headerdef.rs index c8eeec37df..d46c87d758 100644 --- a/src/headerdef.rs +++ b/src/headerdef.rs @@ -122,6 +122,10 @@ pub enum HeaderDef { /// This is an unprotected header. ChatIsPostMessage, + /// Broadcasted reactions for this or other chat messages. + /// See broadcast_reactions.rs for the wire format. + ChatBroadcastReactions, + /// [Autocrypt](https://autocrypt.org/) header. Autocrypt, AutocryptGossip, diff --git a/src/mimefactory.rs b/src/mimefactory.rs index 7abae86212..96ce76bdbe 100644 --- a/src/mimefactory.rs +++ b/src/mimefactory.rs @@ -1698,6 +1698,13 @@ impl MimeFactory { )) } + if let Some(broadcast_reactions) = msg.param.get(Param::BroadcastReactions) { + headers.push(( + "Chat-Broadcast-Reactions", + mail_builder::headers::raw::Raw::new(b_encode(broadcast_reactions)).into(), + )); + } + if msg.viewtype == Viewtype::Voice || msg.viewtype == Viewtype::Audio || msg.viewtype == Viewtype::Video diff --git a/src/mimeparser.rs b/src/mimeparser.rs index 654b3df20e..b826ba3bd1 100644 --- a/src/mimeparser.rs +++ b/src/mimeparser.rs @@ -116,6 +116,10 @@ pub(crate) struct MimeMessage { pub(crate) mdn_reports: Vec, pub(crate) delivery_report: Option, + /// Parsed `Chat-Broadcast-Reactions` header, if any: + /// accumulated reaction updates sent by a broadcast channel owner. + pub(crate) broadcast_reactions: Option, + /// Standard USENET signature, if any. /// /// `None` means no text part was received, empty string means a text part without a footer is @@ -657,6 +661,7 @@ impl MimeMessage { user_avatar: None, group_avatar: None, delivery_report: None, + broadcast_reactions: None, footer: None, is_mime_modified: false, decoded_data: Vec::new(), @@ -793,6 +798,12 @@ impl MimeMessage { } } + fn parse_broadcast_reactions_header(&mut self) { + self.broadcast_reactions = self + .get_header(HeaderDef::ChatBroadcastReactions) + .map(|s| s.to_string()); + } + /// Squashes mutitpart chat messages with attachment into single-part messages. /// /// Delta Chat sends attachments, such as images, in two-part messages, with the first message @@ -874,6 +885,7 @@ impl MimeMessage { self.parse_system_message_headers(); self.parse_avatar_headers(context)?; self.parse_videochat_headers(); + self.parse_broadcast_reactions_header(); if self.delivery_report.is_none() { self.squash_attachment_parts(); } diff --git a/src/param.rs b/src/param.rs index 41360bc1d7..95053ec6df 100644 --- a/src/param.rs +++ b/src/param.rs @@ -70,9 +70,12 @@ pub enum Param { /// For Messages WantsMdn = b'r', - /// For Messages: the message is a reaction. + /// For Messages: Render message as a RFC 9078 reaction. Reaction = b'x', + /// For Messages: Additional reactions that go to the `Chat-Broadcast-Reactions:` header + BroadcastReactions = b'X', + /// For Chats: the timestamp of the last reaction. LastReactionTimestamp = b'y', diff --git a/src/reaction.rs b/src/reaction.rs index 2b4752ca15..a2fa881931 100644 --- a/src/reaction.rs +++ b/src/reaction.rs @@ -14,6 +14,8 @@ //! possible to remove the reaction by sending an empty string as a reaction, //! even though RFC 9078 requires at least one emoji to be sent. +pub(crate) mod broadcast_reactions; + use std::cmp::Ordering; use std::collections::BTreeMap; use std::fmt; @@ -23,11 +25,15 @@ use serde::{Deserialize, Serialize}; use crate::chat::{Chat, ChatId, send_msg}; use crate::chatlist_events; +use crate::constants::Chattype; use crate::contact::ContactId; use crate::context::Context; use crate::events::EventType; use crate::message::{Message, MsgId, rfc724_mid_exists}; use crate::param::Param; +use crate::reaction::broadcast_reactions::{ + load_broadcast_reactions, modify_frequencies, refine_frequencies, save_broadcast_reactions, +}; /// A single reaction. #[derive(Debug, Default, Clone, Deserialize, Eq, PartialEq, Serialize)] @@ -70,78 +76,46 @@ impl Reaction { } } +/// A single reaction with frequency and sender flag. +#[derive(Debug, Clone, PartialEq)] +pub struct ReactionFrequency { + /// The reaction emoji. + pub reaction: Reaction, + + /// Number of contacts that reacted with this emoji. + pub count: usize, + + /// True if `ContactId::SELF` is among the contacts that reacted with this emoji. + pub is_from_self: bool, +} + /// Structure representing all reactions to a particular message. #[derive(Debug)] pub struct Reactions { + /// Unique reactions and their frequencies. + pub frequencies: Vec, + /// Map from a contact to its reaction to message. - reactions: BTreeMap, + /// For channels subscribers, this map is empty or contains `ContactId::SELF` only. + pub by_contact: BTreeMap, } impl Reactions { - /// Returns vector of contacts that reacted to the message. - pub fn contacts(&self) -> Vec { - self.reactions.keys().copied().collect() - } - - /// Returns reaction of a given contact to message. - /// - /// If contact did not react to message or removed the reaction, - /// this method returns an empty reaction. - pub fn get(&self, contact_id: ContactId) -> Reaction { - self.reactions.get(&contact_id).cloned().unwrap_or_default() - } - /// Returns true if the message has no reactions. pub fn is_empty(&self) -> bool { - self.reactions.is_empty() - } - - /// Returns a map from emojis to their frequencies. - #[expect(clippy::arithmetic_side_effects)] - pub fn emoji_frequencies(&self) -> BTreeMap { - let mut emoji_frequencies: BTreeMap = BTreeMap::new(); - for reaction in self.reactions.values() { - emoji_frequencies - .entry(reaction.as_str().to_string()) - .and_modify(|x| *x += 1) - .or_insert(1); - } - emoji_frequencies - } - - /// Returns a vector of emojis - /// sorted in descending order of frequencies. - /// - /// This function can be used to display the reactions in - /// the message bubble in the UIs. - pub fn emoji_sorted_by_frequency(&self) -> Vec<(String, usize)> { - let mut emoji_frequencies: Vec<(String, usize)> = - self.emoji_frequencies().into_iter().collect(); - emoji_frequencies.sort_by(|(a, a_count), (b, b_count)| { - match a_count.cmp(b_count).reverse() { - Ordering::Equal => a.cmp(b), - other => other, - } - }); - emoji_frequencies - } - - /// Returns an iterator of the contacts that reacted and their corresponding reactions. - pub fn iter(&self) -> impl Iterator { - self.reactions.iter() + self.frequencies.is_empty() } } impl fmt::Display for Reactions { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let emoji_frequencies = self.emoji_sorted_by_frequency(); let mut first = true; - for (emoji, frequency) in emoji_frequencies { + for entry in &self.frequencies { if !first { write!(f, " ")?; } first = false; - write!(f, "{emoji}{frequency}")?; + write!(f, "{}{}", entry.reaction.as_str(), entry.count)?; } Ok(()) } @@ -155,6 +129,10 @@ async fn set_msg_id_reaction( timestamp: i64, reaction: &Reaction, ) -> Result<()> { + let mut chat = Chat::load_from_db(context, chat_id).await?; + let old_reactions = get_msg_reactions(context, msg_id).await?; + let old_self_reaction = old_reactions.by_contact.get(&ContactId::SELF); + if reaction.is_empty() { // Simply remove the record instead of setting it to empty string. context @@ -177,7 +155,6 @@ async fn set_msg_id_reaction( (msg_id, contact_id, reaction.as_str()), ) .await?; - let mut chat = Chat::load_from_db(context, chat_id).await?; if chat .param .update_timestamp(Param::LastReactionTimestamp, timestamp)? @@ -190,6 +167,25 @@ async fn set_msg_id_reaction( } } + if chat.typ == Chattype::OutBroadcast { + context + .sql + .execute( + "INSERT INTO reactions_need_broadcast (chat_id, msg_id) + VALUES (?1, ?2) + ON CONFLICT(chat_id, msg_id) DO NOTHING;", + (chat_id, msg_id), + ) + .await?; + } else if chat.typ == Chattype::InBroadcast && contact_id == ContactId::SELF { + // for immediate feedback, alter `broadcasted_reactions` directly. + // this "dirty state" will overwritten on next broadcast, + // however, means that `broadcasted_reactions` counts can be assumemd to include SELF-reaction eventually. + let mut frequencies = load_broadcast_reactions(context, msg_id).await?; + modify_frequencies(&mut frequencies, old_self_reaction, reaction); + save_broadcast_reactions(context, msg_id, &frequencies).await?; + } + context.emit_event(EventType::ReactionsChanged { chat_id, msg_id, @@ -405,9 +401,49 @@ pub(crate) async fn apply_pending_reactions( Ok(()) } +/// Returns unique reactions with their frequency and whether self reacted, +/// sorted in descending order of frequency. +fn calc_frequencies(by_contact: &BTreeMap) -> Vec { + let mut self_reaction = Reaction::new(""); + let mut counts: BTreeMap<&str, usize> = BTreeMap::new(); + for (contact_id, reaction) in by_contact { + let count = counts.entry(reaction.as_str()).or_insert(0); + *count = count.saturating_add(1); + if *contact_id == ContactId::SELF { + self_reaction = reaction.clone(); + } + } + + let mut frequencies: Vec = counts + .into_iter() + .map(|(emoji, count)| ReactionFrequency { + reaction: Reaction::new(emoji), + count, + is_from_self: !self_reaction.is_empty() && self_reaction.as_str() == emoji, + }) + .collect(); + + sort_frequencies(&mut frequencies); + frequencies +} + +/// Sorts reaction frequencies by descending count. when equal, order by emoji string. +/// +/// This is the order UIs shall use to display reactions in the message bubble. +pub(crate) fn sort_frequencies(frequencies: &mut [ReactionFrequency]) { + frequencies.sort_by(|a, b| match b.count.cmp(&a.count) { + Ordering::Equal => a.reaction.as_str().cmp(b.reaction.as_str()), + other => other, + }); +} + /// Returns a structure containing all reactions to the message. +/// +/// For displaying, UI shall use the `frequencies` field, which is already sorted accordingly. +/// `frequencies` should also be used to check for SELF-reaction. +/// For detailed reaction information outside broadcast channel subscribers, UI can use the `by_contact` table. pub async fn get_msg_reactions(context: &Context, msg_id: MsgId) -> Result { - let mut reactions: BTreeMap = context + let mut by_contact: BTreeMap = context .sql .query_map_collect( "SELECT contact_id, reaction FROM reactions WHERE msg_id=?", @@ -419,8 +455,19 @@ pub async fn get_msg_reactions(context: &Context, msg_id: MsgId) -> Result Vec { + self.by_contact.keys().copied().collect() + } + + // Returns reaction of a given contact to message or an empty reaction. + fn get(&self, contact_id: ContactId) -> Reaction { + self.by_contact + .get(&contact_id) + .cloned() + .unwrap_or_default() + } + } + #[test] fn test_parse_reaction() { // Check that basic set of emojis from RFC 9078 is supported. @@ -842,11 +903,9 @@ Content-Disposition: reaction\n\ .unwrap(); let reactions = get_msg_reactions(&alice, alice_msg.sender_msg_id).await?; assert_eq!(reactions.to_string(), "👍2"); - - assert_eq!( - reactions.emoji_sorted_by_frequency(), - vec![("👍".to_string(), 2)] - ); + assert_eq!(reactions.frequencies.len(), 1); + assert_eq!(reactions.frequencies[0].reaction.as_str(), "👍"); + assert_eq!(reactions.frequencies[0].count, 2); Ok(()) } @@ -876,7 +935,7 @@ Content-Disposition: reaction\n\ bob.recv_msg_hidden(&reaction_msg).await; let msg = bob.recv_msg(&alice_msg).await; - assert_eq!(get_msg_reactions(&bob, msg.id).await?.reactions.len(), 1); + assert_eq!(get_msg_reactions(&bob, msg.id).await?.by_contact.len(), 1); } // group @@ -894,7 +953,7 @@ Content-Disposition: reaction\n\ bob.recv_msg_hidden(&reaction_msg_alice).await; bob.recv_msg_hidden(&reaction_msg_charlie).await; let msg = bob.recv_msg(&alice_msg).await; - assert_eq!(get_msg_reactions(&bob, msg.id).await?.reactions.len(), 2); + assert_eq!(get_msg_reactions(&bob, msg.id).await?.by_contact.len(), 2); } // react and remove reaction @@ -911,7 +970,7 @@ Content-Disposition: reaction\n\ bob.recv_msg_hidden(&reaction_msg).await; bob.recv_msg_hidden(&remove_reaction_msg).await; let msg = bob.recv_msg(&alice_msg).await; - assert_eq!(get_msg_reactions(&bob, msg.id).await?.reactions.len(), 0); + assert!(get_msg_reactions(&bob, msg.id).await?.is_empty()); } Ok(()) } @@ -1026,7 +1085,7 @@ Content-Disposition: reaction\n\ send_reaction(&alice, msg_id, "🐫").await?; assert_summary(&alice, "You reacted 🐫 to \"foo\"").await; let reactions = get_msg_reactions(&alice, msg_id).await?; - assert_eq!(reactions.reactions.len(), 1); + assert_eq!(reactions.by_contact.len(), 1); // Alice forwards that message to Bob: Reactions are not forwarded, the message is prefixed by "Forwarded". let bob_id = Contact::create(&alice, "", "bob@example.net").await?; @@ -1036,7 +1095,7 @@ Content-Disposition: reaction\n\ let chatlist = Chatlist::try_load(&alice, 0, None, None).await.unwrap(); let forwarded_msg_id = chatlist.get_msg_id(0)?.unwrap(); let reactions = get_msg_reactions(&alice, forwarded_msg_id).await?; - assert!(reactions.reactions.is_empty()); // reactions are not forwarded + assert!(reactions.is_empty()); // reactions are not forwarded // Alice reacts to forwarded message: // For reaction summary neither original message author nor "Forwarded" prefix is shown @@ -1044,7 +1103,7 @@ Content-Disposition: reaction\n\ send_reaction(&alice, forwarded_msg_id, "🐳").await?; assert_summary(&alice, "You reacted 🐳 to \"foo\"").await; let reactions = get_msg_reactions(&alice, msg_id).await?; - assert_eq!(reactions.reactions.len(), 1); + assert_eq!(reactions.by_contact.len(), 1); Ok(()) } @@ -1231,11 +1290,10 @@ Content-Transfer-Encoding: 7bit\r // MDN request was ignored, but reaction was not. let reactions = get_msg_reactions(bob, bob_msg.id).await?; - assert_eq!(reactions.reactions.len(), 1); - assert_eq!( - reactions.emoji_sorted_by_frequency(), - vec![("👀".to_string(), 1)] - ); + assert_eq!(reactions.by_contact.len(), 1); + assert_eq!(reactions.frequencies.len(), 1); + assert_eq!(reactions.frequencies[0].reaction.as_str(), "👀"); + assert_eq!(reactions.frequencies[0].count, 1); Ok(()) } diff --git a/src/reaction/broadcast_reactions.rs b/src/reaction/broadcast_reactions.rs new file mode 100644 index 0000000000..4a4e0e10a2 --- /dev/null +++ b/src/reaction/broadcast_reactions.rs @@ -0,0 +1,663 @@ +//! # Broadcasting Reactions. +//! +//! For broadcast channels, reactions are sent from the subscriber to the broadcast channel owner as usual. +//! The owner then remembers these changes by adding a record to `reactions_need_broadcast`, +//! and every some minutes sends an update to all subscribers. + +use anyhow::Result; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; + +use crate::chat::{Chat, ChatId, send_msg}; +use crate::config::Config; +use crate::constants::Chattype; +use crate::contact::ContactId; +use crate::context::Context; +use crate::log::warn; +use crate::message::{Message, MsgId, rfc724_mid_exists}; +use crate::param::Param; +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 +/// (sent from broadcast channel owner to subscriber in `Chat-Broadcast-Reactions:` header) +#[derive(Debug, Serialize, Deserialize)] +struct WirePayload { + messages: Vec, +} +#[derive(Debug, Serialize, Deserialize)] +struct WireMessage { + /// RFC 724 Message-ID. + id: String, + + /// Array of reaction entries. + reactions: Vec, +} +#[derive(Debug, Serialize, Deserialize)] +struct WireEntry { + emoji: String, + count: usize, +} + +/// Seconds between sending out accumulated reaction updates for broadcast channels from `reactions_need_broadcast` table +const REACTION_BROADCAST_PERIOD: i64 = 10 * 60; + +/// Starts broadcasting if last broadcasting is more than `REACTION_BROADCAST_PERIOD` seconds in the past. +/// +/// Moreover, also broadcast if `lst_broadcast_time` is in the future: +/// That way we're not stuck e.g. if the clock was accidentally set to one year in the future and then rewinded back. +pub(crate) async fn maybe_broadcast_reactions(context: &Context) -> Result<()> { + let now = time(); + let last_broadcast_time = context + .get_config_i64(Config::LastReactionsBroadcast) + .await?; + let next_broadcast_time = last_broadcast_time.saturating_add(REACTION_BROADCAST_PERIOD); + if next_broadcast_time <= now || last_broadcast_time > now { + context + .set_config_internal(Config::LastReactionsBroadcast, Some(&now.to_string())) + .await?; + broadcast_reactions_for_all_chats(context).await?; + } + Ok(()) +} + +/// Sends out accumulated reactions +/// for all broadcast channels with reactions in `reactions_need_broadcast`. +/// +/// For every affected `chat_id`, +/// a single hidden message is sent to all subscribers containing the full, current reaction state (not a diff) +/// for every message that received a reaction change since the last broadcast. +async fn broadcast_reactions_for_all_chats(context: &Context) -> Result<()> { + let chat_ids: Vec = context + .sql + .query_map_collect( + "SELECT DISTINCT chat_id FROM reactions_need_broadcast", + (), + |row| { + let chat_id: ChatId = row.get(0)?; + Ok(chat_id) + }, + ) + .await?; + + for chat_id in chat_ids { + if let Err(err) = broadcast_reactions_for_one_chat(context, chat_id).await { + warn!( + context, + "Failed to broadcast reactions for chat {chat_id}: {err:#}." + ); + } + } + Ok(()) +} + +/// Sends out accumulated reactions for a single broadcast channel +async fn broadcast_reactions_for_one_chat(context: &Context, chat_id: ChatId) -> Result<()> { + let msg_ids: Vec = context + .sql + .query_map_collect( + "SELECT DISTINCT msg_id FROM reactions_need_broadcast WHERE chat_id=?", + (chat_id,), + |row| { + let msg_id: MsgId = row.get(0)?; + Ok(msg_id) + }, + ) + .await?; + + let mut messages: Vec = Vec::new(); + for msg_id in &msg_ids { + let Some(msg) = Message::load_from_db_optional(context, *msg_id).await? else { + continue; + }; + let reactions = get_msg_reactions(context, *msg_id).await?; + let entries: Vec = reactions + .frequencies + .into_iter() + .map(|entry| WireEntry { + emoji: entry.reaction.as_str().to_string(), + count: entry.count, + }) + .collect(); + messages.push(WireMessage { + id: msg.rfc724_mid, + reactions: entries, // can be empty if all reactions were removed + }); + } + + if !messages.is_empty() { + let payload = WirePayload { messages }; + let json = serde_json::to_string(&payload)?; + let mut reaction_msg = Message::new_text("".to_string()); + reaction_msg.set_reaction(); + reaction_msg.param.set(Param::BroadcastReactions, json); + reaction_msg.hidden = true; + send_msg(context, chat_id, &mut reaction_msg).await?; + } + + context + .sql + .execute( + "DELETE FROM reactions_need_broadcast WHERE chat_id=?", + (chat_id,), + ) + .await?; + + Ok(()) +} + +/// Applies incoming, accumulated reactions received via the `Chat-Broadcast-Reactions:` header +/// to the `broadcasted_reactions` table. +pub(crate) async fn receive_broadcast_reactions(context: &Context, json: &str) -> Result<()> { + let payload: WirePayload = serde_json::from_str(json)?; + + for message in payload.messages { + let Some(msg_id) = rfc724_mid_exists(context, &message.id).await? else { + continue; // no need for a pending reaction, the next periodic update has the state again + }; + let Some(msg) = Message::load_from_db_optional(context, msg_id).await? else { + continue; // there may have been a deletion race, ignore error + }; + let chat = match Chat::load_from_db(context, msg.chat_id).await { + Ok(chat) => chat, + Err(err) => { + warn!(context, "Cannot load chat for broadcast reaction: {err}"); + continue; + } + }; + if chat.typ != Chattype::InBroadcast { + continue; + } + + let frequencies: Vec = message + .reactions + .into_iter() + .map(|entry| ReactionFrequency { + reaction: Reaction::new(&entry.emoji), + count: entry.count, + is_from_self: false, // set in refine_frequencies() + }) + .collect(); + save_broadcast_reactions(context, msg_id, &frequencies).await?; + + context.emit_event(EventType::ReactionsChanged { + // the event is for the subscriber, ReactionsIncoming is not needed + chat_id: msg.chat_id, + msg_id, + contact_id: ContactId::UNDEFINED, + }); + chatlist_events::emit_chatlist_item_changed(context, msg.chat_id); + } + + Ok(()) +} + +/// Load broadcasted reactions from `broadcasted_reactions`. +/// This table is filled only for the broadcast channel subscribers (`Chattype::InBroadcast`), +/// by received reactions from the owner or by or temporarily add SELF-reactions. +/// In there are no broadcasted reactions, an empty array is returned. +pub(crate) async fn load_broadcast_reactions( + context: &Context, + msg_id: MsgId, +) -> Result> { + let mut frequencies: Vec = context + .sql + .query_map_collect( + "SELECT reaction, count FROM broadcasted_reactions WHERE msg_id=?", + (msg_id,), + |row| { + let reaction: String = row.get(0)?; + let count: i64 = row.get(1)?; + Ok(ReactionFrequency { + reaction: Reaction::new(&reaction), + count: count as usize, + is_from_self: false, + }) + }, + ) + .await?; + + sort_frequencies(&mut frequencies); + Ok(frequencies) +} + +/// Save an array of frequencies to the `broadcasted_reactions` table. +pub(crate) async fn save_broadcast_reactions( + context: &Context, + msg_id: MsgId, + frequencies: &Vec, +) -> Result<()> { + context + .sql + .transaction(move |transaction| { + transaction.execute( + "DELETE FROM broadcasted_reactions WHERE msg_id=?", + (msg_id,), + )?; + for entry in frequencies { + transaction.execute( + "INSERT INTO broadcasted_reactions (msg_id, reaction, count) + VALUES (?1, ?2, ?3)", + (msg_id, &entry.reaction.as_str(), entry.count), + )?; + } + Ok(()) + }) + .await?; + Ok(()) +} + +/// Modifies frequencies in-place to reflect a change in the SELF user's reaction. +/// +/// This is used for immediate local feedback in `Chattype::InBroadcast` before the +/// next periodic broadcast overwrites this "dirty state". +pub(crate) fn modify_frequencies( + frequencies: &mut Vec, + old_self_reaction: Option<&Reaction>, + new_self_reaction: &Reaction, +) { + if let Some(old_reaction) = old_self_reaction { + let mut remove_idx = None; + for (idx, entry) in frequencies.iter_mut().enumerate() { + if entry.reaction == *old_reaction { + entry.count = entry.count.saturating_sub(1); + if entry.count == 0 { + remove_idx = Some(idx); + } + break; + } + } + if let Some(idx) = remove_idx { + frequencies.remove(idx); + } + } + + if new_self_reaction.is_empty() { + return; + } + + if let Some(entry) = frequencies + .iter_mut() + .find(|e| e.reaction == *new_self_reaction) + { + entry.count = entry.count.saturating_add(1); + } else { + frequencies.push(ReactionFrequency { + reaction: new_self_reaction.clone(), + count: 1, + is_from_self: false, // Will be correctly set to `true` by `refine_frequencies` + }); + } +} + +/// Merge `by_contact` status to broadcasted reaction frequencies. +pub(crate) fn refine_frequencies( + mut broadcasted_reactions: Vec, + by_contact: &BTreeMap, +) -> Vec { + // Add missing reactions. + // This can happen e.g. for SELF-reactions done during offline when state of owner does not have ones yet. + // It will repair on the next reaction broadcast, until then, the following is good enough. + for reaction in by_contact.values() { + if !broadcasted_reactions + .iter() + .any(|entry| entry.reaction == *reaction) + { + broadcasted_reactions.push(ReactionFrequency { + reaction: reaction.clone(), + count: 1, + is_from_self: false, + }); + } + } + + // Mark SELF-reaction as such + if let Some(self_reaction) = by_contact.get(&ContactId::SELF) { + for entry in &mut broadcasted_reactions { + entry.is_from_self = entry.reaction == *self_reaction; + } + } + + sort_frequencies(&mut broadcasted_reactions); + broadcasted_reactions +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::chat::create_broadcast; + use crate::reaction::send_reaction; + use crate::securejoin::get_securejoin_qr; + use crate::test_utils::TestContextManager; + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_broadcast_reaction_wire_format() { + let payload = WirePayload { + messages: vec![ + WireMessage { + id: "12345678@foo".to_string(), + reactions: vec![ + WireEntry { + emoji: "😎".to_string(), + count: 4, + }, + WireEntry { + emoji: "🕺".to_string(), + count: 2, + }, + ], + }, + WireMessage { + id: "23456789@bar".to_string(), + reactions: vec![], + }, + ], + }; + + 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":[]}]}"# + ); + + let payload: WirePayload = serde_json::from_str(&json).unwrap(); + assert_eq!(payload.messages.len(), 2); + assert_eq!(payload.messages[0].id, "12345678@foo"); + assert_eq!(payload.messages[0].reactions.len(), 2); + assert_eq!(payload.messages[0].reactions[0].emoji, "😎"); + assert_eq!(payload.messages[0].reactions[0].count, 4); + assert_eq!(payload.messages[0].reactions[1].emoji, "🕺"); + assert_eq!(payload.messages[0].reactions[1].count, 2); + assert_eq!(payload.messages[1].id, "23456789@bar"); + assert!(payload.messages[1].reactions.is_empty()); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_modify_frequencies() { + // Helper to create a ReactionFrequency entry + let freq = |emoji: &str, count: usize, is_from_self: bool| -> ReactionFrequency { + ReactionFrequency { + reaction: Reaction::new(emoji), + count, + is_from_self, + } + }; + + // Add entry + let mut frequencies = vec![freq("👍", 2, false)]; + let old: Option<&Reaction> = None; + let new = Reaction::new("❤️"); + modify_frequencies(&mut frequencies, old, &new); + assert_eq!(frequencies.len(), 2); + assert_eq!(frequencies[0].reaction.as_str(), "👍"); + assert_eq!(frequencies[0].count, 2); + assert_eq!(frequencies[1].reaction.as_str(), "❤️"); + assert_eq!(frequencies[1].count, 1); + + // Increase existing entry + let mut frequencies = vec![freq("👍", 2, false)]; + let old: Option<&Reaction> = None; + let new = Reaction::new("👍"); + modify_frequencies(&mut frequencies, old, &new); + assert_eq!(frequencies.len(), 1); + assert_eq!(frequencies[0].reaction.as_str(), "👍"); + assert_eq!(frequencies[0].count, 3); + + // Decreased existing entry + let mut frequencies = vec![freq("👍", 2, false)]; + let old = Some(Reaction::new("👍")); + let new = Reaction::new(""); + modify_frequencies(&mut frequencies, old.as_ref(), &new); + assert_eq!(frequencies.len(), 1); + assert_eq!(frequencies[0].reaction.as_str(), "👍"); + assert_eq!(frequencies[0].count, 1); + + // Remove existing entry + let mut frequencies = vec![freq("👍", 1, false)]; + let old = Some(Reaction::new("👍")); + let new = Reaction::new(""); + modify_frequencies(&mut frequencies, old.as_ref(), &new); + assert_eq!(frequencies.len(), 0); + + // Reaction changed: old reaction removed (count was 1), new reaction added + let mut frequencies = vec![freq("👍", 1, false), freq("❤️", 3, false)]; + let old = Some(Reaction::new("👍")); + let new = Reaction::new("🎉"); + modify_frequencies(&mut frequencies, old.as_ref(), &new); + assert_eq!(frequencies.len(), 2); + assert_eq!(frequencies[0].reaction.as_str(), "❤️"); + assert_eq!(frequencies[0].count, 3); + assert_eq!(frequencies[1].reaction.as_str(), "🎉"); + assert_eq!(frequencies[1].count, 1); + + // Reaction changed: old reaction decreased (count was 2), new reaction added + let mut frequencies = vec![freq("👍", 2, false)]; + let old = Some(Reaction::new("👍")); + let new = Reaction::new("🎉"); + modify_frequencies(&mut frequencies, old.as_ref(), &new); + assert_eq!(frequencies.len(), 2); + assert_eq!(frequencies[0].reaction.as_str(), "👍"); + assert_eq!(frequencies[0].count, 1); + assert_eq!(frequencies[1].reaction.as_str(), "🎉"); + assert_eq!(frequencies[1].count, 1); + + // Old and new reaction are the same + let mut frequencies = vec![freq("👍", 2, false)]; + let old = Some(Reaction::new("👍")); + let new = Reaction::new("👍"); + modify_frequencies(&mut frequencies, old.as_ref(), &new); + assert_eq!(frequencies.len(), 1); + assert_eq!(frequencies[0].reaction.as_str(), "👍"); + assert_eq!(frequencies[0].count, 2); + + // Empty frequencies array, adding a new reaction + let mut frequencies = vec![]; + let old: Option<&Reaction> = None; + let new = Reaction::new("👍"); + modify_frequencies(&mut frequencies, old, &new); + assert_eq!(frequencies.len(), 1); + assert_eq!(frequencies[0].reaction.as_str(), "👍"); + assert_eq!(frequencies[0].count, 1); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_refine_frequencies() { + // Test for empty inputs + let broadcasted = vec![]; + let by_contact: BTreeMap = BTreeMap::new(); + let result = refine_frequencies(broadcasted, &by_contact); + assert!(result.is_empty()); + + // Test broadcasted reactions only, no by_contact reactions + let broadcasted = vec![ReactionFrequency { + reaction: Reaction::new("👍"), + count: 2, + is_from_self: false, + }]; + let by_contact: BTreeMap = BTreeMap::new(); + let result = refine_frequencies(broadcasted, &by_contact); + assert_eq!(result.len(), 1); + assert_eq!(result[0].reaction.as_str(), "👍"); + assert_eq!(result[0].count, 2); + assert_eq!(result[0].is_from_self, false); + + // Test `by_contact` adding a completely new reaction not yet in `broadcasted` + let broadcasted = vec![ReactionFrequency { + reaction: Reaction::new("👍"), + count: 2, + is_from_self: false, + }]; + let mut by_contact: BTreeMap = BTreeMap::new(); + by_contact.insert(ContactId::new(10), Reaction::new("❤️")); + let result = refine_frequencies(broadcasted, &by_contact); + assert_eq!(result.len(), 2); + assert_eq!(result[0].reaction.as_str(), "👍"); + assert_eq!(result[0].count, 2); + assert_eq!(result[1].reaction.as_str(), "❤️"); + assert_eq!(result[1].count, 1); + assert_eq!(result[1].is_from_self, false); + + // Test `by_contact` contains SELF reaction, ensuring it is marked correctly + let broadcasted = vec![ + ReactionFrequency { + reaction: Reaction::new("❤️"), + count: 1, + is_from_self: false, + }, + ReactionFrequency { + reaction: Reaction::new("👍"), + count: 2, + is_from_self: false, + }, + ]; + let mut by_contact: BTreeMap = BTreeMap::new(); + by_contact.insert(ContactId::SELF, Reaction::new("❤️")); + let result = refine_frequencies(broadcasted, &by_contact); + assert_eq!(result.len(), 2); + assert_eq!(result[0].reaction.as_str(), "👍"); + assert_eq!(result[0].is_from_self, false); + assert_eq!(result[1].reaction.as_str(), "❤️"); + assert_eq!(result[1].is_from_self, true); + + // Test `by_contact` contains a reaction already in broadcasted; count must NOT increase + let broadcasted = vec![ReactionFrequency { + reaction: Reaction::new("👍"), + count: 2, + is_from_self: false, + }]; + let mut by_contact: BTreeMap = BTreeMap::new(); + by_contact.insert(ContactId::new(10), Reaction::new("👍")); + let result = refine_frequencies(broadcasted, &by_contact); + assert_eq!(result.len(), 1); + assert_eq!(result[0].reaction.as_str(), "👍"); + assert_eq!(result[0].count, 2); + + // Test scenario with multiple contacts, overlapping reactions, and SELF + let broadcasted = vec![ReactionFrequency { + reaction: Reaction::new("👍"), + count: 3, + is_from_self: false, + }]; + let mut by_contact: BTreeMap = BTreeMap::new(); + by_contact.insert(ContactId::new(11), Reaction::new("👍")); + by_contact.insert(ContactId::SELF, Reaction::new("❤️")); + let result = refine_frequencies(broadcasted, &by_contact); + + assert_eq!(result.len(), 2); + assert_eq!(result[0].reaction.as_str(), "👍"); + assert_eq!(result[0].count, 3); + assert_eq!(result[0].is_from_self, false); + + assert_eq!(result[1].reaction.as_str(), "❤️"); + assert_eq!(result[1].count, 1); + assert_eq!(result[1].is_from_self, true); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_broadcast_channel_reaction() -> Result<()> { + let mut tcm = TestContextManager::new(); + let alice = &tcm.alice().await; + let bob = &tcm.bob().await; + let claire = &tcm.charlie().await; + + // Alice creates a channel + let alice_chat_id = create_broadcast(alice, "Channel".to_string()).await?; + let qr = get_securejoin_qr(alice, Some(alice_chat_id)).await?; + + // Bob and claire join the channel via QR code + let bob_chat_id = tcm.exec_securejoin_qr(bob, alice, &qr).await; + bob_chat_id.accept(bob).await?; + let claire_chat_id = tcm.exec_securejoin_qr(claire, alice, &qr).await; + claire_chat_id.accept(claire).await?; + + // Alice sends a message to the channel + let sent_msg = alice.send_text(alice_chat_id, "hi channel!").await; + let alice_msg_id = sent_msg.load_from_db().await.id; + + // Bob and Claire receive the message + let bob_msg = bob.recv_msg(&sent_msg).await; + let claire_msg = claire.recv_msg(&sent_msg).await; + assert_eq!(bob_msg.get_text(), "hi channel!"); + assert_eq!(claire_msg.get_text(), "hi channel!"); + + // Bob reacts to the message + send_reaction(bob, bob_msg.id, "🏳️‍🌈").await?; + let sent_msg = bob.pop_sent_msg().await; + let reactions = get_msg_reactions(bob, bob_msg.id).await?; + assert_eq!(reactions.to_string(), "🏳️‍🌈1"); + + // Alice receives Bob's reaction + alice.recv_msg_hidden(&sent_msg).await; + let reactions = get_msg_reactions(alice, alice_msg_id).await?; + assert_eq!(reactions.to_string(), "🏳️‍🌈1"); + + // Alice broadcasts recent reaction changes to Bob and Claire. + // On the wire, the hidden message has a header like + // `Chat-Broadcast-Reactions: {"messages":[{"id":"123@adc","reactions":[{"emoji":"🏳️‍🌈","count":1}]}]}` + maybe_broadcast_reactions(alice).await?; + let sent_msg = alice.pop_sent_msg().await; + bob.recv_msg_hidden(&sent_msg).await; + claire.recv_msg_hidden(&sent_msg).await; + + // Check that there is nothing left for Alice to broadcast + maybe_broadcast_reactions(alice).await?; + broadcast_reactions_for_all_chats(alice).await?; + assert!(alice.pop_sent_msg_opt().await.is_none()); + + // Claire got the broadcasted reaction, and then reacts herself. + // This means, her local view on reactions are a mix `broadcasted_reactions`and `reactions`. + let reactions = get_msg_reactions(claire, claire_msg.id).await?; + assert_eq!(reactions.to_string(), "🏳️‍🌈1"); + assert_eq!(reactions.frequencies.len(), 1); + assert_eq!(reactions.by_contact.len(), 0); + + send_reaction(claire, claire_msg.id, "💪").await?; + let reactions = get_msg_reactions(claire, claire_msg.id).await?; + assert_eq!(reactions.to_string(), "🏳️‍🌈1 💪1"); + assert_eq!(reactions.frequencies.len(), 2); + assert_eq!(reactions.frequencies[0].is_from_self, false); + assert_eq!(reactions.frequencies[1].is_from_self, true); + + // Claire's reaction is sent to Alice who in turn broadcast it again to Bob and Claire. + // This must not modify Claire's get_reactions() even tho the reaction is present now in `broadcasted_reactions` and `reactions`. + let sent_msg = claire.pop_sent_msg().await; + alice.recv_msg_hidden(&sent_msg).await; + let reactions = get_msg_reactions(alice, alice_msg_id).await?; + assert_eq!(reactions.to_string(), "🏳️‍🌈1 💪1"); + + broadcast_reactions_for_all_chats(alice).await?; // bypass timer in maybe_broadcast_reactions() + let sent_msg = alice.pop_sent_msg().await; + bob.recv_msg_hidden(&sent_msg).await; + + claire.recv_msg_hidden(&sent_msg).await; + let reactions = get_msg_reactions(claire, claire_msg.id).await?; + assert_eq!(reactions.to_string(), "🏳️‍🌈1 💪1"); + assert_eq!(reactions.frequencies.len(), 2); + assert_eq!(reactions.frequencies[0].is_from_self, false); + assert_eq!(reactions.frequencies[1].is_from_self, true); + + // Claire removes her 💪 reaction, and also reactios with 🏳️‍🌈; + // SELF-changes are immediate even tho not broadcasted yet, the bring broadcasted reactions table to a "dirty state" ... + send_reaction(claire, claire_msg.id, "").await?; + let sent_msg = claire.pop_sent_msg().await; + let reactions = get_msg_reactions(claire, claire_msg.id).await?; + assert_eq!(reactions.to_string(), "🏳️‍🌈1"); + + send_reaction(claire, claire_msg.id, "🏳️‍🌈").await?; + let sent_msg2 = claire.pop_sent_msg().await; + let reactions = get_msg_reactions(claire, claire_msg.id).await?; + assert_eq!(reactions.to_string(), "🏳️‍🌈2"); + + // ... "dirty state" is fixed after next broadcast then, counters should stay the same + alice.recv_msg_hidden(&sent_msg).await; + alice.recv_msg_hidden(&sent_msg2).await; + broadcast_reactions_for_all_chats(alice).await?; // bypass timer in maybe_broadcast_reactions() + let sent_msg = alice.pop_sent_msg().await; + claire.recv_msg_hidden(&sent_msg).await; + let reactions = get_msg_reactions(claire, claire_msg.id).await?; + assert_eq!(reactions.to_string(), "🏳️‍🌈2"); + + Ok(()) + } +} diff --git a/src/receive_imf.rs b/src/receive_imf.rs index 72ced0958a..27317fd503 100644 --- a/src/receive_imf.rs +++ b/src/receive_imf.rs @@ -41,6 +41,7 @@ use crate::mimeparser::{ }; use crate::param::{Param, Params}; use crate::peer_channels::{add_gossip_peer_from_header, insert_topic_stub, iroh_topic_from_str}; +use crate::reaction::broadcast_reactions::receive_broadcast_reactions; use crate::reaction::{Reaction, set_msg_reaction}; use crate::rusqlite::OptionalExtension; use crate::securejoin::{ @@ -901,6 +902,12 @@ UPDATE config SET value=? WHERE keyname='configured_addr' AND value!=?1 } } + if let Some(broadcast_reactions) = &mime_parser.broadcast_reactions + && let Err(err) = receive_broadcast_reactions(context, broadcast_reactions).await + { + warn!(context, "Cannot apply broadcast reactions: {err:#}."); + } + if let Some(avatar_action) = &mime_parser.user_avatar && !matches!(from_id, ContactId::UNDEFINED | ContactId::SELF) && context diff --git a/src/scheduler.rs b/src/scheduler.rs index f44a4021f4..d50f963678 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -20,6 +20,7 @@ use crate::events::EventType; use crate::imap::{Imap, session::Session}; use crate::location; use crate::log::{LogExt, warn}; +use crate::reaction::broadcast_reactions::maybe_broadcast_reactions; use crate::smtp::{Smtp, send_smtp_messages}; use crate::sql; use crate::stats::maybe_send_stats; @@ -448,6 +449,7 @@ async fn inbox_fetch_idle(ctx: &Context, imap: &mut Imap, mut session: Session) } }; + maybe_broadcast_reactions(ctx).await.log_err(ctx).ok(); maybe_send_stats(ctx).await.log_err(ctx).ok(); session diff --git a/src/sql/migrations.rs b/src/sql/migrations.rs index f435cc2dba..6a060756de 100644 --- a/src/sql/migrations.rs +++ b/src/sql/migrations.rs @@ -2529,6 +2529,32 @@ UPDATE msgs SET state=24 WHERE state=18; -- Change OutPreparing to OutFailed. fingerprint TEXT PRIMARY KEY NOT NULL, -- Upper-case fingerprint of the recipient key. attached_timestamp INTEGER NOT NULL ) STRICT", + + migration_version, + ) + .await?; + } + + inc_and_check(&mut migration_version, 161)?; + if dbversion < migration_version { + // `broadcasted_reactions` stores accumulated reactions for broadcast channel subscribers (Chattype::InBroadcast). + // `broadcasted_reactions` is unused for broadcast channel owners (Chattype::OutBroadcast), + // there `reactions_need_broadcast` is used to find out new reactions to be sent to subscribers. + sql.execute_migration( + "CREATE TABLE broadcasted_reactions ( + msg_id INTEGER NOT NULL DEFAULT 0, + reaction TEXT NOT NULL DEFAULT '', + count INTEGER NOT NULL DEFAULT 0, + FOREIGN KEY(msg_id) REFERENCES msgs(id) ON DELETE CASCADE -- delete reactions when message is deleted + ) STRICT; + CREATE INDEX broadcasted_reactions_index1 ON broadcasted_reactions (msg_id); + CREATE TABLE reactions_need_broadcast ( + chat_id INTEGER NOT NULL DEFAULT 0, + msg_id INTEGER NOT NULL DEFAULT 0, + UNIQUE (chat_id, msg_id), + FOREIGN KEY(msg_id) REFERENCES msgs(id) ON DELETE CASCADE -- delete reactions when message is deleted + ) STRICT; + CREATE INDEX reactions_need_broadcast_index1 ON reactions_need_broadcast (chat_id);", migration_version, ) .await?; diff --git a/test-data/golden/test_broadcast_joining_golden_alice b/test-data/golden/test_broadcast_joining_golden_alice index 84602fc4c1..1c83761c86 100644 --- a/test-data/golden/test_broadcast_joining_golden_alice +++ b/test-data/golden/test_broadcast_joining_golden_alice @@ -1,4 +1,4 @@ -OutBroadcast#Chat#1001: My Channel [1 member(s)] Icon: e9b6c7a78aa2e4f415644f55a553e73.png +OutBroadcast#Chat#1001: My Channel [1 member(s)]🔇 Icon: e9b6c7a78aa2e4f415644f55a553e73.png -------------------------------------------------------------------------------- Msg#1001: info (Contact#Contact#Info): Messages are end-to-end encrypted. [NOTICED][INFO] Msg#1002🔒: Me (Contact#Contact#Self): Channel image changed. [INFO] √ diff --git a/test-data/golden/test_sync_broadcast_alice1 b/test-data/golden/test_sync_broadcast_alice1 index 23c767db56..94e15a733c 100644 --- a/test-data/golden/test_sync_broadcast_alice1 +++ b/test-data/golden/test_sync_broadcast_alice1 @@ -1,4 +1,4 @@ -OutBroadcast#Chat#1001: Channel [0 member(s)] +OutBroadcast#Chat#1001: Channel [0 member(s)]🔇 -------------------------------------------------------------------------------- Msg#1001: info (Contact#Contact#Info): Messages are end-to-end encrypted. [NOTICED][INFO] Msg#1006🔒: Me (Contact#Contact#Self): Member bob@example.net added. [INFO] √ diff --git a/test-data/golden/test_sync_broadcast_alice2 b/test-data/golden/test_sync_broadcast_alice2 index 277f0f8e6e..6f624bb0cd 100644 --- a/test-data/golden/test_sync_broadcast_alice2 +++ b/test-data/golden/test_sync_broadcast_alice2 @@ -1,4 +1,4 @@ -OutBroadcast#Chat#1001: Channel [0 member(s)] +OutBroadcast#Chat#1001: Channel [0 member(s)]🔇 -------------------------------------------------------------------------------- Msg#1002: info (Contact#Contact#Info): Messages are end-to-end encrypted. [NOTICED][INFO] Msg#1006🔒: Me (Contact#Contact#Self): Member bob@example.net added. [INFO] √