#![cfg_attr(not(test), deny(clippy::unwrap_used))]
#![cfg_attr(not(test), deny(clippy::expect_used))]

use std::sync::Arc;

use anyhow::Result;
use arc_swap::ArcSwap;
use log::info;
use storage_backend::storage::BackendStorage;
use tici_common::index::IndexKey;
use tici_import::ImportJobMeta;
use tici_service_discovery::election::leadership::Leadership;

mod config_change;
mod ddl_msg;
mod import_msg;
mod internal_msg;

use crate::{
    config::MetaServiceConfig,
    connection_pool::TiDBConnectionPool,
    meta::{
        MetaStore, MetaStoreImplTiDB,
        importer::ImportTaskMgr,
        reader::{CacheEventProcessor, reader_pool::ReaderPool, shard_cache::ShardCacheManager},
        writer::{FTSIndex, ShardEventSubscriber, ShardTaskManager, mgr::WorkerNodeMgr},
    },
};
/// Core implementation of meta service.
pub struct MetaServiceCore {
    pub meta_store: Arc<dyn MetaStore>,
    backend_storage: Arc<dyn BackendStorage>,

    pub worker_node_mgr: Arc<WorkerNodeMgr>,
    reader_pool: Arc<ReaderPool>,
    pub shard_cache_manager: Arc<ShardCacheManager>,
    pub shard_task_manager: Arc<ShardTaskManager>,

    pub(super) import_task_mgr: ImportTaskMgr,

    pub config: Arc<ArcSwap<MetaServiceConfig>>,
    tso_provider: ddl_msg::PdTsoProvider,
    _cache_subscriber: ShardEventSubscriber,
}

impl MetaServiceCore {
    pub fn new<L: Leadership + 'static>(
        config: ArcSwap<MetaServiceConfig>,
        tidb_pools: TiDBConnectionPool,
        leadership: L,
        backend_storage: Arc<dyn BackendStorage>,
    ) -> Self {
        let meta_store = Arc::new(MetaStoreImplTiDB::new(tidb_pools, leadership));
        // A task queue for ShardTasks
        let (shard_task_tx, shard_task_rx) = tokio::sync::mpsc::unbounded_channel();
        // A task queue for WorkerNodeTasks
        let (wn_task_tx, wn_task_rx) = tokio::sync::mpsc::unbounded_channel();

        let config_arc = Arc::new(ArcSwap::new(config.load_full()));
        // WorkerNodeMgr, it may
        // - Generate ShardTasks
        let worker_node_mgr = WorkerNodeMgr::new(
            shard_task_tx.clone(),
            wn_task_tx.clone(),
            wn_task_rx,
            config_arc.clone(),
            meta_store.clone(),
        );

        // Create ShardCacheManager first, which owns its own event channel
        let shard_cache_manager = ShardCacheManager::new(config_arc.load().as_ref(), false);

        // ShardTaskManager, it may
        // - Consume ShardTasks
        let shard_task_manager = ShardTaskManager::new(
            meta_store.clone(),
            worker_node_mgr.clone(),
            config_arc.clone(),
            shard_task_tx,
            shard_task_rx,
            backend_storage.clone(),
        );
        // ImportTaskMgr, it may
        // - Generate WorkerNodeTask
        let import_task_mgr = ImportTaskMgr::new(
            config_arc.load().as_ref(),
            meta_store.clone(),
            shard_task_manager.clone(),
            worker_node_mgr.clone(),
            wn_task_tx.clone(),
            backend_storage.clone(),
        );
        let mut reader_pool = ReaderPool::new(config_arc.load().as_ref());
        reader_pool.init(shard_cache_manager.clone());
        let reader_pool = Arc::new(reader_pool);
        let cache_subscriber =
            CacheEventProcessor::new(shard_cache_manager.clone(), reader_pool.clone());
        let cache_subscriber = shard_task_manager
            .event_pubsub
            .resubscribe(cache_subscriber, "read_cache".to_string());

        Self {
            meta_store: meta_store.clone(),
            backend_storage,
            worker_node_mgr: worker_node_mgr.clone(),
            reader_pool: reader_pool.clone(),
            shard_cache_manager,
            shard_task_manager,
            import_task_mgr,
            config: config_arc,
            tso_provider: ddl_msg::PdTsoProvider::new(),
            _cache_subscriber: cache_subscriber,
        }
    }

    // NOTE: This function will start itself as a leader.
    pub async fn start(&self) -> Result<(), anyhow::Error> {
        self.shard_task_manager.restore_from_start().await?;

        // Start reader pool cleanup task
        self.reader_pool.clone().start().await;

        info!("MetaServiceCore started successfully");
        Ok(())
    }

    pub async fn stop(&self) -> Result<(), anyhow::Error> {
        log::info!("stop meta service: shard_task_manager");
        self.shard_task_manager.stop();
        loop {
            if self.shard_task_manager.is_stopped() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
        }
        log::info!("stop meta service: worker_node_mgr");
        self.worker_node_mgr.stop();
        loop {
            if self.worker_node_mgr.is_stopped() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
        }

        self.import_task_mgr.stop();

        log::info!("stop meta service: finished");
        Ok(())
    }

    // NOTE: This is for restore from a node start. However, if
    pub async fn restore_from_start(&self) -> Result<(), anyhow::Error> {
        self.shard_task_manager.restore_from_start().await
    }

    pub async fn get_reader_count(&self) -> usize {
        self.reader_pool.get_all_readers().await.len()
    }

    /// Start the reader pool (exposed for testing)
    pub async fn start_reader_pool(self: &Arc<Self>) {
        let reader_pool = self.reader_pool.clone();
        reader_pool.start().await;
    }

    pub async fn get_index(&self, index_key: &IndexKey) -> Result<Arc<FTSIndex>, anyhow::Error> {
        self.shard_task_manager.get_index(index_key).await
    }

    pub fn get_reader_pool(&self) -> Arc<ReaderPool> {
        self.reader_pool.clone()
    }

    #[cfg(any(test, feature = "testexport"))]
    pub async fn run_import_job_check_once_for_test(&self) -> bool {
        crate::meta::importer::run_import_job_check_once_for_test(
            self.meta_store.clone(),
            self.shard_task_manager.clone(),
            self.backend_storage.clone(),
            self.config.load().s3.clone(),
        )
        .await
    }

    #[cfg(any(test, feature = "testexport"))]
    pub async fn run_import_cleanup_check_once_for_test(&self) -> bool {
        crate::meta::importer::run_import_cleanup_check_once_for_test(
            self.meta_store.clone(),
            self.shard_task_manager.clone(),
            self.backend_storage.clone(),
            self.config.load().s3.clone(),
        )
        .await
    }

    #[cfg(any(test, feature = "testexport"))]
    pub async fn run_import_recovery_once_for_test(&self) -> bool {
        crate::meta::importer::run_import_recovery_once_for_test(
            self.meta_store.clone(),
            self.worker_node_mgr.clone(),
        )
        .await
    }
}