Skip to main content

max / goingson

Data/DB: non-destructive sync upsert, atomic review upserts, local-day dashboard, sargable dates Ultra-fuzz Run #30 Phase 2 (Data/DB -> A). - S1: sync apply used INSERT OR REPLACE (delete-then-reinsert), which cascades to children if FKs are ever enforced. Switch to INSERT ... ON CONFLICT(id) DO UPDATE so a parent re-upsert mutates in place and children survive. To keep OR REPLACE's tolerance of a null on a NOT NULL column (the changelog never emits one, but several apply tests rely on it), omit such columns, deriving the NOT NULL set from the live schema via PRAGMA table_info so it can't drift. Test: parent re-upsert preserves annotations under FK on. - S3: weekly_reviews.upsert/set_vacation_days and monthly_reflections.upsert were racy get-then-insert/update; convert to ON CONFLICT DO UPDATE on their unique keys. monthly_goals has no safe unique key (a UNIQUE on position would collide across devices on sync), so wrap its get-then-write in a transaction instead. - S2: dashboard "due today/this week" used UTC midnights (date('now')). Compute local-day windows in Rust (system tz) and pass them as bound UTC instants; counts now match the user's calendar day. - S4: drop datetime() wrappers around indexed task date columns in task_repo_state and stats so the predicates stay sargable. - Minors: toggle_subtask flips atomically via UPDATE ... RETURNING (was a read-modify-write race); clamp list_available_for_focus LIMIT to [0,1000]. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 03:48 UTC
Commit: d7eb3b8ebd4271ec9a40f2dda38eaeafffbe06b9
Parent: 415911e
9 files changed, +277 insertions, -170 deletions
M crates/core/src/repository.rs +13 -1
@@ -628,7 +628,19 @@ pub struct HighUrgencyTask {
628 628 #[async_trait]
629 629 pub trait StatsRepository: Send + Sync {
630 630 /// Computes aggregated dashboard statistics for a user.
631 - async fn get_dashboard_stats(&self, user_id: UserId) -> Result<DashboardStats>;
631 + ///
632 + /// Calendar-day windows (`today_start`, `tomorrow_start`, `week_end`) are
633 + /// passed in as UTC instants derived from the user's *local* day, so "due
634 + /// today/this week" respect the user's timezone rather than UTC midnights.
635 + /// `now` is the reference instant for overdue/upcoming comparisons.
636 + async fn get_dashboard_stats(
637 + &self,
638 + user_id: UserId,
639 + now: DateTime<Utc>,
640 + today_start: DateTime<Utc>,
641 + tomorrow_start: DateTime<Utc>,
642 + week_end: DateTime<Utc>,
643 + ) -> Result<DashboardStats>;
632 644 }
633 645
634 646 /// Type of item in search results.
M crates/db-sqlite/src/repository/monthly_review_repo.rs +40 -51
@@ -146,7 +146,13 @@ impl MonthlyReviewRepository for SqliteMonthlyReviewRepository {
146 146 let user_id_str = user_id.to_string();
147 147 let now = format_datetime(&Utc::now());
148 148
149 - // Check if a goal exists at this position for this month
149 + // monthly_goals has no unique key on (user_id, month, position) and one
150 + // can't be added safely (it would make two devices' position-1 goals
151 + // collide on sync apply). So make the get-then-write atomic with a
152 + // transaction instead of an ON CONFLICT upsert — closes the race that
153 + // could double-insert a goal at the same position.
154 + let mut tx = self.pool.begin().await.map_err(CoreError::database)?;
155 +
150 156 let existing: Option<MonthlyGoalRow> = sqlx::query_as(
151 157 "SELECT id, user_id, month, text, status, position, created_at, updated_at
152 158 FROM monthly_goals
@@ -155,11 +161,11 @@ impl MonthlyReviewRepository for SqliteMonthlyReviewRepository {
155 161 .bind(&user_id_str)
156 162 .bind(month)
157 163 .bind(position)
158 - .fetch_optional(&self.pool)
164 + .fetch_optional(&mut *tx)
159 165 .await
160 166 .map_err(CoreError::database)?;
161 167
162 - if let Some(existing) = existing {
168 + let goal = if let Some(existing) = existing {
163 169 let id = existing.id.clone();
164 170 sqlx::query(
165 171 "UPDATE monthly_goals SET text = ?, updated_at = ? WHERE id = ?"
@@ -167,14 +173,14 @@ impl MonthlyReviewRepository for SqliteMonthlyReviewRepository {
167 173 .bind(text)
168 174 .bind(&now)
169 175 .bind(&id)
170 - .execute(&self.pool)
176 + .execute(&mut *tx)
171 177 .await
172 178 .map_err(CoreError::database)?;
173 179
174 180 let mut goal = MonthlyGoal::try_from(existing)?;
175 181 goal.text = text.to_string();
176 182 goal.updated_at = Utc::now();
177 - Ok(goal)
183 + goal
178 184 } else {
179 185 let id = MonthlyGoalId::new();
180 186 sqlx::query(
@@ -188,12 +194,12 @@ impl MonthlyReviewRepository for SqliteMonthlyReviewRepository {
188 194 .bind(position)
189 195 .bind(&now)
190 196 .bind(&now)
191 - .execute(&self.pool)
197 + .execute(&mut *tx)
192 198 .await
193 199 .map_err(CoreError::database)?;
194 200
195 201 let now_dt = Utc::now();
196 - Ok(MonthlyGoal {
202 + MonthlyGoal {
197 203 id,
198 204 user_id,
199 205 month: month.to_string(),
@@ -202,8 +208,11 @@ impl MonthlyReviewRepository for SqliteMonthlyReviewRepository {
202 208 position,
203 209 created_at: now_dt,
204 210 updated_at: now_dt,
205 - })
206 - }
211 + }
212 + };
213 +
214 + tx.commit().await.map_err(CoreError::database)?;
215 + Ok(goal)
207 216 }
208 217
209 218 #[tracing::instrument(skip_all)]
@@ -275,48 +284,28 @@ impl MonthlyReviewRepository for SqliteMonthlyReviewRepository {
275 284 let now = Utc::now();
276 285 let now_str = format_datetime(&now);
277 286
278 - let existing = self.get_reflection(user_id, month).await?;
279 -
280 - let id = if let Some(existing) = existing {
281 - sqlx::query(
282 - "UPDATE monthly_reflections SET highlight_text = ?, change_text = ?, completed_at = ?
283 - WHERE id = ?"
284 - )
285 - .bind(highlight)
286 - .bind(change)
287 - .bind(&now_str)
288 - .bind(existing.id.to_string())
289 - .execute(&self.pool)
290 - .await
291 - .map_err(CoreError::database)?;
292 -
293 - existing.id
294 - } else {
295 - let id = MonthlyReflectionId::new();
296 - sqlx::query(
297 - "INSERT INTO monthly_reflections (id, user_id, month, highlight_text, change_text, completed_at)
298 - VALUES (?, ?, ?, ?, ?, ?)"
299 - )
300 - .bind(id.to_string())
301 - .bind(&user_id_str)
302 - .bind(month)
303 - .bind(highlight)
304 - .bind(change)
305 - .bind(&now_str)
306 - .execute(&self.pool)
307 - .await
308 - .map_err(CoreError::database)?;
309 -
310 - id
311 - };
287 + // Atomic upsert on the (user_id, month) unique key — no get-then-write
288 + // race.
289 + sqlx::query(
290 + "INSERT INTO monthly_reflections (id, user_id, month, highlight_text, change_text, completed_at)
291 + VALUES (?, ?, ?, ?, ?, ?)
292 + ON CONFLICT(user_id, month) DO UPDATE SET
293 + highlight_text = excluded.highlight_text,
294 + change_text = excluded.change_text,
295 + completed_at = excluded.completed_at",
296 + )
297 + .bind(MonthlyReflectionId::new().to_string())
298 + .bind(&user_id_str)
299 + .bind(month)
300 + .bind(highlight)
301 + .bind(change)
302 + .bind(&now_str)
303 + .execute(&self.pool)
304 + .await
305 + .map_err(CoreError::database)?;
312 306
313 - Ok(MonthlyReflection {
314 - id,
315 - user_id,
316 - month: month.to_string(),
317 - highlight_text: highlight.to_string(),
318 - change_text: change.to_string(),
319 - completed_at: now,
320 - })
307 + self.get_reflection(user_id, month)
308 + .await?
309 + .ok_or_else(|| CoreError::database_msg("monthly reflection vanished after upsert"))
321 310 }
322 311 }
M crates/db-sqlite/src/repository/stats_repo.rs +28 -9
@@ -7,10 +7,11 @@
7 7 //! - High-urgency task list
8 8
9 9 use async_trait::async_trait;
10 + use chrono::{DateTime, Duration, Utc};
10 11 use sqlx::SqlitePool;
11 12 use goingson_core::{CoreError, DashboardStats, HighUrgencyTask, Result, StatsRepository, UserId};
12 13
13 - use crate::utils::parse_datetime;
14 + use crate::utils::{format_datetime, parse_datetime};
14 15
15 16 /// SQLite-backed implementation of [`StatsRepository`].
16 17 ///
@@ -27,10 +28,25 @@ impl SqliteStatsRepository {
27 28 #[async_trait]
28 29 impl StatsRepository for SqliteStatsRepository {
29 30 #[tracing::instrument(skip_all)]
30 - async fn get_dashboard_stats(&self, user_id: UserId) -> Result<DashboardStats> {
31 + async fn get_dashboard_stats(
32 + &self,
33 + user_id: UserId,
34 + now: DateTime<Utc>,
35 + today_start: DateTime<Utc>,
36 + tomorrow_start: DateTime<Utc>,
37 + week_end: DateTime<Utc>,
38 + ) -> Result<DashboardStats> {
31 39 let user_id_str = user_id.to_string();
40 + let now_str = format_datetime(&now);
41 + let today_start_str = format_datetime(&today_start);
42 + let tomorrow_start_str = format_datetime(&tomorrow_start);
43 + let week_end_str = format_datetime(&week_end);
44 + let events_end_str = format_datetime(&(now + Duration::days(7)));
32 45
33 46 // Batch all 6 scalar counts into a single query using subqueries.
47 + // Day windows are bound UTC instants of the user's local day; all
48 + // predicates compare the bare indexed column (no date()/datetime()
49 + // wrapper) so they stay sargable on idx_tasks_due / start_time.
34 50 #[derive(sqlx::FromRow)]
35 51 struct StatsRow {
36 52 tasks_due_today: i64,
@@ -45,22 +61,25 @@ impl StatsRepository for SqliteStatsRepository {
45 61 "SELECT \
46 62 (SELECT COUNT(*) FROM tasks WHERE user_id = ?1 \
47 63 AND status NOT IN ('Completed', 'Deleted') \
48 - AND due IS NOT NULL AND date(due) = date('now')) AS tasks_due_today, \
64 + AND due IS NOT NULL AND due >= ?3 AND due < ?4) AS tasks_due_today, \
49 65 (SELECT COUNT(*) FROM tasks WHERE user_id = ?1 \
50 66 AND status NOT IN ('Completed', 'Deleted') \
51 - AND due IS NOT NULL AND date(due) >= date('now') \
52 - AND date(due) <= date('now', '+7 days')) AS tasks_due_this_week, \
67 + AND due IS NOT NULL AND due >= ?3 AND due < ?5) AS tasks_due_this_week, \
53 68 (SELECT COUNT(*) FROM tasks WHERE user_id = ?1 \
54 69 AND status NOT IN ('Completed', 'Deleted') \
55 - AND due IS NOT NULL AND datetime(due) < datetime('now')) AS overdue_count, \
70 + AND due IS NOT NULL AND due < ?2) AS overdue_count, \
56 71 (SELECT COUNT(*) FROM emails WHERE user_id = ?1 \
57 72 AND is_read = 0) AS unread_emails, \
58 73 (SELECT COUNT(*) FROM events WHERE user_id = ?1 \
59 - AND datetime(start_time) >= datetime('now') \
60 - AND datetime(start_time) <= datetime('now', '+7 days')) AS upcoming_events, \
74 + AND start_time >= ?2 AND start_time <= ?6) AS upcoming_events, \
61 75 (SELECT COUNT(*) FROM projects WHERE user_id = ?1 \
62 76 AND status = 'Active') AS active_projects")
63 - .bind(&user_id_str)
77 + .bind(&user_id_str) // ?1
78 + .bind(&now_str) // ?2
79 + .bind(&today_start_str) // ?3
80 + .bind(&tomorrow_start_str) // ?4
81 + .bind(&week_end_str) // ?5
82 + .bind(&events_end_str) // ?6
64 83 .fetch_one(&self.pool).await.map_err(CoreError::database)?;
65 84
66 85 // High-urgency tasks returns rows, so it remains a separate query.
M crates/db-sqlite/src/repository/subtask_repo.rs +8 -15
@@ -149,13 +149,15 @@ pub(crate) async fn toggle_subtask(
149 149 subtask_id: SubtaskId,
150 150 user_id: UserId,
151 151 ) -> Result<Option<Subtask>> {
152 - // First check if subtask exists and belongs to user's task
152 + // Flip atomically (1 - is_completed) and read back the new row in one
153 + // statement, scoped to the user's own tasks. Avoids the read-then-write
154 + // race where two concurrent toggles both read the old value and cancel out.
155 + // No matching row (missing or not owned) yields None.
153 156 let row = sqlx::query_as::<_, SubtaskRow>(
154 157 r#"
155 - SELECT s.id, s.task_id, s.text, s.linked_task_id, s.is_completed, s.position
156 - FROM subtasks s
157 - JOIN tasks t ON s.task_id = t.id
158 - WHERE s.id = ? AND t.user_id = ?
158 + UPDATE subtasks SET is_completed = 1 - is_completed
159 + WHERE id = ? AND task_id IN (SELECT id FROM tasks WHERE user_id = ?)
160 + RETURNING id, task_id, text, linked_task_id, is_completed, position
159 161 "#
160 162 )
161 163 .bind(subtask_id.to_string())
@@ -168,21 +170,12 @@ pub(crate) async fn toggle_subtask(
168 170 return Ok(None);
169 171 };
170 172
171 - let new_completed = if subtask.is_completed != 0 { 0 } else { 1 };
172 -
173 - sqlx::query("UPDATE subtasks SET is_completed = ? WHERE id = ?")
174 - .bind(new_completed)
175 - .bind(subtask_id.to_string())
176 - .execute(pool)
177 - .await
178 - .map_err(CoreError::database)?;
179 -
180 173 Ok(Some(Subtask {
181 174 id: parse_uuid(&subtask.id)?.into(),
182 175 task_id: parse_uuid(&subtask.task_id)?.into(),
183 176 text: subtask.text,
184 177 linked_task_id: parse_uuid_opt(subtask.linked_task_id.as_deref())?.map(Into::into),
185 - is_completed: new_completed != 0,
178 + is_completed: subtask.is_completed != 0,
186 179 position: subtask.position,
187 180 }))
188 181 }
M crates/db-sqlite/src/repository/task_repo_state.rs +12 -9
@@ -79,7 +79,7 @@ pub(crate) async fn list_snoozed(
79 79 user_id: UserId,
80 80 ) -> Result<Vec<Task>> {
81 81 let sql = format!(
82 - "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.user_id = ? AND t.status != 'Deleted' AND t.snoozed_until IS NOT NULL AND datetime(t.snoozed_until) > datetime('now') ORDER BY t.snoozed_until ASC",
82 + "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.user_id = ? AND t.status != 'Deleted' AND t.snoozed_until IS NOT NULL AND t.snoozed_until > datetime('now') ORDER BY t.snoozed_until ASC",
83 83 TASK_SELECT_COLUMNS
84 84 );
85 85 query_tasks(pool, &sql, &[user_id.to_string(), user_id.to_string()]).await
@@ -161,7 +161,7 @@ pub(crate) async fn list_scheduled_for_date(
161 161 let date_end = format!("{} 23:59:59", date);
162 162
163 163 let sql = format!(
164 - "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.user_id = ? AND t.status != 'Deleted' AND t.status != 'Completed' AND t.scheduled_start IS NOT NULL AND datetime(t.scheduled_start) >= datetime(?) AND datetime(t.scheduled_start) <= datetime(?) ORDER BY t.scheduled_start ASC",
164 + "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.user_id = ? AND t.status != 'Deleted' AND t.status != 'Completed' AND t.scheduled_start IS NOT NULL AND t.scheduled_start >= ? AND t.scheduled_start <= ? ORDER BY t.scheduled_start ASC",
165 165 TASK_SELECT_COLUMNS
166 166 );
167 167 query_tasks(pool, &sql, &[user_id.to_string(), user_id.to_string(), date_start, date_end]).await
@@ -177,7 +177,7 @@ pub(crate) async fn list_unscheduled_due_on_date(
177 177 let date_end = format!("{} 23:59:59", date);
178 178
179 179 let sql = format!(
180 - "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.user_id = ? AND t.status != 'Deleted' AND t.status != 'Completed' AND t.scheduled_start IS NULL AND t.due IS NOT NULL AND datetime(t.due) >= datetime(?) AND datetime(t.due) <= datetime(?) ORDER BY t.urgency DESC, t.due ASC",
180 + "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.user_id = ? AND t.status != 'Deleted' AND t.status != 'Completed' AND t.scheduled_start IS NULL AND t.due IS NOT NULL AND t.due >= ? AND t.due <= ? ORDER BY t.urgency DESC, t.due ASC",
181 181 TASK_SELECT_COLUMNS
182 182 );
183 183 query_tasks(pool, &sql, &[user_id.to_string(), user_id.to_string(), date_start, date_end]).await
@@ -196,7 +196,7 @@ pub(crate) async fn list_unscheduled_due_between(
196 196 let end_str = format_datetime(&end);
197 197
198 198 let sql = format!(
199 - "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.user_id = ? AND t.status != 'Deleted' AND t.status != 'Completed' AND t.scheduled_start IS NULL AND t.due IS NOT NULL AND datetime(t.due) >= datetime(?) AND datetime(t.due) <= datetime(?) ORDER BY t.urgency DESC, t.due ASC",
199 + "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.user_id = ? AND t.status != 'Deleted' AND t.status != 'Completed' AND t.scheduled_start IS NULL AND t.due IS NOT NULL AND t.due >= ? AND t.due <= ? ORDER BY t.urgency DESC, t.due ASC",
200 200 TASK_SELECT_COLUMNS
201 201 );
202 202 query_tasks(pool, &sql, &[user_id.to_string(), user_id.to_string(), start_str, end_str]).await
@@ -304,7 +304,7 @@ pub(crate) async fn list_completed_between(
304 304 let end_str = format_datetime(&end);
305 305
306 306 let sql = format!(
307 - "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.user_id = ? AND t.status = 'Completed' AND t.completed_at IS NOT NULL AND datetime(t.completed_at) >= datetime(?) AND datetime(t.completed_at) <= datetime(?) ORDER BY t.completed_at DESC",
307 + "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.user_id = ? AND t.status = 'Completed' AND t.completed_at IS NOT NULL AND t.completed_at >= ? AND t.completed_at <= ? ORDER BY t.completed_at DESC",
308 308 TASK_SELECT_COLUMNS
309 309 );
310 310 query_tasks(pool, &sql, &[user_id.to_string(), user_id.to_string(), start_str, end_str]).await
@@ -321,7 +321,7 @@ pub(crate) async fn list_created_between(
321 321 let end_str = format_datetime(&end);
322 322
323 323 let sql = format!(
324 - "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.user_id = ? AND t.status != 'Deleted' AND datetime(t.created_at) >= datetime(?) AND datetime(t.created_at) <= datetime(?) ORDER BY t.created_at DESC",
324 + "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.user_id = ? AND t.status != 'Deleted' AND t.created_at >= ? AND t.created_at <= ? ORDER BY t.created_at DESC",
325 325 TASK_SELECT_COLUMNS
326 326 );
327 327 query_tasks(pool, &sql, &[user_id.to_string(), user_id.to_string(), start_str, end_str]).await
@@ -339,7 +339,7 @@ pub(crate) async fn list_became_overdue_between(
339 339
340 340 // Tasks whose due date is in the given range and are still pending/started (overdue)
341 341 let sql = format!(
342 - "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.user_id = ? AND t.status NOT IN ('Completed', 'Deleted') AND t.due IS NOT NULL AND datetime(t.due) >= datetime(?) AND datetime(t.due) <= datetime(?) ORDER BY t.due ASC",
342 + "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.user_id = ? AND t.status NOT IN ('Completed', 'Deleted') AND t.due IS NOT NULL AND t.due >= ? AND t.due <= ? ORDER BY t.due ASC",
343 343 TASK_SELECT_COLUMNS
344 344 );
345 345 query_tasks(pool, &sql, &[user_id.to_string(), user_id.to_string(), start_str, end_str]).await
@@ -356,7 +356,7 @@ pub(crate) async fn list_due_between(
356 356 let end_str = format_datetime(&end);
357 357
358 358 let sql = format!(
359 - "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.user_id = ? AND t.status NOT IN ('Completed', 'Deleted') AND t.due IS NOT NULL AND datetime(t.due) >= datetime(?) AND datetime(t.due) <= datetime(?) ORDER BY t.due ASC, t.urgency DESC",
359 + "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.user_id = ? AND t.status NOT IN ('Completed', 'Deleted') AND t.due IS NOT NULL AND t.due >= ? AND t.due <= ? ORDER BY t.due ASC, t.urgency DESC",
360 360 TASK_SELECT_COLUMNS
361 361 );
362 362 query_tasks(pool, &sql, &[user_id.to_string(), user_id.to_string(), start_str, end_str]).await
@@ -368,8 +368,11 @@ pub(crate) async fn list_available_for_focus(
368 368 user_id: UserId,
369 369 limit: i64,
370 370 ) -> Result<Vec<Task>> {
371 + // Clamp the caller-supplied limit: a negative value would otherwise mean
372 + // "unbounded" to SQLite and a huge value an unbounded allocation.
373 + let limit = limit.clamp(0, 1000);
371 374 let sql = format!(
372 - "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.user_id = ? AND t.status NOT IN ('Completed', 'Deleted') AND t.is_focus = 0 AND (t.snoozed_until IS NULL OR datetime(t.snoozed_until) <= datetime('now')) AND t.waiting_for_response = 0 ORDER BY t.urgency DESC, t.priority DESC, t.due ASC NULLS LAST LIMIT ?",
375 + "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.user_id = ? AND t.status NOT IN ('Completed', 'Deleted') AND t.is_focus = 0 AND (t.snoozed_until IS NULL OR t.snoozed_until <= datetime('now')) AND t.waiting_for_response = 0 ORDER BY t.urgency DESC, t.priority DESC, t.due ASC NULLS LAST LIMIT ?",
373 376 TASK_SELECT_COLUMNS
374 377 );
375 378
M crates/db-sqlite/src/repository/weekly_review_repo.rs +39 -76
@@ -110,50 +110,29 @@ impl WeeklyReviewRepository for SqliteWeeklyReviewRepository {
110 110 let now = Utc::now();
111 111 let completed_at_str = format_datetime(&now);
112 112
113 - // Try to get existing review
114 - let existing = self.get_for_week(user_id, week_start).await?;
115 - let existing_vacation_days = existing.as_ref().map(|r| r.vacation_days.clone()).unwrap_or_default();
116 -
117 - let id = if let Some(existing) = existing {
118 - // Update existing
119 - sqlx::query(
120 - "UPDATE weekly_reviews SET notes = ?, completed_at = ? WHERE id = ?"
121 - )
122 - .bind(notes)
123 - .bind(&completed_at_str)
124 - .bind(existing.id.to_string())
125 - .execute(&self.pool)
126 - .await
127 - .map_err(CoreError::database)?;
128 -
129 - existing.id
130 - } else {
131 - // Insert new
132 - let id = WeeklyReviewId::new();
133 - sqlx::query(
134 - "INSERT INTO weekly_reviews (id, user_id, week_start_date, completed_at, notes)
135 - VALUES (?, ?, ?, ?, ?)"
136 - )
137 - .bind(id.to_string())
138 - .bind(&user_id_str)
139 - .bind(&week_start_str)
140 - .bind(&completed_at_str)
141 - .bind(notes)
142 - .execute(&self.pool)
143 - .await
144 - .map_err(CoreError::database)?;
145 -
146 - id
147 - };
113 + // Atomic upsert on the (user_id, week_start_date) unique key. Avoids the
114 + // get-then-insert/update race that could double-insert under a
115 + // concurrent first write (sync apply + a UI write). vacation_days and
116 + // the original id are preserved by only updating notes + completed_at.
117 + sqlx::query(
118 + "INSERT INTO weekly_reviews (id, user_id, week_start_date, completed_at, notes)
119 + VALUES (?, ?, ?, ?, ?)
120 + ON CONFLICT(user_id, week_start_date)
121 + DO UPDATE SET notes = excluded.notes, completed_at = excluded.completed_at",
122 + )
123 + .bind(WeeklyReviewId::new().to_string())
124 + .bind(&user_id_str)
125 + .bind(&week_start_str)
126 + .bind(&completed_at_str)
127 + .bind(notes)
128 + .execute(&self.pool)
129 + .await
130 + .map_err(CoreError::database)?;
148 131
149 - Ok(WeeklyReview {
150 - id,
151 - user_id,
152 - week_start_date: week_start,
153 - completed_at: now,
154 - notes: notes.to_string(),
155 - vacation_days: existing_vacation_days,
156 - })
132 + // Re-read to return the persisted row (its id may predate this call).
133 + self.get_for_week(user_id, week_start)
134 + .await?
135 + .ok_or_else(|| CoreError::database_msg("weekly review vanished after upsert"))
157 136 }
158 137
159 138 #[tracing::instrument(skip_all)]
@@ -168,39 +147,23 @@ impl WeeklyReviewRepository for SqliteWeeklyReviewRepository {
168 147 let user_id_str = user_id.to_string();
169 148 let week_start_str = week_start.format("%Y-%m-%d").to_string();
170 149 let vacation_str = serialize_vacation_days(days);
171 -
172 - // Check if a review row already exists for this week
173 - let existing = self.get_for_week(user_id, week_start).await?;
174 -
175 - if existing.is_some() {
176 - // Update existing row
177 - sqlx::query(
178 - "UPDATE weekly_reviews SET vacation_days = ? WHERE user_id = ? AND week_start_date = ?"
179 - )
180 - .bind(&vacation_str)
181 - .bind(&user_id_str)
182 - .bind(&week_start_str)
183 - .execute(&self.pool)
184 - .await
185 - .map_err(CoreError::database)?;
186 - } else {
187 - // Insert a new row (review not yet completed, just vacation days)
188 - let id = WeeklyReviewId::new();
189 - let now = format_datetime(&Utc::now());
190 -
191 - sqlx::query(
192 - "INSERT INTO weekly_reviews (id, user_id, week_start_date, completed_at, notes, vacation_days)
193 - VALUES (?, ?, ?, ?, '', ?)"
194 - )
195 - .bind(id.to_string())
196 - .bind(&user_id_str)
197 - .bind(&week_start_str)
198 - .bind(&now)
199 - .bind(&vacation_str)
200 - .execute(&self.pool)
201 - .await
202 - .map_err(CoreError::database)?;
203 - }
150 + let now = format_datetime(&Utc::now());
151 +
152 + // Atomic upsert: insert a vacation-only row or update the existing
153 + // week's vacation_days, without the get-then-write race.
154 + sqlx::query(
155 + "INSERT INTO weekly_reviews (id, user_id, week_start_date, completed_at, notes, vacation_days)
156 + VALUES (?, ?, ?, ?, '', ?)
157 + ON CONFLICT(user_id, week_start_date) DO UPDATE SET vacation_days = excluded.vacation_days",
158 + )
159 + .bind(WeeklyReviewId::new().to_string())
160 + .bind(&user_id_str)
161 + .bind(&week_start_str)
162 + .bind(&now)
163 + .bind(&vacation_str)
164 + .execute(&self.pool)
165 + .await
166 + .map_err(CoreError::database)?;
204 167
205 168 Ok(())
206 169 }
M src-tauri/src/commands/stats.rs +18 -1
@@ -73,6 +73,23 @@ impl From<DashboardStats> for DashboardStatsResponse {
73 73 #[tauri::command]
74 74 #[instrument(skip_all)]
75 75 pub async fn get_dashboard_stats(state: State<'_, Arc<AppState>>) -> Result<DashboardStatsResponse, ApiError> {
76 - let stats = state.stats.get_dashboard_stats(DESKTOP_USER_ID).await?;
76 + // Compute the "today"/"this week" windows from the user's *local* calendar
77 + // day so due-date counts match what the user sees, then hand them to the
78 + // repo as UTC instants. "This week" preserves the prior inclusive
79 + // today..today+7 span (an 8-day exclusive upper bound).
80 + let now = chrono::Utc::now();
81 + let tz = crate::tz::system_tz();
82 + let local_today = now.with_timezone(&tz).date_naive();
83 + let midnight = |d: chrono::NaiveDate| {
84 + crate::tz::local_civil_to_utc(d.and_hms_opt(0, 0, 0).expect("valid midnight"))
85 + };
86 + let today_start = midnight(local_today);
87 + let tomorrow_start = midnight(local_today + chrono::Duration::days(1));
88 + let week_end = midnight(local_today + chrono::Duration::days(8));
89 +
90 + let stats = state
91 + .stats
92 + .get_dashboard_stats(DESKTOP_USER_ID, now, today_start, tomorrow_start, week_end)
93 + .await?;
77 94 Ok(DashboardStatsResponse::from(stats))
78 95 }
M src-tauri/src/sync_service/apply.rs +66 -8
@@ -16,7 +16,7 @@
16 16 //! `apply_delete` return an error if `table_columns()` returns `None`.
17 17
18 18 use goingson_core::CoreError;
19 - use sqlx::SqliteConnection;
19 + use sqlx::{Row, SqliteConnection};
20 20
21 21 use super::EMAIL_ACCOUNT_SYNC_COLS;
22 22
@@ -104,14 +104,21 @@ pub(crate) fn table_columns(table: &str) -> Option<&'static [&'static str]> {
104 104 .map(|(_, cols)| *cols)
105 105 }
106 106
107 - /// Apply an INSERT OR REPLACE for a remote change.
107 + /// Apply an upsert for a remote change.
108 + ///
109 + /// Uses `INSERT ... ON CONFLICT(id) DO UPDATE` rather than `INSERT OR REPLACE`.
110 + /// `INSERT OR REPLACE` deletes the conflicting row before reinserting it, which
111 + /// fires `ON DELETE CASCADE` on children (annotations, subtasks, contact_*) and
112 + /// would wipe them on any parent re-upsert. `ON CONFLICT DO UPDATE` mutates the
113 + /// row in place, so children survive and referential integrity holds regardless
114 + /// of FK enforcement. Every syncable table has `id` as its primary key.
108 115 pub(crate) async fn apply_upsert(
109 116 conn: &mut SqliteConnection,
110 117 table: &str,
111 118 _row_id: &str,
112 119 data: &serde_json::Value,
113 120 ) -> Result<(), CoreError> {
114 - // Email accounts use ON CONFLICT to preserve local credentials
121 + // Email accounts use a bespoke ON CONFLICT that preserves local credentials
115 122 if table == "email_accounts" {
116 123 return apply_email_account_upsert(conn, data).await;
117 124 }
@@ -119,16 +126,42 @@ pub(crate) async fn apply_upsert(
119 126 let columns = table_columns(table)
120 127 .ok_or_else(|| CoreError::bad_request(format!("unknown syncable table: {}", table)))?;
121 128
122 - let col_list = columns.join(", ");
123 - let placeholders = columns.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
129 + // A remote null for a NOT NULL column is invalid input the changelog never
130 + // emits. INSERT OR REPLACE silently coerced it to the column default; to
131 + // keep that tolerance under ON CONFLICT, omit such columns: a new row then
132 + // takes the schema default and an existing row keeps its current value.
133 + // Nullable columns keep null (so legitimate clears — e.g. un-snoozing —
134 + // still propagate). NOT NULL columns are derived from the live schema, not
135 + // a hand-maintained list, so they can't drift.
136 + let not_null = not_null_columns(conn, table).await?;
137 + let included: Vec<&str> = columns
138 + .iter()
139 + .copied()
140 + .filter(|c| !(data[*c].is_null() && not_null.contains(*c)))
141 + .collect();
142 +
143 + let col_list = included.join(", ");
144 + let placeholders = included.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
145 + let update_set = included
146 + .iter()
147 + .filter(|c| **c != "id")
148 + .map(|c| format!("{} = excluded.{}", c, c))
149 + .collect::<Vec<_>>()
150 + .join(", ");
151 + // If only `id` survived, there is nothing to update on conflict.
152 + let conflict = if update_set.is_empty() {
153 + "DO NOTHING".to_string()
154 + } else {
155 + format!("DO UPDATE SET {}", update_set)
156 + };
124 157 let sql = format!(
125 - "INSERT OR REPLACE INTO {} ({}) VALUES ({})",
126 - table, col_list, placeholders
158 + "INSERT INTO {} ({}) VALUES ({}) ON CONFLICT(id) {}",
159 + table, col_list, placeholders, conflict
127 160 );
128 161
129 162 let mut query = sqlx::query(&sql);
130 163
131 - for col in columns {
164 + for col in &included {
132 165 query = bind_json_value(query, &data[*col]);
133 166 }
134 167
@@ -140,6 +173,31 @@ pub(crate) async fn apply_upsert(
140 173 Ok(())
141 174 }
142 175
176 + /// Names of the NOT NULL columns of a table, read from the live schema.
177 + ///
178 + /// Derived via `PRAGMA table_info` rather than a static list so it can never
179 + /// drift from the migrations. `table` is already validated against the column
180 + /// whitelist by the caller, so interpolating it is safe.
181 + async fn not_null_columns(
182 + conn: &mut SqliteConnection,
183 + table: &str,
184 + ) -> Result<std::collections::HashSet<String>, CoreError> {
185 + let rows = sqlx::query(&format!("PRAGMA table_info({})", table))
186 + .fetch_all(&mut *conn)
187 + .await
188 + .map_err(CoreError::database)?;
189 +
190 + let mut set = std::collections::HashSet::new();
191 + for row in rows {
192 + let notnull: i64 = row.try_get("notnull").map_err(CoreError::database)?;
193 + if notnull != 0 {
194 + let name: String = row.try_get("name").map_err(CoreError::database)?;
195 + set.insert(name);
196 + }
197 + }
198 + Ok(set)
199 + }
200 +
143 201 /// Apply an upsert for email_accounts that preserves local credentials.
144 202 ///
145 203 /// Uses INSERT ... ON CONFLICT(id) DO UPDATE to only touch the 16 config columns,
M src-tauri/src/sync_service/tests.rs +53
@@ -238,6 +238,59 @@ use sqlx::SqlitePool;
238 238 }
239 239
240 240 #[tokio::test]
241 + async fn apply_upsert_parent_reupsert_preserves_children() {
242 + // Re-upserting a parent task must not cascade-delete its annotations.
243 + // INSERT OR REPLACE would; ON CONFLICT DO UPDATE must not. Enforce FKs
244 + // so the cascade would actually fire if the row were deleted.
245 + let pool = setup_test_db().await;
246 + let user_id = create_test_user(&pool).await;
247 + let task_id = uuid::Uuid::new_v4().to_string();
248 + let annotation_id = uuid::Uuid::new_v4().to_string();
249 + let now = now_sql();
250 +
251 + let task = |status: &str| {
252 + json!({
253 + "id": task_id, "project_id": null, "description": "Parent",
254 + "status": status, "priority": "Medium", "due": null, "tags": "",
255 + "urgency": 0.0, "recurrence": "None", "recurrence_rule": null,
256 + "created_at": now, "user_id": user_id, "recurrence_parent_id": null,
257 + "source_email_id": null, "snoozed_until": null, "waiting_for_response": 0,
258 + "waiting_since": null, "expected_response_date": null, "scheduled_start": null,
259 + "scheduled_duration": null, "is_focus": 0, "focus_set_at": null,
260 + "contact_id": null, "milestone_id": null, "completed_at": null,
261 + "estimated_minutes": null, "actual_minutes": null,
262 + })
263 + };
264 +
265 + let mut conn = pool.acquire().await.unwrap();
266 + sqlx::query("PRAGMA foreign_keys = ON").execute(&mut *conn).await.unwrap();
267 +
268 + apply::apply_upsert(&mut conn, "tasks", &task_id, &task("Active")).await.unwrap();
269 +
270 + let annotation = json!({
271 + "id": annotation_id, "task_id": task_id, "timestamp": now, "note": "child note",
272 + });
273 + apply::apply_upsert(&mut conn, "annotations", &annotation_id, &annotation).await.unwrap();
274 +
275 + // Re-upsert the parent (e.g. a remote status change).
276 + apply::apply_upsert(&mut conn, "tasks", &task_id, &task("Completed")).await.unwrap();
277 +
278 + let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM annotations WHERE task_id = ?")
279 + .bind(&task_id)
280 + .fetch_one(&pool)
281 + .await
282 + .unwrap();
283 + assert_eq!(count.0, 1, "annotation should survive parent re-upsert");
284 +
285 + let status: (String,) = sqlx::query_as("SELECT status FROM tasks WHERE id = ?")
286 + .bind(&task_id)
287 + .fetch_one(&pool)
288 + .await
289 + .unwrap();
290 + assert_eq!(status.0, "Completed");
291 + }
292 +
293 + #[tokio::test]
241 294 async fn apply_upsert_rejects_unknown_table() {
242 295 let pool = setup_test_db().await;
243 296 let data = json!({"id": "abc"});