transfer_family_cli 0.1.0

//! Abstraction over AWS Transfer Family command operations.

use crate::error::{Error, Result};
use crate::listing::DirectoryListing;
use crate::retry;
use crate::transfer_storage::TransferStorage;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};

/// Status of a single file transfer result (from `list_file_transfer_results`).
#[non_exhaustive]
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct TransferResultStatus {
    /// Status code string, e.g. "COMPLETED", "SUCCESS", "FAILED", "ERROR".
    pub status_code: String,
    /// Failure message when status is FAILED or ERROR.
    pub failure_message: Option<String>,
}

/// Result of `start_directory_listing`: IDs so the caller can poll storage for the output file.
#[non_exhaustive]
#[derive(Clone, Debug)]
pub struct DirectoryListingStarted {
    pub listing_id: String,
    pub output_file_name: String,
}

/// Abstraction over AWS Transfer Family: start transfers, start directory listing, poll results.
#[allow(clippy::module_name_repetitions)]
pub trait TransferCommands: Send + Sync {
    /// Starts a retrieve (get) transfer; returns `transfer_id`.
    fn start_file_transfer_retrieve(
        &self,
        connector_id: &str,
        remote_file_path: &str,
        local_directory_path: &str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String>> + Send + '_>>;

    /// Starts a send (put) transfer; returns `transfer_id`.
    fn start_file_transfer_send(
        &self,
        connector_id: &str,
        send_path: &str,
        remote_directory_path: &str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String>> + Send + '_>>;

    /// Returns current status(es) for the transfer (empty if still in progress).
    fn list_file_transfer_results(
        &self,
        connector_id: &str,
        transfer_id: &str,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<Vec<TransferResultStatus>>> + Send + '_>,
    >;

    /// Starts a directory listing; caller polls storage for output at (`output_directory_path` + `output_file_name`).
    fn start_directory_listing(
        &self,
        connector_id: &str,
        remote_directory_path: &str,
        output_directory_path: &str,
        max_items: Option<i32>,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<DirectoryListingStarted>> + Send + '_>,
    >;
}

// -----------------------------------------------------------------------------
// AWS implementation
// -----------------------------------------------------------------------------

/// AWS Transfer Family client wrapper.
#[derive(Clone)]
pub struct AwsTransferCommands {
    client: aws_sdk_transfer::Client,
}

impl AwsTransferCommands {
    #[must_use]
    pub const fn new(client: aws_sdk_transfer::Client) -> Self {
        Self { client }
    }
}

impl TransferCommands for AwsTransferCommands {
    fn start_file_transfer_retrieve(
        &self,
        connector_id: &str,
        remote_file_path: &str,
        local_directory_path: &str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String>> + Send + '_>> {
        let client = self.client.clone();
        let connector_id = connector_id.to_string();
        let remote_file_path = remote_file_path.to_string();
        let local_directory_path = local_directory_path.to_string();
        Box::pin(async move {
            retry::with_retry_and_timeout(|| {
                let client = client.clone();
                let connector_id = connector_id.clone();
                let remote_file_path = remote_file_path.clone();
                let local_directory_path = local_directory_path.clone();
                async move {
                    let output = client
                        .start_file_transfer()
                        .connector_id(&connector_id)
                        .retrieve_file_paths(&remote_file_path)
                        .local_directory_path(&local_directory_path)
                        .send()
                        .await
                        .map_err(|e| {
                            let status = e.raw_response().map(|r| r.status().as_u16());
                            let err = match status {
                                Some(429) | Some(500..=599) => Error::api(e.to_string()),
                                _ => Error::api_permanent(e.to_string()),
                            };
                            err.with("remote_file_path", &remote_file_path)
                                .with("connector_id", &connector_id)
                        })?;
                    let transfer_id = output.transfer_id().to_string();
                    if transfer_id.is_empty() {
                        return Err(Error::invalid_input("Missing TransferId")
                            .with("remote_file_path", &remote_file_path));
                    }
                    Ok(transfer_id)
                }
            })
            .await
        })
    }

    fn start_file_transfer_send(
        &self,
        connector_id: &str,
        send_path: &str,
        remote_directory_path: &str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String>> + Send + '_>> {
        let client = self.client.clone();
        let connector_id = connector_id.to_string();
        let send_path = send_path.to_string();
        let remote_directory_path = remote_directory_path.to_string();
        Box::pin(async move {
            retry::with_retry_and_timeout(|| {
                let client = client.clone();
                let connector_id = connector_id.clone();
                let send_path = send_path.clone();
                let remote_directory_path = remote_directory_path.clone();
                async move {
                    let output = client
                        .start_file_transfer()
                        .connector_id(&connector_id)
                        .send_file_paths(&send_path)
                        .remote_directory_path(&remote_directory_path)
                        .send()
                        .await
                        .map_err(|e| {
                            let status = e.raw_response().map(|r| r.status().as_u16());
                            let err = match status {
                                Some(429) | Some(500..=599) => Error::api(e.to_string()),
                                _ => Error::api_permanent(e.to_string()),
                            };
                            err.with("connector_id", &connector_id)
                                .with("send_path", &send_path)
                        })?;
                    let transfer_id = output.transfer_id().to_string();
                    if transfer_id.is_empty() {
                        return Err(Error::invalid_input("Missing TransferId")
                            .with("remote_directory_path", &remote_directory_path));
                    }
                    Ok(transfer_id)
                }
            })
            .await
        })
    }

    fn list_file_transfer_results(
        &self,
        connector_id: &str,
        transfer_id: &str,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<Vec<TransferResultStatus>>> + Send + '_>,
    > {
        let client = self.client.clone();
        let connector_id = connector_id.to_string();
        let transfer_id = transfer_id.to_string();
        Box::pin(async move {
            retry::with_retry_and_timeout(|| {
                let client = client.clone();
                let connector_id = connector_id.clone();
                let transfer_id = transfer_id.clone();
                async move {
                    let resp = client
                        .list_file_transfer_results()
                        .connector_id(&connector_id)
                        .transfer_id(&transfer_id)
                        .send()
                        .await
                        .map_err(|e| {
                            let status = e.raw_response().map(|r| r.status().as_u16());
                            let err = match status {
                                Some(429) | Some(500..=599) => Error::api(e.to_string()),
                                _ => Error::api_permanent(e.to_string()),
                            };
                            err.with("transfer_id", &transfer_id)
                        })?;
                    let results = resp
                        .file_transfer_results()
                        .iter()
                        .map(|r| TransferResultStatus {
                            status_code: r.status_code().as_str().to_string(),
                            failure_message: r.failure_message().map(String::from),
                        })
                        .collect();
                    Ok(results)
                }
            })
            .await
        })
    }

    fn start_directory_listing(
        &self,
        connector_id: &str,
        remote_directory_path: &str,
        output_directory_path: &str,
        max_items: Option<i32>,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<DirectoryListingStarted>> + Send + '_>,
    > {
        let client = self.client.clone();
        let connector_id = connector_id.to_string();
        let remote_directory_path = remote_directory_path.to_string();
        let output_directory_path = output_directory_path.to_string();
        Box::pin(async move {
            retry::with_retry_and_timeout(|| {
                let client = client.clone();
                let connector_id = connector_id.clone();
                let remote_directory_path = remote_directory_path.clone();
                let output_directory_path = output_directory_path.clone();
                async move {
                    let output = client
                        .start_directory_listing()
                        .connector_id(&connector_id)
                        .remote_directory_path(&remote_directory_path)
                        .output_directory_path(&output_directory_path)
                        .set_max_items(max_items)
                        .send()
                        .await
                        .map_err(|e| {
                            let status = e.raw_response().map(|r| r.status().as_u16());
                            let err = match status {
                                Some(429) | Some(500..=599) => Error::api(e.to_string()),
                                _ => Error::api_permanent(e.to_string()),
                            };
                            err.with("remote_directory_path", &remote_directory_path)
                                .with("connector_id", &connector_id)
                        })?;
                    let listing_id = output.listing_id().to_string();
                    let output_file_name = output.output_file_name().to_string();
                    if listing_id.is_empty() {
                        return Err(Error::invalid_input(
                            "Missing ListingId in StartDirectoryListing response",
                        )
                        .with("remote_directory_path", &remote_directory_path));
                    }
                    if output_file_name.is_empty() {
                        return Err(Error::invalid_input(
                            "Missing OutputFileName in StartDirectoryListing response",
                        )
                        .with("remote_directory_path", &remote_directory_path));
                    }
                    Ok(DirectoryListingStarted {
                        listing_id,
                        output_file_name,
                    })
                }
            })
            .await
        })
    }
}

// -----------------------------------------------------------------------------
// In-memory implementation (for tests)
// -----------------------------------------------------------------------------

/// In-memory transfer commands that coordinate with `MemoryTransferStorage`:
/// writes listing/retrieve data to storage so `list_directory` and `get_file` succeed.
pub struct MemoryTransferCommands {
    storage: std::sync::Arc<crate::transfer_storage::MemoryTransferStorage>,
    transfer_counter: AtomicU64,
    listing_counter: AtomicU64,
    /// Default listing written when `start_directory_listing` is called.
    default_listing: DirectoryListing,
    /// When set, write these bytes instead of serializing `default_listing` (for tests).
    #[allow(dead_code)]
    listing_body_override: Option<Vec<u8>>,
    /// When true, return empty `listing_id`/`output_file_name` (for tests).
    #[allow(dead_code)]
    empty_listing_response: bool,
}

impl MemoryTransferCommands {
    #[allow(clippy::missing_const_for_fn)] // DirectoryListing uses vec![]; not const in stable
    #[must_use]
    pub fn new(storage: std::sync::Arc<crate::transfer_storage::MemoryTransferStorage>) -> Self {
        Self {
            storage,
            transfer_counter: AtomicU64::new(0),
            listing_counter: AtomicU64::new(0),
            default_listing: DirectoryListing {
                files: vec![],
                paths: vec![],
                truncated: false,
            },
            listing_body_override: None,
            empty_listing_response: false,
        }
    }

    /// Sets the default listing JSON written to storage on `start_directory_listing`.
    #[must_use]
    pub fn with_default_listing(mut self, listing: DirectoryListing) -> Self {
        self.default_listing = listing;
        self
    }

    /// Writes the given bytes as the listing result instead of serializing `default_listing` (for tests).
    #[cfg(test)]
    #[must_use]
    pub fn with_listing_body(mut self, body: Vec<u8>) -> Self {
        self.listing_body_override = Some(body);
        self
    }

    /// When set, `start_directory_listing` returns empty `listing_id` and `output_file_name` (for tests).
    #[cfg(test)]
    #[must_use]
    pub const fn with_empty_listing_response(mut self, empty: bool) -> Self {
        self.empty_listing_response = empty;
        self
    }
}

impl Clone for MemoryTransferCommands {
    fn clone(&self) -> Self {
        Self {
            storage: std::sync::Arc::clone(&self.storage),
            transfer_counter: AtomicU64::new(self.transfer_counter.load(Ordering::SeqCst)),
            listing_counter: AtomicU64::new(self.listing_counter.load(Ordering::SeqCst)),
            default_listing: self.default_listing.clone(),
            listing_body_override: self.listing_body_override.clone(),
            empty_listing_response: self.empty_listing_response,
        }
    }
}

impl TransferCommands for MemoryTransferCommands {
    fn start_file_transfer_retrieve(
        &self,
        _connector_id: &str,
        remote_file_path: &str,
        local_directory_path: &str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String>> + Send + '_>> {
        let storage = std::sync::Arc::clone(&self.storage);
        let transfer_id = format!(
            "mem-{}",
            self.transfer_counter.fetch_add(1, Ordering::SeqCst)
        );
        let (bucket, key_prefix) = crate::transfer_storage::split_s3_path(local_directory_path);
        let base_name = Path::new(remote_file_path)
            .file_name()
            .and_then(|p| p.to_str())
            .unwrap_or("file");
        let s3_key = format!("{key_prefix}{base_name}");
        let bucket = bucket.to_string();
        Box::pin(async move {
            // Simulate connector writing the retrieved file to staging.
            storage
                .put_object(&bucket, &s3_key, b"mock retrieved content".to_vec())
                .await?;
            Ok(transfer_id)
        })
    }

    fn start_file_transfer_send(
        &self,
        _connector_id: &str,
        _send_path: &str,
        _remote_directory_path: &str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String>> + Send + '_>> {
        let transfer_id = format!(
            "mem-send-{}",
            self.transfer_counter.fetch_add(1, Ordering::SeqCst)
        );
        Box::pin(async move { Ok(transfer_id) })
    }

    fn list_file_transfer_results(
        &self,
        _connector_id: &str,
        _transfer_id: &str,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<Vec<TransferResultStatus>>> + Send + '_>,
    > {
        Box::pin(async move {
            Ok(vec![TransferResultStatus {
                status_code: "COMPLETED".to_string(),
                failure_message: None,
            }])
        })
    }

    fn start_directory_listing(
        &self,
        _connector_id: &str,
        _remote_directory_path: &str,
        output_directory_path: &str,
        _max_items: Option<i32>,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<DirectoryListingStarted>> + Send + '_>,
    > {
        let storage = std::sync::Arc::clone(&self.storage);
        let listing_id = format!(
            "mem-listing-{}",
            self.listing_counter.fetch_add(1, Ordering::SeqCst)
        );
        let output_file_name = format!("listing-{}.json", listing_id);
        let (bucket, key_prefix) = crate::transfer_storage::split_s3_path(output_directory_path);
        let key = if key_prefix.is_empty() {
            output_file_name.clone()
        } else {
            format!("{key_prefix}/{output_file_name}")
        };
        let bucket = bucket.to_string();
        let default_listing = self.default_listing.clone();
        let listing_body_override = self.listing_body_override.clone();
        let empty_listing_response = self.empty_listing_response;
        Box::pin(async move {
            if empty_listing_response {
                return Ok(DirectoryListingStarted {
                    listing_id: String::new(),
                    output_file_name: String::new(),
                });
            }
            let json_bytes = match listing_body_override {
                Some(bytes) => bytes,
                None => serde_json::to_vec(&default_listing)
                    .map_err(|e| Error::parse("listing serialize", e))?,
            };
            storage.put_object(&bucket, &key, json_bytes).await?;
            Ok(DirectoryListingStarted {
                listing_id,
                output_file_name,
            })
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::listing::{DirectoryListing, ListedFile, ListedPath};
    use crate::transfer_storage::MemoryTransferStorage;
    use std::path::PathBuf;
    use std::sync::Arc;

    fn test_config() -> crate::config::Config {
        crate::config::Config {
            connector_id: "c-test".to_string(),
            s3_root: "bucket/transfer-cli/".to_string(),
            region: None,
            profile: None,
            download_dir: PathBuf::from("/mem"),
        }
    }

    #[tokio::test]
    async fn start_directory_listing_writes_to_storage() {
        let storage = Arc::new(MemoryTransferStorage::new());
        let transfer = MemoryTransferCommands::new(Arc::clone(&storage));
        let config = test_config();
        let output_path = config.listings_prefix();

        let started = transfer
            .start_directory_listing(&config.connector_id, "/remote", &output_path, Some(100))
            .await
            .unwrap();

        assert!(!started.listing_id.is_empty());
        assert!(!started.output_file_name.is_empty());
        assert!(started.output_file_name.ends_with(".json"));

        let (bucket, key_prefix) = crate::transfer_storage::split_s3_path(&output_path);
        let key = if key_prefix.is_empty() {
            started.output_file_name.clone()
        } else {
            format!("{key_prefix}/{}", started.output_file_name)
        };
        let bytes = storage.get_object(bucket, &key).await.unwrap();
        let listing: DirectoryListing = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(listing.files.len(), 0);
        assert_eq!(listing.paths.len(), 0);
        assert!(!listing.truncated);
    }

    #[tokio::test]
    async fn with_default_listing_stores_custom_listing() {
        let storage = Arc::new(MemoryTransferStorage::new());
        let custom_listing = DirectoryListing {
            files: vec![ListedFile {
                file_path: "/remote/custom.txt".to_string(),
                modified_timestamp: None,
                size: Some(99),
            }],
            paths: vec![ListedPath {
                path: "/remote/sub".to_string(),
            }],
            truncated: true,
        };
        let transfer = MemoryTransferCommands::new(Arc::clone(&storage))
            .with_default_listing(custom_listing.clone());
        let config = test_config();
        let output_path = config.listings_prefix();

        let started = transfer
            .start_directory_listing(&config.connector_id, "/remote", &output_path, Some(50))
            .await
            .unwrap();

        let (bucket, key_prefix) = crate::transfer_storage::split_s3_path(&output_path);
        let key = if key_prefix.is_empty() {
            started.output_file_name
        } else {
            format!("{key_prefix}/{}", started.output_file_name)
        };
        let bytes = storage.get_object(bucket, &key).await.unwrap();
        let listing: DirectoryListing = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(listing.files.len(), 1);
        assert_eq!(
            listing.files.first().unwrap().file_path,
            "/remote/custom.txt"
        );
        assert_eq!(listing.files.first().unwrap().size, Some(99));
        assert_eq!(listing.paths.len(), 1);
        assert_eq!(listing.paths.first().unwrap().path, "/remote/sub");
        assert!(listing.truncated);
    }

    #[tokio::test]
    async fn start_file_transfer_retrieve_returns_id_and_writes_storage() {
        let storage = Arc::new(MemoryTransferStorage::new());
        let transfer = MemoryTransferCommands::new(Arc::clone(&storage));
        let config = test_config();
        let local_directory_path = config.retrieve_prefix();

        let transfer_id = transfer
            .start_file_transfer_retrieve(
                &config.connector_id,
                "/remote/data.bin",
                &local_directory_path,
            )
            .await
            .unwrap();

        assert!(!transfer_id.is_empty());
        assert!(transfer_id.starts_with("mem-"));

        let (bucket, key_prefix) = crate::transfer_storage::split_s3_path(&local_directory_path);
        let key = format!("{key_prefix}data.bin");
        let bytes = storage.get_object(bucket, &key).await.unwrap();
        assert_eq!(bytes, b"mock retrieved content");
    }

    #[tokio::test]
    async fn start_file_transfer_send_returns_non_empty_id() {
        let storage = Arc::new(MemoryTransferStorage::new());
        let transfer = MemoryTransferCommands::new(Arc::clone(&storage));
        let config = test_config();

        let transfer_id = transfer
            .start_file_transfer_send(&config.connector_id, "bucket/send/xyz", "/remote")
            .await
            .unwrap();

        assert!(!transfer_id.is_empty());
        assert!(transfer_id.starts_with("mem-send-"));
    }

    #[tokio::test]
    async fn list_file_transfer_results_returns_completed() {
        let storage = Arc::new(MemoryTransferStorage::new());
        let transfer = MemoryTransferCommands::new(Arc::clone(&storage));

        let results = transfer
            .list_file_transfer_results("conn", "mem-0")
            .await
            .unwrap();

        assert_eq!(results.len(), 1);
        assert_eq!(results.first().unwrap().status_code, "COMPLETED");
        assert!(results.first().unwrap().failure_message.is_none());
    }

    #[tokio::test]
    async fn multiple_listing_calls_different_ids_and_keys() {
        let storage = Arc::new(MemoryTransferStorage::new());
        let transfer = MemoryTransferCommands::new(Arc::clone(&storage));
        let config = test_config();
        let output_path = config.listings_prefix();

        let first = transfer
            .start_directory_listing(&config.connector_id, "/remote", &output_path, None)
            .await
            .unwrap();
        let second = transfer
            .start_directory_listing(&config.connector_id, "/remote", &output_path, None)
            .await
            .unwrap();

        assert_ne!(first.listing_id, second.listing_id);
        assert_ne!(first.output_file_name, second.output_file_name);

        let keys = storage.test_keys();
        assert_eq!(keys.len(), 2);
        assert_ne!(keys.first().unwrap(), keys.get(1).unwrap());
    }
}