diff --git a/doctate-client-core/src/case_store.rs b/doctate-client-core/src/case_store.rs index ee51de1..781c771 100644 --- a/doctate-client-core/src/case_store.rs +++ b/doctate-client-core/src/case_store.rs @@ -5,12 +5,20 @@ //! keeps an in-memory copy and broadcasts a sorted snapshot on every //! write via [`tokio::sync::watch`]. //! -//! The merge-with-server-snapshot logic only **adds or updates** markers. -//! It never deletes: `/api/oneliners` is a sliding-window view, not a -//! full case inventory, so "missing from snapshot" is ambiguous (deleted -//! vs. fell out of the window). Cleanup is a separate, later concern. +//! Two mutation paths treat the server snapshot asymmetrically: +//! +//! - [`CaseStore::merge_server_snapshot`] **adds or updates** markers — +//! never deletes. `/api/oneliners` is a sliding-window view, so +//! "missing from snapshot" on its own is ambiguous (could be deleted +//! server-side, could simply have fallen out of the window). +//! - [`CaseStore::reconcile_with_server_snapshot`] **removes** synced +//! markers the server doesn't know about **anymore**, scoped to the +//! window reported by `OnelinersResponse.as_of - window_hours`. Pending +//! markers (`synced_to_server = false`) and markers older than the +//! window are preserved — their absence from the snapshot carries no +//! information about deletion. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -18,9 +26,9 @@ use doctate_common::oneliners::OnelinersResponse; use serde::{Deserialize, Serialize}; use thiserror::Error; use time::format_description::well_known::Rfc3339; -use time::OffsetDateTime; +use time::{Duration, OffsetDateTime}; use tokio::sync::{watch, Mutex}; -use tracing::{debug, warn}; +use tracing::{debug, info, warn}; use uuid::Uuid; /// On-disk representation of a single case known to this client. @@ -53,6 +61,8 @@ pub enum CaseStoreError { Serialize(#[from] serde_json::Error), #[error("format timestamp: {0}")] Format(#[from] time::error::Format), + #[error("parse timestamp: {0}")] + Parse(#[from] time::error::Parse), } type Snapshot = Arc>; @@ -218,6 +228,63 @@ impl CaseStore { Ok(()) } + /// Reconcile local markers against an authoritative server snapshot. + /// + /// Removes exactly those markers that are all of: + /// 1. `synced_to_server == true` (server has seen this case before), + /// 2. `last_activity_at >= as_of - window_hours` (inside the window + /// the response is authoritative for), and + /// 3. `case_id ∉ response.oneliners` (server no longer reports it). + /// + /// In that combination the server's silence is unambiguous: the + /// case was deleted (web UI soft-delete, or future hard-delete by a + /// retention sweep). Pending markers (`synced_to_server = false`) + /// guard unsent uploads and are left alone. Markers older than the + /// window are also preserved — their absence from the snapshot is + /// expected and says nothing about deletion. + /// + /// Uses the server-reported `as_of` (not the client clock) as the + /// window anchor, so clock skew between client and server cannot + /// cause spurious deletions. + pub async fn reconcile_with_server_snapshot( + &self, + response: &OnelinersResponse, + ) -> Result { + let as_of = OffsetDateTime::parse(&response.as_of, &Rfc3339)?; + let window_start = as_of - Duration::hours(response.window_hours as i64); + let window_start_str = format_utc(window_start)?; + + let server_ids: HashSet = response + .oneliners + .iter() + .filter_map(|e| Uuid::parse_str(&e.case_id).ok()) + .collect(); + + let mut state = self.state.lock().await; + let to_remove: Vec = state + .iter() + .filter(|(id, m)| { + m.synced_to_server + && m.last_activity_at >= window_start_str + && !server_ids.contains(id) + }) + .map(|(id, _)| *id) + .collect(); + let count = to_remove.len(); + for id in &to_remove { + let path = self.dir.join(format!("{id}.json")); + match tokio::fs::remove_file(&path).await { + Ok(_) => info!(case_id = %id, "marker removed (server reconcile: case missing from snapshot)"), + Err(e) => warn!(case_id = %id, error = %e, "failed to remove reconciled marker"), + } + state.remove(id); + } + if count > 0 { + self.broadcast(&state); + } + Ok(count) + } + /// Drop markers whose last activity is older than `cutoff` **and** /// that the server has already confirmed (`synced_to_server = true`). /// Pending/unsynced markers are preserved — they guard uploads that @@ -671,6 +738,144 @@ mod tests { assert_eq!(store.list().await.len(), 1); } + /// Helper: build a snapshot with an explicit `as_of` anchor. + fn snapshot_as_of(as_of: &str, entries: Vec) -> OnelinersResponse { + OnelinersResponse { + as_of: as_of.into(), + window_hours: 72, + oneliners: entries, + } + } + + /// Core case: a synced marker whose `last_activity_at` sits inside the + /// server's window, but whose `case_id` is missing from the snapshot → + /// that can only mean server-side deletion, so the marker goes. + #[tokio::test] + async fn reconcile_removes_synced_marker_missing_from_snapshot() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id = Uuid::new_v4(); + + // Seed a synced marker with recent activity. + store.create_local(id, t("2026-04-18T09:00:00Z")).await.unwrap(); + store.mark_synced(id).await.unwrap(); + + // Snapshot authoritative for [2026-04-15T10Z, 2026-04-18T10Z], + // but the case is not listed — server deleted it. + let snap = snapshot_as_of("2026-04-18T10:00:00Z", vec![]); + let removed = store.reconcile_with_server_snapshot(&snap).await.unwrap(); + + assert_eq!(removed, 1); + assert!(store.list().await.is_empty()); + let path = tmp.path().join(format!("{id}.json")); + assert!(!path.exists(), "marker file must be deleted"); + } + + /// Pending markers guard unsent uploads — they must never be touched + /// by reconciliation, even if the server knows nothing about them. + #[tokio::test] + async fn reconcile_keeps_pending_marker_missing_from_snapshot() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id = Uuid::new_v4(); + + // Pending: synced_to_server = false (default after create_local). + store.create_local(id, t("2026-04-18T09:00:00Z")).await.unwrap(); + + let snap = snapshot_as_of("2026-04-18T10:00:00Z", vec![]); + let removed = store.reconcile_with_server_snapshot(&snap).await.unwrap(); + + assert_eq!(removed, 0); + assert_eq!(store.list().await.len(), 1); + } + + /// A synced marker older than the window: server not listing it is + /// expected (fell out of the sliding window), not a deletion signal. + /// Must survive reconciliation. + #[tokio::test] + async fn reconcile_keeps_old_synced_marker_outside_window() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id = Uuid::new_v4(); + + // last_activity_at = 2026-04-10 (8 days ago); window = 72h anchor 2026-04-18T10Z + // → window_start = 2026-04-15T10Z. Marker is outside. + store.create_local(id, t("2026-04-10T10:00:00Z")).await.unwrap(); + store.mark_synced(id).await.unwrap(); + + let snap = snapshot_as_of("2026-04-18T10:00:00Z", vec![]); + let removed = store.reconcile_with_server_snapshot(&snap).await.unwrap(); + + assert_eq!(removed, 0); + assert_eq!(store.list().await.len(), 1); + } + + /// When every local marker is present in the snapshot, reconciliation + /// is a pure no-op. Important negative test — prevents accidental + /// over-deletion from a buggy predicate. + #[tokio::test] + async fn reconcile_is_noop_when_all_markers_present() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id_a = Uuid::new_v4(); + let id_b = Uuid::new_v4(); + store.create_local(id_a, t("2026-04-18T09:00:00Z")).await.unwrap(); + store.create_local(id_b, t("2026-04-18T09:30:00Z")).await.unwrap(); + store.mark_synced(id_a).await.unwrap(); + store.mark_synced(id_b).await.unwrap(); + + let snap = snapshot_as_of( + "2026-04-18T10:00:00Z", + vec![ + server_entry(id_a, Some("a"), "2026-04-18T09:00:00Z"), + server_entry(id_b, Some("b"), "2026-04-18T09:30:00Z"), + ], + ); + let removed = store.reconcile_with_server_snapshot(&snap).await.unwrap(); + + assert_eq!(removed, 0); + assert_eq!(store.list().await.len(), 2); + } + + /// The window anchor is the server's `as_of`, not the local clock. + /// A marker that would look "old" against `now` but sits inside the + /// server's authoritative window must be reconciled correctly based + /// on the server timestamp. + #[tokio::test] + async fn reconcile_uses_server_as_of_not_client_clock() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id = Uuid::new_v4(); + + // Marker last_activity_at 2026-04-18T09:00Z, synced. + store.create_local(id, t("2026-04-18T09:00:00Z")).await.unwrap(); + store.mark_synced(id).await.unwrap(); + + // Server reports as_of 2026-04-18T10:00Z (1h after marker, + // clearly inside 72h window) and does NOT list the case. + let snap = snapshot_as_of("2026-04-18T10:00:00Z", vec![]); + assert_eq!( + store.reconcile_with_server_snapshot(&snap).await.unwrap(), + 1, + "inside server window → must be removed" + ); + + // Seed a second case for the symmetric assertion. + let id2 = Uuid::new_v4(); + store.create_local(id2, t("2026-04-18T09:00:00Z")).await.unwrap(); + store.mark_synced(id2).await.unwrap(); + + // Now the server reports a much later as_of. The marker's + // last_activity_at is further than 72h before as_of, so it's + // *outside* the authoritative window → keep. + let snap2 = snapshot_as_of("2026-04-30T10:00:00Z", vec![]); + assert_eq!( + store.reconcile_with_server_snapshot(&snap2).await.unwrap(), + 0, + "outside server window → must survive" + ); + } + #[tokio::test] async fn mark_synced_flips_flag_without_touching_activity() { let tmp = TempDir::new().unwrap(); diff --git a/doctate-client-core/src/server_sync.rs b/doctate-client-core/src/server_sync.rs index 84d0a61..742bf36 100644 --- a/doctate-client-core/src/server_sync.rs +++ b/doctate-client-core/src/server_sync.rs @@ -428,7 +428,13 @@ pub(crate) async fn poll_once( StatusCode::OK => { let new_etag = extract_etag(&resp); let body: OnelinersResponse = resp.json().await?; + // Merge adds/updates markers the server still knows about; + // reconcile removes the ones it no longer does (within the + // window reported by `as_of` / `window_hours`). Running both + // after every successful poll keeps the client UI in step + // with server-side deletions without any extra round-trips. store.merge_server_snapshot(&body).await?; + store.reconcile_with_server_snapshot(&body).await?; if let Some(tag) = new_etag.clone() { let cached = CachedSnapshot { @@ -484,6 +490,7 @@ mod tests { use crate::upload::write_sidecar; use doctate_common::oneliners::OnelinerEntry; use tempfile::TempDir; + use time::format_description::well_known::Rfc3339; use uuid::Uuid; use wiremock::matchers::{header, method, path as wpath}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -597,6 +604,59 @@ mod tests { )); } + /// End-to-end reconcile: a synced marker that the server omits from + /// its next snapshot must vanish locally after a single poll. This + /// is the scenario the user reported ("deleted cases still show up + /// on the client") — the wiring is: `poll_once` → `merge` → `reconcile`. + #[tokio::test] + async fn poll_once_reconciles_missing_synced_marker() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(wpath(ONELINERS_PATH)) + .respond_with( + ResponseTemplate::new(200) + .insert_header("etag", TEST_ETAG) + // Authoritative snapshot with zero entries — any + // synced local marker inside the window is deleted + // server-side. + .set_body_json(OnelinersResponse { + as_of: "2026-04-18T10:00:00Z".into(), + window_hours: 72, + oneliners: vec![], + }), + ) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let (store, cache) = make_store_and_cache(&tmp).await; + + // Seed a synced marker inside the window. + let stale_id = Uuid::new_v4(); + store + .create_local( + stale_id, + time::OffsetDateTime::parse("2026-04-18T09:00:00Z", &Rfc3339).unwrap(), + ) + .await + .unwrap(); + store.mark_synced(stale_id).await.unwrap(); + assert_eq!(store.list().await.len(), 1); + + // Poll once — the merge produces no add/update, reconcile drops + // the stale marker. + let http = reqwest::Client::new(); + let mut etag: Option = None; + let outcome = poll_once(&http, &cfg(&server), &store, &cache, &mut etag) + .await + .unwrap(); + assert_eq!(outcome, PollOutcome::Success); + assert!( + store.list().await.is_empty(), + "reconcile must remove the synced marker the server no longer lists" + ); + } + #[tokio::test] async fn poll_once_200_merges_and_caches() { let server = MockServer::start().await; diff --git a/server/src/routes/upload.rs b/server/src/routes/upload.rs index f6f1dc4..e6541e8 100644 --- a/server/src/routes/upload.rs +++ b/server/src/routes/upload.rs @@ -72,6 +72,19 @@ pub async fn handle_upload( // Find or create the case directory. let case_dir = resolve_case_dir(&user.data_dir, &case_id).await?; + // Auto-reopen soft-deleted cases: a new upload clears the `.deleted` + // marker so the case reappears in /api/oneliners and the web UI. + // Hard-deleted cases (directory already gone) are handled transparently + // by `resolve_case_dir` above — nothing to clear. + if crate::paths::is_deleted(&case_dir).await { + tokio::fs::remove_file(case_dir.join(crate::paths::DELETE_MARKER)).await?; + info!( + user = %user.slug, + case_id = %case_id, + "Reopened soft-deleted case via new upload" + ); + } + // Build a filesystem-safe filename from the timestamp (colons → hyphens). let safe_timestamp = recorded_at_to_filename_stem(&recorded_at); let file_path = case_dir.join(format!("{safe_timestamp}.m4a")); diff --git a/server/tests/upload_test.rs b/server/tests/upload_test.rs index 87a683e..34d047b 100644 --- a/server/tests/upload_test.rs +++ b/server/tests/upload_test.rs @@ -6,6 +6,7 @@ use axum::http::{Request, StatusCode}; use tower::util::ServiceExt; use doctate_server::config::{Config, User}; +use doctate_server::paths::{write_delete_marker, DeleteMarker, DELETE_MARKER}; fn test_config() -> Arc { let data_path = std::env::temp_dir().join(format!( @@ -250,3 +251,142 @@ async fn upload_second_recording_same_case() { let _ = std::fs::remove_dir_all(&data_path); } +/// Scenario A: a new upload to a *soft-deleted* case must silently +/// remove the `.deleted` marker and store the new audio. Reopens the +/// case so it reappears in /api/oneliners without any manual "undo". +#[tokio::test] +async fn upload_reopens_soft_deleted_case() { + let config = test_config(); + let data_path = config.data_path.clone(); + let app = doctate_server::create_router(config); + + let case_id = "770e8400-e29b-41d4-a716-446655440000"; + + // First upload — creates the case directory. + let (boundary, body) = multipart_body(case_id, "2026-04-13T10:30:00Z", b"first recording"); + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/api/upload") + .header("X-API-Key", "test-key-123") + .header( + "Content-Type", + format!("multipart/form-data; boundary={boundary}"), + ) + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + // Soft-delete the case by writing the marker directly — simulates + // the user hitting the delete button in the web UI without having + // to drive the /web/cases/{id}/delete handler here. + let case_dir = data_path.join("dr_test").join(case_id); + let marker = DeleteMarker { + batch: uuid::Uuid::new_v4(), + deleted_at: "2026-04-13T11:00:00Z".into(), + }; + write_delete_marker(&case_dir, &marker).await.unwrap(); + assert!(case_dir.join(DELETE_MARKER).exists()); + + // Second upload — must auto-reopen. + let (boundary, body) = multipart_body(case_id, "2026-04-13T12:00:00Z", b"reopen recording"); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/upload") + .header("X-API-Key", "test-key-123") + .header( + "Content-Type", + format!("multipart/form-data; boundary={boundary}"), + ) + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + // Marker is gone → case is visible again. + assert!( + !case_dir.join(DELETE_MARKER).exists(), + ".deleted marker must be removed after re-upload" + ); + // Both audios are on disk — the old one was not touched. + assert!(case_dir.join("2026-04-13T10-30-00Z.m4a").exists()); + assert!(case_dir.join("2026-04-13T12-00-00Z.m4a").exists()); + + let _ = std::fs::remove_dir_all(&data_path); +} + +/// Scenario B: a new upload to a case whose server-side directory has +/// been *hard-deleted* (e.g. by a future retention sweep or an +/// out-of-band cleanup) must transparently re-create the directory and +/// accept the audio. This guards the forward-looking scenario where the +/// client was offline long enough that the server has already purged +/// the case. +#[tokio::test] +async fn upload_resurrects_hard_deleted_case() { + let config = test_config(); + let data_path = config.data_path.clone(); + let app = doctate_server::create_router(config); + + let case_id = "880e8400-e29b-41d4-a716-446655440000"; + + // First upload. + let (boundary, body) = multipart_body(case_id, "2026-04-13T10:30:00Z", b"original recording"); + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/api/upload") + .header("X-API-Key", "test-key-123") + .header( + "Content-Type", + format!("multipart/form-data; boundary={boundary}"), + ) + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + // Wipe the case directory entirely — no marker, no audio, nothing. + let case_dir = data_path.join("dr_test").join(case_id); + std::fs::remove_dir_all(&case_dir).unwrap(); + assert!(!case_dir.exists()); + + // Second upload with the same client-generated UUID. + let (boundary, body) = multipart_body(case_id, "2026-04-13T12:00:00Z", b"resurrected recording"); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/upload") + .header("X-API-Key", "test-key-123") + .header( + "Content-Type", + format!("multipart/form-data; boundary={boundary}"), + ) + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + // Directory is back, new audio is inside, old audio stays gone. + assert!(case_dir.exists()); + assert!(case_dir.join("2026-04-13T12-00-00Z.m4a").exists()); + assert!(!case_dir.join("2026-04-13T10-30-00Z.m4a").exists()); + + let _ = std::fs::remove_dir_all(&data_path); +} +