//! Subprocess entrypoint for `nexo-plugin-voice`.
//!
//! Phase voice-plugin-extract Fase M.2 / follow-up #1 — wires the
//! child-side JSON-RPC adapter + the broker subscriber tasks for
//! HTTP / admin / metrics topics.
//!
//! Boot order:
//! 1. `tracing_subscriber` → stderr (stdout is reserved for the
//!    daemon's JSON-RPC framing).
//! 2. `rustls::crypto::ring::default_provider().install_default()`
//!    so any later TLS path uses the same crypto everywhere.
//! 3. Build `PluginAdapter` first → initialize / shutdown reply
//!    immediately even if the broker is slow to come up.
//! 4. Detached task: build [`VoiceToolContext`] (open SQLite,
//!    register the Edge provider, install breakers + cache),
//!    connect to the broker, spawn the three subscriber loops,
//!    park the context in a `OnceCell` so the tool dispatcher can
//!    pick it up on the next `tool.invoke`.
//! 5. `adapter.run_stdio().await` blocks the main task.

use std::sync::Arc;

use nexo_broker::{AnyBroker, BrokerHandle, Event, Message};
use nexo_microapp_sdk::plugin::{
    print_manifest_if_requested, PluginAdapter, ToolInvocation, ToolInvocationError,
};
use nexo_plugin_voice::{
    auto_discovery, boot,
    breaker::BreakerRegistry,
    config::VoicePluginConfig,
    instance_registry,
    metrics::Metrics,
    providers::TtsProvider,
    store::VoiceModeStore,
    tool_defs::voice_tool_defs,
    tools::{self, VoiceToolContext},
    MANIFEST,
};
use once_cell::sync::Lazy;
use std::path::{Path, PathBuf};
use tokio::sync::OnceCell;

/// Lazy flag — flipped on the first successful `apply_configure`
/// so we only spawn the broker subscribers once. Subsequent
/// `plugin.configure` calls update the registry in place; the
/// subscribers re-resolve the catch-all instance per request.
static SUBSCRIBERS_SPAWNED: Lazy<OnceCell<()>> = Lazy::new(OnceCell::new);

/// Process-wide broker handle captured after the first configure
/// payload so subscribers can be spawned. Set once; cloned per
/// task.
static BROKER: Lazy<OnceCell<AnyBroker>> = Lazy::new(OnceCell::new);

/// Shared metrics handle. The tool dispatcher records counters /
/// histograms here; the broker `metrics.scrape` subscriber reads
/// them. Lives at module-scope so every tenant context shares the
/// same registry (per-tenant labels are a Wave B follow-up).
static METRICS: Lazy<Arc<Metrics>> = Lazy::new(|| Arc::new(Metrics::new()));

/// Connect to the daemon's broker once. Cached in `BROKER`.
async fn connect_broker() -> anyhow::Result<AnyBroker> {
    if let Some(b) = BROKER.get() {
        return Ok(b.clone());
    }
    let broker_url = std::env::var("NEXO_BROKER_URL")
        .map_err(|_| anyhow::anyhow!("NEXO_BROKER_URL not set — daemon must seed it"))?;

    let broker_inner = nexo_config::types::broker::BrokerInner {
        kind: if broker_url.starts_with("nats://") {
            nexo_config::types::broker::BrokerKind::Nats
        } else {
            nexo_config::types::broker::BrokerKind::Local
        },
        url: broker_url,
        auth: nexo_config::types::broker::BrokerAuthConfig::default(),
        persistence: nexo_config::types::broker::BrokerPersistenceConfig::default(),
        limits: nexo_config::types::broker::BrokerLimitsConfig::default(),
        fallback: nexo_config::types::broker::BrokerFallbackConfig::default(),
    };

    let broker = AnyBroker::from_config(&broker_inner)
        .await
        .map_err(|e| anyhow::anyhow!("broker connect failed: {e}"))?;
    let _ = BROKER.set(broker.clone());
    Ok(broker)
}

/// Build one `VoiceToolContext` for a tenant. Opens
/// `<data_dir>/voice.db` and instantiates the configured provider
/// stack. Invoked by `boot::apply_configure` once per declared
/// tenant — the caller has already sanitised labels and resolved
/// `<state_root>/voice/<tenant>` for `data_dir`.
async fn build_tenant_context(
    cfg: VoicePluginConfig,
    data_dir: PathBuf,
) -> Result<Arc<VoiceToolContext>, String> {
    if let Err(e) = tokio::fs::create_dir_all(&data_dir).await {
        return Err(format!("create data_dir {data_dir:?}: {e}"));
    }
    let db_path = data_dir.join("voice.db");
    let store = VoiceModeStore::open(&db_path)
        .await
        .map_err(|e| format!("open SQLite at {db_path:?}: {e}"))?;

    let edge = if let Some(edge_cfg) = cfg.providers.edge.as_ref() {
        let mut p = nexo_plugin_voice::providers::edge::EdgeTtsProvider::default();
        if let Some(rate) = edge_cfg.base_rate_pct {
            p.base_rate_pct = rate;
        }
        p
    } else {
        nexo_plugin_voice::providers::edge::EdgeTtsProvider::default()
    };
    let providers: Vec<Arc<dyn TtsProvider>> = vec![Arc::new(edge)];

    Ok(Arc::new(VoiceToolContext {
        store,
        providers,
        fallback: cfg.to_fallback_config(),
        breakers: BreakerRegistry::new(cfg.breakers.clone()),
        policy: cfg.model_overrides_policy.clone(),
        cache: cfg.build_cache(),
        max_text_length: cfg.text.max_length,
    }))
}

/// Drop a `.nexo-paired` marker into the tenant data dir so the
/// daemon's setup wizard `auth_check.candidates` probe sees the
/// tenant as "ready". The dashboard layout walks
/// `<state_root>/voice/<tenant>/` per [`plugin.dashboard.layout`].
async fn touch_paired_marker(data_dir: &Path) {
    let path = data_dir.join(".nexo-paired");
    if let Err(e) = tokio::fs::write(&path, b"").await {
        tracing::warn!(
            target: "plugin.voice",
            path = %path.display(),
            error = %e,
            "failed to write .nexo-paired marker (non-fatal)"
        );
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Plugin auto-discovery handshake: when the daemon's discovery
    // walker invokes us with `--print-manifest`, dump the bundled
    // TOML to stdout and exit 0 before doing anything else. Keeps
    // the daemon from waiting on a full subprocess wire just to
    // read the manifest.
    print_manifest_if_requested(MANIFEST);

    // CRITICAL: tracing MUST write to stderr — stdout is reserved
    // for the daemon's JSON-RPC framing. Same dance as the email
    // plugin's main.rs.
    tracing_subscriber::fmt()
        .with_writer(std::io::stderr)
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .init();

    let _ = rustls::crypto::ring::default_provider().install_default();

    // NOTE: a previous early-return on missing `NEXO_BROKER_URL`
    // was REMOVED — the daemon's stdio JSON-RPC handshake doesn't
    // need the broker, only the detached boot does. Exiting before
    // the adapter armed caused the daemon to timeout waiting for
    // `initialize` ("child did not reply within 5000ms"). The
    // adapter now always runs; if the broker connect fails inside
    // boot_plugin, the tool dispatcher returns Unavailable and the
    // daemon's RemoteToolHandler propagates a soft error.
    let smoke_mode = std::env::var("NEXO_BROKER_URL").is_err();
    if smoke_mode {
        tracing::info!(
            manifest_bytes = MANIFEST.len(),
            "nexo-plugin-voice: NEXO_BROKER_URL not seeded — running adapter only \
             (no broker subscribers). Tool dispatch will return Unavailable until boot completes."
        );
    }

    let adapter = PluginAdapter::new(MANIFEST)?
        .declare_tools(voice_tool_defs())
        .on_configure(|value: serde_yaml::Value| async move {
            // Phase 93 `plugin.configure` JSON-RPC handler. The
            // daemon sends the YAML slice at `cfg.plugins.voice`;
            // we hand it to `boot::apply_configure` which sanitises
            // labels, diffs against the registry, and rebuilds each
            // declared tenant's `VoiceToolContext` via the factory
            // below. Single-object YAML still works via
            // `VoicePluginShape::Single` and becomes one implicit
            // `default` tenant.
            tracing::info!("voice plugin: configure received");

            let data_root = std::env::var("NEXO_PLUGIN_VOICE_DATA_DIR")
                .map(PathBuf::from)
                .unwrap_or_else(|_| std::env::temp_dir().join("nexo-plugin-voice"));

            boot::apply_configure(value, &data_root, |cfg, data_dir| async move {
                let ctx = build_tenant_context(cfg, data_dir.clone()).await?;
                touch_paired_marker(&data_dir).await;
                // Keep the legacy global settings lock populated
                // with the most recently registered tenant — the
                // HTTP / admin / metrics subscribers still read it
                // until Wave B routes per-tenant.
                if let Some(label) = ctx_label_hint(&ctx) {
                    tracing::info!(target: "plugin.voice", tenant = %label, "built tenant context");
                }
                Ok(ctx)
            })
            .await?;

            // Spawn broker subscribers exactly once. They re-resolve
            // the catch-all instance via instance_registry per
            // request so subsequent configure calls don't require
            // re-spawning.
            if SUBSCRIBERS_SPAWNED.get().is_none() && !smoke_flag() {
                match connect_broker().await {
                    Ok(broker) => {
                        spawn_auto_discovery_subscribers(broker);
                        let _ = SUBSCRIBERS_SPAWNED.set(());
                    }
                    Err(e) => {
                        return Err(format!("broker connect: {e}"));
                    }
                }
            }

            tracing::info!(
                target: "plugin.voice",
                tenants = instance_registry::len(),
                "voice plugin: configure applied"
            );
            Ok(())
        })
        .on_tool(|inv: ToolInvocation| async move {
            // Resolve the tenant for this invocation: tenant claim
            // by agent_id → catch-all fallback. If no instance has
            // been registered yet (configure pending), surface
            // Unavailable so the host retries.
            let Some(ctx) = instance_registry::resolve(inv.agent_id.as_deref()) else {
                return Err(ToolInvocationError::Unavailable(
                    "voice plugin: no tenant registered (configure pending)".into(),
                ));
            };
            match tools::dispatch(&ctx, &inv.tool_name, inv.args).await {
                Ok(v) => Ok(v),
                Err(e) => {
                    let msg = e.to_string();
                    if msg.contains("InvalidParam") || msg.contains("invalid_voice") {
                        Err(ToolInvocationError::ArgumentInvalid(msg))
                    } else {
                        Err(ToolInvocationError::ExecutionFailed(msg))
                    }
                }
            }
        })
        .on_shutdown(|| async { Ok(()) });

    adapter.run_stdio().await?;
    Ok(())
}

fn smoke_flag() -> bool {
    std::env::var("NEXO_BROKER_URL").is_err()
}

/// Best-effort tenant label hint for logging. Returns `None` —
/// the registry owns the canonical label; this is a placeholder
/// the caller is free to ignore.
fn ctx_label_hint(_ctx: &Arc<VoiceToolContext>) -> Option<&'static str> {
    None
}

/// Broker subscribers for HTTP routes / admin RPC / metrics
/// scrape. Each runs in its own tokio task with failure isolation
/// — a dropped subscriber doesn't take the whole plugin down.
///
/// Wave A: the HTTP / admin handlers resolve the tenant lazily per
/// request via the registry's catch-all (no agent_id in HTTP
/// surfaces yet). Per-tenant HTTP routing is a Wave B follow-up
/// (`81.33.x.voice-per-tenant-http`).
fn spawn_auto_discovery_subscribers(broker: AnyBroker) {
    // Phase 81.33.b.real Stage 1 — pairing adapter subscribers.
    // Voice has no operator-side pairing flow; the three topics
    // subscribe so the daemon's `GenericBrokerPairingAdapter`
    // routes uniformly, and the handlers return `not_supported`.
    spawn_one(
        broker.clone(),
        "plugin.voice.pairing.normalize_sender",
        |_b, p| async move { auto_discovery::pairing_normalize_sender(&p) },
    );
    spawn_one(
        broker.clone(),
        "plugin.voice.pairing.send_reply",
        |_b, p| async move { auto_discovery::pairing_send_reply(&p).await },
    );
    spawn_one(
        broker.clone(),
        "plugin.voice.pairing.send_qr_image",
        |_b, p| async move { auto_discovery::pairing_send_qr_image(&p).await },
    );

    spawn_one(
        broker.clone(),
        "plugin.voice.http.request",
        |_b, p| async move {
            let Some(ctx) = instance_registry::resolve(None) else {
                return serde_json::json!({
                    "error": "voice plugin: no tenant registered"
                });
            };
            auto_discovery::http_request(&ctx, &p).await
        },
    );

    spawn_one(
        broker.clone(),
        "plugin.voice.metrics.scrape",
        |_b, p| async move { auto_discovery::metrics_scrape(&METRICS, &p).await },
    );

    spawn_one(broker, "plugin.voice.admin.>", |_b, p| async move {
        let Some(ctx) = instance_registry::resolve(None) else {
            return serde_json::json!({
                "error": "voice plugin: no tenant registered"
            });
        };
        auto_discovery::admin_handle(&ctx, &p).await
    });
}

fn spawn_one<F, Fut>(broker: AnyBroker, topic: &'static str, handler: F)
where
    F: Fn(AnyBroker, serde_json::Value) -> Fut + Send + Sync + 'static,
    Fut: std::future::Future<Output = serde_json::Value> + Send + 'static,
{
    tokio::spawn(async move {
        let mut sub = match broker.subscribe(topic).await {
            Ok(s) => s,
            Err(e) => {
                tracing::warn!(
                    target = "voice.auto_discovery",
                    topic,
                    error = %e,
                    "subscribe failed; topic will not receive requests"
                );
                return;
            }
        };
        tracing::info!(target = "voice.auto_discovery", topic, "subscriber up");
        while let Some(event) = sub.next().await {
            let Ok(msg) = serde_json::from_value::<Message>(event.payload) else {
                continue;
            };
            let Some(reply_to) = msg.reply_to.clone() else {
                continue;
            };
            let reply_payload = handler(broker.clone(), msg.payload.clone()).await;
            let reply_msg = Message::new(reply_to.clone(), reply_payload);
            let reply_event = Event::new(
                reply_to.clone(),
                "voice",
                match serde_json::to_value(&reply_msg) {
                    Ok(v) => v,
                    Err(_) => continue,
                },
            );
            if let Err(e) = broker.publish(&reply_to, reply_event).await {
                tracing::warn!(
                    target = "voice.auto_discovery",
                    topic,
                    reply_to = %reply_to,
                    error = %e,
                    "failed to publish reply"
                );
            }
        }
        tracing::debug!(
            target = "voice.auto_discovery",
            topic,
            "subscriber stream ended"
        );
    });
}
