//! The S3 client itself: its handle, the operations it delegates, and its
//! implementation of [`super::StorageBackend`].
//!
//! The trait impl and the inherent methods it delegates to are two halves of
//! one contract, which is why they stay in one file.

use super::backend::StorageBackend;
use super::bucket::PRESIGN_EXPIRY_SECS;
use super::bucket::S3DeleteAuthority;
use super::key::S3Key;
use crate::config::StorageConfig;
use crate::error::{AppError, Result};

/// S3 client wrapper for presigned URL operations.
/// Delegates S3 operations to `s3_storage::S3Client`.
#[derive(Clone)]
pub struct S3Client {
    inner: s3_storage::S3Client,
}

impl S3Client {
    /// Create a new S3 client from storage configuration.
    ///
    /// Configures CORS on the bucket at startup so browser PUT uploads to
    /// presigned URLs work without manual bucket configuration.
    pub async fn new(config: &StorageConfig, host_url: &str) -> Result<Self> {
        let s3_config = s3_storage::S3Config {
            endpoint: config.endpoint.clone(),
            bucket: config.bucket.clone(),
            access_key: config.access_key.clone(),
            secret_key: config.secret_key.clone(),
            region: config.region.clone(),
        };

        let inner = s3_storage::S3Client::new(&s3_config)
            .await
            .map_err(AppError::Storage)?;

        inner.configure_cors(host_url).await;

        Ok(S3Client { inner })
    }

    /// Generate a presigned URL for uploading a file. `max_bytes`, when set,
    /// binds `Content-Length` into the signature, S3 will reject any PUT
    /// whose actual body length differs from `max_bytes`.
    pub async fn presign_upload(
        &self,
        s3_key: &S3Key,
        content_type: &str,
        expiry_secs: Option<u64>,
        cache_control: Option<&str>,
        max_bytes: Option<i64>,
    ) -> Result<String> {
        self.inner
            .presign_upload(
                s3_key.as_str(),
                content_type,
                expiry_secs.unwrap_or(PRESIGN_EXPIRY_SECS),
                cache_control,
                max_bytes,
            )
            .await
            .map_err(AppError::Storage)
    }

    /// Generate a presigned URL for downloading/streaming a file
    pub async fn presign_download(
        &self,
        s3_key: &S3Key,
        expiry_secs: Option<u64>,
    ) -> Result<String> {
        self.inner
            .presign_download(s3_key.as_str(), expiry_secs.unwrap_or(PRESIGN_EXPIRY_SECS))
            .await
            .map_err(AppError::Storage)
    }

    /// Check if an object exists in S3
    pub async fn object_exists(&self, s3_key: &str) -> Result<bool> {
        self.inner
            .object_exists(s3_key)
            .await
            .map_err(AppError::Storage)
    }

    /// Get the size of an object in S3 (bytes), or None if not found.
    pub async fn object_size(&self, s3_key: &str) -> Result<Option<i64>> {
        self.inner
            .object_size(s3_key)
            .await
            .map_err(AppError::Storage)
    }

    /// Download an object's bytes from S3
    pub async fn download_object(&self, s3_key: &str) -> Result<Vec<u8>> {
        self.inner
            .download(s3_key)
            .await
            .map(|(bytes, _content_type)| bytes)
            .map_err(AppError::Storage)
    }

    /// Download an object as `bytes::Bytes` without the `to_vec` copy. See trait docs.
    pub async fn download_object_buf(&self, s3_key: &str) -> Result<bytes::Bytes> {
        self.inner
            .download_buf(s3_key)
            .await
            .map(|(bytes, _content_type)| bytes)
            .map_err(AppError::Storage)
    }

    /// Stream an object's body from S3 without buffering. See trait docs.
    pub async fn download_stream(&self, s3_key: &str) -> Result<s3_storage::ByteStream> {
        self.inner
            .download_stream(s3_key)
            .await
            .map_err(AppError::Storage)
    }

    /// Read the first `len` bytes via a ranged S3 GET (for content sniffing).
    pub async fn download_head(&self, s3_key: &str, len: usize) -> Result<Vec<u8>> {
        self.inner
            .download_head(s3_key, len)
            .await
            .map_err(AppError::Storage)
    }

    /// Upload an object to S3 from bytes
    pub async fn upload_object(
        &self,
        s3_key: &S3Key,
        content_type: &str,
        data: Vec<u8>,
        cache_control: Option<&str>,
    ) -> Result<()> {
        self.inner
            .upload(s3_key.as_str(), content_type, data, cache_control)
            .await
            .map_err(AppError::Storage)
    }

    /// Delete an object from S3
    pub async fn delete_object(&self, s3_key: &S3Key) -> Result<()> {
        self.inner
            .delete(s3_key.as_str())
            .await
            .map_err(AppError::Storage)
    }

    /// Server-side copy within the bucket. See the [`StorageBackend::copy_object`]
    /// trait method for the scan-then-promote rationale.
    pub async fn copy_object(&self, src_key: &S3Key, dst_key: &S3Key) -> Result<()> {
        self.inner
            .copy_object(src_key.as_str(), dst_key.as_str())
            .await
            .map_err(AppError::Storage)
    }

    /// Server-side copy from `src_bucket` into this client's bucket. See the
    /// [`StorageBackend::copy_object_from`] trait method for the cross-bucket
    /// promote rationale.
    pub async fn copy_object_from(
        &self,
        src_bucket: &str,
        src_key: &S3Key,
        dst_key: &S3Key,
    ) -> Result<()> {
        self.inner
            .copy_object_from(src_bucket, src_key.as_str(), dst_key.as_str())
            .await
            .map_err(AppError::Storage)
    }

    /// Batched S3 delete (`DeleteObjects`, up to 1000 keys per call).
    /// Chunks larger slices into 1000-key batches and logs per-key failures
    /// without bubbling, the pending_s3_deletions queue is the safety net.
    pub async fn delete_objects(&self, keys: &[S3Key]) -> Result<()> {
        if keys.is_empty() {
            return Ok(());
        }
        for chunk in keys.chunks(1000) {
            let chunk: Vec<String> = chunk.iter().map(|k| k.as_str().to_string()).collect();
            match self.inner.delete_objects(&chunk).await {
                Ok(failures) => {
                    for (k, msg) in &failures {
                        tracing::warn!(key = %k, error = %msg, "S3 delete_objects: key-level failure");
                    }
                    // A whole-batch failure must not read as success (Run #2
                    // Storage MINOR); partial failures stay logged and the
                    // pending_s3_deletions queue is the retry net.
                    if !chunk.is_empty() && failures.len() == chunk.len() {
                        return Err(AppError::Storage(format!(
                            "S3 delete_objects: all {} keys in batch failed",
                            chunk.len()
                        )));
                    }
                }
                Err(e) => return Err(AppError::Storage(e)),
            }
        }
        Ok(())
    }

    /// Upload a file to S3 using multipart upload (10 MB parts).
    pub async fn upload_multipart(
        &self,
        s3_key: &S3Key,
        content_type: &str,
        file_path: &std::path::Path,
    ) -> Result<()> {
        self.inner
            .upload_multipart(s3_key.as_str(), content_type, file_path, None)
            .await
            .map_err(AppError::Storage)
    }

    /// Server-side multipart copy for sources over the 5 GiB single-part
    /// `CopyObject` limit. See the [`StorageBackend::copy_object_multipart`]
    /// trait method for the promote rationale. `part_size` of `None` lets the
    /// storage layer auto-size parts for `src_size`.
    pub async fn copy_object_multipart(
        &self,
        src_bucket: &str,
        src_key: &S3Key,
        dst_key: &S3Key,
        content_type: &str,
        src_size: u64,
        part_size: Option<usize>,
    ) -> Result<()> {
        self.inner
            .copy_object_multipart(
                src_bucket,
                src_key.as_str(),
                dst_key.as_str(),
                content_type,
                src_size,
                part_size,
            )
            .await
            .map_err(AppError::Storage)
    }

    /// Begin a client-direct multipart upload. See the
    /// [`StorageBackend::create_multipart_upload`] trait method.
    pub async fn create_multipart_upload(
        &self,
        s3_key: &S3Key,
        content_type: &str,
    ) -> Result<String> {
        self.inner
            .create_multipart_upload(s3_key.as_str(), content_type)
            .await
            .map_err(AppError::Storage)
    }

    /// Presign one `UploadPart` request. See the
    /// [`StorageBackend::presign_upload_part`] trait method.
    pub async fn presign_upload_part(
        &self,
        s3_key: &S3Key,
        upload_id: &str,
        part_number: i32,
        expiry_secs: Option<u64>,
        max_bytes: Option<i64>,
        checksum_sha256: Option<&str>,
    ) -> Result<String> {
        self.inner
            .presign_upload_part(
                s3_key.as_str(),
                upload_id,
                part_number,
                expiry_secs.unwrap_or(PRESIGN_EXPIRY_SECS),
                max_bytes,
                checksum_sha256,
            )
            .await
            .map_err(AppError::Storage)
    }

    /// Complete a multipart upload. See the
    /// [`StorageBackend::complete_multipart_upload`] trait method.
    pub async fn complete_multipart_upload(
        &self,
        s3_key: &S3Key,
        upload_id: &str,
        parts: &[(i32, String)],
    ) -> Result<()> {
        self.inner
            .complete_multipart_upload(s3_key.as_str(), upload_id, parts)
            .await
            .map_err(AppError::Storage)
    }

    /// Abort a multipart upload. See the
    /// [`StorageBackend::abort_multipart_upload`] trait method.
    pub async fn abort_multipart_upload(&self, s3_key: &S3Key, upload_id: &str) -> Result<()> {
        self.inner
            .abort_multipart_upload(s3_key.as_str(), upload_id)
            .await
            .map_err(AppError::Storage)
    }

    /// In-progress multipart sessions for a key. See the
    /// [`StorageBackend::list_multipart_uploads_for_key`] trait method.
    pub async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result<Vec<String>> {
        self.inner
            .list_multipart_uploads_for_key(s3_key)
            .await
            .map_err(AppError::Storage)
    }

    /// Lightweight connectivity check, issues a list with max_keys(0).
    pub async fn check_connectivity(&self) -> std::result::Result<(), String> {
        self.inner.check_connectivity().await
    }
}

#[async_trait::async_trait]
impl StorageBackend for S3Client {
    async fn presign_upload(
        &self,
        s3_key: &S3Key,
        content_type: &str,
        expiry_secs: Option<u64>,
        cache_control: Option<&str>,
        max_bytes: Option<i64>,
    ) -> Result<String> {
        self.presign_upload(s3_key, content_type, expiry_secs, cache_control, max_bytes)
            .await
    }

    async fn presign_download(&self, s3_key: &S3Key, expiry_secs: Option<u64>) -> Result<String> {
        self.presign_download(s3_key, expiry_secs).await
    }

    async fn object_exists(&self, s3_key: &str) -> Result<bool> {
        self.object_exists(s3_key).await
    }

    async fn object_size(&self, s3_key: &str) -> Result<Option<i64>> {
        self.object_size(s3_key).await
    }

    async fn download_object(&self, s3_key: &str) -> Result<Vec<u8>> {
        self.download_object(s3_key).await
    }

    async fn download_object_buf(&self, s3_key: &str) -> Result<bytes::Bytes> {
        self.download_object_buf(s3_key).await
    }

    async fn download_stream(&self, s3_key: &str) -> Result<s3_storage::ByteStream> {
        self.download_stream(s3_key).await
    }

    async fn download_head(&self, s3_key: &str, len: usize) -> Result<Vec<u8>> {
        self.download_head(s3_key, len).await
    }

    async fn upload_object(
        &self,
        s3_key: &S3Key,
        content_type: &str,
        data: Vec<u8>,
        cache_control: Option<&str>,
    ) -> Result<()> {
        self.upload_object(s3_key, content_type, data, cache_control)
            .await
    }

    async fn delete_object(&self, _auth: &S3DeleteAuthority, s3_key: &S3Key) -> Result<()> {
        // Authority proven by the caller; delegate to the inherent impl.
        self.delete_object(s3_key).await
    }

    async fn delete_objects(&self, _auth: &S3DeleteAuthority, keys: &[S3Key]) -> Result<()> {
        self.delete_objects(keys).await
    }

    async fn copy_object_from(
        &self,
        src_bucket: &str,
        src_key: &S3Key,
        dst_key: &S3Key,
    ) -> Result<()> {
        // Inherent method; delegate.
        self.copy_object_from(src_bucket, src_key, dst_key).await
    }

    async fn copy_object(&self, src_key: &S3Key, dst_key: &S3Key) -> Result<()> {
        // Inherent method; delegate.
        self.copy_object(src_key, dst_key).await
    }

    async fn delete_prefix(&self, _auth: &S3DeleteAuthority, prefix: &str) -> Result<()> {
        self.inner
            .delete_prefix(prefix)
            .await
            .map_err(AppError::Storage)
    }

    async fn upload_multipart(
        &self,
        s3_key: &S3Key,
        content_type: &str,
        file_path: &std::path::Path,
    ) -> Result<()> {
        self.upload_multipart(s3_key, content_type, file_path).await
    }

    async fn copy_object_multipart(
        &self,
        src_bucket: &str,
        src_key: &S3Key,
        dst_key: &S3Key,
        content_type: &str,
        src_size: u64,
        part_size: Option<usize>,
    ) -> Result<()> {
        // Inherent method; delegate.
        self.copy_object_multipart(
            src_bucket,
            src_key,
            dst_key,
            content_type,
            src_size,
            part_size,
        )
        .await
    }

    async fn create_multipart_upload(&self, s3_key: &S3Key, content_type: &str) -> Result<String> {
        self.create_multipart_upload(s3_key, content_type).await
    }

    async fn presign_upload_part(
        &self,
        s3_key: &S3Key,
        upload_id: &str,
        part_number: i32,
        expiry_secs: Option<u64>,
        max_bytes: Option<i64>,
        checksum_sha256: Option<&str>,
    ) -> Result<String> {
        self.presign_upload_part(
            s3_key,
            upload_id,
            part_number,
            expiry_secs,
            max_bytes,
            checksum_sha256,
        )
        .await
    }

    async fn complete_multipart_upload(
        &self,
        s3_key: &S3Key,
        upload_id: &str,
        parts: &[(i32, String)],
    ) -> Result<()> {
        self.complete_multipart_upload(s3_key, upload_id, parts)
            .await
    }

    async fn abort_multipart_upload(&self, s3_key: &S3Key, upload_id: &str) -> Result<()> {
        self.abort_multipart_upload(s3_key, upload_id).await
    }

    async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result<Vec<String>> {
        self.list_multipart_uploads_for_key(s3_key).await
    }

    async fn check_connectivity(&self) -> std::result::Result<(), String> {
        self.check_connectivity().await
    }

    fn bucket(&self) -> &str {
        self.inner.bucket()
    }
}