Skip to main content

max / goingson

37.2 KB · 772 lines History Blame Raw
1 //! SQLite implementation of the EmailRepository.
2 //!
3 //! Manages email messages with support for:
4 //! - IMAP synchronization tracking (message_id, imap_uid)
5 //! - Threading via in_reply_to and thread_id
6 //! - Read/archived status
7 //! - Project associations
8 //! - Snoozing and waiting-for-response tracking
9
10 use async_trait::async_trait;
11 use chrono::{DateTime, Utc};
12 use sqlx::SqlitePool;
13 use std::collections::HashSet;
14 use goingson_core::{
15 CoreError, Email, EmailAccountId, EmailId, EmailRepository, EmailThread, NewEmail,
16 NewEmailWithTracking, ProjectId, Result, UserId,
17 };
18 use std::collections::HashMap;
19
20 use crate::utils::{bind_placeholders, format_datetime, format_datetime_now, format_datetime_opt, parse_datetime, parse_uuid, parse_uuid_opt};
21
22 /// Column list for SELECT queries - avoids duplication across methods.
23 const EMAIL_SELECT_COLUMNS: &str = r#"e.id, e.project_id, p.name as project_name, e.from_address, e.to_address,
24 e.subject, e.body, e.html_body, e.is_read, e.is_archived, e.received_at, e.message_id,
25 e.in_reply_to, e.thread_id, e.email_account_id, e.is_outgoing, e.imap_uid, e.source_folder,
26 e.attachment_meta, e.labels, e.is_draft, e.cc_address, e.bcc_address, e.draft_account_id,
27 e.snoozed_until, e.waiting_for_response, e.waiting_since, e.expected_response_date,
28 e.body_truncated, e.jmap_id"#;
29
30 /// Same shape as [`EMAIL_SELECT_COLUMNS`] but with the two heavy body columns
31 /// blanked, for flat list views that never render a body (Perf S3). `body_truncated`
32 /// is forced to 1 so the reader knows to re-fetch the full body on open.
33 const EMAIL_LIST_COLUMNS: &str = r#"e.id, e.project_id, p.name as project_name, e.from_address, e.to_address,
34 e.subject, '' AS body, NULL AS html_body, e.is_read, e.is_archived, e.received_at, e.message_id,
35 e.in_reply_to, e.thread_id, e.email_account_id, e.is_outgoing, e.imap_uid, e.source_folder,
36 e.attachment_meta, e.labels, e.is_draft, e.cc_address, e.bcc_address, e.draft_account_id,
37 e.snoozed_until, e.waiting_for_response, e.waiting_since, e.expected_response_date,
38 1 AS body_truncated, e.jmap_id"#;
39
40 /// Upper bound on a flat metadata list so it can never materialize an unbounded
41 /// number of rows (the list UI paginates via `list_threaded`; this flat path is a
42 /// safety-capped fallback).
43 const EMAIL_LIST_CAP: i64 = 1000;
44
45 #[derive(Debug, Clone, sqlx::FromRow)]
46 struct EmailRow {
47 pub id: String,
48 pub project_id: Option<String>,
49 pub project_name: Option<String>,
50 pub from_address: String,
51 pub to_address: String,
52 pub subject: String,
53 pub body: String,
54 pub html_body: Option<String>,
55 pub is_read: i32,
56 pub is_archived: i32,
57 pub received_at: String,
58 pub message_id: Option<String>,
59 pub in_reply_to: Option<String>,
60 pub thread_id: Option<String>,
61 pub email_account_id: Option<String>,
62 pub is_outgoing: i32,
63 pub imap_uid: Option<i64>,
64 pub source_folder: Option<String>,
65 pub attachment_meta: Option<String>,
66 pub labels: String,
67 pub is_draft: i32,
68 pub cc_address: Option<String>,
69 pub bcc_address: Option<String>,
70 pub draft_account_id: Option<String>,
71 pub snoozed_until: Option<String>,
72 pub waiting_for_response: i32,
73 pub waiting_since: Option<String>,
74 pub expected_response_date: Option<String>,
75 pub body_truncated: i32,
76 pub jmap_id: Option<String>,
77 }
78
79 impl TryFrom<EmailRow> for Email {
80 type Error = CoreError;
81
82 fn try_from(row: EmailRow) -> std::result::Result<Self, Self::Error> {
83 Ok(Email {
84 id: parse_uuid(&row.id)?.into(),
85 project_id: parse_uuid_opt(row.project_id.as_deref())?.map(Into::into),
86 project_name: row.project_name,
87 from: row.from_address,
88 to: row.to_address,
89 subject: row.subject,
90 body: row.body,
91 html_body: row.html_body,
92 body_truncated: row.body_truncated != 0,
93 jmap_id: row.jmap_id,
94 is_read: row.is_read != 0,
95 is_archived: row.is_archived != 0,
96 received_at: parse_datetime(&row.received_at)?,
97 message_id: row.message_id,
98 in_reply_to: row.in_reply_to,
99 thread_id: row.thread_id,
100 email_account_id: parse_uuid_opt(row.email_account_id.as_deref())?.map(Into::into),
101 is_outgoing: row.is_outgoing != 0,
102 imap_uid: row.imap_uid,
103 source_folder: row.source_folder,
104 attachment_meta: row.attachment_meta,
105 labels: serde_json::from_str(&row.labels).unwrap_or_default(),
106 is_draft: row.is_draft != 0,
107 cc_address: row.cc_address,
108 bcc_address: row.bcc_address,
109 draft_account_id: parse_uuid_opt(row.draft_account_id.as_deref())?.map(Into::into),
110 snoozed_until: row.snoozed_until.as_ref().map(|s| parse_datetime(s)).transpose()?,
111 waiting_for_response: row.waiting_for_response != 0,
112 waiting_since: row.waiting_since.as_ref().map(|s| parse_datetime(s)).transpose()?,
113 expected_response_date: row.expected_response_date.as_ref().map(|s| parse_datetime(s)).transpose()?,
114 })
115 }
116 }
117
118 /// SQLite-backed implementation of [`EmailRepository`].
119 ///
120 /// Manages email messages with threading support, snoozing, and
121 /// waiting-for-response tracking. Integrates with IMAP sync via message_id.
122 pub struct SqliteEmailRepository { pool: SqlitePool }
123
124 impl SqliteEmailRepository {
125 /// Creates a new repository instance with the given connection pool.
126 #[tracing::instrument(skip_all)]
127 pub fn new(pool: SqlitePool) -> Self { Self { pool } }
128 }
129
130 #[async_trait]
131 impl EmailRepository for SqliteEmailRepository {
132 #[tracing::instrument(skip_all)]
133 async fn list_all(&self, user_id: UserId, include_archived: bool) -> Result<Vec<Email>> {
134 let archived_filter = if include_archived { "" } else { "AND e.is_archived = 0" };
135 let query = format!(
136 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.is_draft = 0 {} ORDER BY e.received_at DESC",
137 EMAIL_SELECT_COLUMNS, archived_filter
138 );
139 let rows = sqlx::query_as::<_, EmailRow>(&query).bind(user_id.to_string()).bind(user_id.to_string()).fetch_all(&self.pool).await.map_err(CoreError::database)?;
140 rows.into_iter().map(Email::try_from).collect()
141 }
142
143 #[tracing::instrument(skip_all)]
144 async fn list_metadata(&self, user_id: UserId, include_archived: bool) -> Result<Vec<Email>> {
145 let archived_filter = if include_archived { "" } else { "AND e.is_archived = 0" };
146 let query = format!(
147 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.is_draft = 0 {} ORDER BY e.received_at DESC LIMIT {}",
148 EMAIL_LIST_COLUMNS, archived_filter, EMAIL_LIST_CAP
149 );
150 let rows = sqlx::query_as::<_, EmailRow>(&query).bind(user_id.to_string()).bind(user_id.to_string()).fetch_all(&self.pool).await.map_err(CoreError::database)?;
151 rows.into_iter().map(Email::try_from).collect()
152 }
153
154 #[tracing::instrument(skip_all)]
155 async fn list_threaded(&self, user_id: UserId, include_archived: bool, offset: Option<i64>, limit: Option<i64>, folder: Option<&str>, label: Option<&str>) -> Result<(Vec<EmailThread>, i64)> {
156 let uid = user_id.to_string();
157 let archived_filter = if include_archived { "" } else { "AND e.is_archived = 0" };
158 let folder_filter = folder.map(|_| "AND e.source_folder = ?").unwrap_or("");
159 let label_filter = label.map(|_| "AND EXISTS (SELECT 1 FROM json_each(e.labels) j WHERE j.value = ?)").unwrap_or("");
160 // Defense-in-depth: clamp before binding. A negative LIMIT means
161 // unbounded in SQLite (would load the whole mailbox); a negative OFFSET
162 // is ignored. Matches task_repo/search_repo.
163 const MAX_PAGE_LIMIT: i64 = 1000;
164 let offset_val = offset.unwrap_or(0).max(0);
165 let limit_val = limit.unwrap_or(50).clamp(0, MAX_PAGE_LIMIT);
166
167 // Query 1: Get total thread count
168 let count_sql = format!(
169 "SELECT COUNT(DISTINCT COALESCE(e.thread_id, e.id)) FROM emails e WHERE e.user_id = ? AND e.is_draft = 0 {} {} {}",
170 archived_filter, folder_filter, label_filter
171 );
172 let mut count_q = sqlx::query_as::<_, (i64,)>(&count_sql).bind(&uid);
173 if let Some(f) = folder { count_q = count_q.bind(f); }
174 if let Some(l) = label { count_q = count_q.bind(l); }
175 let (total,) = count_q.fetch_one(&self.pool).await.map_err(CoreError::database)?;
176
177 if total == 0 {
178 return Ok((vec![], 0));
179 }
180
181 // Query 2: Thread summary — group by thread, get latest received_at, count, unread status
182 #[derive(sqlx::FromRow)]
183 #[allow(dead_code)]
184 struct ThreadSummary {
185 thread_key: String,
186 latest_received_at: String, // needed for SQL ORDER BY
187 thread_count: i64,
188 unread_count: i64,
189 latest_email_id: String,
190 }
191
192 // Rank emails within each thread by recency in a single pass with window
193 // functions, then keep the latest row per thread. This replaces a
194 // correlated subquery that rescanned `emails` once per thread group
195 // (O(threads x emails)); the partition's MAX(received_at) is the rn = 1
196 // row, and the thread-wide counts come from window aggregates.
197 let summary_sql = format!(
198 r#"WITH ranked AS (
199 SELECT
200 e.id AS email_id,
201 COALESCE(e.thread_id, e.id) AS thread_key,
202 e.received_at AS received_at,
203 ROW_NUMBER() OVER (
204 PARTITION BY COALESCE(e.thread_id, e.id)
205 ORDER BY e.received_at DESC, e.id DESC
206 ) AS rn,
207 COUNT(*) OVER (PARTITION BY COALESCE(e.thread_id, e.id)) AS thread_count,
208 SUM(CASE WHEN e.is_read = 0 THEN 1 ELSE 0 END)
209 OVER (PARTITION BY COALESCE(e.thread_id, e.id)) AS unread_count
210 FROM emails e
211 WHERE e.user_id = ? AND e.is_draft = 0 {} {} {}
212 )
213 SELECT
214 thread_key,
215 received_at AS latest_received_at,
216 thread_count,
217 unread_count,
218 email_id AS latest_email_id
219 FROM ranked
220 WHERE rn = 1
221 ORDER BY latest_received_at DESC
222 LIMIT ? OFFSET ?"#,
223 archived_filter, folder_filter, label_filter,
224 );
225
226 let mut summary_q = sqlx::query_as::<_, ThreadSummary>(&summary_sql).bind(&uid);
227 if let Some(f) = folder { summary_q = summary_q.bind(f); }
228 if let Some(l) = label { summary_q = summary_q.bind(l); }
229 let summaries = summary_q
230 .bind(limit_val)
231 .bind(offset_val)
232 .fetch_all(&self.pool)
233 .await
234 .map_err(CoreError::database)?;
235
236 if summaries.is_empty() {
237 return Ok((vec![], total));
238 }
239
240 // Query 3: Fetch full emails for the page's most-recent-email IDs
241 let email_ids: Vec<String> = summaries.iter().map(|s| s.latest_email_id.clone()).collect();
242 let placeholders = bind_placeholders(email_ids.len());
243 let emails_sql = format!(
244 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.id IN ({}) AND e.user_id = ?",
245 EMAIL_SELECT_COLUMNS, placeholders
246 );
247
248 let mut q = sqlx::query_as::<_, EmailRow>(&emails_sql).bind(&uid);
249 for id in &email_ids {
250 q = q.bind(id);
251 }
252 q = q.bind(&uid);
253
254 let rows = q.fetch_all(&self.pool).await.map_err(CoreError::database)?;
255
256 let email_map: HashMap<String, Email> = rows
257 .into_iter()
258 .filter_map(|row| {
259 let id_str = row.id.clone();
260 Email::try_from(row).ok().map(|e| (id_str, e))
261 })
262 .collect();
263
264 // Assemble threads in summary order
265 let threads: Vec<EmailThread> = summaries
266 .into_iter()
267 .filter_map(|s| {
268 let email = email_map.get(&s.latest_email_id)?.clone();
269 Some(EmailThread {
270 thread_id: s.thread_key,
271 most_recent_email: email,
272 thread_count: s.thread_count as usize,
273 has_unread: s.unread_count > 0,
274 })
275 })
276 .collect();
277
278 Ok((threads, total))
279 }
280
281 #[tracing::instrument(skip_all)]
282 async fn list_by_project(&self, user_id: UserId, project_id: ProjectId) -> Result<Vec<Email>> {
283 // Body-less + capped: the project dashboard renders only subject/from/date
284 // and opens the reader (which re-fetches the full body via `get_by_id`) on
285 // click, so this must not materialize every body into RAM (Perf S3, same
286 // rule as `list_metadata`). `EMAIL_LIST_COLUMNS` forces `body_truncated=1`.
287 let query = format!(
288 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.project_id = ? ORDER BY e.received_at DESC LIMIT {}",
289 EMAIL_LIST_COLUMNS, EMAIL_LIST_CAP
290 );
291 let rows = sqlx::query_as::<_, EmailRow>(&query).bind(user_id.to_string()).bind(user_id.to_string()).bind(project_id.to_string()).fetch_all(&self.pool).await.map_err(CoreError::database)?;
292 rows.into_iter().map(Email::try_from).collect()
293 }
294
295 #[tracing::instrument(skip_all)]
296 async fn list_by_addresses(&self, user_id: UserId, addresses: &[&str]) -> Result<Vec<Email>> {
297 if addresses.is_empty() {
298 return Ok(Vec::new());
299 }
300 let placeholders = bind_placeholders(addresses.len());
301 let query = format!(
302 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND (LOWER(e.from_address) IN ({placeholders}) OR LOWER(e.to_address) IN ({placeholders})) ORDER BY e.received_at DESC LIMIT 200",
303 EMAIL_SELECT_COLUMNS
304 );
305 let mut q = sqlx::query_as::<_, EmailRow>(&query)
306 .bind(user_id.to_string())
307 .bind(user_id.to_string());
308 // Bind addresses twice (once for from_address IN, once for to_address IN)
309 for _ in 0..2 {
310 for addr in addresses {
311 q = q.bind(addr.to_lowercase());
312 }
313 }
314 let rows = q.fetch_all(&self.pool).await.map_err(CoreError::database)?;
315 rows.into_iter().map(Email::try_from).collect()
316 }
317
318 #[tracing::instrument(skip_all)]
319 async fn list_unlinked(&self, user_id: UserId) -> Result<Vec<Email>> {
320 // Body-less + capped: the sole caller is the "link email to project" picker,
321 // which shows only subject/from. No body needed, and an unbounded mailbox
322 // must not load every body into RAM (Perf S3). `EMAIL_LIST_COLUMNS` forces
323 // `body_truncated=1` so any later reader re-fetches the full body.
324 let query = format!(
325 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.project_id IS NULL AND e.is_archived = 0 ORDER BY e.received_at DESC LIMIT {}",
326 EMAIL_LIST_COLUMNS, EMAIL_LIST_CAP
327 );
328 let rows = sqlx::query_as::<_, EmailRow>(&query).bind(user_id.to_string()).bind(user_id.to_string()).fetch_all(&self.pool).await.map_err(CoreError::database)?;
329 rows.into_iter().map(Email::try_from).collect()
330 }
331
332 #[tracing::instrument(skip_all)]
333 async fn get_by_id(&self, id: EmailId, user_id: UserId) -> Result<Option<Email>> {
334 let query = format!(
335 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.id = ? AND e.user_id = ?",
336 EMAIL_SELECT_COLUMNS
337 );
338 let row = sqlx::query_as::<_, EmailRow>(&query).bind(user_id.to_string()).bind(id.to_string()).bind(user_id.to_string()).fetch_optional(&self.pool).await.map_err(CoreError::database)?;
339 row.map(Email::try_from).transpose()
340 }
341
342 #[tracing::instrument(skip_all)]
343 async fn create(&self, user_id: UserId, email: NewEmail) -> Result<Email> {
344 let id = EmailId::new();
345 let received_at = format_datetime(&email.received_at.unwrap_or_else(Utc::now));
346 sqlx::query("INSERT INTO emails (id, user_id, project_id, from_address, to_address, subject, body, is_read, received_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)")
347 .bind(id.to_string()).bind(user_id.to_string()).bind(email.project_id.map(|p| p.to_string()))
348 .bind(&email.from_address).bind(&email.to_address).bind(&email.subject).bind(&email.body)
349 .bind(if email.is_read { 1 } else { 0 }).bind(&received_at)
350 .execute(&self.pool).await.map_err(CoreError::database)?;
351 self.get_by_id(id, user_id).await?.ok_or_else(|| CoreError::internal("Failed to retrieve created email"))
352 }
353
354 #[tracing::instrument(skip_all)]
355 async fn restore(&self, user_id: UserId, email: &Email) -> Result<()> {
356 // Durable content fields are round-tripped; account-linked and
357 // sync-transient state (email_account_id, imap_uid, source_folder,
358 // attachment_meta, draft_account_id) is intentionally omitted — those
359 // FK into email_accounts (not part of a backup) or are re-derived on the
360 // next IMAP sync. Preserving the original id + message_id makes a second
361 // restore a no-op.
362 let labels_json = serde_json::to_string(&email.labels).unwrap_or_else(|_| "[]".to_string());
363 sqlx::query(
364 "INSERT OR IGNORE INTO emails (id, user_id, project_id, from_address, to_address, subject, body, html_body, is_read, is_archived, received_at, message_id, in_reply_to, thread_id, is_outgoing, labels, is_draft, cc_address, bcc_address, snoozed_until, waiting_for_response, waiting_since, expected_response_date) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
365 )
366 .bind(email.id.to_string())
367 .bind(user_id.to_string())
368 .bind(email.project_id.map(|p| p.to_string()))
369 .bind(&email.from)
370 .bind(&email.to)
371 .bind(&email.subject)
372 .bind(&email.body)
373 .bind(&email.html_body)
374 .bind(if email.is_read { 1 } else { 0 })
375 .bind(if email.is_archived { 1 } else { 0 })
376 .bind(format_datetime(&email.received_at))
377 .bind(&email.message_id)
378 .bind(&email.in_reply_to)
379 .bind(&email.thread_id)
380 .bind(if email.is_outgoing { 1 } else { 0 })
381 .bind(&labels_json)
382 .bind(if email.is_draft { 1 } else { 0 })
383 .bind(&email.cc_address)
384 .bind(&email.bcc_address)
385 .bind(format_datetime_opt(email.snoozed_until))
386 .bind(if email.waiting_for_response { 1 } else { 0 })
387 .bind(format_datetime_opt(email.waiting_since))
388 .bind(format_datetime_opt(email.expected_response_date))
389 .execute(&self.pool)
390 .await
391 .map_err(CoreError::database)?;
392 Ok(())
393 }
394
395 #[tracing::instrument(skip_all)]
396 async fn create_with_tracking(&self, user_id: UserId, email: NewEmailWithTracking) -> Result<Email> {
397 let id = goingson_core::deterministic_email_id(email.message_id.as_deref());
398 let received_at = format_datetime(&email.received_at.unwrap_or_else(Utc::now));
399 sqlx::query("INSERT INTO emails (id, user_id, project_id, from_address, to_address, subject, body, html_body, is_read, is_archived, received_at, message_id, in_reply_to, thread_id, email_account_id, is_outgoing, imap_uid, source_folder, attachment_meta, body_truncated, jmap_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
400 .bind(id.to_string()).bind(user_id.to_string()).bind(email.project_id.map(|p| p.to_string()))
401 .bind(&email.from_address).bind(&email.to_address).bind(&email.subject).bind(&email.body).bind(&email.html_body)
402 .bind(if email.is_read { 1 } else { 0 }).bind(if email.is_archived { 1 } else { 0 }).bind(&received_at)
403 .bind(&email.message_id).bind(&email.in_reply_to).bind(&email.thread_id)
404 .bind(email.email_account_id.map(|a| a.to_string()))
405 .bind(if email.is_outgoing { 1 } else { 0 }).bind(email.imap_uid).bind(&email.source_folder)
406 .bind(&email.attachment_meta)
407 .bind(if email.body_truncated { 1 } else { 0 }).bind(&email.jmap_id)
408 .execute(&self.pool).await.map_err(CoreError::database)?;
409 self.get_by_id(id, user_id).await?.ok_or_else(|| CoreError::internal("Failed to retrieve created email"))
410 }
411
412 #[tracing::instrument(skip_all)]
413 async fn create_with_tracking_batch(&self, user_id: UserId, emails: Vec<NewEmailWithTracking>) -> Result<usize> {
414 if emails.is_empty() {
415 return Ok(0);
416 }
417
418 let mut count = 0usize;
419 let uid = user_id.to_string();
420
421 let mut tx = self.pool.begin().await.map_err(CoreError::database)?;
422
423 for email in emails {
424 let id = goingson_core::deterministic_email_id(email.message_id.as_deref());
425 let received_at = format_datetime(&email.received_at.unwrap_or_else(Utc::now));
426 let result = sqlx::query("INSERT OR IGNORE INTO emails (id, user_id, project_id, from_address, to_address, subject, body, html_body, is_read, is_archived, received_at, message_id, in_reply_to, thread_id, email_account_id, is_outgoing, imap_uid, source_folder, attachment_meta, body_truncated, jmap_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
427 .bind(id.to_string()).bind(&uid).bind(email.project_id.map(|p| p.to_string()))
428 .bind(&email.from_address).bind(&email.to_address).bind(&email.subject).bind(&email.body).bind(&email.html_body)
429 .bind(if email.is_read { 1 } else { 0 }).bind(if email.is_archived { 1 } else { 0 }).bind(&received_at)
430 .bind(&email.message_id).bind(&email.in_reply_to).bind(&email.thread_id)
431 .bind(email.email_account_id.map(|a| a.to_string()))
432 .bind(if email.is_outgoing { 1 } else { 0 }).bind(email.imap_uid).bind(&email.source_folder)
433 .bind(&email.attachment_meta)
434 .bind(if email.body_truncated { 1 } else { 0 }).bind(&email.jmap_id)
435 .execute(&mut *tx).await.map_err(CoreError::database)?;
436 if result.rows_affected() > 0 {
437 count += 1;
438 }
439 }
440
441 tx.commit().await.map_err(CoreError::database)?;
442 Ok(count)
443 }
444
445 #[tracing::instrument(skip_all)]
446 async fn delete(&self, id: EmailId, user_id: UserId) -> Result<bool> {
447 let result = sqlx::query("DELETE FROM emails WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
448 Ok(result.rows_affected() > 0)
449 }
450
451 #[tracing::instrument(skip_all)]
452 async fn mark_read(&self, id: EmailId, user_id: UserId) -> Result<bool> {
453 let result = sqlx::query("UPDATE emails SET is_read = 1 WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
454 Ok(result.rows_affected() > 0)
455 }
456
457 #[tracing::instrument(skip_all)]
458 async fn mark_unread(&self, id: EmailId, user_id: UserId) -> Result<bool> {
459 let result = sqlx::query("UPDATE emails SET is_read = 0 WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
460 Ok(result.rows_affected() > 0)
461 }
462
463 #[tracing::instrument(skip_all)]
464 async fn archive(&self, id: EmailId, user_id: UserId) -> Result<bool> {
465 let result = sqlx::query("UPDATE emails SET is_archived = 1 WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
466 Ok(result.rows_affected() > 0)
467 }
468
469 #[tracing::instrument(skip_all)]
470 async fn unarchive(&self, id: EmailId, user_id: UserId) -> Result<bool> {
471 let result = sqlx::query("UPDATE emails SET is_archived = 0 WHERE id = ? AND user_id = ?").bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
472 Ok(result.rows_affected() > 0)
473 }
474
475 #[tracing::instrument(skip_all)]
476 async fn update_source_folder(&self, id: EmailId, user_id: UserId, new_folder: &str) -> Result<bool> {
477 let result = sqlx::query("UPDATE emails SET source_folder = ? WHERE id = ? AND user_id = ?").bind(new_folder).bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
478 Ok(result.rows_affected() > 0)
479 }
480
481 #[tracing::instrument(skip_all)]
482 async fn set_full_body(&self, id: EmailId, user_id: UserId, body: &str) -> Result<()> {
483 sqlx::query("UPDATE emails SET body = ?, body_truncated = 0 WHERE id = ? AND user_id = ?")
484 .bind(body).bind(id.to_string()).bind(user_id.to_string())
485 .execute(&self.pool).await.map_err(CoreError::database)?;
486 Ok(())
487 }
488
489 #[tracing::instrument(skip_all)]
490 async fn mark_all_read(&self, user_id: UserId) -> Result<u64> {
491 let result = sqlx::query("UPDATE emails SET is_read = 1 WHERE user_id = ? AND is_read = 0").bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
492 Ok(result.rows_affected())
493 }
494
495 #[tracing::instrument(skip_all)]
496 async fn link_to_project(&self, id: EmailId, user_id: UserId, project_id: Option<ProjectId>) -> Result<bool> {
497 let result = sqlx::query("UPDATE emails SET project_id = ? WHERE id = ? AND user_id = ?").bind(project_id.map(|p| p.to_string())).bind(id.to_string()).bind(user_id.to_string()).execute(&self.pool).await.map_err(CoreError::database)?;
498 Ok(result.rows_affected() > 0)
499 }
500
501 #[tracing::instrument(skip_all)]
502 async fn count_unread(&self, user_id: UserId) -> Result<i64> {
503 let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM emails WHERE user_id = ? AND is_read = 0").bind(user_id.to_string()).fetch_one(&self.pool).await.map_err(CoreError::database)?;
504 Ok(row.0)
505 }
506
507 #[tracing::instrument(skip_all)]
508 async fn exists_by_message_id(&self, user_id: UserId, message_id: &str) -> Result<bool> {
509 let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM emails WHERE user_id = ? AND message_id = ?").bind(user_id.to_string()).bind(message_id).fetch_one(&self.pool).await.map_err(CoreError::database)?;
510 Ok(row.0 > 0)
511 }
512
513 #[tracing::instrument(skip_all)]
514 async fn exists_by_message_ids(&self, user_id: UserId, message_ids: &[&str]) -> Result<HashSet<String>> {
515 if message_ids.is_empty() {
516 return Ok(HashSet::new());
517 }
518
519 let placeholders = bind_placeholders(message_ids.len());
520 let query = format!(
521 "SELECT message_id FROM emails WHERE user_id = ? AND message_id IN ({})",
522 placeholders
523 );
524
525 let mut q = sqlx::query_as::<_, (String,)>(&query).bind(user_id.to_string());
526 for msg_id in message_ids {
527 q = q.bind(*msg_id);
528 }
529
530 let rows = q.fetch_all(&self.pool).await
531 .map_err(CoreError::database)?;
532
533 Ok(rows.into_iter().map(|(id,)| id).collect())
534 }
535
536 #[tracing::instrument(skip_all)]
537 async fn exists_as_senders(&self, user_id: UserId, addresses: &[&str]) -> Result<HashSet<String>> {
538 if addresses.is_empty() {
539 return Ok(HashSet::new());
540 }
541
542 let placeholders = bind_placeholders(addresses.len());
543 let query = format!(
544 "SELECT DISTINCT LOWER(from_address) FROM emails WHERE user_id = ? AND LOWER(from_address) IN ({})",
545 placeholders
546 );
547
548 let mut q = sqlx::query_as::<_, (String,)>(&query).bind(user_id.to_string());
549 for addr in addresses {
550 q = q.bind(addr.to_lowercase());
551 }
552
553 let rows = q.fetch_all(&self.pool).await
554 .map_err(CoreError::database)?;
555
556 Ok(rows.into_iter().map(|(a,)| a).collect())
557 }
558
559 #[tracing::instrument(skip_all)]
560 async fn snooze(&self, id: EmailId, user_id: UserId, until: DateTime<Utc>) -> Result<Option<Email>> {
561 let until_str = format_datetime(&until);
562 let result = sqlx::query("UPDATE emails SET snoozed_until = ? WHERE id = ? AND user_id = ?")
563 .bind(&until_str)
564 .bind(id.to_string())
565 .bind(user_id.to_string())
566 .execute(&self.pool)
567 .await
568 .map_err(CoreError::database)?;
569 if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
570 }
571
572 #[tracing::instrument(skip_all)]
573 async fn unsnooze(&self, id: EmailId, user_id: UserId) -> Result<Option<Email>> {
574 let result = sqlx::query("UPDATE emails SET snoozed_until = NULL WHERE id = ? AND user_id = ?")
575 .bind(id.to_string())
576 .bind(user_id.to_string())
577 .execute(&self.pool)
578 .await
579 .map_err(CoreError::database)?;
580 if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
581 }
582
583 #[tracing::instrument(skip_all)]
584 async fn list_snoozed(&self, user_id: UserId) -> Result<Vec<Email>> {
585 let query = format!(
586 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.snoozed_until IS NOT NULL AND datetime(e.snoozed_until) > datetime('now') ORDER BY e.snoozed_until ASC",
587 EMAIL_SELECT_COLUMNS
588 );
589 let rows = sqlx::query_as::<_, EmailRow>(&query)
590 .bind(user_id.to_string())
591 .bind(user_id.to_string())
592 .fetch_all(&self.pool)
593 .await
594 .map_err(CoreError::database)?;
595 rows.into_iter().map(Email::try_from).collect()
596 }
597
598 #[tracing::instrument(skip_all)]
599 async fn mark_waiting(&self, id: EmailId, user_id: UserId, expected_response: Option<DateTime<Utc>>) -> Result<Option<Email>> {
600 let now = format_datetime_now();
601 let expected = format_datetime_opt(expected_response);
602
603 let result = sqlx::query(
604 "UPDATE emails SET waiting_for_response = 1, waiting_since = ?, expected_response_date = ? WHERE id = ? AND user_id = ?"
605 )
606 .bind(&now)
607 .bind(&expected)
608 .bind(id.to_string())
609 .bind(user_id.to_string())
610 .execute(&self.pool)
611 .await
612 .map_err(CoreError::database)?;
613
614 if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
615 }
616
617 #[tracing::instrument(skip_all)]
618 async fn clear_waiting(&self, id: EmailId, user_id: UserId) -> Result<Option<Email>> {
619 let result = sqlx::query(
620 "UPDATE emails SET waiting_for_response = 0, waiting_since = NULL, expected_response_date = NULL WHERE id = ? AND user_id = ?"
621 )
622 .bind(id.to_string())
623 .bind(user_id.to_string())
624 .execute(&self.pool)
625 .await
626 .map_err(CoreError::database)?;
627
628 if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
629 }
630
631 #[tracing::instrument(skip_all)]
632 async fn list_waiting(&self, user_id: UserId) -> Result<Vec<Email>> {
633 let query = format!(
634 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.waiting_for_response = 1 ORDER BY e.expected_response_date ASC",
635 EMAIL_SELECT_COLUMNS
636 );
637 let rows = sqlx::query_as::<_, EmailRow>(&query)
638 .bind(user_id.to_string())
639 .bind(user_id.to_string())
640 .fetch_all(&self.pool)
641 .await
642 .map_err(CoreError::database)?;
643 rows.into_iter().map(Email::try_from).collect()
644 }
645
646 #[tracing::instrument(skip_all)]
647 async fn list_by_thread(&self, user_id: UserId, thread_id: &str) -> Result<Vec<Email>> {
648 let query = format!(
649 "SELECT {} 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",
650 EMAIL_SELECT_COLUMNS
651 );
652 let rows = sqlx::query_as::<_, EmailRow>(&query)
653 .bind(user_id.to_string())
654 .bind(user_id.to_string())
655 .bind(thread_id)
656 .fetch_all(&self.pool)
657 .await
658 .map_err(CoreError::database)?;
659 rows.into_iter().map(Email::try_from).collect()
660 }
661
662 #[tracing::instrument(skip_all)]
663 async fn list_drafts(&self, user_id: UserId) -> Result<Vec<Email>> {
664 let query = format!(
665 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.is_draft = 1 ORDER BY e.received_at DESC",
666 EMAIL_SELECT_COLUMNS
667 );
668 let rows = sqlx::query_as::<_, EmailRow>(&query)
669 .bind(user_id.to_string())
670 .bind(user_id.to_string())
671 .fetch_all(&self.pool)
672 .await
673 .map_err(CoreError::database)?;
674 rows.into_iter().map(Email::try_from).collect()
675 }
676
677 #[tracing::instrument(skip_all)]
678 async fn save_draft(&self, id: EmailId, user_id: UserId, from: &str, to: &str, cc: Option<&str>, bcc: Option<&str>, subject: &str, body: &str, account_id: Option<EmailAccountId>, in_reply_to: Option<&str>, _references: Option<&str>, thread_id: Option<&str>) -> Result<Email> {
679 let now = format_datetime_now();
680 let account_id_str = account_id.map(|a: EmailAccountId| a.to_string());
681
682 // Upsert: update if exists, insert if not
683 sqlx::query(r#"
684 INSERT INTO emails (id, user_id, from_address, to_address, cc_address, bcc_address, subject, body,
685 is_read, is_archived, is_draft, is_outgoing, received_at, draft_account_id, in_reply_to, thread_id)
686 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, 0, 1, 1, ?, ?, ?, ?)
687 ON CONFLICT(id) DO UPDATE SET
688 from_address = excluded.from_address,
689 to_address = excluded.to_address,
690 cc_address = excluded.cc_address,
691 bcc_address = excluded.bcc_address,
692 subject = excluded.subject,
693 body = excluded.body,
694 received_at = excluded.received_at,
695 draft_account_id = excluded.draft_account_id,
696 in_reply_to = excluded.in_reply_to,
697 thread_id = excluded.thread_id
698 "#)
699 .bind(id.to_string())
700 .bind(user_id.to_string())
701 .bind(from)
702 .bind(to)
703 .bind(cc)
704 .bind(bcc)
705 .bind(subject)
706 .bind(body)
707 .bind(&now)
708 .bind(&account_id_str)
709 .bind(in_reply_to)
710 .bind(thread_id)
711 .execute(&self.pool)
712 .await
713 .map_err(CoreError::database)?;
714
715 self.get_by_id(id, user_id).await?.ok_or_else(|| CoreError::internal("Failed to retrieve saved draft"))
716 }
717
718 #[tracing::instrument(skip_all)]
719 async fn get_by_message_id(&self, user_id: UserId, message_id: &str) -> Result<Option<Email>> {
720 let query = format!(
721 "SELECT {} FROM emails e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? WHERE e.user_id = ? AND e.message_id = ?",
722 EMAIL_SELECT_COLUMNS
723 );
724 let row = sqlx::query_as::<_, EmailRow>(&query)
725 .bind(user_id.to_string())
726 .bind(user_id.to_string())
727 .bind(message_id)
728 .fetch_optional(&self.pool)
729 .await
730 .map_err(CoreError::database)?;
731 row.map(Email::try_from).transpose()
732 }
733
734 #[tracing::instrument(skip_all)]
735 async fn update_labels(&self, id: EmailId, user_id: UserId, labels: &[String]) -> Result<Option<Email>> {
736 let labels_json = serde_json::to_string(labels).unwrap_or_else(|_| "[]".to_string());
737 let result = sqlx::query("UPDATE emails SET labels = ? WHERE id = ? AND user_id = ?")
738 .bind(&labels_json)
739 .bind(id.to_string())
740 .bind(user_id.to_string())
741 .execute(&self.pool)
742 .await
743 .map_err(CoreError::database)?;
744 if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
745 }
746
747 #[tracing::instrument(skip_all)]
748 async fn list_folders(&self, user_id: UserId) -> Result<Vec<String>> {
749 let rows: Vec<(String,)> = sqlx::query_as(
750 "SELECT DISTINCT source_folder FROM emails WHERE user_id = ? AND source_folder IS NOT NULL AND is_draft = 0 ORDER BY source_folder ASC"
751 )
752 .bind(user_id.to_string())
753 .fetch_all(&self.pool)
754 .await
755 .map_err(CoreError::database)?;
756 Ok(rows.into_iter().map(|r| r.0).collect())
757 }
758
759 #[tracing::instrument(skip_all)]
760 async fn list_labels(&self, user_id: UserId) -> Result<Vec<String>> {
761 // Extract all unique labels across all emails via JSON parsing
762 let rows: Vec<(String,)> = sqlx::query_as(
763 "SELECT DISTINCT j.value FROM emails e, json_each(e.labels) j WHERE e.user_id = ? AND e.is_draft = 0 ORDER BY j.value ASC"
764 )
765 .bind(user_id.to_string())
766 .fetch_all(&self.pool)
767 .await
768 .map_err(CoreError::database)?;
769 Ok(rows.into_iter().map(|r| r.0).collect())
770 }
771 }
772