//! 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::collections::HashSet;
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::{ONELINERS_PATH, OnelinersResponse};
use reqwest::StatusCode;
use tokio::sync::{mpsc, watch};
use tokio::task::AbortHandle;
use tracing::{debug, info, warn};
use uuid::Uuid;

use crate::case_store::CaseStore;
use crate::snapshot_cache::{CachedSnapshot, SnapshotCache};
use crate::upload::{PendingUpload, UploadEvent, scan_pending_dir, sidecar_path_for};

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 of `case_id`s that currently have at least one file sitting
    /// in the pending directory. Shared as an `Arc` so the UI can clone
    /// the borrow cheaply and look up per-case without scanning disk.
    pub pending_case_ids: Arc<HashSet<Uuid>>,
    /// Set to `Some(now)` after any successful upload or poll (200/304).
    /// `None` before the first success. Drives Online/Offline heuristics
    /// together with `last_failure`.
    pub last_success: Option<Instant>,
    /// Set to `Some(now)` after any failed upload or poll. Compared to
    /// `last_success` in the UI: if `last_failure` is younger, the
    /// footer shows Offline regardless of the success's age.
    pub last_failure: Option<Instant>,
}

impl Default for WorkerSnapshot {
    fn default() -> Self {
        Self {
            state: WorkerState::Idle,
            queue_len: 0,
            pending_case_ids: Arc::new(HashSet::new()),
            last_success: None,
            last_failure: 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<WorkerSnapshot>,
    /// Kept in an `Arc<StdMutex<_>>` so tests can peek at `last_success`
    /// independently of the watch channel. Not required for production.
    #[allow(dead_code)]
    last_success_mirror: Arc<StdMutex<Option<Instant>>>,
}

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<reqwest::Client>,
        config: SyncConfig,
        store: Arc<CaseStore>,
        cache: Arc<SnapshotCache>,
        pending_dir: PathBuf,
    ) -> (Self, mpsc::UnboundedReceiver<UploadEvent>) {
        let (kick_tx, kick_rx) = mpsc::unbounded_channel::<()>();
        let (events_tx, events_rx) = mpsc::unbounded_channel::<UploadEvent>();
        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<WorkerSnapshot> {
        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<reqwest::Client>,
    config: SyncConfig,
    store: Arc<CaseStore>,
    cache: Arc<SnapshotCache>,
    pending_dir: PathBuf,
    mut kick_rx: mpsc::UnboundedReceiver<()>,
    events_tx: mpsc::UnboundedSender<UploadEvent>,
    snapshot_tx: watch::Sender<WorkerSnapshot>,
    last_success_mirror: Arc<StdMutex<Option<Instant>>>,
) {
    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<String> = 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<Instant> = None;
    let mut last_failure: Option<Instant> = None;

    let publish = |state: WorkerState,
                   queue_len: usize,
                   pending_ids: Arc<HashSet<Uuid>>,
                   last_ok: Option<Instant>,
                   last_fail: Option<Instant>| {
        let snap = WorkerSnapshot {
            state,
            queue_len,
            pending_case_ids: pending_ids,
            last_success: last_ok,
            last_failure: last_fail,
        };
        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 pending_ids: Arc<HashSet<Uuid>> = Arc::new(files.iter().map(|u| u.case_id).collect());
        let next = files.into_iter().next();

        if let Some(upload) = next {
            publish(
                WorkerState::Uploading,
                queue_len,
                pending_ids.clone(),
                last_success,
                last_failure,
            );
            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");
                    last_failure = Some(Instant::now());
                    let _ = events_tx.send(UploadEvent::Failed {
                        case_id: upload.case_id,
                        reason,
                        will_retry: true,
                    });
                    // Flip back to Idle before the backoff sleep so the UI
                    // doesn't misreport "Uploading" during a passive wait.
                    // The footer derives "Offline" from `last_failure` being
                    // younger than `last_success`.
                    publish(
                        WorkerState::Idle,
                        queue_len,
                        pending_ids.clone(),
                        last_success,
                        last_failure,
                    );
                    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;
                    last_failure = Some(Instant::now());
                    backoff = INITIAL_BACKOFF;
                    let _ = events_tx.send(UploadEvent::Failed {
                        case_id: upload.case_id,
                        reason,
                        will_retry: false,
                    });
                }
            }
        } else {
            publish(
                WorkerState::Polling,
                0,
                pending_ids.clone(),
                last_success,
                last_failure,
            );
            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");
                    last_failure = Some(Instant::now());
                }
                Err(e) => {
                    warn!(error = %e, "poll error");
                    last_failure = Some(Instant::now());
                }
            }

            publish(
                WorkerState::Idle,
                0,
                pending_ids.clone(),
                last_success,
                last_failure,
            );

            // 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::<AckResponse>().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<String>,
) -> Result<PollOutcome, PollError> {
    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?;
            // Merge adds/updates markers the server still knows about;
            // reconcile removes the ones it no longer does (within the
            // window reported by `as_of` / `window_hours`). Running both
            // after every successful poll keeps the client UI in step
            // with server-side deletions without any extra round-trips.
            store.merge_server_snapshot(&body).await?;
            store.reconcile_with_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<String> {
    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 time::format_description::well_known::Rfc3339;
    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<CaseStore>, Arc<SnapshotCache>) {
        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(_)
        ));
    }

    /// End-to-end reconcile: a synced marker that the server omits from
    /// its next snapshot must vanish locally after a single poll. This
    /// is the scenario the user reported ("deleted cases still show up
    /// on the client") — the wiring is: `poll_once` → `merge` → `reconcile`.
    #[tokio::test]
    async fn poll_once_reconciles_missing_synced_marker() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(wpath(ONELINERS_PATH))
            .respond_with(
                ResponseTemplate::new(200)
                    .insert_header("etag", TEST_ETAG)
                    // Authoritative snapshot with zero entries — any
                    // synced local marker inside the window is deleted
                    // server-side.
                    .set_body_json(OnelinersResponse {
                        as_of: "2026-04-18T10:00:00Z".into(),
                        window_hours: 72,
                        oneliners: vec![],
                    }),
            )
            .mount(&server)
            .await;

        let tmp = TempDir::new().unwrap();
        let (store, cache) = make_store_and_cache(&tmp).await;

        // Seed a synced marker inside the window.
        let stale_id = Uuid::new_v4();
        store
            .create_local(
                stale_id,
                time::OffsetDateTime::parse("2026-04-18T09:00:00Z", &Rfc3339).unwrap(),
            )
            .await
            .unwrap();
        store.mark_synced(stale_id).await.unwrap();
        assert_eq!(store.list().await.len(), 1);

        // Poll once — the merge produces no add/update, reconcile drops
        // the stale marker.
        let http = reqwest::Client::new();
        let mut etag: Option<String> = None;
        let outcome = poll_once(&http, &cfg(&server), &store, &cache, &mut etag)
            .await
            .unwrap();
        assert_eq!(outcome, PollOutcome::Success);
        assert!(
            store.list().await.is_empty(),
            "reconcile must remove the synced marker the server no longer lists"
        );
    }

    #[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<String> = 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<tokio::sync::Mutex<Vec<Uuid>>> =
            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<Uuid> = 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"
        );
    }
}