feat(sync): upload queue pump w/ backoff, sync_meeting/status/retry, finalize hook, sync://job events (T9.5)

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<String>,
     pub name: Option<String>,
@@ -1909,8 +1924,27 @@ pub async fn begin_oauth_link(_kind: String) -> WaResult<serde_json::Value> {
 }
 
 #[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<String> {
+    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<u32, WaError> {
+    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<u64, crate::sync::SyncError> {
+    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<String>) -> 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<String>,
+) -> 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<MeetingId>) -> WaResult<Vec<SyncJobInfo>> {
-    Err(not_implemented("sync_status"))
+pub async fn sync_status(
+    state: State<'_, AppState>,
+    meeting_id: Option<MeetingId>,
+) -> WaResult<Vec<SyncJobInfo>> {
+    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")];