Skip to main content

max / goingson

7.5 KB · 212 lines History Blame Raw
1 //! Integration tests for `goingson_core::email_sync::process_fetched_emails`.
2 //!
3 //! Exercises the shared dedup-save-clear-waiting loop against the real
4 //! SqliteEmailRepository (in-memory), which is higher fidelity than a hand mock:
5 //! it validates the synthetic-message-id path, cross-call deduplication, the
6 //! thread_id fallback chain, and waiting-status clearing against real SQL.
7
8 mod common;
9
10 use chrono::{Duration, Utc};
11 use goingson_core::email_sync::{process_fetched_emails, FetchedEmail};
12 use goingson_core::{EmailAccountId, EmailAccountRepository, EmailRepository, UserId};
13 use goingson_db_sqlite::{SqliteEmailAccountRepository, SqliteEmailRepository};
14 use sqlx::SqlitePool;
15
16 async fn make_account(pool: &SqlitePool, user_id: UserId) -> EmailAccountId {
17 let repo = SqliteEmailAccountRepository::new(pool.clone());
18 let account = repo
19 .create(
20 user_id,
21 "Test",
22 "me@example.com",
23 "imap.example.com",
24 993,
25 "smtp.example.com",
26 465,
27 "me@example.com",
28 "pw",
29 true,
30 Some("Archive"),
31 )
32 .await
33 .expect("create account");
34 account.id
35 }
36
37 /// A minimal fetched email with only the fields a test cares about set.
38 fn fetched(message_id: Option<&str>, from: &str, subject: &str) -> FetchedEmail {
39 FetchedEmail {
40 message_id: message_id.map(str::to_string),
41 in_reply_to: None,
42 references_root: None,
43 from: from.to_string(),
44 to: "me@example.com".to_string(),
45 subject: subject.to_string(),
46 body: "body".to_string(),
47 html_body: None,
48 is_read: false,
49 date: Utc::now(),
50 source_folder: "INBOX".to_string(),
51 imap_uid: None,
52 is_archived: false,
53 attachment_meta: None,
54 body_truncated: false,
55 jmap_id: None,
56 }
57 }
58
59 #[tokio::test]
60 async fn saves_new_emails_and_dedups_on_second_run() {
61 let pool = common::setup_test_db().await;
62 let user_id = common::create_test_user(&pool).await;
63 let account_id = make_account(&pool, user_id).await;
64 let repo = SqliteEmailRepository::new(pool.clone());
65
66 let batch = vec![
67 fetched(Some("m-1@example.com"), "a@example.com", "One"),
68 fetched(Some("m-2@example.com"), "b@example.com", "Two"),
69 ];
70 let r = process_fetched_emails(&repo, user_id, account_id, batch)
71 .await
72 .expect("process");
73 assert_eq!(r.emails_saved, 2);
74 assert_eq!(r.waiting_cleared, 0);
75
76 // Re-processing the same message IDs must dedup — nothing new saved.
77 let again = vec![
78 fetched(Some("m-1@example.com"), "a@example.com", "One"),
79 fetched(Some("m-2@example.com"), "b@example.com", "Two"),
80 fetched(Some("m-3@example.com"), "c@example.com", "Three"),
81 ];
82 let r2 = process_fetched_emails(&repo, user_id, account_id, again)
83 .await
84 .expect("process");
85 assert_eq!(r2.emails_saved, 1, "only the genuinely-new message is saved");
86 assert_eq!(repo.list_all(user_id, true).await.unwrap().len(), 3);
87 }
88
89 #[tokio::test]
90 async fn synthesizes_message_id_and_dedups_it_across_runs() {
91 let pool = common::setup_test_db().await;
92 let user_id = common::create_test_user(&pool).await;
93 let account_id = make_account(&pool, user_id).await;
94 let repo = SqliteEmailRepository::new(pool.clone());
95
96 // No message_id: process_fetched_emails hashes from/to/subject/date into a
97 // stable "synth-..." id. Fix the date so the hash is identical across runs.
98 let date = Utc::now();
99 let mut e1 = fetched(None, "sender@example.com", "No Message-ID");
100 e1.date = date;
101 let r = process_fetched_emails(&repo, user_id, account_id, vec![e1])
102 .await
103 .expect("process");
104 assert_eq!(r.emails_saved, 1);
105
106 let saved = repo.list_all(user_id, true).await.unwrap();
107 assert_eq!(saved.len(), 1);
108 let synth = saved[0].message_id.clone().expect("synthetic id assigned");
109 assert!(synth.starts_with("synth-"), "got: {synth}");
110
111 // Same content + date -> same synthetic id -> deduped on the next run.
112 let mut e1_again = fetched(None, "sender@example.com", "No Message-ID");
113 e1_again.date = date;
114 let r2 = process_fetched_emails(&repo, user_id, account_id, vec![e1_again])
115 .await
116 .expect("process");
117 assert_eq!(r2.emails_saved, 0, "identical content dedups via synthetic id");
118 }
119
120 #[tokio::test]
121 async fn thread_id_follows_reply_then_falls_back_to_message_id() {
122 let pool = common::setup_test_db().await;
123 let user_id = common::create_test_user(&pool).await;
124 let account_id = make_account(&pool, user_id).await;
125 let repo = SqliteEmailRepository::new(pool.clone());
126
127 // A thread-root email: thread_id falls back to its own message_id.
128 let root = fetched(Some("root@example.com"), "a@example.com", "Root");
129 // A reply: thread_id should be the in_reply_to it points at.
130 let mut reply = fetched(Some("reply@example.com"), "b@example.com", "Re: Root");
131 reply.in_reply_to = Some("root@example.com".to_string());
132
133 process_fetched_emails(&repo, user_id, account_id, vec![root, reply])
134 .await
135 .expect("process");
136
137 let root_saved = repo
138 .get_by_message_id(user_id, "root@example.com")
139 .await
140 .unwrap()
141 .unwrap();
142 assert_eq!(root_saved.thread_id.as_deref(), Some("root@example.com"));
143
144 let reply_saved = repo
145 .get_by_message_id(user_id, "reply@example.com")
146 .await
147 .unwrap()
148 .unwrap();
149 assert_eq!(
150 reply_saved.thread_id.as_deref(),
151 Some("root@example.com"),
152 "a reply threads under the message it answers"
153 );
154 }
155
156 #[tokio::test]
157 async fn clears_waiting_status_when_a_reply_arrives() {
158 let pool = common::setup_test_db().await;
159 let user_id = common::create_test_user(&pool).await;
160 let account_id = make_account(&pool, user_id).await;
161 let repo = SqliteEmailRepository::new(pool.clone());
162
163 // Save an outgoing message we're waiting on a reply to.
164 process_fetched_emails(
165 &repo,
166 user_id,
167 account_id,
168 vec![fetched(Some("sent@example.com"), "me@example.com", "Question")],
169 )
170 .await
171 .expect("process");
172 let sent = repo
173 .get_by_message_id(user_id, "sent@example.com")
174 .await
175 .unwrap()
176 .unwrap();
177 repo.mark_waiting(sent.id, user_id, Some(Utc::now() + Duration::days(3)))
178 .await
179 .unwrap();
180
181 // A reply to it arrives: waiting status must clear.
182 let mut reply = fetched(Some("their-reply@example.com"), "them@example.com", "Re: Question");
183 reply.in_reply_to = Some("sent@example.com".to_string());
184 let r = process_fetched_emails(&repo, user_id, account_id, vec![reply])
185 .await
186 .expect("process");
187
188 assert_eq!(r.emails_saved, 1);
189 assert_eq!(r.waiting_cleared, 1);
190 let sent_after = repo
191 .get_by_message_id(user_id, "sent@example.com")
192 .await
193 .unwrap()
194 .unwrap();
195 assert!(!sent_after.waiting_for_response, "reply clears the waiting flag");
196 }
197
198 #[tokio::test]
199 async fn empty_batch_is_a_noop() {
200 let pool = common::setup_test_db().await;
201 let user_id = common::create_test_user(&pool).await;
202 let account_id = make_account(&pool, user_id).await;
203 let repo = SqliteEmailRepository::new(pool.clone());
204
205 let r = process_fetched_emails(&repo, user_id, account_id, vec![])
206 .await
207 .expect("process");
208 assert_eq!(r.emails_saved, 0);
209 assert_eq!(r.waiting_cleared, 0);
210 assert!(repo.list_all(user_id, true).await.unwrap().is_empty());
211 }
212