//! Library entrypoint for embedding the Pulse Relay inside another process.
//!
//! This module exposes:
//! - [`run`] — full relay startup with CLI-style args, banner, port detection.
//! - [`run_embedded`] — programmatic startup without CLI parsing (for NAPI-RS).

pub mod ai_processor;
pub mod auth;
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;

#[cfg(feature = "queue-postgres")]
pub mod queue_backend_postgres;

#[cfg(feature = "queue-redis")]
pub mod queue_backend_redis;

use clap::ValueEnum;
use pulse_core::queue_storage::QueueStorageBackend;
use pulse_core::transport::PulseEndpoint;
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use serde::Serialize;
use std::io::Write;
use std::net::{SocketAddr, TcpListener};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tracing::{error, info, warn};

// ═══════════════════════════════════════════════════════════════════════
//  Public types for the CLI-driven `run()` entry point.
// ═══════════════════════════════════════════════════════════════════════

/// Deployment mode for the unified Pulse binary.
#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
pub enum Mode {
    /// Everything in one process: SFU + control services + HTTP bridge.
    Standalone,
    /// SFU + HTTP bridge only. Connects to an external control node.
    Relay,
    /// Control services only (no SFU). Central coordinator for relay fleet.
    Control,
}

impl std::fmt::Display for Mode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Mode::Standalone => write!(f, "standalone"),
            Mode::Relay => write!(f, "relay"),
            Mode::Control => write!(f, "control"),
        }
    }
}

/// Queue storage backend selection.
#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
pub enum QueueStorage {
    /// In-memory only (no persistence, current default).
    Memory,
    /// Write-Ahead Log on local disk (recommended for production).
    Wal,
    /// PostgreSQL database (requires --database-url).
    #[cfg(feature = "queue-postgres")]
    Postgres,
    /// Redis (requires --redis-url).
    #[cfg(feature = "queue-redis")]
    Redis,
}

/// Arguments accepted by [`run`]. Mirrors the CLI flags of
/// the standalone `pulse` binary so that `pulse-cli` can construct
/// these from its own `Serve` subcommand.
#[derive(Debug, Clone)]
pub struct ServeArgs {
    pub mode: Mode,
    pub bind: String,
    pub public_endpoint: String,
    pub region: String,
    pub lat: f64,
    pub lon: f64,
    pub control_url: String,
    pub capacity: u32,
    pub stt_endpoint: Option<String>,
    pub vision_endpoint: Option<String>,
    pub queue_storage: QueueStorage,
    pub queue_data_dir: String,
    pub database_url: Option<String>,
    pub redis_url: Option<String>,
    pub queue_encryption_key: Option<String>,
    pub queue_encryption_key_file: Option<String>,
    /// Auth configuration.
    pub auth: auth::AuthConfig,
}

impl Default for ServeArgs {
    fn default() -> Self {
        Self {
            mode: Mode::Standalone,
            bind: "0.0.0.0:4001".into(),
            public_endpoint: "localhost:4001".into(),
            region: "eu-west-1".into(),
            lat: 0.0,
            lon: 0.0,
            control_url: "http://localhost:4002".into(),
            capacity: 100_000,
            stt_endpoint: None,
            vision_endpoint: None,
            queue_storage: QueueStorage::Memory,
            queue_data_dir: "./data/queues".into(),
            database_url: None,
            redis_url: None,
            queue_encryption_key: None,
            queue_encryption_key_file: None,
            auth: auth::AuthConfig::default(),
        }
    }
}

#[derive(Serialize)]
struct RegistrationPayload {
    region: String,
    endpoint: String,
    capacity: u32,
    lat: f64,
    lon: f64,
}

// ═══════════════════════════════════════════════════════════════════════
//  `run()` — full relay startup with banner, port detection, etc.
// ═══════════════════════════════════════════════════════════════════════

/// Start the Pulse Relay with full CLI-style output (banner, port
/// conflict detection, queue backend init). This function blocks
/// until the server is shut down.
pub async fn run(args: ServeArgs) -> Result<(), Box<dyn std::error::Error>> {
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();

    let addr: SocketAddr = args.bind.parse().expect("Invalid bind address");

    // ── Port conflict detection ─────────────────────────────────────
    let prometheus_port: u16 = std::env::var("PROMETHEUS_PORT")
        .unwrap_or_else(|_| "9090".to_string())
        .parse()
        .unwrap_or(9090);

    let ports_to_check: Vec<(u16, &str)> = vec![
        (addr.port(), "Relay (WS + QUIC)"),
        (prometheus_port, "Prometheus exporter"),
    ];

    let mut busy: Vec<(u16, &str, String)> = Vec::new();
    for (port, label) in &ports_to_check {
        if TcpListener::bind(format!("0.0.0.0:{}", port)).is_err() {
            let pid_info = std::process::Command::new("lsof")
                .args(["-ti", &format!("tcp:{}", port)])
                .output()
                .ok()
                .and_then(|o| {
                    let s = String::from_utf8_lossy(&o.stdout).trim().to_string();
                    if s.is_empty() { None } else { Some(s) }
                })
                .unwrap_or_else(|| "unknown".to_string());
            busy.push((*port, label, pid_info));
        }
    }

    if !busy.is_empty() {
        const R: &str = "\x1b[31m";
        const Y: &str = "\x1b[33m";
        const B: &str = "\x1b[1m";
        const D: &str = "\x1b[2m";
        const RST: &str = "\x1b[0m";

        eprintln!();
        eprintln!("  {R}{B}⚠  Port conflict detected!{RST}  The following ports are already in use:");
        eprintln!();
        for (port, label, pids) in &busy {
            eprintln!("   {Y}{B}:{:<5}{RST}  {label}  {D}(PID: {pids}){RST}", port);
        }
        eprintln!();
        eprint!("   Kill these processes and continue? {D}[y/N]{RST} ");
        let _ = std::io::stdout().flush();
        let _ = std::io::stderr().flush();

        let mut input = String::new();
        if std::io::stdin().read_line(&mut input).is_ok() {
            let answer = input.trim().to_lowercase();
            if answer == "y" || answer == "yes" {
                for (port, label, pids) in &busy {
                    for pid in pids.split('\n') {
                        let pid = pid.trim();
                        if !pid.is_empty() && pid != "unknown" {
                            eprintln!("   Killing PID {pid} {D}(:{port} {label}){RST}");
                            let _ = std::process::Command::new("kill")
                                .args(["-9", pid])
                                .status();
                        }
                    }
                }
                std::thread::sleep(std::time::Duration::from_millis(500));
                eprintln!("   Ports freed. Continuing startup…");
                eprintln!();
            } else {
                eprintln!();
                eprintln!("  {R}Aborting.{RST} Free the ports above and try again.");
                std::process::exit(1);
            }
        }
    }

    // ── Observability (binds Prometheus port) ────────────────────────
    observability::init("pulse");

    // ANSI color helpers
    const RED: &str = "\x1b[31m";
    const GREEN: &str = "\x1b[32m";
    const ORANGE: &str = "\x1b[33m";
    const CYAN: &str = "\x1b[36m";
    const BOLD: &str = "\x1b[1m";
    const DIM: &str = "\x1b[2m";
    const RESET: &str = "\x1b[0m";

    // ── Startup banner ──────────────────────────────────────────────
    eprintln!();
    eprintln!("{BOLD}{CYAN}  ╔═══════════════════════════════════════════════════╗{RESET}");
    eprintln!("{BOLD}{CYAN}  ║            Pulse Relay — Starting Up              ║{RESET}");
    eprintln!("{BOLD}{CYAN}  ╚═══════════════════════════════════════════════════╝{RESET}");
    eprintln!();

    let mode_str = format!("{}", args.mode);
    let mode_color = match args.mode {
        Mode::Standalone => GREEN,
        Mode::Relay => CYAN,
        Mode::Control => CYAN,
    };
    eprintln!("  {BOLD}Mode:{RESET}        {mode_color}{BOLD}{mode_str}{RESET}");
    eprintln!("  {BOLD}Bind:{RESET}        {DIM}{addr}{RESET}");
    eprintln!("  {BOLD}Region:{RESET}      {DIM}{}{RESET}", args.region);
    eprintln!();

    // Initialize Hardware & AI Subsystems dynamically
    let gpu_manager = Arc::new(RwLock::new(gpu::GpuManager::new()));
    let ai_config = ai_processor::AiConfig {
        stt_endpoint: args.stt_endpoint.clone(),
        vision_endpoint: args.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!("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!("Noise cancellation initialization failed: {}", e);
    }
    if let Err(e) = media_pipeline.enable_background_blur().await {
        warn!("Background blur initialization failed: {}", e);
    }

    let sfu_instance = Arc::new(sfu::Sfu::new());
    let http_proxy_config = Arc::new(http_bridge::HttpProxyConfig::default());

    // ── Queue storage backend ───────────────────────────────────────
    let queue_backend: Arc<dyn pulse_core::queue_storage::QueueStorageBackend> =
        match args.queue_storage {
            QueueStorage::Memory => {
                eprintln!("  {BOLD}Queue:{RESET}       {ORANGE}{BOLD}memory{RESET} {ORANGE}(no persistence — messages lost on restart){RESET}");
                Arc::new(pulse_core::queue_storage::InMemoryBackend::new())
                    as Arc<dyn pulse_core::queue_storage::QueueStorageBackend>
            }
            QueueStorage::Wal => {
                eprintln!("  {BOLD}Queue:{RESET}       {ORANGE}{BOLD}wal{RESET} {DIM}→ {}{RESET}", args.queue_data_dir);
                Arc::new(
                    pulse_core::wal::WalBackend::new(pulse_core::wal::WalConfig {
                        data_dir: std::path::PathBuf::from(&args.queue_data_dir),
                        sync_policy: pulse_core::wal::SyncPolicy::default(),
                    })
                    .expect("Failed to initialize WAL queue backend"),
                ) as Arc<dyn pulse_core::queue_storage::QueueStorageBackend>
            }
            #[cfg(feature = "queue-postgres")]
            QueueStorage::Postgres => {
                let url = args
                    .database_url
                    .as_deref()
                    .expect("--database-url required for postgres queue storage");
                let display_url = url.split('@').last().unwrap_or(url);
                eprintln!("  {BOLD}Queue:{RESET}       {DIM}postgres → {display_url}{RESET}");
                eprint!("  {BOLD}Status:{RESET}      ");

                match queue_backend_postgres::PostgresQueueBackend::connect(url, 10).await {
                    Ok(pg) => {
                        match pg.migrate().await {
                            Ok(_) => {
                                eprintln!("{GREEN}{BOLD}✓ connected{RESET}");
                                Arc::new(pg) as Arc<dyn pulse_core::queue_storage::QueueStorageBackend>
                            }
                            Err(e) => {
                                eprintln!("{RED}{BOLD}✗ migration failed:{RESET} {RED}{e}{RESET}");
                                eprintln!("  {RED}{BOLD}Fallback:{RESET}    {ORANGE}memory (no persistence){RESET}");
                                Arc::new(pulse_core::queue_storage::InMemoryBackend::new())
                                    as Arc<dyn pulse_core::queue_storage::QueueStorageBackend>
                            }
                        }
                    }
                    Err(e) => {
                        eprintln!("{RED}{BOLD}✗ connection failed:{RESET} {RED}{e}{RESET}");
                        eprintln!("  {RED}{BOLD}Fallback:{RESET}    {ORANGE}memory (no persistence){RESET}");
                        Arc::new(pulse_core::queue_storage::InMemoryBackend::new())
                            as Arc<dyn pulse_core::queue_storage::QueueStorageBackend>
                    }
                }
            }
            #[cfg(feature = "queue-redis")]
            QueueStorage::Redis => {
                let url = args
                    .redis_url
                    .as_deref()
                    .expect("--redis-url required for redis queue storage");
                let display_url = url.split('@').last().unwrap_or(url);
                eprintln!("  {BOLD}Queue:{RESET}       {DIM}redis → {display_url}{RESET}");
                eprint!("  {BOLD}Status:{RESET}      ");

                match queue_backend_redis::RedisQueueBackend::connect(url) {
                    Ok(redis) => {
                        match redis.append("__pulse_healthcheck__", &pulse_core::queue::QueueMessage {
                            sequence: 0,
                            payload: vec![],
                            idempotency_key: None,
                            ttl_secs: None,
                            enqueued_at: 0,
                            delivery_count: 0,
                            state: pulse_core::queue::MessageState::Acked,
                        }) {
                            Ok(_) => {
                                let _ = redis.purge("__pulse_healthcheck__");
                                eprintln!("{GREEN}{BOLD}✓ connected{RESET}");
                                Arc::new(redis) as Arc<dyn pulse_core::queue_storage::QueueStorageBackend>
                            }
                            Err(e) => {
                                eprintln!("{RED}{BOLD}✗ ping failed:{RESET} {RED}{e}{RESET}");
                                eprintln!("  {RED}{BOLD}Fallback:{RESET}    {ORANGE}memory (no persistence){RESET}");
                                Arc::new(pulse_core::queue_storage::InMemoryBackend::new())
                                    as Arc<dyn pulse_core::queue_storage::QueueStorageBackend>
                            }
                        }
                    }
                    Err(e) => {
                        eprintln!("{RED}{BOLD}✗ connection failed:{RESET} {RED}{e}{RESET}");
                        eprintln!("  {RED}{BOLD}Fallback:{RESET}    {ORANGE}memory (no persistence){RESET}");
                        Arc::new(pulse_core::queue_storage::InMemoryBackend::new())
                            as Arc<dyn pulse_core::queue_storage::QueueStorageBackend>
                    }
                }
            }
        };

    // ── Optional encryption at rest ──────────────────────────────────
    let queue_backend: Arc<dyn pulse_core::queue_storage::QueueStorageBackend> =
        if args.queue_encryption_key.is_some() || args.queue_encryption_key_file.is_some() {
            let key_source = if let Some(ref key_file) = args.queue_encryption_key_file {
                pulse_core::queue_encryption::EncryptionKeySource::KeyFile(
                    std::path::PathBuf::from(key_file),
                )
            } else {
                pulse_core::queue_encryption::EncryptionKeySource::EnvVar(
                    "PULSE_QUEUE_KEY".to_string(),
                )
            };
            eprintln!("  {BOLD}Encrypt:{RESET}     {GREEN}{BOLD}✓ enabled{RESET}");
            Arc::new(
                pulse_core::queue_encryption::EncryptedBackend::new(queue_backend, key_source)
                    .expect("Failed to initialize queue encryption"),
            )
        } else {
            eprintln!("  {BOLD}Encrypt:{RESET}     {DIM}off{RESET}");
            queue_backend
        };

    eprintln!();
    eprintln!("{BOLD}{CYAN}  ───────────────────────────────────────────────────{RESET}");
    eprintln!();

    // ── Service state ────────────────────────────────────────────────
    let service_state = Arc::new(service::ServiceState::new(Arc::clone(&queue_backend)));

    let state = Arc::new(AppState {
        gpu_manager,
        ai_processor,
        media_pipeline,
        control_url: args.control_url.clone(),
        sfu: sfu_instance,
        max_payload_bytes: config::RelayConfig::default().max_payload_bytes,
        service_state,
        http_proxy_config,
        queue_backend,
        auth_verifier: auth::AuthVerifier::new(args.auth.clone()),
    });

    // Best-effort WebSocket gateway
    {
        let ws_state = Arc::clone(&state);
        tokio::spawn(async move {
            if let Err(e) = ws_gateway::run(addr, ws_state).await {
                warn!("WebSocket gateway not started: {}", e);
            }
        });
    }

    // Start registration loop (only in relay mode)
    if args.mode == Mode::Relay {
        let hb_args = args.clone();
        tokio::spawn(async move {
            let client = reqwest::Client::new();
            let url = format!("{}/api/v1/mesh/register", hb_args.control_url);
            loop {
                let payload = RegistrationPayload {
                    region: hb_args.region.clone(),
                    endpoint: hb_args.public_endpoint.clone(),
                    capacity: hb_args.capacity,
                    lat: hb_args.lat,
                    lon: hb_args.lon,
                };
                match client.post(&url).json(&payload).send().await {
                    Ok(resp) if resp.status().is_success() => {
                        tracing::debug!("Heartbeat OK");
                    }
                    Ok(resp) => {
                        warn!("Heartbeat failed: HTTP {}", resp.status());
                    }
                    Err(e) => {
                        error!("Heartbeat network error: {}", e);
                    }
                }
                tokio::time::sleep(Duration::from_secs(15)).await;
            }
        });
    }

    // Generate certificates
    let (certs, key) = generate_self_signed_cert_public()?;
    info!("Generated ephemeral development certificates");

    let server_endpoint = PulseEndpoint::new_server(addr, certs, key)?;
    let endpoint = server_endpoint.inner();

    info!("Relay listening for UDP (QUIC) on {}", addr);

    // Connection loop
    while let Some(incoming) = endpoint.accept().await {
        let state_clone = Arc::clone(&state);
        tokio::spawn(async move {
            match incoming.await {
                Ok(connection) => {
                    info!(
                        "Accepting connection from {}. ALPN: {:?}",
                        connection.remote_address(),
                        connection
                            .handshake_data()
                            .unwrap()
                            .downcast_ref::<quinn::crypto::rustls::HandshakeData>()
                            .and_then(|h| h
                                .protocol
                                .as_ref()
                                .map(|p| String::from_utf8_lossy(p).into_owned()))
                    );

                    if let Err(e) = session::handle_connection(connection, state_clone).await {
                        warn!("Connection closed with error: {}", e);
                    }
                }
                Err(e) => warn!("Incoming connection failed: {}", e),
            }
        });
    }

    Ok(())
}

fn generate_self_signed_cert_public()
-> Result<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>), Box<dyn std::error::Error>>
{
    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))
}

/// Configuration for an embedded relay instance (no CLI parsing).
#[derive(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>,
    /// Queue storage backend. Defaults to in-memory.
    pub queue_backend: Option<Arc<dyn pulse_core::queue_storage::QueueStorageBackend>>,
}

impl std::fmt::Debug for EmbeddedRelayConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EmbeddedRelayConfig")
            .field("bind", &self.bind)
            .field("region", &self.region)
            .field("control_url", &self.control_url)
            .field("capacity", &self.capacity)
            .field("max_payload_bytes", &self.max_payload_bytes)
            .field("enable_websocket", &self.enable_websocket)
            .field("queue_backend", &self.queue_backend.as_ref().map(|_| "<custom>"))
            .finish()
    }
}

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,
            queue_backend: 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>,
    /// Queue storage backend (shared across all queues).
    pub queue_backend: Arc<dyn pulse_core::queue_storage::QueueStorageBackend>,
    /// Auth verifier for CONNECT handshake.
    pub auth_verifier: Arc<auth::AuthVerifier>,
}

/// 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 http_proxy_config = Arc::new(http_bridge::HttpProxyConfig::default());
    let queue_backend = cfg.queue_backend.unwrap_or_else(|| {
        Arc::new(pulse_core::queue_storage::InMemoryBackend::new())
    });
    let service_state = Arc::new(service::ServiceState::new(Arc::clone(&queue_backend)));
    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,
        queue_backend,
        auth_verifier: auth::AuthVerifier::new(auth::AuthConfig::default()),
    });

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