diff --git a/src/reaction.rs b/src/reaction.rs index 279c1afa1f..8436db30a9 100644 --- a/src/reaction.rs +++ b/src/reaction.rs @@ -199,6 +199,40 @@ async fn set_msg_id_reaction( Ok(()) } +/// Adds or updates a pending reaction to `pending_reactions` table. +async fn set_pending_reaction( + context: &Context, + rfc724_mid: &str, + chat_id: ChatId, + contact_id: ContactId, + timestamp: i64, + reaction: &Reaction, +) -> Result<()> { + if reaction.is_empty() { + context + .sql + .execute( + "DELETE FROM pending_reactions + WHERE rfc724_mid = ?1 + AND contact_id = ?2", + (rfc724_mid, contact_id), + ) + .await?; + } else { + context + .sql + .execute( + "INSERT INTO pending_reactions (rfc724_mid, contact_id, chat_id, reaction, timestamp) + VALUES (?1, ?2, ?3, ?4, ?5) + ON CONFLICT(rfc724_mid, contact_id) + DO UPDATE SET reaction=excluded.reaction", + (rfc724_mid, contact_id, chat_id, reaction.as_str(), timestamp), + ) + .await?; + } + Ok(()) +} + /// Sends a reaction to message `msg_id`, overriding previously sent reactions. /// /// `reaction` is a string consisting of a single emoji. Use @@ -235,6 +269,9 @@ pub async fn send_reaction(context: &Context, msg_id: MsgId, reaction: &str) -> /// /// `reaction` is string representing the emoji. It can be empty /// if contact wants to remove the reaction. +/// +/// Delegates to [`set_msg_reaction_if_present`], +/// and if message is not present, fallbacks to [`set_pending_reaction`]. pub(crate) async fn set_msg_reaction( context: &Context, in_reply_to: &str, @@ -244,8 +281,46 @@ pub(crate) async fn set_msg_reaction( reaction: Reaction, is_incoming_fresh: bool, ) -> Result<()> { + if !set_msg_reaction_if_present( + context, + in_reply_to, + chat_id, + contact_id, + timestamp, + &reaction, + is_incoming_fresh, + ) + .await? + { + info!( + context, + "Can't assign reaction to unknown message with Message-ID {}; inserting into pending table.", + in_reply_to + ); + let rfc724_mid = in_reply_to.trim_start_matches('<').trim_end_matches('>'); + set_pending_reaction( + context, rfc724_mid, chat_id, contact_id, timestamp, &reaction, + ) + .await? + } + Ok(()) +} + +/// Similar to [`set_msg_reaction`], +/// but does not create a row in `pending_messages` if message is not present. +/// +/// Returns `Ok(true)` if message was present and `Ok(false)` if not. +pub(crate) async fn set_msg_reaction_if_present( + context: &Context, + in_reply_to: &str, + chat_id: ChatId, + contact_id: ContactId, + timestamp: i64, + reaction: &Reaction, + is_incoming_fresh: bool, +) -> Result { if let Some(msg_id) = rfc724_mid_exists(context, in_reply_to).await? { - set_msg_id_reaction(context, msg_id, chat_id, contact_id, timestamp, &reaction).await?; + set_msg_id_reaction(context, msg_id, chat_id, contact_id, timestamp, reaction).await?; if is_incoming_fresh && !reaction.is_empty() @@ -255,15 +330,69 @@ pub(crate) async fn set_msg_reaction( chat_id, contact_id, msg_id, - reaction, + reaction: reaction.clone(), }); } - } else { - info!( - context, - "Can't assign reaction to unknown message with Message-ID {}", in_reply_to - ); + return Ok(true); + } + Ok(false) +} + +/// Applies pending reactions to message `rfc724_mid`. +pub(crate) async fn apply_pending_reactions(context: &Context, rfc724_mid: &str) -> Result<()> { + let pending_reactions: BTreeMap = context + .sql + .query_map_collect( + "SELECT contact_id, chat_id, reaction, timestamp + FROM pending_reactions + WHERE rfc724_mid=?", + (rfc724_mid,), + |row| { + let contact_id: ContactId = row.get(0)?; + let chat_id: ChatId = row.get(1)?; + let reaction: Reaction = Reaction::new(row.get::<_, String>(2)?.as_str()); + let timestamp: i64 = row.get(3)?; + Ok((contact_id, (chat_id, reaction, timestamp))) + }, + ) + .await?; + + if pending_reactions.is_empty() { + return Ok(()); + } + + info!( + context, + "Applying {} pending reactions to {}.", + pending_reactions.len(), + rfc724_mid + ); + + for (contact_id, (chat_id, reaction, timestamp)) in pending_reactions { + if !set_msg_reaction_if_present( + context, rfc724_mid, chat_id, contact_id, timestamp, &reaction, + true, // todo: how to treat this? + ) + .await? + { + return Err(anyhow::Error::msg( + "Message {rfc724_mid} is not present, can't apply pending reactions!", + )); + } } + + // Note: race condition can't happen here, + // as at this point the message is already added to the DB, + // so no new pending reactions with this rfc724_mid will be added in meantime. + // (Assuming this function is used after receiving the message.) + context + .sql + .execute( + "DELETE FROM pending_reactions WHERE rfc724_mid=?", + (rfc724_mid,), + ) + .await?; + Ok(()) } @@ -713,6 +842,26 @@ Content-Disposition: reaction\n\ Ok(()) } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_send_out_of_order_reaction() -> Result<()> { + let alice = TestContext::new_alice().await; + let bob = TestContext::new_bob().await; + + let chat_alice = alice.create_chat(&bob).await; + let alice_msg1 = alice.send_text(chat_alice.id, "1").await; + let bob_msg = bob.recv_msg(&alice_msg1).await; + bob_msg.chat_id.accept(&bob).await?; + + let alice_msg2 = alice.send_text(chat_alice.id, "2").await; + send_reaction(&alice, alice_msg2.sender_msg_id, "👍").await?; + let reaction_msg = alice.pop_sent_msg().await; + + bob.recv_msg_hidden(&reaction_msg).await; + let msg = bob.recv_msg(&alice_msg2).await; + assert_eq!(get_msg_reactions(&bob, msg.id).await?.reactions.len(), 1); + Ok(()) + } + async fn assert_summary(t: &TestContext, expected: &str) { let chatlist = Chatlist::try_load(t, 0, None, None).await.unwrap(); let summary = chatlist.get_summary(t, 0, None).await.unwrap(); diff --git a/src/receive_imf.rs b/src/receive_imf.rs index c91cdea22b..f61902d4f8 100644 --- a/src/receive_imf.rs +++ b/src/receive_imf.rs @@ -743,6 +743,10 @@ pub(crate) async fn receive_imf_inner( .context("add_parts error")? }; + crate::reaction::apply_pending_reactions(context, rfc724_mid) + .await + .context("failed to apply pending reactions")?; + if !from_id.is_special() { contact::update_last_seen(context, from_id, mime_parser.timestamp_sent).await?; } diff --git a/src/sql.rs b/src/sql.rs index ef825e61a1..0d59281ea5 100644 --- a/src/sql.rs +++ b/src/sql.rs @@ -917,10 +917,28 @@ pub async fn housekeeping(context: &Context) -> Result<()> { .log_err(context) .ok(); + remove_old_pending_reactions(context) + .await + .context("Failed to remove old pending reactions") + .log_err(context) + .ok(); + info!(context, "Housekeeping done."); Ok(()) } +/// Removes pending reactions that weren't applied for 90 days. +async fn remove_old_pending_reactions(context: &Context) -> Result { + let three_months_ago = time().saturating_sub(60 * 60 * 24 * 90); + context + .sql + .execute( + "DELETE FROM pending_reactions WHERE timestamp < ?", + (three_months_ago,), + ) + .await +} + /// Removes transports that are hidden (`is_published=0`), /// and haven't been used to receive new messages for [`UNPUBLISHED_TRANSPORT_KEEP_TIME`] seconds. pub(crate) async fn remove_unused_hidden_transports(context: &Context) -> Result { diff --git a/src/sql/migrations.rs b/src/sql/migrations.rs index a0fd79eda6..2e83fd3954 100644 --- a/src/sql/migrations.rs +++ b/src/sql/migrations.rs @@ -2477,6 +2477,24 @@ UPDATE msgs SET state=24 WHERE state=18; -- Change OutPreparing to OutFailed. .await?; } + inc_and_check(&mut migration_version, 156)?; + if dbversion < migration_version { + // Stores reactions to not-yet-received messages. + sql.execute_migration( + "CREATE TABLE pending_reactions ( + rfc724_mid TEXT NOT NULL, + contact_id INTEGER NOT NULL, + chat_id INTEGER NOT NULL, + reaction TEXT DEFAULT '' NOT NULL, + timestamp INTEGER DEFAULT 0 NOT NULL, + PRIMARY KEY(rfc724_mid, contact_id), + FOREIGN KEY(contact_id) REFERENCES contacts(id) ON DELETE CASCADE + )", + migration_version, + ) + .await?; + } + let new_version = sql .get_raw_config_int(VERSION_CFG) .await?