Skip to main content

max / goingson

44.6 KB · 1081 lines History Blame Raw
1 //! SQLite implementation of the TaskRepository.
2 //!
3 //! Manages tasks with full support for:
4 //! - Status tracking (pending, in_progress, completed, deleted)
5 //! - Priority and urgency calculations
6 //! - Due dates and recurrence patterns
7 //! - Annotations and subtasks (delegated to annotation_repo and subtask_repo)
8 //! - Snoozing and waiting-for-response states
9 //! - Day planning with scheduled time blocks
10
11 use async_trait::async_trait;
12 use chrono::{DateTime, NaiveDate, Utc};
13 use sqlx::SqlitePool;
14
15 use goingson_core::{
16 calculate_urgency, AnnotationId, Annotation, ContactId, CoreError, DbValue, MilestoneId, NewTask,
17 ParseableEnum, Priority, PositiveMinutes, ProjectId, Recurrence, Result, SortDirection, SubtaskId,
18 StatusToken, StatusTokenId, Subtask, Task, TaskAnnotations, TaskCrud, TaskFilterQuery, TaskId, TaskScheduling,
19 TaskSortColumn, TaskStatus, TaskTimeTracking, TimeSession,
20 TimeTrackingSummary, TokenState, UpdateTask, UserId,
21 };
22
23 use crate::utils::{bind_placeholders, format_datetime, format_datetime_now, format_datetime_opt, parse_datetime, parse_tags, parse_uuid, parse_uuid_opt};
24
25 use super::annotation_repo;
26 use super::subtask_repo;
27 use super::status_token_repo;
28 use super::task_repo_state;
29 use super::time_session_repo;
30
31 /// Returns the SQL column expression for a [`TaskSortColumn`].
32 fn sort_column_sql(col: &TaskSortColumn) -> &'static str {
33 match col {
34 TaskSortColumn::Description => "t.description",
35 TaskSortColumn::Project => "p.name",
36 TaskSortColumn::Priority => "CASE t.priority WHEN 'High' THEN 3 WHEN 'Medium' THEN 2 WHEN 'Low' THEN 1 ELSE 0 END",
37 TaskSortColumn::Due => "t.due",
38 TaskSortColumn::Urgency => "t.urgency",
39 }
40 }
41
42 /// Returns whether NULLs should sort last for the given column.
43 fn sort_column_nulls_last(col: &TaskSortColumn) -> bool {
44 matches!(col, TaskSortColumn::Project | TaskSortColumn::Due)
45 }
46
47 /// Common SELECT columns for task queries with project JOIN.
48 ///
49 /// This constant ensures consistent column ordering across all task queries.
50 /// Usage: `format!("SELECT {} FROM tasks t LEFT JOIN projects p ON ...", TASK_SELECT_COLUMNS)`
51 pub(crate) const TASK_SELECT_COLUMNS: &str = r#"t.id, t.project_id, p.name as project_name,
52 t.contact_id, ct.display_name as contact_name,
53 t.milestone_id,
54 t.description, t.status,
55 t.priority, t.due, t.tags, t.urgency, t.recurrence, t.recurrence_rule, t.recurrence_parent_id, t.source_email_id,
56 t.snoozed_until, t.waiting_for_response, t.waiting_since, t.expected_response_date,
57 t.scheduled_start, t.scheduled_duration,
58 t.estimated_minutes, t.actual_minutes,
59 t.created_at, t.completed_at, t.is_focus, t.focus_set_at"#;
60
61 /// Row struct for task with project name from JOIN
62 #[derive(Debug, Clone, sqlx::FromRow)]
63 pub(crate) struct TaskRowWithProject {
64 pub id: String,
65 pub project_id: Option<String>,
66 pub project_name: Option<String>,
67 pub contact_id: Option<String>,
68 pub contact_name: Option<String>,
69 pub milestone_id: Option<String>,
70 pub description: String,
71 pub status: String,
72 pub priority: String,
73 pub due: Option<String>,
74 pub tags: String,
75 pub urgency: f64,
76 pub recurrence: String,
77 pub recurrence_rule: Option<String>,
78 pub recurrence_parent_id: Option<String>,
79 pub source_email_id: Option<String>,
80 pub snoozed_until: Option<String>,
81 pub waiting_for_response: i32,
82 pub waiting_since: Option<String>,
83 pub expected_response_date: Option<String>,
84 pub scheduled_start: Option<String>,
85 pub scheduled_duration: Option<i32>,
86 pub estimated_minutes: Option<i32>,
87 pub actual_minutes: i32,
88 pub created_at: String,
89 pub completed_at: Option<String>,
90 pub is_focus: i32,
91 pub focus_set_at: Option<String>,
92 }
93
94 impl TaskRowWithProject {
95 fn into_task(self, annotations: Vec<Annotation>, subtasks: Vec<Subtask>, status_tokens: Vec<StatusToken>) -> Result<Task> {
96 Ok(Task {
97 id: parse_uuid(&self.id)?.into(),
98 project_id: parse_uuid_opt(self.project_id.as_deref())?.map(Into::into),
99 project_name: self.project_name,
100 contact_id: parse_uuid_opt(self.contact_id.as_deref())?.map(Into::into),
101 contact_name: self.contact_name,
102 milestone_id: parse_uuid_opt(self.milestone_id.as_deref())?.map(Into::into),
103 description: self.description,
104 status: TaskStatus::from_str_or_default(&self.status),
105 priority: Priority::from_str_or_default(&self.priority),
106 due: self.due.as_ref().map(|s| parse_datetime(s)).transpose()?,
107 tags: parse_tags(&self.tags),
108 urgency: self.urgency,
109 recurrence: Recurrence::from_str_or_default(&self.recurrence),
110 recurrence_rule: self.recurrence_rule
111 .as_deref()
112 .and_then(|s| serde_json::from_str(s).ok()),
113 recurrence_parent_id: parse_uuid_opt(self.recurrence_parent_id.as_deref())?.map(Into::into),
114 source_email_id: parse_uuid_opt(self.source_email_id.as_deref())?.map(Into::into),
115 snoozed_until: self.snoozed_until.as_ref().map(|s| parse_datetime(s)).transpose()?,
116 waiting_for_response: self.waiting_for_response != 0,
117 waiting_since: self.waiting_since.as_ref().map(|s| parse_datetime(s)).transpose()?,
118 expected_response_date: self.expected_response_date.as_ref().map(|s| parse_datetime(s)).transpose()?,
119 scheduled_start: self.scheduled_start.as_ref().map(|s| parse_datetime(s)).transpose()?,
120 scheduled_duration: self.scheduled_duration,
121 estimated_minutes: self.estimated_minutes,
122 actual_minutes: self.actual_minutes,
123 active_session: None,
124 annotations,
125 subtasks,
126 status_tokens,
127 created_at: parse_datetime(&self.created_at)?,
128 completed_at: self.completed_at.as_ref().map(|s| parse_datetime(s)).transpose()?,
129 is_focus: self.is_focus != 0,
130 focus_set_at: self.focus_set_at.as_ref().map(|s| parse_datetime(s)).transpose()?,
131 })
132 }
133 }
134
135 /// SQLite-backed implementation of [`TaskRepository`].
136 ///
137 /// The most complex repository in the system, handling tasks with all their
138 /// related data (annotations, subtasks) and supporting advanced filtering,
139 /// sorting, and recurrence logic.
140 pub struct SqliteTaskRepository {
141 pool: SqlitePool,
142 }
143
144 impl SqliteTaskRepository {
145 #[tracing::instrument(skip_all)]
146 pub fn new(pool: SqlitePool) -> Self {
147 Self { pool }
148 }
149 }
150
151 /// Converts task rows to Task objects with annotations, subtasks, and active sessions.
152 ///
153 /// This helper encapsulates the common pattern of:
154 /// 1. Extracting task IDs from rows
155 /// 2. Batch-fetching annotations, subtasks, and active sessions for all tasks
156 /// 3. Converting each row to a Task with its related data
157 ///
158 /// Returns an empty vec if rows is empty (no database calls made).
159 pub(crate) async fn rows_to_tasks(pool: &SqlitePool, rows: Vec<TaskRowWithProject>) -> Result<Vec<Task>> {
160 if rows.is_empty() {
161 return Ok(vec![]);
162 }
163
164 let task_ids: Vec<String> = rows.iter().map(|r| r.id.clone()).collect();
165 let annotations_map = annotation_repo::get_annotations_for_tasks(pool, &task_ids).await?;
166 let subtasks_map = subtask_repo::get_subtasks_for_tasks(pool, &task_ids).await?;
167 let tokens_map = status_token_repo::get_tokens_for_tasks(pool, &task_ids).await?;
168 let active_sessions = time_session_repo::get_active_sessions_for_tasks(pool, &task_ids).await?;
169
170 let mut tasks = Vec::with_capacity(rows.len());
171 for row in rows {
172 let id: TaskId = parse_uuid(&row.id)?.into();
173 let annotations = annotations_map.get(&id).cloned().unwrap_or_default();
174 let subtasks = subtasks_map.get(&id).cloned().unwrap_or_default();
175 let status_tokens = tokens_map.get(&id).cloned().unwrap_or_default();
176 let mut task = row.into_task(annotations, subtasks, status_tokens)?;
177 task.active_session = active_sessions.get(&id).cloned();
178 tasks.push(task);
179 }
180
181 Ok(tasks)
182 }
183
184 /// Fetch only the fields needed for update logic — avoids annotation/subtask/session sub-queries.
185 pub(crate) async fn get_task_update_context<'e, E>(executor: E, id: TaskId, user_id: UserId) -> Result<Option<goingson_core::models::TaskUpdateContext>>
186 where
187 E: sqlx::SqliteExecutor<'e>,
188 {
189 #[derive(sqlx::FromRow)]
190 struct Row {
191 created_at: String,
192 status: String,
193 completed_at: Option<String>,
194 scheduled_start: Option<String>,
195 scheduled_duration: Option<i32>,
196 }
197
198 let row = sqlx::query_as::<_, Row>(
199 "SELECT created_at, status, completed_at, scheduled_start, scheduled_duration FROM tasks WHERE id = ? AND user_id = ?"
200 )
201 .bind(id.to_string())
202 .bind(user_id.to_string())
203 .fetch_optional(executor)
204 .await
205 .map_err(CoreError::database)?;
206
207 match row {
208 Some(r) => Ok(Some(goingson_core::models::TaskUpdateContext {
209 created_at: parse_datetime(&r.created_at)?,
210 status: TaskStatus::from_str_or_default(&r.status),
211 completed_at: r.completed_at.as_ref().map(|s| parse_datetime(s)).transpose()?,
212 scheduled_start: r.scheduled_start.as_ref().map(|s| parse_datetime(s)).transpose()?,
213 scheduled_duration: r.scheduled_duration,
214 })),
215 None => Ok(None),
216 }
217 }
218
219 /// Fetch a single task by ID and user, with annotations and subtasks.
220 pub(crate) async fn get_task_by_id(pool: &SqlitePool, id: TaskId, user_id: UserId) -> Result<Option<Task>> {
221 let sql = format!(
222 r#"
223 SELECT {}
224 FROM tasks t
225 LEFT JOIN projects p ON t.project_id = p.id AND p.user_id = ?
226 LEFT JOIN contacts ct ON ct.id = t.contact_id
227 WHERE t.id = ? AND t.user_id = ?
228 "#,
229 TASK_SELECT_COLUMNS
230 );
231 let row = sqlx::query_as::<_, TaskRowWithProject>(&sql)
232 .bind(user_id.to_string())
233 .bind(id.to_string())
234 .bind(user_id.to_string())
235 .fetch_optional(pool)
236 .await
237 .map_err(CoreError::database)?;
238
239 match row {
240 Some(row) => {
241 let annotations = annotation_repo::get_annotations_for_task(pool, id).await?;
242 let subtasks = subtask_repo::get_subtasks_for_task(pool, id).await?;
243 let status_tokens = status_token_repo::get_tokens_for_task(pool, id).await?;
244 let active_sessions = time_session_repo::get_active_sessions_for_tasks(pool, std::slice::from_ref(&row.id)).await?;
245 let mut task = row.into_task(annotations, subtasks, status_tokens)?;
246 task.active_session = active_sessions.get(&task.id).cloned();
247 Ok(Some(task))
248 }
249 None => Ok(None),
250 }
251 }
252
253 /// Run a task query with string bind parameters and convert rows to tasks.
254 ///
255 /// Handles the common pattern of: format SQL with TASK_SELECT_COLUMNS,
256 /// bind string params in order, fetch rows, convert via rows_to_tasks.
257 pub(crate) async fn query_tasks(pool: &SqlitePool, sql: &str, binds: &[String]) -> Result<Vec<Task>> {
258 let mut query = sqlx::query_as::<_, TaskRowWithProject>(sql);
259 for b in binds {
260 query = query.bind(b);
261 }
262 let rows = query.fetch_all(pool).await.map_err(CoreError::database)?;
263 rows_to_tasks(pool, rows).await
264 }
265
266 #[async_trait]
267 impl TaskCrud for SqliteTaskRepository {
268 #[tracing::instrument(skip_all)]
269 async fn list_all(&self, user_id: UserId) -> Result<Vec<Task>> {
270 let sql = format!(
271 r#"
272 SELECT {}
273 FROM tasks t
274 LEFT JOIN projects p ON t.project_id = p.id AND p.user_id = ?
275 LEFT JOIN contacts ct ON ct.id = t.contact_id
276 WHERE t.user_id = ? AND t.status != 'Deleted'
277 ORDER BY t.urgency DESC, t.created_at DESC
278 "#,
279 TASK_SELECT_COLUMNS
280 );
281 query_tasks(&self.pool, &sql, &[user_id.to_string(), user_id.to_string()]).await
282 }
283
284 #[tracing::instrument(skip_all)]
285 async fn list_by_project(&self, user_id: UserId, project_id: ProjectId) -> Result<Vec<Task>> {
286 let sql = format!(
287 r#"
288 SELECT {}
289 FROM tasks t
290 LEFT JOIN projects p ON t.project_id = p.id AND p.user_id = ?
291 LEFT JOIN contacts ct ON ct.id = t.contact_id
292 WHERE t.user_id = ? AND t.project_id = ? AND t.status != 'Deleted'
293 ORDER BY t.urgency DESC, t.created_at DESC
294 "#,
295 TASK_SELECT_COLUMNS
296 );
297 query_tasks(&self.pool, &sql, &[user_id.to_string(), user_id.to_string(), project_id.to_string()]).await
298 }
299
300 #[tracing::instrument(skip_all)]
301 async fn list_by_contact(&self, user_id: UserId, contact_id: ContactId) -> Result<Vec<Task>> {
302 let sql = format!(
303 r#"
304 SELECT {}
305 FROM tasks t
306 LEFT JOIN projects p ON t.project_id = p.id AND p.user_id = ?
307 LEFT JOIN contacts ct ON ct.id = t.contact_id
308 WHERE t.user_id = ? AND t.contact_id = ? AND t.status != 'Deleted'
309 ORDER BY t.created_at DESC
310 "#,
311 TASK_SELECT_COLUMNS
312 );
313 query_tasks(&self.pool, &sql, &[user_id.to_string(), user_id.to_string(), contact_id.to_string()]).await
314 }
315
316 #[tracing::instrument(skip_all)]
317 async fn list_filtered(&self, user_id: UserId, query: TaskFilterQuery) -> Result<(Vec<Task>, i64)> {
318 // Build dynamic WHERE clause
319 let mut conditions = vec!["t.user_id = ?".to_string(), "t.status != 'Deleted'".to_string()];
320 let mut bind_values: Vec<String> = vec![user_id.to_string()];
321
322 // Status filter
323 if let Some(ref status) = query.status {
324 conditions.push("t.status = ?".to_string());
325 bind_values.push(status.db_value().to_string());
326 }
327
328 // Project filter
329 if let Some(ref project_id) = query.project_id {
330 conditions.push("t.project_id = ?".to_string());
331 bind_values.push(project_id.to_string());
332 }
333
334 // Priority filter
335 if let Some(ref priority) = query.priority {
336 conditions.push("t.priority = ?".to_string());
337 bind_values.push(priority.db_value().to_string());
338 }
339
340 // Milestone filter
341 if let Some(ref milestone_id) = query.milestone_id {
342 conditions.push("t.milestone_id = ?".to_string());
343 bind_values.push(milestone_id.to_string());
344 }
345
346 // Snoozed filter - hide snoozed tasks unless explicitly requested
347 if !query.show_snoozed {
348 conditions.push("(t.snoozed_until IS NULL OR datetime(t.snoozed_until) <= datetime('now'))".to_string());
349 }
350
351 // Waiting only filter
352 if query.waiting_only {
353 conditions.push("t.waiting_for_response = 1".to_string());
354 }
355
356 let where_clause = conditions.join(" AND ");
357
358 // First, get total count for pagination
359 let count_sql = format!("SELECT COUNT(*) FROM tasks t WHERE {}", where_clause);
360 let mut count_query = sqlx::query_as::<_, (i64,)>(&count_sql);
361 for value in &bind_values {
362 count_query = count_query.bind(value);
363 }
364 let (total,): (i64,) = count_query.fetch_one(&self.pool).await.map_err(CoreError::database)?;
365
366 if total == 0 {
367 return Ok((vec![], 0));
368 }
369
370 // Build paginated query with parameterized LIMIT/OFFSET.
371 //
372 // Defense-in-depth: clamp before binding. A negative LIMIT means
373 // "unbounded" in SQLite and an absurd LIMIT/negative OFFSET could blow up
374 // the result set. The UI already paginates; this is a backstop against a
375 // bad caller, not the primary guard.
376 const MAX_PAGE_LIMIT: i64 = 1000;
377 let mut pagination_binds: Vec<i64> = Vec::new();
378 let limit = query.limit.map(|l| l.clamp(0, MAX_PAGE_LIMIT));
379 let offset = query.offset.map(|o| o.max(0));
380 let pagination = match (limit, offset) {
381 (Some(limit), Some(offset)) => {
382 pagination_binds.push(limit);
383 pagination_binds.push(offset);
384 " LIMIT ? OFFSET ?".to_string()
385 }
386 (Some(limit), None) => {
387 pagination_binds.push(limit);
388 " LIMIT ?".to_string()
389 }
390 _ => String::new(),
391 };
392
393 // Build dynamic ORDER BY clause
394 let sort_column = query.sort_column.unwrap_or(TaskSortColumn::Urgency);
395 let sort_direction = query.sort_direction.unwrap_or_else(|| {
396 // Default to DESC for urgency (highest first), ASC for others
397 if sort_column == TaskSortColumn::Urgency {
398 SortDirection::Desc
399 } else {
400 SortDirection::Asc
401 }
402 });
403
404 let order_by = if sort_column_nulls_last(&sort_column) {
405 // For nullable columns (project, due), put NULLs last regardless of sort direction
406 format!(
407 "{} {} NULLS LAST, t.created_at DESC",
408 sort_column_sql(&sort_column),
409 sort_direction.sql()
410 )
411 } else {
412 format!(
413 "{} {}, t.created_at DESC",
414 sort_column_sql(&sort_column),
415 sort_direction.sql()
416 )
417 };
418
419 let sql = format!(
420 r#"
421 SELECT {}
422 FROM tasks t
423 LEFT JOIN projects p ON t.project_id = p.id AND p.user_id = ?
424 LEFT JOIN contacts ct ON ct.id = t.contact_id
425 WHERE {}
426 ORDER BY {}{}
427 "#,
428 TASK_SELECT_COLUMNS, where_clause, order_by, pagination
429 );
430
431 // Build query with dynamic bindings
432 let mut sqlx_query = sqlx::query_as::<_, TaskRowWithProject>(&sql);
433
434 // Bind user_id for the JOIN
435 sqlx_query = sqlx_query.bind(user_id.to_string());
436
437 // Bind all WHERE clause values
438 for value in bind_values {
439 sqlx_query = sqlx_query.bind(value);
440 }
441
442 // Bind LIMIT/OFFSET values
443 for value in pagination_binds {
444 sqlx_query = sqlx_query.bind(value);
445 }
446
447 let rows = sqlx_query
448 .fetch_all(&self.pool)
449 .await
450 .map_err(CoreError::database)?;
451
452 let tasks = rows_to_tasks(&self.pool, rows).await?;
453 Ok((tasks, total))
454 }
455
456 #[tracing::instrument(skip_all)]
457 async fn get_by_id(&self, id: TaskId, user_id: UserId) -> Result<Option<Task>> {
458 get_task_by_id(&self.pool, id, user_id).await
459 }
460
461 #[tracing::instrument(skip_all)]
462 async fn get_update_context(&self, id: TaskId, user_id: UserId) -> Result<Option<goingson_core::models::TaskUpdateContext>> {
463 get_task_update_context(&self.pool, id, user_id).await
464 }
465
466 #[tracing::instrument(skip_all)]
467 async fn create(&self, user_id: UserId, task: NewTask) -> Result<Task> {
468 let id = TaskId::new();
469 let now = format_datetime_now();
470 let due_str = format_datetime_opt(task.due);
471 let scheduled_start_str = format_datetime_opt(task.scheduled_start);
472 let tags_json = serde_json::to_string(&task.tags).unwrap_or_else(|_| "[]".to_string());
473
474 sqlx::query(
475 r#"
476 INSERT INTO tasks (id, user_id, project_id, contact_id, milestone_id, description, priority, due, tags, recurrence, recurrence_rule, urgency, source_email_id, scheduled_start, scheduled_duration, estimated_minutes, created_at)
477 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
478 "#,
479 )
480 .bind(id.to_string())
481 .bind(user_id.to_string())
482 .bind(task.project_id.map(|p| p.to_string()))
483 .bind(task.contact_id.map(|c| c.to_string()))
484 .bind(task.milestone_id.map(|m| m.to_string()))
485 .bind(&task.description)
486 .bind(task.priority.db_value())
487 .bind(&due_str)
488 .bind(&tags_json)
489 .bind(task.recurrence.db_value())
490 .bind(task.recurrence_rule.as_ref().map(|r| serde_json::to_string(r).unwrap_or_default()))
491 .bind(task.urgency)
492 .bind(task.source_email_id.map(|e| e.to_string()))
493 .bind(&scheduled_start_str)
494 .bind(task.scheduled_duration)
495 .bind(task.estimated_minutes)
496 .bind(&now)
497 .execute(&self.pool)
498 .await
499 .map_err(CoreError::database)?;
500
501 get_task_by_id(&self.pool, id, user_id)
502 .await?
503 .ok_or_else(|| CoreError::internal("Failed to retrieve created task"))
504 }
505
506 #[tracing::instrument(skip_all)]
507 async fn restore(&self, user_id: UserId, task: &Task) -> Result<()> {
508 let tags_json = serde_json::to_string(&task.tags).unwrap_or_else(|_| "[]".to_string());
509 let recurrence_rule_json = task.recurrence_rule.as_ref()
510 .map(|r| serde_json::to_string(r).unwrap_or_default());
511
512 sqlx::query(
513 r#"
514 INSERT OR IGNORE INTO tasks (
515 id, user_id, project_id, contact_id, milestone_id, description, status,
516 priority, due, tags, urgency, recurrence, recurrence_rule, recurrence_parent_id,
517 source_email_id, snoozed_until, waiting_for_response, waiting_since, expected_response_date,
518 scheduled_start, scheduled_duration, estimated_minutes, actual_minutes,
519 created_at, completed_at, is_focus, focus_set_at
520 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
521 "#,
522 )
523 .bind(task.id.to_string())
524 .bind(user_id.to_string())
525 .bind(task.project_id.map(|p| p.to_string()))
526 .bind(task.contact_id.map(|c| c.to_string()))
527 .bind(task.milestone_id.map(|m| m.to_string()))
528 .bind(&task.description)
529 .bind(task.status.db_value())
530 .bind(task.priority.db_value())
531 .bind(format_datetime_opt(task.due))
532 .bind(&tags_json)
533 .bind(task.urgency)
534 .bind(task.recurrence.db_value())
535 .bind(&recurrence_rule_json)
536 .bind(task.recurrence_parent_id.map(|p| p.to_string()))
537 .bind(task.source_email_id.map(|e| e.to_string()))
538 .bind(format_datetime_opt(task.snoozed_until))
539 .bind(if task.waiting_for_response { 1 } else { 0 })
540 .bind(format_datetime_opt(task.waiting_since))
541 .bind(format_datetime_opt(task.expected_response_date))
542 .bind(format_datetime_opt(task.scheduled_start))
543 .bind(task.scheduled_duration)
544 .bind(task.estimated_minutes)
545 .bind(task.actual_minutes)
546 .bind(format_datetime(&task.created_at))
547 .bind(task.completed_at.map(|d| format_datetime(&d)))
548 .bind(if task.is_focus { 1 } else { 0 })
549 .bind(task.focus_set_at.map(|d| format_datetime(&d)))
550 .execute(&self.pool)
551 .await
552 .map_err(CoreError::database)?;
553 Ok(())
554 }
555
556 #[tracing::instrument(skip_all)]
557 async fn update(&self, id: TaskId, user_id: UserId, task: UpdateTask) -> Result<Option<Task>> {
558 let due_str = format_datetime_opt(task.due);
559 let scheduled_start_str = format_datetime_opt(task.scheduled_start);
560 let tags_json = serde_json::to_string(&task.tags).unwrap_or_else(|_| "[]".to_string());
561
562 // completed_at tracks the status transition, not every edit: stamp it when
563 // a task first becomes Completed, preserve it while it stays Completed,
564 // clear it only when it actually leaves Completed, and otherwise leave the
565 // stored value untouched. Lightweight context query (no sub-queries).
566 // Read the current completion context and write the update in one
567 // transaction: the completed_at derivation is a read-modify-write, so a
568 // concurrent complete()/remote-apply landing between the two checkouts
569 // would otherwise be silently clobbered.
570 let mut tx = self.pool.begin().await.map_err(CoreError::database)?;
571
572 let ctx = get_task_update_context(&mut *tx, id, user_id).await?;
573 let was_completed = ctx.as_ref().is_some_and(|c| c.status == TaskStatus::Completed);
574 let prior_completed_at = ctx.as_ref().and_then(|c| c.completed_at.as_ref().map(format_datetime));
575 let completed_at_str: Option<String> = match task.status {
576 TaskStatus::Completed if was_completed => prior_completed_at, // stays completed
577 TaskStatus::Completed => Some(format_datetime_now()), // transition in
578 _ if was_completed => None, // transition out: clear
579 _ => prior_completed_at, // stays non-completed
580 };
581
582 let result = sqlx::query(
583 r#"
584 UPDATE tasks
585 SET project_id = ?, contact_id = ?, milestone_id = ?, description = ?, status = ?, priority = ?, due = ?, tags = ?, recurrence = ?, recurrence_rule = ?, urgency = ?, scheduled_start = ?, scheduled_duration = ?, estimated_minutes = ?, completed_at = ?
586 WHERE id = ? AND user_id = ?
587 "#,
588 )
589 .bind(task.project_id.map(|p| p.to_string()))
590 .bind(task.contact_id.map(|c| c.to_string()))
591 .bind(task.milestone_id.map(|m| m.to_string()))
592 .bind(&task.description)
593 .bind(task.status.db_value())
594 .bind(task.priority.db_value())
595 .bind(&due_str)
596 .bind(&tags_json)
597 .bind(task.recurrence.db_value())
598 .bind(task.recurrence_rule.as_ref().map(|r| serde_json::to_string(r).unwrap_or_default()))
599 .bind(task.urgency)
600 .bind(&scheduled_start_str)
601 .bind(task.scheduled_duration)
602 .bind(task.estimated_minutes)
603 .bind(&completed_at_str)
604 .bind(id.to_string())
605 .bind(user_id.to_string())
606 .execute(&mut *tx)
607 .await
608 .map_err(CoreError::database)?;
609
610 let affected = result.rows_affected();
611 tx.commit().await.map_err(CoreError::database)?;
612
613 if affected > 0 {
614 get_task_by_id(&self.pool, id, user_id).await
615 } else {
616 Ok(None)
617 }
618 }
619
620 #[tracing::instrument(skip_all)]
621 async fn bulk_set_project(&self, user_id: UserId, ids: &[TaskId], project_id: Option<ProjectId>) -> Result<usize> {
622 if ids.is_empty() {
623 return Ok(0);
624 }
625 let placeholders = bind_placeholders(ids.len());
626 let sql = format!(
627 "UPDATE tasks SET project_id = ? WHERE user_id = ? AND id IN ({placeholders})"
628 );
629 let mut q = sqlx::query(&sql)
630 .bind(project_id.map(|p| p.to_string()))
631 .bind(user_id.to_string());
632 for id in ids {
633 q = q.bind(id.to_string());
634 }
635 let result = q.execute(&self.pool).await.map_err(CoreError::database)?;
636 Ok(result.rows_affected() as usize)
637 }
638
639 #[tracing::instrument(skip_all)]
640 async fn bulk_set_priority(&self, user_id: UserId, ids: &[TaskId], priority: Priority) -> Result<usize> {
641 if ids.is_empty() {
642 return Ok(0);
643 }
644 // Priority feeds urgency, so each task's urgency must be recomputed. Do the
645 // whole batch in one transaction (one connection) instead of N command
646 // round-trips (Perf S4).
647 let mut tx = self.pool.begin().await.map_err(CoreError::database)?;
648 let mut affected = 0usize;
649 for id in ids {
650 let row: Option<(String, Option<String>, String, String)> = sqlx::query_as(
651 "SELECT status, due, created_at, tags FROM tasks WHERE id = ? AND user_id = ?",
652 )
653 .bind(id.to_string())
654 .bind(user_id.to_string())
655 .fetch_optional(&mut *tx)
656 .await
657 .map_err(CoreError::database)?;
658 let Some((status_s, due_s, created_s, tags_s)) = row else {
659 continue;
660 };
661 let status = TaskStatus::from_str_or_default(&status_s);
662 let due = due_s.as_deref().map(parse_datetime).transpose()?;
663 let created = parse_datetime(&created_s)?;
664 let tags = parse_tags(&tags_s);
665 let urgency = calculate_urgency(&priority, &status, due.as_ref(), &created, &tags);
666 let result = sqlx::query(
667 "UPDATE tasks SET priority = ?, urgency = ? WHERE id = ? AND user_id = ?",
668 )
669 .bind(priority.db_value())
670 .bind(urgency)
671 .bind(id.to_string())
672 .bind(user_id.to_string())
673 .execute(&mut *tx)
674 .await
675 .map_err(CoreError::database)?;
676 affected += result.rows_affected() as usize;
677 }
678 tx.commit().await.map_err(CoreError::database)?;
679 Ok(affected)
680 }
681
682 #[tracing::instrument(skip_all)]
683 async fn delete(&self, id: TaskId, user_id: UserId) -> Result<bool> {
684 let result = sqlx::query("UPDATE tasks SET status = 'Deleted' WHERE id = ? AND user_id = ?")
685 .bind(id.to_string())
686 .bind(user_id.to_string())
687 .execute(&self.pool)
688 .await
689 .map_err(CoreError::database)?;
690
691 Ok(result.rows_affected() > 0)
692 }
693
694 #[tracing::instrument(skip_all)]
695 async fn start(&self, id: TaskId, user_id: UserId) -> Result<bool> {
696 let result = sqlx::query(
697 "UPDATE tasks SET status = 'Started' WHERE id = ? AND user_id = ? AND status = 'Pending'"
698 )
699 .bind(id.to_string())
700 .bind(user_id.to_string())
701 .execute(&self.pool)
702 .await
703 .map_err(CoreError::database)?;
704
705 Ok(result.rows_affected() > 0)
706 }
707
708 #[tracing::instrument(skip_all)]
709 async fn complete(&self, id: TaskId, user_id: UserId) -> Result<Option<Task>> {
710 let task = match get_task_by_id(&self.pool, id, user_id).await? {
711 Some(t) => t,
712 None => return Ok(None),
713 };
714
715 if task.status == TaskStatus::Completed {
716 return Ok(None);
717 }
718
719 let now = format_datetime_now();
720
721 let result = sqlx::query("UPDATE tasks SET status = 'Completed', completed_at = ? WHERE id = ? AND user_id = ?")
722 .bind(&now)
723 .bind(id.to_string())
724 .bind(user_id.to_string())
725 .execute(&self.pool)
726 .await
727 .map_err(CoreError::database)?;
728
729 if result.rows_affected() == 0 {
730 return Ok(None);
731 }
732
733 get_task_by_id(&self.pool, id, user_id).await
734 }
735
736 #[tracing::instrument(skip_all)]
737 async fn complete_recurring(&self, id: TaskId, user_id: UserId, next: Option<NewTask>) -> Result<(Option<Task>, Option<Task>)> {
738 let task = match get_task_by_id(&self.pool, id, user_id).await? {
739 Some(t) => t,
740 None => return Ok((None, None)),
741 };
742
743 if task.status == TaskStatus::Completed {
744 return Ok((None, None));
745 }
746
747 let mut tx = self.pool.begin().await.map_err(CoreError::database)?;
748
749 // Mark complete, but only if it is not already Completed. The single
750 // conditional UPDATE serializes on the write lock, so of two concurrent
751 // calls only the one that actually transitions the task matches a row;
752 // the loser affects zero rows and skips the next-instance insert. Without
753 // this both calls would pass the pre-txn guard (WAL snapshot isolation
754 // hides the other's uncommitted write) and each insert a duplicate.
755 let now = format_datetime_now();
756 let marked = sqlx::query(
757 "UPDATE tasks SET status = 'Completed', completed_at = ? WHERE id = ? AND user_id = ? AND status != 'Completed'",
758 )
759 .bind(&now)
760 .bind(id.to_string())
761 .bind(user_id.to_string())
762 .execute(&mut *tx)
763 .await
764 .map_err(CoreError::database)?;
765
766 if marked.rows_affected() == 0 {
767 // Another call completed it first (or it vanished) — do not insert a
768 // second recurring instance.
769 tx.rollback().await.map_err(CoreError::database)?;
770 return Ok((None, None));
771 }
772
773 // Create next recurring instance if provided
774 let next_id = if let Some(new_task) = &next {
775 let nid = TaskId::new();
776 let due_str = format_datetime_opt(new_task.due);
777 let scheduled_start_str = format_datetime_opt(new_task.scheduled_start);
778 let tags_json = serde_json::to_string(&new_task.tags).unwrap_or_else(|_| "[]".to_string());
779
780 sqlx::query(
781 r#"
782 INSERT INTO tasks (id, user_id, project_id, contact_id, milestone_id, description, priority, due, tags, recurrence, recurrence_rule, urgency, source_email_id, scheduled_start, scheduled_duration, estimated_minutes, recurrence_parent_id, created_at)
783 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
784 "#,
785 )
786 .bind(nid.to_string())
787 .bind(user_id.to_string())
788 .bind(new_task.project_id.map(|p| p.to_string()))
789 .bind(new_task.contact_id.map(|c| c.to_string()))
790 .bind(new_task.milestone_id.map(|m| m.to_string()))
791 .bind(&new_task.description)
792 .bind(new_task.priority.db_value())
793 .bind(&due_str)
794 .bind(&tags_json)
795 .bind(new_task.recurrence.db_value())
796 .bind(new_task.recurrence_rule.as_ref().map(|r| serde_json::to_string(r).unwrap_or_default()))
797 .bind(new_task.urgency)
798 .bind(new_task.source_email_id.map(|e| e.to_string()))
799 .bind(&scheduled_start_str)
800 .bind(new_task.scheduled_duration)
801 .bind(new_task.estimated_minutes)
802 .bind(new_task.recurrence_parent_id.map(|p| p.to_string()))
803 .bind(&now)
804 .execute(&mut *tx)
805 .await
806 .map_err(CoreError::database)?;
807
808 Some(nid)
809 } else {
810 None
811 };
812
813 tx.commit().await.map_err(CoreError::database)?;
814
815 // Fetch the completed task and new task (outside transaction, committed)
816 let completed = get_task_by_id(&self.pool, id, user_id).await?;
817 let next_task = match next_id {
818 Some(nid) => get_task_by_id(&self.pool, nid, user_id).await?,
819 None => None,
820 };
821
822 Ok((completed, next_task))
823 }
824
825 #[tracing::instrument(skip_all)]
826 async fn count_incomplete_by_milestone(&self, milestone_id: MilestoneId, user_id: UserId) -> Result<i64> {
827 let (count,): (i64,) = sqlx::query_as(
828 "SELECT COUNT(*) FROM tasks WHERE milestone_id = ? AND user_id = ? AND status != 'Deleted' AND status != 'Completed'"
829 )
830 .bind(milestone_id.to_string())
831 .bind(user_id.to_string())
832 .fetch_one(&self.pool)
833 .await
834 .map_err(CoreError::database)?;
835
836 Ok(count)
837 }
838
839 // ---- Reporting (delegated to task_repo_state) ----
840
841 #[tracing::instrument(skip_all)]
842 async fn list_completed_between(&self, user_id: UserId, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<Task>> {
843 task_repo_state::list_completed_between(&self.pool, user_id, start, end).await
844 }
845
846 #[tracing::instrument(skip_all)]
847 async fn list_became_overdue_between(&self, user_id: UserId, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<Task>> {
848 task_repo_state::list_became_overdue_between(&self.pool, user_id, start, end).await
849 }
850
851 #[tracing::instrument(skip_all)]
852 async fn list_due_between(&self, user_id: UserId, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<Task>> {
853 task_repo_state::list_due_between(&self.pool, user_id, start, end).await
854 }
855
856 #[tracing::instrument(skip_all)]
857 async fn list_created_between(&self, user_id: UserId, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<Task>> {
858 task_repo_state::list_created_between(&self.pool, user_id, start, end).await
859 }
860
861 #[tracing::instrument(skip_all)]
862 async fn list_recurrence_chain(&self, root_id: TaskId, user_id: UserId) -> Result<Vec<Task>> {
863 let sql = format!(
864 "SELECT {} FROM tasks t LEFT JOIN projects p ON t.project_id = p.id AND p.user_id = ? LEFT JOIN contacts ct ON ct.id = t.contact_id WHERE (t.recurrence_parent_id = ? OR t.id = ?) AND t.user_id = ? ORDER BY t.created_at DESC",
865 TASK_SELECT_COLUMNS
866 );
867 query_tasks(&self.pool, &sql, &[
868 user_id.to_string(),
869 root_id.to_string(),
870 root_id.to_string(),
871 user_id.to_string(),
872 ]).await
873 }
874 }
875
876 #[async_trait]
877 impl TaskAnnotations for SqliteTaskRepository {
878 // ---- Annotations (delegated to annotation_repo) ----
879
880 #[tracing::instrument(skip_all)]
881 async fn get_annotations_for_task(&self, task_id: TaskId) -> Result<Vec<Annotation>> {
882 annotation_repo::get_annotations_for_task(&self.pool, task_id).await
883 }
884
885 #[tracing::instrument(skip_all)]
886 async fn add_annotation(&self, task_id: TaskId, user_id: UserId, note: &str) -> Result<Option<Annotation>> {
887 annotation_repo::add_annotation(&self.pool, task_id, user_id, note).await
888 }
889
890 #[tracing::instrument(skip_all)]
891 async fn delete_annotation(&self, annotation_id: AnnotationId, user_id: UserId) -> Result<bool> {
892 annotation_repo::delete_annotation(&self.pool, annotation_id, user_id).await
893 }
894
895 // ---- Subtasks (delegated to subtask_repo) ----
896
897 #[tracing::instrument(skip_all)]
898 async fn get_subtasks_for_task(&self, task_id: TaskId) -> Result<Vec<Subtask>> {
899 subtask_repo::get_subtasks_for_task(&self.pool, task_id).await
900 }
901
902 #[tracing::instrument(skip_all)]
903 async fn add_subtask(&self, task_id: TaskId, user_id: UserId, text: &str) -> Result<Option<Subtask>> {
904 subtask_repo::add_subtask(&self.pool, task_id, user_id, text).await
905 }
906
907 #[tracing::instrument(skip_all)]
908 async fn toggle_subtask(&self, subtask_id: SubtaskId, user_id: UserId) -> Result<Option<Subtask>> {
909 subtask_repo::toggle_subtask(&self.pool, subtask_id, user_id).await
910 }
911
912 #[tracing::instrument(skip_all)]
913 async fn update_subtask(&self, subtask_id: SubtaskId, user_id: UserId, text: &str) -> Result<Option<Subtask>> {
914 subtask_repo::update_subtask(&self.pool, subtask_id, user_id, text).await
915 }
916
917 #[tracing::instrument(skip_all)]
918 async fn delete_subtask(&self, subtask_id: SubtaskId, user_id: UserId) -> Result<bool> {
919 subtask_repo::delete_subtask(&self.pool, subtask_id, user_id).await
920 }
921
922 #[tracing::instrument(skip_all)]
923 async fn add_subtask_link(&self, task_id: TaskId, user_id: UserId, linked_task_id: TaskId) -> Result<Option<Subtask>> {
924 // Verify linked task exists and belongs to user
925 let linked_task = get_task_by_id(&self.pool, linked_task_id, user_id).await?
926 .ok_or_else(|| CoreError::not_found("linked task", linked_task_id.to_string()))?;
927
928 subtask_repo::add_subtask_link(
929 &self.pool,
930 task_id,
931 user_id,
932 linked_task_id,
933 &linked_task.description,
934 &linked_task.status,
935 ).await
936 }
937
938 // ---- Status tokens (delegated to status_token_repo) ----
939
940 #[tracing::instrument(skip_all)]
941 async fn get_status_tokens_for_task(&self, task_id: TaskId) -> Result<Vec<StatusToken>> {
942 status_token_repo::get_tokens_for_task(&self.pool, task_id).await
943 }
944
945 #[tracing::instrument(skip_all)]
946 async fn record_status_token(&self, task_id: TaskId, user_id: UserId, kind: &str, reference: &str, state: TokenState, is_primary: bool) -> Result<Option<StatusToken>> {
947 status_token_repo::record_token(&self.pool, task_id, user_id, kind, reference, state, is_primary).await
948 }
949
950 #[tracing::instrument(skip_all)]
951 async fn delete_status_token(&self, token_id: StatusTokenId, user_id: UserId) -> Result<bool> {
952 status_token_repo::delete_token(&self.pool, token_id, user_id).await
953 }
954 }
955
956 #[async_trait]
957 impl TaskScheduling for SqliteTaskRepository {
958 // ---- Snooze (delegated to task_repo_state) ----
959
960 #[tracing::instrument(skip_all)]
961 async fn snooze(&self, id: TaskId, user_id: UserId, until: DateTime<Utc>) -> Result<Option<Task>> {
962 task_repo_state::snooze(&self.pool, id, user_id, until).await
963 }
964
965 #[tracing::instrument(skip_all)]
966 async fn unsnooze(&self, id: TaskId, user_id: UserId) -> Result<Option<Task>> {
967 task_repo_state::unsnooze(&self.pool, id, user_id).await
968 }
969
970 #[tracing::instrument(skip_all)]
971 async fn list_snoozed(&self, user_id: UserId) -> Result<Vec<Task>> {
972 task_repo_state::list_snoozed(&self.pool, user_id).await
973 }
974
975 // ---- Waiting (delegated to task_repo_state) ----
976
977 #[tracing::instrument(skip_all)]
978 async fn mark_waiting(&self, id: TaskId, user_id: UserId, expected_response: Option<DateTime<Utc>>) -> Result<Option<Task>> {
979 task_repo_state::mark_waiting(&self.pool, id, user_id, expected_response).await
980 }
981
982 #[tracing::instrument(skip_all)]
983 async fn clear_waiting(&self, id: TaskId, user_id: UserId) -> Result<Option<Task>> {
984 task_repo_state::clear_waiting(&self.pool, id, user_id).await
985 }
986
987 #[tracing::instrument(skip_all)]
988 async fn list_waiting(&self, user_id: UserId) -> Result<Vec<Task>> {
989 task_repo_state::list_waiting(&self.pool, user_id).await
990 }
991
992 // ---- Scheduling (delegated to task_repo_state) ----
993
994 #[tracing::instrument(skip_all)]
995 async fn list_scheduled_for_date(&self, user_id: UserId, date: NaiveDate) -> Result<Vec<Task>> {
996 task_repo_state::list_scheduled_for_date(&self.pool, user_id, date).await
997 }
998
999 #[tracing::instrument(skip_all)]
1000 async fn list_unscheduled_due_on_date(&self, user_id: UserId, date: NaiveDate) -> Result<Vec<Task>> {
1001 task_repo_state::list_unscheduled_due_on_date(&self.pool, user_id, date).await
1002 }
1003
1004 #[tracing::instrument(skip_all)]
1005 async fn list_unscheduled_due_between(&self, user_id: UserId, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<Task>> {
1006 task_repo_state::list_unscheduled_due_between(&self.pool, user_id, start, end).await
1007 }
1008
1009 #[tracing::instrument(skip_all)]
1010 async fn update_schedule(&self, id: TaskId, user_id: UserId, start: Option<DateTime<Utc>>, duration: Option<i32>) -> Result<Option<Task>> {
1011 task_repo_state::update_schedule(&self.pool, id, user_id, start, duration).await
1012 }
1013
1014 // ---- Focus (delegated to task_repo_state) ----
1015
1016 #[tracing::instrument(skip_all)]
1017 async fn set_focus(&self, id: TaskId, user_id: UserId, is_focus: bool) -> Result<Option<Task>> {
1018 task_repo_state::set_focus(&self.pool, id, user_id, is_focus).await
1019 }
1020
1021 #[tracing::instrument(skip_all)]
1022 async fn list_focused(&self, user_id: UserId) -> Result<Vec<Task>> {
1023 task_repo_state::list_focused(&self.pool, user_id).await
1024 }
1025
1026 #[tracing::instrument(skip_all)]
1027 async fn clear_all_focus(&self, user_id: UserId) -> Result<u64> {
1028 task_repo_state::clear_all_focus(&self.pool, user_id).await
1029 }
1030
1031 #[tracing::instrument(skip_all)]
1032 async fn list_available_for_focus(&self, user_id: UserId, limit: i64) -> Result<Vec<Task>> {
1033 task_repo_state::list_available_for_focus(&self.pool, user_id, limit).await
1034 }
1035 }
1036
1037 #[async_trait]
1038 impl TaskTimeTracking for SqliteTaskRepository {
1039 // ---- Time Tracking (delegated to time_session_repo) ----
1040
1041 #[tracing::instrument(skip_all)]
1042 async fn start_timer(&self, task_id: TaskId, user_id: UserId) -> Result<TimeSession> {
1043 time_session_repo::start_timer(&self.pool, task_id, user_id).await
1044 }
1045
1046 #[tracing::instrument(skip_all)]
1047 async fn stop_timer(&self, task_id: TaskId, user_id: UserId) -> Result<Option<TimeSession>> {
1048 time_session_repo::stop_timer(&self.pool, task_id, user_id).await
1049 }
1050
1051 #[tracing::instrument(skip_all)]
1052 async fn discard_timer(&self, task_id: TaskId, user_id: UserId) -> Result<bool> {
1053 time_session_repo::discard_timer(&self.pool, task_id, user_id).await
1054 }
1055
1056 #[tracing::instrument(skip_all)]
1057 async fn get_active_timer(&self, user_id: UserId) -> Result<Option<(TimeSession, String)>> {
1058 time_session_repo::get_active_timer(&self.pool, user_id).await
1059 }
1060
1061 #[tracing::instrument(skip_all)]
1062 async fn list_time_sessions(&self, task_id: TaskId, user_id: UserId) -> Result<Vec<TimeSession>> {
1063 time_session_repo::list_time_sessions(&self.pool, task_id, user_id).await
1064 }
1065
1066 #[tracing::instrument(skip_all)]
1067 async fn list_all_time_sessions(&self, user_id: UserId) -> Result<Vec<TimeSession>> {
1068 time_session_repo::list_all_time_sessions(&self.pool, user_id).await
1069 }
1070
1071 #[tracing::instrument(skip_all)]
1072 async fn log_manual_time(&self, task_id: TaskId, user_id: UserId, minutes: PositiveMinutes, date: DateTime<Utc>) -> Result<TimeSession> {
1073 time_session_repo::log_manual_time(&self.pool, task_id, user_id, minutes, date).await
1074 }
1075
1076 #[tracing::instrument(skip_all)]
1077 async fn get_time_summary(&self, user_id: UserId, start: DateTime<Utc>, end: DateTime<Utc>) -> Result<Vec<TimeTrackingSummary>> {
1078 time_session_repo::get_time_summary(&self.pool, user_id, start, end).await
1079 }
1080 }
1081