use std::{net::SocketAddr, pin::Pin, sync::Arc};

use anyhow::Context;
use arc_swap::ArcSwap;
use async_trait::async_trait;
use log::{info, warn};
use nix::NixPath;
use storage_backend::storage::{STORAGE_SERVICE_META, create_backend_storage};
use tici_common::config::GrpcConfig;
use tici_grpc::{
    AppendBaseFragMetaRequest, AppendDeltaFragMetaRequest, AppendFragMetaResponse,
    CreateIndexRequest, CreateIndexResponse, DebugGetShardInfoRequest, DebugGetShardInfoResponse,
    DebugGetShardManifestRequest, DebugGetShardManifestResponse, DropIndexRequest,
    DropIndexResponse, ErrorCode, FinishCompactFragRequest, FinishCompactFragResponse,
    FinishImportPartitionBuildRequest, GetIndexMetaRequest, GetIndexMetaResponse,
    GetIndexProgressRequest, GetIndexProgressResponse, GetShardLocalCacheRequest,
    GetShardLocalCacheResponse, GetShardManifestRequest, GetShardManifestResponse, GrpcMetaService,
    GrpcMetaServiceClient, GrpcMetaServiceInternal, GrpcMetaServiceInternalClient,
    GrpcMetaServiceInternalServer, GrpcMetaServiceServer, PreSplitImportShardsRequest,
    PreSplitImportShardsResponse, ReaderNodeHeartbeatRequest, ReaderNodeHeartbeatResponse,
    WorkerNodeHeartbeatRequest, WorkerNodeHeartbeatResponse, metrics::begin_grpc_handle_metrics,
    prelude_import::*,
};
use tici_service_discovery::{
    EtcdClient, EtcdDiscoveryClient, EtcdElectionClient, META_SERVICE_ELECTION_KEY,
    META_SERVICE_REGISTER_KEY_PREFIX,
    discovery::client::DiscoveryClient,
    election::leadership::{EtcdLeadership, Leadership, LeadershipObserver},
};
use tokio::sync::RwLock;
use tokio_stream::Stream;
use tonic::{Request, Response, Status, transport::Server};

use crate::{
    config::MetaServiceConfig,
    connection_pool::TiDBConnectionPool,
    meta::{Bootstrap, MetaServiceCore, metrics::update_meta_leader_metric},
    status_server::StatusServer,
};

const NOT_LEADER_ERROR_MESSAGE: &str = "meta service is not leader";

trait NotLeaderRpcResponse {
    fn not_leader() -> Self;
}

impl NotLeaderRpcResponse for CreateIndexResponse {
    fn not_leader() -> Self {
        Self {
            status: ErrorCode::NotLeader.into(),
            error_message: NOT_LEADER_ERROR_MESSAGE.to_string(),
            ..Default::default()
        }
    }
}

impl NotLeaderRpcResponse for DropIndexResponse {
    fn not_leader() -> Self {
        Self {
            status: ErrorCode::NotLeader.into(),
            error_message: NOT_LEADER_ERROR_MESSAGE.to_string(),
        }
    }
}

impl NotLeaderRpcResponse for GetImportStoragePrefixResponse {
    fn not_leader() -> Self {
        Self {
            status: ErrorCode::NotLeader.into(),
            error_message: NOT_LEADER_ERROR_MESSAGE.to_string(),
            ..Default::default()
        }
    }
}

impl NotLeaderRpcResponse for FinishImportResponse {
    fn not_leader() -> Self {
        Self {
            status: ErrorCode::NotLeader.into(),
            error_message: NOT_LEADER_ERROR_MESSAGE.to_string(),
        }
    }
}

impl NotLeaderRpcResponse for PreSplitImportShardsResponse {
    fn not_leader() -> Self {
        Self {
            status: ErrorCode::NotLeader.into(),
            error_message: NOT_LEADER_ERROR_MESSAGE.to_string(),
            ..Default::default()
        }
    }
}

impl NotLeaderRpcResponse for ReaderNodeHeartbeatResponse {
    fn not_leader() -> Self {
        Self {
            status: ErrorCode::NotLeader.into(),
            error_message: NOT_LEADER_ERROR_MESSAGE.to_string(),
            ..Default::default()
        }
    }
}

impl NotLeaderRpcResponse for WorkerNodeHeartbeatResponse {
    fn not_leader() -> Self {
        Self {
            status: ErrorCode::NotLeader.into(),
            ..Default::default()
        }
    }
}

impl NotLeaderRpcResponse for AppendFragMetaResponse {
    fn not_leader() -> Self {
        Self {
            status: ErrorCode::NotLeader.into(),
        }
    }
}

impl NotLeaderRpcResponse for FinishCompactFragResponse {
    fn not_leader() -> Self {
        Self {
            status: ErrorCode::NotLeader.into(),
        }
    }
}

fn is_meta_leader(meta_service: &MetaServiceCore) -> bool {
    meta_service
        .meta_store
        .get_leadership()
        .try_get_leader_term()
        .is_some()
}

struct MetaServiceInternalGrpcImpl {
    meta_service: Arc<MetaServiceCore>,
}

unsafe impl Send for MetaServiceInternalGrpcImpl {}
unsafe impl Sync for MetaServiceInternalGrpcImpl {}

impl MetaServiceInternalGrpcImpl {
    pub fn new(meta_service: Arc<MetaServiceCore>) -> Self {
        Self { meta_service }
    }

    fn reject_if_not_leader<R: NotLeaderRpcResponse>(&self, rpc_name: &str) -> Option<Response<R>> {
        if is_meta_leader(&self.meta_service) {
            return None;
        }
        warn!("{rpc_name}: rejected because this meta service is not leader");
        Some(Response::new(R::not_leader()))
    }
}

#[async_trait]
impl GrpcMetaServiceInternal for MetaServiceInternalGrpcImpl {
    type WorkerNodeHeartbeatStream =
        Pin<Box<dyn Stream<Item = Result<WorkerNodeHeartbeatResponse, Status>> + Send>>;

    async fn reader_node_heartbeat(
        &self,
        request: Request<ReaderNodeHeartbeatRequest>,
    ) -> Result<Response<ReaderNodeHeartbeatResponse>, Status> {
        if let Some(resp) = self.reject_if_not_leader("reader_node_heartbeat") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta_internal", "reader_node_heartbeat");
        let req = request.into_inner();
        match self.meta_service.report_reader_node_status(req).await {
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("reader_node_heartbeat: caught err={:?}", e);
                Ok(Response::new(ReaderNodeHeartbeatResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    error_message: e.to_string(),
                    shards_to_keep: vec![],
                    shards_to_remove: vec![],
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    async fn worker_node_heartbeat(
        &self,
        request: Request<WorkerNodeHeartbeatRequest>,
    ) -> Result<Response<Self::WorkerNodeHeartbeatStream>, Status> {
        if !is_meta_leader(&self.meta_service) {
            warn!("worker_node_heartbeat: rejected because this meta service is not leader");
            let stream = tokio_stream::iter(vec![Ok(WorkerNodeHeartbeatResponse {
                status: ErrorCode::NotLeader.into(),
                ..Default::default()
            })]);
            return Ok(Response::new(Box::pin(stream)));
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta_internal", "worker_node_heartbeat");
        let req = request.into_inner();
        match self.meta_service.report_worker_node_status(req).await {
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("worker_node_heartbeat: caught err={:?}", e);
                let stream = tokio_stream::iter(vec![Ok(WorkerNodeHeartbeatResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    ..Default::default()
                })]);
                Ok(Response::new(Box::pin(stream)))
            }
            Ok(responses) => {
                let stream = tokio_stream::iter(responses.into_iter().map(Ok));
                Ok(Response::new(Box::pin(stream)))
            }
        }
    }

    async fn append_delta_frag_meta(
        &self,
        request: Request<AppendDeltaFragMetaRequest>,
    ) -> Result<tonic::Response<AppendFragMetaResponse>, tonic::Status> {
        if let Some(resp) = self.reject_if_not_leader("append_delta_frag_meta") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta_internal", "append_delta_frag_meta");
        match self
            .meta_service
            .worker_node_mgr
            .on_msg_append_delta_frag_meta(request.get_ref())
            .await
        {
            Err(e) => {
                rpc_metrics.mark_error();
                log::error!(
                    "append_delta_frag_meta: caught error: err={:?}, req={:?}",
                    e,
                    request.get_ref()
                );
                Ok(Response::new(AppendFragMetaResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    async fn append_base_frag_meta(
        &self,
        request: Request<AppendBaseFragMetaRequest>,
    ) -> Result<tonic::Response<AppendFragMetaResponse>, tonic::Status> {
        if let Some(resp) = self.reject_if_not_leader("append_base_frag_meta") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta_internal", "append_base_frag_meta");
        match self
            .meta_service
            .shard_task_manager
            .on_msg_append_base_frag_meta(request.get_ref())
            .await
        {
            Err(e) => {
                rpc_metrics.mark_error();
                log::error!(
                    "append_base_frag_meta: caught error: err={:?}, req={:?}",
                    e,
                    request
                );
                Ok(Response::new(AppendFragMetaResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    async fn finish_compact_fragments(
        &self,
        request: Request<FinishCompactFragRequest>,
    ) -> Result<Response<FinishCompactFragResponse>, Status> {
        if let Some(resp) = self.reject_if_not_leader("finish_compact_fragments") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta_internal", "finish_compact_fragments");
        let req = request.into_inner();
        match self.meta_service.finish_compact_fragments(&req).await {
            Err(e) => {
                rpc_metrics.mark_error();
                log::warn!(
                    "finish_compact_fragments: caught error: err={}, req={:?}",
                    e,
                    req
                );
                Ok(Response::new(FinishCompactFragResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    // TODO(P1) This API should not request writer path, make it at least to request
    // from shard cache.
    async fn debug_get_shard_manifest(
        &self,
        request: Request<DebugGetShardManifestRequest>,
    ) -> Result<tonic::Response<DebugGetShardManifestResponse>, tonic::Status> {
        let rpc_metrics = begin_grpc_handle_metrics("meta_internal", "debug_get_shard_manifest");
        match self
            .meta_service
            .shard_task_manager
            .debug_get_shard_manifest(request.get_ref())
            .await
        {
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("debug_get_shard_manifest: caught err={:?}", e);
                Ok(Response::new(DebugGetShardManifestResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    ..Default::default()
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    async fn get_shard_manifest(
        &self,
        request: Request<GetShardManifestRequest>,
    ) -> Result<tonic::Response<GetShardManifestResponse>, tonic::Status> {
        let rpc_metrics = begin_grpc_handle_metrics("meta_internal", "get_shard_manifest");
        match self
            .meta_service
            .shard_cache_manager
            .on_msg_get_shard_manifest(request.get_ref())
            .await
        {
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("get_shard_manifest: caught err={:?}", e);
                Ok(Response::new(GetShardManifestResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    ..Default::default()
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    async fn finish_import_partition_build(
        &self,
        request: Request<FinishImportPartitionBuildRequest>,
    ) -> Result<tonic::Response<FinishImportResponse>, tonic::Status> {
        if let Some(resp) = self.reject_if_not_leader("finish_import_partition_build") {
            return Ok(resp);
        }
        match self
            .meta_service
            .finish_partition_build(request.get_ref())
            .await
        {
            Ok(resp) => Ok(Response::new(resp)),
            Err(e) => {
                warn!("finish_partition_build: caught err={:?}", e);
                Ok(Response::new(FinishImportResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    error_message: e.to_string(),
                }))
            }
        }
    }

    async fn debug_get_shard_info(
        &self,
        request: Request<DebugGetShardInfoRequest>,
    ) -> Result<tonic::Response<DebugGetShardInfoResponse>, tonic::Status> {
        let rpc_metrics = begin_grpc_handle_metrics("meta_internal", "debug_get_shard_info");
        match self
            .meta_service
            .shard_task_manager
            .debug_get_shard_info(request.get_ref())
            .await
        {
            Ok(resp) => Ok(Response::new(resp)),
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("debug_get_shard_info: caught err={:?}", e);
                Ok(Response::new(DebugGetShardInfoResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    shards: vec![],
                }))
            }
        }
    }

    async fn get_index_meta(
        &self,
        request: Request<GetIndexMetaRequest>,
    ) -> Result<tonic::Response<GetIndexMetaResponse>, tonic::Status> {
        let rpc_metrics = begin_grpc_handle_metrics("meta_internal", "get_index_meta");
        match self.meta_service.get_index_meta(request.get_ref()).await {
            Err(e) => {
                rpc_metrics.mark_error();
                log::error!("get_index_meta: caught err={:?}, req={:?}", e, request);
                Ok(Response::new(GetIndexMetaResponse {
                    status: ErrorCode::UnknownError.into(),
                    error_message: e.to_string(),
                    table_info: None,
                    index_info: None,
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }
}

struct MetaServiceGrpcImpl {
    meta_service: Arc<MetaServiceCore>,
}

unsafe impl Send for MetaServiceGrpcImpl {}
unsafe impl Sync for MetaServiceGrpcImpl {}

impl MetaServiceGrpcImpl {
    pub fn new(meta_service: Arc<MetaServiceCore>) -> Self {
        Self { meta_service }
    }

    fn reject_if_not_leader<R: NotLeaderRpcResponse>(&self, rpc_name: &str) -> Option<Response<R>> {
        if is_meta_leader(&self.meta_service) {
            return None;
        }
        warn!("{rpc_name}: rejected because this meta service is not leader");
        Some(Response::new(R::not_leader()))
    }
}

#[async_trait]
impl GrpcMetaService for MetaServiceGrpcImpl {
    async fn create_index(
        &self,
        request: Request<CreateIndexRequest>,
    ) -> Result<tonic::Response<CreateIndexResponse>, tonic::Status> {
        if let Some(resp) = self.reject_if_not_leader("create_index") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta", "create_index");
        match self.meta_service.create_index(request.get_ref()).await {
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("create_index: caught err={:?}, request={:?}", e, request);
                Ok(Response::new(CreateIndexResponse {
                    status: -1,
                    error_message: e.to_string(),
                    index_id: String::new(),
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    async fn drop_index(
        &self,
        request: Request<DropIndexRequest>,
    ) -> Result<tonic::Response<DropIndexResponse>, tonic::Status> {
        if let Some(resp) = self.reject_if_not_leader("drop_index") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta", "drop_index");
        match self.meta_service.drop_index(request.get_ref()).await {
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("drop_index: caught err={:?}", e);
                Ok(Response::new(DropIndexResponse {
                    status: -1,
                    error_message: e.to_string(),
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    async fn get_index_progress(
        &self,
        request: Request<GetIndexProgressRequest>,
    ) -> Result<tonic::Response<GetIndexProgressResponse>, tonic::Status> {
        let rpc_metrics = begin_grpc_handle_metrics("meta", "get_index_progress");
        match self
            .meta_service
            .get_index_progress(request.get_ref())
            .await
        {
            Err(e) => {
                rpc_metrics.mark_error();
                log::error!("get_index_progress: caught err={:?}, req={:?}", e, request);
                Ok(Response::new(GetIndexProgressResponse {
                    status: -1,
                    error_message: e.to_string(),
                    ..Default::default()
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    async fn get_shard_local_cache_info(
        &self,
        request: Request<GetShardLocalCacheRequest>,
    ) -> Result<tonic::Response<GetShardLocalCacheResponse>, tonic::Status> {
        let rpc_metrics = begin_grpc_handle_metrics("meta", "get_shard_local_cache_info");
        match self
            .meta_service
            .get_shard_local_cache_info(request.get_ref())
            .await
        {
            Err(e) => {
                rpc_metrics.mark_error();
                log::error!("get_shard_local_cache_info: caught err={:?}", e);
                Ok(Response::new(GetShardLocalCacheResponse {
                    status: ErrorCode::UnknownError.into(),
                    shard_local_cache_infos: vec![],
                }))
            }
            Ok(resp) => Ok(Response::new(resp)),
        }
    }

    async fn get_import_storage_prefix(
        &self,
        req: Request<GetImportStoragePrefixRequest>,
    ) -> Result<tonic::Response<GetImportStoragePrefixResponse>, tonic::Status> {
        if let Some(resp) = self.reject_if_not_leader("get_import_storage_prefix") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta", "get_import_storage_prefix");
        match self
            .meta_service
            .get_import_storage_prefix(req.get_ref())
            .await
        {
            Ok(resp) => Ok(Response::new(resp)),
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("start_import_index: caught err={:?}", e);
                Ok(Response::new(GetImportStoragePrefixResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    error_message: e.to_string(),
                    ..Default::default()
                }))
            }
        }
    }

    async fn finish_import_partition_upload(
        &self,
        request: Request<FinishImportPartitionUploadRequest>,
    ) -> Result<tonic::Response<FinishImportResponse>, tonic::Status> {
        if let Some(resp) = self.reject_if_not_leader("finish_import_partition_upload") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta", "finish_import_partition_upload");
        match self
            .meta_service
            .finish_partition_upload(request.get_ref())
            .await
        {
            Ok(resp) => Ok(Response::new(resp)),
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("finish_partition_upload: caught err={:?}", e);
                Ok(Response::new(FinishImportResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    error_message: e.to_string(),
                }))
            }
        }
    }

    async fn pre_split_import_shards(
        &self,
        request: Request<PreSplitImportShardsRequest>,
    ) -> Result<tonic::Response<PreSplitImportShardsResponse>, tonic::Status> {
        if let Some(resp) = self.reject_if_not_leader("pre_split_import_shards") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta", "pre_split_import_shards");
        match self
            .meta_service
            .presplit_import_shards(request.get_ref())
            .await
        {
            Ok(resp) => Ok(Response::new(resp)),
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("pre_split_import_shards: caught err={:?}", e);
                Ok(Response::new(PreSplitImportShardsResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    error_message: e.to_string(),
                    ..Default::default()
                }))
            }
        }
    }

    async fn finish_import_index_upload(
        &self,
        request: Request<FinishImportIndexUploadRequest>,
    ) -> Result<tonic::Response<FinishImportResponse>, tonic::Status> {
        if let Some(resp) = self.reject_if_not_leader("finish_import_index_upload") {
            return Ok(resp);
        }
        let rpc_metrics = begin_grpc_handle_metrics("meta", "finish_import_index_upload");
        match self
            .meta_service
            .mark_table_upload_finished(request.get_ref())
            .await
        {
            Ok(resp) => Ok(Response::new(resp)),
            Err(e) => {
                rpc_metrics.mark_error();
                warn!("mark_table_upload_finished: caught err={:?}", e);
                Ok(Response::new(FinishImportResponse {
                    status: tici_grpc::ErrorCode::UnknownError.into(),
                    error_message: e.to_string(),
                }))
            }
        }
    }
}

/// - Make MetaService be the leader in the etcd election for failover
/// - Provide service discovery for other components to find MetaService Uses HA
///   etcd client for automatic member discovery and failover
pub async fn build_leadership_and_discovery(
    config: &MetaServiceConfig,
) -> Result<(EtcdLeadership, Arc<RwLock<dyn DiscoveryClient>>), anyhow::Error> {
    let advertise_addr = config.server.get_advertise_addr();
    let pd_addrs = config.server.get_pd_addresses();
    // Create etcd client
    let client = Arc::new(EtcdClient::new_tls(pd_addrs.as_slice(), &config.security)?);

    let leader_value = advertise_addr.to_string();
    info!(
        "Meta service leadership start with etcd client, pd_addr={} prefix={} value={}",
        pd_addrs.join(","),
        META_SERVICE_ELECTION_KEY,
        leader_value
    );

    let election_client = EtcdElectionClient::new(
        META_SERVICE_ELECTION_KEY,
        leader_value,
        client.clone(),
        config.election.effective_lease_keepalive_interval().into(),
    );
    let leader_metrics_observer: LeadershipObserver = Arc::new(update_meta_leader_metric);
    let leadership = EtcdLeadership::new_with_observer_and_interval(
        Arc::new(election_client),
        config.election.campaign_ttl.into(),
        config.election.campaign_interval.into(),
        Some(leader_metrics_observer),
    );

    let mut discovery_client = EtcdDiscoveryClient::new(
        META_SERVICE_REGISTER_KEY_PREFIX,
        client.clone(),
        config.election.registration_lease_ttl.into(),
        config
            .election
            .effective_registration_keepalive_interval()
            .into(),
    );
    discovery_client
        .register(
            advertise_addr,
            "", // TODO: add more info about the meta service if needed
        )
        .await?;
    Ok((leadership, Arc::new(RwLock::new(discovery_client))))
}

pub struct MetaServiceServer {
    tidb_pools: TiDBConnectionPool,
    leadership: EtcdLeadership,
    discovery_client: Arc<RwLock<dyn DiscoveryClient>>,
    pub meta_service: Arc<MetaServiceCore>,
    pub status_server: Option<StatusServer>,
    pub config: ArcSwap<MetaServiceConfig>,
    grpc_handle: Option<tokio::task::JoinHandle<Result<(), tonic::transport::Error>>>,
}

impl MetaServiceServer {
    pub fn build_status_server(
        addr_str: String,
        meta: Arc<MetaServiceCore>,
    ) -> Option<StatusServer> {
        if !addr_str.is_empty() {
            let folder_path = if meta
                .config
                .load()
                .server
                .status_server_store_path
                .is_empty()
            {
                tici_common::config::ServerConfig::deduce_status_server_store_path()
            } else {
                meta.config
                    .load()
                    .server
                    .status_server_store_path
                    .clone()
                    .into()
            };

            info!("status_server at path={:?}", folder_path);

            if !folder_path.is_empty() {
                if let Ok(a) = addr_str.parse() {
                    let mut st = if let Ok(s) = StatusServer::new_for_meta(folder_path, meta) {
                        s
                    } else {
                        warn!("build_status_server: failed to create status server");
                        return None;
                    };
                    if st.start(a).is_ok() {
                        Some(st)
                    } else {
                        warn!(
                            "build_status_server: failed to start status server at {}",
                            a
                        );
                        None
                    }
                } else {
                    info!("build_status_server: can not parse addr, addr={}", addr_str);
                    None
                }
            } else {
                info!("build_status_server: can not generate store path, and it is not specified");
                None
            }
        } else {
            info!("build_status_server: status server is not enabled");
            None
        }
    }

    pub async fn new(config: MetaServiceConfig) -> anyhow::Result<Self> {
        config.validate()?;
        // Ensure the meta store is bootstrapped correctly.
        info!("connect to tidb={}", config.tidb_server.dsns.join(","));
        let tidb_pools = TiDBConnectionPool::new(&config.tidb_server);
        let bootstrap = Bootstrap::new(tidb_pools.clone());
        bootstrap
            .ensure_bootstrapped()
            .await
            .context("failed to bootstrap")?;

        // Only start campaign after we confirm meta store has bootstrapped, so
        // that MetaService can provide service discovery information to other
        // components.
        let (leadership, discovery_client) = build_leadership_and_discovery(&config)
            .await
            .context("Failed to build service discovery and leadership")?;

        info!(
            "meta service created with config={}",
            serde_json::to_string(&config).unwrap_or_default()
        );
        let backend_storage = create_backend_storage(&config.s3, STORAGE_SERVICE_META)
            .await
            .context("create backend storage failed")?;
        let config_arc = Arc::new(config.clone());
        let meta_service = Arc::new(MetaServiceCore::new(
            ArcSwap::new(config_arc.clone()),
            tidb_pools.clone(),
            leadership.clone(),
            backend_storage,
        ));
        let status_server =
            Self::build_status_server(config.server.status_addr.clone(), meta_service.clone());
        Ok(Self {
            tidb_pools,
            leadership,
            discovery_client,
            meta_service,
            status_server,
            config: ArcSwap::new(config_arc.clone()),
            grpc_handle: None,
        })
    }

    pub async fn wait_become_leader(&self) {
        // log down each step of waiting by interval
        let interval = 5 * 60;
        let lease_id = tici_service_discovery::election::leadership::block_wait_leader_term(
            &self.leadership,
            std::time::Duration::from_secs(interval),
        )
        .await;
        info!("MetaService become leader with lease_id={}", lease_id);
    }

    pub async fn start(
        &mut self,
        rx: tokio::sync::oneshot::Receiver<()>,
    ) -> Result<(), anyhow::Error> {
        info!("Meta service is starting...");
        if let Err(e) = self.meta_service.start().await {
            info!("Meta service start err={:?}", e);
            return Err(e);
        }
        info!("Meta service started successfully");

        // Create gRPC server
        let addr_str: String = self.config.load().server.addr.clone();
        let addr = addr_str.parse()?;
        info!("Starting MetaServiceServer grpc server at {:?}", addr);
        crate::common::server_util::initial_metric();
        self.grpc_handle = Some(tokio::spawn(Self::build_server(
            addr,
            self.meta_service.clone(),
            rx,
        )));
        Ok(())
    }

    pub fn build_server(
        addr: SocketAddr,
        meta_service: Arc<MetaServiceCore>,
        rx: tokio::sync::oneshot::Receiver<()>,
    ) -> impl Future<Output = Result<(), tonic::transport::Error>> {
        let grpc_config = meta_service.config.load().server.grpc.clone();
        Server::builder()
            .add_service(
                GrpcMetaServiceServer::new(MetaServiceGrpcImpl::new(meta_service.clone()))
                    .max_decoding_message_size(grpc_config.max_decoding_message_size.0 as usize)
                    .max_encoding_message_size(grpc_config.max_encoding_message_size.0 as usize),
            )
            .add_service(
                GrpcMetaServiceInternalServer::new(MetaServiceInternalGrpcImpl::new(meta_service))
                    .max_decoding_message_size(grpc_config.max_decoding_message_size.0 as usize)
                    .max_encoding_message_size(grpc_config.max_encoding_message_size.0 as usize),
            )
            .serve_with_shutdown(addr, async {
                if let Err(e) = rx.await {
                    warn!("wait shutdown error, e={:?}", e);
                }
            })
    }

    // Take the ownership of `self` and stop all the components.
    pub async fn stop(
        mut self,
        grpc_server_close_tx: tokio::sync::oneshot::Sender<()>,
    ) -> anyhow::Result<()> {
        info!("MetaServiceServer is stopping...");
        if let Err(_e) = grpc_server_close_tx.send(()) {
            warn!("send grpc_server_close_tx fail");
        }
        // Stop external services (grpc, http) first
        info!("Grpc server is stopping...");
        if let Some(handle) = self.grpc_handle {
            if let Err(e) = handle.await {
                warn!("Grpc server stopped with err={}", e);
            } else {
                info!("Grpc server stopped successfully");
            }
        }
        if let Some(s) = self.status_server.take() {
            s.stop();
        }
        info!("StatusService stopped successfully");

        self.meta_service.stop().await?;
        info!("MetaService stopped successfully");

        // stop the campaign and service discovery
        self.leadership.stop_campaign().await?;
        self.discovery_client
            .write()
            .await
            .deregister(self.config.load().server.get_advertise_addr())
            .await?;
        info!("Service discovery deregistered successfully");
        self.tidb_pools.stop().await?;
        info!("TiDB connection pools stopped successfully");

        info!("MetaServiceServer stopped successfully...");
        Ok(())
    }

    // Create a gRPC client and retry until success.
    pub async fn get_client(
        addr: &str,
        grpc_config: &GrpcConfig,
    ) -> GrpcMetaServiceClient<tonic::transport::Channel> {
        let addr = if addr.starts_with("http://") || addr.starts_with("https://") {
            addr.to_string()
        } else {
            format!("http://{}", addr)
        };
        let mut attempts = 0;
        let max_attempts = 30;
        loop {
            match GrpcMetaServiceClient::connect(addr.to_string()).await {
                std::result::Result::Ok(c) => {
                    info!("Connected to MetaServiceCore gRPC Server at {}", addr);
                    let c = c.max_decoding_message_size(
                        grpc_config.max_decoding_message_size.0 as usize,
                    );
                    let c = c.max_encoding_message_size(
                        grpc_config.max_encoding_message_size.0 as usize,
                    );
                    break c;
                }
                std::result::Result::Err(e) => {
                    info!(
                        "Failed to connect to server at {} (attempt {}/{}), err={}",
                        addr,
                        attempts + 1,
                        max_attempts,
                        e
                    );
                    std::thread::sleep(std::time::Duration::from_secs(1));
                    attempts += 1;
                    if attempts >= max_attempts {
                        panic!("exceed max attempts");
                    }
                }
            }
        }
    }

    // Create a gRPC client and retry until success.
    pub async fn get_internal_client(
        addr: &str,
        grpc_config: &GrpcConfig,
    ) -> GrpcMetaServiceInternalClient<tonic::transport::Channel> {
        let addr = if addr.starts_with("http://") || addr.starts_with("https://") {
            addr.to_string()
        } else {
            format!("http://{}", addr)
        };
        let mut attempts = 0;
        let max_attempts = 30;
        loop {
            match GrpcMetaServiceInternalClient::connect(addr.to_string()).await {
                std::result::Result::Ok(c) => {
                    info!("Connected to MetaServiceInternal gRPC Server at {}", addr);
                    let c = c.max_decoding_message_size(
                        grpc_config.max_decoding_message_size.0 as usize,
                    );
                    let c = c.max_encoding_message_size(
                        grpc_config.max_encoding_message_size.0 as usize,
                    );
                    break c;
                }
                std::result::Result::Err(e) => {
                    info!(
                        "Failed to connect to server at {} (attempt {}/{}), err={}",
                        addr,
                        attempts + 1,
                        max_attempts,
                        e
                    );
                    std::thread::sleep(std::time::Duration::from_secs(1));
                    attempts += 1;
                    if attempts >= max_attempts {
                        panic!("exceed max attempts");
                    }
                }
            }
        }
    }
}