//! Library entrypoint for embedding the Pulse Relay inside another process.
//!
//! This module exposes [`run_embedded`] which spins up the full relay
//! (QUIC + WebSocket + SFU + Media Pipeline) on a caller-provided Tokio
//! runtime, without requiring CLI argument parsing. It is designed for
//! the `@sansavision/pulse-node` NAPI-RS binding but works from any
//! Rust program that owns a Tokio runtime.

pub mod ai_processor;
pub mod config;
pub mod edge_wasm;
pub mod gpu;
pub mod http_bridge;
pub mod http_listener;
pub mod media;
pub mod nat;
pub mod nic;
pub mod observability;
pub mod queues;
pub mod record;
pub mod ringbuf;
pub mod schema_registry;
pub mod service;
pub mod session;
pub mod sfu;
pub mod telemetry;
pub mod webhook;
pub mod ws_gateway;

use pulse_core::transport::PulseEndpoint;
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{error, info, warn};

/// Configuration for an embedded relay instance (no CLI parsing).
#[derive(Debug, Clone)]
pub struct EmbeddedRelayConfig {
    /// Address to bind QUIC + WebSocket listeners.
    pub bind: SocketAddr,
    /// Region identifier for control-plane registration.
    pub region: String,
    /// Optional control-plane URL for heartbeat registration.
    pub control_url: Option<String>,
    /// Maximum concurrent connections advertised to the control-plane.
    pub capacity: u32,
    /// Maximum PLP payload size in bytes (default 1 MiB).
    pub max_payload_bytes: usize,
    /// Whether to start the WebSocket gateway for browser clients.
    pub enable_websocket: bool,
    /// Optional STT inference endpoint for AI processing.
    pub stt_endpoint: Option<String>,
    /// Optional vision inference endpoint for AI processing.
    pub vision_endpoint: Option<String>,
}

impl Default for EmbeddedRelayConfig {
    fn default() -> Self {
        Self {
            bind: "0.0.0.0:4001".parse().unwrap(),
            region: "local".into(),
            control_url: None,
            capacity: 100_000,
            max_payload_bytes: 1024 * 1024,
            enable_websocket: true,
            stt_endpoint: None,
            vision_endpoint: None,
        }
    }
}

/// Shared application state — identical to what `main.rs` constructs,
/// but created from [`EmbeddedRelayConfig`] instead of CLI args.
pub struct AppState {
    pub gpu_manager: Arc<RwLock<gpu::GpuManager>>,
    pub ai_processor: Arc<ai_processor::AiProcessor>,
    pub media_pipeline: Arc<media::MediaPipeline>,
    pub control_url: String,
    pub sfu: Arc<sfu::Sfu>,
    pub max_payload_bytes: usize,
    /// Embedded control-plane service state.
    pub service_state: Arc<service::ServiceState>,
    /// HTTP outbound proxy configuration.
    pub http_proxy_config: Arc<http_bridge::HttpProxyConfig>,
}

/// Token returned by [`run_embedded`]; drop it or call [`stop`] to
/// signal a graceful shutdown of the relay.
pub struct RelayHandle {
    shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
}

impl RelayHandle {
    /// Signal the relay to shut down gracefully.
    pub fn stop(&mut self) {
        if let Some(tx) = self.shutdown_tx.take() {
            let _ = tx.send(());
        }
    }
}

impl Drop for RelayHandle {
    fn drop(&mut self) {
        self.stop();
    }
}

/// Start the full Pulse Relay (QUIC + WS + SFU + media pipeline)
/// on the current Tokio runtime. Returns a handle that can be used
/// to shut down the relay.
///
/// This function spawns background tasks and returns immediately.
pub async fn run_embedded(
    cfg: EmbeddedRelayConfig,
) -> Result<RelayHandle, Box<dyn std::error::Error + Send + Sync>> {
    // Install the rustls crypto provider (idempotent).
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();

    // Initialise hardware & AI subsystems.
    let gpu_manager = Arc::new(RwLock::new(gpu::GpuManager::new()));
    let ai_config = ai_processor::AiConfig {
        stt_endpoint: cfg.stt_endpoint.clone(),
        vision_endpoint: cfg.vision_endpoint.clone(),
        tts_endpoint: None,
    };
    let ai_processor = Arc::new(ai_processor::AiProcessor::new(ai_config));

    if let Err(e) = ai_processor.load_models().await {
        error!("[embedded] Failed to initialize AI models: {}", e);
    }

    let engine =
        Arc::new(edge_wasm::EdgeWasmEngine::new().expect("Failed to create WASM engine"));
    let media_pipeline = Arc::new(media::MediaPipeline::new(engine));

    if let Err(e) = media_pipeline.enable_noise_cancellation().await {
        warn!(
            "[embedded] Noise cancellation initialization failed: {}",
            e
        );
    }
    if let Err(e) = media_pipeline.enable_background_blur().await {
        warn!(
            "[embedded] Background blur initialization failed: {}",
            e
        );
    }

    let sfu_instance = Arc::new(sfu::Sfu::new());
    let service_state = Arc::new(service::ServiceState::new());
    let http_proxy_config = Arc::new(http_bridge::HttpProxyConfig::default());
    let state = Arc::new(AppState {
        gpu_manager,
        ai_processor,
        media_pipeline,
        control_url: cfg.control_url.clone().unwrap_or_default(),
        sfu: sfu_instance,
        max_payload_bytes: cfg.max_payload_bytes,
        service_state,
        http_proxy_config,
    });

    // ── WebSocket gateway ───────────────────────────────────────────
    if cfg.enable_websocket {
        let ws_state = Arc::clone(&state);
        let ws_addr = cfg.bind;
        tokio::spawn(async move {
            if let Err(e) = ws_gateway::run(ws_addr, ws_state).await {
                warn!("[embedded] WebSocket gateway not started: {}", e);
            }
        });
        info!(
            "[embedded] WebSocket gateway listening on ws://{}",
            cfg.bind
        );
    }

    // ── Heartbeat registration loop ─────────────────────────────────
    if let Some(ref control_url) = cfg.control_url {
        let url = format!("{}/api/v1/mesh/register", control_url);
        let region = cfg.region.clone();
        let capacity = cfg.capacity;
        let endpoint_str = cfg.bind.to_string();
        tokio::spawn(async move {
            let client = reqwest::Client::new();
            loop {
                let payload = serde_json::json!({
                    "region": region,
                    "endpoint": endpoint_str,
                    "capacity": capacity,
                    "lat": 0.0,
                    "lon": 0.0,
                });
                match client.post(&url).json(&payload).send().await {
                    Ok(resp) if resp.status().is_success() => {
                        tracing::debug!("[embedded] Heartbeat OK");
                    }
                    Ok(resp) => {
                        warn!(
                            "[embedded] Heartbeat failed: HTTP {}",
                            resp.status()
                        );
                    }
                    Err(e) => {
                        error!("[embedded] Heartbeat network error: {}", e);
                    }
                }
                tokio::time::sleep(std::time::Duration::from_secs(15)).await;
            }
        });
    }

    // ── QUIC server accept loop ─────────────────────────────────────
    let (certs, key) = generate_self_signed_cert()?;
    info!("[embedded] Generated ephemeral development certificates");

    let server_endpoint = PulseEndpoint::new_server(cfg.bind, certs, key)
        .map_err(|e| Box::<dyn std::error::Error + Send + Sync>::from(e.to_string()))?;
    let endpoint = server_endpoint.inner().clone();

    info!(
        "[embedded] Pulse Relay listening on udp://{} (region: {})",
        cfg.bind, cfg.region
    );

    let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel::<()>();

    let quic_state = Arc::clone(&state);
    tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = &mut shutdown_rx => {
                    info!("[embedded] Relay shutting down gracefully");
                    endpoint.close(0u32.into(), b"shutdown");
                    break;
                }
                incoming = endpoint.accept() => {
                    match incoming {
                        Some(conn_incoming) => {
                            let state_clone = Arc::clone(&quic_state);
                            tokio::spawn(async move {
                                match conn_incoming.await {
                                    Ok(connection) => {
                                        info!(
                                            "[embedded] Connection from {}",
                                            connection.remote_address()
                                        );
                                        if let Err(e) =
                                            session::handle_connection(connection, state_clone)
                                                .await
                                        {
                                            warn!(
                                                "[embedded] Connection error: {}",
                                                e
                                            );
                                        }
                                    }
                                    Err(e) => {
                                        warn!("[embedded] Incoming connection failed: {}", e);
                                    }
                                }
                            });
                        }
                        None => {
                            info!("[embedded] QUIC endpoint closed");
                            break;
                        }
                    }
                }
            }
        }
    });

    Ok(RelayHandle {
        shutdown_tx: Some(shutdown_tx),
    })
}

fn generate_self_signed_cert()
-> Result<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>), Box<dyn std::error::Error + Send + Sync>>
{
    let subject_alt_names = vec!["localhost".to_string()];
    let cert = rcgen::generate_simple_self_signed(subject_alt_names)?;
    let key = PrivateKeyDer::try_from(cert.signing_key.serialize_der())?;
    let cert_der = CertificateDer::from(cert.cert.der().to_vec());
    Ok((vec![cert_der], key))
}
