//! Every `S3Client` method, against a real object store.
//!
//! WHY THIS FILE EXISTS. The crate had no integration test of any kind, and a
//! mutation run said what that costs: 114 of 206 mutants survived, 104 of them
//! inside `S3Client` (infra `1498fe2a`). `upload -> Ok(())` passed the whole
//! suite, because nothing in this repo ever called `upload`. The two worth
//! naming: `attempt < 3` can become `true` in all three multipart drivers,
//! which is an infinite retry on a permanent failure, and the part-offset
//! arithmetic in the copy driver can be wrong in either direction, which is
//! silent corruption of creator media -- the object completes and the bytes are
//! the wrong ones. Nothing short of a real round trip observes that.
//!
//! HOW TO RUN IT. The tier is astra, where MinIO is a local service:
//!
//!     set -a; . ~/.config/s3-storage-tests.env; set +a
//!     cargo nextest run --run-ignored all
//!
//! Every test is `#[ignore]`d so a developer machine reports them as SKIPPED
//! rather than passing silently. A test that did not run must never read as one
//! that did, which is the same rule the sweep applies to a cell it could not
//! evaluate.
//!
//! ISOLATION. `S3_TEST_*` names a bucket the credentials can reach and nothing
//! else: the key is scoped to it and MinIO refuses a write to the media buckets
//! (infra `6f7c7b46`, measured). Belt and braces anyway, every test works under
//! its own unique prefix and deletes it on the way out, so two runs cannot
//! collide and a leaked object is traceable to a run.
//!
//! MINIO IS NOT S3, and where they differ these tests assert the property that
//! matters rather than the exact bytes of a header. Multipart ETags are the
//! usual example: their format is not contractual, so nothing here reads one.

use std::io::{Read, Write};
use std::net::TcpStream;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

use s3_storage::{S3Client, S3Config};

const SKIP: &str = "live S3: set S3_TEST_* and run with --run-ignored (see the module docs)";

/// The bucket and credentials, or a panic naming what is missing.
///
/// A panic rather than a skip: these tests are `#[ignore]`d, so the only way to
/// reach this code is to have asked for them, and silently passing an asked-for
/// test because a variable was unset is the failure this whole file is about.
fn config() -> S3Config {
    fn var(name: &str) -> String {
        std::env::var(name)
            .unwrap_or_else(|_| panic!("{name} is not set; source ~/.config/s3-storage-tests.env"))
    }
    S3Config {
        endpoint: var("S3_TEST_ENDPOINT"),
        bucket: var("S3_TEST_BUCKET"),
        access_key: var("S3_TEST_ACCESS_KEY"),
        secret_key: var("S3_TEST_SECRET_KEY"),
        region: std::env::var("S3_TEST_REGION").unwrap_or_else(|_| "us-east-1".into()),
    }
}

async fn client() -> S3Client {
    S3Client::new(&config())
        .await
        .expect("building the client should not need the network")
}

/// A prefix no other run will use: the clock, the process, and a counter.
///
/// All three are needed. The counter because two tests in one process start
/// inside the same millisecond, and the pid because a mutation run has a dozen
/// copies of this suite against one bucket at once (infra `1498fe2a`), where
/// two runs sharing a prefix would delete each other's objects and report it as
/// a killed mutant.
fn prefix(what: &str) -> String {
    static SEQ: AtomicU32 = AtomicU32::new(0);
    let millis = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("the clock is after 1970")
        .as_millis();
    let n = SEQ.fetch_add(1, Ordering::Relaxed);
    format!("run-{millis}-{}-{n}/{what}", std::process::id())
}

/// Bytes that are not all the same, so a wrong offset shows up as wrong content
/// rather than as an identical-looking block.
fn pattern(len: usize) -> Vec<u8> {
    (0..len).map(|i| (i % 251) as u8).collect()
}

/// Compare without printing megabytes on failure: say where they diverge.
fn assert_same_bytes(expected: &[u8], actual: &[u8], what: &str) {
    assert_eq!(
        expected.len(),
        actual.len(),
        "{what}: length differs, expected {} bytes and got {}",
        expected.len(),
        actual.len()
    );
    if let Some(at) = expected.iter().zip(actual).position(|(a, b)| a != b) {
        panic!(
            "{what}: first difference at byte {at} of {}, expected {:#04x} and got {:#04x}",
            expected.len(),
            expected[at],
            actual[at]
        );
    }
}

/// Best-effort teardown. A failed cleanup must not turn a passing test red, and
/// must not hide a failing one, so it reports and moves on.
async fn cleanup(s3: &S3Client, prefix: &str) {
    if let Err(e) = s3.delete_prefix(prefix).await {
        eprintln!("cleanup of {prefix} failed, leaving objects behind: {e}");
    }
}

#[tokio::test]
#[ignore = "live S3"]
async fn an_object_round_trips_with_its_content_type() {
    let s3 = client().await;
    let key = prefix("round-trip/a.bin");
    let body = pattern(4096);

    s3.upload(
        &key,
        "application/octet-stream",
        body.clone(),
        Some("no-store"),
    )
    .await
    .expect("upload");

    let (got, content_type) = s3.download(&key).await.expect("download");
    assert_same_bytes(&body, &got, "round trip");
    assert_eq!(content_type, "application/octet-stream");

    assert!(s3.object_exists(&key).await.expect("exists"));
    assert_eq!(
        s3.object_size(&key).await.expect("size"),
        Some(body.len() as i64)
    );

    s3.delete(&key).await.expect("delete");
    assert!(
        !s3.object_exists(&key).await.expect("exists after delete"),
        "the object survived its own deletion"
    );
    assert_eq!(s3.object_size(&key).await.expect("size after delete"), None);

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn download_head_reads_a_prefix_and_nothing_more() {
    let s3 = client().await;
    let key = prefix("head/a.bin");
    let body = pattern(10_000);
    s3.upload(&key, "application/octet-stream", body.clone(), None)
        .await
        .expect("upload");

    let head = s3.download_head(&key, 512).await.expect("ranged read");
    assert_same_bytes(&body[..512], &head, "ranged read");

    // Asking for more than the object holds is not an error, and the answer is
    // the object.
    let all = s3
        .download_head(&key, body.len() * 2)
        .await
        .expect("over-read");
    assert_same_bytes(&body, &all, "over-read");

    // Zero is answered without a request at all. It has to be: the range header
    // is built as `bytes=0-{len-1}`, which underflows at zero.
    assert!(
        s3.download_head(&key, 0)
            .await
            .expect("zero-length")
            .is_empty()
    );

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn the_streaming_download_carries_the_same_bytes() {
    // `download` is `download_buf` plus a copy, and `download_buf` is the
    // aggregating form of this one, so the three share a path and only this one
    // hands the caller an unread stream.
    let s3 = client().await;
    let key = prefix("stream/a.bin");
    let body = pattern(200_000);
    s3.upload(&key, "application/octet-stream", body.clone(), None)
        .await
        .expect("upload");

    let stream = s3.download_stream(&key).await.expect("download_stream");
    let collected = stream.collect().await.expect("draining the stream");
    assert_same_bytes(&body, &collected.to_vec(), "streamed download");

    let (buffered, content_type) = s3.download_buf(&key).await.expect("download_buf");
    assert_same_bytes(&body, &buffered, "buffered download");
    assert_eq!(content_type, "application/octet-stream");

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_missing_object_is_absent_rather_than_an_error() {
    let s3 = client().await;
    let key = prefix("missing/never-written.bin");
    assert!(!s3.object_exists(&key).await.expect("exists"));
    assert_eq!(s3.object_size(&key).await.expect("size"), None);
    assert!(
        s3.download(&key).await.is_err(),
        "downloading nothing should fail rather than return empty"
    );
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_copy_carries_the_bytes_not_just_a_status() {
    let s3 = client().await;
    let root = prefix("copy");
    let src = format!("{root}/src.bin");
    let dst = format!("{root}/dst.bin");
    let from = format!("{root}/from.bin");
    let body = pattern(65_536);

    s3.upload(&src, "application/octet-stream", body.clone(), None)
        .await
        .expect("upload");

    s3.copy_object(&src, &dst).await.expect("copy_object");
    let (got, _) = s3.download(&dst).await.expect("download copy");
    assert_same_bytes(&body, &got, "copy_object");

    // Same bucket, named explicitly: the cross-bucket form with the only bucket
    // these credentials can reach.
    s3.copy_object_from(s3.bucket(), &src, &from)
        .await
        .expect("copy_object_from");
    let (got, _) = s3.download(&from).await.expect("download copy_from");
    assert_same_bytes(&body, &got, "copy_object_from");

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn deleting_a_batch_reports_no_failures_and_removes_every_key() {
    let s3 = client().await;
    let root = prefix("batch");
    let keys: Vec<String> = (0..5).map(|i| format!("{root}/{i}.bin")).collect();
    for key in &keys {
        s3.upload(key, "application/octet-stream", pattern(32), None)
            .await
            .expect("upload");
    }
    // A key that was never written, because a batch in the wild has them and S3
    // treats deleting an absent key as success.
    let mut with_absent = keys.clone();
    with_absent.push(format!("{root}/never-written.bin"));

    let failures = s3
        .delete_objects(&with_absent)
        .await
        .expect("delete_objects");
    assert!(failures.is_empty(), "delete_objects reported {failures:?}");
    for key in &keys {
        assert!(
            !s3.object_exists(key).await.expect("exists"),
            "{key} survived the batch delete"
        );
    }

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn delete_prefix_stops_at_the_prefix() {
    let s3 = client().await;
    let root = prefix("wipe");
    let doomed = format!("{root}/doomed");
    let kept = format!("{root}/kept");
    for i in 0..3 {
        s3.upload(
            &format!("{doomed}/{i}.bin"),
            "application/octet-stream",
            pattern(16),
            None,
        )
        .await
        .expect("upload");
    }
    s3.upload(
        &format!("{kept}/survivor.bin"),
        "application/octet-stream",
        pattern(16),
        None,
    )
    .await
    .expect("upload");

    s3.delete_prefix(&doomed).await.expect("delete_prefix");

    for i in 0..3 {
        assert!(
            !s3.object_exists(&format!("{doomed}/{i}.bin"))
                .await
                .expect("exists"),
            "object {i} survived delete_prefix"
        );
    }
    assert!(
        s3.object_exists(&format!("{kept}/survivor.bin"))
            .await
            .expect("exists"),
        "delete_prefix reached past its prefix, which is how a wipe becomes an incident"
    );

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_multipart_upload_reads_back_byte_identical() {
    let s3 = client().await;
    let key = prefix("multipart/big.bin");
    // Three parts at the 5 MiB floor, the last one deliberately short: a driver
    // that gets the final part's length wrong passes on an exact multiple.
    let part_size = 5 * 1024 * 1024;
    let body = pattern(part_size * 2 + 1_234_567);

    let file = std::env::temp_dir().join(format!("s3-live-{}.bin", std::process::id()));
    std::fs::write(&file, &body).expect("writing the source file");

    let uploaded = s3
        .upload_multipart(&key, "application/octet-stream", &file, Some(part_size))
        .await;
    std::fs::remove_file(&file).ok();
    uploaded.expect("upload_multipart");

    assert_eq!(
        s3.object_size(&key).await.expect("size"),
        Some(body.len() as i64),
        "the assembled object is the wrong length"
    );
    let (got, _) = s3.download(&key).await.expect("download");
    assert_same_bytes(&body, &got, "multipart round trip");

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn the_default_part_size_uploads_a_file_larger_than_one_part() {
    // `part_size: None` takes the 10 MiB default, which no other test exercises
    // and which nothing asserts: the default is written as an arithmetic
    // expression, and a wrong one is either a refusal at the 5 MiB floor or an
    // upload in more parts than intended. 12 MiB is two parts at the real
    // default and one at any smaller one.
    let s3 = client().await;
    let key = prefix("default-part/big.bin");
    let body = pattern(12 * 1024 * 1024);

    let file = std::env::temp_dir().join(format!("s3-live-default-{}.bin", std::process::id()));
    std::fs::write(&file, &body).expect("writing the source file");
    let uploaded = s3
        .upload_multipart(&key, "application/octet-stream", &file, None)
        .await;
    std::fs::remove_file(&file).ok();
    uploaded.expect("upload_multipart with the default part size");

    let (got, _) = s3.download(&key).await.expect("download");
    assert_same_bytes(&body, &got, "default part size round trip");

    cleanup(&s3, &key).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_part_size_below_the_s3_floor_is_refused_before_anything_is_created() {
    // The floor is 5 MiB and the check is `<`, so one byte under is refused and
    // exactly 5 MiB is not. Refused BEFORE the upload is created, which is why
    // this asserts on the pending-upload list as well as on the error: a
    // pre-flight that creates first and validates second strands an upload
    // nobody will ever complete or be billed for noticing.
    let s3 = client().await;
    let key = prefix("floor/rejected.bin");
    let file = std::env::temp_dir().join(format!("s3-live-floor-{}.bin", std::process::id()));
    std::fs::write(&file, pattern(1024)).expect("writing the source file");

    let err = s3
        .upload_multipart(
            &key,
            "application/octet-stream",
            &file,
            Some(5 * 1024 * 1024 - 1),
        )
        .await
        .expect_err("a part below the floor must be refused");
    assert!(
        err.contains("at least 5 MB"),
        "refused for the wrong reason: {err}"
    );
    std::fs::remove_file(&file).ok();

    assert!(
        s3.list_multipart_uploads_for_key(&key)
            .await
            .expect("list")
            .is_empty(),
        "the refusal created an upload and left it pending"
    );
}

#[tokio::test]
#[ignore = "live S3"]
async fn pending_uploads_are_listed_for_their_own_key_only() {
    // The listing filters a bucket-wide response down to one key. A filter that
    // widened would hand a caller somebody else's upload id, and
    // abort_multipart_upload would then cancel an upload that was going fine.
    let s3 = client().await;
    let root = prefix("listing");
    let mine = format!("{root}/mine.bin");
    let theirs = format!("{root}/theirs.bin");

    let mine_id = s3
        .create_multipart_upload(&mine, "application/octet-stream")
        .await
        .expect("create mine");
    let theirs_id = s3
        .create_multipart_upload(&theirs, "application/octet-stream")
        .await
        .expect("create theirs");

    let listed = s3
        .list_multipart_uploads_for_key(&mine)
        .await
        .expect("list mine");
    assert!(
        listed.contains(&mine_id),
        "our own upload is missing: {listed:?}"
    );
    assert!(
        !listed.contains(&theirs_id),
        "the listing reached past its key: {listed:?}"
    );

    s3.abort_multipart_upload(&mine, &mine_id)
        .await
        .expect("abort mine");
    s3.abort_multipart_upload(&theirs, &theirs_id)
        .await
        .expect("abort theirs");
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_multipart_copy_preserves_every_byte() {
    let s3 = client().await;
    let root = prefix("multipart-copy");
    let src = format!("{root}/src.bin");
    let dst = format!("{root}/dst.bin");
    let part_size = 5 * 1024 * 1024;
    let body = pattern(part_size * 2 + 999_983);

    let file = std::env::temp_dir().join(format!("s3-live-copy-{}.bin", std::process::id()));
    std::fs::write(&file, &body).expect("writing the source file");
    let uploaded = s3
        .upload_multipart(&src, "application/octet-stream", &file, Some(part_size))
        .await;
    std::fs::remove_file(&file).ok();
    uploaded.expect("upload_multipart");

    s3.copy_object_multipart(
        s3.bucket(),
        &src,
        &dst,
        "application/octet-stream",
        body.len() as u64,
        Some(part_size),
    )
    .await
    .expect("copy_object_multipart");

    let (got, _) = s3.download(&dst).await.expect("download");
    // THE ASSERTION THIS FILE WAS WRITTEN FOR. A copy driver whose part offsets
    // are off by one part duplicates or drops a range, the object still
    // completes, and nothing but the bytes says so.
    assert_same_bytes(&body, &got, "multipart copy");

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn an_aborted_upload_leaves_nothing_behind() {
    let s3 = client().await;
    let key = prefix("abort/pending.bin");

    let upload_id = s3
        .create_multipart_upload(&key, "application/octet-stream")
        .await
        .expect("create_multipart_upload");
    let pending = s3
        .list_multipart_uploads_for_key(&key)
        .await
        .expect("list_multipart_uploads_for_key");
    assert!(
        pending.contains(&upload_id),
        "the upload we just created is not listed as pending: {pending:?}"
    );

    s3.abort_multipart_upload(&key, &upload_id)
        .await
        .expect("abort_multipart_upload");
    let after = s3
        .list_multipart_uploads_for_key(&key)
        .await
        .expect("list after abort");
    assert!(
        !after.contains(&upload_id),
        "the aborted upload is still pending, which is a bill nobody sees: {after:?}"
    );
    assert!(
        !s3.object_exists(&key).await.expect("exists"),
        "an aborted upload produced an object"
    );
}

#[tokio::test]
#[ignore = "live S3"]
async fn a_presigned_url_actually_fetches_and_actually_uploads() {
    let s3 = client().await;
    let root = prefix("presign");
    let download_key = format!("{root}/download.bin");
    let upload_key = format!("{root}/upload.bin");
    let body = pattern(2048);

    s3.upload(
        &download_key,
        "application/octet-stream",
        body.clone(),
        None,
    )
    .await
    .expect("upload");

    let url = s3
        .presign_download(&download_key, 300)
        .await
        .expect("presign_download");
    let fetched = http_get(&url).expect("fetching the presigned URL");
    assert_same_bytes(&body, &fetched, "presigned download");

    let put_url = s3
        .presign_upload(
            &upload_key,
            "application/octet-stream",
            300,
            None,
            Some(body.len() as i64),
        )
        .await
        .expect("presign_upload");
    http_put(&put_url, "application/octet-stream", &body).expect("PUT to the presigned URL");
    let (got, _) = s3
        .download(&upload_key)
        .await
        .expect("download what was PUT");
    assert_same_bytes(&body, &got, "presigned upload");

    cleanup(&s3, &root).await;
}

#[tokio::test]
#[ignore = "live S3"]
async fn connectivity_answers_for_the_configured_bucket() {
    let s3 = client().await;
    s3.check_connectivity().await.expect("check_connectivity");

    let mut wrong = config();
    wrong.bucket = format!("{}-does-not-exist", wrong.bucket);
    let s3 = S3Client::new(&wrong).await.expect("client");
    assert!(
        s3.check_connectivity().await.is_err(),
        "connectivity passed against a bucket that does not exist, so it is not checking the bucket"
    );
}

// ---------------------------------------------------------------------------
// A minimal HTTP client, because a presigned URL is only worth anything if
// something outside this crate can use it.
//
// Raw TCP rather than a dependency: the tier's endpoint is MinIO on localhost
// over plain HTTP, and adding an HTTP stack to dev-dependencies to issue two
// requests would pull a TLS backend into a crate that deliberately pins its own
// (see Cargo.toml on `rustls-ring`). If the tier ever points at an https
// endpoint, these two helpers are what has to change, and they will fail loudly
// rather than quietly skip.
// ---------------------------------------------------------------------------

fn split_url(url: &str) -> (String, String) {
    let rest = url
        .strip_prefix("http://")
        .unwrap_or_else(|| panic!("this helper speaks plain HTTP only, got: {url}"));
    match rest.split_once('/') {
        Some((host, path)) => (host.to_string(), format!("/{path}")),
        None => (rest.to_string(), "/".to_string()),
    }
}

fn read_response(mut stream: TcpStream) -> Result<Vec<u8>, String> {
    let mut raw = Vec::new();
    stream
        .read_to_end(&mut raw)
        .map_err(|e| format!("reading the response: {e}"))?;
    let split = raw
        .windows(4)
        .position(|w| w == b"\r\n\r\n")
        .ok_or_else(|| "no header terminator in the response".to_string())?;
    let headers = String::from_utf8_lossy(&raw[..split]).to_string();
    let status = headers
        .lines()
        .next()
        .unwrap_or_default()
        .split_whitespace()
        .nth(1)
        .unwrap_or_default()
        .to_string();
    if !status.starts_with('2') {
        return Err(format!("HTTP {status}: {headers}"));
    }
    Ok(raw[split + 4..].to_vec())
}

fn http_get(url: &str) -> Result<Vec<u8>, String> {
    let (host, path) = split_url(url);
    let mut stream = TcpStream::connect(&host).map_err(|e| format!("connecting to {host}: {e}"))?;
    let req = format!("GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n");
    stream
        .write_all(req.as_bytes())
        .map_err(|e| format!("writing the request: {e}"))?;
    read_response(stream)
}

fn http_put(url: &str, content_type: &str, body: &[u8]) -> Result<Vec<u8>, String> {
    let (host, path) = split_url(url);
    let mut stream = TcpStream::connect(&host).map_err(|e| format!("connecting to {host}: {e}"))?;
    let head = format!(
        "PUT {path} HTTP/1.1\r\nHost: {host}\r\nContent-Type: {content_type}\r\n\
         Content-Length: {}\r\nConnection: close\r\n\r\n",
        body.len()
    );
    stream
        .write_all(head.as_bytes())
        .map_err(|e| format!("writing the request head: {e}"))?;
    stream
        .write_all(body)
        .map_err(|e| format!("writing the body: {e}"))?;
    read_response(stream)
}

/// Named so `cargo nextest list` shows why the file looks empty on a laptop.
#[test]
#[ignore = "live S3"]
fn these_tests_need_a_live_object_store() {
    println!("{SKIP}");
}