//! Tunnel RPC framework - async implementation
//!
//! Provides bidirectional RPC over any async channel pair.
//! Used with `#[tunnel_service]` proc macro to generate Proxy and Dispatcher.

use serde_json::Value;
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::{mpsc, oneshot};

// ── Structured error types ────────────────────────────────────

/// Wire-level tunnel error.
///
/// Returned by `TunnelCaller::call()`. Distinguishes between:
/// - `Service(Value)`: remote handler returned a business error (serialized as JSON)
/// - `Transport(String)`: communication failure (disconnect, timeout, serialization error)
#[derive(Debug, Clone)]
pub enum TunnelError {
    /// Remote handler returned a business error (JSON value).
    Service(Value),
    /// Transport-level failure (disconnect, timeout, etc).
    Transport(String),
}

impl std::fmt::Display for TunnelError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            TunnelError::Service(v) => write!(f, "service error: {}", v),
            TunnelError::Transport(msg) => write!(f, "transport error: {}", msg),
        }
    }
}

impl std::error::Error for TunnelError {}

/// Typed RPC error for tunnel service methods.
///
/// Returned by proxy methods generated by `#[tunnel_service]`.
/// - `Service(E)`: business error from the remote handler
/// - `Transport(String)`: communication failure
///
/// Local implementations only return `Service(E)`.
/// `Transport` is injected by the proxy when communication fails.
#[derive(Debug, Clone)]
pub enum RpcError<E> {
    /// Business error from the service handler.
    Service(E),
    /// Transport-level failure (disconnect, timeout, serialization error).
    Transport(String),
}

impl<E: std::fmt::Display> std::fmt::Display for RpcError<E> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RpcError::Service(e) => write!(f, "service error: {}", e),
            RpcError::Transport(msg) => write!(f, "transport error: {}", msg),
        }
    }
}

impl<E: std::fmt::Debug + std::fmt::Display> std::error::Error for RpcError<E> {}

impl<E> RpcError<E> {
    /// Wrap a business error into an RpcError.
    pub fn service(e: E) -> Self {
        RpcError::Service(e)
    }
}

// ── Message protocol ──────────────────────────────────────────

pub struct RequestMessage {
    pub id: String,
    pub method: String,
    pub params: Value,
}

pub struct ResponseMessage {
    pub id: String,
    pub result: Result<Value, Value>,
}

pub enum ParsedMessage {
    Request(RequestMessage),
    Response(ResponseMessage),
}

pub fn make_request(msg: &RequestMessage) -> String {
    serde_json::to_string(&serde_json::json!({
        "type": "request",
        "id": msg.id,
        "method": msg.method,
        "params": msg.params,
    })).unwrap()
}

pub fn make_response(msg: &ResponseMessage) -> String {
    let value = match &msg.result {
        Ok(v) => serde_json::json!({"type": "response", "id": msg.id, "result": v}),
        Err(e) => serde_json::json!({"type": "response", "id": msg.id, "error": e}),
    };
    serde_json::to_string(&value).unwrap()
}

pub fn parse_message(text: &str) -> Option<ParsedMessage> {
    let v: Value = serde_json::from_str(text).ok()?;
    let msg_type = v.get("type")?.as_str()?;
    match msg_type {
        "request" => {
            let id = v.get("id")?.as_str()?.to_string();
            let method = v.get("method")?.as_str()?.to_string();
            let params = v.get("params")?.clone();
            Some(ParsedMessage::Request(RequestMessage { id, method, params }))
        }
        "response" => {
            let id = v.get("id")?.as_str()?.to_string();
            if let Some(error) = v.get("error") {
                Some(ParsedMessage::Response(ResponseMessage {
                    id,
                    result: Err(error.clone()),
                }))
            } else {
                let result = v.get("result")?.clone();
                Some(ParsedMessage::Response(ResponseMessage {
                    id,
                    result: Ok(result),
                }))
            }
        }
        _ => None,
    }
}

// ── Dispatch trait ─────────────────────────────────────────────

/// Trait for dispatching incoming RPC requests to local handlers.
/// Implementations are generated by `#[tunnel_service]` macro.
#[async_trait::async_trait]
pub trait Dispatch: Send + Sync {
    /// Returns the service prefix this dispatcher handles (e.g., "MyService")
    fn service_name(&self) -> &str;

    /// Dispatch a method call. method is the full "ServiceName.method_name" string.
    ///
    /// Returns `Ok(Value)` on success, `Err(Value)` for business errors
    /// (the error is the serialized service error type).
    async fn dispatch(&self, method: &str, params: Value) -> Result<Value, Value>;
}

// ── TunnelCaller trait ─────────────────────────────────────────

/// Abstraction for making RPC calls through a tunnel.
///
/// Implemented by `TunnelEndpoint` (direct) and `TunnelClient` (managed connection).
/// Proxy structs generated by `#[tunnel_service]` hold `Arc<dyn TunnelCaller>`.
#[async_trait::async_trait]
pub trait TunnelCaller: Send + Sync {
    async fn call(&self, method: &str, params: Value) -> Result<Value, TunnelError>;
}

// ── TunnelEndpoint ─────────────────────────────────────────────

type PendingMap = HashMap<String, oneshot::Sender<Result<Value, Value>>>;

/// Async bidirectional RPC endpoint over channel pairs.
///
/// Create with `TunnelEndpoint::new()`, which spawns an async reader task.
/// Use `call()` to make RPC requests to the remote side.
pub struct TunnelEndpoint {
    outgoing_tx: mpsc::UnboundedSender<String>,
    pending: Arc<Mutex<PendingMap>>,
    timeout: Duration,
}

impl TunnelEndpoint {
    /// Create a new TunnelEndpoint and spawn the reader loop.
    ///
    /// - `incoming_rx`: receives messages from the remote side
    /// - `outgoing_tx`: sends messages to the remote side
    /// - `dispatchers`: local RPC handlers for incoming requests
    /// - `timeout`: how long `call()` waits for a response
    pub fn new(
        incoming_rx: mpsc::UnboundedReceiver<String>,
        outgoing_tx: mpsc::UnboundedSender<String>,
        dispatchers: Vec<Box<dyn Dispatch>>,
        timeout: Duration,
    ) -> Arc<Self> {
        let pending: Arc<Mutex<PendingMap>> = Arc::new(Mutex::new(HashMap::new()));

        let endpoint = Arc::new(Self {
            outgoing_tx: outgoing_tx.clone(),
            pending: pending.clone(),
            timeout,
        });

        // Spawn async reader loop
        let dispatchers = Arc::new(dispatchers);
        tokio::spawn(reader_loop(incoming_rx, outgoing_tx, dispatchers, pending));

        endpoint
    }

    /// Make an RPC call to the remote side.
    ///
    /// Sends a request and waits for the response (with timeout).
    pub async fn call(
        &self,
        method: &str,
        params: Value,
    ) -> Result<Value, TunnelError> {
        let id = generate_id();

        let (tx, rx) = oneshot::channel();

        // Register pending request
        {
            let mut map = self.pending.lock().unwrap();
            map.insert(id.clone(), tx);
        }

        // Send request
        let msg = make_request(&RequestMessage {
            id: id.clone(),
            method: method.to_string(),
            params,
        });

        if self.outgoing_tx.send(msg).is_err() {
            // Clean up pending
            let mut map = self.pending.lock().unwrap();
            map.remove(&id);
            return Err(TunnelError::Transport("tunnel disconnected".to_string()));
        }

        // Wait for response with timeout
        match tokio::time::timeout(self.timeout, rx).await {
            Ok(Ok(result)) => match result {
                Ok(v) => Ok(v),
                Err(e) => Err(TunnelError::Service(e)),
            },
            Ok(Err(_)) => {
                // oneshot sender dropped — endpoint shutting down
                Err(TunnelError::Transport("tunnel endpoint closed".to_string()))
            }
            Err(_) => {
                // Timeout — clean up pending
                let mut map = self.pending.lock().unwrap();
                map.remove(&id);
                Err(TunnelError::Transport(format!("tunnel call timeout: {}", method)))
            }
        }
    }
}

#[async_trait::async_trait]
impl TunnelCaller for TunnelEndpoint {
    async fn call(&self, method: &str, params: Value) -> Result<Value, TunnelError> {
        TunnelEndpoint::call(self, method, params).await
    }
}

/// Async reader loop — processes incoming messages from the remote side.
///
/// Request dispatch is spawned as independent tasks to avoid blocking the
/// reader loop. This prevents deadlocks when a dispatch handler makes a
/// nested RPC call back through the same tunnel (the response must be
/// received by this reader loop while dispatch is still running).
async fn reader_loop(
    mut incoming_rx: mpsc::UnboundedReceiver<String>,
    outgoing_tx: mpsc::UnboundedSender<String>,
    dispatchers: Arc<Vec<Box<dyn Dispatch>>>,
    pending: Arc<Mutex<PendingMap>>,
) {
    while let Some(text) = incoming_rx.recv().await {
        let parsed = match parse_message(&text) {
            Some(p) => p,
            None => continue,
        };

        match parsed {
            ParsedMessage::Request(req) => {
                // Spawn independent task to avoid blocking reader loop.
                // Critical for nested RPC: dispatch may call back through
                // this tunnel, and the response must be readable here.
                let outgoing_tx = outgoing_tx.clone();
                let dispatchers = dispatchers.clone();
                tokio::spawn(async move {
                    let mut result: Result<Value, Value> = Err(
                        serde_json::to_value(format!("no handler for method: {}", req.method))
                            .unwrap_or(Value::String("no handler".to_string()))
                    );
                    for dispatcher in dispatchers.iter() {
                        if req.method.starts_with(dispatcher.service_name()) {
                            result = dispatcher.dispatch(&req.method, req.params.clone()).await;
                            break;
                        }
                    }

                    let response = make_response(&ResponseMessage {
                        id: req.id,
                        result,
                    });
                    let _ = outgoing_tx.send(response);
                });
            }
            ParsedMessage::Response(resp) => {
                // Find and notify pending caller
                let sender = {
                    let mut map = pending.lock().unwrap();
                    map.remove(&resp.id)
                };
                if let Some(sender) = sender {
                    let _ = sender.send(resp.result);
                }
            }
        }
    }

    // Channel closed — clean up all pending requests
    let mut map = pending.lock().unwrap();
    for (_, sender) in map.drain() {
        let _ = sender.send(Err(Value::String("tunnel disconnected".to_string())));
    }
}

// ── Utilities ──────────────────────────────────────────────────

fn generate_id() -> String {
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("rpc_{}", n)
}

/// Create a TunnelEndpoint with wired-up channels for WebSocket bridging.
///
/// Returns:
/// - `endpoint`: for making RPC calls
/// - `ws_to_ep_tx`: feed incoming WS text messages to the endpoint
/// - `ep_to_ws_rx`: read outgoing text messages from endpoint to send over WS
pub fn create_endpoint(
    dispatchers: Vec<Box<dyn Dispatch>>,
    timeout: Duration,
) -> (Arc<TunnelEndpoint>, mpsc::UnboundedSender<String>, mpsc::UnboundedReceiver<String>) {
    let (ws_to_ep_tx, ws_to_ep_rx) = mpsc::unbounded_channel();
    let (ep_to_ws_tx, ep_to_ws_rx) = mpsc::unbounded_channel();
    let endpoint = TunnelEndpoint::new(ws_to_ep_rx, ep_to_ws_tx, dispatchers, timeout);
    (endpoint, ws_to_ep_tx, ep_to_ws_rx)
}

/// Create a pair of connected channel pairs for testing.
///
/// Returns ((incoming_rx_a, outgoing_tx_a), (incoming_rx_b, outgoing_tx_b))
/// where A's outgoing connects to B's incoming and vice versa.
pub fn channel_pair() -> (
    (mpsc::UnboundedReceiver<String>, mpsc::UnboundedSender<String>),
    (mpsc::UnboundedReceiver<String>, mpsc::UnboundedSender<String>),
) {
    let (a_to_b_tx, a_to_b_rx) = mpsc::unbounded_channel();
    let (b_to_a_tx, b_to_a_rx) = mpsc::unbounded_channel();
    // A's outgoing → B's incoming, B's outgoing → A's incoming
    ((b_to_a_rx, a_to_b_tx), (a_to_b_rx, b_to_a_tx))
}
// ── Transaction guard for tunnel RPC ──────────────────────────

use crate::critical::{TxScope, ConceptMapToken};


/// Transaction guard for tunnel RPC handlers.
///
/// Creates a `TxScope<F>`, passes it to the handler closure, and catches panics.
/// Fatal panics (from `propagate()`) are converted to `Err(Value)` with abort info.
///
/// Returns:
/// - `Ok(inner_result)`: handler completed normally (inner_result is the dispatch Result)
/// - `Err(Value)`: fatal abort occurred (panic caught, abort info extracted)
///
/// Used by `#[tunnel_service]` macro for methods with `#[tx(Type)]`.
#[doc(hidden)]
pub async fn tunnel_tx_guard<T, Fut>(
    handler: impl FnOnce(std::sync::Arc<TxScope<T>>) -> Fut,
) -> Result<Result<Value, Value>, Value>
where
    T: crate::critical::Transaction,
    Fut: std::future::Future<Output = Result<Value, Value>> + Send,
{
    let scope = std::sync::Arc::new(TxScope::<T>::new());
    let scope_clone = scope.clone();

    match crate::critical::run_tx_scope(&scope, handler(scope_clone)).await {
        crate::critical::TxOutcome::Completed(inner_result) => Ok(inner_result),
        crate::critical::TxOutcome::Aborted(token, msg) => {
            tracing::error!(
                "[TX-GUARD-TUNNEL] Fatal abort: {} — {}",
                token.operation_name(), msg
            );
            Err(serde_json::json!({
                "tx_abort": true,
                "operation": token.operation_name(),
                "message": msg.to_string()
            }))
        }
        crate::critical::TxOutcome::UnexpectedPanic => {
            tracing::error!(
                "[TX-GUARD-TUNNEL] Unexpected panic in handler (not from propagate)"
            );
            Err(serde_json::json!({
                "tx_abort": true,
                "operation": "unknown",
                "message": "unexpected panic in tunnel handler"
            }))
        }
    }
}


/// Tunnel transaction guard with database binding.
///
/// Like `tunnel_tx_guard`, but creates a `TxScope` bound to the given database.
/// Auto-commits on success; rolls back on panic.
#[doc(hidden)]
pub async fn tunnel_tx_guard_with_db<T, Fut>(
    db: crate::db::Db,
    handler: impl FnOnce(std::sync::Arc<TxScope<T>>) -> Fut,
) -> Result<Result<Value, Value>, Value>
where
    T: crate::critical::Transaction,
    Fut: std::future::Future<Output = Result<Value, Value>> + Send,
{
    let scope = std::sync::Arc::new(TxScope::<T>::new_with_db(db));
    let scope_clone = scope.clone();

    match crate::critical::run_tx_scope(&scope, handler(scope_clone)).await {
        crate::critical::TxOutcome::Completed(inner_result) => Ok(inner_result),
        crate::critical::TxOutcome::Aborted(token, msg) => {
            tracing::error!(
                "[TX-GUARD-TUNNEL] Fatal abort: {} — {}",
                token.operation_name(), msg
            );
            Err(serde_json::json!({
                "tx_abort": true,
                "operation": token.operation_name(),
                "message": msg.to_string()
            }))
        }
        crate::critical::TxOutcome::UnexpectedPanic => {
            tracing::error!(
                "[TX-GUARD-TUNNEL] Unexpected panic in handler (not from propagate)"
            );
            Err(serde_json::json!({
                "tx_abort": true,
                "operation": "unknown",
                "message": "unexpected panic in tunnel handler"
            }))
        }
    }
}

