Merge remote-tracking branch 'origin/master' into hoc/migitate-from-forgery

src/contact.rs

@@ -34,7 +34,9 @@ const SEEN_RECENTLY_SECONDS: i64 = 600;
 ///
 /// Some contact IDs are reserved to identify special contacts.  This
 /// type can represent both the special as well as normal contacts.
-#[derive(Debug, Copy, Clone, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[derive(
+    Debug, Copy, Clone, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
+)]
 pub struct ContactId(u32);
 
 impl ContactId {

src/download.rs

@@ -11,7 +11,7 @@ use crate::imap::{Imap, ImapActionResult};
 use crate::job::{self, Action, Job, Status};
 use crate::message::{Message, MsgId, Viewtype};
 use crate::mimeparser::{MimeMessage, Part};
-use crate::param::{Param, Params};
+use crate::param::Params;
 use crate::tools::time;
 use crate::{job_try, stock_str, EventType};
 use std::cmp::max;
@@ -69,42 +69,6 @@ impl Context {
             Ok(Some(max(MIN_DOWNLOAD_LIMIT, download_limit as u32)))
         }
     }
-
-    // Merges the two messages to `placeholder_msg_id`;
-    // `full_msg_id` is no longer used afterwards.
-    pub(crate) async fn merge_messages(
-        &self,
-        full_msg_id: MsgId,
-        placeholder_msg_id: MsgId,
-    ) -> Result<()> {
-        let placeholder = Message::load_from_db(self, placeholder_msg_id).await?;
-        self.sql
-            .transaction(move |transaction| {
-                transaction
-                    .execute("DELETE FROM msgs WHERE id=?;", paramsv![placeholder_msg_id])?;
-                transaction.execute(
-                    "UPDATE msgs SET id=? WHERE id=?",
-                    paramsv![placeholder_msg_id, full_msg_id],
-                )?;
-                Ok(())
-            })
-            .await?;
-        let mut full = Message::load_from_db(self, placeholder_msg_id).await?;
-
-        for key in [
-            Param::WebxdcSummary,
-            Param::WebxdcSummaryTimestamp,
-            Param::WebxdcDocument,
-            Param::WebxdcDocumentTimestamp,
-        ] {
-            if let Some(value) = placeholder.param.get(key) {
-                full.param.set(key, value);
-            }
-        }
-        full.update_param(self).await?;
-
-        Ok(())
-    }
 }
 
 impl MsgId {

src/events.rs

@@ -173,6 +173,13 @@ pub enum EventType {
         msg_id: MsgId,
     },
 
+    /// Reactions for the message changed.
+    ReactionsChanged {
+        chat_id: ChatId,
+        msg_id: MsgId,
+        contact_id: ContactId,
+    },
+
     /// There is a fresh message. Typically, the user will show an notification
     /// when receiving this message.
     ///

src/imap.rs

@@ -2157,8 +2157,8 @@ pub(crate) async fn set_uid_next(context: &Context, folder: &str, uid_next: u32)
         .sql
         .execute(
             "INSERT INTO imap_sync (folder, uid_next) VALUES (?,?)
-                ON CONFLICT(folder) DO UPDATE SET uid_next=? WHERE folder=?;",
-            paramsv![folder, uid_next, uid_next, folder],
+                ON CONFLICT(folder) DO UPDATE SET uid_next=excluded.uid_next",
+            paramsv![folder, uid_next],
         )
         .await?;
     Ok(())
@@ -2189,8 +2189,8 @@ pub(crate) async fn set_uidvalidity(
         .sql
         .execute(
             "INSERT INTO imap_sync (folder, uidvalidity) VALUES (?,?)
-                ON CONFLICT(folder) DO UPDATE SET uidvalidity=? WHERE folder=?;",
-            paramsv![folder, uidvalidity, uidvalidity, folder],
+                ON CONFLICT(folder) DO UPDATE SET uidvalidity=excluded.uidvalidity",
+            paramsv![folder, uidvalidity],
         )
         .await?;
     Ok(())
@@ -2212,8 +2212,8 @@ pub(crate) async fn set_modseq(context: &Context, folder: &str, modseq: u64) ->
         .sql
         .execute(
             "INSERT INTO imap_sync (folder, modseq) VALUES (?,?)
-                ON CONFLICT(folder) DO UPDATE SET modseq=? WHERE folder=?;",
-            paramsv![folder, modseq, modseq, folder],
+                ON CONFLICT(folder) DO UPDATE SET modseq=excluded.modseq",
+            paramsv![folder, modseq],
         )
         .await?;
     Ok(())

src/lib.rs

@@ -104,6 +104,7 @@ pub mod receive_imf;
 pub mod tools;
 
 pub mod accounts;
+pub mod reaction;
 
 /// if set imap/incoming and smtp/outgoing MIME messages will be printed
 pub const DCC_MIME_DEBUG: &str = "DCC_MIME_DEBUG";

src/message.rs

@@ -22,6 +22,7 @@ use crate::imap::markseen_on_imap_table;
 use crate::mimeparser::{parse_message_id, DeliveryReport, SystemMessage};
 use crate::param::{Param, Params};
 use crate::pgp::split_armored_data;
+use crate::reaction::get_msg_reactions;
 use crate::scheduler::InterruptInfo;
 use crate::sql;
 use crate::stock_str;
@@ -751,6 +752,11 @@ impl Message {
         self.param.set_int(Param::Duration, duration);
     }
 
+    /// Marks the message as reaction.
+    pub(crate) fn set_reaction(&mut self) {
+        self.param.set_int(Param::Reaction, 1);
+    }
+
     pub async fn latefiling_mediasize(
         &mut self,
         context: &Context,
@@ -1082,6 +1088,11 @@ pub async fn get_msg_info(context: &Context, msg_id: MsgId) -> Result<String> {
 
     ret += "\n";
 
+    let reactions = get_msg_reactions(context, msg_id).await?;
+    if !reactions.is_empty() {
+        ret += &format!("Reactions: {}\n", reactions);
+    }
+
     if let Some(error) = msg.error.as_ref() {
         ret += &format!("Error: {}", error);
     }

src/mimefactory.rs

@@ -183,7 +183,10 @@ impl<'a> MimeFactory<'a> {
                 )
                 .await?;
 
-            if !msg.is_system_message() && context.get_config_bool(Config::MdnsEnabled).await? {
+            if !msg.is_system_message()
+                && msg.param.get_int(Param::Reaction).unwrap_or_default() == 0
+                && context.get_config_bool(Config::MdnsEnabled).await?
+            {
                 req_mdn = true;
             }
         }
@@ -1122,6 +1125,11 @@ impl<'a> MimeFactory<'a> {
                 "text/plain; charset=utf-8; format=flowed; delsp=no".to_string(),
             ))
             .body(message_text);
+
+        if self.msg.param.get_int(Param::Reaction).unwrap_or_default() != 0 {
+            main_part = main_part.header(("Content-Disposition", "reaction"));
+        }
+
         let mut parts = Vec::new();
 
         // add HTML-part, this is needed only if a HTML-message from a non-delta-client is forwarded;

src/mimeparser.rs

@@ -557,7 +557,10 @@ impl MimeMessage {
             }
 
             if prepend_subject && !subject.is_empty() {
-                let part_with_text = self.parts.iter_mut().find(|part| !part.msg.is_empty());
+                let part_with_text = self
+                    .parts
+                    .iter_mut()
+                    .find(|part| !part.msg.is_empty() && !part.is_reaction);
                 if let Some(mut part) = part_with_text {
                     part.msg = format!("{} – {}", subject, part.msg);
                 }
@@ -919,6 +922,7 @@ impl MimeMessage {
         Ok(any_part_added)
     }
 
+    /// Returns true if any part was added, false otherwise.
     async fn add_single_part_if_known(
         &mut self,
         context: &Context,
@@ -952,6 +956,30 @@ impl MimeMessage {
                         warn!(context, "Missing attachment");
                         return Ok(false);
                     }
+                    mime::TEXT
+                        if mail.get_content_disposition().disposition
+                            == DispositionType::Extension("reaction".to_string()) =>
+                    {
+                        // Reaction.
+                        let decoded_data = match mail.get_body() {
+                            Ok(decoded_data) => decoded_data,
+                            Err(err) => {
+                                warn!(context, "Invalid body parsed {:?}", err);
+                                // Note that it's not always an error - might be no data
+                                return Ok(false);
+                            }
+                        };
+
+                        let part = Part {
+                            typ: Viewtype::Text,
+                            mimetype: Some(mime_type),
+                            msg: decoded_data,
+                            is_reaction: true,
+                            ..Default::default()
+                        };
+                        self.do_add_single_part(part);
+                        return Ok(true);
+                    }
                     mime::TEXT | mime::HTML => {
                         let decoded_data = match mail.get_body() {
                             Ok(decoded_data) => decoded_data,
@@ -1650,6 +1678,9 @@ pub struct Part {
     /// note that multipart/related may contain further multipart nestings
     /// and all of them needs to be marked with `is_related`.
     pub(crate) is_related: bool,
+
+    /// Part is an RFC 9078 reaction.
+    pub(crate) is_reaction: bool,
 }
 
 /// return mimetype and viewtype for a parsed mail
@@ -3335,4 +3366,39 @@ Message.
 
         Ok(())
     }
+
+    /// Tests parsing of MIME message containing RFC 9078 reaction.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_parse_reaction() -> Result<()> {
+        let alice = TestContext::new_alice().await;
+
+        let mime_message = MimeMessage::from_bytes(
+            &alice,
+            "To: alice@example.org\n\
+From: bob@example.net\n\
+Date: Today, 29 February 2021 00:00:10 -800\n\
+Message-ID: 56789@example.net\n\
+In-Reply-To: 12345@example.org\n\
+Subject: Meeting\n\
+Mime-Version: 1.0 (1.0)\n\
+Content-Type: text/plain; charset=utf-8\n\
+Content-Disposition: reaction\n\
+\n\
+\u{1F44D}"
+                .as_bytes(),
+        )
+        .await?;
+
+        assert_eq!(mime_message.parts.len(), 1);
+        assert_eq!(mime_message.parts[0].is_reaction, true);
+        assert_eq!(
+            mime_message
+                .get_header(HeaderDef::InReplyTo)
+                .and_then(|msgid| parse_message_id(msgid).ok())
+                .unwrap(),
+            "12345@example.org"
+        );
+
+        Ok(())
+    }
 }

src/param.rs

@@ -59,6 +59,9 @@ pub enum Param {
     /// For Messages
     WantsMdn = b'r',
 
+    /// For Messages: the message is a reaction.
+    Reaction = b'x',
+
     /// For Messages: a message with Auto-Submitted header ("bot").
     Bot = b'b',
 

src/reaction.rs

@@ -0,0 +1,551 @@
+//! # Reactions.
+//!
+//! Reactions are short messages consisting of emojis sent in reply to
+//! messages. Unlike normal messages which are added to the end of the chat,
+//! reactions are supposed to be displayed near the original messages.
+//!
+//! RFC 9078 specifies how reactions are transmitted in MIME messages.
+//!
+//! Reaction update semantics is not well-defined in RFC 9078, so
+//! Delta Chat uses the same semantics as in
+//! [XEP-0444](https://xmpp.org/extensions/xep-0444.html) section
+//! "3.2 Updating reactions to a message". Received reactions override
+//! all previously received reactions from the same user and it is
+//! possible to remove all reactions by sending an empty string as a reaction,
+//! even though RFC 9078 requires at least one emoji to be sent.
+
+use std::collections::BTreeMap;
+use std::fmt;
+
+use anyhow::Result;
+
+use crate::chat::{send_msg, ChatId};
+use crate::contact::ContactId;
+use crate::context::Context;
+use crate::events::EventType;
+use crate::message::{rfc724_mid_exists, Message, MsgId, Viewtype};
+
+/// A single reaction consisting of multiple emoji sequences.
+///
+/// It is guaranteed to have all emojis sorted and deduplicated inside.
+#[derive(Debug, Default, Clone)]
+pub struct Reaction {
+    /// Canonical represntation of reaction as a string of space-separated emojis.
+    reaction: String,
+}
+
+// We implement From<&str> instead of std::str::FromStr, because
+// FromStr requires error type and reaction parsing never returns an
+// error.
+impl From<&str> for Reaction {
+    /// Parses a string containing a reaction.
+    ///
+    /// Reaction string is separated by spaces or tabs (`WSP` in ABNF),
+    /// but this function accepts any ASCII whitespace, so even a CRLF at
+    /// the end of string is acceptable.
+    ///
+    /// Any short enough string is accepted as a reaction to avoid the
+    /// complexity of validating emoji sequences as required by RFC
+    /// 9078. On the sender side UI is responsible to provide only
+    /// valid emoji sequences via reaction picker. On the receiver
+    /// side, abuse of the possibility to use arbitrary strings as
+    /// reactions is not different from other kinds of spam attacks
+    /// such as sending large numbers of large messages, and should be
+    /// dealt with the same way, e.g. by blocking the user.
+    fn from(reaction: &str) -> Self {
+        let mut emojis: Vec<&str> = reaction
+            .split_ascii_whitespace()
+            .filter(|&emoji| emoji.len() < 30)
+            .collect();
+        emojis.sort();
+        emojis.dedup();
+        let reaction = emojis.join(" ");
+        Self { reaction }
+    }
+}
+
+impl Reaction {
+    /// Returns true if reaction contains no emojis.
+    pub fn is_empty(&self) -> bool {
+        self.reaction.is_empty()
+    }
+
+    /// Returns a vector of emojis composing a reaction.
+    pub fn emojis(&self) -> Vec<&str> {
+        self.reaction.split(' ').collect()
+    }
+
+    /// Returns space-separated string of emojis
+    pub fn as_str(&self) -> &str {
+        &self.reaction
+    }
+
+    /// Appends emojis from another reaction to this reaction.
+    pub fn add(&self, other: Self) -> Self {
+        let mut emojis: Vec<&str> = self.emojis();
+        emojis.append(&mut other.emojis());
+        emojis.sort();
+        emojis.dedup();
+        let reaction = emojis.join(" ");
+        Self { reaction }
+    }
+}
+
+/// Structure representing all reactions to a particular message.
+#[derive(Debug)]
+pub struct Reactions {
+    /// Map from a contact to its reaction to message.
+    reactions: 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()
+    }
+}
+
+impl fmt::Display for Reactions {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut emoji_frequencies: BTreeMap<String, usize> = BTreeMap::new();
+        for reaction in self.reactions.values() {
+            for emoji in reaction.emojis() {
+                emoji_frequencies
+                    .entry(emoji.to_string())
+                    .and_modify(|x| *x += 1)
+                    .or_insert(1);
+            }
+        }
+        let mut first = true;
+        for (emoji, frequency) in emoji_frequencies {
+            if !first {
+                write!(f, " ")?;
+            }
+            first = false;
+            write!(f, "{}{}", emoji, frequency)?;
+        }
+        Ok(())
+    }
+}
+
+async fn set_msg_id_reaction(
+    context: &Context,
+    msg_id: MsgId,
+    chat_id: ChatId,
+    contact_id: ContactId,
+    reaction: Reaction,
+) -> Result<()> {
+    if reaction.is_empty() {
+        // Simply remove the record instead of setting it to empty string.
+        context
+            .sql
+            .execute(
+                "DELETE FROM reactions
+                 WHERE msg_id = ?1
+                 AND contact_id = ?2",
+                paramsv![msg_id, contact_id],
+            )
+            .await?;
+    } else {
+        context
+            .sql
+            .execute(
+                "INSERT INTO reactions (msg_id, contact_id, reaction)
+                 VALUES (?1, ?2, ?3)
+                 ON CONFLICT(msg_id, contact_id)
+                 DO UPDATE SET reaction=excluded.reaction",
+                paramsv![msg_id, contact_id, reaction.as_str()],
+            )
+            .await?;
+    }
+
+    context.emit_event(EventType::ReactionsChanged {
+        chat_id,
+        msg_id,
+        contact_id,
+    });
+    Ok(())
+}
+
+/// Sends a reaction to message `msg_id`, overriding previously sent reactions.
+///
+/// `reaction` is a string consisting of space-separated emoji. Use
+/// empty string to retract a reaction.
+pub async fn send_reaction(context: &Context, msg_id: MsgId, reaction: &str) -> Result<MsgId> {
+    let msg = Message::load_from_db(context, msg_id).await?;
+    let chat_id = msg.chat_id;
+
+    let reaction: Reaction = reaction.into();
+    let mut reaction_msg = Message::new(Viewtype::Text);
+    reaction_msg.text = Some(reaction.as_str().to_string());
+    reaction_msg.set_reaction();
+    reaction_msg.in_reply_to = Some(msg.rfc724_mid);
+    reaction_msg.hidden = true;
+
+    // Send messsage first.
+    let reaction_msg_id = send_msg(context, chat_id, &mut reaction_msg).await?;
+
+    // Only set reaction if we successfully sent the message.
+    set_msg_id_reaction(context, msg_id, msg.chat_id, ContactId::SELF, reaction).await?;
+    Ok(reaction_msg_id)
+}
+
+/// Adds given reaction to message `msg_id` and sends an update.
+///
+/// This can be used to implement advanced clients that allow reacting
+/// with multiple emojis. For a simple messenger UI, you probably want
+/// to use [`send_reaction()`] instead so reacting with a new emoji
+/// removes previous emoji at the same time.
+pub async fn add_reaction(context: &Context, msg_id: MsgId, reaction: &str) -> Result<MsgId> {
+    let self_reaction = get_self_reaction(context, msg_id).await?;
+    let reaction = self_reaction.add(Reaction::from(reaction));
+    send_reaction(context, msg_id, reaction.as_str()).await
+}
+
+/// Updates reaction of `contact_id` on the message with `in_reply_to`
+/// Message-ID. If no such message is found in the database, reaction
+/// is ignored.
+///
+/// `reaction` is a space-separated string of emojis. It can be empty
+/// if contact wants to remove all reactions.
+pub(crate) async fn set_msg_reaction(
+    context: &Context,
+    in_reply_to: &str,
+    chat_id: ChatId,
+    contact_id: ContactId,
+    reaction: Reaction,
+) -> 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, reaction).await
+    } else {
+        info!(
+            context,
+            "Can't assign reaction to unknown message with Message-ID {}", in_reply_to
+        );
+        Ok(())
+    }
+}
+
+/// Get our own reaction for a given message.
+async fn get_self_reaction(context: &Context, msg_id: MsgId) -> Result<Reaction> {
+    let reaction_str: Option<String> = context
+        .sql
+        .query_get_value(
+            "SELECT reaction
+             FROM reactions
+             WHERE msg_id=? AND contact_id=?",
+            paramsv![msg_id, ContactId::SELF],
+        )
+        .await?;
+    Ok(reaction_str
+        .as_deref()
+        .map(Reaction::from)
+        .unwrap_or_default())
+}
+
+/// Returns a structure containing all reactions to the message.
+pub async fn get_msg_reactions(context: &Context, msg_id: MsgId) -> Result<Reactions> {
+    let reactions = context
+        .sql
+        .query_map(
+            "SELECT contact_id, reaction FROM reactions WHERE msg_id=?",
+            paramsv![msg_id],
+            |row| {
+                let contact_id: ContactId = row.get(0)?;
+                let reaction: String = row.get(1)?;
+                Ok((contact_id, reaction))
+            },
+            |rows| {
+                let mut reactions = Vec::new();
+                for row in rows {
+                    let (contact_id, reaction) = row?;
+                    reactions.push((contact_id, Reaction::from(reaction.as_str())));
+                }
+                Ok(reactions)
+            },
+        )
+        .await?
+        .into_iter()
+        .collect();
+    Ok(Reactions { reactions })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::chat::get_chat_msgs;
+
+    use crate::config::Config;
+    use crate::constants::DC_CHAT_ID_TRASH;
+    use crate::contact::{Contact, Origin};
+    use crate::download::DownloadState;
+    use crate::message::MessageState;
+    use crate::receive_imf::{receive_imf, receive_imf_inner};
+    use crate::test_utils::TestContext;
+
+    #[test]
+    fn test_parse_reaction() {
+        // Check that basic set of emojis from RFC 9078 is supported.
+        assert_eq!(Reaction::from("👍").emojis(), vec!["👍"]);
+        assert_eq!(Reaction::from("👎").emojis(), vec!["👎"]);
+        assert_eq!(Reaction::from("😀").emojis(), vec!["😀"]);
+        assert_eq!(Reaction::from("☹").emojis(), vec!["☹"]);
+        assert_eq!(Reaction::from("😢").emojis(), vec!["😢"]);
+
+        // Empty string can be used to remove all reactions.
+        assert!(Reaction::from("").is_empty());
+
+        // Short strings can be used as emojis, could be used to add
+        // support for custom emojis via emoji shortcodes.
+        assert_eq!(Reaction::from(":deltacat:").emojis(), vec![":deltacat:"]);
+
+        // Check that long strings are not valid emojis.
+        assert!(
+            Reaction::from(":foobarbazquuxaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa:").is_empty()
+        );
+
+        // Multiple reactions separated by spaces or tabs are supported.
+        assert_eq!(Reaction::from("👍 ❤").emojis(), vec!["❤", "👍"]);
+        assert_eq!(Reaction::from("👍\t❤").emojis(), vec!["❤", "👍"]);
+
+        // Invalid emojis are removed, but valid emojis are retained.
+        assert_eq!(
+            Reaction::from("👍\t:foo: ❤").emojis(),
+            vec![":foo:", "❤", "👍"]
+        );
+        assert_eq!(Reaction::from("👍\t:foo: ❤").as_str(), ":foo: ❤ 👍");
+
+        // Duplicates are removed.
+        assert_eq!(Reaction::from("👍 👍").emojis(), vec!["👍"]);
+    }
+
+    #[test]
+    fn test_add_reaction() {
+        let reaction1 = Reaction::from("👍 😀");
+        let reaction2 = Reaction::from("❤");
+        let reaction_sum = reaction1.add(reaction2);
+
+        assert_eq!(reaction_sum.emojis(), vec!["❤", "👍", "😀"]);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_receive_reaction() -> Result<()> {
+        let alice = TestContext::new_alice().await;
+        alice.set_config(Config::ShowEmails, Some("2")).await?;
+
+        // Alice receives BCC-self copy of a message sent to Bob.
+        receive_imf(
+            &alice,
+            "To: bob@example.net\n\
+From: alice@example.org\n\
+Date: Today, 29 February 2021 00:00:00 -800\n\
+Message-ID: 12345@example.org\n\
+Subject: Meeting\n\
+\n\
+Can we chat at 1pm pacific, today?"
+                .as_bytes(),
+            false,
+        )
+        .await?;
+        let msg = alice.get_last_msg().await;
+        assert_eq!(msg.state, MessageState::OutDelivered);
+        let reactions = get_msg_reactions(&alice, msg.id).await?;
+        let contacts = reactions.contacts();
+        assert_eq!(contacts.len(), 0);
+
+        let bob_id = Contact::add_or_lookup(&alice, "", "bob@example.net", Origin::ManuallyCreated)
+            .await?
+            .0;
+        let bob_reaction = reactions.get(bob_id);
+        assert!(bob_reaction.is_empty()); // Bob has not reacted to message yet.
+
+        // Alice receives reaction to her message from Bob.
+        receive_imf(
+            &alice,
+            "To: alice@example.org\n\
+From: bob@example.net\n\
+Date: Today, 29 February 2021 00:00:10 -800\n\
+Message-ID: 56789@example.net\n\
+In-Reply-To: 12345@example.org\n\
+Subject: Meeting\n\
+Mime-Version: 1.0 (1.0)\n\
+Content-Type: text/plain; charset=utf-8\n\
+Content-Disposition: reaction\n\
+\n\
+\u{1F44D}"
+                .as_bytes(),
+            false,
+        )
+        .await?;
+
+        let reactions = get_msg_reactions(&alice, msg.id).await?;
+        assert_eq!(reactions.to_string(), "👍1");
+
+        let contacts = reactions.contacts();
+        assert_eq!(contacts.len(), 1);
+
+        assert_eq!(contacts.get(0), Some(&bob_id));
+        let bob_reaction = reactions.get(bob_id);
+        assert_eq!(bob_reaction.is_empty(), false);
+        assert_eq!(bob_reaction.emojis(), vec!["👍"]);
+        assert_eq!(bob_reaction.as_str(), "👍");
+
+        Ok(())
+    }
+
+    async fn expect_reactions_changed_event(
+        t: &TestContext,
+        expected_chat_id: ChatId,
+        expected_msg_id: MsgId,
+        expected_contact_id: ContactId,
+    ) -> Result<()> {
+        let event = t
+            .evtracker
+            .get_matching(|evt| matches!(evt, EventType::ReactionsChanged { .. }))
+            .await;
+        match event {
+            EventType::ReactionsChanged {
+                chat_id,
+                msg_id,
+                contact_id,
+            } => {
+                assert_eq!(chat_id, expected_chat_id);
+                assert_eq!(msg_id, expected_msg_id);
+                assert_eq!(contact_id, expected_contact_id);
+            }
+            _ => unreachable!(),
+        }
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_send_reaction() -> Result<()> {
+        let alice = TestContext::new_alice().await;
+        let bob = TestContext::new_bob().await;
+
+        let chat_alice = alice.create_chat(&bob).await;
+        let alice_msg = alice.send_text(chat_alice.id, "Hi!").await;
+        let bob_msg = bob.recv_msg(&alice_msg).await;
+        assert_eq!(get_chat_msgs(&alice, chat_alice.id, 0).await?.len(), 1);
+        assert_eq!(get_chat_msgs(&bob, bob_msg.chat_id, 0).await?.len(), 1);
+
+        let alice_msg2 = alice.send_text(chat_alice.id, "Hi again!").await;
+        bob.recv_msg(&alice_msg2).await;
+        assert_eq!(get_chat_msgs(&alice, chat_alice.id, 0).await?.len(), 2);
+        assert_eq!(get_chat_msgs(&bob, bob_msg.chat_id, 0).await?.len(), 2);
+
+        bob_msg.chat_id.accept(&bob).await?;
+
+        send_reaction(&bob, bob_msg.id, "👍").await.unwrap();
+        expect_reactions_changed_event(&bob, bob_msg.chat_id, bob_msg.id, ContactId::SELF).await?;
+        assert_eq!(get_chat_msgs(&bob, bob_msg.chat_id, 0).await?.len(), 2);
+
+        let bob_reaction_msg = bob.pop_sent_msg().await;
+        let alice_reaction_msg = alice.recv_msg_opt(&bob_reaction_msg).await.unwrap();
+        assert_eq!(alice_reaction_msg.chat_id, DC_CHAT_ID_TRASH);
+        assert_eq!(get_chat_msgs(&alice, chat_alice.id, 0).await?.len(), 2);
+
+        let reactions = get_msg_reactions(&alice, alice_msg.sender_msg_id).await?;
+        assert_eq!(reactions.to_string(), "👍1");
+        let contacts = reactions.contacts();
+        assert_eq!(contacts.len(), 1);
+        let bob_id = contacts.get(0).unwrap();
+        let bob_reaction = reactions.get(*bob_id);
+        assert_eq!(bob_reaction.is_empty(), false);
+        assert_eq!(bob_reaction.emojis(), vec!["👍"]);
+        assert_eq!(bob_reaction.as_str(), "👍");
+        expect_reactions_changed_event(&alice, chat_alice.id, alice_msg.sender_msg_id, *bob_id)
+            .await?;
+
+        // Alice reacts to own message.
+        send_reaction(&alice, alice_msg.sender_msg_id, "👍 😀")
+            .await
+            .unwrap();
+        let reactions = get_msg_reactions(&alice, alice_msg.sender_msg_id).await?;
+        assert_eq!(reactions.to_string(), "👍2 😀1");
+
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_partial_download_and_reaction() -> Result<()> {
+        let alice = TestContext::new_alice().await;
+        let bob = TestContext::new_bob().await;
+
+        alice
+            .create_chat_with_contact("Bob", "bob@example.net")
+            .await;
+
+        let msg_header = "From: Bob <bob@example.net>\n\
+                    To: Alice <alice@example.org>\n\
+                    Chat-Version: 1.0\n\
+                    Subject: subject\n\
+                    Message-ID: <first@example.org>\n\
+                    Date: Sun, 14 Nov 2021 00:10:00 +0000\
+                    Content-Type: text/plain";
+        let msg_full = format!("{}\n\n100k text...", msg_header);
+
+        // Alice downloads message from Bob partially.
+        let alice_received_message = receive_imf_inner(
+            &alice,
+            "first@example.org",
+            msg_header.as_bytes(),
+            false,
+            Some(100000),
+            false,
+        )
+        .await?
+        .unwrap();
+        let alice_msg_id = *alice_received_message.msg_ids.get(0).unwrap();
+
+        // Bob downloads own message on the other device.
+        let bob_received_message = receive_imf(&bob, msg_full.as_bytes(), false)
+            .await?
+            .unwrap();
+        let bob_msg_id = *bob_received_message.msg_ids.get(0).unwrap();
+
+        // Bob reacts to own message.
+        send_reaction(&bob, bob_msg_id, "👍").await.unwrap();
+        let bob_reaction_msg = bob.pop_sent_msg().await;
+
+        // Alice receives a reaction.
+        alice.recv_msg_opt(&bob_reaction_msg).await.unwrap();
+
+        let reactions = get_msg_reactions(&alice, alice_msg_id).await?;
+        assert_eq!(reactions.to_string(), "👍1");
+        let msg = Message::load_from_db(&alice, alice_msg_id).await?;
+        assert_eq!(msg.download_state(), DownloadState::Available);
+
+        // Alice downloads full message.
+        receive_imf_inner(
+            &alice,
+            "first@example.org",
+            msg_full.as_bytes(),
+            false,
+            None,
+            false,
+        )
+        .await?;
+
+        // Check that reaction is still on the message after full download.
+        let msg = Message::load_from_db(&alice, alice_msg_id).await?;
+        assert_eq!(msg.download_state(), DownloadState::Done);
+        let reactions = get_msg_reactions(&alice, alice_msg_id).await?;
+        assert_eq!(reactions.to_string(), "👍1");
+
+        Ok(())
+    }
+}

src/receive_imf.rs

@@ -33,6 +33,7 @@ use crate::mimeparser::{
 };
 use crate::param::{Param, Params};
 use crate::peerstate::{Peerstate, PeerstateKeyType, PeerstateVerifiedStatus};
+use crate::reaction::{set_msg_reaction, Reaction};
 use crate::securejoin::{self, handle_securejoin_handshake, observe_securejoin_on_other_device};
 use crate::sql;
 use crate::stock_str;
@@ -404,7 +405,7 @@ async fn add_parts(
     from_id: ContactId,
     seen: bool,
     is_partial_download: Option<u32>,
-    replace_msg_id: Option<MsgId>,
+    mut replace_msg_id: Option<MsgId>,
     fetching_existing_messages: bool,
     prevent_rename: bool,
 ) -> Result<ReceivedMsg> {
@@ -430,8 +431,9 @@ async fn add_parts(
     };
     // incoming non-chat messages may be discarded
 
-    let location_kml_is = mime_parser.location_kml.is_some();
+    let is_location_kml = mime_parser.location_kml.is_some();
     let is_mdn = !mime_parser.mdn_reports.is_empty();
+    let is_reaction = mime_parser.parts.iter().any(|part| part.is_reaction);
     let show_emails =
         ShowEmails::from_i32(context.get_config_int(Config::ShowEmails).await?).unwrap_or_default();
 
@@ -450,7 +452,7 @@ async fn add_parts(
             ShowEmails::All => allow_creation = !is_mdn,
         }
     } else {
-        allow_creation = !is_mdn;
+        allow_creation = !is_mdn && !is_reaction;
     }
 
     // check if the message introduces a new chat:
@@ -689,7 +691,8 @@ async fn add_parts(
         state = if seen
             || fetching_existing_messages
             || is_mdn
-            || location_kml_is
+            || is_reaction
+            || is_location_kml
             || securejoin_seen
             || chat_id_blocked == Blocked::Yes
         {
@@ -841,14 +844,15 @@ async fn add_parts(
         }
     }
 
-    if is_mdn {
-        chat_id = Some(DC_CHAT_ID_TRASH);
-    }
-
-    let chat_id = chat_id.unwrap_or_else(|| {
-        info!(context, "No chat id for message (TRASH)");
+    let orig_chat_id = chat_id;
+    let chat_id = if is_mdn || is_reaction {
         DC_CHAT_ID_TRASH
-    });
+    } else {
+        chat_id.unwrap_or_else(|| {
+            info!(context, "No chat id for message (TRASH)");
+            DC_CHAT_ID_TRASH
+        })
+    };
 
     // Extract ephemeral timer from the message or use the existing timer if the message is not fully downloaded.
     let mut ephemeral_timer = if is_partial_download.is_some() {
@@ -1053,11 +1057,41 @@ async fn add_parts(
     let conn = context.sql.get_conn().await?;
 
     for part in &mime_parser.parts {
+        if part.is_reaction {
+            set_msg_reaction(
+                context,
+                &mime_in_reply_to,
+                orig_chat_id.unwrap_or_default(),
+                from_id,
+                Reaction::from(part.msg.as_str()),
+            )
+            .await?;
+        }
+
+        let mut param = part.param.clone();
+        if is_system_message != SystemMessage::Unknown {
+            param.set_int(Param::Cmd, is_system_message as i32);
+        }
+        if let Some(replace_msg_id) = replace_msg_id {
+            let placeholder = Message::load_from_db(context, replace_msg_id).await?;
+            for key in [
+                Param::WebxdcSummary,
+                Param::WebxdcSummaryTimestamp,
+                Param::WebxdcDocument,
+                Param::WebxdcDocumentTimestamp,
+            ] {
+                if let Some(value) = placeholder.param.get(key) {
+                    param.set(key, value);
+                }
+            }
+        }
+
         let mut txt_raw = "".to_string();
         let mut stmt = conn.prepare_cached(
             r#"
 INSERT INTO msgs
   (
+    id,
     rfc724_mid, chat_id,
     from_id, to_id, timestamp, timestamp_sent, 
     timestamp_rcvd, type, state, msgrmsg, 
@@ -1067,13 +1101,22 @@ INSERT INTO msgs
     ephemeral_timestamp, download_state, hop_info
   )
   VALUES (
+    ?,
     ?, ?, ?, ?,
     ?, ?, ?, ?,
     ?, ?, ?, ?,
     ?, ?, ?, ?,
     ?, ?, ?, ?,
     ?, ?, ?, ?
-  );
+  )
+ON CONFLICT (id) DO UPDATE
+SET rfc724_mid=excluded.rfc724_mid, chat_id=excluded.chat_id,
+    from_id=excluded.from_id, to_id=excluded.to_id, timestamp=excluded.timestamp, timestamp_sent=excluded.timestamp_sent,
+    timestamp_rcvd=excluded.timestamp_rcvd, type=excluded.type, state=excluded.state, msgrmsg=excluded.msgrmsg,
+    txt=excluded.txt, subject=excluded.subject, txt_raw=excluded.txt_raw, param=excluded.param,
+    bytes=excluded.bytes, mime_headers=excluded.mime_headers, mime_in_reply_to=excluded.mime_in_reply_to,
+    mime_references=excluded.mime_references, mime_modified=excluded.mime_modified, error=excluded.error, ephemeral_timer=excluded.ephemeral_timer,
+    ephemeral_timestamp=excluded.ephemeral_timestamp, download_state=excluded.download_state, hop_info=excluded.hop_info
 "#,
         )?;
 
@@ -1095,11 +1138,6 @@ INSERT INTO msgs
             txt_raw = format!("{}\n\n{}", subject, msg_raw);
         }
 
-        let mut param = part.param.clone();
-        if is_system_message != SystemMessage::Unknown {
-            param.set_int(Param::Cmd, is_system_message as i32);
-        }
-
         let ephemeral_timestamp = if in_fresh {
             0
         } else {
@@ -1113,9 +1151,10 @@ INSERT INTO msgs
 
         // If you change which information is skipped if the message is trashed,
         // also change `MsgId::trash()` and `delete_expired_messages()`
-        let trash = chat_id.is_trash() || (location_kml_is && msg.is_empty());
+        let trash = chat_id.is_trash() || (is_location_kml && msg.is_empty());
 
         stmt.execute(paramsv![
+            replace_msg_id,
             rfc724_mid,
             if trash { DC_CHAT_ID_TRASH } else { chat_id },
             if trash { ContactId::UNDEFINED } else { from_id },
@@ -1154,6 +1193,10 @@ INSERT INTO msgs
             },
             mime_parser.hop_info
         ])?;
+
+        // We only replace placeholder with a first part,
+        // afterwards insert additional parts.
+        replace_msg_id = None;
         let row_id = conn.last_insert_rowid();
 
         drop(stmt);
@@ -1162,14 +1205,8 @@ INSERT INTO msgs
     drop(conn);
 
     if let Some(replace_msg_id) = replace_msg_id {
-        if let Some(created_msg_id) = created_db_entries.pop() {
-            context
-                .merge_messages(created_msg_id, replace_msg_id)
-                .await?;
-            created_db_entries.push(replace_msg_id);
-        } else {
-            replace_msg_id.delete_from_db(context).await?;
-        }
+        // "Replace" placeholder with a message that has no parts.
+        replace_msg_id.delete_from_db(context).await?;
     }
 
     chat_id.unarchive_if_not_muted(context).await?;

src/sql.rs

@@ -396,7 +396,7 @@ impl Sql {
     }
 
     /// Used for executing `SELECT COUNT` statements only. Returns the resulting count.
-    pub async fn count(&self, query: &str, params: impl rusqlite::Params) -> anyhow::Result<usize> {
+    pub async fn count(&self, query: &str, params: impl rusqlite::Params) -> Result<usize> {
         let count: isize = self.query_row(query, params, |row| row.get(0)).await?;
         Ok(usize::try_from(count)?)
     }
@@ -429,10 +429,10 @@ impl Sql {
     ///
     /// If the function returns an error, the transaction will be rolled back. If it does not return an
     /// error, the transaction will be committed.
-    pub async fn transaction<G, H>(&self, callback: G) -> anyhow::Result<H>
+    pub async fn transaction<G, H>(&self, callback: G) -> Result<H>
     where
         H: Send + 'static,
-        G: Send + 'static + FnOnce(&mut rusqlite::Transaction<'_>) -> anyhow::Result<H>,
+        G: Send + 'static + FnOnce(&mut rusqlite::Transaction<'_>) -> Result<H>,
     {
         let mut conn = self.get_conn().await?;
         tokio::task::block_in_place(move || {
@@ -453,7 +453,7 @@ impl Sql {
     }
 
     /// Query the database if the requested table already exists.
-    pub async fn table_exists(&self, name: &str) -> anyhow::Result<bool> {
+    pub async fn table_exists(&self, name: &str) -> Result<bool> {
         let conn = self.get_conn().await?;
         tokio::task::block_in_place(move || {
             let mut exists = false;
@@ -468,7 +468,7 @@ impl Sql {
     }
 
     /// Check if a column exists in a given table.
-    pub async fn col_exists(&self, table_name: &str, col_name: &str) -> anyhow::Result<bool> {
+    pub async fn col_exists(&self, table_name: &str, col_name: &str) -> Result<bool> {
         let conn = self.get_conn().await?;
         tokio::task::block_in_place(move || {
             let mut exists = false;
@@ -492,7 +492,7 @@ impl Sql {
         sql: &str,
         params: impl rusqlite::Params,
         f: F,
-    ) -> anyhow::Result<Option<T>>
+    ) -> Result<Option<T>>
     where
         F: FnOnce(&rusqlite::Row) -> rusqlite::Result<T>,
     {
@@ -516,7 +516,7 @@ impl Sql {
         &self,
         query: &str,
         params: impl rusqlite::Params,
-    ) -> anyhow::Result<Option<T>>
+    ) -> Result<Option<T>>
     where
         T: rusqlite::types::FromSql,
     {

src/sql/migrations.rs

@@ -597,9 +597,22 @@ CREATE INDEX smtp_messageid ON imap(rfc724_mid);
         .await?;
     }
     if dbversion < 92 {
+        sql.execute_migration(
+            r#"CREATE TABLE reactions (
+              msg_id INTEGER NOT NULL, -- id of the message reacted to
+              contact_id INTEGER NOT NULL, -- id of the contact reacting to the message
+              reaction TEXT DEFAULT '' NOT NULL, -- a sequence of emojis separated by spaces
+              PRIMARY KEY(msg_id, contact_id),
+              FOREIGN KEY(msg_id) REFERENCES msgs(id) ON DELETE CASCADE -- delete reactions when message is deleted
+              FOREIGN KEY(contact_id) REFERENCES contacts(id) ON DELETE CASCADE -- delete reactions when contact is deleted
+            )"#,
+            92
+        ).await?;
+    }
+    if dbversion < 93 {
         sql.execute_migration(
             "CREATE TABLE sending_domains(domain TEXT PRIMARY KEY, dkim_works INTEGER DEFAULT 0);",
-            92,
+            93,
         )
         .await?;
     }