Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-09-22 13:01:21 +03:00
Code Issues Packages Projects Releases Wiki Activity

broadcast channel reactions (#8450)

Browse Source 
this PR adds support for reactions in broadcast channels.

> the idea of broadcast reactions is that they are sent as usual from
subscribers to owner. after some time, the owner broadcasts them to all
subscribers, who only get to see reaction+count, not who-reacted-what

the PR is quite large, but a good share are tests and otherwise many
things are straight forward.

review should be done by-file, not by-commit. to make review easier,
here is a high-level overview:

first a change in the existing internal `Reactions` object was required.
before this PR, `Reactions` had a "contact to reaction map" only, and
the "frequencies map", that are actually mainly needed for UI, were
calculated as needed. with this PR, the "frequencies map" is the field
that always exist, the "contact map" is only available on top of that.
moreover, this PR shifts that part to the core, it was unfortunately in
the bindings before.

with that preparation things done as follows:

1. reactions from broadcast channel subscriber (`Chattype::InBroadcast`)
to broadcast channel owner (`Chattype::OutBroadcast`) are sent as usual,
only change for that step was to allow sending them at all

2. the owner receives reactions and saves them to the existing
`reactions` table as usual. additionally, the changed message is
remembered in `reactions_need_broadcast` table

3. in the IMAP loop, when ~10 minutes have passed, and
`reactions_need_broadcast` contains entries, a single, hidden message
with accumulated reactions is sent. this message may contain reactions
to different messages. for each message, all known reactions are sent as
reaction+count.

4. subscriber receive that message and save the accumulated reactions in
`reactions_broadcasted`

6. `get_message_reactions` is adapted so that `frequencies` are set
independently of who-reacted-what (the old and only field).
who-reacted-what is called `by_contact` now. it is always set for
compatibility reasons, however, it is not exhaustive for subscribers.
in general, UI should work with frequencies, the API itself, however,
has not changed.

other tweaks:

- outgoing channels are muted on creation, and UI shall allow to
unmute/mute them as all other chats. reason is that reactions are
notified, but in many cases not of large interest. this is also what
telegram is doing

- to have an intermediate feedback when reacting, the local state should
include ones own reaction, even if it is not yet broadcasted. for that,
we modify `reactions_broadcasted` using `modify_frequencies()` as needed
when sending an reaction. there are still some situations where the
update may not include ones own reaction, in this case it is added
lately by `refine_frequencies()`, so that `get_message_reactions()`
always contain SELF.
(in a first implementation, we always increased SELF reaction in
refine_frequencies(), however, that was worse and led to SELF counted
twice once the owner sent broadcast)

<details>
<summary>wire format</summary>

wire format is a JSON in the `Broadcast-Reactions:` header.
additionally, `Content-Disposition: reaction` is set to not show the
hidden message on existing devices.

using a header also allows us to broadcast reactions with resent channel
messages (on joining) later.

```
{
  "messages": [
    {
      "id": "12345678",
      "reactions": [
        { "emoji": "👍", "count": 4 },
        { "emoji": "🎉", "count": 2 }
      ]
    },
    {
      "id": "23456789",
      "reactions": []
    }
  ]
}
```

for `id`, the wire format needs to use `rfc724_mid` as `msg_id` are
local only.

</details>

### known issues

- if the channel owner uses multiple devices, broadcasted reaction
updates are sent from each device. the updates are not that big, so that
is probably not a big deal. if it turns out that this is an issue, we
can think about fixes in another PR. might be done by restarting our
10-minute-wait once we see an update from another device

- we cannot set contact_id for DC_EVENT_REACTIONS_CHANGED - but i doubt
it was ever used

### for another pr

- ~~add `Broadcast-Reactions:` header also for resent channel messages,
so that new subscriber do not only get the latest messages, but also
their reactions. for that, the `Broadcast-Reactions:` header can go to
the corresponding message, no need to send extra messages. we would need
to change the sending part to send all reactions for a given message. on
receiving part, we need to make sure, `receive_broadcast_reactions()` is
called when the message actually exist.~~
EDIT: subsequent PR for resending broadcast reactions at
https://github.com/chatmail/core/pull/8496

- add api to allow only a subset of reactions, fiter incoming reactions
before broadcasting

### misc.

ui pr: https://github.com/deltachat/deltachat-ios/pull/3225 and
https://github.com/deltachat/deltachat-android/pull/4560 , which both
were tested successfully with this core PR already. desktop is meant to
be done once this is merged

---------

Co-authored-by: l <link2xt@testrun.org>
This commit is contained in:
biørn
2026-08-04 18:54:17 +02:00
committed by GitHub
parent 547d22a3e3
commit bd846c6e43
18 changed files with 910 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/chat.rs
View File

@@ -2687,10 +2687,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
@@ -3710,14 +3711,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(),
),
)?;

23
src/chat/chat_tests.rs
View File

@@ -3056,6 +3056,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();

3
src/config.rs
View File

@@ -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,

6
src/context.rs
View File

@@ -956,6 +956,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)

2
src/events/payload.rs
View File

@@ -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,
},

4
src/headerdef.rs
View File

@@ -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,

7
src/mimefactory.rs
View File

@@ -1986,6 +1986,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

12
src/mimeparser.rs
View File

@@ -116,6 +116,10 @@ pub(crate) struct MimeMessage {
pub(crate) mdn_reports: Vec<Report>,
pub(crate) delivery_report: Option<DeliveryReport>,
/// Parsed `Chat-Broadcast-Reactions` header, if any:
/// accumulated reaction updates sent by a broadcast channel owner.
pub(crate) broadcast_reactions: Option<String>,
/// 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();
}

5
src/param.rs
View File

@@ -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',

204
src/reaction.rs
View File

@@ -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<ReactionFrequency>,
/// Map from a contact to its reaction to message.
reactions: BTreeMap<ContactId, Reaction>,
/// For channels subscribers, this map is empty or contains `ContactId::SELF` only.
pub by_contact: BTreeMap<ContactId, Reaction>,
}
impl Reactions {
/// Returns vector of contacts that reacted to the message.
pub fn contacts(&self) -> Vec<ContactId> {
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<String, usize> {
let mut emoji_frequencies: BTreeMap<String, usize> = 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<Item = (&ContactId, &Reaction)> {
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<ContactId, Reaction>) -> Vec<ReactionFrequency> {
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<ReactionFrequency> = 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<Reactions> {
let mut reactions: BTreeMap<ContactId, Reaction> = context
let mut by_contact: BTreeMap<ContactId, Reaction> = 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<React
},
)
.await?;
reactions.retain(|_contact, reaction| !reaction.is_empty());
Ok(Reactions { reactions })
by_contact.retain(|_contact, reaction| !reaction.is_empty());
let broadcasted_reactions = load_broadcast_reactions(context, msg_id).await?;
let frequencies = if !broadcasted_reactions.is_empty() {
refine_frequencies(broadcasted_reactions, &by_contact)
} else {
calc_frequencies(&by_contact)
};
Ok(Reactions {
frequencies,
by_contact,
})
}
impl Chat {
@@ -490,6 +537,20 @@ mod tests {
use crate::tools::SystemTime;
use std::time::Duration;
impl Reactions {
fn contacts(&self) -> Vec<ContactId> {
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(())
}
@@ -1230,11 +1289,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(())
}

663
src/reaction/broadcast_reactions.rs Normal file
View File

@@ -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<WireMessage>,
}
#[derive(Debug, Serialize, Deserialize)]
struct WireMessage {
/// RFC 724 Message-ID.
id: String,
/// Array of reaction entries.
reactions: Vec<WireEntry>,
}
#[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<ChatId> = 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<MsgId> = 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<WireMessage> = 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<WireEntry> = 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<ReactionFrequency> = 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<Vec<ReactionFrequency>> {
let mut frequencies: Vec<ReactionFrequency> = 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<ReactionFrequency>,
) -> 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<ReactionFrequency>,
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<ReactionFrequency>,
by_contact: &BTreeMap<ContactId, Reaction>,
) -> Vec<ReactionFrequency> {
// 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<ContactId, Reaction> = 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<ContactId, Reaction> = 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<ContactId, Reaction> = 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<ContactId, Reaction> = 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<ContactId, Reaction> = 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<ContactId, Reaction> = 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(())
}
}

7
src/receive_imf.rs
View File

@@ -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

2
src/scheduler.rs
View File

@@ -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

25
src/sql/migrations.rs
View File

@@ -2567,6 +2567,31 @@ UPDATE msgs SET state=24 WHERE state=18; -- Change OutPreparing to OutFailed.
.await?;
}
inc_and_check(&mut migration_version, 162)?;
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?;
}
let new_version = sql
.get_raw_config_int(VERSION_CFG)
.await?