yah-object-store 0.8.22

//! Cloudflare R2 implementation of [`ObjectStore`] over S3-compat SigV4.
//!
//! Uses `local_driver::s3_sign` helpers for signature computation and
//! `reqwest::blocking` for HTTP so the [`ObjectStore`] trait stays synchronous.
//! Async consumers wrap calls in `tokio::task::spawn_blocking`.
//!
//! ## Endpoint
//!
//! R2's S3-compat endpoint is `https://<account_id>.r2.cloudflarestorage.com`.
//! Region is always `"auto"`. Bucket lives in the URL path:
//! `https://<account_id>.r2.cloudflarestorage.com/<bucket>/<key>`.
//!
//! ## list_prefix
//!
//! Issues `GET /<bucket>?list-type=2&prefix=<encoded>` (ListObjectsV2) and
//! parses the `<Key>…</Key>` elements out of the XML body. Continues with
//! `&continuation-token=<…>` while `<IsTruncated>true</IsTruncated>` so
//! prefixes larger than the 1000-key page size return complete.
//!
//! @yah:relay(R630, "Object-store correctness + tooling gaps surfaced by standing up the cr.yah.dev registry")
//! @yah:at(2026-07-23T03:06:57Z)
//! @yah:status(open)
//! @yah:next("Both children were found while building yah-cr (the R2-backed OCI registry) on 2026-07-22 and are independent of that work — they are latent defects in shared object-store code that any caller can hit.")
//! @yah:next("Start with the SigV4 child: it is a correctness bug that fails closed but silently constrains every key namespace we can use. The bucket-delete child is additive and can follow.")
//! @yah:gotcha("The SigV4 defect is why cr.yah.dev stores OCI digests as sha256/<hex> instead of the natural sha256:<hex>. That workaround is load-bearing in two files that must stay in lockstep (app/yah/cli/src/cr.rs digest_key, app/yah/workers/yah-cr/src/index.ts digestKey). If the signing bug is fixed, those can be simplified — but only together, and only with a migration for keys already written.")
//! @arch:see(.yah/docs/working/W175-per-publisher-prefix.md)

use std::time::Duration;

use percent_encoding::{utf8_percent_encode, AsciiSet, NON_ALPHANUMERIC};
use reqwest::blocking::Client;
use reqwest::header::{HeaderValue, ETAG, IF_MATCH, IF_NONE_MATCH};
use reqwest::StatusCode;
use sha2::{Digest, Sha256};

use local_driver::s3_sign::{
    sign_s3_empty_body, sign_s3_get_with_query, sign_s3_no_body, sign_s3_put_object,
    sign_s3_put_object_with, S3PutOptions,
};

use crate::{Error, ObjectStore, Precondition};

/// R2's S3-compat region. The endpoint always accepts `"auto"`.
const R2_REGION: &str = "auto";

/// Keystore slot for the R2 S3 access key id.
pub const R2_ACCESS_KEY_SLOT: &str = "cloudflare-r2-access-key-id";
/// Keystore slot for the R2 S3 secret key.
pub const R2_SECRET_KEY_SLOT: &str = "cloudflare-r2-secret-key";
/// Env var fallback for the R2 access key id.
pub const R2_ACCESS_KEY_ENV: &str = "CF_R2_ACCESS_KEY_ID";
/// Env var fallback for the R2 secret key.
pub const R2_SECRET_KEY_ENV: &str = "CF_R2_SECRET_KEY";

/// Percent-encoding set for query-string values. SigV4 requires
/// unreserved characters (A-Z a-z 0-9 - _ . ~) to remain literal;
/// everything else gets percent-encoded.
const QUERY_VALUE: &AsciiSet = &NON_ALPHANUMERIC
    .remove(b'-')
    .remove(b'_')
    .remove(b'.')
    .remove(b'~');

/// Default content-type for keys with no recognized extension.
const DEFAULT_CONTENT_TYPE: &str = "application/octet-stream";

/// Content-type for an object key, inferred from its file extension.
///
/// R2 stores whatever Content-Type we send on PUT and serves it back verbatim
/// (the CDN custom domain does no extension sniffing). An octet-stream default
/// makes browsers *download* html shells instead of rendering them, so we set
/// an explicit type for the extensions a static site actually ships. Unknown
/// or extensionless keys (pointers, the `_yah-manifest.json` sidecar is `.json`
/// and handled) fall back to [`DEFAULT_CONTENT_TYPE`].
fn content_type_for_key(key: &str) -> &'static str {
    let ext = match key.rsplit_once('.') {
        // A `.` in a directory segment is not an extension.
        Some((_, e)) if !e.contains('/') => e,
        _ => "",
    };
    match ext.to_ascii_lowercase().as_str() {
        "html" | "htm" => "text/html; charset=utf-8",
        "css" => "text/css; charset=utf-8",
        "js" | "mjs" => "text/javascript; charset=utf-8",
        "json" | "map" => "application/json",
        "webmanifest" => "application/manifest+json",
        "xml" => "application/xml",
        "txt" => "text/plain; charset=utf-8",
        "svg" => "image/svg+xml",
        "webp" => "image/webp",
        "png" => "image/png",
        "jpg" | "jpeg" => "image/jpeg",
        "gif" => "image/gif",
        "avif" => "image/avif",
        "ico" => "image/x-icon",
        "woff2" => "font/woff2",
        "woff" => "font/woff",
        "ttf" => "font/ttf",
        "otf" => "font/otf",
        "wasm" => "application/wasm",
        "pdf" => "application/pdf",
        _ => DEFAULT_CONTENT_TYPE,
    }
}

/// R2-backed [`ObjectStore`].
///
/// Construct with [`R2ObjectStore::new`] when keys are already in hand,
/// or [`R2ObjectStore::from_vault`] to pull them from the yah keystore
/// (with env-var fallback).
pub struct R2ObjectStore {
    account_id: String,
    bucket: String,
    access_key: String,
    secret_key: String,
    /// Overrides the derived `https://<account_id>.r2.cloudflarestorage.com`.
    /// See [`R2ObjectStore::with_endpoint`].
    endpoint: Option<String>,
    client: Option<Client>,
}

impl Drop for R2ObjectStore {
    fn drop(&mut self) {
        // `reqwest::blocking::Client` owns a background tokio runtime whose
        // Drop panics with "Cannot drop a runtime in a context where blocking
        // is not allowed" when the drop happens inside an async context. This
        // fires when an `Arc<R2ObjectStore>` reaches zero from inside an
        // awaited future (e.g. publish_to_r2). Detach the shutdown onto a
        // fresh OS thread which has no tokio runtime context, so the client's
        // Drop can shut its internal runtime down cleanly. Dep-neutral — this
        // crate keeps its sync/tokio-free profile.
        let Some(client) = self.client.take() else { return };
        std::thread::spawn(move || drop(client));
    }
}

impl R2ObjectStore {
    /// Construct with explicit keys.
    ///
    /// `account_id` is the Cloudflare account id (the subdomain in
    /// `<account_id>.r2.cloudflarestorage.com`).
    pub fn new(
        account_id: impl Into<String>,
        bucket: impl Into<String>,
        access_key: impl Into<String>,
        secret_key: impl Into<String>,
    ) -> Result<Self, Error> {
        let client = Client::builder()
            .timeout(Duration::from_secs(300))
            .build()
            .map_err(|e| Error::Backend(format!("reqwest client: {e}")))?;
        Ok(Self {
            account_id: account_id.into(),
            bucket: bucket.into(),
            access_key: access_key.into(),
            secret_key: secret_key.into(),
            endpoint: None,
            client: Some(client),
        })
    }

    /// Point this store at an S3-compatible endpoint other than R2 — in
    /// practice, the pond tier's local MinIO (`http://127.0.0.1:9000`).
    ///
    /// Everything else about the store is already endpoint-agnostic: the bucket
    /// lives in the URL path (path-style addressing, which MinIO also speaks)
    /// and SigV4 is signed against whatever host the URL names.
    ///
    /// This exists because without it the pond rehearsal could not exercise the
    /// *read* side of a publish at all. `publish_to_pond` uploads a directory
    /// tree and offers no way to read an object back, so the one part of a
    /// release that is a read-modify-write — the accumulating `index.json` that
    /// https://yah.dev/releases renders from — was the one part a green local
    /// rehearsal proved nothing about (R330-T32). A conditional-write loop that
    /// has never run is a conditional-write loop you do not have.
    ///
    /// The region stays `"auto"`; MinIO accepts it.
    pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
        let endpoint = endpoint.into();
        let trimmed = endpoint.trim_end_matches('/');
        self.endpoint = (!trimmed.is_empty()).then(|| trimmed.to_string());
        self
    }

    fn client(&self) -> &Client {
        self.client
            .as_ref()
            .expect("client is Some until Drop takes it")
    }

    /// Construct from the yah keystore (vault), falling back to env vars.
    ///
    /// Reads `cloudflare-r2-access-key-id` / `cloudflare-r2-secret-key` slots
    /// (env fallback `CF_R2_ACCESS_KEY_ID` / `CF_R2_SECRET_KEY`). Returns
    /// [`Error::Auth`] if either is missing.
    pub fn from_vault(
        account_id: impl Into<String>,
        bucket: impl Into<String>,
    ) -> Result<Self, Error> {
        let access_key = fob::get_or_env(R2_ACCESS_KEY_SLOT, R2_ACCESS_KEY_ENV)
            .map_err(|e| Error::Auth(format!("vault read {R2_ACCESS_KEY_SLOT}: {e}")))?
            .ok_or_else(|| {
                Error::Auth(format!(
                    "missing R2 credential: set vault slot {R2_ACCESS_KEY_SLOT} or env {R2_ACCESS_KEY_ENV}"
                ))
            })?;
        let secret_key = fob::get_or_env(R2_SECRET_KEY_SLOT, R2_SECRET_KEY_ENV)
            .map_err(|e| Error::Auth(format!("vault read {R2_SECRET_KEY_SLOT}: {e}")))?
            .ok_or_else(|| {
                Error::Auth(format!(
                    "missing R2 credential: set vault slot {R2_SECRET_KEY_SLOT} or env {R2_SECRET_KEY_ENV}"
                ))
            })?;
        Self::new(account_id, bucket, access_key, secret_key)
    }

    fn endpoint(&self) -> String {
        match &self.endpoint {
            Some(e) => e.clone(),
            None => format!("https://{}.r2.cloudflarestorage.com", self.account_id),
        }
    }

    fn object_url(&self, key: &str) -> String {
        format!("{}/{}/{}", self.endpoint(), self.bucket, key)
    }

    fn bucket_url(&self) -> String {
        format!("{}/{}", self.endpoint(), self.bucket)
    }

    /// The one PUT path, with `Cache-Control` optional (R703-B8).
    ///
    /// `put` and `put_cached` differ only in that header, so they share this
    /// rather than each carrying their own signing + status handling — the
    /// shape where one of two copies quietly stops matching the other.
    fn put_inner(
        &self,
        key: &str,
        data: Vec<u8>,
        cache_control: Option<&str>,
    ) -> Result<(), Error> {
        let url = self.object_url(key);
        let body_sha256 = {
            let mut h = Sha256::new();
            h.update(&data);
            hex::encode(h.finalize())
        };
        let headers = sign_s3_put_object_with(
            &url,
            &body_sha256,
            data.len(),
            R2_REGION,
            &self.access_key,
            &self.secret_key,
            &S3PutOptions {
                content_type: content_type_for_key(key),
                // Generic object-store put — the BLAKE3 stamp is a static-asset
                // catalog concern, not a property of every object (R546-B10).
                blake3_meta: None,
                cache_control,
            },
        )
        .map_err(|e| Error::Backend(format!("sign PUT {key}: {e}")))?;

        let resp = self
            .client()
            .put(&url)
            .headers(headers)
            .body(data)
            .send()
            .map_err(|e| io_err(&format!("PUT {key}"), e))?;
        check_status(resp, "PUT", key)
    }
}

/// Convert a reqwest error into our generic [`Error`].
fn io_err(ctx: &str, e: impl std::fmt::Display) -> Error {
    Error::Io(format!("{ctx}: {e}"))
}

impl ObjectStore for R2ObjectStore {
    fn put(&self, key: &str, data: Vec<u8>) -> Result<(), Error> {
        self.put_inner(key, data, None)
    }

    fn put_cached(&self, key: &str, data: Vec<u8>, cache_control: &str) -> Result<(), Error> {
        self.put_inner(key, data, Some(cache_control))
    }

    fn get(&self, key: &str) -> Result<Option<Vec<u8>>, Error> {
        let url = self.object_url(key);
        // GET has no body: reqwest drops the `content-length: 0` header on the
        // wire, so signing it (as `sign_s3_empty_body` does) yields a signature
        // the server can't reproduce → 403 SignatureDoesNotMatch. Sign with the
        // content-length-free helper instead, exactly like ListObjectsV2. The
        // empty query string is correct for a plain object GET.
        let headers = sign_s3_get_with_query(
            &url,
            "",
            R2_REGION,
            &self.access_key,
            &self.secret_key,
        )
        .map_err(|e| Error::Backend(format!("sign GET {key}: {e}")))?;

        let resp = self
            .client()
            .get(&url)
            .headers(headers)
            .send()
            .map_err(|e| io_err(&format!("GET {key}"), e))?;

        match resp.status() {
            StatusCode::OK => {
                let bytes = resp
                    .bytes()
                    .map_err(|e| io_err(&format!("read GET {key}"), e))?;
                Ok(Some(bytes.to_vec()))
            }
            StatusCode::NOT_FOUND => Ok(None),
            s => Err(status_err("GET", key, s, resp.text().ok())),
        }
    }

    fn head(&self, key: &str) -> Result<bool, Error> {
        let url = self.object_url(key);
        // HEAD is body-less like GET: sign without content-length (see `get`).
        let headers = sign_s3_no_body(
            "HEAD",
            &url,
            "",
            R2_REGION,
            &self.access_key,
            &self.secret_key,
        )
        .map_err(|e| Error::Backend(format!("sign HEAD {key}: {e}")))?;

        let resp = self
            .client()
            .head(&url)
            .headers(headers)
            .send()
            .map_err(|e| io_err(&format!("HEAD {key}"), e))?;

        match resp.status() {
            StatusCode::OK => Ok(true),
            StatusCode::NOT_FOUND => Ok(false),
            s => Err(status_err("HEAD", key, s, None)),
        }
    }

    fn delete(&self, key: &str) -> Result<(), Error> {
        let url = self.object_url(key);
        let headers = sign_s3_empty_body(
            "DELETE",
            &url,
            R2_REGION,
            &self.access_key,
            &self.secret_key,
        )
        .map_err(|e| Error::Backend(format!("sign DELETE {key}: {e}")))?;

        let resp = self
            .client()
            .delete(&url)
            .headers(headers)
            .send()
            .map_err(|e| io_err(&format!("DELETE {key}"), e))?;

        match resp.status() {
            // S3 DELETE on a missing key returns 204 too — both are success
            // semantics for an idempotent delete.
            StatusCode::OK | StatusCode::NO_CONTENT | StatusCode::NOT_FOUND => Ok(()),
            s => Err(status_err("DELETE", key, s, resp.text().ok())),
        }
    }

    fn list_prefix(&self, prefix: &str) -> Result<Vec<String>, Error> {
        Ok(self
            .list_prefix_detailed(prefix)?
            .into_iter()
            .map(|m| m.key)
            .collect())
    }

    fn put_if(&self, key: &str, data: Vec<u8>, cond: Precondition) -> Result<String, Error> {
        let url = self.object_url(key);
        let body_sha256 = {
            let mut h = Sha256::new();
            h.update(&data);
            hex::encode(h.finalize())
        };
        // Sign the same fixed header set as an unconditional PUT. The conditional
        // header (If-Match / If-None-Match) is added *unsigned* afterwards: SigV4
        // only covers the headers in `SignedHeaders`, and S3/R2 honor extra
        // unsigned headers — so the precondition is enforced server-side without
        // touching the signer.
        let mut headers = sign_s3_put_object(
            &url,
            &body_sha256,
            content_type_for_key(key),
            data.len(),
            R2_REGION,
            &self.access_key,
            &self.secret_key,
            None,
        )
        .map_err(|e| Error::Backend(format!("sign PUT {key}: {e}")))?;

        match &cond {
            Precondition::IfAbsent => {
                headers.insert(IF_NONE_MATCH, HeaderValue::from_static("*"));
            }
            Precondition::IfMatch(etag) => {
                let v = HeaderValue::from_str(etag)
                    .map_err(|e| Error::Backend(format!("invalid If-Match etag {etag:?}: {e}")))?;
                headers.insert(IF_MATCH, v);
            }
        }

        let resp = self
            .client()
            .put(&url)
            .headers(headers)
            .body(data)
            .send()
            .map_err(|e| io_err(&format!("PUT(if) {key}"), e))?;

        let status = resp.status();
        if status == StatusCode::PRECONDITION_FAILED {
            return Err(Error::PreconditionFailed(format!(
                "put_if {key}: precondition not met ({cond:?})"
            )));
        }
        if !status.is_success() {
            return Err(status_err("PUT(if)", key, status, resp.text().ok()));
        }
        // Prefer the ETag echoed in the PUT response; fall back to a HEAD if a
        // backend ever omits it (R2 always returns it).
        match resp.headers().get(ETAG).and_then(|v| v.to_str().ok()) {
            Some(e) => Ok(e.to_string()),
            None => self
                .etag(key)?
                .ok_or_else(|| Error::Backend(format!("PUT(if) {key} returned no ETag"))),
        }
    }

    fn etag(&self, key: &str) -> Result<Option<String>, Error> {
        let url = self.object_url(key);
        // HEAD is body-less: sign without content-length (see `head`).
        let headers = sign_s3_no_body(
            "HEAD",
            &url,
            "",
            R2_REGION,
            &self.access_key,
            &self.secret_key,
        )
        .map_err(|e| Error::Backend(format!("sign HEAD {key}: {e}")))?;

        let resp = self
            .client()
            .head(&url)
            .headers(headers)
            .send()
            .map_err(|e| io_err(&format!("HEAD(etag) {key}"), e))?;

        match resp.status() {
            StatusCode::OK => Ok(resp
                .headers()
                .get(ETAG)
                .and_then(|v| v.to_str().ok())
                .map(|s| s.to_string())),
            StatusCode::NOT_FOUND => Ok(None),
            s => Err(status_err("HEAD(etag)", key, s, None)),
        }
    }
}

/// One `<Contents>` entry from an R2 `ListObjectsV2` response.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ObjectMeta {
    /// Object key (full path including any prefix).
    pub key: String,
    /// Object size in bytes.
    pub size: u64,
    /// Last-modified timestamp in ISO-8601 / RFC-3339 (R2's `<LastModified>` value).
    pub last_modified: String,
}

impl R2ObjectStore {
    /// List objects under `prefix` returning key + size + last-modified.
    ///
    /// Same paginated request as [`ObjectStore::list_prefix`] but parses the
    /// `<Size>` and `<LastModified>` siblings of each `<Key>` element. Used by
    /// the data-tab bucket viewer to render a directory-style listing.
    pub fn list_prefix_detailed(&self, prefix: &str) -> Result<Vec<ObjectMeta>, Error> {
        let mut entries = Vec::new();
        let mut continuation_token: Option<String> = None;
        let bucket_url = self.bucket_url();
        let encoded_prefix = utf8_percent_encode(prefix, QUERY_VALUE).to_string();

        loop {
            // Canonical query MUST be sorted by parameter name (SigV4).
            // Parameters: continuation-token (optional), list-type, prefix.
            let mut params: Vec<(String, String)> =
                vec![("list-type".to_string(), "2".to_string())];
            if let Some(token) = &continuation_token {
                let encoded = utf8_percent_encode(token, QUERY_VALUE).to_string();
                params.push(("continuation-token".to_string(), encoded));
            }
            params.push(("prefix".to_string(), encoded_prefix.clone()));
            params.sort_by(|a, b| a.0.cmp(&b.0));
            let canonical_query = params
                .iter()
                .map(|(k, v)| format!("{k}={v}"))
                .collect::<Vec<_>>()
                .join("&");

            let url_with_query = format!("{bucket_url}?{canonical_query}");

            let headers = sign_s3_get_with_query(
                &bucket_url,
                &canonical_query,
                R2_REGION,
                &self.access_key,
                &self.secret_key,
            )
            .map_err(|e| Error::Backend(format!("sign LIST {prefix}: {e}")))?;

            let resp = self
                .client()
                .get(&url_with_query)
                .headers(headers)
                .send()
                .map_err(|e| io_err(&format!("LIST {prefix}"), e))?;

            if !resp.status().is_success() {
                return Err(status_err("LIST", prefix, resp.status(), resp.text().ok()));
            }
            let body = resp
                .text()
                .map_err(|e| io_err(&format!("LIST {prefix} body"), e))?;
            let (page_entries, next_token) = parse_list_v2_detailed(&body);
            entries.extend(page_entries);
            if let Some(t) = next_token {
                continuation_token = Some(t);
            } else {
                break;
            }
        }
        Ok(entries)
    }
}

fn check_status(resp: reqwest::blocking::Response, verb: &str, key: &str) -> Result<(), Error> {
    if resp.status().is_success() {
        Ok(())
    } else {
        let status = resp.status();
        let body = resp.text().ok();
        Err(status_err(verb, key, status, body))
    }
}

fn status_err(verb: &str, key: &str, status: StatusCode, body: Option<String>) -> Error {
    let snippet = body
        .as_deref()
        .map(|s| s.chars().take(200).collect::<String>())
        .unwrap_or_default();
    let msg = format!("{verb} {key} → {status} {snippet}");
    match status {
        StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED => Error::Auth(msg),
        StatusCode::NOT_FOUND => Error::NotFound(msg),
        _ => Error::Backend(msg),
    }
}

/// Parse a `ListObjectsV2` XML response for keys + next continuation token.
///
/// Deliberately tiny — full XML parsing is overkill for the two elements we
/// care about. Looks for `<Key>...</Key>` and `<NextContinuationToken>...`
/// inside the body. If R2 ever changes the element shape (it won't — it's
/// S3-compat), the integration test catches it.
fn parse_list_v2(body: &str) -> (Vec<String>, Option<String>) {
    let keys = extract_all_tags(body, "Key");
    let next = extract_first_tag(body, "NextContinuationToken");
    let truncated = extract_first_tag(body, "IsTruncated")
        .map(|v| v.trim().eq_ignore_ascii_case("true"))
        .unwrap_or(false);
    (keys, if truncated { next } else { None })
}

/// Parse `<Contents>` blocks for key + size + last-modified.
///
/// R2's `<Contents>` always has `<Key>` followed by `<LastModified>` and
/// `<Size>` siblings. We walk `<Contents>...</Contents>` blocks and pull the
/// three tags from each — order-insensitive within the block. Entries missing
/// any of the three are skipped (defensive — R2 always emits all three).
fn parse_list_v2_detailed(body: &str) -> (Vec<ObjectMeta>, Option<String>) {
    let blocks = extract_all_tags(body, "Contents");
    let entries = blocks
        .into_iter()
        .filter_map(|block| {
            let key = extract_first_tag(&block, "Key")?;
            let size = extract_first_tag(&block, "Size")?.trim().parse::<u64>().ok()?;
            let last_modified = extract_first_tag(&block, "LastModified")?;
            Some(ObjectMeta { key, size, last_modified })
        })
        .collect();
    let next = extract_first_tag(body, "NextContinuationToken");
    let truncated = extract_first_tag(body, "IsTruncated")
        .map(|v| v.trim().eq_ignore_ascii_case("true"))
        .unwrap_or(false);
    (entries, if truncated { next } else { None })
}

fn extract_all_tags(body: &str, tag: &str) -> Vec<String> {
    let open = format!("<{tag}>");
    let close = format!("</{tag}>");
    let mut out = Vec::new();
    let mut search = body;
    while let Some(start) = search.find(&open) {
        let content_start = start + open.len();
        if let Some(end) = search[content_start..].find(&close) {
            out.push(search[content_start..content_start + end].to_string());
            search = &search[content_start + end + close.len()..];
        } else {
            break;
        }
    }
    out
}

fn extract_first_tag(body: &str, tag: &str) -> Option<String> {
    extract_all_tags(body, tag).into_iter().next()
}

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

    #[test]
    fn with_endpoint_redirects_every_url_and_leaves_r2_alone() {
        let store = R2ObjectStore::new("acct", "yah-dev", "k", "s").unwrap();
        assert_eq!(
            store.object_url("yah/index.json"),
            "https://acct.r2.cloudflarestorage.com/yah-dev/yah/index.json"
        );

        // Pond tier: same store, MinIO endpoint, path-style bucket preserved.
        let pond = R2ObjectStore::new("pond", "yah-dev", "k", "s")
            .unwrap()
            .with_endpoint("http://127.0.0.1:9000");
        assert_eq!(
            pond.object_url("yah/index.json"),
            "http://127.0.0.1:9000/yah-dev/yah/index.json"
        );
        assert_eq!(pond.bucket_url(), "http://127.0.0.1:9000/yah-dev");
    }

    #[test]
    fn with_endpoint_normalizes_trailing_slash_and_ignores_empty() {
        let s = R2ObjectStore::new("acct", "b", "k", "s")
            .unwrap()
            .with_endpoint("http://127.0.0.1:9000/");
        assert_eq!(s.object_url("k1"), "http://127.0.0.1:9000/b/k1");
        // An empty override is a config mistake, not an instruction to sign
        // against the empty host — fall back to the derived R2 endpoint.
        let s = R2ObjectStore::new("acct", "b", "k", "s")
            .unwrap()
            .with_endpoint("");
        assert_eq!(
            s.object_url("k1"),
            "https://acct.r2.cloudflarestorage.com/b/k1"
        );
    }

    /// Accept exactly one HTTP request on an ephemeral loopback port, answer
    /// `200`, and hand the raw request head back. Enough of a server to prove
    /// what went onto the wire, and no more — the point is the headers, and a
    /// mock at the `reqwest` layer would only re-assert what the signer already
    /// returned rather than what the client actually sent.
    fn one_shot_http() -> (String, std::thread::JoinHandle<String>) {
        use std::io::{BufRead, BufReader, Read, Write};

        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let url = format!("http://{}", listener.local_addr().unwrap());
        let handle = std::thread::spawn(move || {
            let (stream, _) = listener.accept().unwrap();
            let mut reader = BufReader::new(stream);
            let mut head = String::new();
            loop {
                let mut line = String::new();
                if reader.read_line(&mut line).unwrap() == 0 {
                    break;
                }
                let done = line == "\r\n";
                head.push_str(&line);
                if done {
                    break;
                }
            }
            // Drain the body, else the client sees the connection close
            // mid-write and reports a broken pipe instead of our 200.
            let len: usize = head
                .lines()
                .find_map(|l| {
                    l.strip_prefix("content-length: ")
                        .or_else(|| l.strip_prefix("Content-Length: "))
                })
                .and_then(|v| v.trim().parse().ok())
                .unwrap_or(0);
            let mut body = vec![0u8; len];
            reader.read_exact(&mut body).unwrap();
            reader
                .into_inner()
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
                .unwrap();
            head
        });
        (url, handle)
    }

    /// R703-B8, the end of the chain: the signer builds a `Cache-Control` and
    /// `reqwest` has to actually put it on the socket. Every other test here
    /// stops at the `HeaderMap`, which is one `.headers()` call away from being
    /// a test that passes while R2 stores an object with no directive.
    #[test]
    fn put_cached_sends_the_cache_control_header_on_the_wire() {
        let (endpoint, server) = one_shot_http();
        let store = R2ObjectStore::new("acct", "yah-dev", "AK", "SK")
            .unwrap()
            .with_endpoint(endpoint);

        store
            .put_cached(
                "yah-desktop/latest.json",
                b"{\"version\":\"0.8.22\"}".to_vec(),
                crate::CACHE_CONTROL_NO_CACHE,
            )
            .unwrap();

        let head = server.join().unwrap().to_lowercase();
        assert!(
            head.starts_with("put /yah-dev/yah-desktop/latest.json "),
            "{head}"
        );
        assert!(head.contains("cache-control: no-cache, max-age=0\r\n"), "{head}");
        // Sent AND signed — an unsigned header R2 would reject the request over.
        assert!(
            head.contains("signedheaders=cache-control;content-length;content-type;host;"),
            "{head}"
        );
    }

    /// The other half: a plain `put` must still send no directive at all. If it
    /// quietly gained a default, versioned release bytes would start carrying
    /// whatever that default was.
    #[test]
    fn a_plain_put_sends_no_cache_control_header() {
        let (endpoint, server) = one_shot_http();
        let store = R2ObjectStore::new("acct", "yah-dev", "AK", "SK")
            .unwrap()
            .with_endpoint(endpoint);

        store.put("some/blob.bin", b"bytes".to_vec()).unwrap();

        let head = server.join().unwrap().to_lowercase();
        assert!(!head.contains("cache-control"), "{head}");
    }

    #[test]
    fn parse_list_v2_extracts_keys() {
        let body = r#"<?xml version="1.0" encoding="UTF-8"?>
            <ListBucketResult>
                <IsTruncated>false</IsTruncated>
                <Contents><Key>yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz</Key></Contents>
                <Contents><Key>yubaba/release-manifest.json</Key></Contents>
            </ListBucketResult>"#;
        let (keys, next) = parse_list_v2(body);
        assert_eq!(
            keys,
            vec![
                "yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz".to_string(),
                "yubaba/release-manifest.json".to_string(),
            ]
        );
        assert!(next.is_none());
    }

    #[test]
    fn parse_list_v2_returns_continuation_when_truncated() {
        let body = r#"<ListBucketResult>
                <IsTruncated>true</IsTruncated>
                <NextContinuationToken>abc123</NextContinuationToken>
                <Contents><Key>a</Key></Contents>
            </ListBucketResult>"#;
        let (keys, next) = parse_list_v2(body);
        assert_eq!(keys, vec!["a".to_string()]);
        assert_eq!(next.as_deref(), Some("abc123"));
    }

    #[test]
    fn parse_list_v2_ignores_token_when_not_truncated() {
        // Some S3-compat impls emit NextContinuationToken with IsTruncated=false.
        // We treat IsTruncated as load-bearing.
        let body = r#"<ListBucketResult>
                <IsTruncated>false</IsTruncated>
                <NextContinuationToken>stale</NextContinuationToken>
                <Contents><Key>a</Key></Contents>
            </ListBucketResult>"#;
        let (_, next) = parse_list_v2(body);
        assert!(next.is_none());
    }

    #[test]
    fn parse_list_v2_detailed_extracts_size_and_mtime() {
        let body = r#"<?xml version="1.0" encoding="UTF-8"?>
            <ListBucketResult>
                <IsTruncated>false</IsTruncated>
                <Contents>
                    <Key>yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz</Key>
                    <LastModified>2026-06-08T20:14:32.000Z</LastModified>
                    <ETag>"abc"</ETag>
                    <Size>4823104</Size>
                    <StorageClass>STANDARD</StorageClass>
                </Contents>
                <Contents>
                    <Key>yubaba/release-manifest.json</Key>
                    <LastModified>2026-06-08T20:14:35.000Z</LastModified>
                    <Size>412</Size>
                </Contents>
            </ListBucketResult>"#;
        let (entries, next) = parse_list_v2_detailed(body);
        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].key, "yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz");
        assert_eq!(entries[0].size, 4823104);
        assert_eq!(entries[0].last_modified, "2026-06-08T20:14:32.000Z");
        assert_eq!(entries[1].key, "yubaba/release-manifest.json");
        assert_eq!(entries[1].size, 412);
        assert!(next.is_none());
    }

    #[test]
    fn r2_object_store_constructs_with_explicit_keys() {
        let s = R2ObjectStore::new("acct", "yah-dev", "AK", "SK").unwrap();
        assert_eq!(s.object_url("k"), "https://acct.r2.cloudflarestorage.com/yah-dev/k");
        assert_eq!(s.bucket_url(), "https://acct.r2.cloudflarestorage.com/yah-dev");
    }

    #[test]
    fn object_url_preserves_slashes_in_key() {
        let s = R2ObjectStore::new("acct", "b", "AK", "SK").unwrap();
        assert_eq!(
            s.object_url("yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz"),
            "https://acct.r2.cloudflarestorage.com/b/yubaba/0.8.9/x86_64-unknown-linux-musl/yubaba.tar.gz"
        );
    }

    #[test]
    fn content_type_inferred_from_extension() {
        assert_eq!(
            content_type_for_key("yah-marketing/cloud/index.html"),
            "text/html; charset=utf-8"
        );
        assert_eq!(content_type_for_key("app.css"), "text/css; charset=utf-8");
        assert_eq!(content_type_for_key("bundle.mjs"), "text/javascript; charset=utf-8");
        assert_eq!(content_type_for_key("illustrations/horse.webp"), "image/webp");
        assert_eq!(content_type_for_key("manifest.json"), "application/json");
        // Extensionless keys (pointers) and dotted directory segments fall back.
        assert_eq!(content_type_for_key("pointers/releases"), DEFAULT_CONTENT_TYPE);
        assert_eq!(content_type_for_key("v1.2/binary"), DEFAULT_CONTENT_TYPE);
    }
}