r2kit 0.1.0

use std::{
    collections::BTreeMap,
    fmt,
    num::NonZeroU16,
    time::{Duration, SystemTime, UNIX_EPOCH},
};

use aws_sdk_s3::{
    presigning::PresigningConfig,
    types::{CompletedMultipartUpload, CompletedPart},
};
use base64::{Engine as _, engine::general_purpose::STANDARD};

use crate::{Bucket, Error, ObjectUploadOptions, ValidationError, validation};

// https://developers.cloudflare.com/r2/platform/limits/
const MAX_MULTIPART_OBJECT_SIZE: u64 = 5 * 1024 * 1024 * 1024 * 1024 - 5 * 1024 * 1024 * 1024;
const MAX_PARTS: u16 = 10_000;

/// A validated multipart part number in the range `1..=10_000`.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct PartNumber(NonZeroU16);

impl PartNumber {
    /// Returns the numeric part number.
    #[must_use]
    pub const fn get(self) -> u16 {
        self.0.get()
    }
}

impl TryFrom<u16> for PartNumber {
    type Error = Error;

    fn try_from(value: u16) -> Result<Self, Self::Error> {
        let provided = value;
        let value = NonZeroU16::new(value).ok_or(ValidationError::PartNumberOutOfRange {
            provided,
            min: 1,
            max: MAX_PARTS,
        })?;
        if value.get() > MAX_PARTS {
            return Err(ValidationError::PartNumberOutOfRange {
                provided,
                min: 1,
                max: MAX_PARTS,
            }
            .into());
        }
        Ok(Self(value))
    }
}

/// A successfully uploaded multipart part.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct UploadedPart {
    part_number: PartNumber,
    etag: String,
}

/// A validated Base64-encoded 128-bit MD5 digest for one multipart part.
///
/// R2 checks this value against the request body when it is included in a
/// presigned `UploadPart` request.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PartMd5(String);

impl PartMd5 {
    /// Returns the canonical Base64 representation expected by `Content-MD5`.
    #[must_use]
    pub fn as_base64(&self) -> &str {
        &self.0
    }
}

impl TryFrom<String> for PartMd5 {
    type Error = Error;

    fn try_from(value: String) -> Result<Self, Self::Error> {
        let decoded = STANDARD.decode(&value).map_err(|_| Error::InvalidInput {
            field: "content_md5",
            reason: "must be canonical Base64 for a 128-bit MD5 digest",
        })?;
        if decoded.len() != 16 || STANDARD.encode(decoded) != value {
            return Err(Error::InvalidInput {
                field: "content_md5",
                reason: "must be canonical Base64 for a 128-bit MD5 digest",
            });
        }
        Ok(Self(value))
    }
}

impl TryFrom<&str> for PartMd5 {
    type Error = Error;

    fn try_from(value: &str) -> Result<Self, Self::Error> {
        Self::try_from(value.to_owned())
    }
}

/// Untrusted transport data reported by a direct multipart uploader.
///
/// Convert this value with [`MultipartPartReceipt::try_into_uploaded_part`]
/// before using it in a completion manifest.
#[derive(Clone, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
pub struct MultipartPartReceipt {
    part_number: u16,
    etag: String,
}

impl MultipartPartReceipt {
    /// Creates a wire receipt from primitive values.
    #[must_use]
    pub fn new(part_number: u16, etag: impl Into<String>) -> Self {
        Self {
            part_number,
            etag: etag.into(),
        }
    }

    /// Returns the unvalidated numeric part number.
    #[must_use]
    pub const fn part_number(&self) -> u16 {
        self.part_number
    }

    /// Returns the exact ETag reported by R2.
    #[must_use]
    pub fn etag(&self) -> &str {
        &self.etag
    }

    /// Validates this untrusted receipt for use in a completion manifest.
    pub fn try_into_uploaded_part(self) -> Result<UploadedPart, Error> {
        UploadedPart::new(PartNumber::try_from(self.part_number)?, self.etag)
    }
}

impl UploadedPart {
    /// Creates a completion entry from the exact ETag returned by R2.
    pub fn new(part_number: PartNumber, etag: impl Into<String>) -> Result<Self, Error> {
        let etag = etag.into();
        if etag.is_empty() {
            return Err(Error::InvalidInput {
                field: "etag",
                reason: "must not be empty",
            });
        }
        Ok(Self { part_number, etag })
    }

    /// Returns the part number.
    #[must_use]
    pub const fn part_number(&self) -> PartNumber {
        self.part_number
    }

    /// Returns the exact R2 ETag, including quotes when R2 supplied them.
    #[must_use]
    pub fn etag(&self) -> &str {
        &self.etag
    }
}

/// A canonical, duplicate-free completion manifest sorted by part number.
#[derive(Clone, Debug)]
pub struct CompletionManifest(Vec<UploadedPart>);

impl CompletionManifest {
    /// Validates and canonicalizes uploaded parts.
    pub fn try_from_parts(parts: impl IntoIterator<Item = UploadedPart>) -> Result<Self, Error> {
        let mut canonical = BTreeMap::new();
        for part in parts {
            let number = part.part_number();
            if canonical.insert(number, part).is_some() {
                return Err(Error::InvalidInput {
                    field: "parts",
                    reason: "contains a duplicate part number",
                });
            }
        }
        if canonical.is_empty() {
            return Err(Error::InvalidInput {
                field: "parts",
                reason: "must contain at least one uploaded part",
            });
        }
        Ok(Self(canonical.into_values().collect()))
    }

    /// Validates untrusted uploader receipts and builds a canonical manifest.
    pub fn try_from_receipts(
        receipts: impl IntoIterator<Item = MultipartPartReceipt>,
    ) -> Result<Self, Error> {
        receipts
            .into_iter()
            .map(MultipartPartReceipt::try_into_uploaded_part)
            .collect::<Result<Vec<_>, _>>()
            .and_then(Self::try_from_parts)
    }

    /// Iterates over canonical parts in ascending part-number order.
    pub fn parts(&self) -> impl ExactSizeIterator<Item = &UploadedPart> {
        self.0.iter()
    }
}

/// A presigned URL treated as a bearer credential.
#[derive(Clone)]
pub struct SecretUrl(String);

impl SecretUrl {
    /// Deliberately exposes the bearer URL for transmission to an uploader.
    #[must_use]
    pub fn expose(&self) -> &str {
        &self.0
    }

    /// Deliberately consumes and exposes the bearer URL.
    #[must_use]
    pub fn into_exposed_string(self) -> String {
        self.0
    }
}

impl fmt::Debug for SecretUrl {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("SecretUrl([REDACTED PRESIGNED URL])")
    }
}

/// An HTTP request signed for temporary direct access to R2.
#[derive(Clone)]
pub struct PresignedRequest {
    method: String,
    url: SecretUrl,
    required_headers: Vec<(String, String)>,
    expires_at: SystemTime,
}

impl PresignedRequest {
    pub(crate) fn from_sdk(
        signed: aws_sdk_s3::presigning::PresignedRequest,
        expires_in: Duration,
    ) -> Result<Self, Error> {
        let required_headers = signed
            .headers()
            .map(|(name, value)| (name.to_owned(), value.to_owned()))
            .collect();
        Ok(Self {
            method: signed.method().to_owned(),
            url: SecretUrl(signed.uri().to_string()),
            required_headers,
            expires_at: SystemTime::now() + expires_in,
        })
    }

    /// Returns the HTTP method the uploader must use.
    #[must_use]
    pub fn method(&self) -> &str {
        &self.method
    }

    /// Returns the redacted-by-default bearer URL wrapper.
    #[must_use]
    pub fn url(&self) -> &SecretUrl {
        &self.url
    }

    /// Returns headers that must be replayed exactly by the uploader.
    pub fn required_headers(&self) -> impl ExactSizeIterator<Item = (&str, &str)> {
        self.required_headers
            .iter()
            .map(|(name, value)| (name.as_str(), value.as_str()))
    }

    /// Returns the approximate expiration instant.
    #[must_use]
    pub const fn expires_at(&self) -> SystemTime {
        self.expires_at
    }

    /// Deliberately exposes all request components for an HTTP client.
    #[must_use]
    pub fn into_exposed_parts(self) -> (String, String, Vec<(String, String)>) {
        (
            self.method,
            self.url.into_exposed_string(),
            self.required_headers,
        )
    }
}

impl fmt::Debug for PresignedRequest {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let header_names: Vec<&str> = self
            .required_headers
            .iter()
            .map(|(name, _)| name.as_str())
            .collect();
        f.debug_struct("PresignedRequest")
            .field("method", &self.method)
            .field("url", &"[REDACTED PRESIGNED URL]")
            .field("required_header_names", &header_names)
            .field("expires_at", &self.expires_at)
            .finish()
    }
}

/// A presigned upload request for one known multipart part.
#[derive(Clone, Debug)]
pub struct PresignedUploadPart {
    part_number: PartNumber,
    content_length: u64,
    content_md5: Option<PartMd5>,
    request: PresignedRequest,
}

impl PresignedUploadPart {
    /// Returns the part number.
    #[must_use]
    pub const fn part_number(&self) -> PartNumber {
        self.part_number
    }

    /// Returns the exact number of bytes expected for this part.
    #[must_use]
    pub const fn content_length(&self) -> u64 {
        self.content_length
    }

    /// Returns the body checksum enforced by R2, when one was signed.
    #[must_use]
    pub fn content_md5(&self) -> Option<&PartMd5> {
        self.content_md5.as_ref()
    }

    /// Returns the signed request.
    #[must_use]
    pub fn request(&self) -> &PresignedRequest {
        &self.request
    }

    /// Consumes the part wrapper and returns the signed request.
    #[must_use]
    pub fn into_request(self) -> PresignedRequest {
        self.request
    }

    /// Deliberately exposes the bearer request as a serializable protocol DTO.
    ///
    /// The resulting value still redacts its URL and header values from
    /// `Debug`, but serialization exposes them for transport to an uploader.
    pub fn into_protocol_request(self) -> Result<MultipartUploadPartRequest, Error> {
        let expires_at_unix_seconds = self
            .request
            .expires_at()
            .duration_since(UNIX_EPOCH)
            .map_err(|_| Error::Presign)?
            .as_secs();
        let (method, url, required_headers) = self.request.into_exposed_parts();
        Ok(MultipartUploadPartRequest {
            part_number: self.part_number.get(),
            content_length: self.content_length,
            content_md5: self.content_md5.map(|value| value.0),
            method,
            url,
            required_headers,
            expires_at_unix_seconds,
        })
    }
}

/// Serializable protocol DTO sent from a trusted signer to an uploader.
///
/// This value contains a bearer URL. Serialization is therefore an explicit
/// secret-exposure boundary even though `Debug` remains redacted. Its primitive
/// fields are a transport representation of an already validated and signed
/// request; deserializing this DTO does not establish a new trusted validation
/// boundary.
#[derive(Clone)]
#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
pub struct MultipartUploadPartRequest {
    part_number: u16,
    content_length: u64,
    content_md5: Option<String>,
    method: String,
    url: String,
    required_headers: Vec<(String, String)>,
    expires_at_unix_seconds: u64,
}

impl MultipartUploadPartRequest {
    /// Returns the part number this request may upload.
    #[must_use]
    pub const fn part_number(&self) -> u16 {
        self.part_number
    }

    /// Returns the exact required request-body length.
    #[must_use]
    pub const fn content_length(&self) -> u64 {
        self.content_length
    }

    /// Returns the signed Base64 `Content-MD5`, when enabled.
    #[must_use]
    pub fn content_md5(&self) -> Option<&str> {
        self.content_md5.as_deref()
    }

    /// Returns the signed HTTP method.
    #[must_use]
    pub fn method(&self) -> &str {
        &self.method
    }

    /// Deliberately exposes the bearer URL to the uploader.
    #[must_use]
    pub fn expose_url(&self) -> &str {
        &self.url
    }

    /// Returns headers that the uploader must replay exactly.
    pub fn required_headers(&self) -> impl ExactSizeIterator<Item = (&str, &str)> {
        self.required_headers
            .iter()
            .map(|(name, value)| (name.as_str(), value.as_str()))
    }

    /// Returns the approximate Unix expiration timestamp in seconds.
    #[must_use]
    pub const fn expires_at_unix_seconds(&self) -> u64 {
        self.expires_at_unix_seconds
    }
}

impl fmt::Debug for MultipartUploadPartRequest {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let header_names: Vec<&str> = self
            .required_headers
            .iter()
            .map(|(name, _)| name.as_str())
            .collect();
        f.debug_struct("MultipartUploadPartRequest")
            .field("part_number", &self.part_number)
            .field("content_length", &self.content_length)
            .field("content_md5", &self.content_md5)
            .field("method", &self.method)
            .field("url", &"[REDACTED PRESIGNED URL]")
            .field("required_header_names", &header_names)
            .field("expires_at_unix_seconds", &self.expires_at_unix_seconds)
            .finish()
    }
}

/// Persistable state needed to resume a presigned multipart upload.
#[derive(Clone)]
pub struct MultipartSessionSnapshot {
    bucket: String,
    key: String,
    upload_id: String,
    file_size: u64,
    part_size: u64,
}

/// Versioned persistence DTO for resuming a multipart session.
///
/// This value contains an upload ID. Serialization deliberately exposes that
/// credential to the selected persistence layer, while `Debug` redacts it.
#[derive(Clone)]
#[cfg_attr(feature = "serde", derive(serde::Deserialize, serde::Serialize))]
pub struct MultipartSessionRecord {
    version: u8,
    bucket: String,
    key: String,
    upload_id: String,
    file_size: u64,
    part_size: u64,
}

impl MultipartSessionRecord {
    /// Returns the persistence schema version.
    #[must_use]
    pub const fn version(&self) -> u8 {
        self.version
    }

    /// Returns the bucket stored in this untrusted record.
    #[must_use]
    pub fn bucket(&self) -> &str {
        &self.bucket
    }

    /// Returns the key stored in this untrusted record.
    #[must_use]
    pub fn key(&self) -> &str {
        &self.key
    }

    /// Deliberately exposes the persisted upload ID.
    #[must_use]
    pub fn expose_upload_id(&self) -> &str {
        &self.upload_id
    }

    /// Returns the planned object size.
    #[must_use]
    pub const fn file_size(&self) -> u64 {
        self.file_size
    }

    /// Returns the planned part size.
    #[must_use]
    pub const fn part_size(&self) -> u64 {
        self.part_size
    }
}

impl fmt::Debug for MultipartSessionRecord {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("MultipartSessionRecord")
            .field("version", &self.version)
            .field("bucket", &self.bucket)
            .field("key", &self.key)
            .field("upload_id", &"[REDACTED]")
            .field("file_size", &self.file_size)
            .field("part_size", &self.part_size)
            .finish()
    }
}

impl MultipartSessionSnapshot {
    /// Restores validated multipart session state previously returned by [`PresignedMultipart::snapshot`].
    pub fn restore(
        bucket: impl Into<String>,
        key: impl Into<String>,
        upload_id: impl Into<String>,
        file_size: u64,
        part_size: u64,
    ) -> Result<Self, Error> {
        let key = key.into();
        validation::validate_key(&key)?;
        MultipartPlan::new(file_size, part_size)?;
        let upload_id = upload_id.into();
        if upload_id.is_empty() {
            return Err(Error::InvalidInput {
                field: "upload_id",
                reason: "must not be empty",
            });
        }
        Ok(Self {
            bucket: bucket.into(),
            key,
            upload_id,
            file_size,
            part_size,
        })
    }

    /// Validates a deserialized persistence record before it becomes session state.
    pub fn from_persistence_record(record: MultipartSessionRecord) -> Result<Self, Error> {
        if record.version != 1 {
            return Err(Error::InvalidInput {
                field: "version",
                reason: "unsupported multipart session record version",
            });
        }
        Self::restore(
            record.bucket,
            record.key,
            record.upload_id,
            record.file_size,
            record.part_size,
        )
    }

    /// Deliberately exposes resumable state as a versioned persistence DTO.
    #[must_use]
    pub fn into_persistence_record(self) -> MultipartSessionRecord {
        MultipartSessionRecord {
            version: 1,
            bucket: self.bucket,
            key: self.key,
            upload_id: self.upload_id,
            file_size: self.file_size,
            part_size: self.part_size,
        }
    }

    /// Deliberately exposes the upload ID for persistence.
    #[must_use]
    pub fn expose_upload_id(&self) -> &str {
        &self.upload_id
    }

    /// Returns the bucket owning this upload.
    #[must_use]
    pub fn bucket(&self) -> &str {
        &self.bucket
    }

    /// Returns the destination object key.
    #[must_use]
    pub fn key(&self) -> &str {
        &self.key
    }

    /// Returns the complete object size in bytes.
    #[must_use]
    pub const fn file_size(&self) -> u64 {
        self.file_size
    }

    /// Returns the uniform non-final part size in bytes.
    #[must_use]
    pub const fn part_size(&self) -> u64 {
        self.part_size
    }
}

impl fmt::Debug for MultipartSessionSnapshot {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("MultipartSessionSnapshot")
            .field("bucket", &self.bucket)
            .field("key", &self.key)
            .field("upload_id", &"[REDACTED]")
            .field("file_size", &self.file_size)
            .field("part_size", &self.part_size)
            .finish()
    }
}

#[derive(Clone, Debug)]
struct MultipartPlan {
    file_size: u64,
    part_size: u64,
    part_count: u16,
}

impl MultipartPlan {
    fn new(file_size: u64, part_size: u64) -> Result<Self, Error> {
        if file_size == 0 {
            return Err(ValidationError::MultipartFileSizeZero.into());
        }
        if file_size > MAX_MULTIPART_OBJECT_SIZE {
            return Err(ValidationError::MultipartObjectTooLarge {
                provided: file_size,
                max: MAX_MULTIPART_OBJECT_SIZE,
            }
            .into());
        }
        validation::validate_part_size(part_size)?;
        let part_count = file_size.div_ceil(part_size);
        if part_count > u64::from(MAX_PARTS) {
            return Err(ValidationError::TooManyParts {
                required: part_count,
                max: MAX_PARTS,
            }
            .into());
        }
        Ok(Self {
            file_size,
            part_size,
            part_count: part_count as u16,
        })
    }

    fn part_length(&self, number: PartNumber) -> Result<u64, Error> {
        if number.get() > self.part_count {
            return Err(Error::InvalidInput {
                field: "part_number",
                reason: "exceeds this upload's planned part count",
            });
        }
        let start = u64::from(number.get() - 1) * self.part_size;
        Ok((self.file_size - start).min(self.part_size))
    }
}

/// Server-observed state of a live multipart upload.
#[derive(Clone, Debug)]
pub struct MultipartReconciliation {
    uploaded_parts: Vec<UploadedPart>,
    missing_parts: Vec<PartNumber>,
}

impl MultipartReconciliation {
    /// Returns uploaded parts in ascending part-number order.
    pub fn uploaded_parts(&self) -> impl ExactSizeIterator<Item = &UploadedPart> {
        self.uploaded_parts.iter()
    }

    /// Consumes the reconciliation and returns uploaded parts in ascending
    /// part-number order.
    #[must_use]
    pub fn into_uploaded_parts(self) -> Vec<UploadedPart> {
        self.uploaded_parts
    }

    /// Returns planned parts that R2 has not received yet.
    pub fn missing_parts(&self) -> impl ExactSizeIterator<Item = PartNumber> + '_ {
        self.missing_parts.iter().copied()
    }

    /// Returns whether R2 has every planned part with the expected size.
    #[must_use]
    pub fn is_complete(&self) -> bool {
        self.missing_parts.is_empty()
    }

    /// Builds a canonical completion manifest when every part is present.
    pub fn into_completion_manifest(self) -> Result<CompletionManifest, Error> {
        if !self.missing_parts.is_empty() {
            return Err(Error::InvalidInput {
                field: "remote_parts",
                reason: "multipart upload is missing planned parts",
            });
        }
        CompletionManifest::try_from_parts(self.uploaded_parts)
    }
}

/// Builder for a new presigned multipart upload session.
#[derive(Clone, Debug)]
pub struct PresignedMultipartBuilder {
    bucket: Bucket,
    key: String,
    file_size: Option<u64>,
    part_size: Option<u64>,
    options: ObjectUploadOptions,
}

impl PresignedMultipartBuilder {
    /// Sets typed metadata stored on the completed object.
    #[must_use]
    pub fn upload_options(mut self, options: ObjectUploadOptions) -> Self {
        self.options = options;
        self
    }

    /// Sets the completed object's MIME media type.
    #[must_use]
    pub fn content_type(mut self, value: mime::Mime) -> Self {
        self.options = self.options.with_content_type(value);
        self
    }

    /// Sets the completed object's typed HTTP cache policy.
    #[must_use]
    pub fn cache_control(mut self, value: headers::CacheControl) -> Self {
        self.options = self.options.with_cache_control(value);
        self
    }

    /// Sets the completed object's content disposition.
    #[must_use]
    pub fn content_disposition(mut self, value: impl Into<String>) -> Self {
        self.options = self.options.with_content_disposition(value);
        self
    }

    /// Sets the completed object's content encoding.
    #[must_use]
    pub fn content_encoding(mut self, value: impl Into<String>) -> Self {
        self.options = self.options.with_content_encoding(value);
        self
    }

    /// Sets the completed object's content language.
    #[must_use]
    pub fn content_language(mut self, value: impl Into<String>) -> Self {
        self.options = self.options.with_content_language(value);
        self
    }

    /// Sets the completed object's HTTP expiration time.
    #[must_use]
    pub fn expires(mut self, value: SystemTime) -> Self {
        self.options = self.options.with_expires(value);
        self
    }

    /// Adds user-defined metadata to the completed object.
    #[must_use]
    pub fn custom_metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.options = self.options.with_custom_metadata(key, value);
        self
    }

    /// Sets the complete object size in bytes.
    ///
    /// This is required because r2kit plans every part before creating the
    /// remote upload: it determines the final part length, enforces R2's object
    /// and 10,000-part limits, signs exact content lengths, and later verifies
    /// the remote parts. For browser uploads, send the browser's `File.size`
    /// to the trusted server and treat it as untrusted input; r2kit validates
    /// the value before network I/O.
    ///
    /// Managed local-file uploads do not require this setting because
    /// [`crate::ManagedMultipartBuilder::upload_file`] reads file metadata.
    #[must_use]
    pub const fn file_size(mut self, bytes: u64) -> Self {
        self.file_size = Some(bytes);
        self
    }

    /// Sets the uniform multipart part size in bytes.
    #[must_use]
    pub const fn part_size(mut self, bytes: u64) -> Self {
        self.part_size = Some(bytes);
        self
    }

    /// Sets the uniform multipart part size in mebibytes (MiB).
    ///
    /// This is equivalent to [`Self::part_size`] with a binary-unit conversion
    /// and avoids repeating `1024 * 1024` at call sites. Values that cannot be
    /// represented as bytes are rejected by [`Self::create`].
    #[must_use]
    pub const fn part_size_mib(mut self, mebibytes: u64) -> Self {
        self.part_size = Some(validation::mebibytes(mebibytes));
        self
    }

    /// Creates the remote multipart upload after local validation succeeds.
    pub async fn create(self) -> Result<PresignedMultipart, Error> {
        self.options.validate()?;
        let plan = MultipartPlan::new(
            self.file_size.ok_or(Error::InvalidInput {
                field: "file_size",
                reason: "is required",
            })?,
            self.part_size.ok_or(Error::InvalidInput {
                field: "part_size",
                reason: "is required",
            })?,
        )?;
        let output = self
            .bucket
            .client
            .as_sdk()
            .create_multipart_upload()
            .bucket(&self.bucket.name)
            .key(&self.key)
            .set_content_type(self.options.content_type_value())
            .set_cache_control(self.options.cache_control_value())
            .set_content_disposition(self.options.content_disposition_value())
            .set_content_encoding(self.options.content_encoding_value())
            .set_content_language(self.options.content_language_value())
            .set_expires(self.options.expires_value())
            .set_metadata(self.options.custom_metadata_values())
            .send()
            .await
            .map_err(|error| Error::remote("CreateMultipartUpload", &error))?;
        let upload_id = output.upload_id().ok_or(Error::Service {
            operation: "CreateMultipartUpload",
        })?;

        Ok(PresignedMultipart {
            bucket: self.bucket,
            key: self.key,
            upload_id: upload_id.to_owned(),
            plan,
        })
    }
}

/// An active R2 multipart upload that can presign parts, complete, or abort.
#[derive(Clone)]
pub struct PresignedMultipart {
    bucket: Bucket,
    key: String,
    upload_id: String,
    plan: MultipartPlan,
}

impl PresignedMultipart {
    /// Returns the number of planned parts.
    #[must_use]
    pub const fn part_count(&self) -> u16 {
        self.plan.part_count
    }

    /// Returns the expected byte length for a part.
    pub fn part_length(&self, number: PartNumber) -> Result<u64, Error> {
        self.plan.part_length(number)
    }

    /// Creates a temporary signed PUT request for one part.
    pub async fn presign_part(
        &self,
        number: PartNumber,
        expires_in: Duration,
    ) -> Result<PresignedUploadPart, Error> {
        self.presign_part_inner(number, None, expires_in).await
    }

    /// Creates a signed PUT request that makes R2 verify `Content-MD5`.
    ///
    /// The uploader must replay the returned `content-md5` header exactly.
    pub async fn presign_part_with_md5(
        &self,
        number: PartNumber,
        content_md5: PartMd5,
        expires_in: Duration,
    ) -> Result<PresignedUploadPart, Error> {
        self.presign_part_inner(number, Some(content_md5), expires_in)
            .await
    }

    async fn presign_part_inner(
        &self,
        number: PartNumber,
        content_md5: Option<PartMd5>,
        expires_in: Duration,
    ) -> Result<PresignedUploadPart, Error> {
        let content_length = self.plan.part_length(number)?;
        validation::validate_expiry(expires_in)?;
        let config = PresigningConfig::expires_in(expires_in).map_err(|_| Error::Presign)?;
        let request = self
            .bucket
            .client
            .as_sdk()
            .upload_part()
            .bucket(&self.bucket.name)
            .key(&self.key)
            .upload_id(&self.upload_id)
            .part_number(i32::from(number.get()));
        let request = match &content_md5 {
            Some(value) => request.content_md5(value.as_base64()),
            None => request,
        };
        let signed = request
            .presigned(config)
            .await
            .map_err(|_| Error::Presign)?;

        Ok(PresignedUploadPart {
            part_number: number,
            content_length,
            content_md5,
            request: PresignedRequest::from_sdk(signed, expires_in)?,
        })
    }

    /// Reconciles persisted client state with the parts currently stored by R2.
    ///
    /// Every remote part is checked for a valid number, a unique entry, and
    /// the exact size required by the original upload plan.
    pub async fn reconcile(&self) -> Result<MultipartReconciliation, Error> {
        let mut marker = None;
        let mut uploaded: Vec<Option<UploadedPart>> = std::iter::repeat_with(|| None)
            .take(usize::from(self.plan.part_count))
            .collect();
        loop {
            let output = self
                .bucket
                .client
                .as_sdk()
                .list_parts()
                .bucket(&self.bucket.name)
                .key(&self.key)
                .upload_id(&self.upload_id)
                .max_parts(1_000)
                .set_part_number_marker(marker)
                .send()
                .await
                .map_err(|error| Error::remote("ListParts", &error))?;
            for remote in output.parts() {
                let raw_number = remote.part_number().ok_or(Error::Service {
                    operation: "ListParts",
                })?;
                let number = u16::try_from(raw_number)
                    .ok()
                    .and_then(|value| PartNumber::try_from(value).ok())
                    .ok_or(Error::InvalidInput {
                        field: "remote_parts",
                        reason: "contains an invalid part number",
                    })?;
                let expected_size =
                    self.plan
                        .part_length(number)
                        .map_err(|_| Error::InvalidInput {
                            field: "remote_parts",
                            reason: "contains a part outside the upload plan",
                        })?;
                if remote.size().and_then(|size| u64::try_from(size).ok()) != Some(expected_size) {
                    return Err(Error::InvalidInput {
                        field: "remote_parts",
                        reason: "contains a part with an unexpected size",
                    });
                }
                let etag = remote.e_tag().ok_or(Error::Service {
                    operation: "ListParts",
                })?;
                let part = UploadedPart::new(number, etag)?;
                let slot = &mut uploaded[usize::from(number.get() - 1)];
                if slot.replace(part).is_some() {
                    return Err(Error::InvalidInput {
                        field: "remote_parts",
                        reason: "contains a duplicate part number",
                    });
                }
            }
            if output.is_truncated() != Some(true) {
                break;
            }
            marker = output.next_part_number_marker;
            if marker.is_none() {
                return Err(Error::Service {
                    operation: "ListParts",
                });
            }
        }

        let mut uploaded_parts = Vec::with_capacity(uploaded.len());
        let mut missing_parts = Vec::new();
        for (index, part) in uploaded.into_iter().enumerate() {
            match part {
                Some(part) => uploaded_parts.push(part),
                None => missing_parts.push(
                    PartNumber::try_from(index as u16 + 1).expect("planned part numbers are valid"),
                ),
            }
        }
        Ok(MultipartReconciliation {
            uploaded_parts,
            missing_parts,
        })
    }

    /// Verifies client receipts against R2 before completing the upload.
    ///
    /// This extra `ListParts` round trip rejects stale, missing, incorrectly
    /// sized, or mismatched part receipts before `CompleteMultipartUpload`.
    pub async fn complete_verified(
        &self,
        manifest: CompletionManifest,
    ) -> Result<CompletedObject, Error> {
        let reconciliation = self.reconcile().await?;
        if !reconciliation.is_complete()
            || reconciliation.uploaded_parts.as_slice() != manifest.0.as_slice()
        {
            return Err(Error::InvalidInput {
                field: "parts",
                reason: "does not match the parts currently stored by R2",
            });
        }
        self.complete(manifest).await
    }

    /// Completes the upload with the exact ETags returned by every uploaded part.
    pub async fn complete(&self, manifest: CompletionManifest) -> Result<CompletedObject, Error> {
        if manifest.0.len() != usize::from(self.plan.part_count)
            || manifest
                .0
                .iter()
                .enumerate()
                .any(|(index, part)| usize::from(part.part_number().get()) != index + 1)
        {
            return Err(Error::InvalidInput {
                field: "parts",
                reason: "must include every planned part exactly once",
            });
        }

        let parts = manifest
            .0
            .into_iter()
            .map(|part| {
                CompletedPart::builder()
                    .part_number(i32::from(part.part_number.get()))
                    .e_tag(part.etag)
                    .build()
            })
            .collect();
        let upload = CompletedMultipartUpload::builder()
            .set_parts(Some(parts))
            .build();
        let output = self
            .bucket
            .client
            .as_sdk()
            .complete_multipart_upload()
            .bucket(&self.bucket.name)
            .key(&self.key)
            .upload_id(&self.upload_id)
            .multipart_upload(upload)
            .send()
            .await
            .map_err(|error| Error::remote("CompleteMultipartUpload", &error))?;

        Ok(CompletedObject {
            etag: output.e_tag().map(ToOwned::to_owned),
        })
    }

    /// Aborts the remote multipart upload.
    pub async fn abort(&self) -> Result<(), Error> {
        self.bucket
            .client
            .as_sdk()
            .abort_multipart_upload()
            .bucket(&self.bucket.name)
            .key(&self.key)
            .upload_id(&self.upload_id)
            .send()
            .await
            .map_err(|error| Error::remote("AbortMultipartUpload", &error))?;
        Ok(())
    }

    /// Captures resumable session state. Its debug output redacts the upload ID.
    #[must_use]
    pub fn snapshot(&self) -> MultipartSessionSnapshot {
        MultipartSessionSnapshot {
            bucket: self.bucket.name.clone(),
            key: self.key.clone(),
            upload_id: self.upload_id.clone(),
            file_size: self.plan.file_size,
            part_size: self.plan.part_size,
        }
    }
}

impl fmt::Debug for PresignedMultipart {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("PresignedMultipart")
            .field("bucket", &self.bucket.name)
            .field("key", &self.key)
            .field("upload_id", &"[REDACTED]")
            .field("plan", &self.plan)
            .finish()
    }
}

/// Result metadata for a completed multipart object.
#[derive(Clone, Debug)]
pub struct CompletedObject {
    etag: Option<String>,
}

impl CompletedObject {
    /// Returns R2's multipart ETag. It is not the MD5 of the complete object.
    #[must_use]
    pub fn etag(&self) -> Option<&str> {
        self.etag.as_deref()
    }
}

impl Bucket {
    /// Starts configuring a presigned multipart upload for an object key.
    pub fn presigned_multipart(
        &self,
        key: impl Into<String>,
    ) -> Result<PresignedMultipartBuilder, Error> {
        let key = key.into();
        validation::validate_key(&key)?;
        Ok(PresignedMultipartBuilder {
            bucket: self.clone(),
            key,
            file_size: None,
            part_size: None,
            options: ObjectUploadOptions::default(),
        })
    }

    /// Restores a previously captured multipart upload session.
    pub fn resume_presigned_multipart(
        &self,
        snapshot: MultipartSessionSnapshot,
    ) -> Result<PresignedMultipart, Error> {
        if snapshot.bucket != self.name {
            return Err(Error::InvalidInput {
                field: "bucket",
                reason: "snapshot belongs to another bucket",
            });
        }
        let plan = MultipartPlan::new(snapshot.file_size, snapshot.part_size)?;
        Ok(PresignedMultipart {
            bucket: self.clone(),
            key: snapshot.key,
            upload_id: snapshot.upload_id,
            plan,
        })
    }
}

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

    #[test]
    fn plans_exact_and_short_final_parts() {
        let plan = MultipartPlan::new(11 * 1024 * 1024, 5 * 1024 * 1024).unwrap();
        assert_eq!(plan.part_count, 3);
        assert_eq!(
            plan.part_length(PartNumber::try_from(1).unwrap()).unwrap(),
            5 * 1024 * 1024
        );
        assert_eq!(
            plan.part_length(PartNumber::try_from(3).unwrap()).unwrap(),
            1024 * 1024
        );
    }

    #[test]
    fn rejects_too_many_parts() {
        let min_part_size = 5 * 1024 * 1024;
        let result = MultipartPlan::new(10_001 * min_part_size, min_part_size);
        assert!(matches!(
            result,
            Err(Error::Validation(ValidationError::TooManyParts {
                required: 10_001,
                max: 10_000
            }))
        ));
    }

    #[test]
    fn rejects_an_object_over_r2s_effective_limit() {
        let max_part_size = validation::MAX_MULTIPART_PART_SIZE;
        let result = MultipartPlan::new(MAX_MULTIPART_OBJECT_SIZE + 1, max_part_size);
        assert!(matches!(
            result,
            Err(Error::Validation(
                ValidationError::MultipartObjectTooLarge {
                    provided,
                    max: MAX_MULTIPART_OBJECT_SIZE
                }
            )) if provided == MAX_MULTIPART_OBJECT_SIZE + 1
        ));
    }

    #[test]
    fn rejects_subsecond_presign_expiry() {
        assert!(matches!(
            validation::validate_expiry(Duration::from_millis(999)),
            Err(Error::Validation(
                ValidationError::PresignExpiryOutOfRange { provided, .. }
            )) if provided == Duration::from_millis(999)
        ));
    }

    #[test]
    fn canonicalizes_manifest_and_rejects_duplicates() {
        let one = PartNumber::try_from(1).unwrap();
        let two = PartNumber::try_from(2).unwrap();
        let manifest = CompletionManifest::try_from_parts([
            UploadedPart::new(two, "two").unwrap(),
            UploadedPart::new(one, "one").unwrap(),
        ])
        .unwrap();
        assert_eq!(manifest.0[0].part_number(), one);

        let duplicate = CompletionManifest::try_from_parts([
            UploadedPart::new(one, "one").unwrap(),
            UploadedPart::new(one, "again").unwrap(),
        ]);
        assert!(duplicate.is_err());
    }

    #[test]
    fn redacts_secret_wrappers() {
        let min_part_size = 5 * 1024 * 1024;
        let url = SecretUrl("https://example.invalid/?X-Amz-Signature=secret".into());
        assert!(!format!("{url:?}").contains("X-Amz-Signature"));
        let snapshot = MultipartSessionSnapshot::restore(
            "r2kit",
            "key",
            "secret-upload-id",
            min_part_size,
            min_part_size,
        )
        .unwrap();
        assert!(!format!("{snapshot:?}").contains("secret-upload-id"));
    }
}