use std::{
    collections::HashMap,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
};

use anyhow::{Result, bail, ensure};
use intervaltree::IntervalTree;
use log::{info, warn};
use scopeguard::defer;
use tici_common::cdc_file_name::CDCFileName;
use tokio::sync::mpsc;

use crate::{
    utils::{
        key::ShardingKeyRange,
        metrics::{inc_cdc_op_count, start_cdc_op_timer},
        schema::IndexMeta,
    },
    writer::cdc_log::CDCLogEntry,
};

pub struct ShardInfo {
    pub(crate) shard_id: u64,
    pub(crate) range: ShardingKeyRange,
    pub(crate) tracing_id: String,
    pub(crate) index_meta: IndexMeta,

    sender: mpsc::Sender<CDCLogEntry>,
    // Use lock to protect last_file, so that it can be updated independently after the file is
    // processed.
    last_file: parking_lot::Mutex<CDCFileName>,
    // Track if the channel is closed to avoid repeated send attempts.
    closed: AtomicBool,
}

fn entry_type(entry: &CDCLogEntry) -> &'static str {
    match entry {
        CDCLogEntry::CDCData(_) => "cdc_data",
        CDCLogEntry::Filename(_) => "progress",
    }
}

fn send_type(entry_type: &str, s: &str) -> &'static str {
    match entry_type {
        "cdc_data" => match s {
            "ok" => "send_cdc_data_ok",
            "full" => "send_cdc_data_full",
            "closed" => "send_cdc_data_closed",
            _ => "send_cdc_data_unknown",
        },
        "progress" => match s {
            "ok" => "send_progress_ok",
            "full" => "send_progress_full",
            "closed" => "send_progress_closed",
            _ => "send_progress_unknown",
        },
        _ => "send_unknown",
    }
}

fn send_blocked_latency_type(entry_type: &str) -> &'static str {
    match entry_type {
        "cdc_data" => "send_cdc_data_blocked",
        "progress" => "send_progress_blocked",
        _ => "send_unknown_blocked",
    }
}

impl ShardInfo {
    pub fn new(
        shard_id: u64,
        sender: mpsc::Sender<CDCLogEntry>,
        range: ShardingKeyRange,
        tracing_id: String,
        last_file: CDCFileName,
        index_meta: IndexMeta,
    ) -> Result<Self> {
        ensure!(
            !range.is_empty(),
            "Empty shard range is not allowed: shard_id={}, tracing_id={}, range={:?}",
            shard_id,
            tracing_id,
            range
        );
        Ok(Self {
            shard_id,
            sender,
            range,
            tracing_id,
            last_file: parking_lot::Mutex::new(last_file),
            index_meta,
            closed: AtomicBool::new(false),
        })
    }

    /// Check if the channel is closed without attempting to send.
    pub fn is_closed(&self) -> bool {
        self.closed.load(Ordering::Relaxed)
    }

    pub fn should_process_next(&self, prev_file: &CDCFileName) -> bool {
        let last = self.last_file.lock();
        *last == *prev_file
    }

    pub fn update_last_file(&self, new_file: CDCFileName) {
        let mut last = self.last_file.lock();
        *last = new_file;
    }

    pub fn get_last_file(&self) -> CDCFileName {
        self.last_file.lock().clone()
    }

    // If the channel is full, block the send.
    // So the only possible error is when the channel is closed.
    // If the channel is closed, it means the cdc_log_frag_writer is dropped, just
    // ignore the entry, instead of returning an error.
    //
    // Returns true if send succeeded, false if channel is closed.
    pub async fn send(&self, entry: CDCLogEntry) -> (bool, f64) {
        let entry_type = entry_type(&entry);
        let mut blocked_secs = 0.0;
        // Check the result of try_send
        let res = match self.sender.try_send(entry) {
            Ok(()) => {
                inc_cdc_op_count(send_type(entry_type, "ok"));
                return (true, blocked_secs);
            }
            Err(mpsc::error::TrySendError::Full(e)) => {
                inc_cdc_op_count(send_type(entry_type, "full"));
                let blocked_timer = start_cdc_op_timer(send_blocked_latency_type(entry_type));
                defer!(blocked_secs = blocked_timer.stop_and_record());
                warn!("CDC log channel full, blocking send: {}", self.tracing_id);
                self.sender.send(e).await
            }
            Err(mpsc::error::TrySendError::Closed(_)) => {
                inc_cdc_op_count(send_type(entry_type, "closed"));
                info!("CDC log channel closed, ignore entry: {}", self.tracing_id);
                self.closed.store(true, Ordering::Relaxed);
                return (false, blocked_secs);
            }
        };

        // Check the result of send.
        match res {
            Ok(()) => {
                inc_cdc_op_count(send_type(entry_type, "ok"));
                (true, blocked_secs)
            }
            // The only possible error of `send()` is when the channel is closed.
            Err(e) => {
                inc_cdc_op_count(send_type(entry_type, "closed"));
                info!(
                    "CDC log channel closed ({:?}), ignore entry: {}",
                    e, self.tracing_id
                );
                self.closed.store(true, Ordering::Relaxed);
                (false, blocked_secs)
            }
        }
    }
}

pub struct IndexShardInfo {
    pub(crate) index_meta: IndexMeta,
    shard_count: usize,
    shard_infos_by_range: IntervalTree<Vec<u8>, Arc<ShardInfo>>,
}

impl IndexShardInfo {
    fn from_shard_infos(index_meta: IndexMeta, shard_infos: Vec<Arc<ShardInfo>>) -> Self {
        let shard_count = shard_infos.len();
        let shard_infos_by_range = shard_infos
            .into_iter()
            .map(|shard_info| {
                let start = shard_info.range.start().raw().to_vec();
                let end = shard_info.range.end().raw().to_vec();
                assert!(
                    start < end,
                    "shard range must be non-empty: shard_id={}, range={:?}",
                    shard_info.shard_id,
                    shard_info.range
                );
                (start..end, shard_info)
            })
            .collect();
        Self {
            index_meta,
            shard_count,
            shard_infos_by_range,
        }
    }

    pub fn get_shard_infos_by_key(&self, key: &[u8]) -> Vec<&Arc<ShardInfo>> {
        self.shard_infos_by_range
            .query_point(key.to_vec())
            .map(|element| &element.value)
            .collect()
    }

    pub fn get_all_shard_infos(&self) -> impl Iterator<Item = &Arc<ShardInfo>> {
        self.shard_infos_by_range
            .iter()
            .map(|element| &element.value)
    }

    pub fn get_shard_count(&self) -> usize {
        self.shard_count
    }
}

// Internal state that needs to be protected by a single mutex
struct ReaderStateInner {
    // shard_id -> ShardInfo
    shard_infos: HashMap<u64, Arc<ShardInfo>>,
    // Global last processed file
    last_file: CDCFileName,
}

impl ReaderStateInner {
    fn new(last_file: CDCFileName) -> Self {
        Self {
            shard_infos: HashMap::new(),
            last_file,
        }
    }

    fn add_shard_info(&mut self, shard_id: u64, shard_info: Arc<ShardInfo>) {
        let previous = self.shard_infos.insert(shard_id, shard_info.clone());
        match previous {
            None => {
                inc_cdc_op_count("reader_state_add_shard");
                info!(
                    "Added shard {}, range={:?}, last_file={}",
                    shard_info.tracing_id,
                    shard_info.range,
                    shard_info.get_last_file()
                );
            }
            Some(old_shard_info) => {
                inc_cdc_op_count("reader_state_replace_shard");
                info!(
                    "Replaced shard {} (id={}): old_range={:?}, old_last_file={} -> new_range={:?}, new_last_file={}",
                    shard_info.tracing_id,
                    shard_id,
                    old_shard_info.range,
                    old_shard_info.get_last_file(),
                    shard_info.range,
                    shard_info.get_last_file()
                );
            }
        }

        // Update last_file to an older value if the shard's last_file is earlier
        let shard_last_file = shard_info.get_last_file();
        if shard_last_file < self.last_file {
            inc_cdc_op_count("reader_state_rewind_last_file");
            info!(
                "Rewound last file based on {}: {:?} -> {:?}",
                shard_info.tracing_id, self.last_file, shard_last_file
            );
            self.last_file = shard_last_file;
        }
    }

    fn remove_shard_info(&mut self, shard_id: u64) -> Option<Arc<ShardInfo>> {
        let removed = self.shard_infos.remove(&shard_id);
        if removed.is_some() {
            inc_cdc_op_count("reader_state_remove_shard");
            let recomputed_last_file = self
                .shard_infos
                .values()
                .map(|shard_info| shard_info.get_last_file())
                .min()
                .unwrap_or_else(CDCFileName::empty);
            if recomputed_last_file != self.last_file {
                inc_cdc_op_count("reader_state_recompute_last_file");
                info!(
                    "Recomputed last file after removing shard {}: {:?} -> {:?}",
                    shard_id, self.last_file, recomputed_last_file
                );
                self.last_file = recomputed_last_file;
            }
        }
        removed
    }

    fn is_shard_infos_empty(&self) -> bool {
        self.shard_infos.is_empty()
    }

    // Get a snapshot of shard processing state for the next file after prev_file.
    fn get_shards_for_processing(&self, prev_cdc_filename: &CDCFileName) -> Vec<IndexShardInfo> {
        let mut grouped_shards = HashMap::<i64, (IndexMeta, Vec<Arc<ShardInfo>>)>::new();
        for (&_, shard_info) in self.shard_infos.iter() {
            if shard_info.should_process_next(prev_cdc_filename) {
                let index_id = shard_info.index_meta.index_key().index_id;
                grouped_shards
                    .entry(index_id)
                    .or_insert_with(|| (shard_info.index_meta.clone(), Vec::new()))
                    .1
                    .push(shard_info.clone());
            }
        }

        grouped_shards
            .into_iter()
            .map(|(_, (index_meta, shard_infos))| {
                IndexShardInfo::from_shard_infos(index_meta, shard_infos)
            })
            .collect()
    }

    // Ensure all shards have processed up to processed_file, then update last_file
    // to processed_file.
    //
    // Returns true if processing can continue (no stale shards), false if the loop
    // should break to re-list files because there are stale shards.
    fn check_and_update_last_file(&mut self, processed_file: &CDCFileName) -> Result<bool> {
        // Sanity check: processed_file should be after last_file
        if self.last_file >= *processed_file {
            bail!(
                "Processed file {} is not after last_file {}",
                processed_file,
                self.last_file
            );
        }
        // Check if any shard is still behind processed_file.
        // If any shard is behind, we should skip the processing loop and list files for
        // these behind shards.
        let has_stale_shards = self
            .shard_infos
            .iter()
            .any(|(_, shard_info)| shard_info.get_last_file() < *processed_file);
        let can_advance = !has_stale_shards;
        if can_advance {
            inc_cdc_op_count("reader_state_advance_last_file");
            // All shards have moved forward, advance last_file to processed_file.
            self.last_file = processed_file.clone();
        } else {
            inc_cdc_op_count("reader_state_found_stale_shard");
        }
        Ok(can_advance)
    }
}

// Thread-safe wrapper around ReaderState
pub struct ReaderState {
    state: parking_lot::Mutex<ReaderStateInner>,
}

impl ReaderState {
    pub fn new(last_file: CDCFileName) -> Self {
        let state = parking_lot::Mutex::new(ReaderStateInner::new(last_file));
        Self { state }
    }

    pub fn add_shard_info(&self, shard_id: u64, shard_info: Arc<ShardInfo>) {
        self.state.lock().add_shard_info(shard_id, shard_info)
    }
    pub fn remove_shard_info(&self, shard_id: u64) -> Option<Arc<ShardInfo>> {
        self.state.lock().remove_shard_info(shard_id)
    }

    pub fn is_shard_infos_empty(&self) -> bool {
        self.state.lock().is_shard_infos_empty()
    }

    pub fn get_shards_for_processing(
        &self,
        prev_cdc_filename: &CDCFileName,
    ) -> Vec<IndexShardInfo> {
        self.state
            .lock()
            .get_shards_for_processing(prev_cdc_filename)
    }

    pub fn check_and_update_last_file(&self, processed_file: &CDCFileName) -> Result<bool> {
        self.state.lock().check_and_update_last_file(processed_file)
    }

    pub fn has_table(&self, table_id: i64) -> bool {
        self.state
            .lock()
            .shard_infos
            .values()
            .any(|info| info.index_meta.index_key().table_id == table_id)
    }

    pub fn get_last_file(&self) -> CDCFileName {
        self.state.lock().last_file.clone()
    }

    // Helper methods for tests (used by CDCFileReader tests)
    #[cfg(test)]
    pub fn get_shard_info(&self, shard_id: u64) -> Option<Arc<ShardInfo>> {
        self.state.lock().shard_infos.get(&shard_id).cloned()
    }

    #[cfg(test)]
    pub fn get_shard_infos(&self) -> HashMap<u64, Arc<ShardInfo>> {
        self.state.lock().shard_infos.clone()
    }

    #[cfg(test)]
    pub fn contains_shard(&self, shard_id: u64) -> bool {
        self.state.lock().shard_infos.contains_key(&shard_id)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_utils::{TEST_HDFSLOGS_INDEX_META, TEST_INF_END, TEST_INF_START};

    fn mock_cdc_file_name() -> (String, String) {
        (
            "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000001.json"
                .to_string(),
            "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000002.json"
                .to_string(),
        )
    }

    fn mock_cdc_file_name_reversed() -> (String, String) {
        (
            "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000003.json"
                .to_string(),
            "tici_default_prefix/cdc/10313/461869219056451591/CDC00000000000000000002.json"
                .to_string(),
        )
    }

    fn mock_shard_info(shard_id: u64, start: &[u8], end: &[u8], last_file: &str) -> Arc<ShardInfo> {
        let (tx, _) = mpsc::channel(1);
        Arc::new(
            ShardInfo::new(
                shard_id,
                tx,
                ShardingKeyRange::half_open(start.to_vec(), end.to_vec()).unwrap(),
                format!("shard_{shard_id}"),
                CDCFileName::parse(last_file).unwrap(),
                TEST_HDFSLOGS_INDEX_META.clone(),
            )
            .unwrap(),
        )
    }

    fn shard_ids(shard_infos: Vec<&Arc<ShardInfo>>) -> Vec<u64> {
        let mut shard_ids = shard_infos
            .into_iter()
            .map(|shard_info| shard_info.shard_id)
            .collect::<Vec<_>>();
        shard_ids.sort_unstable();
        shard_ids
    }

    async fn _test_reader_state_basic_operations(
        (cdc_file_name1, _cdc_file_name2): (String, String),
    ) {
        let state = ReaderStateInner::new(CDCFileName::parse(&cdc_file_name1).unwrap());

        assert!(state.is_shard_infos_empty());
        assert_eq!(
            state.last_file,
            CDCFileName::parse(&cdc_file_name1).unwrap()
        );
    }

    #[tokio::test]
    async fn test_reader_state_basic_operations() {
        _test_reader_state_basic_operations(mock_cdc_file_name()).await;
    }

    async fn _test_reader_state_add_shard_updates_last_file(
        (cdc_file_name1, cdc_file_name2): (String, String),
    ) {
        let mut state = ReaderStateInner::new(CDCFileName::parse(&cdc_file_name1).unwrap());

        let (tx, _) = mpsc::channel(1);
        let shard_info = Arc::new(
            ShardInfo::new(
                1,
                tx,
                ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec())
                    .unwrap(),
                "test_shard".to_string(),
                CDCFileName::parse(&cdc_file_name2).unwrap(),
                TEST_HDFSLOGS_INDEX_META.clone(),
            )
            .unwrap(),
        );

        state.add_shard_info(1, shard_info);

        // last_file should be updated to the older file from the shard
        assert_eq!(
            state.last_file,
            CDCFileName::parse(&cdc_file_name2).unwrap()
        );
    }

    #[tokio::test]
    async fn test_reader_state_add_shard_updates_last_file() {
        _test_reader_state_add_shard_updates_last_file(mock_cdc_file_name_reversed()).await;
    }

    async fn _test_reader_state_add_shard_no_update_if_newer(
        (cdc_file_name1, cdc_file_name2): (String, String),
    ) {
        let mut state = ReaderStateInner::new(CDCFileName::parse(&cdc_file_name1).unwrap());

        let (tx, _) = mpsc::channel(1);
        let shard_info = Arc::new(
            ShardInfo::new(
                1,
                tx,
                ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec())
                    .unwrap(),
                "test_shard".to_string(),
                CDCFileName::parse(&cdc_file_name2).unwrap(),
                TEST_HDFSLOGS_INDEX_META.clone(),
            )
            .unwrap(),
        );

        state.add_shard_info(1, shard_info);

        // last_file should remain unchanged
        assert_eq!(
            state.last_file,
            CDCFileName::parse(&cdc_file_name1).unwrap()
        );
    }

    #[tokio::test]
    async fn test_reader_state_add_shard_no_update_if_newer() {
        _test_reader_state_add_shard_no_update_if_newer(mock_cdc_file_name()).await;
    }

    async fn _test_reader_state_remove_shard_recomputes_last_file(
        (cdc_file_name1, cdc_file_name2): (String, String),
    ) {
        let mut state = ReaderStateInner::new(CDCFileName::parse(&cdc_file_name2).unwrap());

        let (tx1, _) = mpsc::channel(1);
        let shard1 = Arc::new(
            ShardInfo::new(
                1,
                tx1,
                ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec())
                    .unwrap(),
                "test_shard_1".to_string(),
                CDCFileName::parse(&cdc_file_name1).unwrap(),
                TEST_HDFSLOGS_INDEX_META.clone(),
            )
            .unwrap(),
        );
        state.add_shard_info(1, shard1);

        let (tx2, _) = mpsc::channel(1);
        let shard2 = Arc::new(
            ShardInfo::new(
                2,
                tx2,
                ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec())
                    .unwrap(),
                "test_shard_2".to_string(),
                CDCFileName::parse(&cdc_file_name2).unwrap(),
                TEST_HDFSLOGS_INDEX_META.clone(),
            )
            .unwrap(),
        );
        state.add_shard_info(2, shard2);

        assert_eq!(
            state.last_file,
            CDCFileName::parse(&cdc_file_name1).unwrap()
        );

        state.remove_shard_info(1);
        assert_eq!(
            state.last_file,
            CDCFileName::parse(&cdc_file_name2).unwrap()
        );

        state.remove_shard_info(2);
        assert_eq!(state.last_file, CDCFileName::empty());
    }

    #[tokio::test]
    async fn test_reader_state_remove_shard_recomputes_last_file() {
        _test_reader_state_remove_shard_recomputes_last_file(mock_cdc_file_name()).await;
    }

    #[test]
    fn test_shard_info_new_rejects_empty_range() {
        let (tx, _) = mpsc::channel(1);
        let err = ShardInfo::new(
            1,
            tx,
            ShardingKeyRange::half_open(vec![], vec![]).unwrap(),
            "test_shard".to_string(),
            CDCFileName::empty(),
            TEST_HDFSLOGS_INDEX_META.clone(),
        )
        .err()
        .unwrap();

        assert!(err.to_string().contains("Empty shard range is not allowed"));
    }

    async fn _test_reader_state_get_shards_for_processing(
        (cdc_file_name1, cdc_file_name2): (String, String),
    ) {
        let mut state = ReaderStateInner::new(CDCFileName::parse(&cdc_file_name1).unwrap());

        let prev_file = CDCFileName::parse(&cdc_file_name1).unwrap();
        let next_file = CDCFileName::parse(&cdc_file_name2).unwrap();

        // Add shard 1 with last_file = prev_file
        let (tx1, _) = mpsc::channel(1);
        let shard1 = Arc::new(
            ShardInfo::new(
                1,
                tx1,
                ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec())
                    .unwrap(),
                "shard1".to_string(),
                prev_file.clone(),
                TEST_HDFSLOGS_INDEX_META.clone(),
            )
            .unwrap(),
        );
        state.add_shard_info(1, shard1);

        // Add shard 2 with last_file = next_file
        let (tx2, _) = mpsc::channel(1);
        let shard2 = Arc::new(
            ShardInfo::new(
                2,
                tx2,
                ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec())
                    .unwrap(),
                "shard2".to_string(),
                next_file.clone(),
                TEST_HDFSLOGS_INDEX_META.clone(),
            )
            .unwrap(),
        );
        state.add_shard_info(2, shard2);

        // Only shard1 should be ready to process next file after prev_file
        let shards = state.get_shards_for_processing(&prev_file);
        assert_eq!(shards.len(), 1);
        let index_shard_info = shards.first().unwrap();
        let all_shard_infos: Vec<_> = index_shard_info.get_all_shard_infos().collect();
        assert_eq!(all_shard_infos.len(), 1);
        assert_eq!(all_shard_infos[0].shard_id, 1);
        assert!(
            all_shard_infos
                .iter()
                .all(|shard_info| shard_info.shard_id != 2)
        );
    }

    #[tokio::test]
    async fn test_reader_state_get_shards_for_processing() {
        _test_reader_state_get_shards_for_processing(mock_cdc_file_name()).await;
    }

    async fn _test_reader_state_check_and_update_last_file_success(
        (cdc_file_name1, cdc_file_name2): (String, String),
    ) {
        let mut state = ReaderStateInner::new(CDCFileName::parse(&cdc_file_name1).unwrap());

        let processed_file = CDCFileName::parse(&cdc_file_name2).unwrap();

        // Add shard with last_file = processed_file
        let (tx, _) = mpsc::channel(1);
        let shard = Arc::new(
            ShardInfo::new(
                1,
                tx,
                ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec())
                    .unwrap(),
                "shard1".to_string(),
                processed_file.clone(),
                TEST_HDFSLOGS_INDEX_META.clone(),
            )
            .unwrap(),
        );
        state.add_shard_info(1, shard);

        // All shards are at processed_file, so update should succeed
        let result = state.check_and_update_last_file(&processed_file).unwrap();
        assert!(result); // Should return true
        assert_eq!(state.last_file, processed_file);
    }

    #[tokio::test]
    async fn test_reader_state_check_and_update_last_file_success() {
        _test_reader_state_check_and_update_last_file_success(mock_cdc_file_name()).await;
    }

    async fn _test_reader_state_check_and_update_last_file_has_stale(
        (cdc_file_name1, cdc_file_name2): (String, String),
    ) {
        let mut state = ReaderStateInner::new(CDCFileName::parse(&cdc_file_name1).unwrap());

        let old_file = CDCFileName::parse(&cdc_file_name1).unwrap();
        let processed_file = CDCFileName::parse(&cdc_file_name2).unwrap();

        // Add shard with last_file < processed_file
        let (tx, _) = mpsc::channel(1);
        let shard = Arc::new(
            ShardInfo::new(
                1,
                tx,
                ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec())
                    .unwrap(),
                "shard1".to_string(),
                old_file.clone(),
                TEST_HDFSLOGS_INDEX_META.clone(),
            )
            .unwrap(),
        );
        state.add_shard_info(1, shard);

        // Some shards are still behind, so update should not happen
        let result = state.check_and_update_last_file(&processed_file).unwrap();
        assert!(!result); // Should return false (has stale)
        assert_eq!(state.last_file, old_file); // Should remain unchanged
    }

    #[tokio::test]
    async fn test_reader_state_check_and_update_last_file_has_stale() {
        _test_reader_state_check_and_update_last_file_has_stale(mock_cdc_file_name()).await;
    }

    async fn _test_reader_state_check_and_update_last_file_error(
        (cdc_file_name1, cdc_file_name2): (String, String),
    ) {
        let mut state = ReaderStateInner::new(CDCFileName::parse(&cdc_file_name2).unwrap());

        let old_file = CDCFileName::parse(&cdc_file_name1).unwrap();

        // Try to update with file that's not after last_file - should fail
        let result = state.check_and_update_last_file(&old_file);
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_reader_state_check_and_update_last_file_error() {
        _test_reader_state_check_and_update_last_file_error(mock_cdc_file_name()).await;
    }

    async fn _test_shard_info_should_process_next(
        (cdc_file_name1, cdc_file_name2): (String, String),
    ) {
        let (tx, _) = mpsc::channel(1);
        let last_file = CDCFileName::parse(&cdc_file_name1).unwrap();
        let shard_info = ShardInfo::new(
            1,
            tx,
            ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec()).unwrap(),
            "test_shard".to_string(),
            last_file.clone(),
            TEST_HDFSLOGS_INDEX_META.clone(),
        )
        .unwrap();

        // Should process next file when prev_file matches last_file
        assert!(shard_info.should_process_next(&last_file));

        // Should not process when prev_file is different
        let other_file = CDCFileName::parse(&cdc_file_name2).unwrap();
        assert!(!shard_info.should_process_next(&other_file));
    }

    #[tokio::test]
    async fn test_shard_info_should_process_next() {
        _test_shard_info_should_process_next(mock_cdc_file_name()).await;
    }

    async fn _test_shard_info_update_last_file((cdc_file_name1, cdc_file_name2): (String, String)) {
        let (tx, _) = mpsc::channel(1);
        let initial_file = CDCFileName::parse(&cdc_file_name1).unwrap();
        let new_file = CDCFileName::parse(&cdc_file_name2).unwrap();

        let shard_info = ShardInfo::new(
            1,
            tx,
            ShardingKeyRange::half_open(TEST_INF_START.to_vec(), TEST_INF_END.to_vec()).unwrap(),
            "test_shard".to_string(),
            initial_file.clone(),
            TEST_HDFSLOGS_INDEX_META.clone(),
        )
        .unwrap();

        assert_eq!(shard_info.get_last_file(), initial_file);

        shard_info.update_last_file(new_file.clone());
        assert_eq!(shard_info.get_last_file(), new_file);
    }

    #[tokio::test]
    async fn test_shard_info_update_last_file() {
        _test_shard_info_update_last_file(mock_cdc_file_name()).await;
    }

    #[test]
    fn test_index_shard_info_get_shard_infos_by_key_with_overlap() {
        let (last_file, _) = mock_cdc_file_name();
        let index_shard_info = IndexShardInfo::from_shard_infos(
            TEST_HDFSLOGS_INDEX_META.clone(),
            vec![
                mock_shard_info(1, b"a", b"m", &last_file),
                mock_shard_info(2, b"f", b"t", &last_file),
                mock_shard_info(3, b"t", b"z", &last_file),
            ],
        );

        assert_eq!(
            shard_ids(index_shard_info.get_shard_infos_by_key(b"b")),
            vec![1]
        );
        assert_eq!(
            shard_ids(index_shard_info.get_shard_infos_by_key(b"f")),
            vec![1, 2]
        );
        assert_eq!(
            shard_ids(index_shard_info.get_shard_infos_by_key(b"s")),
            vec![2]
        );
        assert_eq!(
            shard_ids(index_shard_info.get_shard_infos_by_key(b"t")),
            vec![3]
        );
        assert!(index_shard_info.get_shard_infos_by_key(b"z").is_empty());
    }
}