Centralised_VideoVerification_Dashboard/server/src/worker.rs

//! 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> {
    #[allow(clippy::type_complexity)]
    let row: Option<(i32, String, String, Option<i32>, Option<i32>, Option<f64>, Option<i32>, chrono::DateTime<chrono::Utc>, Option<Value>, String, Option<f64>, Option<f64>, Option<f64>, Option<f64>)> =
        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,
                      road_type, chainage_start_m, chainage_end_m, sr_chainage_start_m, sr_chainage_end_m
            "#,
        )
        .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,
              road_type, ch_s, ch_e, sr_s, sr_e)) = 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;

    // Inject the server-authoritative road_type + chainage into the doc the worker
    // pulls (only when a doc exists — a pending video legitimately has none; push
    // re-stamps regardless, so the metadata always persists). Each annotation also
    // gets its frame-exact lat/lon + chainage.
    let annotations = match raw_json {
        Some(mut doc) => {
            crate::stamp_road_meta(&mut doc, &road_type, (ch_s, ch_e, sr_s, sr_e));
            if let Some(geo) = crate::chainage::video_geo(&s.db, video_id).await {
                crate::chainage::stamp_annotation_geo(&mut doc, &geo);
            }
            doc
        }
        None => Value::Null,
    };

    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,
    }))
}

/// `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> {
    // A non-admin pusher must currently hold an ACTIVE claim on this video. This blocks
    // a stale/expired claim from overwriting newer authoritative server state: if the
    // lease lapsed (auto-released, or re-claimed by someone else) the worker must
    // re-claim before pushing — their local work is never lost, it just can't clobber a
    // claim they no longer hold. Admins may always push.
    #[allow(clippy::type_complexity)]
    let row: Option<(Option<String>, Option<chrono::DateTime<chrono::Utc>>, String, Option<f64>, Option<f64>, Option<f64>, Option<f64>)> =
        sqlx::query_as(
            r#"SELECT claimed_by, lease_expires_at, road_type,
                      chainage_start_m, chainage_end_m, sr_chainage_start_m, sr_chainage_end_m
               FROM videos WHERE id=$1"#,
        )
            .bind(id)
            .fetch_optional(&s.db)
            .await
            .map_err(err)?;
    let (claimed_by, lease_expires_at, road_type, ch_s, ch_e, sr_s, sr_e) =
        row.ok_or((StatusCode::NOT_FOUND, "video not found".to_string()))?;
    if !user.is_admin() {
        let holds_active_claim = claimed_by.as_deref() == Some(user.username.as_str())
            && lease_expires_at.map(|l| l > chrono::Utc::now()).unwrap_or(false);
        if !holds_active_claim {
            return Err((
                StatusCode::CONFLICT,
                "your claim has expired or is held by someone else — re-claim the video before pushing".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}")))?;

    // Adopt the doc's fps/frame_count/resolution BEFORE stamping. A raw (from-scratch)
    // NAS video has no fps in our DB — but the desktop decoded the real file, so its
    // fps is authoritative. Without this, per-annotation lat/lon/chainage can't be
    // computed (frame→time needs fps) and the geo silently wouldn't be written.
    // COALESCE(doc, existing) so a doc that omits a field never wipes it.
    if let Some(v) = doc.video.as_ref() {
        let _ = sqlx::query(
            r#"UPDATE videos SET fps=COALESCE($2, fps), frame_count=COALESCE($3, frame_count),
                 width=COALESCE($4, width), height=COALESCE($5, height) WHERE id=$1"#,
        )
        .bind(id)
        .bind(v.fps)
        .bind(v.frame_count.map(|x| x as i32))
        .bind(v.width.map(|x| x as i32))
        .bind(v.height.map(|x| x as i32))
        .execute(&s.db)
        .await;
    }

    // Re-stamp the server-authoritative road_type + chainage onto the stored doc —
    // whatever the worker's client sent (or dropped), the server's values win. Every
    // annotation gets its frame-exact lat/lon + chainage too (from the fps just set).
    let mut body = body;
    crate::stamp_road_meta(&mut body, &road_type, (ch_s, ch_e, sr_s, sr_e));
    if let Some(geo) = crate::chainage::video_geo(&s.db, id).await {
        crate::chainage::stamp_annotation_geo(&mut body, &geo);
    }

    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);

    // Replace the annotation set AND flip the video to verified in ONE transaction:
    // either the pushed annotations + the verified status both commit, or neither does.
    // This makes a push all-or-nothing — a verifier can never end up "verified" with a
    // half-written annotation set, and a failed push leaves the prior pass intact.
    let mut tx = s.db.begin().await.map_err(err)?;
    crate::ingest::replace_annotations(&mut tx, id, &doc)
        .await
        .map_err(err)?;

    // 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 is the ORIGINAL auto-imported baseline (set at ingest).
               -- Keep it fixed once set: verification/re-verification must NOT fold added
               -- annotations into it (otherwise "imported 200" would drift to 210). Only
               -- establish it from the claim-time count if it was never set.
               imported_count = COALESCE(NULLIF(videos.imported_count, 0), $8),
               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(&mut *tx)
    .await
    .map_err(err)?;
    tx.commit().await.map_err(err)?;

    // Record this pass's resulting annotation count so the dashboard can show a
    // per-pass breakdown (imported baseline → verify Δ → re-verify Δ …), not just a net.
    log_event(
        &s.db,
        id,
        &user.username,
        "push",
        json!({ "verify_time_ms": verify_time_ms, "annotation_count": ann_count }),
    )
    .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 })))
}