use crate::workspace::{
    cache_key, chmod_safe, file_hash, resolve_workspace_file, HashCache, READONLY_MODE,
    WRITABLE_MODE,
};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::io::AsyncWriteExt;

/// Files at or below this size ride archive batches; larger ones stream
/// through their own GET (spec: Transfer v2).
pub const ARCHIVE_MAX_FILE_BYTES: u64 = 32 * 1024 * 1024;
pub const ARCHIVE_MAX_BATCH_FILES: usize = 500;
pub const ARCHIVE_MAX_BATCH_BYTES: u64 = 256 * 1024 * 1024;
/// Budget charged against a batch for a file whose size the caller didn't know.
const UNKNOWN_SIZE_BUDGET: u64 = 1024 * 1024;
/// Concurrent transfer streams (archive batches and large files alike).
const TRANSFER_CONCURRENCY: usize = 4;
/// Emit a mid-file progress event at least every this many bytes.
const PROGRESS_BYTE_STRIDE: u64 = 8 * 1024 * 1024;

pub fn build_headers(map: &HashMap<String, String>) -> reqwest::header::HeaderMap {
    let mut headers = reqwest::header::HeaderMap::new();
    for (k, v) in map {
        if let (Ok(name), Ok(value)) = (
            reqwest::header::HeaderName::from_bytes(k.as_bytes()),
            reqwest::header::HeaderValue::from_str(v),
        ) {
            headers.insert(name, value);
        }
    }
    headers
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TransferProgress {
    pub kind: String,
    pub path: String,
    pub done: u32,
    pub total: u32,
    pub bytes_done: u64,
    pub bytes_total: u64,
}

/// Progress consumer shared across the parallel transfer tasks. The Tauri
/// commands wrap an IPC channel in one of these; tests pass a capturing
/// closure or `None`.
pub type ProgressSink = Option<Arc<dyn Fn(TransferProgress) + Send + Sync>>;

struct ProgressState {
    kind: &'static str,
    total: u32,
    bytes_total: u64,
    // Counter updates and their emission happen under one lock so events
    // leave in counter order — separate atomics let two tasks emit a
    // higher-done event before a lower one and the UI counter would tick
    // backwards. Uncontended in practice (one event per file / per 8 MiB).
    counters: std::sync::Mutex<(u32, u64)>,
    sink: ProgressSink,
}

impl ProgressState {
    fn new(kind: &'static str, total: u32, bytes_total: u64, sink: ProgressSink) -> Arc<Self> {
        Arc::new(Self {
            kind,
            total,
            bytes_total,
            counters: std::sync::Mutex::new((0, 0)),
            sink,
        })
    }

    fn update(&self, path: &str, done_inc: u32, bytes_inc: u64) {
        let mut guard = self.counters.lock().unwrap();
        guard.0 += done_inc;
        guard.1 += bytes_inc;
        let (done, bytes_done) = *guard;
        if let Some(sink) = &self.sink {
            sink(TransferProgress {
                kind: self.kind.to_string(),
                path: path.to_string(),
                done,
                total: self.total,
                bytes_done,
                bytes_total: self.bytes_total,
            });
        }
    }

    /// Bytes streamed inside a still-in-flight file (large-file heartbeat).
    fn stream_bytes(&self, path: &str, delta: u64) {
        self.update(path, 0, delta);
    }

    /// A file finished processing (landed or errored); `tail_bytes` are the
    /// bytes it transferred since its last `stream_bytes` emission.
    fn file_done(&self, path: &str, tail_bytes: u64) {
        self.update(path, 1, tail_bytes);
    }
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PullFile {
    pub path: String,
    pub url: String,
    #[serde(default)]
    pub hash: Option<String>,
    #[serde(default)]
    pub size_bytes: Option<u64>,
}

#[derive(Serialize, Debug)]
pub struct FileError {
    pub path: String,
    pub error: String,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PullResult {
    pub ok: bool,
    pub changed: Vec<String>,
    pub created: Vec<String>,
    pub dirs_created: Vec<String>,
    pub errors: Vec<FileError>,
}

/// One file that survived the prep phase and needs downloading.
struct PullWork {
    index: usize,
    path: String,
    url: String,
    expect_hash: Option<String>,
    size_bytes: Option<u64>,
    full: PathBuf,
    existed: bool,
}

/// A finished download attempt. `result` carries the hash-cache insertion
/// (key, hash) so the coordinator — which owns the cache reference — can
/// record it; spawned tasks never touch the cache directly.
struct FileDone {
    index: usize,
    path: String,
    created: bool,
    result: Result<(String, String), String>,
}

/// How a pull's work list splits into archive batches and individual GETs
/// (indices into the work list). Pure so the thresholds are unit-testable.
pub(crate) struct PullPlan {
    pub batches: Vec<Vec<usize>>,
    pub individual: Vec<usize>,
}

pub(crate) fn plan_pull(sizes: &[Option<u64>], batching_available: bool) -> PullPlan {
    let mut plan = PullPlan { batches: Vec::new(), individual: Vec::new() };
    let mut current: Vec<usize> = Vec::new();
    let mut current_bytes: u64 = 0;
    for (i, size) in sizes.iter().enumerate() {
        if !batching_available || size.is_some_and(|s| s > ARCHIVE_MAX_FILE_BYTES) {
            plan.individual.push(i);
            continue;
        }
        let cost = size.unwrap_or(UNKNOWN_SIZE_BUDGET);
        if !current.is_empty()
            && (current.len() >= ARCHIVE_MAX_BATCH_FILES
                || current_bytes + cost > ARCHIVE_MAX_BATCH_BYTES)
        {
            plan.batches.push(std::mem::take(&mut current));
            current_bytes = 0;
        }
        current.push(i);
        current_bytes += cost;
    }
    if !current.is_empty() {
        plan.batches.push(current);
    }
    plan
}

fn temp_path(full: &Path) -> PathBuf {
    full.with_file_name(format!(
        "{}.rc-download",
        full.file_name().unwrap_or_default().to_string_lossy()
    ))
}

/// Shared landing tail for both download paths: swap the verified temp into
/// place, set the writable bit, and return the hash-cache entry for the
/// final file's path+size+mtime.
async fn finalize_pulled_file(
    tmp: &Path,
    full: &Path,
    existed: bool,
    writable: bool,
    hash: &str,
) -> Result<(String, String), String> {
    if existed {
        chmod_safe(full, WRITABLE_MODE).await; // unlock read-only before overwrite
    }
    if let Err(e) = tokio::fs::rename(tmp, full).await {
        return Err(e.to_string());
    }
    chmod_safe(full, if writable { WRITABLE_MODE } else { READONLY_MODE }).await;
    let meta = tokio::fs::metadata(full).await.map_err(|e| e.to_string())?;
    Ok((cache_key(full, &meta), hash.to_string()))
}

const INTERRUPTED: &str = "archive stream interrupted";

#[derive(Deserialize)]
struct ArchiveHeader {
    path: String,
    size: u64,
    hash: String,
}

/// Incremental reader over an archive response body: buffers chunks and
/// hands out exact byte counts so frame parsing never over-reads.
struct FrameReader {
    resp: reqwest::Response,
    buf: Vec<u8>,
}

impl FrameReader {
    /// Pull one more chunk into the buffer. Ok(false) = clean EOF.
    async fn fill(&mut self) -> Result<bool, String> {
        match self.resp.chunk().await.map_err(|_| INTERRUPTED.to_string())? {
            Some(chunk) => {
                self.buf.extend_from_slice(&chunk);
                Ok(true)
            }
            None => Ok(false),
        }
    }

    async fn read_exact(&mut self, n: usize) -> Result<Vec<u8>, String> {
        while self.buf.len() < n {
            if !self.fill().await? {
                return Err(INTERRUPTED.to_string());
            }
        }
        let rest = self.buf.split_off(n);
        Ok(std::mem::replace(&mut self.buf, rest))
    }

    /// At least one byte, at most `max`, from the buffer (refilling on empty).
    async fn read_up_to(&mut self, max: usize) -> Result<Vec<u8>, String> {
        while self.buf.is_empty() {
            if !self.fill().await? {
                return Err(INTERRUPTED.to_string());
            }
        }
        let n = self.buf.len().min(max);
        let rest = self.buf.split_off(n);
        Ok(std::mem::replace(&mut self.buf, rest))
    }
}

/// Stream one framed file body to `tmp`, hashing while writing. Returns the
/// hex hash and the bytes transferred since the last progress emission.
async fn receive_archive_file(
    reader: &mut FrameReader,
    tmp: &Path,
    size: u64,
    progress: &ProgressState,
    path: &str,
) -> Result<(String, u64), String> {
    // Buffered: every unbuffered write_all round-trips tokio's blocking pool,
    // which at network-chunk granularity dominates the whole transfer.
    let file = tokio::fs::File::create(tmp).await.map_err(|e| e.to_string())?;
    let mut out = tokio::io::BufWriter::with_capacity(1024 * 1024, file);
    let mut hasher = Sha256::new();
    let mut remaining = size;
    let mut unreported: u64 = 0;
    while remaining > 0 {
        let max = usize::try_from(remaining).unwrap_or(usize::MAX);
        let chunk = reader.read_up_to(max).await?;
        hasher.update(&chunk);
        out.write_all(&chunk).await.map_err(|e| e.to_string())?;
        remaining -= chunk.len() as u64;
        unreported += chunk.len() as u64;
        if unreported >= PROGRESS_BYTE_STRIDE && remaining > 0 {
            progress.stream_bytes(path, unreported);
            unreported = 0;
        }
    }
    out.flush().await.map_err(|e| e.to_string())?;
    Ok((hex::encode(hasher.finalize()), unreported))
}

/// POST one archive batch and land every framed file. A broken stream marks
/// the current and all not-yet-received files as errors — the frame boundary
/// is lost, so nothing after the break is salvageable.
async fn run_archive_batch(
    client: reqwest::Client,
    archive_url: String,
    header_map: reqwest::header::HeaderMap,
    batch: Vec<PullWork>,
    writable: bool,
    progress: Arc<ProgressState>,
) -> Vec<FileDone> {
    let paths: Vec<String> = batch.iter().map(|w| w.path.clone()).collect();
    let mut by_path: HashMap<String, PullWork> =
        batch.into_iter().map(|w| (w.path.clone(), w)).collect();
    let expected = by_path.len();
    let mut dones = Vec::new();

    let fail_all = |by_path: &mut HashMap<String, PullWork>, error: String| -> Vec<FileDone> {
        by_path
            .drain()
            .map(|(_, w)| FileDone {
                index: w.index,
                path: w.path,
                created: !w.existed,
                result: Err(error.clone()),
            })
            .collect()
    };

    let body = match serde_json::to_string(&serde_json::json!({ "paths": paths })) {
        Ok(b) => b,
        Err(e) => return fail_all(&mut by_path, e.to_string()),
    };
    let resp = match client
        .post(&archive_url)
        .headers(header_map)
        .header(reqwest::header::CONTENT_TYPE, "application/json")
        .body(body)
        .send()
        .await
    {
        Ok(r) => r,
        Err(e) => return fail_all(&mut by_path, e.to_string()),
    };
    let status = resp.status().as_u16();
    if status != 200 {
        return fail_all(&mut by_path, format!("Server error ({status})."));
    }

    let mut reader = FrameReader { resp, buf: Vec::new() };
    let stream_result: Result<(), String> = async {
        for _ in 0..expected {
            let len_bytes = reader.read_exact(8).await?;
            let header_len = u64::from_le_bytes(len_bytes.try_into().unwrap());
            if header_len == 0 || header_len > 1_000_000 {
                return Err("malformed archive frame".into());
            }
            let header_bytes = reader.read_exact(header_len as usize).await?;
            let header: ArchiveHeader = serde_json::from_slice(&header_bytes)
                .map_err(|_| "malformed archive frame".to_string())?;
            let Some(work) = by_path.remove(&header.path) else {
                return Err("malformed archive frame".into());
            };
            let tmp = temp_path(&work.full);
            let (hash, tail) =
                match receive_archive_file(&mut reader, &tmp, header.size, &progress, &work.path)
                    .await
                {
                    Ok(v) => v,
                    Err(e) => {
                        let _ = tokio::fs::remove_file(&tmp).await;
                        // Re-file it so the interrupted sweep below covers it.
                        by_path.insert(header.path, work);
                        return Err(e);
                    }
                };
            let result = if hash != header.hash {
                let _ = tokio::fs::remove_file(&tmp).await;
                Err("Downloaded content failed hash verification.".to_string())
            } else {
                let r = finalize_pulled_file(&tmp, &work.full, work.existed, writable, &hash).await;
                if r.is_err() {
                    let _ = tokio::fs::remove_file(&tmp).await;
                }
                r
            };
            progress.file_done(&work.path, tail);
            dones.push(FileDone {
                index: work.index,
                path: work.path,
                created: !work.existed,
                result,
            });
        }
        Ok(())
    }
    .await;

    if let Err(e) = stream_result {
        dones.extend(fail_all(&mut by_path, e));
    }
    dones
}

/// GET one (large) file with hash-while-write, ≥8 MiB progress heartbeats,
/// verification against the expected depot hash, and the shared landing tail.
async fn download_single(
    client: reqwest::Client,
    header_map: reqwest::header::HeaderMap,
    work: PullWork,
    writable: bool,
    progress: Arc<ProgressState>,
) -> FileDone {
    let tmp = temp_path(&work.full);
    let mut tail: u64 = 0;
    let result = async {
        let mut resp = client
            .get(&work.url)
            .headers(header_map)
            .send()
            .await
            .map_err(|e| e.to_string())?;
        let status = resp.status().as_u16();
        if status != 200 {
            return Err(format!("Server error ({status})."));
        }
        let file = tokio::fs::File::create(&tmp).await.map_err(|e| e.to_string())?;
        let mut out = tokio::io::BufWriter::with_capacity(1024 * 1024, file);
        let mut hasher = Sha256::new();
        while let Some(chunk) = resp.chunk().await.map_err(|e| e.to_string())? {
            hasher.update(&chunk);
            out.write_all(&chunk).await.map_err(|e| e.to_string())?;
            tail += chunk.len() as u64;
            if tail >= PROGRESS_BYTE_STRIDE {
                progress.stream_bytes(&work.path, tail);
                tail = 0;
            }
        }
        out.flush().await.map_err(|e| e.to_string())?;
        drop(out);
        let hash = hex::encode(hasher.finalize());
        if let Some(expect) = &work.expect_hash {
            if expect != &hash {
                return Err("Downloaded content failed hash verification.".to_string());
            }
        }
        finalize_pulled_file(&tmp, &work.full, work.existed, writable, &hash).await
    }
    .await;
    if result.is_err() {
        let _ = tokio::fs::remove_file(&tmp).await;
    }
    progress.file_done(&work.path, tail);
    FileDone { index: work.index, path: work.path, created: !work.existed, result }
}

pub const PULL_IN_PROGRESS: &str = "A pull is already running for this workspace.";

/// One pull per workspace folder, process-wide. Temp names are deterministic
/// (`<path>.rc-download`), so two overlapping pulls of the same folder write
/// the same temp files — and after the winner's rename, the loser's still-open
/// handle keeps writing into the RENAMED FINAL FILE. Stream-hash verification
/// can't catch that (it hashes the network bytes, not the landed file), so the
/// file lands corrupted and scans as "edited". The UI's busy flag can't be the
/// only guard: it lives in component state that resets when the panel
/// remounts (page swaps) while the engine keeps running.
static ACTIVE_PULLS: std::sync::OnceLock<std::sync::Mutex<HashSet<PathBuf>>> =
    std::sync::OnceLock::new();

struct PullClaim(PathBuf);

impl Drop for PullClaim {
    fn drop(&mut self) {
        if let Some(set) = ACTIVE_PULLS.get() {
            set.lock().unwrap().remove(&self.0);
        }
    }
}

fn claim_pull(folder: &Path) -> Option<PullClaim> {
    // Canonicalized so `/ws` and `/ws/../ws` count as the same workspace.
    let key = folder.canonicalize().unwrap_or_else(|_| folder.to_path_buf());
    let set = ACTIVE_PULLS.get_or_init(|| std::sync::Mutex::new(HashSet::new()));
    let mut guard = set.lock().unwrap();
    if !guard.insert(key.clone()) {
        return None;
    }
    Some(PullClaim(key))
}

/// Skip up-to-date files by hash, then land the rest through 4-way-parallel
/// archive batches (small files) and individual streaming GETs (large files):
/// download to `<path>.rc-download`, hash while writing, verify, rename,
/// record the hash in the shared cache, and report progress through `sink`.
pub async fn pull_files_impl(
    folder_path: &str,
    files: &[PullFile],
    headers: &HashMap<String, String>,
    writable: bool,
    archive_url: Option<&str>,
    cache: &HashCache,
    sink: ProgressSink,
    pin: Option<&str>,
) -> PullResult {
    let root = Path::new(folder_path);
    let Some(_claim) = claim_pull(root) else {
        return PullResult {
            ok: false,
            changed: vec![],
            created: vec![],
            dirs_created: vec![],
            errors: vec![FileError { path: String::new(), error: PULL_IN_PROGRESS.into() }],
        };
    };
    let header_map = build_headers(headers);
    // Every file/archive URL in one pull shares the server's origin, so any
    // representative URL carries the right scheme for the pinned-HTTPS check
    // (the pin verifier itself is host-agnostic — see pinned_http::build_client).
    let representative_url = archive_url
        .or_else(|| files.first().map(|f| f.url.as_str()))
        .unwrap_or("");
    let client = match crate::pinned_http::build_client(representative_url, pin, header_map.clone())
    {
        Ok(c) => c,
        Err(e) => {
            return PullResult {
                ok: false,
                changed: vec![],
                created: vec![],
                dirs_created: vec![],
                errors: vec![FileError { path: String::new(), error: e }],
            };
        }
    };
    let mut dirs_created = Vec::new();
    let mut prep_errors: HashMap<usize, FileError> = HashMap::new();
    let mut work: Vec<PullWork> = Vec::new();

    // Prep phase stays sequential: the up-to-date skip reads the hash cache,
    // and folder creation order determines the dirs_created report.
    for (index, file) in files.iter().enumerate() {
        let Some(full) = resolve_workspace_file(root, &file.path) else {
            prep_errors.insert(
                index,
                FileError { path: file.path.clone(), error: "Invalid path.".into() },
            );
            continue;
        };
        let existing = tokio::fs::metadata(&full).await.ok();
        if let (Some(meta), Some(expect)) = (&existing, &file.hash) {
            if let Ok(actual) = file_hash(cache, &full, meta).await {
                if &actual == expect {
                    continue; // up to date
                }
            }
        }

        // Create missing parent folders, remembering which ones are new.
        if let Some(dir) = full.parent() {
            if tokio::fs::metadata(dir).await.is_err() {
                let rel = file.path.replace('\\', "/");
                let mut probe = Path::new(&rel).parent();
                let mut missing = Vec::new();
                while let Some(p) = probe {
                    if p.as_os_str().is_empty() {
                        break;
                    }
                    let rel_str = p.to_string_lossy().to_string();
                    if let Some(abs) = resolve_workspace_file(root, &rel_str) {
                        if tokio::fs::metadata(&abs).await.is_err() {
                            missing.push(rel_str);
                        }
                    }
                    probe = p.parent();
                }
                if tokio::fs::create_dir_all(dir).await.is_err() {
                    prep_errors.insert(
                        index,
                        FileError {
                            path: file.path.clone(),
                            error: "Could not create folder.".into(),
                        },
                    );
                    continue;
                }
                dirs_created.extend(missing);
            }
        }

        work.push(PullWork {
            index,
            path: file.path.clone(),
            url: file.url.clone(),
            expect_hash: file.hash.clone(),
            size_bytes: file.size_bytes,
            full,
            existed: existing.is_some(),
        });
    }

    let bytes_total: u64 = work.iter().map(|w| w.size_bytes.unwrap_or(0)).sum();
    let progress = ProgressState::new("pull", work.len() as u32, bytes_total, sink);
    let plan = plan_pull(
        &work.iter().map(|w| w.size_bytes).collect::<Vec<_>>(),
        archive_url.is_some(),
    );

    let work_indices: HashSet<usize> = work.iter().map(|w| w.index).collect();
    let mut slots: Vec<Option<PullWork>> = work.into_iter().map(Some).collect();
    let sem = Arc::new(tokio::sync::Semaphore::new(TRANSFER_CONCURRENCY));
    let mut set = tokio::task::JoinSet::new();

    for batch_indices in plan.batches {
        let batch: Vec<PullWork> =
            batch_indices.iter().map(|&i| slots[i].take().unwrap()).collect();
        let (client, hm, progress, sem) =
            (client.clone(), header_map.clone(), progress.clone(), sem.clone());
        let url = archive_url.expect("batches exist only with an archive url").to_string();
        set.spawn(async move {
            let _permit = sem.acquire_owned().await;
            run_archive_batch(client, url, hm, batch, writable, progress).await
        });
    }
    for i in plan.individual {
        let w = slots[i].take().unwrap();
        let (client, hm, progress, sem) =
            (client.clone(), header_map.clone(), progress.clone(), sem.clone());
        set.spawn(async move {
            let _permit = sem.acquire_owned().await;
            vec![download_single(client, hm, w, writable, progress).await]
        });
    }

    let mut outcomes: HashMap<usize, FileDone> = HashMap::new();
    while let Some(res) = set.join_next().await {
        let Ok(dones) = res else { continue };
        let mut guard = cache.lock().await;
        for d in dones {
            if let Ok((key, hash)) = &d.result {
                guard.entries.insert(key.clone(), hash.clone());
            }
            outcomes.insert(d.index, d);
        }
    }

    // Assemble results in input order, matching the sequential engine's shape.
    let mut changed = Vec::new();
    let mut created = Vec::new();
    let mut errors = Vec::new();
    for (index, file) in files.iter().enumerate() {
        if let Some(err) = prep_errors.remove(&index) {
            errors.push(err);
            continue;
        }
        match outcomes.remove(&index) {
            Some(done) => match done.result {
                Ok(_) => {
                    changed.push(done.path.clone());
                    if done.created {
                        created.push(done.path);
                    }
                }
                Err(error) => errors.push(FileError { path: done.path, error }),
            },
            // Only reachable if a transfer task panicked.
            None if work_indices.contains(&index) => errors.push(FileError {
                path: file.path.clone(),
                error: "Transfer task failed.".into(),
            }),
            None => {} // skipped as up to date
        }
    }

    PullResult { ok: errors.is_empty(), changed, created, dirs_created, errors }
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PullPayload {
    pub folder_path: String,
    pub files: Vec<PullFile>,
    #[serde(default)]
    pub headers: HashMap<String, String>,
    #[serde(default)]
    pub writable: bool,
    #[serde(default)]
    pub archive_url: Option<String>,
}

// Return an early `{ ok:false, error:'Bad pull request.', changed:[],... }`
// for a missing/blank folderPath. PullResult has no error field, but the
// renderer only checks ok/errors, so an empty-errors ok:false is equivalent.
// Split out so the guard is testable without a live tauri::State.
pub fn blank_pull_guard(folder_path: &str) -> Option<PullResult> {
    if folder_path.trim().is_empty() {
        Some(PullResult {
            ok: false,
            changed: vec![],
            created: vec![],
            dirs_created: vec![],
            errors: vec![],
        })
    } else {
        None
    }
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UploadFile {
    pub path: String,
    pub url: String,
    #[serde(default)]
    pub size_bytes: Option<u64>,
}

#[derive(Serialize, Debug)]
#[serde(rename_all = "camelCase")]
pub struct UploadOutcome {
    pub path: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub hash: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub size_bytes: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

#[derive(Serialize)]
pub struct UploadResult {
    pub ok: bool,
    pub results: Vec<UploadOutcome>,
}

/// POST one workspace file as a raw body with an explicit Content-Length
/// (the stdlib Python server can't decode chunked encoding), streaming from
/// disk so large game assets never sit in memory. Returns the outcome and
/// the bytes actually shipped (for progress accounting).
async fn upload_single(
    client: reqwest::Client,
    header_map: reqwest::header::HeaderMap,
    full: Option<PathBuf>,
    path: String,
    url: String,
) -> (UploadOutcome, u64) {
    let meta = match &full {
        Some(p) => tokio::fs::metadata(p).await.ok(),
        None => None,
    };
    let (Some(full), Some(meta)) = (full, meta) else {
        return (
            UploadOutcome {
                path,
                hash: None,
                size_bytes: None,
                error: Some("File not found in workspace.".into()),
            },
            0,
        );
    };
    let size = meta.len();

    let outcome = async {
        let fh = tokio::fs::File::open(&full).await.map_err(|e| e.to_string())?;
        let stream = tokio_util::io::ReaderStream::new(fh);
        let resp = client
            .post(&url)
            .headers(header_map)
            .header(reqwest::header::CONTENT_TYPE, "application/octet-stream")
            .header(reqwest::header::CONTENT_LENGTH, size)
            .body(reqwest::Body::wrap_stream(stream))
            .send()
            .await
            .map_err(|e| e.to_string())?;
        let status = resp.status().as_u16();
        let raw = resp.text().await.unwrap_or_default();
        Ok::<(u16, String), String>((status, raw))
    }
    .await;

    match outcome {
        Ok((status, raw)) => {
            let parsed: Option<serde_json::Value> = serde_json::from_str(&raw).ok();
            match parsed {
                Some(body) => {
                    if let Some(hash) = body.get("hash").and_then(|h| h.as_str()) {
                        (
                            UploadOutcome {
                                path,
                                hash: Some(hash.to_string()),
                                size_bytes: Some(size),
                                error: None,
                            },
                            size,
                        )
                    } else {
                        let err = body
                            .get("error")
                            .and_then(|e| e.as_str())
                            .map(String::from)
                            .unwrap_or_else(|| "Upload failed.".into());
                        (UploadOutcome { path, hash: None, size_bytes: None, error: Some(err) }, 0)
                    }
                }
                None => (
                    UploadOutcome {
                        path,
                        hash: None,
                        size_bytes: None,
                        error: Some(format!("Server error ({status}).")),
                    },
                    0,
                ),
            }
        }
        Err(msg) => (UploadOutcome { path, hash: None, size_bytes: None, error: Some(msg) }, 0),
    }
}

/// 4-way-parallel uploads; results come back in input order regardless of
/// completion order.
pub async fn upload_files_impl(
    folder_path: &str,
    files: &[UploadFile],
    headers: &HashMap<String, String>,
    sink: ProgressSink,
    pin: Option<&str>,
) -> UploadResult {
    let root = Path::new(folder_path);
    let header_map = build_headers(headers);
    // Every upload target shares the server's origin; see the matching
    // comment in pull_files_impl for why any one file's URL is representative.
    let representative_url = files.first().map(|f| f.url.as_str()).unwrap_or("");
    let client = match crate::pinned_http::build_client(representative_url, pin, header_map.clone())
    {
        Ok(c) => c,
        Err(e) => {
            return UploadResult {
                ok: false,
                results: files
                    .iter()
                    .map(|f| UploadOutcome {
                        path: f.path.clone(),
                        hash: None,
                        size_bytes: None,
                        error: Some(e.clone()),
                    })
                    .collect(),
            };
        }
    };
    let bytes_total: u64 = files.iter().map(|f| f.size_bytes.unwrap_or(0)).sum();
    let progress = ProgressState::new("upload", files.len() as u32, bytes_total, sink);
    let sem = Arc::new(tokio::sync::Semaphore::new(TRANSFER_CONCURRENCY));
    let mut set = tokio::task::JoinSet::new();

    for (index, file) in files.iter().enumerate() {
        let full = resolve_workspace_file(root, &file.path);
        let (path, url) = (file.path.clone(), file.url.clone());
        let (client, hm, progress, sem) =
            (client.clone(), header_map.clone(), progress.clone(), sem.clone());
        set.spawn(async move {
            let _permit = sem.acquire_owned().await;
            let (outcome, bytes) = upload_single(client, hm, full, path, url).await;
            progress.file_done(&outcome.path, bytes);
            (index, outcome)
        });
    }

    let mut slots: Vec<Option<UploadOutcome>> = (0..files.len()).map(|_| None).collect();
    while let Some(res) = set.join_next().await {
        if let Ok((index, outcome)) = res {
            slots[index] = Some(outcome);
        }
    }
    let results: Vec<UploadOutcome> = slots
        .into_iter()
        .enumerate()
        .map(|(i, slot)| {
            slot.unwrap_or_else(|| UploadOutcome {
                path: files[i].path.clone(),
                hash: None,
                size_bytes: None,
                error: Some("Transfer task failed.".into()),
            })
        })
        .collect();

    UploadResult { ok: results.iter().all(|r| r.error.is_none()), results }
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UploadPayload {
    pub folder_path: String,
    pub files: Vec<UploadFile>,
    #[serde(default)]
    pub headers: HashMap<String, String>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::workspace::HashCache;
    use std::collections::HashMap;
    use std::io::{Read, Write};
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Mutex;

    const MIB: u64 = 1024 * 1024;

    fn sha_hex(data: &[u8]) -> String {
        hex::encode(Sha256::digest(data))
    }

    /// Build the server-side archive framing for the given files.
    fn frame_archive(files: &[(&str, &[u8])]) -> Vec<u8> {
        let mut out = Vec::new();
        for (path, content) in files {
            let header = serde_json::json!({
                "path": path, "size": content.len(), "hash": sha_hex(content)
            })
            .to_string();
            out.extend_from_slice(&(header.len() as u64).to_le_bytes());
            out.extend_from_slice(header.as_bytes());
            out.extend_from_slice(content);
        }
        out
    }

    fn capture_sink() -> (Arc<Mutex<Vec<TransferProgress>>>, ProgressSink) {
        let events: Arc<Mutex<Vec<TransferProgress>>> = Arc::new(Mutex::new(Vec::new()));
        let sink = {
            let events = events.clone();
            Some(Arc::new(move |p: TransferProgress| {
                events.lock().unwrap().push(p);
            }) as Arc<dyn Fn(TransferProgress) + Send + Sync>)
        };
        (events, sink)
    }

    /// Serve fixed bodies from a background tiny_http server; returns base URL.
    fn spawn_server(routes: Vec<(&'static str, u16, Vec<u8>)>) -> String {
        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
        let port = server.server_addr().to_ip().unwrap().port();
        std::thread::spawn(move || {
            for req in server.incoming_requests() {
                let url = req.url().to_string();
                let hit = routes.iter().find(|(p, _, _)| url == *p);
                match hit {
                    Some((_, status, body)) => {
                        let resp = tiny_http::Response::from_data(body.clone())
                            .with_status_code(*status);
                        let _ = req.respond(resp);
                    }
                    None => {
                        let _ = req.respond(tiny_http::Response::empty(404));
                    }
                }
            }
        });
        format!("http://127.0.0.1:{port}")
    }

    /// Archive server: answers POST /api/files/archive with the framing for
    /// the requested paths (in request order) from `contents`, counting
    /// requests. Unknown routes 404.
    fn spawn_archive_server(
        contents: Vec<(&'static str, Vec<u8>)>,
        requests: Arc<AtomicUsize>,
    ) -> String {
        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
        let port = server.server_addr().to_ip().unwrap().port();
        std::thread::spawn(move || {
            for mut req in server.incoming_requests() {
                if req.url() != "/api/files/archive" {
                    let _ = req.respond(tiny_http::Response::empty(404));
                    continue;
                }
                requests.fetch_add(1, Ordering::SeqCst);
                let mut body = String::new();
                req.as_reader().read_to_string(&mut body).unwrap();
                let parsed: serde_json::Value = serde_json::from_str(&body).unwrap();
                let paths: Vec<&str> = parsed["paths"]
                    .as_array()
                    .unwrap()
                    .iter()
                    .map(|p| p.as_str().unwrap())
                    .collect();
                let files: Vec<(&str, &[u8])> = paths
                    .iter()
                    .map(|p| {
                        let (name, data) =
                            contents.iter().find(|(name, _)| name == p).unwrap();
                        (*name, data.as_slice())
                    })
                    .collect();
                let _ = req.respond(tiny_http::Response::from_data(frame_archive(&files)));
            }
        });
        format!("http://127.0.0.1:{port}")
    }

    fn pull_file(path: &str, url: String, content: &[u8]) -> PullFile {
        PullFile {
            path: path.into(),
            url,
            hash: Some(sha_hex(content)),
            size_bytes: Some(content.len() as u64),
        }
    }

    #[test]
    fn plan_pull_partitions_by_size_count_and_budget() {
        // No archive URL → everything individual.
        let plan = plan_pull(&[Some(10), None], false);
        assert!(plan.batches.is_empty());
        assert_eq!(plan.individual, vec![0, 1]);

        // Size boundary: ≤ 32 MiB (and unknown) batch, above goes individual.
        let plan =
            plan_pull(&[Some(32 * MIB), Some(32 * MIB + 1), None, Some(1)], true);
        assert_eq!(plan.batches, vec![vec![0, 2, 3]]);
        assert_eq!(plan.individual, vec![1]);

        // Count split: 501 tiny files → 500 + 1.
        let sizes: Vec<Option<u64>> = (0..501).map(|_| Some(1)).collect();
        let plan = plan_pull(&sizes, true);
        assert_eq!(plan.batches.len(), 2);
        assert_eq!(plan.batches[0].len(), 500);
        assert_eq!(plan.batches[1], vec![500]);

        // Byte budget: 9 × 32 MiB = 288 MiB → 8 files (256 MiB) then 1.
        let sizes: Vec<Option<u64>> = (0..9).map(|_| Some(32 * MIB)).collect();
        let plan = plan_pull(&sizes, true);
        assert_eq!(plan.batches.len(), 2);
        assert_eq!(plan.batches[0].len(), 8);
        assert_eq!(plan.batches[1].len(), 1);

        // Unknown sizes cost 1 MiB against the budget: 255 × 1 MiB + 2
        // unknowns → the first unknown fills the 256 MiB budget exactly and
        // the second starts a new batch.
        let mut sizes: Vec<Option<u64>> = (0..255).map(|_| Some(MIB)).collect();
        sizes.push(None);
        sizes.push(None);
        let plan = plan_pull(&sizes, true);
        assert_eq!(plan.batches.len(), 2);
        assert_eq!(plan.batches[0].len(), 256);
        assert_eq!(plan.batches[1], vec![256]);
    }

    #[tokio::test]
    async fn archive_pull_batches_small_files_in_one_request() {
        let requests = Arc::new(AtomicUsize::new(0));
        let contents: Vec<(&'static str, Vec<u8>)> = vec![
            ("a.txt", b"alpha".to_vec()),
            ("Sub/Deep/b.bin", vec![7u8; 4096]),
            ("c.txt", b"gamma content".to_vec()),
        ];
        let base = spawn_archive_server(contents.clone(), requests.clone());
        let archive_url = format!("{base}/api/files/archive");

        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_str().unwrap().to_string();
        let files: Vec<PullFile> = contents
            .iter()
            .map(|(p, c)| pull_file(p, format!("{base}/unused/{p}"), c))
            .collect();

        let cache = HashCache::default();
        let (events, sink) = capture_sink();
        let res = pull_files_impl(
            &ws,
            &files,
            &HashMap::new(),
            false,
            Some(&archive_url),
            &cache,
            sink,
            None,
        )
        .await;

        assert!(res.ok, "errors: {:?}", res.errors);
        assert_eq!(res.changed, vec!["a.txt", "Sub/Deep/b.bin", "c.txt"]);
        assert_eq!(res.created, res.changed);
        assert!(res.dirs_created.contains(&"Sub".to_string()));
        assert!(res.dirs_created.contains(&"Sub/Deep".to_string()));
        // The three small files rode ONE archive request.
        assert_eq!(requests.load(Ordering::SeqCst), 1);

        let entries = &cache.lock().await.entries;
        for (path, content) in &contents {
            let full = dir.path().join(path);
            let mut buf = Vec::new();
            std::fs::File::open(&full).unwrap().read_to_end(&mut buf).unwrap();
            assert_eq!(&buf, content, "{path} content");
            assert!(!crate::workspace::is_writable(&std::fs::metadata(&full).unwrap()));
            assert!(!temp_path(&full).exists());
            // Hash landed in the cache under the final path+size+mtime key,
            // so the next scan is stat-only for pulled files.
            let key = cache_key(&full, &std::fs::metadata(&full).unwrap());
            assert_eq!(entries.get(&key), Some(&sha_hex(content)), "{path} cached");
        }

        let events = events.lock().unwrap();
        assert!(!events.is_empty());
        let last = events.last().unwrap();
        assert_eq!(last.kind, "pull");
        assert_eq!((last.done, last.total), (3, 3));
        let bytes_total: u64 = contents.iter().map(|(_, c)| c.len() as u64).sum();
        assert_eq!(last.bytes_total, bytes_total);
        assert_eq!(last.bytes_done, bytes_total);
        // done counts are monotonically non-decreasing across events.
        assert!(events.windows(2).all(|w| w[0].done <= w[1].done));
    }

    #[tokio::test]
    async fn archive_hash_mismatch_is_a_per_file_error() {
        // Server frames bad.txt with a hash that doesn't match its bytes.
        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
        let port = server.server_addr().to_ip().unwrap().port();
        std::thread::spawn(move || {
            for mut req in server.incoming_requests() {
                let mut body = String::new();
                req.as_reader().read_to_string(&mut body).unwrap();
                let mut frames = Vec::new();
                let bad_header = serde_json::json!({
                    "path": "bad.txt", "size": 9, "hash": sha_hex(b"different")
                })
                .to_string();
                frames.extend_from_slice(&(bad_header.len() as u64).to_le_bytes());
                frames.extend_from_slice(bad_header.as_bytes());
                frames.extend_from_slice(b"bad bytes");
                frames.extend_from_slice(&frame_archive(&[("good.txt", b"good data")]));
                let _ = req.respond(tiny_http::Response::from_data(frames));
            }
        });
        let archive_url = format!("http://127.0.0.1:{port}/api/files/archive");

        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_str().unwrap().to_string();
        let files = vec![
            pull_file("bad.txt", "http://unused/".into(), b"bad bytes"),
            pull_file("good.txt", "http://unused/".into(), b"good data"),
        ];
        let cache = HashCache::default();
        let res =
            pull_files_impl(
                &ws, &files, &HashMap::new(), false, Some(&archive_url), &cache, None, None,
            )
            .await;

        assert!(!res.ok);
        assert_eq!(res.errors.len(), 1);
        assert_eq!(res.errors[0].path, "bad.txt");
        assert!(res.errors[0].error.contains("hash verification"));
        assert!(!dir.path().join("bad.txt").exists());
        assert!(!dir.path().join("bad.txt.rc-download").exists());
        // The stream stayed aligned: the following file still landed.
        assert_eq!(res.changed, vec!["good.txt"]);
        assert_eq!(std::fs::read(dir.path().join("good.txt")).unwrap(), b"good data");
    }

    #[tokio::test]
    async fn archive_interrupted_stream_errors_remaining_files() {
        // Raw TCP server: declares the full Content-Length but sends only the
        // first file's frames plus half of the second body, then hangs up.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        std::thread::spawn(move || {
            let (mut conn, _) = listener.accept().unwrap();
            conn.set_read_timeout(Some(std::time::Duration::from_secs(5))).unwrap();
            // Drain the request (headers + JSON body) best-effort.
            let mut buf = [0u8; 65536];
            let mut seen = Vec::new();
            loop {
                match conn.read(&mut buf) {
                    Ok(0) => break,
                    Ok(n) => {
                        seen.extend_from_slice(&buf[..n]);
                        let headers_end =
                            seen.windows(4).position(|w| w == b"\r\n\r\n").map(|p| p + 4);
                        if let Some(end) = headers_end {
                            let head = String::from_utf8_lossy(&seen[..end]).to_lowercase();
                            let len: usize = head
                                .lines()
                                .find_map(|l| l.strip_prefix("content-length:"))
                                .map(|v| v.trim().parse().unwrap())
                                .unwrap_or(0);
                            if seen.len() >= end + len {
                                break;
                            }
                        }
                    }
                    Err(_) => break,
                }
            }
            let full = frame_archive(&[
                ("one.txt", b"first file"),
                ("two.txt", b"second file body"),
                ("three.txt", b"third file"),
            ]);
            let first = frame_archive(&[("one.txt", b"first file")]);
            let truncated = &full[..first.len() + 30]; // header + part of two.txt
            let _ = conn.write_all(
                format!(
                    "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nContent-Type: application/x-rc-archive\r\n\r\n",
                    full.len()
                )
                .as_bytes(),
            );
            let _ = conn.write_all(truncated);
            let _ = conn.flush();
            // Drop = RST/FIN long before Content-Length is satisfied.
        });
        let archive_url = format!("http://127.0.0.1:{port}/api/files/archive");

        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_str().unwrap().to_string();
        let files = vec![
            pull_file("one.txt", "http://unused/".into(), b"first file"),
            pull_file("two.txt", "http://unused/".into(), b"second file body"),
            pull_file("three.txt", "http://unused/".into(), b"third file"),
        ];
        let cache = HashCache::default();
        let res = tokio::time::timeout(
            std::time::Duration::from_secs(30),
            pull_files_impl(&ws, &files, &HashMap::new(), false, Some(&archive_url), &cache, None, None),
        )
        .await
        .expect("pull must not hang on a truncated stream");

        assert!(!res.ok);
        assert_eq!(res.changed, vec!["one.txt"]);
        assert_eq!(std::fs::read(dir.path().join("one.txt")).unwrap(), b"first file");
        let errored: Vec<&str> = res.errors.iter().map(|e| e.path.as_str()).collect();
        assert!(errored.contains(&"two.txt"));
        assert!(errored.contains(&"three.txt"));
        for e in &res.errors {
            assert!(e.error.contains("interrupted"), "{}: {}", e.path, e.error);
        }
        // No temp litter for the interrupted files.
        assert!(!dir.path().join("two.txt").exists());
        assert!(!dir.path().join("two.txt.rc-download").exists());
        assert!(!dir.path().join("three.txt.rc-download").exists());
    }

    #[tokio::test]
    async fn large_files_stream_individually_with_progress_heartbeats() {
        // size_bytes above the 32 MiB threshold forces the per-file GET even
        // though an archive URL is available; a 9 MiB body crosses the 8 MiB
        // stride so at least one mid-file heartbeat fires.
        let body = vec![42u8; 9 * MIB as usize];
        let hash = sha_hex(&body);
        let archive_hits = Arc::new(AtomicUsize::new(0));
        let base = spawn_server(vec![("/big.bin", 200, body.clone())]);
        let archive_url = spawn_archive_server(vec![], archive_hits.clone());

        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_str().unwrap().to_string();
        let files = vec![PullFile {
            path: "big.bin".into(),
            url: format!("{base}/big.bin"),
            hash: Some(hash.clone()),
            size_bytes: Some(40 * MIB),
        }];
        let cache = HashCache::default();
        let (events, sink) = capture_sink();
        let res = pull_files_impl(
            &ws,
            &files,
            &HashMap::new(),
            false,
            Some(&archive_url),
            &cache,
            sink,
            None,
        )
        .await;

        assert!(res.ok, "errors: {:?}", res.errors);
        assert_eq!(res.changed, vec!["big.bin"]);
        assert_eq!(archive_hits.load(Ordering::SeqCst), 0);
        let full = dir.path().join("big.bin");
        assert_eq!(std::fs::metadata(&full).unwrap().len(), 9 * MIB);
        let key = cache_key(&full, &std::fs::metadata(&full).unwrap());
        assert_eq!(cache.lock().await.entries.get(&key), Some(&hash));

        let events = events.lock().unwrap();
        let mid: Vec<_> = events.iter().filter(|e| e.done == 0).collect();
        assert!(!mid.is_empty(), "expected a mid-file heartbeat before completion");
        assert!(mid[0].bytes_done >= 8 * MIB);
        assert_eq!(events.last().unwrap().done, 1);
    }

    #[tokio::test]
    async fn overlapping_pulls_on_one_workspace_are_rejected() {
        // Slow server: accepts the connection, then serves the file only after
        // a delay — long enough that the second pull provably starts while the
        // first still holds the workspace claim.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        std::thread::spawn(move || {
            for conn in listener.incoming() {
                let Ok(mut conn) = conn else { break };
                std::thread::spawn(move || {
                    let mut buf = [0u8; 8192];
                    let _ = conn.read(&mut buf);
                    std::thread::sleep(std::time::Duration::from_millis(1200));
                    let body = b"slow content";
                    let _ = conn.write_all(
                        format!(
                            "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                            body.len()
                        )
                        .as_bytes(),
                    );
                    let _ = conn.write_all(body);
                });
            }
        });

        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_str().unwrap().to_string();

        let first = {
            let ws = ws.clone();
            let url = format!("http://127.0.0.1:{port}/slow.txt");
            tokio::spawn(async move {
                let cache = HashCache::default();
                let files = vec![PullFile {
                    path: "slow.txt".into(),
                    url,
                    hash: None,
                    size_bytes: None,
                }];
                pull_files_impl(&ws, &files, &HashMap::new(), false, None, &cache, None, None).await
            })
        };
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;

        // Second pull of the SAME workspace while the first is in flight:
        // rejected outright — overlapping pulls share deterministic temp
        // names and corrupt each other's landed files.
        let cache = HashCache::default();
        let files = vec![PullFile {
            path: "other.txt".into(),
            url: format!("http://127.0.0.1:{port}/other.txt"),
            hash: None,
            size_bytes: None,
        }];
        let second =
            pull_files_impl(&ws, &files, &HashMap::new(), false, None, &cache, None, None).await;
        assert!(!second.ok);
        assert_eq!(second.errors.len(), 1);
        assert_eq!(second.errors[0].error, PULL_IN_PROGRESS);
        assert!(second.changed.is_empty());

        // A DIFFERENT workspace is not blocked by the first pull's claim.
        let other_dir = tempfile::tempdir().unwrap();
        let other_ws = other_dir.path().to_str().unwrap().to_string();
        let res = pull_files_impl(
            &other_ws,
            &[PullFile {
                path: "b.txt".into(),
                url: format!("http://127.0.0.1:{port}/b.txt"),
                hash: None,
                size_bytes: None,
            }],
            &HashMap::new(),
            false,
            None,
            &cache,
            None,
            None,
        )
        .await;
        assert!(res.errors.iter().all(|e| e.error != PULL_IN_PROGRESS));

        // Once the first pull finishes, its claim is released.
        let first_res = first.await.unwrap();
        assert!(first_res.ok, "first pull errors: {:?}", first_res.errors);
        let res =
            pull_files_impl(&ws, &files, &HashMap::new(), false, None, &cache, None, None).await;
        assert!(res.errors.iter().all(|e| e.error != PULL_IN_PROGRESS));
    }

    #[tokio::test]
    async fn individual_download_hash_mismatch_is_rejected() {
        let base = spawn_server(vec![("/f.bin", 200, b"actual bytes".to_vec())]);
        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_str().unwrap().to_string();
        let files = vec![PullFile {
            path: "f.bin".into(),
            url: format!("{base}/f.bin"),
            hash: Some(sha_hex(b"something else")),
            size_bytes: Some(40 * MIB), // force the individual path
        }];
        let cache = HashCache::default();
        let res =
            pull_files_impl(&ws, &files, &HashMap::new(), false, None, &cache, None, None).await;
        assert!(!res.ok);
        assert!(res.errors[0].error.contains("hash verification"));
        assert!(!dir.path().join("f.bin").exists());
        assert!(!dir.path().join("f.bin.rc-download").exists());
    }

    #[tokio::test]
    async fn pull_downloads_skips_matching_and_reports_errors() {
        let base = spawn_server(vec![
            ("/new.txt", 200, b"new content".to_vec()),
            ("/broken.txt", 500, b"".to_vec()),
        ]);
        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_str().unwrap().to_string();

        // Pre-existing file whose hash matches → must be skipped untouched.
        std::fs::write(dir.path().join("same.txt"), b"hello").unwrap();
        let same_hash = "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824";

        let cache = HashCache::default();
        let files = vec![
            PullFile {
                path: "same.txt".into(),
                url: format!("{base}/same.txt"),
                hash: Some(same_hash.into()),
                size_bytes: None,
            },
            PullFile {
                path: "Assets/Deep/new.txt".into(),
                url: format!("{base}/new.txt"),
                hash: None,
                size_bytes: None,
            },
            PullFile {
                path: "broken.txt".into(),
                url: format!("{base}/broken.txt"),
                hash: None,
                size_bytes: None,
            },
            PullFile {
                path: "../escape.txt".into(),
                url: format!("{base}/new.txt"),
                hash: None,
                size_bytes: None,
            },
        ];
        let res =
            pull_files_impl(&ws, &files, &HashMap::new(), false, None, &cache, None, None).await;

        assert_eq!(res.changed, vec!["Assets/Deep/new.txt".to_string()]);
        assert_eq!(res.created, vec!["Assets/Deep/new.txt".to_string()]);
        assert!(res.dirs_created.contains(&"Assets".to_string()));
        assert!(res.dirs_created.contains(&"Assets/Deep".to_string()));
        assert_eq!(res.errors.len(), 2); // broken.txt (500) + ../escape.txt
        assert!(!res.ok);

        // Downloaded file is read-only (writable=false) with the right content.
        let full = dir.path().join("Assets/Deep/new.txt");
        let mut s = String::new();
        std::fs::File::open(&full).unwrap().read_to_string(&mut s).unwrap();
        assert_eq!(s, "new content");
        assert!(!crate::workspace::is_writable(&std::fs::metadata(&full).unwrap()));
        // No stray temp file.
        assert!(!dir.path().join("broken.txt.rc-download").exists());
        // The pulled file's hash is recorded without a rescan.
        let key = cache_key(&full, &std::fs::metadata(&full).unwrap());
        assert_eq!(cache.lock().await.entries.get(&key), Some(&sha_hex(b"new content")));
    }

    #[tokio::test]
    async fn uploads_run_parallel_and_preserve_input_order() {
        // The server holds every request until all 4 have arrived, then
        // answers each — a sequential engine would deadlock (caught by the
        // timeout), and completion order is scrambled vs input order.
        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
        let port = server.server_addr().to_ip().unwrap().port();
        std::thread::spawn(move || {
            let mut held = Vec::new();
            for mut req in server.incoming_requests() {
                let mut body = Vec::new();
                req.as_reader().read_to_end(&mut body).unwrap();
                held.push((req, body.len()));
                if held.len() == 4 {
                    for (req, n) in held.drain(..).rev() {
                        let name = req.url().rsplit('/').next().unwrap().to_string();
                        let _ = req.respond(
                            tiny_http::Response::from_string(format!(
                                r#"{{"hash":"h-{name}","n":{n}}}"#
                            ))
                            .with_status_code(200),
                        );
                    }
                    break;
                }
            }
        });

        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_str().unwrap().to_string();
        let mut files = Vec::new();
        for i in 0..4 {
            let name = format!("f{i}.bin");
            std::fs::write(dir.path().join(&name), vec![i as u8; 100 + i]).unwrap();
            files.push(UploadFile {
                path: name.clone(),
                url: format!("http://127.0.0.1:{port}/stage/{name}"),
                size_bytes: Some(100 + i as u64),
            });
        }

        let (events, sink) = capture_sink();
        let res = tokio::time::timeout(
            std::time::Duration::from_secs(30),
            upload_files_impl(&ws, &files, &HashMap::new(), sink, None),
        )
        .await
        .expect("parallel uploads must not serialize (server holds until 4 arrive)");

        assert!(res.ok);
        let paths: Vec<&str> = res.results.iter().map(|r| r.path.as_str()).collect();
        assert_eq!(paths, vec!["f0.bin", "f1.bin", "f2.bin", "f3.bin"]);
        for (i, r) in res.results.iter().enumerate() {
            assert_eq!(r.hash.as_deref(), Some(format!("h-f{i}.bin").as_str()));
            assert_eq!(r.size_bytes, Some(100 + i as u64));
        }

        let events = events.lock().unwrap();
        assert_eq!(events.len(), 4);
        let last = events.last().unwrap();
        assert_eq!(last.kind, "upload");
        assert_eq!((last.done, last.total), (4, 4));
        assert_eq!(last.bytes_done, 100 + 101 + 102 + 103);
        assert_eq!(last.bytes_total, 100 + 101 + 102 + 103);
    }

    #[tokio::test]
    async fn upload_streams_with_content_length() {
        let server = tiny_http::Server::http("127.0.0.1:0").unwrap();
        let port = server.server_addr().to_ip().unwrap().port();
        let handle = std::thread::spawn(move || {
            let mut req = server.recv().unwrap();
            let len = req
                .headers()
                .iter()
                .find(|h| h.field.equiv("Content-Length"))
                .map(|h| h.value.as_str().to_string());
            let chunked = req
                .headers()
                .iter()
                .any(|h| h.field.equiv("Transfer-Encoding"));
            let mut body = Vec::new();
            req.as_reader().read_to_end(&mut body).unwrap();
            let _ = req.respond(
                tiny_http::Response::from_string(r#"{"hash":"abc123"}"#).with_status_code(200),
            );
            (len, chunked, body)
        });

        let dir = tempfile::tempdir().unwrap();
        let ws = dir.path().to_str().unwrap().to_string();
        std::fs::write(dir.path().join("asset.bin"), vec![7u8; 1024]).unwrap();

        let files = vec![UploadFile {
            path: "asset.bin".into(),
            url: format!("http://127.0.0.1:{port}/stage"),
            size_bytes: None,
        }];
        let res = upload_files_impl(&ws, &files, &HashMap::new(), None, None).await;

        let (len, chunked, body) = handle.join().unwrap();
        assert_eq!(len.as_deref(), Some("1024"));
        assert!(!chunked, "must not use chunked transfer encoding");
        assert_eq!(body.len(), 1024);
        assert!(res.ok);
        assert_eq!(res.results[0].hash.as_deref(), Some("abc123"));
        assert_eq!(res.results[0].size_bytes, Some(1024));

        // Missing file → per-file error, ok:false, no panic.
        let res = upload_files_impl(
            &ws,
            &[UploadFile {
                path: "nope.bin".into(),
                url: "http://127.0.0.1:1/x".into(),
                size_bytes: None,
            }],
            &HashMap::new(),
            None,
            None,
        )
        .await;
        assert!(!res.ok);
        assert_eq!(res.results[0].error.as_deref(), Some("File not found in workspace."));
    }

    #[test]
    fn pull_rejects_blank_folder_path() {
        // Exercises the same guard pull_workspace_files (shims.rs) calls
        // before touching tauri::State, so it covers the command's
        // blank-path behavior without needing a live app handle.
        for blank in ["", "   ", "\t\n"] {
            let res = blank_pull_guard(blank).expect("blank path must be guarded");
            assert!(!res.ok);
            assert!(res.changed.is_empty());
            assert!(res.created.is_empty());
            assert!(res.dirs_created.is_empty());
            assert!(res.errors.is_empty());
        }
        assert!(blank_pull_guard("/tmp/ws").is_none());
    }
}
