oneio 0.27.0

//! S3 operations using rusty-s3 for signing and reqwest for HTTP transport.
//!
//! # Environment Variables
//!
//! Required:
//! - `AWS_ACCESS_KEY_ID`
//! - `AWS_SECRET_ACCESS_KEY`
//! - `AWS_REGION` - Use `"auto"` for Cloudflare R2
//! - `AWS_ENDPOINT` - e.g. `https://xxx.r2.cloudflarestorage.com`
//!
//! Optional:
//! - `AWS_SESSION_TOKEN` - Temporary session token
//! - `ONEIO_S3_CHUNK_SIZE` - Multipart part size in bytes (default: 8MB)
//! - `ONEIO_S3_MULTIPART_THRESHOLD` - File size threshold for multipart upload (default: 5MB)
//! - `ONEIO_S3_MAX_RETRIES` - Retry attempts after the initial request for transient transport errors (default: 3)
//! - `ONEIO_S3_RETRY_BACKOFF_MS` - Initial retry backoff in ms, doubles each attempt (default: 1000)
//! - `ONEIO_S3_MAX_CONCURRENCY` - Multipart parts uploaded in parallel (default: 4, max: 16)
//! - `ONEIO_S3_REQUEST_TIMEOUT_SECS` - Per-request timeout for upload requests (default: 300)
//!
//! # Upload Behavior
//!
//! Files smaller than the multipart threshold use a single PUT request.
//! Larger files are uploaded via multipart with auto-calculated part sizing
//! to stay within S3's 10,000 part limit, and with parts uploaded in parallel.
//! Peak memory for a multipart upload is `concurrency * part_size`.

pub mod config;

pub use config::{S3Config, S3Credentials};

use crate::OneIoError;
use hmac::{Hmac, Mac};
use percent_encoding::{percent_decode_str, utf8_percent_encode, AsciiSet, CONTROLS};
use quick_xml::{events::Event, Reader};
use reqwest::blocking::Response;
use rusty_s3::S3Action;
use sha2::{Digest, Sha256};
use std::io::{Read, Seek, SeekFrom};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Mutex, OnceLock};
use std::time::Duration;

type HmacSha256 = Hmac<Sha256>;

/// Multipart parts uploaded in parallel when the caller does not override it.
const DEFAULT_MULTIPART_CONCURRENCY: usize = 4;
/// Upper bound on multipart workers, so one upload cannot exhaust the caller.
const MAX_MULTIPART_CONCURRENCY: usize = 16;
/// Per-request timeout for S3 upload requests (single PUT, part, initiate, complete).
const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 300;

const COPY_SOURCE_ENCODE_SET: &AsciiSet = &CONTROLS
    .add(b':')
    .add(b'?')
    .add(b'#')
    .add(b'[')
    .add(b']')
    .add(b'@')
    .add(b'!')
    .add(b'$')
    .add(b'&')
    .add(b'\'')
    .add(b'(')
    .add(b')')
    .add(b'*')
    .add(b'+')
    .add(b',')
    .add(b';')
    .add(b'=')
    .add(b'"')
    .add(b' ')
    .add(b'<')
    .add(b'>')
    .add(b'%')
    .add(b'{')
    .add(b'}')
    .add(b'|')
    .add(b'\\')
    .add(b'^')
    .add(b'`');

const S3_QUERY_ENCODE_SET: &AsciiSet = &COPY_SOURCE_ENCODE_SET.add(b'/');

fn uses_path_style(config: &config::S3Config) -> bool {
    !config.endpoint.contains("amazonaws.com") || config.bucket.contains('.')
}

fn path_style_object_url(
    config: &config::S3Config,
    encoded_object_path: &str,
) -> Result<reqwest::Url, OneIoError> {
    format!(
        "{}/{bucket}/{encoded_object_path}",
        config.endpoint,
        bucket = config.bucket
    )
    .parse()
    .map_err(|e| OneIoError::NotSupported(format!("Invalid S3 endpoint: {e}")))
}

/// Restore a path-style bucket component lost by rusty-s3 for leading-slash keys.
///
/// rusty-s3 resolves object keys with `Url::join()`. A leading slash therefore
/// replaces the path-style bucket component before the action is signed. Rebuild
/// the object path and re-sign the presigned request with the original action
/// query parameters.
fn repair_leading_slash_action_url(
    mut url: reqwest::Url,
    config: &config::S3Config,
    key: &str,
    method: &str,
) -> Result<reqwest::Url, OneIoError> {
    if !key.starts_with('/') || !uses_path_style(config) {
        return Ok(url);
    }

    let encoded_object_path = url.path().to_string();
    let action_query = url.query().map(str::to_owned);
    url = path_style_object_url(config, &encoded_object_path)?;
    url.set_query(action_query.as_deref());

    let mut query: Vec<(String, String)> = url
        .query_pairs()
        .filter(|(name, _)| name != "X-Amz-Signature")
        .map(|(name, value)| (name.into_owned(), value.into_owned()))
        .collect();
    query.sort_unstable();

    let datetime = query
        .iter()
        .find(|(name, _)| name == "X-Amz-Date")
        .map(|(_, value)| value.as_str())
        .ok_or_else(|| OneIoError::NotSupported("Missing X-Amz-Date in S3 action".to_string()))?;
    let datestamp = datetime
        .get(..8)
        .ok_or_else(|| OneIoError::NotSupported("Malformed X-Amz-Date in S3 action".to_string()))?;

    let default_port = match url.scheme() {
        "https" => 443,
        "http" => 80,
        _ => 0,
    };
    let host = match url.port() {
        Some(port) if port != default_port => format!(
            "{}:{port}",
            url.host_str()
                .ok_or_else(|| OneIoError::NotSupported("Invalid URL: no host".to_string()))?
        ),
        _ => url
            .host_str()
            .ok_or_else(|| OneIoError::NotSupported("Invalid URL: no host".to_string()))?
            .to_string(),
    };

    let canonical_query = query
        .iter()
        .map(|(name, value)| {
            format!(
                "{}={}",
                utf8_percent_encode(name, S3_QUERY_ENCODE_SET),
                utf8_percent_encode(value, S3_QUERY_ENCODE_SET)
            )
        })
        .collect::<Vec<_>>()
        .join("&");
    let canonical_request = format!(
        "{method}\n{}\n{canonical_query}\nhost:{host}\n\nhost\nUNSIGNED-PAYLOAD",
        url.path()
    );
    let credential_scope = format!("{datestamp}/{}/s3/aws4_request", config.region);
    let string_to_sign = format!(
        "AWS4-HMAC-SHA256\n{datetime}\n{credential_scope}\n{}",
        hex::encode(Sha256::digest(canonical_request.as_bytes()))
    );
    let signing_key = derive_signing_key(&config.credentials.secret_key, datestamp, &config.region);
    let signature = hex::encode(hmac_sha256(&signing_key, string_to_sign.as_bytes()));

    url.set_query(Some(&format!(
        "{canonical_query}&X-Amz-Signature={signature}"
    )));

    Ok(url)
}

fn s3_object_url(config: &config::S3Config, key: &str) -> Result<reqwest::Url, OneIoError> {
    let url = config
        .rusty_bucket()?
        .object_url(key)
        .map_err(|e| OneIoError::NotSupported(format!("Invalid object key: {e}")))?;

    if key.starts_with('/') && uses_path_style(config) {
        path_style_object_url(config, url.path())
    } else {
        Ok(url)
    }
}

// Shared HTTP and retry configuration for S3 operations.
static S3_HTTP_CLIENT: OnceLock<reqwest::blocking::Client> = OnceLock::new();
static S3_RETRY_CONFIG: OnceLock<(u32, u64)> = OnceLock::new();

fn get_s3_client() -> &'static reqwest::blocking::Client {
    S3_HTTP_CLIENT.get_or_init(|| {
        #[cfg(feature = "rustls")]
        if let Err(e) = crate::crypto::ensure_default_provider() {
            eprintln!("Warning: failed to initialize rustls crypto provider: {e}");
        }

        let mut builder =
            reqwest::blocking::Client::builder().connect_timeout(Duration::from_secs(30));

        #[cfg(all(feature = "http", any(feature = "rustls", feature = "native-tls")))]
        {
            if let Ok(ca_bundle_path) = std::env::var("ONEIO_CA_BUNDLE") {
                if let Ok(pem) = std::fs::read(&ca_bundle_path) {
                    if let Ok(cert) = reqwest::Certificate::from_pem(&pem) {
                        builder = builder.add_root_certificate(cert);
                    }
                }
            }

            let accept_invalid = matches!(
                std::env::var("ONEIO_ACCEPT_INVALID_CERTS")
                    .unwrap_or_default()
                    .to_lowercase()
                    .as_str(),
                "true" | "yes" | "y" | "1"
            );
            builder = builder.danger_accept_invalid_certs(accept_invalid);
        }

        builder.build().expect("Failed to create S3 HTTP client")
    })
}

/// Metadata returned by s3_stats().
#[derive(Debug, Clone)]
pub struct S3ObjectMetadata {
    /// Content length in bytes.
    pub content_length: u64,
    /// Content type (MIME type), if available.
    pub content_type: Option<String>,
    /// Last modified timestamp, if available.
    pub last_modified: Option<String>,
    /// ETag of the object, if available.
    pub etag: Option<String>,
}

/// Bucket handle returned by s3_bucket().
#[derive(Debug, Clone)]
pub struct S3Bucket {
    /// Bucket name.
    pub name: String,
    /// Endpoint URL.
    pub endpoint: String,
    /// Region.
    pub region: String,
}

/// Checks if the necessary environment variables for AWS S3 are set.
pub fn s3_env_check() -> Result<(), OneIoError> {
    let _ = config::S3Config::from_env("test")?;
    Ok(())
}

/// Parse an S3 URL into a bucket and key.
pub fn s3_url_parse(path: &str) -> Result<(String, String), OneIoError> {
    let (_, remaining) = path
        .split_once("://")
        .ok_or_else(|| OneIoError::NotSupported(format!("Invalid S3 URL: {path}")))?;
    let (bucket, key) = remaining
        .split_once('/')
        .ok_or_else(|| OneIoError::NotSupported(format!("Invalid S3 URL: {path}")))?;
    if bucket.is_empty() || key.is_empty() {
        return Err(OneIoError::NotSupported(format!("Invalid S3 URL: {path}")));
    }
    Ok((bucket.to_string(), key.to_string()))
}

/// Creates an S3 bucket handle with the specified bucket name.
pub fn s3_bucket(name: &str) -> Result<S3Bucket, OneIoError> {
    let config = config::S3Config::from_env(name)?;
    Ok(S3Bucket {
        name: config.bucket,
        endpoint: config.endpoint,
        region: config.region,
    })
}

/// Reads a file from an S3 bucket and returns a boxed reader implementing `Read` trait.
pub fn s3_reader(bucket: &str, key: &str) -> Result<Box<dyn Read + Send>, OneIoError> {
    let config = config::S3Config::from_env(bucket)?;
    let bucket = config.rusty_bucket()?;
    let creds = config.rusty_credentials();
    let action = bucket.get_object(Some(&creds), key);
    let url = repair_leading_slash_action_url(action.sign(config.ttl), &config, key, "GET")?;
    let response = ensure_s3_success(get_s3_client().get(url).send()?)?;
    Ok(Box::new(response))
}

/// Downloads a file from an S3 bucket and saves it locally.
pub fn s3_download(bucket: &str, key: &str, file_path: &str) -> Result<(), OneIoError> {
    let mut reader = s3_reader(bucket, key)?;
    let mut writer = crate::get_writer_raw_impl(file_path)?;
    std::io::copy(&mut reader, &mut writer)?;
    Ok(())
}

/// Uploads a file to an S3 bucket at the specified path.
pub fn s3_upload(bucket: &str, key: &str, file_path: &str) -> Result<(), OneIoError> {
    // Early validation: check if file exists before attempting S3 operations
    if !std::path::Path::new(file_path).exists() {
        return Err(OneIoError::Io(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            format!("File not found: {file_path}"),
        )));
    }

    let metadata = std::fs::metadata(file_path)?;
    let size = metadata.len();

    let config = config::S3Config::from_env(bucket)?;

    if size < config.multipart_threshold {
        upload_single(&config, key, file_path)
    } else {
        upload_multipart(&config, key, file_path)
    }
}

fn upload_single(config: &config::S3Config, key: &str, file_path: &str) -> Result<(), OneIoError> {
    let bucket = config.rusty_bucket()?;
    let creds = config.rusty_credentials();

    let file = std::fs::File::open(file_path)?;
    let action = bucket.put_object(Some(&creds), key);
    let url = repair_leading_slash_action_url(action.sign(config.ttl), config, key, "PUT")?;
    ensure_s3_success(
        get_s3_client()
            .put(url)
            .timeout(upload_request_timeout())
            .body(file)
            .send()?,
    )?;
    Ok(())
}

fn calculate_chunk_size(file_size: u64, requested_chunk_size: u64) -> (u64, usize) {
    const MAX_PARTS: u64 = 10_000;
    const MIN_PART_SIZE: u64 = 5 * 1024 * 1024;

    #[allow(clippy::manual_div_ceil)]
    let chunk_size = {
        let required = (file_size + MAX_PARTS - 1) / MAX_PARTS;
        requested_chunk_size.max(required).max(MIN_PART_SIZE)
    };
    #[allow(clippy::manual_div_ceil)]
    let total_parts = ((file_size + chunk_size - 1) / chunk_size) as usize;

    (chunk_size, total_parts)
}

fn s3_retry_config() -> (u32, u64) {
    *S3_RETRY_CONFIG.get_or_init(|| {
        let max_retries = std::env::var("ONEIO_S3_MAX_RETRIES")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(3);
        let retry_backoff_ms = std::env::var("ONEIO_S3_RETRY_BACKOFF_MS")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(1000);
        (max_retries, retry_backoff_ms)
    })
}

/// Concurrency and request timeout for S3 uploads, read once from the
/// environment so repeated uploads in one process behave identically.
fn s3_upload_tuning() -> (usize, Duration) {
    static S3_UPLOAD_CONFIG: OnceLock<(usize, Duration)> = OnceLock::new();
    *S3_UPLOAD_CONFIG.get_or_init(|| {
        let concurrency = std::env::var("ONEIO_S3_MAX_CONCURRENCY")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(DEFAULT_MULTIPART_CONCURRENCY);
        let timeout_secs = std::env::var("ONEIO_S3_REQUEST_TIMEOUT_SECS")
            .ok()
            .and_then(|value| value.parse().ok())
            .filter(|value| *value > 0)
            .unwrap_or(DEFAULT_REQUEST_TIMEOUT_SECS);
        (concurrency, Duration::from_secs(timeout_secs))
    })
}

fn upload_request_timeout() -> Duration {
    s3_upload_tuning().1
}

/// Number of parts to keep in flight. Bounded by the configured value, the
/// per-upload ceiling and the number of parts there are to send.
fn multipart_concurrency(requested: usize, total_parts: usize) -> usize {
    requested
        .clamp(1, MAX_MULTIPART_CONCURRENCY)
        .min(total_parts.max(1))
}

/// Turn per-part ETag slots into the ordered part list for
/// CompleteMultipartUpload. A missing slot means a worker never succeeded.
fn collect_part_etags(slots: &[Option<String>]) -> Result<Vec<String>, OneIoError> {
    slots
        .iter()
        .enumerate()
        .map(|(index, slot)| {
            slot.clone().ok_or_else(|| {
                OneIoError::NotSupported(format!("Missing ETag for multipart part {}", index + 1))
            })
        })
        .collect()
}

/// Shared state of one multipart part pool: which part to take next, whether a
/// worker has already failed, and which error to report.
struct PartUploadState {
    next_part: AtomicUsize,
    stop: AtomicBool,
    first_error: Mutex<Option<OneIoError>>,
}

impl PartUploadState {
    fn new() -> Self {
        Self {
            next_part: AtomicUsize::new(0),
            stop: AtomicBool::new(false),
            first_error: Mutex::new(None),
        }
    }

    /// Claim the next part, or `None` once every part is taken or the pool has
    /// already failed.
    fn claim(&self, total_parts: usize) -> Option<usize> {
        if self.stop.load(Ordering::Relaxed) {
            return None;
        }
        let index = self.next_part.fetch_add(1, Ordering::Relaxed);
        (index < total_parts).then_some(index)
    }

    #[cfg(test)]
    fn cancelled(&self) -> bool {
        self.stop.load(Ordering::Relaxed)
    }

    /// Record a failure and stop the pool.
    ///
    /// Workers call this themselves: the coordinator joins them in creation
    /// order, so signalling from the join loop would let a failure in a later
    /// worker go unnoticed while the remaining parts are still uploaded.
    fn fail(&self, error: OneIoError) {
        let mut slot = self
            .first_error
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if slot.is_none() {
            *slot = Some(error);
        }
        self.stop.store(true, Ordering::Relaxed);
    }

    /// The first error reported by any worker, if one was.
    fn into_error(self) -> Option<OneIoError> {
        self.first_error
            .into_inner()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }
}

/// Execute an S3 request with retry on transient transport errors.
///
/// S3-compatible services (especially Cloudflare R2) occasionally reset
/// connections during large multipart uploads. Without retry, a single
/// transient failure on any part aborts the entire multipart upload,
/// wasting all successfully uploaded parts. This helper retries transport
/// errors (connection reset, timeout) with exponential backoff. HTTP status
/// errors (4xx/5xx) are returned immediately without retry — they arrive
/// as `Response`, not `Error`, and are handled by the caller.
fn send_with_retry<F>(request: F) -> Result<Response, OneIoError>
where
    F: Fn() -> Result<Response, reqwest::Error>,
{
    let (max_retries, mut backoff_ms) = s3_retry_config();

    for _attempt in 0..=max_retries {
        match request() {
            Ok(response) => return Ok(response),
            Err(e) if _attempt < max_retries && is_retryable_error(&e) => {
                std::thread::sleep(Duration::from_millis(backoff_ms));
                backoff_ms = backoff_ms.saturating_mul(2);
            }
            Err(e) => return Err(e.into()),
        }
    }
    unreachable!()
}

/// Determine if a reqwest error is likely transient and worth retrying.
///
/// Retryable: timeouts, connection resets, broken pipes, DNS failures.
/// Non-retryable: invalid URLs, TLS errors, HTTP status errors (which
/// arrive as `Response`, not `Error`).
fn is_retryable_error(err: &reqwest::Error) -> bool {
    if err.is_timeout() || err.is_connect() {
        return true;
    }
    // Body decode errors (connection reset mid-stream) are transient
    if err.is_body() || err.is_decode() {
        return true;
    }
    false
}

/// Single open handle for the multipart source, read under a lock.
///
/// Reopening the path in each worker would let an atomic replacement of the
/// file between two opens mix bytes from two versions into one object: the
/// per-part length checks still pass when both versions have the same size.
/// Handles cloned from one open would share a seek position instead, so reads
/// go through this lock rather than a raw file per worker.
struct PartSource {
    file: Mutex<std::fs::File>,
}

impl PartSource {
    fn open(path: &str) -> Result<Self, OneIoError> {
        Ok(Self {
            file: Mutex::new(std::fs::File::open(path)?),
        })
    }

    /// Size of the open file, not of whatever the path resolves to now.
    fn len(&self) -> Result<u64, OneIoError> {
        let file = self
            .file
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        Ok(file.metadata()?.len())
    }

    /// Read exactly `len` bytes at `offset` from the open file.
    fn read_range(&self, offset: u64, len: u64) -> Result<Vec<u8>, OneIoError> {
        let mut file = self
            .file
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let mut buf = Vec::with_capacity(len as usize);
        file.seek(SeekFrom::Start(offset))?;
        file.by_ref().take(len).read_to_end(&mut buf)?;
        if buf.len() as u64 != len {
            return Err(OneIoError::Io(std::io::Error::new(
                std::io::ErrorKind::UnexpectedEof,
                format!(
                    "multipart part read was short: expected {len} bytes, got {}",
                    buf.len()
                ),
            )));
        }
        Ok(buf)
    }
}

/// Upload a single multipart part with retry, avoiding unnecessary clones.
///
/// On the first attempt, `body` is moved into the request with zero copy.
/// On retry (transient transport error), the part bytes are re-read from
/// `source` at `offset` to reconstruct the request body.
fn upload_part_with_retry(
    url: &reqwest::Url,
    body: Vec<u8>,
    source: &PartSource,
    offset: u64,
    part_len: u64,
) -> Result<Response, OneIoError> {
    let (max_retries, mut backoff_ms) = s3_retry_config();

    // First attempt: move the body, no clone.
    // Retry attempts: re-read the same range from the open source.
    let mut body = Some(body);

    for _attempt in 0..=max_retries {
        let request_body = match body.take() {
            Some(b) => b,
            None => source.read_range(offset, part_len)?,
        };

        match get_s3_client()
            .put(url.clone())
            .timeout(upload_request_timeout())
            .body(request_body)
            .send()
        {
            Ok(response) => return Ok(response),
            Err(e) if _attempt < max_retries && is_retryable_error(&e) => {
                std::thread::sleep(Duration::from_millis(backoff_ms));
                backoff_ms = backoff_ms.saturating_mul(2);
            }
            Err(e) => return Err(e.into()),
        }
    }
    unreachable!()
}

fn upload_multipart(
    config: &config::S3Config,
    key: &str,
    file_path: &str,
) -> Result<(), OneIoError> {
    // One open handle covers the part ranges, the total size and the retry
    // reads, so every part comes from the same file version.
    let source = PartSource::open(file_path)?;
    let size = source.len()?;
    let (chunk_size, total_parts) = calculate_chunk_size(size, config.multipart_chunk_size);

    let bucket = config.rusty_bucket()?;
    let creds = config.rusty_credentials();

    // 1. Initiate multipart upload
    let action = bucket.create_multipart_upload(Some(&creds), key);
    let url = repair_leading_slash_action_url(action.sign(config.ttl), config, key, "POST")?;
    let response = ensure_s3_success(send_with_retry(|| {
        get_s3_client()
            .post(url.clone())
            .timeout(upload_request_timeout())
            .send()
    })?)?;
    let init_response =
        rusty_s3::actions::CreateMultipartUpload::parse_response(response.text()?.as_str())
            .map_err(|e| OneIoError::Network(Box::new(e)))?;
    let upload_id = init_response.upload_id().to_string();

    // 2. Upload parts concurrently. Part ranges are read from the one open
    //    source, serialized by its lock, and each part is then uploaded after
    //    the read returns. One stalled part therefore no longer stops the whole
    //    upload: the other workers keep filling the pipe with parts they have
    //    already read.
    let concurrency = multipart_concurrency(s3_upload_tuning().0, total_parts);
    let state = PartUploadState::new();
    let slots: Mutex<Vec<Option<String>>> = Mutex::new(vec![None; total_parts]);

    std::thread::scope(|scope| {
        let mut handles = Vec::with_capacity(concurrency);
        for _ in 0..concurrency {
            // `Scope::spawn` panics when the OS refuses a thread, which would
            // unwind past the abort path below and leave the multipart upload
            // behind. Report the failure through the pool state instead.
            let spawned = std::thread::Builder::new()
                .name("oneio-s3-part".to_string())
                .spawn_scoped(scope, || {
                    let outcome = (|| -> Result<(), OneIoError> {
                        while let Some(index) = state.claim(total_parts) {
                            let offset = index as u64 * chunk_size;
                            let part_len = (size - offset).min(chunk_size);
                            let body = source.read_range(offset, part_len)?;

                            let action = bucket.upload_part(
                                Some(&creds),
                                key,
                                (index + 1) as u16,
                                &upload_id,
                            );

                            let url = repair_leading_slash_action_url(
                                action.sign(config.ttl),
                                config,
                                key,
                                "PUT",
                            )?;

                            // On retry this re-reads the same range from the
                            // shared source; the read is taken under the source
                            // lock, outside the network wait.
                            let response = ensure_s3_success(upload_part_with_retry(
                                &url, body, &source, offset, part_len,
                            )?)?;
                            let etag = extract_etag(response.headers()).ok_or_else(|| {
                                OneIoError::NotSupported(
                                    "Missing ETag in UploadPart response".into(),
                                )
                            })?;
                            slots
                                .lock()
                                .unwrap_or_else(|poisoned| poisoned.into_inner())[index] =
                                Some(etag);
                        }
                        Ok(())
                    })();
                    if let Err(e) = outcome {
                        state.fail(e);
                    }
                });

            match spawned {
                Ok(handle) => handles.push(handle),
                Err(e) => {
                    // Keep whatever workers already started; they stop at the
                    // next part boundary because `fail` sets the stop flag.
                    state.fail(OneIoError::Io(e));
                    break;
                }
            }
        }

        // Workers stop themselves through `state.fail`; joining is only needed
        // to surface a panic, which cannot report through that path.
        for handle in handles {
            if handle.join().is_err() {
                state.fail(OneIoError::NotSupported(
                    "multipart part worker panicked".to_string(),
                ));
            }
        }
    });

    if let Some(e) = state.into_error() {
        abort_multipart_upload(&bucket, &creds, config, key, &upload_id);
        return Err(e);
    }

    let parts = collect_part_etags(
        &slots
            .into_inner()
            .unwrap_or_else(|poisoned| poisoned.into_inner()),
    )?;

    // 3. Complete multipart upload
    let action = bucket.complete_multipart_upload(
        Some(&creds),
        key,
        &upload_id,
        parts.iter().map(|s| s.as_str()),
    );
    let url = repair_leading_slash_action_url(action.sign(config.ttl), config, key, "POST")?;
    let body = action.body();
    let response = match send_with_retry(|| {
        get_s3_client()
            .post(url.clone())
            .timeout(upload_request_timeout())
            .header("content-type", "application/xml")
            .body(body.clone())
            .send()
    }) {
        Ok(response) => response,
        Err(e) => {
            abort_multipart_upload(&bucket, &creds, config, key, &upload_id);
            return Err(e);
        }
    };

    // CompleteMultipartUpload can return 200 OK with an embedded <Error> body.
    // Validate HTTP status first, then parse the body to confirm success.
    if !response.status().is_success() {
        abort_multipart_upload(&bucket, &creds, config, key, &upload_id);
        return Err(s3_error_from_response(response));
    }
    let complete_body = response.text().unwrap_or_default();
    if let Some(parsed) = parse_s3_error_xml(&complete_body) {
        abort_multipart_upload(&bucket, &creds, config, key, &upload_id);
        return Err(map_parsed_s3_error(200, parsed));
    }

    Ok(())
}

fn abort_multipart_upload(
    bucket: &rusty_s3::Bucket,
    creds: &rusty_s3::Credentials,
    config: &config::S3Config,
    key: &str,
    upload_id: &str,
) {
    let action = bucket.abort_multipart_upload(Some(creds), key, upload_id);
    if let Ok(url) = repair_leading_slash_action_url(action.sign(config.ttl), config, key, "DELETE")
    {
        let _ = get_s3_client().delete(url).send();
    }
}

/// Copies an object within the same S3 bucket.
///
/// Uses AWS Signature V4 with Authorization header (not presigned URL).
/// This is required by some S3-compatible services like Cloudflare R2
/// that reject presigned URLs for CopyObject operations.
///
/// TODO: Upstream CopyObject support to rusty-s3 and remove manual signing.
/// rusty-s3 v0.9 does not provide a CopyObject action or header-based signing.
///
/// # Limitations
///
/// Single-request copy is limited to 5 GiB. For larger objects, use
/// multipart copy (not yet implemented).
pub fn s3_copy(bucket: &str, src_key: &str, dst_key: &str) -> Result<(), OneIoError> {
    let config = config::S3Config::from_env(bucket)?;
    // Get the base URL for the destination object
    let url = s3_object_url(&config, dst_key)?;
    let url_str = url.as_str();

    // Extract host and path for signing, including non-default port
    let default_port = match url.scheme() {
        "https" => 443,
        "http" => 80,
        _ => 0,
    };
    let host = match url.port() {
        Some(port) if port != default_port => format!(
            "{}:{}",
            url.host_str()
                .ok_or_else(|| OneIoError::NotSupported("Invalid URL: no host".to_string()))?,
            port
        ),
        _ => url
            .host_str()
            .ok_or_else(|| OneIoError::NotSupported("Invalid URL: no host".to_string()))?
            .to_string(),
    };
    let canonical_uri = url.path();

    // Build x-amz-copy-source header value (/bucket/key)
    let copy_source = format!(
        "/{}/{}",
        config.bucket,
        utf8_percent_encode(src_key, COPY_SOURCE_ENCODE_SET)
    );

    // Generate timestamp and datestamp
    let now = std::time::SystemTime::now();
    let datetime = format_timestamp(now);
    let datestamp = datetime[..8].to_string();

    // Empty payload hash for COPY (no request body)
    let payload_hash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";

    // Build signed headers list (alphabetical order for canonical request)
    let host_str = host.as_str();
    let mut signed_headers = vec![
        ("host", host_str),
        ("x-amz-content-sha256", payload_hash),
        ("x-amz-copy-source", copy_source.as_str()),
        ("x-amz-date", datetime.as_str()),
    ];

    // Add session token if present
    if let Some(ref token) = config.credentials.session_token {
        signed_headers.push(("x-amz-security-token", token.as_str()));
    }

    let signed_headers_str = signed_headers
        .iter()
        .map(|(k, _)| *k)
        .collect::<Vec<_>>()
        .join(";");

    // Build canonical headers string
    let canonical_headers = signed_headers
        .iter()
        .map(|(k, v)| format!("{}:{}\n", k.to_lowercase(), v))
        .collect::<String>();

    // Build canonical request (empty query string for header-based auth)
    let canonical_request = format!(
        "PUT\n{}\n\n{}\n{}\n{}",
        canonical_uri, canonical_headers, signed_headers_str, payload_hash
    );

    // Build string to sign
    let credential_scope = format!("{}/{}/s3/aws4_request", datestamp, config.region);
    let string_to_sign = format!(
        "AWS4-HMAC-SHA256\n{}\n{}\n{}",
        datetime,
        credential_scope,
        hex::encode(Sha256::digest(canonical_request.as_bytes()))
    );

    // Calculate signature
    let signing_key =
        derive_signing_key(&config.credentials.secret_key, &datestamp, &config.region);
    let signature = hex::encode(hmac_sha256(&signing_key, string_to_sign.as_bytes()));

    // Build Authorization header
    let authorization = format!(
        "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
        config.credentials.access_key, credential_scope, signed_headers_str, signature
    );

    // Build and send request
    let mut request_builder = get_s3_client()
        .put(url_str)
        .header("host", host_str)
        .header("x-amz-date", datetime)
        .header("x-amz-content-sha256", payload_hash)
        .header("x-amz-copy-source", copy_source)
        .header("Authorization", authorization);

    // Add session token if present
    if let Some(token) = &config.credentials.session_token {
        request_builder = request_builder.header("x-amz-security-token", token);
    }

    let response = request_builder.send()?;

    // CopyObject can return 200 OK with an embedded <Error> body.
    // Validate HTTP status first, then parse the body to confirm success.
    if !response.status().is_success() {
        return Err(s3_error_from_response(response));
    }
    let body = response.text().unwrap_or_default();
    if let Some(parsed) = parse_s3_error_xml(&body) {
        return Err(map_parsed_s3_error(200, parsed));
    }

    Ok(())
}

/// Format system time as ISO 8601 timestamp (YYYYMMDD'T'HHMMSS'Z').
fn format_timestamp(time: std::time::SystemTime) -> String {
    let duration = time
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default();
    let secs = duration.as_secs();

    // Simple UTC conversion (no leap second handling needed for AWS SigV4)
    let days_since_epoch = secs / 86400;
    let seconds_of_day = secs % 86400;

    let (year, month, day) = epoch_days_to_ymd(days_since_epoch);
    let hour = (seconds_of_day / 3600) % 24;
    let minute = (seconds_of_day / 60) % 60;
    let second = seconds_of_day % 60;

    format!(
        "{:04}{:02}{:02}T{:02}{:02}{:02}Z",
        year, month, day, hour, minute, second
    )
}

/// Convert days since Unix epoch to year, month, day.
fn epoch_days_to_ymd(mut days: u64) -> (u32, u32, u32) {
    let mut year = 1970u32;
    loop {
        let days_in_year = if is_leap_year(year) { 366 } else { 365 };
        if days < days_in_year {
            break;
        }
        days -= days_in_year;
        year += 1;
    }

    let month_lengths = if is_leap_year(year) {
        [31u64, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    } else {
        [31u64, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    };

    let mut month = 1u32;
    for &len in &month_lengths {
        if days < len {
            break;
        }
        days -= len;
        month += 1;
    }

    (year, month, days as u32 + 1)
}

/// Check if a year is a leap year.
fn is_leap_year(year: u32) -> bool {
    year.is_multiple_of(4) && (!year.is_multiple_of(100) || year.is_multiple_of(400))
}

/// Derive AWS Signature V4 signing key.
fn derive_signing_key(secret_key: &str, datestamp: &str, region: &str) -> Vec<u8> {
    let k_date = hmac_sha256(
        format!("AWS4{}", secret_key).as_bytes(),
        datestamp.as_bytes(),
    );
    let k_region = hmac_sha256(&k_date, region.as_bytes());
    let k_service = hmac_sha256(&k_region, b"s3");
    hmac_sha256(&k_service, b"aws4_request")
}

/// Compute HMAC-SHA256.
fn hmac_sha256(key: &[u8], data: &[u8]) -> Vec<u8> {
    let mut mac = HmacSha256::new_from_slice(key).expect("HMAC accepts any key length");
    mac.update(data);
    mac.finalize().into_bytes().to_vec()
}

/// Deletes an object from an S3 bucket.
pub fn s3_delete(bucket: &str, key: &str) -> Result<(), OneIoError> {
    let config = config::S3Config::from_env(bucket)?;
    let bucket_obj = config.rusty_bucket()?;
    let creds = config.rusty_credentials();
    let action = bucket_obj.delete_object(Some(&creds), key);
    let url = repair_leading_slash_action_url(action.sign(config.ttl), &config, key, "DELETE")?;
    ensure_s3_success(get_s3_client().delete(url).send()?)?;
    Ok(())
}

/// Perform a HEAD request for an S3 object and return the raw response.
fn s3_head_object(bucket: &str, key: &str) -> Result<reqwest::blocking::Response, OneIoError> {
    let config = config::S3Config::from_env(bucket)?;
    let bucket_obj = config.rusty_bucket()?;
    let creds = config.rusty_credentials();
    let action = bucket_obj.head_object(Some(&creds), key);
    let url = repair_leading_slash_action_url(action.sign(config.ttl), &config, key, "HEAD")?;
    Ok(get_s3_client().head(url).send()?)
}

/// Retrieves the head object result for a given bucket and path in Amazon S3.
pub fn s3_stats(bucket: &str, key: &str) -> Result<S3ObjectMetadata, OneIoError> {
    let response = s3_head_object(bucket, key)?;

    if response.status().is_success() {
        let content_length = response
            .headers()
            .get("content-length")
            .and_then(|v| v.to_str().ok())
            .and_then(|s| s.parse().ok())
            .ok_or_else(|| {
                OneIoError::NotSupported(
                    "Missing or invalid content-length header in S3 response".to_string(),
                )
            })?;
        let content_type = response
            .headers()
            .get("content-type")
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());
        let last_modified = response
            .headers()
            .get("last-modified")
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());
        let etag = extract_etag(response.headers());

        Ok(S3ObjectMetadata {
            content_length,
            content_type,
            last_modified,
            etag,
        })
    } else {
        Err(s3_error_from_response(response))
    }
}

/// Check if a file exists in an S3 bucket.
pub fn s3_exists(bucket: &str, key: &str) -> Result<bool, OneIoError> {
    let response = s3_head_object(bucket, key)?;
    match response.status().as_u16() {
        200..=299 => Ok(true),
        404 => Ok(false),
        _ => Err(s3_error_from_response(response)),
    }
}

/// Lists objects in the specified Amazon S3 bucket with given prefix and delimiter.
pub fn s3_list(
    bucket: &str,
    prefix: &str,
    delimiter: Option<String>,
    dirs: bool,
) -> Result<Vec<String>, OneIoError> {
    let config = config::S3Config::from_env(bucket)?;
    let bucket_obj = config.rusty_bucket()?;
    let creds = config.rusty_credentials();

    let fixed_delimiter = match dirs && delimiter.is_none() {
        true => Some("/"),
        false => delimiter.as_deref(),
    };

    let mut result = Vec::new();
    let mut continuation_token: Option<String> = None;

    loop {
        let mut action = bucket_obj.list_objects_v2(Some(&creds));
        action.with_prefix(prefix);
        if let Some(delim) = fixed_delimiter {
            action.with_delimiter(delim);
        }
        if let Some(token) = &continuation_token {
            action.with_continuation_token(token);
        }

        let url = action.sign(config.ttl);
        let response = ensure_s3_success(get_s3_client().get(url).send()?)?;

        let parsed = rusty_s3::actions::ListObjectsV2::parse_response(response.text()?.as_str())
            .map_err(|e| OneIoError::Network(Box::new(e)))?;

        if dirs {
            result.extend(
                parsed
                    .common_prefixes
                    .into_iter()
                    .map(|p| decode_s3_path(&p.prefix)),
            );
        } else {
            result.extend(parsed.contents.into_iter().map(|c| decode_s3_path(&c.key)));
        }

        match parsed.next_continuation_token {
            Some(token) => continuation_token = Some(token),
            None => break,
        }
    }

    Ok(result)
}

/// Check an S3 HTTP response for errors and preserve the response body for callers.
fn ensure_s3_success(response: Response) -> Result<Response, OneIoError> {
    if response.status().is_success() {
        Ok(response)
    } else {
        Err(s3_error_from_response(response))
    }
}

/// Extract ETag from response headers.
fn extract_etag(headers: &reqwest::header::HeaderMap) -> Option<String> {
    headers
        .get(reqwest::header::ETAG)
        .and_then(|v| v.to_str().ok())
        .map(|s| s.trim_matches('"').to_string())
}

fn decode_s3_path(path: &str) -> String {
    percent_decode_str(path).decode_utf8_lossy().into_owned()
}

#[derive(Debug, Default)]
struct ParsedS3Error {
    code: Option<String>,
    message: Option<String>,
    key: Option<String>,
    bucket_name: Option<String>,
}

fn s3_error_from_response(response: Response) -> OneIoError {
    let status = response.status().as_u16();
    let body_text = response.text().unwrap_or_default();

    if let Some(parsed) = parse_s3_error_xml(&body_text) {
        return map_parsed_s3_error(status, parsed);
    }

    match status {
        404 => OneIoError::Status {
            service: "s3",
            code: 404,
            message: Some("Object not found".to_string()),
        },
        403 => OneIoError::Status {
            service: "s3",
            code: 403,
            message: Some("Access denied".to_string()),
        },
        code => OneIoError::Status {
            service: "s3",
            code,
            message: None,
        },
    }
}

fn map_parsed_s3_error(status: u16, parsed: ParsedS3Error) -> OneIoError {
    let code = parsed.code.unwrap_or_else(|| format!("S3Status{status}"));
    let message = parsed.message.unwrap_or_default();
    let key = parsed.key.unwrap_or_default();
    let bucket = parsed.bucket_name.unwrap_or_default();

    let detail = match code.as_str() {
        "NoSuchKey" if !key.is_empty() => format!("Object not found: {key}"),
        "NoSuchKey" => "Object not found".to_string(),
        "NoSuchBucket" if !bucket.is_empty() => format!("Bucket not found: {bucket}"),
        "NoSuchBucket" => "Bucket not found".to_string(),
        "AccessDenied" => format!("Access denied: {message}"),
        "InvalidAccessKeyId" | "SignatureDoesNotMatch" => format!("{code}: {message}"),
        _ => format!("{code}: {message}"),
    };

    OneIoError::Status {
        service: "s3",
        code: status,
        message: Some(detail),
    }
}

fn parse_s3_error_xml(body: &str) -> Option<ParsedS3Error> {
    let mut reader = Reader::from_str(body);
    reader.config_mut().trim_text(true);

    let mut parsed = ParsedS3Error::default();
    let mut current_field: Option<&[u8]> = None;

    loop {
        match reader.read_event() {
            Ok(Event::Start(e)) => {
                current_field = Some(match e.name().as_ref() {
                    b"Code" => b"Code",
                    b"Message" => b"Message",
                    b"Key" => b"Key",
                    b"BucketName" => b"BucketName",
                    _ => b"",
                });
            }
            Ok(Event::Text(e)) => {
                let Some(field) = current_field.take() else {
                    continue;
                };
                if field.is_empty() {
                    continue;
                }

                let value = match e.decode() {
                    Ok(value) => value.into_owned(),
                    Err(_) => return None,
                };

                match field {
                    b"Code" => parsed.code = Some(value),
                    b"Message" => parsed.message = Some(value),
                    b"Key" => parsed.key = Some(value),
                    b"BucketName" => parsed.bucket_name = Some(value),
                    _ => {}
                }
            }
            Ok(Event::End(_)) => current_field = None,
            Ok(Event::Eof) => break,
            Err(_) => return None,
            _ => {}
        }
    }

    if parsed.code.is_some() || parsed.message.is_some() {
        Some(parsed)
    } else {
        None
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn path_style_test_config() -> config::S3Config {
        config::S3Config {
            bucket: "test-bucket".to_string(),
            credentials: config::S3Credentials {
                access_key: "test-access-key".to_string(),
                secret_key: "test-secret-key".to_string(),
                session_token: None,
            },
            endpoint: "https://s3.example.test/base".to_string(),
            region: "us-east-1".to_string(),
            ttl: Duration::from_secs(60),
            multipart_chunk_size: 8 * 1024 * 1024,
            multipart_threshold: 5 * 1024 * 1024,
        }
    }

    #[test]
    fn test_leading_slash_path_style_urls_preserve_bucket_and_key() {
        let config = path_style_test_config();
        let bucket = config.rusty_bucket().unwrap();
        let creds = config.rusty_credentials();
        let action = bucket.get_object(Some(&creds), "/folder/file name.txt");
        let action_url = action.sign(config.ttl);

        let repaired =
            repair_leading_slash_action_url(action_url, &config, "/folder/file name.txt", "GET")
                .unwrap();
        assert_eq!(repaired.path(), "/base/test-bucket//folder/file%20name.txt");

        let copy_url = s3_object_url(&config, "/folder/file name.txt").unwrap();
        assert_eq!(copy_url.path(), "/base/test-bucket//folder/file%20name.txt");
    }

    #[test]
    fn test_s3_url_parse() {
        const S3_URL: &str = "s3://test-bucket/test-path/test-file.txt";
        let (bucket, path) = s3_url_parse(S3_URL).unwrap();
        assert_eq!(bucket, "test-bucket");
        assert_eq!(path, "test-path/test-file.txt");

        const NON_S3_URL: &str = "s3:/test-bucket";
        assert!(s3_url_parse(NON_S3_URL).is_err());
    }

    #[test]
    fn test_s3_upload_nonexistent_file_early_validation() {
        let non_existent_file = "/tmp/oneio_test_nonexistent_file_12345.txt";
        let _ = std::fs::remove_file(non_existent_file);
        assert!(!std::path::Path::new(non_existent_file).exists());

        let start = std::time::Instant::now();
        match s3_upload("test-bucket", "test-path", non_existent_file) {
            Ok(_) => panic!("Upload should have failed for non-existent file"),
            Err(OneIoError::Io(e)) => {
                let duration = start.elapsed();
                assert!(
                    duration < std::time::Duration::from_millis(100),
                    "Early validation should be instant"
                );
                assert_eq!(e.kind(), std::io::ErrorKind::NotFound);
            }
            Err(_) => {
                let duration = start.elapsed();
                assert!(duration < std::time::Duration::from_secs(1));
            }
        }
    }

    #[test]
    fn test_extract_etag() {
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            "etag",
            reqwest::header::HeaderValue::from_static("\"abc123\""),
        );
        assert_eq!(extract_etag(&headers), Some("abc123".to_string()));

        let mut headers2 = reqwest::header::HeaderMap::new();
        headers2.insert(
            "ETag",
            reqwest::header::HeaderValue::from_static("\"def456\""),
        );
        assert_eq!(extract_etag(&headers2), Some("def456".to_string()));
    }

    #[test]
    fn test_decode_s3_path() {
        assert_eq!(
            decode_s3_path("test%2Fpath%20file.txt"),
            "test/path file.txt"
        );
    }

    #[test]
    fn test_parse_s3_error_xml() {
        let parsed = parse_s3_error_xml(
            r#"<?xml version="1.0" encoding="UTF-8"?>
            <Error>
              <Code>NoSuchKey</Code>
              <Message>The specified key does not exist.</Message>
              <Key>test-file.txt</Key>
            </Error>"#,
        )
        .unwrap();

        assert_eq!(parsed.code.as_deref(), Some("NoSuchKey"));
        assert_eq!(
            parsed.message.as_deref(),
            Some("The specified key does not exist.")
        );
        assert_eq!(parsed.key.as_deref(), Some("test-file.txt"));
    }

    #[test]
    fn test_calculate_chunk_size() {
        let chunk_size = 8 * 1024 * 1024; // 8MB default

        // 0 bytes -> 8MB default chunk (default > min), 0 parts
        let (cs, tp) = calculate_chunk_size(0, chunk_size);
        assert_eq!(cs, 8 * 1024 * 1024);
        assert_eq!(tp, 0);

        // 1 byte -> 8MB chunk, 1 part
        let (cs, tp) = calculate_chunk_size(1, chunk_size);
        assert_eq!(cs, 8 * 1024 * 1024);
        assert_eq!(tp, 1);

        // 5MB - 1 -> 8MB chunk, 1 part
        let (cs, tp) = calculate_chunk_size(5 * 1024 * 1024 - 1, chunk_size);
        assert_eq!(cs, 8 * 1024 * 1024);
        assert_eq!(tp, 1);

        // Exactly 5MB -> 8MB chunk, 1 part
        let (cs, tp) = calculate_chunk_size(5 * 1024 * 1024, chunk_size);
        assert_eq!(cs, 8 * 1024 * 1024);
        assert_eq!(tp, 1);

        // 5MB + 1 -> 8MB chunk (default), 1 part
        let (cs, tp) = calculate_chunk_size(5 * 1024 * 1024 + 1, chunk_size);
        assert_eq!(cs, 8 * 1024 * 1024);
        assert_eq!(tp, 1);

        // 10MB -> 8MB chunk, 2 parts
        let (cs, tp) = calculate_chunk_size(10 * 1024 * 1024, chunk_size);
        assert_eq!(cs, 8 * 1024 * 1024);
        assert_eq!(tp, 2);

        // 80MB -> 8MB chunk, 10 parts
        let (cs, tp) = calculate_chunk_size(80 * 1024 * 1024, chunk_size);
        assert_eq!(cs, 8 * 1024 * 1024);
        assert_eq!(tp, 10);

        // Very large file: 100GB -> chunk size auto-increases to stay under 10,000 parts
        let hundred_gb = 100u64 * 1024 * 1024 * 1024;
        let (cs, tp) = calculate_chunk_size(hundred_gb, chunk_size);
        assert!(tp <= 10_000);
        assert!(cs >= 8 * 1024 * 1024);
        assert_eq!(tp, ((hundred_gb + cs - 1) / cs) as usize);
    }

    #[test]
    fn test_multipart_concurrency_bounds() {
        // Requested value is used when there is enough work.
        assert_eq!(multipart_concurrency(4, 100), 4);
        // Never more workers than parts.
        assert_eq!(multipart_concurrency(4, 2), 2);
        // At least one worker, even for a single-part upload.
        assert_eq!(multipart_concurrency(4, 1), 1);
        assert_eq!(multipart_concurrency(0, 10), 1);
        // Never above the per-upload ceiling.
        assert_eq!(
            multipart_concurrency(1_000, 10_000),
            MAX_MULTIPART_CONCURRENCY
        );
    }

    #[test]
    fn test_collect_part_etags_preserves_part_order() {
        let slots = vec![
            Some("etag-1".to_string()),
            Some("etag-2".to_string()),
            Some("etag-3".to_string()),
        ];
        assert_eq!(
            collect_part_etags(&slots).map_err(|e| e.to_string()),
            Ok(vec![
                "etag-1".to_string(),
                "etag-2".to_string(),
                "etag-3".to_string()
            ])
        );
    }

    #[test]
    fn test_collect_part_etags_rejects_missing_part() {
        let slots = vec![Some("etag-1".to_string()), None];
        match collect_part_etags(&slots) {
            Ok(parts) => panic!("expected a missing-part error, got {parts:?}"),
            Err(crate::OneIoError::NotSupported(message)) => {
                assert_eq!(message, "Missing ETag for multipart part 2");
            }
            Err(e) => panic!("unexpected error: {e}"),
        }
    }

    fn temp_path(name: &str) -> std::path::PathBuf {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        std::env::temp_dir().join(format!("oneio-{}-{nanos}-{name}", std::process::id()))
    }

    #[test]
    fn part_source_reads_exact_ranges() {
        let path = temp_path("ranges.bin");
        std::fs::write(&path, b"0123456789abcdef").unwrap();
        let source = PartSource::open(path.to_str().unwrap()).unwrap();

        assert_eq!(source.len().unwrap(), 16);
        assert_eq!(source.read_range(0, 4).unwrap(), b"0123".to_vec());
        assert_eq!(source.read_range(10, 6).unwrap(), b"abcdef".to_vec());
        // A range past the end is an error, not a short body.
        assert!(source.read_range(10, 7).is_err());

        let _ = std::fs::remove_file(&path);
    }

    /// A rotating writer replaces the path atomically; the upload must keep
    /// reading the file version it opened, not whichever version the path
    /// resolves to now. Equal sizes make a mixed upload otherwise invisible.
    #[cfg(unix)]
    #[test]
    fn part_source_keeps_the_opened_file_after_path_replacement() {
        let path = temp_path("replace.bin");
        let replacement = temp_path("replacement.bin");
        std::fs::write(&path, b"AAAAAAAAAAAA").unwrap();
        std::fs::write(&replacement, b"BBBBBBBBBBBB").unwrap();

        let source = PartSource::open(path.to_str().unwrap()).unwrap();
        std::fs::rename(&replacement, &path).unwrap();

        assert_eq!(source.len().unwrap(), 12);
        assert_eq!(source.read_range(0, 12).unwrap(), b"AAAAAAAAAAAA".to_vec());

        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn part_state_claims_each_part_once() {
        let state = PartUploadState::new();
        let claimed: Vec<usize> = std::iter::from_fn(|| state.claim(3)).collect();
        assert_eq!(claimed, vec![0, 1, 2]);
        assert_eq!(state.claim(3), None);
    }

    #[test]
    fn part_state_keeps_first_failure_and_stops_claims() {
        let state = PartUploadState::new();
        assert_eq!(state.claim(10), Some(0));
        state.fail(crate::OneIoError::NotSupported("part 2 failed".into()));
        state.fail(crate::OneIoError::NotSupported("part 5 failed".into()));
        assert!(state.cancelled());
        assert_eq!(state.claim(10), None);
        match state.into_error() {
            Some(crate::OneIoError::NotSupported(message)) => assert_eq!(message, "part 2 failed"),
            other => panic!("expected the first failure, got {other:?}"),
        }
    }

    /// The coordinator joins workers in creation order, so a failing worker has
    /// to publish cancellation itself. Otherwise a failure in a later worker
    /// waits behind a slow earlier worker while the remaining parts keep
    /// uploading.
    #[test]
    fn part_state_failure_is_visible_while_another_worker_runs() {
        let state = PartUploadState::new();
        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();

        std::thread::scope(|scope| {
            let blocked = scope.spawn(move || {
                let _ = release_rx.recv_timeout(std::time::Duration::from_secs(10));
            });
            let failing = scope.spawn(|| {
                state.fail(crate::OneIoError::NotSupported("part 2 failed".into()));
            });

            let _ = failing.join();
            let cancelled_while_blocked = state.cancelled();
            let _ = release_tx.send(());
            let _ = blocked.join();

            assert!(
                cancelled_while_blocked,
                "a failing worker must set the stop flag before other workers finish"
            );
        });

        assert!(state.into_error().is_some());
    }
}