Skip to main content

max / goingson

37.4 KB · 1107 lines History Blame Raw
1 //! SQLite implementation of the ContactRepository.
2 //!
3 //! Manages contacts with sub-collections (emails, phones, social handles).
4 //! Sub-collections are stored in separate tables and batch-loaded for list operations.
5
6 use async_trait::async_trait;
7 use sqlx::SqlitePool;
8 use std::collections::{HashMap, HashSet};
9 use goingson_core::{
10 Contact, ContactCustomField, ContactEmail, ContactEmailEntry, ContactEmailId, ContactId, ContactPhone,
11 ContactPhoneId, ContactRepository, CoreError, CustomFieldId, NewContact,
12 NewContactCustomField, NewContactEmail, NewContactPhone, NewSocialHandle, Result,
13 SocialHandle, SocialHandleId, UpdateContact, UserId,
14 };
15
16 use crate::utils::{bind_placeholders, escape_like, format_datetime, format_datetime_now, parse_datetime, parse_tags, parse_uuid};
17
18 // ============ Row Structs ============
19
20 #[derive(Debug, Clone, sqlx::FromRow)]
21 struct ContactRow {
22 pub id: String,
23 pub display_name: String,
24 pub nickname: Option<String>,
25 pub company: Option<String>,
26 pub title: Option<String>,
27 pub notes: String,
28 pub tags: String,
29 pub birthday: Option<String>,
30 pub timezone: Option<String>,
31 pub external_source: Option<String>,
32 pub external_id: Option<String>,
33 pub is_implicit: i32,
34 pub created_at: String,
35 pub updated_at: String,
36 }
37
38 #[derive(Debug, Clone, sqlx::FromRow)]
39 struct ContactEmailRow {
40 pub id: String,
41 pub contact_id: String,
42 pub address: String,
43 pub label: String,
44 pub is_primary: i32,
45 }
46
47 #[derive(Debug, Clone, sqlx::FromRow)]
48 struct ContactPhoneRow {
49 pub id: String,
50 pub contact_id: String,
51 pub number: String,
52 pub label: String,
53 pub is_primary: i32,
54 }
55
56 #[derive(Debug, Clone, sqlx::FromRow)]
57 struct SocialHandleRow {
58 pub id: String,
59 pub contact_id: String,
60 pub platform: String,
61 pub handle: String,
62 pub url: Option<String>,
63 }
64
65 #[derive(Debug, Clone, sqlx::FromRow)]
66 struct CustomFieldRow {
67 pub id: String,
68 pub contact_id: String,
69 pub label: String,
70 pub value: String,
71 pub url: Option<String>,
72 }
73
74 // ============ Row Conversions ============
75
76 fn contact_email_from_row(row: ContactEmailRow) -> Result<ContactEmail> {
77 Ok(ContactEmail {
78 id: parse_uuid(&row.id)?.into(),
79 contact_id: parse_uuid(&row.contact_id)?.into(),
80 address: row.address,
81 label: row.label,
82 is_primary: row.is_primary != 0,
83 })
84 }
85
86 fn contact_phone_from_row(row: ContactPhoneRow) -> Result<ContactPhone> {
87 Ok(ContactPhone {
88 id: parse_uuid(&row.id)?.into(),
89 contact_id: parse_uuid(&row.contact_id)?.into(),
90 number: row.number,
91 label: row.label,
92 is_primary: row.is_primary != 0,
93 })
94 }
95
96 fn social_handle_from_row(row: SocialHandleRow) -> Result<SocialHandle> {
97 Ok(SocialHandle {
98 id: parse_uuid(&row.id)?.into(),
99 contact_id: parse_uuid(&row.contact_id)?.into(),
100 platform: row.platform,
101 handle: row.handle,
102 url: row.url,
103 })
104 }
105
106 fn custom_field_from_row(row: CustomFieldRow) -> Result<ContactCustomField> {
107 Ok(ContactCustomField {
108 id: parse_uuid(&row.id)?.into(),
109 contact_id: parse_uuid(&row.contact_id)?.into(),
110 label: row.label,
111 value: row.value,
112 url: row.url,
113 })
114 }
115
116 fn contact_from_row(
117 row: ContactRow,
118 emails: Vec<ContactEmail>,
119 phones: Vec<ContactPhone>,
120 social_handles: Vec<SocialHandle>,
121 custom_fields: Vec<ContactCustomField>,
122 ) -> Result<Contact> {
123 let birthday = row
124 .birthday
125 .as_deref()
126 .map(|s| {
127 chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d")
128 .map_err(|e| CoreError::database_msg(format!("Invalid birthday: {}", e)))
129 })
130 .transpose()?;
131
132 Ok(Contact {
133 id: parse_uuid(&row.id)?.into(),
134 display_name: row.display_name,
135 nickname: row.nickname,
136 company: row.company,
137 title: row.title,
138 notes: row.notes,
139 tags: parse_tags(&row.tags),
140 birthday,
141 timezone: row.timezone,
142 external_source: row.external_source,
143 external_id: row.external_id,
144 is_implicit: row.is_implicit != 0,
145 emails,
146 phones,
147 social_handles,
148 custom_fields,
149 created_at: parse_datetime(&row.created_at)?,
150 updated_at: parse_datetime(&row.updated_at)?,
151 })
152 }
153
154 // ============ Repository ============
155
156 /// SQLite-backed implementation of [`ContactRepository`].
157 pub struct SqliteContactRepository {
158 pool: SqlitePool,
159 }
160
161 impl SqliteContactRepository {
162 /// Creates a new repository instance with the given connection pool.
163 #[tracing::instrument(skip_all)]
164 pub fn new(pool: SqlitePool) -> Self {
165 Self { pool }
166 }
167
168 /// Batch-load emails for a set of contact IDs.
169 async fn load_emails_for_contacts(
170 &self,
171 ids: &[String],
172 ) -> Result<HashMap<ContactId, Vec<ContactEmail>>> {
173 if ids.is_empty() {
174 return Ok(HashMap::new());
175 }
176
177 let placeholders = bind_placeholders(ids.len());
178 let sql = format!(
179 "SELECT id, contact_id, address, label, is_primary FROM contact_emails WHERE contact_id IN ({}) ORDER BY is_primary DESC, rowid ASC",
180 placeholders
181 );
182
183 let mut query = sqlx::query_as::<_, ContactEmailRow>(&sql);
184 for id in ids {
185 query = query.bind(id);
186 }
187
188 let rows = query.fetch_all(&self.pool).await.map_err(CoreError::database)?;
189 let mut map: HashMap<ContactId, Vec<ContactEmail>> = HashMap::new();
190 for row in rows {
191 let contact_id: ContactId = parse_uuid(&row.contact_id)?.into();
192 let email = contact_email_from_row(row)?;
193 map.entry(contact_id).or_default().push(email);
194 }
195 Ok(map)
196 }
197
198 /// Batch-load phones for a set of contact IDs.
199 async fn load_phones_for_contacts(
200 &self,
201 ids: &[String],
202 ) -> Result<HashMap<ContactId, Vec<ContactPhone>>> {
203 if ids.is_empty() {
204 return Ok(HashMap::new());
205 }
206
207 let placeholders = bind_placeholders(ids.len());
208 let sql = format!(
209 "SELECT id, contact_id, number, label, is_primary FROM contact_phones WHERE contact_id IN ({}) ORDER BY is_primary DESC, rowid ASC",
210 placeholders
211 );
212
213 let mut query = sqlx::query_as::<_, ContactPhoneRow>(&sql);
214 for id in ids {
215 query = query.bind(id);
216 }
217
218 let rows = query.fetch_all(&self.pool).await.map_err(CoreError::database)?;
219 let mut map: HashMap<ContactId, Vec<ContactPhone>> = HashMap::new();
220 for row in rows {
221 let contact_id: ContactId = parse_uuid(&row.contact_id)?.into();
222 let phone = contact_phone_from_row(row)?;
223 map.entry(contact_id).or_default().push(phone);
224 }
225 Ok(map)
226 }
227
228 /// Batch-load social handles for a set of contact IDs.
229 async fn load_social_handles_for_contacts(
230 &self,
231 ids: &[String],
232 ) -> Result<HashMap<ContactId, Vec<SocialHandle>>> {
233 if ids.is_empty() {
234 return Ok(HashMap::new());
235 }
236
237 let placeholders = bind_placeholders(ids.len());
238 let sql = format!(
239 "SELECT id, contact_id, platform, handle, url FROM contact_social_handles WHERE contact_id IN ({}) ORDER BY rowid ASC",
240 placeholders
241 );
242
243 let mut query = sqlx::query_as::<_, SocialHandleRow>(&sql);
244 for id in ids {
245 query = query.bind(id);
246 }
247
248 let rows = query.fetch_all(&self.pool).await.map_err(CoreError::database)?;
249 let mut map: HashMap<ContactId, Vec<SocialHandle>> = HashMap::new();
250 for row in rows {
251 let contact_id: ContactId = parse_uuid(&row.contact_id)?.into();
252 let handle = social_handle_from_row(row)?;
253 map.entry(contact_id).or_default().push(handle);
254 }
255 Ok(map)
256 }
257
258 /// Batch-load custom fields for a set of contact IDs.
259 async fn load_custom_fields_for_contacts(
260 &self,
261 ids: &[String],
262 ) -> Result<HashMap<ContactId, Vec<ContactCustomField>>> {
263 if ids.is_empty() {
264 return Ok(HashMap::new());
265 }
266
267 let placeholders = bind_placeholders(ids.len());
268 let sql = format!(
269 "SELECT id, contact_id, label, value, url FROM contact_custom_fields WHERE contact_id IN ({}) ORDER BY rowid ASC",
270 placeholders
271 );
272
273 let mut query = sqlx::query_as::<_, CustomFieldRow>(&sql);
274 for id in ids {
275 query = query.bind(id);
276 }
277
278 let rows = query.fetch_all(&self.pool).await.map_err(CoreError::database)?;
279 let mut map: HashMap<ContactId, Vec<ContactCustomField>> = HashMap::new();
280 for row in rows {
281 let contact_id: ContactId = parse_uuid(&row.contact_id)?.into();
282 let field = custom_field_from_row(row)?;
283 map.entry(contact_id).or_default().push(field);
284 }
285 Ok(map)
286 }
287
288 /// Hydrate a list of contact rows with their sub-collections.
289 async fn hydrate_contacts(&self, rows: Vec<ContactRow>) -> Result<Vec<Contact>> {
290 if rows.is_empty() {
291 return Ok(vec![]);
292 }
293
294 let ids: Vec<String> = rows.iter().map(|r| r.id.clone()).collect();
295 let mut emails_map = self.load_emails_for_contacts(&ids).await?;
296 let mut phones_map = self.load_phones_for_contacts(&ids).await?;
297 let mut social_map = self.load_social_handles_for_contacts(&ids).await?;
298 let mut custom_map = self.load_custom_fields_for_contacts(&ids).await?;
299
300 let mut contacts = Vec::with_capacity(rows.len());
301 for row in rows {
302 let id: ContactId = parse_uuid(&row.id)?.into();
303 let emails = emails_map.remove(&id).unwrap_or_default();
304 let phones = phones_map.remove(&id).unwrap_or_default();
305 let social_handles = social_map.remove(&id).unwrap_or_default();
306 let custom_fields = custom_map.remove(&id).unwrap_or_default();
307 contacts.push(contact_from_row(row, emails, phones, social_handles, custom_fields)?);
308 }
309 Ok(contacts)
310 }
311 }
312
313 #[async_trait]
314 impl ContactRepository for SqliteContactRepository {
315 #[tracing::instrument(skip_all)]
316 async fn list_all(&self, user_id: UserId) -> Result<Vec<Contact>> {
317 let rows = sqlx::query_as::<_, ContactRow>(
318 r#"
319 SELECT id, display_name, nickname, company, title, notes, tags, birthday, timezone, external_source, external_id, is_implicit, created_at, updated_at
320 FROM contacts
321 WHERE user_id = ? AND is_implicit = 0
322 ORDER BY display_name ASC
323 "#,
324 )
325 .bind(user_id.to_string())
326 .fetch_all(&self.pool)
327 .await
328 .map_err(CoreError::database)?;
329
330 self.hydrate_contacts(rows).await
331 }
332
333 #[tracing::instrument(skip_all)]
334 async fn get_by_id(&self, id: ContactId, user_id: UserId) -> Result<Option<Contact>> {
335 let row = sqlx::query_as::<_, ContactRow>(
336 r#"
337 SELECT id, display_name, nickname, company, title, notes, tags, birthday, timezone, external_source, external_id, is_implicit, created_at, updated_at
338 FROM contacts
339 WHERE id = ? AND user_id = ?
340 "#,
341 )
342 .bind(id.to_string())
343 .bind(user_id.to_string())
344 .fetch_optional(&self.pool)
345 .await
346 .map_err(CoreError::database)?;
347
348 match row {
349 Some(r) => {
350 let contacts = self.hydrate_contacts(vec![r]).await?;
351 Ok(contacts.into_iter().next())
352 }
353 None => Ok(None),
354 }
355 }
356
357 #[tracing::instrument(skip_all)]
358 async fn create(&self, user_id: UserId, contact: NewContact) -> Result<Contact> {
359 let id = ContactId::new();
360 let now = format_datetime_now();
361 let tags_json = serde_json::to_string(&contact.tags).unwrap_or_else(|_| "[]".to_string());
362 let birthday_str = contact.birthday.map(|d| d.format("%Y-%m-%d").to_string());
363
364 sqlx::query(
365 r#"
366 INSERT INTO contacts (id, user_id, display_name, nickname, company, title, notes, tags, birthday, timezone, is_implicit, created_at, updated_at)
367 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
368 "#,
369 )
370 .bind(id.to_string())
371 .bind(user_id.to_string())
372 .bind(&contact.display_name)
373 .bind(&contact.nickname)
374 .bind(&contact.company)
375 .bind(&contact.title)
376 .bind(&contact.notes)
377 .bind(&tags_json)
378 .bind(&birthday_str)
379 .bind(&contact.timezone)
380 .bind(if contact.is_implicit { 1 } else { 0 })
381 .bind(&now)
382 .bind(&now)
383 .execute(&self.pool)
384 .await
385 .map_err(CoreError::database)?;
386
387 self.get_by_id(id, user_id)
388 .await?
389 .ok_or_else(|| CoreError::internal("Failed to retrieve created contact"))
390 }
391
392 #[tracing::instrument(skip_all)]
393 async fn restore(&self, user_id: UserId, contact: &Contact) -> Result<()> {
394 let tags_json = serde_json::to_string(&contact.tags).unwrap_or_else(|_| "[]".to_string());
395 let birthday_str = contact.birthday.map(|d| d.format("%Y-%m-%d").to_string());
396
397 sqlx::query(
398 r#"
399 INSERT OR IGNORE INTO contacts (id, user_id, display_name, nickname, company, title, notes, tags, birthday, timezone, external_source, external_id, is_implicit, created_at, updated_at)
400 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
401 "#,
402 )
403 .bind(contact.id.to_string())
404 .bind(user_id.to_string())
405 .bind(&contact.display_name)
406 .bind(&contact.nickname)
407 .bind(&contact.company)
408 .bind(&contact.title)
409 .bind(&contact.notes)
410 .bind(&tags_json)
411 .bind(&birthday_str)
412 .bind(&contact.timezone)
413 .bind(&contact.external_source)
414 .bind(&contact.external_id)
415 .bind(if contact.is_implicit { 1 } else { 0 })
416 .bind(format_datetime(&contact.created_at))
417 .bind(format_datetime(&contact.updated_at))
418 .execute(&self.pool)
419 .await
420 .map_err(CoreError::database)?;
421 Ok(())
422 }
423
424 #[tracing::instrument(skip_all)]
425 async fn update(&self, id: ContactId, user_id: UserId, contact: UpdateContact) -> Result<Option<Contact>> {
426 let now = format_datetime_now();
427 let tags_json = serde_json::to_string(&contact.tags).unwrap_or_else(|_| "[]".to_string());
428 let birthday_str = contact.birthday.map(|d| d.format("%Y-%m-%d").to_string());
429
430 let result = sqlx::query(
431 r#"
432 UPDATE contacts
433 SET display_name = ?, nickname = ?, company = ?, title = ?, notes = ?, tags = ?, birthday = ?, timezone = ?, updated_at = ?
434 WHERE id = ? AND user_id = ?
435 "#,
436 )
437 .bind(&contact.display_name)
438 .bind(&contact.nickname)
439 .bind(&contact.company)
440 .bind(&contact.title)
441 .bind(&contact.notes)
442 .bind(&tags_json)
443 .bind(&birthday_str)
444 .bind(&contact.timezone)
445 .bind(&now)
446 .bind(id.to_string())
447 .bind(user_id.to_string())
448 .execute(&self.pool)
449 .await
450 .map_err(CoreError::database)?;
451
452 if result.rows_affected() > 0 {
453 self.get_by_id(id, user_id).await
454 } else {
455 Ok(None)
456 }
457 }
458
459 #[tracing::instrument(skip_all)]
460 async fn delete(&self, id: ContactId, user_id: UserId) -> Result<bool> {
461 let result = sqlx::query("DELETE FROM contacts WHERE id = ? AND user_id = ?")
462 .bind(id.to_string())
463 .bind(user_id.to_string())
464 .execute(&self.pool)
465 .await
466 .map_err(CoreError::database)?;
467
468 Ok(result.rows_affected() > 0)
469 }
470
471 #[tracing::instrument(skip_all)]
472 async fn set_external_ref(
473 &self,
474 id: ContactId,
475 user_id: UserId,
476 source: &str,
477 external_id: &str,
478 ) -> Result<()> {
479 sqlx::query(
480 "UPDATE contacts SET external_source = ?, external_id = ? WHERE id = ? AND user_id = ?",
481 )
482 .bind(source)
483 .bind(external_id)
484 .bind(id.to_string())
485 .bind(user_id.to_string())
486 .execute(&self.pool)
487 .await
488 .map_err(CoreError::database)?;
489
490 Ok(())
491 }
492
493 #[tracing::instrument(skip_all)]
494 async fn delete_many(&self, ids: &[ContactId], user_id: UserId) -> Result<u64> {
495 if ids.is_empty() {
496 return Ok(0);
497 }
498 let user_id_str = user_id.to_string();
499 let placeholders = bind_placeholders(ids.len());
500 let sql = format!("DELETE FROM contacts WHERE user_id = ? AND id IN ({placeholders})");
501 let mut query = sqlx::query(&sql).bind(&user_id_str);
502 for id in ids {
503 query = query.bind(id.to_string());
504 }
505 let result = query.execute(&self.pool).await.map_err(CoreError::database)?;
506 Ok(result.rows_affected())
507 }
508
509 #[tracing::instrument(skip_all)]
510 async fn tag_many(&self, ids: &[ContactId], user_id: UserId, tag: &str) -> Result<u64> {
511 if ids.is_empty() || tag.is_empty() {
512 return Ok(0);
513 }
514 let user_id_str = user_id.to_string();
515 let like_pattern = format!("%\"{}\"%" , escape_like(tag));
516 let placeholders = bind_placeholders(ids.len());
517 // Append tag to JSON array where not already present.
518 let sql = format!(
519 r#"UPDATE contacts
520 SET tags = CASE
521 WHEN tags IS NULL OR tags = '' OR tags = '[]'
522 THEN json_array(?)
523 ELSE json_insert(tags, '$[#]', ?)
524 END,
525 updated_at = datetime('now')
526 WHERE user_id = ? AND id IN ({placeholders})
527 AND (tags IS NULL OR tags NOT LIKE ? ESCAPE '\')"#,
528 );
529 let mut query = sqlx::query(&sql)
530 .bind(tag)
531 .bind(tag)
532 .bind(&user_id_str);
533 for id in ids {
534 query = query.bind(id.to_string());
535 }
536 query = query.bind(&like_pattern);
537 let result = query.execute(&self.pool).await.map_err(CoreError::database)?;
538 Ok(result.rows_affected())
539 }
540
541 #[tracing::instrument(skip_all)]
542 async fn list_by_tag(&self, user_id: UserId, tag: &str) -> Result<Vec<Contact>> {
543 // Tags stored as JSON array, use LIKE for matching
544 let pattern = format!("%\"{}\"%" , escape_like(tag));
545 let rows = sqlx::query_as::<_, ContactRow>(
546 r#"
547 SELECT id, display_name, nickname, company, title, notes, tags, birthday, timezone, external_source, external_id, is_implicit, created_at, updated_at
548 FROM contacts
549 WHERE user_id = ? AND tags LIKE ? ESCAPE '\'
550 ORDER BY display_name ASC
551 "#,
552 )
553 .bind(user_id.to_string())
554 .bind(&pattern)
555 .fetch_all(&self.pool)
556 .await
557 .map_err(CoreError::database)?;
558
559 self.hydrate_contacts(rows).await
560 }
561
562 #[tracing::instrument(skip_all)]
563 async fn list_filtered(&self, user_id: UserId, search: Option<&str>, tag: Option<&str>, include_implicit: bool) -> Result<Vec<Contact>> {
564 let has_search = search.is_some_and(|s| !s.is_empty());
565 let has_tag = tag.is_some_and(|t| !t.is_empty());
566
567 if !has_search && !has_tag && !include_implicit {
568 return self.list_all(user_id).await;
569 }
570
571 let mut conditions = vec!["c.user_id = ?".to_string()];
572 if !include_implicit {
573 conditions.push("c.is_implicit = 0".to_string());
574 }
575 let mut binds: Vec<String> = vec![user_id.to_string()];
576
577 if let Some(t) = tag.filter(|t| !t.is_empty()) {
578 conditions.push("c.tags LIKE ? ESCAPE '\\'".to_string());
579 binds.push(format!("%\"{}\"%" , escape_like(t)));
580 }
581
582 if let Some(s) = search.filter(|s| !s.is_empty()) {
583 let search_pattern = format!("%{}%", escape_like(&s.to_lowercase()));
584 // Search across contact fields and email addresses using a subquery
585 conditions.push(
586 "(LOWER(c.display_name) LIKE ? ESCAPE '\\' OR LOWER(COALESCE(c.nickname, '')) LIKE ? ESCAPE '\\' OR LOWER(COALESCE(c.company, '')) LIKE ? ESCAPE '\\' OR LOWER(COALESCE(c.title, '')) LIKE ? ESCAPE '\\' OR LOWER(c.notes) LIKE ? ESCAPE '\\' OR EXISTS (SELECT 1 FROM contact_emails ce WHERE ce.contact_id = c.id AND LOWER(ce.address) LIKE ? ESCAPE '\\'))".to_string()
587 );
588 // 6 binds for the search pattern
589 for _ in 0..6 {
590 binds.push(search_pattern.clone());
591 }
592 }
593
594 let sql = format!(
595 "SELECT c.id, c.display_name, c.nickname, c.company, c.title, c.notes, c.tags, c.birthday, c.timezone, c.external_source, c.external_id, c.is_implicit, c.created_at, c.updated_at FROM contacts c WHERE {} ORDER BY c.display_name ASC",
596 conditions.join(" AND ")
597 );
598
599 let mut query = sqlx::query_as::<_, ContactRow>(&sql);
600 for bind in &binds {
601 query = query.bind(bind);
602 }
603
604 let rows = query.fetch_all(&self.pool).await.map_err(CoreError::database)?;
605 self.hydrate_contacts(rows).await
606 }
607
608 #[tracing::instrument(skip_all)]
609 async fn list_email_directory(&self, user_id: UserId, include_implicit: bool) -> Result<Vec<ContactEmailEntry>> {
610 // One JOIN, one row per email address — no per-contact sub-collection
611 // hydration (the compose autocomplete only needs name + address).
612 let implicit_filter = if include_implicit { "" } else { "AND c.is_implicit = 0" };
613 let sql = format!(
614 "SELECT c.display_name, ce.address, c.is_implicit \
615 FROM contacts c JOIN contact_emails ce ON ce.contact_id = c.id \
616 WHERE c.user_id = ? {} \
617 ORDER BY c.display_name ASC, ce.is_primary DESC",
618 implicit_filter
619 );
620 let rows: Vec<(String, String, i64)> = sqlx::query_as(&sql)
621 .bind(user_id.to_string())
622 .fetch_all(&self.pool)
623 .await
624 .map_err(CoreError::database)?;
625 Ok(rows
626 .into_iter()
627 .map(|(name, email, is_implicit)| ContactEmailEntry {
628 name,
629 email,
630 is_implicit: is_implicit != 0,
631 })
632 .collect())
633 }
634
635 #[tracing::instrument(skip_all)]
636 async fn find_by_email(&self, user_id: UserId, email: &str) -> Result<Option<Contact>> {
637 let row = sqlx::query_as::<_, ContactRow>(
638 r#"
639 SELECT c.id, c.display_name, c.nickname, c.company, c.title, c.notes, c.tags, c.birthday, c.timezone, c.external_source, c.external_id, c.is_implicit, c.created_at, c.updated_at
640 FROM contacts c
641 JOIN contact_emails ce ON ce.contact_id = c.id
642 WHERE c.user_id = ? AND LOWER(ce.address) = LOWER(?)
643 LIMIT 1
644 "#,
645 )
646 .bind(user_id.to_string())
647 .bind(email)
648 .fetch_optional(&self.pool)
649 .await
650 .map_err(CoreError::database)?;
651
652 match row {
653 Some(r) => {
654 let contacts = self.hydrate_contacts(vec![r]).await?;
655 Ok(contacts.into_iter().next())
656 }
657 None => Ok(None),
658 }
659 }
660
661 #[tracing::instrument(skip_all)]
662 async fn find_emails_in_contacts(&self, user_id: UserId, addresses: &[&str]) -> Result<HashSet<String>> {
663 if addresses.is_empty() {
664 return Ok(HashSet::new());
665 }
666
667 let placeholders = bind_placeholders(addresses.len());
668 let query = format!(
669 "SELECT DISTINCT LOWER(ce.address) FROM contact_emails ce JOIN contacts c ON ce.contact_id = c.id WHERE c.user_id = ? AND LOWER(ce.address) IN ({})",
670 placeholders
671 );
672
673 let mut q = sqlx::query_as::<_, (String,)>(&query).bind(user_id.to_string());
674 for addr in addresses {
675 q = q.bind(addr.to_lowercase());
676 }
677
678 let rows = q.fetch_all(&self.pool).await
679 .map_err(CoreError::database)?;
680
681 Ok(rows.into_iter().map(|(a,)| a).collect())
682 }
683
684 #[tracing::instrument(skip_all)]
685 async fn promote_contact(&self, id: ContactId, user_id: UserId) -> Result<Option<Contact>> {
686 let now = format_datetime_now();
687 let result = sqlx::query(
688 "UPDATE contacts SET is_implicit = 0, updated_at = ? WHERE id = ? AND user_id = ?"
689 )
690 .bind(&now)
691 .bind(id.to_string())
692 .bind(user_id.to_string())
693 .execute(&self.pool)
694 .await
695 .map_err(CoreError::database)?;
696
697 if result.rows_affected() > 0 {
698 self.get_by_id(id, user_id).await
699 } else {
700 Ok(None)
701 }
702 }
703
704 #[tracing::instrument(skip_all)]
705 async fn find_by_external_id(&self, source: &str, ext_id: &str, user_id: UserId) -> Result<Option<Contact>> {
706 let row = sqlx::query_as::<_, ContactRow>(
707 r#"
708 SELECT id, display_name, nickname, company, title, notes, tags, birthday, timezone, external_source, external_id, is_implicit, created_at, updated_at
709 FROM contacts
710 WHERE user_id = ? AND external_source = ? AND external_id = ?
711 LIMIT 1
712 "#,
713 )
714 .bind(user_id.to_string())
715 .bind(source)
716 .bind(ext_id)
717 .fetch_optional(&self.pool)
718 .await
719 .map_err(CoreError::database)?;
720
721 match row {
722 Some(r) => {
723 let contacts = self.hydrate_contacts(vec![r]).await?;
724 Ok(contacts.into_iter().next())
725 }
726 None => Ok(None),
727 }
728 }
729
730 #[tracing::instrument(skip_all)]
731 async fn add_email(&self, contact_id: ContactId, user_id: UserId, email: NewContactEmail) -> Result<ContactEmail> {
732 // Verify contact ownership
733 let exists = sqlx::query_scalar::<_, i32>(
734 "SELECT COUNT(*) FROM contacts WHERE id = ? AND user_id = ?"
735 )
736 .bind(contact_id.to_string())
737 .bind(user_id.to_string())
738 .fetch_one(&self.pool)
739 .await
740 .map_err(CoreError::database)?;
741
742 if exists == 0 {
743 return Err(CoreError::not_found("contact", contact_id));
744 }
745
746 let id = ContactEmailId::new();
747 sqlx::query(
748 "INSERT INTO contact_emails (id, contact_id, address, label, is_primary) VALUES (?, ?, ?, ?, ?)"
749 )
750 .bind(id.to_string())
751 .bind(contact_id.to_string())
752 .bind(&email.address)
753 .bind(&email.label)
754 .bind(email.is_primary as i32)
755 .execute(&self.pool)
756 .await
757 .map_err(CoreError::database)?;
758
759 Ok(ContactEmail {
760 id,
761 contact_id,
762 address: email.address,
763 label: email.label,
764 is_primary: email.is_primary,
765 })
766 }
767
768 #[tracing::instrument(skip_all)]
769 async fn remove_email(&self, email_id: ContactEmailId, user_id: UserId) -> Result<bool> {
770 // Verify ownership via JOIN
771 let result = sqlx::query(
772 r#"
773 DELETE FROM contact_emails
774 WHERE id = ? AND contact_id IN (SELECT id FROM contacts WHERE user_id = ?)
775 "#
776 )
777 .bind(email_id.to_string())
778 .bind(user_id.to_string())
779 .execute(&self.pool)
780 .await
781 .map_err(CoreError::database)?;
782
783 Ok(result.rows_affected() > 0)
784 }
785
786 #[tracing::instrument(skip_all)]
787 async fn add_phone(&self, contact_id: ContactId, user_id: UserId, phone: NewContactPhone) -> Result<ContactPhone> {
788 // Verify contact ownership
789 let exists = sqlx::query_scalar::<_, i32>(
790 "SELECT COUNT(*) FROM contacts WHERE id = ? AND user_id = ?"
791 )
792 .bind(contact_id.to_string())
793 .bind(user_id.to_string())
794 .fetch_one(&self.pool)
795 .await
796 .map_err(CoreError::database)?;
797
798 if exists == 0 {
799 return Err(CoreError::not_found("contact", contact_id));
800 }
801
802 let id = ContactPhoneId::new();
803 sqlx::query(
804 "INSERT INTO contact_phones (id, contact_id, number, label, is_primary) VALUES (?, ?, ?, ?, ?)"
805 )
806 .bind(id.to_string())
807 .bind(contact_id.to_string())
808 .bind(&phone.number)
809 .bind(&phone.label)
810 .bind(phone.is_primary as i32)
811 .execute(&self.pool)
812 .await
813 .map_err(CoreError::database)?;
814
815 Ok(ContactPhone {
816 id,
817 contact_id,
818 number: phone.number,
819 label: phone.label,
820 is_primary: phone.is_primary,
821 })
822 }
823
824 #[tracing::instrument(skip_all)]
825 async fn remove_phone(&self, phone_id: ContactPhoneId, user_id: UserId) -> Result<bool> {
826 let result = sqlx::query(
827 r#"
828 DELETE FROM contact_phones
829 WHERE id = ? AND contact_id IN (SELECT id FROM contacts WHERE user_id = ?)
830 "#
831 )
832 .bind(phone_id.to_string())
833 .bind(user_id.to_string())
834 .execute(&self.pool)
835 .await
836 .map_err(CoreError::database)?;
837
838 Ok(result.rows_affected() > 0)
839 }
840
841 #[tracing::instrument(skip_all)]
842 async fn add_social_handle(&self, contact_id: ContactId, user_id: UserId, handle: NewSocialHandle) -> Result<SocialHandle> {
843 // Verify contact ownership
844 let exists = sqlx::query_scalar::<_, i32>(
845 "SELECT COUNT(*) FROM contacts WHERE id = ? AND user_id = ?"
846 )
847 .bind(contact_id.to_string())
848 .bind(user_id.to_string())
849 .fetch_one(&self.pool)
850 .await
851 .map_err(CoreError::database)?;
852
853 if exists == 0 {
854 return Err(CoreError::not_found("contact", contact_id));
855 }
856
857 let id = SocialHandleId::new();
858 sqlx::query(
859 "INSERT INTO contact_social_handles (id, contact_id, platform, handle, url) VALUES (?, ?, ?, ?, ?)"
860 )
861 .bind(id.to_string())
862 .bind(contact_id.to_string())
863 .bind(&handle.platform)
864 .bind(&handle.handle)
865 .bind(&handle.url)
866 .execute(&self.pool)
867 .await
868 .map_err(CoreError::database)?;
869
870 Ok(SocialHandle {
871 id,
872 contact_id,
873 platform: handle.platform,
874 handle: handle.handle,
875 url: handle.url,
876 })
877 }
878
879 #[tracing::instrument(skip_all)]
880 async fn remove_social_handle(&self, handle_id: SocialHandleId, user_id: UserId) -> Result<bool> {
881 let result = sqlx::query(
882 r#"
883 DELETE FROM contact_social_handles
884 WHERE id = ? AND contact_id IN (SELECT id FROM contacts WHERE user_id = ?)
885 "#
886 )
887 .bind(handle_id.to_string())
888 .bind(user_id.to_string())
889 .execute(&self.pool)
890 .await
891 .map_err(CoreError::database)?;
892
893 Ok(result.rows_affected() > 0)
894 }
895
896 #[tracing::instrument(skip_all)]
897 async fn add_custom_field(&self, contact_id: ContactId, user_id: UserId, field: NewContactCustomField) -> Result<ContactCustomField> {
898 // Verify contact ownership
899 let exists = sqlx::query_scalar::<_, i32>(
900 "SELECT COUNT(*) FROM contacts WHERE id = ? AND user_id = ?"
901 )
902 .bind(contact_id.to_string())
903 .bind(user_id.to_string())
904 .fetch_one(&self.pool)
905 .await
906 .map_err(CoreError::database)?;
907
908 if exists == 0 {
909 return Err(CoreError::not_found("contact", contact_id));
910 }
911
912 let id = CustomFieldId::new();
913 sqlx::query(
914 "INSERT INTO contact_custom_fields (id, contact_id, label, value, url) VALUES (?, ?, ?, ?, ?)"
915 )
916 .bind(id.to_string())
917 .bind(contact_id.to_string())
918 .bind(&field.label)
919 .bind(&field.value)
920 .bind(&field.url)
921 .execute(&self.pool)
922 .await
923 .map_err(CoreError::database)?;
924
925 Ok(ContactCustomField {
926 id,
927 contact_id,
928 label: field.label,
929 value: field.value,
930 url: field.url,
931 })
932 }
933
934 #[tracing::instrument(skip_all)]
935 async fn remove_custom_field(&self, field_id: CustomFieldId, user_id: UserId) -> Result<bool> {
936 let result = sqlx::query(
937 r#"
938 DELETE FROM contact_custom_fields
939 WHERE id = ? AND contact_id IN (SELECT id FROM contacts WHERE user_id = ?)
940 "#
941 )
942 .bind(field_id.to_string())
943 .bind(user_id.to_string())
944 .execute(&self.pool)
945 .await
946 .map_err(CoreError::database)?;
947
948 Ok(result.rows_affected() > 0)
949 }
950
951 #[tracing::instrument(skip_all)]
952 async fn update_email(&self, email_id: ContactEmailId, user_id: UserId, email: NewContactEmail) -> Result<Option<ContactEmail>> {
953 let result = sqlx::query(
954 r#"
955 UPDATE contact_emails
956 SET address = ?, label = ?, is_primary = ?
957 WHERE id = ? AND contact_id IN (SELECT id FROM contacts WHERE user_id = ?)
958 "#
959 )
960 .bind(&email.address)
961 .bind(&email.label)
962 .bind(email.is_primary as i32)
963 .bind(email_id.to_string())
964 .bind(user_id.to_string())
965 .execute(&self.pool)
966 .await
967 .map_err(CoreError::database)?;
968
969 if result.rows_affected() == 0 {
970 return Ok(None);
971 }
972
973 let row = sqlx::query_as::<_, (String, String, String, i32)>(
974 "SELECT contact_id, address, label, is_primary FROM contact_emails WHERE id = ?"
975 )
976 .bind(email_id.to_string())
977 .fetch_one(&self.pool)
978 .await
979 .map_err(CoreError::database)?;
980
981 Ok(Some(ContactEmail {
982 id: email_id,
983 contact_id: crate::utils::parse_uuid(&row.0)?.into(),
984 address: row.1,
985 label: row.2,
986 is_primary: row.3 != 0,
987 }))
988 }
989
990 #[tracing::instrument(skip_all)]
991 async fn update_phone(&self, phone_id: ContactPhoneId, user_id: UserId, phone: NewContactPhone) -> Result<Option<ContactPhone>> {
992 let result = sqlx::query(
993 r#"
994 UPDATE contact_phones
995 SET number = ?, label = ?, is_primary = ?
996 WHERE id = ? AND contact_id IN (SELECT id FROM contacts WHERE user_id = ?)
997 "#
998 )
999 .bind(&phone.number)
1000 .bind(&phone.label)
1001 .bind(phone.is_primary as i32)
1002 .bind(phone_id.to_string())
1003 .bind(user_id.to_string())
1004 .execute(&self.pool)
1005 .await
1006 .map_err(CoreError::database)?;
1007
1008 if result.rows_affected() == 0 {
1009 return Ok(None);
1010 }
1011
1012 let row = sqlx::query_as::<_, (String, String, String, i32)>(
1013 "SELECT contact_id, number, label, is_primary FROM contact_phones WHERE id = ?"
1014 )
1015 .bind(phone_id.to_string())
1016 .fetch_one(&self.pool)
1017 .await
1018 .map_err(CoreError::database)?;
1019
1020 Ok(Some(ContactPhone {
1021 id: phone_id,
1022 contact_id: crate::utils::parse_uuid(&row.0)?.into(),
1023 number: row.1,
1024 label: row.2,
1025 is_primary: row.3 != 0,
1026 }))
1027 }
1028
1029 #[tracing::instrument(skip_all)]
1030 async fn update_social_handle(&self, handle_id: SocialHandleId, user_id: UserId, handle: NewSocialHandle) -> Result<Option<SocialHandle>> {
1031 let result = sqlx::query(
1032 r#"
1033 UPDATE contact_social_handles
1034 SET platform = ?, handle = ?, url = ?
1035 WHERE id = ? AND contact_id IN (SELECT id FROM contacts WHERE user_id = ?)
1036 "#
1037 )
1038 .bind(&handle.platform)
1039 .bind(&handle.handle)
1040 .bind(&handle.url)
1041 .bind(handle_id.to_string())
1042 .bind(user_id.to_string())
1043 .execute(&self.pool)
1044 .await
1045 .map_err(CoreError::database)?;
1046
1047 if result.rows_affected() == 0 {
1048 return Ok(None);
1049 }
1050
1051 let row = sqlx::query_as::<_, (String, String, String, Option<String>)>(
1052 "SELECT contact_id, platform, handle, url FROM contact_social_handles WHERE id = ?"
1053 )
1054 .bind(handle_id.to_string())
1055 .fetch_one(&self.pool)
1056 .await
1057 .map_err(CoreError::database)?;
1058
1059 Ok(Some(SocialHandle {
1060 id: handle_id,
1061 contact_id: crate::utils::parse_uuid(&row.0)?.into(),
1062 platform: row.1,
1063 handle: row.2,
1064 url: row.3,
1065 }))
1066 }
1067
1068 #[tracing::instrument(skip_all)]
1069 async fn update_custom_field(&self, field_id: CustomFieldId, user_id: UserId, field: NewContactCustomField) -> Result<Option<ContactCustomField>> {
1070 let result = sqlx::query(
1071 r#"
1072 UPDATE contact_custom_fields
1073 SET label = ?, value = ?, url = ?
1074 WHERE id = ? AND contact_id IN (SELECT id FROM contacts WHERE user_id = ?)
1075 "#
1076 )
1077 .bind(&field.label)
1078 .bind(&field.value)
1079 .bind(&field.url)
1080 .bind(field_id.to_string())
1081 .bind(user_id.to_string())
1082 .execute(&self.pool)
1083 .await
1084 .map_err(CoreError::database)?;
1085
1086 if result.rows_affected() == 0 {
1087 return Ok(None);
1088 }
1089
1090 let row = sqlx::query_as::<_, (String, String, String, Option<String>)>(
1091 "SELECT contact_id, label, value, url FROM contact_custom_fields WHERE id = ?"
1092 )
1093 .bind(field_id.to_string())
1094 .fetch_one(&self.pool)
1095 .await
1096 .map_err(CoreError::database)?;
1097
1098 Ok(Some(ContactCustomField {
1099 id: field_id,
1100 contact_id: crate::utils::parse_uuid(&row.0)?.into(),
1101 label: row.1,
1102 value: row.2,
1103 url: row.3,
1104 }))
1105 }
1106 }
1107