diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index 8f77450..bc8937e 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -547,6 +547,22 @@ pub async fn stop_recording( let _ = std::fs::remove_file(&session.wav_path); } + // Sync-on-finalize (T9.5, FR-SYNC-5): enqueue configured artifacts for + // finalize-trigger targets and pump in the background so stop returns + // promptly. Runs after the WAV-retention decision above, so a deleted + // working recording is simply skipped (its file is gone). + if load_settings().sync_enabled { + let store = state.store.clone(); + let app_sync = app.clone(); + let mid = meeting_id.clone(); + tauri::async_runtime::spawn(async move { + if let Err(e) = enqueue_meeting_sync(store.as_ref(), &mid, None, true).await { + tracing::warn!("sync enqueue on finalize failed: {e:?}"); + } + pump_sync(&app_sync, store.as_ref()).await; + }); + } + crate::update_tray_tooltip(&app, "WhispAssist — idle"); let _ = app.emit( "recording://state", @@ -1750,11 +1766,10 @@ pub async fn attach_meeting_to_event( // ---- Sync / upload (Phase 9, ADR-0010) ---- -/// Config payload for add/update. Snake→camel so the TS side stays idiomatic; -/// the `secret` (app password) is write-only — it goes to the OS store, never -/// back to the UI (FR-SYNC-6). +/// Config payload for add/update (snake_case, matching the TS `SyncTargetConfig` +/// and the other serde structs). The `secret` (app password) is write-only — it +/// goes to the OS store, never back to the UI (FR-SYNC-6). #[derive(Deserialize, Default)] -#[serde(rename_all = "camelCase")] pub struct SyncTargetConfigInput { pub id: Option, pub name: Option, @@ -1909,8 +1924,27 @@ pub async fn begin_oauth_link(_kind: String) -> WaResult { } #[tauri::command] -pub async fn retry_sync_job(_job_id: String) -> WaResult<()> { - Err(not_implemented("retry_sync_job")) +pub async fn retry_sync_job( + app: AppHandle, + state: State<'_, AppState>, + job_id: String, +) -> WaResult<()> { + let mut job = state + .store + .get_sync_job(&job_id) + .await + .map_err(|e| WaError::new("sync", e.to_string()))?; + // Clear the backoff so the pump picks it up immediately. + job.status = "pending".to_string(); + job.next_attempt_at = None; + job.updated_at = now_unix(); + state + .store + .update_sync_job(job) + .await + .map_err(|e| WaError::new("sync", e.to_string()))?; + pump_sync(&app, state.store.as_ref()).await; + Ok(()) } /// Provider-specific setup tips surfaced on a failed/attempted test (T9.4). @@ -1990,15 +2024,231 @@ pub async fn set_sync_enabled(enabled: bool) -> WaResult<()> { save_settings(&settings) } +// ---- Sync durable queue (T9.5) ---- + +/// Give up auto-retrying after this many failures; the job stays `failed` with +/// no `next_attempt_at` until the user hits "retry". +const SYNC_MAX_ATTEMPTS: i64 = 6; + +/// Exponential backoff: 30s, 60s, 120s … capped at 1h. +fn sync_backoff_secs(attempts: i64) -> i64 { + (30_i64.saturating_mul(1_i64 << attempts.min(7))).min(3600) +} + +fn file_sha256(path: &Path) -> Option { + use sha2::{Digest, Sha256}; + let bytes = std::fs::read(path).ok()?; + let mut hasher = Sha256::new(); + hasher.update(&bytes); + Some(format!("{:x}", hasher.finalize())) +} + +/// Which (artifact, filename) pairs this target is configured to upload. +fn artifact_plan(target: &SyncTargetRow) -> Vec<(&'static str, &'static str)> { + let mut plan = Vec::new(); + if target.upload_transcript { + plan.push(("transcript", "transcript.json")); + } + if target.upload_notes { + plan.push(("notes", "notes.md")); + } + if target.upload_summary { + plan.push(("summary", "summary.json")); + } + if target.upload_recording { + plan.push(("recording", "audio.wav")); + } + plan +} + +fn job_to_info(row: &crate::storage::SyncJobRow) -> SyncJobInfo { + SyncJobInfo { + id: row.id.clone(), + target_id: row.target_id.clone(), + meeting_id: row.meeting_id.clone(), + artifact: row.artifact.clone(), + status: row.status.clone(), + attempts: row.attempts as u32, + bytes_sent: row.bytes_sent as u64, + bytes_total: row.bytes_total.map(|b| b as u64), + last_error: row.last_error.clone(), + } +} + +fn emit_sync_job(app: &AppHandle, row: &crate::storage::SyncJobRow) { + // camelCase event shape per docs/04-api-contracts.md (`onSyncJob`). + let _ = app.emit( + "sync://job", + serde_json::json!({ + "jobId": row.id, + "meetingId": row.meeting_id, + "targetId": row.target_id, + "artifact": row.artifact, + "status": row.status, + "bytesSent": row.bytes_sent, + "bytesTotal": row.bytes_total, + "attempts": row.attempts, + "error": row.last_error, + }), + ); +} + +/// Enqueue a meeting's configured artifacts for one or all enabled targets +/// (T9.5). `finalize_only` limits to targets with `trigger_on_finalize`. Files +/// that don't exist are skipped; unchanged files are deduped by SHA-256 in the +/// storage upsert. Returns the number of jobs (re)queued. +pub(crate) async fn enqueue_meeting_sync( + store: &dyn crate::storage::Store, + meeting_id: &MeetingId, + target_id: Option<&str>, + finalize_only: bool, +) -> Result { + let targets = store + .list_sync_targets() + .await + .map_err(|e| WaError::new("sync", e.to_string()))?; + let mut queued = 0u32; + for target in targets.iter().filter(|t| t.enabled) { + if target_id.is_some_and(|id| id != target.id) { + continue; + } + if finalize_only && !target.trigger_on_finalize { + continue; + } + let base = target.remote_base_path.trim_end_matches('/'); + for (artifact, filename) in artifact_plan(target) { + let local = meeting_dir(meeting_id).join(filename); + if !local.exists() { + continue; + } + let bytes_total = std::fs::metadata(&local).ok().map(|m| m.len() as i64); + let job = crate::storage::SyncJobRow { + id: uuid::Uuid::new_v4().to_string(), + target_id: target.id.clone(), + meeting_id: meeting_id.clone(), + artifact: artifact.to_string(), + local_path: local.to_string_lossy().into_owned(), + remote_path: format!("{base}/{meeting_id}/{filename}"), + sha256: file_sha256(&local), + status: "pending".to_string(), + attempts: 0, + last_error: None, + next_attempt_at: None, + bytes_total, + bytes_sent: 0, + updated_at: now_unix(), + }; + if store + .upsert_sync_job(job) + .await + .map_err(|e| WaError::new("sync", e.to_string()))? + { + queued += 1; + } + } + } + Ok(queued) +} + +/// Upload one job's file: ensure the remote dir, PUT, report bytes sent. +async fn upload_job( + target: &crate::sync::WebDavTarget, + row: &crate::storage::SyncJobRow, +) -> Result { + use crate::sync::SyncTarget; + let remote_dir = row + .remote_path + .rsplit_once('/') + .map(|(dir, _)| dir) + .unwrap_or(""); + target.ensure_dir(remote_dir).await?; + let (tx, rx) = std::sync::mpsc::channel(); + target + .put(Path::new(&row.local_path), &row.remote_path, tx) + .await?; + let sent = rx + .try_iter() + .last() + .map(|(s, _)| s) + .unwrap_or(row.bytes_total.unwrap_or(0) as u64); + Ok(sent) +} + +/// Drive all due jobs once: upload each, emit `sync://job` on every transition, +/// apply exponential backoff on failure (T9.5). Called on finalize, on startup, +/// and on manual upload/retry — never on an idle timer (NFR-RES-1). +pub(crate) async fn pump_sync(app: &AppHandle, store: &dyn crate::storage::Store) { + let due = match store.claim_due_sync_jobs(now_unix()).await { + Ok(jobs) => jobs, + Err(e) => { + tracing::warn!("sync pump: claim failed: {e}"); + return; + } + }; + for mut job in due { + let target = match store.get_sync_target(&job.target_id).await { + Ok(t) => t, + Err(_) => continue, // target removed under us — its jobs cascaded away + }; + + job.status = "uploading".to_string(); + job.updated_at = now_unix(); + let _ = store.update_sync_job(job.clone()).await; + emit_sync_job(app, &job); + + let webdav = crate::sync::WebDavTarget::from_row(&target); + match upload_job(&webdav, &job).await { + Ok(sent) => { + job.status = "done".to_string(); + job.bytes_sent = sent as i64; + job.last_error = None; + job.next_attempt_at = None; + } + Err(e) => { + job.attempts += 1; + job.last_error = Some(e.to_string()); + job.status = "failed".to_string(); + job.next_attempt_at = (job.attempts < SYNC_MAX_ATTEMPTS) + .then(|| now_unix() + sync_backoff_secs(job.attempts)); + } + } + job.updated_at = now_unix(); + let _ = store.update_sync_job(job.clone()).await; + emit_sync_job(app, &job); + } +} + +/// Manual "Upload now": (re)enqueue the meeting's artifacts and pump immediately. +/// Streams progress via `sync://job` events. #[tauri::command] -pub async fn sync_meeting(_meeting_id: MeetingId, _target_id: Option) -> WaResult<()> { - // Manual "Upload now"; streams progress via "sync://job" events. - Err(not_implemented("sync_meeting")) +pub async fn sync_meeting( + app: AppHandle, + state: State<'_, AppState>, + meeting_id: MeetingId, + target_id: Option, +) -> WaResult<()> { + enqueue_meeting_sync( + state.store.as_ref(), + &meeting_id, + target_id.as_deref(), + false, + ) + .await?; + pump_sync(&app, state.store.as_ref()).await; + Ok(()) } #[tauri::command] -pub async fn sync_status(_meeting_id: Option) -> WaResult> { - Err(not_implemented("sync_status")) +pub async fn sync_status( + state: State<'_, AppState>, + meeting_id: Option, +) -> WaResult> { + let rows = state + .store + .list_sync_jobs(meeting_id.as_ref()) + .await + .map_err(|e| WaError::new("sync", e.to_string()))?; + Ok(rows.iter().map(job_to_info).collect()) } // ---- Feature briefs + MCP server (Phase 10b, ADR-0011) ---- @@ -2184,6 +2434,20 @@ mod tests { } } + #[test] + fn sync_backoff_grows_then_caps_at_one_hour() { + assert_eq!(sync_backoff_secs(0), 30); + assert_eq!(sync_backoff_secs(1), 60); + assert_eq!(sync_backoff_secs(2), 120); + // Monotonic non-decreasing and never above the 1h cap. + let mut prev = 0; + for a in 0..12 { + let b = sync_backoff_secs(a); + assert!(b >= prev && b <= 3600); + prev = b; + } + } + #[test] fn speaker_infos_lists_distinct_speakers_in_first_appearance_order() { let segments = vec![segment("S2"), segment("S1"), segment("S2")];