//! Live-server smoke test for the Tauri transfer layer (Task 7).
//!
//! This is NOT run by a plain `cargo test` — every test here is `#[ignore]`
//! so the hermetic unit suite (tested against tiny_http, see transfer.rs) stays
//! fast and offline. Run explicitly against a real `CambiumServer/server.py`:
//!
//!   cd ../CambiumServer && python3 server.py &
//!   export RC_LIVE_SERVER=http://127.0.0.1:8666
//!   cargo test --manifest-path TauriClient/src-tauri/Cargo.toml \
//!       --test live_server -- --ignored --nocapture
//!
//! Why this exists: transfer.rs streams uploads with an explicit
//! Content-Length (not chunked encoding) specifically because the Python
//! stdlib `http.server` cannot decode chunked bodies. That assumption was
//! previously verified only against a tiny_http fixture in-process — never
//! against the real server. This test closes that gap.

use cambium_core::transfer::{
    pull_files_impl, upload_files_impl, PullFile, UploadFile,
};
use cambium_core::workspace::HashCache;
use serde_json::Value;
use std::collections::HashMap;

/// Read RC_LIVE_SERVER or panic with a clear message (so `--ignored` runs
/// without the env var fail loudly instead of silently no-op-ing).
fn base_url() -> String {
    std::env::var("RC_LIVE_SERVER")
        .expect("set RC_LIVE_SERVER=http://127.0.0.1:8666 to run this test")
}

/// Log in as @alice via plain reqwest (mirrors RcApi.login in client.js) and
/// return (token, headers-map-with-auth-only). Panics on any failure — this
/// is a smoke test, not a negative-path test.
async fn login(client: &reqwest::Client, base: &str) -> String {
    let payload = serde_json::json!({
        "username": "@alice",
        "password": "password",
        "deviceId": "live-server-smoke-test",
        "deviceName": "rust-integration-test",
    });
    let body = post_json(client, &format!("{base}/api/login"), None, None, &payload).await;
    body["token"]
        .as_str()
        .expect("login response missing token")
        .to_string()
}

/// reqwest is built without the "json" feature (see transfer.rs — this
/// project keeps that dependency lean), so POST/parse JSON manually rather
/// than pulling the feature in just for this test harness.
async fn post_json(
    client: &reqwest::Client,
    url: &str,
    token: Option<&str>,
    project: Option<&str>,
    payload: &Value,
) -> Value {
    let mut req = client
        .post(url)
        .header(reqwest::header::CONTENT_TYPE, "application/json")
        .body(serde_json::to_vec(payload).unwrap());
    if let Some(t) = token {
        req = req.bearer_auth(t);
    }
    if let Some(p) = project {
        req = req.header("X-Project", p);
    }
    let resp = req.send().await.expect("request failed");
    let status = resp.status();
    let bytes = resp.bytes().await.expect("failed to read response body");
    let body: Value = serde_json::from_slice(&bytes).expect("response not JSON");
    assert!(status.is_success(), "{url} returned {status}: {body}");
    body
}

async fn get_json(client: &reqwest::Client, url: &str, token: &str, project: Option<&str>) -> Value {
    let mut req = client.get(url).bearer_auth(token);
    if let Some(p) = project {
        req = req.header("X-Project", p);
    }
    let resp = req.send().await.expect("request failed");
    let status = resp.status();
    let bytes = resp.bytes().await.expect("failed to read response body");
    let body: Value = serde_json::from_slice(&bytes).expect("response not JSON");
    assert!(status.is_success(), "{url} returned {status}: {body}");
    body
}

/// GET /api/projects and return the id of the project to test against:
/// RC_LIVE_PROJECT (a name) when set — needed on servers that carry copied,
/// content-less metadata for other projects — else the project with the most
/// content so the pull step has something real to fetch.
async fn pick_project_with_files(client: &reqwest::Client, base: &str, token: &str) -> String {
    let body = get_json(client, &format!("{base}/api/projects"), token, None).await;
    let projects = body["projects"].as_array().expect("projects not an array");
    assert!(!projects.is_empty(), "no projects visible to @alice");
    let picked = match std::env::var("RC_LIVE_PROJECT") {
        Ok(name) => projects
            .iter()
            .find(|p| p["name"].as_str() == Some(name.as_str()))
            .unwrap_or_else(|| panic!("no project named {name}")),
        Err(_) => projects
            .iter()
            .max_by_key(|p| p["sizeBytes"].as_u64().unwrap_or(0))
            .expect("no project found"),
    };
    picked["id"].as_str().expect("project id missing").to_string()
}

/// Walk GET /api/depot/tree and return the depot path of the first file
/// entry found that is NOT currently checked out (so the download is not
/// contending with lock state left over from other runs).
fn first_unlocked_file(node: &Value, prefix: &str, out: &mut Option<String>) {
    if out.is_some() {
        return;
    }
    let children = match node.get("children").and_then(|c| c.as_array()) {
        Some(c) => c,
        None => return,
    };
    for child in children {
        let name = child["name"].as_str().unwrap_or("");
        let path = if prefix.is_empty() {
            name.to_string()
        } else {
            format!("{prefix}/{name}")
        };
        if child["type"].as_str() == Some("file") {
            if child.get("checkedOutBy").is_none() {
                *out = Some(path);
                return;
            }
        } else {
            first_unlocked_file(child, &path, out);
        }
        if out.is_some() {
            return;
        }
    }
}

async fn depot_tree(client: &reqwest::Client, base: &str, token: &str, project: &str) -> Value {
    get_json(client, &format!("{base}/api/depot/tree"), token, Some(project)).await
}

fn auth_headers(token: &str, project: &str) -> HashMap<String, String> {
    let mut headers = HashMap::new();
    headers.insert("Authorization".to_string(), format!("Bearer {token}"));
    headers.insert("X-Project".to_string(), project.to_string());
    headers
}

/// Downloads a real depot file through `pull_files_impl` into a scratch
/// tempdir workspace and verifies it lands read-only with the expected hash.
#[tokio::test]
#[ignore]
async fn pull_real_depot_file_lands_readonly() {
    let base = base_url();
    let client = reqwest::Client::new();
    let token = login(&client, &base).await;
    let project = pick_project_with_files(&client, &base, &token).await;
    let tree = depot_tree(&client, &base, &token, &project).await;

    let mut path = None;
    first_unlocked_file(&tree, "", &mut path);
    let path = path.expect("no unlocked file found in depot tree to test with");

    let hash = {
        // Re-walk to find the matching node's hash (kept separate from
        // first_unlocked_file for clarity).
        fn find_hash(node: &Value, target: &str, prefix: &str) -> Option<String> {
            let children = node.get("children")?.as_array()?;
            for child in children {
                let name = child["name"].as_str().unwrap_or("");
                let p = if prefix.is_empty() { name.to_string() } else { format!("{prefix}/{name}") };
                if p == target && child["type"].as_str() == Some("file") {
                    return child["hash"].as_str().map(String::from);
                }
                if let Some(h) = find_hash(child, target, &p) {
                    return Some(h);
                }
            }
            None
        }
        find_hash(&tree, &path, "")
    };

    let dir = tempfile::tempdir().expect("tempdir");
    let ws = dir.path().to_str().unwrap().to_string();
    let url = format!(
        "{base}/api/files/content?path={}",
        urlencoding_light(&path)
    );
    let headers = auth_headers(&token, &project);
    let cache = HashCache::default();

    let files = vec![PullFile { path: path.clone(), url, hash: None, size_bytes: None }];
    let result = pull_files_impl(&ws, &files, &headers, false, None, &cache, None, None).await;

    assert!(result.errors.is_empty(), "pull errors: {:?}", result.errors);
    assert_eq!(result.changed, vec![path.clone()]);

    let full = dir.path().join(&path);
    assert!(full.is_file(), "downloaded file missing at {:?}", full);

    let meta = std::fs::metadata(&full).unwrap();
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let mode = meta.permissions().mode() & 0o777;
        assert_eq!(mode, 0o444, "downloaded file must be read-only, got {mode:o}");
    }

    if let Some(expect_hash) = hash {
        let bytes = std::fs::read(&full).unwrap();
        use sha2::{Digest, Sha256};
        let actual = hex::encode(Sha256::digest(&bytes));
        assert_eq!(actual, expect_hash, "downloaded content hash mismatch");
    }

    println!("PASS: pulled {path} into {ws}, read-only, hash verified");
}

/// Uploads a small unique file through `upload_files_impl` straight to the
/// staging endpoint and asserts a `hash` comes back — proving the Python
/// stdlib server accepts our non-chunked, explicit-Content-Length body.
#[tokio::test]
#[ignore]
async fn upload_small_file_returns_hash() {
    let base = base_url();
    let client = reqwest::Client::new();
    let token = login(&client, &base).await;
    let project = pick_project_with_files(&client, &base, &token).await;

    let dir = tempfile::tempdir().expect("tempdir");
    let ws = dir.path().to_str().unwrap().to_string();
    let content = format!(
        "live-server smoke test upload at {:?}\n",
        std::time::SystemTime::now()
    );
    std::fs::write(dir.path().join("smoke.txt"), &content).unwrap();

    let headers = auth_headers(&token, &project);
    let upload_url = format!("{base}/api/files/upload");
    let files = vec![UploadFile { path: "smoke.txt".to_string(), url: upload_url, size_bytes: None }];

    let result = upload_files_impl(&ws, &files, &headers, None, None).await;

    assert!(result.ok, "upload failed: {:?}", result.results);
    let outcome = &result.results[0];
    assert!(outcome.error.is_none(), "upload error: {:?}", outcome.error);
    let hash = outcome.hash.as_deref().expect("no hash returned from upload");

    use sha2::{Digest, Sha256};
    let expect = hex::encode(Sha256::digest(content.as_bytes()));
    assert_eq!(hash, expect, "returned hash does not match sha256 of uploaded content");

    println!("PASS: uploaded smoke.txt ({} bytes), server returned matching hash {hash}", content.len());
}

/// Best-effort full round trip: upload a new file's content, then submit it
/// as an "added" change, then confirm the new revision shows up in the
/// project's revision list. Skipped (not failed) if submit is rejected for
/// reasons unrelated to the transfer layer (e.g. auth/locking edge cases) —
/// the two tests above already prove the thing this task cares about
/// (streaming transfer correctness against the real server).
#[tokio::test]
#[ignore]
async fn upload_then_submit_creates_revision() {
    let base = base_url();
    let client = reqwest::Client::new();
    let token = login(&client, &base).await;
    let project = pick_project_with_files(&client, &base, &token).await;

    let dir = tempfile::tempdir().expect("tempdir");
    let ws = dir.path().to_str().unwrap().to_string();
    let stamp = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis();
    let rel_path = format!("SmokeTest/live-{stamp}.txt");
    let content = format!("submitted by live_server.rs at {stamp}\n");
    let full_local = dir.path().join(&rel_path);
    std::fs::create_dir_all(full_local.parent().unwrap()).unwrap();
    std::fs::write(&full_local, &content).unwrap();

    let headers = auth_headers(&token, &project);
    let upload_url = format!("{base}/api/files/upload");
    let files = vec![UploadFile { path: rel_path.clone(), url: upload_url, size_bytes: None }];
    let upload_result = upload_files_impl(&ws, &files, &headers, None, None).await;

    let Some(hash) = upload_result.results.first().and_then(|o| o.hash.clone()) else {
        println!("SKIP: upload for submit test failed: {:?}", upload_result.results);
        return;
    };

    let submit_body = serde_json::json!({
        "description": "live_server.rs smoke test submit",
        "changes": [{ "path": rel_path, "change": "added", "hash": hash }],
        "folders": [],
    });
    let resp = client
        .post(format!("{base}/api/submit"))
        .bearer_auth(&token)
        .header("X-Project", &project)
        .header(reqwest::header::CONTENT_TYPE, "application/json")
        .body(serde_json::to_vec(&submit_body).unwrap())
        .send()
        .await
        .expect("submit request failed");
    let submit_bytes = resp.bytes().await.expect("failed to read submit response body");
    let submit_json: Value = serde_json::from_slice(&submit_bytes).unwrap_or(Value::Null);
    // apply_submit's response shape: { ok, revision, submitted: [...], conflicts: [...], ... }
    let accepted = submit_json
        .get("submitted")
        .and_then(|a| a.as_array())
        .map(|a| a.iter().any(|s| s.as_str() == Some(rel_path.as_str())))
        .unwrap_or(false);
    if !accepted {
        println!("SKIP: submit not accepted (response: {submit_json}), auth/locking made full round trip disproportionate to chase here");
        return;
    }

    let history = get_json(
        &client,
        &format!("{base}/api/files/history?path={}", urlencoding_light(&rel_path)),
        &token,
        Some(&project),
    )
    .await;
    let entries = history["history"].as_array().expect("history not an array");
    assert!(!entries.is_empty(), "submitted file has no history entries");

    println!("PASS: submitted {rel_path}, new revision confirmed in history ({} entries)", entries.len());
}

/// Whole-project pull through the batched parallel engine, timed, with the
/// two Transfer-v2 acceptance checks: every file lands verified, and a scan
/// straight after the pull re-hashes ZERO files (the pull already fed the
/// hash cache). Uses RC_LIVE_PROJECT (name) when set — the E2E server keeps
/// content-less copied metadata for other projects, so max-size picking
/// would grab an unpullable project there.
// Multi-thread flavor: the engine runs on Tauri's multi-threaded runtime in
// production; the default single-thread test runtime would serialize the
// 4-way transfer parallelism and understate throughput badly.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn timed_full_project_pull_feeds_cache() {
    let base = base_url();
    let client = reqwest::Client::new();
    let token = login(&client, &base).await;
    let project = pick_project_with_files(&client, &base, &token).await;
    let tree = depot_tree(&client, &base, &token, &project).await;

    fn flatten(node: &Value, prefix: &str, out: &mut Vec<(String, String, u64)>) {
        let Some(children) = node.get("children").and_then(|c| c.as_array()) else { return };
        for child in children {
            let name = child["name"].as_str().unwrap_or("");
            let path =
                if prefix.is_empty() { name.to_string() } else { format!("{prefix}/{name}") };
            if child["type"].as_str() == Some("file") {
                out.push((
                    path,
                    child["hash"].as_str().unwrap_or("").to_string(),
                    child["sizeBytes"].as_u64().unwrap_or(0),
                ));
            } else {
                flatten(child, &path, out);
            }
        }
    }
    let mut head = Vec::new();
    flatten(&tree, "", &mut head);
    assert!(!head.is_empty(), "project has no files to pull");
    let total_bytes: u64 = head.iter().map(|(_, _, s)| s).sum();

    let dir = tempfile::tempdir().expect("tempdir");
    let ws = dir.path().to_str().unwrap().to_string();
    let headers = auth_headers(&token, &project);
    let archive_url = format!("{base}/api/files/archive");
    let cache = HashCache::default();

    let files: Vec<PullFile> = head
        .iter()
        .map(|(path, hash, size)| PullFile {
            path: path.clone(),
            url: format!("{base}/api/files/content?path={}", urlencoding_light(path)),
            hash: Some(hash.clone()),
            size_bytes: Some(*size),
        })
        .collect();

    let events = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
    let sink = {
        let events = events.clone();
        Some(std::sync::Arc::new(move |p: cambium_core::transfer::TransferProgress| {
            events.lock().unwrap().push((p.done, p.total, p.bytes_done));
        })
            as std::sync::Arc<
                dyn Fn(cambium_core::transfer::TransferProgress) + Send + Sync,
            >)
    };

    let t0 = std::time::Instant::now();
    let result =
        pull_files_impl(&ws, &files, &headers, false, Some(&archive_url), &cache, sink, None).await;
    let pull_secs = t0.elapsed().as_secs_f64();

    assert!(result.errors.is_empty(), "pull errors (first 3): {:?}", &result.errors[..result.errors.len().min(3)]);
    assert_eq!(result.changed.len(), head.len(), "not every file landed");

    let events = events.lock().unwrap();
    let last = events.last().expect("no progress events");
    assert_eq!(last.0 as usize, head.len(), "final done != total files");
    assert!(events.windows(2).all(|w| w[0].0 <= w[1].0), "done went backwards");

    // The acceptance check behind "refreshes after the first are fast":
    // a hashed scan straight after the pull must find every hash already
    // cached — zero files re-read.
    let mut entries = cache.lock().await.entries.clone();
    let t1 = std::time::Instant::now();
    let (_, live_keys, newly_hashed) =
        cambium_core::workspace::scan_workspace(dir.path(), true, &mut entries).await;
    let scan_secs = t1.elapsed().as_secs_f64();
    assert_eq!(live_keys.len(), head.len());
    assert_eq!(newly_hashed, 0, "post-pull scan re-hashed files — cache insertion broken");

    println!(
        "PASS: pulled {} files / {:.2} GiB in {:.1}s ({:.0} MiB/s); post-pull hashed scan {:.2}s with 0 re-hashes; {} progress events",
        head.len(),
        total_bytes as f64 / (1u64 << 30) as f64,
        pull_secs,
        total_bytes as f64 / (1u64 << 20) as f64 / pull_secs,
        scan_secs,
        events.len(),
    );
}

/// Minimal percent-encoding for query-string path segments (path separators
/// kept literal since the server expects a normal depot path string).
fn urlencoding_light(s: &str) -> String {
    let mut out = String::new();
    for b in s.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' | b'/' => {
                out.push(b as char)
            }
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}
