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

use std::sync::Arc;

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

use crate::request::RequestPart;
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 [`Self::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 [`Self::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 [`Self::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 [`Self::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 [`Self::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 [`Self::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
    }

    /// Like [`execute_streamed_with_body`], but the request body is a parsed
    /// `multipart/form-data` stream rather than opaque bytes.
    ///
    /// `parts_rx` yields [`RequestPart`]s (Start / Data / End per field); PHP
    /// pulls them via `folk_next_part()` + `folk_part_read()`. The channel
    /// closing signals "no more parts".
    ///
    /// The default implementation discards `parts_rx` and delegates to
    /// [`execute_streamed`]; the worker-pool runtime overrides it.
    ///
    /// [`execute_streamed_with_body`]: Self::execute_streamed_with_body
    /// [`execute_streamed`]: Self::execute_streamed
    async fn execute_streamed_with_parts(
        &self,
        method: &str,
        payload: serde_json::Value,
        parts_rx: mpsc::Receiver<RequestPart>,
    ) -> Result<(mpsc::Receiver<ResponseChunk>, Arc<str>)> {
        drop(parts_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);

    // Ordered pairs, expanding array-valued headers (e.g. multiple Set-Cookie)
    // into one entry each so they are not folded into a single value (#86).
    let mut headers: Vec<(String, String)> = Vec::new();
    if let Some(obj) = value.get("headers").and_then(|v| v.as_object()) {
        for (k, v) in obj {
            match v {
                serde_json::Value::String(s) => headers.push((k.clone(), s.clone())),
                serde_json::Value::Array(items) => {
                    for item in items {
                        if let Some(s) = item.as_str() {
                            headers.push((k.clone(), s.to_string()));
                        }
                    }
                }
                _ => {}
            }
        }
    }

    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;
}

/// Drain a [`ResponseChunk`] stream into a single `serde_json::Value`.
///
/// The inverse of [`value_to_chunks`], used by the non-streaming callers (gRPC,
/// jobs, and the backward-compat `execute` path). The result depends on how the
/// worker finalised the request:
///
/// * [`ResponseChunk::Return`] — the value is returned **verbatim** (gRPC/jobs
///   need their payload intact, not coerced into the HTTP shape);
/// * [`ResponseChunk::Error`] — propagated as an `Err`;
/// * a `Headers`/`Body` stream with no explicit `Return` is reconstructed into
///   the legacy `{status, headers, body}` shape for backward compatibility.
pub async fn chunks_to_value(rx: &mut mpsc::Receiver<ResponseChunk>) -> Result<serde_json::Value> {
    let mut status: u16 = 200;
    let mut headers: Vec<(String, String)> = Vec::new();
    let mut body_bytes: Vec<u8> = Vec::new();
    let mut return_value: Option<serde_json::Value> = None;

    while let Some(chunk) = rx.recv().await {
        match chunk {
            ResponseChunk::Headers {
                status: s,
                headers: h,
            } => {
                status = s;
                headers = h;
            }
            ResponseChunk::Body(b) => body_bytes.extend_from_slice(&b),
            ResponseChunk::Return(v) => return_value = Some(v),
            ResponseChunk::Error(e) => return Err(anyhow::Error::new(e)),
            ResponseChunk::End => break,
        }
    }

    if let Some(value) = return_value {
        return Ok(value);
    }

    // Reconstruct the legacy Value format that HTTP plugin / callers expect.
    // A header seen once maps to a string; a repeated header (e.g. Set-Cookie)
    // maps to an array so no value is lost (#86).
    let mut headers_map: serde_json::Map<String, serde_json::Value> = serde_json::Map::new();
    for (k, v) in headers {
        match headers_map.get_mut(&k) {
            Some(serde_json::Value::Array(arr)) => arr.push(serde_json::Value::String(v)),
            Some(existing) => {
                let prev = existing.take();
                *existing = serde_json::Value::Array(vec![prev, serde_json::Value::String(v)]);
            }
            None => {
                headers_map.insert(k, serde_json::Value::String(v));
            }
        }
    }
    let headers_value = serde_json::Value::Object(headers_map);

    Ok(serde_json::json!({
        "status": status,
        "headers": headers_value,
        "body": String::from_utf8_lossy(&body_bytes),
    }))
}

/// An [`Executor`] that has no PHP runtime behind it.
///
/// Installed in the **master** process's [`PluginContext`](crate::PluginContext)
/// (phase 79): the master supervises and owns shared state but never executes
/// PHP. A `MasterOnly` plugin should never call its executor; if it does, every
/// method logs an error and returns `Err` rather than silently doing nothing —
/// surfacing the misuse (a plugin dispatching PHP while placed in the master)
/// loudly but without crashing the supervisor.
#[derive(Debug, Clone, Copy, Default)]
pub struct NoopExecutor;

impl NoopExecutor {
    fn refuse<T>() -> Result<T> {
        tracing::error!(
            "NoopExecutor called in the master process: a plugin dispatched PHP \
             but ran with role=Master (no PHP runtime). Check its placement()."
        );
        Err(anyhow::anyhow!(
            "no PHP runtime in the master process (NoopExecutor)"
        ))
    }
}

#[async_trait]
impl Executor for NoopExecutor {
    async fn execute_method(&self, _method: &str, _payload: Bytes) -> Result<Bytes> {
        Self::refuse()
    }
}

/// 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
    }

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