use thiserror::Error;
use tonic::transport::{Channel, Endpoint};

use crate::proto::kafkalite_client::KafkaliteClient as ProtoKafkaliteClient;
use crate::proto::{
    self, CommitOffsetsRequest, CommitOffsetsResponse, GetCommittedOffsetRequest,
    GetCommittedOffsetResponse, HealthCheckRequest, HealthCheckResponse, HeartbeatRequest,
    HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaveGroupRequest,
    LeaveGroupResponse, PollMessagesRequest, PollMessagesResponse,
};
use crate::{AppendAck, Header, NewMessage, StoredMessage};

#[derive(Debug, Clone)]
pub struct KafkaliteClient {
    inner: ProtoKafkaliteClient<Channel>,
}

#[derive(Debug, Error)]
pub enum KafkaliteClientError {
    #[error("invalid endpoint: {0}")]
    Endpoint(String),
    #[error("grpc status failed: {0}")]
    Status(Box<tonic::Status>),
}

pub type Result<T> = std::result::Result<T, KafkaliteClientError>;

impl KafkaliteClient {
    pub fn new(host: &str, port: u16) -> Result<Self> {
        let endpoint = Endpoint::from_shared(format!("http://{host}:{port}"))
            .map_err(|err| KafkaliteClientError::Endpoint(err.to_string()))?;
        Ok(Self {
            inner: ProtoKafkaliteClient::new(endpoint.connect_lazy()),
        })
    }

    pub async fn health(&mut self) -> Result<HealthCheckResponse> {
        Ok(self
            .inner
            .health(HealthCheckRequest {})
            .await
            .map_err(|err| KafkaliteClientError::Status(Box::new(err)))?
            .into_inner())
    }

    pub async fn publish(&mut self, message: &NewMessage) -> Result<AppendAck> {
        Ok(self
            .inner
            .publish(proto::NewMessage::from(message.clone()))
            .await
            .map_err(|err| KafkaliteClientError::Status(Box::new(err)))?
            .into_inner()
            .into())
    }

    pub async fn join_group(
        &mut self,
        request: &JoinGroupRequest,
    ) -> Result<JoinGroupResponse> {
        Ok(self
            .inner
            .join_group(request.clone())
            .await
            .map_err(|err| KafkaliteClientError::Status(Box::new(err)))?
            .into_inner())
    }

    pub async fn heartbeat(&mut self, request: &HeartbeatRequest) -> Result<HeartbeatResponse> {
        Ok(self
            .inner
            .heartbeat(request.clone())
            .await
            .map_err(|err| KafkaliteClientError::Status(Box::new(err)))?
            .into_inner())
    }

    pub async fn poll_messages(
        &mut self,
        request: &PollMessagesRequest,
    ) -> Result<PollMessagesResponse> {
        Ok(self
            .inner
            .poll_messages(request.clone())
            .await
            .map_err(|err| KafkaliteClientError::Status(Box::new(err)))?
            .into_inner())
    }

    pub async fn commit_offsets(
        &mut self,
        request: &CommitOffsetsRequest,
    ) -> Result<CommitOffsetsResponse> {
        Ok(self
            .inner
            .commit_offsets(request.clone())
            .await
            .map_err(|err| KafkaliteClientError::Status(Box::new(err)))?
            .into_inner())
    }

    pub async fn leave_group(
        &mut self,
        request: &LeaveGroupRequest,
    ) -> Result<LeaveGroupResponse> {
        Ok(self
            .inner
            .leave_group(request.clone())
            .await
            .map_err(|err| KafkaliteClientError::Status(Box::new(err)))?
            .into_inner())
    }

    pub async fn get_committed_offset(
        &mut self,
        request: &GetCommittedOffsetRequest,
    ) -> Result<GetCommittedOffsetResponse> {
        Ok(self
            .inner
            .get_committed_offset(request.clone())
            .await
            .map_err(|err| KafkaliteClientError::Status(Box::new(err)))?
            .into_inner())
    }
}

impl From<Header> for proto::Header {
    fn from(value: Header) -> Self {
        Self {
            key: value.key,
            value: value.value,
        }
    }
}

impl From<proto::Header> for Header {
    fn from(value: proto::Header) -> Self {
        Self {
            key: value.key,
            value: value.value,
        }
    }
}

impl From<NewMessage> for proto::NewMessage {
    fn from(value: NewMessage) -> Self {
        Self {
            topic: value.topic,
            key: value.key.unwrap_or_default(),
            payload_json: value.payload_json,
            headers: value.headers.into_iter().map(Into::into).collect(),
            published_at_unix_ms: value.published_at_unix_ms,
        }
    }
}

impl From<proto::AppendAck> for AppendAck {
    fn from(value: proto::AppendAck) -> Self {
        Self {
            topic: value.topic,
            partition: value.partition,
            offset: value.offset,
            published_at_unix_ms: value.published_at_unix_ms,
        }
    }
}

impl From<StoredMessage> for proto::StoredMessage {
    fn from(value: StoredMessage) -> Self {
        Self {
            topic: value.topic,
            partition: value.partition,
            offset: value.offset,
            key: value.key.unwrap_or_default(),
            payload_json: value.payload_json,
            headers: value.headers.into_iter().map(Into::into).collect(),
            published_at_unix_ms: value.published_at_unix_ms,
        }
    }
}

impl From<proto::StoredMessage> for StoredMessage {
    fn from(value: proto::StoredMessage) -> Self {
        Self {
            topic: value.topic,
            partition: value.partition,
            offset: value.offset,
            key: if value.key.is_empty() {
                None
            } else {
                Some(value.key)
            },
            payload_json: value.payload_json,
            headers: value.headers.into_iter().map(Into::into).collect(),
            published_at_unix_ms: value.published_at_unix_ms,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn header_conversion_preserves_raw_bytes() {
        let header = Header {
            key: "x-bin".to_string(),
            value: vec![0, 159, 146, 150],
        };

        let proto_header = proto::Header::from(header.clone());
        assert_eq!(proto_header.value, header.value);

        let roundtrip = Header::from(proto_header);
        assert_eq!(roundtrip, header);
    }
}
