deltachat/
download.rs

1//! # Download large messages manually.
2
3use std::cmp::max;
4use std::collections::BTreeMap;
5
6use anyhow::{anyhow, bail, ensure, Result};
7use deltachat_derive::{FromSql, ToSql};
8use serde::{Deserialize, Serialize};
9
10use crate::config::Config;
11use crate::context::Context;
12use crate::imap::session::Session;
13use crate::log::info;
14use crate::message::{Message, MsgId, Viewtype};
15use crate::mimeparser::{MimeMessage, Part};
16use crate::tools::time;
17use crate::{chatlist_events, stock_str, EventType};
18
19/// Download limits should not be used below `MIN_DOWNLOAD_LIMIT`.
20///
21/// For better UX, some messages as add-member, non-delivery-reports (NDN) or read-receipts (MDN)
22/// should always be downloaded completely to handle them correctly,
23/// also in larger groups and if group and contact avatar are attached.
24/// Most of these cases are caught by `MIN_DOWNLOAD_LIMIT`.
25pub(crate) const MIN_DOWNLOAD_LIMIT: u32 = 163840;
26
27/// If a message is downloaded only partially
28/// and `delete_server_after` is set to small timeouts (eg. "at once"),
29/// the user might have no chance to actually download that message.
30/// `MIN_DELETE_SERVER_AFTER` increases the timeout in this case.
31pub(crate) const MIN_DELETE_SERVER_AFTER: i64 = 48 * 60 * 60;
32
33/// Download state of the message.
34#[derive(
35    Debug,
36    Default,
37    Display,
38    Clone,
39    Copy,
40    PartialEq,
41    Eq,
42    FromPrimitive,
43    ToPrimitive,
44    FromSql,
45    ToSql,
46    Serialize,
47    Deserialize,
48)]
49#[repr(u32)]
50pub enum DownloadState {
51    /// Message is fully downloaded.
52    #[default]
53    Done = 0,
54
55    /// Message is partially downloaded and can be fully downloaded at request.
56    Available = 10,
57
58    /// Failed to fully download the message.
59    Failure = 20,
60
61    /// Undecipherable message.
62    Undecipherable = 30,
63
64    /// Full download of the message is in progress.
65    InProgress = 1000,
66}
67
68impl Context {
69    // Returns validated download limit or `None` for "no limit".
70    pub(crate) async fn download_limit(&self) -> Result<Option<u32>> {
71        let download_limit = self.get_config_int(Config::DownloadLimit).await?;
72        if download_limit <= 0 {
73            Ok(None)
74        } else {
75            Ok(Some(max(MIN_DOWNLOAD_LIMIT, download_limit as u32)))
76        }
77    }
78}
79
80impl MsgId {
81    /// Schedules full message download for partially downloaded message.
82    pub async fn download_full(self, context: &Context) -> Result<()> {
83        let msg = Message::load_from_db(context, self).await?;
84        match msg.download_state() {
85            DownloadState::Done | DownloadState::Undecipherable => {
86                return Err(anyhow!("Nothing to download."))
87            }
88            DownloadState::InProgress => return Err(anyhow!("Download already in progress.")),
89            DownloadState::Available | DownloadState::Failure => {
90                self.update_download_state(context, DownloadState::InProgress)
91                    .await?;
92                context
93                    .sql
94                    .execute("INSERT INTO download (msg_id) VALUES (?)", (self,))
95                    .await?;
96                context.scheduler.interrupt_inbox().await;
97            }
98        }
99        Ok(())
100    }
101
102    /// Updates the message download state. Returns `Ok` if the message doesn't exist anymore.
103    pub(crate) async fn update_download_state(
104        self,
105        context: &Context,
106        download_state: DownloadState,
107    ) -> Result<()> {
108        if context
109            .sql
110            .execute(
111                "UPDATE msgs SET download_state=? WHERE id=?;",
112                (download_state, self),
113            )
114            .await?
115            == 0
116        {
117            return Ok(());
118        }
119        let Some(msg) = Message::load_from_db_optional(context, self).await? else {
120            return Ok(());
121        };
122        context.emit_event(EventType::MsgsChanged {
123            chat_id: msg.chat_id,
124            msg_id: self,
125        });
126        chatlist_events::emit_chatlist_item_changed(context, msg.chat_id);
127        Ok(())
128    }
129}
130
131impl Message {
132    /// Returns the download state of the message.
133    pub fn download_state(&self) -> DownloadState {
134        self.download_state
135    }
136}
137
138/// Actually download a message partially downloaded before.
139///
140/// Most messages are downloaded automatically on fetch instead.
141pub(crate) async fn download_msg(
142    context: &Context,
143    msg_id: MsgId,
144    session: &mut Session,
145) -> Result<()> {
146    let Some(msg) = Message::load_from_db_optional(context, msg_id).await? else {
147        // If partially downloaded message was already deleted
148        // we do not know its Message-ID anymore
149        // so cannot download it.
150        //
151        // Probably the message expired due to `delete_device_after`
152        // setting or was otherwise removed from the device,
153        // so we don't want it to reappear anyway.
154        return Ok(());
155    };
156
157    let row = context
158        .sql
159        .query_row_optional(
160            "SELECT uid, folder, uidvalidity FROM imap WHERE rfc724_mid=? AND target!=''",
161            (&msg.rfc724_mid,),
162            |row| {
163                let server_uid: u32 = row.get(0)?;
164                let server_folder: String = row.get(1)?;
165                let uidvalidity: u32 = row.get(2)?;
166                Ok((server_uid, server_folder, uidvalidity))
167            },
168        )
169        .await?;
170
171    let Some((server_uid, server_folder, uidvalidity)) = row else {
172        // No IMAP record found, we don't know the UID and folder.
173        return Err(anyhow!("Call download_full() again to try over."));
174    };
175
176    session
177        .fetch_single_msg(
178            context,
179            &server_folder,
180            uidvalidity,
181            server_uid,
182            msg.rfc724_mid.clone(),
183        )
184        .await?;
185    Ok(())
186}
187
188impl Session {
189    /// Download a single message and pipe it to receive_imf().
190    ///
191    /// receive_imf() is not directly aware that this is a result of a call to download_msg(),
192    /// however, implicitly knows that as the existing message is flagged as being partly.
193    async fn fetch_single_msg(
194        &mut self,
195        context: &Context,
196        folder: &str,
197        uidvalidity: u32,
198        uid: u32,
199        rfc724_mid: String,
200    ) -> Result<()> {
201        if uid == 0 {
202            bail!("Attempt to fetch UID 0");
203        }
204
205        let create = false;
206        let folder_exists = self
207            .select_with_uidvalidity(context, folder, create)
208            .await?;
209        ensure!(folder_exists, "No folder {folder}");
210
211        // we are connected, and the folder is selected
212        info!(context, "Downloading message {}/{} fully...", folder, uid);
213
214        let mut uid_message_ids: BTreeMap<u32, String> = BTreeMap::new();
215        uid_message_ids.insert(uid, rfc724_mid);
216        let (last_uid, _received) = self
217            .fetch_many_msgs(
218                context,
219                folder,
220                uidvalidity,
221                vec![uid],
222                &uid_message_ids,
223                false,
224            )
225            .await?;
226        if last_uid.is_none() {
227            bail!("Failed to fetch UID {uid}");
228        }
229        Ok(())
230    }
231}
232
233impl MimeMessage {
234    /// Creates a placeholder part and add that to `parts`.
235    ///
236    /// To create the placeholder, only the outermost header can be used,
237    /// the mime-structure itself is not available.
238    ///
239    /// The placeholder part currently contains a text with size and availability of the message;
240    /// in the future, we may do more advanced things as previews here.
241    pub(crate) async fn create_stub_from_partial_download(
242        &mut self,
243        context: &Context,
244        org_bytes: u32,
245    ) -> Result<()> {
246        let mut text = format!(
247            "[{}]",
248            stock_str::partial_download_msg_body(context, org_bytes).await
249        );
250        if let Some(delete_server_after) = context.get_config_delete_server_after().await? {
251            let until = stock_str::download_availability(
252                context,
253                time() + max(delete_server_after, MIN_DELETE_SERVER_AFTER),
254            )
255            .await;
256            text += format!(" [{until}]").as_str();
257        };
258
259        info!(context, "Partial download: {}", text);
260
261        self.parts.push(Part {
262            typ: Viewtype::Text,
263            msg: text,
264            ..Default::default()
265        });
266
267        Ok(())
268    }
269}
270
271#[cfg(test)]
272mod tests {
273    use num_traits::FromPrimitive;
274
275    use super::*;
276    use crate::chat::{get_chat_msgs, send_msg};
277    use crate::ephemeral::Timer;
278    use crate::receive_imf::receive_imf_from_inbox;
279    use crate::test_utils::TestContext;
280
281    #[test]
282    fn test_downloadstate_values() {
283        // values may be written to disk and must not change
284        assert_eq!(DownloadState::Done, DownloadState::default());
285        assert_eq!(DownloadState::Done, DownloadState::from_i32(0).unwrap());
286        assert_eq!(
287            DownloadState::Available,
288            DownloadState::from_i32(10).unwrap()
289        );
290        assert_eq!(DownloadState::Failure, DownloadState::from_i32(20).unwrap());
291        assert_eq!(
292            DownloadState::InProgress,
293            DownloadState::from_i32(1000).unwrap()
294        );
295    }
296
297    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
298    async fn test_download_limit() -> Result<()> {
299        let t = TestContext::new_alice().await;
300
301        assert_eq!(t.download_limit().await?, None);
302
303        t.set_config(Config::DownloadLimit, Some("200000")).await?;
304        assert_eq!(t.download_limit().await?, Some(200000));
305
306        t.set_config(Config::DownloadLimit, Some("20000")).await?;
307        assert_eq!(t.download_limit().await?, Some(MIN_DOWNLOAD_LIMIT));
308
309        t.set_config(Config::DownloadLimit, None).await?;
310        assert_eq!(t.download_limit().await?, None);
311
312        for val in &["0", "-1", "-100", "", "foo"] {
313            t.set_config(Config::DownloadLimit, Some(val)).await?;
314            assert_eq!(t.download_limit().await?, None);
315        }
316
317        Ok(())
318    }
319
320    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
321    async fn test_update_download_state() -> Result<()> {
322        let t = TestContext::new_alice().await;
323        let chat = t.create_chat_with_contact("Bob", "bob@example.org").await;
324
325        let mut msg = Message::new_text("Hi Bob".to_owned());
326        let msg_id = send_msg(&t, chat.id, &mut msg).await?;
327        let msg = Message::load_from_db(&t, msg_id).await?;
328        assert_eq!(msg.download_state(), DownloadState::Done);
329
330        for s in &[
331            DownloadState::Available,
332            DownloadState::InProgress,
333            DownloadState::Failure,
334            DownloadState::Done,
335            DownloadState::Done,
336        ] {
337            msg_id.update_download_state(&t, *s).await?;
338            let msg = Message::load_from_db(&t, msg_id).await?;
339            assert_eq!(msg.download_state(), *s);
340        }
341        t.sql
342            .execute("DELETE FROM msgs WHERE id=?", (msg_id,))
343            .await?;
344        // Nothing to do is ok.
345        msg_id
346            .update_download_state(&t, DownloadState::Done)
347            .await?;
348
349        Ok(())
350    }
351
352    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
353    async fn test_partial_receive_imf() -> Result<()> {
354        let t = TestContext::new_alice().await;
355
356        let header =
357            "Received: (Postfix, from userid 1000); Mon, 4 Dec 2006 14:51:39 +0100 (CET)\n\
358             From: bob@example.com\n\
359             To: alice@example.org\n\
360             Subject: foo\n\
361             Message-ID: <Mr.12345678901@example.com>\n\
362             Chat-Version: 1.0\n\
363             Date: Sun, 22 Mar 2020 22:37:57 +0000\
364             Content-Type: text/plain";
365
366        receive_imf_from_inbox(
367            &t,
368            "Mr.12345678901@example.com",
369            header.as_bytes(),
370            false,
371            Some(100000),
372        )
373        .await?;
374        let msg = t.get_last_msg().await;
375        assert_eq!(msg.download_state(), DownloadState::Available);
376        assert_eq!(msg.get_subject(), "foo");
377        assert!(msg
378            .get_text()
379            .contains(&stock_str::partial_download_msg_body(&t, 100000).await));
380
381        receive_imf_from_inbox(
382            &t,
383            "Mr.12345678901@example.com",
384            format!("{header}\n\n100k text...").as_bytes(),
385            false,
386            None,
387        )
388        .await?;
389        let msg = t.get_last_msg().await;
390        assert_eq!(msg.download_state(), DownloadState::Done);
391        assert_eq!(msg.get_subject(), "foo");
392        assert_eq!(msg.get_text(), "100k text...");
393
394        Ok(())
395    }
396
397    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
398    async fn test_partial_download_and_ephemeral() -> Result<()> {
399        let t = TestContext::new_alice().await;
400        let chat_id = t
401            .create_chat_with_contact("bob", "bob@example.org")
402            .await
403            .id;
404        chat_id
405            .set_ephemeral_timer(&t, Timer::Enabled { duration: 60 })
406            .await?;
407
408        // download message from bob partially, this must not change the ephemeral timer
409        receive_imf_from_inbox(
410            &t,
411            "first@example.org",
412            b"From: Bob <bob@example.org>\n\
413                    To: Alice <alice@example.org>\n\
414                    Chat-Version: 1.0\n\
415                    Subject: subject\n\
416                    Message-ID: <first@example.org>\n\
417                    Date: Sun, 14 Nov 2021 00:10:00 +0000\
418                    Content-Type: text/plain",
419            false,
420            Some(100000),
421        )
422        .await?;
423        assert_eq!(
424            chat_id.get_ephemeral_timer(&t).await?,
425            Timer::Enabled { duration: 60 }
426        );
427
428        Ok(())
429    }
430
431    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
432    async fn test_status_update_expands_to_nothing() -> Result<()> {
433        let alice = TestContext::new_alice().await;
434        let bob = TestContext::new_bob().await;
435        let chat_id = alice.create_chat(&bob).await.id;
436
437        let file = alice.get_blobdir().join("minimal.xdc");
438        tokio::fs::write(&file, include_bytes!("../test-data/webxdc/minimal.xdc")).await?;
439        let mut instance = Message::new(Viewtype::File);
440        instance.set_file_and_deduplicate(&alice, &file, None, None)?;
441        let _sent1 = alice.send_msg(chat_id, &mut instance).await;
442
443        alice
444            .send_webxdc_status_update(instance.id, r#"{"payload":7}"#)
445            .await?;
446        alice.flush_status_updates().await?;
447        let sent2 = alice.pop_sent_msg().await;
448        let sent2_rfc724_mid = sent2.load_from_db().await.rfc724_mid;
449
450        // not downloading the status update results in an placeholder
451        receive_imf_from_inbox(
452            &bob,
453            &sent2_rfc724_mid,
454            sent2.payload().as_bytes(),
455            false,
456            Some(sent2.payload().len() as u32),
457        )
458        .await?;
459        let msg = bob.get_last_msg().await;
460        let chat_id = msg.chat_id;
461        assert_eq!(get_chat_msgs(&bob, chat_id).await?.len(), 1);
462        assert_eq!(msg.download_state(), DownloadState::Available);
463
464        // downloading the status update afterwards expands to nothing and moves the placeholder to trash-chat
465        // (usually status updates are too small for not being downloaded directly)
466        receive_imf_from_inbox(
467            &bob,
468            &sent2_rfc724_mid,
469            sent2.payload().as_bytes(),
470            false,
471            None,
472        )
473        .await?;
474        assert_eq!(get_chat_msgs(&bob, chat_id).await?.len(), 0);
475        assert!(Message::load_from_db_optional(&bob, msg.id)
476            .await?
477            .is_none());
478
479        Ok(())
480    }
481
482    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
483    async fn test_mdn_expands_to_nothing() -> Result<()> {
484        let bob = TestContext::new_bob().await;
485        let raw = b"Subject: Message opened\n\
486            Date: Mon, 10 Jan 2020 00:00:00 +0000\n\
487            Chat-Version: 1.0\n\
488            Message-ID: <bar@example.org>\n\
489            To: Alice <alice@example.org>\n\
490            From: Bob <bob@example.org>\n\
491            Content-Type: multipart/report; report-type=disposition-notification;\n\t\
492            boundary=\"kJBbU58X1xeWNHgBtTbMk80M5qnV4N\"\n\
493            \n\
494            \n\
495            --kJBbU58X1xeWNHgBtTbMk80M5qnV4N\n\
496            Content-Type: text/plain; charset=utf-8\n\
497            \n\
498            bla\n\
499            \n\
500            \n\
501            --kJBbU58X1xeWNHgBtTbMk80M5qnV4N\n\
502            Content-Type: message/disposition-notification\n\
503            \n\
504            Reporting-UA: Delta Chat 1.88.0\n\
505            Original-Recipient: rfc822;bob@example.org\n\
506            Final-Recipient: rfc822;bob@example.org\n\
507            Original-Message-ID: <foo@example.org>\n\
508            Disposition: manual-action/MDN-sent-automatically; displayed\n\
509            \n\
510            \n\
511            --kJBbU58X1xeWNHgBtTbMk80M5qnV4N--\n\
512            ";
513
514        // not downloading the mdn results in an placeholder
515        receive_imf_from_inbox(&bob, "bar@example.org", raw, false, Some(raw.len() as u32)).await?;
516        let msg = bob.get_last_msg().await;
517        let chat_id = msg.chat_id;
518        assert_eq!(get_chat_msgs(&bob, chat_id).await?.len(), 1);
519        assert_eq!(msg.download_state(), DownloadState::Available);
520
521        // downloading the mdn afterwards expands to nothing and deletes the placeholder directly
522        // (usually mdn are too small for not being downloaded directly)
523        receive_imf_from_inbox(&bob, "bar@example.org", raw, false, None).await?;
524        assert_eq!(get_chat_msgs(&bob, chat_id).await?.len(), 0);
525        assert!(Message::load_from_db_optional(&bob, msg.id)
526            .await?
527            .is_none());
528
529        Ok(())
530    }
531}