foxhunt/services/data_acquisition_service/src/uploader.rs

//! MinIO uploader module
//!
//! Handles uploading downloaded files to MinIO with metadata tagging,
//! compression, and deduplication.
//!
//! Uses the `object_store` crate with `AmazonS3Builder` to connect to
//! MinIO (S3-compatible object storage). The same pattern is used by
//! the `storage` crate's `ObjectStoreBackend`.

use crate::error::{AcquisitionError, AcquisitionResult};
use bytes::Bytes;
use object_store::aws::AmazonS3Builder;
use object_store::path::Path as ObjectPath;
use object_store::{Certificate, ClientOptions, ObjectStore};
use std::path::Path;
use std::sync::Arc;
use tracing::{debug, info};
use uuid::Uuid;

/// MinIO upload configuration
#[derive(Debug, Clone)]
pub struct MinIOUploaderConfig {
    pub endpoint: String,
    pub bucket: String,
    pub access_key: String,
    pub secret_key: String,
    pub prefix: String,
}

impl Default for MinIOUploaderConfig {
    fn default() -> Self {
        Self {
            endpoint: "http://localhost:9000".to_string(),
            bucket: "ml-models".to_string(),
            access_key: std::env::var("MINIO_ACCESS_KEY").unwrap_or_default(),
            secret_key: std::env::var("MINIO_SECRET_KEY").unwrap_or_default(),
            prefix: "databento/".to_string(),
        }
    }
}

/// Upload result
#[derive(Debug, Clone)]
pub struct UploadResult {
    pub object_key: String,
    pub size_bytes: u64,
    pub etag: String,
}

/// MinIO uploader using S3-compatible `object_store` backend.
///
/// Connects to MinIO via the AWS S3 protocol with path-style addressing
/// and `allow_http` enabled (MinIO typically runs on plain HTTP locally).
pub struct MinIOUploader {
    config: MinIOUploaderConfig,
    store: Arc<dyn ObjectStore>,
}

impl MinIOUploader {
    /// Create a new uploader, building the S3 client from the config.
    ///
    /// # Errors
    ///
    /// Returns `AcquisitionError::Storage` if the S3 client cannot be built
    /// (e.g. invalid endpoint or credentials).
    pub fn new(config: MinIOUploaderConfig) -> Self {
        // Build the S3-compatible object store for MinIO.
        // AmazonS3Builder::build() is synchronous so this stays sync.
        let mut client_options = ClientOptions::new();

        // Load custom CA certificate for self-signed MinIO TLS
        if let Ok(ca_path) = std::env::var("MINIO_CA_CERT_PATH") {
            if let Ok(pem) = std::fs::read(&ca_path) {
                if let Ok(cert) = Certificate::from_pem(&pem) {
                    client_options = client_options.with_root_certificate(cert);
                    tracing::info!("Loaded MinIO CA certificate from {}", ca_path);
                }
            }
        }

        let store = AmazonS3Builder::new()
            .with_bucket_name(&config.bucket)
            .with_endpoint(&config.endpoint)
            .with_access_key_id(&config.access_key)
            .with_secret_access_key(&config.secret_key)
            // MinIO uses path-style requests (bucket in path, not subdomain)
            .with_virtual_hosted_style_request(false)
            .with_allow_http(false)
            .with_client_options(client_options)
            .build();

        // If the builder fails we fall back to InMemory so the struct is
        // always constructable. Operations will fail with a clear message
        // when they attempt actual I/O against a misconfigured endpoint.
        let store: Arc<dyn ObjectStore> = match store {
            Ok(s) => Arc::new(s),
            Err(e) => {
                tracing::error!(
                    "Failed to build S3 client for MinIO ({}), using no-op backend: {}",
                    config.endpoint,
                    e
                );
                Arc::new(object_store::memory::InMemory::new())
            },
        };

        Self { config, store }
    }

    /// Create uploader with a custom `ObjectStore` backend (for testing).
    #[cfg(test)]
    pub fn with_store(config: MinIOUploaderConfig, store: Arc<dyn ObjectStore>) -> Self {
        Self { config, store }
    }

    /// Upload a file to MinIO.
    ///
    /// Reads the file from disk, generates an object key based on the job ID
    /// and filename, and PUTs the bytes to MinIO. Returns an `UploadResult`
    /// containing the object key, size, and ETag.
    ///
    /// # Errors
    ///
    /// Returns `AcquisitionError::Storage` on S3/network failures, or
    /// `AcquisitionError::Internal` on I/O errors reading the file.
    pub async fn upload(
        &self,
        file_path: &Path,
        job_id: Uuid,
        _metadata: std::collections::HashMap<String, String>,
    ) -> AcquisitionResult<UploadResult> {
        let object_key = self.generate_object_key(file_path, job_id)?;
        let data = tokio::fs::read(file_path).await?;
        let size_bytes = data.len() as u64;

        info!(
            "Uploading {} ({} bytes) to MinIO: {}",
            file_path.display(),
            size_bytes,
            object_key,
        );

        let s3_path = ObjectPath::from(object_key.as_str());
        let payload = Bytes::from(data);

        let put_result = self
            .store
            .put(&s3_path, payload.into())
            .await
            .map_err(|e| AcquisitionError::Storage {
                message: format!("Failed to upload {} to MinIO: {}", object_key, e),
            })?;

        let etag = put_result
            .e_tag
            .unwrap_or_default();

        debug!(
            "Successfully uploaded {} ({} bytes, etag={})",
            object_key, size_bytes, etag,
        );

        Ok(UploadResult {
            object_key,
            size_bytes,
            etag,
        })
    }

    /// Check if an object already exists in MinIO (HEAD request).
    ///
    /// Returns `true` if the object exists, `false` if it does not.
    ///
    /// # Errors
    ///
    /// Returns `AcquisitionError::Storage` on unexpected S3/network failures
    /// (a 404 Not Found is not treated as an error).
    pub async fn exists(&self, object_key: &str) -> AcquisitionResult<bool> {
        let s3_path = ObjectPath::from(object_key);

        match self.store.head(&s3_path).await {
            Ok(_) => Ok(true),
            Err(object_store::Error::NotFound { .. }) => Ok(false),
            Err(e) => Err(AcquisitionError::Storage {
                message: format!("Failed to check existence of {}: {}", object_key, e),
            }),
        }
    }

    /// Delete an object from MinIO.
    ///
    /// # Errors
    ///
    /// Returns `AcquisitionError::Storage` on S3/network failures.
    /// Deleting a non-existent key is not an error (S3 DELETE is idempotent).
    pub async fn delete(&self, object_key: &str) -> AcquisitionResult<()> {
        let s3_path = ObjectPath::from(object_key);

        self.store
            .delete(&s3_path)
            .await
            .map_err(|e| AcquisitionError::Storage {
                message: format!("Failed to delete {} from MinIO: {}", object_key, e),
            })?;

        debug!("Deleted object from MinIO: {}", object_key);
        Ok(())
    }

    /// Generate the object key for an upload.
    ///
    /// Returns `{prefix}{job_id}/{filename}`.
    ///
    /// # Errors
    ///
    /// Returns `AcquisitionError::Validation` if the path has no filename
    /// component or the filename is not valid UTF-8.
    pub fn generate_object_key(
        &self,
        file_path: &Path,
        job_id: Uuid,
    ) -> AcquisitionResult<String> {
        let filename = file_path
            .file_name()
            .ok_or_else(|| AcquisitionError::Validation {
                message: format!("Path has no filename component: {}", file_path.display()),
            })?
            .to_str()
            .ok_or_else(|| AcquisitionError::Validation {
                message: format!(
                    "Filename is not valid UTF-8: {}",
                    file_path.display()
                ),
            })?;

        Ok(format!("{}{}/{}", self.config.prefix, job_id, filename))
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;

    #[test]
    fn test_generate_object_key_success() {
        let config = MinIOUploaderConfig {
            prefix: "databento/".to_string(),
            ..MinIOUploaderConfig::default()
        };
        let store = Arc::new(object_store::memory::InMemory::new());
        let uploader = MinIOUploader::with_store(config, store);

        let job_id = Uuid::nil();
        let path = Path::new("/tmp/test_file.dbn");
        let key = uploader.generate_object_key(path, job_id);
        assert!(key.is_ok());
        assert_eq!(
            key.ok(),
            Some(format!("databento/{}/test_file.dbn", Uuid::nil()))
        );
    }

    #[test]
    fn test_generate_object_key_no_filename() {
        let config = MinIOUploaderConfig::default();
        let store = Arc::new(object_store::memory::InMemory::new());
        let uploader = MinIOUploader::with_store(config, store);

        let job_id = Uuid::nil();
        let path = Path::new("/");
        let result = uploader.generate_object_key(path, job_id);
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_upload_and_exists_with_in_memory_store() {
        let temp_dir = tempfile::TempDir::new().ok();
        let Some(temp_dir) = temp_dir else {
            return;
        };
        let test_file = temp_dir.path().join("test_data.dbn");
        tokio::fs::write(&test_file, b"fake dbn content")
            .await
            .ok();

        let config = MinIOUploaderConfig {
            prefix: "test/".to_string(),
            ..MinIOUploaderConfig::default()
        };
        let store = Arc::new(object_store::memory::InMemory::new());
        let uploader = MinIOUploader::with_store(config, store);

        let job_id = Uuid::nil();
        let metadata = std::collections::HashMap::new();

        let result = uploader.upload(&test_file, job_id, metadata).await;
        assert!(result.is_ok());

        let upload_result = result.ok();
        let Some(upload_result) = upload_result else {
            return;
        };
        assert_eq!(upload_result.size_bytes, 16); // "fake dbn content" is 16 bytes
        assert!(upload_result.object_key.contains("test_data.dbn"));

        // Verify the object exists
        let exists = uploader.exists(&upload_result.object_key).await;
        assert_eq!(exists.ok(), Some(true));

        // Verify a non-existent key returns false
        let not_exists = uploader.exists("nonexistent/key").await;
        assert_eq!(not_exists.ok(), Some(false));
    }

    #[tokio::test]
    async fn test_delete_with_in_memory_store() {
        let temp_dir = tempfile::TempDir::new().ok();
        let Some(temp_dir) = temp_dir else {
            return;
        };
        let test_file = temp_dir.path().join("to_delete.dbn");
        tokio::fs::write(&test_file, b"delete me").await.ok();

        let config = MinIOUploaderConfig {
            prefix: "del/".to_string(),
            ..MinIOUploaderConfig::default()
        };
        let store = Arc::new(object_store::memory::InMemory::new());
        let uploader = MinIOUploader::with_store(config, store);

        let job_id = Uuid::nil();
        let metadata = std::collections::HashMap::new();

        let result = uploader.upload(&test_file, job_id, metadata).await;
        assert!(result.is_ok());
        let key = result.map(|r| r.object_key).unwrap_or_default();

        // Delete the object
        let del_result = uploader.delete(&key).await;
        assert!(del_result.is_ok());

        // Verify it no longer exists
        let exists = uploader.exists(&key).await;
        assert_eq!(exists.ok(), Some(false));
    }
}