loonfs_api/v0/

uploads.rs

//! Upload-session shapes for the v0 HTTP API: transport modes, session
//! begin/append/complete requests and responses, and the direct-put
//! presigned-access envelope. Content moves through these shapes; the
//! metadata that later references it commits through [`super::commits`].

use crate::{ContentRef, NamespaceId, UploadId};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

/// What a `direct_put` client promises about bytes it has not written yet.
///
/// The server mints the content object's identity — a client cannot name a
/// key it has not been given — so a direct upload declares only what it can
/// know about its own bytes. The server signs both into the provider write
/// and verifies them again at completion.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct DirectPutContentClaim {
    /// Complete byte length the client will write.
    pub size_bytes: u64,
    /// SHA-256 over the complete payload, lowercase hex.
    pub sha256: String,
}

/// What a `direct_multipart` client says about the object it finished
/// writing, supplied at completion rather than at begin.
///
/// The claim arrives last because that is the only place a one-pass
/// uploader can produce it: a client that had to declare the length and
/// digest up front would have to read its payload twice, and a client
/// reading from a pipe could not start at all. Nothing is lost by waiting —
/// the claim was never trusted, only verified, and verification happens at
/// completion either way.
///
/// The digest is CRC-64/NVME rather than SHA-256 because that is the
/// checksum an S3-compatible provider computes over a multipart object: it
/// is the only full-object evidence the provider will ever be able to show
/// back, so it is the only thing worth claiming at all.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct DirectMultipartContentClaim {
    /// Complete byte length the client wrote across every part.
    pub size_bytes: u64,
    /// CRC-64/NVME over the complete assembled payload, lowercase hex.
    pub crc64nvme: String,
}

/// What a `direct_multipart` client asks for when it opens a session.
///
/// A begin request declares no length and no digest: the session exists to
/// receive bytes whose length may not be known yet. All it settles is the
/// geometry the client cuts to.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct DirectMultipartUploadOptions {
    /// Byte length of every part except the last, or `None` for the
    /// server's default.
    ///
    /// The value bounds the object: a provider accepts at most 10,000
    /// parts, so this session can carry at most `part_size_bytes × 10_000`
    /// bytes. A client that knows its payload is very large asks for a
    /// larger part size; one that does not know its length at all takes the
    /// default and keeps asking for part URLs until its stream ends.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub part_size_bytes: Option<u64>,
}

/// Upload transport mode.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "snake_case")]
pub enum UploadMode {
    /// The service receives bytes and writes content to object storage.
    #[default]
    ServiceProxied,
    /// The service mints a short-lived presigned PUT URL for the content object.
    DirectPut,
    /// The service opens a provider multipart upload for the content object
    /// and signs one PUT per part, so a large object crosses the network
    /// once, in parallel, without passing through the server.
    DirectMultipart,
}

/// Request for starting an upload session, tagged by the transport it asks
/// for.
///
/// Each transport carries only what it needs, so the combinations a flat
/// request could spell — a proxied begin carrying multipart geometry, a
/// direct put with no claim to sign — are refused when the body is decoded
/// rather than by a handler reading them back. `mode` is required: a
/// request that does not say how it intends to move its bytes is not a
/// request this API can answer.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(tag = "mode", rename_all = "snake_case", deny_unknown_fields)]
pub enum BeginUploadRequest {
    // The empty braces are load-bearing: serde lets a *unit* variant of a
    // tagged enum swallow whatever else the body carried, so spelling this
    // as `ServiceProxied` would quietly accept a proxied begin holding
    // another transport's fields. A variant with no fields refuses them.
    /// Send the bytes to the service, which writes the content object.
    #[cfg_attr(feature = "openapi", schema(title = "BeginUploadServiceProxied"))]
    ServiceProxied {},
    /// Write the whole object through one presigned request. The server
    /// signs exactly these bytes into the write it authorizes, so the claim
    /// is required.
    #[cfg_attr(feature = "openapi", schema(title = "BeginUploadDirectPut"))]
    DirectPut {
        /// Byte length and digest of the payload about to be written.
        content: DirectPutContentClaim,
    },
    /// Write the object in parts through presigned part uploads.
    #[cfg_attr(feature = "openapi", schema(title = "BeginUploadDirectMultipart"))]
    DirectMultipart {
        /// Selects the part geometry; absent takes the server's default.
        /// A multipart upload claims its content at completion, so nothing
        /// about the payload is declared here.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        multipart: Option<DirectMultipartUploadOptions>,
    },
}

impl BeginUploadRequest {
    /// The transport this request asks for.
    pub fn mode(&self) -> UploadMode {
        match self {
            Self::ServiceProxied {} => UploadMode::ServiceProxied,
            Self::DirectPut { .. } => UploadMode::DirectPut,
            Self::DirectMultipart { .. } => UploadMode::DirectMultipart,
        }
    }
}

/// Client-facing direct transfer capability.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ObjectTransferAccess {
    /// Short-lived URL plus required headers for one object-store write.
    #[cfg_attr(
        feature = "openapi",
        schema(title = "ObjectTransferAccessPresignedUrl")
    )]
    PresignedUrl {
        /// HTTP method the client must use.
        method: String,
        /// Full presigned URL.
        url: String,
        /// Headers that are covered by the signature and must be sent.
        #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
        headers: BTreeMap<String, String>,
        /// Expiration timestamp in Unix milliseconds.
        expires_at_ms: u64,
    },
}

/// Presigned direct_put upload details. The raw object key is intentionally not public.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct DirectPutUpload {
    /// Immutable object identity the server minted, plus the byte length and
    /// checksum covered by the signed request. Completion and the later
    /// commit both name exactly this reference.
    pub content_ref: ContentRef,
    /// Short-lived write capability the client uses without learning the raw object key.
    pub access: ObjectTransferAccess,
}

/// Direct multipart upload details: the geometry a client cuts its payload
/// into parts with, and nothing else.
///
/// There is no content reference here, and no part count. The session has
/// not been told what it is about to receive, so there is no identity to
/// echo and no arithmetic to do — the server mints the content object
/// behind the session and names it in the completion response. The
/// provider's upload id is absent for the same reason it always was: a
/// client asks this server for part URLs by part number and never talks to
/// the provider's multipart API in its own words.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct DirectMultipartUpload {
    /// Byte length of every part except the last. At most 10,000 parts may
    /// be uploaded, so this bounds the object at `part_size_bytes × 10_000`.
    pub part_size_bytes: u64,
}

/// One part's checksum, supplied by the client so the server can sign it
/// into that part's upload URL.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct UploadPartChecksumClaim {
    /// One-based part number, at most the provider's 10,000-part limit.
    pub part_number: u32,
    /// CRC-64/NVME over this part's bytes, lowercase hex.
    pub crc64nvme: String,
}

/// Request for part-upload capabilities on an open multipart session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct SignUploadPartsRequest {
    /// Parts to authorize, each with the checksum the provider will enforce
    /// on it. Asking again for a part already uploaded is how a client
    /// retries one: a repeated part is last-write-wins at the provider.
    pub parts: Vec<UploadPartChecksumClaim>,
}

/// One authorized part upload.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct SignedUploadPart {
    /// Part number this capability writes.
    pub part_number: u32,
    /// Short-lived write capability for that part.
    pub access: ObjectTransferAccess,
}

/// Response carrying one capability per requested part.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct SignUploadPartsResponse {
    /// Namespace that owns the upload session.
    pub namespace_id: NamespaceId,
    /// Session the parts belong to.
    pub upload_id: UploadId,
    /// Capabilities in the order the request asked for them.
    pub parts: Vec<SignedUploadPart>,
}

/// One uploaded part, as the client observed the provider accept it.
///
/// The server keeps no durable record of any part. Part bookkeeping is the
/// client's, exactly as it is in the provider's own multipart API, and this
/// is where the client hands it back.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(deny_unknown_fields)]
pub struct CompletedUploadPart {
    /// One-based part number.
    pub part_number: u32,
    /// Entity tag the provider returned for the accepted part.
    pub etag: String,
    /// CRC-64/NVME the part was signed and accepted with, lowercase hex.
    pub crc64nvme: String,
}

/// Stateless proof that a LoonFS server already validated a content ref.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct ValidatedContentToken {
    /// Content identity the server attests it already verified.
    pub content_ref: ContentRef,
    /// Opaque, server-signed token. Clients must not parse it.
    pub token: String,
}

/// Response for starting an upload session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct BeginUploadResponse {
    /// Namespace authorized to consume the eventual staged content.
    pub namespace_id: NamespaceId,
    /// Durable session identity used by subsequent append and completion calls.
    pub upload_id: UploadId,
    /// Transport selected after applying server capability and request validation.
    pub mode: UploadMode,
    /// Presigned write details for `DirectPut`, or `None` for `ServiceProxied`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub direct_put: Option<DirectPutUpload>,
    /// Part geometry for `DirectMultipart`, or `None` for every other mode.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub direct_multipart: Option<DirectMultipartUpload>,
}

/// Response after uploading bytes into a session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct UploadContentResponse {
    /// Namespace that owns the upload session.
    pub namespace_id: NamespaceId,
    /// Session into which the service staged these bytes.
    pub upload_id: UploadId,
    /// Digest and byte length computed from the accepted body.
    pub content_ref: ContentRef,
}

/// Request to complete an upload, tagged by which of its two shapes it is.
///
/// The shapes correspond to who knew the content identity first. A
/// service-proxied or `direct_put` session was handed its reference before
/// any byte moved, so its completion names that reference back. A
/// `direct_multipart` session was never told one — there was nothing to
/// tell — so its completion carries the claim and its parts instead, and
/// the server builds the reference from the identity it has held all along.
///
/// The two share no fields, so a completion carrying one shape's fields
/// under the other's tag does not decode. What decoding cannot settle is
/// whether the shape matches the *session*, because only the server knows
/// which transport the session was opened with; that one check is made
/// against the durable record.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(tag = "completion", rename_all = "snake_case", deny_unknown_fields)]
pub enum CompleteUploadRequest {
    /// Completes a session the server named a content object for: the
    /// caller names it back and the server proves the object matches.
    #[cfg_attr(feature = "openapi", schema(title = "CompleteUploadContentRef"))]
    ContentRef {
        /// Content identity the caller expects the session to have settled
        /// on.
        content_ref: ContentRef,
    },
    /// Completes a `direct_multipart` session with what it uploaded.
    #[cfg_attr(feature = "openapi", schema(title = "CompleteUploadMultipart"))]
    Multipart {
        /// The assembled object's length and CRC-64/NVME, which completion
        /// verifies against the provider's own reading of the object.
        multipart: DirectMultipartContentClaim,
        /// Every part the client uploaded, in ascending part order. The
        /// server holds no part records of its own, so this list is what it
        /// assembles the object from.
        parts: Vec<CompletedUploadPart>,
    },
}

impl CompleteUploadRequest {
    /// Completes a session that already knows its content reference.
    pub fn for_content_ref(content_ref: ContentRef) -> Self {
        Self::ContentRef { content_ref }
    }

    /// Completes a `direct_multipart` session with what it assembled.
    pub fn for_multipart(
        claim: DirectMultipartContentClaim,
        parts: Vec<CompletedUploadPart>,
    ) -> Self {
        Self::Multipart {
            multipart: claim,
            parts,
        }
    }
}

/// Response after an upload session is completed.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct CompleteUploadResponse {
    /// Namespace that owns the completed session.
    pub namespace_id: NamespaceId,
    /// Session whose result is now frozen for idempotent completion retries.
    pub upload_id: UploadId,
    /// Verified immutable content selected by the completed session.
    pub content_ref: ContentRef,
    /// Opaque server proof for a later commit, or `None` when the backend needs no token.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub validated_content_token: Option<String>,
}

/// Observed state of an upload session.
///
/// A session is `open`, then `completed` or `aborted`, and both of those are
/// final. Reading a completed session mints a fresh receipt for content that
/// is already durable, which is why losing a commit response never costs a
/// retransfer.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum UploadSessionStatus {
    /// Accepting content until its lease passes.
    #[cfg_attr(feature = "openapi", schema(title = "UploadSessionStatusOpen"))]
    Open {
        /// Unix-millisecond instant after which the session is abandoned and
        /// may be aborted by server-side cleanup.
        expires_at_ms: u64,
    },
    /// Final: the content is durable and verified.
    #[cfg_attr(feature = "openapi", schema(title = "UploadSessionStatusCompleted"))]
    Completed {
        /// Unix-millisecond stamp of the completion.
        completed_at_ms: u64,
        /// Verified immutable content this session settled on.
        content_ref: ContentRef,
        /// Freshly minted proof for a following commit, or `None` once the
        /// session has stopped minting them.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        validated_content_token: Option<String>,
    },
    /// Final: the session selected no content and its object is gone.
    #[cfg_attr(feature = "openapi", schema(title = "UploadSessionStatusAborted"))]
    Aborted {
        /// Unix-millisecond stamp of the abort.
        aborted_at_ms: u64,
    },
}

/// Response for reading one upload session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct UploadStatusResponse {
    /// Namespace that owns the session.
    pub namespace_id: NamespaceId,
    /// Session that was read.
    pub upload_id: UploadId,
    /// The session's state, with a fresh receipt when it is completed.
    pub status: UploadSessionStatus,
}

/// Response after aborting an upload session.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
pub struct AbortUploadResponse {
    /// Namespace that owns the session.
    pub namespace_id: NamespaceId,
    /// Session that is now final.
    pub upload_id: UploadId,
    /// Unix-millisecond stamp of the abort that stands, which for a repeated
    /// abort is the first one's.
    pub aborted_at_ms: u64,
}

#[cfg(test)]
mod tests {
    use super::{
        BeginUploadRequest, BeginUploadResponse, CompleteUploadRequest, DirectPutContentClaim,
        DirectPutUpload, ObjectTransferAccess, UploadMode, UploadSessionStatus,
    };
    use crate::{ContentId, ContentRef, NamespaceId, UploadId};
    use std::collections::BTreeMap;

    #[test]
    fn direct_put_upload_mode_serializes_as_expected() {
        assert_eq!(
            serde_json::to_string(&UploadMode::DirectPut).expect("serialize mode"),
            r#""direct_put""#
        );
    }

    /// A begin request says how it means to move its bytes, or it is not a
    /// request. Nothing is inferred from what the body left out.
    #[test]
    fn a_begin_request_without_a_mode_does_not_decode() {
        assert!(serde_json::from_str::<BeginUploadRequest>("{}").is_err());
        assert_eq!(
            serde_json::from_str::<BeginUploadRequest>(r#"{"mode":"service_proxied"}"#)
                .expect("decode proxied begin request"),
            BeginUploadRequest::ServiceProxied {}
        );
    }

    /// The combinations a flat begin request could spell are refused where
    /// the body is read, not by a handler comparing fields afterwards.
    #[test]
    fn a_begin_request_carrying_another_modes_fields_does_not_decode() {
        for body in [
            r#"{"mode":"service_proxied","multipart":{"part_size_bytes":8388608}}"#,
            r#"{"mode":"service_proxied","content":{"size_bytes":5,"sha256":"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"}}"#,
            r#"{"mode":"direct_multipart","content":{"size_bytes":5,"sha256":"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"}}"#,
            // A direct put with nothing to sign is not a direct put.
            r#"{"mode":"direct_put"}"#,
        ] {
            assert!(
                serde_json::from_str::<BeginUploadRequest>(body).is_err(),
                "decoded a begin request that mixes modes: {body}"
            );
        }
    }

    /// A completion carries one shape's fields under that shape's tag.
    #[test]
    fn a_completion_mixing_its_two_shapes_does_not_decode() {
        for body in [
            r#"{"completion":"multipart","multipart":{"size_bytes":5,"crc64nvme":"0123456789abcdef"},"parts":[],"content_ref":{"kind":"blob_v1","content_id":"con_0123456789abcdef0123456789abcdef","size_bytes":5,"storage_checksum":{"algorithm":"sha256","value":"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"}}}"#,
            // Neither multipart field stands without the other.
            r#"{"completion":"multipart","multipart":{"size_bytes":5,"crc64nvme":"0123456789abcdef"}}"#,
            r#"{"completion":"multipart","parts":[]}"#,
            r#"{"completion":"content_ref"}"#,
        ] {
            assert!(
                serde_json::from_str::<CompleteUploadRequest>(body).is_err(),
                "decoded a completion that mixes shapes: {body}"
            );
        }
    }

    #[test]
    fn direct_put_response_exposes_only_presigned_access() {
        let response = BeginUploadResponse {
            namespace_id: NamespaceId::parse("demo").expect("namespace id"),
            upload_id: UploadId::parse("upl_00000000000000000000000000000001")
                .expect("valid upload id"),
            mode: UploadMode::DirectPut,
            direct_put: Some(DirectPutUpload {
                content_ref: ContentRef::blob_v1(ContentId::generate(), b"hello"),
                access: ObjectTransferAccess::PresignedUrl {
                    method: "PUT".to_owned(),
                    url: "https://bucket.example/object?X-Amz-Signature=abc".to_owned(),
                    headers: BTreeMap::from([
                        ("if-none-match".to_owned(), "*".to_owned()),
                        (
                            "x-provider-checksum".to_owned(),
                            "LPJNul+wow4m6DsqxbninhsWHlwfp0JecwQzYpOLmCQ=".to_owned(),
                        ),
                    ]),
                    expires_at_ms: 1,
                },
            }),
            direct_multipart: None,
        };

        let json = serde_json::to_string(&response).expect("serialize response");
        assert!(json.contains(r#""kind":"presigned_url""#));
        assert!(!json.contains("object_key"));
    }

    /// A direct-put client declares what it is about to write; it cannot
    /// declare *where*, because the server owns content identity.
    #[test]
    fn a_direct_put_claim_names_only_size_and_digest() {
        let request: BeginUploadRequest = serde_json::from_str(
            r#"{"mode":"direct_put","content":{"size_bytes":5,"sha256":"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"}}"#,
        )
        .expect("decode direct-put begin request");
        assert_eq!(
            request,
            BeginUploadRequest::DirectPut {
                content: DirectPutContentClaim {
                    size_bytes: 5,
                    sha256: "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
                        .to_owned(),
                },
            }
        );

        assert!(
            serde_json::from_str::<DirectPutContentClaim>(
                r#"{"size_bytes":5,"sha256":"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824","content_id":"con_0123456789abcdef0123456789abcdef"}"#
            )
            .is_err(),
            "a client must not be able to name the content object"
        );
    }

    #[test]
    fn upload_status_names_its_state_on_the_wire() {
        let open = serde_json::to_value(UploadSessionStatus::Open {
            expires_at_ms: 1_000,
        })
        .expect("serialize open status");
        assert_eq!(open["state"], "open");

        let aborted = serde_json::to_value(UploadSessionStatus::Aborted {
            aborted_at_ms: 2_000,
        })
        .expect("serialize aborted status");
        assert_eq!(aborted["state"], "aborted");

        let completed = serde_json::to_value(UploadSessionStatus::Completed {
            completed_at_ms: 3_000,
            content_ref: ContentRef::blob_v1(ContentId::generate(), b"hello"),
            validated_content_token: None,
        })
        .expect("serialize completed status");
        assert_eq!(completed["state"], "completed");
        assert!(
            completed.get("validated_content_token").is_none(),
            "a session past its receipt window reports no token at all"
        );
    }
}