use std::sync::Arc;

use anyhow::Result;
use arc_swap::ArcSwap;
use log::warn;
use tici_grpc::{
    AppendBaseFragMetaRequest, AppendFragMetaResponse, CreateIndexRequest, CreateIndexRequestExt,
    CreateIndexResponse, DropIndexRequest, DropIndexResponse, IndexInfo, TableInfo,
};
use tici_shard::utils::{frag_path_generator::TSOGenerator, schema::IndexMeta};
use tokio::sync::OnceCell;

use crate::meta::MetaServiceCore;

pub(super) struct PdTsoProvider {
    generator: OnceCell<TSOGenerator>,
}

impl PdTsoProvider {
    pub(super) fn new() -> Self {
        Self {
            generator: OnceCell::new(),
        }
    }

    pub(super) async fn alloc_tso(
        &self,
        config: &Arc<ArcSwap<crate::config::MetaServiceConfig>>,
    ) -> Result<u64> {
        let generator = self
            .generator
            .get_or_try_init(|| async {
                let config = config.load();
                TSOGenerator::new(config.server.get_pd_addresses(), config.security.clone()).await
            })
            .await?;

        generator.gen_tso().await
    }
}

impl MetaServiceCore {
    pub async fn append_base_frag_meta(
        &self,
        request: &AppendBaseFragMetaRequest,
    ) -> Result<AppendFragMetaResponse> {
        self.shard_task_manager
            .on_msg_append_base_frag_meta(request)
            .await
    }
}

impl MetaServiceCore {
    pub async fn create_index(&self, request: &CreateIndexRequest) -> Result<CreateIndexResponse> {
        log::info!("create_index: request={:?}", request);
        let (table_info, mut index_info, partition_table_info) = {
            match request.get_grpc_table_info_and_index_info() {
                Ok((table_info, index_info, partition_table_info)) => {
                    (table_info, index_info, partition_table_info)
                }
                Err(e) => {
                    warn!(
                        "create_index: failed to parse request, err={:?}, request={:?}",
                        e, request
                    );
                    return Ok(CreateIndexResponse {
                        status: 2,
                        error_message: format!("Failed to parse request, err={}", e),
                        index_id: "".to_string(),
                    });
                }
            }
        };

        if let Err(e) = IndexMeta::new(&table_info, &index_info) {
            warn!(
                "create_index: TableInfo={:?}, IndexInfo={:?}, failed to create index meta={:?}",
                table_info, index_info, e
            );
            return Ok(CreateIndexResponse {
                status: 2,
                error_message: format!("Failed to create index, err={}", e),
                index_id: "".to_string(),
            });
        }

        index_info.created_ts = self.tso_provider.alloc_tso(&self.config).await?;

        self.shard_task_manager
            .on_msg_create_index(table_info, index_info, partition_table_info, request)
            .await
    }

    pub async fn drop_index(&self, request: &DropIndexRequest) -> Result<DropIndexResponse> {
        self.shard_task_manager.on_msg_drop_index(request).await
    }

    pub async fn bootstrap_index(
        &self,
        table_info: TableInfo,
        index_info: IndexInfo,
    ) -> Result<bool> {
        self.shard_task_manager
            .bootstrap_index(table_info, index_info)
            .await
    }
}