//! gRPC transport — implements [`Transport`] over gRPC.
//!
//! Provides [`GrpcTransport`] as a first-class alternative to
//! [`JsonRpcTransport`](super::JsonRpcTransport). Lives in the `client` module
//! alongside other transport implementations.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use futures::Stream;
use tokio::sync::Mutex;
use tonic::transport::Channel;

use super::{EventStream, Transport};
use crate::error::{A2AError, Result};
use crate::grpc::convert::{hashmap_to_struct, struct_to_hashmap};
use crate::grpc::proto::{
    self, CancelTaskRequest, GetTaskRequest, SendMessageRequest, SubscribeToTaskRequest,
    a2a_service_client::A2aServiceClient,
};
use crate::types::{
    AgentCard, Artifact, DeleteTaskPushConfigParams, Event, GetTaskPushConfigParams,
    ListTaskPushConfigParams, ListTasksRequest, ListTasksResponse, Message, MessageSendParams,
    SendMessageResult, Task, TaskArtifactUpdateEvent, TaskIdParams, TaskPushConfig,
    TaskQueryParams, TaskState, TaskStatus, TaskStatusUpdateEvent,
};

/// gRPC transport for A2A client operations.
///
/// Implements [`Transport`] so it can be used interchangeably with
/// [`JsonRpcTransport`](super::JsonRpcTransport) via [`Client`](super::Client).
///
/// # Example
///
/// ```no_run
/// use ra2a::client::{Client, GrpcTransport};
///
/// # async fn example() -> ra2a::error::Result<()> {
/// let transport = GrpcTransport::connect("http://localhost:50051").await?;
/// let client = Client::new(Box::new(transport));
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone)]
pub struct GrpcTransport {
    // Arc<Mutex> because tonic client methods take &mut self
    client: Arc<Mutex<A2aServiceClient<Channel>>>,
}

impl GrpcTransport {
    /// Connects to a gRPC endpoint.
    pub async fn connect(endpoint: impl Into<String>) -> Result<Self> {
        let endpoint = endpoint.into();
        let channel = Channel::from_shared(endpoint)
            .map_err(|e| A2AError::Other(e.to_string()))?
            .connect()
            .await
            .map_err(|e| A2AError::Other(e.to_string()))?;
        Ok(Self::from_channel(channel))
    }

    /// Creates a transport from an existing [`Channel`].
    #[must_use]
    pub fn from_channel(channel: Channel) -> Self {
        Self {
            client: Arc::new(Mutex::new(A2aServiceClient::new(channel))),
        }
    }

    /// Builds a proto `SendMessageRequest` from native params.
    fn build_send_request(params: &MessageSendParams) -> SendMessageRequest {
        let message = proto::Message::from(params.message.clone());
        let configuration =
            params
                .configuration
                .as_ref()
                .map(|config| proto::SendMessageConfiguration {
                    accepted_output_modes: config.accepted_output_modes.clone(),
                    push_notification_config: config
                        .push_notification_config
                        .as_ref()
                        .map(|pc| proto::PushNotificationConfig::from(pc.clone())),
                    history_length: config.history_length,
                    blocking: config.blocking.unwrap_or(false),
                });
        let metadata = if params.metadata.is_empty() {
            None
        } else {
            hashmap_to_struct(params.metadata.clone())
        };

        SendMessageRequest {
            tenant: String::new(),
            message: Some(message),
            configuration,
            metadata,
        }
    }
}

impl Transport for GrpcTransport {
    fn send_message<'a>(
        &'a self,
        params: &'a MessageSendParams,
    ) -> Pin<Box<dyn Future<Output = Result<SendMessageResult>> + Send + 'a>> {
        Box::pin(async move {
            let request = Self::build_send_request(params);
            let response = self
                .client
                .lock()
                .await
                .send_message(request)
                .await
                .map_err(|e| A2AError::Other(e.to_string()))?;

            match response.into_inner().payload {
                Some(proto::send_message_response::Payload::Task(task)) => {
                    Ok(SendMessageResult::Task(Task::from(task)))
                }
                Some(proto::send_message_response::Payload::Message(msg)) => {
                    Ok(SendMessageResult::Message(Message::from(msg)))
                }
                None => Err(A2AError::InternalError("empty gRPC response".into())),
            }
        })
    }

    fn send_message_stream<'a>(
        &'a self,
        params: &'a MessageSendParams,
    ) -> Pin<Box<dyn Future<Output = Result<EventStream>> + Send + 'a>> {
        Box::pin(async move {
            let request = Self::build_send_request(params);
            let response = self
                .client
                .lock()
                .await
                .send_streaming_message(request)
                .await
                .map_err(|e| A2AError::Other(e.to_string()))?;

            Ok(Box::pin(GrpcEventStream {
                inner: response.into_inner(),
            }) as EventStream)
        })
    }

    fn get_task<'a>(
        &'a self,
        params: &'a TaskQueryParams,
    ) -> Pin<Box<dyn Future<Output = Result<Task>> + Send + 'a>> {
        Box::pin(async move {
            let request = GetTaskRequest {
                tenant: String::new(),
                id: params.id.clone(),
                history_length: params.history_length,
            };
            let response = self
                .client
                .lock()
                .await
                .get_task(request)
                .await
                .map_err(|e| A2AError::Other(e.to_string()))?;
            Ok(Task::from(response.into_inner()))
        })
    }

    fn list_tasks<'a>(
        &'a self,
        _params: &'a ListTasksRequest,
    ) -> Pin<Box<dyn Future<Output = Result<ListTasksResponse>> + Send + 'a>> {
        Box::pin(async move {
            // gRPC proto does not define a ListTasks RPC yet
            Err(A2AError::UnsupportedOperation(
                "list_tasks not available over gRPC".into(),
            ))
        })
    }

    fn cancel_task<'a>(
        &'a self,
        params: &'a TaskIdParams,
    ) -> Pin<Box<dyn Future<Output = Result<Task>> + Send + 'a>> {
        Box::pin(async move {
            let request = CancelTaskRequest {
                tenant: String::new(),
                id: params.id.clone(),
            };
            let response = self
                .client
                .lock()
                .await
                .cancel_task(request)
                .await
                .map_err(|e| A2AError::Other(e.to_string()))?;
            Ok(Task::from(response.into_inner()))
        })
    }

    fn resubscribe<'a>(
        &'a self,
        params: &'a TaskIdParams,
    ) -> Pin<Box<dyn Future<Output = Result<EventStream>> + Send + 'a>> {
        Box::pin(async move {
            let request = SubscribeToTaskRequest {
                tenant: String::new(),
                id: params.id.clone(),
            };
            let response = self
                .client
                .lock()
                .await
                .subscribe_to_task(request)
                .await
                .map_err(|e| A2AError::Other(e.to_string()))?;

            Ok(Box::pin(GrpcEventStream {
                inner: response.into_inner(),
            }) as EventStream)
        })
    }

    fn set_task_push_config<'a>(
        &'a self,
        _params: &'a TaskPushConfig,
    ) -> Pin<Box<dyn Future<Output = Result<TaskPushConfig>> + Send + 'a>> {
        Box::pin(async move {
            Err(A2AError::UnsupportedOperation(
                "push config not available over gRPC".into(),
            ))
        })
    }

    fn get_task_push_config<'a>(
        &'a self,
        _params: &'a GetTaskPushConfigParams,
    ) -> Pin<Box<dyn Future<Output = Result<TaskPushConfig>> + Send + 'a>> {
        Box::pin(async move {
            Err(A2AError::UnsupportedOperation(
                "push config not available over gRPC".into(),
            ))
        })
    }

    fn list_task_push_config<'a>(
        &'a self,
        _params: &'a ListTaskPushConfigParams,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<TaskPushConfig>>> + Send + 'a>> {
        Box::pin(async move {
            Err(A2AError::UnsupportedOperation(
                "push config not available over gRPC".into(),
            ))
        })
    }

    fn delete_task_push_config<'a>(
        &'a self,
        _params: &'a DeleteTaskPushConfigParams,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>> {
        Box::pin(async move {
            Err(A2AError::UnsupportedOperation(
                "push config not available over gRPC".into(),
            ))
        })
    }

    fn get_agent_card(&self) -> Pin<Box<dyn Future<Output = Result<AgentCard>> + Send + '_>> {
        Box::pin(async move {
            Err(A2AError::UnsupportedOperation(
                "agent card discovery not available over gRPC".into(),
            ))
        })
    }
}

/// Adapts a tonic streaming response into a [`Stream`] of [`Event`]s.
struct GrpcEventStream {
    inner: tonic::Streaming<proto::StreamResponse>,
}

impl Stream for GrpcEventStream {
    type Item = Result<Event>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match Pin::new(&mut self.inner).poll_next(cx) {
            Poll::Ready(Some(Ok(response))) => match convert_stream_response(response) {
                Some(event) => Poll::Ready(Some(Ok(event))),
                None => {
                    // Skip unknown payload types, poll again
                    cx.waker().wake_by_ref();
                    Poll::Pending
                }
            },
            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(A2AError::Other(e.to_string())))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

/// Converts a proto `StreamResponse` to a native [`Event`].
fn convert_stream_response(response: proto::StreamResponse) -> Option<Event> {
    match response.payload {
        Some(proto::stream_response::Payload::StatusUpdate(update)) => {
            let status = update
                .status
                .map_or_else(|| TaskStatus::new(TaskState::Unknown), TaskStatus::from);
            let mut event =
                TaskStatusUpdateEvent::new(update.task_id, update.context_id, status, false);
            event.metadata = update
                .metadata
                .and_then(struct_to_hashmap)
                .unwrap_or_default();
            Some(Event::StatusUpdate(event))
        }
        Some(proto::stream_response::Payload::ArtifactUpdate(update)) => {
            let artifact = update
                .artifact
                .map_or_else(|| Artifact::new("", vec![]), Artifact::from);
            let mut event =
                TaskArtifactUpdateEvent::new(update.task_id, update.context_id, artifact);
            event.append = update.append;
            event.last_chunk = update.last_chunk;
            event.metadata = update
                .metadata
                .and_then(struct_to_hashmap)
                .unwrap_or_default();
            Some(Event::ArtifactUpdate(event))
        }
        _ => None,
    }
}
