Simplify logic

Add a test on what happens currently when apps call `markseen_msgs()` for not downloaded
messages. Such messages are marked as seen, but MDNs aren't sent for them. This is probably correct,
but when a message is downloaded, its state should be downgraded to `InNoticed` and when the user
really sees it, the app should issue `markseen_msgs()` again and that should trigger sending an
MDN. Currently when such a message is downloaded, it remains `InSeen`.

Cargo.lock

@@ -273,24 +273,26 @@ dependencies = [
 
 [[package]]
 name = "async-compression"
-version = "0.4.14"
+version = "0.4.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "998282f8f49ccd6116b0ed8a4de0fbd3151697920e7c7533416d6e25e76434a7"
+checksum = "e26a9844c659a2a293d239c7910b752f8487fe122c6c8bd1659bf85a6507c302"
 dependencies = [
  "flate2",
  "futures-core",
  "futures-io",
  "memchr",
  "pin-project-lite",
+ "tokio",
 ]
 
 [[package]]
 name = "async-imap"
-version = "0.10.1"
+version = "0.10.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2162818f7b394e342a6591864bfc960824ed13e3f6609fee0d19e770ebcd99e1"
+checksum = "5488cd022c3c7bc41a9b34a540d9ac0d9c5cd42fdb106a67616521b7592d5b4e"
 dependencies = [
  "async-channel 2.3.1",
+ "async-compression",
  "base64 0.21.7",
  "bytes",
  "chrono",
@@ -299,6 +301,7 @@ dependencies = [
  "log",
  "nom",
  "once_cell",
+ "pin-project",
  "pin-utils",
  "self_cell",
  "stop-token",

Cargo.toml

@@ -41,7 +41,7 @@ ratelimit = { path = "./deltachat-ratelimit" }
 anyhow = { workspace = true }
 async-broadcast = "0.7.1"
 async-channel = { workspace = true }
-async-imap = { version = "0.10.1", default-features = false, features = ["runtime-tokio"] }
+async-imap = { version = "0.10.2", default-features = false, features = ["runtime-tokio", "compress"] }
 async-native-tls = { version = "0.5", default-features = false, features = ["runtime-tokio"] }
 async-smtp = { version = "0.9", default-features = false, features = ["runtime-tokio"] }
 async_zip = { version = "0.0.17", default-features = false, features = ["deflate", "tokio-fs"] }

src/imap.rs

@@ -39,6 +39,7 @@ use crate::login_param::{
 use crate::message::{self, Message, MessageState, MessengerMessage, MsgId, Viewtype};
 use crate::mimeparser;
 use crate::net::proxy::ProxyConfig;
+use crate::net::session::SessionStream;
 use crate::oauth2::get_oauth2_access_token;
 use crate::receive_imf::{
     from_field_to_contact_id, get_prefetch_parent_message, receive_imf_inner, ReceivedMsg,
@@ -55,7 +56,7 @@ pub mod scan_folders;
 pub mod select_folder;
 pub(crate) mod session;
 
-use client::Client;
+use client::{determine_capabilities, Client};
 use mailparse::SingleInfo;
 use session::Session;
 
@@ -376,7 +377,23 @@ impl Imap {
             };
 
             match login_res {
-                Ok(session) => {
+                Ok(mut session) => {
+                    let capabilities = determine_capabilities(&mut session).await?;
+
+                    let session = if capabilities.can_compress {
+                        info!(context, "Enabling IMAP compression.");
+                        let compressed_session = session
+                            .compress(|s| {
+                                let session_stream: Box<dyn SessionStream> = Box::new(s);
+                                session_stream
+                            })
+                            .await
+                            .context("Failed to enable IMAP compression")?;
+                        Session::new(compressed_session, capabilities)
+                    } else {
+                        Session::new(session, capabilities)
+                    };
+
                     // Store server ID in the context to display in account info.
                     let mut lock = context.server_id.write().await;
                     lock.clone_from(&session.capabilities.server_id);

src/imap/capabilities.rs

@@ -25,6 +25,10 @@ pub(crate) struct Capabilities {
     /// <https://tools.ietf.org/html/rfc5464>
     pub can_metadata: bool,
 
+    /// True if the server has COMPRESS=DEFLATE capability as defined in
+    /// <https://tools.ietf.org/html/rfc4978>
+    pub can_compress: bool,
+
     /// True if the server supports XDELTAPUSH capability.
     /// This capability means setting /private/devicetoken IMAP METADATA
     /// on the INBOX results in new mail notifications

src/imap/client.rs

@@ -7,7 +7,6 @@ use async_imap::Session as ImapSession;
 use tokio::io::BufWriter;
 
 use super::capabilities::Capabilities;
-use super::session::Session;
 use crate::context::Context;
 use crate::login_param::{ConnectionCandidate, ConnectionSecurity};
 use crate::net::dns::{lookup_host_with_cache, update_connect_timestamp};
@@ -51,7 +50,7 @@ fn alpn(port: u16) -> &'static [&'static str] {
 /// Determine server capabilities.
 ///
 /// If server supports ID capability, send our client ID.
-async fn determine_capabilities(
+pub(crate) async fn determine_capabilities(
     session: &mut ImapSession<Box<dyn SessionStream>>,
 ) -> Result<Capabilities> {
     let caps = session
@@ -69,6 +68,7 @@ async fn determine_capabilities(
         can_check_quota: caps.has_str("QUOTA"),
         can_condstore: caps.has_str("CONDSTORE"),
         can_metadata: caps.has_str("METADATA"),
+        can_compress: caps.has_str("COMPRESS=DEFLATE"),
         can_push: caps.has_str("XDELTAPUSH"),
         is_chatmail: caps.has_str("XCHATMAIL"),
         server_id,
@@ -83,28 +83,31 @@ impl Client {
         }
     }
 
-    pub(crate) async fn login(self, username: &str, password: &str) -> Result<Session> {
+    pub(crate) async fn login(
+        self,
+        username: &str,
+        password: &str,
+    ) -> Result<ImapSession<Box<dyn SessionStream>>> {
         let Client { inner, .. } = self;
-        let mut session = inner
+
+        let session = inner
             .login(username, password)
             .await
             .map_err(|(err, _client)| err)?;
-        let capabilities = determine_capabilities(&mut session).await?;
-        Ok(Session::new(session, capabilities))
+        Ok(session)
     }
 
     pub(crate) async fn authenticate(
         self,
         auth_type: &str,
         authenticator: impl async_imap::Authenticator,
-    ) -> Result<Session> {
+    ) -> Result<ImapSession<Box<dyn SessionStream>>> {
         let Client { inner, .. } = self;
-        let mut session = inner
+        let session = inner
             .authenticate(auth_type, authenticator)
             .await
             .map_err(|(err, _client)| err)?;
-        let capabilities = determine_capabilities(&mut session).await?;
-        Ok(Session::new(session, capabilities))
+        Ok(session)
     }
 
     async fn connection_attempt(

src/message.rs

@@ -1826,8 +1826,8 @@ pub(crate) async fn update_msg_state(
     context
         .sql
         .execute(
-            &format!("UPDATE msgs SET state=?1 {error_subst} WHERE id=?2 AND (?1!=?3 OR state<?3)"),
-            (state, msg_id, MessageState::OutDelivered),
+            &format!("UPDATE msgs SET state=? {error_subst} WHERE id=?"),
+            (state, msg_id),
         )
         .await?;
     Ok(())
@@ -2590,6 +2590,60 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_markseen_not_downloaded_msg() -> Result<()> {
+        let mut tcm = TestContextManager::new();
+        let alice = &tcm.alice().await;
+        alice.set_config(Config::DownloadLimit, Some("1")).await?;
+        let bob = &tcm.bob().await;
+        let bob_chat_id = tcm.send_recv_accept(alice, bob, "hi").await.chat_id;
+
+        let file_bytes = include_bytes!("../test-data/image/screenshot.png");
+        let mut msg = Message::new(Viewtype::Image);
+        msg.set_file_from_bytes(bob, "a.jpg", file_bytes, None)
+            .await?;
+        let sent_msg = bob.send_msg(bob_chat_id, &mut msg).await;
+        let msg = alice.recv_msg(&sent_msg).await;
+        assert_eq!(msg.download_state, DownloadState::Available);
+        assert!(!msg.param.get_bool(Param::WantsMdn).unwrap_or_default());
+        assert_eq!(msg.state, MessageState::InFresh);
+        markseen_msgs(alice, vec![msg.id]).await?;
+        let msg = Message::load_from_db(alice, msg.id).await?;
+        assert_eq!(msg.state, MessageState::InSeen);
+        assert!(
+            !alice
+                .sql
+                .exists("SELECT COUNT(*) FROM smtp_mdns", ())
+                .await?
+        );
+
+        alice.set_config(Config::DownloadLimit, None).await?;
+        // Simulate that the message is even marked as `\Seen` on IMAP.
+        let rcvd_msg = receive_imf(alice, sent_msg.payload().as_bytes(), true)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(rcvd_msg.chat_id, msg.chat_id);
+        let msg = Message::load_from_db(alice, *rcvd_msg.msg_ids.last().unwrap())
+            .await
+            .unwrap();
+        assert_eq!(msg.download_state, DownloadState::Done);
+        assert!(msg.param.get_bool(Param::WantsMdn).unwrap_or_default());
+        assert!(msg.get_showpadlock());
+        assert_eq!(msg.state, MessageState::InNoticed);
+        markseen_msgs(alice, vec![msg.id]).await?;
+        let msg = Message::load_from_db(alice, msg.id).await?;
+        assert_eq!(msg.state, MessageState::InSeen);
+        assert_eq!(
+            alice
+                .sql
+                .count("SELECT COUNT(*) FROM smtp_mdns", ())
+                .await?,
+            1
+        );
+        Ok(())
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_get_state() -> Result<()> {
         let alice = TestContext::new_alice().await;

src/net/session.rs

@@ -53,6 +53,11 @@ impl<T: SessionStream> SessionStream for shadowsocks::ProxyClientStream<T> {
         self.get_mut().set_read_timeout(timeout)
     }
 }
+impl<T: SessionStream> SessionStream for async_imap::DeflateStream<T> {
+    fn set_read_timeout(&mut self, timeout: Option<Duration>) {
+        self.get_mut().set_read_timeout(timeout)
+    }
+}
 
 /// Session stream with a read buffer.
 pub(crate) trait SessionBufStream: SessionStream + AsyncBufRead {}

src/receive_imf.rs

@@ -1014,7 +1014,10 @@ async fn add_parts(
             }
         }
 
-        state = if seen
+        state = if replace_msg_id.is_some() {
+            // TODO a comment explaining why we're doing this would be nice
+            MessageState::InNoticed
+        } else if seen
             || fetching_existing_messages
             || is_mdn
             || is_reaction
@@ -1562,7 +1565,7 @@ INSERT INTO msgs
 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_sent=excluded.timestamp_sent,
-    type=excluded.type, msgrmsg=excluded.msgrmsg,
+    type=excluded.type, state=min(state,excluded.state), msgrmsg=excluded.msgrmsg,
     txt=excluded.txt, txt_normalized=excluded.txt_normalized, subject=excluded.subject,
     txt_raw=excluded.txt_raw, param=excluded.param,
     hidden=excluded.hidden,bytes=excluded.bytes, mime_headers=excluded.mime_headers,
@@ -1613,7 +1616,7 @@ RETURNING id
                     } else {
                         DownloadState::Done
                     },
-                    mime_parser.hop_info
+                    mime_parser.hop_info,
                 ],
                 |row| {
                     let msg_id: MsgId = row.get(0)?;