//! Chunked, resumable, deduplicated uploads: `PUT /chunks/{hash}`, `POST /chunks/check`
//! and `PUT /files/{path}` with `X-Chunks`. Real HTTP with shrunken limits.

mod common;

use std::path::{Path, PathBuf};

use common::{ADMIN, Server, temp_dir};
use notescraft_sync::hash_hex;
use notescraft_sync_server::{AppState, Store, TOMBSTONE_GRACE_MS, now_ms};
use serde_json::{Value, json};

/// Tests upload 1 KiB chunks instead of the 4 MiB default.
const CHUNK_MAX: usize = 1024;
/// Files may reach 8 KiB in tests instead of the 1 GiB default.
const FILE_CAP: u64 = 8 * 1024;
/// A 64 hex digit shape for hostile or placeholder hashes.
fn hex_zeros(n: usize) -> String {
    "0".repeat(n)
}

fn chunks_server(name: &str, quota: Option<u64>) -> (Server, PathBuf) {
    let data = temp_dir(name);
    let state = AppState::with_quota(Store::open(&data.join("sync.db")).expect("db"), data.join("blobs"), ADMIN, quota);
    state.set_chunk_limits(CHUNK_MAX, FILE_CAP);
    (common::start_with_state(state), data)
}

/// `len` bytes of one value, so chunks of different seeds differ.
fn chunk_data(seed: u8, len: usize) -> Vec<u8> {
    vec![seed; len]
}

/// The chunk files of a vault, by name.
fn chunk_names(data: &Path, vault: &str) -> Vec<String> {
    let dir = data.join("blobs").join(vault).join("chunks");
    let mut names: Vec<String> = std::fs::read_dir(&dir)
        .map_or_else(|_| Vec::new(), |entries| entries.filter_map(|entry| entry.ok().and_then(|e| e.file_name().into_string().ok())).collect());
    names.sort();
    names
}

/// `PUT /chunks/{hash}`: the status.
fn put_chunk(server: &Server, vault: &str, token: &str, hash: &str, body: &[u8]) -> u16 {
    server.send("PUT", &format!("/v1/vaults/{vault}/chunks/{hash}"), Some(token), &[], Some(body)).0
}

/// Uploads the seeds as [`CHUNK_MAX`]-sized chunks; answers their joined bytes and hashes.
fn upload_chunks(server: &Server, vault: &str, token: &str, seeds: &[u8]) -> (Vec<u8>, Vec<String>) {
    let mut content = Vec::new();
    let mut hashes = Vec::new();
    for seed in seeds {
        let bytes = chunk_data(*seed, CHUNK_MAX);
        assert_eq!(put_chunk(server, vault, token, &hash_hex(&bytes), &bytes), 201);
        hashes.push(hash_hex(&bytes));
        content.extend_from_slice(&bytes);
    }
    (content, hashes)
}

/// `POST /chunks/check`: `(status, missing)`.
fn check_chunks(server: &Server, vault: &str, token: &str, hashes: &[String]) -> (u16, Vec<String>) {
    let (status, reply) = server.json("POST", &format!("/v1/vaults/{vault}/chunks/check"), Some(token), &[], Some(&json!({"hashes": hashes})));
    let missing = reply["missing"].as_array().map_or_else(Vec::new, |rows| rows.iter().filter_map(|v| v.as_str().map(str::to_owned)).collect());
    (status, missing)
}

/// `PUT /files/{path}` with `X-Chunks`: `(status, reply body)`.
fn assemble(server: &Server, vault: &str, token: &str, path: &str, hashes: &[String], base_seq: u64, content_hash: Option<&str>) -> (u16, Value) {
    let joined = hashes.join(",");
    let base = base_seq.to_string();
    let mut named: Vec<(&str, &str)> = vec![("X-Chunks", joined.as_str()), ("X-Base-Seq", base.as_str())];
    named.extend(content_hash.map(|hash| ("X-Content-Hash", hash)));
    let (status, reply) = server.send("PUT", &format!("/v1/vaults/{vault}/files/{path}"), Some(token), &named, None);
    (status, serde_json::from_slice(&reply).unwrap_or(Value::Null))
}

#[test]
fn chunks_upload_idempotently_assemble_and_read_back_exactly() {
    let (server, data) = chunks_server("chunks", None);
    let (vault, token) = server.paired("pc");
    let (content, hashes) = upload_chunks(&server, &vault, &token, &[1, 2, 3, 4, 5]);
    assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()), "five chunks stored once each");

    // Everything is there; the resume endpoint answers complete.
    let (status, missing) = check_chunks(&server, &vault, &token, &hashes);
    assert_eq!((status, missing), (200, Vec::<String>::new()));

    let expected_hash = hash_hex(&content);
    let (status, record) = assemble(&server, &vault, &token, "big.bin", &hashes, 0, None);
    assert_eq!(status, 201);
    assert_eq!(record["hash"].as_str(), Some(expected_hash.as_str()));
    assert_eq!(record["size"].as_u64(), Some(u64::try_from(content.len()).expect("small")));
    let (status, bytes) = server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None);
    assert_eq!((status, bytes.as_slice()), (200, content.as_slice()));

    // A second assembly over a fresh path reuses the chunk files.
    let (status, _) = assemble(&server, &vault, &token, "same.bin", &hashes, 0, None);
    assert_eq!(status, 201, "a fresh path is a new file");
    assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()), "chunks untouched by assembly");
}

/// The hashes a folder holds, ascending, for comparisons with `chunk_names`.
fn sorted(hashes: &[String]) -> Vec<String> {
    let mut sorted = hashes.to_vec();
    sorted.sort();
    sorted
}

/// `PUT /files/{path}` with an `X-Chunks` header through a raw HTTP connection, closed
/// after the answer; `(status, body)`.
fn raw_put(server: &Server, token: &str, vault: &str, path: &str, chunks: &str) -> (u16, String) {
    use std::io::{Read, Write};
    let port: u16 = server.base.strip_prefix("http://127.0.0.1:").and_then(|raw| raw.parse().ok()).expect("port");
    let request = format!(
        "PUT /v1/vaults/{vault}/files{path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nAuthorization: Bearer {token}\r\nX-Chunks: {chunks}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
    );
    let mut stream = std::net::TcpStream::connect(("127.0.0.1", port)).expect("connect");
    stream.write_all(request.as_bytes()).expect("write");
    let mut reply = Vec::new();
    stream.read_to_end(&mut reply).expect("read");
    let text = String::from_utf8_lossy(&reply).into_owned();
    let code = text.split_whitespace().nth(1).and_then(|word| word.parse().ok()).unwrap_or(0);
    (code, text)
}

#[test]
fn resume_after_a_cut_connection_only_sends_missing_chunks() {
    let (server, _) = chunks_server("resume", None);
    let (vault, token) = server.paired("pc");
    let seeds = [1u8, 2, 3, 4, 5];
    let hashes: Vec<String> = seeds.iter().map(|seed| hash_hex(&chunk_data(*seed, CHUNK_MAX))).collect();
    let content: Vec<u8> = seeds.iter().flat_map(|seed| chunk_data(*seed, CHUNK_MAX)).collect();

    // Three of five arrived before the cut.
    for (index, seed) in seeds.iter().enumerate().take(3) {
        assert_eq!(put_chunk(&server, &vault, &token, &hashes[index], &chunk_data(*seed, CHUNK_MAX)), 201);
    }
    let (status, missing) = check_chunks(&server, &vault, &token, &hashes);
    assert_eq!((status, missing), (200, hashes[3..5].to_vec()));

    // The resumed connection sends only what is missing, then assembles.
    for (index, seed) in seeds.iter().enumerate().skip(3) {
        assert_eq!(put_chunk(&server, &vault, &token, &hashes[index], &chunk_data(*seed, CHUNK_MAX)), 201);
    }
    assert_eq!(assemble(&server, &vault, &token, "big.bin", &hashes, 0, None).0, 201);
    let (status, bytes) = server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None);
    assert_eq!((status, bytes.as_slice()), (200, content.as_slice()));
}

#[test]
fn a_chunk_whose_hash_does_not_match_is_refused_and_stores_nothing() {
    let (server, data) = chunks_server("badhash", None);
    let (vault, token) = server.paired("pc");
    let body = chunk_data(1, 100);
    for wrong in ["a".repeat(64), "A".repeat(64), "0".repeat(63), String::from("x")] {
        let (status, _) = server.send("PUT", &format!("/v1/vaults/{vault}/chunks/{wrong}"), Some(&token), &[], Some(body.as_slice()));
        assert_eq!(status, 400, "hash {wrong}");
    }
    // A well formed hash that does not match the body.
    assert_eq!(put_chunk(&server, &vault, &token, &hash_hex(b"other"), &body), 400);
    // The empty body is refused as well.
    assert_eq!(put_chunk(&server, &vault, &token, &hash_hex(&body), b""), 400);
    assert_eq!(chunk_names(&data, &vault), Vec::<String>::new(), "nothing was stored");
}

#[test]
fn an_oversized_chunk_is_refused() {
    let (server, data) = chunks_server("bigchunk", None);
    let (vault, token) = server.paired("pc");
    let body = chunk_data(1, CHUNK_MAX + 1);
    assert_eq!(put_chunk(&server, &vault, &token, &hex_zeros(64), &body), 413);
    assert_eq!(chunk_names(&data, &vault), Vec::<String>::new());
}

#[test]
fn assembling_with_a_missing_chunk_answers_409_and_makes_no_file() {
    let (server, _) = chunks_server("missing", None);
    let (vault, token) = server.paired("pc");
    let bytes = chunk_data(1, CHUNK_MAX);
    let hash = hash_hex(&bytes);
    assert_eq!(put_chunk(&server, &vault, &token, &hash, &bytes), 201);
    let unknown = "f".repeat(64);
    let (status, reply) = assemble(&server, &vault, &token, "big.bin", &[hash, unknown.clone()], 0, None);
    assert_eq!(status, 409);
    assert_eq!(reply["missing"], json!([unknown]), "the absent hash is named");
    assert_eq!(server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None).0, 404);
}

#[test]
fn assembling_with_a_wrong_content_hash_is_400() {
    let (server, _) = chunks_server("wrongcontent", None);
    let (vault, token) = server.paired("pc");
    let (_, hashes) = upload_chunks(&server, &vault, &token, &[1, 2]);
    let (status, _) = assemble(&server, &vault, &token, "big.bin", &hashes, 0, Some(&"f".repeat(64)));
    assert_eq!(status, 400, "the header must match the computed hash");
    let content: Vec<u8> = chunk_data(1, CHUNK_MAX).into_iter().chain(chunk_data(2, CHUNK_MAX)).collect();
    let (status, record) = assemble(&server, &vault, &token, "big.bin", &hashes, 0, Some(&hash_hex(&content)));
    assert_eq!((status, record["size"].as_u64()), (201, Some(u64::try_from(content.len()).expect("small"))));
}

#[test]
fn an_assembly_over_the_file_cap_is_413() {
    let (server, _) = chunks_server("filecap", None);
    let (vault, token) = server.paired("pc");
    let seeds: Vec<u8> = (1..=10).collect();
    let (_, hashes) = upload_chunks(&server, &vault, &token, &seeds);
    let (status, reply) = assemble(&server, &vault, &token, "huge.bin", &hashes, 0, None);
    assert_eq!(status, 413, "10 KiB of chunks exceed the 8 KiB file cap");
    assert!(reply["error"].is_string());
    assert_eq!(server.send("GET", &format!("/v1/vaults/{vault}/files/huge.bin"), Some(&token), &[], None).0, 404);
}

#[test]
fn the_second_file_sharing_chunks_stores_them_once() {
    let (server, data) = chunks_server("dedup", None);
    let (vault, token) = server.paired("pc");
    let (_, hashes) = upload_chunks(&server, &vault, &token, &[1, 2, 3]);
    assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()));

    // A second file with the same chunks in a different order: a different file hash,
    // and the chunk folder does not grow.
    let reversed: Vec<String> = hashes.iter().rev().cloned().collect();
    assert_eq!(assemble(&server, &vault, &token, "a.bin", &hashes, 0, None).0, 201);
    assert_eq!(assemble(&server, &vault, &token, "b.bin", &reversed, 0, None).0, 201);
    assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()), "three chunks, two files");

    // Re-uploading a stored chunk is idempotent.
    let bytes = chunk_data(1, CHUNK_MAX);
    assert_eq!(put_chunk(&server, &vault, &token, &hash_hex(&bytes), &bytes), 200, "already there");
    assert_eq!(chunk_names(&data, &vault), sorted(hashes.as_slice()));
}
#[test]
fn a_viewer_cannot_upload_chunks_or_assemble() {
    let (server, _) = chunks_server("viewer", None);
    let (vault, owner) = server.paired("pc");
    let (_, viewer) = server.paired("vw");
    let (status, code) = server.invite(&vault, &owner, "viewer");
    assert_eq!(status, 200);
    assert_eq!(server.accept(&viewer, &code).0, 200);

    let bytes = chunk_data(1, 100);
    assert_eq!(put_chunk(&server, &vault, &viewer, &hash_hex(&bytes), &bytes), 403);
    let (_, hashes) = upload_chunks(&server, &vault, &owner, &[1, 2]);
    assert_eq!(assemble(&server, &vault, &viewer, "big.bin", &hashes, 0, None).0, 403, "viewer assembly refused");
    assert_eq!(assemble(&server, &vault, &owner, "big.bin", &hashes, 0, None).0, 201);
}

#[test]
fn the_quota_refuses_a_chunk_and_an_assembly() {
    // A chunk larger than what the quota still allows: refused at upload.
    let (server, _) = chunks_server("quota-chunk", Some(512));
    let (vault, token) = server.paired("pc");
    let bytes = chunk_data(1, CHUNK_MAX);
    assert_eq!(put_chunk(&server, &vault, &token, &hash_hex(&bytes), &bytes), 413);
    assert!(server.audit().iter().any(|row| row["kind"] == "quota_rejected"));

    // Two chunks fit (their upload counts them), the assembly does not.
    let (server, _) = chunks_server("quota-assemble", Some(1536));
    let (vault, token) = server.paired("pc");
    let (_, hashes) = upload_chunks(&server, &vault, &token, &[1, 2]);
    let (status, reply) = assemble(&server, &vault, &token, "big.bin", &hashes, 0, None);
    assert_eq!(status, 413);
    assert_eq!(reply["error"], json!("vault storage quota exceeded"));
    assert!(server.audit().iter().any(|row| row["kind"] == "quota_rejected"));
    assert_eq!(server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None).0, 404);
}

#[test]
fn a_chunked_file_has_revisions_and_their_bytes_stay_restorable() {
    let (server, _) = chunks_server("history", None);
    let (vault, token) = server.paired("pc");
    let (v1, hashes1) = upload_chunks(&server, &vault, &token, &[1, 2]);
    let (status, _) = assemble(&server, &vault, &token, "doc.bin", &hashes1, 0, None);
    assert_eq!(status, 201);

    // The edit arrives as its own chunk set on a fresh base.
    let mut hashes2 = hashes1.clone();
    let (extra_bytes, extra_hash) = (chunk_data(3, CHUNK_MAX), hash_hex(&chunk_data(3, CHUNK_MAX)));
    assert_eq!(put_chunk(&server, &vault, &token, &extra_hash, &extra_bytes), 201);
    hashes2[0] = extra_hash;
    let (status, record) = assemble(&server, &vault, &token, "doc.bin", &hashes2, 1, None);
    assert_eq!(status, 201, "the edited version wins with a fresh base");
    let seq2 = record["seq"].as_u64().expect("seq");
    assert_eq!(seq2, 2);

    let (status, history) = server.json("GET", &format!("/v1/vaults/{vault}/files-history/doc.bin"), Some(&token), &[], None);
    assert_eq!((status, history.as_array().map(Vec::len)), (200, Some(2)));
    let (status, old) = server.send("GET", &format!("/v1/vaults/{vault}/files/doc.bin?seq=1"), Some(&token), &[], None);
    assert_eq!((status, old.as_slice()), (200, v1.as_slice()));
}

#[test]
fn gc_removes_unreferenced_chunks_and_keeps_assembled_ones() {
    let (server, data) = chunks_server("gc", None);
    let (vault, token) = server.paired("pc");
    let (_, referenced) = upload_chunks(&server, &vault, &token, &[1, 2]);
    let _spare = upload_chunks(&server, &vault, &token, &[3, 4]);
    assert_eq!(assemble(&server, &vault, &token, "big.bin", &referenced, 0, None).0, 201);
    let content: Vec<u8> = chunk_data(1, CHUNK_MAX).into_iter().chain(chunk_data(2, CHUNK_MAX)).collect();

    // Unreferenced chunks of a dead upload and an orphan file: collected after the grace.
    let orphan = data.join("blobs").join(&vault).join("chunks").join("garbage");
    std::fs::write(&orphan, b"waste").expect("orphan file");
    let aged = now_ms() + TOMBSTONE_GRACE_MS + 1;
    let report = server.state.gc(&vault, TOMBSTONE_GRACE_MS, aged).expect("gc");
    assert_eq!(report.removed, 3, "two spare chunks plus the orphan file");
    assert_eq!(chunk_names(&data, &vault), sorted(referenced.as_slice()), "the assembled file's chunks stay");
    let (status, bytes) = server.send("GET", &format!("/v1/vaults/{vault}/files/big.bin"), Some(&token), &[], None);
    assert_eq!((status, bytes.as_slice()), (200, content.as_slice()), "the assembled file survives GC");
}

#[test]
fn hostile_input_is_refused_without_panic() {
    let (server, data) = chunks_server("hostile", None);
    let (vault, token) = server.paired("pc");

    // A chunk check of 5001 hashes is refused; 5000 are the cap.
    let many: Vec<String> = (0..5001).map(|i| format!("{i:064x}")).collect();
    let (status, _) = server.json("POST", &format!("/v1/vaults/{vault}/chunks/check"), Some(&token), &[], Some(&json!({"hashes": many})));
    assert_eq!(status, 400);
    let (status, _) = server.json("POST", &format!("/v1/vaults/{vault}/chunks/check"), Some(&token), &[], Some(&json!({"hashes": []})));
    assert_eq!(status, 200, "an empty check answers complete");
    let (_, hashes) = upload_chunks(&server, &vault, &token, &[1, 2, 3]);
    let (status, missing) = check_chunks(&server, &vault, &token, &hashes);
    assert_eq!((status, missing), (200, Vec::<String>::new()));

    // An X-Chunks of 5000 entries is refused: beyond the 4096 list limit. The header
    // fits hyper's 408 KiB parse buffer, so it reaches the handler — ureq cannot send
    // headers this large, so the request goes through a raw HTTP connection.
    let beyond: Vec<String> = (0..5000).map(|i| format!("{i:064x}")).collect();
    let (status, _) = raw_put(&server, &token, &vault, "/x.bin", &beyond.join(","));
    assert_eq!(status, 400);

    // Malformed check entries, a hostile path and a non-empty body.

    // Malformed check entries, a hostile path and a non-empty body.
    let (status, _) = server.json("POST", &format!("/v1/vaults/{vault}/chunks/check"), Some(&token), &[], Some(&json!({"hashes": ["zz"]})));
    assert_eq!(status, 400);
    let (_, hashes) = upload_chunks(&server, &vault, &token, &[8]);
    assert_eq!(assemble(&server, &vault, &token, "a/../x.bin", &hashes, 0, None).0, 400);
    let (status, _) =
        server.send("PUT", &format!("/v1/vaults/{vault}/files/x.bin"), Some(&token), &[("X-Chunks", &hash_hex(&chunk_data(8, CHUNK_MAX)))], Some(b"not empty"));
    assert_eq!(status, 400, "a chunked upload carries no body");
    let (status, _) = server.send("PUT", &format!("/v1/vaults/{vault}/files/x.bin"), Some(&token), &[("X-Chunks", "")], None);
    assert_eq!(status, 400, "an empty X-Chunks names no chunks");
    assert!(chunk_names(&data, &vault).len() >= 2, "the honest chunks still exist");
}