doctate/client-desktop/src/uploader.rs

//! 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<PendingUpload>,
}

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<UploadEvent>) {
        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<PendingUpload>> {
        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.
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)?;
    std::fs::write(&sidecar, json)?;
    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<PendingUpload> {
    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<PendingUpload, UploaderError> {
    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<PendingUpload>,
    events_tx: mpsc::UnboundedSender<UploadEvent>,
) {
    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<UploadEvent>,
) {
    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)) => {
                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<AckResponse, ErrorKind> {
    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::<AckResponse>()
            .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());
    }

    #[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 { .. })));
    }

    #[tokio::test]
    async fn terminal_401_keeps_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 preserved 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");
    }
}