//! HTTP transport for ACP
//!
//! This module provides a transport implementation that communicates with ACP agents
//! via HTTP POST for requests and Server-Sent Events (SSE) for notifications.
//!
//! # Protocol
//!
//! - **Requests**: HTTP POST to `/jsonrpc` endpoint with JSON-RPC payload
//! - **Notifications**: SSE stream from `/events` endpoint for real-time updates
//!
//! # Architecture
//!
//! The transport maintains two connections:
//! 1. HTTP client for sending requests (reqwest::Client)
//! 2. SSE client for receiving notifications (eventsource stream)
//!
//! Responses and notifications are multiplexed into a single receive channel
//! using a background task that routes SSE events based on their ID.
//!
//! # Lifecycle
//!
//! 1. Create HttpTransport with base URL
//! 2. Call `connect()` to establish SSE connection
//! 3. Use `send()` for requests, `recv()` for responses and notifications
//! 4. Call `close()` to shut down connections

use super::{AcpTransport, TransportResult};
use async_trait::async_trait;
use futures::StreamExt;
use reqwest::Client;
use serde_json::Value;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, Mutex};
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};

/// HTTP transport implementation
///
/// Communicates with ACP agents via HTTP POST (requests) and SSE (notifications).
///
/// # Example
///
/// ```no_run
/// use dirigent_core::connectors::acp::transport::{AcpTransport, HttpTransport};
/// use serde_json::json;
///
/// # async fn example() -> anyhow::Result<()> {
/// // Connect to ACP agent HTTP server
/// let mut transport = HttpTransport::new("http://localhost:8080");
///
/// // Connect (establishes SSE stream)
/// transport.connect().await?;
///
/// // Send initialize request
/// transport.send(json!({
///     "jsonrpc": "2.0",
///     "method": "initialize",
///     "id": 1,
///     "params": {
///         "protocolVersion": 1,
///         "clientCapabilities": {}
///     }
/// })).await?;
///
/// // Receive response
/// if let Some(response) = transport.recv().await? {
///     println!("Initialized: {:?}", response);
/// }
///
/// // Clean up
/// transport.close().await?;
/// # Ok(())
/// # }
/// ```
pub struct HttpTransport {
    /// Base URL of the ACP server (e.g., "http://localhost:8080")
    base_url: String,

    /// Client ID for SSE subscription and RPC identification
    ///
    /// This UUID is generated when connecting to the ACP server and is used:
    /// - As query parameter for SSE endpoint: `/events?client_id=...`
    /// - In RPC requests to identify this client (future: via X-Client-Id header)
    client_id: String,

    /// Request timeout
    timeout: Duration,

    /// HTTP client for sending requests
    client: Client,

    /// Channel for receiving messages (responses + notifications)
    ///
    /// This channel is fed by the SSE background task and HTTP responses.
    /// The recv() method reads from this channel.
    rx: Arc<Mutex<Option<mpsc::Receiver<Value>>>>,

    /// Sender for the receive channel (used internally)
    tx: Arc<Mutex<Option<mpsc::Sender<Value>>>>,

    /// Join handle for the SSE background task
    sse_task: Arc<Mutex<Option<JoinHandle<()>>>>,

    /// Flag to indicate if the transport is connected
    connected: Arc<Mutex<bool>>,
}

impl HttpTransport {
    /// Create a new HttpTransport
    ///
    /// # Arguments
    ///
    /// * `base_url` - Base URL of the ACP server (e.g., "http://localhost:8080")
    /// * `client_id` - Client ID for SSE subscription (UUID string)
    ///
    /// # Returns
    ///
    /// A new HttpTransport that is not yet connected. Call `connect()` to
    /// establish the SSE connection.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use dirigent_core::connectors::acp::transport::HttpTransport;
    ///
    /// let client_id = uuid::Uuid::new_v4().to_string();
    /// let transport = HttpTransport::new("http://localhost:8080", client_id);
    /// ```
    pub fn new(base_url: impl Into<String>, client_id: impl Into<String>) -> Self {
        let base_url = base_url.into();
        let client_id = client_id.into();

        info!(
            base_url = %base_url,
            client_id = %client_id,
            "Creating HTTP transport"
        );

        let (tx, rx) = mpsc::channel(1000); // Buffer up to 1000 messages

        Self {
            base_url,
            client_id,
            timeout: Duration::from_secs(30), // Default 30 seconds
            client: Client::new(),
            rx: Arc::new(Mutex::new(Some(rx))),
            tx: Arc::new(Mutex::new(Some(tx))),
            sse_task: Arc::new(Mutex::new(None)),
            connected: Arc::new(Mutex::new(false)),
        }
    }

    /// Set the request timeout
    ///
    /// Must be called before `connect()`.
    pub fn set_timeout(&mut self, timeout: Duration) {
        self.timeout = timeout;
    }

    /// Start the SSE background task
    ///
    /// This task connects to the `/events` endpoint and forwards all SSE events
    /// to the receive channel.
    async fn start_sse_task(&self) -> TransportResult<()> {
        let url = format!("{}/events?client_id={}", self.base_url, self.client_id);
        let tx = self
            .tx
            .lock()
            .await
            .clone()
            .ok_or("Transport already closed")?;

        info!(url = %url, client_id = %self.client_id, "Starting SSE subscription");

        // Create SSE client using reqwest
        let client = self.client.clone();
        let sse_url = url.clone();

        let handle = tokio::spawn(async move {
            debug!("SSE task started");

            // Connect to SSE endpoint
            let response = match client.get(&sse_url).send().await {
                Ok(r) => r,
                Err(e) => {
                    error!(error = %e, "Failed to connect to SSE endpoint");
                    return;
                }
            };

            if !response.status().is_success() {
                error!(status = %response.status(), "SSE endpoint returned error");
                return;
            }

            // Read SSE stream using eventsource-client
            use eventsource_client::{Client as EventSourceClient, ClientBuilder, SSE};

            let sse_client = match ClientBuilder::for_url(&sse_url) {
                Ok(c) => c.build(),
                Err(e) => {
                    error!(error = %e, "Failed to create SSE client");
                    return;
                }
            };

            let mut stream = sse_client.stream();

            // Process SSE events
            while let Some(event_result) = stream.next().await {
                match event_result {
                    Ok(SSE::Connected(_)) => {
                        info!("SSE stream connected");
                    }
                    Ok(SSE::Event(event)) => {
                        debug!(event = ?event, "Received SSE event");

                        // Parse the event data as JSON
                        match serde_json::from_str::<Value>(&event.data) {
                            Ok(message) => {
                                let masked_message = dirigent_protocol::log_utils::format_for_log(&message);
                                debug!(message = %masked_message, "Parsed SSE message");

                                // Send to receive channel
                                if let Err(e) = tx.send(message).await {
                                    warn!(error = %e, "Failed to forward SSE message (receiver dropped)");
                                    break;
                                }
                            }
                            Err(e) => {
                                warn!(error = %e, data = %event.data, "Failed to parse SSE event data");
                            }
                        }
                    }
                    Ok(SSE::Comment(comment)) => {
                        debug!(comment = %comment, "Received SSE comment");
                    }
                    Err(e) => {
                        error!(error = %e, "SSE stream error");
                        break;
                    }
                }
            }

            info!("SSE task ended");
        });

        // Store the task handle
        *self.sse_task.lock().await = Some(handle);

        Ok(())
    }
}

#[async_trait]
impl AcpTransport for HttpTransport {
    async fn connect(&mut self) -> TransportResult<()> {
        let mut connected = self.connected.lock().await;

        if *connected {
            return Err("Transport already connected".into());
        }

        info!(base_url = %self.base_url, "Connecting HTTP transport");

        // Start SSE background task
        self.start_sse_task().await?;

        *connected = true;

        info!("HTTP transport connected");

        Ok(())
    }

    async fn send(&mut self, message: Value) -> TransportResult<()> {
        let connected = self.connected.lock().await;
        if !*connected {
            return Err("Transport not connected".into());
        }
        drop(connected); // Release lock

        // Send HTTP POST to /jsonrpc
        let url = format!("{}/jsonrpc", self.base_url);

        let masked_message = dirigent_protocol::log_utils::format_for_log(&message);
        debug!(
            url = %url,
            client_id = %self.client_id,
            message = %masked_message,
            "Sending HTTP request with X-Client-Id header"
        );

        let response = self
            .client
            .post(&url)
            .header("X-Client-ID", &self.client_id)
            .json(&message)
            .send()
            .await
            .map_err(|e| format!("Failed to send HTTP request: {}", e))?;

        let status = response.status();

        if !status.is_success() {
            let error_body = response
                .text()
                .await
                .unwrap_or_else(|_| "<failed to read body>".to_string());
            return Err(format!("HTTP request failed with status {}: {}", status, error_body).into());
        }

        // Parse the response
        let response_json: Value = response
            .json()
            .await
            .map_err(|e| format!("Failed to parse response JSON: {}", e))?;

        let masked_response = dirigent_protocol::log_utils::format_for_log(&response_json);
        debug!(response = %masked_response, "Received HTTP response");

        // Forward the response to the receive channel
        let tx = self
            .tx
            .lock()
            .await
            .clone()
            .ok_or("Transport closed")?;
        tx.send(response_json)
            .await
            .map_err(|e| format!("Failed to forward response: {}", e))?;

        Ok(())
    }

    async fn recv(&mut self) -> TransportResult<Option<Value>> {
        let mut rx_lock = self.rx.lock().await;
        let rx = rx_lock
            .as_mut()
            .ok_or("Transport not connected (no receiver)")?;

        // Receive from the channel (blocks until message available)
        match rx.recv().await {
            Some(message) => {
                let masked_message = dirigent_protocol::log_utils::format_for_log(&message);
                debug!(message = %masked_message, "Received message");
                Ok(Some(message))
            }
            None => {
                // Channel closed (SSE task ended)
                debug!("Receive channel closed");
                Ok(None)
            }
        }
    }

    async fn close(&mut self) -> TransportResult<()> {
        info!("Closing HTTP transport");

        // Mark as disconnected
        *self.connected.lock().await = false;

        // Drop the sender to close the channel
        *self.tx.lock().await = None;

        // Cancel the SSE task
        let mut sse_task_lock = self.sse_task.lock().await;
        if let Some(handle) = sse_task_lock.take() {
            debug!("Aborting SSE task");
            handle.abort();

            // Wait for task to finish (with timeout)
            match tokio::time::timeout(std::time::Duration::from_secs(2), handle).await {
                Ok(Ok(())) => {
                    info!("SSE task finished gracefully");
                }
                Ok(Err(e)) if e.is_cancelled() => {
                    info!("SSE task was cancelled");
                }
                Ok(Err(e)) => {
                    warn!(error = %e, "SSE task panicked");
                }
                Err(_) => {
                    warn!("SSE task did not finish within timeout");
                }
            }
        }

        // Drop the receiver
        *self.rx.lock().await = None;

        info!("HTTP transport closed");

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use std::net::SocketAddr;
    use axum::{
        routing::{get, post},
        Json, Router,
    };
    use axum::response::{sse::Event, Sse};
    use futures::stream;
    use std::convert::Infallible;
    use std::time::Duration;

    /// Helper to find a free port for the test server
    async fn find_free_port() -> u16 {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("Failed to bind to random port");
        listener
            .local_addr()
            .expect("Failed to get local addr")
            .port()
    }

    /// Helper to start a mock ACP HTTP server for testing
    async fn start_mock_server() -> (String, tokio::task::JoinHandle<()>) {

        let port = find_free_port().await;
        let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();

        // Create router
        let app = Router::new()
            .route("/jsonrpc", post(handle_jsonrpc))
            .route("/events", get(handle_sse));

        // Spawn server
        let handle = tokio::spawn(async move {
            let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
            axum::serve(listener, app).await.unwrap();
        });

        // Wait a bit for server to start
        tokio::time::sleep(Duration::from_millis(100)).await;

        (format!("http://127.0.0.1:{}", port), handle)
    }

    async fn handle_jsonrpc(Json(request): Json<Value>) -> Json<Value> {
        // Echo back a successful response
        let id = request.get("id").cloned().unwrap_or(json!(null));
        let method = request
            .get("method")
            .and_then(|m| m.as_str())
            .unwrap_or("");

        let result = match method {
            "initialize" => json!({
                "protocolVersion": 1,
                "agentCapabilities": {}
            }),
            "session/new" => json!({
                "sessionId": "test-session-123"
            }),
            _ => json!({}),
        };

        Json(json!({
            "jsonrpc": "2.0",
            "id": id,
            "result": result
        }))
    }

    async fn handle_sse() -> Sse<impl futures::Stream<Item = Result<Event, Infallible>>> {
        // Send a few test events
        let stream = stream::iter(vec![
            Ok(Event::default().data(
                json!({
                    "jsonrpc": "2.0",
                    "method": "session/update",
                    "params": {
                        "sessionId": "test-session-123",
                        "update": {
                            "type": "messageStarted"
                        }
                    }
                })
                .to_string(),
            )),
            Ok(Event::default().data(
                json!({
                    "jsonrpc": "2.0",
                    "method": "session/update",
                    "params": {
                        "sessionId": "test-session-123",
                        "update": {
                            "type": "messageCompleted"
                        }
                    }
                })
                .to_string(),
            )),
        ]);

        Sse::new(stream)
    }

    #[tokio::test]
    async fn test_http_transport_basic_flow() {
        let (base_url, _server_handle) = start_mock_server().await;

        // Create transport
        let mut transport = HttpTransport::new(&base_url, "test-client");

        // Connect
        transport.connect().await.expect("Failed to connect");

        // Test 1: Initialize
        let init_request = json!({
            "jsonrpc": "2.0",
            "method": "initialize",
            "id": 1,
            "params": {
                "protocolVersion": 1,
                "clientCapabilities": {}
            }
        });

        transport
            .send(init_request)
            .await
            .expect("Failed to send init request");

        let init_response = transport
            .recv()
            .await
            .expect("Failed to receive init response")
            .expect("Got None instead of response");

        assert_eq!(init_response["jsonrpc"], "2.0");
        assert_eq!(init_response["id"], 1);
        assert!(init_response["result"]["protocolVersion"].is_number());

        // Test 2: Create session
        let session_request = json!({
            "jsonrpc": "2.0",
            "method": "session/new",
            "id": 2,
            "params": {
                "cwd": ".",
                "mcpServers": []
            }
        });

        transport
            .send(session_request)
            .await
            .expect("Failed to send session request");

        let session_response = transport
            .recv()
            .await
            .expect("Failed to receive session response")
            .expect("Got None instead of response");

        assert_eq!(session_response["jsonrpc"], "2.0");
        assert_eq!(session_response["id"], 2);

        // Test 3: Receive SSE notifications
        // The mock server sends 2 notifications
        let notif1 = transport
            .recv()
            .await
            .expect("Failed to receive notification")
            .expect("Got None instead of notification");
        assert_eq!(notif1["method"], "session/update");

        let notif2 = transport
            .recv()
            .await
            .expect("Failed to receive notification")
            .expect("Got None instead of notification");
        assert_eq!(notif2["method"], "session/update");

        // Close
        transport.close().await.expect("Failed to close");
    }

    #[tokio::test]
    async fn test_http_transport_sse_notifications() {
        let (base_url, _server_handle) = start_mock_server().await;

        let mut transport = HttpTransport::new(&base_url, "test-client");
        transport.connect().await.expect("Failed to connect");

        // Receive SSE notifications without sending any requests
        let notif1 = transport
            .recv()
            .await
            .expect("Failed to receive notification")
            .expect("Got None instead of notification");
        assert_eq!(notif1["method"], "session/update");

        transport.close().await.expect("Failed to close");
    }

    #[tokio::test]
    async fn test_http_transport_concurrent_requests() {
        let (base_url, _server_handle) = start_mock_server().await;

        let mut transport = HttpTransport::new(&base_url, "test-client");
        transport.connect().await.expect("Failed to connect");

        // Send multiple requests concurrently
        let transport = Arc::new(Mutex::new(transport));

        let mut handles = vec![];
        for i in 0..5 {
            let t = Arc::clone(&transport);
            let handle = tokio::spawn(async move {
                let mut transport = t.lock().await;
                transport
                    .send(json!({
                        "jsonrpc": "2.0",
                        "method": "session/new",
                        "id": i + 10,
                        "params": {
                            "cwd": ".",
                            "mcpServers": []
                        }
                    }))
                    .await
            });
            handles.push(handle);
        }

        // All sends should succeed
        for handle in handles {
            handle
                .await
                .expect("Task panicked")
                .expect("Send failed");
        }

        // Close
        let mut transport = transport.lock().await;
        transport.close().await.expect("Failed to close");
    }

    #[tokio::test]
    async fn test_http_transport_error_response() {
        // This test would require a mock server that returns errors
        // For now, we'll skip it as the basic mock server always succeeds
    }
}
