Skip to main content

max / goingson

17.5 KB · 477 lines History Blame Raw
1 //! Email synchronization commands.
2 //!
3 //! Provides IMAP and JMAP email sync, using `goingson_core::email_sync` for the
4 //! shared dedup-save-clear-waiting loop.
5
6 use serde::Serialize;
7 use sha2::{Sha256, Digest};
8 use std::path::Path;
9 use std::sync::Arc;
10 use tauri::State;
11 use tracing::{instrument, warn};
12 use uuid::Uuid;
13
14 use goingson_core::email_sync::{FetchedEmail, process_fetched_emails};
15 use goingson_core::{AttachmentMeta, EmailAccount, EmailAccountId};
16
17 use crate::email::{AttachmentPart, ParsedEmail};
18
19 use crate::email::{uses_jmap, ImapClient};
20 use crate::jmap::JmapClient;
21 use crate::oauth::{CredentialStore, TokenManager};
22 use crate::state::{AppState, DESKTOP_USER_ID};
23 use super::{ApiError, OptionApiError, OptionNotFound, ResultApiError};
24 use super::email_account::{uses_oauth_imap, get_account_password, get_valid_access_token};
25
26 // ============ Types ============
27
28 #[derive(Debug, Serialize)]
29 #[serde(rename_all = "camelCase")]
30 pub struct SyncResponse {
31 pub emails_fetched: usize,
32 pub emails_saved: usize,
33 pub inbox_fetched: usize,
34 pub archive_fetched: usize,
35 pub message: String,
36 pub debug_info: Option<String>,
37 }
38
39 // ============ Commands ============
40
41 /// Inner sync function callable from both the command and the scheduler.
42 #[instrument(skip_all)]
43 pub async fn sync_email_account_inner(
44 state: &Arc<AppState>,
45 id: EmailAccountId,
46 full_sync: Option<bool>,
47 ) -> Result<SyncResponse, ApiError> {
48 // Acquire per-account lock to prevent concurrent syncs on the same account
49 if !state.email_sync_locks.lock().unwrap_or_else(|e| e.into_inner()).insert(id) {
50 return Err(ApiError::bad_request("Sync already in progress for this account"));
51 }
52
53 let result = async {
54 let account = state.email_accounts
55 .get_by_id(id, DESKTOP_USER_ID)
56 .await?
57 .or_not_found("emailAccount", id)?;
58
59 let result = if uses_jmap(&account) {
60 sync_jmap_account_inner(state, &account, id, full_sync).await?
61 } else {
62 sync_imap_account_inner(state, &account, id, full_sync).await?
63 };
64
65 state.email_accounts.update_last_sync(id, DESKTOP_USER_ID).await?;
66
67 Ok(result)
68 }.await;
69
70 // Always release the lock. The lock is not held across the await, so it cannot be poisoned
71 // by an async-block panic; unwrap_or_else handles the unreachable poison case defensively.
72 state.email_sync_locks
73 .lock()
74 .unwrap_or_else(|p| p.into_inner())
75 .remove(&id);
76
77 result
78 }
79
80 /// Synchronizes emails for an account.
81 #[tauri::command]
82 #[instrument(skip_all)]
83 pub async fn sync_email_account(state: State<'_, Arc<AppState>>, id: EmailAccountId, full_sync: Option<bool>) -> Result<SyncResponse, ApiError> {
84 sync_email_account_inner(&state, id, full_sync).await
85 }
86
87 /// Syncs an IMAP email account.
88 ///
89 /// Fetches from INBOX and Archive sequentially (single IMAP connection), then
90 /// processes both folders concurrently — hashing attachments, saving to DB, and
91 /// writing blobs in parallel for each folder.
92 #[instrument(skip_all)]
93 async fn sync_imap_account_inner(
94 state: &Arc<AppState>,
95 account: &EmailAccount,
96 id: EmailAccountId,
97 full_sync: Option<bool>,
98 ) -> Result<SyncResponse, ApiError> {
99 let imap_client = if uses_oauth_imap(account) {
100 let access_token = get_valid_access_token(state, account).await?;
101 ImapClient::with_oauth(
102 &account.imap_server,
103 account.imap_port as u16,
104 &account.email_address,
105 &access_token,
106 )
107 } else {
108 let password = get_account_password(account)?;
109 ImapClient::with_password(account, &password)
110 };
111
112 let archive_folder = account.archive_folder_name.clone().unwrap_or_else(|| "Archive".to_string());
113 let is_full_sync = full_sync.unwrap_or(false);
114
115 let mut debug_parts = Vec::new();
116 let since = if is_full_sync { None } else { account.last_sync_at };
117 debug_parts.push(format!("since filter: {:?}, full_sync: {:?}", since.map(|d| d.to_rfc3339()), full_sync));
118
119 let blobs_dir = state.data_dir.join("blobs");
120
121 // --- Fetch sync states ---
122 let inbox_sync_state = if is_full_sync {
123 None
124 } else {
125 state.email_accounts.get_folder_sync_state(id, "INBOX").await.ok().flatten()
126 };
127
128 let archive_sync_state = if is_full_sync {
129 None
130 } else {
131 state.email_accounts.get_folder_sync_state(id, &archive_folder).await.ok().flatten()
132 };
133
134 // --- IMAP fetches (sequential, single connection) ---
135 let inbox_result = imap_client
136 .fetch_emails_incremental("INBOX", inbox_sync_state.as_ref(), since)
137 .await
138 .map_api_err("Failed to fetch INBOX", ApiError::external_service)?;
139
140 let inbox_fetched = inbox_result.emails.len();
141 debug_parts.push(format!("INBOX: {}", inbox_result.debug_info));
142
143 let archive_result = match imap_client
144 .fetch_emails_incremental(&archive_folder, archive_sync_state.as_ref(), since)
145 .await
146 {
147 Ok(result) => {
148 debug_parts.push(format!("Archive: {}", result.debug_info));
149 Some(result)
150 }
151 Err(e) => {
152 warn!("Could not sync archive folder '{}': {}", archive_folder, e);
153 None
154 }
155 };
156
157 let archive_fetched = archive_result.as_ref().map_or(0, |r| r.emails.len());
158 let emails_fetched = inbox_fetched + archive_fetched;
159
160 // Extract sync metadata before consuming emails
161 let archive_uid_validity = archive_result.as_ref().and_then(|r| r.uid_validity);
162 let archive_max_uid = archive_result.as_ref().and_then(|r| r.max_uid_fetched);
163 let archive_emails = archive_result.map(|r| r.emails).unwrap_or_default();
164
165 // --- Process both folders concurrently (hash + DB save + blob write) ---
166 let inbox_process = process_folder_emails(state, &blobs_dir, id, inbox_result.emails, false);
167 let archive_process = process_folder_emails(state, &blobs_dir, id, archive_emails, true);
168
169 let (inbox_saved, archive_saved) = tokio::try_join!(inbox_process, archive_process)?;
170 let total_saved = inbox_saved + archive_saved;
171
172 // --- Update sync states ---
173 if let Some(validity) = inbox_result.uid_validity {
174 if let Some(ref prev) = inbox_sync_state
175 && prev.uid_validity != validity {
176 state.email_accounts.delete_folder_sync_state(id, "INBOX").await.ok();
177 }
178 let new_max = inbox_result.max_uid_fetched
179 .unwrap_or_else(|| inbox_sync_state.as_ref().map_or(0, |s| s.last_seen_uid));
180 if new_max > 0 {
181 state.email_accounts.upsert_folder_sync_state(id, "INBOX", validity, new_max).await.ok();
182 }
183 }
184
185 if let Some(validity) = archive_uid_validity {
186 if let Some(ref prev) = archive_sync_state
187 && prev.uid_validity != validity {
188 state.email_accounts.delete_folder_sync_state(id, &archive_folder).await.ok();
189 }
190 let new_max = archive_max_uid
191 .unwrap_or_else(|| archive_sync_state.as_ref().map_or(0, |s| s.last_seen_uid));
192 if new_max > 0 {
193 state.email_accounts.upsert_folder_sync_state(id, &archive_folder, validity, new_max).await.ok();
194 }
195 }
196
197 Ok(SyncResponse {
198 emails_fetched,
199 emails_saved: total_saved,
200 inbox_fetched,
201 archive_fetched,
202 message: format!("Found {} in INBOX, {} in Archive. Saved {} new emails.", inbox_fetched, archive_fetched, total_saved),
203 debug_info: Some(debug_parts.join(" | ")),
204 })
205 }
206
207 /// Process fetched emails for a single folder: hash attachments, save to DB,
208 /// then write blobs to disk. Blob I/O is deferred until after the DB save so
209 /// large attachments don't block the sync pipeline.
210 async fn process_folder_emails(
211 state: &Arc<AppState>,
212 blobs_dir: &Path,
213 account_id: EmailAccountId,
214 emails: Vec<ParsedEmail>,
215 is_archived: bool,
216 ) -> Result<usize, ApiError> {
217 if emails.is_empty() {
218 return Ok(0);
219 }
220
221 // Phase 1: Hash attachments and build FetchedEmail structs (CPU-bound, no disk I/O)
222 let (fetched_emails, pending_blobs) = tokio::task::spawn_blocking(move || {
223 let mut all_pending: Vec<(String, Vec<u8>)> = Vec::new();
224 let fetched: Vec<FetchedEmail> = emails.into_iter().map(|p| {
225 let (attachment_meta, pending) = build_attachment_meta_deferred(p.attachments);
226 all_pending.extend(pending);
227 FetchedEmail {
228 message_id: p.message_id,
229 in_reply_to: p.in_reply_to,
230 references_root: p.references_root,
231 from: p.from,
232 to: p.to,
233 subject: p.subject,
234 body: p.body,
235 html_body: p.html_body,
236 is_read: p.is_read,
237 date: p.date,
238 source_folder: p.source_folder,
239 imap_uid: Some(p.imap_uid as i64),
240 is_archived,
241 attachment_meta,
242 body_truncated: false,
243 jmap_id: None,
244 }
245 }).collect();
246 (fetched, all_pending)
247 }).await.map_err(|e| ApiError::internal(format!("Attachment processing failed: {e}")))?;
248
249 // Phase 2: Save to DB immediately (metadata is ready)
250 let save_result = process_fetched_emails(
251 state.emails.as_ref(), DESKTOP_USER_ID, account_id, fetched_emails,
252 ).await?;
253
254 // Phase 3: Write blobs to disk (deferred — large file I/O after DB commit)
255 if !pending_blobs.is_empty() {
256 let blobs_dir = blobs_dir.to_path_buf();
257 tokio::task::spawn_blocking(move || {
258 write_pending_blobs(&blobs_dir, pending_blobs);
259 }).await.ok();
260 }
261
262 Ok(save_result.emails_saved)
263 }
264
265 /// Builds a JMAP client for an account, resolving and (if needed) refreshing the
266 /// OAuth access token. Shared by the sync path and on-demand fetches (e.g. lazy
267 /// full-body load) so token handling lives in exactly one place.
268 #[instrument(skip_all)]
269 pub(crate) async fn build_jmap_client(
270 state: &Arc<AppState>,
271 account: &EmailAccount,
272 ) -> Result<JmapClient, ApiError> {
273 let session_url = account.jmap_session_url.as_ref()
274 .or_api_err(|| ApiError::bad_request("No JMAP session URL configured"))?;
275 let id = account.id;
276
277 // Token resolution order: (1) in-memory keychain cache, (2) DB-stored token,
278 // (3) account field. The keychain is preferred because token refreshes update
279 // it first, and the DB may lag behind during the same session.
280 let raw_id: Uuid = account.id.into();
281 let mut access_token = CredentialStore::get_oauth(raw_id)
282 .map(|c| c.access_token)
283 .or_else(|| account.oauth2_access_token.clone())
284 .or_api_err(|| ApiError::auth("No access token available"))?;
285
286 // Check if token needs refresh (serialized per-account to prevent thundering herd)
287 if account.needs_token_refresh() {
288 let refresh_lock = state.token_refresh_lock(raw_id);
289 let _guard = refresh_lock.lock().await;
290 // Re-read the account after acquiring the lock (GO-8): a prior waiter may
291 // have refreshed and persisted a new token while we waited, leaving the
292 // in-memory `account` and its expiry stale. Passing the fresh copy lets
293 // refresh_if_needed correctly no-op instead of firing a redundant refresh.
294 let fresh = state.email_accounts
295 .get_by_id(id, DESKTOP_USER_ID)
296 .await?
297 .or_api_err(|| ApiError::auth("Email account no longer exists"))?;
298 let token_manager = TokenManager::from_env();
299 match token_manager.refresh_if_needed(&fresh).await {
300 Ok(Some((new_token, new_refresh, expires_at))) => {
301 state.email_accounts
302 .update_oauth_tokens(id, DESKTOP_USER_ID, &new_token, new_refresh.as_deref(), expires_at)
303 .await?;
304 let _ = CredentialStore::update_oauth_tokens(raw_id, &new_token, new_refresh.as_deref());
305 access_token = new_token;
306 }
307 // Already fresh (a prior waiter refreshed): adopt the latest token,
308 // keychain first since refreshes write there before the DB.
309 Ok(None) => {
310 if let Some(c) = CredentialStore::get_oauth(raw_id) {
311 access_token = c.access_token;
312 } else if let Some(tok) = fresh.oauth2_access_token.clone() {
313 access_token = tok;
314 }
315 }
316 Err(_) => {}
317 }
318 }
319
320 JmapClient::new(session_url, &access_token).map_err(ApiError::external_service)
321 }
322
323 /// Syncs a JMAP email account.
324 #[instrument(skip_all)]
325 async fn sync_jmap_account_inner(
326 state: &Arc<AppState>,
327 account: &EmailAccount,
328 id: EmailAccountId,
329 full_sync: Option<bool>,
330 ) -> Result<SyncResponse, ApiError> {
331 let mut client = build_jmap_client(state, account).await?;
332 let mut debug_parts = Vec::new();
333
334 let since = if full_sync.unwrap_or(false) { None } else { account.last_sync_at };
335 debug_parts.push(format!("JMAP sync, since: {:?}, full_sync: {:?}", since.map(|d| d.to_rfc3339()), full_sync));
336
337 let limit = if full_sync.unwrap_or(false) { 100 } else { 50 };
338
339 let mut emails_fetched = 0;
340 let mut total_saved = 0;
341
342 // Sync inbox
343 let inbox_emails = client.fetch_inbox(since, limit).await
344 .map_api_err("Failed to fetch JMAP inbox", ApiError::external_service)?;
345
346 let inbox_fetched = inbox_emails.len();
347 emails_fetched += inbox_emails.len();
348 debug_parts.push(format!("JMAP Inbox: {} emails", inbox_fetched));
349
350 // JMAP returns plain-text body only — html_body is not yet extracted from
351 // the JMAP Email/get response (would require requesting bodyValues with
352 // text/html type). IMAP UIDs don't apply to JMAP (uses opaque blob IDs).
353 let fetched_inbox: Vec<FetchedEmail> = inbox_emails.into_iter().map(|p| FetchedEmail {
354 message_id: p.message_id,
355 in_reply_to: p.in_reply_to,
356 references_root: p.references_root,
357 from: p.from,
358 to: p.to,
359 subject: p.subject,
360 body: p.body,
361 html_body: None,
362 is_read: p.is_read,
363 date: p.date,
364 source_folder: p.source_folder,
365 imap_uid: None,
366 is_archived: false,
367 attachment_meta: None,
368 body_truncated: p.body_truncated,
369 jmap_id: Some(p.jmap_id),
370 }).collect();
371
372 let inbox_result = process_fetched_emails(
373 state.emails.as_ref(), DESKTOP_USER_ID, id, fetched_inbox,
374 ).await?;
375 total_saved += inbox_result.emails_saved;
376
377 // Sync archive
378 let archive_emails = match client.fetch_archive(since, limit).await {
379 Ok(emails) => emails,
380 Err(e) => {
381 debug_parts.push(format!("Archive error: {}", e));
382 Vec::new()
383 }
384 };
385
386 let archive_fetched = archive_emails.len();
387 emails_fetched += archive_emails.len();
388 debug_parts.push(format!("JMAP Archive: {} emails", archive_fetched));
389
390 let fetched_archive: Vec<FetchedEmail> = archive_emails.into_iter().map(|p| FetchedEmail {
391 message_id: p.message_id,
392 in_reply_to: p.in_reply_to,
393 references_root: p.references_root,
394 from: p.from,
395 to: p.to,
396 subject: p.subject,
397 body: p.body,
398 html_body: None,
399 is_read: true,
400 date: p.date,
401 source_folder: p.source_folder,
402 imap_uid: None,
403 is_archived: true,
404 attachment_meta: None,
405 body_truncated: p.body_truncated,
406 jmap_id: Some(p.jmap_id),
407 }).collect();
408
409 let archive_result = process_fetched_emails(
410 state.emails.as_ref(), DESKTOP_USER_ID, id, fetched_archive,
411 ).await?;
412 total_saved += archive_result.emails_saved;
413
414 Ok(SyncResponse {
415 emails_fetched,
416 emails_saved: total_saved,
417 inbox_fetched,
418 archive_fetched,
419 message: format!("JMAP: Found {} in Inbox, {} in Archive. Saved {} new emails.", inbox_fetched, archive_fetched, total_saved),
420 debug_info: Some(debug_parts.join(" | ")),
421 })
422 }
423
424 // ============ Attachment Helpers ============
425
426 /// Hash attachments and build metadata JSON without writing blobs to disk.
427 ///
428 /// Returns (metadata JSON string, pending blob writes as (hash, data) pairs).
429 /// The caller is responsible for writing blobs via [`write_pending_blobs`].
430 fn build_attachment_meta_deferred(
431 attachments: Vec<AttachmentPart>,
432 ) -> (Option<String>, Vec<(String, Vec<u8>)>) {
433 if attachments.is_empty() {
434 return (None, Vec::new());
435 }
436
437 let mut pending_writes = Vec::new();
438 let metas: Vec<AttachmentMeta> = attachments.into_iter().map(|part| {
439 let hash = {
440 let mut hasher = Sha256::new();
441 hasher.update(&part.data);
442 format!("{:x}", hasher.finalize())
443 };
444
445 let size = part.data.len();
446 pending_writes.push((hash.clone(), part.data));
447
448 AttachmentMeta {
449 filename: part.filename,
450 mime_type: part.mime_type,
451 size,
452 blob_hash: hash,
453 }
454 }).collect();
455
456 if metas.is_empty() {
457 return (None, Vec::new());
458 }
459
460 (serde_json::to_string(&metas).ok(), pending_writes)
461 }
462
463 /// Write pending blob data to disk with content-addressed dedup.
464 fn write_pending_blobs(blobs_dir: &Path, pending: Vec<(String, Vec<u8>)>) {
465 if let Err(e) = std::fs::create_dir_all(blobs_dir) {
466 warn!("Failed to create blobs dir: {}", e);
467 return;
468 }
469 for (hash, data) in pending {
470 let blob_path = blobs_dir.join(&hash);
471 if !blob_path.exists()
472 && let Err(e) = std::fs::write(&blob_path, &data) {
473 warn!(blob_hash = %hash, "Failed to write attachment blob: {}", e);
474 }
475 }
476 }
477