mod admin;
mod auth;
mod ingest;
mod models;
mod nfs;
mod worker;

use auth::AuthUser;
use axum::{
    body::Body,
    extract::{Path as AxPath, Query, State},
    http::{header, StatusCode},
    response::Response,
    routing::{delete, get, post},
    Json, Router,
};
use models::*;
use serde_json::json;
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::time::Duration;
use tower_http::cors::CorsLayer;

const SCHEMA: &str = include_str!("../../db/init.sql");

/// Max rows kept in each append-only log (audit_events + non-push video_events).
/// Trimmed nightly so the DB doesn't grow unbounded. 'push' video_events are kept
/// regardless (they're load-bearing: verify_count / verifiers derive from them).
const MAX_LOG_ROWS: i64 = 5000;

/// Window (days) for the "verified per day" verification-pace metric.
const RECENT_WINDOW_DAYS: i64 = 7;

/// SQL fragment: restrict to videos in a "considered" folder, or all folders when
/// none are configured for the project. References the bound project_id as `$1` and
/// the unaliased `videos` table (so `rel_path` resolves). Append to a query whose
/// first bind is the project id.
const FOLDER_FILTER: &str = " AND (NOT EXISTS (SELECT 1 FROM project_folders pf WHERE pf.project_id = $1) \
    OR (CASE WHEN rel_path LIKE '%/%' THEN regexp_replace(rel_path, '/[^/]*$', '') ELSE '' END) \
       IN (SELECT folder FROM project_folders WHERE project_id = $1)) ";

#[derive(Clone)]
pub struct AppState {
    pub db: PgPool,
    pub ingest_dir: PathBuf,
    /// Master token granting admin role (bootstrap / break-glass).
    pub admin_token: String,
    /// Claim lease length in seconds.
    pub lease_secs: i32,
    /// NAS host for `nfs`-kind projects (e.g. 192.168.1.199). Server-side only — never returned.
    pub nfs_host: String,
    /// Mount options for `nfs`-kind projects.
    pub nfs_opts: String,
}

pub type ApiResult<T> = Result<Json<T>, (StatusCode, String)>;
pub fn err<E: std::fmt::Display>(e: E) -> (StatusCode, String) {
    (StatusCode::INTERNAL_SERVER_ERROR, e.to_string())
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into()))
        .init();

    let db_url = std::env::var("DATABASE_URL")
        .unwrap_or_else(|_| "postgres://central:central@localhost:5433/central".into());
    let bind = std::env::var("BIND_ADDR").unwrap_or_else(|_| "0.0.0.0:8080".into());
    let ingest_dir = PathBuf::from(std::env::var("INGEST_DIR").unwrap_or_else(|_| "./sample-data".into()));
    let project_name = std::env::var("PROJECT_NAME").unwrap_or_else(|_| "default".into());
    let admin_token = std::env::var("ADMIN_TOKEN").unwrap_or_else(|_| "dev-admin-token".into());
    if admin_token == "dev-admin-token" {
        tracing::warn!("ADMIN_TOKEN is the insecure default 'dev-admin-token' — set it in production");
    }
    let lease_secs: i32 = std::env::var("LEASE_SECS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(900);
    let nfs_host = std::env::var("NFS_HOST").unwrap_or_default();
    let nfs_opts = std::env::var("NFS_OPTS")
        .unwrap_or_else(|_| "nfsvers=4,ro,soft,timeo=30,retrans=2,retry=0".into());

    // Connect with retry — the DB container may still be starting.
    let db = connect_with_retry(&db_url, 30).await?;
    sqlx::raw_sql(SCHEMA).execute(&db).await?;

    // On a fresh DB only, seed a 'default' client/project at INGEST_DIR for local dev.
    // Once real clients exist, we leave the data alone (no 'default' reappearing).
    let client_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM clients")
        .fetch_one(&db)
        .await?;
    if client_count == 0 {
        let default_client_id: i32 =
            sqlx::query_scalar("INSERT INTO clients (name) VALUES ('default') RETURNING id")
                .fetch_one(&db)
                .await?;
        let project_id: i32 = sqlx::query_scalar(
            r#"INSERT INTO projects (name, source_path, client_id) VALUES ($1,$2,$3)
               ON CONFLICT (client_id, name) DO UPDATE SET source_path=EXCLUDED.source_path
               RETURNING id"#,
        )
        .bind(&project_name)
        .bind(ingest_dir.to_string_lossy().to_string())
        .bind(default_client_id)
        .fetch_one(&db)
        .await?;
        match ingest::ingest_project(&db, project_id, &ingest_dir).await {
            Ok(n) => tracing::info!("seeded default project; startup ingest: {n} videos from {}", ingest_dir.display()),
            Err(e) => tracing::warn!("startup ingest failed (non-fatal): {e}"),
        }
    } else {
        tracing::info!("{client_count} client(s) present; skipping default seed");
    }

    let state = AppState { db, ingest_dir, admin_token, lease_secs, nfs_host, nfs_opts };

    // Phase 3: background task that auto-releases expired claims back to the pool.
    {
        let db = state.db.clone();
        tokio::spawn(async move {
            let mut tick = tokio::time::interval(Duration::from_secs(60));
            loop {
                tick.tick().await;
                match release_expired_leases(&db).await {
                    Ok(n) if n > 0 => tracing::info!("auto-released {n} expired claim(s)"),
                    Ok(_) => {}
                    Err(e) => tracing::warn!("lease sweep failed: {e}"),
                }
                match auto_stop_idle_timers(&db).await {
                    Ok(n) if n > 0 => tracing::info!("auto-stopped {n} idle project timer(s) (72h)"),
                    Ok(_) => {}
                    Err(e) => tracing::warn!("timer sweep failed: {e}"),
                }
            }
        });
    }

    // Nightly retention: trim the append-only logs to MAX_LOG_ROWS at ~00:00 UTC
    // (run once on startup too, to bound an already-large DB immediately).
    {
        let db = state.db.clone();
        tokio::spawn(async move {
            loop {
                match retention_sweep(&db).await {
                    Ok((a, e)) if a + e > 0 => tracing::info!("retention: trimmed {a} audit + {e} event row(s)"),
                    Ok(_) => {}
                    Err(e) => tracing::warn!("retention sweep failed: {e}"),
                }
                tokio::time::sleep(Duration::from_secs(secs_until_next_utc_midnight())).await;
            }
        });
    }

    let app = Router::new()
        .route("/health", get(|| async { "ok" }))
        // Dashboard read APIs — ALL require a valid token now (no anonymous access):
        // both admins and ml_support must sign in before any data is returned.
        .route("/api/clients", get(list_clients).post(admin::create_client))
        .route("/api/clients/:id", delete(admin::delete_client))
        .route("/api/projects", get(list_projects).post(admin::create_project))
        .route("/api/projects/:id", delete(admin::delete_project))
        .route("/api/projects/:id/source", post(admin::update_project_source))
        .route("/api/projects/:id/ingest", post(trigger_ingest))
        .route("/api/projects/:id/videos", get(list_videos))
        .route("/api/projects/:id/stats", get(stats))
        .route("/api/projects/:id/activity", get(activity))
        .route("/api/audit", get(list_audit))
        .route("/api/videos/:id/remarks", get(list_remarks).post(add_remark))
        .route("/api/videos/:id/hand", post(set_hand))
        .route("/api/videos/:id/ignore", post(set_ignore))
        // Export downloads are hit via <a download>, which can't set a header — they
        // authenticate with a `?token=` query param instead (validated in-handler).
        .route("/api/videos/:id/export", get(export_video))
        .route("/api/projects/:id/export", get(export_project))
        .route("/api/projects/:id/assign", post(assign_videos))
        .route("/api/projects/:id/members", get(admin::list_members).post(admin::add_member))
        .route("/api/projects/:id/members/:username", delete(admin::remove_member))
        .route("/api/projects/:id/timer", post(project_timer))
        .route("/api/projects/:id/folders", get(list_folders).post(set_folders))
        // Phase 3 admin (token-gated).
        .route("/api/admin/storage", get(storage_info))
        .route("/api/users", get(admin::list_users).post(admin::create_user))
        .route("/api/users/:id", delete(admin::delete_user))
        .route("/api/users/:id/update", post(admin::update_user))
        .route("/api/users/:id/token", post(admin::rotate_token))
        // Phase 3 worker pull/push (token-gated).
        .route("/api/auth/whoami", get(worker::whoami))
        .route("/api/auth/login", post(worker::login))
        .route("/api/videos/:id/claim", post(worker::claim))
        .route("/api/videos/:id/download", get(worker::download))
        .route("/api/videos/:id/heartbeat", post(worker::heartbeat))
        .route("/api/videos/:id/release", post(worker::release))
        .route("/api/videos/:id/push", post(worker::push))
        .layer(CorsLayer::permissive())
        .with_state(state);

    let addr: SocketAddr = bind.parse()?;
    tracing::info!("listening on {addr}");
    let listener = tokio::net::TcpListener::bind(addr).await?;
    axum::serve(listener, app).await?;
    Ok(())
}

/// Trim the append-only logs to the most-recent `MAX_LOG_ROWS` rows so the DB stays
/// bounded. `audit_events` is trimmed wholesale; `video_events` keeps ALL 'push'
/// rows (load-bearing) and trims the rest. Returns (audit_deleted, events_deleted).
async fn retention_sweep(db: &PgPool) -> Result<(u64, u64), sqlx::Error> {
    let audit = sqlx::query(
        "DELETE FROM audit_events WHERE id NOT IN (SELECT id FROM audit_events ORDER BY at DESC, id DESC LIMIT $1)",
    )
    .bind(MAX_LOG_ROWS)
    .execute(db)
    .await?
    .rows_affected();
    let events = sqlx::query(
        r#"DELETE FROM video_events
           WHERE event <> 'push'
             AND id NOT IN (
               SELECT id FROM video_events WHERE event <> 'push' ORDER BY at DESC, id DESC LIMIT $1
             )"#,
    )
    .bind(MAX_LOG_ROWS)
    .execute(db)
    .await?
    .rows_affected();
    Ok((audit, events))
}

/// Seconds from now until the next 00:00 UTC (the nightly retention tick).
fn secs_until_next_utc_midnight() -> u64 {
    let now = chrono::Utc::now();
    let next = (now + chrono::Duration::days(1))
        .date_naive()
        .and_hms_opt(0, 0, 0)
        .unwrap()
        .and_utc();
    (next - now).num_seconds().max(60) as u64
}

/// `GET /api/admin/storage` — database size + row counts (admin only).
async fn storage_info(State(s): State<AppState>, user: AuthUser) -> ApiResult<StorageInfo> {
    user.require_admin()?;
    let (db_bytes, videos, annotations, video_events, audit_events): (i64, i64, i64, i64, i64) =
        sqlx::query_as(
            r#"SELECT pg_database_size(current_database())::bigint,
                      (SELECT count(*) FROM videos)::bigint,
                      (SELECT count(*) FROM annotations)::bigint,
                      (SELECT count(*) FROM video_events)::bigint,
                      (SELECT count(*) FROM audit_events)::bigint"#,
        )
        .fetch_one(&s.db)
        .await
        .map_err(err)?;
    Ok(Json(StorageInfo {
        db_bytes, videos, annotations, video_events, audit_events,
        max_log_rows: MAX_LOG_ROWS,
    }))
}

/// Flip every expired, unfinished claim back to available and log an `auto_release`
/// event for each. Returns the number released.
async fn release_expired_leases(db: &PgPool) -> Result<u64, sqlx::Error> {
    let res = sqlx::query(
        r#"
        WITH expired AS (
            UPDATE videos
            SET claimed_by=NULL, claimed_at=NULL, lease_expires_at=NULL
            WHERE claimed_by IS NOT NULL AND lease_expires_at < now() AND status <> 'verified'
            RETURNING id
        )
        INSERT INTO video_events (video_id, username, event)
        SELECT id, '', 'auto_release' FROM expired
        "#,
    )
    .execute(db)
    .await?;
    Ok(res.rows_affected())
}

/// Auto-stop project timers idle for >72h: accumulate up to the last activity (so the
/// idle stretch isn't counted) and pause. Resumes automatically on the next push.
async fn auto_stop_idle_timers(db: &PgPool) -> Result<u64, sqlx::Error> {
    let res = sqlx::query(
        r#"UPDATE projects SET
             time_accumulated_ms = time_accumulated_ms + CASE WHEN time_started_at IS NOT NULL
               THEN GREATEST(0, (EXTRACT(EPOCH FROM (LEAST(last_activity_at, now()) - time_started_at)) * 1000)::bigint)
               ELSE 0 END,
             time_running = FALSE,
             time_started_at = NULL
           WHERE time_running AND last_activity_at IS NOT NULL
             AND now() - last_activity_at > interval '72 hours'"#,
    )
    .execute(db)
    .await?;
    Ok(res.rows_affected())
}

/// `POST /api/projects/:id/timer {action}` — start | stop | reset the project timer
/// (admin only). start = resume (no-op if already running); stop = pause; reset = zero.
async fn project_timer(
    State(s): State<AppState>,
    user: AuthUser,
    AxPath(id): AxPath<i32>,
    Json(req): Json<TimerRequest>,
) -> ApiResult<serde_json::Value> {
    user.require_admin()?;
    let exists: Option<i32> = sqlx::query_scalar("SELECT id FROM projects WHERE id=$1")
        .bind(id)
        .fetch_optional(&s.db)
        .await
        .map_err(err)?;
    if exists.is_none() {
        return Err((StatusCode::NOT_FOUND, "project not found".into()));
    }
    let sql = match req.action.as_str() {
        "start" => {
            "UPDATE projects SET time_running=TRUE, last_activity_at=now(),
               time_started_at = CASE WHEN time_running THEN time_started_at ELSE now() END
             WHERE id=$1"
        }
        "stop" => {
            "UPDATE projects SET
               time_accumulated_ms = time_accumulated_ms + CASE WHEN time_running AND time_started_at IS NOT NULL
                 THEN GREATEST(0, (EXTRACT(EPOCH FROM (now() - time_started_at)) * 1000)::bigint) ELSE 0 END,
               time_running=FALSE, time_started_at=NULL
             WHERE id=$1"
        }
        "reset" => {
            "UPDATE projects SET time_accumulated_ms=0, last_activity_at=now(),
               time_started_at = CASE WHEN time_running THEN now() ELSE NULL END
             WHERE id=$1"
        }
        _ => return Err((StatusCode::BAD_REQUEST, "action must be start, stop or reset".into())),
    };
    sqlx::query(sql).bind(id).execute(&s.db).await.map_err(err)?;
    admin::audit(&s.db, &user.username, "project_timer", json!({ "project_id": id, "action": req.action })).await;
    Ok(Json(json!({ "action": req.action })))
}

/// `GET /api/projects/:id/folders` — folders in the project + video counts + whether
/// each is currently included in verification. Admin or member.
async fn list_folders(State(s): State<AppState>, user: AuthUser, AxPath(id): AxPath<i32>) -> ApiResult<Vec<FolderRow>> {
    require_project_access(&s, &user, id).await?;
    let rows = sqlx::query_as::<_, FolderRow>(
        r#"SELECT d.folder, d.video_count, (pf.folder IS NOT NULL) AS included
           FROM (SELECT CASE WHEN rel_path LIKE '%/%' THEN regexp_replace(rel_path, '/[^/]*$', '') ELSE '' END AS folder,
                        count(*)::bigint AS video_count
                 FROM videos WHERE project_id=$1 GROUP BY 1) d
           LEFT JOIN project_folders pf ON pf.project_id=$1 AND pf.folder=d.folder
           ORDER BY d.folder"#,
    )
    .bind(id)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    Ok(Json(rows))
}

/// `POST /api/projects/:id/folders {folders}` — replace the considered-folders set
/// (admin only). Empty list = consider all folders again.
async fn set_folders(
    State(s): State<AppState>,
    user: AuthUser,
    AxPath(id): AxPath<i32>,
    Json(req): Json<FoldersRequest>,
) -> ApiResult<serde_json::Value> {
    user.require_admin()?;
    let mut tx = s.db.begin().await.map_err(err)?;
    sqlx::query("DELETE FROM project_folders WHERE project_id=$1")
        .bind(id)
        .execute(&mut *tx)
        .await
        .map_err(err)?;
    for f in &req.folders {
        sqlx::query("INSERT INTO project_folders (project_id, folder) VALUES ($1,$2) ON CONFLICT DO NOTHING")
            .bind(id)
            .bind(f)
            .execute(&mut *tx)
            .await
            .map_err(err)?;
    }
    tx.commit().await.map_err(err)?;
    admin::audit(&s.db, &user.username, "folders_updated", json!({ "project_id": id, "count": req.folders.len() })).await;
    Ok(Json(json!({ "folders": req.folders.len() })))
}

async fn connect_with_retry(url: &str, attempts: u32) -> anyhow::Result<PgPool> {
    let mut last = None;
    for i in 1..=attempts {
        match PgPoolOptions::new().max_connections(10).connect(url).await {
            Ok(pool) => return Ok(pool),
            Err(e) => {
                tracing::warn!("db connect attempt {i}/{attempts} failed: {e}");
                last = Some(e);
                tokio::time::sleep(Duration::from_secs(2)).await;
            }
        }
    }
    Err(anyhow::anyhow!("db connect failed: {:?}", last))
}

/// True if the user may access this project: admins always; others only if a member.
async fn is_member_or_admin(s: &AppState, user: &AuthUser, project_id: i32) -> Result<bool, (StatusCode, String)> {
    if user.is_admin() {
        return Ok(true);
    }
    let m: Option<i32> = sqlx::query_scalar(
        "SELECT 1 FROM project_members WHERE project_id=$1 AND username=$2",
    )
    .bind(project_id)
    .bind(&user.username)
    .fetch_optional(&s.db)
    .await
    .map_err(err)?;
    Ok(m.is_some())
}

/// 403 unless the caller is an admin or a member of this project.
async fn require_project_access(s: &AppState, user: &AuthUser, project_id: i32) -> Result<(), (StatusCode, String)> {
    if is_member_or_admin(s, user, project_id).await? {
        Ok(())
    } else {
        Err((StatusCode::FORBIDDEN, "you are not assigned to this project".into()))
    }
}

async fn list_projects(State(s): State<AppState>, user: AuthUser) -> ApiResult<Vec<ProjectSummary>> {
    // Non-admins only see projects they're a member of (picked by an admin).
    let rows: Vec<(i32, String, String, String, Option<i32>, String)> = sqlx::query_as(
        r#"SELECT p.id, p.name, p.source_path, p.source_kind, p.client_id, COALESCE(c.name,'')
           FROM projects p LEFT JOIN clients c ON p.client_id=c.id
           WHERE $1 OR EXISTS (SELECT 1 FROM project_members m WHERE m.project_id=p.id AND m.username=$2)
           ORDER BY c.name NULLS FIRST, p.name"#,
    )
    .bind(user.is_admin())
    .bind(&user.username)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    let mut out = Vec::new();
    for (id, name, source_path, source_kind, client_id, client_name) in rows {
        let (total, pending, annotated, verified) = counts(&s.db, id).await.map_err(err)?;
        out.push(ProjectSummary {
            id, name, source_path, source_kind, client_id, client_name,
            total, pending, annotated, verified,
        });
    }
    Ok(Json(out))
}

/// `GET /api/clients` — list clients with their project counts (auth required).
async fn list_clients(State(s): State<AppState>, _user: AuthUser) -> ApiResult<Vec<ClientRow>> {
    let rows: Vec<(i32, String, chrono::DateTime<chrono::Utc>, i64)> = sqlx::query_as(
        r#"SELECT c.id, c.name, c.created_at, COUNT(p.id)
           FROM clients c LEFT JOIN projects p ON p.client_id=c.id
           GROUP BY c.id, c.name, c.created_at
           ORDER BY c.name"#,
    )
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    Ok(Json(
        rows.into_iter()
            .map(|(id, name, created_at, project_count)| ClientRow { id, name, created_at, project_count })
            .collect(),
    ))
}

async fn trigger_ingest(
    State(s): State<AppState>,
    user: AuthUser,
    AxPath(id): AxPath<i32>,
) -> ApiResult<serde_json::Value> {
    // Resolve the project's source folder (mounting an NFS share if `source_kind='nfs'`).
    let row: Option<(String, String)> =
        sqlx::query_as("SELECT source_path, source_kind FROM projects WHERE id=$1")
            .bind(id)
            .fetch_optional(&s.db)
            .await
            .map_err(err)?;
    let (source_path, source_kind) = row.unwrap_or_else(|| {
        (s.ingest_dir.to_string_lossy().to_string(), "local".to_string())
    });

    let (sp, nfs_host, nfs_opts) = (source_path.clone(), s.nfs_host.clone(), s.nfs_opts.clone());
    let dir = tokio::task::spawn_blocking(move || nfs::resolve_dir(&sp, &source_kind, &nfs_host, &nfs_opts))
        .await
        .map_err(err)?
        .map_err(|e| (StatusCode::BAD_REQUEST, e))?;

    let n = ingest::ingest_project(&s.db, id, &dir)
        .await
        .map_err(|e| (StatusCode::BAD_REQUEST, e.to_string()))?;

    admin::audit(&s.db, &user.username, "synced", serde_json::json!({ "project_id": id, "ingested": n })).await;
    // Report the export path, never the resolved local mountpoint / host.
    Ok(Json(serde_json::json!({ "ingested": n, "source_path": source_path })))
}

#[derive(serde::Deserialize)]
struct AuditQuery {
    limit: Option<i64>,
}

/// Query-param auth for file downloads (`<a download>` can't send an auth header).
#[derive(serde::Deserialize)]
struct TokenQuery {
    token: Option<String>,
}

/// `GET /api/audit?limit=100` — recent org-level audit events (auth required).
async fn list_audit(State(s): State<AppState>, _user: AuthUser, Query(q): Query<AuditQuery>) -> ApiResult<Vec<AuditRow>> {
    let limit = q.limit.unwrap_or(100).clamp(1, 1000);
    let rows = sqlx::query_as::<_, AuditRow>(
        "SELECT at, username, action, detail FROM audit_events ORDER BY at DESC LIMIT $1",
    )
    .bind(limit)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    Ok(Json(rows))
}

async fn list_videos(State(s): State<AppState>, user: AuthUser, AxPath(id): AxPath<i32>) -> ApiResult<Vec<VideoRow>> {
    require_project_access(&s, &user, id).await?;
    let q = format!(r#"SELECT id, file_name, rel_path, has_json, width, height, fps, frame_count,
                  annotation_count, imported_count, annotation_time_ms, primary_annotator, status, annotated_at,
                  claimed_by, claimed_at, lease_expires_at, completed_by, completed_at, verify_time_ms,
                  -- annotations flagged for review (review_status='flagged') still open on this video
                  COALESCE((SELECT count(*) FROM annotations a WHERE a.video_id = videos.id AND a.review_status='flagged'), 0) AS review_count,
                  -- pass count + distinct verifiers (in first-push order), derived from the push log
                  COALESCE((SELECT count(*) FROM video_events e WHERE e.video_id = videos.id AND e.event='push'), 0) AS verify_count,
                  COALESCE((SELECT string_agg(s.u, ', ' ORDER BY s.first_at)
                            FROM (SELECT username AS u, min(at) AS first_at FROM video_events
                                  WHERE video_id = videos.id AND event='push' GROUP BY username) s), '') AS verifiers,
                  assigned_to,
                  -- derived lifecycle label for the UI badge
                  CASE
                    WHEN videos.claimed_by IS NOT NULL AND videos.lease_expires_at > now() THEN 'in-progress'
                    WHEN videos.status = 'verified' THEN
                      CASE WHEN (SELECT count(*) FROM video_events e WHERE e.video_id = videos.id AND e.event='push') >= 2
                           THEN 're-verified' ELSE 'verified' END
                    WHEN videos.assigned_to IS NOT NULL AND videos.assigned_to <> '' THEN 'assigned'
                    ELSE videos.status
                  END AS workflow_status,
                  hand_raised, hand_raised_by, hand_raised_at, ignored, ignored_by,
                  COALESCE((SELECT json_agg(json_build_object('username', r.username, 'body', r.body, 'created_at', r.created_at)
                                            ORDER BY r.created_at)
                            FROM video_remarks r WHERE r.video_id = videos.id), '[]'::json) AS remarks
           FROM videos WHERE project_id=$1 {FOLDER_FILTER} ORDER BY file_name"#);
    let rows = sqlx::query_as::<_, VideoRow>(&q)
    .bind(id)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    Ok(Json(rows))
}

/// `GET /api/videos/:id/export` — download one video's annotations as the canonical
/// export JSON (`videos.raw_json`, the doc pushed by the worker / ingested from NAS).
/// Auth required (via `?token=`); served as a file attachment.
async fn export_video(State(s): State<AppState>, Query(q): Query<TokenQuery>, AxPath(id): AxPath<i32>) -> Result<Response, (StatusCode, String)> {
    if auth::validate_token(&s, q.token.as_deref().unwrap_or("")).await.is_none() {
        return Err((StatusCode::UNAUTHORIZED, "missing or invalid token".into()));
    }
    let row: Option<(String, Option<serde_json::Value>)> =
        sqlx::query_as("SELECT file_name, raw_json FROM videos WHERE id=$1")
            .bind(id)
            .fetch_optional(&s.db)
            .await
            .map_err(err)?;
    let (file_name, raw_json) = row.ok_or((StatusCode::NOT_FOUND, "video not found".to_string()))?;
    let doc = raw_json.unwrap_or_else(|| json!({ "fixed_annotations": [], "range_annotations": [] }));
    let body = serde_json::to_vec_pretty(&doc).map_err(err)?;
    let stem = file_name.rsplit_once('.').map(|(a, _)| a).unwrap_or(&file_name);
    let fname = format!("{stem}_annotations.json");
    Response::builder()
        .header(header::CONTENT_TYPE, "application/json")
        .header(header::CONTENT_DISPOSITION, format!("attachment; filename=\"{fname}\""))
        .body(Body::from(body))
        .map_err(err)
}

/// `GET /api/projects/:id/export` — download ALL annotated videos in a project as one
/// JSON ({project_id, video_count, videos:[{video_id, file_name, rel_path, annotations}]}).
async fn export_project(State(s): State<AppState>, Query(q): Query<TokenQuery>, AxPath(id): AxPath<i32>) -> Result<Response, (StatusCode, String)> {
    if auth::validate_token(&s, q.token.as_deref().unwrap_or("")).await.is_none() {
        return Err((StatusCode::UNAUTHORIZED, "missing or invalid token".into()));
    }
    let rows: Vec<(i32, String, String, Option<serde_json::Value>)> = sqlx::query_as(
        "SELECT id, file_name, rel_path, raw_json FROM videos WHERE project_id=$1 AND raw_json IS NOT NULL ORDER BY rel_path",
    )
    .bind(id)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    let videos: Vec<serde_json::Value> = rows
        .into_iter()
        .map(|(vid, fname, rel, raw)| json!({ "video_id": vid, "file_name": fname, "rel_path": rel, "annotations": raw }))
        .collect();
    let doc = json!({ "project_id": id, "video_count": videos.len(), "videos": videos });
    let body = serde_json::to_vec_pretty(&doc).map_err(err)?;
    Response::builder()
        .header(header::CONTENT_TYPE, "application/json")
        .header(header::CONTENT_DISPOSITION, format!("attachment; filename=\"project_{id}_annotations.json\""))
        .body(Body::from(body))
        .map_err(err)
}

/// `POST /api/projects/:id/assign {video_ids, assignee}` — assign (or unassign with an
/// empty assignee) a set of videos to a user. Any authenticated user (admins/collaborators).
async fn assign_videos(
    State(s): State<AppState>,
    user: AuthUser,
    AxPath(id): AxPath<i32>,
    Json(req): Json<AssignRequest>,
) -> ApiResult<serde_json::Value> {
    if req.video_ids.is_empty() {
        return Err((StatusCode::BAD_REQUEST, "no videos selected".into()));
    }
    let assignee = req.assignee.trim().to_string();
    if !assignee.is_empty() {
        // The assignee must be a picked member of this project.
        let member: Option<String> = sqlx::query_scalar(
            "SELECT username FROM project_members WHERE project_id=$1 AND username=$2",
        )
        .bind(id)
        .bind(&assignee)
        .fetch_optional(&s.db)
        .await
        .map_err(err)?;
        if member.is_none() {
            return Err((StatusCode::BAD_REQUEST, format!("'{assignee}' is not a member of this project")));
        }
    }
    let value: Option<String> = if assignee.is_empty() { None } else { Some(assignee.clone()) };
    let n = sqlx::query("UPDATE videos SET assigned_to=$3 WHERE project_id=$1 AND id = ANY($2)")
        .bind(id)
        .bind(&req.video_ids)
        .bind(&value)
        .execute(&s.db)
        .await
        .map_err(err)?
        .rows_affected();
    admin::audit(&s.db, &user.username, "videos_assigned",
        json!({ "project_id": id, "assignee": assignee, "count": n })).await;
    Ok(Json(json!({ "assigned": n, "assignee": assignee })))
}

/// `GET /api/videos/:id/remarks` — the full thread (auth required, like the video list).
async fn list_remarks(State(s): State<AppState>, _user: AuthUser, AxPath(id): AxPath<i32>) -> ApiResult<Vec<Remark>> {
    let rows = sqlx::query_as::<_, Remark>(
        "SELECT username, body, created_at FROM video_remarks WHERE video_id=$1 ORDER BY created_at",
    )
    .bind(id)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    Ok(Json(rows))
}

/// `POST /api/videos/:id/remarks {body}` — append a remark. Any authenticated user
/// (admin or worker); attributed to their username. Returns the updated thread.
async fn add_remark(
    State(s): State<AppState>,
    user: AuthUser,
    AxPath(id): AxPath<i32>,
    Json(req): Json<NewRemarkRequest>,
) -> ApiResult<Vec<Remark>> {
    let body = req.body.trim();
    if body.is_empty() {
        return Err((StatusCode::BAD_REQUEST, "remark is required".into()));
    }
    let exists: Option<i32> = sqlx::query_scalar("SELECT id FROM videos WHERE id=$1")
        .bind(id)
        .fetch_optional(&s.db)
        .await
        .map_err(err)?;
    if exists.is_none() {
        return Err((StatusCode::NOT_FOUND, "video not found".into()));
    }
    sqlx::query("INSERT INTO video_remarks (video_id, username, body) VALUES ($1,$2,$3)")
        .bind(id)
        .bind(&user.username)
        .bind(body)
        .execute(&s.db)
        .await
        .map_err(err)?;
    let rows = sqlx::query_as::<_, Remark>(
        "SELECT username, body, created_at FROM video_remarks WHERE video_id=$1 ORDER BY created_at",
    )
    .bind(id)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    Ok(Json(rows))
}

/// `POST /api/videos/:id/hand {raised}` — raise or lower the "hand" on a video.
/// Any authenticated user (ml_support or admin) can raise (to start a conversation)
/// or lower it. Raising records who/when; lowering clears that.
async fn set_hand(
    State(s): State<AppState>,
    user: AuthUser,
    AxPath(id): AxPath<i32>,
    Json(req): Json<HandRequest>,
) -> ApiResult<serde_json::Value> {
    let updated: Option<i32> = if req.raised {
        sqlx::query_scalar(
            "UPDATE videos SET hand_raised=TRUE, hand_raised_by=$2, hand_raised_at=now() WHERE id=$1 RETURNING id",
        )
        .bind(id)
        .bind(&user.username)
        .fetch_optional(&s.db)
        .await
        .map_err(err)?
    } else {
        sqlx::query_scalar(
            "UPDATE videos SET hand_raised=FALSE, hand_raised_by=NULL, hand_raised_at=NULL WHERE id=$1 RETURNING id",
        )
        .bind(id)
        .fetch_optional(&s.db)
        .await
        .map_err(err)?
    };
    if updated.is_none() {
        return Err((StatusCode::NOT_FOUND, "video not found".into()));
    }
    worker::log_event(&s.db, id, &user.username,
        if req.raised { "hand_raised" } else { "hand_lowered" }, json!({})).await;
    Ok(Json(json!({ "hand_raised": req.raised, "by": if req.raised { user.username } else { String::new() } })))
}

/// `POST /api/videos/:id/ignore {ignored}` — mark a video ignored / un-ignored.
/// **Admin only.** Ignored videos are crossed out in the UI (excluded from the
/// active workflow); the flag is orthogonal to status/claim.
async fn set_ignore(
    State(s): State<AppState>,
    user: AuthUser,
    AxPath(id): AxPath<i32>,
    Json(req): Json<IgnoreRequest>,
) -> ApiResult<serde_json::Value> {
    user.require_admin()?;
    let by: Option<String> = if req.ignored { Some(user.username.clone()) } else { None };
    let updated: Option<i32> = sqlx::query_scalar(
        "UPDATE videos SET ignored=$2, ignored_by=$3 WHERE id=$1 RETURNING id",
    )
    .bind(id)
    .bind(req.ignored)
    .bind(&by)
    .fetch_optional(&s.db)
    .await
    .map_err(err)?;
    if updated.is_none() {
        return Err((StatusCode::NOT_FOUND, "video not found".into()));
    }
    worker::log_event(&s.db, id, &user.username,
        if req.ignored { "ignored" } else { "unignored" }, json!({})).await;
    Ok(Json(json!({ "ignored": req.ignored })))
}

async fn counts(db: &PgPool, id: i32) -> Result<(i64, i64, i64, i64), sqlx::Error> {
    let q = format!(
        r#"SELECT COUNT(*),
                  COUNT(*) FILTER (WHERE status='pending'),
                  COUNT(*) FILTER (WHERE status='annotated'),
                  COUNT(*) FILTER (WHERE status='verified')
           FROM videos WHERE project_id=$1 {FOLDER_FILTER}"#,
    );
    sqlx::query_as(&q).bind(id).fetch_one(db).await
}

async fn stats(State(s): State<AppState>, user: AuthUser, AxPath(id): AxPath<i32>) -> ApiResult<Stats> {
    require_project_access(&s, &user, id).await?;
    let (total, pending, annotated, verified) = counts(&s.db, id).await.map_err(err)?;
    let pct_done = if total > 0 { verified as f64 / total as f64 * 100.0 } else { 0.0 };

    // Annotation leaderboard: credited per ANNOTATION author (annotated_by), so a
    // verifier is credited only for annotations they actually DREW — imported
    // annotations keep their original author and credit them, not the verifier.
    // videos = distinct videos they drew in; time = annotation_time_ms of videos
    // where they're the primary annotator.
    let lead_rows: Vec<(String, i64, i64, i64)> = sqlx::query_as(
        r#"SELECT a.annotated_by,
                  COUNT(*)::bigint,
                  COUNT(DISTINCT a.video_id)::bigint,
                  COALESCE((SELECT SUM(v2.annotation_time_ms) FROM videos v2
                            WHERE v2.project_id=$1 AND v2.primary_annotator = a.annotated_by), 0)::bigint
           FROM annotations a JOIN videos v ON a.video_id=v.id
           WHERE v.project_id=$1 AND a.annotated_by <> '' AND lower(a.annotated_by) <> 'auto'
           GROUP BY a.annotated_by"#,
    )
    .bind(id)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    let mut leaderboard: Vec<LeaderRow> = lead_rows
        .into_iter()
        .map(|(user, annotations, videos, total_time_ms)| LeaderRow {
            avg_time_ms: if videos > 0 { total_time_ms / videos } else { 0 },
            annotations,
            videos,
            total_time_ms,
            user,
        })
        .collect();
    leaderboard.sort_by(|a, b| b.annotations.cmp(&a.annotations));
    // Annotations authored per user (for the verification leaderboard's "Annotations" col).
    let ann_by_user: std::collections::HashMap<String, i64> =
        leaderboard.iter().map(|r| (r.user.clone(), r.annotations)).collect();

    // Verification leaderboard: ALL assigned users (members) ranked by ANNOTATION
    // THROUGHPUT — annotations added per minute of work (most annotations in the least
    // time first), not by video count. Anyone who has completed videos but isn't a
    // member is included too; users with no annotations/time sort last (unranked).
    // The synthetic 'auto' author (auto-segmented/imported) is excluded. Refreshed
    // every poll → "live" ranking.
    let ver_rows: Vec<(String, i64, i64)> = sqlx::query_as(
        r#"SELECT completed_by, COUNT(*), COALESCE(SUM(verify_time_ms),0)::bigint
           FROM videos
           WHERE project_id=$1 AND status='verified' AND completed_by IS NOT NULL
                 AND completed_by <> '' AND lower(completed_by) <> 'auto'
           GROUP BY completed_by"#,
    )
    .bind(id)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;
    let members: Vec<String> = sqlx::query_scalar("SELECT username FROM project_members WHERE project_id=$1")
        .bind(id)
        .fetch_all(&s.db)
        .await
        .map_err(err)?;
    let mut stats_by_user: std::collections::HashMap<String, (i64, i64)> =
        ver_rows.into_iter().map(|(u, v, t)| (u, (v, t))).collect();
    for m in &members {
        if m.eq_ignore_ascii_case("auto") {
            continue;
        }
        stats_by_user.entry(m.clone()).or_insert((0, 0));
    }
    let mut verifiers: Vec<VerifierRow> = stats_by_user
        .into_iter()
        .map(|(user, (videos, total_time_ms))| {
            let annotations = ann_by_user.get(&user).copied().unwrap_or(0);
            // annotations added per minute of work; 0 when no time recorded yet.
            let annotations_per_min = if total_time_ms > 0 {
                annotations as f64 / (total_time_ms as f64 / 60_000.0)
            } else {
                0.0
            };
            VerifierRow {
                avg_time_ms: if videos > 0 { total_time_ms / videos } else { 0 },
                videos,
                total_time_ms,
                annotations,
                annotations_per_min,
                rank: 0,
                user,
            }
        })
        .collect();
    // Highest throughput first (most annotations per minute). A user is rankable only
    // once they have both annotations AND recorded time; everyone else sorts last by name.
    let rankable = |v: &VerifierRow| v.annotations > 0 && v.annotations_per_min > 0.0;
    verifiers.sort_by(|a, b| match (rankable(a), rankable(b)) {
        (true, false) => std::cmp::Ordering::Less,
        (false, true) => std::cmp::Ordering::Greater,
        (true, true) => b
            .annotations_per_min
            .partial_cmp(&a.annotations_per_min)
            .unwrap_or(std::cmp::Ordering::Equal)
            .then_with(|| a.user.cmp(&b.user)),
        (false, false) => a.user.cmp(&b.user),
    });
    let mut rank = 0i64;
    for v in verifiers.iter_mut() {
        if rankable(v) {
            rank += 1;
            v.rank = rank;
        }
    }

    // Verification pace: push (verify) events in the recent window (folder-aware).
    let recent_q = format!(
        r#"SELECT COUNT(*)::bigint FROM video_events e JOIN videos v ON e.video_id=v.id
           WHERE v.project_id=$1 AND e.event='push' AND e.at > now() - ($2::int * interval '1 day')
           {FOLDER_FILTER}"#,
    );
    let recent_verified: i64 = sqlx::query_scalar(&recent_q)
        .bind(id)
        .bind(RECENT_WINDOW_DAYS as i32)
        .fetch_one(&s.db)
        .await
        .map_err(err)?;

    // Project time metrics: aggregate effort + avg per verified video (over the
    // considered folders), the admin-controlled timer's elapsed, and a worker-
    // parallelised ETA for the remaining (un-verified) videos.
    let active_workers: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM project_members WHERE project_id=$1")
        .bind(id)
        .fetch_one(&s.db)
        .await
        .map_err(err)?;

    let agg_q = format!(
        r#"SELECT COALESCE(SUM(annotation_time_ms + verify_time_ms), 0)::bigint,
                  COALESCE(AVG(annotation_time_ms + verify_time_ms) FILTER (WHERE status='verified'), 0)::bigint
           FROM videos WHERE project_id=$1 {FOLDER_FILTER}"#,
    );
    let (total_spent_ms, avg_video_ms): (i64, i64) =
        sqlx::query_as(&agg_q).bind(id).fetch_one(&s.db).await.map_err(err)?;

    // Project timer: elapsed = accumulated + (running ? now - started_at : 0).
    let (time_running, t_acc, t_started, last_activity_at): (
        bool, i64,
        Option<chrono::DateTime<chrono::Utc>>,
        Option<chrono::DateTime<chrono::Utc>>,
    ) = sqlx::query_as(
        "SELECT time_running, time_accumulated_ms, time_started_at, last_activity_at FROM projects WHERE id=$1",
    )
    .bind(id)
    .fetch_one(&s.db)
    .await
    .map_err(err)?;
    let live = if time_running {
        t_started.map(|st| (chrono::Utc::now() - st).num_milliseconds().max(0)).unwrap_or(0)
    } else {
        0
    };
    let elapsed_ms = t_acc + live;

    let all_verified = total > 0 && verified == total;
    // Remaining work spread across the picked workers (≥1). Needs at least one
    // verified video to have an average to extrapolate from.
    let remaining = (total - verified).max(0);
    let eta_ms = if avg_video_ms > 0 && remaining > 0 {
        Some(remaining * avg_video_ms / active_workers.max(1))
    } else {
        None
    };

    Ok(Json(Stats {
        overview: Overview { total, pending, annotated, verified, pct_done },
        leaderboard,
        verifiers,
        timeline: ProjectTime {
            active_workers,
            avg_video_ms,
            total_spent_ms,
            elapsed_ms,
            all_verified,
            eta_ms,
            time_running,
            last_activity_at,
            recent_verified,
            recent_window_days: RECENT_WINDOW_DAYS,
        },
    }))
}

/// `GET /api/projects/:id/activity` — live claims + recent events for the dashboard.
async fn activity(State(s): State<AppState>, user: AuthUser, AxPath(id): AxPath<i32>) -> ApiResult<Activity> {
    require_project_access(&s, &user, id).await?;
    let active_claims = sqlx::query_as::<_, ActiveClaim>(
        r#"SELECT v.id AS video_id, v.file_name, v.claimed_by, v.claimed_at, v.lease_expires_at
           FROM videos v
           WHERE v.project_id=$1 AND v.claimed_by IS NOT NULL AND v.lease_expires_at > now()
           ORDER BY v.claimed_at DESC"#,
    )
    .bind(id)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;

    let recent_events = sqlx::query_as::<_, EventRow>(
        r#"SELECT e.video_id, v.file_name, e.username, e.event, e.at
           FROM video_events e JOIN videos v ON e.video_id=v.id
           WHERE v.project_id=$1
           ORDER BY e.at DESC LIMIT 50"#,
    )
    .bind(id)
    .fetch_all(&s.db)
    .await
    .map_err(err)?;

    Ok(Json(Activity { active_claims, recent_events }))
}