From a1ff410d1b119fcbf3e5443395c9c49c391ff4c8 Mon Sep 17 00:00:00 2001 From: Brummel Date: Sun, 19 Apr 2026 14:47:30 +0200 Subject: [PATCH] Refactor ETag generation for oneliners API Remove the per-user oneliner watermark, which was previously used to generate ETags for the `/api/oneliners` endpoint. The watermark was intended to track the last successful oneliner write for each user. The ETag generation has been refactored to use a deterministic FNV-1a hash. This hash is computed over a sorted list of tuples containing `(case_id, last_recording_at_ns, oneliner_mtime_ns)` for all visible cases. This approach ensures that the ETag changes whenever any listen-relevant file system modification occurs, such as a new recording, oneliner regeneration, soft-delete, undo, or upload-auto-reopen. This new method is more robust and avoids the complexity of managing per-user state. The `OnelinerWatermark` type alias and its associated logic have been removed from `AppState` and the relevant modules (`transcribe::worker`, `transcribe::recovery`, `routes::oneliners`, `main`). New integration tests have been added to verify that various delete operations (single delete, undo delete, bulk delete) correctly trigger an ETag update, ensuring cache invalidation. --- doctate-client-core/src/case_store.rs | 8 +- server/src/lib.rs | 18 -- server/src/main.rs | 14 +- server/src/routes/oneliners.rs | 115 ++++++++++-- server/src/routes/upload.rs | 4 +- server/src/transcribe/recovery.rs | 8 +- server/src/transcribe/worker.rs | 20 +-- server/tests/delete_watermark_test.rs | 250 ++++++++++++++++++++++++++ server/tests/oneliners_api_test.rs | 3 +- server/tests/transcribe_test.rs | 10 +- server/tests/upload_test.rs | 55 +++++- 11 files changed, 425 insertions(+), 80 deletions(-) create mode 100644 server/tests/delete_watermark_test.rs diff --git a/doctate-client-core/src/case_store.rs b/doctate-client-core/src/case_store.rs index 781c771..fdfa946 100644 --- a/doctate-client-core/src/case_store.rs +++ b/doctate-client-core/src/case_store.rs @@ -41,8 +41,12 @@ pub struct CaseMarker { /// updates. Drives the UI sort order (newest activity first). pub last_activity_at: String, /// `true` once this case has been seen in any `/api/oneliners` - /// response. Reserved for future delete-propagation logic; today only - /// useful for diagnostics. + /// response or after a successful upload ACK. Drives two clean-up + /// paths: [`CaseStore::reconcile_with_server_snapshot`] only removes + /// synced markers the server no longer lists, and + /// [`CaseStore::cleanup_stale`] only ages out synced markers. Pending + /// (`= false`) markers guard unsent uploads and are never touched by + /// either — losing one would silently erase a patient recording. pub synced_to_server: bool, /// `None` while transcription / oneliner generation is pending; set /// from the server snapshot when available. diff --git a/server/src/lib.rs b/server/src/lib.rs index 1a615fc..4ca8123 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -9,27 +9,17 @@ pub mod routes; pub mod transcribe; pub mod web_session; -use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use axum::extract::FromRef; use axum::Router; -use time::OffsetDateTime; -use tokio::sync::RwLock; use analyze::AnalyzeSender; use config::Config; use transcribe::TranscribeSender; use web_session::SessionStore; -/// Per-user UTC timestamp of the last successful oneliner write. -/// Read on every `/api/oneliners` request to build the ETag; written -/// from the transcribe worker and the boot recovery scan. Missing entry -/// → the user has no known oneliner activity yet; the endpoint returns -/// a full response and sets an ETag based on a `0` seed. -pub type OnelinerWatermark = Arc>>; - /// Live "is this worker currently processing a job?" flag. /// Set true before each job, false after. UI uses it to distinguish a real /// in-flight job from a stale on-disk marker (orphan input, missing @@ -70,7 +60,6 @@ pub struct AppState { pub session_store: SessionStore, pub analyze_busy: AnalyzeBusy, pub transcribe_busy: TranscribeBusy, - pub oneliner_watermark: OnelinerWatermark, } impl FromRef for Arc { @@ -109,12 +98,6 @@ impl FromRef for TranscribeBusy { } } -impl FromRef for OnelinerWatermark { - fn from_ref(state: &AppState) -> Self { - state.oneliner_watermark.clone() - } -} - /// Test/simple entrypoint: jobs pushed into either channel are dropped /// because the receivers are not retained. Use [`create_router_with_state`] /// from `main.rs` where real workers own the receivers. @@ -128,7 +111,6 @@ pub fn create_router(config: Arc) -> Router { session_store: web_session::new_store(), analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))), transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), - oneliner_watermark: Arc::new(RwLock::new(HashMap::new())), }) } diff --git a/server/src/main.rs b/server/src/main.rs index 52a180a..d971318 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -12,7 +12,7 @@ use doctate_server::analyze; use doctate_server::config::Config; use doctate_server::gazetteer::{Gazetteer, SpellbookDict}; use doctate_server::transcribe; -use doctate_server::{AppState, OnelinerWatermark}; +use doctate_server::AppState; #[tokio::main] async fn main() { @@ -110,13 +110,6 @@ async fn main() { } }); - // Oneliner watermark: per-user "last successful oneliner write (UTC)". - // Feeds the ETag on `/api/oneliners`. Shared across the transcribe - // worker (writer) and the oneliners route handler (reader). - let oneliner_watermark: OnelinerWatermark = Arc::new(tokio::sync::RwLock::new( - std::collections::HashMap::new(), - )); - // Transcription pipeline: channel + worker + recovery scan. let (transcribe_tx, transcribe_rx) = transcribe::channel(); let transcribe_busy: doctate_server::WorkerBusy = @@ -127,7 +120,6 @@ async fn main() { http_client.clone(), transcribe_busy.clone(), vocab.clone(), - oneliner_watermark.clone(), )); { let tx = transcribe_tx.clone(); @@ -135,13 +127,12 @@ async fn main() { let client = http_client.clone(); let cfg = config.clone(); let vocab = vocab.clone(); - let watermark = oneliner_watermark.clone(); tokio::spawn(async move { transcribe::recovery::scan_and_enqueue(&data_path, &tx).await; // Then catch up oneliners for cases that crashed between // transcript-write and oneliner-write in a previous run. transcribe::recovery::regenerate_missing_oneliners( - &data_path, &client, &cfg, &vocab, &watermark, + &data_path, &client, &cfg, &vocab, ) .await; }); @@ -173,7 +164,6 @@ async fn main() { session_store: doctate_server::web_session::new_store(), analyze_busy: doctate_server::AnalyzeBusy(analyze_busy), transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy), - oneliner_watermark, }; let addr = format!("0.0.0.0:{}", config.server_port); diff --git a/server/src/routes/oneliners.rs b/server/src/routes/oneliners.rs index 155dbcc..d278ab5 100644 --- a/server/src/routes/oneliners.rs +++ b/server/src/routes/oneliners.rs @@ -5,16 +5,22 @@ //! text (or null if not yet generated), and UTC timestamps for when the //! case started and when its oneliner was last written. //! -//! Conditional-GET: the response carries a weak ETag `W/"-"` -//! derived from the per-user watermark. Clients that send back a matching -//! `If-None-Match` get `304 Not Modified` without touching the filesystem. +//! Conditional-GET: the response carries a weak ETag +//! `W/"-"`. The fingerprint is a deterministic +//! FNV-1a hash over the sorted tuples +//! `(case_id, last_recording_at_ns, oneliner_mtime_ns)` of every +//! *visible* case. Any listen-relevant FS change — new recording, +//! oneliner regeneration, soft-delete, undo, upload-auto-reopen — +//! shifts at least one tuple and the hash moves. Server restart is a +//! no-op: same FS, same hash. +//! //! The `hours` component is included so a client swapping window sizes //! does not get a stale 304. use std::path::{Path, PathBuf}; use std::time::SystemTime; -use axum::extract::{Query, State}; +use axum::extract::Query; use axum::http::{header, HeaderMap, StatusCode}; use axum::response::{IntoResponse, Response}; use axum::Json; @@ -26,7 +32,6 @@ use tracing::warn; use crate::auth::AuthenticatedUser; use crate::error::AppError; -use crate::OnelinerWatermark; const DEFAULT_HOURS: u32 = 16; const MAX_HOURS: u32 = 168; @@ -38,7 +43,6 @@ pub struct OnelinersQuery { pub async fn handle_oneliners( user: AuthenticatedUser, - State(watermark): State, Query(params): Query, headers: HeaderMap, ) -> Result { @@ -46,18 +50,21 @@ pub async fn handle_oneliners( let now = OffsetDateTime::now_utc(); let since = now - Duration::hours(hours as i64); - let current_watermark = watermark.read().await.get(&user.slug).copied(); - let etag = build_etag(current_watermark, hours); + // One scan pass yields both the case list and the fingerprint — + // the fingerprint is the authoritative signal for "has the view + // changed?". Body serialization is skipped if the client already + // has this version (304). + let scan = scan_cases(&user.data_dir, since).await; + let etag = build_etag(scan.fingerprint, hours); if client_has_matching_etag(&headers, &etag) { return Ok(not_modified(&etag)); } - let oneliners = enumerate_recent_oneliners(&user.data_dir, since).await; let body = OnelinersResponse { as_of: now.format(&Rfc3339).unwrap_or_default(), window_hours: hours, - oneliners, + oneliners: scan.entries, }; let mut response = Json(body).into_response(); @@ -67,9 +74,8 @@ pub async fn handle_oneliners( Ok(response) } -fn build_etag(watermark: Option, hours: u32) -> String { - let seconds = watermark.map(|t| t.unix_timestamp()).unwrap_or(0); - format!("W/\"{seconds}-{hours}\"") +fn build_etag(fingerprint: u64, hours: u32) -> String { + format!("W/\"{fingerprint:016x}-{hours}\"") } /// Case-sensitive match against the raw `If-None-Match` header value. @@ -91,19 +97,36 @@ fn not_modified(etag: &str) -> Response { response } +/// Output of [`scan_cases`]: the sorted entry list that becomes the +/// response body, and a fingerprint derived from the same scan. +struct ScanResult { + entries: Vec, + fingerprint: u64, +} + /// Scan `//` for cases whose latest recording /// sits inside the `[since, now]` window. Returns entries sorted by /// `last_recording_at` descending (most recently active first); cases /// without recordings fall back to `created_at`. Soft-deleted cases and /// non-UUID directories are filtered out. -async fn enumerate_recent_oneliners( - user_data_dir: &Path, - since: OffsetDateTime, -) -> Vec { +/// +/// Alongside the entries we compute a deterministic fingerprint +/// (`fnv1a_64`) over the sorted set of +/// `(case_id, last_recording_at_ns, oneliner_mtime_ns)` tuples — cheap +/// to compute during the scan, and it changes exactly when the visible +/// list changes. No RAM-local state, so server restart is a no-op. +async fn scan_cases(user_data_dir: &Path, since: OffsetDateTime) -> ScanResult { let mut entries: Vec = Vec::new(); let Ok(mut cases) = tokio::fs::read_dir(user_data_dir).await else { - return entries; + return ScanResult { + entries, + fingerprint: fnv1a_64(b""), + }; }; + // Parallel array keyed 1:1 with `entries` carrying the raw nanosecond + // timestamps needed for the fingerprint. Kept separate so the + // response DTO stays flat. + let mut fp_tuples: Vec<(String, i128, i128)> = Vec::new(); while let Ok(Some(case_entry)) = cases.next_entry().await { let case_dir = case_entry.path(); if !case_dir.is_dir() { @@ -139,6 +162,10 @@ async fn enumerate_recent_oneliners( let last_recording_at_str = last_recording_at.format(&Rfc3339).ok(); let updated_at_str = updated_at.and_then(|t| t.format(&Rfc3339).ok()); + let last_rec_ns = last_recording_at.unix_timestamp_nanos(); + let oneliner_ns = updated_at.map(|t| t.unix_timestamp_nanos()).unwrap_or(0); + fp_tuples.push((case_id.to_owned(), last_rec_ns, oneliner_ns)); + entries.push(OnelinerEntry { case_id: case_id.to_owned(), oneliner: oneliner_text, @@ -156,9 +183,59 @@ async fn enumerate_recent_oneliners( let b_key = b.last_recording_at.as_deref().unwrap_or(&b.created_at); b_key.cmp(a_key) }); - entries + + // Fingerprint: independent of entries' display order, so sort the + // tuple list by case_id and hash. + fp_tuples.sort_by(|a, b| a.0.cmp(&b.0)); + let fingerprint = fingerprint_tuples(&fp_tuples); + + ScanResult { + entries, + fingerprint, + } } +/// FNV-1a 64-bit over the concatenation of the tuples. Stable across +/// server restarts — no `RandomState`, no HashMap hashing. Ample +/// resistance against accidental collisions for ETag purposes. +fn fingerprint_tuples(tuples: &[(String, i128, i128)]) -> u64 { + let mut h: u64 = FNV_OFFSET; + for (case_id, last_rec_ns, oneliner_ns) in tuples { + for b in case_id.as_bytes() { + h ^= *b as u64; + h = h.wrapping_mul(FNV_PRIME); + } + // Separator so `"ab" || "12"` ≠ `"a" || "b12"`. + h ^= b'|' as u64; + h = h.wrapping_mul(FNV_PRIME); + for b in last_rec_ns.to_le_bytes() { + h ^= b as u64; + h = h.wrapping_mul(FNV_PRIME); + } + h ^= b'|' as u64; + h = h.wrapping_mul(FNV_PRIME); + for b in oneliner_ns.to_le_bytes() { + h ^= b as u64; + h = h.wrapping_mul(FNV_PRIME); + } + h ^= b'\n' as u64; + h = h.wrapping_mul(FNV_PRIME); + } + h +} + +fn fnv1a_64(bytes: &[u8]) -> u64 { + let mut h: u64 = FNV_OFFSET; + for b in bytes { + h ^= *b as u64; + h = h.wrapping_mul(FNV_PRIME); + } + h +} + +const FNV_OFFSET: u64 = 0xcbf29ce4_84222325; +const FNV_PRIME: u64 = 0x00000100_000001b3; + /// Return `(earliest, latest)` `.m4a` mtime pair in `case_dir`. /// `.m4a.failed` entries count — a failed recording still marks activity. /// `None` on empty directory or read error. A single recording yields diff --git a/server/src/routes/upload.rs b/server/src/routes/upload.rs index e6541e8..45df68f 100644 --- a/server/src/routes/upload.rs +++ b/server/src/routes/upload.rs @@ -75,7 +75,9 @@ pub async fn handle_upload( // 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. + // by `resolve_case_dir` above — nothing to clear. No explicit ETag + // bump is needed: the subsequent m4a write changes `last_recording_at`, + // which flows into the `/api/oneliners` fingerprint automatically. if crate::paths::is_deleted(&case_dir).await { tokio::fs::remove_file(case_dir.join(crate::paths::DELETE_MARKER)).await?; info!( diff --git a/server/src/transcribe/recovery.rs b/server/src/transcribe/recovery.rs index 1e27998..c7ddeab 100644 --- a/server/src/transcribe/recovery.rs +++ b/server/src/transcribe/recovery.rs @@ -5,8 +5,6 @@ use tracing::{info, warn}; use super::{TranscribeJob, TranscribeSender}; use crate::config::Config; use crate::gazetteer::Gazetteer; -use crate::OnelinerWatermark; - /// Walk `data_path/*/` and enqueue every recording pending transcription for /// every user. Intended for startup; the same primitive backs the per-user /// self-heal triggered by web handlers. @@ -87,8 +85,7 @@ pub async fn enqueue_pending_for_user( /// one non-empty `*.transcript.txt` but no `oneliner.txt`, paired with the /// user-slug (parent directory name). Pure filesystem scan, no LLM calls — /// separated from the regeneration wrapper so it can be unit-tested without -/// mocking Ollama. The slug is carried through so the caller can bump the -/// per-user watermark when regeneration succeeds. +/// mocking Ollama. pub(crate) async fn cases_needing_oneliner(data_path: &Path) -> Vec<(PathBuf, String)> { let mut out: Vec<(PathBuf, String)> = Vec::new(); let Ok(mut users) = tokio::fs::read_dir(data_path).await else { @@ -152,7 +149,6 @@ pub async fn regenerate_missing_oneliners( client: &reqwest::Client, config: &Config, vocab: &Gazetteer, - watermark: &OnelinerWatermark, ) { let cases = cases_needing_oneliner(data_path).await; if cases.is_empty() { @@ -160,7 +156,7 @@ pub async fn regenerate_missing_oneliners( } info!(count = cases.len(), "Regenerating missing oneliners"); for (case_dir, slug) in cases { - super::worker::update_oneliner(&case_dir, &slug, client, config, vocab, watermark).await; + super::worker::update_oneliner(&case_dir, &slug, client, config, vocab).await; } } diff --git a/server/src/transcribe/worker.rs b/server/src/transcribe/worker.rs index d42a5c9..b768d2f 100644 --- a/server/src/transcribe/worker.rs +++ b/server/src/transcribe/worker.rs @@ -2,13 +2,12 @@ use std::path::Path; use std::sync::Arc; use std::time::Duration; -use time::OffsetDateTime; use tracing::{error, info, warn}; use super::{ffmpeg, ollama, whisper, TranscribeReceiver}; use crate::config::{Config, WhisperUserSettings}; use crate::gazetteer::Gazetteer; -use crate::{BusyGuard, OnelinerWatermark, WorkerBusy}; +use crate::{BusyGuard, WorkerBusy}; const ONELINER_TIMEOUT: Duration = Duration::from_secs(60); @@ -20,7 +19,6 @@ pub async fn run( client: reqwest::Client, worker_busy: WorkerBusy, vocab: Arc, - watermark: OnelinerWatermark, ) { info!("Transcription worker started"); let timeout = Duration::from_secs(config.whisper_timeout_seconds); @@ -90,7 +88,7 @@ pub async fn run( if let Some(case_dir) = audio_path.parent() && !has_pending_recordings(case_dir).await { - update_oneliner(case_dir, &job.user_slug, &client, &config, &vocab, &watermark).await; + update_oneliner(case_dir, &job.user_slug, &client, &config, &vocab).await; } } @@ -133,7 +131,6 @@ pub(crate) async fn update_oneliner( client: &reqwest::Client, config: &Config, vocab: &Gazetteer, - watermark: &OnelinerWatermark, ) { let path = case_dir.join("oneliner.txt"); @@ -157,13 +154,12 @@ pub(crate) async fn update_oneliner( if let Err(e) = tokio::fs::write(&path, &line).await { error!(path = %path.display(), error = %e, "writing oneliner failed"); } else { - info!(path = %path.display(), chars = line.chars().count(), "Oneliner updated"); - // Bump the per-user watermark so the next `/api/oneliners` - // poll sees a fresh ETag and fetches the new content. - watermark - .write() - .await - .insert(user_slug.to_string(), OffsetDateTime::now_utc()); + info!( + user = %user_slug, + path = %path.display(), + chars = line.chars().count(), + "Oneliner updated" + ); } } Err(e) => { diff --git a/server/tests/delete_watermark_test.rs b/server/tests/delete_watermark_test.rs new file mode 100644 index 0000000..55d9088 --- /dev/null +++ b/server/tests/delete_watermark_test.rs @@ -0,0 +1,250 @@ +//! Listen-relevante Mutationen (Einzel-Delete, Undo, Bulk-Delete) müssen +//! den pro-User Watermark bumpen. Sonst bleibt der ETag von +//! `/api/oneliners` stehen, jeder Poll liefert 304, und der Client +//! läuft dauerhaft mit einem Marker für einen Fall, den der Server +//! bereits gelöscht hat (ursprünglicher Bugreport: "Polymyalgia +//! rheumatica" blieb im Client sichtbar, obwohl im Web-UI bereits +//! gelöscht). + +use std::collections::HashMap; +use std::sync::Arc; + +use axum::body::Body; +use axum::http::{header, Request, StatusCode}; +use doctate_common::constants::API_KEY_HEADER; +use doctate_common::oneliners::ONELINERS_PATH; +use doctate_server::config::{Config, User}; +use tower::util::ServiceExt; + +fn make_user(slug: &str, password_plain: &str) -> User { + User { + slug: slug.into(), + api_key: format!("key-{slug}"), + web_password: bcrypt::hash(password_plain, 4).unwrap(), + role: "doctor".into(), + whisper: Default::default(), + } +} + +fn test_config_with_users(users: Vec) -> Arc { + let data_path = std::env::temp_dir().join(format!( + "doctate-delete-wm-test-{}-{}", + std::process::id(), + uuid::Uuid::new_v4() + )); + let api_keys: HashMap = + users.iter().map(|u| (u.api_key.clone(), u.slug.clone())).collect(); + Arc::new(Config { + data_path, + users, + api_keys, + ..Config::test_default() + }) +} + +fn extract_session_cookie(resp: &axum::response::Response) -> Option { + for v in resp.headers().get_all(header::SET_COOKIE).iter() { + let s = v.to_str().ok()?; + if let Some(pair) = s.split(';').next() + && pair.starts_with("session=") + { + return Some(pair.to_string()); + } + } + None +} + +fn login_request(slug: &str, password: &str) -> Request { + let body = format!("slug={slug}&password={password}"); + Request::builder() + .method("POST") + .uri("/web/login") + .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded") + .body(Body::from(body)) + .unwrap() +} + +fn oneliners_request(api_key: &str, if_none_match: Option<&str>) -> Request { + let mut b = Request::builder() + .method("GET") + .uri(ONELINERS_PATH) + .header(API_KEY_HEADER, api_key); + if let Some(tag) = if_none_match { + b = b.header(header::IF_NONE_MATCH, tag); + } + b.body(Body::empty()).unwrap() +} + +/// Create `///.m4a` so the case is +/// visible to `enumerate_recent_oneliners` (window defaults to 16h; +/// the fake mtime is "now" because write happens right now). +fn seed_case(data_path: &std::path::Path, slug: &str, case_id: &str) { + let case_dir = data_path.join(slug).join(case_id); + std::fs::create_dir_all(&case_dir).unwrap(); + std::fs::write(case_dir.join("2026-04-19T10-00-00Z.m4a"), b"fake audio").unwrap(); +} + +async fn capture_etag( + app: &axum::Router, + api_key: &str, + if_none_match: Option<&str>, +) -> (StatusCode, Option) { + let resp = app + .clone() + .oneshot(oneliners_request(api_key, if_none_match)) + .await + .unwrap(); + let status = resp.status(); + let etag = resp + .headers() + .get(header::ETAG) + .and_then(|v| v.to_str().ok()) + .map(str::to_owned); + (status, etag) +} + +#[tokio::test] +async fn delete_bumps_watermark() { + let config = test_config_with_users(vec![make_user("dr_a", "s")]); + let data_path = config.data_path.clone(); + let case_id = "550e8400-e29b-41d4-a716-446655440000"; + seed_case(&data_path, "dr_a", case_id); + + let app = doctate_server::create_router(config); + + let login = app.clone().oneshot(login_request("dr_a", "s")).await.unwrap(); + let cookie = extract_session_cookie(&login).expect("login should set cookie"); + + // First GET: capture baseline ETag (watermark empty → seeded on 0). + let (s1, etag_before) = capture_etag(&app, "key-dr_a", None).await; + assert_eq!(s1, StatusCode::OK); + let etag_before = etag_before.expect("server must set ETag"); + + // Soft-delete via the web handler. + let del = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/web/cases/{case_id}/delete")) + .header(header::COOKIE, &cookie) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + del.status().as_u16() / 100, + 3, + "delete should redirect (3xx)" + ); + + // Same If-None-Match now must NOT match — watermark was bumped. + let (s2, etag_after) = capture_etag(&app, "key-dr_a", Some(&etag_before)).await; + assert_eq!( + s2, + StatusCode::OK, + "watermark bump must invalidate cached ETag", + ); + let etag_after = etag_after.expect("200 response must set ETag"); + assert_ne!( + etag_after, etag_before, + "delete handler must change the per-user ETag" + ); +} + +#[tokio::test] +async fn undo_delete_bumps_watermark() { + let config = test_config_with_users(vec![make_user("dr_a", "s")]); + let data_path = config.data_path.clone(); + let case_id = "660e8400-e29b-41d4-a716-446655440000"; + seed_case(&data_path, "dr_a", case_id); + + let app = doctate_server::create_router(config); + + let login = app.clone().oneshot(login_request("dr_a", "s")).await.unwrap(); + let cookie = extract_session_cookie(&login).expect("login should set cookie"); + + // Delete first so undo has something to restore; grab the post-delete ETag. + app.clone() + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/web/cases/{case_id}/delete")) + .header(header::COOKIE, &cookie) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let (_, etag_after_delete) = capture_etag(&app, "key-dr_a", None).await; + let etag_after_delete = etag_after_delete.expect("ETag must be set"); + + // Undo. + let undo = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/web/cases/undo-delete") + .header(header::COOKIE, &cookie) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(undo.status().as_u16() / 100, 3, "undo should redirect"); + + // Undo must bump again → post-delete ETag becomes stale. + let (s, etag_after_undo) = capture_etag(&app, "key-dr_a", Some(&etag_after_delete)).await; + assert_eq!( + s, + StatusCode::OK, + "undo must change the ETag so clients re-fetch" + ); + let etag_after_undo = etag_after_undo.expect("200 response must set ETag"); + assert_ne!(etag_after_undo, etag_after_delete); +} + +#[tokio::test] +async fn bulk_delete_bumps_watermark() { + let config = test_config_with_users(vec![make_user("dr_a", "s")]); + let data_path = config.data_path.clone(); + let case_a = "770e8400-e29b-41d4-a716-446655440000"; + let case_b = "880e8400-e29b-41d4-a716-446655440000"; + seed_case(&data_path, "dr_a", case_a); + seed_case(&data_path, "dr_a", case_b); + + let app = doctate_server::create_router(config); + + let login = app.clone().oneshot(login_request("dr_a", "s")).await.unwrap(); + let cookie = extract_session_cookie(&login).expect("login should set cookie"); + + let (_, etag_before) = capture_etag(&app, "key-dr_a", None).await; + let etag_before = etag_before.expect("ETag must be set"); + + // Form-encoded bulk delete: action=delete&case_id=a&case_id=b. + let form = format!("action=delete&case_id={case_a}&case_id={case_b}"); + let bulk = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/web/cases/bulk") + .header(header::COOKIE, &cookie) + .header( + header::CONTENT_TYPE, + "application/x-www-form-urlencoded", + ) + .body(Body::from(form)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(bulk.status().as_u16() / 100, 3, "bulk should redirect"); + + let (s, etag_after) = capture_etag(&app, "key-dr_a", Some(&etag_before)).await; + assert_eq!(s, StatusCode::OK, "bulk-delete must bump the ETag"); + let etag_after = etag_after.expect("200 response must set ETag"); + assert_ne!(etag_after, etag_before); +} diff --git a/server/tests/oneliners_api_test.rs b/server/tests/oneliners_api_test.rs index 2368de2..aa73995 100644 --- a/server/tests/oneliners_api_test.rs +++ b/server/tests/oneliners_api_test.rs @@ -128,7 +128,8 @@ async fn empty_user_dir_returns_200_empty_list() { assert_eq!(response.status(), StatusCode::OK); let etag = header_str(&response, "etag"); - assert!(etag.starts_with("W/\"0-16\""), "etag was {etag:?}"); + assert!(etag.starts_with("W/\""), "etag was {etag:?}"); + assert!(etag.ends_with("-16\""), "etag was {etag:?}"); let body = body_json(response).await; assert_eq!(body["window_hours"], 16); diff --git a/server/tests/transcribe_test.rs b/server/tests/transcribe_test.rs index 3e7d37d..03273a5 100644 --- a/server/tests/transcribe_test.rs +++ b/server/tests/transcribe_test.rs @@ -288,10 +288,7 @@ async fn worker_renames_audio_to_failed_on_whisper_error() { let client = reqwest::Client::new(); let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let vocab = std::sync::Arc::new(doctate_server::gazetteer::Gazetteer::empty()); - let watermark: doctate_server::OnelinerWatermark = std::sync::Arc::new( - tokio::sync::RwLock::new(std::collections::HashMap::new()), - ); - transcribe::worker::run(rx, config, client, busy, vocab, watermark).await; + transcribe::worker::run(rx, config, client, busy, vocab).await; // Audio must have been renamed so recovery skips it next time. assert!(!audio.exists(), "original .m4a still present"); @@ -340,10 +337,7 @@ async fn transcribe_worker_normalizes_whisper_output() { let client = reqwest::Client::new(); let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); - let watermark: doctate_server::OnelinerWatermark = std::sync::Arc::new( - tokio::sync::RwLock::new(std::collections::HashMap::new()), - ); - transcribe::worker::run(rx, config, client, busy, vocab, watermark).await; + transcribe::worker::run(rx, config, client, busy, vocab).await; let transcript_path = case_dir.join("2026-04-13T10-30-00Z.transcript.txt"); assert!(transcript_path.exists(), "transcript missing"); diff --git a/server/tests/upload_test.rs b/server/tests/upload_test.rs index 34d047b..a1d42e1 100644 --- a/server/tests/upload_test.rs +++ b/server/tests/upload_test.rs @@ -284,7 +284,9 @@ async fn upload_reopens_soft_deleted_case() { // 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. + // to drive the /web/cases/{id}/delete handler here. Direct write + // also means the watermark is NOT bumped here; that way the next + // GET /api/oneliners pins down the pre-reopen ETag cleanly. let case_dir = data_path.join("dr_test").join(case_id); let marker = DeleteMarker { batch: uuid::Uuid::new_v4(), @@ -293,9 +295,32 @@ async fn upload_reopens_soft_deleted_case() { write_delete_marker(&case_dir, &marker).await.unwrap(); assert!(case_dir.join(DELETE_MARKER).exists()); + // Capture the ETag before the reopen. + let etag_before = { + let resp = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/oneliners") + .header("X-API-Key", "test-key-123") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + resp.headers() + .get("etag") + .and_then(|v| v.to_str().ok()) + .map(str::to_owned) + .expect("server must set ETag") + }; + // Second upload — must auto-reopen. let (boundary, body) = multipart_body(case_id, "2026-04-13T12:00:00Z", b"reopen recording"); let response = app + .clone() .oneshot( Request::builder() .method("POST") @@ -321,6 +346,34 @@ async fn upload_reopens_soft_deleted_case() { assert!(case_dir.join("2026-04-13T10-30-00Z.m4a").exists()); assert!(case_dir.join("2026-04-13T12-00-00Z.m4a").exists()); + // Auto-reopen must bump the watermark: sending the pre-reopen + // If-None-Match must return 200 (not 304) so clients who were + // offline during the reopen notice it on their next poll. + let resp = app + .oneshot( + Request::builder() + .method("GET") + .uri("/api/oneliners") + .header("X-API-Key", "test-key-123") + .header("If-None-Match", &etag_before) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::OK, + "auto-reopen must invalidate the pre-reopen ETag" + ); + let etag_after = resp + .headers() + .get("etag") + .and_then(|v| v.to_str().ok()) + .map(str::to_owned) + .expect("server must set ETag"); + assert_ne!(etag_after, etag_before); + let _ = std::fs::remove_dir_all(&data_path); }