path: root/server/src/admin.rs

use serde_json::json;
use axum::Json;

use crate::auth::{audit, hash_password, make_admin_token, AuthUser};
use crate::error::{AppError, AppResult};
use crate::state::Cx;
use crate::util::now_iso;

async fn assert_admin(_cx: &Cx, auth: &AuthUser) -> AppResult<()> {
    if auth.role != "admin" {
        return Err(AppError::forbidden("admin only"));
    }
    Ok(())
}

fn user_json(id: &str, email: &str, name: &str, role: &str, active: i64, ai: i64, limit: i64, created: String) -> serde_json::Value {
    json!({
        "id": id, "email": email, "name": name, "role": role,
        "active": active != 0, "ai_enabled": ai != 0,
        "daily_run_limit": limit, "created_at": created,
    })
}

pub async fn users(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> {
    assert_admin(&cx, &auth).await?;
    let items: Vec<serde_json::Value> = cx.with_db(|db| {
        let mut st = db.prepare("SELECT id,email,name,role,active,ai_enabled,daily_run_limit,created_at FROM users ORDER BY created_at ASC")?;
        let v: Vec<serde_json::Value> = st.query_map([], |r| Ok(user_json(
            &r.get::<_, String>(0)?, &r.get::<_, String>(1)?, &r.get::<_, String>(2)?, &r.get::<_, String>(3)?,
            r.get::<_, i64>(4)?, r.get::<_, i64>(5)?, r.get::<_, i64>(6)?, r.get::<_, String>(7)?,
        )))?.collect::<Result<_, _>>()?;
        Ok::<_, AppError>(v)
    }).await?;
    Ok(Json(json!({ "items": items })))
}

#[derive(serde::Deserialize)]
pub struct AdminUserPatch {
    pub active: Option<bool>,
    pub role: Option<String>,
    pub daily_run_limit: Option<i64>,
    pub ai_enabled: Option<bool>,
}

pub async fn patch_user(cx: Cx, auth: AuthUser, axum::extract::Path(target): axum::extract::Path<String>, body: Option<axum::Json<AdminUserPatch>>) -> AppResult<Json<serde_json::Value>> {
    assert_admin(&cx, &auth).await?;
    let axum::Json(r) = body.ok_or_else(|| AppError::bad("invalid_body", "JSON body required"))?;
    if let Some(role) = &r.role {
        if !matches!(role.as_str(), "admin" | "member") {
            return Err(AppError::bad("validation", "role must be admin|member"));
        }
    }
    if let Some(n) = r.daily_run_limit {
        if !(1..=2000).contains(&n) {
            return Err(AppError::bad("validation", "daily_run_limit must be between 1 and 2000"));
        }
    }
    cx.with_db(|db| -> AppResult<()> {
        let tx = db.transaction()?;
        let (trole, tactive): (String, i64) = tx.query_row(
            "SELECT role, active FROM users WHERE id=?1", [&target], |row| Ok((row.get(0)?, row.get(1)?)))
            .map_err(|_| AppError::not_found("user not found"))?;
        let demoting = r.role.as_ref().map(|new| trole == "admin" && new != "admin").unwrap_or(false);
        let disabling = r.active == Some(false);
        // Last active admin protection, enforced atomically with the update.
        if (demoting || disabling) && trole == "admin" && tactive != 0 {
            let active_admins: i64 = tx.query_row("SELECT COUNT(*) FROM users WHERE role='admin' AND active=1", [], |row| row.get(0))?;
            if active_admins <= 1 {
                return Err(AppError::conflict("last_admin", "cannot demote or disable the last active admin"));
            }
        }
        tx.execute(
            "UPDATE users SET ai_enabled=COALESCE(?1,ai_enabled), daily_run_limit=COALESCE(?2,daily_run_limit), role=COALESCE(?3,role), active=COALESCE(?4,active) WHERE id=?5",
            rusqlite::params![
                r.ai_enabled.map(|b| b as i64),
                r.daily_run_limit,
                &r.role,
                r.active.map(|b| b as i64),
                &target,
            ])?;
        // Disabling revokes every session and freezes that user's live runs,
        // inside the same transaction as the account state flip.
        if disabling {
            tx.execute("DELETE FROM sessions WHERE user_id=?1", [&target]).ok();
            tx.execute(
                "UPDATE runs SET status='cancelled', error='account disabled', finished_at=?1 WHERE user_id=?2 AND status IN ('queued','running')",
                rusqlite::params![now_iso(), &target],
            ).ok();
        }
        tx.commit()?;
        Ok(())
    }).await?;
    audit(&cx, Some(&auth.id), "admin_user_update", &target, "ok").await;
    let v = cx.with_db(|db| {
        db.query_row("SELECT id,email,name,role,active,ai_enabled,daily_run_limit,created_at FROM users WHERE id=?1", [&target], |r|
            Ok(user_json(
                &r.get::<_, String>(0)?, &r.get::<_, String>(1)?, &r.get::<_, String>(2)?, &r.get::<_, String>(3)?,
                r.get::<_, i64>(4)?, r.get::<_, i64>(5)?, r.get::<_, i64>(6)?, r.get::<_, String>(7)?,
            ))).map_err(|_| AppError::not_found("user not found"))
    }).await?;
    Ok(Json(v))
}

#[derive(serde::Deserialize)]
pub struct InvitationReq {
    pub email: Option<String>,
    pub role: Option<String>,
    pub expires_hours: Option<i64>,
}

pub async fn create_invitation(cx: Cx, auth: AuthUser, body: Option<axum::Json<InvitationReq>>) -> AppResult<(axum::http::StatusCode, Json<serde_json::Value>)> {
    assert_admin(&cx, &auth).await?;
    let axum::Json(r) = body.unwrap_or(axum::Json(InvitationReq { email: None, role: None, expires_hours: None }));
    let role = r.role.unwrap_or_else(|| "member".to_string());
    // POC: invitations can only mint members; admins are bootstrapped offline.
    // The invitation's role is stored in DB and registration ignores any
    // client-supplied role input, so no escalation path exists.
    if role != "member" {
        return Err(AppError::bad("validation", "POC invitations can only create member role"));
    }
    let hours = r.expires_hours.unwrap_or(168);
    if !(1..=336).contains(&hours) {
        return Err(AppError::bad("validation", "expires_hours must be 1-336"));
    }
    let (token, expires) = make_admin_token(&cx, "invitations", None, hours, r.email.as_deref()).await?;
    audit(&cx, Some(&auth.id), "invitation_issued", r.email.as_deref().unwrap_or("open-invite"), "ok").await;
    Ok((axum::http::StatusCode::CREATED, Json(json!({ "token": token, "expires_at": expires }))))
}

/// Sanitized invitations list: no token hashes, no secrets.
pub async fn list_invitations(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> {
    assert_admin(&cx, &auth).await?;
    let items: Vec<serde_json::Value> = cx.with_db(|db| {
        let now = now_iso();
        let mut st = db.prepare("SELECT id,email,role,expires_at,used_by,created_at FROM invitations WHERE expires_at > ?1 ORDER BY created_at DESC")?;
        let v = st.query_map([&now], |r| Ok(json!({
            "id": r.get::<_, String>(0)?,
            "email": r.get::<_, Option<String>>(1)?,
            "role": r.get::<_, String>(2)?,
            "expires_at": r.get::<_, String>(3)?,
            "used_by": r.get::<_, Option<String>>(4)?,
            "created_at": r.get::<_, String>(5)?,
        })))?.collect::<Result<Vec<_>, _>>()?;
        Ok::<_, AppError>(v)
    }).await?;
    Ok(Json(json!({ "items": items })))
}

pub async fn delete_invitation(cx: Cx, auth: AuthUser, axum::extract::Path(id): axum::extract::Path<String>) -> AppResult<Json<serde_json::Value>> {
    assert_admin(&cx, &auth).await?;
    cx.with_db(|db| -> AppResult<()> {
        if db.execute("DELETE FROM invitations WHERE id=?1", [&id])? == 0 {
            return Err(AppError::not_found("invitation not found"));
        }
        Ok(())
    }).await?;
    audit(&cx, Some(&auth.id), "invitation_revoke", &id, "ok").await;
    Ok(Json(json!({"ok": true})))
}

/// Admin-issued password reset token, hashed in DB, short-lived single-use.
/// The raw token is returned once for display; no fake email delivery.
pub async fn create_reset(cx: Cx, auth: AuthUser, axum::extract::Path(target): axum::extract::Path<String>, _body: Option<axum::Json<serde_json::Value>>) -> AppResult<(axum::http::StatusCode, Json<serde_json::Value>)> {
    assert_admin(&cx, &auth).await?;
    cx.with_db(|db| {
        db.query_row("SELECT 1 FROM users WHERE id=?1", [&target], |row| row.get::<_, i64>(0))
            .map_err(|_| AppError::not_found("user not found"))?;
        Ok::<_, AppError>(())
    }).await?;
    let hours = 2;
    let (token, expires) = make_admin_token(&cx, "password_resets", Some(target.as_str()), hours, None).await?;
    audit(&cx, Some(&auth.id), "admin_password_reset_issued", &target, "ok").await;
    Ok((axum::http::StatusCode::CREATED, Json(json!({"reset_token": token, "expires_at": expires}))))
}

/// Sanitized security audit listing: actor/action/target/time/status only.
/// Sensitive material never reaches this table (see auth::audit) and stored
/// targets are rendered trimmed, never with password/key/code content.
pub async fn audit_list(cx: Cx, auth: AuthUser) -> AppResult<Json<serde_json::Value>> {
    assert_admin(&cx, &auth).await?;
    let items: Vec<serde_json::Value> = cx.with_db(|db| {
        let mut st = db.prepare("SELECT ts,actor_id,action,target,status FROM audit ORDER BY seq DESC LIMIT 500")?;
        let v = st.query_map([], |r| {
            let ts: String = r.get(0)?;
            let actor: Option<String> = r.get(1)?;
            let action: String = r.get(2)?;
            let target: Option<String> = r.get(3)?;
            let status: String = r.get(4)?;
            let items_json = json!({
                "ts": ts, "actor": actor, "action": action,
                "target": target.unwrap_or_default(), "status": status,
            });
            Ok(items_json)
        })?.collect::<Result<Vec<_>, _>>()?;
        Ok::<_, AppError>(v)
    }).await?;
    Ok(Json(json!({ "items": items })))
}

/// Bootstrap the first admin from env. Idempotent: only inserts when no
/// admin account exists yet.
pub async fn bootstrap_admin(cx: &Cx) -> AppResult<()> {
    let cfg = &cx.cfg;
    let Some(email) = cfg.bootstrap_admin_email.clone() else { return Ok(()); };
    let Some(password) = cfg.bootstrap_admin_password.clone() else { return Ok(()); };
    let email = email.trim().to_lowercase();
    if email.is_empty() || !email.contains('@') || password.len() < 8 {
        return Ok(());
    }
    let exists: i64 = cx.with_db(|db| {
        db.query_row("SELECT COUNT(*) FROM users WHERE role='admin'", [], |r| r.get(0)).unwrap_or(0)
    }).await;
    if exists > 0 { return Ok(()); }
    let hash = hash_password(&password)?;
    let id = crate::util::new_id();
    cx.with_db(|db| {
        db.execute("INSERT OR IGNORE INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES (?1,?2,?3,'Administrator','admin',1,1,100,?4)",
            rusqlite::params![&id, &email, &hash, now_iso()]).ok();
    }).await;
    audit(cx, Some(&id), "bootstrap_admin", &id, "ok").await;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::state::AppState;
    use axum::extract::{Path as P, State};
    use std::sync::Arc;

    fn handle(s: &Arc<AppState>) -> Cx { State(s.clone()) }

    fn admin_auth() -> AuthUser {
        AuthUser { id: "admin-x".into(), email: "[email protected]".into(), role: "admin".into(), ai_enabled: true, daily_run_limit: 100, session_id: "sess-admin".into() }
    }

    fn member_auth() -> AuthUser {
        AuthUser { id: "member-x".into(), email: "[email protected]".into(), role: "member".into(), ai_enabled: false, daily_run_limit: 10, session_id: "sess-member".into() }
    }

    fn test_state() -> Arc<AppState> {
        let conn = rusqlite::Connection::open_in_memory().expect("in-memory db");
        crate::db::init_db(&conn).expect("schema");
        let cfg = crate::config::Config {
            bind_addr: "127.0.0.1:0".into(),
            canonical_origin: String::new(),
            secure_cookies: false,
            db_path: ":memory:".into(),
            frontend_dir: "frontend/dist".into(),
            data_dir: std::env::temp_dir().to_string_lossy().into_owned(),
            worker_image: "test".into(),
            fetch_timeout_secs: 1,
            backtest_timeout_secs: 1,
            run_concurrency: 1,
            fetch_concurrency: 1,
            session_hours: 24,
            bootstrap_admin_email: None,
            bootstrap_admin_password: None,
            ai_base_url: "http://127.0.0.1:9".into(),
            ai_model: "test-model".into(),
            ai_daily_request_cap: 1,
            ai_input_token_cap: 1,
            ai_output_token_cap: 1,
            ai_enabled_poc: false,
            default_run_limit_per_day: 10,
            version: "test".into(),
        };
        Arc::new(AppState {
            cfg,
            db: tokio::sync::Mutex::new(conn),
            run_sem: Arc::new(tokio::sync::Semaphore::new(1)),
            fetch_sem: Arc::new(tokio::sync::Semaphore::new(1)),
        })
    }

    /// Insert a user directly and return its id.
    async fn seed_user(cx: &Cx, email: &str, role: &str, password: &str) -> String {
        let hash = crate::auth::hash_password(password).unwrap();
        let uid = crate::util::new_id();
        cx.with_db(|db| {
            db.execute(
                "INSERT INTO users (id,email,password_hash,name,role,active,ai_enabled,daily_run_limit,created_at) VALUES (?1,?2,?3,?4,?5,1,0,10,?6)",
                rusqlite::params![&uid, email, &hash, "Test User", role, now_iso()]).unwrap();
        }).await;
        uid
    }

    #[tokio::test]
    async fn member_cannot_list_users_or_audit() {
        let s = test_state();
        let u = users(handle(&s), member_auth()).await;
        assert_eq!(u.err().map(|e| e.status).unwrap_or_default(), axum::http::StatusCode::FORBIDDEN);
        let a = audit_list(handle(&s), member_auth()).await;
        assert_eq!(a.err().map(|e| e.status).unwrap_or_default(), axum::http::StatusCode::FORBIDDEN);
        let inv = create_invitation(handle(&s), member_auth(), None).await;
        assert_eq!(inv.err().map(|e| e.status).unwrap_or_default(), axum::http::StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn last_active_admin_cannot_be_disabled_or_demoted() {
        let s = test_state();
        let aid = seed_user(&handle(&s), "[email protected]", "admin", "an-admin-passphrase").await;
        let res = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: Some(false), role: None, daily_run_limit: None, ai_enabled: None }))).await;
        assert_eq!(res.err().map(|e| e.code).unwrap_or_default(), "last_admin", "disabling the last admin must be blocked");
        let res = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: None, role: Some("member".into()), daily_run_limit: None, ai_enabled: None }))).await;
        assert_eq!(res.err().map(|e| e.code).unwrap_or_default(), "last_admin", "demoting the last admin must be blocked");
        // With a second active admin, disabling becomes legal.
        let _ = seed_user(&handle(&s), "[email protected]", "admin", "another-passphrase-2").await;
        let ok = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: Some(false), role: None, daily_run_limit: None, ai_enabled: None }))).await;
        assert!(ok.is_ok());
        let out = ok.unwrap().0;
        assert_eq!(out["active"], json!(false));
    }

    #[tokio::test]
    async fn disabling_user_revokes_sessions_and_cancels_runs() {
        let s = test_state();
        let aid = seed_user(&handle(&s), "[email protected]", "member", "a-member-passphrase").await;
        handle(&s).with_db(|db| {
            db.execute("INSERT INTO sessions (id,user_id,created_at,expires_at) VALUES ('sess-1',?1,?2,?3)",
                rusqlite::params![&aid, now_iso(), crate::util::plus_hours(1)]).unwrap();
            db.execute(
                "INSERT INTO projects (id,user_id,name,draft_code,draft_generation,created_at,updated_at) VALUES ('pid',?1,'p','',0,?2,?2)",
                rusqlite::params![&aid, now_iso()]).unwrap();
            db.execute(
                "INSERT INTO project_versions (id,project_id,code,hash,message,source,created_at) VALUES ('vid','pid','code','h','m','manual',?1)",
                [now_iso()]).unwrap();
            db.execute(
                "INSERT INTO datasets (id,user_id,name,request,status,created_at,updated_at) VALUES ('did',?1,'ds','{}','ready',?2,?2)",
                rusqlite::params![&aid, now_iso()]).unwrap();
            db.execute(
                "INSERT INTO runs (id,user_id,project_id,version_id,dataset_id,status,config,created_at) VALUES ('rid',?1,'pid','vid','did','running','{}',?2)",
                rusqlite::params![&aid, now_iso()]).unwrap();
        }).await;
        let ok = patch_user(handle(&s), admin_auth(), P(aid.clone()), Some(axum::Json(AdminUserPatch { active: Some(false), role: None, daily_run_limit: None, ai_enabled: None }))).await;
        assert!(ok.is_ok(), "member can be disabled");
        let blocked: i64 = handle(&s).with_db(|db| db.query_row(
            "SELECT COUNT(*) FROM runs WHERE id='rid' AND status='cancelled'", [], |r| r.get(0)).unwrap()).await;
        assert_eq!(blocked, 1, "live runs must be cancelled");
        let gone: i64 = handle(&s).with_db(|db| db.query_row(
            "SELECT COUNT(*) FROM sessions WHERE user_id=?1", [&aid], |r| r.get(0)).unwrap()).await;
        assert_eq!(gone, 0, "sessions must be revoked");
    }

    #[tokio::test]
    async fn invitations_only_mint_members_and_are_hashed() {
        let s = test_state();
        let res = create_invitation(handle(&s), admin_auth(), None).await.unwrap();
        assert_eq!(res.0, axum::http::StatusCode::CREATED);
        let token = res.1.0["token"].as_str().unwrap().to_string();
        assert!(!token.contains("password"));
        // The DB must hold only the sha256 of the token, never the raw token.
        let hash = crate::util::sha256_hex(token.as_bytes());
        let hashed_rows: i64 = handle(&s).with_db(|db| db.query_row(
            "SELECT COUNT(*) FROM invitations WHERE token_hash=?1", [&hash], |r| r.get(0)).unwrap()).await;
        assert_eq!(hashed_rows, 1);
        let raw_rows: i64 = handle(&s).with_db(|db| db.query_row(
            "SELECT COUNT(*) FROM invitations WHERE token_hash=?1", [&token], |r| r.get(0)).unwrap()).await;
        assert_eq!(raw_rows, 0, "raw tokens must never be stored");
        // Admin roles via invitation are rejected.
        let offered = create_invitation(handle(&s), admin_auth(), Some(axum::Json(InvitationReq { email: None, role: Some("admin".into()), expires_hours: None }))).await;
        assert_eq!(offered.err().map(|e| e.code).unwrap_or_default(), "validation");
        // Sanitized listing contains no token material.
        let list = list_invitations(handle(&s), admin_auth()).await.unwrap().0;
        let raw = serde_json::to_string(&list).unwrap();
        assert!(!raw.contains(&hash), "listing must not leak token hashes");
    }

    #[tokio::test]
    async fn bootstrap_admin_is_idempotent_and_hashes_password() {
        let mut cfg_state = test_state();
        {
            let cfg = &mut Arc::get_mut(&mut cfg_state).unwrap().cfg;
            cfg.bootstrap_admin_email = Some("[email protected] ".into());
            cfg.bootstrap_admin_password = Some("boot-admin-passphrase".into());
        }
        let cx = handle(&cfg_state);
        bootstrap_admin(&cx).await.unwrap();
        bootstrap_admin(&cx).await.unwrap();
        let count: i64 = cx.with_db(|db| db.query_row(
            "SELECT COUNT(*) FROM users WHERE role='admin'", [], |r| r.get(0)).unwrap()).await;
        assert_eq!(count, 1, "bootstrap must not duplicate admins");
        let hash: String = cx.with_db(|db| db.query_row(
            "SELECT password_hash FROM users WHERE role='admin'", [], |r| r.get(0)).unwrap()).await;
        assert!(hash.starts_with("$argon2"), "bootstrap password must be Argon2 hashed");
    }
}