//! Broker-subscriber dispatchers for pairing / HTTP / admin /
//! metrics.
//!
//! Phase voice-plugin-extract follow-up #1 — mirrors
//! `nexo-rs-plugin-email::auto_discovery`. Each entry-point fn
//! takes the broker `Message.payload` JSON, runs the matching
//! handler, and returns the reply payload JSON. The subscriber
//! loop in `main.rs` wraps each topic with `Message`/`Event` framing.

use serde_json::{json, Value};

use crate::admin;
use crate::http::{self, HttpRequest};
use crate::metrics::Metrics;
use crate::tools::VoiceToolContext;

// ── Stage 1 — pairing adapter ──────────────────────────────────
//
// Voice is a transform plugin (TTS / STT). There is no operator-
// facing pairing flow — voices are configured via `voices.toml`
// + provider credentials in `secrets/voice.toml`. The three
// pairing topics are declared in the manifest for contract
// completeness so the daemon's `GenericBrokerPairingAdapter`
// constructs uniformly across every plugin; each handler returns
// a typed `not_supported` so a stray daemon-side
// `PairingChannelAdapter` trait call degrades gracefully.

/// `plugin.voice.pairing.normalize_sender` — voice has no concept
/// of a remote "sender" address; the parameter has no meaningful
/// canonical form. Return the raw value unchanged (or `null` when
/// empty) so the daemon's cache stays consistent.
pub fn pairing_normalize_sender(request: &Value) -> Value {
    let raw = request
        .get("raw")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty());
    json!({ "normalized": raw })
}

/// `plugin.voice.pairing.send_reply` — voice is transform-only;
/// outbound reply via voice happens through the regular
/// `voice_reply_transform` LLM tool (or `plugin_channel_send` in
/// the autonomous worker). Pairing-time reply is `not_supported`.
pub async fn pairing_send_reply(_request: &Value) -> Value {
    json!({
        "ok": false,
        "error": "voice plugin has no pairing flow; use the voice_reply_transform tool for outbound voice replies",
    })
}

/// `plugin.voice.pairing.send_qr_image` — N/A. Voice doesn't
/// authenticate against a remote peer; there is no QR.
pub async fn pairing_send_qr_image(_request: &Value) -> Value {
    json!({
        "ok": false,
        "error": "voice plugin is transform-only; QR pairing not supported",
    })
}

/// Handle `plugin.voice.http.request` payload.
///
/// Expected shape: serialise of [`HttpRequest`].
/// Reply shape: `{ status: u16, body: <json> }` so the daemon's
/// HTTP proxy can rebuild a real HTTP response.
pub async fn http_request(ctx: &VoiceToolContext, payload: &Value) -> Value {
    let req: HttpRequest = match serde_json::from_value(payload.clone()) {
        Ok(r) => r,
        Err(e) => {
            use base64::Engine;
            let err_json = json!({ "error": "bad_request", "detail": e.to_string() });
            let body_base64 = base64::engine::general_purpose::STANDARD
                .encode(err_json.to_string().as_bytes());
            return json!({
                "status": 400,
                "headers": [["content-type", "application/json"]],
                "body_base64": body_base64,
            });
        }
    };
    // Wrap into the daemon's `PluginHttpResponse` wire shape:
    //   { status, headers: [(k,v)], body_base64 }
    // The daemon decodes body_base64 + uses the headers verbatim
    // when relaying the response back to the original HTTP caller.
    use base64::Engine;
    match http::dispatch(ctx, req).await {
        Ok((status, body)) => {
            let body_str = body.to_string();
            let body_base64 = base64::engine::general_purpose::STANDARD.encode(body_str.as_bytes());
            json!({
                "status": status,
                "headers": [["content-type", "application/json"]],
                "body_base64": body_base64,
            })
        }
        Err(e) => {
            let err_json = json!({ "error": "internal", "detail": e.to_string() });
            let body_bytes = err_json.to_string().into_bytes();
            let body_base64 = base64::engine::general_purpose::STANDARD.encode(body_bytes);
            json!({
                "status": 500,
                "headers": [["content-type", "application/json"]],
                "body_base64": body_base64,
            })
        }
    }
}

/// Handle `plugin.voice.admin.<verb>` payload.
///
/// The broker subject already encodes the verb — the daemon's
/// admin forwarder strips the `nexo/admin/voice/` prefix and
/// places the verb in `payload.verb` when forwarding. Args under
/// `payload.args`.
pub async fn admin_handle(ctx: &VoiceToolContext, payload: &Value) -> Value {
    let verb = payload.get("verb").and_then(|v| v.as_str()).unwrap_or("");
    let args = payload.get("args").cloned().unwrap_or(Value::Null);
    if verb.is_empty() {
        return json!({ "error": "missing_verb" });
    }
    match admin::dispatch(ctx, verb, args).await {
        Ok(v) => v,
        Err(e) => json!({ "error": "internal", "detail": e.to_string() }),
    }
}

/// Handle `plugin.voice.metrics.scrape` payload. Daemon's
/// `nexo_pairing::plugin_metrics::scrape_one` reads the reply at
/// `.payload.text` — the field name is fixed by the contract, not
/// arbitrary. Returns Prometheus exposition format text.
pub async fn metrics_scrape(metrics: &Metrics, _payload: &Value) -> Value {
    json!({ "text": metrics.render() })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::breaker::{BreakerRegistry, BreakerSettings};
    use crate::cache::AudioCache;
    use crate::fallback::FallbackConfig;
    use crate::policy::ModelOverridesPolicy;
    use crate::providers::TtsProvider;
    use crate::store::VoiceModeStore;

    async fn fresh_ctx() -> VoiceToolContext {
        use sqlx::sqlite::SqliteConnectOptions;
        use std::str::FromStr;
        use std::sync::Arc;
        let opts = SqliteConnectOptions::from_str("sqlite::memory:").unwrap();
        let pool = sqlx::sqlite::SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(opts)
            .await
            .unwrap();
        let store = VoiceModeStore::open_with_pool(pool).await.unwrap();
        let providers: Vec<Arc<dyn TtsProvider>> = vec![];
        VoiceToolContext {
            store,
            providers,
            fallback: FallbackConfig::default(),
            breakers: BreakerRegistry::new(BreakerSettings::default()),
            policy: ModelOverridesPolicy::default(),
            cache: AudioCache::new(0, std::time::Duration::from_secs(60)),
            max_text_length: 1500,
        }
    }

    #[tokio::test]
    async fn http_request_handles_bad_payload() {
        let ctx = fresh_ctx().await;
        let reply = http_request(&ctx, &json!("not an object")).await;
        assert_eq!(reply["status"], 400);
    }

    #[tokio::test]
    async fn http_request_routes_to_voices_catalog() {
        use base64::Engine;
        let ctx = fresh_ctx().await;
        let payload = json!({
            "method": "GET",
            "path": "/api/voice/voices",
            "query": "provider=edge",
            "body_base64": ""
        });
        let reply = http_request(&ctx, &payload).await;
        assert_eq!(reply["status"], 200);
        // Reply now uses PluginHttpResponse shape: { status, headers,
        // body_base64 }. Decode body_base64 to inspect the JSON body.
        let body_base64 = reply["body_base64"].as_str().expect("body_base64 string");
        let body_bytes = base64::engine::general_purpose::STANDARD
            .decode(body_base64)
            .expect("body_base64 decodes");
        let body_json: serde_json::Value =
            serde_json::from_slice(&body_bytes).expect("body_base64 contents are JSON");
        assert!(body_json["voices"].is_array());
    }

    #[tokio::test]
    async fn admin_handle_returns_error_for_missing_verb() {
        let ctx = fresh_ctx().await;
        let reply = admin_handle(&ctx, &json!({})).await;
        assert_eq!(reply["error"], "missing_verb");
    }

    #[tokio::test]
    async fn admin_handle_dispatches_stats_verb() {
        let ctx = fresh_ctx().await;
        let reply = admin_handle(
            &ctx,
            &json!({
                "verb": "stats",
                "args": {}
            }),
        )
        .await;
        assert_eq!(reply["ok"], true);
        assert!(reply["providers"].is_array());
    }

    #[tokio::test]
    async fn metrics_scrape_returns_prometheus_text_field() {
        let m = Metrics::new();
        let reply = metrics_scrape(&m, &json!({})).await;
        // Daemon's `scrape_one` reads `.payload.text` — wire field
        // name fixed by `nexo_pairing::plugin_metrics::scrape_one`.
        assert!(reply["text"].as_str().unwrap().contains("# HELP nexo_voice_"));
    }
}
