mxr_store/
thread.rs

1use mxr_core::id::*;
2use mxr_core::types::*;
3
4use crate::message::{future_date_cutoff_timestamp, record_to_envelope};
5
6impl super::Store {
7    pub async fn get_thread(&self, thread_id: &ThreadId) -> Result<Option<Thread>, sqlx::Error> {
8        let tid = thread_id.as_str();
9        let cutoff = future_date_cutoff_timestamp();
10        let row = sqlx::query!(
11            r#"SELECT
12                thread_id as "thread_id!",
13                account_id as "account_id!",
14                MIN(subject) as "subject!: String",
15                COUNT(*) as "message_count!: i64",
16                SUM(CASE WHEN (flags & 1) = 0 THEN 1 ELSE 0 END) as "unread_count!: i64",
17                MAX(CASE WHEN date > ? THEN 0 ELSE date END) as "latest_date!: i64",
18                snippet as "snippet!"
19             FROM messages
20             WHERE thread_id = ?
21             GROUP BY thread_id"#,
22            cutoff,
23            tid,
24        )
25        .fetch_optional(self.reader())
26        .await?;
27
28        let row = match row {
29            Some(r) => r,
30            None => return Ok(None),
31        };
32
33        // Get participants
34        let tid2 = thread_id.as_str();
35        let participant_rows = sqlx::query!(
36            r#"SELECT DISTINCT from_name, from_email as "from_email!" FROM messages WHERE thread_id = ?"#,
37            tid2,
38        )
39        .fetch_all(self.reader())
40        .await?;
41
42        let participants: Vec<Address> = participant_rows
43            .into_iter()
44            .map(|r| Address {
45                name: r.from_name,
46                email: r.from_email,
47            })
48            .collect();
49
50        Ok(Some(Thread {
51            id: ThreadId::from_uuid(uuid::Uuid::parse_str(&row.thread_id).unwrap()),
52            account_id: AccountId::from_uuid(uuid::Uuid::parse_str(&row.account_id).unwrap()),
53            subject: row.subject,
54            participants,
55            message_count: row.message_count as u32,
56            unread_count: row.unread_count as u32,
57            latest_date: chrono::DateTime::from_timestamp(row.latest_date, 0).unwrap_or_default(),
58            snippet: row.snippet,
59        }))
60    }
61
62    pub async fn get_thread_envelopes(
63        &self,
64        thread_id: &ThreadId,
65    ) -> Result<Vec<Envelope>, sqlx::Error> {
66        let tid = thread_id.as_str();
67        let cutoff = future_date_cutoff_timestamp();
68        let rows = sqlx::query!(
69            r#"SELECT
70                id as "id!", account_id as "account_id!", provider_id as "provider_id!",
71                thread_id as "thread_id!", message_id_header, in_reply_to,
72                reference_headers, from_name, from_email as "from_email!",
73                to_addrs as "to_addrs!", cc_addrs as "cc_addrs!", bcc_addrs as "bcc_addrs!",
74                subject as "subject!", date as "date!", flags as "flags!",
75                snippet as "snippet!", has_attachments as "has_attachments!: bool",
76                size_bytes as "size_bytes!", unsubscribe_method,
77                COALESCE((
78                    SELECT GROUP_CONCAT(labels.provider_id, char(31))
79                    FROM message_labels
80                    JOIN labels ON labels.id = message_labels.label_id
81                    WHERE message_labels.message_id = messages.id
82                ), '') as "label_provider_ids!: String"
83             FROM messages
84             WHERE thread_id = ?
85             ORDER BY CASE WHEN date > ? THEN 0 ELSE date END ASC, id ASC"#,
86            tid,
87            cutoff,
88        )
89        .fetch_all(self.reader())
90        .await?;
91
92        Ok(rows
93            .into_iter()
94            .map(|r| {
95                record_to_envelope(
96                    &r.id,
97                    &r.account_id,
98                    &r.provider_id,
99                    &r.thread_id,
100                    r.message_id_header.as_deref(),
101                    r.in_reply_to.as_deref(),
102                    r.reference_headers.as_deref(),
103                    r.from_name.as_deref(),
104                    &r.from_email,
105                    &r.to_addrs,
106                    &r.cc_addrs,
107                    &r.bcc_addrs,
108                    &r.subject,
109                    r.date,
110                    r.flags,
111                    &r.snippet,
112                    r.has_attachments,
113                    r.size_bytes,
114                    r.unsubscribe_method.as_deref(),
115                    &r.label_provider_ids,
116                )
117            })
118            .collect())
119    }
120}
mxr_store/thread.rs

mxr_store/
thread.rs
