use std::sync::Arc;

use log::info;
use tici::config::MetaServiceConfig;
use tici_search_lib::{
    config::{ReaderNodeConfig, TiCIConfig, TiFlashServerConfig},
    reader_server::ReaderServer,
};
use tici_shard::{
    config::{FragReaderConfig, StorageConfig},
    reader::shard_mirror_manager::ShardMirrorManager,
};
use tici_system::readable::ReadableSize;
use tokio::sync::RwLock;

pub fn make_default_reader_config(
    meta_config: &MetaServiceConfig,
    addr: String,
    tiflash_addr: String,
) -> tici_search_lib::config::Config {
    tici_search_lib::config::Config {
        tici: TiCIConfig {
            s3: meta_config.s3.clone(),
            reader_node: ReaderNodeConfig {
                addr,
                ..Default::default()
            },
            storage: StorageConfig {
                data_dir: format!("tici-reader-data/{}", uuid::Uuid::new_v4()),
            },
            frag_reader: FragReaderConfig {
                disk_reserve_size: ReadableSize::mb(64),
                ..Default::default()
            },
            ..Default::default()
        },
        flash: TiFlashServerConfig {
            service_addr: tiflash_addr,
        },
        pd: tici_search_lib::config::PDConfig {
            pd_addr: meta_config.server.pd_addr.clone(),
        },
        logger: Default::default(),
    }
}

pub struct ReaderNodeHandle {
    pub reader: Arc<RwLock<ReaderServer>>,
    pub tiflash_addr: String,
    grpc_server_close_tx: Option<tokio::sync::oneshot::Sender<()>>,
}

impl ReaderNodeHandle {
    pub async fn new(searchlib_config: tici_search_lib::config::Config) -> Self {
        let shard_mirror_manager =
            ShardMirrorManager::new(searchlib_config.tici.frag_reader.clone());
        let shard_mirror_manager = Arc::new(shard_mirror_manager);
        let reader = ReaderServer::new(&searchlib_config, shard_mirror_manager)
            .await
            .unwrap();
        let reader = Arc::new(RwLock::new(reader));
        let reader_clone = reader.clone();
        let (tx, rx) = tokio::sync::oneshot::channel();
        info!("ReaderNodeHandle: start reader server");
        reader.write().await.start(rx).unwrap();
        info!("ReaderNodeHandle: start heartbeat manager");

        Self {
            reader: reader_clone,
            tiflash_addr: searchlib_config.flash.service_addr.clone(),
            grpc_server_close_tx: Some(tx),
        }
    }

    pub async fn stop(&mut self) {
        info!("ReaderNodeHandle stop heartbeat manager");
        self.reader
            .read()
            .await
            .get_service()
            .heartbeat_manager
            .stop()
            .await;
        info!("ReaderNodeHandle stop server");
        let sender = self.grpc_server_close_tx.take().unwrap();
        self.reader.write().await.shutdown(sender).await.unwrap();
        info!("ReaderNodeHandle stop finished");
    }
}