Skip to main content

max / goingson

20.8 KB · 497 lines History Blame Raw
1 //! SQLite implementation of the EventRepository.
2 //!
3 //! Manages calendar events with support for:
4 //! - All-day and timed events
5 //! - Recurrence patterns
6 //! - Project associations
7 //! - Date range queries for dashboard views
8
9 use async_trait::async_trait;
10 use chrono::{DateTime, NaiveDate, Utc};
11 use sqlx::SqlitePool;
12 use goingson_core::{
13 BlockType, ContactId, CoreError, DbValue, Event, EventId, EventRepository, NewEvent, ParseableEnum,
14 ProjectId, Recurrence, RecurrenceRule, Result, TaskId, UpdateEvent, UserId,
15 };
16
17 use crate::utils::{bind_placeholders, format_datetime, format_datetime_opt, parse_datetime, parse_uuid, parse_uuid_opt};
18
19 /// Column list for SELECT queries - avoids duplication across methods.
20 const EVENT_SELECT_COLUMNS: &str = r#"e.id, e.user_id, e.project_id, p.name as project_name,
21 e.title, e.description, e.start_time, e.end_time, e.location,
22 e.linked_task_id, e.recurrence, e.recurrence_rule, e.recurrence_parent_id,
23 e.contact_id, ct.display_name as contact_name, e.block_type,
24 e.external_source, e.external_id, e.is_read_only, e.snoozed_until,
25 e.reminder_offsets_seconds"#;
26
27 #[derive(Debug, Clone, sqlx::FromRow)]
28 struct EventRow {
29 pub id: String,
30 pub user_id: Option<String>,
31 pub project_id: Option<String>,
32 pub project_name: Option<String>,
33 pub title: String,
34 pub description: String,
35 pub start_time: String,
36 pub end_time: Option<String>,
37 pub location: Option<String>,
38 pub linked_task_id: Option<String>,
39 pub recurrence: String,
40 pub recurrence_rule: Option<String>,
41 pub recurrence_parent_id: Option<String>,
42 pub contact_id: Option<String>,
43 pub contact_name: Option<String>,
44 pub block_type: Option<String>,
45 pub external_source: Option<String>,
46 pub external_id: Option<String>,
47 pub is_read_only: i32,
48 pub snoozed_until: Option<String>,
49 pub reminder_offsets_seconds: Option<String>,
50 }
51
52 impl TryFrom<EventRow> for Event {
53 type Error = CoreError;
54
55 fn try_from(row: EventRow) -> std::result::Result<Self, Self::Error> {
56 Ok(Event {
57 id: parse_uuid(&row.id)?.into(),
58 user_id: parse_uuid_opt(row.user_id.as_deref())?.map(Into::into),
59 project_id: parse_uuid_opt(row.project_id.as_deref())?.map(Into::into),
60 project_name: row.project_name,
61 title: row.title,
62 description: row.description,
63 start_time: parse_datetime(&row.start_time)?,
64 end_time: row.end_time.as_ref().map(|s| parse_datetime(s)).transpose()?,
65 location: row.location,
66 linked_task_id: parse_uuid_opt(row.linked_task_id.as_deref())?.map(Into::into),
67 recurrence: Recurrence::from_str_or_default(&row.recurrence),
68 recurrence_rule: row.recurrence_rule
69 .as_deref()
70 .and_then(|s| serde_json::from_str::<RecurrenceRule>(s).ok()),
71 recurrence_parent_id: parse_uuid_opt(row.recurrence_parent_id.as_deref())?.map(Into::into),
72 is_recurring_instance: false,
73 contact_id: parse_uuid_opt(row.contact_id.as_deref())?.map(Into::into),
74 contact_name: row.contact_name,
75 block_type: row.block_type.as_deref().and_then(BlockType::from_str_opt),
76 external_source: row.external_source,
77 external_id: row.external_id,
78 is_read_only: row.is_read_only != 0,
79 snoozed_until: row.snoozed_until.as_deref().map(parse_datetime).transpose()?,
80 reminder_offsets_seconds: row.reminder_offsets_seconds
81 .as_deref()
82 .and_then(|s| serde_json::from_str::<Vec<i64>>(s).ok())
83 .unwrap_or_default(),
84 })
85 }
86 }
87
88 /// SQLite-backed implementation of [`EventRepository`].
89 ///
90 /// Manages calendar events with date range queries optimized for
91 /// dashboard and day planning views.
92 pub struct SqliteEventRepository {
93 pool: SqlitePool,
94 }
95
96 impl SqliteEventRepository {
97 /// Creates a new repository instance with the given connection pool.
98 #[tracing::instrument(skip_all)]
99 pub fn new(pool: SqlitePool) -> Self {
100 Self { pool }
101 }
102 }
103
104 #[async_trait]
105 impl EventRepository for SqliteEventRepository {
106 #[tracing::instrument(skip_all)]
107 async fn list_all(&self, user_id: UserId) -> Result<Vec<Event>> {
108 let query = format!(
109 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? ORDER BY e.start_time ASC",
110 EVENT_SELECT_COLUMNS
111 );
112 let rows = sqlx::query_as::<_, EventRow>(&query)
113 .bind(user_id.to_string())
114 .bind(user_id.to_string())
115 .fetch_all(&self.pool)
116 .await
117 .map_err(CoreError::database)?;
118 rows.into_iter().map(Event::try_from).collect()
119 }
120
121 #[tracing::instrument(skip_all)]
122 async fn list_by_project(&self, user_id: UserId, project_id: ProjectId) -> Result<Vec<Event>> {
123 let query = format!(
124 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? AND e.project_id = ? ORDER BY e.start_time ASC",
125 EVENT_SELECT_COLUMNS
126 );
127 let rows = sqlx::query_as::<_, EventRow>(&query)
128 .bind(user_id.to_string())
129 .bind(user_id.to_string())
130 .bind(project_id.to_string())
131 .fetch_all(&self.pool)
132 .await
133 .map_err(CoreError::database)?;
134 rows.into_iter().map(Event::try_from).collect()
135 }
136
137 #[tracing::instrument(skip_all)]
138 async fn list_by_contact(&self, user_id: UserId, contact_id: ContactId) -> Result<Vec<Event>> {
139 let query = format!(
140 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? AND e.contact_id = ? ORDER BY e.start_time DESC",
141 EVENT_SELECT_COLUMNS
142 );
143 let rows = sqlx::query_as::<_, EventRow>(&query)
144 .bind(user_id.to_string())
145 .bind(user_id.to_string())
146 .bind(contact_id.to_string())
147 .fetch_all(&self.pool)
148 .await
149 .map_err(CoreError::database)?;
150 rows.into_iter().map(Event::try_from).collect()
151 }
152
153 #[tracing::instrument(skip_all)]
154 async fn get_by_id(&self, id: EventId, user_id: UserId) -> Result<Option<Event>> {
155 let query = format!(
156 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.id = ? AND e.user_id = ?",
157 EVENT_SELECT_COLUMNS
158 );
159 let row = sqlx::query_as::<_, EventRow>(&query)
160 .bind(user_id.to_string())
161 .bind(id.to_string())
162 .bind(user_id.to_string())
163 .fetch_optional(&self.pool)
164 .await
165 .map_err(CoreError::database)?;
166 row.map(Event::try_from).transpose()
167 }
168
169 #[tracing::instrument(skip_all)]
170 async fn create(&self, user_id: UserId, event: NewEvent) -> Result<Event> {
171 let id = EventId::new();
172 let start_str = format_datetime(&event.start_time);
173 let end_str = format_datetime_opt(event.end_time);
174
175 let recurrence_rule_json = event.recurrence_rule.as_ref()
176 .map(|r| serde_json::to_string(r).unwrap_or_default());
177 let reminder_offsets_json = if event.reminder_offsets_seconds.is_empty() {
178 None
179 } else {
180 Some(serde_json::to_string(&event.reminder_offsets_seconds).unwrap_or_default())
181 };
182
183 sqlx::query(
184 "INSERT INTO events (id, user_id, project_id, title, description, start_time, end_time, location, linked_task_id, recurrence, recurrence_rule, contact_id, block_type, reminder_offsets_seconds) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
185 )
186 .bind(id.to_string())
187 .bind(user_id.to_string())
188 .bind(event.project_id.map(|p| p.to_string()))
189 .bind(&event.title)
190 .bind(&event.description)
191 .bind(&start_str)
192 .bind(&end_str)
193 .bind(&event.location)
194 .bind(event.linked_task_id.map(|t| t.to_string()))
195 .bind(event.recurrence.db_value())
196 .bind(&recurrence_rule_json)
197 .bind(event.contact_id.map(|c| c.to_string()))
198 .bind(event.block_type.as_ref().map(|b| b.db_value()))
199 .bind(&reminder_offsets_json)
200 .execute(&self.pool)
201 .await
202 .map_err(CoreError::database)?;
203
204 self.get_by_id(id, user_id).await?.ok_or_else(|| CoreError::internal("Failed to retrieve created event"))
205 }
206
207 #[tracing::instrument(skip_all)]
208 async fn restore(&self, user_id: UserId, event: &Event) -> Result<()> {
209 let recurrence_rule_json = event.recurrence_rule.as_ref()
210 .map(|r| serde_json::to_string(r).unwrap_or_default());
211 let reminder_offsets_json = if event.reminder_offsets_seconds.is_empty() {
212 None
213 } else {
214 Some(serde_json::to_string(&event.reminder_offsets_seconds).unwrap_or_default())
215 };
216
217 sqlx::query(
218 "INSERT OR IGNORE INTO events (id, user_id, project_id, title, description, start_time, end_time, location, linked_task_id, recurrence, recurrence_rule, recurrence_parent_id, contact_id, block_type, external_source, external_id, is_read_only, snoozed_until, reminder_offsets_seconds) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
219 )
220 .bind(event.id.to_string())
221 .bind(user_id.to_string())
222 .bind(event.project_id.map(|p| p.to_string()))
223 .bind(&event.title)
224 .bind(&event.description)
225 .bind(format_datetime(&event.start_time))
226 .bind(format_datetime_opt(event.end_time))
227 .bind(&event.location)
228 .bind(event.linked_task_id.map(|t| t.to_string()))
229 .bind(event.recurrence.db_value())
230 .bind(&recurrence_rule_json)
231 .bind(event.recurrence_parent_id.map(|p| p.to_string()))
232 .bind(event.contact_id.map(|c| c.to_string()))
233 .bind(event.block_type.as_ref().map(|b| b.db_value()))
234 .bind(&event.external_source)
235 .bind(&event.external_id)
236 .bind(if event.is_read_only { 1 } else { 0 })
237 .bind(format_datetime_opt(event.snoozed_until))
238 .bind(&reminder_offsets_json)
239 .execute(&self.pool)
240 .await
241 .map_err(CoreError::database)?;
242 Ok(())
243 }
244
245 #[tracing::instrument(skip_all)]
246 async fn update(&self, id: EventId, user_id: UserId, event: UpdateEvent) -> Result<Option<Event>> {
247 let start_str = format_datetime(&event.start_time);
248 let end_str = format_datetime_opt(event.end_time);
249
250 let recurrence_rule_json = event.recurrence_rule.as_ref()
251 .map(|r| serde_json::to_string(r).unwrap_or_default());
252 let reminder_offsets_json = if event.reminder_offsets_seconds.is_empty() {
253 None
254 } else {
255 Some(serde_json::to_string(&event.reminder_offsets_seconds).unwrap_or_default())
256 };
257
258 let result = sqlx::query(
259 "UPDATE events SET project_id = ?, title = ?, description = ?, start_time = ?, end_time = ?, location = ?, linked_task_id = ?, recurrence = ?, recurrence_rule = ?, contact_id = ?, block_type = ?, reminder_offsets_seconds = ? WHERE id = ? AND user_id = ?",
260 )
261 .bind(event.project_id.map(|p| p.to_string()))
262 .bind(&event.title)
263 .bind(&event.description)
264 .bind(&start_str)
265 .bind(&end_str)
266 .bind(&event.location)
267 .bind(event.linked_task_id.map(|t| t.to_string()))
268 .bind(event.recurrence.db_value())
269 .bind(&recurrence_rule_json)
270 .bind(event.contact_id.map(|c| c.to_string()))
271 .bind(event.block_type.as_ref().map(|b| b.db_value()))
272 .bind(&reminder_offsets_json)
273 .bind(id.to_string())
274 .bind(user_id.to_string())
275 .execute(&self.pool)
276 .await
277 .map_err(CoreError::database)?;
278
279 if result.rows_affected() > 0 { self.get_by_id(id, user_id).await } else { Ok(None) }
280 }
281
282 #[tracing::instrument(skip_all)]
283 async fn delete(&self, id: EventId, user_id: UserId) -> Result<bool> {
284 let result = sqlx::query("DELETE FROM events WHERE id = ? AND user_id = ?")
285 .bind(id.to_string())
286 .bind(user_id.to_string())
287 .execute(&self.pool)
288 .await
289 .map_err(CoreError::database)?;
290
291 Ok(result.rows_affected() > 0)
292 }
293
294 #[tracing::instrument(skip_all)]
295 async fn set_external_ref(
296 &self,
297 id: EventId,
298 user_id: UserId,
299 source: &str,
300 external_id: &str,
301 ) -> Result<()> {
302 sqlx::query(
303 "UPDATE events SET external_source = ?, external_id = ? WHERE id = ? AND user_id = ?",
304 )
305 .bind(source)
306 .bind(external_id)
307 .bind(id.to_string())
308 .bind(user_id.to_string())
309 .execute(&self.pool)
310 .await
311 .map_err(CoreError::database)?;
312
313 Ok(())
314 }
315
316 #[tracing::instrument(skip_all)]
317 async fn delete_many(&self, ids: &[EventId], user_id: UserId) -> Result<u64> {
318 if ids.is_empty() {
319 return Ok(0);
320 }
321 let user_id_str = user_id.to_string();
322 let placeholders = bind_placeholders(ids.len());
323 let sql = format!("DELETE FROM events WHERE user_id = ? AND id IN ({placeholders})");
324 let mut query = sqlx::query(&sql).bind(&user_id_str);
325 for id in ids {
326 query = query.bind(id.to_string());
327 }
328 let result = query.execute(&self.pool).await.map_err(CoreError::database)?;
329 Ok(result.rows_affected())
330 }
331
332 #[tracing::instrument(skip_all)]
333 async fn get_upcoming(&self, user_id: UserId, days: i64) -> Result<Vec<Event>> {
334 let query = format!(
335 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? AND e.start_time >= datetime('now') AND e.start_time <= datetime('now', ? || ' days') ORDER BY e.start_time ASC",
336 EVENT_SELECT_COLUMNS
337 );
338 let rows = sqlx::query_as::<_, EventRow>(&query)
339 .bind(user_id.to_string())
340 .bind(user_id.to_string())
341 .bind(format!("+{}", days))
342 .fetch_all(&self.pool)
343 .await
344 .map_err(CoreError::database)?;
345 rows.into_iter().map(Event::try_from).collect()
346 }
347
348 #[tracing::instrument(skip_all)]
349 async fn get_by_linked_task(&self, user_id: UserId, task_id: TaskId) -> Result<Option<Event>> {
350 let query = format!(
351 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? AND e.linked_task_id = ?",
352 EVENT_SELECT_COLUMNS
353 );
354 let row = sqlx::query_as::<_, EventRow>(&query)
355 .bind(user_id.to_string())
356 .bind(user_id.to_string())
357 .bind(task_id.to_string())
358 .fetch_optional(&self.pool)
359 .await
360 .map_err(CoreError::database)?;
361 row.map(Event::try_from).transpose()
362 }
363
364 #[tracing::instrument(skip_all)]
365 async fn delete_by_linked_task(&self, user_id: UserId, task_id: TaskId) -> Result<bool> {
366 let result = sqlx::query("DELETE FROM events WHERE user_id = ? AND linked_task_id = ?")
367 .bind(user_id.to_string())
368 .bind(task_id.to_string())
369 .execute(&self.pool)
370 .await
371 .map_err(CoreError::database)?;
372
373 Ok(result.rows_affected() > 0)
374 }
375
376 #[tracing::instrument(skip_all)]
377 async fn list_for_date(&self, user_id: UserId, date: NaiveDate) -> Result<Vec<Event>> {
378 let date_start = format!("{} 00:00:00", date);
379 let date_end = format!("{} 23:59:59", date);
380 let query = format!(
381 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? AND e.start_time <= ? AND (e.end_time IS NULL OR e.end_time >= ?) ORDER BY e.start_time ASC",
382 EVENT_SELECT_COLUMNS
383 );
384 let rows = sqlx::query_as::<_, EventRow>(&query)
385 .bind(user_id.to_string())
386 .bind(user_id.to_string())
387 .bind(&date_end)
388 .bind(&date_start)
389 .fetch_all(&self.pool)
390 .await
391 .map_err(CoreError::database)?;
392 rows.into_iter().map(Event::try_from).collect()
393 }
394
395 #[tracing::instrument(skip_all)]
396 async fn list_between(&self, user_id: UserId, start: chrono::DateTime<chrono::Utc>, end: chrono::DateTime<chrono::Utc>) -> Result<Vec<Event>> {
397 let start_str = format_datetime(&start);
398 let end_str = format_datetime(&end);
399 let query = format!(
400 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? AND e.start_time <= ? AND (e.end_time IS NULL OR e.end_time >= ?) ORDER BY e.start_time ASC",
401 EVENT_SELECT_COLUMNS
402 );
403 let rows = sqlx::query_as::<_, EventRow>(&query)
404 .bind(user_id.to_string())
405 .bind(user_id.to_string())
406 .bind(&end_str)
407 .bind(&start_str)
408 .fetch_all(&self.pool)
409 .await
410 .map_err(CoreError::database)?;
411 rows.into_iter().map(Event::try_from).collect()
412 }
413
414 #[tracing::instrument(skip_all)]
415 async fn list_recurring(&self, user_id: UserId) -> Result<Vec<Event>> {
416 let query = format!(
417 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? AND (e.recurrence != 'None' OR e.recurrence_rule IS NOT NULL) ORDER BY e.start_time ASC",
418 EVENT_SELECT_COLUMNS
419 );
420 let rows = sqlx::query_as::<_, EventRow>(&query)
421 .bind(user_id.to_string())
422 .bind(user_id.to_string())
423 .fetch_all(&self.pool)
424 .await
425 .map_err(CoreError::database)?;
426 rows.into_iter().map(Event::try_from).collect()
427 }
428
429 #[tracing::instrument(skip_all)]
430 async fn find_by_external_id(&self, source: &str, ext_id: &str, user_id: UserId) -> Result<Option<Event>> {
431 let query = format!(
432 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? AND e.external_source = ? AND e.external_id = ?",
433 EVENT_SELECT_COLUMNS
434 );
435 let row = sqlx::query_as::<_, EventRow>(&query)
436 .bind(user_id.to_string())
437 .bind(user_id.to_string())
438 .bind(source)
439 .bind(ext_id)
440 .fetch_optional(&self.pool)
441 .await
442 .map_err(CoreError::database)?;
443 row.map(Event::try_from).transpose()
444 }
445
446 #[tracing::instrument(skip_all)]
447 async fn snooze(&self, id: EventId, user_id: UserId, until: DateTime<Utc>) -> Result<Option<Event>> {
448 let until_str = format_datetime(&until);
449 let result = sqlx::query(
450 "UPDATE events SET snoozed_until = ? WHERE id = ? AND user_id = ?"
451 )
452 .bind(&until_str)
453 .bind(id.to_string())
454 .bind(user_id.to_string())
455 .execute(&self.pool)
456 .await
457 .map_err(CoreError::database)?;
458
459 if result.rows_affected() == 0 {
460 return Ok(None);
461 }
462 EventRepository::get_by_id(self, id, user_id).await
463 }
464
465 #[tracing::instrument(skip_all)]
466 async fn unsnooze(&self, id: EventId, user_id: UserId) -> Result<Option<Event>> {
467 let result = sqlx::query(
468 "UPDATE events SET snoozed_until = NULL WHERE id = ? AND user_id = ?"
469 )
470 .bind(id.to_string())
471 .bind(user_id.to_string())
472 .execute(&self.pool)
473 .await
474 .map_err(CoreError::database)?;
475
476 if result.rows_affected() == 0 {
477 return Ok(None);
478 }
479 EventRepository::get_by_id(self, id, user_id).await
480 }
481
482 #[tracing::instrument(skip_all)]
483 async fn list_snoozed(&self, user_id: UserId) -> Result<Vec<Event>> {
484 let query = format!(
485 "SELECT {} FROM events e LEFT JOIN projects p ON e.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = e.contact_id WHERE e.user_id = ? AND e.snoozed_until IS NOT NULL AND e.snoozed_until > datetime('now') ORDER BY e.snoozed_until ASC",
486 EVENT_SELECT_COLUMNS
487 );
488 let rows = sqlx::query_as::<_, EventRow>(&query)
489 .bind(user_id.to_string())
490 .bind(user_id.to_string())
491 .fetch_all(&self.pool)
492 .await
493 .map_err(CoreError::database)?;
494 rows.into_iter().map(Event::try_from).collect()
495 }
496 }
497