Skip to main content

max / goingson

14.1 KB · 446 lines History Blame Raw
1 //! SQLite implementation of time session operations.
2 //!
3 //! Manages start/stop/discard timer sessions and aggregation queries.
4 //! At most one session per user can be active (ended_at IS NULL) at any time.
5
6 use std::collections::HashMap;
7 use chrono::{DateTime, Utc};
8 use sqlx::SqlitePool;
9
10 use goingson_core::{
11 CoreError, PositiveMinutes, Result, TaskId, TimeSession, TimeSessionId, TimeTrackingSummary, UserId,
12 };
13
14 use crate::utils::{bind_placeholders, format_datetime, format_datetime_now, parse_datetime, parse_uuid};
15
16 /// Row struct for time session queries.
17 #[derive(Debug, sqlx::FromRow)]
18 struct TimeSessionRow {
19 id: String,
20 task_id: String,
21 user_id: String,
22 started_at: String,
23 ended_at: Option<String>,
24 duration_minutes: Option<i32>,
25 created_at: String,
26 }
27
28 impl TimeSessionRow {
29 fn into_session(self) -> Result<TimeSession> {
30 Ok(TimeSession {
31 id: parse_uuid(&self.id)?.into(),
32 task_id: parse_uuid(&self.task_id)?.into(),
33 user_id: parse_uuid(&self.user_id)?.into(),
34 started_at: parse_datetime(&self.started_at)?,
35 ended_at: self.ended_at.as_ref().map(|s| parse_datetime(s)).transpose()?,
36 duration_minutes: self.duration_minutes,
37 created_at: parse_datetime(&self.created_at)?,
38 })
39 }
40 }
41
42 /// Batch-fetch active sessions for a set of tasks.
43 /// Returns a map from TaskId to the active TimeSession (if any).
44 pub(crate) async fn get_active_sessions_for_tasks(
45 pool: &SqlitePool,
46 task_ids: &[String],
47 ) -> Result<HashMap<TaskId, TimeSession>> {
48 if task_ids.is_empty() {
49 return Ok(HashMap::new());
50 }
51
52 let placeholders = bind_placeholders(task_ids.len());
53 let sql = format!(
54 "SELECT id, task_id, user_id, started_at, ended_at, duration_minutes, created_at
55 FROM time_sessions WHERE task_id IN ({}) AND ended_at IS NULL",
56 placeholders
57 );
58
59 let mut query = sqlx::query_as::<_, TimeSessionRow>(&sql);
60 for id in task_ids {
61 query = query.bind(id);
62 }
63
64 let rows = query.fetch_all(pool).await.map_err(CoreError::database)?;
65
66 let mut map = HashMap::new();
67 for row in rows {
68 let task_id: TaskId = parse_uuid(&row.task_id)?.into();
69 map.insert(task_id, row.into_session()?);
70 }
71
72 Ok(map)
73 }
74
75 /// Start a timer on a task. Fails if any session is already active for the user.
76 /// Uses a transaction to prevent double-start from concurrent requests.
77 pub(crate) async fn start_timer(
78 pool: &SqlitePool,
79 task_id: TaskId,
80 user_id: UserId,
81 ) -> Result<TimeSession> {
82 let mut tx = pool.begin().await.map_err(CoreError::database)?;
83
84 // Check for existing active session (inside transaction to prevent race)
85 let existing: Option<(String,)> = sqlx::query_as(
86 "SELECT id FROM time_sessions WHERE user_id = ? AND ended_at IS NULL LIMIT 1"
87 )
88 .bind(user_id.to_string())
89 .fetch_optional(&mut *tx)
90 .await
91 .map_err(CoreError::database)?;
92
93 if existing.is_some() {
94 return Err(CoreError::validation(
95 "timer",
96 "A timer is already running. Stop or discard it first.",
97 ));
98 }
99
100 let id = TimeSessionId::new();
101 let now = format_datetime_now();
102
103 sqlx::query(
104 "INSERT INTO time_sessions (id, task_id, user_id, started_at, created_at) VALUES (?, ?, ?, ?, ?)"
105 )
106 .bind(id.to_string())
107 .bind(task_id.to_string())
108 .bind(user_id.to_string())
109 .bind(&now)
110 .bind(&now)
111 .execute(&mut *tx)
112 .await
113 .map_err(|e| {
114 // The partial unique index on (user_id) WHERE ended_at IS NULL (migration
115 // 053) is the real TOCTOU backstop if a concurrent caller slips past the
116 // existence check above. Surface the same friendly message instead of a
117 // raw database error (ultra-fuzz Run #28).
118 match e {
119 sqlx::Error::Database(ref db) if db.is_unique_violation() => {
120 CoreError::validation("timer", "A timer is already running. Stop or discard it first.")
121 }
122 other => CoreError::database(other),
123 }
124 })?;
125
126 let row = sqlx::query_as::<_, TimeSessionRow>(
127 "SELECT id, task_id, user_id, started_at, ended_at, duration_minutes, created_at FROM time_sessions WHERE id = ?"
128 )
129 .bind(id.to_string())
130 .fetch_one(&mut *tx)
131 .await
132 .map_err(CoreError::database)?;
133
134 tx.commit().await.map_err(CoreError::database)?;
135 row.into_session()
136 }
137
138 /// Stop the active timer on a task. Updates duration_minutes and the task's actual_minutes cache.
139 /// Uses a transaction so session end and task actual_minutes are updated atomically.
140 pub(crate) async fn stop_timer(
141 pool: &SqlitePool,
142 task_id: TaskId,
143 user_id: UserId,
144 ) -> Result<Option<TimeSession>> {
145 // Find active session for this task
146 let row = sqlx::query_as::<_, TimeSessionRow>(
147 "SELECT id, task_id, user_id, started_at, ended_at, duration_minutes, created_at
148 FROM time_sessions WHERE task_id = ? AND user_id = ? AND ended_at IS NULL"
149 )
150 .bind(task_id.to_string())
151 .bind(user_id.to_string())
152 .fetch_optional(pool)
153 .await
154 .map_err(CoreError::database)?;
155
156 let row = match row {
157 Some(r) => r,
158 None => return Ok(None),
159 };
160
161 let started_at = parse_datetime(&row.started_at)?;
162 let now = Utc::now();
163 let duration = (now - started_at).num_minutes().max(0) as i32;
164 let now_str = format_datetime(&now);
165
166 let mut tx = pool.begin().await.map_err(CoreError::database)?;
167
168 // Close the session only if it is still active. The active-session SELECT
169 // above ran outside this transaction, so two concurrent stops could both
170 // see it open; gating on `ended_at IS NULL` means the loser updates 0 rows
171 // and we skip the actual_minutes increment, preventing a double-count.
172 let closed = sqlx::query(
173 "UPDATE time_sessions SET ended_at = ?, duration_minutes = ? WHERE id = ? AND ended_at IS NULL"
174 )
175 .bind(&now_str)
176 .bind(duration)
177 .bind(&row.id)
178 .execute(&mut *tx)
179 .await
180 .map_err(CoreError::database)?;
181
182 if closed.rows_affected() == 0 {
183 // Another stop already closed this session; nothing to add.
184 tx.rollback().await.ok();
185 return Ok(None);
186 }
187
188 // Update the task's actual_minutes cache (only when we closed the session).
189 sqlx::query(
190 "UPDATE tasks SET actual_minutes = actual_minutes + ? WHERE id = ?"
191 )
192 .bind(duration)
193 .bind(task_id.to_string())
194 .execute(&mut *tx)
195 .await
196 .map_err(CoreError::database)?;
197
198 tx.commit().await.map_err(CoreError::database)?;
199
200 // Fetch updated session
201 let updated = sqlx::query_as::<_, TimeSessionRow>(
202 "SELECT id, task_id, user_id, started_at, ended_at, duration_minutes, created_at FROM time_sessions WHERE id = ?"
203 )
204 .bind(&row.id)
205 .fetch_one(pool)
206 .await
207 .map_err(CoreError::database)?;
208
209 Ok(Some(updated.into_session()?))
210 }
211
212 /// Discard the active timer without updating actual_minutes.
213 pub(crate) async fn discard_timer(
214 pool: &SqlitePool,
215 task_id: TaskId,
216 user_id: UserId,
217 ) -> Result<bool> {
218 let result = sqlx::query(
219 "DELETE FROM time_sessions WHERE task_id = ? AND user_id = ? AND ended_at IS NULL"
220 )
221 .bind(task_id.to_string())
222 .bind(user_id.to_string())
223 .execute(pool)
224 .await
225 .map_err(CoreError::database)?;
226
227 Ok(result.rows_affected() > 0)
228 }
229
230 /// Get the currently active timer for a user with the task description.
231 pub(crate) async fn get_active_timer(
232 pool: &SqlitePool,
233 user_id: UserId,
234 ) -> Result<Option<(TimeSession, String)>> {
235 #[derive(sqlx::FromRow)]
236 #[allow(dead_code)]
237 struct ActiveTimerRow {
238 id: String,
239 task_id: String,
240 user_id: String,
241 started_at: String,
242 ended_at: Option<String>,
243 duration_minutes: Option<i32>,
244 created_at: String,
245 task_description: String,
246 }
247
248 let row = sqlx::query_as::<_, ActiveTimerRow>(
249 "SELECT ts.id, ts.task_id, ts.user_id, ts.started_at, ts.ended_at, ts.duration_minutes, ts.created_at,
250 t.description as task_description
251 FROM time_sessions ts
252 JOIN tasks t ON t.id = ts.task_id
253 WHERE ts.user_id = ? AND ts.ended_at IS NULL
254 LIMIT 1"
255 )
256 .bind(user_id.to_string())
257 .fetch_optional(pool)
258 .await
259 .map_err(CoreError::database)?;
260
261 match row {
262 Some(r) => {
263 let session = TimeSession {
264 id: parse_uuid(&r.id)?.into(),
265 task_id: parse_uuid(&r.task_id)?.into(),
266 user_id: parse_uuid(&r.user_id)?.into(),
267 started_at: parse_datetime(&r.started_at)?,
268 ended_at: None,
269 duration_minutes: r.duration_minutes,
270 created_at: parse_datetime(&r.created_at)?,
271 };
272 Ok(Some((session, r.task_description)))
273 }
274 None => Ok(None),
275 }
276 }
277
278 /// List all time sessions for a task.
279 pub(crate) async fn list_time_sessions(
280 pool: &SqlitePool,
281 task_id: TaskId,
282 user_id: UserId,
283 ) -> Result<Vec<TimeSession>> {
284 let rows = sqlx::query_as::<_, TimeSessionRow>(
285 "SELECT id, task_id, user_id, started_at, ended_at, duration_minutes, created_at
286 FROM time_sessions WHERE task_id = ? AND user_id = ?
287 ORDER BY started_at DESC"
288 )
289 .bind(task_id.to_string())
290 .bind(user_id.to_string())
291 .fetch_all(pool)
292 .await
293 .map_err(CoreError::database)?;
294
295 rows.into_iter().map(|r| r.into_session()).collect()
296 }
297
298 /// List every time session for a user across all tasks (for full backup export).
299 pub(crate) async fn list_all_time_sessions(
300 pool: &SqlitePool,
301 user_id: UserId,
302 ) -> Result<Vec<TimeSession>> {
303 let rows = sqlx::query_as::<_, TimeSessionRow>(
304 "SELECT id, task_id, user_id, started_at, ended_at, duration_minutes, created_at
305 FROM time_sessions WHERE user_id = ?
306 ORDER BY started_at ASC"
307 )
308 .bind(user_id.to_string())
309 .fetch_all(pool)
310 .await
311 .map_err(CoreError::database)?;
312
313 rows.into_iter().map(|r| r.into_session()).collect()
314 }
315
316 /// Get aggregated time tracking summary grouped by project and date.
317 pub(crate) async fn get_time_summary(
318 pool: &SqlitePool,
319 user_id: UserId,
320 start: DateTime<Utc>,
321 end: DateTime<Utc>,
322 ) -> Result<Vec<TimeTrackingSummary>> {
323 #[derive(sqlx::FromRow)]
324 struct SummaryRow {
325 project_id: Option<String>,
326 project_name: Option<String>,
327 date: String,
328 total_minutes: i32,
329 session_count: i32,
330 }
331
332 let start_str = format_datetime(&start);
333 let end_str = format_datetime(&end);
334
335 let rows = sqlx::query_as::<_, SummaryRow>(
336 "SELECT t.project_id, p.name as project_name,
337 date(ts.started_at) as date,
338 CAST(COALESCE(SUM(ts.duration_minutes), 0) AS INTEGER) as total_minutes,
339 CAST(COUNT(*) AS INTEGER) as session_count
340 FROM time_sessions ts
341 JOIN tasks t ON t.id = ts.task_id
342 LEFT JOIN projects p ON p.id = t.project_id
343 WHERE ts.user_id = ? AND ts.ended_at IS NOT NULL
344 AND ts.started_at >= ? AND ts.started_at < ?
345 GROUP BY t.project_id, date(ts.started_at)
346 ORDER BY date DESC, total_minutes DESC"
347 )
348 .bind(user_id.to_string())
349 .bind(&start_str)
350 .bind(&end_str)
351 .fetch_all(pool)
352 .await
353 .map_err(CoreError::database)?;
354
355 rows.into_iter().map(|r| {
356 Ok(TimeTrackingSummary {
357 project_id: r.project_id.as_ref().map(|s| parse_uuid(s)).transpose()?.map(Into::into),
358 project_name: r.project_name,
359 date: r.date,
360 total_minutes: r.total_minutes,
361 session_count: r.session_count,
362 })
363 }).collect()
364 }
365
366 /// Log a manual time entry (completed session, no live timer).
367 pub(crate) async fn log_manual_time(
368 pool: &SqlitePool,
369 task_id: TaskId,
370 user_id: UserId,
371 minutes: PositiveMinutes,
372 date: DateTime<Utc>,
373 ) -> Result<TimeSession> {
374 use chrono::Duration;
375
376 let minutes = minutes.as_i32();
377
378 let id = TimeSessionId::new();
379 let started_at = date;
380 let ended_at = date + Duration::minutes(minutes as i64);
381 let now = Utc::now();
382
383 let started_str = format_datetime(&started_at);
384 let ended_str = format_datetime(&ended_at);
385 let created_str = format_datetime(&now);
386
387 // Single transaction so the session row and the task's cached actual_minutes
388 // either both land or neither does. Ownership of `task_id` is checked first,
389 // inside the same tx, so a foreign/bogus id can never leave an orphan session
390 // row nor a silently no-op'd cache update.
391 let mut tx = pool.begin().await.map_err(CoreError::database)?;
392
393 let owns_task: Option<(i64,)> = sqlx::query_as(
394 "SELECT 1 FROM tasks WHERE id = ? AND user_id = ?"
395 )
396 .bind(task_id.to_string())
397 .bind(user_id.to_string())
398 .fetch_optional(&mut *tx)
399 .await
400 .map_err(CoreError::database)?;
401 if owns_task.is_none() {
402 return Err(CoreError::not_found("task", task_id));
403 }
404
405 sqlx::query(
406 "INSERT INTO time_sessions (id, task_id, user_id, started_at, ended_at, duration_minutes, created_at)
407 VALUES (?, ?, ?, ?, ?, ?, ?)"
408 )
409 .bind(id.to_string())
410 .bind(task_id.to_string())
411 .bind(user_id.to_string())
412 .bind(&started_str)
413 .bind(&ended_str)
414 .bind(minutes)
415 .bind(&created_str)
416 .execute(&mut *tx)
417 .await
418 .map_err(CoreError::database)?;
419
420 // Update task's cached actual_minutes. NB: the `tasks` table has no
421 // `updated_at` column (only `created_at`), so this must not reference one —
422 // the working stop_timer path updates the same cache the same way.
423 sqlx::query(
424 "UPDATE tasks SET actual_minutes = COALESCE(actual_minutes, 0) + ?
425 WHERE id = ? AND user_id = ?"
426 )
427 .bind(minutes)
428 .bind(task_id.to_string())
429 .bind(user_id.to_string())
430 .execute(&mut *tx)
431 .await
432 .map_err(CoreError::database)?;
433
434 tx.commit().await.map_err(CoreError::database)?;
435
436 Ok(TimeSession {
437 id,
438 task_id,
439 user_id,
440 started_at,
441 ended_at: Some(ended_at),
442 duration_minutes: Some(minutes),
443 created_at: now,
444 })
445 }
446