//! The `Executor` trait — how plugins send work to PHP workers.
//!
//! `folk-core` provides the concrete implementation, backed by the worker pool.

use std::collections::HashMap;
use std::sync::Arc;

use anyhow::Result;
use async_trait::async_trait;
use bytes::Bytes;
use tokio::sync::mpsc;

use crate::response::ResponseChunk;

/// Sends a serialized payload to a PHP worker and returns the response.
///
/// Plugins call this; they never see the worker pool directly.
#[async_trait]
pub trait Executor: Send + Sync + 'static {
    /// Send `payload` to a worker and return the response bytes.
    /// Uses the RPC method name "dispatch" by default.
    async fn execute(&self, payload: Bytes) -> Result<Bytes> {
        self.execute_method("dispatch", payload).await
    }

    /// Send `payload` to a worker using a specific RPC method name.
    async fn execute_method(&self, method: &str, payload: Bytes) -> Result<Bytes>;

    /// Send structured data directly (no serialization).
    ///
    /// The default implementation falls back to JSON serialization
    /// via [`execute_method`]. Extension runtimes override this to
    /// pass values through the channel without JSON encode/decode.
    async fn execute_value(
        &self,
        method: &str,
        payload: serde_json::Value,
    ) -> Result<serde_json::Value> {
        let bytes = serde_json::to_vec(&payload)?;
        let result = self.execute_method(method, Bytes::from(bytes)).await?;
        Ok(serde_json::from_slice(&result)?)
    }

    /// Like [`execute_value`], but also returns the `request_id` under which the
    /// request was executed — the same id exposed to PHP via `folk_request_id()`.
    ///
    /// Plugins that maintain an access log (e.g. HTTP) use this to record the id
    /// for log correlation. The default implementation calls [`execute_value`]
    /// and returns an empty id (runtimes that don't track one).
    async fn execute_value_traced(
        &self,
        method: &str,
        payload: serde_json::Value,
    ) -> Result<(serde_json::Value, Arc<str>)> {
        let value = self.execute_value(method, payload).await?;
        Ok((value, Arc::from("")))
    }

    /// Send a request and receive the response as a stream of [`ResponseChunk`]s.
    ///
    /// The returned receiver emits zero or more [`ResponseChunk::Body`] chunks
    /// preceded by a [`ResponseChunk::Headers`] chunk, and terminated by
    /// [`ResponseChunk::End`]. The `Arc<str>` is the `request_id` (UUID v7) for
    /// log correlation — the same id exposed to PHP via `folk_request_id()`.
    ///
    /// Callers must drain the receiver to completion (`End` or channel close) to
    /// allow the worker to accept the next request.
    ///
    /// The default implementation wraps [`execute_value_traced`] into a
    /// single-shot stream for runtimes that do not support true streaming. The
    /// value is emitted verbatim as a [`ResponseChunk::Return`] — consumers
    /// (HTTP / gRPC / jobs) interpret it themselves.
    async fn execute_streamed(
        &self,
        method: &str,
        payload: serde_json::Value,
    ) -> Result<(mpsc::Receiver<ResponseChunk>, Arc<str>)> {
        let (value, id) = self.execute_value_traced(method, payload).await?;
        let (tx, rx) = mpsc::channel(4);
        let _ = tx.send(ResponseChunk::Return(value)).await;
        let _ = tx.send(ResponseChunk::End).await;
        Ok((rx, id))
    }

    /// Like [`execute_streamed`], but the request body is delivered to PHP as a
    /// stream of [`Bytes`] chunks rather than embedded in `payload`.
    ///
    /// `body_rx` yields raw body chunks as they arrive from the transport; PHP
    /// pulls them via `folk_read()` / `folk_read_all()`. The channel closing
    /// signals end-of-body. This is the request-side counterpart to the
    /// [`ResponseChunk`] response stream.
    ///
    /// The default implementation does not support request body streaming: it
    /// discards `body_rx` and delegates to [`execute_streamed`]. Runtimes backed
    /// by a real worker pool override this to thread `body_rx` to the worker.
    async fn execute_streamed_with_body(
        &self,
        method: &str,
        payload: serde_json::Value,
        body_rx: mpsc::Receiver<Bytes>,
    ) -> Result<(mpsc::Receiver<ResponseChunk>, Arc<str>)> {
        drop(body_rx);
        self.execute_streamed(method, payload).await
    }
}

/// Convert a single-shot PHP response Value into a stream of ResponseChunks.
///
/// Sends `Headers`, optionally `Body`, then `End`. Errors are logged and
/// result in an empty body.
pub async fn value_to_chunks(value: serde_json::Value, tx: &mpsc::Sender<ResponseChunk>) {
    let status =
        u16::try_from(value.get("status").and_then(|v| v.as_u64()).unwrap_or(200)).unwrap_or(200);

    let headers: HashMap<String, String> = value
        .get("headers")
        .and_then(|v| v.as_object())
        .map(|obj| {
            obj.iter()
                .filter_map(|(k, v)| Some((k.clone(), v.as_str()?.to_string())))
                .collect()
        })
        .unwrap_or_default();

    let _ = tx.send(ResponseChunk::Headers { status, headers }).await;

    let body_str = value
        .get("body")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string();
    let body_encoding = value
        .get("body_encoding")
        .and_then(|v| v.as_str())
        .map(str::to_string);

    let body_bytes: Bytes = if body_encoding.as_deref() == Some("base64") {
        use base64::Engine;
        match base64::engine::general_purpose::STANDARD.decode(&body_str) {
            Ok(b) => Bytes::from(b),
            Err(e) => {
                tracing::error!(error = ?e, "value_to_chunks: base64 decode failed");
                Bytes::new()
            }
        }
    } else {
        Bytes::from(body_str.into_bytes())
    };

    if !body_bytes.is_empty() {
        let _ = tx.send(ResponseChunk::Body(body_bytes)).await;
    }

    let _ = tx.send(ResponseChunk::End).await;
}

/// Blanket impl: an `Arc<dyn Executor>` is also an `Executor`.
#[async_trait]
impl<T: Executor + ?Sized> Executor for Arc<T> {
    async fn execute(&self, payload: Bytes) -> Result<Bytes> {
        (**self).execute(payload).await
    }

    async fn execute_method(&self, method: &str, payload: Bytes) -> Result<Bytes> {
        (**self).execute_method(method, payload).await
    }

    async fn execute_value(
        &self,
        method: &str,
        payload: serde_json::Value,
    ) -> Result<serde_json::Value> {
        (**self).execute_value(method, payload).await
    }

    async fn execute_value_traced(
        &self,
        method: &str,
        payload: serde_json::Value,
    ) -> Result<(serde_json::Value, Arc<str>)> {
        (**self).execute_value_traced(method, payload).await
    }

    async fn execute_streamed(
        &self,
        method: &str,
        payload: serde_json::Value,
    ) -> Result<(mpsc::Receiver<ResponseChunk>, Arc<str>)> {
        (**self).execute_streamed(method, payload).await
    }

    async fn execute_streamed_with_body(
        &self,
        method: &str,
        payload: serde_json::Value,
        body_rx: mpsc::Receiver<Bytes>,
    ) -> Result<(mpsc::Receiver<ResponseChunk>, Arc<str>)> {
        (**self)
            .execute_streamed_with_body(method, payload, body_rx)
            .await
    }
}
