use std::str::FromStr;

use pd_proto_grpc::{pdpb, tsopb};
use tonic::metadata::MetadataValue;
use tonic::{Code, Status};

use crate::error::{GrpcCode, GrpcStatus};
use crate::forwarding::bootstrap_keyspace_id;

pub(crate) fn status_from_core(status: GrpcStatus) -> Status {
    let mut transport = Status::new(map_grpc_code(status.code), status.message);
    if let Some(retry_after_ms) = status.retry_after_ms
        && let Ok(value) = MetadataValue::from_str(&retry_after_ms.to_string())
    {
        transport.metadata_mut().insert("retry-after-ms", value);
    }
    transport
}

fn map_grpc_code(code: GrpcCode) -> Code {
    match code {
        GrpcCode::Ok => Code::Ok,
        GrpcCode::Canceled => Code::Cancelled,
        GrpcCode::Unknown => Code::Unknown,
        GrpcCode::InvalidArgument => Code::InvalidArgument,
        GrpcCode::DeadlineExceeded => Code::DeadlineExceeded,
        GrpcCode::NotFound => Code::NotFound,
        GrpcCode::FailedPrecondition => Code::FailedPrecondition,
        GrpcCode::Unavailable => Code::Unavailable,
        GrpcCode::ResourceExhausted => Code::ResourceExhausted,
        GrpcCode::Unimplemented => Code::Unimplemented,
    }
}

pub(crate) fn from_tsopb_request_header(
    header: Option<tsopb::RequestHeader>,
) -> pd_proto_tso::RequestHeader {
    let header = header.unwrap_or_default();
    pd_proto_tso::RequestHeader {
        cluster_id: header.cluster_id,
        sender_id: header.sender_id,
        keyspace_id: header.keyspace_id,
        keyspace_group_id: header.keyspace_group_id,
    }
}

pub(crate) fn from_tsopb_tso_request(request: tsopb::TsoRequest) -> pd_proto_tso::TsoRequest {
    pd_proto_tso::TsoRequest {
        header: from_tsopb_request_header(request.header),
        count: request.count,
        dc_location: request.dc_location,
    }
}

pub(crate) fn from_pdpb_request_header(
    header: Option<pdpb::RequestHeader>,
) -> pd_proto_tso::RequestHeader {
    let header = header.unwrap_or_default();
    pd_proto_tso::RequestHeader {
        cluster_id: header.cluster_id,
        sender_id: header.sender_id,
        keyspace_id: bootstrap_keyspace_id(),
        keyspace_group_id: 0,
    }
}

pub(crate) fn from_pdpb_tso_request(request: pdpb::TsoRequest) -> pd_proto_tso::TsoRequest {
    pd_proto_tso::TsoRequest {
        header: from_pdpb_request_header(request.header),
        count: request.count,
        // Legacy forwarded request-shape parity keeps dc_location default-empty.
        dc_location: String::new(),
    }
}

pub(crate) fn to_tsopb_error(error: Option<pd_proto_tso::Error>) -> Option<tsopb::Error> {
    let error = error?;
    Some(tsopb::Error {
        r#type: match error.typ {
            pd_proto_tso::ErrorType::Ok => tsopb::ErrorType::Ok as i32,
            pd_proto_tso::ErrorType::Unknown => tsopb::ErrorType::Unknown as i32,
            pd_proto_tso::ErrorType::NotBootstrapped => tsopb::ErrorType::NotBootstrapped as i32,
            pd_proto_tso::ErrorType::AlreadyBootstrapped => {
                tsopb::ErrorType::AlreadyBootstrapped as i32
            }
            pd_proto_tso::ErrorType::InvalidValue => tsopb::ErrorType::InvalidValue as i32,
            pd_proto_tso::ErrorType::ClusterMismatched => {
                tsopb::ErrorType::ClusterMismatched as i32
            }
        },
        message: error.message,
    })
}

pub(crate) fn to_tsopb_response_header(
    header: pd_proto_tso::ResponseHeader,
) -> tsopb::ResponseHeader {
    tsopb::ResponseHeader {
        cluster_id: header.cluster_id,
        error: to_tsopb_error(header.error),
        keyspace_id: header.keyspace_id,
        keyspace_group_id: header.keyspace_group_id,
    }
}

pub(crate) fn to_tsopb_timestamp(timestamp: pd_proto_tso::Timestamp) -> pdpb::Timestamp {
    pdpb::Timestamp {
        physical: timestamp.physical,
        logical: timestamp.logical,
        suffix_bits: timestamp.suffix_bits,
    }
}

pub(crate) fn to_tsopb_tso_response(response: pd_proto_tso::TsoResponse) -> tsopb::TsoResponse {
    tsopb::TsoResponse {
        header: Some(to_tsopb_response_header(response.header)),
        count: response.count,
        timestamp: Some(to_tsopb_timestamp(response.timestamp)),
    }
}

pub(crate) fn to_pdpb_error(error: Option<pd_proto_tso::Error>) -> Option<pdpb::Error> {
    let error = error?;
    let typ = match error.typ {
        pd_proto_tso::ErrorType::Ok => pdpb::ErrorType::Ok,
        pd_proto_tso::ErrorType::Unknown
        | pd_proto_tso::ErrorType::NotBootstrapped
        | pd_proto_tso::ErrorType::AlreadyBootstrapped
        | pd_proto_tso::ErrorType::InvalidValue
        | pd_proto_tso::ErrorType::ClusterMismatched => pdpb::ErrorType::Unknown,
    };
    Some(pdpb::Error {
        r#type: typ as i32,
        message: error.message,
    })
}

pub(crate) fn to_pdpb_response_header(
    header: pd_proto_tso::ResponseHeader,
) -> pdpb::ResponseHeader {
    pdpb::ResponseHeader {
        cluster_id: header.cluster_id,
        error: to_pdpb_error(header.error),
    }
}

pub(crate) fn to_pdpb_timestamp(timestamp: pd_proto_tso::Timestamp) -> pdpb::Timestamp {
    pdpb::Timestamp {
        physical: timestamp.physical,
        logical: timestamp.logical,
        suffix_bits: timestamp.suffix_bits,
    }
}

pub(crate) fn to_pdpb_tso_response(response: pd_proto_tso::TsoResponse) -> pdpb::TsoResponse {
    pdpb::TsoResponse {
        header: Some(to_pdpb_response_header(response.header)),
        count: response.count,
        timestamp: Some(to_pdpb_timestamp(response.timestamp)),
    }
}

pub(crate) fn to_tsopb_keyspace_group_member(
    member: pd_proto_tso::KeyspaceGroupMember,
) -> tsopb::KeyspaceGroupMember {
    tsopb::KeyspaceGroupMember {
        address: member.address,
        is_primary: member.is_primary,
    }
}

pub(crate) fn to_tsopb_keyspace_group(group: pd_proto_tso::KeyspaceGroup) -> tsopb::KeyspaceGroup {
    tsopb::KeyspaceGroup {
        id: group.id,
        user_kind: group.user_kind,
        split_state: group.split_state.map(|state| tsopb::SplitState {
            split_source: state.split_source,
        }),
        members: group
            .members
            .into_iter()
            .map(to_tsopb_keyspace_group_member)
            .collect(),
    }
}

pub(crate) fn to_tsopb_find_group_response(
    response: pd_proto_tso::FindGroupByKeyspaceIdResponse,
) -> tsopb::FindGroupByKeyspaceIdResponse {
    tsopb::FindGroupByKeyspaceIdResponse {
        header: Some(to_tsopb_response_header(response.header)),
        keyspace_group: response.keyspace_group.map(to_tsopb_keyspace_group),
        mod_revision: response.mod_revision,
    }
}

pub(crate) fn to_tsopb_get_min_ts_response(
    response: pd_proto_tso::GetMinTsResponse,
) -> tsopb::GetMinTsResponse {
    tsopb::GetMinTsResponse {
        header: Some(to_tsopb_response_header(response.header)),
        timestamp: Some(to_tsopb_timestamp(response.timestamp)),
        keyspace_groups_serving: response.keyspace_groups_serving,
        keyspace_groups_total: response.keyspace_groups_total,
    }
}