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

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

/// Deployment mode for the unified Pulse binary.
#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
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"),
        }
    }
}

#[derive(Parser, Debug, Clone)]
#[command(author, version, about = "Pulse — Unified Real-Time Server")]
struct Args {
    /// Deployment mode: standalone (default), relay, or control.
    #[arg(long, value_enum, default_value_t = Mode::Standalone)]
    mode: Mode,

    #[arg(short, long, default_value = "0.0.0.0:4001")]
    bind: String,

    #[arg(short, long, default_value = "localhost:4001")]
    public_endpoint: String,

    #[arg(short, long, default_value = "eu-west-1")]
    region: String,

    #[arg(long, default_value_t = 0.0)]
    lat: f64,

    #[arg(long, default_value_t = 0.0)]
    lon: f64,

    /// Control plane URL (only used in --mode relay).
    #[arg(long, default_value = "http://localhost:4002")]
    control_url: String,

    #[arg(long, default_value_t = 100000)]
    capacity: u32,

    #[arg(long, env = "AI_STT_ENDPOINT")]
    stt_endpoint: Option<String>,

    #[arg(long, env = "AI_VISION_ENDPOINT")]
    vision_endpoint: Option<String>,
}

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

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<crate::sfu::Sfu>,
    /// Maximum allowed payload size per PLP packet (bytes).
    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>,
}

async fn run_heartbeat(args: Args) {
    let client = reqwest::Client::new();
    let url = format!("{}/api/v1/mesh/register", args.control_url);

    loop {
        let payload = RegistrationPayload {
            region: args.region.clone(),
            endpoint: args.public_endpoint.clone(),
            capacity: args.capacity,
            lat: args.lat,
            lon: args.lon,
        };

        match client.post(&url).json(&payload).send().await {
            Ok(resp) if resp.status().is_success() => {
                tracing::debug!("Successfully registered/heartbeated with control plane");
            }
            Ok(resp) => {
                warn!(
                    "Failed to register with control plane: HTTP {}",
                    resp.status()
                );
            }
            Err(e) => {
                error!("Network error registering with control plane: {}", e);
            }
        }

        tokio::time::sleep(Duration::from_secs(15)).await;
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    rustls::crypto::ring::default_provider()
        .install_default()
        .ok();
    observability::init("pulse");

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

    info!(
        "Starting Pulse ({} mode) on {} (Region: {})",
        args.mode, addr, args.region
    );

    // 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 = Arc::new(crate::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: args.control_url.clone(),
        sfu,
        max_payload_bytes: config::RelayConfig::default().max_payload_bytes,
        service_state,
        http_proxy_config,
    });

    // Best-effort WebSocket gateway for browsers (TCP on the same bind addr).
    // If binding fails (port already used on TCP), continue running the QUIC relay.
    {
        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 — standalone has control embedded)
    if args.mode == Mode::Relay {
        tokio::spawn(run_heartbeat(args.clone()));
    }

    // Generate certificates just for development
    let (certs, key) = generate_self_signed_cert()?;
    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()
-> 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))
}
