//! # rutster — slice-1 binary
//!
//! axum signaling server + the media-core poll driver + a static HTML
//! test client (spec §4). Binds `0.0.0.0:8080` plaintext — no TLS in
//! slice 1 (out of scope per §1.2). DTLS-SRTP is mandatory on the media
//! surface (str0m handles natively); TLS on the HTTP surface lands with
//! the deployment posture in step 5.
//!
//! ## Architecture refs
//!
//! - [slice-1 spec §4](../../docs/superpowers/specs/2026-06-28-slice-1-webrtc-loopback-design.md)
//! - [ARCHITECTURE.md](../../docs/ARCHITECTURE.md) — fused vertical.

use std::net::SocketAddr;

use rutster::routes::router;
use rutster::session_map::AppState;
use tokio::sync::mpsc;
use tracing::info;

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "rutster=info".into()),
        )
        .init();

    let default_tap_url: url::Url = std::env::var("RUTSTER_TAP_URL")
        .unwrap_or_else(|_| "ws://127.0.0.1:8081/echo".to_string())
        .parse()
        .expect(
            "RUTSTER_TAP_URL must be a valid ws:// URL (slice-2 dev loop; wss:// lands in step 6)",
        );
    // Placeholder sender so we can build an AppState; the real sender comes
    // from the spawned MediaThread below.
    let (placeholder_tx, _placeholder_rx) = mpsc::channel(1);
    // Deploy slice A §5.3: config::twilio_credentials was parsed-but-
    // never-called since slice-5 — wired into startup validation here.
    // Partial or malformed RUTSTER_TWILIO_* config now fails the boot
    // loudly instead of silently running WebRTC-only. (The credentials
    // themselves stay green-zone: only webhook_base is threaded into the
    // routes; TwilioCredentials' Debug redacts auth_token — ADR-0009.)
    let twilio_credentials = rutster::config::twilio_credentials(
        std::env::var("RUTSTER_TWILIO_ACCOUNT_SID").ok(),
        std::env::var("RUTSTER_TWILIO_AUTH_TOKEN").ok(),
        std::env::var("RUTSTER_TWILIO_MEDIA_BIND").ok(),
        std::env::var("RUTSTER_TWILIO_WEBHOOK_BASE").ok(),
    )
    .expect("RUTSTER_TWILIO_* env config invalid");
    if let Some(creds) = twilio_credentials.as_ref() {
        info!(webhook_base = %creds.webhook_base, "trunk configured");
    }
    // Deploy slice A §5.3: trusted-proxy CIDR list. Empty/unset is the
    // default fail-closed posture — forwarded headers are ignored unless
    // the direct peer is in this list.
    let trusted_proxies =
        rutster::config::trusted_proxies(std::env::var("RUTSTER_TRUSTED_PROXIES").ok())
            .expect("RUTSTER_TRUSTED_PROXIES must be a comma-separated CIDR list");
    let app_state = AppState::new(placeholder_tx, default_tap_url)
        .with_trunk_webhook_base(twilio_credentials.as_ref().map(|c| c.webhook_base.clone()))
        .with_trusted_proxies(trusted_proxies);

    let media_cfg = rutster::config::media_address_config(
        std::env::var("RUTSTER_MEDIA_BIND_IP").ok(),
        std::env::var("RUTSTER_MEDIA_ADVERTISED_IP").ok(),
        std::env::var("RUTSTER_MEDIA_PORT_RANGE").ok(),
    )
    .expect("RUTSTER_MEDIA_* env config invalid");
    let valkey_url = rutster::config::valkey_url(std::env::var("RUTSTER_VALKEY_URL").ok())
        .expect("RUTSTER_VALKEY_URL must be redis:// or rediss:// when set");
    let sink: rutster::event_sink::SharedEventSink = match valkey_url {
        Some(url) => {
            info!(%url, "ValkeyEventSink enabled");
            rutster::valkey_sink::ValkeyEventSink::spawn(url, tokio::runtime::Handle::current())
        }
        None => {
            info!("TracingEventSink (default; set RUTSTER_VALKEY_URL for durable events)");
            std::sync::Arc::new(rutster::event_sink::TracingEventSink)
        }
    };

    let opts = rutster::media_thread::MediaThreadOpts {
        media_cfg,
        max_sessions: rutster::config::max_sessions(std::env::var("RUTSTER_MAX_SESSIONS").ok())
            .expect("RUTSTER_MAX_SESSIONS must be a number"),
        sink,
    };
    let (app_state, media_thread) = app_state
        .spawn_media_thread_with(opts, tokio::runtime::Handle::current())
        .expect("media thread spawn");

    // RUTSTER_HTTP_BIND: per-instance bind is the first thing any LB /
    // compose template parametrizes (slice-5; matches the existing
    // RUTSTER_TAP_BIND pattern in rutster-brain-realtime).
    let addr: SocketAddr = rutster::config::http_bind(std::env::var("RUTSTER_HTTP_BIND").ok())
        .expect("RUTSTER_HTTP_BIND must be host:port");
    let drain_deadline =
        rutster::config::drain_deadline(std::env::var("RUTSTER_DRAIN_DEADLINE_SECS").ok())
            .expect("RUTSTER_DRAIN_DEADLINE_SECS must be seconds");
    info!(%addr, "listening");

    // Two-phase shutdown (slice-5, review M1):
    //   SIGTERM → Drain (stop admitting; keep serving HTTP for in-flight
    //   calls' DELETE/offer; readyz flips so the LB pulls us) → deadline
    //   → HTTP quiesce → media_thread.shutdown() (the hard stop).
    let (http_stop_tx, http_stop_rx) = tokio::sync::oneshot::channel::<()>();
    let drain_cmd_tx = app_state.cmd_tx.clone();
    tokio::spawn(async move {
        shutdown_signal().await;
        if !drain_deadline.is_zero() {
            let (reply, drained) = tokio::sync::oneshot::channel();
            // One deadline bounds the WHOLE drain round trip — send AND
            // completion. A wedged media thread can stall the send itself
            // (full command channel), and the shutdown promise is "bounded
            // by the deadline" against exactly that node. Same principle
            // as readyz's whole-round-trip timeout (commit 5ce18bf).
            let outcome = tokio::time::timeout(drain_deadline, async {
                if drain_cmd_tx
                    .send(rutster::media_thread::MediaCmd::Drain { reply })
                    .await
                    .is_ok()
                {
                    let _ = drained.await;
                }
            })
            .await;
            match outcome {
                Ok(()) => info!("drain complete; proceeding to shutdown"),
                Err(_) => info!(?drain_deadline, "drain deadline hit; shutting down anyway"),
            }
        }
        let _ = http_stop_tx.send(());
    });

    // slice-5 T5: mount the TwilioMediaStreamsServer WS sub-router.
    // The pump task sends RegisterTrunkInboundChannel via a relay mpsc;
    // the relay task translates the channel to MediaCmd::RegisterTrunk
    // + awaits the oneshot reply before continuing the pump task.
    //
    // Step A (this slice): relay + mount wired; the actual
    // RegisterTrunk handler returns the placeholder NACK on the media
    // thread side (see crates/rutster/src/media_thread.rs Step A).
    let (trunk_register_tx, mut trunk_register_rx) =
        mpsc::channel::<rutster_trunk::twilio_media_streams::RegisterTrunkInboundChannel>(32);
    let mount_cmd_tx = app_state.cmd_tx.clone();
    tokio::spawn(async move {
        while let Some(register) = trunk_register_rx.recv().await {
            let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
            let register_parts = register;
            let trunk_register_cmd = rutster::media_thread::MediaCmd::RegisterTrunk {
                call_sid: register_parts.call_sid,
                inbound_from_twilio_rx: register_parts.inbound_from_twilio_rx,
                outbound_to_twilio_tx: register_parts.outbound_to_twilio_tx,
                tap_url: register_parts.tap_url,
                reply: reply_tx,
            };
            if mount_cmd_tx.send(trunk_register_cmd).await.is_ok() {
                // Wait for ChannelId ack OR Err (the placeholder returns Err
                // via drop(reply)). Either way, the pump resumes.
                let _ = reply_rx.await;
            }
        }
    });
    // Deploy slice A §5.2: app-level keepalive on the only long-lived WS
    // surface. Fail-fast on garbage, like every RUTSTER_* knob.
    let ws_ping_interval =
        rutster::config::ws_ping_interval(std::env::var("RUTSTER_WS_PING_SECS").ok())
            .expect("RUTSTER_WS_PING_SECS must be a positive integer (seconds)");
    let trunk_router = rutster_trunk::twilio_media_streams::TwilioMediaStreamsServer::router(
        trunk_register_tx,
        ws_ping_interval,
    );

    // Deploy slice C §5.5: /metrics on its own listener, never through
    // the TLS edge. The app_state clone is the same one the main router
    // uses; the media thread behind it is shared.
    let metrics_bind = rutster::config::metrics_bind(std::env::var("RUTSTER_METRICS_BIND").ok())
        .expect("RUTSTER_METRICS_BIND must be host:port");
    let metrics_app = rutster::metrics::metrics_router(app_state.clone());
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(metrics_bind)
            .await
            .expect("RUTSTER_METRICS_BIND bind failed");
        info!(%metrics_bind, "metrics listening");
        axum::serve(listener, metrics_app)
            .tcp_nodelay(true)
            .await
            .unwrap();
    });

    let app = router(app_state).merge(trunk_router);

    // Deploy slice C §5.4: branch to TLS or plaintext. ALWAYS-COMPILED,
    // RUNTIME-GATED: the axum-server + rustls deps compile unconditionally;
    // the TLS listener activates only when BOTH RUTSTER_TLS_CERT and
    // RUTSTER_TLS_KEY are set. Plaintext :8080 (plan A) is the artifact
    // default.
    let tls_cert = rutster::config::tls_cert_path(std::env::var("RUTSTER_TLS_CERT").ok())
        .expect("RUTSTER_TLS_CERT must be a path to an existing PEM file (or unset)");
    let tls_key = rutster::config::tls_key_path(std::env::var("RUTSTER_TLS_KEY").ok())
        .expect("RUTSTER_TLS_KEY must be a path to an existing PEM file (or unset)");

    match (tls_cert, tls_key) {
        (Some(cert_path), Some(key_path)) => {
            info!(cert = %cert_path.display(), key = %key_path.display(), "TLS listener enabled");
            let rustls_config = axum_server::tls_rustls::RustlsConfig::from_pem_file(
                &cert_path, &key_path,
            )
            .await
            .expect("RUTSTER_TLS_CERT / RUTSTER_TLS_KEY load failed: bad PEM or key/cert mismatch");

            // Hot-reload watcher (deploy-C §5.4 — reload on cert-file
            // change WITHOUT dropping live WSS). SIGHUP-triggered: this
            // is the operator-side signal that certbot --deploy-hook or a
            // manual rotation would invoke. `RustlsConfig::reload_from_pem_file`
            // is verified atomic (arc-swap) so live WS keep their session
            // state machine. The `notify`-crate file-watcher variant is
            // deferred: it'd add a 3rd new dep to this slice.
            let reload_cfg = rustls_config.clone();
            let reload_cert = cert_path.clone();
            let reload_key = key_path.clone();
            tokio::spawn(async move {
                let mut sighup =
                    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup())
                        .expect("install SIGHUP handler");
                loop {
                    sighup.recv().await;
                    info!("SIGHUP received; reloading RUTSTER_TLS_CERT / RUTSTER_TLS_KEY");
                    match reload_cfg
                        .reload_from_pem_file(&reload_cert, &reload_key)
                        .await
                    {
                        Ok(()) => info!("TLS cert reloaded"),
                        Err(e) => tracing::error!(
                            error = %e,
                            "TLS cert reload failed; keeping old cert"
                        ),
                    }
                }
            });

            // `axum_server::bind(addr)` owns its own listener — pass
            // the SocketAddr, not the existing TcpListener.
            rutster::serve_tls::serve_tls_with_nodelay(addr, rustls_config, app, async {
                let _ = http_stop_rx.await;
            })
            .await
            .unwrap();
        }
        (None, None) => {
            // Plaintext path (plan A). Behind Caddy (default) or any
            // TLS-terminating edge. Listener binds here (only the
            // plaintext arm owns the bind; the TLS arm at
            // serve_tls_with_nodelay above binds addr itself — two
            // binders would EADDRINUSE).
            let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
            rutster::serve::serve_with_nodelay(listener, app, async {
                let _ = http_stop_rx.await;
            })
            .await
            .unwrap();
        }
        (cert_opt, key_opt) => {
            panic!(
                "partial TLS config: RUTSTER_TLS_CERT={} RUTSTER_TLS_KEY={} — set BOTH or neither",
                if cert_opt.is_some() { "set" } else { "unset" },
                if key_opt.is_some() { "set" } else { "unset" },
            );
        }
    }

    media_thread.shutdown();
}

/// Ctrl-C / SIGTERM handler (spec §4.5). This future resolves the instant
/// the signal arrives — it does NOT wait for the drain. The caller (`main`,
/// slice-5 review M1) is the two-phase shutdown: after this resolves, it
/// sends `MediaCmd::Drain` and keeps `axum::serve` running so in-flight
/// calls can still hit DELETE/offer while readyz flips the LB away from
/// this node; once the map empties or `RUTSTER_DRAIN_DEADLINE_SECS` elapses
/// (whichever first — default 0 skips this wait entirely, preserving the
/// old instant-shutdown dev loop), HTTP quiesces and `media_thread.shutdown()`
/// sends `MediaCmd::Shutdown` for the hard stop.
async fn shutdown_signal() {
    let ctrl_c = async {
        tokio::signal::ctrl_c()
            .await
            .expect("installed ctrl-c handler");
    };

    #[cfg(unix)]
    let sigterm = async {
        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("installed SIGTERM handler")
            .recv()
            .await;
    };
    #[cfg(not(unix))]
    let sigterm = std::future::pending::<()>();

    tokio::select! {
        _ = ctrl_c => info!("received Ctrl-C, shutting down"),
        _ = sigterm => info!("received SIGTERM, shutting down"),
    }
}
