use argon2::{
    Argon2, PasswordHash, PasswordHasher, PasswordVerifier,
    password_hash::{SaltString, rand_core::OsRng},
};
use axum::{
    Json, Router,
    extract::{
        ConnectInfo, Path, Query, Request, State,
        ws::{Message, WebSocket, WebSocketUpgrade},
    },
    http::{HeaderMap, StatusCode, header},
    middleware::{self, Next},
    response::{IntoResponse, Response},
    routing::{delete, get, post},
};
use futures_util::{SinkExt, StreamExt};
use hmac::{Hmac, Mac};
use serde::Deserialize;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use sqlx::{Row, SqlitePool, sqlite::SqlitePoolOptions};
use std::{
    collections::HashMap,
    net::SocketAddr,
    path::PathBuf,
    sync::Arc,
    time::{SystemTime, UNIX_EPOCH},
};
use tokio::sync::{Mutex, mpsc};
use tower_http::services::{ServeDir, ServeFile};
use uuid::Uuid;

const IDLE: i64 = 30 * 60;
const ABSOLUTE: i64 = 12 * 60 * 60;
fn now() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64
}
fn id() -> String {
    Uuid::new_v4().to_string()
}
fn digest(s: &str) -> String {
    hex::encode(Sha256::digest(s.as_bytes()))
}
type ApiResult<T> = Result<T, ApiError>;
#[derive(Debug)]
struct ApiError(StatusCode, &'static str);
impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        (self.0, Json(json!({"error": self.1}))).into_response()
    }
}
impl From<sqlx::Error> for ApiError {
    fn from(e: sqlx::Error) -> Self {
        eprintln!("database: {e}");
        Self(StatusCode::INTERNAL_SERVER_ERROR, "server")
    }
}
fn bad(code: &'static str) -> ApiError {
    ApiError(StatusCode::BAD_REQUEST, code)
}
fn unauthorized() -> ApiError {
    ApiError(StatusCode::UNAUTHORIZED, "unauthorized")
}

#[derive(Clone)]
struct App {
    db: SqlitePool,
    key: Arc<Vec<u8>>,
    setup: Arc<String>,
    origin: Arc<String>,
    secure: bool,
    peers: Arc<Mutex<Peers>>,
    limits: Arc<Mutex<HashMap<String, (i64, u32)>>>,
    hashes: Arc<tokio::sync::Semaphore>,
}
#[derive(Default)]
struct Peers {
    clients: HashMap<String, Client>,
    host: Option<String>,
    since: Option<i64>,
}
struct Client {
    session: String,
    user: String,
    name: String,
    mode: String,
    tx: mpsc::Sender<Message>,
    shutdown: tokio::sync::watch::Sender<bool>,
}
#[derive(Clone)]
struct Auth {
    session: String,
    user: String,
    name: String,
    role: String,
}
impl Auth {
    fn admin(&self) -> ApiResult<()> {
        if self.role == "admin" {
            Ok(())
        } else {
            Err(ApiError(StatusCode::FORBIDDEN, "forbidden"))
        }
    }
}

async fn database(url: &str) -> Result<SqlitePool, sqlx::Error> {
    let db = SqlitePoolOptions::new()
        .max_connections(5)
        .connect_with(
            url.parse::<sqlx::sqlite::SqliteConnectOptions>()?
                .create_if_missing(true)
                .foreign_keys(true)
                .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
                .busy_timeout(std::time::Duration::from_secs(5)),
        )
        .await?;
    sqlx::raw_sql("CREATE TABLE IF NOT EXISTS accounts(id TEXT PRIMARY KEY, username TEXT NOT NULL UNIQUE COLLATE NOCASE, password TEXT NOT NULL, role TEXT NOT NULL CHECK(role IN ('admin','viewer')), last_login INTEGER);
    CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY, token_hash TEXT NOT NULL UNIQUE, user_id TEXT NOT NULL REFERENCES accounts(id) ON DELETE CASCADE, created INTEGER NOT NULL, last_seen INTEGER NOT NULL);
    CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
    CREATE TABLE IF NOT EXISTS settings(key TEXT PRIMARY KEY, value TEXT NOT NULL);
    PRAGMA optimize;").execute(&db).await?;
    Ok(db)
}
fn password_valid(p: &str) -> bool {
    p.chars().count() >= 12
        && p.len() <= 256
        && p.chars().any(char::is_alphabetic)
        && p.chars().any(|c| c.is_ascii_digit())
}
fn credentials(c: &Credentials) -> ApiResult<()> {
    if c.username.len() < 3
        || c.username.len() > 40
        || !c
            .username
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || "_.-".contains(c))
    {
        return Err(bad("username"));
    }
    if !password_valid(&c.password) {
        return Err(bad("password"));
    }
    Ok(())
}
async fn hash_password(app: &App, password: String) -> ApiResult<String> {
    let permit = app
        .hashes
        .clone()
        .acquire_owned()
        .await
        .map_err(|_| bad("server"))?;
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        Argon2::default()
            .hash_password(password.as_bytes(), &SaltString::generate(&mut OsRng))
            .map(|h| h.to_string())
            .map_err(|_| bad("server"))
    })
    .await
    .map_err(|_| bad("server"))?
}
async fn verify_password(app: &App, password: String, hash: String) -> bool {
    let Ok(permit) = app.hashes.clone().acquire_owned().await else {
        return false;
    };
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        PasswordHash::new(&hash)
            .map(|h| {
                Argon2::default()
                    .verify_password(password.as_bytes(), &h)
                    .is_ok()
            })
            .unwrap_or(false)
    })
    .await
    .unwrap_or(false)
}
fn sign(key: &[u8], token: &str) -> String {
    let mut mac = Hmac::<Sha256>::new_from_slice(key).unwrap();
    mac.update(token.as_bytes());
    format!("{token}.{}", hex::encode(mac.finalize().into_bytes()))
}
fn cookie_token(headers: &HeaderMap, key: &[u8]) -> ApiResult<String> {
    let raw = headers
        .get(header::COOKIE)
        .and_then(|h| h.to_str().ok())
        .and_then(|s| {
            s.split(';')
                .find_map(|c| c.trim().strip_prefix("webka_session="))
        })
        .ok_or_else(unauthorized)?;
    let (token, sig) = raw.split_once('.').ok_or_else(unauthorized)?;
    let mut mac = Hmac::<Sha256>::new_from_slice(key).unwrap();
    mac.update(token.as_bytes());
    mac.verify_slice(&hex::decode(sig).map_err(|_| unauthorized())?)
        .map_err(|_| unauthorized())?;
    Ok(token.to_owned())
}
async fn session_by_id(app: &App, session: &str) -> ApiResult<Auth> {
    let row = sqlx::query("SELECT s.id, a.id AS user_id, a.username, a.role FROM sessions s JOIN accounts a ON a.id=s.user_id WHERE s.id=? AND s.last_seen>? AND s.created>?").bind(session).bind(now()-IDLE).bind(now()-ABSOLUTE).fetch_optional(&app.db).await?.ok_or_else(unauthorized)?;
    Ok(Auth {
        session: row.get("id"),
        user: row.get("user_id"),
        name: row.get("username"),
        role: row.get("role"),
    })
}
async fn auth(app: &App, headers: &HeaderMap, touch: bool) -> ApiResult<Auth> {
    let token = cookie_token(headers, &app.key)?;
    let session: Option<String> = sqlx::query_scalar("SELECT id FROM sessions WHERE token_hash=?")
        .bind(digest(&token))
        .fetch_optional(&app.db)
        .await?;
    let a = session_by_id(app, &session.ok_or_else(unauthorized)?).await?;
    if touch {
        sqlx::query("UPDATE sessions SET last_seen=? WHERE id=?")
            .bind(now())
            .bind(&a.session)
            .execute(&app.db)
            .await?;
    }
    Ok(a)
}
async fn guard(State(app): State<App>, req: Request, next: Next) -> Response {
    let mutation = !matches!(
        *req.method(),
        axum::http::Method::GET | axum::http::Method::HEAD | axum::http::Method::OPTIONS
    );
    if (mutation || req.uri().path() == "/api/signal")
        && req
            .headers()
            .get(header::ORIGIN)
            .and_then(|v| v.to_str().ok())
            != Some(app.origin.as_str())
    {
        return ApiError(StatusCode::FORBIDDEN, "origin").into_response();
    }
    let mut res = next.run(req).await;
    res.headers_mut()
        .insert(header::CACHE_CONTROL, "no-store".parse().unwrap());
    res.headers_mut()
        .insert(header::X_CONTENT_TYPE_OPTIONS, "nosniff".parse().unwrap());
    res.headers_mut()
        .insert(header::REFERRER_POLICY, "no-referrer".parse().unwrap());
    res.headers_mut().insert(header::CONTENT_SECURITY_POLICY, "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data: blob:; media-src 'self' blob:; connect-src 'self' ws: wss:; frame-ancestors 'none'; base-uri 'self'; form-action 'self'".parse().unwrap());
    res
}
async fn throttle(app: &App, key: String) -> ApiResult<()> {
    let mut limits = app.limits.lock().await;
    limits.retain(|_, (start, _)| now() - *start < 60);
    if limits.len() >= 4096 && !limits.contains_key(&key) {
        return Err(ApiError(StatusCode::TOO_MANY_REQUESTS, "rateLimited"));
    }
    let entry = limits.entry(key).or_insert((now(), 0));
    entry.1 += 1;
    if entry.1 > 10 {
        Err(ApiError(StatusCode::TOO_MANY_REQUESTS, "rateLimited"))
    } else {
        Ok(())
    }
}
#[derive(Deserialize)]
struct Credentials {
    username: String,
    password: String,
    setup_token: Option<String>,
}
async fn bootstrap_status(State(app): State<App>) -> ApiResult<Json<Value>> {
    let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM accounts")
        .fetch_one(&app.db)
        .await?;
    Ok(Json(json!({"needs_setup": count == 0})))
}
async fn bootstrap(
    State(app): State<App>,
    ConnectInfo(addr): ConnectInfo<SocketAddr>,
    Json(c): Json<Credentials>,
) -> ApiResult<Response> {
    throttle(&app, addr.ip().to_string()).await?;
    if c.setup_token.as_deref().map(digest) != Some(digest(&app.setup)) {
        return Err(ApiError(StatusCode::FORBIDDEN, "setupToken"));
    }
    credentials(&c)?;
    let hash = hash_password(&app, c.password).await?;
    let result = sqlx::query("INSERT INTO accounts(id,username,password,role) SELECT ?,?,?,'admin' WHERE NOT EXISTS (SELECT 1 FROM accounts)").bind(id()).bind(&c.username).bind(hash).execute(&app.db).await?;
    if result.rows_affected() == 0 {
        return Err(ApiError(StatusCode::CONFLICT, "alreadySetup"));
    }
    Ok(StatusCode::CREATED.into_response())
}
async fn login(
    State(app): State<App>,
    ConnectInfo(addr): ConnectInfo<SocketAddr>,
    Json(c): Json<Credentials>,
) -> ApiResult<Response> {
    throttle(&app, addr.ip().to_string()).await?;
    if c.password.len() > 256 || c.username.len() > 40 {
        return Err(unauthorized());
    }
    let row = sqlx::query("SELECT id,password FROM accounts WHERE username=? COLLATE NOCASE")
        .bind(&c.username)
        .fetch_optional(&app.db)
        .await?;
    // Unknown users perform the same expensive Argon2 operation as existing users.
    let dummy: String = sqlx::query_scalar("SELECT value FROM settings WHERE key='dummy_hash'")
        .fetch_one(&app.db)
        .await?;
    let hash = row
        .as_ref()
        .map(|r| r.get::<String, _>("password"))
        .unwrap_or(dummy);
    if !verify_password(&app, c.password, hash.clone()).await || row.is_none() {
        return Err(unauthorized());
    }
    let row = row.unwrap();
    let uid: String = row.get("id");
    let token = format!("{}{}", id(), id());
    sqlx::query("DELETE FROM sessions WHERE last_seen<=? OR created<=?")
        .bind(now() - IDLE)
        .bind(now() - ABSOLUTE)
        .execute(&app.db)
        .await?;
    // A reset/delete that won the race with password verification must invalidate this login.
    let mut transaction = app.db.begin().await?;
    let inserted = sqlx::query("INSERT INTO sessions(id,token_hash,user_id,created,last_seen) SELECT ?,?,id,?,? FROM accounts WHERE id=? AND password=?")
        .bind(id()).bind(digest(&token)).bind(now()).bind(now()).bind(&uid).bind(&hash)
        .execute(&mut *transaction).await?;
    if inserted.rows_affected() == 0 {
        return Err(unauthorized());
    }
    sqlx::query("UPDATE accounts SET last_login=? WHERE id=?")
        .bind(now())
        .bind(uid)
        .execute(&mut *transaction)
        .await?;
    transaction.commit().await?;
    let cookie = format!(
        "webka_session={}; Path=/; HttpOnly; SameSite=Strict; Max-Age={ABSOLUTE}{}",
        sign(&app.key, &token),
        if app.secure { "; Secure" } else { "" }
    );
    Ok(([(header::SET_COOKIE, cookie)], Json(json!({"ok":true}))).into_response())
}
async fn me(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
    let a = auth(&app, &headers, true).await?;
    Ok(Json(
        json!({"id":a.user,"username":a.name,"role":a.role,"session_id":a.session}),
    ))
}
async fn logout(State(app): State<App>, headers: HeaderMap) -> ApiResult<Response> {
    let a = auth(&app, &headers, false).await?;
    revoke(&app, &a.session).await?;
    Ok((
        [(
            header::SET_COOKIE,
            "webka_session=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0",
        )],
        StatusCode::NO_CONTENT,
    )
        .into_response())
}
async fn accounts(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
    auth(&app, &headers, true).await?.admin()?;
    let rows = sqlx::query("SELECT a.id,a.username,a.role,a.last_login,(SELECT COUNT(*) FROM sessions s WHERE s.user_id=a.id AND s.last_seen>? AND s.created>?) AS active_sessions FROM accounts a ORDER BY a.role,a.username").bind(now()-IDLE).bind(now()-ABSOLUTE).fetch_all(&app.db).await?;
    Ok(Json(json!(rows.iter().map(|r| json!({"id":r.get::<String,_>("id"),"username":r.get::<String,_>("username"),"role":r.get::<String,_>("role"),"last_login":r.get::<Option<i64>,_>("last_login"),"active_sessions":r.get::<i64,_>("active_sessions")})).collect::<Vec<_>>())))
}
async fn create_account(
    State(app): State<App>,
    headers: HeaderMap,
    Json(c): Json<Credentials>,
) -> ApiResult<StatusCode> {
    auth(&app, &headers, true).await?.admin()?;
    credentials(&c)?;
    let hash = hash_password(&app, c.password).await?;
    let result =
        sqlx::query("INSERT INTO accounts(id,username,password,role) VALUES(?,?,?,'viewer')")
            .bind(id())
            .bind(c.username)
            .bind(hash)
            .execute(&app.db)
            .await;
    match result {
        Ok(_) => Ok(StatusCode::CREATED),
        Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
            Err(ApiError(StatusCode::CONFLICT, "duplicate"))
        }
        Err(e) => Err(e.into()),
    }
}
async fn delete_account(
    State(app): State<App>,
    headers: HeaderMap,
    Path(uid): Path<String>,
) -> ApiResult<StatusCode> {
    auth(&app, &headers, true).await?.admin()?;
    let mut tx = app.db.begin().await?;
    let sessions: Vec<String> = sqlx::query_scalar("SELECT id FROM sessions WHERE user_id=?")
        .bind(&uid)
        .fetch_all(&mut *tx)
        .await?;
    let result = sqlx::query("DELETE FROM accounts WHERE id=? AND role='viewer'")
        .bind(&uid)
        .execute(&mut *tx)
        .await?;
    if result.rows_affected() == 0 {
        return Err(bad("protectedAccount"));
    }
    tx.commit().await?;
    for sid in sessions {
        disconnect_session(&app, &sid).await;
    }
    Ok(StatusCode::NO_CONTENT)
}
#[derive(Deserialize)]
struct PasswordChange {
    password: String,
}
async fn reset_password(
    State(app): State<App>,
    headers: HeaderMap,
    Path(uid): Path<String>,
    Json(c): Json<PasswordChange>,
) -> ApiResult<StatusCode> {
    auth(&app, &headers, true).await?.admin()?;
    if !password_valid(&c.password) {
        return Err(bad("password"));
    }
    let hash = hash_password(&app, c.password).await?;
    let mut tx = app.db.begin().await?;
    let result = sqlx::query("UPDATE accounts SET password=? WHERE id=?")
        .bind(hash)
        .bind(&uid)
        .execute(&mut *tx)
        .await?;
    if result.rows_affected() == 0 {
        return Err(bad("notFound"));
    }
    let sessions: Vec<String> = sqlx::query_scalar("SELECT id FROM sessions WHERE user_id=?")
        .bind(&uid)
        .fetch_all(&mut *tx)
        .await?;
    sqlx::query("DELETE FROM sessions WHERE user_id=?")
        .bind(&uid)
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    for sid in sessions {
        disconnect_session(&app, &sid).await;
    }
    Ok(StatusCode::NO_CONTENT)
}
async fn sessions(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
    auth(&app, &headers, true).await?.admin()?;
    let rows = sqlx::query("SELECT s.id,s.created,s.last_seen,a.username FROM sessions s JOIN accounts a ON a.id=s.user_id WHERE s.last_seen>? AND s.created>? ORDER BY s.last_seen DESC").bind(now()-IDLE).bind(now()-ABSOLUTE).fetch_all(&app.db).await?;
    Ok(Json(json!(rows.iter().map(|r| json!({"id":r.get::<String,_>("id"),"username":r.get::<String,_>("username"),"created":r.get::<i64,_>("created"),"last_seen":r.get::<i64,_>("last_seen")})).collect::<Vec<_>>())))
}
async fn revoke(app: &App, sid: &str) -> ApiResult<()> {
    sqlx::query("DELETE FROM sessions WHERE id=?")
        .bind(sid)
        .execute(&app.db)
        .await?;
    disconnect_session(app, sid).await;
    Ok(())
}
async fn revoke_session(
    State(app): State<App>,
    headers: HeaderMap,
    Path(sid): Path<String>,
) -> ApiResult<StatusCode> {
    auth(&app, &headers, true).await?.admin()?;
    revoke(&app, &sid).await?;
    Ok(StatusCode::NO_CONTENT)
}
fn send(client: &Client, value: Value) {
    if client
        .tx
        .try_send(Message::Text(value.to_string().into()))
        .is_err()
    {
        let _ = client.shutdown.send(true);
    }
}
async fn disconnect_session(app: &App, sid: &str) {
    let mut peers = app.peers.lock().await;
    let ids: Vec<_> = peers
        .clients
        .iter()
        .filter(|(_, c)| c.session == sid)
        .map(|(id, _)| id.clone())
        .collect();
    for cid in ids {
        remove_peer(&mut peers, &cid);
    }
}
fn remove_peer(peers: &mut Peers, cid: &str) {
    let Some(client) = peers.clients.remove(cid) else {
        return;
    };
    let _ = client.tx.try_send(Message::Close(None));
    let _ = client.shutdown.send(true);
    if peers.host.as_deref() == Some(cid) {
        peers.host = None;
        peers.since = None;
        for c in peers.clients.values() {
            send(c, json!({"type":"host-offline"}));
        }
    } else if let Some(host) = peers.host.as_ref().and_then(|h| peers.clients.get(h)) {
        send(host, json!({"type":"viewer-left","peer":cid}));
    }
}
async fn status(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
    let a = auth(&app, &headers, true).await?;
    let peers = app.peers.lock().await;
    let viewers: Vec<_> = peers
        .clients
        .iter()
        .filter(|(_, c)| c.mode == "viewer")
        .map(|(id, c)| {
            if a.role == "admin" {
                json!({"id":id,"session_id":c.session,"username":c.name,"user_id":c.user})
            } else {
                json!({})
            }
        })
        .collect();
    Ok(Json(
        json!({"live":peers.host.is_some(),"since":peers.since,"viewer_count":viewers.len(),"viewers":if a.role=="admin" {viewers} else {vec![]},"local_url":format!("{}/watch",app.origin)}),
    ))
}
#[derive(Deserialize)]
struct SignalQuery {
    mode: Option<String>,
}
async fn signal(
    State(app): State<App>,
    headers: HeaderMap,
    Query(q): Query<SignalQuery>,
    ws: WebSocketUpgrade,
) -> ApiResult<Response> {
    let a = auth(&app, &headers, true).await?;
    let mode = q.mode.unwrap_or("viewer".into());
    if mode != "host" && mode != "viewer" {
        return Err(bad("mode"));
    }
    if mode == "host" {
        a.admin()?;
    }
    Ok(ws
        .max_message_size(64 * 1024)
        .max_frame_size(64 * 1024)
        .on_upgrade(move |socket| connection(app, a, mode, socket)))
}
async fn connection(app: App, a: Auth, mode: String, socket: WebSocket) {
    let cid = id();
    let (tx, mut rx) = mpsc::channel(64);
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false);
    let (mut sink, mut source) = socket.split();
    {
        let mut peers = app.peers.lock().await;
        if peers.clients.len() >= 32
            || (mode == "host" && peers.host.is_some())
            || peers.clients.values().any(|c| c.session == a.session)
        {
            drop(peers);
            let _ = tokio::time::timeout(
                std::time::Duration::from_secs(2),
                sink.send(Message::Text(
                    json!({"type":"error","error":"hostBusy"})
                        .to_string()
                        .into(),
                )),
            )
            .await;
            return;
        }
        peers.clients.insert(
            cid.clone(),
            Client {
                session: a.session.clone(),
                user: a.user,
                name: a.name,
                mode: mode.clone(),
                tx,
                shutdown: shutdown_tx,
            },
        );
        if mode == "host" {
            peers.host = Some(cid.clone());
            peers.since = Some(now());
            for (vid, c) in &peers.clients {
                if c.mode == "viewer" {
                    send(
                        peers.clients.get(&cid).unwrap(),
                        json!({"type":"viewer-joined","peer":vid}),
                    );
                    send(c, json!({"type":"host-online"}));
                }
            }
        } else if let Some(host) = peers.host.as_ref().and_then(|h| peers.clients.get(h)) {
            send(host, json!({"type":"viewer-joined","peer":cid}));
        }
        send(
            peers.clients.get(&cid).unwrap(),
            json!({"type":"welcome","peer":cid,"live":peers.host.is_some()}),
        );
    }
    let mut tick = tokio::time::interval(std::time::Duration::from_secs(2));
    let mut count = 0;
    let mut window = now();
    loop {
        tokio::select! {
            _ = shutdown_rx.changed() => break,
            _ = tick.tick() => { if session_by_id(&app,&a.session).await.is_err() { break; } }
            msg = rx.recv() => { match msg { Some(msg) => { let close = matches!(msg, Message::Close(_)); if !tokio::time::timeout(std::time::Duration::from_secs(2), sink.send(msg)).await.is_ok_and(|result| result.is_ok()) || close { break; } }, None => break } }
            msg = source.next() => {
                match msg {
                    Some(Ok(Message::Text(text))) => {
                        if now() != window { window = now(); count = 0; } count += 1; if count > 120 { break; }
                        if session_by_id(&app,&a.session).await.is_err() { break; }
                        let Ok(v) = serde_json::from_str::<Value>(&text) else { continue; };
                        let Some(kind) = v["type"].as_str() else { continue; };
                        let peers = app.peers.lock().await;
                        if !peers.clients.contains_key(&cid) { break; }
                        let Some(target_id) = v["target"].as_str() else { continue; };
                        let Some(target) = peers.clients.get(target_id) else { continue; };
                        let allowed = if mode == "host" { peers.host.as_deref() == Some(&cid) && target.mode=="viewer" && matches!(kind,"offer"|"ice") } else { peers.host.as_deref()==Some(target_id) && matches!(kind,"answer"|"ice"|"renegotiate") };
                        if allowed { send(target,json!({"type":kind,"peer":cid,"data":v["data"]})); }
                    }
                    Some(Ok(Message::Ping(data))) => { if !tokio::time::timeout(std::time::Duration::from_secs(2), sink.send(Message::Pong(data))).await.is_ok_and(|result| result.is_ok()) { break; } }
                    Some(Ok(Message::Close(_))) | None | Some(Err(_)) => break,
                    _ => {}
                }
            }
        }
    }
    remove_peer(&mut *app.peers.lock().await, &cid);
    let _ = tokio::time::timeout(std::time::Duration::from_secs(2), sink.close()).await;
}
fn router(app: App, assets: PathBuf) -> Router {
    Router::new()
        .route("/api/bootstrap", get(bootstrap_status).post(bootstrap))
        .route("/api/login", post(login))
        .route("/api/logout", post(logout))
        .route("/api/session", get(me))
        .route("/api/accounts", get(accounts).post(create_account))
        .route("/api/accounts/{id}", delete(delete_account))
        .route("/api/accounts/{id}/password", post(reset_password))
        .route("/api/sessions", get(sessions))
        .route("/api/sessions/{id}", delete(revoke_session))
        .route("/api/status", get(status))
        .route("/api/signal", get(signal))
        .fallback_service(
            ServeDir::new(&assets).not_found_service(ServeFile::new(assets.join("index.html"))),
        )
        .layer(axum::extract::DefaultBodyLimit::max(8192))
        .layer(middleware::from_fn_with_state(app.clone(), guard))
        .with_state(app)
}
fn secret_file(path: &std::path::Path) -> std::io::Result<String> {
    use std::io::Write;
    if path.exists() {
        return std::fs::read_to_string(path).map(|s| s.trim().to_owned());
    }
    let mut options = std::fs::OpenOptions::new();
    options.write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.mode(0o600);
    }
    let mut file = options.open(path)?;
    let secret = format!("{}{}", id(), id());
    file.write_all(secret.as_bytes())?;
    Ok(secret)
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let data = PathBuf::from(std::env::var("WEBKA_DATA").unwrap_or("data".into()));
    std::fs::create_dir_all(&data)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&data, std::fs::Permissions::from_mode(0o700))?;
    }
    let key = secret_file(&data.join("session-key"))?;
    let setup = secret_file(&data.join("setup-token"))?;
    let origin = std::env::var("WEBKA_ORIGIN")
        .unwrap_or("http://localhost:3000".into())
        .trim_end_matches('/')
        .to_owned();
    let addr: SocketAddr = std::env::var("WEBKA_BIND")
        .unwrap_or("127.0.0.1:3001".into())
        .parse()?;
    let app = App {
        db: database(&format!("sqlite://{}", data.join("webka.db").display())).await?,
        key: Arc::new(key.into_bytes()),
        setup: Arc::new(setup),
        secure: origin.starts_with("https://"),
        origin: Arc::new(origin),
        peers: Default::default(),
        limits: Default::default(),
        hashes: Arc::new(tokio::sync::Semaphore::new(2)),
    };
    let dummy = hash_password(&app, id())
        .await
        .map_err(|_| "hash initialization failed")?;
    sqlx::query("INSERT OR IGNORE INTO settings(key,value) VALUES('dummy_hash',?)")
        .bind(dummy)
        .execute(&app.db)
        .await?;
    let mdns = if std::env::var("WEBKA_MDNS").as_deref() == Ok("1") {
        match (mdns_sd::ServiceDaemon::new(), local_ip_address::local_ip()) {
            (Ok(daemon), Ok(ip)) => {
                let port: u16 = std::env::var("WEBKA_PUBLIC_PORT")
                    .unwrap_or(if app.secure { "443" } else { "3000" }.into())
                    .parse()?;
                let service = mdns_sd::ServiceInfo::new(
                    "_webka._tcp.local.",
                    "Webka",
                    "webka.local.",
                    ip,
                    port,
                    [
                        ("path", "/watch"),
                        ("scheme", if app.secure { "https" } else { "http" }),
                    ]
                    .as_slice(),
                )?;
                daemon.register(service)?;
                Some(daemon)
            }
            _ => {
                eprintln!("mDNS unavailable; use the configured local URL.");
                None
            }
        }
    } else {
        None
    };
    let listener = tokio::net::TcpListener::bind(addr).await?;
    println!(
        "Webka listening at {addr}. Open {}. First-run setup token: {}",
        app.origin,
        data.join("setup-token").display()
    );
    let assets = std::env::var("WEBKA_ASSETS").unwrap_or("frontend/.output/public".into());
    axum::serve(
        listener,
        router(app, assets.into()).into_make_service_with_connect_info::<SocketAddr>(),
    )
    .with_graceful_shutdown(async {
        let _ = tokio::signal::ctrl_c().await;
    })
    .await?;
    if let Some(daemon) = mdns {
        let _ = daemon.shutdown();
    }
    Ok(())
}

#[cfg(test)]
mod tests;