Mangalord/backend/src/api/admin/analysis.rs

//! Admin controls for the AI content-analysis worker.
//!
//! Two endpoints, both admin-only (`RequireAdmin`, cookie-only) and gated
//! on `AppState.analysis_enabled` (503 when the feature is off):
//!
//! * `POST /admin/analysis/reenqueue` — bulk backfill: enqueue
//!   `analyze_page` jobs for existing pages. Body `{ only_unanalyzed }`
//!   (default true) skips pages that already have a completed analysis.
//! * `POST /admin/pages/:id/analyze` — force re-analysis of a single page
//!   (re-runs even if already `done`).

use std::convert::Infallible;

use axum::extract::{Path, Query, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::routing::{get, post};
use axum::{Json, Router};
use futures_util::Stream;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tokio::sync::broadcast::error::RecvError;
use uuid::Uuid;

use crate::analysis::events::AnalysisEvent;
use crate::api::pagination::PagedResponse;
use crate::app::AppState;
use crate::auth::extractor::RequireAdmin;
use crate::domain::page_analysis::{
    AnalysisHistoryRow, AnalysisMetrics, ChapterCoverage, MangaCoverage, PageAnalysisDetail,
    PageStatusItem,
};
use crate::error::{AppError, AppResult};
use crate::repo;

pub fn routes() -> Router<AppState> {
    Router::new()
        .route("/admin/analysis/reenqueue", post(reenqueue))
        .route("/admin/pages/:id/analyze", post(analyze_page))
        // Coverage / inspection (admin-gated, but NOT analysis-enabled-gated
        // — an operator can browse coverage with the worker off).
        .route("/admin/analysis/mangas", get(coverage_mangas))
        .route("/admin/analysis/mangas/:id/chapters", get(coverage_chapters))
        .route("/admin/analysis/chapters/:id/pages", get(chapter_pages))
        .route("/admin/analysis/pages/:id", get(page_detail))
        .route("/admin/analysis/history", get(list_history))
        .route("/admin/analysis/metrics", get(metrics))
        .route("/admin/analysis/metrics/series", get(metrics_series))
        .route("/admin/analysis/status/stream", get(stream_status))
}

/// Live analysis events (SSE). Forwards each `AnalysisEvent` as a named
/// `analysis` event; on broadcast lag (a slow client) emits a `lagged`
/// event so the dashboard can refresh. EventSource sends the session
/// cookie, so `RequireAdmin` gates the stream like any other admin route.
async fn stream_status(
    State(state): State<AppState>,
    _admin: RequireAdmin,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let rx = state.analysis_events.subscribe();
    let stream = futures_util::stream::unfold(rx, |mut rx| async move {
        loop {
            match rx.recv().await {
                Ok(ev) => {
                    let event = Event::default()
                        .event("analysis")
                        .json_data(&ev)
                        .unwrap_or_else(|_| Event::default().comment("serialize error"));
                    return Some((Ok(event), rx));
                }
                Err(RecvError::Lagged(_)) => {
                    return Some((Ok(Event::default().event("lagged").data("")), rx));
                }
                Err(RecvError::Closed) => return None,
            }
        }
    });
    Sse::new(stream).keep_alive(KeepAlive::default())
}

#[derive(Debug, Deserialize)]
pub struct CoverageParams {
    #[serde(default)]
    pub search: Option<String>,
    #[serde(default = "default_coverage_limit")]
    pub limit: i64,
    #[serde(default)]
    pub offset: i64,
}

fn default_coverage_limit() -> i64 {
    25
}

async fn coverage_mangas(
    State(state): State<AppState>,
    _admin: RequireAdmin,
    Query(params): Query<CoverageParams>,
) -> AppResult<Json<PagedResponse<MangaCoverage>>> {
    let limit = params.limit.clamp(1, 100);
    let offset = params.offset.max(0);
    let search = params
        .search
        .as_deref()
        .map(str::trim)
        .filter(|s| !s.is_empty());
    let (items, total) =
        repo::page_analysis::manga_coverage(&state.db, search, limit, offset).await?;
    Ok(Json(PagedResponse::with_total(items, limit, offset, total)))
}

async fn coverage_chapters(
    State(state): State<AppState>,
    _admin: RequireAdmin,
    Path(manga_id): Path<Uuid>,
) -> AppResult<Json<ItemsResponse<ChapterCoverage>>> {
    if !repo::manga::exists(&state.db, manga_id).await? {
        return Err(AppError::NotFound);
    }
    let items = repo::page_analysis::chapter_coverage(&state.db, manga_id).await?;
    Ok(Json(ItemsResponse { items }))
}

async fn chapter_pages(
    State(state): State<AppState>,
    _admin: RequireAdmin,
    Path(chapter_id): Path<Uuid>,
) -> AppResult<Json<ItemsResponse<PageStatusItem>>> {
    let exists: bool =
        sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM chapters WHERE id = $1)")
            .bind(chapter_id)
            .fetch_one(&state.db)
            .await?;
    if !exists {
        return Err(AppError::NotFound);
    }
    let items = repo::page_analysis::chapter_page_status(&state.db, chapter_id).await?;
    Ok(Json(ItemsResponse { items }))
}

async fn page_detail(
    State(state): State<AppState>,
    _admin: RequireAdmin,
    Path(page_id): Path<Uuid>,
) -> AppResult<Json<PageAnalysisDetail>> {
    let detail = repo::page_analysis::page_detail(&state.db, page_id)
        .await?
        .ok_or(AppError::NotFound)?;
    Ok(Json(detail))
}

#[derive(Debug, Serialize)]
struct ItemsResponse<T> {
    items: Vec<T>,
}

/// Status/NSFW filters + pagination/search for the analysis history list.
#[derive(Debug, Deserialize, Default)]
pub struct HistoryParams {
    #[serde(default)]
    pub status: Option<String>,
    #[serde(default)]
    pub nsfw: bool,
    #[serde(default)]
    pub search: Option<String>,
    #[serde(default = "default_coverage_limit")]
    pub limit: i64,
    #[serde(default)]
    pub offset: i64,
}

async fn list_history(
    State(state): State<AppState>,
    _admin: RequireAdmin,
    Query(params): Query<HistoryParams>,
) -> AppResult<Json<PagedResponse<AnalysisHistoryRow>>> {
    let limit = params.limit.clamp(1, 100);
    let offset = params.offset.max(0);
    let status = params
        .status
        .as_deref()
        .map(str::trim)
        .filter(|s| !s.is_empty());
    let search = params
        .search
        .as_deref()
        .map(str::trim)
        .filter(|s| !s.is_empty());
    let (items, total) = repo::page_analysis::list_history(
        &state.db,
        repo::page_analysis::AnalysisHistoryFilter {
            status,
            nsfw_only: params.nsfw,
            search,
        },
        limit,
        offset,
    )
    .await?;
    Ok(Json(PagedResponse::with_total(items, limit, offset, total)))
}

/// `?days=` window for the metrics endpoints. `0` (or absent) = all time.
#[derive(Debug, Deserialize, Default)]
pub struct MetricsParams {
    #[serde(default)]
    pub days: i64,
}

/// Convert a `days` window param into a lower-bound timestamp. `<= 0` → all
/// time (`None`); otherwise `now - days` (capped at a year).
pub(super) fn window_since(days: i64) -> Option<chrono::DateTime<chrono::Utc>> {
    if days <= 0 {
        None
    } else {
        Some(chrono::Utc::now() - chrono::Duration::days(days.min(365)))
    }
}

// --- shared trend-series plumbing (crawler + analysis charts) ---------------

/// Query params for the bucketed trend-series endpoints. `bucket` is
/// optional; when absent it's derived from `days`.
#[derive(Debug, Deserialize, Default)]
pub struct SeriesParams {
    #[serde(default)]
    pub days: i64,
    #[serde(default)]
    pub bucket: Option<String>,
}

/// Envelope for a trend series.
#[derive(Debug, Serialize)]
pub struct SeriesResponse<T> {
    pub buckets: Vec<T>,
}

/// Resolve the `date_trunc` granularity: an explicit `bucket` (validated to
/// `hour`|`day`, else 400) wins; otherwise short windows bucket by hour and
/// longer ones by day so the point count stays chart-friendly.
pub(super) fn resolve_bucket(
    days: i64,
    explicit: Option<&str>,
) -> Result<crate::repo::crawl_metrics::Bucket, AppError> {
    use crate::repo::crawl_metrics::Bucket;
    match explicit {
        Some("hour") => Ok(Bucket::Hour),
        Some("day") => Ok(Bucket::Day),
        Some(other) => Err(AppError::InvalidInput(format!(
            "invalid bucket '{other}' (expected 'hour' or 'day')"
        ))),
        None => Ok(if days > 0 && days <= 2 {
            Bucket::Hour
        } else {
            Bucket::Day
        }),
    }
}

async fn metrics(
    State(state): State<AppState>,
    _admin: RequireAdmin,
    Query(params): Query<MetricsParams>,
) -> AppResult<Json<AnalysisMetrics>> {
    let since = window_since(params.days);
    let m = repo::page_analysis::analysis_metrics(&state.db, since).await?;
    Ok(Json(m))
}

async fn metrics_series(
    State(state): State<AppState>,
    _admin: RequireAdmin,
    Query(params): Query<SeriesParams>,
) -> AppResult<Json<SeriesResponse<crate::domain::crawl_metrics::MetricsBucket>>> {
    let bucket = resolve_bucket(params.days, params.bucket.as_deref())?;
    let buckets =
        repo::page_analysis::analysis_series(&state.db, bucket, window_since(params.days)).await?;
    Ok(Json(SeriesResponse { buckets }))
}

#[derive(Debug, Deserialize)]
pub struct ReenqueueBody {
    /// Skip pages that already have a `done` analysis row. Defaults to
    /// true so a routine backfill only touches the gaps; `false` forces a
    /// full re-analysis of in-scope pages.
    #[serde(default = "default_true")]
    pub only_unanalyzed: bool,
    /// Limit the re-enqueue to one manga (all its chapters' pages).
    #[serde(default)]
    pub manga_id: Option<Uuid>,
    /// Limit the re-enqueue to one chapter's pages. Mutually exclusive
    /// with `manga_id`.
    #[serde(default)]
    pub chapter_id: Option<Uuid>,
}

impl Default for ReenqueueBody {
    fn default() -> Self {
        Self { only_unanalyzed: true, manga_id: None, chapter_id: None }
    }
}

fn default_true() -> bool {
    true
}

#[derive(Debug, Serialize)]
pub struct ReenqueueResponse {
    pub enqueued: u64,
}

#[derive(Debug, Serialize)]
pub struct AnalyzePageResponse {
    pub enqueued: bool,
}

/// Reject the request with 503 when the analysis worker is disabled, so
/// an operator doesn't pile up jobs that nothing will ever drain.
fn ensure_enabled(state: &AppState) -> AppResult<()> {
    if state.analysis_enabled() {
        Ok(())
    } else {
        Err(AppError::ServiceUnavailable(
            "content-analysis worker is disabled (ANALYSIS_ENABLED=false)".into(),
        ))
    }
}

async fn reenqueue(
    State(state): State<AppState>,
    admin: RequireAdmin,
    body: Option<Json<ReenqueueBody>>,
) -> AppResult<Json<ReenqueueResponse>> {
    ensure_enabled(&state)?;
    let body = body.map(|b| b.0).unwrap_or_default();

    // Resolve the scope. manga_id and chapter_id are mutually exclusive;
    // an unknown target is a 404 rather than a silent zero-enqueue.
    if body.manga_id.is_some() && body.chapter_id.is_some() {
        return Err(AppError::ValidationFailed {
            message: "manga_id and chapter_id are mutually exclusive".into(),
            details: json!({ "scope": "pick at most one" }),
        });
    }
    let (scope, target_type, target_id) = if let Some(chapter_id) = body.chapter_id {
        let exists: bool =
            sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM chapters WHERE id = $1)")
                .bind(chapter_id)
                .fetch_one(&state.db)
                .await?;
        if !exists {
            return Err(AppError::NotFound);
        }
        (
            repo::page_analysis::ReenqueueScope::Chapter(chapter_id),
            "chapter",
            Some(chapter_id),
        )
    } else if let Some(manga_id) = body.manga_id {
        if !repo::manga::exists(&state.db, manga_id).await? {
            return Err(AppError::NotFound);
        }
        (
            repo::page_analysis::ReenqueueScope::Manga(manga_id),
            "manga",
            Some(manga_id),
        )
    } else {
        (repo::page_analysis::ReenqueueScope::All, "analysis", None)
    };

    let enqueued =
        repo::page_analysis::enqueue_pages(&state.db, scope, body.only_unanalyzed).await?;

    // Push a live event so connected dashboards mark the in-scope pages as
    // queued. Skip the no-op (nothing actually enqueued).
    if enqueued > 0 {
        state.analysis_events.publish(AnalysisEvent::Enqueued {
            count: enqueued,
            manga_id: body.manga_id,
            chapter_id: body.chapter_id,
        });
    }

    repo::admin_audit::insert(
        &state.db,
        admin.0.id,
        "analysis_reenqueue",
        target_type,
        target_id,
        json!({
            "only_unanalyzed": body.only_unanalyzed,
            "manga_id": body.manga_id,
            "chapter_id": body.chapter_id,
            "enqueued": enqueued,
        }),
    )
    .await?;

    Ok(Json(ReenqueueResponse { enqueued }))
}

async fn analyze_page(
    State(state): State<AppState>,
    admin: RequireAdmin,
    Path(page_id): Path<Uuid>,
) -> AppResult<Json<AnalyzePageResponse>> {
    ensure_enabled(&state)?;

    let exists: bool =
        sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pages WHERE id = $1)")
            .bind(page_id)
            .fetch_one(&state.db)
            .await?;
    if !exists {
        return Err(AppError::NotFound);
    }

    // Surface the actual outcome so the admin (and audit log) sees
    // whether the click actually moved anything. Before 0.87.11 the
    // partial unique index could silently swallow a force request
    // when a `force=false` job was already pending — the worker would
    // then pick up the non-force row, hit skip-if-done, ack done, and
    // the admin saw "queued for re-analysis" with no re-analysis.
    let outcome =
        repo::page_analysis::enqueue_for_page(&state.db, page_id, true).await?;

    let outcome_label = match outcome {
        repo::page_analysis::EnqueueForPageOutcome::Inserted => "inserted",
        repo::page_analysis::EnqueueForPageOutcome::UpgradedToForce => "upgraded_pending_to_force",
        repo::page_analysis::EnqueueForPageOutcome::AlreadyEnqueued => "already_force_enqueued",
    };
    repo::admin_audit::insert(
        &state.db,
        admin.0.id,
        "analysis_force_page",
        "page",
        Some(page_id),
        json!({ "force": true, "outcome": outcome_label }),
    )
    .await?;

    Ok(Json(AnalyzePageResponse { enqueued: true }))
}