Skip to main content

max / goingson

6.9 KB · 196 lines History Blame Raw
1 //! Threaded list assembly: the paginated thread view and the per-thread fetch.
2 //!
3 //! `list_threaded` is four queries in sequence (count, ranked thread summary,
4 //! the page's full emails, then the assembly pass), which is why it has a
5 //! module to itself.
6
7 use std::collections::HashMap;
8
9 use goingson_core::{CoreError, Email, EmailThread, Result, UserId};
10 use rusqlite::{Connection, params, params_from_iter};
11
12 use crate::utils::{bind_placeholders, query_all};
13
14 use super::row::{EMAIL_SELECT_COLUMNS, EmailRow};
15
16 /// One page of threads, newest first, with the total thread count.
17 pub(super) fn list_threaded(
18 conn: &Connection,
19 user_id: UserId,
20 include_archived: bool,
21 offset: Option<i64>,
22 limit: Option<i64>,
23 folder: Option<&str>,
24 label: Option<&str>,
25 ) -> Result<(Vec<EmailThread>, i64)> {
26 let uid = user_id.to_string();
27 let archived_filter = if include_archived {
28 ""
29 } else {
30 "AND e.is_archived = 0"
31 };
32 let folder_filter = folder.map_or("", |_| "AND e.source_folder = ?");
33 let label_filter = label.map_or(
34 "",
35 |_| "AND EXISTS (SELECT 1 FROM json_each(e.labels) j WHERE j.value = ?)",
36 );
37 // Defense-in-depth: clamp before binding. A negative LIMIT means
38 // unbounded in SQLite (would load the whole mailbox); a negative OFFSET
39 // is ignored. Matches task_repo/search_repo.
40 const MAX_PAGE_LIMIT: i64 = 1000;
41 let offset_val = offset.unwrap_or(0).max(0);
42 let limit_val = limit.unwrap_or(50).clamp(0, MAX_PAGE_LIMIT);
43
44 // Query 1: Get total thread count
45 let count_sql = format!(
46 "SELECT COUNT(DISTINCT COALESCE(e.thread_id, e.id)) FROM emails e WHERE e.user_id = ? AND e.is_draft = 0 {archived_filter} {folder_filter} {label_filter}"
47 );
48 // The optional folder/label filters add a placeholder each, so the binds are
49 // assembled in the same order the filters were spliced into the SQL.
50 let mut filter_binds: Vec<String> = vec![uid.clone()];
51 if let Some(f) = folder {
52 filter_binds.push(f.to_string());
53 }
54 if let Some(l) = label {
55 filter_binds.push(l.to_string());
56 }
57
58 let total: i64 = conn
59 .query_row(&count_sql, params_from_iter(&filter_binds), |row| {
60 row.get(0)
61 })
62 .map_err(CoreError::database)?;
63
64 if total == 0 {
65 return Ok((vec![], 0));
66 }
67
68 // Query 2: Thread summary, group by thread, get latest received_at, count, unread status
69 #[allow(dead_code)]
70 struct ThreadSummary {
71 thread_key: String,
72 latest_received_at: String, // needed for SQL ORDER BY
73 thread_count: i64,
74 unread_count: i64,
75 latest_email_id: String,
76 }
77
78 // Rank emails within each thread by recency in a single pass with window
79 // functions, then keep the latest row per thread. This replaces a
80 // correlated subquery that rescanned `emails` once per thread group
81 // (O(threads x emails)); the partition's MAX(received_at) is the rn = 1
82 // row, and the thread-wide counts come from window aggregates.
83 let summary_sql = format!(
84 r"WITH ranked AS (
85 SELECT
86 e.id AS email_id,
87 COALESCE(e.thread_id, e.id) AS thread_key,
88 e.received_at AS received_at,
89 ROW_NUMBER() OVER (
90 PARTITION BY COALESCE(e.thread_id, e.id)
91 ORDER BY e.received_at DESC, e.id DESC
92 ) AS rn,
93 COUNT(*) OVER (PARTITION BY COALESCE(e.thread_id, e.id)) AS thread_count,
94 SUM(CASE WHEN e.is_read = 0 THEN 1 ELSE 0 END)
95 OVER (PARTITION BY COALESCE(e.thread_id, e.id)) AS unread_count
96 FROM emails e
97 WHERE e.user_id = ? AND e.is_draft = 0 {archived_filter} {folder_filter} {label_filter}
98 )
99 SELECT
100 thread_key,
101 received_at AS latest_received_at,
102 thread_count,
103 unread_count,
104 email_id AS latest_email_id
105 FROM ranked
106 WHERE rn = 1
107 ORDER BY latest_received_at DESC
108 LIMIT ? OFFSET ?",
109 );
110
111 let mut summary_binds: Vec<rusqlite::types::Value> = filter_binds
112 .iter()
113 .map(|s| rusqlite::types::Value::from(s.clone()))
114 .collect();
115 summary_binds.push(limit_val.into());
116 summary_binds.push(offset_val.into());
117
118 let summaries = query_all(conn, &summary_sql, params_from_iter(summary_binds), |row| {
119 Ok(ThreadSummary {
120 thread_key: row.get("thread_key")?,
121 latest_received_at: row.get("latest_received_at")?,
122 thread_count: row.get("thread_count")?,
123 unread_count: row.get("unread_count")?,
124 latest_email_id: row.get("latest_email_id")?,
125 })
126 })?;
127
128 if summaries.is_empty() {
129 return Ok((vec![], total));
130 }
131
132 // Query 3: Fetch full emails for the page's most-recent-email IDs
133 let email_ids: Vec<String> = summaries
134 .iter()
135 .map(|s| s.latest_email_id.clone())
136 .collect();
137 let placeholders = bind_placeholders(email_ids.len());
138 let emails_sql = format!(
139 "SELECT {EMAIL_SELECT_COLUMNS} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.id IN ({placeholders}) AND e.user_id = ?"
140 );
141
142 let mut email_binds: Vec<String> = Vec::with_capacity(email_ids.len() + 2);
143 email_binds.push(uid.clone());
144 email_binds.extend(email_ids.iter().cloned());
145 email_binds.push(uid.clone());
146
147 let rows = query_all(
148 conn,
149 &emails_sql,
150 params_from_iter(email_binds),
151 EmailRow::from_row,
152 )?;
153
154 let email_map: HashMap<String, Email> = rows
155 .into_iter()
156 .filter_map(|row| {
157 let id_str = row.id.clone();
158 Email::try_from(row).ok().map(|e| (id_str, e))
159 })
160 .collect();
161
162 // Assemble threads in summary order
163 let threads: Vec<EmailThread> = summaries
164 .into_iter()
165 .filter_map(|s| {
166 let email = email_map.get(&s.latest_email_id)?.clone();
167 Some(EmailThread {
168 thread_id: s.thread_key,
169 most_recent_email: email,
170 thread_count: s.thread_count as usize,
171 has_unread: s.unread_count > 0,
172 })
173 })
174 .collect();
175
176 Ok((threads, total))
177 }
178
179 /// Every email in one thread, oldest first.
180 pub(super) fn list_by_thread(
181 conn: &Connection,
182 user_id: UserId,
183 thread_id: &str,
184 ) -> Result<Vec<Email>> {
185 let query = format!(
186 "SELECT {EMAIL_SELECT_COLUMNS} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.thread_id = ? ORDER BY e.received_at ASC"
187 );
188 let rows = query_all(
189 conn,
190 &query,
191 params![user_id.to_string(), user_id.to_string(), thread_id],
192 EmailRow::from_row,
193 )?;
194 rows.into_iter().map(Email::try_from).collect()
195 }
196