diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 098e9a4..33b6948 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1470,7 +1470,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.65.0" +version = "0.66.0" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index e364e9f..aad8afc 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.65.0" +version = "0.66.0" edition = "2021" default-run = "mangalord" diff --git a/backend/src/api/admin/analysis.rs b/backend/src/api/admin/analysis.rs new file mode 100644 index 0000000..e31e041 --- /dev/null +++ b/backend/src/api/admin/analysis.rs @@ -0,0 +1,116 @@ +//! 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 axum::extract::{Path, State}; +use axum::routing::post; +use axum::{Json, Router}; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use uuid::Uuid; + +use crate::app::AppState; +use crate::auth::extractor::RequireAdmin; +use crate::error::{AppError, AppResult}; +use crate::repo; + +pub fn routes() -> Router { + Router::new() + .route("/admin/analysis/reenqueue", post(reenqueue)) + .route("/admin/pages/:id/analyze", post(analyze_page)) +} + +#[derive(Debug, Default, Deserialize)] +pub struct ReenqueueBody { + /// Skip pages that already have a `done` analysis row. Defaults to + /// true so a routine backfill only touches the gaps. + #[serde(default = "default_true")] + pub only_unanalyzed: bool, +} + +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, + admin: RequireAdmin, + body: Option>, +) -> AppResult> { + ensure_enabled(&state)?; + let only_unanalyzed = body.map(|b| b.0.only_unanalyzed).unwrap_or(true); + + let enqueued = repo::page_analysis::enqueue_all_pages(&state.db, only_unanalyzed).await?; + + repo::admin_audit::insert( + &state.db, + admin.0.id, + "analysis_reenqueue", + "analysis", + None, + json!({ "only_unanalyzed": only_unanalyzed, "enqueued": enqueued }), + ) + .await?; + + Ok(Json(ReenqueueResponse { enqueued })) +} + +async fn analyze_page( + State(state): State, + admin: RequireAdmin, + Path(page_id): Path, +) -> AppResult> { + 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 })) +} diff --git a/backend/src/api/admin/mod.rs b/backend/src/api/admin/mod.rs index 2048cf1..3cdb271 100644 --- a/backend/src/api/admin/mod.rs +++ b/backend/src/api/admin/mod.rs @@ -4,6 +4,7 @@ //! bot/API tokens cannot reach admin routes (see //! `crate::auth::extractor::RequireAdmin`). +pub mod analysis; pub mod crawler; pub mod mangas; pub mod resync; @@ -21,4 +22,5 @@ pub fn routes() -> Router { .merge(resync::routes()) .merge(system::routes()) .merge(crawler::routes()) + .merge(analysis::routes()) } diff --git a/backend/src/api/chapters.rs b/backend/src/api/chapters.rs index 26578f9..fdcb67a 100644 --- a/backend/src/api/chapters.rs +++ b/backend/src/api/chapters.rs @@ -137,6 +137,7 @@ async fn create( ) .await?; + let mut page_ids: Vec = Vec::with_capacity(pages.len()); for (idx, page) in pages.iter().enumerate() { let page_number = (idx + 1) as i32; let nnnn = format!("{:04}", page_number); @@ -145,7 +146,9 @@ async fn create( manga_id, chapter.id, nnnn, page.ext ); state.storage.put(&key, &page.bytes).await?; - repo::page::create(&mut *tx, chapter.id, page_number, &key, page.mime).await?; + let created = + repo::page::create(&mut *tx, chapter.id, page_number, &key, page.mime).await?; + page_ids.push(created.id); } let page_count = pages.len() as i32; @@ -154,6 +157,20 @@ async fn create( tx.commit().await?; + // Enqueue AI content-analysis for each new page. Done after commit so a + // rolled-back upload never leaves jobs pointing at nonexistent pages; a + // failed enqueue is logged but doesn't fail the upload (the admin + // re-enqueue endpoint can backfill). + if state.analysis_enabled { + for page_id in page_ids { + if let Err(e) = + repo::page_analysis::enqueue_for_page(&state.db, page_id, false).await + { + tracing::warn!(%page_id, error = %e, "failed to enqueue page analysis"); + } + } + } + Ok((StatusCode::CREATED, Json(chapter))) } diff --git a/backend/src/app.rs b/backend/src/app.rs index b5250e6..b2f3856 100644 --- a/backend/src/app.rs +++ b/backend/src/app.rs @@ -56,6 +56,11 @@ pub struct AppState { /// `Arc` keeps the clone cheap. Empty list → check is skipped /// (operator opt-out documented in `.env.example`). pub admin_allowed_origins: Arc>, + /// When `true`, page-create paths (upload + crawler) enqueue + /// `analyze_page` jobs and the admin re-enqueue endpoints are active. + /// Mirrors `config.analysis.enabled`; defaults to `false` so the + /// feature is opt-in and the test harness inherits a no-op gate. + pub analysis_enabled: bool, } /// Shared handle the admin crawler endpoints use to observe and control @@ -113,7 +118,13 @@ pub async fn build(config: Config) -> anyhow::Result { let storage: Arc = Arc::new(LocalStorage::new(config.storage_dir.clone())); let (daemon, resync, crawler) = if config.crawler.daemon_enabled { - let spawned = spawn_crawler_daemon(db.clone(), Arc::clone(&storage), &config.crawler).await?; + let spawned = spawn_crawler_daemon( + db.clone(), + Arc::clone(&storage), + &config.crawler, + config.analysis.enabled, + ) + .await?; (Some(spawned.handle), Some(spawned.resync), Some(spawned.crawler)) } else { tracing::info!("crawler daemon disabled (CRAWLER_DAEMON=false)"); @@ -130,6 +141,7 @@ pub async fn build(config: Config) -> anyhow::Result { resync, crawler, admin_allowed_origins: Arc::new(config.admin_allowed_origins.clone()), + analysis_enabled: config.analysis.enabled, }; let router = router(state).layer(cors_layer(&config.cors_allowed_origins)); Ok(AppHandle { router, daemon }) @@ -149,6 +161,7 @@ async fn spawn_crawler_daemon( db: PgPool, storage: Arc, cfg: &CrawlerConfig, + analysis_enabled: bool, ) -> anyhow::Result { // Reqwest client with a shared cookie jar so CDN image fetches include // PHPSESSID. The same `Arc` is held by the SessionController, so a @@ -288,6 +301,7 @@ async fn spawn_crawler_daemon( rate: Arc::clone(&rate), download_allowlist: cfg.download_allowlist.clone(), max_image_bytes: cfg.max_image_bytes, + analysis_enabled, transient_failures: Arc::new(AtomicU32::new(0)), restart_threshold: cfg.browser_restart_threshold, drain_deadline: cfg.job_timeout, @@ -446,6 +460,9 @@ struct RealChapterDispatcher { rate: Arc, download_allowlist: DownloadAllowlist, max_image_bytes: usize, + /// Enqueue `analyze_page` jobs for freshly-crawled pages. Mirrors + /// `config.analysis.enabled`. + analysis_enabled: bool, /// Consecutive transient chapter failures; resets on any success. /// Drives the automatic coordinated browser restart. transient_failures: Arc, @@ -501,6 +518,7 @@ impl ChapterDispatcher for RealChapterDispatcher { self.max_image_bytes, self.tor.as_deref(), Some(&self.status), + self.analysis_enabled, ) .await; drop(lease); diff --git a/backend/src/bin/crawler.rs b/backend/src/bin/crawler.rs index f36b3b8..0128788 100644 --- a/backend/src/bin/crawler.rs +++ b/backend/src/bin/crawler.rs @@ -419,6 +419,8 @@ async fn sync_bookmarked_chapter_content( tor.as_deref(), // CLI one-shot — no live status surface. None, + // Standalone CLI doesn't drive the analysis worker. + false, ) .await; drop(lease); diff --git a/backend/src/config.rs b/backend/src/config.rs index 37effaf..d7224dd 100644 --- a/backend/src/config.rs +++ b/backend/src/config.rs @@ -66,6 +66,31 @@ impl Default for UploadConfig { } } +/// AI content-analysis worker configuration. Phase 2 only needs the +/// `enabled` gate (so page-create paths know whether to enqueue analysis +/// jobs); later phases extend this with the vision endpoint, model, and +/// worker knobs. +#[derive(Clone, Debug)] +pub struct AnalysisConfig { + /// Master switch (`ANALYSIS_ENABLED`). When `false`, no analysis jobs + /// are enqueued and no worker runs. Defaults to `false`. + pub enabled: bool, +} + +impl Default for AnalysisConfig { + fn default() -> Self { + Self { enabled: false } + } +} + +impl AnalysisConfig { + pub fn from_env() -> Self { + Self { + enabled: env_bool("ANALYSIS_ENABLED", false), + } + } +} + #[derive(Clone, Debug)] pub struct Config { pub database_url: String, @@ -89,6 +114,7 @@ pub struct Config { /// the check. pub admin_allowed_origins: Vec, pub crawler: CrawlerConfig, + pub analysis: AnalysisConfig, /// `(username, password)` for the admin user provisioned at startup /// when both `ADMIN_USERNAME` and `ADMIN_PASSWORD` are set. `None` /// skips the bootstrap entirely. See `repo::user::bootstrap_admin` @@ -245,6 +271,7 @@ impl Config { }) .unwrap_or_default(), crawler: CrawlerConfig::from_env()?, + analysis: AnalysisConfig::from_env(), admin_bootstrap: admin_bootstrap_from_env(), }) } diff --git a/backend/src/crawler/content.rs b/backend/src/crawler/content.rs index 13ffa88..f6a89be 100644 --- a/backend/src/crawler/content.rs +++ b/backend/src/crawler/content.rs @@ -215,6 +215,10 @@ pub async fn sync_chapter_content( // dispatcher passes the shared handle (the chapter has already been // registered via `begin_chapter`); the CLI / admin resync pass `None`. progress: Option<&crate::crawler::status::StatusHandle>, + // When `true`, enqueue an `analyze_page` job per persisted page. The + // daemon dispatcher passes the configured flag; CLI / resync pass + // `false`. + enqueue_analysis: bool, ) -> anyhow::Result { // Skip if already fetched, unless caller explicitly forces. if !force_refetch { @@ -316,7 +320,7 @@ pub async fn sync_chapter_content( // Short transaction: page rows + page_count only, no network I/O. On // failure, roll back the stored bytes so the chapter stays at // page_count=0 and is retried cleanly next run. - if let Err(e) = persist_pages(db, chapter_id, &stored).await { + if let Err(e) = persist_pages(db, chapter_id, &stored, enqueue_analysis).await { cleanup_orphans(storage, &written_keys).await; return Err(e); } @@ -450,23 +454,27 @@ pub(crate) async fn persist_pages( db: &PgPool, chapter_id: Uuid, stored: &[StoredPage], + enqueue_analysis: bool, ) -> anyhow::Result<()> { let mut tx = db.begin().await.context("open chapter sync tx")?; + let mut page_ids: Vec = Vec::with_capacity(stored.len()); for page in stored { - sqlx::query( + let (id,): (Uuid,) = sqlx::query_as( "INSERT INTO pages (chapter_id, page_number, storage_key, content_type) VALUES ($1, $2, $3, $4) ON CONFLICT (chapter_id, page_number) DO UPDATE SET storage_key = EXCLUDED.storage_key, - content_type = EXCLUDED.content_type", + content_type = EXCLUDED.content_type + RETURNING id", ) .bind(chapter_id) .bind(page.page_number) .bind(&page.storage_key) .bind(&page.content_type) - .execute(&mut *tx) + .fetch_one(&mut *tx) .await .with_context(|| format!("insert page row {}", page.page_number))?; + page_ids.push(id); } sqlx::query("UPDATE chapters SET page_count = $1 WHERE id = $2") .bind(stored.len() as i32) @@ -475,6 +483,20 @@ pub(crate) async fn persist_pages( .await .context("update page_count")?; tx.commit().await.context("commit chapter sync")?; + + // Enqueue AI content-analysis for the crawled pages once their rows + // are committed. Best-effort: a failed enqueue is logged, never fatal + // (the admin re-enqueue endpoint can backfill). Gated by the caller so + // jobs don't pile up when the analysis worker is disabled. + if enqueue_analysis { + for page_id in page_ids { + if let Err(e) = + crate::repo::page_analysis::enqueue_for_page(db, page_id, false).await + { + tracing::warn!(%page_id, error = %e, "failed to enqueue page analysis after crawl"); + } + } + } Ok(()) } @@ -561,7 +583,7 @@ mod tests { content_type: "image/jpeg".into(), }, ]; - persist_pages(&pool, chapter_id, &stored).await.unwrap(); + persist_pages(&pool, chapter_id, &stored, false).await.unwrap(); let page_count: i32 = sqlx::query_scalar("SELECT page_count FROM chapters WHERE id = $1") @@ -579,7 +601,7 @@ mod tests { assert_eq!(rows, 2); // Idempotent re-run (force refetch path): same rows, page_count stable. - persist_pages(&pool, chapter_id, &stored).await.unwrap(); + persist_pages(&pool, chapter_id, &stored, false).await.unwrap(); let rows2: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pages WHERE chapter_id = $1") .bind(chapter_id) @@ -589,6 +611,48 @@ mod tests { assert_eq!(rows2, 2, "re-run is idempotent via ON CONFLICT"); } + #[sqlx::test(migrations = "./migrations")] + async fn persist_pages_enqueues_analysis_only_when_flag_set(pool: PgPool) { + let manga_id = Uuid::new_v4(); + let chapter_id = Uuid::new_v4(); + sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, 'T')") + .bind(manga_id) + .execute(&pool) + .await + .unwrap(); + sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)") + .bind(chapter_id) + .bind(manga_id) + .execute(&pool) + .await + .unwrap(); + let stored = vec![StoredPage { + page_number: 1, + storage_key: "k/0001.jpg".into(), + content_type: "image/jpeg".into(), + }]; + + // Flag off: no analyze_page jobs. + persist_pages(&pool, chapter_id, &stored, false).await.unwrap(); + let jobs_off: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(jobs_off, 0, "flag off must not enqueue analysis"); + + // Flag on: one analyze_page job for the upserted page. + persist_pages(&pool, chapter_id, &stored, true).await.unwrap(); + let jobs_on: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(jobs_on, 1, "flag on enqueues one job per persisted page"); + } + #[test] fn parse_chapter_pages_skips_loader_and_sorts_by_id() { // Loader image, two real pages out of order, and one with no id. diff --git a/backend/src/crawler/resync.rs b/backend/src/crawler/resync.rs index f964c93..2b22740 100644 --- a/backend/src/crawler/resync.rs +++ b/backend/src/crawler/resync.rs @@ -259,6 +259,9 @@ impl ResyncService for RealResyncService { self.tor.as_deref(), // Admin resync isn't a daemon worker slot — no live status. None, + // Resync re-fetches existing pages (same ids); analysis isn't + // re-enqueued here — use the admin force re-analyze endpoint. + false, ) .await; drop(lease); diff --git a/backend/src/repo/page_analysis.rs b/backend/src/repo/page_analysis.rs index a406a0d..62169aa 100644 --- a/backend/src/repo/page_analysis.rs +++ b/backend/src/repo/page_analysis.rs @@ -31,6 +31,34 @@ pub async fn enqueue_for_page(pool: &PgPool, page_id: Uuid, force: bool) -> AppR Ok(()) } +/// Bulk-enqueue `analyze_page` jobs for existing pages — the admin +/// backfill path for content uploaded/crawled before analysis was +/// enabled. When `only_unanalyzed` is true, pages that already have a +/// `done` analysis row are skipped. Pages with a pending/running +/// `analyze_page` job are always skipped so repeated calls don't pile up +/// duplicates. Returns the number of jobs enqueued. +pub async fn enqueue_all_pages(pool: &PgPool, only_unanalyzed: bool) -> AppResult { + let result = sqlx::query( + r#" + INSERT INTO crawler_jobs (payload) + SELECT jsonb_build_object('kind', 'analyze_page', 'page_id', p.id, 'force', false) + FROM pages p + WHERE ($1 = false OR NOT EXISTS ( + SELECT 1 FROM page_analysis pa + WHERE pa.page_id = p.id AND pa.status = 'done')) + AND NOT EXISTS ( + SELECT 1 FROM crawler_jobs j + WHERE j.payload->>'kind' = 'analyze_page' + AND j.payload->>'page_id' = p.id::text + AND j.state IN ('pending', 'running')) + "#, + ) + .bind(only_unanalyzed) + .execute(pool) + .await?; + Ok(result.rows_affected()) +} + /// Load a page's analysis row, if it has one. pub async fn load(pool: &PgPool, page_id: Uuid) -> AppResult> { let row = sqlx::query_as::<_, PageAnalysis>( diff --git a/backend/tests/api_admin_analysis.rs b/backend/tests/api_admin_analysis.rs new file mode 100644 index 0000000..2dc77fc --- /dev/null +++ b/backend/tests/api_admin_analysis.rs @@ -0,0 +1,269 @@ +//! Integration tests for the AI-analysis enqueue wiring: chapter upload +//! enqueues `analyze_page` jobs, and the admin backfill / force-reanalyze +//! endpoints. All gated on `analysis_enabled` (the `harness_with_analysis` +//! variant turns it on; the default harness leaves it off → 503). + +mod common; + +use axum::http::StatusCode; +use axum::Router; +use serde_json::json; +use sqlx::PgPool; +use tower::ServiceExt; + +use mangalord::repo; + +async fn seed_admin(pool: &PgPool, app: &Router) -> String { + let (username, cookie) = common::register_user(app).await; + let u = repo::user::find_by_username(pool, &username) + .await + .unwrap() + .unwrap(); + repo::user::set_is_admin_unchecked(pool, u.id, true) + .await + .unwrap(); + cookie +} + +/// Seed a manga → chapter with `n` pages directly in the DB, returning the +/// page ids. Used by the backfill tests (no upload round-trip needed). +async fn seed_pages(pool: &PgPool, n: i32) -> Vec { + let manga_id: uuid::Uuid = + sqlx::query_scalar("INSERT INTO mangas (title) VALUES ('M') RETURNING id") + .fetch_one(pool) + .await + .unwrap(); + let chapter_id: uuid::Uuid = sqlx::query_scalar( + "INSERT INTO chapters (manga_id, number) VALUES ($1, 1) RETURNING id", + ) + .bind(manga_id) + .fetch_one(pool) + .await + .unwrap(); + let mut ids = Vec::new(); + for i in 1..=n { + let id: uuid::Uuid = sqlx::query_scalar( + "INSERT INTO pages (chapter_id, page_number, storage_key, content_type) \ + VALUES ($1, $2, $3, 'image/png') RETURNING id", + ) + .bind(chapter_id) + .bind(i) + .bind(format!("k/{i}.png")) + .fetch_one(pool) + .await + .unwrap(); + ids.push(id); + } + ids +} + +async fn analyze_job_count(pool: &PgPool) -> i64 { + sqlx::query_scalar( + "SELECT COUNT(*) FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'", + ) + .fetch_one(pool) + .await + .unwrap() +} + +#[sqlx::test(migrations = "./migrations")] +async fn chapter_upload_enqueues_one_analysis_job_per_page(pool: PgPool) { + let h = common::harness_with_analysis(pool.clone()); + let (_user, cookie) = common::register_user(&h.app).await; + let manga_id = common::seed_manga_via_api(&h.app, &cookie, "Uploaded").await; + + let resp = h + .app + .clone() + .oneshot(common::post_multipart_with_cookie( + &format!("/api/v1/mangas/{manga_id}/chapters"), + common::MultipartBuilder::new() + .add_json("metadata", json!({ "number": 1, "title": "Ch 1" })) + .add_file("page", "p1.png", "image/png", &common::fake_png_bytes()) + .add_file("page", "p2.png", "image/png", &common::fake_png_bytes()), + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + + assert_eq!(analyze_job_count(&pool).await, 2); +} + +#[sqlx::test(migrations = "./migrations")] +async fn chapter_upload_does_not_enqueue_when_analysis_disabled(pool: PgPool) { + // Default harness has analysis_enabled = false. + let h = common::harness(pool.clone()); + let (_user, cookie) = common::register_user(&h.app).await; + let manga_id = common::seed_manga_via_api(&h.app, &cookie, "Uploaded").await; + + let resp = h + .app + .clone() + .oneshot(common::post_multipart_with_cookie( + &format!("/api/v1/mangas/{manga_id}/chapters"), + common::MultipartBuilder::new() + .add_json("metadata", json!({ "number": 1 })) + .add_file("page", "p1.png", "image/png", &common::fake_png_bytes()), + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + assert_eq!(analyze_job_count(&pool).await, 0); +} + +#[sqlx::test(migrations = "./migrations")] +async fn reenqueue_is_forbidden_for_non_admin(pool: PgPool) { + let h = common::harness_with_analysis(pool.clone()); + let (_user, cookie) = common::register_user(&h.app).await; + + let resp = h + .app + .clone() + .oneshot(common::post_json_with_cookie( + "/api/v1/admin/analysis/reenqueue", + json!({}), + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::FORBIDDEN); +} + +#[sqlx::test(migrations = "./migrations")] +async fn reenqueue_returns_503_when_analysis_disabled(pool: PgPool) { + let h = common::harness(pool.clone()); // analysis disabled + let cookie = seed_admin(&pool, &h.app).await; + + let resp = h + .app + .clone() + .oneshot(common::post_json_with_cookie( + "/api/v1/admin/analysis/reenqueue", + json!({}), + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); +} + +#[sqlx::test(migrations = "./migrations")] +async fn reenqueue_backfills_existing_pages(pool: PgPool) { + let h = common::harness_with_analysis(pool.clone()); + let cookie = seed_admin(&pool, &h.app).await; + seed_pages(&pool, 3).await; + + let resp = h + .app + .clone() + .oneshot(common::post_json_with_cookie( + "/api/v1/admin/analysis/reenqueue", + json!({ "only_unanalyzed": true }), + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = common::body_json(resp).await; + assert_eq!(body["enqueued"], 3); + assert_eq!(analyze_job_count(&pool).await, 3); + + // Idempotent: a second call enqueues nothing (pending jobs already exist). + let resp2 = h + .app + .clone() + .oneshot(common::post_json_with_cookie( + "/api/v1/admin/analysis/reenqueue", + json!({ "only_unanalyzed": true }), + &cookie, + )) + .await + .unwrap(); + let body2 = common::body_json(resp2).await; + assert_eq!(body2["enqueued"], 0); + assert_eq!(analyze_job_count(&pool).await, 3); +} + +#[sqlx::test(migrations = "./migrations")] +async fn reenqueue_only_unanalyzed_skips_done_pages(pool: PgPool) { + use mangalord::domain::page_analysis::{SafetyFlag, VisionAnalysis}; + + let h = common::harness_with_analysis(pool.clone()); + let cookie = seed_admin(&pool, &h.app).await; + let ids = seed_pages(&pool, 2).await; + + // Mark the first page as already analyzed. + repo::page_analysis::persist_analysis( + &pool, + ids[0], + &VisionAnalysis { + ocr_results: vec![], + tagging_results: vec![], + scene_description: String::new(), + safety_flag: SafetyFlag::default(), + }, + "m", + ) + .await + .unwrap(); + + let resp = h + .app + .clone() + .oneshot(common::post_json_with_cookie( + "/api/v1/admin/analysis/reenqueue", + json!({ "only_unanalyzed": true }), + &cookie, + )) + .await + .unwrap(); + let body = common::body_json(resp).await; + assert_eq!(body["enqueued"], 1, "only the un-analyzed page is enqueued"); +} + +#[sqlx::test(migrations = "./migrations")] +async fn force_analyze_page_enqueues_with_force_flag(pool: PgPool) { + let h = common::harness_with_analysis(pool.clone()); + let cookie = seed_admin(&pool, &h.app).await; + let ids = seed_pages(&pool, 1).await; + + let resp = h + .app + .clone() + .oneshot(common::post_json_with_cookie( + &format!("/api/v1/admin/pages/{}/analyze", ids[0]), + json!({}), + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let payload: serde_json::Value = sqlx::query_scalar( + "SELECT payload FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(payload["force"].as_bool(), Some(true)); +} + +#[sqlx::test(migrations = "./migrations")] +async fn force_analyze_unknown_page_is_404(pool: PgPool) { + let h = common::harness_with_analysis(pool.clone()); + let cookie = seed_admin(&pool, &h.app).await; + + let resp = h + .app + .clone() + .oneshot(common::post_json_with_cookie( + &format!("/api/v1/admin/pages/{}/analyze", uuid::Uuid::new_v4()), + json!({}), + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); +} diff --git a/backend/tests/api_admin_role.rs b/backend/tests/api_admin_role.rs index 05d51ed..5d5dc9b 100644 --- a/backend/tests/api_admin_role.rs +++ b/backend/tests/api_admin_role.rs @@ -52,6 +52,7 @@ fn admin_test_router(pool: PgPool) -> (Router, TempDir) { resync: None, crawler: None, admin_allowed_origins: Arc::new(Vec::new()), + analysis_enabled: false, }; let app = Router::new() .nest("/api/v1", api::routes()) diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 78e75e3..c5fb4ef 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -82,6 +82,7 @@ fn harness_with_auth_config( // Empty allowlist = CSRF check skipped. The CSRF-specific test // harness `harness_with_admin_origins` overrides this. admin_allowed_origins: Arc::new(Vec::new()), + analysis_enabled: false, }; Harness { app: router(state), _storage_dir: storage_dir } } @@ -158,6 +159,38 @@ pub fn harness_with_resync( resync: Some(resync), crawler: None, admin_allowed_origins: Arc::new(Vec::new()), + analysis_enabled: false, + }; + Harness { + app: router(state), + _storage_dir: storage_dir, + } +} + +/// Like [`harness`] but flips `analysis_enabled` on so the page-create +/// paths enqueue `analyze_page` jobs and the admin analysis endpoints are +/// active (rather than returning 503). +pub fn harness_with_analysis(pool: PgPool) -> Harness { + let storage_dir = tempfile::tempdir().expect("tempdir"); + let storage = Arc::new(LocalStorage::new(storage_dir.path())); + let auth = AuthConfig { + cookie_secure: false, + ..AuthConfig::default() + }; + let auth_limiter = Arc::new(AuthRateLimiter::new(auth.rate_limit)); + let state = AppState { + db: pool, + storage, + auth, + upload: UploadConfig { + max_request_bytes: 4 * 1024 * 1024, + max_file_bytes: 256 * 1024, + }, + auth_limiter, + resync: None, + crawler: None, + admin_allowed_origins: Arc::new(Vec::new()), + analysis_enabled: true, }; Harness { app: router(state), @@ -187,6 +220,7 @@ pub fn harness_with_admin_origins(pool: PgPool, origins: Vec) -> Harness resync: None, crawler: None, admin_allowed_origins: Arc::new(origins), + analysis_enabled: false, }; Harness { app: router(state), diff --git a/frontend/package.json b/frontend/package.json index b75f678..c5d442d 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.65.0", + "version": "0.66.0", "private": true, "type": "module", "scripts": {