Central Verification Dashboard

server/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "central-server"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+axum = "0.7"
+tokio = { version = "1", features = ["full"] }
+tower-http = { version = "0.5", features = ["cors", "trace"] }
+sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "chrono", "json"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+chrono = { version = "0.4", features = ["serde"] }
+walkdir = "2"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+anyhow = "1"
+# Phase 3: token hashing (sha256) + random token generation + file-stream download.
+sha2 = "0.10"
+rand = "0.8"
+hex = "0.4"
+tokio-util = { version = "0.7", features = ["io"] }

server/Dockerfile

@@ -0,0 +1,18 @@
+# Build context is the `central/` dir (see docker-compose) so we can copy both
+# server/ and db/ — the server embeds db/init.sql via include_str!("../../db/init.sql").
+FROM rust:1-bookworm AS build
+WORKDIR /build
+COPY server/Cargo.toml server/Cargo.toml
+COPY server/src server/src
+COPY db db
+WORKDIR /build/server
+RUN cargo build --release
+
+FROM debian:bookworm-slim
+# nfs-common provides mount.nfs for `nfs`-kind projects (mounted on demand at sync/download time).
+RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates nfs-common \
+    && rm -rf /var/lib/apt/lists/*
+COPY --from=build /build/server/target/release/central-server /usr/local/bin/central-server
+ENV BIND_ADDR=0.0.0.0:8080
+EXPOSE 8080
+CMD ["central-server"]

server/src/admin.rs

@@ -0,0 +1,435 @@
+//! Phase 3 admin API — provisioning. Gated by `AuthUser::require_admin()`, which is
+//! satisfied by a user with role 'admin' or by the master `ADMIN_TOKEN`.
+
+use crate::auth::{gen_token, sha256_hex, AuthUser};
+use crate::models::*;
+use crate::{err, AppState, ApiResult};
+use axum::{
+    extract::{Path as AxPath, State},
+    http::StatusCode,
+    Json,
+};
+use serde_json::{json, Value};
+
+fn is_unique_violation(e: &sqlx::Error) -> bool {
+    e.as_database_error().map(|d| d.is_unique_violation()).unwrap_or(false)
+}
+
+/// Record an org-level audit event (best-effort: log on failure, never block the action).
+pub async fn audit(db: &sqlx::PgPool, username: &str, action: &str, detail: Value) {
+    if let Err(e) =
+        sqlx::query("INSERT INTO audit_events (username, action, detail) VALUES ($1,$2,$3)")
+            .bind(username)
+            .bind(action)
+            .bind(detail)
+            .execute(db)
+            .await
+    {
+        tracing::warn!("failed to record audit '{action}' by '{username}': {e}");
+    }
+}
+
+/// `POST /api/users` — provision a worker/admin. Generates the token, stores only its
+/// hash, and returns the clear token **once** (it can never be retrieved again).
+pub async fn create_user(
+    State(s): State<AppState>,
+    user: AuthUser,
+    Json(req): Json<NewUserRequest>,
+) -> ApiResult<NewUserResponse> {
+    user.require_admin()?;
+    if req.username.trim().is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "username is required".into()));
+    }
+    // Only two roles exist: 'admin' and 'ml_support' (anything else normalizes to ml_support).
+    let role = if req.role == "admin" { "admin" } else { "ml_support" };
+    let display_name = if req.display_name.trim().is_empty() {
+        req.username.clone()
+    } else {
+        req.display_name.clone()
+    };
+    let token = gen_token();
+    let token_hash = sha256_hex(&token);
+
+    let res = sqlx::query(
+        "INSERT INTO users (username, display_name, token_hash, role) VALUES ($1,$2,$3,$4)",
+    )
+    .bind(&req.username)
+    .bind(&display_name)
+    .bind(&token_hash)
+    .bind(role)
+    .execute(&s.db)
+    .await;
+
+    match res {
+        Ok(_) => {
+            audit(&s.db, &user.username, "user_created",
+                json!({ "username": req.username, "role": role })).await;
+            Ok(Json(NewUserResponse {
+                username: req.username,
+                display_name,
+                role: role.into(),
+                token,
+            }))
+        }
+        Err(e) if is_unique_violation(&e) => {
+            Err((StatusCode::CONFLICT, "username already exists".into()))
+        }
+        Err(e) => Err(err(e)),
+    }
+}
+
+/// `POST /api/users/:id/token` — regenerate a user's access token (admin only), e.g.
+/// when they forget it. The old token stops working immediately; the new clear token
+/// is returned **once**. The master `ADMIN_TOKEN` env is unaffected (it's not a row).
+pub async fn rotate_token(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+) -> ApiResult<NewUserResponse> {
+    user.require_admin()?;
+    let token = gen_token();
+    let token_hash = sha256_hex(&token);
+    let row: Option<(String, String, String)> = sqlx::query_as(
+        "UPDATE users SET token_hash=$2 WHERE id=$1 RETURNING username, display_name, role",
+    )
+    .bind(id)
+    .bind(&token_hash)
+    .fetch_optional(&s.db)
+    .await
+    .map_err(err)?;
+    let Some((username, display_name, role)) = row else {
+        return Err((StatusCode::NOT_FOUND, "user not found".into()));
+    };
+    audit(&s.db, &user.username, "user_token_reset",
+        json!({ "user_id": id, "username": username })).await;
+    Ok(Json(NewUserResponse { username, display_name, role, token }))
+}
+
+/// `POST /api/users/:id/update {display_name, role}` — edit a user's display name +
+/// role (admin only). Empty display_name falls back to the username. Tokens are not
+/// affected (they're hashed; use rotate_token to issue a new one).
+pub async fn update_user(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+    Json(req): Json<UpdateUserRequest>,
+) -> ApiResult<Value> {
+    user.require_admin()?;
+    let username: Option<String> = sqlx::query_scalar("SELECT username FROM users WHERE id=$1")
+        .bind(id)
+        .fetch_optional(&s.db)
+        .await
+        .map_err(err)?;
+    let Some(username) = username else {
+        return Err((StatusCode::NOT_FOUND, "user not found".into()));
+    };
+    let role = if req.role == "admin" { "admin" } else { "ml_support" };
+    let display_name = if req.display_name.trim().is_empty() {
+        username.clone()
+    } else {
+        req.display_name.trim().to_string()
+    };
+    sqlx::query("UPDATE users SET display_name=$2, role=$3 WHERE id=$1")
+        .bind(id)
+        .bind(&display_name)
+        .bind(role)
+        .execute(&s.db)
+        .await
+        .map_err(err)?;
+    audit(&s.db, &user.username, "user_updated",
+        json!({ "user_id": id, "username": username, "display_name": display_name, "role": role })).await;
+    Ok(Json(json!({ "updated": true, "display_name": display_name, "role": role })))
+}
+
+/// `GET /api/users` — list provisioned users (never exposes tokens).
+pub async fn list_users(State(s): State<AppState>, user: AuthUser) -> ApiResult<Vec<UserRow>> {
+    user.require_admin()?;
+    let rows = sqlx::query_as::<_, UserRow>(
+        "SELECT id, username, display_name, role, active, created_at FROM users ORDER BY id",
+    )
+    .fetch_all(&s.db)
+    .await
+    .map_err(err)?;
+    Ok(Json(rows))
+}
+
+/// `POST /api/clients` — create a client. Unique on name.
+pub async fn create_client(
+    State(s): State<AppState>,
+    user: AuthUser,
+    Json(req): Json<NewClientRequest>,
+) -> ApiResult<ClientRow> {
+    user.require_admin()?;
+    if req.name.trim().is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "client name is required".into()));
+    }
+    let res: Result<(i32, chrono::DateTime<chrono::Utc>), sqlx::Error> = sqlx::query_as(
+        "INSERT INTO clients (name) VALUES ($1) RETURNING id, created_at",
+    )
+    .bind(&req.name)
+    .fetch_one(&s.db)
+    .await;
+
+    match res {
+        Ok((id, created_at)) => {
+            audit(&s.db, &user.username, "client_created", json!({ "client_id": id, "name": req.name })).await;
+            Ok(Json(ClientRow {
+                id,
+                name: req.name,
+                created_at,
+                project_count: 0,
+            }))
+        }
+        Err(e) if is_unique_violation(&e) => {
+            Err((StatusCode::CONFLICT, "client name already exists".into()))
+        }
+        Err(e) => Err(err(e)),
+    }
+}
+
+fn norm_kind(k: &str) -> &'static str {
+    if k == "nfs" { "nfs" } else { "local" }
+}
+
+/// `POST /api/projects` — create (or repoint) a project under a client. Allowed for **any
+/// authenticated user** (collaborators add projects); idempotent on `(client_id, name)`.
+pub async fn create_project(
+    State(s): State<AppState>,
+    user: AuthUser,
+    Json(req): Json<NewProjectRequest>,
+) -> ApiResult<Value> {
+    if req.name.trim().is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "project name is required".into()));
+    }
+    if req.source_path.trim().is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "source path is required".into()));
+    }
+    let kind = norm_kind(&req.source_kind);
+    let client_exists: Option<i32> = sqlx::query_scalar("SELECT id FROM clients WHERE id=$1")
+        .bind(req.client_id)
+        .fetch_optional(&s.db)
+        .await
+        .map_err(err)?;
+    if client_exists.is_none() {
+        return Err((StatusCode::NOT_FOUND, "client not found".into()));
+    }
+    let id: i32 = sqlx::query_scalar(
+        r#"INSERT INTO projects (name, source_path, source_kind, client_id) VALUES ($1,$2,$3,$4)
+           ON CONFLICT (client_id, name)
+           DO UPDATE SET source_path=EXCLUDED.source_path, source_kind=EXCLUDED.source_kind
+           RETURNING id"#,
+    )
+    .bind(&req.name)
+    .bind(&req.source_path)
+    .bind(kind)
+    .bind(req.client_id)
+    .fetch_one(&s.db)
+    .await
+    .map_err(err)?;
+    audit(&s.db, &user.username, "project_created",
+        json!({ "project_id": id, "client_id": req.client_id, "name": req.name,
+                "source_path": req.source_path, "source_kind": kind })).await;
+    Ok(Json(json!({
+        "id": id, "client_id": req.client_id, "name": req.name,
+        "source_path": req.source_path, "source_kind": kind
+    })))
+}
+
+/// `POST /api/projects/:id/source` — change a project's sync directory (any authenticated user).
+pub async fn update_project_source(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+    Json(req): Json<UpdateSourceRequest>,
+) -> ApiResult<Value> {
+    if req.source_path.trim().is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "source path is required".into()));
+    }
+    let kind = norm_kind(&req.source_kind);
+    let updated: Option<i32> = sqlx::query_scalar(
+        "UPDATE projects SET source_path=$2, source_kind=$3 WHERE id=$1 RETURNING id",
+    )
+    .bind(id)
+    .bind(&req.source_path)
+    .bind(kind)
+    .fetch_optional(&s.db)
+    .await
+    .map_err(err)?;
+    if updated.is_none() {
+        return Err((StatusCode::NOT_FOUND, "project not found".into()));
+    }
+    audit(&s.db, &user.username, "project_source_changed",
+        json!({ "project_id": id, "source_path": req.source_path, "source_kind": kind })).await;
+    Ok(Json(json!({ "id": id, "source_path": req.source_path, "source_kind": kind })))
+}
+
+/// `DELETE /api/projects/:id` — delete a project and everything under it (videos,
+/// annotations, events, remarks all cascade). Admin only.
+pub async fn delete_project(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+) -> ApiResult<Value> {
+    user.require_admin()?;
+    let name: Option<String> = sqlx::query_scalar("SELECT name FROM projects WHERE id=$1")
+        .bind(id)
+        .fetch_optional(&s.db)
+        .await
+        .map_err(err)?;
+    let Some(name) = name else {
+        return Err((StatusCode::NOT_FOUND, "project not found".into()));
+    };
+    sqlx::query("DELETE FROM projects WHERE id=$1")
+        .bind(id)
+        .execute(&s.db)
+        .await
+        .map_err(err)?;
+    audit(&s.db, &user.username, "project_deleted", json!({ "project_id": id, "name": name })).await;
+    Ok(Json(json!({ "deleted": true, "id": id })))
+}
+
+/// `DELETE /api/clients/:id` — delete a client and ALL its projects + their data
+/// (cascades through projects → videos → annotations). Admin only. The dashboard
+/// re-affirms (type-the-name) before calling this.
+pub async fn delete_client(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+) -> ApiResult<Value> {
+    user.require_admin()?;
+    let name: Option<String> = sqlx::query_scalar("SELECT name FROM clients WHERE id=$1")
+        .bind(id)
+        .fetch_optional(&s.db)
+        .await
+        .map_err(err)?;
+    let Some(name) = name else {
+        return Err((StatusCode::NOT_FOUND, "client not found".into()));
+    };
+    let project_count: i64 =
+        sqlx::query_scalar("SELECT COUNT(*) FROM projects WHERE client_id=$1")
+            .bind(id)
+            .fetch_one(&s.db)
+            .await
+            .map_err(err)?;
+    sqlx::query("DELETE FROM clients WHERE id=$1")
+        .bind(id)
+        .execute(&s.db)
+        .await
+        .map_err(err)?;
+    audit(&s.db, &user.username, "client_deleted",
+        json!({ "client_id": id, "name": name, "projects_removed": project_count })).await;
+    Ok(Json(json!({ "deleted": true, "id": id, "projects_removed": project_count })))
+}
+
+/// `GET /api/projects/:id/members` — users picked for a project. Admin or member.
+pub async fn list_members(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+) -> ApiResult<Vec<MemberRow>> {
+    crate::require_project_access(&s, &user, id).await?;
+    let rows = sqlx::query_as::<_, MemberRow>(
+        r#"SELECT m.username,
+                  COALESCE(u.display_name, m.username) AS display_name,
+                  COALESCE(u.role, '') AS role,
+                  m.added_at
+           FROM project_members m
+           LEFT JOIN users u ON u.username = m.username
+           WHERE m.project_id=$1
+           ORDER BY m.username"#,
+    )
+    .bind(id)
+    .fetch_all(&s.db)
+    .await
+    .map_err(err)?;
+    Ok(Json(rows))
+}
+
+/// `POST /api/projects/:id/members {username}` — pick a user for a project. Admin only.
+pub async fn add_member(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+    Json(req): Json<AddMemberRequest>,
+) -> ApiResult<Value> {
+    user.require_admin()?;
+    let username = req.username.trim().to_string();
+    if username.is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "username is required".into()));
+    }
+    let exists: Option<i32> = sqlx::query_scalar("SELECT id FROM users WHERE username=$1 AND active")
+        .bind(&username)
+        .fetch_optional(&s.db)
+        .await
+        .map_err(err)?;
+    if exists.is_none() {
+        return Err((StatusCode::BAD_REQUEST, format!("unknown user '{username}'")));
+    }
+    sqlx::query(
+        "INSERT INTO project_members (project_id, username) VALUES ($1,$2) ON CONFLICT DO NOTHING",
+    )
+    .bind(id)
+    .bind(&username)
+    .execute(&s.db)
+    .await
+    .map_err(err)?;
+    audit(&s.db, &user.username, "project_member_added",
+        json!({ "project_id": id, "username": username })).await;
+    Ok(Json(json!({ "added": true, "username": username })))
+}
+
+/// `DELETE /api/projects/:id/members/:username` — remove a user from a project (admin
+/// only). Also clears any of that user's video assignments within the project.
+pub async fn remove_member(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath((id, username)): AxPath<(i32, String)>,
+) -> ApiResult<Value> {
+    user.require_admin()?;
+    sqlx::query("DELETE FROM project_members WHERE project_id=$1 AND username=$2")
+        .bind(id)
+        .bind(&username)
+        .execute(&s.db)
+        .await
+        .map_err(err)?;
+    // Unassign their videos in this project so nothing stays assigned to a non-member.
+    sqlx::query("UPDATE videos SET assigned_to=NULL WHERE project_id=$1 AND assigned_to=$2")
+        .bind(id)
+        .bind(&username)
+        .execute(&s.db)
+        .await
+        .map_err(err)?;
+    audit(&s.db, &user.username, "project_member_removed",
+        json!({ "project_id": id, "username": username })).await;
+    Ok(Json(json!({ "removed": true, "username": username })))
+}
+
+/// `DELETE /api/users/:id` — remove a collaborator (admin or ml_support). Admin only.
+/// You cannot delete the account you're signed in as (use a different admin / the
+/// master token). Annotations/events keep their text `annotated_by`/`username`
+/// attribution (those are not FKs), so history is preserved.
+pub async fn delete_user(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+) -> ApiResult<Value> {
+    user.require_admin()?;
+    let target: Option<String> = sqlx::query_scalar("SELECT username FROM users WHERE id=$1")
+        .bind(id)
+        .fetch_optional(&s.db)
+        .await
+        .map_err(err)?;
+    let Some(username) = target else {
+        return Err((StatusCode::NOT_FOUND, "user not found".into()));
+    };
+    if username == user.username {
+        return Err((StatusCode::BAD_REQUEST, "you cannot delete your own account".into()));
+    }
+    sqlx::query("DELETE FROM users WHERE id=$1")
+        .bind(id)
+        .execute(&s.db)
+        .await
+        .map_err(err)?;
+    audit(&s.db, &user.username, "user_deleted", json!({ "user_id": id, "username": username })).await;
+    Ok(Json(json!({ "deleted": true, "id": id, "username": username })))
+}

server/src/auth.rs

@@ -0,0 +1,93 @@
+//! Phase 3 token auth. Workers/admins authenticate with `Authorization: Bearer <token>`.
+//! Tokens are random 32-byte hex strings; only their SHA-256 hash is stored
+//! (`users.token_hash`). A master `ADMIN_TOKEN` env var maps to a synthetic admin
+//! identity so the first real user can be provisioned (bootstrap).
+
+use crate::AppState;
+use axum::{
+    async_trait,
+    extract::FromRequestParts,
+    http::{header::AUTHORIZATION, request::Parts, StatusCode},
+};
+use rand::RngCore;
+use sha2::{Digest, Sha256};
+
+/// Lowercase hex SHA-256 of `s` — used to hash tokens before storage/lookup.
+pub fn sha256_hex(s: &str) -> String {
+    let mut h = Sha256::new();
+    h.update(s.as_bytes());
+    hex::encode(h.finalize())
+}
+
+/// Generate a fresh 256-bit token as a 64-char hex string.
+pub fn gen_token() -> String {
+    let mut bytes = [0u8; 32];
+    rand::thread_rng().fill_bytes(&mut bytes);
+    hex::encode(bytes)
+}
+
+/// The authenticated caller. Use as a handler argument to require a valid token.
+pub struct AuthUser {
+    pub username: String,
+    pub role: String,
+}
+
+impl AuthUser {
+    pub fn is_admin(&self) -> bool {
+        self.role == "admin"
+    }
+    /// 403 unless this caller is an admin.
+    pub fn require_admin(&self) -> Result<(), (StatusCode, String)> {
+        if self.is_admin() {
+            Ok(())
+        } else {
+            Err((StatusCode::FORBIDDEN, "admin only".into()))
+        }
+    }
+}
+
+fn bearer(parts: &Parts) -> Option<String> {
+    let raw = parts.headers.get(AUTHORIZATION)?.to_str().ok()?;
+    raw.strip_prefix("Bearer ")
+        .or_else(|| raw.strip_prefix("bearer "))
+        .map(|t| t.trim().to_string())
+        .filter(|t| !t.is_empty())
+}
+
+/// Resolve a raw token to an `AuthUser`, or `None` if it's empty/unknown/inactive.
+/// Used both by the `FromRequestParts` extractor (header auth) and by endpoints that
+/// must accept a `?token=` query param (file downloads via `<a download>`, which can't
+/// set an Authorization header).
+pub async fn validate_token(state: &AppState, token: &str) -> Option<AuthUser> {
+    let token = token.trim();
+    if token.is_empty() {
+        return None;
+    }
+    // Master admin token (bootstrap / break-glass).
+    if !state.admin_token.is_empty() && token == state.admin_token {
+        return Some(AuthUser { username: "admin".into(), role: "admin".into() });
+    }
+    let hash = sha256_hex(token);
+    sqlx::query_as::<_, (String, String)>(
+        "SELECT username, role FROM users WHERE token_hash=$1 AND active",
+    )
+    .bind(&hash)
+    .fetch_optional(&state.db)
+    .await
+    .ok()
+    .flatten()
+    .map(|(username, role)| AuthUser { username, role })
+}
+
+#[async_trait]
+impl FromRequestParts<AppState> for AuthUser {
+    type Rejection = (StatusCode, String);
+
+    async fn from_request_parts(parts: &mut Parts, state: &AppState) -> Result<Self, Self::Rejection> {
+        let token = bearer(parts)
+            .ok_or((StatusCode::UNAUTHORIZED, "missing bearer token".to_string()))?;
+        validate_token(state, &token)
+            .await
+            .ok_or((StatusCode::UNAUTHORIZED, "invalid or inactive token".into()))
+    }
+}

server/src/ingest.rs

@@ -0,0 +1,254 @@
+use crate::models::ExportDoc;
+use chrono::{DateTime, NaiveDateTime, Utc};
+use sqlx::PgPool;
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use walkdir::WalkDir;
+
+const VIDEO_EXTS: &[&str] = &["mp4", "mkv", "avi", "mov", "m4v", "webm"];
+
+fn is_video(p: &Path) -> bool {
+    p.extension()
+        .and_then(|e| e.to_str())
+        .map(|e| VIDEO_EXTS.contains(&e.to_lowercase().as_str()))
+        .unwrap_or(false)
+}
+
+/// Find the sibling export-JSON for a video. Supports `<file>_annotations.json`
+/// (video-annotator's default export name), `<file>.json`, and `<stem>.json`.
+fn sibling_json(video: &Path) -> Option<PathBuf> {
+    let dir = video.parent()?;
+    let file_name = video.file_name()?.to_str()?;
+    let stem = video.file_stem()?.to_str()?;
+    let candidates = [
+        format!("{file_name}_annotations.json"),
+        format!("{file_name}.json"),
+        format!("{stem}.json"),
+        format!("{stem}_annotations.json"),
+    ];
+    for c in candidates {
+        let p = dir.join(c);
+        if p.is_file() {
+            return Some(p);
+        }
+    }
+    None
+}
+
+fn parse_ts(s: &str) -> Option<DateTime<Utc>> {
+    if s.is_empty() {
+        return None;
+    }
+    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
+        return Some(dt.with_timezone(&Utc));
+    }
+    NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S")
+        .ok()
+        .map(|n| n.and_utc())
+}
+
+/// Scan `dir`, upsert every video (and its annotations) into the project.
+/// Idempotent: re-running updates rows; never downgrades a 'verified' video.
+pub async fn ingest_project(db: &PgPool, project_id: i32, dir: &Path) -> anyhow::Result<usize> {
+    if !dir.is_dir() {
+        anyhow::bail!("ingest dir not found: {}", dir.display());
+    }
+    let mut count = 0usize;
+
+    for entry in WalkDir::new(dir).into_iter().filter_map(|e| e.ok()) {
+        let path = entry.path();
+        if !entry.file_type().is_file() || !is_video(path) {
+            continue;
+        }
+        let rel_path = path
+            .strip_prefix(dir)
+            .unwrap_or(path)
+            .to_string_lossy()
+            .to_string();
+        let file_name = path
+            .file_name()
+            .map(|n| n.to_string_lossy().to_string())
+            .unwrap_or_default();
+
+        // Never clobber a verified push (Phase 3): a worker's pushed annotations +
+        // completer/verify-time are authoritative, so skip re-ingesting this video.
+        let existing_status: Option<String> =
+            sqlx::query_scalar("SELECT status FROM videos WHERE project_id=$1 AND rel_path=$2")
+                .bind(project_id)
+                .bind(&rel_path)
+                .fetch_optional(db)
+                .await?;
+        if existing_status.as_deref() == Some("verified") {
+            tracing::debug!("skip re-ingest of verified video: {rel_path}");
+            continue;
+        }
+
+        let json_path = sibling_json(path);
+        let has_json = json_path.is_some();
+
+        // Parse the sibling JSON if present.
+        let doc: Option<ExportDoc> = json_path.as_ref().and_then(|jp| {
+            std::fs::read_to_string(jp)
+                .ok()
+                .and_then(|t| serde_json::from_str::<ExportDoc>(&t).ok())
+        });
+
+        let (width, height, fps, frame_count, time_ms) = match doc.as_ref().and_then(|d| d.video.as_ref()) {
+            Some(v) => (
+                v.width.map(|x| x as i32),
+                v.height.map(|x| x as i32),
+                v.fps,
+                v.frame_count.map(|x| x as i32),
+                v.annotation_time_ms,
+            ),
+            None => (None, None, None, None, 0i64),
+        };
+
+        // Aggregate annotation_count, primary annotator, annotated_at.
+        let mut ann_count: i32 = 0;
+        let mut by_freq: HashMap<String, i32> = HashMap::new();
+        let mut latest: Option<DateTime<Utc>> = None;
+        if let Some(d) = doc.as_ref() {
+            for a in &d.fixed_annotations {
+                ann_count += 1;
+                if !a.annotated_by.is_empty() {
+                    *by_freq.entry(a.annotated_by.clone()).or_default() += 1;
+                }
+                if let Some(t) = parse_ts(&a.created_at) {
+                    latest = Some(latest.map_or(t, |l| l.max(t)));
+                }
+            }
+            for a in &d.range_annotations {
+                ann_count += 1;
+                if !a.annotated_by.is_empty() {
+                    *by_freq.entry(a.annotated_by.clone()).or_default() += 1;
+                }
+                if let Some(t) = parse_ts(&a.created_at) {
+                    latest = Some(latest.map_or(t, |l| l.max(t)));
+                }
+            }
+        }
+        let primary_annotator = by_freq
+            .into_iter()
+            .max_by_key(|(_, c)| *c)
+            .map(|(u, _)| u)
+            .unwrap_or_default();
+        let status = if has_json && ann_count > 0 { "annotated" } else { "pending" };
+        let raw_json: Option<serde_json::Value> = json_path.as_ref().and_then(|jp| {
+            std::fs::read_to_string(jp).ok().and_then(|t| serde_json::from_str(&t).ok())
+        });
+
+        // Upsert the video. Keep 'verified' status sticky (Phase 3 pushes).
+        let video_id: i32 = sqlx::query_scalar(
+            r#"
+            INSERT INTO videos
+              (project_id, file_name, rel_path, has_json, width, height, fps, frame_count,
+               annotation_count, annotation_time_ms, primary_annotator, status, annotated_at, raw_json,
+               imported_count)
+            VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)
+            ON CONFLICT (project_id, rel_path) DO UPDATE SET
+              file_name=EXCLUDED.file_name,
+              has_json=EXCLUDED.has_json,
+              width=EXCLUDED.width, height=EXCLUDED.height, fps=EXCLUDED.fps,
+              frame_count=EXCLUDED.frame_count,
+              annotation_count=EXCLUDED.annotation_count,
+              annotation_time_ms=EXCLUDED.annotation_time_ms,
+              primary_annotator=EXCLUDED.primary_annotator,
+              status = CASE WHEN videos.status='verified' THEN videos.status ELSE EXCLUDED.status END,
+              annotated_at=EXCLUDED.annotated_at,
+              raw_json=EXCLUDED.raw_json,
+              -- imported baseline = the NAS-ingested count (skipped for verified videos
+              -- via the early continue above, so a pushed pass's baseline is preserved).
+              imported_count=EXCLUDED.imported_count,
+              ingested_at=now()
+            RETURNING id
+            "#,
+        )
+        .bind(project_id)
+        .bind(&file_name)
+        .bind(&rel_path)
+        .bind(has_json)
+        .bind(width)
+        .bind(height)
+        .bind(fps)
+        .bind(frame_count)
+        .bind(ann_count)
+        .bind(time_ms)
+        .bind(&primary_annotator)
+        .bind(status)
+        .bind(latest)
+        .bind(raw_json)
+        .bind(ann_count)
+        .fetch_one(db)
+        .await?;
+
+        // Replace detailed annotation rows for this video.
+        if let Some(d) = doc.as_ref() {
+            replace_annotations(db, video_id, d).await?;
+        } else {
+            sqlx::query("DELETE FROM annotations WHERE video_id=$1")
+                .bind(video_id)
+                .execute(db)
+                .await?;
+        }
+        count += 1;
+    }
+    Ok(count)
+}
+
+/// Replace all annotation rows for a video with the contents of an export doc.
+/// Shared by ingest (NAS JSON) and the Phase 3 worker push. Returns the row count
+/// written (each range annotation contributes 1–2 rows: start, and end if present).
+pub async fn replace_annotations(
+    db: &PgPool,
+    video_id: i32,
+    d: &ExportDoc,
+) -> anyhow::Result<i32> {
+    sqlx::query("DELETE FROM annotations WHERE video_id=$1")
+        .bind(video_id)
+        .execute(db)
+        .await?;
+    let mut written = 0i32;
+    for a in &d.fixed_annotations {
+        insert_ann(db, video_id, a.frame_number, &a.label, &a.side, &a.shape_type,
+            &a.vertices, &a.annotated_by, &a.review_status, &a.remark, &a.subclass, &a.created_at).await?;
+        written += 1;
+    }
+    for a in &d.range_annotations {
+        insert_ann(db, video_id, a.start_frame, &a.label, &a.side, &a.shape_type,
+            &a.start_vertices, &a.annotated_by, &a.review_status, &a.remark, &a.subclass, &a.created_at).await?;
+        written += 1;
+        if let (Some(ef), Some(ev)) = (a.end_frame, a.end_vertices.as_ref()) {
+            insert_ann(db, video_id, ef, &a.label, &a.side, &a.shape_type,
+                ev, &a.annotated_by, &a.review_status, &a.remark, &a.subclass, &a.created_at).await?;
+            written += 1;
+        }
+    }
+    Ok(written)
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn insert_ann(
+    db: &PgPool, video_id: i32, frame: i64, label: &str, side: &str, shape: &str,
+    vertices: &serde_json::Value, by: &str, review: &str, remark: &str, subclass: &str, created_at: &str,
+) -> anyhow::Result<()> {
+    sqlx::query(
+        r#"INSERT INTO annotations
+           (video_id, frame_number, label, side, shape_type, vertices, annotated_by, review_status, remark, subclass, created_at)
+           VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)"#,
+    )
+    .bind(video_id)
+    .bind(frame as i32)
+    .bind(label)
+    .bind(side)
+    .bind(if shape.is_empty() { "bbox" } else { shape })
+    .bind(vertices)
+    .bind(by)
+    .bind(if review.is_empty() { "none" } else { review })
+    .bind(remark)
+    .bind(subclass)
+    .bind(created_at)
+    .execute(db)
+    .await?;
+    Ok(())
+}

server/src/main.rs

@@ -0,0 +1,969 @@
+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 <> ''
+           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 completion
+    // SPEED (avg verify time per video, fastest first). Anyone who has completed
+    // videos but isn't a member is included too; members with no completions sort
+    // last (unranked). 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 <> ''
+           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 {
+        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))| VerifierRow {
+            avg_time_ms: if videos > 0 { total_time_ms / videos } else { 0 },
+            videos,
+            total_time_ms,
+            annotations: ann_by_user.get(&user).copied().unwrap_or(0),
+            rank: 0,
+            user,
+        })
+        .collect();
+    // Fastest first: completers by avg ascending; non-completers last by name.
+    verifiers.sort_by(|a, b| match (a.videos > 0, b.videos > 0) {
+        (true, false) => std::cmp::Ordering::Less,
+        (false, true) => std::cmp::Ordering::Greater,
+        (true, true) => a.avg_time_ms.cmp(&b.avg_time_ms).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 v.videos > 0 {
+            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 }))
+}

server/src/models.rs

@@ -0,0 +1,426 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+// ---- Parsed from the sibling export-JSON (video-annotator export format) ----
+
+#[derive(Deserialize, Default)]
+pub struct ExportDoc {
+    #[serde(default)]
+    pub video: Option<ExportVideo>,
+    #[serde(default)]
+    pub fixed_annotations: Vec<FixedAnn>,
+    #[serde(default)]
+    pub range_annotations: Vec<RangeAnn>,
+}
+
+#[derive(Deserialize, Default)]
+pub struct ExportVideo {
+    #[serde(default)]
+    pub file_name: String,
+    #[serde(default)]
+    pub width: Option<i64>,
+    #[serde(default)]
+    pub height: Option<i64>,
+    #[serde(default)]
+    pub fps: Option<f64>,
+    #[serde(default)]
+    pub frame_count: Option<i64>,
+    #[serde(default)]
+    pub annotation_time_ms: i64,
+}
+
+#[derive(Deserialize)]
+pub struct FixedAnn {
+    #[serde(default)]
+    pub label: String,
+    #[serde(default)]
+    pub side: String,
+    #[serde(default)]
+    pub frame_number: i64,
+    #[serde(default, rename = "type")]
+    pub shape_type: String,
+    #[serde(default)]
+    pub vertices: Value,
+    #[serde(default)]
+    pub annotated_by: String,
+    #[serde(default)]
+    pub remark: String,
+    #[serde(default)]
+    pub review_status: String,
+    #[serde(default)]
+    pub subclass: String,
+    #[serde(default)]
+    pub created_at: String,
+}
+
+#[derive(Deserialize)]
+pub struct RangeAnn {
+    #[serde(default)]
+    pub label: String,
+    #[serde(default)]
+    pub side: String,
+    #[serde(default)]
+    pub start_frame: i64,
+    #[serde(default)]
+    pub start_vertices: Value,
+    #[serde(default)]
+    pub end_frame: Option<i64>,
+    #[serde(default)]
+    pub end_vertices: Option<Value>,
+    #[serde(default, rename = "type")]
+    pub shape_type: String,
+    #[serde(default)]
+    pub annotated_by: String,
+    #[serde(default)]
+    pub remark: String,
+    #[serde(default)]
+    pub review_status: String,
+    #[serde(default)]
+    pub subclass: String,
+    #[serde(default)]
+    pub created_at: String,
+}
+
+// ---- API response shapes ----
+
+#[derive(Serialize)]
+pub struct ProjectSummary {
+    pub id: i32,
+    pub name: String,
+    pub source_path: String,
+    pub source_kind: String,
+    pub client_id: Option<i32>,
+    pub client_name: String,
+    pub total: i64,
+    pub pending: i64,
+    pub annotated: i64,
+    pub verified: i64,
+}
+
+/// A client owning one or more projects (admin-created from the dashboard).
+#[derive(Serialize)]
+pub struct ClientRow {
+    pub id: i32,
+    pub name: String,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    pub project_count: i64,
+}
+
+/// Admin: `POST /api/clients` — create a client.
+#[derive(Deserialize)]
+pub struct NewClientRequest {
+    pub name: String,
+}
+
+#[derive(Serialize, sqlx::FromRow)]
+pub struct VideoRow {
+    pub id: i32,
+    pub file_name: String,
+    pub rel_path: String,
+    pub has_json: bool,
+    pub width: Option<i32>,
+    pub height: Option<i32>,
+    pub fps: Option<f64>,
+    pub frame_count: Option<i32>,
+    pub annotation_count: i32,
+    // Baseline count this video started a verification pass with: the NAS-ingested
+    // count, or (after a push) the count the worker had at claim time. Compared with
+    // annotation_count it shows how many the worker added/deleted.
+    pub imported_count: i32,
+    pub annotation_time_ms: i64,
+    pub primary_annotator: String,
+    pub status: String,
+    pub annotated_at: Option<chrono::DateTime<chrono::Utc>>,
+    // Phase 3: claim/verify dimension (orthogonal to `status`).
+    pub claimed_by: Option<String>,
+    pub claimed_at: Option<chrono::DateTime<chrono::Utc>>,
+    pub lease_expires_at: Option<chrono::DateTime<chrono::Utc>>,
+    pub completed_by: Option<String>,
+    pub completed_at: Option<chrono::DateTime<chrono::Utc>>,
+    pub verify_time_ms: i64,
+    // Annotations still flagged for review (review_status='flagged') on this video.
+    pub review_count: i64,
+    // How many times pushed (1 = verified, ≥2 = re-verified) + the distinct verifiers
+    // ("ravi, john"), both derived from the push event log.
+    pub verify_count: i64,
+    pub verifiers: String,
+    // Admin assignment (advisory owner) + the derived lifecycle label for the UI badge:
+    // pending | annotated | assigned | in-progress | verified | re-verified.
+    pub assigned_to: Option<String>,
+    pub workflow_status: String,
+    // "Raise hand" signal — anyone may raise (to start a conversation) or lower it.
+    pub hand_raised: bool,
+    pub hand_raised_by: Option<String>,
+    pub hand_raised_at: Option<chrono::DateTime<chrono::Utc>>,
+    // "Ignore" flag — admin-only; the UI crosses out ignored rows.
+    pub ignored: bool,
+    pub ignored_by: Option<String>,
+    // Remark thread (json array of {username, body, created_at}); '[]' when none.
+    pub remarks: serde_json::Value,
+}
+
+/// `POST /api/videos/:id/hand {raised}` — raise/lower the hand on a video.
+#[derive(Deserialize)]
+pub struct HandRequest {
+    pub raised: bool,
+}
+
+/// `POST /api/videos/:id/ignore {ignored}` — admin-only ignore toggle.
+#[derive(Deserialize)]
+pub struct IgnoreRequest {
+    pub ignored: bool,
+}
+
+#[derive(Deserialize)]
+pub struct AssignRequest {
+    pub video_ids: Vec<i32>,
+    pub assignee: String, // empty = unassign
+}
+
+/// A user picked for a project (project_members joined with users).
+#[derive(Serialize, sqlx::FromRow)]
+pub struct MemberRow {
+    pub username: String,
+    pub display_name: String,
+    pub role: String,
+    pub added_at: chrono::DateTime<chrono::Utc>,
+}
+
+#[derive(Deserialize)]
+pub struct AddMemberRequest {
+    pub username: String,
+}
+
+/// One line in a video's remark thread.
+#[derive(Serialize, sqlx::FromRow)]
+pub struct Remark {
+    pub username: String,
+    pub body: String,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+}
+
+#[derive(Deserialize)]
+pub struct NewRemarkRequest {
+    pub body: String,
+}
+
+#[derive(Serialize)]
+pub struct Overview {
+    pub total: i64,
+    pub pending: i64,
+    pub annotated: i64,
+    pub verified: i64,
+    pub pct_done: f64,
+}
+
+#[derive(Serialize)]
+pub struct LeaderRow {
+    pub user: String,
+    pub videos: i64,
+    pub annotations: i64,
+    pub total_time_ms: i64,
+    pub avg_time_ms: i64,
+}
+
+#[derive(Serialize)]
+pub struct Stats {
+    pub overview: Overview,
+    pub leaderboard: Vec<LeaderRow>,
+    // Phase 3: verification leaderboard (by `completed_by`).
+    pub verifiers: Vec<VerifierRow>,
+    // Project time, verification pace + worker-based ETA.
+    pub timeline: ProjectTime,
+}
+
+/// Aggregate project time metrics + a worker-parallelised ETA for the remaining work.
+#[derive(Serialize)]
+pub struct ProjectTime {
+    /// Users picked for the project (the parallelism factor for the ETA).
+    pub active_workers: i64,
+    /// Average effort per verified video (annotation + verify time).
+    pub avg_video_ms: i64,
+    /// Total time spent by all users across all videos (annotation + verify).
+    pub total_spent_ms: i64,
+    /// Wall-clock from project inception to last completion (or now if ongoing).
+    pub elapsed_ms: i64,
+    /// Whether every video is verified.
+    pub all_verified: bool,
+    /// Estimated remaining work time = remaining × avg_video ÷ active_workers.
+    pub eta_ms: Option<i64>,
+    /// Whether the project timer is currently running (accumulating).
+    pub time_running: bool,
+    /// Last verify/re-verify activity (drives the 72h auto-stop).
+    pub last_activity_at: Option<chrono::DateTime<chrono::Utc>>,
+    /// Verification pace: push (verify) events in the recent window, and the window
+    /// length in days. videos/day = recent_verified ÷ recent_window_days.
+    pub recent_verified: i64,
+    pub recent_window_days: i64,
+}
+
+/// `POST /api/projects/:id/timer {action}` — start | stop | reset (admin only).
+#[derive(Deserialize)]
+pub struct TimerRequest {
+    pub action: String,
+}
+
+/// `POST /api/projects/:id/folders {folders}` — set the folders to consider (admin).
+#[derive(Deserialize)]
+pub struct FoldersRequest {
+    pub folders: Vec<String>,
+}
+
+/// `GET /api/projects/:id/folders` — a folder in the project + its video count +
+/// whether it's currently included in verification.
+#[derive(Serialize, sqlx::FromRow)]
+pub struct FolderRow {
+    pub folder: String,
+    pub video_count: i64,
+    pub included: bool,
+}
+
+#[derive(Serialize)]
+pub struct VerifierRow {
+    pub user: String,
+    pub videos: i64,
+    pub total_time_ms: i64,
+    pub avg_time_ms: i64,
+    /// Total annotations this user authored (annotated_by) in the project.
+    pub annotations: i64,
+    /// 1-based rank by completion speed (fastest avg first); 0 = no completions yet.
+    pub rank: i64,
+}
+
+// ---- Phase 3: auth + worker pull/push shapes ----
+
+/// `GET /api/auth/whoami` — identifies the bearer.
+#[derive(Serialize)]
+pub struct WhoAmI {
+    pub username: String,
+    pub role: String,
+}
+
+/// `POST /api/videos/:id/claim` response — everything the desktop needs to verify.
+#[derive(Serialize)]
+pub struct ClaimResponse {
+    pub video_id: i32,
+    pub file_name: String,
+    pub rel_path: String,
+    pub width: Option<i32>,
+    pub height: Option<i32>,
+    pub fps: Option<f64>,
+    pub frame_count: Option<i32>,
+    pub lease_expires_at: chrono::DateTime<chrono::Utc>,
+    pub download_url: String,
+    /// Full preloaded export doc (`{video, fixed_annotations[], range_annotations[]}`)
+    /// for the desktop to import via its existing import path.
+    pub annotations: Value,
+}
+
+/// Admin: `POST /api/users` — provision a worker/admin.
+#[derive(Deserialize)]
+pub struct NewUserRequest {
+    pub username: String,
+    #[serde(default)]
+    pub display_name: String,
+    #[serde(default = "default_role")]
+    pub role: String,
+}
+fn default_role() -> String {
+    "ml_support".into()
+}
+
+/// Admin: `POST /api/users/:id/update` — edit a user's display name + role.
+#[derive(Deserialize)]
+pub struct UpdateUserRequest {
+    #[serde(default)]
+    pub display_name: String,
+    #[serde(default)]
+    pub role: String,
+}
+
+/// Admin user-provision response — the **only** time the clear token is shown.
+#[derive(Serialize)]
+pub struct NewUserResponse {
+    pub username: String,
+    pub display_name: String,
+    pub role: String,
+    pub token: String,
+}
+
+#[derive(Serialize, sqlx::FromRow)]
+pub struct UserRow {
+    pub id: i32,
+    pub username: String,
+    pub display_name: String,
+    pub role: String,
+    pub active: bool,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+}
+
+/// `POST /api/projects` — create a project under a client (any authenticated user).
+#[derive(Deserialize)]
+pub struct NewProjectRequest {
+    pub client_id: i32,
+    pub name: String,
+    pub source_path: String,
+    /// 'local' (default) or 'nfs'.
+    #[serde(default = "default_source_kind")]
+    pub source_kind: String,
+}
+fn default_source_kind() -> String {
+    "local".into()
+}
+
+/// `POST /api/projects/:id/source` — change a project's sync directory (any authenticated user).
+#[derive(Deserialize)]
+pub struct UpdateSourceRequest {
+    pub source_path: String,
+    #[serde(default = "default_source_kind")]
+    pub source_kind: String,
+}
+
+/// `GET /api/admin/storage` — DB size + row counts (admin only).
+#[derive(Serialize)]
+pub struct StorageInfo {
+    pub db_bytes: i64,
+    pub videos: i64,
+    pub annotations: i64,
+    pub video_events: i64,
+    pub audit_events: i64,
+    /// Retention cap applied nightly to the event/audit logs.
+    pub max_log_rows: i64,
+}
+
+/// `GET /api/audit` — an org-level audit row.
+#[derive(Serialize, sqlx::FromRow)]
+pub struct AuditRow {
+    pub at: chrono::DateTime<chrono::Utc>,
+    pub username: String,
+    pub action: String,
+    pub detail: Value,
+}
+
+/// `GET /api/projects/:id/activity` — live claims + recent events.
+#[derive(Serialize)]
+pub struct Activity {
+    pub active_claims: Vec<ActiveClaim>,
+    pub recent_events: Vec<EventRow>,
+}
+
+#[derive(Serialize, sqlx::FromRow)]
+pub struct ActiveClaim {
+    pub video_id: i32,
+    pub file_name: String,
+    pub claimed_by: String,
+    pub claimed_at: chrono::DateTime<chrono::Utc>,
+    pub lease_expires_at: chrono::DateTime<chrono::Utc>,
+}
+
+#[derive(Serialize, sqlx::FromRow)]
+pub struct EventRow {
+    pub video_id: i32,
+    pub file_name: String,
+    pub username: String,
+    pub event: String,
+    pub at: chrono::DateTime<chrono::Utc>,
+}

server/src/nfs.rs

@@ -0,0 +1,146 @@
+//! Resolve a project's source folder to a real local directory the server can scan.
+//!
+//! - `local` projects: `source_path` is already a path inside the container (a plain folder, or a
+//!   compose-managed NFS volume) — returned as-is.
+//! - `nfs` projects: the share is mounted on demand at `/mnt/nfs/<sanitized>`. `source_path` may be:
+//!     * a bare export path `/volume4/Share` — the host comes from the `NFS_HOST` env (IP hidden), or
+//!     * `host:/volume4/Share`, or `nfs://host/volume4/Share` — host taken from the path itself.
+
+use std::io::Read;
+use std::net::{TcpStream, ToSocketAddrs};
+use std::path::{Path, PathBuf};
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant};
+
+const NFS_MOUNT_ROOT: &str = "/mnt/nfs";
+
+/// Fast reachability probe: can we open a TCP connection to the NAS's NFS port?
+/// `mount.nfs` can hang for minutes on a filtered host (the connect is uninterruptible),
+/// so we bound the unreachable case here to a few seconds and fail with a clean message.
+fn probe_reachable(nfs_host: &str) -> Result<(), String> {
+    let addr = format!("{nfs_host}:2049");
+    let sock = addr
+        .to_socket_addrs()
+        .map_err(|_| "could not resolve the configured NAS host".to_string())?
+        .next()
+        .ok_or_else(|| "could not resolve the configured NAS host".to_string())?;
+    TcpStream::connect_timeout(&sock, Duration::from_secs(3))
+        .map(|_| ())
+        .map_err(|_| "NAS not reachable (no NFS service on the configured host)".to_string())
+}
+
+/// Turn a string into a stable, filesystem-safe mountpoint dir name.
+fn sanitize(s: &str) -> String {
+    s.trim_matches('/')
+        .chars()
+        .map(|c| if c.is_ascii_alphanumeric() || c == '-' || c == '.' { c } else { '_' })
+        .collect()
+}
+
+/// Parse an nfs `source_path` into `(host, export)`. Accepts:
+///   `nfs://host/volume4/Share`, `host:/volume4/Share`, or bare `/volume4/Share`
+/// (bare → host falls back to the server's `NFS_HOST`). The export is always absolute.
+fn parse_nfs(source_path: &str, default_host: &str) -> Result<(String, String), String> {
+    let s = source_path.trim();
+
+    // nfs://host/export
+    let scheme = s.strip_prefix("nfs://").or_else(|| s.strip_prefix("NFS://"));
+    if let Some(rest) = scheme {
+        let (host, export) = rest
+            .split_once('/')
+            .ok_or_else(|| "nfs:// path must include an export, e.g. nfs://host/volume4/Share".to_string())?;
+        if host.is_empty() {
+            return Err("nfs:// path is missing the host".into());
+        }
+        return Ok((host.to_string(), format!("/{}", export.trim_start_matches('/'))));
+    }
+
+    // host:/export  (standard NFS device notation)
+    if let Some((host, export)) = s.split_once(':') {
+        if !host.is_empty() && export.starts_with('/') {
+            return Ok((host.to_string(), export.to_string()));
+        }
+    }
+
+    // bare /export → host from server env
+    if s.starts_with('/') {
+        if default_host.is_empty() {
+            return Err("NFS host is not configured on the server (set NFS_HOST)".into());
+        }
+        return Ok((default_host.to_string(), s.to_string()));
+    }
+
+    Err(format!(
+        "invalid NFS path '{s}' — use /export, host:/export, or nfs://host/export"
+    ))
+}
+
+/// Is `mp` already an active mountpoint? (Reads /proc/mounts — no extra binaries needed.)
+fn is_mounted(mp: &Path) -> bool {
+    let target = mp.to_string_lossy();
+    std::fs::read_to_string("/proc/mounts")
+        .map(|s| s.lines().any(|l| l.split(' ').nth(1) == Some(target.as_ref())))
+        .unwrap_or(false)
+}
+
+/// Resolve `(source_path, source_kind)` to a local directory, mounting an NFS share if needed.
+/// `nfs_host`/`nfs_opts` come from server env. Blocking (runs a `mount` for nfs) — call inside
+/// `spawn_blocking`.
+pub fn resolve_dir(
+    source_path: &str,
+    source_kind: &str,
+    nfs_host: &str,
+    nfs_opts: &str,
+) -> Result<PathBuf, String> {
+    if source_kind != "nfs" {
+        return Ok(PathBuf::from(source_path));
+    }
+
+    let (host, export) = parse_nfs(source_path, nfs_host)?;
+
+    let mp = PathBuf::from(NFS_MOUNT_ROOT).join(sanitize(&format!("{host}_{export}")));
+    if is_mounted(&mp) {
+        return Ok(mp);
+    }
+
+    // Fail fast if the NAS isn't reachable, before the (potentially long-hanging) mount.
+    probe_reachable(&host)?;
+
+    std::fs::create_dir_all(&mp).map_err(|e| format!("mkdir {}: {e}", mp.display()))?;
+    let device = format!("{host}:{export}");
+    // Spawn the mount and enforce the deadline ourselves: a `mount.nfs` stuck mid-handshake goes
+    // uninterruptible (D state) and ignores SIGTERM/SIGKILL, so `timeout` can't bound it. We poll
+    // and return a clean error at the deadline; an orphaned mount that later succeeds will be picked
+    // up by `is_mounted` on the next sync.
+    let mut child = Command::new("mount")
+        .args(["-t", "nfs", "-o", nfs_opts, &device, &mp.to_string_lossy()])
+        .stdout(Stdio::null())
+        .stderr(Stdio::piped())
+        .spawn()
+        .map_err(|e| format!("failed to run mount (is nfs-common installed?): {e}"))?;
+
+    let deadline = Instant::now() + Duration::from_secs(15);
+    loop {
+        match child.try_wait() {
+            Ok(Some(status)) if status.success() => return Ok(mp),
+            Ok(Some(_)) => {
+                let mut err = String::new();
+                if let Some(mut e) = child.stderr.take() {
+                    let _ = e.read_to_string(&mut err);
+                }
+                let safe = err.replace(&host, "<nas>");
+                return Err(format!("mount of '{export}' failed: {}", safe.trim()));
+            }
+            Ok(None) => {
+                if Instant::now() >= deadline {
+                    let _ = child.kill();
+                    return Err(format!(
+                        "mount of '{export}' timed out — check the NAS exports this path to the server"
+                    ));
+                }
+                std::thread::sleep(Duration::from_millis(300));
+            }
+            Err(e) => return Err(format!("mount wait failed: {e}")),
+        }
+    }
+}

server/src/worker.rs

@@ -0,0 +1,301 @@
+//! Phase 3 worker pull/push API. All routes require a valid bearer token (`AuthUser`).
+//! Flow: claim (atomic, leased) → download local copy → verify in the desktop → push.
+
+use crate::auth::AuthUser;
+use crate::models::*;
+use crate::{err, AppState, ApiResult};
+use axum::{
+    body::Body,
+    extract::{Path as AxPath, State},
+    http::{header, StatusCode},
+    response::Response,
+    Json,
+};
+use serde_json::{json, Value};
+use tokio_util::io::ReaderStream;
+
+/// Record an audit/analytics event. Best-effort: errors are logged, not surfaced.
+pub async fn log_event(db: &sqlx::PgPool, video_id: i32, username: &str, event: &str, meta: Value) {
+    if let Err(e) = sqlx::query(
+        "INSERT INTO video_events (video_id, username, event, meta) VALUES ($1,$2,$3,$4)",
+    )
+    .bind(video_id)
+    .bind(username)
+    .bind(event)
+    .bind(meta)
+    .execute(db)
+    .await
+    {
+        tracing::warn!("failed to log {event} event for video {video_id}: {e}");
+    }
+}
+
+/// `GET /api/auth/whoami` — validate the token and identify the caller.
+pub async fn whoami(user: AuthUser) -> ApiResult<WhoAmI> {
+    Ok(Json(WhoAmI { username: user.username, role: user.role }))
+}
+
+/// `POST /api/auth/login` — like whoami, but records a `signed_in` audit event so
+/// admins can see who logged in (Activity log). Called once on an explicit sign-in,
+/// not on every token check, so the log isn't spammed.
+pub async fn login(State(s): State<AppState>, user: AuthUser) -> ApiResult<WhoAmI> {
+    crate::admin::audit(&s.db, &user.username, "signed_in", json!({ "role": user.role })).await;
+    Ok(Json(WhoAmI { username: user.username, role: user.role }))
+}
+
+/// `POST /api/videos/:id/claim` — atomically claim a video. **Any** status is claimable:
+/// `pending` (annotate from scratch), `annotated` (verify), or `verified` (re-verify —
+/// anyone may re-pull a completed video and push again). Unclaimed or lease-expired only;
+/// same-user re-claim is allowed (resume). Returns metadata + any preloaded annotations.
+pub async fn claim(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+) -> ApiResult<ClaimResponse> {
+    let row: Option<(i32, String, String, Option<i32>, Option<i32>, Option<f64>, Option<i32>, chrono::DateTime<chrono::Utc>, Option<Value>)> =
+        sqlx::query_as(
+            r#"
+            UPDATE videos
+            SET claimed_by=$2, claimed_at=now(), lease_expires_at=now() + ($3::int * interval '1 second')
+            WHERE id=$1
+              AND (claimed_by IS NULL OR claimed_by=$2 OR lease_expires_at < now())
+            RETURNING id, file_name, rel_path, width, height, fps, frame_count, lease_expires_at, raw_json
+            "#,
+        )
+        .bind(id)
+        .bind(&user.username)
+        .bind(s.lease_secs)
+        .fetch_optional(&s.db)
+        .await
+        .map_err(err)?;
+
+    let Some((video_id, file_name, rel_path, width, height, fps, frame_count, lease_expires_at, raw_json)) = row
+    else {
+        // Distinguish "doesn't exist" from "not claimable".
+        let exists: Option<i32> = sqlx::query_scalar("SELECT id FROM videos WHERE id=$1")
+            .bind(id)
+            .fetch_optional(&s.db)
+            .await
+            .map_err(err)?;
+        return Err(match exists {
+            None => (StatusCode::NOT_FOUND, "video not found".into()),
+            Some(_) => (
+                StatusCode::CONFLICT,
+                "video is held by another worker".into(),
+            ),
+        });
+    };
+
+    log_event(&s.db, video_id, &user.username, "claim", json!({})).await;
+
+    Ok(Json(ClaimResponse {
+        video_id,
+        file_name,
+        rel_path,
+        width,
+        height,
+        fps,
+        frame_count,
+        lease_expires_at,
+        download_url: format!("/api/videos/{video_id}/download"),
+        annotations: raw_json.unwrap_or(Value::Null),
+    }))
+}
+
+/// `GET /api/videos/:id/download` — stream the video bytes from the source folder
+/// (the NAS mount on the VM). Only the current claimant (or an admin) may download,
+/// so NAS credentials never leave the server.
+pub async fn download(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+) -> Result<Response, (StatusCode, String)> {
+    let row: Option<(String, String, Option<String>)> = sqlx::query_as(
+        r#"SELECT v.rel_path, v.file_name, v.claimed_by
+           FROM videos v WHERE v.id=$1"#,
+    )
+    .bind(id)
+    .fetch_optional(&s.db)
+    .await
+    .map_err(err)?;
+    let (rel_path, file_name, claimed_by) =
+        row.ok_or((StatusCode::NOT_FOUND, "video not found".to_string()))?;
+
+    if !user.is_admin() && claimed_by.as_deref() != Some(user.username.as_str()) {
+        return Err((StatusCode::FORBIDDEN, "you do not hold this claim".into()));
+    }
+
+    // Resolve against the project's source folder (mount NFS if `source_kind='nfs'`).
+    let (source_path, source_kind): (String, String) = sqlx::query_as(
+        "SELECT p.source_path, p.source_kind FROM projects p JOIN videos v ON v.project_id=p.id WHERE v.id=$1",
+    )
+    .bind(id)
+    .fetch_one(&s.db)
+    .await
+    .map_err(err)?;
+
+    let (sp, nfs_host, nfs_opts) = (source_path, s.nfs_host.clone(), s.nfs_opts.clone());
+    let dir = tokio::task::spawn_blocking(move || crate::nfs::resolve_dir(&sp, &source_kind, &nfs_host, &nfs_opts))
+        .await
+        .map_err(err)?
+        .map_err(|e| (StatusCode::BAD_REQUEST, e))?;
+    let full = dir.join(&rel_path);
+    let file = tokio::fs::File::open(&full).await.map_err(|e| {
+        (StatusCode::NOT_FOUND, format!("source file unavailable ({}): {e}", full.display()))
+    })?;
+    let len = file.metadata().await.map_err(err)?.len();
+    let body = Body::from_stream(ReaderStream::new(file));
+
+    Response::builder()
+        .header(header::CONTENT_TYPE, "application/octet-stream")
+        .header(header::CONTENT_LENGTH, len)
+        .header(
+            header::CONTENT_DISPOSITION,
+            format!("attachment; filename=\"{file_name}\""),
+        )
+        .body(body)
+        .map_err(err)
+}
+
+/// `POST /api/videos/:id/heartbeat` — extend the lease while verifying.
+pub async fn heartbeat(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+) -> ApiResult<Value> {
+    let new_lease: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
+        r#"UPDATE videos
+           SET lease_expires_at = now() + ($3::int * interval '1 second')
+           WHERE id=$1 AND claimed_by=$2
+           RETURNING lease_expires_at"#,
+    )
+    .bind(id)
+    .bind(&user.username)
+    .bind(s.lease_secs)
+    .fetch_optional(&s.db)
+    .await
+    .map_err(err)?;
+
+    match new_lease {
+        Some(lease_expires_at) => Ok(Json(json!({ "lease_expires_at": lease_expires_at }))),
+        None => Err((StatusCode::CONFLICT, "you do not hold this claim".into())),
+    }
+}
+
+/// `POST /api/videos/:id/release` — abandon a claim, returning the video to the pool.
+pub async fn release(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+) -> ApiResult<Value> {
+    let released: Option<i32> = sqlx::query_scalar(
+        r#"UPDATE videos
+           SET claimed_by=NULL, claimed_at=NULL, lease_expires_at=NULL
+           WHERE id=$1 AND claimed_by=$2
+           RETURNING id"#,
+    )
+    .bind(id)
+    .bind(&user.username)
+    .fetch_optional(&s.db)
+    .await
+    .map_err(err)?;
+
+    match released {
+        Some(_) => {
+            log_event(&s.db, id, &user.username, "release", json!({})).await;
+            Ok(Json(json!({ "released": true })))
+        }
+        None => Err((StatusCode::CONFLICT, "you do not hold this claim".into())),
+    }
+}
+
+/// `POST /api/videos/:id/push` — submit the verified export doc + verify time.
+/// Replaces the annotations, marks the video `verified`, records the verifier and
+/// time, and clears the claim. Reuses the same export wire format as ingest.
+pub async fn push(
+    State(s): State<AppState>,
+    user: AuthUser,
+    AxPath(id): AxPath<i32>,
+    Json(body): Json<Value>,
+) -> ApiResult<Value> {
+    // The current holder (or an admin) may push. A null holder (auto-released, not
+    // re-claimed) is tolerated; a different holder is rejected.
+    let claimed_by: Option<Option<String>> =
+        sqlx::query_scalar("SELECT claimed_by FROM videos WHERE id=$1")
+            .bind(id)
+            .fetch_optional(&s.db)
+            .await
+            .map_err(err)?;
+    let claimed_by = claimed_by.ok_or((StatusCode::NOT_FOUND, "video not found".to_string()))?;
+    if !user.is_admin() {
+        if let Some(holder) = claimed_by.as_deref() {
+            if holder != user.username {
+                return Err((StatusCode::CONFLICT, "claim is held by another worker".into()));
+            }
+        }
+    }
+
+    let verify_time_ms = body.get("verify_time_ms").and_then(|v| v.as_i64()).unwrap_or(0);
+    // The desktop reports how many annotations it had at claim time (the preloaded
+    // baseline). 0/absent → keep the existing baseline (set at ingest), so we never
+    // wipe it with an old client that doesn't send the field.
+    let imported_count = body.get("imported_count").and_then(|v| v.as_i64()).unwrap_or(0) as i32;
+    let doc: ExportDoc = serde_json::from_value(body.clone())
+        .map_err(|e| (StatusCode::BAD_REQUEST, format!("invalid export doc: {e}")))?;
+
+    crate::ingest::replace_annotations(&s.db, id, &doc)
+        .await
+        .map_err(err)?;
+    let ann_count = (doc.fixed_annotations.len() + doc.range_annotations.len()) as i32;
+
+    // Recompute the video-level annotator fields from the pushed doc (same idea as
+    // ingest) so the annotation leaderboard + throughput populate for pushed videos:
+    // primary_annotator = the most-frequent drawer; annotation_time_ms from the doc.
+    let mut freq: std::collections::HashMap<String, i32> = std::collections::HashMap::new();
+    for a in &doc.fixed_annotations {
+        if !a.annotated_by.is_empty() { *freq.entry(a.annotated_by.clone()).or_default() += 1; }
+    }
+    for a in &doc.range_annotations {
+        if !a.annotated_by.is_empty() { *freq.entry(a.annotated_by.clone()).or_default() += 1; }
+    }
+    let primary_annotator = freq.into_iter().max_by_key(|(_, c)| *c).map(|(u, _)| u).unwrap_or_default();
+    let ann_time_ms = doc.video.as_ref().map(|v| v.annotation_time_ms).unwrap_or(0);
+
+    // Accumulate verify_time_ms across passes (re-verification adds up). The list
+    // of verifiers + the pass count are derived from video_events (each push logs one).
+    sqlx::query(
+        r#"UPDATE videos
+           SET status='verified', completed_by=$2, completed_at=now(),
+               verify_time_ms = COALESCE(videos.verify_time_ms, 0) + $3,
+               annotation_count=$4, raw_json=$5,
+               primary_annotator=$6, annotation_time_ms=$7, annotated_at=now(),
+               imported_count = COALESCE(NULLIF($8, 0), videos.imported_count),
+               claimed_by=NULL, claimed_at=NULL, lease_expires_at=NULL
+           WHERE id=$1"#,
+    )
+    .bind(id)
+    .bind(&user.username)
+    .bind(verify_time_ms)
+    .bind(ann_count)
+    .bind(&body)
+    .bind(&primary_annotator)
+    .bind(ann_time_ms)
+    .bind(imported_count)
+    .execute(&s.db)
+    .await
+    .map_err(err)?;
+
+    log_event(&s.db, id, &user.username, "push", json!({ "verify_time_ms": verify_time_ms })).await;
+
+    // Mark project activity + auto-resume its timer (a verify/re-verify restarts it).
+    let _ = sqlx::query(
+        r#"UPDATE projects SET last_activity_at=now(), time_running=TRUE,
+             time_started_at = CASE WHEN time_running THEN time_started_at ELSE now() END
+           WHERE id = (SELECT project_id FROM videos WHERE id=$1)"#,
+    )
+    .bind(id)
+    .execute(&s.db)
+    .await;
+
+    Ok(Json(json!({ "verified": true, "annotation_count": ann_count })))
+}