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::{
    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/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>,
}

#[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);
    }

    repo::page_analysis::enqueue_for_page(&state.db, page_id, true).await?;

    repo::admin_audit::insert(
        &state.db,
        admin.0.id,
        "analysis_force_page",
        "page",
        Some(page_id),
        json!({ "force": true }),
    )
    .await?;

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