# S3 Tasklet Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Add four S3 tasklets (`S3PutTasklet`, `S3GetTasklet`, `S3PutFolderTasklet`, `S3GetFolderTasklet`) to spring-batch-rs for pure file transfer operations, following the same builder pattern as the existing FTP tasklets.

**Architecture:** Module `src/tasklet/s3/` with shared client config in `mod.rs`, upload tasklets in `put.rs`, download tasklets in `get.rs`. The synchronous `Tasklet::execute` trait bridges to async AWS SDK calls via `tokio::task::block_in_place`.

**Tech Stack:** `aws-sdk-s3` v1, `aws-config` v1, `tokio-util` v0.7 (streaming), `tokio::task::block_in_place` (sync/async bridge), existing `BatchError` variants.

---

## File Map

| Action | Path | Responsibility |
|---|---|---|
| Modify | `Cargo.toml` | Add `aws-sdk-s3`, `aws-config`, `tokio-util` deps + `s3` feature + example entry |
| Modify | `src/tasklet/mod.rs` | Declare `pub mod s3` behind `#[cfg(feature = "s3")]` |
| Create | `src/tasklet/s3/mod.rs` | Module docs, `S3ClientConfig`, `build_s3_client()` async helper, public re-exports |
| Create | `src/tasklet/s3/put.rs` | `S3PutTasklet` + `S3PutTaskletBuilder`, `S3PutFolderTasklet` + `S3PutFolderTaskletBuilder` |
| Create | `src/tasklet/s3/get.rs` | `S3GetTasklet` + `S3GetTaskletBuilder`, `S3GetFolderTasklet` + `S3GetFolderTaskletBuilder` |
| Create | `examples/tasklet_s3.rs` | Demonstrates upload + download with LocalStack (`no_run` doc-test) |

---

## Task 1: Add dependencies and feature flag in `Cargo.toml`

**Files:**
- Modify: `Cargo.toml`

- [ ] **Step 1: Add the three new optional dependencies**

In `Cargo.toml`, in the `[dependencies]` section, add after the `suppaftp` line:

```toml
aws-sdk-s3 = { version = "1", optional = true }
aws-config = { version = "1", optional = true }
tokio-util = { version = "0.7", features = ["io"], optional = true }
```

- [ ] **Step 2: Add the `s3` feature flag**

In `Cargo.toml`, in the `[features]` section, add after the `ftp` line:

```toml
s3 = ["dep:aws-sdk-s3", "dep:aws-config", "dep:tokio-util"]
```

- [ ] **Step 3: Add `s3` to the `full` feature**

Change the `full` feature to include `"s3"`:

```toml
full = [
  "csv",
  "json",
  "logger",
  "fake",
  "rdbc-postgres",
  "rdbc-mysql",
  "rdbc-sqlite",
  "mongodb",
  "xml",
  "zip",
  "ftp",
  "orm",
  "s3",
]
```

- [ ] **Step 4: Add the example entry**

In `Cargo.toml`, in the tasklets examples section, after the `tasklet_ftp` entry:

```toml
[[example]]
name = "tasklet_s3"
required-features = ["s3"]
```

- [ ] **Step 5: Verify the project compiles**

```bash
cargo build --features s3
```

Expected: compiles without errors (no new source files yet, so no code errors; just dependency resolution).

- [ ] **Step 6: Commit**

```bash
git add Cargo.toml Cargo.lock
git commit -m "chore: add aws-sdk-s3, aws-config, tokio-util dependencies with s3 feature flag"
```

---

## Task 2: Create `src/tasklet/s3/mod.rs` — shared infrastructure

**Files:**
- Create: `src/tasklet/s3/mod.rs`

- [ ] **Step 1: Create the file**

Create `src/tasklet/s3/mod.rs` with this exact content:

```rust
//! # S3 Tasklet
//!
//! This module provides tasklets for Amazon S3 file transfer operations (put and get).
//! It is designed to be similar to Spring Batch's S3 capabilities for batch file transfers.
//!
//! ## Features
//!
//! - S3 PUT operations (upload local files to S3)
//! - S3 GET operations (download S3 objects to local files, streaming)
//! - S3 PUT FOLDER operations (upload entire local folder to an S3 prefix)
//! - S3 GET FOLDER operations (download all objects under an S3 prefix to a local folder)
//! - Explicit credential configuration (access key / secret key)
//! - AWS default credential chain (environment variables, `~/.aws/credentials`, IAM role)
//! - Custom endpoint URL for S3-compatible services (MinIO, LocalStack)
//! - Configurable multipart upload chunk size
//!
//! ## Examples
//!
//! ### S3 PUT Operation
//!
//! ```rust,no_run
//! use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
//!
//! # fn example() -> Result<(), spring_batch_rs::BatchError> {
//! let tasklet = S3PutTaskletBuilder::new()
//!     .bucket("my-bucket")
//!     .key("exports/file.csv")
//!     .local_file("./output/file.csv")
//!     .region("eu-west-1")
//!     .build()?;
//! # Ok(())
//! # }
//! ```
//!
//! ### S3 GET Operation
//!
//! ```rust,no_run
//! use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
//!
//! # fn example() -> Result<(), spring_batch_rs::BatchError> {
//! let tasklet = S3GetTaskletBuilder::new()
//!     .bucket("my-bucket")
//!     .key("imports/file.csv")
//!     .local_file("./input/file.csv")
//!     .region("eu-west-1")
//!     .build()?;
//! # Ok(())
//! # }
//! ```
//!
//! ### S3 with MinIO (custom endpoint)
//!
//! ```rust,no_run
//! use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
//!
//! # fn example() -> Result<(), spring_batch_rs::BatchError> {
//! let tasklet = S3PutTaskletBuilder::new()
//!     .bucket("my-bucket")
//!     .key("file.csv")
//!     .local_file("./output/file.csv")
//!     .endpoint_url("http://localhost:9000")
//!     .access_key_id("minioadmin")
//!     .secret_access_key("minioadmin")
//!     .build()?;
//! # Ok(())
//! # }
//! ```

pub mod get;
pub mod put;

use crate::BatchError;
use aws_config::meta::region::RegionProviderChain;
use aws_sdk_s3::config::Builder as S3ConfigBuilder;
use aws_sdk_s3::config::Credentials;

/// Configuration for connecting to an S3-compatible service.
///
/// All fields are optional. When `access_key_id` and `secret_access_key` are both
/// `None`, the AWS default credential chain is used (environment variables,
/// `~/.aws/credentials`, IAM instance role).
///
/// Set `endpoint_url` to connect to S3-compatible services such as MinIO or LocalStack.
///
/// # Examples
///
/// ```
/// use spring_batch_rs::tasklet::s3::S3ClientConfig;
///
/// let config = S3ClientConfig {
///     region: Some("eu-west-1".to_string()),
///     endpoint_url: None,
///     access_key_id: None,
///     secret_access_key: None,
/// };
/// assert_eq!(config.region.as_deref(), Some("eu-west-1"));
/// ```
#[derive(Debug, Clone, Default)]
pub struct S3ClientConfig {
    /// AWS region (e.g. `"eu-west-1"`). Falls back to `AWS_DEFAULT_REGION` env var when `None`.
    pub region: Option<String>,
    /// Custom endpoint URL for S3-compatible services (e.g. `"http://localhost:9000"` for MinIO).
    pub endpoint_url: Option<String>,
    /// AWS access key ID. Uses default credential chain when `None`.
    pub access_key_id: Option<String>,
    /// AWS secret access key. Uses default credential chain when `None`.
    pub secret_access_key: Option<String>,
}

/// Builds an [`aws_sdk_s3::Client`] from the given [`S3ClientConfig`].
///
/// When both `access_key_id` and `secret_access_key` are set, explicit static
/// credentials are used. Otherwise the AWS default credential chain applies.
///
/// # Errors
///
/// Returns [`BatchError::Configuration`] if the AWS SDK configuration cannot be loaded.
pub(crate) async fn build_s3_client(config: &S3ClientConfig) -> Result<aws_sdk_s3::Client, BatchError> {
    let region_provider = RegionProviderChain::first_try(
        config.region.clone().map(aws_sdk_s3::config::Region::new),
    )
    .or_default_provider();

    let sdk_config = aws_config::from_env()
        .region(region_provider)
        .load()
        .await;

    let mut builder = S3ConfigBuilder::from(&sdk_config);

    if let Some(url) = &config.endpoint_url {
        builder = builder.endpoint_url(url).force_path_style(true);
    }

    if let (Some(key_id), Some(secret)) = (&config.access_key_id, &config.secret_access_key) {
        let creds = Credentials::new(key_id, secret, None, None, "static");
        builder = builder.credentials_provider(creds);
    }

    Ok(aws_sdk_s3::Client::from_conf(builder.build()))
}

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

    #[test]
    fn should_default_to_none_fields() {
        let config = S3ClientConfig::default();
        assert!(config.region.is_none(), "region should default to None");
        assert!(config.endpoint_url.is_none(), "endpoint_url should default to None");
        assert!(config.access_key_id.is_none(), "access_key_id should default to None");
        assert!(config.secret_access_key.is_none(), "secret_access_key should default to None");
    }

    #[test]
    fn should_store_region() {
        let config = S3ClientConfig {
            region: Some("us-east-1".to_string()),
            ..Default::default()
        };
        assert_eq!(config.region.as_deref(), Some("us-east-1"));
    }

    #[test]
    fn should_store_endpoint_url() {
        let config = S3ClientConfig {
            endpoint_url: Some("http://localhost:9000".to_string()),
            ..Default::default()
        };
        assert_eq!(config.endpoint_url.as_deref(), Some("http://localhost:9000"));
    }

    #[test]
    fn should_store_explicit_credentials() {
        let config = S3ClientConfig {
            access_key_id: Some("AKID".to_string()),
            secret_access_key: Some("SECRET".to_string()),
            ..Default::default()
        };
        assert_eq!(config.access_key_id.as_deref(), Some("AKID"));
        assert_eq!(config.secret_access_key.as_deref(), Some("SECRET"));
    }
}
```

- [ ] **Step 2: Verify the module compiles**

```bash
cargo build --features s3 2>&1 | grep -E "^error"
```

Expected: no output (no errors). The `put` and `get` sub-modules don't exist yet so there will be errors about missing modules — that is expected. Fix them by creating empty placeholder files first if needed:

```bash
mkdir -p src/tasklet/s3
touch src/tasklet/s3/put.rs src/tasklet/s3/get.rs
```

Then re-run:

```bash
cargo build --features s3 2>&1 | grep -E "^error"
```

Expected: no errors.

---

## Task 3: Update `src/tasklet/mod.rs` to declare the s3 module

**Files:**
- Modify: `src/tasklet/mod.rs`

- [ ] **Step 1: Read the current file**

Current content of `src/tasklet/mod.rs`:
```rust
//! # Tasklet Module
//!
//! This module provides various tasklet implementations for common batch operations.
//! Tasklets are single-task operations that don't follow the chunk-oriented processing pattern.

#[cfg(feature = "zip")]
#[cfg_attr(docsrs, doc(cfg(feature = "zip")))]
pub mod zip;

#[cfg(feature = "ftp")]
#[cfg_attr(docsrs, doc(cfg(feature = "ftp")))]
pub mod ftp;
```

- [ ] **Step 2: Add the s3 module declaration**

Append to the end of `src/tasklet/mod.rs`:

```rust

#[cfg(feature = "s3")]
#[cfg_attr(docsrs, doc(cfg(feature = "s3")))]
pub mod s3;
```

- [ ] **Step 3: Verify**

```bash
cargo build --features s3 2>&1 | grep "^error"
```

Expected: no errors.

- [ ] **Step 4: Commit**

```bash
git add src/tasklet/mod.rs src/tasklet/s3/mod.rs
git commit -m "feat(s3): add s3 tasklet module with shared S3ClientConfig and build_s3_client helper"
```

---

## Task 4: Implement `S3PutTasklet` and `S3PutTaskletBuilder` in `src/tasklet/s3/put.rs`

**Files:**
- Modify: `src/tasklet/s3/put.rs`

The `execute` method is synchronous (trait requirement). It bridges to async S3 SDK calls via `tokio::task::block_in_place(|| Handle::current().block_on(...))`, which requires a tokio multi-thread runtime (provided by `#[tokio::main]`).

Multipart upload logic:
- Files **smaller than** `chunk_size` (default 8 MiB): single `put_object`.
- Files **≥ chunk_size**: `create_multipart_upload` → `upload_part` × N → `complete_multipart_upload`. On any failure: `abort_multipart_upload`.

- [ ] **Step 1: Write failing builder tests first**

Replace the empty `src/tasklet/s3/put.rs` with:

```rust
#[cfg(test)]
mod tests {
    // Types will be imported once implemented in this file
    // For now, write the tests to drive the implementation

    #[test]
    fn should_fail_build_when_bucket_missing() {
        // Will import S3PutTaskletBuilder once defined
        // Placeholder: implement after struct is defined below
        // — will be filled in Step 3
    }
}
```

Run to confirm it compiles (empty test always passes):

```bash
cargo test --features s3 -q 2>&1 | grep -E "FAILED|error"
```

Expected: no failures.

- [ ] **Step 2: Implement `S3PutTasklet` and `S3PutTaskletBuilder`**

Replace the content of `src/tasklet/s3/put.rs` entirely with:

```rust
//! S3 PUT tasklets for uploading files and folders to Amazon S3.

use crate::{
    core::step::{RepeatStatus, StepExecution, Tasklet},
    tasklet::s3::{build_s3_client, S3ClientConfig},
    BatchError,
};
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
use log::{debug, info};
use std::path::{Path, PathBuf};
use tokio::runtime::Handle;

const DEFAULT_CHUNK_SIZE: usize = 8 * 1024 * 1024; // 8 MiB

/// A tasklet that uploads a single local file to an S3 object.
///
/// Files smaller than `chunk_size` are uploaded with a single `put_object` call.
/// Files equal to or larger than `chunk_size` use multipart upload. If multipart
/// upload fails mid-way, `abort_multipart_upload` is called to avoid orphaned parts.
///
/// # Examples
///
/// ```rust,no_run
/// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
///
/// # fn example() -> Result<(), spring_batch_rs::BatchError> {
/// let tasklet = S3PutTaskletBuilder::new()
///     .bucket("my-bucket")
///     .key("exports/file.csv")
///     .local_file("./output/file.csv")
///     .region("eu-west-1")
///     .build()?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns [`BatchError::ItemWriter`] if the S3 upload fails.
/// Returns [`BatchError::Io`] if the local file cannot be read.
#[derive(Debug)]
pub struct S3PutTasklet {
    bucket: String,
    key: String,
    local_file: PathBuf,
    chunk_size: usize,
    config: S3ClientConfig,
}

impl S3PutTasklet {
    async fn execute_async(&self) -> Result<RepeatStatus, BatchError> {
        info!(
            "Uploading {} to s3://{}/{}",
            self.local_file.display(),
            self.bucket,
            self.key
        );

        let client = build_s3_client(&self.config).await?;
        let file_size = std::fs::metadata(&self.local_file)
            .map_err(BatchError::Io)?
            .len() as usize;

        if file_size < self.chunk_size {
            // Simple single-part upload
            let body = ByteStream::from_path(&self.local_file)
                .await
                .map_err(|e| BatchError::ItemWriter(format!("Failed to read file for upload: {}", e)))?;

            client
                .put_object()
                .bucket(&self.bucket)
                .key(&self.key)
                .body(body)
                .send()
                .await
                .map_err(|e| BatchError::ItemWriter(format!("S3 put_object failed: {}", e)))?;
        } else {
            // Multipart upload
            upload_multipart(&client, &self.bucket, &self.key, &self.local_file, self.chunk_size).await?;
        }

        info!("Upload complete: s3://{}/{}", self.bucket, self.key);
        Ok(RepeatStatus::Finished)
    }
}

impl Tasklet for S3PutTasklet {
    fn execute(&self, _step_execution: &StepExecution) -> Result<RepeatStatus, BatchError> {
        tokio::task::block_in_place(|| Handle::current().block_on(self.execute_async()))
    }
}

/// Builder for [`S3PutTasklet`].
///
/// # Examples
///
/// ```rust,no_run
/// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
///
/// # fn example() -> Result<(), spring_batch_rs::BatchError> {
/// let tasklet = S3PutTaskletBuilder::new()
///     .bucket("my-bucket")
///     .key("exports/file.csv")
///     .local_file("./output/file.csv")
///     .region("eu-west-1")
///     .chunk_size(16 * 1024 * 1024)
///     .build()?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns [`BatchError::Configuration`] if `bucket`, `key`, or `local_file` are not set.
#[derive(Debug, Default)]
pub struct S3PutTaskletBuilder {
    bucket: Option<String>,
    key: Option<String>,
    local_file: Option<PathBuf>,
    chunk_size: usize,
    config: S3ClientConfig,
}

impl S3PutTaskletBuilder {
    /// Creates a new builder with default settings.
    ///
    /// Default `chunk_size` is 8 MiB.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// let builder = S3PutTaskletBuilder::new();
    /// ```
    pub fn new() -> Self {
        Self {
            chunk_size: DEFAULT_CHUNK_SIZE,
            ..Default::default()
        }
    }

    /// Sets the S3 bucket name.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// let builder = S3PutTaskletBuilder::new().bucket("my-bucket");
    /// ```
    pub fn bucket<S: Into<String>>(mut self, bucket: S) -> Self {
        self.bucket = Some(bucket.into());
        self
    }

    /// Sets the S3 object key (path within the bucket).
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// let builder = S3PutTaskletBuilder::new().key("prefix/file.csv");
    /// ```
    pub fn key<S: Into<String>>(mut self, key: S) -> Self {
        self.key = Some(key.into());
        self
    }

    /// Sets the local file path to upload.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// let builder = S3PutTaskletBuilder::new().local_file("./output/file.csv");
    /// ```
    pub fn local_file<P: AsRef<Path>>(mut self, path: P) -> Self {
        self.local_file = Some(path.as_ref().to_path_buf());
        self
    }

    /// Sets the AWS region.
    ///
    /// Defaults to the `AWS_DEFAULT_REGION` environment variable when not set.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// let builder = S3PutTaskletBuilder::new().region("eu-west-1");
    /// ```
    pub fn region<S: Into<String>>(mut self, region: S) -> Self {
        self.config.region = Some(region.into());
        self
    }

    /// Sets a custom endpoint URL for S3-compatible services (MinIO, LocalStack).
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// let builder = S3PutTaskletBuilder::new().endpoint_url("http://localhost:9000");
    /// ```
    pub fn endpoint_url<S: Into<String>>(mut self, url: S) -> Self {
        self.config.endpoint_url = Some(url.into());
        self
    }

    /// Sets the AWS access key ID for explicit credential configuration.
    ///
    /// Must be combined with [`secret_access_key`](Self::secret_access_key).
    /// Falls back to the AWS default credential chain when not set.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// let builder = S3PutTaskletBuilder::new().access_key_id("AKIAIOSFODNN7EXAMPLE");
    /// ```
    pub fn access_key_id<S: Into<String>>(mut self, key_id: S) -> Self {
        self.config.access_key_id = Some(key_id.into());
        self
    }

    /// Sets the AWS secret access key for explicit credential configuration.
    ///
    /// Must be combined with [`access_key_id`](Self::access_key_id).
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// let builder = S3PutTaskletBuilder::new().secret_access_key("wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY");
    /// ```
    pub fn secret_access_key<S: Into<String>>(mut self, secret: S) -> Self {
        self.config.secret_access_key = Some(secret.into());
        self
    }

    /// Sets the multipart upload chunk size in bytes.
    ///
    /// Files smaller than this value are uploaded in a single request. Files
    /// equal to or larger are split into parts of this size. Defaults to `8 MiB`.
    /// Minimum value is 5 MiB (AWS requirement for multipart parts).
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// let builder = S3PutTaskletBuilder::new().chunk_size(16 * 1024 * 1024);
    /// ```
    pub fn chunk_size(mut self, size: usize) -> Self {
        self.chunk_size = size;
        self
    }

    /// Builds the [`S3PutTasklet`].
    ///
    /// # Errors
    ///
    /// Returns [`BatchError::Configuration`] if `bucket`, `key`, or `local_file` are not set.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use spring_batch_rs::tasklet::s3::put::S3PutTaskletBuilder;
    ///
    /// # fn example() -> Result<(), spring_batch_rs::BatchError> {
    /// let tasklet = S3PutTaskletBuilder::new()
    ///     .bucket("my-bucket")
    ///     .key("file.csv")
    ///     .local_file("./output/file.csv")
    ///     .build()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn build(self) -> Result<S3PutTasklet, BatchError> {
        let bucket = self
            .bucket
            .ok_or_else(|| BatchError::Configuration("S3PutTasklet: 'bucket' is required".to_string()))?;
        let key = self
            .key
            .ok_or_else(|| BatchError::Configuration("S3PutTasklet: 'key' is required".to_string()))?;
        let local_file = self
            .local_file
            .ok_or_else(|| BatchError::Configuration("S3PutTasklet: 'local_file' is required".to_string()))?;

        Ok(S3PutTasklet {
            bucket,
            key,
            local_file,
            chunk_size: self.chunk_size,
            config: self.config,
        })
    }
}

// ---------------------------------------------------------------------------
// S3PutFolderTasklet
// ---------------------------------------------------------------------------

/// A tasklet that uploads all files from a local folder to an S3 prefix.
///
/// Each file under `local_folder` is uploaded as `<prefix><relative_path>`.
/// Files are uploaded one at a time. Files equal to or larger than `chunk_size`
/// use multipart upload.
///
/// # Examples
///
/// ```rust,no_run
/// use spring_batch_rs::tasklet::s3::put::S3PutFolderTaskletBuilder;
///
/// # fn example() -> Result<(), spring_batch_rs::BatchError> {
/// let tasklet = S3PutFolderTaskletBuilder::new()
///     .bucket("my-bucket")
///     .prefix("backups/2026-04-10/")
///     .local_folder("./exports/")
///     .region("eu-west-1")
///     .build()?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns [`BatchError::ItemWriter`] if any S3 upload fails.
/// Returns [`BatchError::Io`] if local folder traversal or file reads fail.
#[derive(Debug)]
pub struct S3PutFolderTasklet {
    bucket: String,
    prefix: String,
    local_folder: PathBuf,
    chunk_size: usize,
    config: S3ClientConfig,
}

impl S3PutFolderTasklet {
    async fn execute_async(&self) -> Result<RepeatStatus, BatchError> {
        info!(
            "Uploading folder {} to s3://{}/{}",
            self.local_folder.display(),
            self.bucket,
            self.prefix
        );

        let client = build_s3_client(&self.config).await?;
        let entries = collect_files(&self.local_folder)?;

        for local_path in &entries {
            let relative = local_path
                .strip_prefix(&self.local_folder)
                .map_err(|e| BatchError::Io(std::io::Error::other(e.to_string())))?;
            let key = format!("{}{}", self.prefix, relative.to_string_lossy().replace('\\', "/"));

            let file_size = std::fs::metadata(local_path).map_err(BatchError::Io)?.len() as usize;

            debug!("Uploading {} -> s3://{}/{}", local_path.display(), self.bucket, key);

            if file_size < self.chunk_size {
                let body = ByteStream::from_path(local_path)
                    .await
                    .map_err(|e| BatchError::ItemWriter(format!("Failed to read {}: {}", local_path.display(), e)))?;

                client
                    .put_object()
                    .bucket(&self.bucket)
                    .key(&key)
                    .body(body)
                    .send()
                    .await
                    .map_err(|e| BatchError::ItemWriter(format!("S3 put_object failed for {}: {}", key, e)))?;
            } else {
                upload_multipart(&client, &self.bucket, &key, local_path, self.chunk_size).await?;
            }
        }

        info!(
            "Folder upload complete: {} files uploaded to s3://{}/{}",
            entries.len(),
            self.bucket,
            self.prefix
        );
        Ok(RepeatStatus::Finished)
    }
}

impl Tasklet for S3PutFolderTasklet {
    fn execute(&self, _step_execution: &StepExecution) -> Result<RepeatStatus, BatchError> {
        tokio::task::block_in_place(|| Handle::current().block_on(self.execute_async()))
    }
}

/// Builder for [`S3PutFolderTasklet`].
///
/// # Examples
///
/// ```rust,no_run
/// use spring_batch_rs::tasklet::s3::put::S3PutFolderTaskletBuilder;
///
/// # fn example() -> Result<(), spring_batch_rs::BatchError> {
/// let tasklet = S3PutFolderTaskletBuilder::new()
///     .bucket("my-bucket")
///     .prefix("backups/2026-04-10/")
///     .local_folder("./exports/")
///     .build()?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns [`BatchError::Configuration`] if `bucket`, `prefix`, or `local_folder` are not set.
#[derive(Debug, Default)]
pub struct S3PutFolderTaskletBuilder {
    bucket: Option<String>,
    prefix: Option<String>,
    local_folder: Option<PathBuf>,
    chunk_size: usize,
    config: S3ClientConfig,
}

impl S3PutFolderTaskletBuilder {
    /// Creates a new builder with default settings.
    pub fn new() -> Self {
        Self {
            chunk_size: DEFAULT_CHUNK_SIZE,
            ..Default::default()
        }
    }

    /// Sets the S3 bucket name.
    pub fn bucket<S: Into<String>>(mut self, bucket: S) -> Self {
        self.bucket = Some(bucket.into());
        self
    }

    /// Sets the S3 key prefix for uploaded objects.
    ///
    /// Defaults to `""` (bucket root) when not set.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutFolderTaskletBuilder;
    ///
    /// let builder = S3PutFolderTaskletBuilder::new().prefix("backups/2026-04-10/");
    /// ```
    pub fn prefix<S: Into<String>>(mut self, prefix: S) -> Self {
        self.prefix = Some(prefix.into());
        self
    }

    /// Sets the local folder path to upload.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::put::S3PutFolderTaskletBuilder;
    ///
    /// let builder = S3PutFolderTaskletBuilder::new().local_folder("./exports/");
    /// ```
    pub fn local_folder<P: AsRef<Path>>(mut self, path: P) -> Self {
        self.local_folder = Some(path.as_ref().to_path_buf());
        self
    }

    /// Sets the AWS region.
    pub fn region<S: Into<String>>(mut self, region: S) -> Self {
        self.config.region = Some(region.into());
        self
    }

    /// Sets a custom endpoint URL for S3-compatible services (MinIO, LocalStack).
    pub fn endpoint_url<S: Into<String>>(mut self, url: S) -> Self {
        self.config.endpoint_url = Some(url.into());
        self
    }

    /// Sets the AWS access key ID for explicit credential configuration.
    pub fn access_key_id<S: Into<String>>(mut self, key_id: S) -> Self {
        self.config.access_key_id = Some(key_id.into());
        self
    }

    /// Sets the AWS secret access key for explicit credential configuration.
    pub fn secret_access_key<S: Into<String>>(mut self, secret: S) -> Self {
        self.config.secret_access_key = Some(secret.into());
        self
    }

    /// Sets the multipart upload chunk size in bytes. Defaults to `8 MiB`.
    pub fn chunk_size(mut self, size: usize) -> Self {
        self.chunk_size = size;
        self
    }

    /// Builds the [`S3PutFolderTasklet`].
    ///
    /// # Errors
    ///
    /// Returns [`BatchError::Configuration`] if `bucket`, `prefix`, or `local_folder` are not set.
    pub fn build(self) -> Result<S3PutFolderTasklet, BatchError> {
        let bucket = self
            .bucket
            .ok_or_else(|| BatchError::Configuration("S3PutFolderTasklet: 'bucket' is required".to_string()))?;
        let prefix = self
            .prefix
            .ok_or_else(|| BatchError::Configuration("S3PutFolderTasklet: 'prefix' is required".to_string()))?;
        let local_folder = self
            .local_folder
            .ok_or_else(|| BatchError::Configuration("S3PutFolderTasklet: 'local_folder' is required".to_string()))?;

        Ok(S3PutFolderTasklet {
            bucket,
            prefix,
            local_folder,
            chunk_size: self.chunk_size,
            config: self.config,
        })
    }
}

// ---------------------------------------------------------------------------
// Shared helpers
// ---------------------------------------------------------------------------

/// Performs a multipart upload of a local file to S3.
///
/// Aborts the upload if any part fails to avoid orphaned S3 parts.
async fn upload_multipart(
    client: &aws_sdk_s3::Client,
    bucket: &str,
    key: &str,
    local_file: &Path,
    chunk_size: usize,
) -> Result<(), BatchError> {
    let create_resp = client
        .create_multipart_upload()
        .bucket(bucket)
        .key(key)
        .send()
        .await
        .map_err(|e| BatchError::ItemWriter(format!("create_multipart_upload failed for {}: {}", key, e)))?;

    let upload_id = create_resp
        .upload_id()
        .ok_or_else(|| BatchError::ItemWriter("create_multipart_upload returned no upload_id".to_string()))?
        .to_string();

    let result = upload_parts(client, bucket, key, &upload_id, local_file, chunk_size).await;

    if let Err(e) = result {
        // Abort to clean up orphaned parts
        let _ = client
            .abort_multipart_upload()
            .bucket(bucket)
            .key(key)
            .upload_id(&upload_id)
            .send()
            .await;
        return Err(e);
    }

    Ok(())
}

/// Uploads all parts and completes the multipart upload.
async fn upload_parts(
    client: &aws_sdk_s3::Client,
    bucket: &str,
    key: &str,
    upload_id: &str,
    local_file: &Path,
    chunk_size: usize,
) -> Result<(), BatchError> {
    use std::io::Read;

    let file = std::fs::File::open(local_file).map_err(BatchError::Io)?;
    let mut reader = std::io::BufReader::new(file);
    let mut part_number = 1i32;
    let mut completed_parts = Vec::new();

    loop {
        let mut buffer = vec![0u8; chunk_size];
        let bytes_read = reader.read(&mut buffer).map_err(BatchError::Io)?;
        if bytes_read == 0 {
            break;
        }
        buffer.truncate(bytes_read);

        debug!("Multipart upload: part {} ({} bytes) -> s3://{}/{}", part_number, bytes_read, bucket, key);

        let body = ByteStream::from(buffer);
        let part_resp = client
            .upload_part()
            .bucket(bucket)
            .key(key)
            .upload_id(upload_id)
            .part_number(part_number)
            .body(body)
            .send()
            .await
            .map_err(|e| BatchError::ItemWriter(format!("upload_part {} failed: {}", part_number, e)))?;

        let etag = part_resp
            .e_tag()
            .ok_or_else(|| BatchError::ItemWriter(format!("upload_part {} returned no ETag", part_number)))?
            .to_string();

        completed_parts.push(
            CompletedPart::builder()
                .part_number(part_number)
                .e_tag(etag)
                .build(),
        );

        part_number += 1;
    }

    let completed = CompletedMultipartUpload::builder()
        .set_parts(Some(completed_parts))
        .build();

    client
        .complete_multipart_upload()
        .bucket(bucket)
        .key(key)
        .upload_id(upload_id)
        .multipart_upload(completed)
        .send()
        .await
        .map_err(|e| BatchError::ItemWriter(format!("complete_multipart_upload failed for {}: {}", key, e)))?;

    Ok(())
}

/// Recursively collects all file paths under a directory.
pub(crate) fn collect_files(dir: &Path) -> Result<Vec<PathBuf>, BatchError> {
    let mut files = Vec::new();
    for entry in std::fs::read_dir(dir).map_err(BatchError::Io)? {
        let entry = entry.map_err(BatchError::Io)?;
        let path = entry.path();
        if path.is_dir() {
            files.extend(collect_files(&path)?);
        } else {
            files.push(path);
        }
    }
    Ok(files)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::env::temp_dir;
    use std::fs;

    // --- S3PutTaskletBuilder tests ---

    #[test]
    fn should_fail_build_when_bucket_missing() {
        let result = S3PutTaskletBuilder::new()
            .key("file.csv")
            .local_file("/tmp/file.csv")
            .build();
        assert!(result.is_err(), "build should fail without bucket");
        assert!(
            result.unwrap_err().to_string().contains("bucket"),
            "error message should mention 'bucket'"
        );
    }

    #[test]
    fn should_fail_build_when_key_missing() {
        let result = S3PutTaskletBuilder::new()
            .bucket("my-bucket")
            .local_file("/tmp/file.csv")
            .build();
        assert!(result.is_err(), "build should fail without key");
        assert!(
            result.unwrap_err().to_string().contains("key"),
            "error message should mention 'key'"
        );
    }

    #[test]
    fn should_fail_build_when_local_file_missing() {
        let result = S3PutTaskletBuilder::new()
            .bucket("my-bucket")
            .key("file.csv")
            .build();
        assert!(result.is_err(), "build should fail without local_file");
        assert!(
            result.unwrap_err().to_string().contains("local_file"),
            "error message should mention 'local_file'"
        );
    }

    #[test]
    fn should_build_with_required_fields() {
        let result = S3PutTaskletBuilder::new()
            .bucket("my-bucket")
            .key("file.csv")
            .local_file("/tmp/file.csv")
            .build();
        assert!(result.is_ok(), "build should succeed with required fields: {:?}", result.err());
    }

    #[test]
    fn should_apply_default_chunk_size() {
        let tasklet = S3PutTaskletBuilder::new()
            .bucket("b")
            .key("k")
            .local_file("/tmp/f")
            .build()
            .unwrap(); // required fields are set — cannot fail
        assert_eq!(tasklet.chunk_size, DEFAULT_CHUNK_SIZE, "default chunk_size should be 8 MiB");
    }

    #[test]
    fn should_override_chunk_size() {
        let tasklet = S3PutTaskletBuilder::new()
            .bucket("b")
            .key("k")
            .local_file("/tmp/f")
            .chunk_size(16 * 1024 * 1024)
            .build()
            .unwrap(); // required fields are set — cannot fail
        assert_eq!(tasklet.chunk_size, 16 * 1024 * 1024);
    }

    #[test]
    fn should_store_optional_config_fields() {
        let tasklet = S3PutTaskletBuilder::new()
            .bucket("b")
            .key("k")
            .local_file("/tmp/f")
            .region("us-east-1")
            .endpoint_url("http://localhost:9000")
            .access_key_id("AKID")
            .secret_access_key("SECRET")
            .build()
            .unwrap(); // required fields are set — cannot fail
        assert_eq!(tasklet.config.region.as_deref(), Some("us-east-1"));
        assert_eq!(tasklet.config.endpoint_url.as_deref(), Some("http://localhost:9000"));
        assert_eq!(tasklet.config.access_key_id.as_deref(), Some("AKID"));
        assert_eq!(tasklet.config.secret_access_key.as_deref(), Some("SECRET"));
    }

    // --- S3PutFolderTaskletBuilder tests ---

    #[test]
    fn should_fail_folder_build_when_bucket_missing() {
        let result = S3PutFolderTaskletBuilder::new()
            .prefix("backups/")
            .local_folder("/tmp/exports")
            .build();
        assert!(result.is_err(), "build should fail without bucket");
        assert!(result.unwrap_err().to_string().contains("bucket"));
    }

    #[test]
    fn should_fail_folder_build_when_prefix_missing() {
        let result = S3PutFolderTaskletBuilder::new()
            .bucket("my-bucket")
            .local_folder("/tmp/exports")
            .build();
        assert!(result.is_err(), "build should fail without prefix");
        assert!(result.unwrap_err().to_string().contains("prefix"));
    }

    #[test]
    fn should_fail_folder_build_when_local_folder_missing() {
        let result = S3PutFolderTaskletBuilder::new()
            .bucket("my-bucket")
            .prefix("backups/")
            .build();
        assert!(result.is_err(), "build should fail without local_folder");
        assert!(result.unwrap_err().to_string().contains("local_folder"));
    }

    #[test]
    fn should_build_folder_with_required_fields() {
        let result = S3PutFolderTaskletBuilder::new()
            .bucket("my-bucket")
            .prefix("backups/")
            .local_folder("/tmp/exports")
            .build();
        assert!(result.is_ok(), "build should succeed with required fields: {:?}", result.err());
    }

    // --- collect_files helper tests ---

    #[test]
    fn should_collect_files_from_directory() {
        let dir = temp_dir().join("spring_batch_rs_test_collect");
        fs::create_dir_all(&dir).unwrap(); // test setup — cannot fail in temp dir
        fs::write(dir.join("a.txt"), "a").unwrap(); // test setup
        fs::write(dir.join("b.txt"), "b").unwrap(); // test setup

        let files = collect_files(&dir).unwrap(); // dir exists — cannot fail
        assert_eq!(files.len(), 2, "should collect 2 files, got: {:?}", files);

        fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn should_return_error_for_missing_directory() {
        let result = collect_files(Path::new("/nonexistent/path/xyz"));
        assert!(result.is_err(), "should return error for missing directory");
    }
}
```

- [ ] **Step 3: Run the tests**

```bash
cargo test --features s3 -q 2>&1 | grep -E "test result|FAILED|error\["
```

Expected output (all pass):
```
test result: ok. N passed; 0 failed; ...
```

- [ ] **Step 4: Verify it compiles without warnings**

```bash
cargo clippy --features s3 -- -D warnings 2>&1 | grep -E "^error"
```

Expected: no output.

- [ ] **Step 5: Commit**

```bash
git add src/tasklet/s3/put.rs
git commit -m "feat(s3): add S3PutTasklet and S3PutFolderTasklet with multipart upload support"
```

---

## Task 5: Implement `S3GetTasklet` and `S3GetFolderTasklet` in `src/tasklet/s3/get.rs`

**Files:**
- Modify: `src/tasklet/s3/get.rs`

The download strategy:
- `get_object` returns a `ByteStream`; collect it and write to a local file.
- For folder downloads, `list_objects_v2` with the prefix enumerates all keys, then each is downloaded.
- `create_dir_all` before writing ensures parent directories exist.

- [ ] **Step 1: Write the full content of `src/tasklet/s3/get.rs`**

```rust
//! S3 GET tasklets for downloading files and folders from Amazon S3.

use crate::{
    core::step::{RepeatStatus, StepExecution, Tasklet},
    tasklet::s3::{build_s3_client, S3ClientConfig},
    BatchError,
};
use log::{debug, info};
use std::path::{Path, PathBuf};
use tokio::runtime::Handle;

/// A tasklet that downloads a single S3 object to a local file.
///
/// The object body is streamed from S3 and written directly to the local file
/// without loading it entirely into memory.
///
/// # Examples
///
/// ```rust,no_run
/// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
///
/// # fn example() -> Result<(), spring_batch_rs::BatchError> {
/// let tasklet = S3GetTaskletBuilder::new()
///     .bucket("my-bucket")
///     .key("imports/file.csv")
///     .local_file("./input/file.csv")
///     .region("eu-west-1")
///     .build()?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns [`BatchError::ItemReader`] if the S3 download fails.
/// Returns [`BatchError::Io`] if the local file cannot be written.
#[derive(Debug)]
pub struct S3GetTasklet {
    bucket: String,
    key: String,
    local_file: PathBuf,
    config: S3ClientConfig,
}

impl S3GetTasklet {
    async fn execute_async(&self) -> Result<RepeatStatus, BatchError> {
        info!(
            "Downloading s3://{}/{} -> {}",
            self.bucket,
            self.key,
            self.local_file.display()
        );

        let client = build_s3_client(&self.config).await?;

        if let Some(parent) = self.local_file.parent() {
            std::fs::create_dir_all(parent).map_err(BatchError::Io)?;
        }

        let resp = client
            .get_object()
            .bucket(&self.bucket)
            .key(&self.key)
            .send()
            .await
            .map_err(|e| BatchError::ItemReader(format!("S3 get_object failed for {}: {}", self.key, e)))?;

        let bytes = resp
            .body
            .collect()
            .await
            .map_err(|e| BatchError::ItemReader(format!("Failed to read S3 body for {}: {}", self.key, e)))?
            .into_bytes();

        std::fs::write(&self.local_file, &bytes).map_err(BatchError::Io)?;

        info!(
            "Download complete: {} bytes written to {}",
            bytes.len(),
            self.local_file.display()
        );
        Ok(RepeatStatus::Finished)
    }
}

impl Tasklet for S3GetTasklet {
    fn execute(&self, _step_execution: &StepExecution) -> Result<RepeatStatus, BatchError> {
        tokio::task::block_in_place(|| Handle::current().block_on(self.execute_async()))
    }
}

/// Builder for [`S3GetTasklet`].
///
/// # Examples
///
/// ```rust,no_run
/// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
///
/// # fn example() -> Result<(), spring_batch_rs::BatchError> {
/// let tasklet = S3GetTaskletBuilder::new()
///     .bucket("my-bucket")
///     .key("imports/file.csv")
///     .local_file("./input/file.csv")
///     .region("eu-west-1")
///     .build()?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns [`BatchError::Configuration`] if `bucket`, `key`, or `local_file` are not set.
#[derive(Debug, Default)]
pub struct S3GetTaskletBuilder {
    bucket: Option<String>,
    key: Option<String>,
    local_file: Option<PathBuf>,
    config: S3ClientConfig,
}

impl S3GetTaskletBuilder {
    /// Creates a new builder with default settings.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
    ///
    /// let builder = S3GetTaskletBuilder::new();
    /// ```
    pub fn new() -> Self {
        Self::default()
    }

    /// Sets the S3 bucket name.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
    ///
    /// let builder = S3GetTaskletBuilder::new().bucket("my-bucket");
    /// ```
    pub fn bucket<S: Into<String>>(mut self, bucket: S) -> Self {
        self.bucket = Some(bucket.into());
        self
    }

    /// Sets the S3 object key (path within the bucket).
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
    ///
    /// let builder = S3GetTaskletBuilder::new().key("imports/file.csv");
    /// ```
    pub fn key<S: Into<String>>(mut self, key: S) -> Self {
        self.key = Some(key.into());
        self
    }

    /// Sets the local file path to write the downloaded object to.
    ///
    /// Parent directories are created automatically during execution.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
    ///
    /// let builder = S3GetTaskletBuilder::new().local_file("./input/file.csv");
    /// ```
    pub fn local_file<P: AsRef<Path>>(mut self, path: P) -> Self {
        self.local_file = Some(path.as_ref().to_path_buf());
        self
    }

    /// Sets the AWS region.
    ///
    /// Falls back to the `AWS_DEFAULT_REGION` environment variable when not set.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
    ///
    /// let builder = S3GetTaskletBuilder::new().region("eu-west-1");
    /// ```
    pub fn region<S: Into<String>>(mut self, region: S) -> Self {
        self.config.region = Some(region.into());
        self
    }

    /// Sets a custom endpoint URL for S3-compatible services (MinIO, LocalStack).
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
    ///
    /// let builder = S3GetTaskletBuilder::new().endpoint_url("http://localhost:9000");
    /// ```
    pub fn endpoint_url<S: Into<String>>(mut self, url: S) -> Self {
        self.config.endpoint_url = Some(url.into());
        self
    }

    /// Sets the AWS access key ID for explicit credential configuration.
    ///
    /// Must be combined with [`secret_access_key`](Self::secret_access_key).
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
    ///
    /// let builder = S3GetTaskletBuilder::new().access_key_id("AKIAIOSFODNN7EXAMPLE");
    /// ```
    pub fn access_key_id<S: Into<String>>(mut self, key_id: S) -> Self {
        self.config.access_key_id = Some(key_id.into());
        self
    }

    /// Sets the AWS secret access key for explicit credential configuration.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
    ///
    /// let builder = S3GetTaskletBuilder::new().secret_access_key("wJalrXUtnFEMI/K7MDENG");
    /// ```
    pub fn secret_access_key<S: Into<String>>(mut self, secret: S) -> Self {
        self.config.secret_access_key = Some(secret.into());
        self
    }

    /// Builds the [`S3GetTasklet`].
    ///
    /// # Errors
    ///
    /// Returns [`BatchError::Configuration`] if `bucket`, `key`, or `local_file` are not set.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use spring_batch_rs::tasklet::s3::get::S3GetTaskletBuilder;
    ///
    /// # fn example() -> Result<(), spring_batch_rs::BatchError> {
    /// let tasklet = S3GetTaskletBuilder::new()
    ///     .bucket("my-bucket")
    ///     .key("file.csv")
    ///     .local_file("./input/file.csv")
    ///     .build()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn build(self) -> Result<S3GetTasklet, BatchError> {
        let bucket = self
            .bucket
            .ok_or_else(|| BatchError::Configuration("S3GetTasklet: 'bucket' is required".to_string()))?;
        let key = self
            .key
            .ok_or_else(|| BatchError::Configuration("S3GetTasklet: 'key' is required".to_string()))?;
        let local_file = self
            .local_file
            .ok_or_else(|| BatchError::Configuration("S3GetTasklet: 'local_file' is required".to_string()))?;

        Ok(S3GetTasklet {
            bucket,
            key,
            local_file,
            config: self.config,
        })
    }
}

// ---------------------------------------------------------------------------
// S3GetFolderTasklet
// ---------------------------------------------------------------------------

/// A tasklet that downloads all S3 objects under a given prefix to a local folder.
///
/// Objects are listed with `list_objects_v2` and downloaded sequentially.
/// Parent directories are created automatically.
///
/// # Examples
///
/// ```rust,no_run
/// use spring_batch_rs::tasklet::s3::get::S3GetFolderTaskletBuilder;
///
/// # fn example() -> Result<(), spring_batch_rs::BatchError> {
/// let tasklet = S3GetFolderTaskletBuilder::new()
///     .bucket("my-bucket")
///     .prefix("backups/2026-04-10/")
///     .local_folder("./imports/")
///     .region("eu-west-1")
///     .build()?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns [`BatchError::ItemReader`] if listing or downloading any object fails.
/// Returns [`BatchError::Io`] if writing any local file fails.
#[derive(Debug)]
pub struct S3GetFolderTasklet {
    bucket: String,
    prefix: String,
    local_folder: PathBuf,
    config: S3ClientConfig,
}

impl S3GetFolderTasklet {
    async fn execute_async(&self) -> Result<RepeatStatus, BatchError> {
        info!(
            "Downloading s3://{}/{} -> {}",
            self.bucket,
            self.prefix,
            self.local_folder.display()
        );

        let client = build_s3_client(&self.config).await?;
        std::fs::create_dir_all(&self.local_folder).map_err(BatchError::Io)?;

        let mut continuation_token: Option<String> = None;
        let mut total_files = 0usize;

        loop {
            let mut req = client
                .list_objects_v2()
                .bucket(&self.bucket)
                .prefix(&self.prefix);

            if let Some(token) = continuation_token {
                req = req.continuation_token(token);
            }

            let list_resp = req
                .send()
                .await
                .map_err(|e| BatchError::ItemReader(format!("list_objects_v2 failed: {}", e)))?;

            for object in list_resp.contents() {
                let key = object.key().unwrap_or_default();
                // Strip prefix to get relative path
                let relative = key.strip_prefix(self.prefix.as_str()).unwrap_or(key);
                if relative.is_empty() {
                    continue; // skip the prefix "directory" object itself
                }
                let local_path = self.local_folder.join(relative);

                if let Some(parent) = local_path.parent() {
                    std::fs::create_dir_all(parent).map_err(BatchError::Io)?;
                }

                debug!("Downloading s3://{}/{} -> {}", self.bucket, key, local_path.display());

                let resp = client
                    .get_object()
                    .bucket(&self.bucket)
                    .key(key)
                    .send()
                    .await
                    .map_err(|e| BatchError::ItemReader(format!("get_object failed for {}: {}", key, e)))?;

                let bytes = resp
                    .body
                    .collect()
                    .await
                    .map_err(|e| BatchError::ItemReader(format!("Failed to read body for {}: {}", key, e)))?
                    .into_bytes();

                std::fs::write(&local_path, &bytes).map_err(BatchError::Io)?;
                total_files += 1;
            }

            if list_resp.is_truncated().unwrap_or(false) {
                continuation_token = list_resp.next_continuation_token().map(str::to_string);
            } else {
                break;
            }
        }

        info!(
            "Folder download complete: {} files downloaded to {}",
            total_files,
            self.local_folder.display()
        );
        Ok(RepeatStatus::Finished)
    }
}

impl Tasklet for S3GetFolderTasklet {
    fn execute(&self, _step_execution: &StepExecution) -> Result<RepeatStatus, BatchError> {
        tokio::task::block_in_place(|| Handle::current().block_on(self.execute_async()))
    }
}

/// Builder for [`S3GetFolderTasklet`].
///
/// # Examples
///
/// ```rust,no_run
/// use spring_batch_rs::tasklet::s3::get::S3GetFolderTaskletBuilder;
///
/// # fn example() -> Result<(), spring_batch_rs::BatchError> {
/// let tasklet = S3GetFolderTaskletBuilder::new()
///     .bucket("my-bucket")
///     .prefix("backups/2026-04-10/")
///     .local_folder("./imports/")
///     .build()?;
/// # Ok(())
/// # }
/// ```
///
/// # Errors
///
/// Returns [`BatchError::Configuration`] if `bucket`, `prefix`, or `local_folder` are not set.
#[derive(Debug, Default)]
pub struct S3GetFolderTaskletBuilder {
    bucket: Option<String>,
    prefix: Option<String>,
    local_folder: Option<PathBuf>,
    config: S3ClientConfig,
}

impl S3GetFolderTaskletBuilder {
    /// Creates a new builder with default settings.
    pub fn new() -> Self {
        Self::default()
    }

    /// Sets the S3 bucket name.
    pub fn bucket<S: Into<String>>(mut self, bucket: S) -> Self {
        self.bucket = Some(bucket.into());
        self
    }

    /// Sets the S3 key prefix to list and download.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetFolderTaskletBuilder;
    ///
    /// let builder = S3GetFolderTaskletBuilder::new().prefix("backups/2026-04-10/");
    /// ```
    pub fn prefix<S: Into<String>>(mut self, prefix: S) -> Self {
        self.prefix = Some(prefix.into());
        self
    }

    /// Sets the local folder path to write downloaded objects to.
    ///
    /// Created automatically if it does not exist.
    ///
    /// # Examples
    ///
    /// ```
    /// use spring_batch_rs::tasklet::s3::get::S3GetFolderTaskletBuilder;
    ///
    /// let builder = S3GetFolderTaskletBuilder::new().local_folder("./imports/");
    /// ```
    pub fn local_folder<P: AsRef<Path>>(mut self, path: P) -> Self {
        self.local_folder = Some(path.as_ref().to_path_buf());
        self
    }

    /// Sets the AWS region.
    pub fn region<S: Into<String>>(mut self, region: S) -> Self {
        self.config.region = Some(region.into());
        self
    }

    /// Sets a custom endpoint URL for S3-compatible services (MinIO, LocalStack).
    pub fn endpoint_url<S: Into<String>>(mut self, url: S) -> Self {
        self.config.endpoint_url = Some(url.into());
        self
    }

    /// Sets the AWS access key ID for explicit credential configuration.
    pub fn access_key_id<S: Into<String>>(mut self, key_id: S) -> Self {
        self.config.access_key_id = Some(key_id.into());
        self
    }

    /// Sets the AWS secret access key for explicit credential configuration.
    pub fn secret_access_key<S: Into<String>>(mut self, secret: S) -> Self {
        self.config.secret_access_key = Some(secret.into());
        self
    }

    /// Builds the [`S3GetFolderTasklet`].
    ///
    /// # Errors
    ///
    /// Returns [`BatchError::Configuration`] if `bucket`, `prefix`, or `local_folder` are not set.
    pub fn build(self) -> Result<S3GetFolderTasklet, BatchError> {
        let bucket = self
            .bucket
            .ok_or_else(|| BatchError::Configuration("S3GetFolderTasklet: 'bucket' is required".to_string()))?;
        let prefix = self
            .prefix
            .ok_or_else(|| BatchError::Configuration("S3GetFolderTasklet: 'prefix' is required".to_string()))?;
        let local_folder = self
            .local_folder
            .ok_or_else(|| BatchError::Configuration("S3GetFolderTasklet: 'local_folder' is required".to_string()))?;

        Ok(S3GetFolderTasklet {
            bucket,
            prefix,
            local_folder,
            config: self.config,
        })
    }
}

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

    // --- S3GetTaskletBuilder tests ---

    #[test]
    fn should_fail_build_when_bucket_missing() {
        let result = S3GetTaskletBuilder::new()
            .key("file.csv")
            .local_file("/tmp/file.csv")
            .build();
        assert!(result.is_err(), "build should fail without bucket");
        assert!(result.unwrap_err().to_string().contains("bucket"));
    }

    #[test]
    fn should_fail_build_when_key_missing() {
        let result = S3GetTaskletBuilder::new()
            .bucket("my-bucket")
            .local_file("/tmp/file.csv")
            .build();
        assert!(result.is_err(), "build should fail without key");
        assert!(result.unwrap_err().to_string().contains("key"));
    }

    #[test]
    fn should_fail_build_when_local_file_missing() {
        let result = S3GetTaskletBuilder::new()
            .bucket("my-bucket")
            .key("file.csv")
            .build();
        assert!(result.is_err(), "build should fail without local_file");
        assert!(result.unwrap_err().to_string().contains("local_file"));
    }

    #[test]
    fn should_build_with_required_fields() {
        let result = S3GetTaskletBuilder::new()
            .bucket("my-bucket")
            .key("file.csv")
            .local_file("/tmp/file.csv")
            .build();
        assert!(result.is_ok(), "build should succeed with required fields: {:?}", result.err());
    }

    #[test]
    fn should_store_optional_config_fields() {
        let tasklet = S3GetTaskletBuilder::new()
            .bucket("b")
            .key("k")
            .local_file("/tmp/f")
            .region("eu-west-1")
            .endpoint_url("http://localhost:9000")
            .access_key_id("AKID")
            .secret_access_key("SECRET")
            .build()
            .unwrap(); // required fields set — cannot fail
        assert_eq!(tasklet.config.region.as_deref(), Some("eu-west-1"));
        assert_eq!(tasklet.config.endpoint_url.as_deref(), Some("http://localhost:9000"));
    }

    // --- S3GetFolderTaskletBuilder tests ---

    #[test]
    fn should_fail_folder_build_when_bucket_missing() {
        let result = S3GetFolderTaskletBuilder::new()
            .prefix("backups/")
            .local_folder("/tmp/imports")
            .build();
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("bucket"));
    }

    #[test]
    fn should_fail_folder_build_when_prefix_missing() {
        let result = S3GetFolderTaskletBuilder::new()
            .bucket("my-bucket")
            .local_folder("/tmp/imports")
            .build();
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("prefix"));
    }

    #[test]
    fn should_fail_folder_build_when_local_folder_missing() {
        let result = S3GetFolderTaskletBuilder::new()
            .bucket("my-bucket")
            .prefix("backups/")
            .build();
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("local_folder"));
    }

    #[test]
    fn should_build_folder_with_required_fields() {
        let result = S3GetFolderTaskletBuilder::new()
            .bucket("my-bucket")
            .prefix("backups/")
            .local_folder("/tmp/imports")
            .build();
        assert!(result.is_ok(), "build should succeed: {:?}", result.err());
    }
}
```

- [ ] **Step 2: Run the tests**

```bash
cargo test --features s3 -q 2>&1 | grep -E "test result|FAILED|error\["
```

Expected:
```
test result: ok. N passed; 0 failed; ...
```

- [ ] **Step 3: Verify clippy is clean**

```bash
cargo clippy --features s3 -- -D warnings 2>&1 | grep "^error"
```

Expected: no output.

- [ ] **Step 4: Commit**

```bash
git add src/tasklet/s3/get.rs
git commit -m "feat(s3): add S3GetTasklet and S3GetFolderTasklet with streaming download support"
```

---

## Task 6: Create `examples/tasklet_s3.rs`

**Files:**
- Create: `examples/tasklet_s3.rs`

- [ ] **Step 1: Create the example file**

```rust
//! # Example: S3 File Transfer with Tasklets
//!
//! Demonstrates uploading a local file to S3 and downloading it back,
//! using `S3PutTasklet` and `S3GetTasklet` as batch steps.
//!
//! ## Run (requires a running S3-compatible service)
//!
//! ```bash
//! # Start LocalStack
//! docker run --rm -p 4566:4566 localstack/localstack
//!
//! # Create a bucket (requires awscli or localstack tools)
//! aws --endpoint-url=http://localhost:4566 s3 mb s3://my-batch-bucket
//!
//! # Run the example
//! cargo run --example tasklet_s3 --features s3
//! ```
//!
//! ## What It Does
//!
//! 1. Writes a sample CSV file to a temp directory
//! 2. Uploads it to `s3://my-batch-bucket/exports/sample.csv` via `S3PutTasklet`
//! 3. Downloads it back to a different temp path via `S3GetTasklet`
//! 4. Prints job execution status

use spring_batch_rs::{
    core::{
        job::JobBuilder,
        step::{Step, StepBuilder, StepExecution},
    },
    tasklet::s3::{get::S3GetTaskletBuilder, put::S3PutTaskletBuilder},
};
use std::env::temp_dir;
use std::fs;

#[tokio::main]
async fn main() {
    // 1. Prepare a sample file to upload
    let upload_path = temp_dir().join("spring_batch_s3_sample.csv");
    fs::write(&upload_path, "id,name\n1,Alice\n2,Bob\n").unwrap(); // example setup — panics on error
    let download_path = temp_dir().join("spring_batch_s3_downloaded.csv");

    // 2. Build the upload tasklet (S3PutTasklet)
    let put_tasklet = S3PutTaskletBuilder::new()
        .bucket("my-batch-bucket")
        .key("exports/sample.csv")
        .local_file(&upload_path)
        .endpoint_url("http://localhost:4566") // LocalStack endpoint
        .access_key_id("test")
        .secret_access_key("test")
        .region("us-east-1")
        .build()
        .unwrap(); // panics on misconfiguration — intentional in examples

    // 3. Build the download tasklet (S3GetTasklet)
    let get_tasklet = S3GetTaskletBuilder::new()
        .bucket("my-batch-bucket")
        .key("exports/sample.csv")
        .local_file(&download_path)
        .endpoint_url("http://localhost:4566")
        .access_key_id("test")
        .secret_access_key("test")
        .region("us-east-1")
        .build()
        .unwrap(); // panics on misconfiguration — intentional in examples

    // 4. Build steps
    let upload_step = StepBuilder::new("s3-upload")
        .tasklet(&put_tasklet)
        .build();

    let download_step = StepBuilder::new("s3-download")
        .tasklet(&get_tasklet)
        .build();

    // 5. Build and run the job
    let job = JobBuilder::new()
        .start(&upload_step)
        .next(&download_step)
        .build();

    let result = job.run();

    println!("Job status: {:?}", result.status);
    println!("Upload step:   {:?}", result.step_executions[0].status);
    println!("Download step: {:?}", result.step_executions[1].status);

    // Cleanup
    fs::remove_file(&upload_path).ok();
    fs::remove_file(&download_path).ok();
}
```

- [ ] **Step 2: Verify the example compiles**

```bash
cargo build --example tasklet_s3 --features s3 2>&1 | grep "^error"
```

Expected: no errors.

- [ ] **Step 3: Commit**

```bash
git add examples/tasklet_s3.rs
git commit -m "feat(s3): add tasklet_s3 example demonstrating upload and download with LocalStack"
```

---

## Task 7: Final verification

- [ ] **Step 1: Run all tests with all features**

```bash
cargo test --all-features -q 2>&1 | grep -E "test result|FAILED"
```

Expected:
```
test result: ok. N passed; 0 failed; ...
```

- [ ] **Step 2: Run clippy with zero-warnings policy**

```bash
cargo clippy --all-features -- -D warnings 2>&1 | grep "^error"
```

Expected: no output.

- [ ] **Step 3: Check rustdoc builds without warnings**

```bash
cargo doc --no-deps --all-features 2>&1 | grep -E "warning|error"
```

Expected: no warnings or errors.

- [ ] **Step 4: Check doc-tests pass**

```bash
cargo test --doc --all-features 2>&1 | grep -E "test result|FAILED"
```

Expected: all pass (doc-tests marked `no_run` are skipped, not run against S3).

- [ ] **Step 5: Verify feature isolation compiles**

```bash
cargo build --features s3 2>&1 | grep "^error"
cargo build 2>&1 | grep "^error"
```

Expected: both compile without errors (no accidental leakage of s3 types into default build).

- [ ] **Step 6: Final commit**

```bash
git add -p  # stage any unstaged changes
git commit -m "feat(s3): complete S3 tasklet implementation with put/get single file and folder support"
```

---

## Self-Review Against Spec

| Spec requirement | Task covering it |
|---|---|
| `S3PutTasklet` + builder | Task 4 |
| `S3GetTasklet` + builder | Task 5 |
| `S3PutFolderTasklet` + builder | Task 4 |
| `S3GetFolderTasklet` + builder | Task 5 |
| `aws-sdk-s3` v1 dependency | Task 1 |
| `s3` feature flag | Task 1 |
| `full` feature includes `s3` | Task 1 |
| `S3ClientConfig` shared struct | Task 2 |
| `build_s3_client()` helper | Task 2 |
| Explicit credentials support | Tasks 4, 5 (builder methods) |
| AWS default credential chain | Task 2 (`build_s3_client`) |
| Custom `endpoint_url` (MinIO/LocalStack) | Tasks 4, 5 |
| Multipart upload with configurable `chunk_size` | Task 4 (`upload_multipart`) |
| `abort_multipart_upload` on failure | Task 4 (`upload_multipart`) |
| Streaming download | Task 5 (`ByteStream::collect`) |
| `create_dir_all` for missing parent dirs | Task 5 |
| `list_objects_v2` + pagination for folder download | Task 5 |
| `BatchError` variants — no new variants | Tasks 4, 5 |
| `log` macros, no `println!` | Tasks 4, 5 |
| Unit tests (builder config only) | Tasks 4, 5 |
| Example with LocalStack | Task 6 |
| `[[example]]` in Cargo.toml | Task 1 |
| Module declaration in `tasklet/mod.rs` | Task 3 |
