post3_server/s3/handlers/

objects.rs

use std::collections::HashMap;

use axum::{
    body::Body,
    extract::{Path, Query, State},
    http::{header::HeaderName, HeaderMap, HeaderValue, StatusCode},
    response::{IntoResponse, Response},
};
use bytes::Bytes;
use post3::{Post3Error, StorageBackend};

use crate::s3::extractors::{BucketGetQuery, ObjectKeyQuery};
use crate::s3::handlers::multipart;
use crate::s3::responses;
use crate::state::State as AppState;

// --- Dispatch functions ---

/// PUT /{bucket}/{*key} — dispatches to upload_part or put_object based on query params.
pub async fn put_dispatch<B: StorageBackend>(
    state: State<AppState<B>>,
    path: Path<(String, String)>,
    query: Query<ObjectKeyQuery>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    if query.upload_id.is_some() && query.part_number.is_some() {
        multipart::upload_part(state, path, query, body).await
    } else {
        put_object(state, path, headers, body).await
    }
}

/// GET /{bucket}/{*key} — dispatches to list_parts or get_object based on query params.
pub async fn get_dispatch<B: StorageBackend>(
    state: State<AppState<B>>,
    path: Path<(String, String)>,
    query: Query<ObjectKeyQuery>,
) -> Response {
    if query.upload_id.is_some() {
        multipart::list_parts(state, path, query).await
    } else {
        get_object(state, path).await
    }
}

/// DELETE /{bucket}/{*key} — dispatches to abort_multipart_upload or delete_object.
pub async fn delete_dispatch<B: StorageBackend>(
    state: State<AppState<B>>,
    path: Path<(String, String)>,
    query: Query<ObjectKeyQuery>,
) -> Response {
    if query.upload_id.is_some() {
        multipart::abort_multipart_upload(state, path, query).await
    } else {
        delete_object(state, path).await
    }
}

/// POST /{bucket}/{*key} — dispatches to create_multipart_upload or complete_multipart_upload.
pub async fn post_dispatch<B: StorageBackend>(
    state: State<AppState<B>>,
    path: Path<(String, String)>,
    query: Query<ObjectKeyQuery>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    if query.uploads.is_some() {
        multipart::create_multipart_upload(state, path, headers).await
    } else if query.upload_id.is_some() {
        multipart::complete_multipart_upload(state, path, query, body).await
    } else {
        (
            StatusCode::BAD_REQUEST,
            [("Content-Type", "application/xml")],
            responses::error_xml(
                "InvalidRequest",
                "POST requires ?uploads or ?uploadId parameter",
                &format!("/{}/{}", path.0 .0, path.0 .1),
            ),
        )
            .into_response()
    }
}

// --- Object handlers ---

pub async fn put_object<B: StorageBackend>(
    State(state): State<AppState<B>>,
    Path((bucket, key)): Path<(String, String)>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    let content_type = headers
        .get("content-type")
        .and_then(|v| v.to_str().ok())
        .map(|s| s.to_string());

    // Extract x-amz-meta-* user metadata
    let mut metadata = HashMap::new();
    for (name, value) in headers.iter() {
        let name_str = name.as_str();
        if let Some(meta_key) = name_str.strip_prefix("x-amz-meta-") {
            if let Ok(v) = value.to_str() {
                metadata.insert(meta_key.to_string(), v.to_string());
            }
        }
    }

    match state
        .store
        .put_object(&bucket, &key, content_type.as_deref(), metadata, body)
        .await
    {
        Ok(result) => {
            let mut response_headers = HeaderMap::new();
            response_headers.insert("ETag", result.etag.parse().unwrap());
            response_headers.insert(
                "x-amz-request-id",
                HeaderValue::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
            );
            (StatusCode::OK, response_headers).into_response()
        }
        Err(Post3Error::BucketNotFound(b)) => (
            StatusCode::NOT_FOUND,
            [("Content-Type", "application/xml")],
            responses::error_xml(
                "NoSuchBucket",
                "The specified bucket does not exist",
                &b,
            ),
        )
            .into_response(),
        Err(e) => {
            tracing::error!("put_object error: {e}");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                [("Content-Type", "application/xml")],
                responses::error_xml(
                    "InternalError",
                    &e.to_string(),
                    &format!("/{bucket}/{key}"),
                ),
            )
                .into_response()
        }
    }
}

pub async fn get_object<B: StorageBackend>(
    State(state): State<AppState<B>>,
    Path((bucket, key)): Path<(String, String)>,
) -> Response {
    match state.store.get_object(&bucket, &key).await {
        Ok(result) => {
            let mut headers = HeaderMap::new();
            headers.insert(
                "Content-Type",
                HeaderValue::from_str(&result.metadata.content_type).unwrap(),
            );
            headers.insert(
                "Content-Length",
                HeaderValue::from_str(&result.metadata.size.to_string()).unwrap(),
            );
            headers.insert("ETag", HeaderValue::from_str(&result.metadata.etag).unwrap());
            headers.insert(
                "Last-Modified",
                HeaderValue::from_str(
                    &result
                        .metadata
                        .last_modified
                        .format("%a, %d %b %Y %H:%M:%S GMT")
                        .to_string(),
                )
                .unwrap(),
            );
            headers.insert(
                "x-amz-request-id",
                HeaderValue::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
            );

            // Return user metadata as x-amz-meta-* headers
            for (k, v) in &result.user_metadata {
                let header_name = format!("x-amz-meta-{k}");
                if let (Ok(name), Ok(val)) = (
                    header_name.parse::<HeaderName>(),
                    HeaderValue::from_str(v),
                ) {
                    headers.insert(name, val);
                }
            }

            (StatusCode::OK, headers, Body::from(result.body)).into_response()
        }
        Err(Post3Error::BucketNotFound(b)) => (
            StatusCode::NOT_FOUND,
            [("Content-Type", "application/xml")],
            responses::error_xml(
                "NoSuchBucket",
                "The specified bucket does not exist",
                &b,
            ),
        )
            .into_response(),
        Err(Post3Error::ObjectNotFound { bucket: b, key: k }) => (
            StatusCode::NOT_FOUND,
            [("Content-Type", "application/xml")],
            responses::error_xml(
                "NoSuchKey",
                "The specified key does not exist.",
                &format!("/{b}/{k}"),
            ),
        )
            .into_response(),
        Err(e) => {
            tracing::error!("get_object error: {e}");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                [("Content-Type", "application/xml")],
                responses::error_xml(
                    "InternalError",
                    &e.to_string(),
                    &format!("/{bucket}/{key}"),
                ),
            )
                .into_response()
        }
    }
}

pub async fn head_object<B: StorageBackend>(
    State(state): State<AppState<B>>,
    Path((bucket, key)): Path<(String, String)>,
) -> Response {
    match state.store.head_object(&bucket, &key).await {
        Ok(Some(result)) => {
            let mut headers = HeaderMap::new();
            headers.insert(
                "Content-Type",
                HeaderValue::from_str(&result.object.content_type).unwrap(),
            );
            headers.insert(
                "Content-Length",
                HeaderValue::from_str(&result.object.size.to_string()).unwrap(),
            );
            headers.insert("ETag", HeaderValue::from_str(&result.object.etag).unwrap());
            headers.insert(
                "Last-Modified",
                HeaderValue::from_str(
                    &result
                        .object
                        .last_modified
                        .format("%a, %d %b %Y %H:%M:%S GMT")
                        .to_string(),
                )
                .unwrap(),
            );
            headers.insert(
                "x-amz-request-id",
                HeaderValue::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
            );

            for (k, v) in &result.user_metadata {
                let header_name = format!("x-amz-meta-{k}");
                if let (Ok(name), Ok(val)) = (
                    header_name.parse::<HeaderName>(),
                    HeaderValue::from_str(v),
                ) {
                    headers.insert(name, val);
                }
            }

            (StatusCode::OK, headers).into_response()
        }
        Ok(None) => {
            let mut headers = HeaderMap::new();
            headers.insert(
                "x-amz-request-id",
                HeaderValue::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
            );
            (StatusCode::NOT_FOUND, headers).into_response()
        }
        Err(e) => {
            tracing::error!("head_object error: {e}");
            StatusCode::INTERNAL_SERVER_ERROR.into_response()
        }
    }
}

pub async fn delete_object<B: StorageBackend>(
    State(state): State<AppState<B>>,
    Path((bucket, key)): Path<(String, String)>,
) -> Response {
    match state.store.delete_object(&bucket, &key).await {
        Ok(()) => {
            let mut headers = HeaderMap::new();
            headers.insert(
                "x-amz-request-id",
                HeaderValue::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
            );
            (StatusCode::NO_CONTENT, headers).into_response()
        }
        Err(Post3Error::BucketNotFound(b)) => (
            StatusCode::NOT_FOUND,
            [("Content-Type", "application/xml")],
            responses::error_xml(
                "NoSuchBucket",
                "The specified bucket does not exist",
                &b,
            ),
        )
            .into_response(),
        Err(e) => {
            tracing::error!("delete_object error: {e}");
            StatusCode::INTERNAL_SERVER_ERROR.into_response()
        }
    }
}

/// Handles GET /{bucket} — dispatches to ListMultipartUploads, ListObjectVersions,
/// GetBucketLocation, or ListObjects (v1/v2).
pub async fn list_or_get<B: StorageBackend>(
    State(state): State<AppState<B>>,
    Path(bucket): Path<String>,
    Query(query): Query<BucketGetQuery>,
) -> Response {
    // ?uploads → ListMultipartUploads
    if query.uploads.is_some() {
        return multipart::list_multipart_uploads(
            State(state),
            Path(bucket),
            Query(query),
        )
        .await;
    }

    // ?location → GetBucketLocation
    if query.location.is_some() {
        return get_bucket_location(State(state), Path(bucket)).await;
    }

    // ?versions → ListObjectVersions
    if query.versions.is_some() {
        return list_object_versions(State(state), Path(bucket), Query(query)).await;
    }

    // Default: ListObjects (v1 or v2)
    let is_v2 = query.list_type == Some(2);
    let continuation_token = if is_v2 {
        // v2: use continuation-token if present, else start-after
        query
            .continuation_token
            .as_deref()
            .or(query.start_after.as_deref())
    } else {
        query.marker.as_deref()
    };

    // Treat empty delimiter as absent (S3 spec: empty delimiter = no delimiter)
    let delimiter = query
        .delimiter
        .as_deref()
        .filter(|d| !d.is_empty());

    match state
        .store
        .list_objects_v2(
            &bucket,
            query.prefix.as_deref(),
            continuation_token,
            query.max_keys,
            delimiter,
        )
        .await
    {
        Ok(result) => {
            let max_keys = query.max_keys.unwrap_or(1000);
            let mut headers = HeaderMap::new();
            headers.insert("Content-Type", HeaderValue::from_static("application/xml"));
            headers.insert(
                "x-amz-request-id",
                HeaderValue::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
            );

            let xml = if is_v2 {
                responses::list_objects_v2_xml(
                    &bucket,
                    &result,
                    max_keys,
                    query.continuation_token.as_deref(),
                    query.start_after.as_deref(),
                )
            } else {
                responses::list_objects_v1_xml(
                    &bucket,
                    &result,
                    max_keys,
                    query.marker.as_deref(),
                )
            };

            (StatusCode::OK, headers, xml).into_response()
        }
        Err(Post3Error::BucketNotFound(b)) => (
            StatusCode::NOT_FOUND,
            [("Content-Type", "application/xml")],
            responses::error_xml(
                "NoSuchBucket",
                "The specified bucket does not exist",
                &b,
            ),
        )
            .into_response(),
        Err(e) => {
            tracing::error!("list_objects error: {e}");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                [("Content-Type", "application/xml")],
                responses::error_xml("InternalError", &e.to_string(), &bucket),
            )
                .into_response()
        }
    }
}

/// GET /{bucket}?versions — ListObjectVersions (stub: returns all as version "null").
async fn list_object_versions<B: StorageBackend>(
    State(state): State<AppState<B>>,
    Path(bucket): Path<String>,
    Query(query): Query<BucketGetQuery>,
) -> Response {
    let delimiter = query.delimiter.as_deref().filter(|d| !d.is_empty());
    match state
        .store
        .list_objects_v2(
            &bucket,
            query.prefix.as_deref(),
            query.key_marker.as_deref(),
            query.max_keys,
            delimiter,
        )
        .await
    {
        Ok(result) => {
            let max_keys = query.max_keys.unwrap_or(1000);
            let mut headers = HeaderMap::new();
            headers.insert("Content-Type", HeaderValue::from_static("application/xml"));
            headers.insert(
                "x-amz-request-id",
                HeaderValue::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
            );
            (
                StatusCode::OK,
                headers,
                responses::list_object_versions_xml(
                    &bucket,
                    &result,
                    max_keys,
                    query.key_marker.as_deref(),
                ),
            )
                .into_response()
        }
        Err(Post3Error::BucketNotFound(b)) => (
            StatusCode::NOT_FOUND,
            [("Content-Type", "application/xml")],
            responses::error_xml("NoSuchBucket", "The specified bucket does not exist", &b),
        )
            .into_response(),
        Err(e) => {
            tracing::error!("list_object_versions error: {e}");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                [("Content-Type", "application/xml")],
                responses::error_xml("InternalError", &e.to_string(), &bucket),
            )
                .into_response()
        }
    }
}

/// GET /{bucket}?location — GetBucketLocation.
async fn get_bucket_location<B: StorageBackend>(
    State(state): State<AppState<B>>,
    Path(bucket): Path<String>,
) -> Response {
    match state.store.head_bucket(&bucket).await {
        Ok(Some(_)) => {
            let mut headers = HeaderMap::new();
            headers.insert("Content-Type", HeaderValue::from_static("application/xml"));
            headers.insert(
                "x-amz-request-id",
                HeaderValue::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
            );
            (StatusCode::OK, headers, responses::get_bucket_location_xml()).into_response()
        }
        Ok(None) => (
            StatusCode::NOT_FOUND,
            [("Content-Type", "application/xml")],
            responses::error_xml(
                "NoSuchBucket",
                "The specified bucket does not exist",
                &bucket,
            ),
        )
            .into_response(),
        Err(e) => {
            tracing::error!("get_bucket_location error: {e}");
            StatusCode::INTERNAL_SERVER_ERROR.into_response()
        }
    }
}

/// POST /{bucket} — dispatches to DeleteObjects based on ?delete query param.
pub async fn bucket_post_dispatch<B: StorageBackend>(
    state: State<AppState<B>>,
    path: Path<String>,
    query: Query<crate::s3::extractors::BucketPostQuery>,
    body: Bytes,
) -> Response {
    if query.delete.is_some() {
        delete_objects(state, path, body).await
    } else {
        (
            StatusCode::BAD_REQUEST,
            [("Content-Type", "application/xml")],
            responses::error_xml(
                "InvalidRequest",
                "POST on bucket requires ?delete parameter",
                &format!("/{}", path.0),
            ),
        )
            .into_response()
    }
}

/// POST /{bucket}?delete — DeleteObjects (batch delete).
async fn delete_objects<B: StorageBackend>(
    State(state): State<AppState<B>>,
    Path(bucket): Path<String>,
    body: Bytes,
) -> Response {
    let (keys, quiet) = match responses::parse_delete_objects_xml(&body) {
        Ok(result) => result,
        Err(msg) => {
            return (
                StatusCode::BAD_REQUEST,
                [("Content-Type", "application/xml")],
                responses::error_xml("MalformedXML", &msg, &format!("/{bucket}")),
            )
                .into_response();
        }
    };

    // S3 limits DeleteObjects to 1000 keys
    if keys.len() > 1000 {
        return (
            StatusCode::BAD_REQUEST,
            [("Content-Type", "application/xml")],
            responses::error_xml(
                "MalformedXML",
                "The number of keys in a DeleteObjects request cannot exceed 1000",
                &format!("/{bucket}"),
            ),
        )
            .into_response();
    }

    let mut deleted = Vec::new();
    let mut errors: Vec<(String, String, String)> = Vec::new();

    for key in keys {
        match state.store.delete_object(&bucket, &key).await {
            Ok(()) => {
                if !quiet {
                    deleted.push(key);
                }
            }
            Err(e) => {
                errors.push((key, "InternalError".to_string(), e.to_string()));
            }
        }
    }

    let mut headers = HeaderMap::new();
    headers.insert("Content-Type", HeaderValue::from_static("application/xml"));
    headers.insert(
        "x-amz-request-id",
        HeaderValue::from_str(&uuid::Uuid::new_v4().to_string()).unwrap(),
    );

    (
        StatusCode::OK,
        headers,
        responses::delete_objects_result_xml(&deleted, &errors),
    )
        .into_response()
}