git_remote_object_store/object_store/

azure.rs

//! Azure Blob Storage backend for the [`ObjectStore`] trait.
//!
//! [`ObjectStore`]: super::ObjectStore
//!
//! `AzureStore` wraps `azure_storage_blob`. Like the S3 backend, this
//! module owns the URL → SDK config translation, the error-code
//! classifier ([`classify`]), and the credential resolution plumbing.
//! Unlike S3, the SDK already does parallel range downloads inside
//! `BlobClient::download()`, so there is no hand-rolled multipart
//! orchestrator (asymmetric with S3 by design).
//!
//! ## Authentication
//!
//! The official `azure_storage_blob` 1.0 crate currently exposes only
//! `Arc<dyn TokenCredential>` (Entra ID) on its constructors. Azurite
//! does not implement Entra ID without an `--oauth basic` HTTPS setup,
//! and many production accounts still authenticate with shared keys.
//! To bridge both, we install our own [`auth::SharedKeySigningPolicy`]
//! as a per-try [`azure_core::http::policies::Policy`] and pass `None`
//! for the SDK's `credential` parameter. The SDK then forwards every
//! request through our policy, which signs the request using the Azure
//! Storage shared-key v2 scheme. Tracking issue:
//! `Azure/azure-sdk-for-rust#2975`.
//!
//! Resolution order for `?credential=<NAME>` in the URL:
//!
//! 1. `AZSTORE_<NAME>_KEY` — base64 account key → shared-key signing.
//! 2. `AZSTORE_<NAME>_CONNECTION_STRING` — connection string with
//!    `AccountName=` / `AccountKey=` → shared-key signing.
//! 3. `AZSTORE_<NAME>_SAS` — SAS query string appended verbatim to
//!    every outgoing request URL.
//!
//! When no `?credential=` flag is set we fall back to
//! `azure_identity::DeveloperToolsCredential` (env, workload identity,
//! managed identity, Azure CLI, ...).
//!
//! ## Conditional writes
//!
//! [`put_if_absent`][super::ObjectStore::put_if_absent] uses
//! `If-None-Match: "*"` (the SDK's
//! `BlockBlobClientUploadOptions::if_not_exists` convenience).
//! Azure returns 409 (`BlobAlreadyExists`) or 412
//! (`ConditionNotMet`) for the contention case; both collapse to
//! `Ok(false)`.
//!
//! ## Atomic `get_to_file`
//!
//! Identical to the S3 path: `head` → tempfile → `download(if_match)` →
//! persist. The SDK's `download()` aggregates parallel range fetches
//! internally, so no per-chunk semaphore here. A single retry with a
//! fresh `ETag` covers the head-then-`GET` race (412 mid-download).
//!
//! ## `copy(src, dst)`
//!
//! `azure_storage_blob` 1.0 does not expose a `BlobClient::copy_from_url`
//! method (only `BlockBlobClient::upload_blob_from_url`, which requires
//! a SAS-tokened source URL or an `x-ms-copy-source-authorization`
//! header — neither integrates cleanly with our credential model). We
//! implement `copy` as a stream-through-tempfile round trip:
//! `get_to_file` writes `src` to a `NamedTempFile`, then `put_path`
//! uploads it to `dst`. Both legs already stream — `get_to_file`
//! consumes the SDK's chunked download into the file without buffering
//! the body, and `put_path` switches to our explicit
//! `stage_block` + `commit_block_list` orchestrator (see
//! [`AzureStore::multipart_put_path`]) once the body crosses
//! [`super::multipart::MULTIPART_PUT_THRESHOLD`]. Peak in-flight bytes
//! are bounded by
//! [`super::multipart::MULTIPART_PUT_MAX_CONCURRENCY`] ×
//! [`super::multipart::MULTIPART_PUT_PART_SIZE`] regardless of blob
//! size, which matters for `manage doctor`'s duplicate-bundle
//! quarantine path ([`crate::manage::doctor::Doctor::evict_losing_bundle`])
//! — that path can copy multi-GiB bundles. Zero-byte lock files still
//! round-trip fast: `get_to_file` short-circuits the GET on `size == 0`
//! and `put_path` issues a single zero-byte `Put Blob`. Body is
//! preserved; user metadata is not propagated, matching the S3 backend's
//! `CopyObject` path which similarly carries only body bytes.
//!
//! This is asymmetric with the S3 backend, which uses `CopyObject` for
//! a true server-side copy — Azure's equivalent (`Copy Blob`,
//! `Put Blob From URL`) requires a SAS-signed source URL or an
//! `x-ms-copy-source-authorization` header that the 1.0 SDK does not
//! ergonomically expose. The download+reupload path is the safe
//! correct fallback until the SDK closes that gap.
//!
//! ## A note on `Range` and zero-byte blobs
//!
//! A `Range` request against a zero-byte blob returns HTTP 416. We
//! never issue Range requests directly — `BlobClient::download()`
//! owns that — but the zero-size short-circuit in
//! [`get_to_file`](ObjectStore::get_to_file) also avoids any download
//! SDK call against a known-empty blob, which sidesteps the issue
//! entirely.
//!
//! ## Size limits
//!
//! Azure caps a block blob at 50 000 committed blocks (~4.75 TiB at
//! the SDK's default block size) and a single `Put Blob` body at
//! 5000 MiB; above [`super::multipart::MULTIPART_PUT_THRESHOLD`] the
//! helper switches to explicit `stage_block` + `commit_block_list`,
//! so callers do not have to reason about the single-call cutoff.
//! The upload path is **not resumable** across process death — see
//! the README "Known limitations" section.
//!
//! ## HTTP transport tuning
//!
//! `azure_core` 1.1's default transport keeps idle pooled connections
//! forever and never sets TCP keepalive, so a pooled connection to a
//! rotated VIP would hang an in-flight request until the OS-level TCP
//! retransmit timeout fires (~15 minutes on Linux). [`AzureStore`]
//! installs a custom [`reqwest::Client`] via [`Transport`] on
//! [`ClientOptions::transport`] with four bounds:
//!
//! - [`POOL_IDLE_TIMEOUT`] (30 s) — drops idle pooled connections
//!   before a typical DNS rotation makes them stale.
//! - [`TCP_KEEPALIVE`] (30 s) — detects a dead-but-not-closed TCP
//!   session in seconds rather than the 2-hour Linux default; covers
//!   *hot* pooled connections that pool-idle alone cannot.
//! - [`CONNECT_TIMEOUT`] (10 s) — bounds a fresh-connect attempt to
//!   a dead VIP rather than waiting on the OS connect timeout.
//! - [`READ_TIMEOUT`] (30 s) — per-read timeout that resets after a
//!   successful read, so a stuck transfer fails fast without limiting
//!   total body size.
//!
//! Together these cap a DNS-rotation hang at tens of seconds rather
//! than minutes. The custom transport leaves
//! [`ClientOptions::per_try_policies`] (where the shared-key signing
//! lives) untouched — the SDK pipeline runs per-try policies
//! independently of the transport. Tracking issue: #26.
//!
//! ## Stdout discipline
//!
//! Per `.claude/rules/protocol-stdout.md`, this module never writes to
//! stdout. Diagnostics go through `tracing` (which the helper binaries
//! configure to write to stderr).

pub mod auth;
pub(crate) mod sas;

use std::path::Path;
use std::sync::Arc;
use std::time::Duration;

use azure_core::http::headers::{HeaderName, Headers};
use azure_core::http::request::RequestContent;
use azure_core::http::{ClientOptions, Transport};
use azure_storage_blob::clients::{
    BlobClient, BlobContainerClient, BlobContainerClientOptions, BlockBlobClient,
};
use azure_storage_blob::models::{
    BlobClientDeleteOptions, BlobClientDownloadOptions, BlobClientGetPropertiesOptions,
    BlobContainerClientListBlobsOptions, BlockBlobClientCommitBlockListOptions,
    BlockBlobClientUploadOptions, BlockLookupList, HttpRange,
};
use azure_storage_blob::stream::tokio::FileStream;
use bytes::Bytes;
use futures::StreamExt;
use tempfile::NamedTempFile;
use time::OffsetDateTime;
use tokio::io::AsyncWriteExt;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;
use url::Url;

use crate::url::{AzureAddressing, RemoteUrl};

use super::error::{network_boxed, other_boxed};
use super::multipart::{
    AZURE_MAX_BLOCKS, MULTIPART_PUT_MAX_CONCURRENCY, MULTIPART_PUT_PART_SIZE, UploadPart,
    plan_upload_parts, read_file_part, should_use_multipart, slice_bytes_part,
};
use super::{
    GetOpts, ObjectMeta, ObjectStore, ObjectStoreError, ProgressSink, PutOpts, persist_temp,
};

/// Azure Blob's hard ceiling on a single Put Blob body for the wire
/// versions we negotiate (2019-12-12+). Reported in
/// [`ObjectStoreError::PayloadTooLarge`] when the SDK surfaces HTTP 413
/// or `RequestBodyTooLarge`, so the wire-line names a concrete number
/// rather than dumping an opaque SDK chain.
pub(crate) const SINGLE_PUT_BLOB_LIMIT_BYTES: u64 = 5_000 * (1 << 20);

/// Bound on how long an idle pooled HTTPS connection lingers before
/// the [`reqwest`] connection pool drops it. Short enough that DNS
/// rotation rarely hits a stale pooled connection; long enough that
/// bursty fetch / push batches still benefit from connection reuse.
/// See module-level "HTTP transport tuning" docs and issue #26.
pub(crate) const POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(30);

/// TCP keepalive interval for the custom [`reqwest`] transport.
/// Detects dead-but-not-closed sessions in seconds rather than the
/// 2-hour Linux default. See module-level "HTTP transport tuning"
/// docs and issue #26.
pub(crate) const TCP_KEEPALIVE: Duration = Duration::from_secs(30);

/// Bound on a fresh TCP-connect attempt. `reqwest` defaults to no
/// connect timeout, so an unreachable IP would otherwise wait on the
/// OS-level connect timeout (~75 s on Linux defaults). 10 s is
/// comfortable for an in-region or even cross-region handshake while
/// failing fast on a dead VIP. See module-level "HTTP transport
/// tuning" docs and issue #26.
pub(crate) const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);

/// Per-read timeout for the custom [`reqwest`] transport. Resets after
/// each successful read, so it caps how long a stuck connection can
/// hold a transfer without limiting total body size. Sized to match
/// [`POOL_IDLE_TIMEOUT`] / [`TCP_KEEPALIVE`] so a single rotation
/// budget covers all three knobs. See module-level "HTTP transport
/// tuning" docs and issue #26.
pub(crate) const READ_TIMEOUT: Duration = Duration::from_secs(30);

/// Production [`ObjectStore`] backed by `azure_storage_blob`.
pub struct AzureStore {
    container: BlobContainerClient,
    /// Container name as parsed from the URL — needed by SAS-token
    /// construction (issue #76) because the SDK's
    /// `BlobContainerClient::container_name()` is private. Held
    /// regardless of credential type so the field shape doesn't
    /// branch on whether SAS is reachable.
    container_name: String,
    /// Storage-key material for service-blob SAS generation
    /// ([`presigned_get_url`](ObjectStore::presigned_get_url)).
    /// `Some` when the credential alias resolves to a shared
    /// account key (KEY env var or connection string); `None` for
    /// SAS-env-var or Entra-ID paths, which return
    /// [`ObjectStoreError::Unsupported`] for presigning.
    sas_signing: Option<auth::SasSigningKey>,
}

impl std::fmt::Debug for AzureStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // `BlobContainerClient` is opaque (private fields, no `Debug`);
        // surface the container URL instead so error / log lines remain
        // useful.
        f.debug_struct("AzureStore")
            .field("url", &self.container.url().as_str())
            .field("container", &self.container_name)
            .field("sas_signing", &self.sas_signing)
            .finish()
    }
}

impl AzureStore {
    /// Build an `AzureStore` from a parsed [`RemoteUrl`].
    ///
    /// Like the S3 backend, the [`RemoteUrl::Azure::prefix`] field is
    /// intentionally **not** consumed here; callers compose it into keys
    /// themselves.
    ///
    /// Marked `async` for symmetry with `S3Store::from_remote_url`,
    /// which awaits the AWS provider chain. The Azure path resolves
    /// credentials synchronously today; the signature stays `async` so
    /// future credential providers (e.g. one that fetches an OIDC
    /// token at construction) can plug in without breaking callers.
    ///
    /// # Errors
    ///
    /// Returns [`ObjectStoreError::Other`] if `url` is not the Azure
    /// variant or if credential resolution fails.
    #[allow(clippy::unused_async)]
    pub async fn from_remote_url(url: &RemoteUrl) -> Result<Self, ObjectStoreError> {
        let RemoteUrl::Azure {
            endpoint,
            account,
            container,
            addressing,
            flags,
            ..
        } = url
        else {
            return Err(ObjectStoreError::Other(
                format!("AzureStore::from_remote_url called with non-Azure URL: {url}").into(),
            ));
        };

        let container_url = build_container_url(endpoint, account, container, *addressing);
        let resolved = auth::resolve(account, flags)?;
        let sas_signing = resolved.sas_signing_key.clone();

        let client_options = build_client_options(&resolved)?;

        let container_options = BlobContainerClientOptions {
            client_options,
            ..Default::default()
        };

        let container_client = BlobContainerClient::new(
            container_url,
            resolved.token_credential,
            Some(container_options),
        )
        .map_err(other_boxed)?;

        Ok(Self {
            container: container_client,
            container_name: container.clone(),
            sas_signing,
        })
    }

    /// Construct a [`BlobClient`] for an individual blob.
    fn blob_client(&self, key: &str) -> BlobClient {
        self.container.blob_client(key)
    }

    /// Verify the container is reachable with the configured credentials
    /// by listing one blob (`maxresults=1`) and consuming only the first
    /// page of results. Used by [`crate::protocol::backend::build`] to
    /// fold credential / missing-container / authorization failures into
    /// categorical [`crate::protocol::backend::BackendError`] variants
    /// before the helper REPL runs its first command. Counterpart to
    /// [`crate::object_store::s3::S3Store::probe`].
    pub(crate) async fn probe(&self, prefix: &str) -> Result<(), ObjectStoreError> {
        // Pass `None` for an empty prefix per the same Azurite quirk
        // documented at the top of `list` above: a signed empty prefix
        // returns 403 from Azurite.
        let prefix_opt = (!prefix.is_empty()).then(|| prefix.to_owned());
        let opts = BlobContainerClientListBlobsOptions {
            prefix: prefix_opt,
            maxresults: Some(1),
            ..Default::default()
        };
        let mut pages = self
            .container
            .list_blobs(Some(opts))
            .map_err(|e| classify(e, prefix))?
            .into_pages();
        // Consume only the first page: probing does not need the full
        // listing — we only care that the request succeeded.
        if let Some(page_result) = pages.next().await {
            page_result.map_err(|e| classify(e, prefix))?;
        }
        Ok(())
    }
}

/// Build the [`reqwest::Client`] used by [`AzureStore`]'s custom
/// [`Transport`].
///
/// Bounds the connection pool's idle window, enables TCP keepalive,
/// and sets connect / per-read timeouts so a rotated VIP cannot wedge
/// a long-running session (see [`POOL_IDLE_TIMEOUT`] / [`TCP_KEEPALIVE`]
/// / [`CONNECT_TIMEOUT`] / [`READ_TIMEOUT`] for rationale). Returns
/// [`ObjectStoreError::Other`] if the TLS / DNS resolver layer fails
/// to initialise, which the SDK would otherwise surface as a cryptic
/// per-request error.
pub(crate) fn build_http_client() -> Result<Arc<reqwest::Client>, ObjectStoreError> {
    reqwest::Client::builder()
        .pool_idle_timeout(POOL_IDLE_TIMEOUT)
        .tcp_keepalive(TCP_KEEPALIVE)
        .connect_timeout(CONNECT_TIMEOUT)
        .read_timeout(READ_TIMEOUT)
        .build()
        .map(Arc::new)
        .map_err(other_boxed)
}

/// Build the [`ClientOptions`] [`AzureStore`] hands to the SDK.
///
/// Installs the custom [`Transport`] (see [`build_http_client`]) and
/// preserves the credential resolver's per-try signing policy. The
/// helper is split out (rather than inlined into [`AzureStore::from_remote_url`])
/// so unit tests can assert that both invariants hold without
/// constructing a real `BlobContainerClient`.
pub(crate) fn build_client_options(
    resolved: &auth::ResolvedCredentials,
) -> Result<ClientOptions, ObjectStoreError> {
    let mut opts = ClientOptions {
        transport: Some(Transport::new(build_http_client()?)),
        ..Default::default()
    };
    if let Some(policy) = &resolved.per_try_policy {
        opts.per_try_policies.push(Arc::clone(policy));
    }
    Ok(opts)
}

/// Construct the container-level URL [`BlobContainerClient::new`] expects.
///
/// The SDK addresses a container purely by URL, so any prefix segments
/// carried by the parsed remote URL are dropped and the path is rebuilt
/// from the account and container. For virtual-hosted addressing the
/// path becomes `/<container>`; for path-style addressing (Azurite,
/// custom endpoints) it becomes `/<account>/<container>`.
pub(crate) fn build_container_url(
    endpoint: &Url,
    account: &str,
    container: &str,
    addressing: AzureAddressing,
) -> Url {
    let mut rewritten = endpoint.clone();
    rewritten.set_query(None);
    rewritten.set_fragment(None);
    let path = match addressing {
        AzureAddressing::VirtualHosted => format!("/{container}"),
        AzureAddressing::PathStyle => format!("/{account}/{container}"),
    };
    rewritten.set_path(&path);
    rewritten
}

/// Map an [`azure_core::Error`] into the trait's [`ObjectStoreError`] enum.
///
/// `key` is the operation's key/prefix context; it appears in the
/// resulting [`ObjectStoreError::NotFound`] / [`ObjectStoreError::AccessDenied`] /
/// [`ObjectStoreError::PreconditionFailed`] / [`ObjectStoreError::Conflict`] payload.
fn classify(err: azure_core::Error, key: &str) -> ObjectStoreError {
    if let azure_core::error::ErrorKind::HttpResponse {
        status, error_code, ..
    } = err.kind()
        && let Some(mapped) =
            classify_status_and_code(u16::from(*status), error_code.as_deref(), key)
    {
        return mapped;
    }
    if matches!(err.kind(), azure_core::error::ErrorKind::Io) {
        return network_boxed(err);
    }
    other_boxed(err)
}

/// Pure status/code classifier (key context, no SDK types) so unit
/// tests can exercise every branch without synthesising an SDK error.
fn classify_status_and_code(
    status: u16,
    code: Option<&str>,
    key: &str,
) -> Option<ObjectStoreError> {
    match status {
        404 => return Some(ObjectStoreError::NotFound(key.to_owned())),
        403 => return Some(ObjectStoreError::AccessDenied(key.to_owned())),
        412 => return Some(ObjectStoreError::PreconditionFailed(key.to_owned())),
        409 => return Some(ObjectStoreError::Conflict(key.to_owned())),
        // Azure surfaces a Put Blob body over the single-PUT ceiling as
        // HTTP 413 with code `RequestBodyTooLarge`; the status alone is
        // sufficient (HTTP 413 is the canonical "Payload Too Large").
        413 => {
            return Some(ObjectStoreError::PayloadTooLarge {
                limit_bytes: SINGLE_PUT_BLOB_LIMIT_BYTES,
            });
        }
        _ => {}
    }
    // Defensive backstop for the (rare) case where the SDK exposes the
    // service code without a 413 status: route on the code alone.
    match code {
        Some("RequestBodyTooLarge") => Some(ObjectStoreError::PayloadTooLarge {
            limit_bytes: SINGLE_PUT_BLOB_LIMIT_BYTES,
        }),
        _ => None,
    }
}

/// Convert the relevant `Get Blob Properties` headers into the trait's
/// [`ObjectMeta`].
///
/// Extracted so unit tests can drive the missing-content-length and
/// missing-last-modified guard branches without synthesising a full
/// `BlobClientGetPropertiesResultHeaders` value.
///
/// A missing `Content-Length` is an error rather than silent zero: a
/// 0-byte size is semantically meaningful (lock files are intentionally
/// empty) and downstream `head_then_download` takes a fast path on
/// `size == 0` that writes an empty destination file. Treating "header
/// absent" as 0 would silently produce empty bundles instead of
/// surfacing the malformed response.
fn properties_to_meta(
    key: &str,
    content_length: Option<u64>,
    last_modified: Option<OffsetDateTime>,
    etag: Option<&str>,
) -> Result<ObjectMeta, ObjectStoreError> {
    let size = content_length.ok_or_else(|| {
        ObjectStoreError::Other(
            format!("get_properties on `{key}` returned no content-length").into(),
        )
    })?;
    let last_modified = last_modified.ok_or_else(|| {
        ObjectStoreError::Other(
            format!("get_properties on `{key}` returned no last-modified").into(),
        )
    })?;
    Ok(ObjectMeta {
        key: key.to_owned(),
        size,
        last_modified,
        etag: etag.map(str::to_owned),
    })
}

/// Convert a `BlobItem`-shaped record into the trait's [`ObjectMeta`].
///
/// Extracted so unit tests can drive the missing-field guards without
/// synthesising a full `ListBlobsResponse`.
fn item_to_meta(
    name: Option<&str>,
    content_length: Option<u64>,
    last_modified: Option<OffsetDateTime>,
    etag: Option<&str>,
) -> Result<ObjectMeta, ObjectStoreError> {
    let key = name
        .ok_or_else(|| ObjectStoreError::Other("list_blobs returned a blob without a name".into()))?
        .to_owned();
    let size = content_length.unwrap_or(0);
    let last_modified = last_modified.ok_or_else(|| {
        ObjectStoreError::Other(
            format!("list_blobs returned blob `{key}` without last_modified").into(),
        )
    })?;
    Ok(ObjectMeta {
        key,
        size,
        last_modified,
        etag: etag.map(str::to_owned),
    })
}

#[async_trait::async_trait]
impl ObjectStore for AzureStore {
    async fn list(&self, prefix: &str) -> Result<Vec<ObjectMeta>, ObjectStoreError> {
        // Pass `None` for an empty prefix: Azure list_blobs URL-encodes
        // `prefix=` and Azurite signs an empty value differently than
        // an absent one (treats it as a tampered query and returns
        // 403). Skipping the parameter is the wire-equivalent of "no
        // prefix filter" anyway.
        let prefix_opt = (!prefix.is_empty()).then(|| prefix.to_owned());
        let opts = BlobContainerClientListBlobsOptions {
            prefix: prefix_opt,
            ..Default::default()
        };
        let mut pages = self
            .container
            .list_blobs(Some(opts))
            .map_err(|e| classify(e, prefix))?
            .into_pages();

        let mut out = Vec::new();
        while let Some(page_result) = pages.next().await {
            let response = page_result.map_err(|e| classify(e, prefix))?;
            let body = response
                .into_body()
                .xml::<azure_storage_blob::models::ListBlobsResponse>()
                .map_err(|e| classify(e, prefix))?;
            for item in body.blob_items {
                let props = item.properties.unwrap_or_default();
                let meta = item_to_meta(
                    item.name.as_deref(),
                    props.content_length,
                    props.last_modified,
                    // Listing omits ETag for parity with S3 (avoid
                    // inflating per-object metadata for callers that
                    // only need a key/size enumeration).
                    None,
                )?;
                out.push(meta);
            }
        }
        Ok(out)
    }

    async fn get_to_file(
        &self,
        key: &str,
        dest: &Path,
        opts: GetOpts,
    ) -> Result<(), ObjectStoreError> {
        let parent = dest.parent().ok_or_else(|| {
            ObjectStoreError::Other(
                format!("destination `{}` has no parent directory", dest.display()).into(),
            )
        })?;

        // Mirror S3: try once, retry once on 412 (the head→GET race).
        // After the second attempt any error — including a repeated
        // 412 — propagates.
        let progress = opts.progress.as_ref();
        match self.head_then_download(key, dest, parent, progress).await {
            Err(ObjectStoreError::PreconditionFailed(_)) => {
                tracing::warn!(key, "blob changed between head and GET; retrying");
                self.head_then_download(key, dest, parent, progress).await
            }
            other => other,
        }
    }

    async fn get_bytes(&self, key: &str) -> Result<Bytes, ObjectStoreError> {
        let blob = self.blob_client(key);
        let result = blob.download(None).await.map_err(|e| classify(e, key))?;
        let bytes = result.body.collect().await.map_err(network_boxed)?;
        Ok(bytes)
    }

    /// Issue a Get Blob with an `HttpRange` covering `[start, end)`.
    /// HTTP 416 maps to [`ObjectStoreError::RangeNotSatisfiable`] with
    /// the original `Range<u64>` so the wire-line names what the
    /// caller asked for. All other failures route through [`classify`].
    ///
    /// Azure silently truncates a ranged GET to EOF when the requested
    /// range overruns the blob — `start < body.len() <= end` returns
    /// `start..body.len()` bytes with HTTP 206 and no error. The
    /// post-flight length check via [`super::verify_range_response_length`]
    /// elevates that mismatch to [`ObjectStoreError::RangeNotSatisfiable`]
    /// so callers (notably the packchain reader) cannot mistake a
    /// truncated slice for the full requested range.
    async fn get_bytes_range(
        &self,
        key: &str,
        range: std::ops::Range<u64>,
    ) -> Result<Bytes, ObjectStoreError> {
        if let Some(empty) = super::precheck_range(key, &range)? {
            return Ok(empty);
        }
        // `precheck_range` has already rejected `start > end`, so the
        // `HttpRange` length below cannot underflow.
        let opts = BlobClientDownloadOptions {
            range: Some(HttpRange::from(range.clone())),
            ..Default::default()
        };
        let blob = self.blob_client(key);
        let result = match blob.download(Some(opts)).await {
            Ok(result) => result,
            Err(err) => {
                if let azure_core::error::ErrorKind::HttpResponse { status, .. } = err.kind()
                    && u16::from(*status) == 416
                {
                    return Err(ObjectStoreError::RangeNotSatisfiable {
                        key: key.to_owned(),
                        requested: range,
                    });
                }
                return Err(classify(err, key));
            }
        };
        let bytes = result.body.collect().await.map_err(network_boxed)?;
        super::verify_range_response_length(key, &range, bytes)
    }

    async fn put_bytes(
        &self,
        key: &str,
        body: Bytes,
        opts: PutOpts,
    ) -> Result<(), ObjectStoreError> {
        // Same threshold as the S3 backend: above
        // [`MULTIPART_PUT_THRESHOLD`] use explicit `stage_block` +
        // `commit_block_list` so each block has its own retry budget,
        // predictable concurrency, and per-block progress events. Below
        // the threshold keep the single `Put Blob` round trip. Issue #53.
        let size = body.len() as u64;
        if should_use_multipart(size) {
            return self.multipart_put_bytes(key, body, size, opts).await;
        }
        let progress = opts.progress.clone();
        let blob = self.blob_client(key);
        let upload_opts = upload_options_from(opts);
        blob.upload(bytes_to_request_content(body), Some(upload_opts))
            .await
            .map_err(|e| classify(e, key))?;
        if let Some(sink) = progress
            && size > 0
        {
            sink.report(size);
        }
        Ok(())
    }

    /// Stream a local file to `key` without buffering its full body.
    ///
    /// Above [`super::multipart::MULTIPART_PUT_THRESHOLD`] this routes through explicit
    /// `stage_block` + `commit_block_list`, paralleling the S3 backend
    /// (issue #53). Below the threshold the single `Put Blob` path
    /// preserves the one-round-trip cost for small bundles and lock
    /// files.
    ///
    /// On the multipart path each task opens its own
    /// `tokio::fs::File`, seeks to its part offset, reads the part
    /// into a `Bytes`, then calls `BlockBlobClient::stage_block`. With
    /// `MULTIPART_PUT_MAX_CONCURRENCY = 8` and
    /// `MULTIPART_PUT_PART_SIZE = 16 MiB`, peak memory is bounded at
    /// 128 MiB regardless of file size.
    ///
    /// On the single-PUT path we wrap `tokio::fs::File` in
    /// [`FileStream`] so the body is delivered as
    /// `Body::SeekableStream`. The per-try signing policy reads
    /// `request.body().len()`, which `SeekableStream` reports faithfully
    /// via `len()`.
    async fn put_path(&self, key: &str, src: &Path, opts: PutOpts) -> Result<(), ObjectStoreError> {
        // Open the file once and read size from the open handle. This
        // closes the metadata/upload race that would let a concurrent
        // truncate or rename produce a body whose length disagrees
        // with the size we used for multipart planning.
        let file = tokio::fs::File::open(src).await.map_err(other_boxed)?;
        let body_len = file.metadata().await.map_err(other_boxed)?.len();
        if should_use_multipart(body_len) {
            return self.multipart_put_path(key, file, body_len, opts).await;
        }
        // Below the threshold: single `Put Blob`. Wrap our already-
        // open handle in `FileStream`; the SDK does not re-open by
        // path (which would re-introduce the race).
        let stream = FileStream::builder(file)
            .build()
            .await
            .map_err(other_boxed)?;
        let body: azure_core::http::Body = stream.into();

        let blob = self.blob_client(key);
        let progress = opts.progress.clone();
        let upload_opts = upload_options_from(opts);
        blob.upload(body.into(), Some(upload_opts))
            .await
            .map_err(|e| classify(e, key))?;
        if let Some(sink) = progress
            && body_len > 0
        {
            sink.report(body_len);
        }
        Ok(())
    }

    async fn put_if_absent(&self, key: &str, body: Bytes) -> Result<bool, ObjectStoreError> {
        let blob = self.blob_client(key);
        let upload_opts = BlockBlobClientUploadOptions::default().if_not_exists();
        let resp = blob
            .upload(bytes_to_request_content(body), Some(upload_opts))
            .await;
        match resp.map_err(|e| classify(e, key)) {
            Ok(_) => Ok(true),
            Err(ObjectStoreError::PreconditionFailed(_) | ObjectStoreError::Conflict(_)) => {
                Ok(false)
            }
            Err(other) => Err(other),
        }
    }

    async fn head(&self, key: &str) -> Result<ObjectMeta, ObjectStoreError> {
        let blob = self.blob_client(key);
        let resp = blob
            .get_properties(None::<BlobClientGetPropertiesOptions<'_>>)
            .await
            .map_err(|e| classify(e, key))?;
        let headers = resp.headers();
        properties_to_meta(
            key,
            header_u64(headers, &HeaderName::from_static("content-length")),
            header_http_date(headers, &HeaderName::from_static("last-modified")),
            headers.get_optional_str(&HeaderName::from_static("etag")),
        )
    }

    async fn copy(&self, src: &str, dst: &str) -> Result<(), ObjectStoreError> {
        // Server-side copy via `Put Blob From URL` requires a SAS-tokened
        // source URL or `x-ms-copy-source-authorization`, neither of
        // which integrates with our credential model in a clean way
        // for the SDK 1.0 surface. Stream `src` to a temp file via
        // `get_to_file` (chunked download, no body buffer), then
        // `put_path` it back to `dst` (block-uploaded for large bodies
        // via `multipart_put_path`). Peak in-flight bytes are bounded
        // by `MULTIPART_PUT_MAX_CONCURRENCY` × `MULTIPART_PUT_PART_SIZE`
        // regardless of blob size — necessary because `manage doctor`'s
        // duplicate-bundle quarantine path uses `copy()` and bundles
        // can be multi-GiB.
        let temp = NamedTempFile::new().map_err(other_boxed)?;
        // `get_to_file` propagates `NotFound(src)` if the source is
        // absent — exactly the trait contract for `copy`.
        self.get_to_file(src, temp.path(), GetOpts::default())
            .await?;
        // A NotFound on the upload is destination-side — re-shape it
        // so callers don't mistake it for "src absent".
        match self.put_path(dst, temp.path(), PutOpts::default()).await {
            Ok(()) => Ok(()),
            Err(ObjectStoreError::NotFound(_)) => Err(ObjectStoreError::Other(
                format!("copy `{src}` → `{dst}`: upload returned NotFound").into(),
            )),
            Err(other) => Err(other),
        }
    }

    async fn delete(&self, key: &str) -> Result<(), ObjectStoreError> {
        let blob = self.blob_client(key);
        blob.delete(None::<BlobClientDeleteOptions<'_>>)
            .await
            .map_err(|e| classify(e, key))?;
        Ok(())
    }

    /// Build a service-blob SAS URL for `key` valid for `ttl`.
    /// Used by the `bundle-uri` capability (issue #76) to advertise
    /// time-limited download URLs against private containers.
    ///
    /// Only the shared-key / connection-string credential paths can
    /// produce a SAS — the SAS env-var path has no key to re-sign
    /// with, and the Entra-ID `TokenCredential` path requires
    /// user-delegation SAS (out of scope per the issue). Both
    /// fall through to [`ObjectStoreError::Unsupported`].
    ///
    /// # Errors
    ///
    /// - [`ObjectStoreError::Unsupported`] when the credential is
    ///   not a shared key.
    /// - [`ObjectStoreError::Other`] when SAS construction fails
    ///   (HMAC init / base64 decode / time overflow).
    async fn presigned_get_url(
        &self,
        key: &str,
        ttl: std::time::Duration,
    ) -> Result<String, ObjectStoreError> {
        let signing = self.sas_signing.as_ref().ok_or_else(|| {
            ObjectStoreError::Unsupported(
                "Azure presigned URLs require a shared account key (KEY env var or \
                 connection string); SAS-env-var and Entra-ID credentials cannot \
                 derive per-blob SAS"
                    .to_owned(),
            )
        })?;
        // The SDK's `BlobClient::url()` returns the fully-qualified
        // blob URL including the container path segment. Reuse it
        // rather than re-deriving the URL shape per addressing
        // mode here.
        let blob = self.blob_client(key);
        let base = blob.url();
        sas::build_blob_sas_url(base, &self.container_name, key, signing, ttl)
    }
}

impl AzureStore {
    /// One head→tempfile→download→persist round trip.
    ///
    /// Factored out so [`get_to_file`](ObjectStore::get_to_file) can
    /// invoke it twice: once normally, once more on a 412 retry.
    async fn head_then_download(
        &self,
        key: &str,
        dest: &Path,
        parent: &Path,
        progress: Option<&ProgressSink>,
    ) -> Result<(), ObjectStoreError> {
        let meta = self.head(key).await?;
        let temp = NamedTempFile::new_in(parent).map_err(other_boxed)?;
        if meta.size == 0 {
            // Skip the GET entirely for zero-byte blobs (lock files):
            // `download_streaming` would issue a plain GET for an empty
            // body — correct but a wasted round trip.
            return persist_temp(temp, dest);
        }
        self.download_streaming(key, temp.path(), meta.etag.as_deref(), progress)
            .await?;
        persist_temp(temp, dest)
    }

    /// Stream a blob body to `temp_path` with optional `If-Match`
    /// guarding against mid-download mutation. When `progress` is
    /// `Some`, fires once per SDK body chunk read off the wire.
    async fn download_streaming(
        &self,
        key: &str,
        temp_path: &Path,
        etag: Option<&str>,
        progress: Option<&ProgressSink>,
    ) -> Result<(), ObjectStoreError> {
        let blob = self.blob_client(key);
        let mut opts = BlobClientDownloadOptions::default();
        if let Some(etag) = etag {
            opts.if_match = Some(etag.into());
        }
        let mut result = blob
            .download(Some(opts))
            .await
            .map_err(|e| classify(e, key))?;

        let mut file = tokio::fs::OpenOptions::new()
            .write(true)
            .truncate(true)
            .open(temp_path)
            .await
            .map_err(other_boxed)?;

        while let Some(chunk) = result.body.next().await {
            let bytes = chunk.map_err(network_boxed)?;
            let chunk_len = bytes.len() as u64;
            file.write_all(&bytes).await.map_err(other_boxed)?;
            if let Some(sink) = progress
                && chunk_len > 0
            {
                sink.report(chunk_len);
            }
        }
        file.flush().await.map_err(other_boxed)?;
        Ok(())
    }

    /// Drive a multipart upload from a fully-buffered `Bytes` body.
    ///
    /// `Bytes::slice` is zero-copy — every block borrows into the same
    /// underlying allocation, so peak memory equals the caller's body
    /// rather than `body × blocks`.
    async fn multipart_put_bytes(
        &self,
        key: &str,
        body: Bytes,
        size: u64,
        opts: PutOpts,
    ) -> Result<(), ObjectStoreError> {
        let parts = plan_upload_parts(size, MULTIPART_PUT_PART_SIZE, AZURE_MAX_BLOCKS);
        let progress = opts.progress.clone();
        let staged = self
            .stage_blocks_with_bodies(key, &parts, progress, |part| slice_bytes_part(&body, part))
            .await?;
        let blob = self.blob_client(key).block_blob_client();
        commit_block_list(&blob, key, staged, opts).await
    }

    /// Drive a multipart upload by streaming a local file block-by-block.
    ///
    /// All tasks share one `Arc<std::fs::File>`; per-task
    /// `read_file_part` uses `pread` so reads are concurrent without
    /// offset contention. Sharing one open file description closes
    /// the metadata/upload race. With `MULTIPART_PUT_MAX_CONCURRENCY
    /// = 8` and `MULTIPART_PUT_PART_SIZE = 16 MiB`, peak memory is
    /// bounded at 128 MiB regardless of file size.
    async fn multipart_put_path(
        &self,
        key: &str,
        file: tokio::fs::File,
        size: u64,
        opts: PutOpts,
    ) -> Result<(), ObjectStoreError> {
        let parts = plan_upload_parts(size, MULTIPART_PUT_PART_SIZE, AZURE_MAX_BLOCKS);
        let progress = opts.progress.clone();
        let file: Arc<std::fs::File> = Arc::new(file.into_std().await);
        let staged = self
            .stage_blocks_from_file(key, file, &parts, progress)
            .await?;
        let blob = self.blob_client(key).block_blob_client();
        commit_block_list(&blob, key, staged, opts).await
    }

    /// Spawn parallel `stage_block` tasks with bodies sourced from a
    /// closure (used by `multipart_put_bytes`).
    ///
    /// Returns the per-block IDs in part order so the caller can build
    /// a `BlockLookupList` for `commit_block_list`. On error,
    /// already-staged blocks are simply not committed and Azure
    /// auto-expires them after seven days; there is no client-side
    /// abort call.
    ///
    /// `BlockBlobClient` does not implement `Clone`, so each spawned
    /// task constructs its own via `self.blob_client(...)
    /// .block_blob_client()`. The container's `blob_client(&self, ..)`
    /// returns an owned `BlobClient` already (cheap-clone of internal
    /// `Arc` state), so this stays allocation-light.
    async fn stage_blocks_with_bodies<F>(
        &self,
        key: &str,
        parts: &[UploadPart],
        progress: Option<ProgressSink>,
        make_body: F,
    ) -> Result<Vec<Vec<u8>>, ObjectStoreError>
    where
        F: Fn(UploadPart) -> Result<Bytes, ObjectStoreError>,
    {
        let semaphore = Arc::new(Semaphore::new(MULTIPART_PUT_MAX_CONCURRENCY));
        let mut tasks: JoinSet<Result<(usize, Vec<u8>), ObjectStoreError>> = JoinSet::new();
        for (idx, part) in parts.iter().enumerate() {
            let part = *part;
            let part_index = idx;
            let block_id = block_id_for(idx);
            let body = make_body(part)?;
            let blob = self.blob_client(key).block_blob_client();
            let key = key.to_owned();
            let semaphore = Arc::clone(&semaphore);
            let progress = progress.clone();
            tasks.spawn(async move {
                let _permit = semaphore.acquire_owned().await.map_err(other_boxed)?;
                blob.stage_block(&block_id, part.length, bytes_to_request_content(body), None)
                    .await
                    .map_err(|e| classify(e, &key))?;
                if let Some(sink) = &progress {
                    sink.report(part.length);
                }
                Ok((part_index, block_id))
            });
        }
        join_staged_blocks(tasks, parts.len()).await
    }

    /// Spawn parallel `stage_block` tasks that each read their
    /// block from the shared `Arc<std::fs::File>` via `pread`. The
    /// shared open file description gives every task a stable view
    /// of the same inode (used by `multipart_put_path`).
    async fn stage_blocks_from_file(
        &self,
        key: &str,
        file: Arc<std::fs::File>,
        parts: &[UploadPart],
        progress: Option<ProgressSink>,
    ) -> Result<Vec<Vec<u8>>, ObjectStoreError> {
        let semaphore = Arc::new(Semaphore::new(MULTIPART_PUT_MAX_CONCURRENCY));
        let mut tasks: JoinSet<Result<(usize, Vec<u8>), ObjectStoreError>> = JoinSet::new();
        for (idx, part) in parts.iter().enumerate() {
            let part = *part;
            let part_index = idx;
            let block_id = block_id_for(idx);
            let blob = self.blob_client(key).block_blob_client();
            let key = key.to_owned();
            let task_file = Arc::clone(&file);
            let semaphore = Arc::clone(&semaphore);
            let progress = progress.clone();
            tasks.spawn(async move {
                let _permit = semaphore.acquire_owned().await.map_err(other_boxed)?;
                let body = read_file_part(task_file, part).await?;
                blob.stage_block(&block_id, part.length, bytes_to_request_content(body), None)
                    .await
                    .map_err(|e| classify(e, &key))?;
                if let Some(sink) = &progress {
                    sink.report(part.length);
                }
                Ok((part_index, block_id))
            });
        }
        join_staged_blocks(tasks, parts.len()).await
    }
}

/// Build a deterministic Azure block ID for the `idx`-th part
/// (zero-indexed).
///
/// Azure requires that all block IDs in a single
/// `commit_block_list` request share a length pre-base64. 32 bytes
/// of zero-padded ASCII digits accommodates up to 10^32 parts —
/// vastly above [`AZURE_MAX_BLOCKS`] = 50 000.
fn block_id_for(idx: usize) -> Vec<u8> {
    format!("{:032}", idx + 1).into_bytes()
}

/// Drain a `JoinSet` of `stage_block` tasks into a Vec of block IDs
/// indexed by part order. Short-circuits on the first error.
async fn join_staged_blocks(
    mut tasks: JoinSet<Result<(usize, Vec<u8>), ObjectStoreError>>,
    expected: usize,
) -> Result<Vec<Vec<u8>>, ObjectStoreError> {
    let mut staged: Vec<Option<Vec<u8>>> = (0..expected).map(|_| None).collect();
    while let Some(joined) = tasks.join_next().await {
        let (idx, block_id) = joined.map_err(other_boxed)??;
        staged[idx] = Some(block_id);
    }
    staged
        .into_iter()
        .enumerate()
        .map(|(idx, slot)| {
            slot.ok_or_else(|| {
                ObjectStoreError::Other(
                    format!("internal: stage_block task for part {idx} did not return").into(),
                )
            })
        })
        .collect()
}

/// Commit the staged blocks in order, applying any
/// `content_disposition` / `user_metadata` from the original `PutOpts`.
///
/// Azure has no `AbortMultipartUpload` equivalent: if commit fails
/// the staged blocks remain on the storage account and expire
/// automatically (default seven days). Surface the commit error
/// directly — the caller's error handling already understands the
/// "operation did not succeed" outcome.
async fn commit_block_list(
    blob: &BlockBlobClient,
    key: &str,
    block_ids: Vec<Vec<u8>>,
    opts: PutOpts,
) -> Result<(), ObjectStoreError> {
    let block_list = BlockLookupList {
        latest: Some(block_ids),
        ..Default::default()
    };
    let body: RequestContent<_, _> = block_list.try_into().map_err(other_boxed)?;
    let (cd, metadata) = put_opts_blob_fields(opts);
    let commit_opts = BlockBlobClientCommitBlockListOptions {
        blob_content_disposition: cd,
        metadata,
        ..Default::default()
    };
    blob.commit_block_list(body, Some(commit_opts))
        .await
        .map_err(|e| classify(e, key))?;
    Ok(())
}

/// Wrap `Bytes` in a `RequestContent` without copying the buffer.
///
/// `RequestContent` has an inherent `from(Vec<u8>)` constructor that
/// shadows the generic `From<Bytes>` trait impl, so a bare
/// `RequestContent::from(body)` resolves to the `Vec<u8>` overload and
/// re-allocates. Going through `Into` instead picks up the trait impl
/// and keeps the `Bytes` payload zero-copy. The return type is left
/// generic so the call site (which pins `Bytes` + `NoFormat` via the
/// `BlobClient::upload` signature) drives type inference.
fn bytes_to_request_content<F>(body: Bytes) -> RequestContent<Bytes, F>
where
    Bytes: Into<RequestContent<Bytes, F>>,
{
    body.into()
}

/// Pull the blob-shaped `content_disposition` and `metadata` fields
/// out of [`PutOpts`].
///
/// Both `BlockBlobClientUploadOptions` (single `Put Blob`) and
/// `BlockBlobClientCommitBlockListOptions` (multipart commit) carry
/// the same two fields by the same names. Centralising the
/// conversion here keeps a single source of truth for "how a
/// `PutOpts` becomes Azure blob metadata."
fn put_opts_blob_fields(
    opts: PutOpts,
) -> (
    Option<String>,
    Option<std::collections::HashMap<String, String>>,
) {
    let metadata = (!opts.user_metadata.is_empty()).then(|| {
        opts.user_metadata
            .into_iter()
            .collect::<std::collections::HashMap<_, _>>()
    });
    (opts.content_disposition, metadata)
}

/// Build a [`BlockBlobClientUploadOptions`] from the trait's [`PutOpts`].
fn upload_options_from(opts: PutOpts) -> BlockBlobClientUploadOptions<'static> {
    let (cd, metadata) = put_opts_blob_fields(opts);
    BlockBlobClientUploadOptions {
        blob_content_disposition: cd,
        metadata,
        ..Default::default()
    }
}

fn header_u64(headers: &Headers, name: &HeaderName) -> Option<u64> {
    headers.get_optional_str(name).and_then(|s| s.parse().ok())
}

fn header_http_date(headers: &Headers, name: &HeaderName) -> Option<OffsetDateTime> {
    let raw = headers.get_optional_str(name)?;
    OffsetDateTime::parse(raw, &time::format_description::well_known::Rfc2822).ok()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::url::{AzureAddressing, RemoteFlags};

    fn parse_endpoint(s: &str) -> Url {
        Url::parse(s).expect("test endpoint URL parses")
    }

    fn s3_url() -> RemoteUrl {
        RemoteUrl::S3 {
            endpoint: parse_endpoint("https://my-bucket.s3.us-west-2.amazonaws.com/"),
            bucket: "my-bucket".to_owned(),
            prefix: None,
            addressing: crate::url::S3Addressing::VirtualHosted,
            flags: RemoteFlags::default(),
        }
    }

    // --- build_container_url ------------------------------------------

    #[test]
    fn build_container_url_virtual_hosted_strips_prefix() {
        let url = parse_endpoint("https://acct.blob.core.windows.net/my-container/some/prefix");
        let out = build_container_url(&url, "acct", "my-container", AzureAddressing::VirtualHosted);
        assert_eq!(
            out.as_str(),
            "https://acct.blob.core.windows.net/my-container"
        );
    }

    #[test]
    fn build_container_url_path_style_keeps_account() {
        let url = parse_endpoint("http://127.0.0.1:10000/devstoreaccount1/my-container/repo");
        let out = build_container_url(
            &url,
            "devstoreaccount1",
            "my-container",
            AzureAddressing::PathStyle,
        );
        assert_eq!(
            out.as_str(),
            "http://127.0.0.1:10000/devstoreaccount1/my-container"
        );
    }

    #[test]
    fn build_container_url_strips_query_and_fragment() {
        let url = parse_endpoint("https://acct.blob.core.windows.net/c/r?credential=foo#frag");
        let out = build_container_url(&url, "acct", "c", AzureAddressing::VirtualHosted);
        assert_eq!(out.as_str(), "https://acct.blob.core.windows.net/c");
    }

    // --- classify_status_and_code -------------------------------------

    #[test]
    fn classify_404_is_not_found() {
        assert!(matches!(
            classify_status_and_code(404, None, "k"),
            Some(ObjectStoreError::NotFound(s)) if s == "k"
        ));
    }

    #[test]
    fn classify_403_is_access_denied() {
        assert!(matches!(
            classify_status_and_code(403, None, "k"),
            Some(ObjectStoreError::AccessDenied(s)) if s == "k"
        ));
    }

    #[test]
    fn classify_412_is_precondition_failed() {
        assert!(matches!(
            classify_status_and_code(412, None, "k"),
            Some(ObjectStoreError::PreconditionFailed(s)) if s == "k"
        ));
    }

    #[test]
    fn classify_409_is_conflict() {
        // 409 covers Azure's `BlobAlreadyExists` (the put-if-absent
        // contention path). Without this branch, `put_if_absent` would
        // surface contention as a hard error instead of `Ok(false)`.
        assert!(matches!(
            classify_status_and_code(409, None, "k"),
            Some(ObjectStoreError::Conflict(s)) if s == "k"
        ));
    }

    #[test]
    fn classify_413_is_payload_too_large() {
        // Pass `code=None` so the assertion isolates the 413-status
        // branch; passing a recognised code would still pass even if
        // the status arm regressed (the code arm would catch it). The
        // canonical "Payload Too Large" status alone suffices.
        assert!(matches!(
            classify_status_and_code(413, None, "k"),
            Some(ObjectStoreError::PayloadTooLarge { limit_bytes })
                if limit_bytes == SINGLE_PUT_BLOB_LIMIT_BYTES
        ));
    }

    #[test]
    fn classify_request_body_too_large_code_is_payload_too_large() {
        // Defensive backstop: if the SDK delivers the service code on a
        // non-413 status (e.g. 400), the code branch still catches it.
        assert!(matches!(
            classify_status_and_code(400, Some("RequestBodyTooLarge"), "k"),
            Some(ObjectStoreError::PayloadTooLarge { limit_bytes })
                if limit_bytes == SINGLE_PUT_BLOB_LIMIT_BYTES
        ));
    }

    #[test]
    fn classify_unrecognised_status_returns_none() {
        assert!(classify_status_and_code(500, None, "k").is_none());
        assert!(classify_status_and_code(429, None, "k").is_none());
        assert!(classify_status_and_code(500, Some("InternalError"), "k").is_none());
    }

    // --- properties_to_meta ------------------------------------------

    #[test]
    fn properties_to_meta_round_trips_well_formed_response() {
        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap();
        let meta = properties_to_meta("k", Some(42), Some(now), Some("\"abc\""))
            .expect("conversion succeeds");
        assert_eq!(meta.key, "k");
        assert_eq!(meta.size, 42);
        assert_eq!(meta.last_modified.unix_timestamp(), 1_700_000_000);
        assert_eq!(meta.etag.as_deref(), Some("\"abc\""));
    }

    #[test]
    fn properties_to_meta_preserves_legitimate_zero_size() {
        // Zero-byte lock files are legitimate; a present
        // `Content-Length: 0` header (`Some(0)`) must round-trip as
        // `size == 0`, distinct from the missing-header error.
        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap();
        let meta =
            properties_to_meta("LOCK", Some(0), Some(now), None).expect("conversion succeeds");
        assert_eq!(meta.size, 0);
    }

    #[test]
    fn properties_to_meta_rejects_missing_content_length() {
        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap();
        let err = properties_to_meta("k", None, Some(now), None)
            .expect_err("missing content-length must error");
        match err {
            ObjectStoreError::Other(inner) => {
                let msg = inner.to_string();
                assert!(msg.contains("no content-length"), "names failure: {msg}");
                assert!(msg.contains("`k`"), "includes the key for context: {msg}");
            }
            other => {
                panic!("expected ObjectStoreError::Other for missing content-length, got {other:?}")
            }
        }
    }

    #[test]
    fn properties_to_meta_rejects_missing_last_modified() {
        let err = properties_to_meta("k", Some(0), None, None)
            .expect_err("missing last_modified must error");
        match err {
            ObjectStoreError::Other(inner) => {
                let msg = inner.to_string();
                assert!(msg.contains("no last-modified"), "names failure: {msg}");
                assert!(msg.contains("`k`"), "includes the key for context: {msg}");
            }
            other => {
                panic!("expected ObjectStoreError::Other for missing last_modified, got {other:?}")
            }
        }
    }

    // --- item_to_meta -------------------------------------------------

    #[test]
    fn item_to_meta_round_trips_well_formed_item() {
        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap();
        let meta = item_to_meta(Some("k"), Some(42), Some(now), Some("\"abc\"")).unwrap();
        assert_eq!(meta.key, "k");
        assert_eq!(meta.size, 42);
        assert_eq!(meta.last_modified.unix_timestamp(), 1_700_000_000);
        assert_eq!(meta.etag.as_deref(), Some("\"abc\""));
    }

    #[test]
    fn item_to_meta_rejects_missing_name() {
        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap();
        let err = item_to_meta(None, Some(0), Some(now), None).unwrap_err();
        match err {
            ObjectStoreError::Other(inner) => {
                assert!(
                    inner.to_string().contains("without a name"),
                    "names failure: {inner}"
                );
            }
            other => panic!("expected ObjectStoreError::Other, got {other:?}"),
        }
    }

    #[test]
    fn item_to_meta_rejects_missing_last_modified() {
        let err = item_to_meta(Some("k"), Some(0), None, None).unwrap_err();
        match err {
            ObjectStoreError::Other(inner) => {
                let msg = inner.to_string();
                assert!(
                    msg.contains("without last_modified"),
                    "names failure: {msg}"
                );
                assert!(msg.contains("`k`"), "includes the key: {msg}");
            }
            other => panic!("expected ObjectStoreError::Other, got {other:?}"),
        }
    }

    #[test]
    fn item_to_meta_treats_missing_size_as_zero() {
        // The Azure SDK types content_length as Option<u64>; missing
        // values default to 0 (rather than `None` propagating through
        // every caller's arithmetic).
        let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap();
        let meta = item_to_meta(Some("k"), None, Some(now), None).unwrap();
        assert_eq!(meta.size, 0);
    }

    // --- upload_options_from ------------------------------------------

    #[test]
    fn upload_options_from_default_is_empty() {
        let out = upload_options_from(PutOpts::default());
        assert!(out.blob_content_disposition.is_none());
        assert!(out.metadata.is_none());
    }

    #[test]
    fn upload_options_from_carries_content_disposition() {
        let opts = PutOpts {
            content_disposition: Some("attachment; filename=x".into()),
            user_metadata: Vec::new(),
            progress: None,
        };
        let out = upload_options_from(opts);
        let cd: String = out
            .blob_content_disposition
            .expect("content_disposition should be set");
        assert!(cd.contains("attachment"));
    }

    #[test]
    fn upload_options_from_collects_metadata() {
        let opts = PutOpts {
            content_disposition: None,
            user_metadata: vec![("x-foo".into(), "1".into()), ("x-bar".into(), "2".into())],
            progress: None,
        };
        let out = upload_options_from(opts);
        let map = out.metadata.expect("metadata set");
        assert_eq!(map.get("x-foo").map(String::as_str), Some("1"));
        assert_eq!(map.get("x-bar").map(String::as_str), Some("2"));
    }

    // --- from_remote_url constructor branch ---------------------------

    #[tokio::test]
    async fn from_remote_url_rejects_s3() {
        let result = AzureStore::from_remote_url(&s3_url()).await;
        match result {
            Err(ObjectStoreError::Other(_)) => {}
            Err(other) => panic!("expected ObjectStoreError::Other, got {other:?}"),
            Ok(_) => panic!("expected S3 URL to be rejected"),
        }
    }

    // --- HTTP transport tuning (#26 / #28) ----------------------------

    /// Pin the timeout values. A future copy-paste mistake (`from_millis`
    /// instead of `from_secs`, an accidental zero) silently disables
    /// the very behaviour these constants exist for; fail fast instead.
    /// If the constants are deliberately changed, update the expected
    /// values on the right-hand side together — the test exists to make
    /// such changes deliberate, not to lock the values forever.
    #[test]
    fn transport_timeout_constants_have_expected_values() {
        assert_eq!(POOL_IDLE_TIMEOUT, Duration::from_secs(30));
        assert_eq!(TCP_KEEPALIVE, Duration::from_secs(30));
        assert_eq!(CONNECT_TIMEOUT, Duration::from_secs(10));
        assert_eq!(READ_TIMEOUT, Duration::from_secs(30));
    }

    #[test]
    fn build_http_client_succeeds() {
        build_http_client().expect("reqwest client builds with the configured timeouts");
    }

    /// The meaningful regression check: if a future refactor drops the
    /// `transport = Some(...)` line in `build_client_options`, the
    /// Azure backend silently reverts to `azure_core`'s default
    /// (unbounded) HTTP transport. This test fails when that happens.
    /// Also pins the empty-policies invariant on the no-credential
    /// branch, so a refactor that injects a fallback policy when
    /// `per_try_policy` is `None` is caught.
    #[test]
    fn build_client_options_installs_custom_transport() {
        let resolved = auth::ResolvedCredentials {
            token_credential: None,
            per_try_policy: None,
            sas_signing_key: None,
        };
        let opts = build_client_options(&resolved).expect("client options build");
        assert!(
            opts.transport.is_some(),
            "ClientOptions::transport must be Some so the SDK uses our \
             pool_idle_timeout / tcp_keepalive client (issue #26)",
        );
        assert!(
            opts.per_try_policies.is_empty(),
            "no per-try policy was supplied; the helper must not inject \
             a fallback signer of its own",
        );
    }

    /// Issue #28's Notes section explicitly calls out: the per-try
    /// signing policy must continue to fire after we install a custom
    /// transport. The SDK pipeline runs them independently of the
    /// transport, but a future refactor that confuses the two fields
    /// would silently drop signing — surface that here. The
    /// [`Arc::ptr_eq`] check pins identity so a refactor that
    /// silently *replaces* the caller's policy with a fresh one
    /// (rather than dropping it outright) also fails.
    #[test]
    fn build_client_options_preserves_per_try_policy() {
        // Azurite's published well-known account key — base64-valid
        // and safe to embed.
        const AZURITE_KEY: &str = "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
        let policy: Arc<dyn azure_core::http::policies::Policy> = Arc::new(
            auth::SharedKeySigningPolicy::new("devstoreaccount1", AZURITE_KEY)
                .expect("shared-key policy constructs"),
        );
        let resolved = auth::ResolvedCredentials {
            token_credential: None,
            per_try_policy: Some(Arc::clone(&policy)),
            sas_signing_key: None,
        };
        let opts = build_client_options(&resolved).expect("client options build");
        assert!(opts.transport.is_some(), "transport still wired");
        assert_eq!(
            opts.per_try_policies.len(),
            1,
            "exactly one per-try policy is wired",
        );
        assert!(
            Arc::ptr_eq(&policy, &opts.per_try_policies[0]),
            "the policy at index 0 must be the same Arc the caller \
             supplied — not a fresh policy constructed inside the helper",
        );
    }

    /// Pin the `should_use_multipart` predicate at and around the
    /// shared threshold (issue #53).
    ///
    /// `put_bytes` and `put_path` route through this predicate. The
    /// integration test `multipart_put_emits_per_block_progress_events`
    /// covers the dispatch *call* (only multipart emits per-block
    /// events). This unit test pins the predicate's boundary semantics
    /// so the constant can't be moved out from under that test
    /// without something failing. The Azure backend uses the same
    /// shared `MULTIPART_PUT_THRESHOLD` as S3 so a future refactor
    /// cannot accidentally raise the threshold for one backend alone.
    #[test]
    fn should_use_multipart_pins_threshold_boundary() {
        use super::super::multipart::MULTIPART_PUT_THRESHOLD;
        assert!(!should_use_multipart(MULTIPART_PUT_THRESHOLD - 1));
        assert!(should_use_multipart(MULTIPART_PUT_THRESHOLD));
        assert!(should_use_multipart(MULTIPART_PUT_THRESHOLD + 1));
        assert!(should_use_multipart(6 * (1 << 30)));
    }

    /// Pin `block_id_for(idx)` so two parts can never collide on the
    /// same block ID, and so all IDs in a single `commit_block_list`
    /// share a length pre-base64 (Azure's hard requirement).
    #[test]
    fn block_id_for_is_unique_and_uniform_length() {
        let id_a = block_id_for(0);
        let id_b = block_id_for(1);
        let id_c = block_id_for(99_999);
        assert_eq!(id_a.len(), id_b.len(), "all IDs share length");
        assert_eq!(id_a.len(), id_c.len(), "even at the upper end");
        assert_ne!(id_a, id_b, "two parts get distinct IDs");
        // 32 ASCII bytes accommodates up to 10^32 parts — vastly above
        // [`AZURE_MAX_BLOCKS`] = 50 000.
        assert_eq!(id_a.len(), 32);
    }
}