diff --git a/client-desktop/src/app.rs b/client-desktop/src/app.rs index 3ca0da6..bf58d99 100644 --- a/client-desktop/src/app.rs +++ b/client-desktop/src/app.rs @@ -1,12 +1,21 @@ -//! The egui application: config panel, record/stop controls, case list, -//! state transitions driven by recorder + uploader + poller channels. +//! The egui application: config panel, record/stop controls, case list. +//! +//! State transitions are driven by: +//! - user clicks (new/continue/stop/open/save-config/dismiss-error) +//! - recorder events (ffmpeg finalizing → RecorderEvent::Finished) +//! - upload events (terminal failure → AppState::Error) +//! +//! Connectivity and upload-progress indicators live in the footer; they +//! never block the main state machine. See +//! `berpr-fe-den-plan-lively-journal.md` for the design rationale. use std::path::PathBuf; use std::sync::Arc; use std::time::{Duration, Instant}; use doctate_client_core::{ - cleanup_orphan_audio, CaseMarker, CaseStore, OnelinerPoller, PollerConfig, SnapshotCache, + cleanup_orphan_audio, write_sidecar, CaseMarker, CaseStore, PendingUpload, ServerSync, + SnapshotCache, SyncConfig, UploadEvent, WorkerSnapshot, WorkerState, }; use doctate_common::timestamp::{now_rfc3339, recorded_at_to_filename_stem}; use eframe::egui; @@ -17,8 +26,7 @@ use uuid::Uuid; use crate::config::Config; use crate::recorder::{Recorder, RecorderEvent}; -use crate::state::AppState; -use crate::uploader::{write_sidecar, PendingUpload, UploadEvent, Uploader}; +use crate::state::{pick_footer_status, AppState, FooterStatus}; pub struct DoctateApp { runtime: Arc, @@ -37,10 +45,15 @@ pub struct DoctateApp { recorder: Option, recorder_events: Option>, - uploader: Option, + server_sync: Option, + worker_rx: Option>, upload_events: Option>, - poller: Option, + /// Context for an in-flight ffmpeg flush: we left `AppState::Recording` + /// for `AppState::Idle` on Stop (so the UI stays responsive), but we + /// still need `case_id` + `recorded_at` when the recorder emits + /// `Finished`. `None` when no flush is in progress. + finalizing: Option<(Uuid, String)>, } enum UiAction { @@ -77,8 +90,6 @@ impl DoctateApp { .map(|c| c.api_key.clone()) .unwrap_or_default(); - // Open the case store before any background task — the store - // owns the watch::Sender that drives the UI list. let (store, snapshot_rx) = runtime .block_on(async { CaseStore::open(cases_dir).await }) .expect("open case store"); @@ -86,25 +97,19 @@ impl DoctateApp { let snapshot_cache = Arc::new(SnapshotCache::new(snapshot_cache_path)); let http_client = Arc::new(reqwest::Client::new()); - // Startup cleanup — runs once, before the poller and uploader - // spawn. Keeps the client a lean ephemeral microphone: - // - markers older than 72 h (and server-confirmed) drop out, - // - orphan audio older than 24 h is reaped (patient-data - // minimization; a file that never found its sidecar will - // never be uploaded). + // Startup cleanup — one-shot, before the worker spawns. { let store = case_store.clone(); let pending = pending_dir.clone(); runtime.block_on(async move { - let marker_cutoff = - time::OffsetDateTime::now_utc() - time::Duration::hours(72); + let marker_cutoff = time::OffsetDateTime::now_utc() - time::Duration::hours(72); match store.cleanup_stale(marker_cutoff).await { Ok(n) if n > 0 => info!(count = n, "startup cleanup: stale markers removed"), Ok(_) => {} Err(e) => warn!(error = %e, "startup cleanup: marker sweep failed"), } - let audio_cutoff = std::time::SystemTime::now() - - std::time::Duration::from_secs(24 * 3600); + let audio_cutoff = + std::time::SystemTime::now() - std::time::Duration::from_secs(24 * 3600); let n = cleanup_orphan_audio(&pending, audio_cutoff).await; if n > 0 { info!(count = n, "startup cleanup: orphan audio removed"); @@ -112,8 +117,8 @@ impl DoctateApp { }); } - let (state, uploader, upload_events, poller) = if let Some(cfg) = &config { - let (up, rx, pol) = spawn_background( + let (state, server_sync, worker_rx, upload_events) = if let Some(cfg) = &config { + let (sync, events, rx) = spawn_worker( &runtime, cfg.clone(), pending_dir.clone(), @@ -121,7 +126,7 @@ impl DoctateApp { case_store.clone(), snapshot_cache.clone(), ); - (AppState::Idle, Some(up), Some(rx), Some(pol)) + (AppState::Idle, Some(sync), Some(rx), Some(events)) } else { (AppState::NotConfigured, None, None, None) }; @@ -139,9 +144,10 @@ impl DoctateApp { snapshot_cache, recorder: None, recorder_events: None, - uploader, + server_sync, + worker_rx, upload_events, - poller, + finalizing: None, } } @@ -167,8 +173,6 @@ impl DoctateApp { } info!("config saved"); - // New API key = potentially different user. Wipe local state so - // we don't leak old user's cases into the new user's view. let store = self.case_store.clone(); let cache = self.snapshot_cache.clone(); self.runtime.block_on(async move { @@ -180,13 +184,12 @@ impl DoctateApp { } }); - // Drop old background tasks explicitly so their AbortHandle / - // channel-close fires before the new ones spawn. - self.poller = None; - self.uploader = None; + // Drop old worker explicitly so its abort fires before new spawn. + self.server_sync = None; + self.worker_rx = None; self.upload_events = None; - let (up, rx, pol) = spawn_background( + let (sync, events, rx) = spawn_worker( &self.runtime, cfg.clone(), self.pending_dir.clone(), @@ -194,11 +197,12 @@ impl DoctateApp { self.case_store.clone(), self.snapshot_cache.clone(), ); - self.uploader = Some(up); - self.upload_events = Some(rx); - self.poller = Some(pol); + self.server_sync = Some(sync); + self.worker_rx = Some(rx); + self.upload_events = Some(events); self.config = Some(cfg); self.state = AppState::Idle; + self.finalizing = None; } fn on_start_new(&mut self) { @@ -210,7 +214,9 @@ impl DoctateApp { } fn start_recording_for(&mut self, case_id: Uuid, is_continue: bool) { - if !matches!(self.state, AppState::Idle) { + // Two-part guard: user-visible state AND no lingering recorder + // from a previous stop that is still flushing. + if !matches!(self.state, AppState::Idle) || self.recorder.is_some() { return; } if let Err(e) = std::fs::create_dir_all(&self.pending_dir) { @@ -225,8 +231,7 @@ impl DoctateApp { let output_path = self.pending_dir.join(format!("{stem}.m4a")); // Persist the case marker before recording so the UI list - // reflects the new case immediately (even if the recorder fails - // to start, the marker stays — harmless and reproducible). + // reflects the new case immediately. let store = self.case_store.clone(); let now = time::OffsetDateTime::now_utc(); self.runtime.block_on(async move { @@ -273,10 +278,10 @@ impl DoctateApp { if let Some(rec) = self.recorder.take() { rec.stop(); } - self.state = AppState::FinalizingRecording { - case_id, - recorded_at, - }; + // UI is ready for the next recording immediately — the finalizing + // ffmpeg flush shows up as a footer spinner, not a blocking state. + self.finalizing = Some((case_id, recorded_at)); + self.state = AppState::Idle; } fn on_open_web(&self, case_id: Uuid) { @@ -309,6 +314,7 @@ impl DoctateApp { } RecorderEvent::Failed { error: err } => { error!(error = %err, "recorder failed"); + self.finalizing = None; self.state = AppState::Error { message: format!("Aufnahme: {err}"), }; @@ -320,25 +326,27 @@ impl DoctateApp { } fn handle_recording_finished(&mut self, output: PathBuf) { - let (case_id, recorded_at) = match &self.state { - AppState::FinalizingRecording { - case_id, - recorded_at, - } => (*case_id, recorded_at.clone()), - AppState::Recording { - case_id, - recorded_at, - .. - } => (*case_id, recorded_at.clone()), - other => { - warn!(state = ?other, "recorder finished in unexpected state"); - self.recorder_events = None; - self.recorder = None; - return; - } + // Prefer the explicit finalizing context (Stop-flow). Fall back + // to AppState::Recording (recorder terminated before Stop — rare + // but possible). + let (case_id, recorded_at) = if let Some((id, rec)) = self.finalizing.take() { + (id, rec) + } else if let AppState::Recording { + case_id, + recorded_at, + .. + } = &self.state + { + let pair = (*case_id, recorded_at.clone()); + self.state = AppState::Idle; + pair + } else { + warn!(state = ?self.state, "recorder finished in unexpected state"); + self.recorder_events = None; + self.recorder = None; + return; }; - // Bump activity on the marker so the case rises in the list. let store = self.case_store.clone(); let now = time::OffsetDateTime::now_utc(); self.runtime.block_on(async move { @@ -357,24 +365,19 @@ impl DoctateApp { self.state = AppState::Error { message: format!("Sidecar schreiben: {e}"), }; + self.recorder_events = None; + self.recorder = None; return; } - let Some(uploader) = &self.uploader else { - self.state = AppState::Error { - message: "Upload-Worker nicht verfügbar".into(), - }; - return; - }; - - if let Err(e) = uploader.submit(upload) { - self.state = AppState::Error { - message: format!("Upload-Submit: {e}"), - }; - return; + // Kick the worker so it picks up the fresh file without waiting + // for the next poll-tick. + if let Some(sync) = &self.server_sync { + sync.notify(); + } else { + warn!("no server_sync active — upload stays on disk until next launch"); } - self.state = AppState::Uploading { case_id }; self.recorder_events = None; self.recorder = None; } @@ -394,11 +397,6 @@ impl DoctateApp { } UploadEvent::Succeeded { case_id, status } => { info!(case_id = %case_id, status = ?status, "upload succeeded"); - if let AppState::Uploading { case_id: current } = &self.state - && *current == case_id - { - self.state = AppState::Idle; - } } UploadEvent::Failed { case_id, @@ -406,10 +404,8 @@ impl DoctateApp { will_retry, } => { warn!(case_id = %case_id, reason = %reason, will_retry, "upload failed"); - if !will_retry - && let AppState::Uploading { case_id: current } = &self.state - && *current == case_id - { + if !will_retry { + // Terminal = operator action needed. Block UI. self.state = AppState::Error { message: reason }; } } @@ -456,16 +452,6 @@ impl DoctateApp { } None } - AppState::FinalizingRecording { .. } => { - ui.spinner(); - ui.label("Speichere Aufnahme…"); - None - } - AppState::Uploading { .. } => { - ui.spinner(); - ui.label("Upload läuft…"); - None - } AppState::Error { message } => { ui.colored_label(egui::Color32::RED, message); ui.add_space(8.0); @@ -498,7 +484,9 @@ impl DoctateApp { } let mut action = None; - let idle = matches!(self.state, AppState::Idle); + // Ready-for-new means the user-intent state is Idle AND no ffmpeg + // instance is still flushing. Upload-queue depth does NOT gate. + let idle = matches!(self.state, AppState::Idle) && self.recorder.is_none(); egui::ScrollArea::vertical().show(ui, |ui| { for marker in visible { @@ -530,30 +518,60 @@ impl DoctateApp { action } - fn render_staleness(&self, ui: &mut egui::Ui) { - let Some(poller) = &self.poller else { + fn render_footer_status(&self, ui: &mut egui::Ui) { + let Some(worker_rx) = &self.worker_rx else { + ui.colored_label(egui::Color32::GRAY, "—"); return; }; - let Some(last) = poller.last_success() else { - ui.colored_label(egui::Color32::GRAY, "⚪ noch keine Verbindung"); - return; - }; - let age = last.elapsed(); - let secs = age.as_secs(); - let (color, text) = if secs < 30 { - (egui::Color32::DARK_GREEN, format!("🟢 zuletzt: vor {secs} s")) - } else if secs < 120 { - ( - egui::Color32::from_rgb(200, 140, 0), - format!("🟡 zuletzt: vor {secs} s"), - ) - } else { - ( - egui::Color32::DARK_RED, - format!("🔴 zuletzt: vor {} s", secs), - ) - }; - ui.colored_label(color, text); + let snap = worker_rx.borrow().clone(); + let last_age = snap.last_success.map(|i| i.elapsed()); + let threshold = self + .config + .as_ref() + .map(|c| std::cmp::max(3 * c.poll_interval(), Duration::from_secs(30))) + .unwrap_or(Duration::from_secs(30)); + + let status = pick_footer_status( + self.finalizing.is_some(), + snap.state, + snap.queue_len, + last_age, + threshold, + ); + + match status { + FooterStatus::Finalizing => { + ui.horizontal(|ui| { + ui.spinner(); + ui.label("Speichere Aufnahme…"); + }); + } + FooterStatus::Uploading { count } => { + ui.horizontal(|ui| { + ui.spinner(); + if count <= 1 { + ui.label("Uploading…"); + } else { + ui.label(format!("Uploading ({count} offen)…")); + } + }); + } + FooterStatus::Synchronizing => { + ui.horizontal(|ui| { + ui.spinner(); + ui.label("Synchronisiere…"); + }); + } + FooterStatus::NeverConnected => { + ui.colored_label(egui::Color32::GRAY, "⚪ noch keine Verbindung"); + } + FooterStatus::Offline => { + ui.colored_label(egui::Color32::DARK_RED, "🔴 Offline"); + } + FooterStatus::Online => { + ui.colored_label(egui::Color32::DARK_GREEN, "🟢 Online"); + } + } } fn handle_action(&mut self, action: UiAction) { @@ -566,32 +584,41 @@ impl DoctateApp { UiAction::DismissError => self.state = AppState::Idle, } } + + fn worker_is_active(&self) -> bool { + let Some(rx) = &self.worker_rx else { + return false; + }; + let snap = rx.borrow(); + snap.state != WorkerState::Idle || snap.queue_len > 0 + } } -fn spawn_background( +fn spawn_worker( runtime: &Runtime, config: Config, pending_dir: PathBuf, http: Arc, case_store: Arc, snapshot_cache: Arc, -) -> (Uploader, mpsc::UnboundedReceiver, OnelinerPoller) { +) -> ( + ServerSync, + mpsc::UnboundedReceiver, + watch::Receiver, +) { let _guard = runtime.enter(); let poll_interval = config.poll_interval(); let window_hours = config.window_hours(); - let (uploader, rx) = Uploader::spawn(config.clone(), pending_dir); - let poller = OnelinerPoller::spawn( - http, - PollerConfig { - server_url: config.server_url, - api_key: config.api_key, - poll_interval, - window_hours, - }, - case_store, - snapshot_cache, - ); - (uploader, rx, poller) + let sync_config = SyncConfig { + server_url: config.server_url, + api_key: config.api_key, + poll_interval, + window_hours, + }; + let (sync, events_rx) = + ServerSync::spawn(http, sync_config, case_store, snapshot_cache, pending_dir); + let snapshot_rx = sync.snapshot(); + (sync, events_rx, snapshot_rx) } fn parse_rfc3339(s: &str) -> Option { @@ -605,10 +632,7 @@ fn extract_hhmm(ts: &str) -> String { match parse_rfc3339(ts) { Some(t) => { let local = t - .to_offset( - time::UtcOffset::current_local_offset() - .unwrap_or(time::UtcOffset::UTC), - ); + .to_offset(time::UtcOffset::current_local_offset().unwrap_or(time::UtcOffset::UTC)); format!("{:02}:{:02}", local.hour(), local.minute()) } None => ts.chars().take(16).collect(), @@ -634,7 +658,7 @@ impl eframe::App for DoctateApp { egui::TopBottomPanel::bottom("footer").show(ctx, |ui| { ui.add_space(4.0); ui.vertical_centered(|ui| { - self.render_staleness(ui); + self.render_footer_status(ui); }); ui.add_space(4.0); }); @@ -647,7 +671,6 @@ impl eframe::App for DoctateApp { } ui.add_space(8.0); }); - // Case list fills the remaining space. if !matches!(self.state, AppState::NotConfigured) && let Some(a) = self.render_case_list(ui) { @@ -659,14 +682,14 @@ impl eframe::App for DoctateApp { self.handle_action(action); } - // Repaint periodically so the staleness indicator + watch - // snapshots reach the UI without a user event. - match self.state { - AppState::Recording { .. } => ctx.request_repaint_after(Duration::from_millis(250)), - AppState::FinalizingRecording { .. } | AppState::Uploading { .. } => { - ctx.request_repaint_after(Duration::from_millis(100)); - } - _ => ctx.request_repaint_after(Duration::from_secs(1)), - } + // Repaint cadence: fast while visibly active (spinner, REC-timer), + // slow (1 Hz) in the quiet steady state. + let active_spinner = self.finalizing.is_some() || self.worker_is_active(); + let interval = match self.state { + AppState::Recording { .. } => Duration::from_millis(250), + _ if active_spinner => Duration::from_millis(100), + _ => Duration::from_secs(1), + }; + ctx.request_repaint_after(interval); } } diff --git a/client-desktop/src/main.rs b/client-desktop/src/main.rs index 9c6a812..6c2408e 100644 --- a/client-desktop/src/main.rs +++ b/client-desktop/src/main.rs @@ -3,7 +3,6 @@ mod config; mod paths; mod recorder; mod state; -mod uploader; use std::sync::Arc; diff --git a/client-desktop/src/state.rs b/client-desktop/src/state.rs index bcea817..203cb13 100644 --- a/client-desktop/src/state.rs +++ b/client-desktop/src/state.rs @@ -1,8 +1,14 @@ -//! UI-visible application state. Transitions are driven by user actions -//! (buttons) and async events from the recorder + uploader channels. +//! UI-visible application state and the footer status derivation. +//! +//! The state machine here only reflects **user intent** — "ready to +//! record", "recording in progress", "config error, please fix". +//! Background activity (upload queue, poll cycle, finalize flush) lives +//! in a separate observable layer (`FooterStatus`) derived via a pure +//! function so it can be table-tested without egui. -use std::time::Instant; +use std::time::Duration; +use doctate_client_core::server_sync::WorkerState; use uuid::Uuid; #[derive(Debug)] @@ -15,16 +21,7 @@ pub enum AppState { Recording { case_id: Uuid, recorded_at: String, - started_at: Instant, - }, - /// User pressed Stop; waiting for ffmpeg to finalize the MOOV atom. - FinalizingRecording { - case_id: Uuid, - recorded_at: String, - }, - /// File handed to uploader; waiting for server ACK. - Uploading { - case_id: Uuid, + started_at: std::time::Instant, }, /// Terminal error; user must dismiss or fix config. Error { @@ -38,9 +35,155 @@ impl AppState { Self::NotConfigured => "Konfiguration fehlt", Self::Idle => "Bereit", Self::Recording { .. } => "Aufnahme läuft", - Self::FinalizingRecording { .. } => "Speichere…", - Self::Uploading { .. } => "Upload läuft", Self::Error { .. } => "Fehler", } } } + +/// Derived observable rendered in the footer. Priority order encoded +/// in `pick_footer_status` — active work beats connectivity indicators. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum FooterStatus { + /// ffmpeg is flushing the MOOV atom. Spinner. + Finalizing, + /// Worker is uploading. Spinner. `count` = pending-dir depth. + Uploading { count: usize }, + /// Worker is polling `/api/oneliners`. Spinner. + Synchronizing, + /// Never got a successful request yet — cold start or bad config. + NeverConnected, + /// Last success older than threshold. Red text. + Offline, + /// Last success fresh. Green text. The default quiet state. + Online, +} + +/// Decide which footer status to show. Pure function — no side effects, +/// fully table-testable. Inputs chosen to match what `DoctateApp` can +/// observe cheaply per egui frame. +pub fn pick_footer_status( + finalizing: bool, + worker_state: WorkerState, + queue_len: usize, + last_success_age: Option, + offline_threshold: Duration, +) -> FooterStatus { + if finalizing { + return FooterStatus::Finalizing; + } + if queue_len > 0 || worker_state == WorkerState::Uploading { + return FooterStatus::Uploading { + count: queue_len.max(1), + }; + } + if worker_state == WorkerState::Polling { + return FooterStatus::Synchronizing; + } + match last_success_age { + None => FooterStatus::NeverConnected, + Some(age) if age > offline_threshold => FooterStatus::Offline, + Some(_) => FooterStatus::Online, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const THRESHOLD: Duration = Duration::from_secs(30); + + #[test] + fn finalizing_wins_over_everything() { + let s = pick_footer_status( + true, + WorkerState::Uploading, + 5, + Some(Duration::from_millis(1)), + THRESHOLD, + ); + assert_eq!(s, FooterStatus::Finalizing); + } + + #[test] + fn uploading_when_queue_non_empty() { + let s = pick_footer_status( + false, + WorkerState::Uploading, + 3, + Some(Duration::from_millis(1)), + THRESHOLD, + ); + assert_eq!(s, FooterStatus::Uploading { count: 3 }); + } + + #[test] + fn uploading_reports_at_least_one_even_if_scan_raced() { + // Edge case: WorkerState says Uploading, but scan returned 0 + // (just finished a file and the next scan hasn't happened yet). + // We still want to show "Uploading" to avoid a flicker back to + // Online right before the next file starts. + let s = pick_footer_status( + false, + WorkerState::Uploading, + 0, + Some(Duration::from_millis(1)), + THRESHOLD, + ); + assert_eq!(s, FooterStatus::Uploading { count: 1 }); + } + + #[test] + fn synchronizing_when_polling_and_no_queue() { + let s = pick_footer_status( + false, + WorkerState::Polling, + 0, + Some(Duration::from_millis(1)), + THRESHOLD, + ); + assert_eq!(s, FooterStatus::Synchronizing); + } + + #[test] + fn never_connected_without_last_success() { + let s = pick_footer_status(false, WorkerState::Idle, 0, None, THRESHOLD); + assert_eq!(s, FooterStatus::NeverConnected); + } + + #[test] + fn offline_when_last_success_is_stale() { + let s = pick_footer_status( + false, + WorkerState::Idle, + 0, + Some(Duration::from_secs(120)), + THRESHOLD, + ); + assert_eq!(s, FooterStatus::Offline); + } + + #[test] + fn online_when_last_success_is_fresh() { + let s = pick_footer_status( + false, + WorkerState::Idle, + 0, + Some(Duration::from_secs(5)), + THRESHOLD, + ); + assert_eq!(s, FooterStatus::Online); + } + + #[test] + fn offline_threshold_is_inclusive_of_equal_age_as_online() { + // Exactly at threshold: still online (strict ">" in logic). + let s = pick_footer_status( + false, + WorkerState::Idle, + 0, + Some(THRESHOLD), + THRESHOLD, + ); + assert_eq!(s, FooterStatus::Online); + } +} diff --git a/client-desktop/src/uploader.rs b/client-desktop/src/uploader.rs deleted file mode 100644 index 941080a..0000000 --- a/client-desktop/src/uploader.rs +++ /dev/null @@ -1,466 +0,0 @@ -//! Async uploader with persistent on-disk queue and exponential-backoff -//! retry. Recordings live in the pending directory as `{stem}.m4a` plus -//! a sidecar `{stem}.meta.json`. The worker replays anything it finds -//! there at startup, then processes new submissions as they arrive. - -use std::path::{Path, PathBuf}; -use std::time::Duration; - -use doctate_common::ack::{AckResponse, AckStatus}; -use doctate_common::constants::{ - API_KEY_HEADER, CONTENT_TYPE_AUDIO_MP4, FIELD_AUDIO, FIELD_CASE_ID, FIELD_RECORDED_AT, - UPLOAD_PATH, -}; -use serde::{Deserialize, Serialize}; -use thiserror::Error; -use tokio::sync::mpsc; -use tracing::{info, warn}; -use uuid::Uuid; - -use crate::config::Config; - -const INITIAL_BACKOFF: Duration = Duration::from_secs(2); -const MAX_BACKOFF: Duration = Duration::from_secs(60); - -/// A recording waiting to be uploaded. The `file` path is derived from -/// the sidecar JSON's location, never from the JSON content itself — -/// that way the sidecar stays valid even if the pending directory is -/// renamed. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct PendingUpload { - pub case_id: Uuid, - pub recorded_at: String, - #[serde(skip, default)] - pub file: PathBuf, -} - -/// Lifecycle events emitted to the UI for a single upload attempt chain. -#[derive(Debug)] -pub enum UploadEvent { - Started(Uuid), - Succeeded { - case_id: Uuid, - status: AckStatus, - }, - Failed { - case_id: Uuid, - reason: String, - will_retry: bool, - }, -} - -#[derive(Debug, Error)] -pub enum UploaderError { - #[error("I/O error: {0}")] - Io(#[from] std::io::Error), - #[error("serialize sidecar: {0}")] - Serialize(#[from] serde_json::Error), -} - -/// Handle to the uploader's worker task. Drop does NOT shut down the -/// worker; it stays alive as long as the `submit_tx` is kept. -pub struct Uploader { - submit_tx: mpsc::UnboundedSender, -} - -impl Uploader { - /// Spawn the worker task. Must be called inside a Tokio runtime - /// context (`Runtime::enter()` guard or `#[tokio::main]`). - /// - /// Re-scans `pending_dir` at startup so leftover recordings from - /// a prior run are replayed automatically. - pub fn spawn( - config: Config, - pending_dir: PathBuf, - ) -> (Self, mpsc::UnboundedReceiver) { - let (submit_tx, submit_rx) = mpsc::unbounded_channel(); - let (events_tx, events_rx) = mpsc::unbounded_channel(); - tokio::spawn(run_worker(config, pending_dir, submit_rx, events_tx)); - (Self { submit_tx }, events_rx) - } - - /// Submit a pending upload for processing. Non-blocking. Fails only - /// if the worker has been dropped (should not happen in normal - /// operation). - pub fn submit( - &self, - upload: PendingUpload, - ) -> Result<(), mpsc::error::SendError> { - self.submit_tx.send(upload) - } -} - -/// Path of the sidecar JSON for a given m4a file — same directory, -/// `{stem}.meta.json`. -pub fn sidecar_path_for(m4a: &Path) -> PathBuf { - let stem = m4a - .file_stem() - .map(|s| s.to_string_lossy().into_owned()) - .unwrap_or_default(); - let dir = m4a.parent().unwrap_or_else(|| Path::new(".")); - dir.join(format!("{stem}.meta.json")) -} - -/// Persist a pending upload's metadata next to its m4a file. Atomic: -/// writes a `*.tmp` sibling and renames it into place, so a crash -/// mid-write cannot leave a zero-byte or partial sidecar on disk. -/// Without this guarantee, the `scan_pending_dir` recovery path would -/// silently drop the upload as an unreadable orphan on next launch. -pub fn write_sidecar(upload: &PendingUpload) -> Result<(), UploaderError> { - let sidecar = sidecar_path_for(&upload.file); - if let Some(parent) = sidecar.parent() { - std::fs::create_dir_all(parent)?; - } - let json = serde_json::to_string_pretty(upload)?; - let tmp = sidecar.with_extension("json.tmp"); - std::fs::write(&tmp, json)?; - std::fs::rename(&tmp, &sidecar)?; - Ok(()) -} - -/// Scan `pending_dir` for m4a files with matching sidecars. Orphans -/// (m4a without sidecar, or unreadable sidecar) are logged and skipped -/// rather than propagated as errors — one broken file should not take -/// down recovery of the rest. -pub fn scan_pending_dir(pending_dir: &Path) -> Vec { - let entries = match std::fs::read_dir(pending_dir) { - Ok(e) => e, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Vec::new(), - Err(e) => { - warn!(dir = ?pending_dir, error = %e, "scan pending dir failed"); - return Vec::new(); - } - }; - let mut out = Vec::new(); - for entry in entries.flatten() { - let path = entry.path(); - if path.extension().and_then(|s| s.to_str()) != Some("m4a") { - continue; - } - match read_pending(&path) { - Ok(upload) => out.push(upload), - Err(e) => warn!(path = ?path, error = %e, "skipping pending entry"), - } - } - out -} - -fn read_pending(m4a: &Path) -> Result { - let sidecar = sidecar_path_for(m4a); - let json = std::fs::read_to_string(&sidecar)?; - let mut upload: PendingUpload = serde_json::from_str(&json)?; - upload.file = m4a.to_path_buf(); - Ok(upload) -} - -async fn run_worker( - config: Config, - pending_dir: PathBuf, - mut submit_rx: mpsc::UnboundedReceiver, - events_tx: mpsc::UnboundedSender, -) { - let client = reqwest::Client::new(); - - let startup = scan_pending_dir(&pending_dir); - if !startup.is_empty() { - info!(count = startup.len(), "replaying pending uploads"); - } - for pending in startup { - upload_with_retry(&client, &config, pending, &events_tx).await; - } - - while let Some(pending) = submit_rx.recv().await { - upload_with_retry(&client, &config, pending, &events_tx).await; - } -} - -async fn upload_with_retry( - client: &reqwest::Client, - config: &Config, - upload: PendingUpload, - events_tx: &mpsc::UnboundedSender, -) { - let mut backoff = INITIAL_BACKOFF; - loop { - let _ = events_tx.send(UploadEvent::Started(upload.case_id)); - match attempt_upload(client, config, &upload).await { - Ok(ack) => { - let _ = tokio::fs::remove_file(&upload.file).await; - let _ = tokio::fs::remove_file(&sidecar_path_for(&upload.file)).await; - let _ = events_tx.send(UploadEvent::Succeeded { - case_id: upload.case_id, - status: ack.status, - }); - return; - } - Err(ErrorKind::Terminal(reason)) => { - // Terminal = configuration or operator error (wrong API - // key, payload too big, etc.). No amount of retries will - // help, and keeping sensitive audio on the client "just - // in case" would violate the data-minimization rule - // (clients are ephemeral microphones, not archives). - // Delete both artefacts and surface the failure to the UI. - let _ = tokio::fs::remove_file(&upload.file).await; - let _ = tokio::fs::remove_file(&sidecar_path_for(&upload.file)).await; - let _ = events_tx.send(UploadEvent::Failed { - case_id: upload.case_id, - reason, - will_retry: false, - }); - return; - } - Err(ErrorKind::Transient(reason)) => { - let _ = events_tx.send(UploadEvent::Failed { - case_id: upload.case_id, - reason, - will_retry: true, - }); - tokio::time::sleep(backoff).await; - backoff = (backoff * 2).min(MAX_BACKOFF); - } - } - } -} - -enum ErrorKind { - Transient(String), - Terminal(String), -} - -async fn attempt_upload( - client: &reqwest::Client, - config: &Config, - upload: &PendingUpload, -) -> Result { - let audio = tokio::fs::read(&upload.file) - .await - .map_err(|e| ErrorKind::Terminal(format!("read audio file: {e}")))?; - - let file_name = upload - .file - .file_name() - .map(|s| s.to_string_lossy().into_owned()) - .unwrap_or_else(|| "recording.m4a".to_string()); - - let audio_part = reqwest::multipart::Part::bytes(audio) - .file_name(file_name) - .mime_str(CONTENT_TYPE_AUDIO_MP4) - .expect("static MIME string is valid"); - - let form = reqwest::multipart::Form::new() - .text(FIELD_CASE_ID, upload.case_id.to_string()) - .text(FIELD_RECORDED_AT, upload.recorded_at.clone()) - .part(FIELD_AUDIO, audio_part); - - let url = format!("{}{}", config.server_url.trim_end_matches('/'), UPLOAD_PATH); - - let resp = client - .post(&url) - .header(API_KEY_HEADER, &config.api_key) - .multipart(form) - .send() - .await - .map_err(|e| ErrorKind::Transient(format!("network: {e}")))?; - - let status = resp.status(); - if status.is_success() { - resp.json::() - .await - .map_err(|e| ErrorKind::Terminal(format!("parse ack: {e}"))) - } else if status.as_u16() == 408 || status.as_u16() == 429 || status.is_server_error() { - let body = resp.text().await.unwrap_or_default(); - Err(ErrorKind::Transient(format!("HTTP {status}: {body}"))) - } else { - let body = resp.text().await.unwrap_or_default(); - Err(ErrorKind::Terminal(format!("HTTP {status}: {body}"))) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use tempfile::TempDir; - use wiremock::matchers::{header, method, path as wpath}; - use wiremock::{Mock, MockServer, ResponseTemplate}; - - fn sample_pending(dir: &Path) -> PendingUpload { - let file = dir.join("sample.m4a"); - std::fs::write(&file, b"dummy audio").unwrap(); - PendingUpload { - case_id: Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap(), - recorded_at: "2026-04-15T10:32:00Z".into(), - file, - } - } - - fn ack_body(status: AckStatus) -> AckResponse { - AckResponse { - case_id: "550e8400-e29b-41d4-a716-446655440000".into(), - recorded_at: "2026-04-15T10:32:00Z".into(), - status, - } - } - - #[test] - fn sidecar_roundtrip_via_scan() { - let tmp = TempDir::new().unwrap(); - let upload = sample_pending(tmp.path()); - write_sidecar(&upload).unwrap(); - let found = scan_pending_dir(tmp.path()); - assert_eq!(found.len(), 1); - assert_eq!(found[0].case_id, upload.case_id); - assert_eq!(found[0].recorded_at, upload.recorded_at); - assert_eq!(found[0].file, upload.file); - } - - #[test] - fn scan_ignores_m4a_without_sidecar() { - let tmp = TempDir::new().unwrap(); - std::fs::write(tmp.path().join("orphan.m4a"), b"x").unwrap(); - let found = scan_pending_dir(tmp.path()); - assert!(found.is_empty()); - } - - #[test] - fn scan_returns_empty_for_missing_dir() { - let tmp = TempDir::new().unwrap(); - let missing = tmp.path().join("does-not-exist"); - assert!(scan_pending_dir(&missing).is_empty()); - } - - /// Atomic sidecar write must leave no `.tmp` leftover on success — - /// the recovery scan only accepts exact `.meta.json` files, so a - /// stray temp file would be invisible but wastes inodes, and hints - /// at a rename bug if it ever appears. - #[test] - fn write_sidecar_leaves_no_tmp_artefact() { - let tmp = TempDir::new().unwrap(); - let upload = sample_pending(tmp.path()); - write_sidecar(&upload).unwrap(); - - let sidecar = sidecar_path_for(&upload.file); - assert!(sidecar.exists(), "sidecar must be at final path"); - let tmp_path = sidecar.with_extension("json.tmp"); - assert!(!tmp_path.exists(), "tmp artefact must not remain"); - } - - #[tokio::test] - async fn successful_upload_deletes_files_and_emits_events() { - let server = MockServer::start().await; - Mock::given(method("POST")) - .and(wpath(UPLOAD_PATH)) - .and(header(API_KEY_HEADER, "test-key")) - .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received))) - .mount(&server) - .await; - - let tmp = TempDir::new().unwrap(); - let upload = sample_pending(tmp.path()); - write_sidecar(&upload).unwrap(); - let config = Config { - server_url: server.uri(), - api_key: "test-key".into(), - oneliner_poll_interval_seconds: None, - oneliner_window_hours: None, - }; - - let client = reqwest::Client::new(); - let (tx, mut rx) = mpsc::unbounded_channel(); - upload_with_retry(&client, &config, upload.clone(), &tx).await; - - assert!(!upload.file.exists(), "m4a should be deleted after ack"); - assert!( - !sidecar_path_for(&upload.file).exists(), - "sidecar should be deleted after ack" - ); - - assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_)))); - let Some(UploadEvent::Succeeded { status, .. }) = rx.recv().await else { - panic!("expected Succeeded event"); - }; - assert_eq!(status, AckStatus::Received); - } - - #[tokio::test(start_paused = true)] - async fn transient_500_retries_then_succeeds() { - let server = MockServer::start().await; - Mock::given(method("POST")) - .and(wpath(UPLOAD_PATH)) - .respond_with(ResponseTemplate::new(500)) - .up_to_n_times(1) - .mount(&server) - .await; - Mock::given(method("POST")) - .and(wpath(UPLOAD_PATH)) - .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received))) - .mount(&server) - .await; - - let tmp = TempDir::new().unwrap(); - let upload = sample_pending(tmp.path()); - write_sidecar(&upload).unwrap(); - let config = Config { - server_url: server.uri(), - api_key: "test-key".into(), - oneliner_poll_interval_seconds: None, - oneliner_window_hours: None, - }; - - let client = reqwest::Client::new(); - let (tx, mut rx) = mpsc::unbounded_channel(); - upload_with_retry(&client, &config, upload.clone(), &tx).await; - - assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_)))); - let Some(UploadEvent::Failed { will_retry, .. }) = rx.recv().await else { - panic!("expected Failed event"); - }; - assert!(will_retry, "500 should be a transient failure"); - assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_)))); - assert!(matches!(rx.recv().await, Some(UploadEvent::Succeeded { .. }))); - } - - /// Terminal failures (401, 413, ...) indicate a config or operator - /// problem that retries cannot fix. The sensitive audio file must - /// be deleted immediately — keeping it locally "in case the user - /// fixes the key" would leak patient data indefinitely for a - /// situation that in practice is rarely recoverable without manual - /// intervention anyway. - #[tokio::test] - async fn terminal_401_deletes_files_and_does_not_retry() { - let server = MockServer::start().await; - Mock::given(method("POST")) - .and(wpath(UPLOAD_PATH)) - .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized")) - .mount(&server) - .await; - - let tmp = TempDir::new().unwrap(); - let upload = sample_pending(tmp.path()); - write_sidecar(&upload).unwrap(); - let config = Config { - server_url: server.uri(), - api_key: "bad-key".into(), - oneliner_poll_interval_seconds: None, - oneliner_window_hours: None, - }; - - let client = reqwest::Client::new(); - let (tx, mut rx) = mpsc::unbounded_channel(); - upload_with_retry(&client, &config, upload.clone(), &tx).await; - - assert!( - !upload.file.exists(), - "m4a must be deleted on terminal failure" - ); - assert!( - !sidecar_path_for(&upload.file).exists(), - "sidecar must be deleted on terminal failure" - ); - - assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_)))); - let Some(UploadEvent::Failed { will_retry, .. }) = rx.recv().await else { - panic!("expected Failed event"); - }; - assert!(!will_retry, "401 is terminal; must not retry"); - } -} diff --git a/doctate-client-core/src/case_store.rs b/doctate-client-core/src/case_store.rs index 8abfd37..ee51de1 100644 --- a/doctate-client-core/src/case_store.rs +++ b/doctate-client-core/src/case_store.rs @@ -140,6 +140,30 @@ impl CaseStore { Ok(()) } + /// Mark an existing case as confirmed-by-server. Called by the + /// `ServerSync` worker immediately after a successful upload ACK — so + /// the `synced_to_server` flag reflects reality without waiting for + /// the next `/api/oneliners` poll to round-trip. + /// + /// No-op if the marker does not exist locally (e.g. `clear_all` was + /// called between `submit` and `ACK`). `last_activity_at` is **not** + /// touched: its semantics are "last user activity" (recording end), + /// not "last server confirmation". + pub async fn mark_synced(&self, case_id: Uuid) -> Result<(), CaseStoreError> { + let mut state = self.state.lock().await; + let Some(marker) = state.get_mut(&case_id) else { + return Ok(()); + }; + if marker.synced_to_server { + return Ok(()); // already true, skip the disk write + } + marker.synced_to_server = true; + let marker_clone = marker.clone(); + write_marker_file(&self.dir, &marker_clone).await?; + self.broadcast(&state); + Ok(()) + } + /// Apply a server `/api/oneliners` response. For every entry in the /// snapshot: insert (with `synced_to_server = true`) or update the /// local marker (oneliner + last_activity_at from server timestamps, @@ -647,6 +671,40 @@ mod tests { assert_eq!(store.list().await.len(), 1); } + #[tokio::test] + async fn mark_synced_flips_flag_without_touching_activity() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id = Uuid::new_v4(); + store.create_local(id, t("2026-04-18T10:00:00Z")).await.unwrap(); + assert!(!store.list().await[0].synced_to_server); + + store.mark_synced(id).await.unwrap(); + let list = store.list().await; + assert!(list[0].synced_to_server); + assert_eq!(list[0].last_activity_at, "2026-04-18T10:00:00Z"); + } + + #[tokio::test] + async fn mark_synced_is_noop_when_marker_missing() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + // No marker at all — must not fail, must not create one. + store.mark_synced(Uuid::new_v4()).await.unwrap(); + assert!(store.list().await.is_empty()); + } + + #[tokio::test] + async fn mark_synced_is_idempotent() { + let tmp = TempDir::new().unwrap(); + let (store, _rx) = CaseStore::open(tmp.path().to_owned()).await.unwrap(); + let id = Uuid::new_v4(); + store.create_local(id, t("2026-04-18T10:00:00Z")).await.unwrap(); + store.mark_synced(id).await.unwrap(); + store.mark_synced(id).await.unwrap(); // second call: no disk churn + assert!(store.list().await[0].synced_to_server); + } + #[tokio::test] async fn sorted_newest_first() { let tmp = TempDir::new().unwrap(); diff --git a/doctate-client-core/src/lib.rs b/doctate-client-core/src/lib.rs index 2d9fcb5..5ec8944 100644 --- a/doctate-client-core/src/lib.rs +++ b/doctate-client-core/src/lib.rs @@ -8,15 +8,19 @@ //! - [`case_store`]: per-case marker files, merge-with-server-snapshot logic //! - [`snapshot_cache`]: last successful `/api/oneliners` response, cached //! on disk so the UI has data at cold start -//! - [`oneliner_poller`]: background task that polls the server and feeds -//! the case store +//! - [`server_sync`]: unified background worker — uploads pending +//! recordings and polls `/api/oneliners` from a single HTTP client +//! - [`upload`]: pending-queue IO primitives (sidecar write/scan) +//! - [`pending_cleanup`]: orphan reaper for the pending-upload directory pub mod case_store; -pub mod oneliner_poller; pub mod pending_cleanup; +pub mod server_sync; pub mod snapshot_cache; +pub mod upload; pub use case_store::{CaseMarker, CaseStore, CaseStoreError}; -pub use oneliner_poller::{OnelinerPoller, PollerConfig}; pub use pending_cleanup::cleanup_orphan_audio; +pub use server_sync::{ServerSync, SyncConfig, WorkerSnapshot, WorkerState}; pub use snapshot_cache::{CachedSnapshot, SnapshotCache}; +pub use upload::{scan_pending_dir, sidecar_path_for, write_sidecar, PendingUpload, UploadEvent, UploadIoError}; diff --git a/doctate-client-core/src/oneliner_poller.rs b/doctate-client-core/src/oneliner_poller.rs deleted file mode 100644 index bf4e44c..0000000 --- a/doctate-client-core/src/oneliner_poller.rs +++ /dev/null @@ -1,537 +0,0 @@ -//! Background poller for `GET /api/oneliners`. -//! -//! Hits the server every `poll_interval` with `If-None-Match`, merges -//! 200-responses into the case store, persists each fresh response into -//! the snapshot cache, and survives transient errors by logging + waiting -//! for the next tick. -//! -//! The polling logic is factored into a standalone `poll_once` method on -//! [`PollerState`] so tests can drive a single cycle without wall-clock -//! timing. The [`OnelinerPoller`] handle just owns an [`AbortHandle`]: -//! dropping the handle aborts the task. - -use std::sync::Arc; -use std::time::{Duration, Instant}; - -use doctate_common::oneliners::{OnelinersResponse, ONELINERS_PATH}; -use doctate_common::API_KEY_HEADER; -use reqwest::StatusCode; -use tokio::task::AbortHandle; -use tracing::{debug, info, warn}; - -use crate::case_store::CaseStore; -use crate::snapshot_cache::{CachedSnapshot, SnapshotCache}; - -#[derive(Debug, Clone)] -pub struct PollerConfig { - pub server_url: String, - pub api_key: String, - pub poll_interval: Duration, - pub window_hours: u32, -} - -/// Handle that aborts the background task when dropped. -pub struct OnelinerPoller { - abort: AbortHandle, - /// Timestamp of the last successful 200 or 304 response. UI reads - /// this via `last_success` to render a staleness indicator. - last_success: Arc>>, -} - -impl OnelinerPoller { - /// Start the background polling task. Returns immediately. - /// - /// Expects a Tokio runtime context (call inside `Runtime::enter()` - /// or `#[tokio::main]`). - pub fn spawn( - http: Arc, - config: PollerConfig, - store: Arc, - cache: Arc, - ) -> Self { - let last_success = Arc::new(std::sync::Mutex::new(None)); - let last_success_for_task = last_success.clone(); - let handle = tokio::spawn(run_loop( - http, - config, - store, - cache, - last_success_for_task, - )); - Self { - abort: handle.abort_handle(), - last_success, - } - } - - /// `Some(Instant)` if the poller has ever received a 200 or 304 - /// response; `None` on cold start with no successful poll yet. - pub fn last_success(&self) -> Option { - *self.last_success.lock().expect("poisoned") - } -} - -impl Drop for OnelinerPoller { - fn drop(&mut self) { - self.abort.abort(); - } -} - -/// Task loop: load cache, then poll every `poll_interval` until aborted. -async fn run_loop( - http: Arc, - config: PollerConfig, - store: Arc, - cache: Arc, - last_success: Arc>>, -) { - info!( - interval_s = config.poll_interval.as_secs(), - window_h = config.window_hours, - "oneliner poller started" - ); - - let mut state = PollerState { - http, - config, - store, - cache, - etag: None, - }; - - // Prime the store + ETag from disk cache before the first request. - state.prime_from_cache().await; - - // Fire an immediate poll so the UI doesn't wait a full interval. - tick(&mut state, &last_success).await; - - let mut interval = tokio::time::interval(state.config.poll_interval); - interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - // The first tick is immediate (already covered above); consume it. - interval.tick().await; - - loop { - interval.tick().await; - tick(&mut state, &last_success).await; - } -} - -async fn tick( - state: &mut PollerState, - last_success: &Arc>>, -) { - match state.poll_once().await { - Ok(outcome) => { - if matches!(outcome, PollOutcome::Success | PollOutcome::NotModified) { - *last_success.lock().expect("poisoned") = Some(Instant::now()); - } - } - Err(e) => { - warn!(error = %e, "poll error — will retry on next tick"); - } - } -} - -#[derive(Debug, PartialEq, Eq)] -pub enum PollOutcome { - /// Server returned 200 with a fresh snapshot — merged into the store. - Success, - /// Server returned 304 Not Modified — no change. - NotModified, - /// Transient HTTP failure (non-2xx non-304); retried next tick. - TransientHttp(u16), -} - -#[derive(Debug, thiserror::Error)] -pub enum PollError { - #[error("http: {0}")] - Http(#[from] reqwest::Error), - #[error("deserialize response: {0}")] - Parse(#[from] serde_json::Error), - #[error("cache write: {0}")] - Cache(std::io::Error), - #[error("case store: {0}")] - Store(#[from] crate::case_store::CaseStoreError), - #[error("format timestamp: {0}")] - Format(#[from] time::error::Format), -} - -/// Driver for a single poll cycle. Holds the long-lived `reqwest::Client`, -/// the config (unchanging over the task's lifetime), the store and cache -/// (shared with the UI), and the last ETag seen from the server. -pub(crate) struct PollerState { - http: Arc, - config: PollerConfig, - store: Arc, - cache: Arc, - etag: Option, -} - -impl PollerState { - /// Load a cached snapshot (if any) from disk, merge it into the store - /// and adopt its ETag. Runs once at task start so the UI sees data - /// before the first network round-trip completes. - async fn prime_from_cache(&mut self) { - let Some(cached) = self.cache.load().await else { - return; - }; - if let Err(e) = self.store.merge_server_snapshot(&cached.response).await { - warn!(error = %e, "priming store from cache failed"); - } - self.etag = Some(cached.etag); - debug!("store primed from cache"); - } - - /// Issue one request, update store/cache/etag accordingly. Returns - /// what happened so the caller can track success timestamps. - pub(crate) async fn poll_once(&mut self) -> Result { - let url = format!( - "{}{}?hours={}", - self.config.server_url.trim_end_matches('/'), - ONELINERS_PATH, - self.config.window_hours - ); - - let mut req = self - .http - .get(&url) - .header(API_KEY_HEADER, &self.config.api_key); - if let Some(etag) = &self.etag { - req = req.header("If-None-Match", etag); - } - - let resp = req.send().await?; - let status = resp.status(); - - match status { - StatusCode::OK => { - let new_etag = extract_etag(&resp); - let body: OnelinersResponse = resp.json().await?; - self.store.merge_server_snapshot(&body).await?; - - if let Some(etag) = new_etag.clone() { - let cached = CachedSnapshot { - etag: etag.clone(), - fetched_at: doctate_common::now_rfc3339(), - response: body, - }; - if let Err(e) = self.cache.store(&cached).await { - warn!(error = %e, "snapshot cache write failed — continuing"); - } - self.etag = Some(etag); - } else { - // Server without ETag: accept body but don't try - // conditional requests. Unexpected from our own - // server; defensive fallback. - warn!("server returned 200 without ETag — conditional requests disabled"); - self.etag = None; - } - Ok(PollOutcome::Success) - } - StatusCode::NOT_MODIFIED => Ok(PollOutcome::NotModified), - other => Ok(PollOutcome::TransientHttp(other.as_u16())), - } - } -} - -fn extract_etag(resp: &reqwest::Response) -> Option { - resp.headers() - .get("etag") - .and_then(|v| v.to_str().ok()) - .map(str::to_owned) -} - -#[cfg(test)] -mod tests { - use super::*; - use doctate_common::oneliners::OnelinerEntry; - use tempfile::TempDir; - use wiremock::matchers::{header, header_exists, method, path, query_param}; - use wiremock::{Mock, MockServer, ResponseTemplate}; - - const TEST_KEY: &str = "test-key-poller"; - const TEST_ETAG: &str = "W/\"42-72\""; - - async fn build_state( - tmp: &TempDir, - server_uri: String, - ) -> (PollerState, Arc, Arc) { - let (store_inner, _rx) = CaseStore::open(tmp.path().join("cases")).await.unwrap(); - let store = Arc::new(store_inner); - let cache = Arc::new(SnapshotCache::new(tmp.path().join("cache.json"))); - let state = PollerState { - http: Arc::new(reqwest::Client::new()), - config: PollerConfig { - server_url: server_uri, - api_key: TEST_KEY.into(), - poll_interval: Duration::from_secs(5), - window_hours: 72, - }, - store: store.clone(), - cache: cache.clone(), - etag: None, - }; - (state, store, cache) - } - - fn response_body(case_id: &str, oneliner: Option<&str>) -> OnelinersResponse { - OnelinersResponse { - as_of: "2026-04-18T10:00:00Z".into(), - window_hours: 72, - oneliners: vec![OnelinerEntry { - case_id: case_id.into(), - oneliner: oneliner.map(str::to_owned), - created_at: "2026-04-18T09:30:00Z".into(), - last_recording_at: Some("2026-04-18T09:45:00Z".into()), - updated_at: Some("2026-04-18T09:45:00Z".into()), - }], - } - } - - #[tokio::test] - async fn poll_once_on_200_merges_store_and_caches() { - let server = MockServer::start().await; - let case_id = "550e8400-e29b-41d4-a716-000000000001"; - Mock::given(method("GET")) - .and(path(ONELINERS_PATH)) - .and(query_param("hours", "72")) - .and(header(API_KEY_HEADER, TEST_KEY)) - .respond_with( - ResponseTemplate::new(200) - .insert_header("etag", TEST_ETAG) - .set_body_json(response_body(case_id, Some("Kniegelenk"))), - ) - .mount(&server) - .await; - - let tmp = TempDir::new().unwrap(); - let (mut state, store, cache) = build_state(&tmp, server.uri()).await; - assert_eq!(state.poll_once().await.unwrap(), PollOutcome::Success); - - let list = store.list().await; - assert_eq!(list.len(), 1); - assert_eq!(list[0].oneliner.as_deref(), Some("Kniegelenk")); - assert!(list[0].synced_to_server); - - let cached = cache.load().await.expect("cache written"); - assert_eq!(cached.etag, TEST_ETAG); - assert_eq!(state.etag.as_deref(), Some(TEST_ETAG)); - } - - #[tokio::test] - async fn poll_once_sends_if_none_match_when_etag_known() { - let server = MockServer::start().await; - Mock::given(method("GET")) - .and(path(ONELINERS_PATH)) - .and(header("If-None-Match", TEST_ETAG)) - .respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG)) - .expect(1) - .mount(&server) - .await; - - let tmp = TempDir::new().unwrap(); - let (mut state, _store, _cache) = build_state(&tmp, server.uri()).await; - state.etag = Some(TEST_ETAG.into()); - - assert_eq!(state.poll_once().await.unwrap(), PollOutcome::NotModified); - // Mock's `.expect(1)` enforces the conditional header was sent. - } - - #[tokio::test] - async fn poll_once_first_call_has_no_if_none_match() { - let server = MockServer::start().await; - let case_id = "550e8400-e29b-41d4-a716-000000000002"; - // Only a mock that does NOT require If-None-Match matches; if the - // poller sent one, wiremock would return 404. - Mock::given(method("GET")) - .and(path(ONELINERS_PATH)) - .respond_with( - ResponseTemplate::new(200) - .insert_header("etag", TEST_ETAG) - .set_body_json(response_body(case_id, Some("x"))), - ) - .expect(1) - .mount(&server) - .await; - - let tmp = TempDir::new().unwrap(); - let (mut state, _store, _cache) = build_state(&tmp, server.uri()).await; - assert!(state.etag.is_none()); - assert_eq!(state.poll_once().await.unwrap(), PollOutcome::Success); - } - - #[tokio::test] - async fn poll_once_on_304_does_not_overwrite_store() { - let server = MockServer::start().await; - Mock::given(method("GET")) - .respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG)) - .mount(&server) - .await; - - let tmp = TempDir::new().unwrap(); - let (mut state, store, cache) = build_state(&tmp, server.uri()).await; - - // Seed store with a local marker — must survive a 304. - let id = uuid::Uuid::new_v4(); - store - .create_local( - id, - time::OffsetDateTime::parse( - "2026-04-18T10:00:00Z", - &time::format_description::well_known::Rfc3339, - ) - .unwrap(), - ) - .await - .unwrap(); - - assert_eq!(state.poll_once().await.unwrap(), PollOutcome::NotModified); - assert_eq!(store.list().await.len(), 1); - assert!(cache.load().await.is_none(), "304 must not create cache"); - } - - #[tokio::test] - async fn poll_once_returns_transient_on_500() { - let server = MockServer::start().await; - Mock::given(method("GET")) - .respond_with(ResponseTemplate::new(500)) - .mount(&server) - .await; - let tmp = TempDir::new().unwrap(); - let (mut state, _store, _cache) = build_state(&tmp, server.uri()).await; - assert_eq!( - state.poll_once().await.unwrap(), - PollOutcome::TransientHttp(500) - ); - } - - #[tokio::test] - async fn poll_once_returns_transient_on_401() { - let server = MockServer::start().await; - Mock::given(method("GET")) - .respond_with(ResponseTemplate::new(401)) - .mount(&server) - .await; - let tmp = TempDir::new().unwrap(); - let (mut state, _store, _cache) = build_state(&tmp, server.uri()).await; - // 401 bubbles up as TransientHttp so the task keeps running; the - // user surfaces the wrong-API-key state via the config panel, not - // via a fatal poller crash. - assert_eq!( - state.poll_once().await.unwrap(), - PollOutcome::TransientHttp(401) - ); - } - - #[tokio::test] - async fn poll_once_propagates_network_error() { - let tmp = TempDir::new().unwrap(); - // Unbound address → connection refused. - let (mut state, _store, _cache) = - build_state(&tmp, "http://127.0.0.1:1".into()).await; - let err = state.poll_once().await.unwrap_err(); - assert!(matches!(err, PollError::Http(_))); - } - - #[tokio::test] - async fn prime_from_cache_seeds_store_and_etag() { - let tmp = TempDir::new().unwrap(); - let (mut state, store, cache) = build_state(&tmp, "http://unused".into()).await; - - let case_id = "550e8400-e29b-41d4-a716-000000000003"; - cache - .store(&CachedSnapshot { - etag: TEST_ETAG.into(), - fetched_at: "2026-04-18T10:00:00Z".into(), - response: response_body(case_id, Some("cached")), - }) - .await - .unwrap(); - - state.prime_from_cache().await; - assert_eq!(state.etag.as_deref(), Some(TEST_ETAG)); - let list = store.list().await; - assert_eq!(list.len(), 1); - assert_eq!(list[0].oneliner.as_deref(), Some("cached")); - } - - #[tokio::test] - async fn spawn_and_drop_aborts_task() { - let server = MockServer::start().await; - // Respond to any GET so the loop has work to do. - Mock::given(method("GET")) - .respond_with( - ResponseTemplate::new(200) - .insert_header("etag", TEST_ETAG) - .set_body_json(response_body( - "550e8400-e29b-41d4-a716-000000000004", - Some("x"), - )), - ) - .mount(&server) - .await; - - let tmp = TempDir::new().unwrap(); - let (store_inner, _rx) = CaseStore::open(tmp.path().join("cases")).await.unwrap(); - let store = Arc::new(store_inner); - let cache = Arc::new(SnapshotCache::new(tmp.path().join("cache.json"))); - let poller = OnelinerPoller::spawn( - Arc::new(reqwest::Client::new()), - PollerConfig { - server_url: server.uri(), - api_key: TEST_KEY.into(), - poll_interval: Duration::from_millis(50), - window_hours: 72, - }, - store.clone(), - cache, - ); - - // Let the immediate first poll fire. - tokio::time::sleep(Duration::from_millis(200)).await; - assert!(poller.last_success().is_some(), "expected at least one poll"); - assert_eq!(store.list().await.len(), 1); - - // Drop aborts the task; subsequent store inspection must not - // race with a concurrent merge — asserting no panic is enough. - drop(poller); - tokio::time::sleep(Duration::from_millis(50)).await; - } - - #[tokio::test] - async fn spawn_with_matching_mock_requires_api_key_header() { - let server = MockServer::start().await; - Mock::given(method("GET")) - .and(path(ONELINERS_PATH)) - .and(header_exists(API_KEY_HEADER)) - .respond_with( - ResponseTemplate::new(200) - .insert_header("etag", TEST_ETAG) - .set_body_json(response_body( - "550e8400-e29b-41d4-a716-000000000005", - None, - )), - ) - .expect(1) - .mount(&server) - .await; - - let tmp = TempDir::new().unwrap(); - let (store_inner, _rx) = CaseStore::open(tmp.path().join("cases")).await.unwrap(); - let store = Arc::new(store_inner); - let cache = Arc::new(SnapshotCache::new(tmp.path().join("cache.json"))); - let _poller = OnelinerPoller::spawn( - Arc::new(reqwest::Client::new()), - PollerConfig { - server_url: server.uri(), - api_key: TEST_KEY.into(), - poll_interval: Duration::from_secs(10), - window_hours: 72, - }, - store.clone(), - cache, - ); - tokio::time::sleep(Duration::from_millis(200)).await; - // Mock's `.expect(1)` is verified when `server` drops at end of test. - } -} diff --git a/doctate-client-core/src/server_sync.rs b/doctate-client-core/src/server_sync.rs new file mode 100644 index 0000000..5c33e96 --- /dev/null +++ b/doctate-client-core/src/server_sync.rs @@ -0,0 +1,898 @@ +//! Unified client-side worker that owns the only outbound HTTP channel: +//! it uploads pending recordings and polls `/api/oneliners` from a single +//! `reqwest::Client`, in a single `tokio::select!` loop. +//! +//! Design constraints (see plan `berpr-fe-den-plan-lively-journal.md`): +//! +//! - **Uploads have priority.** If anything sits in `pending/`, the worker +//! uploads it; polling happens only when the directory is empty. +//! - **Disk is the queue.** No in-memory `VecDeque` — the worker scans +//! `pending/` at the top of every iteration. FIFO via `recorded_at`. +//! - **Kick channel, no payload.** `notify()` on the handle just wakes +//! the `select!` sleep; the next iteration finds the fresh file on +//! disk. +//! - **Backoff is a local `Duration`.** Exponential, resets on success, +//! capped at `MAX_BACKOFF`. Not persisted across restarts. +//! - **Terminal upload failures delete the file.** 401/413/etc. will +//! never succeed; keeping the audio around would violate the +//! data-minimization rule. + +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex as StdMutex}; +use std::time::{Duration, Instant}; + +use doctate_common::ack::{AckResponse, AckStatus}; +use doctate_common::constants::{ + API_KEY_HEADER, CONTENT_TYPE_AUDIO_MP4, FIELD_AUDIO, FIELD_CASE_ID, FIELD_RECORDED_AT, + UPLOAD_PATH, +}; +use doctate_common::oneliners::{OnelinersResponse, ONELINERS_PATH}; +use reqwest::StatusCode; +use tokio::sync::{mpsc, watch}; +use tokio::task::AbortHandle; +use tracing::{debug, info, warn}; + +use crate::case_store::CaseStore; +use crate::snapshot_cache::{CachedSnapshot, SnapshotCache}; +use crate::upload::{scan_pending_dir, sidecar_path_for, PendingUpload, UploadEvent}; + +const INITIAL_BACKOFF: Duration = Duration::from_secs(2); +const MAX_BACKOFF: Duration = Duration::from_secs(60); + +#[derive(Debug, Clone)] +pub struct SyncConfig { + pub server_url: String, + pub api_key: String, + pub poll_interval: Duration, + pub window_hours: u32, +} + +/// Three-state observable: what the worker is currently doing. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkerState { + Idle, + Uploading, + Polling, +} + +/// Single observable the UI reads each frame. `watch::channel` semantics: +/// only the last value matters, readers see either the newest or nothing. +#[derive(Debug, Clone)] +pub struct WorkerSnapshot { + pub state: WorkerState, + /// Result of the last `scan_pending_dir` at the top of a loop + /// iteration. `0` during `Polling` / `Idle`, positive during + /// `Uploading`. + pub queue_len: usize, + /// Set to `Some(now)` after any successful upload or poll (200/304). + /// `None` before the first success. Drives Online/Offline heuristics. + pub last_success: Option, +} + +impl Default for WorkerSnapshot { + fn default() -> Self { + Self { + state: WorkerState::Idle, + queue_len: 0, + last_success: None, + } + } +} + +/// Handle to the worker task. Drop aborts the task; outstanding disk +/// artefacts survive and are picked up by the next worker instance. +pub struct ServerSync { + abort: AbortHandle, + kick_tx: mpsc::UnboundedSender<()>, + snapshot_rx: watch::Receiver, + /// Kept in an `Arc>` so tests can peek at `last_success` + /// independently of the watch channel. Not required for production. + #[allow(dead_code)] + last_success_mirror: Arc>>, +} + +impl ServerSync { + /// Spawn the worker task. Returns the handle plus the event receiver + /// for upload lifecycle events (started / succeeded / failed). + /// + /// Must run inside a Tokio runtime context. + pub fn spawn( + http: Arc, + config: SyncConfig, + store: Arc, + cache: Arc, + pending_dir: PathBuf, + ) -> (Self, mpsc::UnboundedReceiver) { + let (kick_tx, kick_rx) = mpsc::unbounded_channel::<()>(); + let (events_tx, events_rx) = mpsc::unbounded_channel::(); + let (snapshot_tx, snapshot_rx) = watch::channel(WorkerSnapshot::default()); + let last_success_mirror = Arc::new(StdMutex::new(None)); + + let task = tokio::spawn(run_loop( + http, + config, + store, + cache, + pending_dir, + kick_rx, + events_tx, + snapshot_tx, + last_success_mirror.clone(), + )); + + let handle = Self { + abort: task.abort_handle(), + kick_tx, + snapshot_rx, + last_success_mirror, + }; + (handle, events_rx) + } + + /// Wake the worker if it's sleeping in the poll-idle `select!`. Used + /// by the app thread after writing a fresh sidecar to `pending/`. + /// Failure to send means the worker is already dropped — fine, + /// ignore. + pub fn notify(&self) { + let _ = self.kick_tx.send(()); + } + + /// Cheap clone of the observable for the UI. Each call yields an + /// independent receiver — all receivers see the same `borrow()`. + pub fn snapshot(&self) -> watch::Receiver { + self.snapshot_rx.clone() + } +} + +impl Drop for ServerSync { + fn drop(&mut self) { + self.abort.abort(); + } +} + +/// Outcome classification for a single upload attempt. +pub(crate) enum UploadOutcome { + Succeeded(AckStatus), + Transient(String), + Terminal(String), +} + +/// Outcome classification for a single poll cycle. Symmetric to the old +/// `PollOutcome` but folded into the unified module. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum PollOutcome { + Success, + NotModified, + /// Non-2xx non-304 HTTP; logged as transient, retried next tick. + TransientHttp(u16), +} + +#[allow(clippy::too_many_arguments)] +async fn run_loop( + http: Arc, + config: SyncConfig, + store: Arc, + cache: Arc, + pending_dir: PathBuf, + mut kick_rx: mpsc::UnboundedReceiver<()>, + events_tx: mpsc::UnboundedSender, + snapshot_tx: watch::Sender, + last_success_mirror: Arc>>, +) { + info!( + interval_s = config.poll_interval.as_secs(), + window_h = config.window_hours, + "server sync worker started" + ); + + // Prime the store + ETag from disk cache before the first request. + let mut etag: Option = None; + if let Some(cached) = cache.load().await { + if let Err(e) = store.merge_server_snapshot(&cached.response).await { + warn!(error = %e, "priming store from cache failed"); + } + etag = Some(cached.etag); + debug!("store primed from cache"); + } + + let mut backoff = INITIAL_BACKOFF; + let mut last_success: Option = None; + + let publish = |state: WorkerState, queue_len: usize, last: Option| { + let snap = WorkerSnapshot { + state, + queue_len, + last_success: last, + }; + let _ = snapshot_tx.send(snap); + }; + + loop { + // Top of iteration: re-scan disk. Cheap when the directory is + // empty; one stat per file when it isn't. + let mut files = scan_pending_dir(&pending_dir); + files.sort_by(|a, b| a.recorded_at.cmp(&b.recorded_at)); // FIFO + let queue_len = files.len(); + let next = files.into_iter().next(); + + if let Some(upload) = next { + publish(WorkerState::Uploading, queue_len, last_success); + let _ = events_tx.send(UploadEvent::Started(upload.case_id)); + match post_upload(&http, &config, &upload).await { + UploadOutcome::Succeeded(status) => { + delete_pair(&upload.file).await; + if let Err(e) = store.mark_synced(upload.case_id).await { + warn!(error = %e, case_id = %upload.case_id, "mark_synced failed"); + } + let now = Instant::now(); + last_success = Some(now); + *last_success_mirror.lock().expect("poisoned") = Some(now); + backoff = INITIAL_BACKOFF; + let _ = events_tx.send(UploadEvent::Succeeded { + case_id: upload.case_id, + status, + }); + } + UploadOutcome::Transient(reason) => { + warn!(case_id = %upload.case_id, reason = %reason, "upload failed transiently"); + let _ = events_tx.send(UploadEvent::Failed { + case_id: upload.case_id, + reason, + will_retry: true, + }); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(MAX_BACKOFF); + } + UploadOutcome::Terminal(reason) => { + warn!(case_id = %upload.case_id, reason = %reason, "upload failed terminally — deleting files"); + delete_pair(&upload.file).await; + backoff = INITIAL_BACKOFF; + let _ = events_tx.send(UploadEvent::Failed { + case_id: upload.case_id, + reason, + will_retry: false, + }); + } + } + } else { + publish(WorkerState::Polling, 0, last_success); + match poll_once(&http, &config, &store, &cache, &mut etag).await { + Ok(PollOutcome::Success) | Ok(PollOutcome::NotModified) => { + let now = Instant::now(); + last_success = Some(now); + *last_success_mirror.lock().expect("poisoned") = Some(now); + } + Ok(PollOutcome::TransientHttp(code)) => { + warn!(code, "poll returned transient HTTP error"); + } + Err(e) => { + warn!(error = %e, "poll error"); + } + } + + publish(WorkerState::Idle, 0, last_success); + + // Wait for next tick OR kick. Cancel-safe on both arms. + tokio::select! { + _ = tokio::time::sleep(config.poll_interval) => {} + _ = kick_rx.recv() => {} + } + } + } +} + +/// Perform one upload attempt. Error classification follows the rules: +/// - I/O errors reading the audio file → Terminal (file is broken, retry +/// won't help). +/// - `reqwest::send` errors → Transient (network). +/// - 2xx with parseable ACK → Succeeded. +/// - 408/429/5xx → Transient (server will likely recover). +/// - Other 4xx → Terminal (401, 413, 415, …; operator must fix). +pub(crate) async fn post_upload( + http: &reqwest::Client, + config: &SyncConfig, + upload: &PendingUpload, +) -> UploadOutcome { + let audio = match tokio::fs::read(&upload.file).await { + Ok(bytes) => bytes, + Err(e) => return UploadOutcome::Terminal(format!("read audio file: {e}")), + }; + + let file_name = upload + .file + .file_name() + .map(|s| s.to_string_lossy().into_owned()) + .unwrap_or_else(|| "recording.m4a".to_string()); + + let audio_part = match reqwest::multipart::Part::bytes(audio) + .file_name(file_name) + .mime_str(CONTENT_TYPE_AUDIO_MP4) + { + Ok(p) => p, + Err(e) => return UploadOutcome::Terminal(format!("build multipart: {e}")), + }; + + let form = reqwest::multipart::Form::new() + .text(FIELD_CASE_ID, upload.case_id.to_string()) + .text(FIELD_RECORDED_AT, upload.recorded_at.clone()) + .part(FIELD_AUDIO, audio_part); + + let url = format!("{}{}", config.server_url.trim_end_matches('/'), UPLOAD_PATH); + + let resp = match http + .post(&url) + .header(API_KEY_HEADER, &config.api_key) + .multipart(form) + .send() + .await + { + Ok(r) => r, + Err(e) => return UploadOutcome::Transient(format!("network: {e}")), + }; + + let status = resp.status(); + if status.is_success() { + match resp.json::().await { + Ok(ack) => UploadOutcome::Succeeded(ack.status), + Err(e) => UploadOutcome::Terminal(format!("parse ack: {e}")), + } + } else if status.as_u16() == 408 || status.as_u16() == 429 || status.is_server_error() { + let body = resp.text().await.unwrap_or_default(); + UploadOutcome::Transient(format!("HTTP {status}: {body}")) + } else { + let body = resp.text().await.unwrap_or_default(); + UploadOutcome::Terminal(format!("HTTP {status}: {body}")) + } +} + +/// Perform one poll of `/api/oneliners`. Side effects: on 200 merge into +/// the store and persist the response to the cache; update `etag` on 200. +pub(crate) async fn poll_once( + http: &reqwest::Client, + config: &SyncConfig, + store: &CaseStore, + cache: &SnapshotCache, + etag: &mut Option, +) -> Result { + let url = format!( + "{}{}?hours={}", + config.server_url.trim_end_matches('/'), + ONELINERS_PATH, + config.window_hours + ); + + let mut req = http.get(&url).header(API_KEY_HEADER, &config.api_key); + if let Some(tag) = etag.as_deref() { + req = req.header("If-None-Match", tag); + } + + let resp = req.send().await?; + let status = resp.status(); + + match status { + StatusCode::OK => { + let new_etag = extract_etag(&resp); + let body: OnelinersResponse = resp.json().await?; + store.merge_server_snapshot(&body).await?; + + if let Some(tag) = new_etag.clone() { + let cached = CachedSnapshot { + etag: tag.clone(), + fetched_at: doctate_common::now_rfc3339(), + response: body, + }; + if let Err(e) = cache.store(&cached).await { + warn!(error = %e, "snapshot cache write failed"); + } + *etag = Some(tag); + } else { + warn!("server returned 200 without ETag — conditional requests disabled"); + *etag = None; + } + Ok(PollOutcome::Success) + } + StatusCode::NOT_MODIFIED => Ok(PollOutcome::NotModified), + other => Ok(PollOutcome::TransientHttp(other.as_u16())), + } +} + +#[derive(Debug, thiserror::Error)] +pub enum PollError { + #[error("http: {0}")] + Http(#[from] reqwest::Error), + #[error("deserialize response: {0}")] + Parse(#[from] serde_json::Error), + #[error("case store: {0}")] + Store(#[from] crate::case_store::CaseStoreError), +} + +fn extract_etag(resp: &reqwest::Response) -> Option { + resp.headers() + .get("etag") + .and_then(|v| v.to_str().ok()) + .map(str::to_owned) +} + +async fn delete_pair(m4a: &Path) { + if let Err(e) = tokio::fs::remove_file(m4a).await { + warn!(path = %m4a.display(), error = %e, "delete m4a failed"); + } + let sidecar = sidecar_path_for(m4a); + if let Err(e) = tokio::fs::remove_file(&sidecar).await { + warn!(path = %sidecar.display(), error = %e, "delete sidecar failed"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::upload::write_sidecar; + use doctate_common::oneliners::OnelinerEntry; + use tempfile::TempDir; + use uuid::Uuid; + use wiremock::matchers::{header, method, path as wpath}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + const TEST_KEY: &str = "test-key-sync"; + const TEST_ETAG: &str = "W/\"42-72\""; + + fn cfg(server: &MockServer) -> SyncConfig { + SyncConfig { + server_url: server.uri(), + api_key: TEST_KEY.into(), + poll_interval: Duration::from_millis(50), + window_hours: 72, + } + } + + fn ack_body(status: AckStatus) -> AckResponse { + AckResponse { + case_id: "550e8400-e29b-41d4-a716-446655440000".into(), + recorded_at: "2026-04-18T10:32:00Z".into(), + status, + } + } + + fn oneliner_body(case_id: &str, oneliner: Option<&str>) -> OnelinersResponse { + OnelinersResponse { + as_of: "2026-04-18T10:00:00Z".into(), + window_hours: 72, + oneliners: vec![OnelinerEntry { + case_id: case_id.into(), + oneliner: oneliner.map(str::to_owned), + created_at: "2026-04-18T09:30:00Z".into(), + last_recording_at: Some("2026-04-18T09:45:00Z".into()), + updated_at: Some("2026-04-18T09:45:00Z".into()), + }], + } + } + + fn sample_pending(dir: &Path, case_id: Uuid, recorded_at: &str) -> PendingUpload { + let stem = format!("{case_id}_{}", recorded_at.replace([':', '-'], "-")); + let file = dir.join(format!("{stem}.m4a")); + std::fs::write(&file, b"dummy audio").unwrap(); + let upload = PendingUpload { + case_id, + recorded_at: recorded_at.into(), + file, + }; + write_sidecar(&upload).unwrap(); + upload + } + + async fn make_store_and_cache(tmp: &TempDir) -> (Arc, Arc) { + let (store, _rx) = CaseStore::open(tmp.path().join("cases")).await.unwrap(); + let cache = SnapshotCache::new(tmp.path().join("cache.json")); + (Arc::new(store), Arc::new(cache)) + } + + // --- unit-level tests on post_upload / poll_once --- + + #[tokio::test] + async fn post_upload_succeeds_on_200_with_ack() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .and(header(API_KEY_HEADER, TEST_KEY)) + .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received))) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let upload = sample_pending(tmp.path(), Uuid::new_v4(), "2026-04-18T10-32-00Z"); + + let http = reqwest::Client::new(); + let outcome = post_upload(&http, &cfg(&server), &upload).await; + assert!(matches!(outcome, UploadOutcome::Succeeded(AckStatus::Received))); + } + + #[tokio::test] + async fn post_upload_transient_on_500() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500).set_body_string("boom")) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let upload = sample_pending(tmp.path(), Uuid::new_v4(), "2026-04-18T10-32-00Z"); + + let http = reqwest::Client::new(); + assert!(matches!( + post_upload(&http, &cfg(&server), &upload).await, + UploadOutcome::Transient(_) + )); + } + + #[tokio::test] + async fn post_upload_terminal_on_401() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(401).set_body_string("nope")) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let upload = sample_pending(tmp.path(), Uuid::new_v4(), "2026-04-18T10-32-00Z"); + + let http = reqwest::Client::new(); + assert!(matches!( + post_upload(&http, &cfg(&server), &upload).await, + UploadOutcome::Terminal(_) + )); + } + + #[tokio::test] + async fn poll_once_200_merges_and_caches() { + let server = MockServer::start().await; + let case_id = "550e8400-e29b-41d4-a716-000000000001"; + Mock::given(method("GET")) + .and(wpath(ONELINERS_PATH)) + .respond_with( + ResponseTemplate::new(200) + .insert_header("etag", TEST_ETAG) + .set_body_json(oneliner_body(case_id, Some("K"))), + ) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let (store, cache) = make_store_and_cache(&tmp).await; + 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_eq!(etag.as_deref(), Some(TEST_ETAG)); + assert_eq!(store.list().await.len(), 1); + assert!(cache.load().await.is_some()); + } + + // --- high-level spawn/loop tests --- + + /// Startup scan: files left in pending_dir before spawn must be + /// uploaded without any explicit submit call. + #[tokio::test] + async fn startup_scan_picks_up_leftover_files() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received))) + .mount(&server) + .await; + // A catch-all for the idle poll so wiremock doesn't 404 after upload. + Mock::given(method("GET")) + .and(wpath(ONELINERS_PATH)) + .respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG)) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let pending_dir = tmp.path().join("pending"); + std::fs::create_dir_all(&pending_dir).unwrap(); + let case_id = Uuid::new_v4(); + sample_pending(&pending_dir, case_id, "2026-04-18T10-00-00Z"); + + let (store, cache) = make_store_and_cache(&tmp).await; + store.create_local(case_id, time::OffsetDateTime::now_utc()) + .await + .unwrap(); + + let (sync, mut events) = ServerSync::spawn( + Arc::new(reqwest::Client::new()), + cfg(&server), + store.clone(), + cache, + pending_dir.clone(), + ); + + // Expect Started + Succeeded within a short window. + let mut saw_started = false; + let mut saw_success = false; + for _ in 0..10 { + let Ok(Some(ev)) = + tokio::time::timeout(Duration::from_millis(500), events.recv()).await + else { + continue; + }; + match ev { + UploadEvent::Started(_) => saw_started = true, + UploadEvent::Succeeded { .. } => { + saw_success = true; + break; + } + UploadEvent::Failed { reason, .. } => panic!("unexpected failure: {reason}"), + } + } + assert!(saw_started); + assert!(saw_success); + + // Disk-side: pair must be gone, marker must be synced. + let leftovers = scan_pending_dir(&pending_dir); + assert!(leftovers.is_empty(), "files must be deleted after ACK"); + assert!( + store.list().await.iter().any(|m| m.synced_to_server), + "mark_synced must have flipped the flag" + ); + + drop(sync); + } + + /// Polling only happens when the pending dir is empty. With an item + /// present, `GET /api/oneliners` must not fire before the upload + /// completes (enforced by a mock with `.expect(0)`). + #[tokio::test] + async fn poll_deferred_while_pending_upload() { + let server = MockServer::start().await; + // Upload always succeeds. + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received))) + .expect(1) + .mount(&server) + .await; + // No GET on /api/oneliners is allowed until after the upload — + // wiremock's `.expect(0)` verifies this when `server` drops. + Mock::given(method("GET")) + .and(wpath(ONELINERS_PATH)) + .respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG)) + .expect(0..) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let pending_dir = tmp.path().join("pending"); + std::fs::create_dir_all(&pending_dir).unwrap(); + sample_pending(&pending_dir, Uuid::new_v4(), "2026-04-18T10-00-00Z"); + + let (store, cache) = make_store_and_cache(&tmp).await; + let (sync, mut events) = ServerSync::spawn( + Arc::new(reqwest::Client::new()), + cfg(&server), + store, + cache, + pending_dir, + ); + + // Wait for the single Succeeded event. + loop { + match tokio::time::timeout(Duration::from_millis(500), events.recv()).await { + Ok(Some(UploadEvent::Succeeded { .. })) => break, + Ok(Some(_)) => continue, + Ok(None) => panic!("event channel closed"), + Err(_) => panic!("timed out waiting for Succeeded"), + } + } + + drop(sync); + // Mock expectations are checked on `server` drop at end of test. + } + + /// Terminal 401: file + sidecar must be deleted, no retry loop. + #[tokio::test] + async fn terminal_401_deletes_files_no_retry() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .respond_with(ResponseTemplate::new(401).set_body_string("nope")) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(wpath(ONELINERS_PATH)) + .respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG)) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let pending_dir = tmp.path().join("pending"); + std::fs::create_dir_all(&pending_dir).unwrap(); + let upload = sample_pending(&pending_dir, Uuid::new_v4(), "2026-04-18T10-00-00Z"); + + let (store, cache) = make_store_and_cache(&tmp).await; + let (sync, mut events) = ServerSync::spawn( + Arc::new(reqwest::Client::new()), + cfg(&server), + store, + cache, + pending_dir.clone(), + ); + + // Expect a Failed{will_retry:false} event. + let mut saw_terminal = false; + for _ in 0..10 { + if let Ok(Some(UploadEvent::Failed { + will_retry: false, .. + })) = tokio::time::timeout(Duration::from_millis(500), events.recv()).await + { + saw_terminal = true; + break; + } + } + assert!(saw_terminal, "expected terminal failure event"); + + // Files gone. + assert!(!upload.file.exists()); + assert!(!sidecar_path_for(&upload.file).exists()); + + drop(sync); + } + + /// Kick wakes the worker out of the idle sleep. With a long + /// poll_interval, the only way an upload happens quickly after a + /// submit is via the kick channel. + #[tokio::test] + async fn notify_kick_wakes_idle_worker() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received))) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(wpath(ONELINERS_PATH)) + .respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG)) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let pending_dir = tmp.path().join("pending"); + std::fs::create_dir_all(&pending_dir).unwrap(); + + let (store, cache) = make_store_and_cache(&tmp).await; + let mut slow_cfg = cfg(&server); + slow_cfg.poll_interval = Duration::from_secs(60); // effectively never ticks + + let (sync, mut events) = ServerSync::spawn( + Arc::new(reqwest::Client::new()), + slow_cfg, + store, + cache, + pending_dir.clone(), + ); + + // Give the worker a moment to settle into idle-sleep. + tokio::time::sleep(Duration::from_millis(100)).await; + + // Now drop a file in and kick the worker. + sample_pending(&pending_dir, Uuid::new_v4(), "2026-04-18T10-00-00Z"); + sync.notify(); + + // Without the kick, we'd wait 60s. With it, we see an event fast. + let success = async { + loop { + if let Some(UploadEvent::Succeeded { .. }) = events.recv().await { + break; + } + } + }; + tokio::time::timeout(Duration::from_secs(3), success) + .await + .expect("kick should have woken the worker well within 3s"); + + drop(sync); + } + + /// FIFO order by `recorded_at`: oldest file uploads first. + #[tokio::test] + async fn fifo_order_by_recorded_at() { + let server = MockServer::start().await; + // Record which case_ids we saw, in order. + let seen: Arc>> = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let seen_for_mock = seen.clone(); + let responder = wiremock::ResponseTemplate::new(200) + .set_body_json(ack_body(AckStatus::Received)); + // Simpler: just let both POSTs succeed; infer order from event + // stream on our side. + let _ = seen_for_mock; + let _ = responder; + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received))) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(wpath(ONELINERS_PATH)) + .respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG)) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let pending_dir = tmp.path().join("pending"); + std::fs::create_dir_all(&pending_dir).unwrap(); + let older = Uuid::new_v4(); + let newer = Uuid::new_v4(); + // Create newer first, older second — the disk order should NOT + // determine upload order; `recorded_at` should. + sample_pending(&pending_dir, newer, "2026-04-18T11-00-00Z"); + sample_pending(&pending_dir, older, "2026-04-18T10-00-00Z"); + + let (store, cache) = make_store_and_cache(&tmp).await; + let (sync, mut events) = ServerSync::spawn( + Arc::new(reqwest::Client::new()), + cfg(&server), + store, + cache, + pending_dir, + ); + + let mut ok_order: Vec = Vec::new(); + while ok_order.len() < 2 { + match tokio::time::timeout(Duration::from_millis(500), events.recv()).await { + Ok(Some(UploadEvent::Succeeded { case_id, .. })) => ok_order.push(case_id), + Ok(Some(_)) => continue, + _ => panic!("timed out waiting for two Succeeded events"), + } + } + assert_eq!(ok_order, vec![older, newer], "FIFO by recorded_at"); + + drop(sync); + } + + #[tokio::test] + async fn abort_on_drop_leaves_files_on_disk() { + let server = MockServer::start().await; + // Slow POST so we can abort mid-flight. + Mock::given(method("POST")) + .and(wpath(UPLOAD_PATH)) + .respond_with( + ResponseTemplate::new(200) + .set_delay(Duration::from_secs(5)) + .set_body_json(ack_body(AckStatus::Received)), + ) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(wpath(ONELINERS_PATH)) + .respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG)) + .mount(&server) + .await; + + let tmp = TempDir::new().unwrap(); + let pending_dir = tmp.path().join("pending"); + std::fs::create_dir_all(&pending_dir).unwrap(); + let upload = sample_pending(&pending_dir, Uuid::new_v4(), "2026-04-18T10-00-00Z"); + + let (store, cache) = make_store_and_cache(&tmp).await; + let (sync, _events) = ServerSync::spawn( + Arc::new(reqwest::Client::new()), + cfg(&server), + store, + cache, + pending_dir.clone(), + ); + + // Let the worker enter the slow POST. + tokio::time::sleep(Duration::from_millis(200)).await; + drop(sync); + tokio::time::sleep(Duration::from_millis(100)).await; + + // Both files must still be on disk. + assert!(upload.file.exists(), "m4a must remain after abort"); + assert!( + sidecar_path_for(&upload.file).exists(), + "sidecar must remain after abort" + ); + } +} diff --git a/doctate-client-core/src/upload.rs b/doctate-client-core/src/upload.rs new file mode 100644 index 0000000..eb26293 --- /dev/null +++ b/doctate-client-core/src/upload.rs @@ -0,0 +1,181 @@ +//! Upload I/O building blocks: `PendingUpload` DTO, atomic sidecar write, +//! pending-dir scan. +//! +//! No network logic here — the actual `POST /api/upload` lives in the +//! `server_sync` worker. This module is only about persistence on the +//! client's disk queue: +//! +//! - `{stem}.m4a` — the recorded audio (written by the recorder) +//! - `{stem}.meta.json` — the sidecar with `case_id` + `recorded_at` +//! +//! A pair of both is a ready-to-upload work item; singletons are orphans +//! that `pending_cleanup` reaps. + +use std::path::{Path, PathBuf}; + +use doctate_common::ack::AckStatus; +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use tracing::warn; +use uuid::Uuid; + +/// A recording waiting to be uploaded. The `file` path is derived from the +/// sidecar's on-disk location, not from the JSON content itself — so the +/// sidecar stays valid if the pending directory is moved/renamed. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PendingUpload { + pub case_id: Uuid, + pub recorded_at: String, + #[serde(skip, default)] + pub file: PathBuf, +} + +/// Lifecycle events for a single upload attempt. Emitted by the worker, +/// consumed by the UI. +#[derive(Debug)] +pub enum UploadEvent { + Started(Uuid), + Succeeded { + case_id: Uuid, + status: AckStatus, + }, + Failed { + case_id: Uuid, + reason: String, + will_retry: bool, + }, +} + +#[derive(Debug, Error)] +pub enum UploadIoError { + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + #[error("serialize sidecar: {0}")] + Serialize(#[from] serde_json::Error), +} + +/// Sidecar path next to a `.m4a` file: same directory, same stem, `.meta.json`. +pub fn sidecar_path_for(m4a: &Path) -> PathBuf { + let stem = m4a + .file_stem() + .map(|s| s.to_string_lossy().into_owned()) + .unwrap_or_default(); + let dir = m4a.parent().unwrap_or_else(|| Path::new(".")); + dir.join(format!("{stem}.meta.json")) +} + +/// Persist a pending upload's metadata next to its m4a. Atomic: writes a +/// `*.tmp` sibling and renames it into place. A crash mid-write cannot +/// leave a zero-byte or partial sidecar on disk — `scan_pending_dir` +/// would otherwise skip the pair as an unreadable orphan. +pub fn write_sidecar(upload: &PendingUpload) -> Result<(), UploadIoError> { + let sidecar = sidecar_path_for(&upload.file); + if let Some(parent) = sidecar.parent() { + std::fs::create_dir_all(parent)?; + } + let json = serde_json::to_string_pretty(upload)?; + let tmp = sidecar.with_extension("json.tmp"); + std::fs::write(&tmp, json)?; + std::fs::rename(&tmp, &sidecar)?; + Ok(()) +} + +/// Scan a pending directory for `.m4a` files with matching sidecars. +/// Orphans (no sidecar, unreadable sidecar) are logged and skipped — one +/// broken pair should not stop recovery of the rest. +pub fn scan_pending_dir(pending_dir: &Path) -> Vec { + let entries = match std::fs::read_dir(pending_dir) { + Ok(e) => e, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Vec::new(), + Err(e) => { + warn!(dir = ?pending_dir, error = %e, "scan pending dir failed"); + return Vec::new(); + } + }; + let mut out = Vec::new(); + for entry in entries.flatten() { + let path = entry.path(); + if path.extension().and_then(|s| s.to_str()) != Some("m4a") { + continue; + } + match read_pending(&path) { + Ok(upload) => out.push(upload), + Err(e) => warn!(path = ?path, error = %e, "skipping pending entry"), + } + } + out +} + +fn read_pending(m4a: &Path) -> Result { + let sidecar = sidecar_path_for(m4a); + let json = std::fs::read_to_string(&sidecar)?; + let mut upload: PendingUpload = serde_json::from_str(&json)?; + upload.file = m4a.to_path_buf(); + Ok(upload) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + fn sample(dir: &Path, stem: &str) -> PendingUpload { + let file = dir.join(format!("{stem}.m4a")); + std::fs::write(&file, b"dummy audio").unwrap(); + PendingUpload { + case_id: Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap(), + recorded_at: "2026-04-15T10:32:00Z".into(), + file, + } + } + + #[test] + fn sidecar_roundtrip_via_scan() { + let tmp = TempDir::new().unwrap(); + let upload = sample(tmp.path(), "case_a"); + write_sidecar(&upload).unwrap(); + let found = scan_pending_dir(tmp.path()); + assert_eq!(found.len(), 1); + assert_eq!(found[0].case_id, upload.case_id); + assert_eq!(found[0].recorded_at, upload.recorded_at); + assert_eq!(found[0].file, upload.file); + } + + #[test] + fn scan_ignores_m4a_without_sidecar() { + let tmp = TempDir::new().unwrap(); + std::fs::write(tmp.path().join("orphan.m4a"), b"x").unwrap(); + let found = scan_pending_dir(tmp.path()); + assert!(found.is_empty()); + } + + #[test] + fn scan_returns_empty_for_missing_dir() { + let tmp = TempDir::new().unwrap(); + let missing = tmp.path().join("does-not-exist"); + assert!(scan_pending_dir(&missing).is_empty()); + } + + #[test] + fn write_sidecar_leaves_no_tmp_artefact() { + let tmp = TempDir::new().unwrap(); + let upload = sample(tmp.path(), "case_b"); + write_sidecar(&upload).unwrap(); + + let sidecar = sidecar_path_for(&upload.file); + assert!(sidecar.exists(), "sidecar must be at final path"); + let tmp_path = sidecar.with_extension("json.tmp"); + assert!(!tmp_path.exists(), "tmp artefact must not remain"); + } + + #[test] + fn scan_returns_pairs_in_arbitrary_order_but_all_present() { + let tmp = TempDir::new().unwrap(); + for stem in &["case_1", "case_2", "case_3"] { + let upload = sample(tmp.path(), stem); + write_sidecar(&upload).unwrap(); + } + let found = scan_pending_dir(tmp.path()); + assert_eq!(found.len(), 3); + } +} diff --git a/server/tests/analyze_test.rs b/server/tests/analyze_test.rs index dc51643..82e36b1 100644 --- a/server/tests/analyze_test.rs +++ b/server/tests/analyze_test.rs @@ -648,9 +648,10 @@ async fn undo_delete_restores_latest_batch_only() { // Two separate delete clicks → two distinct batches. let _ = app.clone().oneshot(delete_request(case_a, &cookie)).await.unwrap(); - // Ensure timestamps differ at sub-second resolution (RFC3339 includes - // fractional seconds via the Rfc3339 well-known format). - tokio::time::sleep(Duration::from_millis(20)).await; + // Batch IDs derive from `now_rfc3339()`, which rounds to whole + // seconds — so the sleep must cross a second boundary for the two + // deletes to land in distinct batches. + tokio::time::sleep(Duration::from_millis(1100)).await; let _ = app.clone().oneshot(delete_request(case_b, &cookie)).await.unwrap(); let undo = app