use std::sync::Arc;

use anyhow::{Context, Result, ensure};
use backon::{ExponentialBuilder, Retryable};
use log::{debug, warn};
use storage_backend::storage::{BackendStorage, StorageError};
use tokio::io::AsyncReadExt;

use crate::utils::metrics::{inc_cdc_op_count, observe_cdc_op_latency, start_cdc_op_timer};

pub struct FileContent {
    pub data: String,
    pub stats: FileContentStats,
}

#[derive(Clone, Copy, Debug, Default)]
pub struct FileContentStats {
    pub bytes_len: usize,
    pub get_object_elapsed: f64,
    pub read_object_elapsed: f64,
    pub utf8_decode_elapsed: f64,
    pub downstream_delay: Option<f64>,
    pub prefetch_wait_elapsed: Option<f64>,
}

pub struct CDCFileCache {
    backend_storage: Arc<dyn BackendStorage>,
    bucket: String,
    head_object_last_modified_fallback: bool,
}

impl CDCFileCache {
    pub fn new(
        backend_storage: Arc<dyn BackendStorage>,
        bucket: impl Into<String>,
        head_object_last_modified_fallback: bool,
    ) -> Self {
        Self {
            backend_storage,
            bucket: bucket.into(),
            head_object_last_modified_fallback,
        }
    }

    async fn get_last_modified(
        &self,
        filename: &str,
        current_last_modified: Option<std::time::SystemTime>,
    ) -> Option<std::time::SystemTime> {
        if let Some(last_modified) = current_last_modified {
            return Some(last_modified);
        }
        if !self.head_object_last_modified_fallback {
            return None;
        }

        match self
            .backend_storage
            .head_object(&self.bucket, filename)
            .await
        {
            Ok(meta) => meta.last_modified,
            Err(err) => {
                warn!(
                    "Failed to fetch object metadata for bucket={}, key={}, proceeding without last_modified: {:?}",
                    self.bucket, filename, err
                );
                None
            }
        }
    }

    /// Read a CDC file from S3 directly without caching
    async fn read_file_from_s3(&self, filename: &str) -> Result<FileContent> {
        let get_object_timer = start_cdc_op_timer("get_object");
        let (mut reader, meta) = self
            .backend_storage
            .get_object(&self.bucket, filename, None)
            .await
            .with_context(|| {
                inc_cdc_op_count("get_object_err");
                format!(
                    "Failed to get object from bucket={}, key={}",
                    &self.bucket, filename
                )
            })?;
        let get_object_elapsed = get_object_timer.stop_and_record();

        let read_object_timer = start_cdc_op_timer("read_object");
        let mut bytes = Vec::new();
        reader.read_to_end(&mut bytes).await.with_context(|| {
            inc_cdc_op_count("read_object_err");
            format!(
                "Failed to read bytes from bucket={}, key={}",
                &self.bucket, filename
            )
        })?;
        let read_object_elapsed = read_object_timer.stop_and_record();

        let read_len = bytes.len();
        if let Some(content_len) = meta.content_length {
            ensure!(
                read_len as i64 == content_len,
                "Read length mismatch: expected={}, got={}, bucket={}, key={}",
                content_len,
                read_len,
                &self.bucket,
                filename
            );
        }

        let utf8_decode_timer = start_cdc_op_timer("utf8_decode");
        let data = String::from_utf8(bytes).with_context(|| {
            inc_cdc_op_count("utf8_decode_err");
            format!(
                "UTF-8 conversion failed from bucket={}, key={}",
                &self.bucket, filename
            )
        })?;
        let utf8_decode_elapsed = utf8_decode_timer.stop_and_record();

        inc_cdc_op_count("get_object_succ");

        let last_modified = self.get_last_modified(filename, meta.last_modified).await;
        let delay = last_modified
            .and_then(|t| std::time::SystemTime::now().duration_since(t).ok())
            .inspect(|d| {
                observe_cdc_op_latency("downstream_delay", *d);
            });

        Ok(FileContent {
            data,
            stats: FileContentStats {
                bytes_len: read_len,
                get_object_elapsed,
                read_object_elapsed,
                utf8_decode_elapsed,
                downstream_delay: delay.map(|d| d.as_secs_f64()),
                prefetch_wait_elapsed: None,
            },
        })
    }

    async fn try_read_file_from_s3(&self, filename: &str) -> Result<FileContent> {
        let bucket = self.bucket.clone();
        let key = filename.to_string();
        let operation = || async { self.read_file_from_s3(filename).await };

        operation
            .retry(&ExponentialBuilder::default().with_max_times(3))
            .when(|e| {
                match e.downcast_ref::<StorageError>() {
                    // NotFound and PermissionDenied are not retryable
                    Some(StorageError::NotFound(_)) | Some(StorageError::PermissionDenied(_)) => {
                        false
                    }
                    Some(_) => true, // Other StorageError retryable
                    None => false,   // Other errors not retryable
                }
            })
            .notify(move |e, dur| {
                warn!(
                    "Retry read cdc file from bucket={}, key={}, wait={:?}, err={:?}",
                    bucket, key, dur, e
                );
            })
            .await
    }

    /// Prefetch returns file content to the caller instead of storing another
    /// in-memory copy inside this helper. The reader already bounds memory by
    /// capping the number of inflight prefetch tasks, so duplicating completed
    /// task results in an internal cache would only increase peak memory usage.
    pub async fn prefetch_file(&self, filename: &str) -> Result<FileContent> {
        debug!("Prefetch CDC file: {}", filename);
        self.try_read_file_from_s3(filename).await
    }

    /// Read a CDC file immediately without going through the reader's prefetch
    /// queue. Tests still use this helper to fetch fixture content directly.
    pub async fn read_file(&self, filename: &str) -> Result<FileContent> {
        debug!("Read CDC file on demand: {}", filename);
        self.try_read_file_from_s3(filename).await
    }
}

#[cfg(test)]
mod tests {
    use std::time::{Duration, SystemTime};

    use storage_backend::storage::ObjectMeta;
    use tici_common::cdc_file_name::CDC_FILENAME_MATCH;

    use super::*;
    use crate::test_utils::{
        MockBackendStorage, TEST_S3_BUCKET, TEST_S3_PREFIX, test_create_backend_storage,
    };

    async fn list_cdc_files(backend_storage: &Arc<dyn BackendStorage>) -> Vec<String> {
        let all_keys = backend_storage
            .list(TEST_S3_BUCKET, TEST_S3_PREFIX.as_str(), None)
            .await
            .unwrap();
        let mut files = all_keys
            .into_iter()
            .filter(|key| CDC_FILENAME_MATCH.is_match(key))
            .collect::<Vec<_>>();
        files.sort_unstable();
        files
    }

    #[tokio::test]
    async fn test_prefetch_file_reads_content() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let cache = CDCFileCache::new(backend_storage.clone(), TEST_S3_BUCKET, true);

        let files = list_cdc_files(&backend_storage).await;
        assert!(!files.is_empty(), "Need at least 1 test file in S3 bucket");

        let content = cache.prefetch_file(&files[0]).await.unwrap();
        assert!(!content.data.is_empty());
        assert_eq!(content.stats.bytes_len, content.data.len());
        assert!(content.stats.prefetch_wait_elapsed.is_none());
    }

    #[tokio::test]
    async fn test_read_file_reads_content() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let cache = CDCFileCache::new(backend_storage.clone(), TEST_S3_BUCKET, true);

        let files = list_cdc_files(&backend_storage).await;
        assert!(!files.is_empty(), "Need at least 1 test file in S3 bucket");

        let content = cache.read_file(&files[0]).await.unwrap();
        assert!(!content.data.is_empty());
        assert_eq!(content.stats.bytes_len, content.data.len());
        assert!(content.stats.prefetch_wait_elapsed.is_none());
    }

    #[tokio::test]
    async fn test_prefetch_and_direct_read_return_same_content() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let cache = CDCFileCache::new(backend_storage.clone(), TEST_S3_BUCKET, true);

        let files = list_cdc_files(&backend_storage).await;
        assert!(!files.is_empty(), "Need at least 1 test file in S3 bucket");

        let prefetched = cache.prefetch_file(&files[0]).await.unwrap();
        let direct = cache.read_file(&files[0]).await.unwrap();
        assert_eq!(prefetched.data, direct.data);
    }

    #[tokio::test]
    async fn test_read_nonexistent_file() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let cache = CDCFileCache::new(backend_storage.clone(), TEST_S3_BUCKET, true);

        let result = cache
            .read_file("nonexistent/path/CDC00000000000000000000.json")
            .await;

        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_concurrent_reads() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let cache = Arc::new(CDCFileCache::new(
            backend_storage.clone(),
            TEST_S3_BUCKET,
            true,
        ));

        let files = list_cdc_files(&backend_storage).await;
        assert!(files.len() >= 3, "Need at least 3 test files in S3 bucket");

        let mut handles = vec![];
        for file in files.iter().take(3) {
            let cache_clone = cache.clone();
            let file = file.clone();
            let handle = tokio::spawn(async move { cache_clone.prefetch_file(&file).await });
            handles.push(handle);
        }

        for handle in handles {
            let result = handle.await.unwrap();
            assert!(result.is_ok());
        }
    }

    #[tokio::test]
    async fn test_read_file_falls_back_to_head_object_for_last_modified() {
        let backend_storage = Arc::new(MockBackendStorage::new());
        let filename = "test/1/CDC00000000000000000001.json";
        let expected_last_modified = SystemTime::now() - Duration::from_secs(30);
        backend_storage.push_object_response(TEST_S3_BUCKET, filename, Ok(b"{}".to_vec()));
        backend_storage.push_head_object_response(
            TEST_S3_BUCKET,
            filename,
            Ok(ObjectMeta {
                content_length: Some(2),
                last_modified: Some(expected_last_modified),
            }),
        );

        let cache = CDCFileCache::new(backend_storage, TEST_S3_BUCKET, true);
        let content = cache.read_file(filename).await.unwrap();

        assert_eq!(content.data, "{}");
        assert!(content.stats.downstream_delay.is_some());
    }

    #[tokio::test]
    async fn test_read_file_ignores_head_object_errors_for_last_modified() {
        let backend_storage = Arc::new(MockBackendStorage::new());
        let filename = "test/1/CDC00000000000000000002.json";
        backend_storage.push_object_response(TEST_S3_BUCKET, filename, Ok(b"{}".to_vec()));

        let cache = CDCFileCache::new(backend_storage, TEST_S3_BUCKET, true);
        let content = cache.read_file(filename).await.unwrap();

        assert_eq!(content.data, "{}");
        assert!(content.stats.downstream_delay.is_none());
    }

    #[tokio::test]
    async fn test_read_file_skips_head_object_for_last_modified_when_disabled() {
        let backend_storage = Arc::new(MockBackendStorage::new());
        let filename = "test/1/CDC00000000000000000003.json";
        let expected_last_modified = SystemTime::now() - Duration::from_secs(30);
        backend_storage.push_object_response(TEST_S3_BUCKET, filename, Ok(b"{}".to_vec()));
        backend_storage.push_head_object_response(
            TEST_S3_BUCKET,
            filename,
            Ok(ObjectMeta {
                content_length: Some(2),
                last_modified: Some(expected_last_modified),
            }),
        );

        let cache = CDCFileCache::new(backend_storage, TEST_S3_BUCKET, false);
        let content = cache.read_file(filename).await.unwrap();

        assert_eq!(content.data, "{}");
        assert!(content.stats.downstream_delay.is_none());
    }
}