//! EXPORT-S3-RETRY: pin the retry-budget contract for `PresignedS3ExportSink`.
//!
//! Scope: this file does NOT spin up an HTTP server (deliberately — keeps
//! dev-deps lean per the EXPORT-S3-RETRY task constraints). It exercises the
//! retry-budget contract via the public constructor and via behavioural
//! observation of `push()` against a connect-refused endpoint.
//!
//! What is locked down here:
//!   1. `max_attempts == 0` is rejected at construction (no "infinite retry by
//!      misconfig" foot-gun).
//!   2. A malformed presigned URL surfaces as `CellosError::ExportSink`, i.e.
//!      the public surface does not silently swallow config errors.
//!   3. A non-http(s) scheme is rejected — the sink does not silently fall
//!      through to a transport that has no retry policy.
//!   4. With `max_attempts = N` and `retry_backoff_ms = 0` against a refused
//!      TCP endpoint, `push()` returns `Err` in bounded wall-clock time. If
//!      the retry loop were unbounded the call would never return; the test
//!      timeout proves the budget is finite.
//!   5. A larger `max_attempts` still terminates in bounded time — i.e. the
//!      loop scales linearly with the configured budget rather than going
//!      exponential or unbounded.
//!   6. A missing artifact file is reported as an error from `push()` and is
//!      NOT retried as a transport failure (file read happens before the HTTP
//!      loop).
//!
//! What is intentionally NOT covered here (no in-process HTTP seam exists):
//!   - The 5xx-retry vs 4xx-terminal branch inside `should_retry_status` is
//!     a private fn and is only reachable via a real HTTP response. Adding a
//!     mock-HTTP dev-dep was explicitly out of scope. If/when the crate
//!     grows a `trait HttpTransport` seam, that branch should get a direct
//!     unit test alongside these behavioural ones.
//!   - The crate currently relies on `reqwest`'s default timeouts (no
//!     explicit per-request or total deadline). That gap is documented in
//!     the EXPORT-S3-RETRY report rather than asserted here.

use std::net::TcpListener;
use std::time::{Duration, Instant};

use cellos_core::ports::ExportSink;
use cellos_core::ExportArtifactMetadata;
use cellos_export_s3::PresignedS3ExportSink;

/// Hard upper bound for any single `push()` call in this file. Connect-refused
/// fails in microseconds on Linux; a multi-second budget here is generous and
/// only exists to fail loudly if the retry loop ever regresses to unbounded.
const PUSH_HARD_DEADLINE: Duration = Duration::from_secs(30);

/// Acquire a `127.0.0.1` port that is (almost certainly) refused. We bind a
/// listener on an ephemeral port and immediately drop it; reusing the port
/// before the kernel re-allocates it gives us a fast `ECONNREFUSED`. There is
/// a tiny theoretical race, but the test wraps the whole thing in
/// `PUSH_HARD_DEADLINE` so a regression is loud either way.
fn refused_port() -> u16 {
    let listener = TcpListener::bind("127.0.0.1:0").expect("retry-semantics: bind ephemeral port");
    let port = listener
        .local_addr()
        .expect("retry-semantics: read local_addr")
        .port();
    drop(listener);
    port
}

#[test]
fn budget_zero_attempts_rejected_at_construction() {
    // Lock down: a zero-attempt budget is not a valid configuration. This
    // prevents an "I disabled retry but actually disabled the upload" foot-gun
    // and ensures the budget is always >= 1.
    let err = match PresignedS3ExportSink::new(
        "https://bucket.s3.amazonaws.com/object",
        "cell",
        "bucket",
        None,
        None,
        None,
        0,
        0,
    ) {
        Ok(_) => panic!("max_attempts == 0 must be rejected"),
        Err(e) => e,
    };
    let msg = err.to_string();
    assert!(
        msg.contains("max_attempts"),
        "expected error to mention max_attempts, got: {msg}"
    );
}

#[test]
fn budget_one_attempt_constructs_ok() {
    // Lock down: the minimum valid budget (1 attempt, no retry) constructs
    // cleanly. This is the "retry disabled" configuration and must remain
    // representable.
    PresignedS3ExportSink::new(
        "https://bucket.s3.amazonaws.com/object",
        "cell",
        "bucket",
        None,
        None,
        None,
        1,
        0,
    )
    .expect("max_attempts == 1 must be accepted");
}

#[test]
fn malformed_url_surfaces_as_export_sink_error() {
    // Lock down: configuration errors from the public constructor are
    // returned as typed `CellosError::ExportSink`, not panicked or silently
    // swallowed.
    let err =
        match PresignedS3ExportSink::new("::not a url::", "cell", "bucket", None, None, None, 1, 0)
        {
            Ok(_) => panic!("malformed URL must be rejected"),
            Err(e) => e,
        };
    let msg = err.to_string().to_lowercase();
    assert!(
        msg.contains("url") || msg.contains("scheme"),
        "expected URL-related error, got: {err}"
    );
}

#[test]
fn non_http_scheme_rejected() {
    // Lock down: only http/https schemes are accepted. The retry policy is
    // tied to the HTTP PUT transport — falling through to e.g. ftp would
    // bypass the bounded-retry contract entirely.
    let err = match PresignedS3ExportSink::new(
        "ftp://example.com/object",
        "cell",
        "bucket",
        None,
        None,
        None,
        1,
        0,
    ) {
        Ok(_) => panic!("non-http scheme must be rejected"),
        Err(e) => e,
    };
    let msg = err.to_string();
    assert!(
        msg.contains("scheme") || msg.to_lowercase().contains("http"),
        "expected scheme error, got: {err}"
    );
}

#[tokio::test]
async fn push_against_refused_port_terminates_in_bounded_time() {
    // Lock down: with a finite `max_attempts` budget and zero backoff, a
    // persistent transport failure causes `push()` to return `Err` in
    // bounded wall-clock time. If the retry loop were unbounded this would
    // never return and the outer `tokio::time::timeout` would fail the test.
    let port = refused_port();
    let url = format!("http://127.0.0.1:{port}/object");

    // Stage an artifact body; we use a tempdir-style path under the OS temp
    // dir so the test does not depend on repo layout.
    let tmp = std::env::temp_dir().join(format!(
        "cellos-export-s3-retry-{}-{}.bin",
        std::process::id(),
        port
    ));
    std::fs::write(&tmp, b"retry-budget-fixture").expect("write fixture");
    // Ensure cleanup even on panic.
    struct Cleanup(std::path::PathBuf);
    impl Drop for Cleanup {
        fn drop(&mut self) {
            let _ = std::fs::remove_file(&self.0);
        }
    }
    let _cleanup = Cleanup(tmp.clone());

    let sink = PresignedS3ExportSink::new(
        url, "cell", "bucket", None, None, None, 3, // max_attempts
        0, // retry_backoff_ms — keep the test fast
    )
    .expect("construct sink");

    let metadata = ExportArtifactMetadata::default();
    let started = Instant::now();
    let result = tokio::time::timeout(
        PUSH_HARD_DEADLINE,
        sink.push(
            "artifact.bin",
            tmp.to_str().expect("utf8 tmp path"),
            &metadata,
        ),
    )
    .await
    .expect("retry loop must terminate within PUSH_HARD_DEADLINE");
    let elapsed = started.elapsed();

    let err = result.expect_err("connect-refused must propagate as Err after retries exhaust");
    // Lock down: the error is a typed `ExportSink` failure — not a panic and
    // not silently dropped.
    let msg = err.to_string().to_lowercase();
    assert!(
        msg.contains("s3 put") || msg.contains("connection") || msg.contains("refused"),
        "expected transport-failure error, got: {err}"
    );

    // Sanity: terminating well inside the hard deadline is the actual proof
    // that the budget is bounded.
    assert!(
        elapsed < PUSH_HARD_DEADLINE,
        "retry loop took {elapsed:?} which is >= hard deadline {PUSH_HARD_DEADLINE:?}"
    );
}

#[tokio::test]
async fn push_with_larger_budget_still_terminates() {
    // Lock down: the retry budget scales linearly — bumping `max_attempts`
    // from 3 to 10 must still terminate inside the hard deadline. This
    // catches regressions that would turn the loop unbounded only for
    // larger budgets (e.g. wrong loop bound or off-by-one in `attempt <
    // self.max_attempts`).
    let port = refused_port();
    let url = format!("http://127.0.0.1:{port}/object");

    let tmp = std::env::temp_dir().join(format!(
        "cellos-export-s3-retry-large-{}-{}.bin",
        std::process::id(),
        port
    ));
    std::fs::write(&tmp, b"retry-budget-fixture-large").expect("write fixture");
    struct Cleanup(std::path::PathBuf);
    impl Drop for Cleanup {
        fn drop(&mut self) {
            let _ = std::fs::remove_file(&self.0);
        }
    }
    let _cleanup = Cleanup(tmp.clone());

    let sink = PresignedS3ExportSink::new(url, "cell", "bucket", None, None, None, 10, 0)
        .expect("construct sink");

    let metadata = ExportArtifactMetadata::default();
    let result = tokio::time::timeout(
        PUSH_HARD_DEADLINE,
        sink.push(
            "artifact.bin",
            tmp.to_str().expect("utf8 tmp path"),
            &metadata,
        ),
    )
    .await
    .expect("retry loop must terminate within PUSH_HARD_DEADLINE");

    result.expect_err("connect-refused must propagate as Err after retries exhaust");
}

#[tokio::test]
async fn missing_artifact_file_returns_err_without_hanging() {
    // Lock down: a missing artifact path returns `Err` from `push()` rather
    // than being misclassified as a transport failure and retried. The file
    // read happens before the HTTP loop, so this should fail fast on the
    // first attempt regardless of `max_attempts`.
    let port = refused_port();
    let url = format!("http://127.0.0.1:{port}/object");

    let sink = PresignedS3ExportSink::new(url, "cell", "bucket", None, None, None, 5, 0)
        .expect("construct sink");

    let metadata = ExportArtifactMetadata::default();
    let result = tokio::time::timeout(
        PUSH_HARD_DEADLINE,
        sink.push(
            "artifact.bin",
            "/this/path/should/not/exist/cellos-export-s3-retry-test",
            &metadata,
        ),
    )
    .await
    .expect("read-error path must return promptly");

    let err = result.expect_err("missing artifact must produce Err");
    let msg = err.to_string();
    assert!(
        msg.contains("read artifact"),
        "expected file-read error, got: {err}"
    );
}
