use std::{pin::Pin, sync::Arc, time::SystemTime};

use anyhow::ensure;
use async_trait::async_trait;
use bytes::Bytes;
use futures::Stream;
use google_cloud_storage::model_ext::ReadRange as GcsReadRange;
use log::info;
use thiserror::Error;
use tici_common::config::S3Config;
use tokio::io::AsyncRead;

pub const STORAGE_BACKEND_S3: &str = "s3";
pub const STORAGE_BACKEND_GCS: &str = "gcs";
pub const STORAGE_SERVICE_META: &str = "meta";
pub const STORAGE_SERVICE_READER: &str = "reader";
pub const STORAGE_SERVICE_UNKNOWN: &str = "unknown";
pub const STORAGE_SERVICE_WORKER: &str = "worker";

#[derive(Debug, Error)]
pub enum StorageError {
    #[error("IO Error: {0}")]
    Io(#[from] std::io::Error),

    #[error("Object not found: {0}")]
    NotFound(String),

    #[error("Permission denied: {0}")]
    PermissionDenied(String),

    #[error("Configuration error: {0}")]
    ConfigError(String),

    #[error("Upstream Service Error: {0}")]
    ServiceError(String),

    #[error("Invalid input: {0}")]
    InputError(String),

    #[error("Range not satisfiable: {0}")]
    RangeNotSatisfiable(String),

    #[error("Unknown error: {0}")]
    Unknown(String),
}

pub type Result<T> = std::result::Result<T, StorageError>;

fn map_http_status_code(status: u16, err_msg: String) -> StorageError {
    match status {
        404 => StorageError::NotFound(err_msg),
        401 | 403 => StorageError::PermissionDenied(err_msg),
        // 416 Range Not Satisfiable
        416 => StorageError::RangeNotSatisfiable(err_msg),
        // Retryable 4xx: 408 Request Timeout, 429 Too Many Requests
        408 | 429 => StorageError::ServiceError(err_msg),
        // Other 4xx: non-retryable client errors
        400..=499 => StorageError::InputError(format!("Client error: {}", err_msg)),
        // 5xx: retryable server errors
        500..=599 => StorageError::ServiceError(err_msg),
        _ => StorageError::ServiceError(err_msg),
    }
}

#[derive(Debug, Clone, Default)]
pub struct ObjectMeta {
    pub content_length: Option<i64>,
    pub last_modified: Option<SystemTime>,
}

/// Specifies a byte range to read from an object.
#[derive(Debug, Clone, Copy)]
pub struct ReadRange {
    /// The byte offset to start reading from.
    offset: u64,
    /// The number of bytes to read. If `None`, read to the end of the object.
    size: Option<u64>,
}

impl ReadRange {
    /// Create a range starting from `offset` and reading to the end.
    pub fn from_offset(offset: u64) -> Self {
        Self { offset, size: None }
    }

    /// Create a range starting from `offset` with a specific `size`.
    pub fn new(offset: u64, size: u64) -> anyhow::Result<Self> {
        ensure!(size > 0, "ReadRange size must be greater than 0");
        ensure!(
            offset.checked_add(size - 1).is_some(),
            "ReadRange offset + size would overflow"
        );
        Ok(Self {
            offset,
            size: Some(size),
        })
    }

    /// Convert to S3 Range header value.
    /// Format: "bytes=offset-end" or "bytes=offset-" (to end of file).
    pub fn to_s3_read_range(&self) -> String {
        match self.size {
            // Safety: overflow is checked in `new()`
            Some(size) => format!("bytes={}-{}", self.offset, self.offset + size - 1),
            None => format!("bytes={}-", self.offset),
        }
    }

    /// Convert to GCS ReadRange for google-cloud-storage library.
    pub fn to_gcs_read_range(&self) -> GcsReadRange {
        match self.size {
            Some(size) => GcsReadRange::segment(self.offset, size),
            _ => GcsReadRange::offset(self.offset),
        }
    }

    /// Calculate the actual content length based on the object's total size.
    /// Returns an error if:
    /// - The offset is beyond the object size
    /// - A specific size was requested but not enough bytes are available
    pub fn content_length(&self, total_size: u64) -> std::result::Result<u64, StorageError> {
        if self.offset >= total_size {
            return Err(StorageError::RangeNotSatisfiable(format!(
                "offset {} is beyond object size {}",
                self.offset, total_size
            )));
        }
        let remaining = total_size - self.offset;
        match self.size {
            Some(size) if remaining < size => Err(StorageError::RangeNotSatisfiable(format!(
                "requested {} bytes at offset {}, but only {} bytes available (object size {})",
                size, self.offset, remaining, total_size
            ))),
            Some(size) => Ok(size),
            None => Ok(remaining),
        }
    }

    pub fn offset(&self) -> u64 {
        self.offset
    }

    pub fn size(&self) -> Option<u64> {
        self.size
    }
}

// BackendStorage represents a kind of file system read / write
// interface abstraction. It hide the details of different storage backends.
#[async_trait]
pub trait BackendStorage: Send + Sync {
    /// Upload object from a stream
    async fn put_object(
        &self,
        bucket: &str,
        key: &str,
        body: Pin<
            Box<
                dyn Stream<Item = std::result::Result<Bytes, std::io::Error>>
                    + Send
                    + Sync
                    + 'static,
            >,
        >,
        content_len: u64,
    ) -> Result<()>;

    /// Upload data from a byte buffer
    async fn put_data(&self, bucket: &str, key: &str, data: Vec<u8>) -> Result<()>;

    /// Get object from storage, optionally with a byte range.
    ///
    /// - `range: None` - Read the entire object.
    /// - `range: Some(ReadRange { offset, size: None })` - Read from offset to
    ///   end.
    /// - `range: Some(ReadRange { offset, size: Some(n) })` - Read n bytes from
    ///   offset.
    ///
    /// The `content_length` in `ObjectMeta` will reflect the number of bytes
    /// being returned (either the range size or remaining bytes).
    ///
    /// `last_modified` may be `None` for backends where fetching it requires an
    /// extra metadata request. Call [`BackendStorage::head_object`] when the
    /// caller requires `last_modified`.
    async fn get_object(
        &self,
        bucket: &str,
        key: &str,
        range: Option<ReadRange>,
    ) -> Result<(Box<dyn AsyncRead + Unpin + Send>, ObjectMeta)>;

    /// Get object metadata without reading the object body.
    ///
    /// The returned `content_length` is the full object size.
    async fn head_object(&self, bucket: &str, key: &str) -> Result<ObjectMeta>;

    /// List objects with a single request.
    async fn list(
        &self,
        bucket: &str,
        prefix: &str,
        start_after: Option<&str>,
    ) -> Result<Vec<String>>;

    /// List all objects with automatic pagination.
    ///
    /// This convenience method loads *all* matching objects into memory.
    /// For buckets with a large number of objects, prefer using [`list`]
    /// with `start_after` for explicit pagination to avoid excessive memory
    /// use and long-running operations. `list_all` is best suited for small
    /// datasets or one-off administrative tasks.
    async fn list_all(&self, bucket: &str, prefix: &str) -> Result<Vec<String>>;

    /// List common prefixes with a delimiter and automatic pagination
    async fn list_with_delimiter(
        &self,
        bucket: &str,
        prefix: &str,
        start_after: Option<&str>,
        delimiter: &str,
    ) -> Result<Vec<String>>;

    /// Delete multiple objects by their keys
    async fn delete_objects(&self, bucket: &str, keys: &[String]) -> Result<()>;
}

pub mod gcs_storage;
pub mod s3_storage;

// Factory function to create BackendStorage based on S3Config.
pub async fn create_backend_storage(
    config: &S3Config,
    service: &'static str,
) -> Result<Arc<dyn BackendStorage>> {
    // HMAC Key -> S3 compatible mode, e.g. MinIO, AWS S3, GCS S3 compatible mode
    // etc.
    if config.has_key() {
        info!("Create backend storage: S3 compatible storage, service={service}");
        return Ok(Arc::new(
            s3_storage::S3Storage::new(config.clone(), service).await?,
        ));
    }

    // GCS native mode
    if gcs_storage::is_gcs_endpoint(&config.endpoint) {
        info!("Create backend storage: GCS, service={service}");
        return Ok(Arc::new(
            gcs_storage::GcsStorage::new(config.clone(), service).await?,
        ));
    }

    info!("Create backend storage: S3, service={service}");
    Ok(Arc::new(
        s3_storage::S3Storage::new(config.clone(), service).await?,
    ))
}

pub async fn create_backend_storage_for_test(config: &S3Config) -> Result<Arc<dyn BackendStorage>> {
    create_backend_storage(config, STORAGE_SERVICE_UNKNOWN).await
}

#[cfg(test)]
mod tests {
    use anyhow::Result;

    use super::*;

    #[test]
    fn test_read_range_from_offset() {
        let range = ReadRange::from_offset(100);
        assert_eq!(range.offset, 100);
        assert!(range.size.is_none());
    }

    #[test]
    fn test_read_range_new_valid() -> Result<()> {
        let range = ReadRange::new(100, 50)?;
        assert_eq!(range.offset, 100);
        assert_eq!(range.size, Some(50));
        Ok(())
    }

    #[test]
    fn test_read_range_new_size_one() -> Result<()> {
        // Edge case: size=1 should work
        let range = ReadRange::new(0, 1)?;
        assert_eq!(range.offset, 0);
        assert_eq!(range.size, Some(1));
        Ok(())
    }

    #[test]
    fn test_read_range_new_size_zero_returns_error() {
        // size=0 should return an error
        let result = ReadRange::new(100, 0);
        assert!(result.is_err());
        let err_msg = result.err().map(|e| e.to_string()).unwrap_or_default();
        assert!(err_msg.contains("size must be greater than 0"));
    }

    #[test]
    fn test_read_range_new_overflow_returns_error() {
        // offset + size - 1 would overflow u64
        let result = ReadRange::new(u64::MAX, 2);
        assert!(result.is_err());
        let err_msg = result.err().map(|e| e.to_string()).unwrap_or_default();
        assert!(err_msg.contains("overflow"));

        // Edge case: offset + size - 1 == u64::MAX should succeed
        let result = ReadRange::new(u64::MAX - 1, 2);
        assert!(result.is_ok());
    }

    #[test]
    fn test_to_s3_read_range_with_size() -> Result<()> {
        // Range with specific size: "bytes=offset-end"
        let range = ReadRange::new(100, 50)?;
        assert_eq!(range.to_s3_read_range(), "bytes=100-149");

        let range = ReadRange::new(0, 1)?;
        assert_eq!(range.to_s3_read_range(), "bytes=0-0");

        let range = ReadRange::new(0, 1000)?;
        assert_eq!(range.to_s3_read_range(), "bytes=0-999");
        Ok(())
    }

    #[test]
    fn test_to_s3_read_range_to_end() {
        // Range to end of file: "bytes=offset-"
        let range = ReadRange::from_offset(100);
        assert_eq!(range.to_s3_read_range(), "bytes=100-");

        let range = ReadRange::from_offset(0);
        assert_eq!(range.to_s3_read_range(), "bytes=0-");
    }

    #[test]
    fn test_to_gcs_read_range_with_size() -> Result<()> {
        let range = ReadRange::new(100, 50)?;
        let gcs_range = range.to_gcs_read_range();
        assert_eq!(gcs_range, GcsReadRange::segment(100, 50));
        Ok(())
    }

    #[test]
    fn test_to_gcs_read_range_to_end() {
        let range = ReadRange::from_offset(100);
        let gcs_range = range.to_gcs_read_range();
        assert_eq!(gcs_range, GcsReadRange::offset(100));
    }

    #[test]
    fn test_content_length_with_size() -> Result<()> {
        // Normal case: range within object
        let range = ReadRange::new(0, 100)?;
        assert_eq!(range.content_length(1000)?, 100);

        let range = ReadRange::new(900, 100)?;
        assert_eq!(range.content_length(1000)?, 100);
        Ok(())
    }

    #[test]
    fn test_content_length_to_end() -> Result<()> {
        // Range to end of file
        let range = ReadRange::from_offset(0);
        assert_eq!(range.content_length(1000)?, 1000);

        let range = ReadRange::from_offset(100);
        assert_eq!(range.content_length(1000)?, 900);
        Ok(())
    }

    #[test]
    fn test_content_length_offset_at_end_returns_error() -> Result<()> {
        // Offset at end of object should return error
        let range = ReadRange::from_offset(1000);
        assert!(range.content_length(1000).is_err());

        let range = ReadRange::new(1000, 100)?;
        assert!(range.content_length(1000).is_err());
        Ok(())
    }

    #[test]
    fn test_content_length_offset_beyond_size_returns_error() -> Result<()> {
        // Offset beyond object size should return error
        let range = ReadRange::from_offset(1500);
        assert!(range.content_length(1000).is_err());

        let range = ReadRange::new(1500, 100)?;
        assert!(range.content_length(1000).is_err());
        Ok(())
    }

    #[test]
    fn test_content_length_exceeds_object_size_returns_error() -> Result<()> {
        // Requested range exceeds object size - should return error
        // Object is 1000 bytes, requesting 200 bytes starting at offset 900
        // Only 100 bytes available (900..1000), but 200 was requested
        let range = ReadRange::new(900, 200)?;
        assert!(range.content_length(1000).is_err());
        Ok(())
    }
}