use std::error::Error as StdError;
use std::fmt::{Display, Formatter};

use pd_proto_tso::{Error, ErrorType};

/// Transport-level status codes used by the service facade.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GrpcCode {
    /// Success.
    Ok,
    /// Operation canceled by caller/context.
    Canceled,
    /// Unknown internal failure.
    Unknown,
    /// Invalid request argument.
    InvalidArgument,
    /// Deadline exceeded.
    DeadlineExceeded,
    /// Requested resource not found.
    NotFound,
    /// Request failed precondition.
    FailedPrecondition,
    /// Service unavailable.
    Unavailable,
    /// Resource exhaustion/backpressure.
    ResourceExhausted,
    /// API not implemented on this surface.
    Unimplemented,
}

/// Transport-level status returned by streaming/unary entry points.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GrpcStatus {
    /// Canonical transport status code.
    pub code: GrpcCode,
    /// Human-readable error message.
    pub message: String,
    /// Optional retry hint in milliseconds for compatibility clients.
    pub retry_after_ms: Option<u64>,
}

impl GrpcStatus {
    /// Builds one transport status payload.
    #[must_use]
    pub fn new(code: GrpcCode, message: impl Into<String>) -> Self {
        Self {
            code,
            message: message.into(),
            retry_after_ms: None,
        }
    }

    /// Attaches retry-hint metadata for compatibility paths that expose backoff hints.
    #[must_use]
    pub fn with_retry_after_ms(mut self, retry_after_ms: u64) -> Self {
        self.retry_after_ms = Some(retry_after_ms);
        self
    }
}

/// PD-compatible transport error text for cluster-id mismatches.
#[must_use]
pub(crate) fn transport_cluster_mismatch_message(expected: u64, got: u64) -> String {
    format!("mismatch cluster id, need {expected} but got {got}")
}

/// Internal service-layer errors.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ServiceError {
    /// Service is not started.
    NotStarted,
    /// Request cluster-id does not match local cluster id.
    ClusterMismatched {
        /// Local expected cluster id.
        expected: u64,
        /// Cluster id carried by request.
        got: u64,
    },
    /// Requested keyspace-group id is outside serving range.
    InvalidGroupId {
        /// Offending keyspace-group id.
        group_id: u32,
    },
    /// Request `count` is invalid.
    InvalidCount,
    /// Group is overloaded and requests should back off.
    Overloaded {
        /// Group id that is currently overloaded.
        group_id: u32,
        /// Recommended backoff in milliseconds.
        retry_after_ms: u64,
    },
    /// Local node is not the PD leader for this request path.
    NotLeader,
    /// Requested group is not currently served on this node.
    NotServed {
        /// Group id not served on this node.
        group_id: u32,
    },
    /// Keyspace has no assigned keyspace-group route.
    KeyspaceNotAssigned {
        /// Keyspace id with missing group assignment.
        keyspace_id: u32,
    },
    /// Keyspace-group is currently merging.
    KeyspaceGroupMerging {
        /// Group id currently blocked by merge workflow.
        group_id: u32,
    },
    /// Caller provided stale `mod_revision`.
    StaleRevision,
    /// Logical component overflowed configured range.
    LogicalOverflow,
    /// Storage-layer failure.
    Storage(String),
}

impl ServiceError {
    /// Converts internal errors to stream/unary transport status surface.
    #[must_use]
    pub fn to_grpc_status(&self) -> GrpcStatus {
        let code = match self {
            Self::NotStarted => GrpcCode::Unavailable,
            Self::ClusterMismatched { .. } => GrpcCode::FailedPrecondition,
            Self::Overloaded { .. } => GrpcCode::ResourceExhausted,
            Self::InvalidGroupId { .. }
            | Self::InvalidCount
            | Self::NotLeader
            | Self::NotServed { .. }
            | Self::KeyspaceNotAssigned { .. }
            | Self::KeyspaceGroupMerging { .. }
            | Self::StaleRevision
            | Self::LogicalOverflow
            | Self::Storage(_) => GrpcCode::Unknown,
        };
        let message = match self {
            Self::ClusterMismatched { expected, got } => {
                transport_cluster_mismatch_message(*expected, *got)
            }
            _ => self.to_string(),
        };
        let status = GrpcStatus::new(code, message);
        match self {
            Self::Overloaded { retry_after_ms, .. } => status.with_retry_after_ms(*retry_after_ms),
            _ => status,
        }
    }

    /// Converts internal errors to handled unary header error payload.
    #[must_use]
    pub fn to_unary_error(&self) -> Error {
        let typ = match self {
            Self::NotStarted => ErrorType::NotBootstrapped,
            Self::ClusterMismatched { .. } => ErrorType::ClusterMismatched,
            Self::InvalidGroupId { .. } | Self::InvalidCount | Self::StaleRevision => {
                ErrorType::InvalidValue
            }
            Self::Overloaded { .. } => ErrorType::Unknown,
            Self::KeyspaceNotAssigned { .. } => ErrorType::Unknown,
            Self::NotLeader
            | Self::NotServed { .. }
            | Self::KeyspaceGroupMerging { .. }
            | Self::LogicalOverflow
            | Self::Storage(_) => ErrorType::Unknown,
        };
        Error::new(typ, self.to_string())
    }
}

impl Display for ServiceError {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NotStarted => write!(f, "server not started"),
            Self::ClusterMismatched { expected, got } => {
                write!(f, "cluster id mismatch: expected {expected}, got {got}")
            }
            Self::InvalidGroupId { group_id } => {
                write!(f, "keyspace group id {group_id} is out of serving range")
            }
            Self::InvalidCount => write!(f, "tso count should be positive"),
            Self::Overloaded {
                group_id,
                retry_after_ms,
            } => write!(
                f,
                "group {group_id} is overloaded; retry after {retry_after_ms} ms"
            ),
            Self::NotLeader => write!(f, "requested pd is not leader of cluster"),
            Self::NotServed { group_id } => {
                write!(
                    f,
                    "requested keyspace group {group_id} is not served by this node"
                )
            }
            Self::KeyspaceNotAssigned { keyspace_id } => {
                write!(
                    f,
                    "keyspace {keyspace_id} is not assigned to a keyspace group"
                )
            }
            Self::KeyspaceGroupMerging { group_id } => {
                write!(f, "keyspace group {group_id} is merging")
            }
            Self::StaleRevision => write!(f, "keyspace group mod_revision stale"),
            Self::LogicalOverflow => write!(f, "logical part outside max logical interval"),
            Self::Storage(msg) => write!(f, "storage error: {msg}"),
        }
    }
}

impl StdError for ServiceError {}

#[cfg(test)]
mod tests {
    use pd_proto_tso::ErrorType;

    use super::ServiceError;

    #[test]
    fn keyspace_not_assigned_maps_to_unknown_for_unary_surface() {
        let mapped = ServiceError::KeyspaceNotAssigned { keyspace_id: 7 }.to_unary_error();
        assert_eq!(mapped.typ, ErrorType::Unknown);
    }

    #[test]
    fn overloaded_maps_to_resource_exhausted_with_retry_hint() {
        let status = ServiceError::Overloaded {
            group_id: 3,
            retry_after_ms: 55,
        }
        .to_grpc_status();
        assert_eq!(status.code, super::GrpcCode::ResourceExhausted);
        assert_eq!(status.retry_after_ms, Some(55));
    }

    #[test]
    fn cluster_mismatched_grpc_status_uses_pd_transport_wording() {
        let status = ServiceError::ClusterMismatched {
            expected: 42,
            got: 7,
        }
        .to_grpc_status();
        assert_eq!(status.code, super::GrpcCode::FailedPrecondition);
        assert_eq!(status.message, "mismatch cluster id, need 42 but got 7");
    }
}