Skip to main content

max / goingson

84.3 KB · 2100 lines History Blame Raw
1 use crate::sync_service::*;
2 use crate::sync_service::{apply, pull, UPSERT_ORDER, DELETE_ORDER};
3 use goingson_db_sqlite::{init_pool, run_migrations};
4 use serde_json::json;
5 use sqlx::SqlitePool;
6
7 /// Format the current UTC time as a SQL-compatible timestamp string.
8 fn now_sql() -> String {
9 chrono::Utc::now().format("%Y-%m-%d %H:%M:%S").to_string()
10 }
11
12 /// Creates an in-memory test database with all migrations applied.
13 async fn setup_test_db() -> SqlitePool {
14 let pool = init_pool(Some(":memory:"))
15 .await
16 .expect("Failed to create in-memory pool");
17 run_migrations(&pool)
18 .await
19 .expect("Failed to run migrations");
20 pool
21 }
22
23 /// Creates a test user and returns their UUID string.
24 async fn create_test_user(pool: &SqlitePool) -> String {
25 let user_id = uuid::Uuid::new_v4().to_string();
26 let now = now_sql();
27 sqlx::query(
28 "INSERT INTO users (id, email, password_hash, display_name, created_at) VALUES (?, ?, ?, ?, ?)"
29 )
30 .bind(&user_id)
31 .bind(format!("test-{}@example.com", user_id))
32 .bind("test-password-hash")
33 .bind("Test User")
34 .bind(&now)
35 .execute(pool)
36 .await
37 .expect("Failed to create test user");
38 user_id
39 }
40
41 /// Helper: build a ChangeEntry for tests.
42 fn change(table: &str, op: synckit_client::ChangeOp, row_id: &str, data: Option<serde_json::Value>) -> synckit_client::ChangeEntry {
43 synckit_client::ChangeEntry {
44 table: table.to_string(),
45 op,
46 row_id: row_id.to_string(),
47 timestamp: chrono::Utc::now(),
48 hlc: synckit_client::Hlc::zero(synckit_client::DeviceId::nil()),
49 data,
50 extra: Default::default(),
51 }
52 }
53
54 // -- FK ordering tests --
55
56 #[test]
57 fn upsert_order_has_parents_before_children() {
58 let pos = |table: &str| UPSERT_ORDER.iter().position(|t| *t == table);
59
60 // projects before tasks, events, milestones
61 assert!(pos("projects").unwrap() < pos("tasks").unwrap());
62 assert!(pos("projects").unwrap() < pos("events").unwrap());
63 assert!(pos("projects").unwrap() < pos("milestones").unwrap());
64
65 // contacts before contact_emails, contact_phones, etc.
66 assert!(pos("contacts").unwrap() < pos("contact_emails").unwrap());
67 assert!(pos("contacts").unwrap() < pos("contact_phones").unwrap());
68 assert!(pos("contacts").unwrap() < pos("contact_social_handles").unwrap());
69 assert!(pos("contacts").unwrap() < pos("contact_custom_fields").unwrap());
70
71 // tasks before annotations, subtasks
72 assert!(pos("tasks").unwrap() < pos("annotations").unwrap());
73 assert!(pos("tasks").unwrap() < pos("subtasks").unwrap());
74
75 // milestones before tasks (tasks.milestone_id FK)
76 assert!(pos("milestones").unwrap() < pos("tasks").unwrap());
77 }
78
79 #[test]
80 fn delete_order_has_children_before_parents() {
81 let pos = |table: &str| DELETE_ORDER.iter().position(|t| *t == table);
82
83 // children before parents (reverse of upsert)
84 assert!(pos("tasks").unwrap() < pos("projects").unwrap());
85 assert!(pos("events").unwrap() < pos("projects").unwrap());
86 assert!(pos("milestones").unwrap() < pos("projects").unwrap());
87
88 assert!(pos("contact_emails").unwrap() < pos("contacts").unwrap());
89 assert!(pos("contact_phones").unwrap() < pos("contacts").unwrap());
90 assert!(pos("contact_social_handles").unwrap() < pos("contacts").unwrap());
91 assert!(pos("contact_custom_fields").unwrap() < pos("contacts").unwrap());
92
93 assert!(pos("annotations").unwrap() < pos("tasks").unwrap());
94 assert!(pos("subtasks").unwrap() < pos("tasks").unwrap());
95
96 assert!(pos("tasks").unwrap() < pos("milestones").unwrap());
97 }
98
99 #[test]
100 fn upsert_and_delete_orders_are_exact_reverses() {
101 let reversed: Vec<&str> = UPSERT_ORDER.iter().rev().copied().collect();
102 assert_eq!(reversed, DELETE_ORDER);
103 }
104
105 #[test]
106 fn all_upsert_tables_have_column_whitelists() {
107 for table in UPSERT_ORDER {
108 assert!(
109 apply::table_columns(table).is_some(),
110 "table_columns missing for syncable table: {}",
111 table
112 );
113 }
114 }
115
116 #[test]
117 fn unknown_table_returns_none() {
118 assert!(apply::table_columns("nonexistent").is_none());
119 assert!(apply::table_columns("users").is_none());
120 assert!(apply::table_columns("emails").is_none());
121 }
122
123 #[test]
124 fn every_table_whitelist_starts_with_id() {
125 for table in UPSERT_ORDER {
126 let cols = apply::table_columns(table).unwrap();
127 assert_eq!(
128 cols[0], "id",
129 "table {} column whitelist should start with 'id'",
130 table
131 );
132 }
133 }
134
135 // -- sync_state helpers --
136
137 #[tokio::test]
138 async fn get_sync_state_returns_empty_for_missing_key() {
139 let pool = setup_test_db().await;
140 let val = get_sync_state(&pool, "nonexistent_key").await.unwrap();
141 assert_eq!(val, "");
142 }
143
144 #[tokio::test]
145 async fn set_and_get_sync_state() {
146 let pool = setup_test_db().await;
147 set_sync_state(&pool, "test_key", "test_value").await.unwrap();
148 let val = get_sync_state(&pool, "test_key").await.unwrap();
149 assert_eq!(val, "test_value");
150 }
151
152 #[tokio::test]
153 async fn set_sync_state_overwrites_existing() {
154 let pool = setup_test_db().await;
155 set_sync_state(&pool, "key", "first").await.unwrap();
156 set_sync_state(&pool, "key", "second").await.unwrap();
157 let val = get_sync_state(&pool, "key").await.unwrap();
158 assert_eq!(val, "second");
159 }
160
161 #[tokio::test]
162 async fn migration_seeds_default_sync_state() {
163 let pool = setup_test_db().await;
164
165 assert_eq!(get_sync_state(&pool, "device_id").await.unwrap(), "");
166 assert_eq!(get_sync_state(&pool, "pull_cursor").await.unwrap(), "0");
167 assert_eq!(get_sync_state(&pool, "applying_remote").await.unwrap(), "0");
168 assert_eq!(get_sync_state(&pool, "initial_snapshot_done").await.unwrap(), "0");
169 }
170
171 // -- committed HLC store (CleanChanges gate backing) --
172
173 #[tokio::test]
174 async fn committed_hlc_records_keeps_max_and_loads() {
175 use crate::sync_service::hlc::{load_committed_hlcs, record_committed_hlc};
176 use synckit_client::Hlc;
177
178 let pool = setup_test_db().await;
179 let node = uuid::Uuid::new_v4();
180 let lo = Hlc { wall_ms: 100, counter: 0, node: synckit_client::DeviceId::new(node) };
181 let hi = Hlc { wall_ms: 200, counter: 0, node: synckit_client::DeviceId::new(node) };
182 let key = ("tasks".to_string(), "r1".to_string());
183
184 record_committed_hlc(&pool, "tasks", "r1", lo).await.unwrap();
185 let m = load_committed_hlcs(&pool, std::slice::from_ref(&key)).await.unwrap();
186 assert_eq!(m.get(&key).copied(), Some(lo));
187
188 // A higher HLC overwrites; a subsequent lower one does NOT (max-keeping).
189 record_committed_hlc(&pool, "tasks", "r1", hi).await.unwrap();
190 record_committed_hlc(&pool, "tasks", "r1", lo).await.unwrap();
191 let m = load_committed_hlcs(&pool, std::slice::from_ref(&key)).await.unwrap();
192 assert_eq!(m.get(&key).copied(), Some(hi), "older HLC must not overwrite newer committed clock");
193
194 // A row never recorded is absent (the gate then keeps that change).
195 let missing = ("tasks".to_string(), "nope".to_string());
196 let m = load_committed_hlcs(&pool, std::slice::from_ref(&missing)).await.unwrap();
197 assert!(m.is_empty());
198 }
199
200 // (The clean-change gate itself — detect_conflicts -> CleanChanges::gated —
201 // is exercised end-to-end in synckit-client's own conflict tests, which can
202 // construct the non-exhaustive PulledChange internally. Here we only cover
203 // GO's committed-clock store that backs the gate.)
204
205 // -- apply_upsert --
206
207 #[tokio::test]
208 async fn apply_upsert_inserts_project() {
209 let pool = setup_test_db().await;
210 let user_id = create_test_user(&pool).await;
211 let project_id = uuid::Uuid::new_v4().to_string();
212 let now = now_sql();
213
214 let data = json!({
215 "id": project_id,
216 "name": "Synced Project",
217 "description": "From remote",
218 "project_type": "SideProject",
219 "status": "Active",
220 "created_at": now,
221 "user_id": user_id,
222 });
223
224 let mut conn = pool.acquire().await.unwrap();
225 apply::apply_upsert(&mut conn, "projects", &project_id, &data).await.unwrap();
226
227 let row: (String,) = sqlx::query_as("SELECT name FROM projects WHERE id = ?")
228 .bind(&project_id)
229 .fetch_one(&pool)
230 .await
231 .unwrap();
232 assert_eq!(row.0, "Synced Project");
233 }
234
235 #[tokio::test]
236 async fn apply_upsert_replaces_existing_row() {
237 let pool = setup_test_db().await;
238 let user_id = create_test_user(&pool).await;
239 let project_id = uuid::Uuid::new_v4().to_string();
240 let now = now_sql();
241
242 let data1 = json!({
243 "id": project_id,
244 "name": "Original",
245 "description": "",
246 "project_type": "Job",
247 "status": "Active",
248 "created_at": now,
249 "user_id": user_id,
250 });
251 let mut conn = pool.acquire().await.unwrap();
252 apply::apply_upsert(&mut conn, "projects", &project_id, &data1).await.unwrap();
253
254 let data2 = json!({
255 "id": project_id,
256 "name": "Updated",
257 "description": "changed",
258 "project_type": "Job",
259 "status": "OnHold",
260 "created_at": now,
261 "user_id": user_id,
262 });
263 apply::apply_upsert(&mut conn, "projects", &project_id, &data2).await.unwrap();
264
265 let row: (String, String) =
266 sqlx::query_as("SELECT name, status FROM projects WHERE id = ?")
267 .bind(&project_id)
268 .fetch_one(&pool)
269 .await
270 .unwrap();
271 assert_eq!(row.0, "Updated");
272 assert_eq!(row.1, "OnHold");
273 }
274
275 #[tokio::test]
276 async fn apply_upsert_parent_reupsert_preserves_children() {
277 // Re-upserting a parent task must not cascade-delete its annotations.
278 // INSERT OR REPLACE would; ON CONFLICT DO UPDATE must not. Enforce FKs
279 // so the cascade would actually fire if the row were deleted.
280 let pool = setup_test_db().await;
281 let user_id = create_test_user(&pool).await;
282 let task_id = uuid::Uuid::new_v4().to_string();
283 let annotation_id = uuid::Uuid::new_v4().to_string();
284 let now = now_sql();
285
286 let task = |status: &str| {
287 json!({
288 "id": task_id, "project_id": null, "description": "Parent",
289 "status": status, "priority": "Medium", "due": null, "tags": "",
290 "urgency": 0.0, "recurrence": "None", "recurrence_rule": null,
291 "created_at": now, "user_id": user_id, "recurrence_parent_id": null,
292 "source_email_id": null, "snoozed_until": null, "waiting_for_response": 0,
293 "waiting_since": null, "expected_response_date": null, "scheduled_start": null,
294 "scheduled_duration": null, "is_focus": 0, "focus_set_at": null,
295 "contact_id": null, "milestone_id": null, "completed_at": null,
296 "estimated_minutes": null, "actual_minutes": null,
297 })
298 };
299
300 let mut conn = pool.acquire().await.unwrap();
301 sqlx::query("PRAGMA foreign_keys = ON").execute(&mut *conn).await.unwrap();
302
303 apply::apply_upsert(&mut conn, "tasks", &task_id, &task("Active")).await.unwrap();
304
305 let annotation = json!({
306 "id": annotation_id, "task_id": task_id, "timestamp": now, "note": "child note",
307 });
308 apply::apply_upsert(&mut conn, "annotations", &annotation_id, &annotation).await.unwrap();
309
310 // Re-upsert the parent (e.g. a remote status change).
311 apply::apply_upsert(&mut conn, "tasks", &task_id, &task("Completed")).await.unwrap();
312
313 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM annotations WHERE task_id = ?")
314 .bind(&task_id)
315 .fetch_one(&pool)
316 .await
317 .unwrap();
318 assert_eq!(count.0, 1, "annotation should survive parent re-upsert");
319
320 let status: (String,) = sqlx::query_as("SELECT status FROM tasks WHERE id = ?")
321 .bind(&task_id)
322 .fetch_one(&pool)
323 .await
324 .unwrap();
325 assert_eq!(status.0, "Completed");
326 }
327
328 #[tokio::test]
329 async fn apply_upsert_rejects_unknown_table() {
330 let pool = setup_test_db().await;
331 let data = json!({"id": "abc"});
332 let mut conn = pool.acquire().await.unwrap();
333 let result = apply::apply_upsert(&mut conn, "nonexistent", "abc", &data).await;
334 assert!(result.is_err());
335 let err_msg = result.unwrap_err().to_string();
336 assert!(err_msg.contains("unknown syncable table"));
337 }
338
339 #[tokio::test]
340 async fn apply_upsert_handles_null_fields() {
341 let pool = setup_test_db().await;
342 let user_id = create_test_user(&pool).await;
343 let project_id = uuid::Uuid::new_v4().to_string();
344 let now = now_sql();
345
346 let mut conn = pool.acquire().await.unwrap();
347
348 // Insert a project to be a FK parent for the task
349 let data_project = json!({
350 "id": project_id,
351 "name": "Parent",
352 "description": "",
353 "project_type": "Job",
354 "status": "Active",
355 "created_at": now,
356 "user_id": user_id,
357 });
358 apply::apply_upsert(&mut conn, "projects", &project_id, &data_project).await.unwrap();
359
360 let task_id = uuid::Uuid::new_v4().to_string();
361 let data = json!({
362 "id": task_id,
363 "project_id": project_id,
364 "description": "Null test",
365 "status": "Pending",
366 "priority": "Medium",
367 "due": null,
368 "tags": null,
369 "urgency": 50,
370 "recurrence": "None",
371 "created_at": now,
372 "user_id": user_id,
373 "recurrence_parent_id": null,
374 "source_email_id": null,
375 "snoozed_until": null,
376 "waiting_for_response": false,
377 "waiting_since": null,
378 "expected_response_date": null,
379 "scheduled_start": null,
380 "scheduled_duration": null,
381 "is_focus": false,
382 "focus_set_at": null,
383 "contact_id": null,
384 "milestone_id": null,
385 });
386
387 apply::apply_upsert(&mut conn, "tasks", &task_id, &data).await.unwrap();
388
389 let row: (String,) = sqlx::query_as("SELECT description FROM tasks WHERE id = ?")
390 .bind(&task_id)
391 .fetch_one(&pool)
392 .await
393 .unwrap();
394 assert_eq!(row.0, "Null test");
395 }
396
397 // -- apply_delete --
398
399 #[tokio::test]
400 async fn apply_delete_removes_row() {
401 let pool = setup_test_db().await;
402 let user_id = create_test_user(&pool).await;
403 let project_id = uuid::Uuid::new_v4().to_string();
404 let now = now_sql();
405
406 let mut conn = pool.acquire().await.unwrap();
407
408 let data = json!({
409 "id": project_id,
410 "name": "To Delete",
411 "description": "",
412 "project_type": "Other",
413 "status": "Active",
414 "created_at": now,
415 "user_id": user_id,
416 });
417 apply::apply_upsert(&mut conn, "projects", &project_id, &data).await.unwrap();
418
419 apply::apply_delete(&mut conn, "projects", &project_id).await.unwrap();
420
421 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM projects WHERE id = ?")
422 .bind(&project_id)
423 .fetch_one(&pool)
424 .await
425 .unwrap();
426 assert_eq!(count.0, 0);
427 }
428
429 #[tokio::test]
430 async fn apply_delete_rejects_unknown_table() {
431 let pool = setup_test_db().await;
432 let mut conn = pool.acquire().await.unwrap();
433 let result = apply::apply_delete(&mut conn, "not_a_table", "abc").await;
434 assert!(result.is_err());
435 let err_msg = result.unwrap_err().to_string();
436 assert!(err_msg.contains("unknown syncable table"));
437 }
438
439 #[tokio::test]
440 async fn apply_delete_is_idempotent() {
441 let pool = setup_test_db().await;
442 let mut conn = pool.acquire().await.unwrap();
443 // Deleting a non-existent row should succeed silently.
444 let result = apply::apply_delete(&mut conn, "projects", "nonexistent-id").await;
445 assert!(result.is_ok());
446 }
447
448 // -- apply_changes_inner ordering --
449
450 #[tokio::test]
451 async fn apply_changes_inner_processes_upserts_in_fk_order() {
452 let pool = setup_test_db().await;
453 let user_id = create_test_user(&pool).await;
454 let project_id = uuid::Uuid::new_v4().to_string();
455 let task_id = uuid::Uuid::new_v4().to_string();
456 let now = now_sql();
457
458 // Provide changes in WRONG order (child before parent) to prove re-ordering works.
459 let changes = vec![
460 change("tasks", synckit_client::ChangeOp::Insert, &task_id, Some(json!({
461 "id": task_id,
462 "project_id": project_id,
463 "description": "Child task",
464 "status": "Pending",
465 "priority": "Low",
466 "due": null,
467 "tags": null,
468 "urgency": 10,
469 "recurrence": "None",
470 "created_at": now,
471 "user_id": user_id,
472 "recurrence_parent_id": null,
473 "source_email_id": null,
474 "snoozed_until": null,
475 "waiting_for_response": false,
476 "waiting_since": null,
477 "expected_response_date": null,
478 "scheduled_start": null,
479 "scheduled_duration": null,
480 "is_focus": false,
481 "focus_set_at": null,
482 "contact_id": null,
483 "milestone_id": null,
484 }))),
485 change("projects", synckit_client::ChangeOp::Insert, &project_id, Some(json!({
486 "id": project_id,
487 "name": "Parent project",
488 "description": "",
489 "project_type": "Job",
490 "status": "Active",
491 "created_at": now,
492 "user_id": user_id,
493 }))),
494 ];
495
496 // Should succeed despite task coming before project in the input --
497 // apply_changes_inner iterates UPSERT_ORDER so projects is applied first.
498 let mut conn = pool.acquire().await.unwrap();
499 pull::apply_changes_inner(&mut conn, changes).await.unwrap();
500
501 let project: (String,) = sqlx::query_as("SELECT name FROM projects WHERE id = ?")
502 .bind(&project_id)
503 .fetch_one(&pool)
504 .await
505 .unwrap();
506 assert_eq!(project.0, "Parent project");
507
508 let task: (String,) = sqlx::query_as("SELECT description FROM tasks WHERE id = ?")
509 .bind(&task_id)
510 .fetch_one(&pool)
511 .await
512 .unwrap();
513 assert_eq!(task.0, "Child task");
514 }
515
516 #[tokio::test]
517 async fn apply_changes_inner_processes_deletes_in_child_first_order() {
518 let pool = setup_test_db().await;
519 let user_id = create_test_user(&pool).await;
520 let project_id = uuid::Uuid::new_v4().to_string();
521 let task_id = uuid::Uuid::new_v4().to_string();
522 let subtask_id = uuid::Uuid::new_v4().to_string();
523 let now = now_sql();
524
525 let mut conn = pool.acquire().await.unwrap();
526
527 // Insert parent -> child -> grandchild
528 apply::apply_upsert(&mut conn, "projects", &project_id, &json!({
529 "id": project_id, "name": "P", "description": "",
530 "project_type": "Job", "status": "Active",
531 "created_at": now, "user_id": user_id,
532 })).await.unwrap();
533
534 apply::apply_upsert(&mut conn, "tasks", &task_id, &json!({
535 "id": task_id, "project_id": project_id, "description": "T",
536 "status": "Pending", "priority": "Low", "due": null,
537 "tags": null, "urgency": 10, "recurrence": "None",
538 "created_at": now, "user_id": user_id,
539 "recurrence_parent_id": null, "source_email_id": null,
540 "snoozed_until": null, "waiting_for_response": false,
541 "waiting_since": null, "expected_response_date": null,
542 "scheduled_start": null, "scheduled_duration": null,
543 "is_focus": false, "focus_set_at": null,
544 "contact_id": null, "milestone_id": null,
545 })).await.unwrap();
546
547 apply::apply_upsert(&mut conn, "subtasks", &subtask_id, &json!({
548 "id": subtask_id, "task_id": task_id, "text": "Sub",
549 "is_completed": false, "position": 0,
550 "created_at": now, "linked_task_id": null,
551 })).await.unwrap();
552
553 // Delete in WRONG input order (parent first) -- apply_changes_inner
554 // should re-order to subtask -> task -> project.
555 let changes = vec![
556 change("projects", synckit_client::ChangeOp::Delete, &project_id, None),
557 change("tasks", synckit_client::ChangeOp::Delete, &task_id, None),
558 change("subtasks", synckit_client::ChangeOp::Delete, &subtask_id, None),
559 ];
560
561 pull::apply_changes_inner(&mut conn, changes).await.unwrap();
562
563 let count: (i64,) =
564 sqlx::query_as("SELECT COUNT(*) FROM projects WHERE id = ?")
565 .bind(&project_id)
566 .fetch_one(&pool)
567 .await
568 .unwrap();
569 assert_eq!(count.0, 0);
570
571 let count: (i64,) =
572 sqlx::query_as("SELECT COUNT(*) FROM tasks WHERE id = ?")
573 .bind(&task_id)
574 .fetch_one(&pool)
575 .await
576 .unwrap();
577 assert_eq!(count.0, 0);
578 }
579
580 #[tokio::test]
581 async fn apply_changes_inner_handles_mixed_ops() {
582 let pool = setup_test_db().await;
583 let user_id = create_test_user(&pool).await;
584 let now = now_sql();
585
586 let mut conn = pool.acquire().await.unwrap();
587
588 // Create a project to delete later
589 let delete_id = uuid::Uuid::new_v4().to_string();
590 apply::apply_upsert(&mut conn, "projects", &delete_id, &json!({
591 "id": delete_id, "name": "Will Delete", "description": "",
592 "project_type": "Job", "status": "Active",
593 "created_at": now, "user_id": user_id,
594 })).await.unwrap();
595
596 // Mix upsert and delete in a single batch
597 let insert_id = uuid::Uuid::new_v4().to_string();
598 let changes = vec![
599 change("projects", synckit_client::ChangeOp::Delete, &delete_id, None),
600 change("projects", synckit_client::ChangeOp::Insert, &insert_id, Some(json!({
601 "id": insert_id, "name": "New One", "description": "",
602 "project_type": "Other", "status": "Active",
603 "created_at": now, "user_id": user_id,
604 }))),
605 ];
606
607 pull::apply_changes_inner(&mut conn, changes).await.unwrap();
608
609 // The old project was deleted
610 let count: (i64,) =
611 sqlx::query_as("SELECT COUNT(*) FROM projects WHERE id = ?")
612 .bind(&delete_id)
613 .fetch_one(&pool)
614 .await
615 .unwrap();
616 assert_eq!(count.0, 0);
617
618 // The new project was inserted
619 let row: (String,) = sqlx::query_as("SELECT name FROM projects WHERE id = ?")
620 .bind(&insert_id)
621 .fetch_one(&pool)
622 .await
623 .unwrap();
624 assert_eq!(row.0, "New One");
625 }
626
627 #[tokio::test]
628 async fn apply_changes_inner_skips_upsert_without_data() {
629 let pool = setup_test_db().await;
630
631 // INSERT with no data should be skipped (data is None)
632 let changes = vec![change("projects", synckit_client::ChangeOp::Insert, "abc", None)];
633 let mut conn = pool.acquire().await.unwrap();
634 let result = pull::apply_changes_inner(&mut conn, changes).await;
635 assert!(result.is_ok());
636
637 let count: (i64,) =
638 sqlx::query_as("SELECT COUNT(*) FROM projects WHERE id = ?")
639 .bind("abc")
640 .fetch_one(&pool)
641 .await
642 .unwrap();
643 assert_eq!(count.0, 0);
644 }
645
646 #[tokio::test]
647 async fn apply_changes_inner_skips_bad_entry_and_applies_the_rest() {
648 // One un-appliable remote row must not abort the whole batch (ultra-fuzz
649 // Run #27 Data S-1: a single bad row used to wedge the pull cursor forever).
650 let pool = setup_test_db().await;
651 let user_id = create_test_user(&pool).await;
652 let now = now_sql();
653 let good_id = uuid::Uuid::new_v4().to_string();
654 let bad_id = uuid::Uuid::new_v4().to_string();
655
656 let changes = vec![
657 // Malformed: `name` is NOT NULL but omitted -> bound NULL -> fails.
658 change("projects", synckit_client::ChangeOp::Insert, &bad_id, Some(json!({
659 "id": bad_id, "description": "", "project_type": "Job",
660 "status": "Active", "created_at": now, "user_id": user_id,
661 }))),
662 // Well-formed: must still be applied.
663 change("projects", synckit_client::ChangeOp::Insert, &good_id, Some(json!({
664 "id": good_id, "name": "Good project", "description": "",
665 "project_type": "Job", "status": "Active",
666 "created_at": now, "user_id": user_id,
667 }))),
668 ];
669
670 let mut conn = pool.acquire().await.unwrap();
671 let skipped = pull::apply_changes_inner(&mut conn, changes).await.unwrap();
672 assert_eq!(skipped, 1, "the malformed row is skipped, not fatal");
673
674 let good: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM projects WHERE id = ?")
675 .bind(&good_id)
676 .fetch_one(&pool)
677 .await
678 .unwrap();
679 assert_eq!(good.0, 1, "well-formed row in the same batch still applied");
680 let bad: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM projects WHERE id = ?")
681 .bind(&bad_id)
682 .fetch_one(&pool)
683 .await
684 .unwrap();
685 assert_eq!(bad.0, 0, "malformed row not applied");
686 }
687
688 // -- Trigger suppression --
689
690 #[tokio::test]
691 async fn apply_remote_changes_suppresses_triggers() {
692 let pool = setup_test_db().await;
693 let user_id = create_test_user(&pool).await;
694 let project_id = uuid::Uuid::new_v4().to_string();
695 let now = now_sql();
696
697 // Clear any existing changelog entries
698 sqlx::query("DELETE FROM sync_changelog")
699 .execute(&pool)
700 .await
701 .unwrap();
702
703 let changes = vec![
704 change("projects", synckit_client::ChangeOp::Insert, &project_id, Some(json!({
705 "id": project_id, "name": "Remote Project", "description": "",
706 "project_type": "Job", "status": "Active",
707 "created_at": now, "user_id": user_id,
708 }))),
709 ];
710
711 pull::apply_remote_changes(&pool, changes).await.unwrap();
712
713 // The project should exist
714 let row: (String,) = sqlx::query_as("SELECT name FROM projects WHERE id = ?")
715 .bind(&project_id)
716 .fetch_one(&pool)
717 .await
718 .unwrap();
719 assert_eq!(row.0, "Remote Project");
720
721 // But the changelog should be empty -- triggers were suppressed
722 let count: (i64,) =
723 sqlx::query_as("SELECT COUNT(*) FROM sync_changelog")
724 .fetch_one(&pool)
725 .await
726 .unwrap();
727 assert_eq!(count.0, 0, "trigger should not fire during remote apply");
728
729 // The applying_remote flag should be cleared
730 let flag = get_sync_state(&pool, "applying_remote").await.unwrap();
731 assert_eq!(flag, "0", "applying_remote should be reset after apply");
732 }
733
734 #[tokio::test]
735 async fn apply_remote_changes_clears_flag_after_success() {
736 let pool = setup_test_db().await;
737
738 // Verify the flag is set to "1" during apply and cleared to "0" after.
739 // We can test this by calling apply_remote_changes with an empty batch.
740 let changes = vec![];
741 pull::apply_remote_changes(&pool, changes).await.unwrap();
742
743 let flag = get_sync_state(&pool, "applying_remote").await.unwrap();
744 assert_eq!(flag, "0", "applying_remote flag should be cleared after apply");
745 }
746
747 #[tokio::test]
748 async fn apply_remote_changes_clears_flag_on_error() {
749 let pool = setup_test_db().await;
750
751 // Test: apply_upsert on unknown table returns BadRequest
752 let mut conn = pool.acquire().await.unwrap();
753 let result = apply::apply_upsert(&mut conn, "fake_table", "id", &json!({"id": "x"})).await;
754 assert!(result.is_err());
755 match result.unwrap_err() {
756 goingson_core::CoreError::BadRequest(msg) => assert!(msg.contains("unknown syncable table")),
757 other => panic!("Expected BadRequest, got: {:?}", other),
758 }
759 }
760
761 #[tokio::test]
762 async fn local_insert_fires_trigger_normally() {
763 let pool = setup_test_db().await;
764 let user_id = create_test_user(&pool).await;
765 let project_id = uuid::Uuid::new_v4().to_string();
766 let now = now_sql();
767
768 // Clear changelog
769 sqlx::query("DELETE FROM sync_changelog")
770 .execute(&pool)
771 .await
772 .unwrap();
773
774 // Insert directly (not via apply_remote_changes) -- trigger should fire
775 sqlx::query(
776 "INSERT INTO projects (id, name, description, project_type, status, created_at, user_id) VALUES (?, ?, ?, ?, ?, ?, ?)"
777 )
778 .bind(&project_id)
779 .bind("Local Project")
780 .bind("")
781 .bind("Job")
782 .bind("Active")
783 .bind(&now)
784 .bind(&user_id)
785 .execute(&pool)
786 .await
787 .unwrap();
788
789 // Trigger should have created a changelog entry
790 let count: (i64,) =
791 sqlx::query_as("SELECT COUNT(*) FROM sync_changelog WHERE table_name = 'projects' AND row_id = ?")
792 .bind(&project_id)
793 .fetch_one(&pool)
794 .await
795 .unwrap();
796 assert_eq!(count.0, 1, "local insert should fire sync trigger");
797 }
798
799 // -- Initial snapshot --
800
801 #[tokio::test]
802 async fn create_initial_snapshot_captures_all_rows() {
803 let pool = setup_test_db().await;
804 let user_id = create_test_user(&pool).await;
805 let now = now_sql();
806
807 // Suppress triggers during setup so only snapshot entries end up in changelog
808 set_sync_state(&pool, "applying_remote", "1").await.unwrap();
809
810 // Insert 2 projects
811 for i in 1..=2 {
812 let pid = uuid::Uuid::new_v4().to_string();
813 sqlx::query(
814 "INSERT INTO projects (id, name, description, project_type, status, created_at, user_id) VALUES (?, ?, ?, ?, ?, ?, ?)"
815 )
816 .bind(&pid)
817 .bind(format!("Project {}", i))
818 .bind("")
819 .bind("Job")
820 .bind("Active")
821 .bind(&now)
822 .bind(&user_id)
823 .execute(&pool)
824 .await
825 .unwrap();
826 }
827
828 set_sync_state(&pool, "applying_remote", "0").await.unwrap();
829
830 // Clear any stray changelog entries
831 sqlx::query("DELETE FROM sync_changelog")
832 .execute(&pool)
833 .await
834 .unwrap();
835
836 let total = create_initial_snapshot(&pool).await.unwrap();
837 assert_eq!(total, 2);
838
839 // Verify changelog entries
840 let count: (i64,) =
841 sqlx::query_as("SELECT COUNT(*) FROM sync_changelog WHERE table_name = 'projects'")
842 .fetch_one(&pool)
843 .await
844 .unwrap();
845 assert_eq!(count.0, 2);
846
847 // Verify snapshot flag was set
848 let flag = get_sync_state(&pool, "initial_snapshot_done").await.unwrap();
849 assert_eq!(flag, "1");
850 }
851
852 #[tokio::test]
853 async fn create_initial_snapshot_empty_db() {
854 let pool = setup_test_db().await;
855 let _user_id = create_test_user(&pool).await;
856
857 // No data rows -- snapshot should report 0
858 sqlx::query("DELETE FROM sync_changelog")
859 .execute(&pool)
860 .await
861 .unwrap();
862
863 let total = create_initial_snapshot(&pool).await.unwrap();
864 assert_eq!(total, 0);
865 }
866
867 // -- Changelog cleanup --
868
869 #[tokio::test]
870 async fn cleanup_changelog_removes_old_pushed_entries() {
871 let pool = setup_test_db().await;
872
873 // Insert a pushed entry with an old timestamp
874 sqlx::query(
875 "INSERT INTO sync_changelog (table_name, op, row_id, timestamp, pushed) VALUES (?, ?, ?, datetime('now', '-10 days'), 1)"
876 )
877 .bind("projects")
878 .bind("INSERT")
879 .bind("old-id")
880 .execute(&pool)
881 .await
882 .unwrap();
883
884 // Insert a recent pushed entry
885 sqlx::query(
886 "INSERT INTO sync_changelog (table_name, op, row_id, timestamp, pushed) VALUES (?, ?, ?, datetime('now'), 1)"
887 )
888 .bind("projects")
889 .bind("INSERT")
890 .bind("recent-id")
891 .execute(&pool)
892 .await
893 .unwrap();
894
895 // Insert an unpushed entry (should never be cleaned)
896 sqlx::query(
897 "INSERT INTO sync_changelog (table_name, op, row_id, timestamp, pushed) VALUES (?, ?, ?, datetime('now', '-10 days'), 0)"
898 )
899 .bind("projects")
900 .bind("INSERT")
901 .bind("unpushed-id")
902 .execute(&pool)
903 .await
904 .unwrap();
905
906 let deleted = cleanup_changelog(&pool).await.unwrap();
907 assert_eq!(deleted, 1, "should only delete old pushed entries");
908
909 // Verify the remaining entries
910 let count: (i64,) =
911 sqlx::query_as("SELECT COUNT(*) FROM sync_changelog")
912 .fetch_one(&pool)
913 .await
914 .unwrap();
915 assert_eq!(count.0, 2, "recent pushed + unpushed should remain");
916 }
917
918 // -- Count pending changes --
919
920 #[tokio::test]
921 async fn count_pending_changes_counts_unpushed() {
922 let pool = setup_test_db().await;
923
924 // Clear any existing entries
925 sqlx::query("DELETE FROM sync_changelog")
926 .execute(&pool)
927 .await
928 .unwrap();
929
930 // Insert 3 unpushed
931 for i in 0..3 {
932 sqlx::query(
933 "INSERT INTO sync_changelog (table_name, op, row_id, pushed) VALUES (?, ?, ?, 0)"
934 )
935 .bind("projects")
936 .bind("INSERT")
937 .bind(format!("id-{}", i))
938 .execute(&pool)
939 .await
940 .unwrap();
941 }
942
943 // Insert 2 pushed
944 for i in 0..2 {
945 sqlx::query(
946 "INSERT INTO sync_changelog (table_name, op, row_id, pushed) VALUES (?, ?, ?, 1)"
947 )
948 .bind("projects")
949 .bind("INSERT")
950 .bind(format!("pushed-{}", i))
951 .execute(&pool)
952 .await
953 .unwrap();
954 }
955
956 let count = count_pending_changes(&pool).await.unwrap();
957 assert_eq!(count, 3);
958 }
959
960 // -- Contact child table upserts --
961
962 #[tokio::test]
963 async fn apply_upsert_contact_with_children() {
964 let pool = setup_test_db().await;
965 let user_id = create_test_user(&pool).await;
966 let contact_id = uuid::Uuid::new_v4().to_string();
967 let email_id = uuid::Uuid::new_v4().to_string();
968 let now = now_sql();
969
970 let mut conn = pool.acquire().await.unwrap();
971
972 // Insert contact (parent)
973 apply::apply_upsert(&mut conn, "contacts", &contact_id, &json!({
974 "id": contact_id, "user_id": user_id,
975 "display_name": "Test Contact", "nickname": null,
976 "company": null, "title": null, "notes": null,
977 "tags": null, "birthday": null, "timezone": null,
978 "created_at": now, "updated_at": now,
979 })).await.unwrap();
980
981 // Insert contact_email (child)
982 apply::apply_upsert(&mut conn, "contact_emails", &email_id, &json!({
983 "id": email_id, "contact_id": contact_id,
984 "address": "test@example.com", "label": "work",
985 "is_primary": true,
986 })).await.unwrap();
987
988 let row: (String,) = sqlx::query_as("SELECT address FROM contact_emails WHERE id = ?")
989 .bind(&email_id)
990 .fetch_one(&pool)
991 .await
992 .unwrap();
993 assert_eq!(row.0, "test@example.com");
994 }
995
996 // -- Boolean handling --
997
998 #[tokio::test]
999 async fn apply_upsert_handles_booleans_as_integers() {
1000 let pool = setup_test_db().await;
1001 let user_id = create_test_user(&pool).await;
1002 let contact_id = uuid::Uuid::new_v4().to_string();
1003 let email_id = uuid::Uuid::new_v4().to_string();
1004 let now = now_sql();
1005
1006 let mut conn = pool.acquire().await.unwrap();
1007
1008 apply::apply_upsert(&mut conn, "contacts", &contact_id, &json!({
1009 "id": contact_id, "user_id": user_id,
1010 "display_name": "Bool Test", "nickname": null,
1011 "company": null, "title": null, "notes": null,
1012 "tags": null, "birthday": null, "timezone": null,
1013 "created_at": now, "updated_at": now,
1014 })).await.unwrap();
1015
1016 // is_primary as a JSON boolean -- should be stored as integer
1017 apply::apply_upsert(&mut conn, "contact_emails", &email_id, &json!({
1018 "id": email_id, "contact_id": contact_id,
1019 "address": "bool@test.com", "label": "home",
1020 "is_primary": true,
1021 })).await.unwrap();
1022
1023 let row: (i32,) = sqlx::query_as("SELECT is_primary FROM contact_emails WHERE id = ?")
1024 .bind(&email_id)
1025 .fetch_one(&pool)
1026 .await
1027 .unwrap();
1028 assert_eq!(row.0, 1);
1029 }
1030
1031 // -- email_accounts sync triggers --
1032
1033 #[tokio::test]
1034 async fn email_account_insert_fires_trigger() {
1035 let pool = setup_test_db().await;
1036 let user_id = create_test_user(&pool).await;
1037 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1038
1039 let account_id = uuid::Uuid::new_v4().to_string();
1040 sqlx::query(
1041 "INSERT INTO email_accounts (id, user_id, account_name, email_address, \
1042 imap_server, imap_port, smtp_server, smtp_port, username, password, \
1043 use_tls, created_at) \
1044 VALUES (?, ?, 'Work', 'test@example.com', 'imap.example.com', 993, \
1045 'smtp.example.com', 587, 'user', 'pass', 1, datetime('now'))"
1046 )
1047 .bind(&account_id)
1048 .bind(&user_id)
1049 .execute(&pool)
1050 .await
1051 .unwrap();
1052
1053 let count: (i64,) = sqlx::query_as(
1054 "SELECT COUNT(*) FROM sync_changelog WHERE table_name = 'email_accounts' AND op = 'INSERT'"
1055 )
1056 .fetch_one(&pool)
1057 .await
1058 .unwrap();
1059 assert_eq!(count.0, 1);
1060
1061 let data: (String,) = sqlx::query_as(
1062 "SELECT data FROM sync_changelog WHERE table_name = 'email_accounts' AND row_id = ?"
1063 )
1064 .bind(&account_id)
1065 .fetch_one(&pool)
1066 .await
1067 .unwrap();
1068 let parsed: serde_json::Value = serde_json::from_str(&data.0).unwrap();
1069 let obj = parsed.as_object().unwrap();
1070 assert_eq!(obj.len(), 17, "trigger data should have 17 keys (config only)");
1071 assert!(!obj.contains_key("password"), "password should not be in trigger data");
1072 assert!(!obj.contains_key("oauth2_access_token"), "oauth2_access_token should not be in trigger data");
1073 assert!(!obj.contains_key("oauth2_refresh_token"), "oauth2_refresh_token should not be in trigger data");
1074 assert!(!obj.contains_key("oauth2_token_expires_at"), "oauth2_token_expires_at should not be in trigger data");
1075 assert!(!obj.contains_key("last_sync_at"), "last_sync_at should not be in trigger data");
1076 assert_eq!(parsed["account_name"], "Work");
1077 }
1078
1079 #[tokio::test]
1080 async fn email_account_update_fires_trigger() {
1081 let pool = setup_test_db().await;
1082 let user_id = create_test_user(&pool).await;
1083
1084 // Insert with triggers suppressed
1085 set_sync_state(&pool, "applying_remote", "1").await.unwrap();
1086 let account_id = uuid::Uuid::new_v4().to_string();
1087 sqlx::query(
1088 "INSERT INTO email_accounts (id, user_id, account_name, email_address, \
1089 imap_server, imap_port, smtp_server, smtp_port, username, password, \
1090 use_tls, created_at) \
1091 VALUES (?, ?, 'Work', 'test@example.com', 'imap.example.com', 993, \
1092 'smtp.example.com', 587, 'user', 'pass', 1, datetime('now'))"
1093 )
1094 .bind(&account_id)
1095 .bind(&user_id)
1096 .execute(&pool)
1097 .await
1098 .unwrap();
1099 set_sync_state(&pool, "applying_remote", "0").await.unwrap();
1100 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1101
1102 // Update account_name
1103 sqlx::query("UPDATE email_accounts SET account_name = 'Personal' WHERE id = ?")
1104 .bind(&account_id)
1105 .execute(&pool)
1106 .await
1107 .unwrap();
1108
1109 let row: (String, String) = sqlx::query_as(
1110 "SELECT op, data FROM sync_changelog WHERE table_name = 'email_accounts' AND row_id = ?"
1111 )
1112 .bind(&account_id)
1113 .fetch_one(&pool)
1114 .await
1115 .unwrap();
1116 assert_eq!(row.0, "UPDATE");
1117
1118 let parsed: serde_json::Value = serde_json::from_str(&row.1).unwrap();
1119 assert_eq!(parsed["account_name"], "Personal");
1120 assert!(!parsed.as_object().unwrap().contains_key("password"), "update trigger should not include password");
1121 }
1122
1123 #[tokio::test]
1124 async fn email_account_delete_fires_trigger() {
1125 let pool = setup_test_db().await;
1126 let user_id = create_test_user(&pool).await;
1127
1128 // Insert with triggers suppressed
1129 set_sync_state(&pool, "applying_remote", "1").await.unwrap();
1130 let account_id = uuid::Uuid::new_v4().to_string();
1131 sqlx::query(
1132 "INSERT INTO email_accounts (id, user_id, account_name, email_address, \
1133 imap_server, imap_port, smtp_server, smtp_port, username, password, \
1134 use_tls, created_at) \
1135 VALUES (?, ?, 'Work', 'test@example.com', 'imap.example.com', 993, \
1136 'smtp.example.com', 587, 'user', 'pass', 1, datetime('now'))"
1137 )
1138 .bind(&account_id)
1139 .bind(&user_id)
1140 .execute(&pool)
1141 .await
1142 .unwrap();
1143 set_sync_state(&pool, "applying_remote", "0").await.unwrap();
1144 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1145
1146 // Delete the account
1147 sqlx::query("DELETE FROM email_accounts WHERE id = ?")
1148 .bind(&account_id)
1149 .execute(&pool)
1150 .await
1151 .unwrap();
1152
1153 let row: (String,) = sqlx::query_as(
1154 "SELECT op FROM sync_changelog WHERE table_name = 'email_accounts' AND row_id = ?"
1155 )
1156 .bind(&account_id)
1157 .fetch_one(&pool)
1158 .await
1159 .unwrap();
1160 assert_eq!(row.0, "DELETE");
1161 }
1162
1163 #[test]
1164 fn email_account_table_columns_has_17_entries() {
1165 let cols = apply::table_columns("email_accounts").unwrap();
1166 assert_eq!(cols.len(), 17, "email_accounts should sync 17 config columns");
1167 assert!(!cols.contains(&"password"), "password should not be in synced columns");
1168 assert!(!cols.contains(&"oauth2_access_token"), "oauth2_access_token should not be in synced columns");
1169 assert!(!cols.contains(&"oauth2_refresh_token"), "oauth2_refresh_token should not be in synced columns");
1170 assert!(!cols.contains(&"oauth2_token_expires_at"), "oauth2_token_expires_at should not be in synced columns");
1171 assert!(!cols.contains(&"last_sync_at"), "last_sync_at should not be in synced columns");
1172 }
1173
1174 // -- email_accounts credential preservation --
1175
1176 #[tokio::test]
1177 async fn apply_email_account_upsert_creates_with_empty_password() {
1178 let pool = setup_test_db().await;
1179 let user_id = create_test_user(&pool).await;
1180 let account_id = uuid::Uuid::new_v4().to_string();
1181 let now = now_sql();
1182
1183 let data = json!({
1184 "id": account_id,
1185 "user_id": user_id,
1186 "account_name": "Remote Work",
1187 "email_address": "work@example.com",
1188 "imap_server": "imap.example.com",
1189 "imap_port": 993,
1190 "smtp_server": "smtp.example.com",
1191 "smtp_port": 587,
1192 "username": "user",
1193 "use_tls": true,
1194 "created_at": now,
1195 "archive_folder_name": null,
1196 "auth_type": "password",
1197 "jmap_session_url": null,
1198 "jmap_account_id": null,
1199 "sync_interval_minutes": 5,
1200 });
1201
1202 let mut conn = pool.acquire().await.unwrap();
1203 apply::apply_upsert(&mut conn, "email_accounts", &account_id, &data).await.unwrap();
1204
1205 let row: (String, String) = sqlx::query_as(
1206 "SELECT account_name, password FROM email_accounts WHERE id = ?"
1207 )
1208 .bind(&account_id)
1209 .fetch_one(&pool)
1210 .await
1211 .unwrap();
1212 assert_eq!(row.0, "Remote Work");
1213 assert_eq!(row.1, "", "new remote account should have empty password");
1214 }
1215
1216 #[tokio::test]
1217 async fn apply_email_account_upsert_preserves_local_credentials() {
1218 let pool = setup_test_db().await;
1219 let user_id = create_test_user(&pool).await;
1220 let account_id = uuid::Uuid::new_v4().to_string();
1221 let now = now_sql();
1222
1223 // Insert a local account with real credentials
1224 sqlx::query(
1225 "INSERT INTO email_accounts (id, user_id, account_name, email_address, \
1226 imap_server, imap_port, smtp_server, smtp_port, username, password, \
1227 use_tls, created_at, auth_type, oauth2_access_token, oauth2_refresh_token, \
1228 oauth2_token_expires_at, sync_interval_minutes) \
1229 VALUES (?, ?, 'Work', 'test@example.com', 'imap.example.com', 993, \
1230 'smtp.example.com', 587, 'user', 'secret-password', 1, ?, 'oauth2', \
1231 'access-tok-123', 'refresh-tok-456', '2026-12-31T00:00:00Z', 5)"
1232 )
1233 .bind(&account_id)
1234 .bind(&user_id)
1235 .bind(&now)
1236 .execute(&pool)
1237 .await
1238 .unwrap();
1239
1240 // Remote update changes account_name but has no credential fields
1241 let remote_data = json!({
1242 "id": account_id,
1243 "user_id": user_id,
1244 "account_name": "Personal",
1245 "email_address": "test@example.com",
1246 "imap_server": "imap.example.com",
1247 "imap_port": 993,
1248 "smtp_server": "smtp.example.com",
1249 "smtp_port": 587,
1250 "username": "user",
1251 "use_tls": true,
1252 "created_at": now,
1253 "archive_folder_name": null,
1254 "auth_type": "oauth2",
1255 "jmap_session_url": null,
1256 "jmap_account_id": null,
1257 "sync_interval_minutes": 5,
1258 });
1259
1260 let mut conn = pool.acquire().await.unwrap();
1261 apply::apply_upsert(&mut conn, "email_accounts", &account_id, &remote_data).await.unwrap();
1262
1263 // Config should be updated
1264 let row: (String, String, Option<String>, Option<String>, Option<String>) = sqlx::query_as(
1265 "SELECT account_name, password, oauth2_access_token, oauth2_refresh_token, \
1266 oauth2_token_expires_at FROM email_accounts WHERE id = ?"
1267 )
1268 .bind(&account_id)
1269 .fetch_one(&pool)
1270 .await
1271 .unwrap();
1272
1273 assert_eq!(row.0, "Personal", "account_name should be updated");
1274 assert_eq!(row.1, "secret-password", "password should be preserved");
1275 assert_eq!(row.2.as_deref(), Some("access-tok-123"), "oauth2_access_token should be preserved");
1276 assert_eq!(row.3.as_deref(), Some("refresh-tok-456"), "oauth2_refresh_token should be preserved");
1277 assert_eq!(row.4.as_deref(), Some("2026-12-31T00:00:00Z"), "oauth2_token_expires_at should be preserved");
1278 }
1279
1280 #[tokio::test]
1281 async fn apply_upsert_task_with_dangling_source_email_id() {
1282 let pool = setup_test_db().await;
1283 let user_id = create_test_user(&pool).await;
1284 let project_id = uuid::Uuid::new_v4().to_string();
1285 let task_id = uuid::Uuid::new_v4().to_string();
1286 let fake_email_id = uuid::Uuid::new_v4().to_string();
1287 let now = now_sql();
1288
1289 // Apply via apply_remote_changes which disables FK enforcement
1290 let changes = vec![
1291 change("projects", synckit_client::ChangeOp::Insert, &project_id, Some(json!({
1292 "id": project_id, "name": "P", "description": "",
1293 "project_type": "Job", "status": "Active",
1294 "created_at": now, "user_id": user_id,
1295 }))),
1296 change("tasks", synckit_client::ChangeOp::Insert, &task_id, Some(json!({
1297 "id": task_id,
1298 "project_id": project_id,
1299 "description": "Task from email",
1300 "status": "Pending",
1301 "priority": "Medium",
1302 "due": null,
1303 "tags": null,
1304 "urgency": 50,
1305 "recurrence": "None",
1306 "created_at": now,
1307 "user_id": user_id,
1308 "recurrence_parent_id": null,
1309 "source_email_id": fake_email_id,
1310 "snoozed_until": null,
1311 "waiting_for_response": false,
1312 "waiting_since": null,
1313 "expected_response_date": null,
1314 "scheduled_start": null,
1315 "scheduled_duration": null,
1316 "is_focus": false,
1317 "focus_set_at": null,
1318 "contact_id": null,
1319 "milestone_id": null,
1320 }))),
1321 ];
1322
1323 // Should succeed -- FK enforcement is OFF during remote apply
1324 pull::apply_remote_changes(&pool, changes).await.unwrap();
1325
1326 let row: (String, Option<String>) = sqlx::query_as(
1327 "SELECT description, source_email_id FROM tasks WHERE id = ?"
1328 )
1329 .bind(&task_id)
1330 .fetch_one(&pool)
1331 .await
1332 .unwrap();
1333 assert_eq!(row.0, "Task from email");
1334 assert_eq!(row.1.as_deref(), Some(fake_email_id.as_str()), "dangling source_email_id should be stored");
1335 }
1336
1337 #[tokio::test]
1338 async fn email_account_trigger_excludes_credentials() {
1339 let pool = setup_test_db().await;
1340 let user_id = create_test_user(&pool).await;
1341 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1342
1343 let account_id = uuid::Uuid::new_v4().to_string();
1344 sqlx::query(
1345 "INSERT INTO email_accounts (id, user_id, account_name, email_address, \
1346 imap_server, imap_port, smtp_server, smtp_port, username, password, \
1347 use_tls, created_at, auth_type, oauth2_access_token, oauth2_refresh_token, \
1348 oauth2_token_expires_at, sync_interval_minutes) \
1349 VALUES (?, ?, 'Work', 'test@example.com', 'imap.example.com', 993, \
1350 'smtp.example.com', 587, 'user', 'secret', 1, datetime('now'), 'oauth2', \
1351 'at-123', 'rt-456', '2026-12-31', 5)"
1352 )
1353 .bind(&account_id)
1354 .bind(&user_id)
1355 .execute(&pool)
1356 .await
1357 .unwrap();
1358
1359 let data: (String,) = sqlx::query_as(
1360 "SELECT data FROM sync_changelog WHERE table_name = 'email_accounts' AND row_id = ?"
1361 )
1362 .bind(&account_id)
1363 .fetch_one(&pool)
1364 .await
1365 .unwrap();
1366
1367 let parsed: serde_json::Value = serde_json::from_str(&data.0).unwrap();
1368 let obj = parsed.as_object().unwrap();
1369
1370 // Should have exactly the 17 config columns
1371 assert_eq!(obj.len(), 17);
1372
1373 // Credential fields must NOT be present
1374 assert!(!obj.contains_key("password"));
1375 assert!(!obj.contains_key("oauth2_access_token"));
1376 assert!(!obj.contains_key("oauth2_refresh_token"));
1377 assert!(!obj.contains_key("oauth2_token_expires_at"));
1378
1379 // Config fields should be present
1380 assert_eq!(parsed["account_name"], "Work");
1381 assert_eq!(parsed["auth_type"], "oauth2");
1382 assert_eq!(parsed["sync_interval_minutes"], 5);
1383 }
1384
1385 // -- Attachment sync tests --
1386
1387 /// Helper: insert a project and task, return (project_id, task_id).
1388 async fn setup_project_and_task(pool: &SqlitePool, user_id: &str) -> (String, String) {
1389 let project_id = uuid::Uuid::new_v4().to_string();
1390 let task_id = uuid::Uuid::new_v4().to_string();
1391 let now = now_sql();
1392
1393 sqlx::query(
1394 "INSERT INTO projects (id, name, description, project_type, status, created_at, user_id) VALUES (?, ?, ?, ?, ?, ?, ?)"
1395 )
1396 .bind(&project_id).bind("Test Project").bind("").bind("Job").bind("Active").bind(&now).bind(user_id)
1397 .execute(pool).await.unwrap();
1398
1399 sqlx::query(
1400 "INSERT INTO tasks (id, project_id, description, status, priority, urgency, recurrence, created_at, user_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
1401 )
1402 .bind(&task_id).bind(&project_id).bind("Test Task").bind("Pending").bind("Medium").bind(50).bind("None").bind(&now).bind(user_id)
1403 .execute(pool).await.unwrap();
1404
1405 (project_id, task_id)
1406 }
1407
1408 #[test]
1409 fn attachment_table_columns_whitelist_has_10_entries() {
1410 let cols = apply::table_columns("attachments").unwrap();
1411 assert_eq!(cols.len(), 10);
1412 assert_eq!(cols[0], "id");
1413 assert!(cols.contains(&"blob_hash"), "blob_hash must be synced for blob resolution");
1414 assert!(cols.contains(&"filename"));
1415 assert!(cols.contains(&"file_size"));
1416 assert!(cols.contains(&"mime_type"));
1417 }
1418
1419 #[test]
1420 fn time_sessions_table_columns_whitelist_has_7_entries() {
1421 let cols = apply::table_columns("time_sessions").unwrap();
1422 assert_eq!(cols.len(), 7);
1423 assert_eq!(cols[0], "id");
1424 assert!(cols.contains(&"task_id"));
1425 assert!(cols.contains(&"duration_minutes"));
1426 }
1427
1428 #[test]
1429 fn upsert_order_has_attachments_after_tasks() {
1430 let pos = |table: &str| UPSERT_ORDER.iter().position(|t| *t == table);
1431 // attachments reference tasks.id
1432 assert!(pos("tasks").unwrap() < pos("attachments").unwrap());
1433 }
1434
1435 #[test]
1436 fn delete_order_has_attachments_before_tasks() {
1437 let pos = |table: &str| DELETE_ORDER.iter().position(|t| *t == table);
1438 assert!(pos("attachments").unwrap() < pos("tasks").unwrap());
1439 }
1440
1441 #[test]
1442 fn upsert_order_has_time_sessions_after_tasks() {
1443 let pos = |table: &str| UPSERT_ORDER.iter().position(|t| *t == table);
1444 assert!(pos("tasks").unwrap() < pos("time_sessions").unwrap());
1445 }
1446
1447 #[test]
1448 fn delete_order_has_time_sessions_before_tasks() {
1449 let pos = |table: &str| DELETE_ORDER.iter().position(|t| *t == table);
1450 assert!(pos("time_sessions").unwrap() < pos("tasks").unwrap());
1451 }
1452
1453 #[tokio::test]
1454 async fn attachment_insert_fires_sync_trigger() {
1455 let pool = setup_test_db().await;
1456 let user_id = create_test_user(&pool).await;
1457 let (_project_id, task_id) = setup_project_and_task(&pool, &user_id).await;
1458 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1459
1460 let att_id = uuid::Uuid::new_v4().to_string();
1461 sqlx::query(
1462 "INSERT INTO attachments (id, user_id, task_id, filename, file_size, mime_type, blob_hash, created_at) \
1463 VALUES (?, ?, ?, 'report.pdf', 12345, 'application/pdf', 'abc123hash', datetime('now'))"
1464 )
1465 .bind(&att_id).bind(&user_id).bind(&task_id)
1466 .execute(&pool).await.unwrap();
1467
1468 let count: (i64,) = sqlx::query_as(
1469 "SELECT COUNT(*) FROM sync_changelog WHERE table_name = 'attachments' AND row_id = ?"
1470 ).bind(&att_id).fetch_one(&pool).await.unwrap();
1471 assert_eq!(count.0, 1, "attachment insert should fire sync trigger");
1472
1473 // Verify trigger data contains blob_hash (critical for blob resolution)
1474 let data: (String,) = sqlx::query_as(
1475 "SELECT data FROM sync_changelog WHERE table_name = 'attachments' AND row_id = ?"
1476 ).bind(&att_id).fetch_one(&pool).await.unwrap();
1477 let parsed: serde_json::Value = serde_json::from_str(&data.0).unwrap();
1478 assert_eq!(parsed["blob_hash"], "abc123hash");
1479 assert_eq!(parsed["filename"], "report.pdf");
1480 assert_eq!(parsed["file_size"], 12345);
1481 }
1482
1483 #[tokio::test]
1484 async fn attachment_delete_fires_sync_trigger() {
1485 let pool = setup_test_db().await;
1486 let user_id = create_test_user(&pool).await;
1487 let (_project_id, task_id) = setup_project_and_task(&pool, &user_id).await;
1488
1489 let att_id = uuid::Uuid::new_v4().to_string();
1490 sqlx::query(
1491 "INSERT INTO attachments (id, user_id, task_id, filename, file_size, mime_type, blob_hash, created_at) \
1492 VALUES (?, ?, ?, 'file.txt', 100, 'text/plain', 'hash1', datetime('now'))"
1493 )
1494 .bind(&att_id).bind(&user_id).bind(&task_id)
1495 .execute(&pool).await.unwrap();
1496 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1497
1498 sqlx::query("DELETE FROM attachments WHERE id = ?")
1499 .bind(&att_id).execute(&pool).await.unwrap();
1500
1501 let row: (String,) = sqlx::query_as(
1502 "SELECT op FROM sync_changelog WHERE table_name = 'attachments' AND row_id = ?"
1503 ).bind(&att_id).fetch_one(&pool).await.unwrap();
1504 assert_eq!(row.0, "DELETE");
1505 }
1506
1507 #[tokio::test]
1508 async fn apply_upsert_attachment_round_trip() {
1509 let pool = setup_test_db().await;
1510 let user_id = create_test_user(&pool).await;
1511 let (project_id, task_id) = setup_project_and_task(&pool, &user_id).await;
1512 let att_id = uuid::Uuid::new_v4().to_string();
1513 let now = now_sql();
1514
1515 let data = json!({
1516 "id": att_id,
1517 "user_id": user_id,
1518 "task_id": task_id,
1519 "project_id": project_id,
1520 "filename": "slides.pptx",
1521 "file_size": 98765,
1522 "mime_type": "application/vnd.ms-powerpoint",
1523 "blob_hash": "sha256-deadbeef",
1524 "source_email_id": null,
1525 "created_at": now,
1526 });
1527
1528 let mut conn = pool.acquire().await.unwrap();
1529 apply::apply_upsert(&mut conn, "attachments", &att_id, &data).await.unwrap();
1530
1531 let row: (String, i64, String) = sqlx::query_as(
1532 "SELECT filename, file_size, blob_hash FROM attachments WHERE id = ?"
1533 ).bind(&att_id).fetch_one(&pool).await.unwrap();
1534 assert_eq!(row.0, "slides.pptx");
1535 assert_eq!(row.1, 98765);
1536 assert_eq!(row.2, "sha256-deadbeef");
1537 }
1538
1539 #[tokio::test]
1540 async fn apply_delete_attachment() {
1541 let pool = setup_test_db().await;
1542 let user_id = create_test_user(&pool).await;
1543 let (_project_id, task_id) = setup_project_and_task(&pool, &user_id).await;
1544 let att_id = uuid::Uuid::new_v4().to_string();
1545 let now = now_sql();
1546
1547 let mut conn = pool.acquire().await.unwrap();
1548 apply::apply_upsert(&mut conn, "attachments", &att_id, &json!({
1549 "id": att_id, "user_id": user_id, "task_id": task_id,
1550 "project_id": null, "filename": "delete-me.txt",
1551 "file_size": 10, "mime_type": "text/plain", "blob_hash": "h",
1552 "source_email_id": null, "created_at": now,
1553 })).await.unwrap();
1554
1555 apply::apply_delete(&mut conn, "attachments", &att_id).await.unwrap();
1556
1557 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM attachments WHERE id = ?")
1558 .bind(&att_id).fetch_one(&pool).await.unwrap();
1559 assert_eq!(count.0, 0);
1560 }
1561
1562 #[tokio::test]
1563 async fn time_session_insert_fires_sync_trigger() {
1564 let pool = setup_test_db().await;
1565 let user_id = create_test_user(&pool).await;
1566 let (_project_id, task_id) = setup_project_and_task(&pool, &user_id).await;
1567 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1568
1569 let session_id = uuid::Uuid::new_v4().to_string();
1570 sqlx::query(
1571 "INSERT INTO time_sessions (id, task_id, user_id, started_at, created_at) \
1572 VALUES (?, ?, ?, datetime('now'), datetime('now'))"
1573 )
1574 .bind(&session_id).bind(&task_id).bind(&user_id)
1575 .execute(&pool).await.unwrap();
1576
1577 let count: (i64,) = sqlx::query_as(
1578 "SELECT COUNT(*) FROM sync_changelog WHERE table_name = 'time_sessions' AND row_id = ?"
1579 ).bind(&session_id).fetch_one(&pool).await.unwrap();
1580 assert_eq!(count.0, 1, "time_session insert should fire sync trigger");
1581
1582 let data: (String,) = sqlx::query_as(
1583 "SELECT data FROM sync_changelog WHERE table_name = 'time_sessions' AND row_id = ?"
1584 ).bind(&session_id).fetch_one(&pool).await.unwrap();
1585 let parsed: serde_json::Value = serde_json::from_str(&data.0).unwrap();
1586 assert_eq!(parsed["task_id"], task_id);
1587 }
1588
1589 #[tokio::test]
1590 async fn time_session_update_fires_sync_trigger() {
1591 let pool = setup_test_db().await;
1592 let user_id = create_test_user(&pool).await;
1593 let (_project_id, task_id) = setup_project_and_task(&pool, &user_id).await;
1594
1595 let session_id = uuid::Uuid::new_v4().to_string();
1596 sqlx::query(
1597 "INSERT INTO time_sessions (id, task_id, user_id, started_at, created_at) \
1598 VALUES (?, ?, ?, datetime('now'), datetime('now'))"
1599 )
1600 .bind(&session_id).bind(&task_id).bind(&user_id)
1601 .execute(&pool).await.unwrap();
1602 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1603
1604 // Stop the timer
1605 sqlx::query("UPDATE time_sessions SET ended_at = datetime('now'), duration_minutes = 25 WHERE id = ?")
1606 .bind(&session_id).execute(&pool).await.unwrap();
1607
1608 let row: (String, String) = sqlx::query_as(
1609 "SELECT op, data FROM sync_changelog WHERE table_name = 'time_sessions' AND row_id = ?"
1610 ).bind(&session_id).fetch_one(&pool).await.unwrap();
1611 assert_eq!(row.0, "UPDATE");
1612
1613 let parsed: serde_json::Value = serde_json::from_str(&row.1).unwrap();
1614 assert_eq!(parsed["duration_minutes"], 25);
1615 }
1616
1617 #[tokio::test]
1618 async fn apply_upsert_time_session_round_trip() {
1619 let pool = setup_test_db().await;
1620 let user_id = create_test_user(&pool).await;
1621 let (_project_id, task_id) = setup_project_and_task(&pool, &user_id).await;
1622 let session_id = uuid::Uuid::new_v4().to_string();
1623 let now = now_sql();
1624
1625 let data = json!({
1626 "id": session_id,
1627 "task_id": task_id,
1628 "user_id": user_id,
1629 "started_at": now,
1630 "ended_at": now,
1631 "duration_minutes": 42,
1632 "created_at": now,
1633 });
1634
1635 let mut conn = pool.acquire().await.unwrap();
1636 apply::apply_upsert(&mut conn, "time_sessions", &session_id, &data).await.unwrap();
1637
1638 let row: (String, i32) = sqlx::query_as(
1639 "SELECT task_id, duration_minutes FROM time_sessions WHERE id = ?"
1640 ).bind(&session_id).fetch_one(&pool).await.unwrap();
1641 assert_eq!(row.0, task_id);
1642 assert_eq!(row.1, 42);
1643 }
1644
1645 #[tokio::test]
1646 async fn apply_remote_changes_attachment_with_task_in_same_batch() {
1647 let pool = setup_test_db().await;
1648 let user_id = create_test_user(&pool).await;
1649 let project_id = uuid::Uuid::new_v4().to_string();
1650 let task_id = uuid::Uuid::new_v4().to_string();
1651 let att_id = uuid::Uuid::new_v4().to_string();
1652 let now = now_sql();
1653
1654 // Send task + attachment in same batch (attachment references task via FK)
1655 // apply_remote_changes must handle FK ordering correctly
1656 let changes = vec![
1657 change("attachments", synckit_client::ChangeOp::Insert, &att_id, Some(json!({
1658 "id": att_id, "user_id": user_id, "task_id": task_id,
1659 "project_id": project_id, "filename": "batch.pdf",
1660 "file_size": 500, "mime_type": "application/pdf",
1661 "blob_hash": "batchhash", "source_email_id": null, "created_at": now,
1662 }))),
1663 change("tasks", synckit_client::ChangeOp::Insert, &task_id, Some(json!({
1664 "id": task_id, "project_id": project_id, "description": "Batch task",
1665 "status": "Pending", "priority": "Low", "due": null,
1666 "tags": null, "urgency": 10, "recurrence": "None",
1667 "created_at": now, "user_id": user_id,
1668 "recurrence_parent_id": null, "source_email_id": null,
1669 "snoozed_until": null, "waiting_for_response": false,
1670 "waiting_since": null, "expected_response_date": null,
1671 "scheduled_start": null, "scheduled_duration": null,
1672 "is_focus": false, "focus_set_at": null,
1673 "contact_id": null, "milestone_id": null,
1674 "completed_at": null, "estimated_minutes": null, "actual_minutes": 0,
1675 }))),
1676 change("projects", synckit_client::ChangeOp::Insert, &project_id, Some(json!({
1677 "id": project_id, "name": "Batch project", "description": "",
1678 "project_type": "Job", "status": "Active",
1679 "created_at": now, "user_id": user_id,
1680 }))),
1681 ];
1682
1683 // Should succeed despite wrong input order (attachment before task before project)
1684 pull::apply_remote_changes(&pool, changes).await.unwrap();
1685
1686 let att: (String,) = sqlx::query_as("SELECT filename FROM attachments WHERE id = ?")
1687 .bind(&att_id).fetch_one(&pool).await.unwrap();
1688 assert_eq!(att.0, "batch.pdf");
1689
1690 // Triggers should be suppressed -- no changelog entries
1691 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1692 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM sync_changelog")
1693 .fetch_one(&pool).await.unwrap();
1694 assert_eq!(count.0, 0);
1695 }
1696
1697 #[tokio::test]
1698 async fn initial_snapshot_captures_attachments_and_time_sessions() {
1699 let pool = setup_test_db().await;
1700 let user_id = create_test_user(&pool).await;
1701 let (_project_id, task_id) = setup_project_and_task(&pool, &user_id).await;
1702
1703 // Insert attachment and time_session with triggers suppressed
1704 set_sync_state(&pool, "applying_remote", "1").await.unwrap();
1705
1706 let att_id = uuid::Uuid::new_v4().to_string();
1707 sqlx::query(
1708 "INSERT INTO attachments (id, user_id, task_id, filename, file_size, mime_type, blob_hash, created_at) \
1709 VALUES (?, ?, ?, 'snap.pdf', 100, 'application/pdf', 'snaphash', datetime('now'))"
1710 )
1711 .bind(&att_id).bind(&user_id).bind(&task_id)
1712 .execute(&pool).await.unwrap();
1713
1714 let session_id = uuid::Uuid::new_v4().to_string();
1715 sqlx::query(
1716 "INSERT INTO time_sessions (id, task_id, user_id, started_at, ended_at, duration_minutes, created_at) \
1717 VALUES (?, ?, ?, datetime('now', '-1 hour'), datetime('now'), 60, datetime('now'))"
1718 )
1719 .bind(&session_id).bind(&task_id).bind(&user_id)
1720 .execute(&pool).await.unwrap();
1721
1722 set_sync_state(&pool, "applying_remote", "0").await.unwrap();
1723 sqlx::query("DELETE FROM sync_changelog").execute(&pool).await.unwrap();
1724
1725 let total = create_initial_snapshot(&pool).await.unwrap();
1726
1727 // Should capture: 1 project + 1 task + 1 attachment + 1 time_session = 4
1728 assert_eq!(total, 4, "snapshot should capture projects, tasks, attachments, and time_sessions");
1729
1730 let att_count: (i64,) = sqlx::query_as(
1731 "SELECT COUNT(*) FROM sync_changelog WHERE table_name = 'attachments'"
1732 ).fetch_one(&pool).await.unwrap();
1733 assert_eq!(att_count.0, 1);
1734
1735 let ts_count: (i64,) = sqlx::query_as(
1736 "SELECT COUNT(*) FROM sync_changelog WHERE table_name = 'time_sessions'"
1737 ).fetch_one(&pool).await.unwrap();
1738 assert_eq!(ts_count.0, 1);
1739 }
1740
1741 // ── CHRONIC-C round-trip invariant (UF-1 regression guard) ──────────────────
1742 //
1743 // Every column a sync changelog trigger emits into `sync_changelog.data` must be
1744 // in the apply-side whitelist (`apply::SYNCED_COLUMNS`), and every whitelisted
1745 // column must be emitted by the trigger, for every synced table. This is read
1746 // from the live trigger DDL in a migrated DB so the two lists cannot silently
1747 // diverge again -- the exact failure mode of UF-1, where the trigger emitted
1748 // `recurrence_rule` but the apply whitelist dropped it, so it arrived NULL on pull.
1749
1750 /// First single-quoted literal after `VALUES (` in a trigger body -- the changelog
1751 /// target table name.
1752 fn extract_changelog_table(sql: &str) -> Option<String> {
1753 let after_values = &sql[sql.find("VALUES")?..];
1754 let after_paren = &after_values[after_values.find('(')? + 1..];
1755 let start = after_paren.find('\'')? + 1;
1756 let rest = &after_paren[start..];
1757 let end = rest.find('\'')?;
1758 Some(rest[..end].to_string())
1759 }
1760
1761 /// Keys of the `json_object(...)` payload: every single-quoted identifier that is
1762 /// immediately followed by `, NEW.` or `, OLD.`. Scoped to the substring after
1763 /// `json_object(`, so the table-name and op literals that precede it (which are
1764 /// not followed by a column reference) are never captured.
1765 fn extract_json_object_keys(sql: &str) -> std::collections::BTreeSet<String> {
1766 let mut keys = std::collections::BTreeSet::new();
1767 let Some(jidx) = sql.find("json_object(") else {
1768 return keys;
1769 };
1770 let body = &sql[jidx..];
1771 let bytes = body.as_bytes();
1772 let is_ws = |b: u8| b == b' ' || b == b'\n' || b == b'\t' || b == b'\r';
1773 let mut i = 0;
1774 while i < bytes.len() {
1775 if bytes[i] != b'\'' {
1776 i += 1;
1777 continue;
1778 }
1779 let key_start = i + 1;
1780 let mut j = key_start;
1781 while j < bytes.len() && bytes[j] != b'\'' {
1782 j += 1;
1783 }
1784 if j >= bytes.len() {
1785 break;
1786 }
1787 let key = &body[key_start..j];
1788 let mut k = j + 1;
1789 while k < bytes.len() && is_ws(bytes[k]) {
1790 k += 1;
1791 }
1792 if k < bytes.len() && bytes[k] == b',' {
1793 k += 1;
1794 while k < bytes.len() && is_ws(bytes[k]) {
1795 k += 1;
1796 }
1797 if body[k..].starts_with("NEW.") || body[k..].starts_with("OLD.") {
1798 keys.insert(key.to_string());
1799 }
1800 }
1801 i = j + 1;
1802 }
1803 keys
1804 }
1805
1806 #[tokio::test]
1807 async fn trigger_columns_match_whitelist() {
1808 use std::collections::{BTreeMap, BTreeSet};
1809
1810 let pool = setup_test_db().await;
1811
1812 let triggers: Vec<(String, String)> = sqlx::query_as(
1813 "SELECT name, sql FROM sqlite_master \
1814 WHERE type = 'trigger' AND sql LIKE '%sync_changelog%'",
1815 )
1816 .fetch_all(&pool)
1817 .await
1818 .expect("query changelog triggers");
1819
1820 // table -> union of columns emitted by its INSERT/UPDATE triggers.
1821 let mut emitted: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
1822 for (name, sql) in &triggers {
1823 let upper = sql.to_uppercase();
1824 // DELETE triggers carry no row payload; only INSERT/UPDATE define columns.
1825 if !(upper.contains("AFTER INSERT") || upper.contains("AFTER UPDATE")) {
1826 continue;
1827 }
1828 let table = extract_changelog_table(sql)
1829 .unwrap_or_else(|| panic!("trigger {name} has no VALUES ('<table>', ...) literal"));
1830 let keys = extract_json_object_keys(sql);
1831 assert!(!keys.is_empty(), "trigger {name} emitted no json_object keys");
1832 emitted.entry(table).or_default().extend(keys);
1833 }
1834
1835 // Every synced table must have a trigger and match its whitelist exactly.
1836 for (table, cols) in apply::SYNCED_COLUMNS {
1837 let whitelist: BTreeSet<String> = cols.iter().map(|c| c.to_string()).collect();
1838 let got = emitted.get(*table).unwrap_or_else(|| {
1839 panic!("synced table `{table}` has no INSERT/UPDATE changelog trigger")
1840 });
1841 assert_eq!(
1842 got, &whitelist,
1843 "column drift for `{table}`: trigger emits {got:?}, apply whitelist is {whitelist:?}",
1844 );
1845 }
1846
1847 // And no trigger may emit for a table absent from the whitelist (orphan).
1848 for table in emitted.keys() {
1849 assert!(
1850 apply::table_columns(table).is_some(),
1851 "trigger emits for `{table}` but it is not in SYNCED_COLUMNS",
1852 );
1853 }
1854 }
1855
1856 // ── HLC assignment monotonicity ─────────────────────────────────────────────
1857 //
1858 // assign_pending_hlcs stamps unpushed local rows lazily; the stamps must sort
1859 // strictly after each other in stamp order so push and conflict-detection agree
1860 // on a total order, and each stamp must land in the committed-HLC store as the
1861 // row's committed clock (so a later older remote edit gates out).
1862
1863 #[tokio::test]
1864 async fn assign_pending_hlcs_stamps_are_strictly_increasing_and_recorded_committed() {
1865 use crate::sync_service::hlc::{assign_pending_hlcs, load_committed_hlcs};
1866
1867 let pool = setup_test_db().await;
1868 let device_id = uuid::Uuid::new_v4();
1869
1870 // Seed a batch of unstamped pending rows in a known id order.
1871 let row_ids: Vec<String> = (0..6).map(|i| format!("row-{i}")).collect();
1872 for rid in &row_ids {
1873 sqlx::query(
1874 "INSERT INTO sync_changelog (table_name, op, row_id, pushed, hlc_wall, hlc_counter) \
1875 VALUES ('tasks', 'INSERT', ?, 0, NULL, NULL)",
1876 )
1877 .bind(rid)
1878 .execute(&pool)
1879 .await
1880 .unwrap();
1881 }
1882
1883 assign_pending_hlcs(&pool, device_id).await.unwrap();
1884
1885 // Read the stamps back in the order they were stamped (id ASC).
1886 let stamped: Vec<(i64, i64, String)> = sqlx::query_as(
1887 "SELECT hlc_wall, hlc_counter, row_id FROM sync_changelog \
1888 WHERE pushed = 0 ORDER BY id ASC",
1889 )
1890 .fetch_all(&pool)
1891 .await
1892 .unwrap();
1893 assert_eq!(stamped.len(), row_ids.len(), "every pending row is stamped");
1894
1895 // Each stamp is strictly greater than the previous by (wall, counter).
1896 for w in stamped.windows(2) {
1897 let (a_wall, a_ctr, _) = &w[0];
1898 let (b_wall, b_ctr, _) = &w[1];
1899 assert!(
1900 (*b_wall, *b_ctr) > (*a_wall, *a_ctr),
1901 "HLC stamps must be strictly increasing: {:?} then {:?}",
1902 (a_wall, a_ctr),
1903 (b_wall, b_ctr),
1904 );
1905 }
1906
1907 // Each row's committed HLC equals its stamp (recorded as the max), with this
1908 // device's node.
1909 let keys: Vec<(String, String)> = stamped
1910 .iter()
1911 .map(|(_, _, r)| ("tasks".to_string(), r.clone()))
1912 .collect();
1913 let committed = load_committed_hlcs(&pool, &keys).await.unwrap();
1914 for (wall, ctr, row_id) in &stamped {
1915 let hlc = committed
1916 .get(&("tasks".to_string(), row_id.clone()))
1917 .copied()
1918 .unwrap_or_else(|| panic!("no committed HLC for {row_id}"));
1919 assert_eq!(hlc.wall_ms, *wall);
1920 assert_eq!(hlc.counter as i64, *ctr);
1921 assert_eq!(hlc.node.as_uuid(), device_id, "committed HLC carries this device's node");
1922 }
1923
1924 // The persistent clock advanced exactly to the last stamp.
1925 let (clock_wall, clock_ctr): (i64, i64) =
1926 sqlx::query_as("SELECT wall_ms, counter FROM hlc_state WHERE id = 1")
1927 .fetch_one(&pool)
1928 .await
1929 .unwrap();
1930 let last = stamped.last().unwrap();
1931 assert_eq!(
1932 (clock_wall, clock_ctr),
1933 (last.0, last.1),
1934 "persistent clock advanced to the final stamp",
1935 );
1936 }
1937
1938 // ── Apply idempotency + committed-HLC gate backing ──────────────────────────
1939 //
1940 // Applying the same upsert twice must be a no-op: no duplicate row, no child
1941 // cascade, and the committed-HLC store still holds the change's HLC (max-kept),
1942 // so the CleanChanges gate would drop the already-seen change on a later pull.
1943
1944 #[tokio::test]
1945 async fn apply_changes_inner_upsert_is_idempotent_and_gate_holds_committed_hlc() {
1946 use crate::sync_service::hlc::load_committed_hlcs;
1947 use synckit_client::{ChangeEntry, ChangeOp, Hlc};
1948
1949 let pool = setup_test_db().await;
1950 let user_id = create_test_user(&pool).await;
1951 let task_id = uuid::Uuid::new_v4().to_string();
1952 let annotation_id = uuid::Uuid::new_v4().to_string();
1953 let now = now_sql();
1954 let node = uuid::Uuid::new_v4();
1955 let hlc = Hlc { wall_ms: 5000, counter: 2, node: synckit_client::DeviceId::new(node) };
1956
1957 let task_data = json!({
1958 "id": task_id, "project_id": null, "description": "Idempotent",
1959 "status": "Active", "priority": "Medium", "due": null, "tags": "",
1960 "urgency": 0.0, "recurrence": "None", "recurrence_rule": null,
1961 "created_at": now, "user_id": user_id, "recurrence_parent_id": null,
1962 "source_email_id": null, "snoozed_until": null, "waiting_for_response": 0,
1963 "waiting_since": null, "expected_response_date": null, "scheduled_start": null,
1964 "scheduled_duration": null, "is_focus": 0, "focus_set_at": null,
1965 "contact_id": null, "milestone_id": null, "completed_at": null,
1966 "estimated_minutes": null, "actual_minutes": null,
1967 });
1968
1969 let entry = ChangeEntry {
1970 table: "tasks".to_string(),
1971 op: ChangeOp::Insert,
1972 row_id: task_id.clone(),
1973 timestamp: chrono::Utc::now(),
1974 hlc,
1975 data: Some(task_data),
1976 extra: Default::default(),
1977 };
1978
1979 let mut conn = pool.acquire().await.unwrap();
1980 sqlx::query("PRAGMA foreign_keys = ON").execute(&mut *conn).await.unwrap();
1981
1982 // First apply: inserts the task and records its committed HLC.
1983 pull::apply_changes_inner(&mut conn, vec![entry.clone()]).await.unwrap();
1984
1985 // Attach a child annotation to prove a re-apply does not cascade-delete it.
1986 apply::apply_upsert(&mut conn, "annotations", &annotation_id, &json!({
1987 "id": annotation_id, "task_id": task_id, "timestamp": now, "note": "child",
1988 })).await.unwrap();
1989
1990 // Second apply of the identical change: a no-op, not an error.
1991 let skipped = pull::apply_changes_inner(&mut conn, vec![entry.clone()]).await.unwrap();
1992 assert_eq!(skipped, 0, "identical re-apply is not a skip/error");
1993
1994 // Exactly one task row, child annotation intact -- no duplicate, no cascade.
1995 let task_count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM tasks WHERE id = ?")
1996 .bind(&task_id).fetch_one(&pool).await.unwrap();
1997 assert_eq!(task_count.0, 1, "re-apply must not duplicate the row");
1998 let ann_count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM annotations WHERE task_id = ?")
1999 .bind(&task_id).fetch_one(&pool).await.unwrap();
2000 assert_eq!(ann_count.0, 1, "child annotation survives the re-apply");
2001
2002 // The committed-HLC store holds exactly the change's HLC (max-kept across both
2003 // applies). Its HLC is therefore not strictly greater than the committed clock,
2004 // so SyncKit's CleanChanges gate would drop this already-seen change next pull.
2005 let key = ("tasks".to_string(), task_id.clone());
2006 let committed = load_committed_hlcs(&pool, std::slice::from_ref(&key)).await.unwrap();
2007 let committed_hlc = committed.get(&key).copied().expect("committed HLC recorded");
2008 assert_eq!(committed_hlc, hlc);
2009 assert!(
2010 entry.hlc <= committed_hlc,
2011 "already-seen HLC is gated out (not newer than committed)",
2012 );
2013 }
2014
2015 // ── LWW clock-poison guard ──────────────────────────────────────────────────
2016 //
2017 // A remote HLC beyond MAX_HLC_DRIFT_MS in the future is clock-poisoned and, per
2018 // resolve_lww_at's (honest local, poisoned remote) => KeepLocal rule, cannot win
2019 // LWW over an honest local write -- the local value survives. resolve_lww_at
2020 // takes a PulledChange (which is #[non_exhaustive] and not constructible outside
2021 // synckit-client, so the full resolution runs in synckit-client's own conflict
2022 // tests); here we pin the load-bearing predicate GO's pull relies on.
2023
2024 #[test]
2025 fn clock_poison_guard_flags_far_future_remote_but_not_honest_local() {
2026 use synckit_client::Hlc;
2027 use synckit_client::conflict::{is_clock_poisoned, MAX_HLC_DRIFT_MS};
2028
2029 let now = chrono::Utc::now();
2030 let now_ms = now.timestamp_millis();
2031 let node = uuid::Uuid::new_v4();
2032
2033 // An honest local write at ~now is never flagged.
2034 let local = Hlc { wall_ms: now_ms, counter: 0, node: synckit_client::DeviceId::new(node) };
2035 assert!(!is_clock_poisoned(&local, now), "an at-now HLC must not be poisoned");
2036
2037 // Just inside the drift window is still honest (inter-device skew tolerated).
2038 let within = Hlc { wall_ms: now_ms + MAX_HLC_DRIFT_MS - 1_000, counter: 0, node: synckit_client::DeviceId::new(node) };
2039 assert!(!is_clock_poisoned(&within, now), "skew within the drift cap is allowed");
2040
2041 // A remote HLC far beyond the drift cap is poisoned. It sorts ABOVE the local
2042 // write, so it would win raw LWW ordering -- but the poison guard makes it lose,
2043 // so the local value survives.
2044 let poisoned = Hlc { wall_ms: now_ms + MAX_HLC_DRIFT_MS + 60_000, counter: 0, node: synckit_client::DeviceId::new(node) };
2045 assert!(is_clock_poisoned(&poisoned, now), "a far-future remote HLC must be flagged");
2046 assert!(poisoned > local, "the poisoned HLC would otherwise win raw LWW ordering");
2047 }
2048
2049 // ── bind_json_value ─────────────────────────────────────────────────────────
2050 //
2051 // The apply-side binder: JSON null (and an absent object field, which indexes to
2052 // Value::Null) must bind SQL NULL; scalars must bind through; a JSON bool binds
2053 // as an integer.
2054
2055 #[tokio::test]
2056 async fn bind_json_value_binds_null_absent_and_passthrough_values() {
2057 let pool = setup_test_db().await;
2058 let mut conn = pool.acquire().await.unwrap();
2059
2060 sqlx::query("CREATE TABLE bjv_probe (id INTEGER PRIMARY KEY, v)")
2061 .execute(&mut *conn).await.unwrap();
2062
2063 // Explicit JSON null -> SQL NULL.
2064 let null_val = serde_json::Value::Null;
2065 apply::bind_json_value(
2066 sqlx::query("INSERT INTO bjv_probe (id, v) VALUES (1, ?)"),
2067 &null_val,
2068 ).execute(&mut *conn).await.unwrap();
2069
2070 // Absent object field indexes to Value::Null -> SQL NULL.
2071 let obj = json!({"present": "x"});
2072 apply::bind_json_value(
2073 sqlx::query("INSERT INTO bjv_probe (id, v) VALUES (2, ?)"),
2074 &obj["missing"],
2075 ).execute(&mut *conn).await.unwrap();
2076
2077 // A normal string binds through unchanged.
2078 let s = json!("hello");
2079 apply::bind_json_value(
2080 sqlx::query("INSERT INTO bjv_probe (id, v) VALUES (3, ?)"),
2081 &s,
2082 ).execute(&mut *conn).await.unwrap();
2083
2084 // A JSON bool binds as integer (0/1).
2085 let b = json!(true);
2086 apply::bind_json_value(
2087 sqlx::query("INSERT INTO bjv_probe (id, v) VALUES (4, ?)"),
2088 &b,
2089 ).execute(&mut *conn).await.unwrap();
2090
2091 // CAST to TEXT so every affinity decodes uniformly (NULL stays NULL).
2092 let rows: Vec<(i64, Option<String>)> =
2093 sqlx::query_as("SELECT id, CAST(v AS TEXT) FROM bjv_probe ORDER BY id")
2094 .fetch_all(&pool).await.unwrap();
2095 assert_eq!(rows[0].1, None, "JSON null binds SQL NULL");
2096 assert_eq!(rows[1].1, None, "absent object field binds SQL NULL");
2097 assert_eq!(rows[2].1.as_deref(), Some("hello"), "string binds through");
2098 assert_eq!(rows[3].1.as_deref(), Some("1"), "bool binds as integer 1");
2099 }
2100