cavs-server 0.8.0

//! `cavs-server` — stateful CAVS-1 origin.
//!
//! Serves one or more `.cavs` files over HTTP:
//!
//! - Control plane (JSON): asset list, manifests, session open.
//! - Data plane (binary `CVSP` batches): per-session inline/ref planning
//!   against the client's have-set — the session-aware dedup layer.
//! - Content-addressable chunk endpoint: stable, edge-cacheable objects.
//! - HLS passthrough: reconstructed `media.m3u8` / `init.mp4` / `seg_*.m4s`
//!   so any standard player (ffplay, Safari, VLC, hls.js) can stream
//!   directly from the origin.
//! - Prometheus-style `/metrics`.

mod state;

use anyhow::{Context, Result};
use axum::extract::{Path, Query, State};
use axum::http::{header, HeaderMap, StatusCode};
use axum::response::{Html, IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use clap::Parser;
use state::{AppState, SharedState};
use std::path::PathBuf;
use std::sync::Arc;

#[derive(Parser)]
#[command(
    name = "cavs-server",
    version,
    about = "CAVS-1 streaming origin server"
)]
struct Cli {
    /// .cavs files to serve (asset name = file stem). Omit when using --store.
    assets: Vec<PathBuf>,
    /// Serve every asset from a shared global content-addressable store
    /// (chunks deduplicated across all assets and versions) instead of from
    /// individual .cavs files. Populate it with `cavs store <dir> add ...`.
    #[arg(long)]
    store: Option<PathBuf>,
    /// Listen address (port 0 picks a free port).
    #[arg(long, default_value = "127.0.0.1:8990")]
    listen: String,
    /// Collapse threshold: if a segment has more cold chunks than this,
    /// deliver it fully inline as a self-sufficient bundle (0 = disabled).
    #[arg(long, default_value_t = 0)]
    max_cold: usize,
    /// Path to the compiled cavs-web WASM module served at /web/cavs_web.wasm.
    #[arg(
        long,
        default_value = "target/wasm32-unknown-unknown/release/cavs_web.wasm"
    )]
    web_wasm: PathBuf,
    /// Serve HTTPS using this PEM certificate (requires --tls-key).
    #[arg(long, requires = "tls_key")]
    tls_cert: Option<PathBuf>,
    /// PEM private key for --tls-cert.
    #[arg(long, requires = "tls_cert")]
    tls_key: Option<PathBuf>,
    /// Serve HTTPS with a self-signed certificate generated into this
    /// directory (cert.pem / key.pem; reused if already present). For
    /// development: point clients at cert.pem via --ca.
    #[arg(long, conflicts_with = "tls_cert")]
    tls_self_signed: Option<PathBuf>,
}

#[tokio::main]
async fn main() -> Result<()> {
    // Two rustls crypto providers exist in the dependency tree; pick one
    // explicitly (idempotent — ignore if something already installed one).
    let _ = rustls::crypto::ring::default_provider().install_default();
    let cli = Cli::parse();
    let state = Arc::new(match &cli.store {
        Some(dir) => AppState::load_store(dir, cli.max_cold, cli.web_wasm)?,
        None => {
            if cli.assets.is_empty() {
                anyhow::bail!("pass .cavs files, or --store <dir> to serve from a global store");
            }
            AppState::load(&cli.assets, cli.max_cold, cli.web_wasm)?
        }
    });
    for name in state.asset_names() {
        eprintln!("[server] asset loaded: {name}");
    }

    let app = Router::new()
        .route("/", get(index))
        .route("/api/assets", get(list_assets))
        .route("/api/assets/{asset}/manifest", get(manifest))
        .route("/api/assets/{asset}/sessions", post(open_session))
        .route("/api/assets/{asset}/bootstrap", get(get_bootstrap))
        .route("/api/assets/{asset}/chunks/{hash}", get(get_chunk))
        .route("/api/sessions/{session}/batch", post(batch))
        .route("/hls/{asset}/{track}/{file}", get(hls_file))
        .route("/web", get(web_index))
        .route("/web/player.js", get(web_js))
        .route("/web/cavs_web.wasm", get(web_wasm))
        .route("/metrics", get(metrics))
        .with_state(state);

    let tls_files: Option<(PathBuf, PathBuf)> = match (&cli.tls_cert, &cli.tls_self_signed) {
        (Some(cert), _) => Some((cert.clone(), cli.tls_key.clone().unwrap())),
        (None, Some(dir)) => Some(ensure_self_signed(dir)?),
        (None, None) => None,
    };

    // Bind with std first so port 0 resolves before printing the banner
    // (tests and scripts parse the "listening on" line).
    let listener = std::net::TcpListener::bind(&cli.listen)
        .with_context(|| format!("cannot bind {}", cli.listen))?;
    listener.set_nonblocking(true)?;
    let addr = listener.local_addr()?;

    match tls_files {
        Some((cert, key)) => {
            let config = axum_server::tls_rustls::RustlsConfig::from_pem_file(&cert, &key)
                .await
                .with_context(|| {
                    format!(
                        "loading TLS cert {} / key {}",
                        cert.display(),
                        key.display()
                    )
                })?;
            println!("listening on https://{addr}");
            axum_server::from_tcp_rustls(listener, config)
                .serve(app.into_make_service())
                .await?;
        }
        None => {
            println!("listening on http://{addr}");
            axum::serve(tokio::net::TcpListener::from_std(listener)?, app).await?;
        }
    }
    Ok(())
}

/// Generate (or reuse) a self-signed localhost certificate for development.
fn ensure_self_signed(dir: &PathBuf) -> Result<(PathBuf, PathBuf)> {
    let cert_path = dir.join("cert.pem");
    let key_path = dir.join("key.pem");
    if !cert_path.exists() || !key_path.exists() {
        std::fs::create_dir_all(dir)?;
        let cert = rcgen::generate_simple_self_signed(vec![
            "localhost".to_string(),
            "127.0.0.1".to_string(),
        ])
        .context("generating self-signed certificate")?;
        std::fs::write(&cert_path, cert.cert.pem())?;
        std::fs::write(&key_path, cert.key_pair.serialize_pem())?;
        eprintln!(
            "[server] self-signed TLS cert written to {}",
            cert_path.display()
        );
    }
    Ok((cert_path, key_path))
}

type AppError = (StatusCode, String);

fn not_found(what: impl std::fmt::Display) -> AppError {
    (StatusCode::NOT_FOUND, format!("{what} not found"))
}

async fn index(State(state): State<SharedState>) -> Html<String> {
    let mut html = String::from(
        "<h1>cavs-server</h1><p><a href=\"/web\">reproductor web (WASM + MSE)</a></p><ul>",
    );
    for name in state.asset_names() {
        html.push_str(&format!(
            "<li>{name} — <a href=\"/api/assets/{name}/manifest\">manifest</a>"
        ));
        for track in state.video_track_names(&name) {
            html.push_str(&format!(
                " | <a href=\"/hls/{name}/{track}/media.m3u8\">hls:{track}</a>"
            ));
        }
        html.push_str("</li>");
    }
    html.push_str("</ul>");
    Html(html)
}

async fn list_assets(State(state): State<SharedState>) -> Json<Vec<cavs_proto::AssetSummary>> {
    Json(state.summaries())
}

#[derive(serde::Deserialize)]
struct ManifestQuery {
    /// `binary-v2` or `json-v1`; overrides content negotiation.
    format: Option<String>,
}

/// v0.3.0 compact manifest: one endpoint, two wire formats. Binary v2 is
/// served when the client asks for it (Accept header or `?format=`);
/// JSON v1 stays the default so v0.2.x clients keep working unchanged.
async fn manifest(
    State(state): State<SharedState>,
    Path(asset): Path<String>,
    Query(query): Query<ManifestQuery>,
    headers: HeaderMap,
) -> Result<Response, AppError> {
    let manifest = state
        .manifest(&asset)
        .ok_or_else(|| not_found(format!("asset {asset}")))?;

    let binary = match query.format.as_deref() {
        Some("binary-v2") => true,
        Some("json-v1") => false,
        Some(other) => {
            return Err((
                StatusCode::BAD_REQUEST,
                format!("unknown manifest format {other} (binary-v2 | json-v1)"),
            ))
        }
        None => headers
            .get(header::ACCEPT)
            .and_then(|v| v.to_str().ok())
            .is_some_and(|accept| accept.contains(cavs_manifest::MANIFEST_V2_CONTENT_TYPE)),
    };

    if binary {
        // Packfile-store assets also carry chunk location hints (0.4.0).
        let bytes = cavs_manifest::encode_manifest_v2_with_locations(
            &manifest,
            state.manifest_locations(&asset),
        )
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
        state.count_manifest_request("binary-v2", bytes.len() as u64);
        Ok((
            [(
                header::CONTENT_TYPE,
                cavs_manifest::MANIFEST_V2_CONTENT_TYPE,
            )],
            bytes,
        )
            .into_response())
    } else {
        let bytes = serde_json::to_vec(&manifest)
            .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
        state.count_manifest_request("json-v1", bytes.len() as u64);
        Ok(([(header::CONTENT_TYPE, "application/json")], bytes).into_response())
    }
}

async fn open_session(
    State(state): State<SharedState>,
    Path(asset): Path<String>,
    Json(req): Json<cavs_proto::SessionOpenRequest>,
) -> Result<Json<cavs_proto::SessionOpenResponse>, AppError> {
    state
        .open_session(&asset, &req.have, req.have_bloom.as_ref())
        .map(Json)
        .ok_or_else(|| not_found(format!("asset {asset}")))
}

async fn batch(
    State(state): State<SharedState>,
    Path(session): Path<String>,
    Json(req): Json<cavs_proto::BatchRequest>,
) -> Result<Response, AppError> {
    let bytes = state
        .plan_batch(&session, &req)
        .map_err(|e| (StatusCode::BAD_REQUEST, e))?;
    Ok(([(header::CONTENT_TYPE, "application/octet-stream")], bytes).into_response())
}

/// Full bootstrap artifact (whole asset, zstd): the cold-install fast path.
/// Streamed from disk so a multi-hundred-MiB artifact never sits in RAM.
/// Supports single `Range: bytes=start[-end]` requests (v0.5.0) so an
/// interrupted download resumes instead of restarting; the artifact is
/// immutable, so a resumed range always continues the same bytes.
async fn get_bootstrap(
    State(state): State<SharedState>,
    Path(asset): Path<String>,
    headers: HeaderMap,
) -> Result<Response, AppError> {
    let (path, size, blake3_hex) = state
        .bootstrap_file(&asset)
        .ok_or_else(|| not_found(format!("bootstrap for {asset}")))?;
    let mut file = tokio::fs::File::open(&path)
        .await
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
    let common = [
        (header::CONTENT_TYPE, "application/zstd".to_string()),
        (header::ACCEPT_RANGES, "bytes".to_string()),
        // Tied to the packed content: immutable, edge-cacheable.
        (
            header::CACHE_CONTROL,
            "public, max-age=31536000, immutable".to_string(),
        ),
        (header::ETAG, format!("\"blake3-{blake3_hex}\"")),
    ];

    let range = headers
        .get(header::RANGE)
        .and_then(|v| v.to_str().ok())
        .map(|s| parse_byte_range(s, size));
    match range {
        // Malformed/multi/suffix ranges: answer with the full body (200).
        None | Some(RangeParse::Full) => {
            let stream = tokio_util::io::ReaderStream::new(file);
            Ok((
                common,
                [(header::CONTENT_LENGTH, size.to_string())],
                axum::body::Body::from_stream(stream),
            )
                .into_response())
        }
        Some(RangeParse::Unsatisfiable) => Ok((
            StatusCode::RANGE_NOT_SATISFIABLE,
            [(header::CONTENT_RANGE, format!("bytes */{size}"))],
        )
            .into_response()),
        Some(RangeParse::Range(start, end)) => {
            use tokio::io::AsyncSeekExt as _;
            file.seek(std::io::SeekFrom::Start(start))
                .await
                .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
            let len = end - start + 1;
            let stream =
                tokio_util::io::ReaderStream::new(tokio::io::AsyncReadExt::take(file, len));
            Ok((
                StatusCode::PARTIAL_CONTENT,
                common,
                [
                    (header::CONTENT_LENGTH, len.to_string()),
                    (header::CONTENT_RANGE, format!("bytes {start}-{end}/{size}")),
                ],
                axum::body::Body::from_stream(stream),
            )
                .into_response())
        }
    }
}

enum RangeParse {
    /// No usable single range: serve the whole body as 200.
    Full,
    /// Valid inclusive byte range within the resource.
    Range(u64, u64),
    /// Syntactically valid but beyond the resource: 416.
    Unsatisfiable,
}

/// Parse a single `bytes=start[-end]` header value. Suffix ranges
/// (`bytes=-N`) and multi-range requests fall back to the full body.
fn parse_byte_range(value: &str, size: u64) -> RangeParse {
    let Some(spec) = value.strip_prefix("bytes=") else {
        return RangeParse::Full;
    };
    if spec.contains(',') {
        return RangeParse::Full;
    }
    let Some((start_s, end_s)) = spec.split_once('-') else {
        return RangeParse::Full;
    };
    let Ok(start) = start_s.trim().parse::<u64>() else {
        return RangeParse::Full; // suffix range or garbage
    };
    let end = if end_s.trim().is_empty() {
        size.saturating_sub(1)
    } else {
        match end_s.trim().parse::<u64>() {
            Ok(e) => e.min(size.saturating_sub(1)),
            Err(_) => return RangeParse::Full,
        }
    };
    if size == 0 || start >= size || start > end {
        return RangeParse::Unsatisfiable;
    }
    RangeParse::Range(start, end)
}

async fn get_chunk(
    State(state): State<SharedState>,
    Path((asset, hash)): Path<(String, String)>,
) -> Result<Response, AppError> {
    let bytes = state
        .chunk_by_hash(&asset, &hash)
        .ok_or_else(|| not_found(format!("chunk {hash}")))?;
    Ok((
        [
            (header::CONTENT_TYPE, "application/octet-stream".to_string()),
            // Content-addressed: immutable forever, ideal for edge caches.
            (
                header::CACHE_CONTROL,
                "public, max-age=31536000, immutable".to_string(),
            ),
            (header::ETAG, format!("\"blake3-{hash}\"")),
        ],
        bytes,
    )
        .into_response())
}

async fn hls_file(
    State(state): State<SharedState>,
    Path((asset, track, file)): Path<(String, String, String)>,
) -> Result<Response, AppError> {
    let (bytes, content_type) = state
        .hls_file(&asset, &track, &file)
        .ok_or_else(|| not_found(format!("{asset}/{track}/{file}")))?;
    Ok(([(header::CONTENT_TYPE, content_type)], bytes).into_response())
}

async fn web_index() -> Html<&'static str> {
    Html(include_str!("../static/index.html"))
}

async fn web_js() -> Response {
    (
        [(header::CONTENT_TYPE, "application/javascript")],
        include_str!("../static/player.js"),
    )
        .into_response()
}

async fn web_wasm(State(state): State<SharedState>) -> Result<Response, AppError> {
    let path = state.web_wasm_path();
    // Small module read on demand; std read keeps tokio features minimal.
    let bytes = std::fs::read(path).map_err(|_| {
        (
            StatusCode::NOT_FOUND,
            format!(
                "WASM module not found at {}. Build it with:\n  cargo build -p cavs-web \
                 --target wasm32-unknown-unknown --release\nor pass --web-wasm <path>.",
                path.display()
            ),
        )
    })?;
    Ok(([(header::CONTENT_TYPE, "application/wasm")], bytes).into_response())
}

async fn metrics(State(state): State<SharedState>) -> Response {
    (
        [(header::CONTENT_TYPE, "text/plain; version=0.0.4")],
        state.render_metrics(),
    )
        .into_response()
}