use std::{sync::Arc, time::Duration};

use anyhow::{Context, Result, anyhow, bail, ensure};
use log::{info, warn};
use storage_backend::storage::{BackendStorage, StorageError};
use tici_common::cdc_file_name::{CDC_FILENAME_MATCH, CDCFileName};
use tokio::io::AsyncReadExt;

use crate::{
    utils::metrics::{inc_cdc_op_count, inc_cdc_op_count_by, start_cdc_op_timer},
    writer::cdc_file_prefix::CDCPrefix,
};

const CDC_INDEX_READ_MAX_ATTEMPTS: usize = 3;
const CDC_META_DIR_NAME: &str = "meta";
const CDC_INDEX_READ_RETRY_INTERVAL: Duration = Duration::from_millis(500);

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
struct CDCLeafDirectory {
    table_version: u64,
}

impl CDCLeafDirectory {
    fn data_dir_path(&self, prefix: &CDCPrefix) -> String {
        let table_prefix = prefix.table_prefix().trim_end_matches('/');
        format!("{table_prefix}/{}", self.table_version)
    }

    fn index_file_path(&self, prefix: &CDCPrefix) -> String {
        format!("{}/meta/CDC.index", self.data_dir_path(prefix))
    }
}

// List S3 files from the last processed file.
// Not thread-safe, should be used within a single async task.
pub struct CDCFilePoller {
    backend_storage: Arc<dyn BackendStorage>,
    bucket: String,
}

pub(crate) struct CDCListResult {
    pub unread_files: Vec<String>,
    pub list_secs: f64,
}

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

    async fn validate_unread_files_for_table(
        &self,
        last_file: &CDCFileName,
        prefix: &str,
        unread_files: &[String],
    ) -> Result<Vec<String>> {
        ensure!(
            !unread_files.is_empty(),
            "No unread files to validate, bucket={}, prefix={}, last_file={:?}",
            self.bucket,
            prefix,
            last_file
        );

        if (!last_file.is_empty() && last_file >= &CDCFileName::parse(&unread_files[0])?)
            || !unread_files.is_sorted()
        {
            bail!(
                "Files are not sorted, bucket={}, prefix={}, last_file={:?}, unread_files={:?}",
                self.bucket,
                prefix,
                last_file,
                unread_files
            );
        }

        // Only validate and return the first unread directory in one poll.
        // If the storage response already spans into a later directory, we
        // leave that directory for the next poll round so the subsequent
        // directory switch can be validated against the previous directory's
        // CDC.index before any file in the new directory is processed.
        let mut files_in_same_dir: Vec<CDCFileName> = vec![];
        for file in unread_files {
            let file = CDCFileName::parse(file).with_context(|| {
                format!(
                    "Failed to parse unread CDC file, bucket={}, prefix={}, file={}",
                    self.bucket, prefix, file
                )
            })?;
            if files_in_same_dir.is_empty() || files_in_same_dir[0].same_directory(&file) {
                files_in_same_dir.push(file);
            } else {
                break;
            }
        }

        // On the first poll, the first visible CDC file may already be in the middle
        // of a directory, but the remaining files still need to be consecutive.
        if last_file.is_empty() {
            self.validate_sequence(
                prefix,
                None,
                &files_in_same_dir,
                "Detected CDC file hole within directory",
                "Detected CDC file hole within directory",
            )?;
        } else if last_file.same_directory(&files_in_same_dir[0]) {
            self.validate_sequence(
                prefix,
                Some(last_file.sequence_number()?),
                &files_in_same_dir,
                "Detected CDC file hole within directory",
                "Detected CDC file hole within directory",
            )?;
        } else {
            // If the directory is switched, sequence is started from 1.
            self.validate_directory_switch(prefix, last_file, &files_in_same_dir)
                .await?;
        }
        Ok(files_in_same_dir
            .iter()
            .map(|file| file.to_string())
            .collect())
    }

    fn validate_sequence(
        &self,
        prefix: &str,
        mut last_seq: Option<u64>,
        unread_files: &[CDCFileName],
        first_mismatch_msg: &str,
        later_mismatch_msg: &str,
    ) -> Result<()> {
        for (idx, file) in unread_files.iter().enumerate() {
            let seq = file.sequence_number()?;
            if let Some(prev_seq) = last_seq {
                if seq != prev_seq + 1 {
                    let mismatch_msg = if idx == 0 {
                        first_mismatch_msg
                    } else {
                        later_mismatch_msg
                    };
                    bail!(
                        "{}, bucket={}, prefix={}, last_seq={}, file={}, expected={}, actual={}",
                        mismatch_msg,
                        self.bucket,
                        prefix,
                        prev_seq,
                        file,
                        prev_seq + 1,
                        seq
                    );
                }
            }
            last_seq = Some(seq);
        }
        Ok(())
    }

    async fn validate_directory_switch(
        &self,
        prefix: &str,
        last_file: &CDCFileName,
        unread_files: &[CDCFileName],
    ) -> Result<()> {
        self.validate_sequence(
            prefix,
            Some(0),
            unread_files,
            "New CDC directory does not start from sequence 1",
            "Detected CDC file hole within directory",
        )?;
        if !last_file.is_empty() {
            self.validate_directory_sealed(prefix, last_file).await?;
        }
        Ok(())
    }

    async fn validate_directory_sealed(&self, prefix: &str, last_file: &CDCFileName) -> Result<()> {
        let index_path = last_file.index_file_path().with_context(|| {
            format!(
                "Failed to build CDC.index path, bucket={}, prefix={}, last_file={}",
                self.bucket, prefix, last_file
            )
        })?;
        let index_content = self.read_index_file(prefix, &index_path).await?;
        let index_content = index_content.trim();
        let last_file_name = last_file.file_name()?;
        if index_content == last_file_name {
            info!(
                "Directory is switched, bucket={}, prefix={}, last_file={}",
                self.bucket, prefix, last_file
            );
            return Ok(());
        }

        bail!(
            "CDC.index does not match the last visible CDC file, bucket={}, prefix={}, index_path={}, index_content={}, last_file_name={}",
            self.bucket,
            prefix,
            index_path,
            index_content,
            last_file_name,
        )
    }

    async fn read_index_file_once(&self, prefix: &str, index_path: &str) -> Result<Option<String>> {
        let (mut reader, _) = match self
            .backend_storage
            .get_object(&self.bucket, index_path, None)
            .await
        {
            Ok(result) => result,
            Err(StorageError::NotFound(_)) => return Ok(None),
            Err(err) => {
                return Err(err).with_context(|| {
                    format!(
                        "Failed to get CDC.index object, bucket={}, prefix={}, index_path={}",
                        self.bucket, prefix, index_path
                    )
                });
            }
        };

        let mut bytes = Vec::new();
        reader.read_to_end(&mut bytes).await.with_context(|| {
            format!(
                "Failed to read CDC.index bytes, bucket={}, prefix={}, index_path={}",
                self.bucket, prefix, index_path
            )
        })?;
        String::from_utf8(bytes)
            .with_context(|| {
                format!(
                    "CDC.index is not valid UTF-8, bucket={}, prefix={}, index_path={}",
                    self.bucket, prefix, index_path
                )
            })
            .map(Some)
    }

    // Since TiCDC first writes JSON files and then writes the CDC.index file,
    // reading CDC.index immediately after the first JSON file is written may return
    // a not found error. We adopt a wait-and-retry strategy here to avoid this
    // problem as much as possible.
    async fn read_index_file(&self, prefix: &str, index_path: &str) -> Result<String> {
        for attempt in 1..=CDC_INDEX_READ_MAX_ATTEMPTS {
            if let Some(content) = self.read_index_file_once(prefix, index_path).await? {
                return Ok(content);
            }

            if attempt < CDC_INDEX_READ_MAX_ATTEMPTS {
                warn!(
                    "CDC.index is missing, retrying after {:?}, bucket={}, prefix={}, index_path={}, attempt={}/{}",
                    CDC_INDEX_READ_RETRY_INTERVAL,
                    self.bucket,
                    prefix,
                    index_path,
                    attempt,
                    CDC_INDEX_READ_MAX_ATTEMPTS
                );
                tokio::time::sleep(CDC_INDEX_READ_RETRY_INTERVAL).await;
            }
        }

        bail!(
            "CDC.index is missing after {} attempts, bucket={}, prefix={}, index_path={}",
            CDC_INDEX_READ_MAX_ATTEMPTS,
            self.bucket,
            prefix,
            index_path,
        );
    }

    async fn last_leaf_directory(&self, prefix: &CDCPrefix) -> Result<Option<CDCLeafDirectory>> {
        // This function lists the following directories:
        // <table-prefix>/<table_version>/
        // <table-prefix>/meta/
        let table_version_prefixes = self
            .backend_storage
            .list_with_delimiter(&self.bucket, prefix.table_prefix(), None, "/")
            .await
            .with_context(|| {
                format!(
                    "Failed to list CDC table_version directories, bucket={}, prefix={:?}",
                    self.bucket, prefix
                )
            })?
            .into_iter();

        let mut last_table_version = None;
        for table_version_prefix in table_version_prefixes {
            let table_version_name = table_version_prefix
                .trim_end_matches('/')
                .rsplit('/')
                .next()
                .ok_or_else(|| anyhow!("Missing table_version in {table_version_prefix}"))?;
            if table_version_name == CDC_META_DIR_NAME {
                continue;
            }
            let table_version = table_version_name.parse::<u64>().with_context(|| {
                format!("Invalid table_version directory: {table_version_prefix}")
            })?;

            if last_table_version
                .as_ref()
                .is_none_or(|(last_version, _)| table_version > *last_version)
            {
                last_table_version = Some((table_version, table_version_prefix));
            }
        }

        let Some((table_version, table_version_prefix)) = last_table_version else {
            return Ok(None);
        };

        // The next level directories should only contain metadata:
        // <table-prefix>/<table-version>/CDC<sequence>.json -> not directory name
        // <table-prefix>/<table-version>/meta/
        let child_prefixes = self
            .backend_storage
            .list_with_delimiter(&self.bucket, &table_version_prefix, None, "/")
            .await
            .with_context(|| {
                format!(
                    "Failed to list CDC child directories, bucket={}, prefix={}",
                    self.bucket, table_version_prefix
                )
            })?;

        for child_prefix in child_prefixes {
            let child_name = child_prefix
                .trim_end_matches('/')
                .rsplit('/')
                .next()
                .ok_or_else(|| anyhow!("Missing child directory in {child_prefix}"))?;
            if child_name == CDC_META_DIR_NAME {
                continue;
            }
            bail!(
                "unexpected CDC subdirectory, bucket={}, prefix={}, child_name={}, child={}",
                self.bucket,
                table_version_prefix,
                child_name,
                child_prefix
            );
        }

        Ok(Some(CDCLeafDirectory { table_version }))
    }

    async fn directory_last_file_from_index(
        &self,
        prefix: &CDCPrefix,
        directory: &CDCLeafDirectory,
    ) -> Result<CDCFileName> {
        let index_path = directory.index_file_path(prefix);
        let index_content = self
            .read_index_file(prefix.table_prefix(), &index_path)
            .await?;
        let file_name = index_content.trim();
        let file = CDCFileName::parse(&format!("{}/{}", directory.data_dir_path(prefix), file_name))
            .with_context(|| {
                format!(
                    "Failed to parse latest CDC file from index, bucket={}, prefix={:?}, index_path={}, index_content={}",
                    self.bucket, prefix, index_path, file_name
                )
            })?;
        Ok(file)
    }

    pub async fn latest_file_for_table(&self, prefix: &CDCPrefix) -> Result<Option<CDCFileName>> {
        let timer = start_cdc_op_timer("list_latest");
        let directory = match self
            .last_leaf_directory(prefix)
            .await
            .inspect_err(|_| inc_cdc_op_count("list_latest_err"))?
        {
            None => {
                inc_cdc_op_count("list_latest_empty");
                return Ok(None);
            }
            Some(d) => d,
        };

        let latest_file = self
            .directory_last_file_from_index(prefix, &directory)
            .await
            .inspect_err(|_| inc_cdc_op_count("list_latest_err"))?;
        inc_cdc_op_count("list_latest_succ");
        info!(
            "Listed latest CDC file, bucket={}, prefix={:?}, latest={}, elapsed={:.3}s",
            self.bucket,
            prefix,
            latest_file,
            timer.stop_and_record()
        );
        Ok(Some(latest_file))
    }

    pub async fn list_unread_files(
        &self,
        prev_file: &CDCFileName,
        prefix: &CDCPrefix,
    ) -> Result<Vec<String>> {
        self.list_unread_files_for_table_with_stats(prev_file, prefix)
            .await
            .map(|result| result.unread_files)
    }

    pub async fn list_unread_files_for_table(
        &self,
        last_file: &CDCFileName,
        prefix: &CDCPrefix,
    ) -> Result<Vec<String>> {
        self.list_unread_files_for_table_with_stats(last_file, prefix)
            .await
            .map(|result| result.unread_files)
    }

    pub(crate) async fn list_unread_files_for_table_with_stats(
        &self,
        last_file: &CDCFileName,
        prefix: &CDCPrefix,
    ) -> Result<CDCListResult> {
        let timer = start_cdc_op_timer("list_unread");
        let start_after = if last_file.is_empty() {
            None
        } else {
            Some(last_file.to_string())
        };

        let keys = self
            .backend_storage
            .list(&self.bucket, prefix.table_prefix(), start_after.as_deref())
            .await
            .with_context(|| {
                inc_cdc_op_count("list_unread_err");
                format!(
                    "Failed to list unread files from bucket={}, prefix={:?}, start_after={:?}",
                    &self.bucket, prefix, start_after
                )
            })?;

        let mut unread_files = vec![];
        for key in keys {
            if key.starts_with(prefix.table_prefix()) && CDC_FILENAME_MATCH.is_match(&key) {
                unread_files.push(key);
            }
        }

        if unread_files.is_empty() {
            inc_cdc_op_count("list_unread_empty");
            return Ok(CDCListResult {
                unread_files,
                list_secs: timer.stop_and_record(),
            });
        }
        // `validate_unread_files_for_table()` intentionally narrows the batch
        // to the first unread directory so every directory switch is validated
        // before later-directory files are observed by the reader.
        unread_files = self
            .validate_unread_files_for_table(last_file, prefix.table_prefix(), &unread_files)
            .await
            .inspect_err(|_| {
                inc_cdc_op_count("list_unread_validate_err");
            })?;
        ensure!(
            !unread_files.is_empty(),
            "No unread files after validation, bucket={}, prefix={:?}, last_file={:?}",
            self.bucket,
            prefix,
            last_file
        );
        let unread_count = unread_files.len() as u64;
        inc_cdc_op_count("list_unread_succ");
        inc_cdc_op_count_by("list_unread_count", unread_count);
        Ok(CDCListResult {
            unread_files,
            list_secs: timer.stop_and_record(),
        })
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use storage_backend::storage::StorageError;

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

    const MOCK_BUCKET: &str = "mock-bucket";
    const MOCK_PREFIX: &str = "mock/prefix/123";
    const MOCK_TABLE_VERSION: u64 = 465019258146652181;

    fn mock_table_version_prefix(table_version: u64) -> String {
        format!("{MOCK_PREFIX}/{table_version}/")
    }

    fn mock_date_prefix(table_version: u64, date: &str) -> String {
        format!("{MOCK_PREFIX}/{table_version}/{date}/")
    }

    fn mock_table_meta_prefix() -> String {
        format!("{MOCK_PREFIX}/meta/")
    }

    fn mock_cdc_file_with_version(table_version: u64, sequence: u64) -> String {
        format!("{}/{}/CDC{:020}.json", MOCK_PREFIX, table_version, sequence)
    }

    fn mock_cdc_file(sequence: u64) -> String {
        mock_cdc_file_with_version(MOCK_TABLE_VERSION, sequence)
    }

    fn mock_index_path_with_version(table_version: u64) -> String {
        format!("{}/{}/meta/CDC.index", MOCK_PREFIX, table_version)
    }

    fn mock_index_path() -> String {
        mock_index_path_with_version(MOCK_TABLE_VERSION)
    }

    fn mock_prefix() -> CDCPrefix {
        CDCPrefix::new(MOCK_PREFIX.to_string())
    }

    fn parse_last_file(last_file: &str) -> CDCFileName {
        CDCFileName::parse(last_file).unwrap()
    }

    #[tokio::test]
    async fn test_latest_file_for_table_uses_directory_index() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let older_table_version_prefix = mock_table_version_prefix(MOCK_TABLE_VERSION);
        let latest_table_version = MOCK_TABLE_VERSION + 1;
        let latest_table_version_prefix = mock_table_version_prefix(latest_table_version);
        let latest_index_path = mock_index_path_with_version(latest_table_version);
        let latest = mock_cdc_file_with_version(latest_table_version, 3);

        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            None,
            "/",
            Ok(vec![
                older_table_version_prefix,
                latest_table_version_prefix.clone(),
            ]),
        );
        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            &mock_table_version_prefix(MOCK_TABLE_VERSION),
            None,
            "/",
            Ok(vec![format!("{MOCK_PREFIX}/{MOCK_TABLE_VERSION}/meta/")]),
        );
        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            &latest_table_version_prefix,
            None,
            "/",
            Ok(vec![format!("{MOCK_PREFIX}/{latest_table_version}/meta/")]),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &latest_index_path,
            Ok(b"CDC00000000000000000003.json\n".to_vec()),
        );

        let latest_file = poller
            .latest_file_for_table(&prefix)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(latest_file.to_string(), latest);
    }

    #[tokio::test]
    async fn test_latest_file_for_table_lists_only_latest_table_version() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let older_table_version_prefix = mock_table_version_prefix(MOCK_TABLE_VERSION);
        let latest_table_version = MOCK_TABLE_VERSION + 1;
        let latest_table_version_prefix = mock_table_version_prefix(latest_table_version);
        let latest_index_path = mock_index_path_with_version(latest_table_version);
        let latest = mock_cdc_file_with_version(latest_table_version, 3);

        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            None,
            "/",
            Ok(vec![
                older_table_version_prefix,
                latest_table_version_prefix.clone(),
            ]),
        );
        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            &latest_table_version_prefix,
            None,
            "/",
            Ok(vec![format!("{MOCK_PREFIX}/{latest_table_version}/meta/")]),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &latest_index_path,
            Ok(b"CDC00000000000000000003.json\n".to_vec()),
        );

        let latest_file = poller
            .latest_file_for_table(&prefix)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(latest_file.to_string(), latest);
    }

    #[tokio::test]
    async fn test_latest_file_for_table_ignores_table_meta_directory() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let latest_table_version = MOCK_TABLE_VERSION + 1;
        let latest_table_version_prefix = mock_table_version_prefix(latest_table_version);
        let latest_index_path = mock_index_path_with_version(latest_table_version);
        let latest = mock_cdc_file_with_version(latest_table_version, 3);

        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            None,
            "/",
            Ok(vec![
                mock_table_meta_prefix(),
                latest_table_version_prefix.clone(),
            ]),
        );
        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            &latest_table_version_prefix,
            None,
            "/",
            Ok(vec![format!("{MOCK_PREFIX}/{latest_table_version}/meta/")]),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &latest_index_path,
            Ok(b"CDC00000000000000000003.json\n".to_vec()),
        );

        let latest_file = poller
            .latest_file_for_table(&prefix)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(latest_file.to_string(), latest);
    }

    #[tokio::test]
    async fn test_latest_file_for_table_rejects_date_directory() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let table_version_prefix = mock_table_version_prefix(MOCK_TABLE_VERSION);

        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            None,
            "/",
            Ok(vec![table_version_prefix.clone()]),
        );
        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            &table_version_prefix,
            None,
            "/",
            Ok(vec![mock_date_prefix(MOCK_TABLE_VERSION, "2026-01-02")]),
        );

        let err = poller.latest_file_for_table(&prefix).await.unwrap_err();
        let err_msg = err.to_string();
        assert!(err_msg.contains("unexpected CDC subdirectory"));
        assert!(err_msg.contains("2026-01-02"));
    }

    #[tokio::test]
    async fn test_latest_file_for_table_handles_table_version_index() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let table_version_prefix = mock_table_version_prefix(MOCK_TABLE_VERSION);
        let table_version_meta_prefix = format!("{table_version_prefix}meta/");
        let index_path = format!("{table_version_prefix}meta/CDC.index");
        let latest = mock_cdc_file_with_version(MOCK_TABLE_VERSION, 3);

        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            None,
            "/",
            Ok(vec![table_version_prefix.clone()]),
        );
        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            &table_version_prefix,
            None,
            "/",
            Ok(vec![table_version_meta_prefix]),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &index_path,
            Ok(b"CDC00000000000000000003.json\n".to_vec()),
        );

        let latest_file = poller
            .latest_file_for_table(&prefix)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(latest_file.to_string(), latest);
    }

    #[tokio::test]
    async fn test_latest_file_for_table_retries_missing_index() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let latest_table_version = MOCK_TABLE_VERSION + 1;
        let latest_table_version_prefix = mock_table_version_prefix(latest_table_version);
        let latest_index_path = mock_index_path_with_version(latest_table_version);
        let latest = mock_cdc_file_with_version(latest_table_version, 3);

        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            None,
            "/",
            Ok(vec![latest_table_version_prefix.clone()]),
        );
        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            &latest_table_version_prefix,
            None,
            "/",
            Ok(vec![format!("{MOCK_PREFIX}/{latest_table_version}/meta/")]),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &latest_index_path,
            Err(StorageError::NotFound("missing index".to_string())),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &latest_index_path,
            Ok(b"CDC00000000000000000003.json\n".to_vec()),
        );

        let latest_file = poller
            .latest_file_for_table(&prefix)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(latest_file.to_string(), latest);
    }

    #[tokio::test]
    async fn test_latest_file_for_table_rejects_missing_index_after_retries() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let table_version_prefix = mock_table_version_prefix(MOCK_TABLE_VERSION);
        let index_path = mock_index_path();

        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            None,
            "/",
            Ok(vec![table_version_prefix.clone()]),
        );
        storage.push_list_with_delimiter_response(
            MOCK_BUCKET,
            &table_version_prefix,
            None,
            "/",
            Ok(vec![format!("{MOCK_PREFIX}/{MOCK_TABLE_VERSION}/meta/")]),
        );
        for _ in 0..CDC_INDEX_READ_MAX_ATTEMPTS {
            storage.push_object_response(
                MOCK_BUCKET,
                &index_path,
                Err(StorageError::NotFound("missing index".to_string())),
            );
        }

        let err = poller.latest_file_for_table(&prefix).await.unwrap_err();
        assert!(
            err.to_string()
                .contains("CDC.index is missing after 3 attempts")
        );
    }

    #[tokio::test]
    async fn test_list_unread_files() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let poller = CDCFilePoller::new(backend_storage.clone(), TEST_S3_BUCKET);

        // Get all CDC files using ObjectStore
        let all_keys = backend_storage
            .list(TEST_S3_BUCKET, TEST_S3_PREFIX.as_str(), None)
            .await
            .unwrap();
        let mut files = vec![];
        for key in all_keys {
            if CDC_FILENAME_MATCH.is_match(&key) {
                files.push(key);
            }
        }
        assert!(!files.is_empty());

        let prefix = CDCPrefix::new(
            files
                .first()
                .unwrap()
                .clone()
                .split("/CDC")
                .next()
                .unwrap()
                .to_string(),
        );
        // Test listing unread files from different last_file positions
        assert_eq!(
            poller
                .list_unread_files_for_table(&CDCFileName::empty(), &prefix)
                .await
                .unwrap()
                .len(),
            files.len()
        );
        assert_eq!(
            poller
                .list_unread_files_for_table(&parse_last_file(files.first().unwrap()), &prefix)
                .await
                .unwrap()
                .len(),
            files.len() - 1
        );
        assert_eq!(
            poller
                .list_unread_files_for_table(&parse_last_file(files.last().unwrap()), &prefix)
                .await
                .unwrap()
                .len(),
            0
        );
    }

    #[tokio::test]
    async fn test_list_unread_files_empty_result() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let poller = CDCFilePoller::new(backend_storage.clone(), TEST_S3_BUCKET);

        // Use a prefix that doesn't exist
        let non_existent_prefix = CDCPrefix::new("non/existent/prefix/path".to_string());
        let result = poller
            .list_unread_files_for_table(&CDCFileName::empty(), &non_existent_prefix)
            .await
            .unwrap();

        assert_eq!(result.len(), 0);
    }

    #[tokio::test]
    async fn test_list_unread_files_filters_non_cdc_files() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let poller = CDCFilePoller::new(backend_storage.clone(), TEST_S3_BUCKET);

        let all_keys = backend_storage
            .list(TEST_S3_BUCKET, TEST_S3_PREFIX.as_str(), None)
            .await
            .unwrap();

        // Count all files vs CDC files
        let all_files_count = all_keys.len();
        let mut cdc_files_count = 0;
        for key in &all_keys {
            if CDC_FILENAME_MATCH.is_match(key) {
                cdc_files_count += 1;
            }
        }

        let prefix = CDCPrefix::new(TEST_S3_PREFIX.to_string());
        let result = poller
            .list_unread_files_for_table(&CDCFileName::empty(), &prefix)
            .await
            .unwrap();

        // Verify that only CDC files are returned (should match or be less than all
        // files)
        assert_eq!(result.len(), cdc_files_count);
        assert!(result.len() <= all_files_count);

        // Verify all returned files match CDC pattern
        for file in result {
            assert!(CDC_FILENAME_MATCH.is_match(&file));
        }
    }

    #[tokio::test]
    async fn test_poller_with_different_buckets() {
        let backend_storage = test_create_backend_storage().await.unwrap();

        // Create pollers with different buckets
        let poller1 = CDCFilePoller::new(backend_storage.clone(), TEST_S3_BUCKET);
        let poller2 = CDCFilePoller::new(backend_storage.clone(), "different-bucket");
        let existing_prefix = CDCPrefix::new(TEST_S3_PREFIX.to_string());
        let missing_prefix = CDCPrefix::new("some_prefix".to_string());

        // First poller should work with test bucket
        let result1 = poller1
            .list_unread_files_for_table(&CDCFileName::empty(), &existing_prefix)
            .await;
        assert!(result1.is_ok());

        // Second poller with non-existent bucket should fail
        let result2 = poller2
            .list_unread_files_for_table(&CDCFileName::empty(), &missing_prefix)
            .await;
        assert!(result2.is_err());
    }

    #[tokio::test]
    async fn test_list_unread_files_for_normal_table() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let poller = CDCFilePoller::new(backend_storage.clone(), TEST_S3_BUCKET);

        let all_keys = backend_storage
            .list(TEST_S3_BUCKET, TEST_S3_PREFIX.as_str(), None)
            .await
            .unwrap();

        let mut files = vec![];
        for key in all_keys {
            if CDC_FILENAME_MATCH.is_match(&key) {
                files.push(key);
            }
        }
        assert!(!files.is_empty());

        let prefix = CDCPrefix::new(TEST_S3_PREFIX.to_string());
        let result = poller
            .list_unread_files(&CDCFileName::empty(), &prefix)
            .await
            .unwrap();
        assert_eq!(result.len(), files.len());
    }

    #[tokio::test]
    async fn test_list_unread_files_for_partition_table() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let poller = CDCFilePoller::new(backend_storage.clone(), TEST_S3_BUCKET_PARTITION);

        let all_keys = backend_storage
            .list(
                TEST_S3_BUCKET_PARTITION,
                TEST_PARTITION_TABLE_PREFIX.as_str(),
                None,
            )
            .await
            .unwrap();
        assert!(!all_keys.is_empty());

        let prefix = CDCPrefix::new(TEST_PARTITION_TABLE_PREFIX.clone());
        let result = poller
            .list_unread_files(&CDCFileName::empty(), &prefix)
            .await
            .unwrap();

        assert_eq!(result.len(), 4);

        let start_file = result.first().unwrap();
        let end_file = result.last().unwrap();

        let result = poller
            .list_unread_files(&parse_last_file(start_file), &prefix)
            .await
            .unwrap();
        assert_eq!(result.len(), 3);

        let result = poller
            .list_unread_files(&parse_last_file(end_file), &prefix)
            .await
            .unwrap();
        assert_eq!(result.len(), 0);
    }

    #[tokio::test]
    async fn test_list_unread_files_rejects_same_directory_hole() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let last_file = mock_cdc_file(17_429);
        storage.push_list_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            Some(last_file.as_str()),
            Ok(vec![
                mock_cdc_file(17_430),
                mock_cdc_file(17_431),
                mock_cdc_file(17_433),
            ]),
        );

        let err = poller
            .list_unread_files_for_table(&parse_last_file(&last_file), &prefix)
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("Detected CDC file hole within directory")
        );
    }

    #[tokio::test]
    async fn test_list_unread_files_allows_initial_poll_mid_sequence() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let unread_files = vec![
            mock_cdc_file(17_430),
            mock_cdc_file(17_431),
            mock_cdc_file(17_432),
        ];
        storage.push_list_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            None,
            Ok(unread_files.clone()),
        );

        let result = poller
            .list_unread_files_for_table(&CDCFileName::empty(), &prefix)
            .await
            .unwrap();
        assert_eq!(result, unread_files);
    }

    #[tokio::test]
    async fn test_list_unread_files_rejects_new_directory_nonzero_start() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let last_file = mock_cdc_file(17_429);
        let next_table_version = MOCK_TABLE_VERSION + 1;
        storage.push_list_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            Some(last_file.as_str()),
            Ok(vec![mock_cdc_file_with_version(next_table_version, 2)]),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &mock_index_path(),
            Ok(b"CDC00000000000000017429.json\n".to_vec()),
        );

        let err = poller
            .list_unread_files_for_table(&parse_last_file(&last_file), &prefix)
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("New CDC directory does not start from sequence 1")
        );
    }

    #[tokio::test]
    async fn test_list_unread_files_rejects_directory_switch_with_index_mismatch() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let last_file = mock_cdc_file(17_429);
        let next_table_version = MOCK_TABLE_VERSION + 1;
        storage.push_list_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            Some(last_file.as_str()),
            Ok(vec![mock_cdc_file_with_version(next_table_version, 1)]),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &mock_index_path(),
            Ok(b"CDC00000000000000017432.json\n".to_vec()),
        );

        let err = poller
            .list_unread_files_for_table(&parse_last_file(&last_file), &prefix)
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("CDC.index does not match the last visible CDC file")
        );
    }

    #[tokio::test]
    async fn test_list_unread_files_allows_directory_switch_with_matching_index() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let last_file = mock_cdc_file(17_429);
        let next_table_version = MOCK_TABLE_VERSION + 1;
        let unread_files = vec![
            mock_cdc_file_with_version(next_table_version, 1),
            mock_cdc_file_with_version(next_table_version, 2),
        ];
        storage.push_list_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            Some(last_file.as_str()),
            Ok(unread_files.clone()),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &mock_index_path(),
            Ok(b"CDC00000000000000017429.json\n".to_vec()),
        );

        let result = poller
            .list_unread_files_for_table(&parse_last_file(&last_file), &prefix)
            .await
            .unwrap();
        assert_eq!(result, unread_files);
    }

    #[tokio::test]
    async fn test_list_unread_files_returns_only_first_unread_directory() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let last_file = mock_cdc_file(17_429);
        let next_table_version = MOCK_TABLE_VERSION + 1;
        let third_table_version = MOCK_TABLE_VERSION + 2;
        let first_dir_files = vec![
            mock_cdc_file_with_version(next_table_version, 1),
            mock_cdc_file_with_version(next_table_version, 2),
        ];
        let second_dir_files = vec![mock_cdc_file_with_version(third_table_version, 1)];
        storage.push_list_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            Some(last_file.as_str()),
            Ok(vec![
                first_dir_files[0].clone(),
                first_dir_files[1].clone(),
                second_dir_files[0].clone(),
            ]),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &mock_index_path(),
            Ok(b"CDC00000000000000017429.json\n".to_vec()),
        );

        let first_batch = poller
            .list_unread_files_for_table(&parse_last_file(&last_file), &prefix)
            .await
            .unwrap();
        assert_eq!(first_batch, first_dir_files);

        storage.push_list_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            Some(first_batch.last().unwrap().as_str()),
            Ok(second_dir_files.clone()),
        );
        storage.push_object_response(
            MOCK_BUCKET,
            &mock_index_path_with_version(next_table_version),
            Ok(b"CDC00000000000000000002.json\n".to_vec()),
        );

        let second_batch = poller
            .list_unread_files_for_table(&parse_last_file(first_batch.last().unwrap()), &prefix)
            .await
            .unwrap();
        assert_eq!(second_batch, second_dir_files);
    }

    #[tokio::test]
    async fn test_list_unread_files_rejects_missing_index_file() {
        let storage = Arc::new(MockBackendStorage::new());
        let poller = CDCFilePoller::new(storage.clone(), MOCK_BUCKET);
        let prefix = mock_prefix();
        let last_file = mock_cdc_file(17_429);
        let next_table_version = MOCK_TABLE_VERSION + 1;
        storage.push_list_response(
            MOCK_BUCKET,
            prefix.table_prefix(),
            Some(last_file.as_str()),
            Ok(vec![mock_cdc_file_with_version(next_table_version, 1)]),
        );
        for _ in 0..CDC_INDEX_READ_MAX_ATTEMPTS {
            storage.push_object_response(
                MOCK_BUCKET,
                &mock_index_path(),
                Err(StorageError::NotFound("missing index".to_string())),
            );
        }

        let err = poller
            .list_unread_files_for_table(&parse_last_file(&last_file), &prefix)
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("CDC.index is missing after 3 attempts")
        );
    }

    #[tokio::test]
    async fn test_list_unread_files_ignores_numeric_prefix_boundary_collision() {
        let backend_storage = test_create_backend_storage().await.unwrap();
        let poller = CDCFilePoller::new(backend_storage.clone(), TEST_S3_BUCKET);

        let base_prefix = format!(
            "isolated/test/numeric_prefix_boundary/{}/cdc",
            rand::random_range(1_u64..=u64::MAX)
        );
        let table_prefix = format!("{base_prefix}/128");
        let prefix = CDCPrefix::new(table_prefix.clone());
        let table_128_file =
            format!("{table_prefix}/465312161095221285/CDC00000000000000000001.json");
        let table_128_meta_file = format!("{table_prefix}/465312161095221285/meta/CDC.index");
        let wrong_table_file =
            format!("{base_prefix}/1280/465312161095221285/CDC00000000000000000001.json");
        let wrong_table_meta_file = format!("{base_prefix}/1280/465312161095221285/meta/CDC.index");

        // Upload the table 128 files to the storage
        backend_storage
            .put_data(TEST_S3_BUCKET, &table_128_file, b"{\"id\":1}\n".to_vec())
            .await
            .unwrap();
        backend_storage
            .put_data(
                TEST_S3_BUCKET,
                &table_128_meta_file,
                b"CDC00000000000000000001.json".to_vec(),
            )
            .await
            .unwrap();
        // Upload the wrong table 1280 files to the storage
        backend_storage
            .put_data(TEST_S3_BUCKET, &wrong_table_file, b"{\"id\":2}\n".to_vec())
            .await
            .unwrap();
        backend_storage
            .put_data(
                TEST_S3_BUCKET,
                &wrong_table_meta_file,
                b"CDC00000000000000000001.json".to_vec(),
            )
            .await
            .unwrap();

        // List the unread files for the table 128
        let result = poller
            .list_unread_files_for_table(&parse_last_file(&table_128_file), &prefix)
            .await
            .unwrap();
        assert!(result.is_empty(), "unexpected files: {:?}", result);
    }
}