fabi/Mangalord
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(analysis): enqueue page-analysis on upload/crawl + admin backfill endpoints

Browse Source 
Wires the analyze_page job kind into the page-create paths and adds admin
controls, all gated on a new ANALYSIS_ENABLED flag (AppState.analysis_enabled,
config::AnalysisConfig):

- Chapter upload (api::chapters) enqueues one analyze_page job per page
  after the tx commits (best-effort; failures logged, never fatal).
- Crawler content sync (persist_pages → RETURNING id) enqueues per
  persisted page when the daemon dispatcher's analysis_enabled flag is set;
  threaded through sync_chapter_content (CLI/resync pass false).
- repo::page_analysis::enqueue_all_pages: bulk backfill via INSERT..SELECT,
  skipping done pages (only_unanalyzed) and existing pending jobs.
- New admin endpoints (RequireAdmin, 503 when disabled, audited):
  POST /admin/analysis/reenqueue and POST /admin/pages/:id/analyze (force).

Tests: upload enqueues N jobs / no-op when disabled; crawler persist_pages
enqueue gate; admin reenqueue (backfill, idempotent, only-unanalyzed, 503,
non-admin 403) and force-analyze (force flag, 404).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-13 18:39:03 +02:00
parent 63e1aa5484
commit 2bbf1595ff
15 changed files with 592 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
backend/src/api/admin/analysis.rs Normal file
View File

@@ -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<AppState> {
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<AppState>,
admin: RequireAdmin,
body: Option<Json<ReenqueueBody>>,
) -> AppResult<Json<ReenqueueResponse>> {
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<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 }))
}

2
backend/src/api/admin/mod.rs
View File

@@ -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<AppState> {
.merge(resync::routes())
.merge(system::routes())
.merge(crawler::routes())
.merge(analysis::routes())
}

19
backend/src/api/chapters.rs
View File

@@ -137,6 +137,7 @@ async fn create(
)
.await?;
let mut page_ids: Vec<Uuid> = 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)))
}

20
backend/src/app.rs
View File

@@ -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<Vec<String>>,
/// 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<AppHandle> {
let storage: Arc<dyn Storage> = 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<AppHandle> {
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<dyn Storage>,
cfg: &CrawlerConfig,
analysis_enabled: bool,
) -> anyhow::Result<SpawnedDaemon> {
// Reqwest client with a shared cookie jar so CDN image fetches include
// PHPSESSID. The same `Arc<Jar>` 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<HostRateLimiters>,
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<std::sync::atomic::AtomicU32>,
@@ -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);

2
backend/src/bin/crawler.rs
View File

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

27
backend/src/config.rs
View File

@@ -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<String>,
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(),
})
}

76
backend/src/crawler/content.rs
View File

@@ -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<SyncOutcome> {
// 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<Uuid> = 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.

3
backend/src/crawler/resync.rs
View File

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

28
backend/src/repo/page_analysis.rs
View File

@@ -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<u64> {
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<Option<PageAnalysis>> {
let row = sqlx::query_as::<_, PageAnalysis>(