use anyhow::{Context, Result};
use aws_sdk_s3::{
    Client as S3Client, operation::delete_objects::DeleteObjectsOutput, primitives::ByteStream,
};
use tokio::{fs::File, io::AsyncReadExt};

pub async fn s3_list_objects(
    s3_client: &S3Client,
    bucket: &str,
    prefix: &str,
) -> Result<Vec<String>> {
    let mut all_keys = Vec::new();
    let mut continuation_token = None;

    loop {
        let response = s3_client
            .list_objects_v2()
            .bucket(bucket)
            .prefix(prefix)
            .set_continuation_token(continuation_token)
            .send()
            .await
            .with_context(|| format!("{}/{}", bucket, prefix))?;

        let objects = response.contents.unwrap_or_default();
        let keys: Vec<String> = objects.into_iter().filter_map(|obj| obj.key).collect();
        all_keys.extend(keys);

        continuation_token = response.next_continuation_token;
        if continuation_token.is_none() {
            break;
        }
    }

    Ok(all_keys)
}

pub async fn s3_delete_objects_by_prefix(
    s3_client: &S3Client,
    bucket: String,
    prefix: String,
    batch_size: Option<usize>,
) -> Result<usize> {
    let batch_size = batch_size.unwrap_or(1000);
    let mut total_deleted = 0;

    let all_keys = s3_list_objects(s3_client, &bucket, &prefix).await?;
    if all_keys.is_empty() {
        log::info!("No objects found with prefix={}", prefix);
        return Ok(0);
    }

    for chunk in all_keys.chunks(batch_size) {
        let response = s3_delete_objects(s3_client, &bucket, chunk).await?;

        if let Some(deleted) = response.deleted {
            total_deleted += deleted.len();
        }

        if let Some(errors) = response.errors {
            for error in errors {
                log::warn!(
                    "Failed to delete object {}: {}",
                    error.key.unwrap_or_default(),
                    error.message.unwrap_or_default()
                );
            }
        } else {
            log::info!("s3_delete_objects: delete {} files", batch_size);
        }
    }

    Ok(total_deleted)
}

pub async fn s3_delete_objects(
    s3_client: &S3Client,
    bucket: &str,
    keys: &[String],
) -> Result<DeleteObjectsOutput> {
    let mut objects_to_delete = Vec::with_capacity(keys.len());
    for key in keys {
        let obj = aws_sdk_s3::types::ObjectIdentifier::builder()
            .key(key)
            .build()
            .with_context(|| format!("Failed to build ObjectIdentifier: key={}", key))?;
        objects_to_delete.push(obj);
    }

    let delete_request = aws_sdk_s3::types::Delete::builder()
        .set_objects(Some(objects_to_delete))
        .build()
        .context("Failed to build Delete request")?;

    let response = s3_client
        .delete_objects()
        .bucket(bucket)
        .delete(delete_request)
        .send()
        .await?;

    Ok(response)
}

pub async fn s3_upload_data(
    s3_client: &S3Client,
    s3_bucket: &str,
    s3_key: &str,
    buffer: Vec<u8>,
) -> Result<()> {
    let body = ByteStream::from(buffer);
    s3_client
        .put_object()
        .bucket(s3_bucket)
        .key(s3_key)
        .body(body)
        .send()
        .await?;
    Ok(())
}

pub async fn s3_upload_object(
    s3_client: &S3Client,
    s3_bucket: &str,
    s3_key: &str,
    local_fname: &str,
) -> Result<usize> {
    log::debug!(
        "Upload file to S3, bucket={} key={} local_fname={}",
        s3_bucket,
        s3_key,
        local_fname
    );
    let mut file = File::open(&local_fname)
        .await
        .with_context(|| format!("Open {} failed", local_fname))?;
    let mut buffer = Vec::new();
    file.read_to_end(&mut buffer)
        .await
        .with_context(|| format!("Read {} failed", local_fname))?;
    let file_size = buffer.len();
    s3_upload_data(s3_client, s3_bucket, s3_key, buffer)
        .await
        .with_context(|| {
            format!(
                "Upload {} to S3 as {} failed, bucket={}",
                local_fname, s3_key, s3_bucket
            )
        })?;
    log::debug!(
        "Upload file to S3 success, bucket={} key={} local_fname={} size={}",
        s3_bucket,
        s3_key,
        local_fname,
        file_size
    );
    Ok(file_size)
}