use anyhow::Result;
use tici_common::index::IndexKey;
use tici_grpc::{
    AppendDeltaFragMetaRequest, AppendFragMetaResponse, ErrorCode, FinishCompactFragRequest,
    FinishCompactFragResponse, GetIndexMetaRequest, GetIndexMetaResponse,
    ReaderNodeHeartbeatRequest, ReaderNodeHeartbeatResponse, WorkerNodeHeartbeatRequest,
    WorkerNodeHeartbeatResponse,
};

use crate::meta::{MetaServiceCore, reader::ShardCacheManager};

impl MetaServiceCore {
    pub async fn append_delta_frag_meta(
        &self,
        request: &AppendDeltaFragMetaRequest,
    ) -> Result<AppendFragMetaResponse> {
        self.worker_node_mgr
            .on_msg_append_delta_frag_meta(request)
            .await
    }

    pub async fn finish_compact_fragments(
        &self,
        req: &FinishCompactFragRequest,
    ) -> Result<FinishCompactFragResponse> {
        self.worker_node_mgr.on_compaction_finish(req).await
    }
}

impl MetaServiceCore {
    pub async fn report_worker_node_status(
        &self,
        req: WorkerNodeHeartbeatRequest,
    ) -> Result<Vec<WorkerNodeHeartbeatResponse>> {
        self.worker_node_mgr.report_worker_node_status(req).await
    }

    pub async fn report_reader_node_status(
        &self,
        req: ReaderNodeHeartbeatRequest,
    ) -> Result<ReaderNodeHeartbeatResponse> {
        self.reader_pool.on_msg_reader_heartbeat(req, None).await
    }

    /// Get shard local cache information
    pub async fn get_shard_local_cache_info(
        &self,
        request: &tici_grpc::GetShardLocalCacheRequest,
    ) -> Result<tici_grpc::GetShardLocalCacheResponse> {
        // Check for invalid table_id/index_id first
        if request.table_id == 0 || request.index_id == 0 {
            return Ok(tici_grpc::GetShardLocalCacheResponse {
                status: ErrorCode::InvalidArguments.into(), // Error status
                shard_local_cache_infos: vec![],
            });
        }

        ShardCacheManager::on_msg_get_shard_local_cache(
            &self.shard_cache_manager,
            request,
            self.reader_pool.clone(),
        )
        .await
    }

    /// Get TableInfo and IndexInfo by keyspace_id, table_id and index_id.
    pub async fn get_index_meta(
        &self,
        request: &GetIndexMetaRequest,
    ) -> Result<GetIndexMetaResponse> {
        let index_key = IndexKey::new(request.keyspace_id, request.table_id, request.index_id);
        let meta = self
            .shard_cache_manager
            .get_index_meta_cloned(&index_key)
            .await;
        match meta {
            Some(m) => Ok(GetIndexMetaResponse {
                status: ErrorCode::Success.into(),
                error_message: String::new(),
                table_info: Some(m.table_info),
                index_info: Some(m.index_info),
            }),
            None => Ok(GetIndexMetaResponse {
                status: ErrorCode::IndexNotFound.into(),
                error_message: format!(
                    "index not found: keyspace_id={}, table_id={}, index_id={}",
                    request.keyspace_id, request.table_id, request.index_id
                ),
                table_info: None,
                index_info: None,
            }),
        }
    }
}