fix: Apply reactions that arrived before the message at later time

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<bool> {
     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<ContactId, (ChatId, Reaction, i64)> = 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();

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?;
     }

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<usize> {
+    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<usize> {

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?