use std::sync::Arc;

use log::{info, warn};

use crate::{
    format_cdc_last_file,
    meta::{
        metrics::SHARD_CACHE_EVENT_COUNTER,
        reader::{AllocateReaderReason, ShardCacheManager, reader_pool::ReaderPool},
        writer::{ShardEvent, ShardEventSubscriberTrait},
    },
};

/// Background event processor for cache operations
pub struct CacheEventProcessor {
    cache_manager: Arc<ShardCacheManager>,
    reader_pool: Arc<ReaderPool>,
}

impl ShardEventSubscriberTrait for CacheEventProcessor {
    async fn on_event(&mut self, event: ShardEvent) {
        match event {
            ShardEvent::FullSync(index_infos, shards) => {
                let index_info_len = index_infos.len();
                let shard_len = shards.len();
                if let Err(e) = self
                    .cache_manager
                    .handle_full_sync(index_infos, shards)
                    .await
                {
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_failed_fullsync"])
                        .inc();
                    warn!(
                        "failed to process full sync event, index_count={}, shards_count={}, err={:?}",
                        index_info_len, shard_len, e
                    );
                } else {
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_suc_fullsync"])
                        .inc();
                    info!("processed full sync event");
                }
            }
            ShardEvent::Add(index_info, shard) => {
                if let Err(e) = self
                    .cache_manager
                    .handle_shard_add(index_info.clone(), &shard)
                    .await
                {
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_failed_add"])
                        .inc();
                    warn!("failed to process shard add event, err={:?}", e);
                } else {
                    self.maybe_schedule_eager_warmup(&index_info, &shard).await;
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_suc_add"])
                        .inc();
                    info!("processed shard add event, shard={}", shard.to_long_ident());
                }
            }
            ShardEvent::Remove(index_info, shard_ids) => {
                let shard_ids_str = shard_ids
                    .iter()
                    .map(|x| format!("shard_id={}", x))
                    .collect::<Vec<_>>()
                    .join(";");
                if let Err(e) = self
                    .cache_manager
                    .handle_shard_remove(index_info, shard_ids.clone(), self.reader_pool.clone())
                    .await
                {
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_failed_remove"])
                        .inc();
                    warn!(
                        "failed to process shard remove event, shard_ids={}, err={:?}",
                        shard_ids_str, e
                    );
                } else {
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_suc_remove"])
                        .inc();
                    info!("processed shard remove event, shard_ids={}", shard_ids_str);
                }
            }
            ShardEvent::Update(index_info, shard) => {
                if let Err(e) = self
                    .cache_manager
                    .handle_shard_update(index_info, &shard)
                    .await
                {
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_failed_update"])
                        .inc();
                    warn!(
                        "failed to process shard update event, shard={}, err={:?}",
                        shard.to_long_ident(),
                        e
                    );
                } else {
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_suc_update"])
                        .inc();
                    info!(
                        "processed shard update event, shard={}, seq={}, delta_len={}",
                        shard.to_long_ident(),
                        shard.seq,
                        shard.manifest.delta_slice().len()
                    );
                }
            }
            ShardEvent::Append(index_info, shard) => {
                if let Err(e) = self
                    .cache_manager
                    .handle_shard_append(index_info, &shard)
                    .await
                {
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_failed_append"])
                        .inc();
                    warn!(
                        "failed to process shard append event, shard_id={}, seq={}, err={:?}",
                        shard.shard_id, shard.new_seq, e
                    );
                } else {
                    SHARD_CACHE_EVENT_COUNTER
                        .with_label_values(&["event_suc_append"])
                        .inc();
                    info!(
                        "processed shard append event, shard_id={}, seq={}, cdc_last_file={}",
                        shard.shard_id,
                        shard.new_seq,
                        format_cdc_last_file(&shard.cdc_s3_last_file)
                    );
                }
            }
            ShardEvent::Split(index, original, new_shards) => {
                info!(
                    "processing shard split event, original={}, new_shards={}",
                    original.to_long_ident(),
                    new_shards
                        .iter()
                        .map(|x| format!("{}", x.shard_id))
                        .collect::<Vec<_>>()
                        .join(","),
                );
                self.cache_manager
                    .handle_shard_split(
                        index.clone(),
                        original.clone(),
                        new_shards,
                        self.reader_pool.clone(),
                    )
                    .await;
                SHARD_CACHE_EVENT_COUNTER
                    .with_label_values(&["event_suc_split"])
                    .inc();
                info!(
                    "processed shard split event, index_key={:?}, original_shard={}",
                    index.index_key(),
                    original.to_long_ident(),
                );
            }
            ShardEvent::Merge(index, old_shards, new_shard) => {
                info!(
                    "processing shard merge event, old_shards={}, new_shard={}",
                    old_shards
                        .iter()
                        .map(|x| format!("{}", x.shard_id))
                        .collect::<Vec<_>>()
                        .join(","),
                    new_shard.to_long_ident(),
                );
                self.cache_manager
                    .handle_shard_merge(
                        index,
                        old_shards,
                        new_shard.clone(),
                        self.reader_pool.clone(),
                    )
                    .await;
                SHARD_CACHE_EVENT_COUNTER
                    .with_label_values(&["event_suc_merge"])
                    .inc();
                info!(
                    "processed shard merge event, new_shard={}",
                    new_shard.to_long_ident(),
                );
            }
            ShardEvent::Stop => {
                info!("CacheEventProcessor: received stop event");
            }
        }
    }
}

impl CacheEventProcessor {
    pub fn new(cache_manager: Arc<ShardCacheManager>, reader_pool: Arc<ReaderPool>) -> Self {
        info!("CacheEventProcessor: starting event processing task");
        Self {
            cache_manager,
            reader_pool,
        }
    }

    async fn maybe_schedule_eager_warmup(
        &self,
        index_info: &crate::meta::writer::ShardEventIndexInfo,
        shard: &crate::ShardInfo,
    ) {
        if !self.cache_manager.config.cache_scheduling.eager_schedule {
            return;
        }

        let Some(unified_shard) = self.cache_manager.get_shard_by_id(&shard.shard_id).await else {
            SHARD_CACHE_EVENT_COUNTER
                .with_label_values(&["warm_f_eager_failed"])
                .inc();
            warn!(
                "skip eager warmup after shard add because shard metadata is missing, shard={}",
                shard.to_long_ident()
            );
            return;
        };

        if self.reader_pool.get_readers_len().await == 0 {
            SHARD_CACHE_EVENT_COUNTER
                .with_label_values(&["warm_f_eager_failed"])
                .inc();
            info!(
                "skip eager warmup after shard add because no reader is available, shard={}",
                unified_shard.to_long_ident()
            );
            return;
        }

        match self.reader_pool.background_warmup(
            &unified_shard,
            self.reader_pool.clone(),
            index_info.table_info.clone(),
            index_info.index_info.clone(),
            AllocateReaderReason::EagerSchedule,
        ) {
            Ok(_) => {
                SHARD_CACHE_EVENT_COUNTER
                    .with_label_values(&["warm_f_suc_eager"])
                    .inc();
            }
            Err(e) => {
                SHARD_CACHE_EVENT_COUNTER
                    .with_label_values(&["warm_f_failed_eager"])
                    .inc();
                warn!(
                    "failed to schedule eager warmup after shard add, shard={}, err={:?}",
                    unified_shard.to_long_ident(),
                    e
                );
            }
        }
    }
}

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

    use tici_common::index::IndexKey;
    use tici_grpc::test_utils::create_test_heartbeat_request;
    use tici_system::readable::ReadableDuration;
    use tici_test_utils::test_utils::{init_logger_for_ut, must_finally};

    use super::*;
    use crate::{
        config::make_default_test_meta_config,
        meta::reader::{
            reader_pool::warmup::{WarmupMgr, WarmupTestAction, WarmupTestController},
            test_utils::{create_test_shard_event_index_info, create_test_shard_info2},
        },
    };

    async fn setup_processor(eager_schedule: bool) -> (CacheEventProcessor, IndexKey) {
        init_logger_for_ut();

        let index_key = IndexKey::new(1, 1, 1);
        let mut config = make_default_test_meta_config(false);
        config.cache_scheduling.eager_schedule = eager_schedule;
        config.reader_pool.ttl_duration = ReadableDuration::secs(10000);

        let warmup_controller = Arc::new(WarmupTestController::new());
        warmup_controller
            .set_fallback_action(Some(WarmupTestAction::Succeed))
            .await;
        let warmup_mgr = WarmupMgr::with_test_controller(warmup_controller);

        let manager = ShardCacheManager::new(&config, false);
        let mut reader_pool = ReaderPool::new_with_warmup_mgr(&config, warmup_mgr);
        reader_pool.init(manager.clone());
        reader_pool
            .register_or_update_reader(&create_test_heartbeat_request(
                "addr0".to_string(),
                "addr0_0".to_string(),
                vec![],
            ))
            .await
            .unwrap();

        (
            CacheEventProcessor::new(manager, Arc::new(reader_pool)),
            index_key,
        )
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_add_event_eager_schedule_warmup() {
        let (mut processor, index_key) = setup_processor(true).await;
        let shard = create_test_shard_info2(&index_key, 1, b"", b"", 1);

        processor
            .on_event(ShardEvent::Add(
                create_test_shard_event_index_info(&index_key),
                shard.clone(),
            ))
            .await;

        must_finally(async || {
            processor
                .cache_manager
                .get_shard_by_id(&shard.shard_id)
                .await
                .unwrap()
                .has_ready_readers()
                .await
        })
        .await;
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_add_event_without_eager_schedule_does_not_warmup() {
        let (mut processor, index_key) = setup_processor(false).await;
        let shard = create_test_shard_info2(&index_key, 1, b"", b"", 1);

        processor
            .on_event(ShardEvent::Add(
                create_test_shard_event_index_info(&index_key),
                shard.clone(),
            ))
            .await;

        let unified_shard = processor
            .cache_manager
            .get_shard_by_id(&shard.shard_id)
            .await
            .unwrap();
        assert!(!unified_shard.has_ready_readers().await);
        assert_eq!(
            unified_shard
                .in_warming
                .load(std::sync::atomic::Ordering::SeqCst),
            0
        );
    }
}