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

src/reaction.rs

@@ -199,6 +199,39 @@ async fn set_msg_id_reaction(
     Ok(())
 }
 
+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
@@ -244,8 +277,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
+        );
+        set_pending_reaction(
+            context,
+            in_reply_to,
+            chat_id,
+            contact_id,
+            timestamp,
+            &reaction,
+        )
+        .await?
+    }
+    Ok(())
+}
+
+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 +326,57 @@ 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)
+}
+
+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?;
+
+    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,
+            false, // todo: how to treat this, should it also be stored in pending_reactions?
+        )
+        .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(())
 }
 

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/migrations.rs

@@ -2456,6 +2456,23 @@ UPDATE msgs SET state=24 WHERE state=18; -- Change OutPreparing to OutFailed.
         .await?;
     }
 
+    inc_and_check(&mut migration_version, 155)?;
+    if dbversion < migration_version {
+        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?