From d0cd31c9a7bc5c81eef415ff1b738e46c9e33b1c Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 13 Jun 2026 21:06:20 +0200 Subject: [PATCH] feat(analysis): live SSE event stream for the admin dashboard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Broadcasts analysis progress so the dashboard updates live: - analysis::events: AnalysisEvents broadcaster + AnalysisEvent (Enqueued / Started / Completed / Failed), carrying the manga/chapter/page breadcrumb. - The worker daemon resolves each page's breadcrumb (repo::page::locate) and publishes Started before dispatch and Completed/Failed after. - The admin reenqueue publishes Enqueued (scoped by manga/chapter). - GET /v1/admin/analysis/status/stream — SSE (RequireAdmin) forwarding each event as a named `analysis` frame; broadcast lag emits a `lagged` frame. AppState carries the always-present events bus. Tests: worker publishes started+completed (with breadcrumb) and failed; SSE route is admin-gated (403 non-admin) and returns text/event-stream. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/Cargo.lock | 2 +- backend/Cargo.toml | 2 +- backend/src/analysis/daemon.rs | 28 +++++++++ backend/src/analysis/events.rs | 78 +++++++++++++++++++++++++ backend/src/analysis/mod.rs | 1 + backend/src/api/admin/analysis.rs | 46 +++++++++++++++ backend/src/app.rs | 11 ++++ backend/src/repo/page.rs | 16 ++++++ backend/tests/analysis_worker.rs | 89 +++++++++++++++++++++++++++++ backend/tests/api_admin_analysis.rs | 37 ++++++++++++ backend/tests/api_admin_role.rs | 1 + backend/tests/common/mod.rs | 4 ++ frontend/package.json | 2 +- 13 files changed, 314 insertions(+), 3 deletions(-) create mode 100644 backend/src/analysis/events.rs diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 468ad59..c622309 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.76.0" +version = "0.77.0" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 1ab6695..446d0b6 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.76.0" +version = "0.77.0" edition = "2021" default-run = "mangalord" diff --git a/backend/src/analysis/daemon.rs b/backend/src/analysis/daemon.rs index 87a00a9..a5624df 100644 --- a/backend/src/analysis/daemon.rs +++ b/backend/src/analysis/daemon.rs @@ -22,6 +22,7 @@ use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; use uuid::Uuid; +use crate::analysis::events::{AnalysisEvent, AnalysisEvents}; use crate::analysis::vision::VisionClient; use crate::crawler::jobs::{self, JobPayload, Lease, KIND_ANALYZE_PAGE}; use crate::repo; @@ -43,6 +44,8 @@ pub struct AnalysisDaemonConfig { pub dispatcher: Arc, pub workers: usize, pub job_timeout: Duration, + /// Live-event sink (Started/Completed/Failed) for admin SSE. + pub events: Arc, } pub struct AnalysisDaemonHandle { @@ -71,6 +74,7 @@ pub fn spawn( cancel: cancel.clone(), dispatcher: Arc::clone(&cfg.dispatcher), job_timeout: cfg.job_timeout, + events: Arc::clone(&cfg.events), id, }; join.spawn(async move { ctx.run().await }); @@ -83,6 +87,7 @@ struct WorkerContext { cancel: CancellationToken, dispatcher: Arc, job_timeout: Duration, + events: Arc, id: usize, } @@ -143,6 +148,20 @@ impl WorkerContext { } } + // Resolve the page breadcrumb once so live events carry the + // manga/chapter/number the dashboard keys on. A missing breadcrumb + // (page deleted) just suppresses events — the dispatch still runs + // and acks normally. + let breadcrumb = repo::page::locate(&self.pool, page_id).await.ok().flatten(); + if let Some((manga_id, chapter_id, page_number)) = breadcrumb { + self.events.publish(AnalysisEvent::Started { + page_id, + manga_id, + chapter_id, + page_number, + }); + } + // Heartbeat the lease while the vision call runs. let heartbeat = { let hb_pool = self.pool.clone(); @@ -171,6 +190,15 @@ impl WorkerContext { Ok(Ok(Ok(()))) => Ok(()), }; + if let Some((manga_id, chapter_id, page_number)) = breadcrumb { + let event = if result.is_ok() { + AnalysisEvent::Completed { page_id, manga_id, chapter_id, page_number } + } else { + AnalysisEvent::Failed { page_id, manga_id, chapter_id, page_number } + }; + self.events.publish(event); + } + match result { Ok(()) => { let _ = jobs::ack_done(&self.pool, lease.id).await; diff --git a/backend/src/analysis/events.rs b/backend/src/analysis/events.rs new file mode 100644 index 0000000..632c5d4 --- /dev/null +++ b/backend/src/analysis/events.rs @@ -0,0 +1,78 @@ +//! Live analysis events broadcast to admin SSE subscribers. +//! +//! A thin wrapper over a `tokio::sync::broadcast` channel. The worker +//! publishes per-page progress (`Started`/`Completed`/`Failed`) and the +//! admin enqueue path publishes `Enqueued`; the +//! `GET /v1/admin/analysis/status/stream` SSE endpoint forwards each event. +//! Discrete events (rather than the crawler's snapshot-on-change) map +//! directly onto "this page/chapter/manga just changed state", which the +//! dashboard applies incrementally. + +use serde::Serialize; +use tokio::sync::broadcast; +use uuid::Uuid; + +/// One live analysis event. Serialized with a `kind` discriminator so the +/// frontend can switch on it. +#[derive(Clone, Debug, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum AnalysisEvent { + /// A bulk re-enqueue happened. `manga_id` / `chapter_id` scope it (both + /// `None` = whole library). The UI optimistically marks in-scope pages + /// as queued. + Enqueued { + count: u64, + manga_id: Option, + chapter_id: Option, + }, + /// The worker began analyzing a page. + Started { + page_id: Uuid, + manga_id: Uuid, + chapter_id: Uuid, + page_number: i32, + }, + /// The worker finished a page successfully. + Completed { + page_id: Uuid, + manga_id: Uuid, + chapter_id: Uuid, + page_number: i32, + }, + /// The worker failed a page (this attempt). + Failed { + page_id: Uuid, + manga_id: Uuid, + chapter_id: Uuid, + page_number: i32, + }, +} + +/// Broadcaster shared on `AppState`. Cheap to clone via `Arc`. Publishing +/// when there are no subscribers is a no-op (the send error is ignored). +pub struct AnalysisEvents { + tx: broadcast::Sender, +} + +impl AnalysisEvents { + pub fn new() -> Self { + // Buffer is generous; a slow subscriber that lags is dropped frames + // (the SSE handler treats `Lagged` as "skip and continue"). + let (tx, _rx) = broadcast::channel(512); + Self { tx } + } + + pub fn publish(&self, event: AnalysisEvent) { + let _ = self.tx.send(event); + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.tx.subscribe() + } +} + +impl Default for AnalysisEvents { + fn default() -> Self { + Self::new() + } +} diff --git a/backend/src/analysis/mod.rs b/backend/src/analysis/mod.rs index 53899a8..5ea2d7c 100644 --- a/backend/src/analysis/mod.rs +++ b/backend/src/analysis/mod.rs @@ -8,5 +8,6 @@ //! * [`daemon`] — the job-leasing worker loop (added in the worker phase). pub mod daemon; +pub mod events; pub mod prompt; pub mod vision; diff --git a/backend/src/api/admin/analysis.rs b/backend/src/api/admin/analysis.rs index 7f054a8..39ce75e 100644 --- a/backend/src/api/admin/analysis.rs +++ b/backend/src/api/admin/analysis.rs @@ -9,13 +9,19 @@ //! * `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; @@ -35,6 +41,36 @@ pub fn routes() -> Router { .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, + _admin: RequireAdmin, +) -> Sse>> { + 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)] @@ -207,6 +243,16 @@ async fn reenqueue( 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, diff --git a/backend/src/app.rs b/backend/src/app.rs index d76036e..7c1fb58 100644 --- a/backend/src/app.rs +++ b/backend/src/app.rs @@ -61,6 +61,10 @@ pub struct AppState { /// 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, + /// Live analysis-event broadcaster. Always present (the worker and the + /// admin enqueue path publish here; the SSE endpoint subscribes). When + /// the worker is disabled the channel simply stays quiet. + pub analysis_events: Arc, } /// Shared handle the admin crawler endpoints use to observe and control @@ -137,6 +141,11 @@ pub async fn build(config: Config) -> anyhow::Result { (None, None, None) }; + // Live-event bus shared by the worker (progress) and the admin enqueue + // path; the SSE endpoint subscribes. Created unconditionally so the + // stream exists even with the worker disabled. + let analysis_events = Arc::new(crate::analysis::events::AnalysisEvents::new()); + // AI content-analysis worker. Independent of the crawler daemon (works // for uploads with the crawler off), gated on ANALYSIS_ENABLED. Uses a // plain reqwest client — no cookie jar / proxy, unlike the crawler's. @@ -160,6 +169,7 @@ pub async fn build(config: Config) -> anyhow::Result { dispatcher, workers: config.analysis.workers, job_timeout: config.analysis.job_timeout, + events: Arc::clone(&analysis_events), }, ); tracing::info!( @@ -183,6 +193,7 @@ pub async fn build(config: Config) -> anyhow::Result { crawler, admin_allowed_origins: Arc::new(config.admin_allowed_origins.clone()), analysis_enabled: config.analysis.enabled, + analysis_events, }; let router = router(state).layer(cors_layer(&config.cors_allowed_origins)); Ok(AppHandle { router, daemon, analysis_daemon }) diff --git a/backend/src/repo/page.rs b/backend/src/repo/page.rs index fa524f5..b5e5edc 100644 --- a/backend/src/repo/page.rs +++ b/backend/src/repo/page.rs @@ -42,6 +42,22 @@ pub async fn find_by_id(pool: &PgPool, id: Uuid) -> AppResult> { Ok(row) } +/// Resolve a page's breadcrumb `(manga_id, chapter_id, page_number)` for +/// live event payloads. `None` if the page doesn't exist. +pub async fn locate( + pool: &PgPool, + page_id: Uuid, +) -> AppResult> { + let row: Option<(Uuid, Uuid, i32)> = sqlx::query_as( + "SELECT c.manga_id, p.chapter_id, p.page_number \ + FROM pages p JOIN chapters c ON c.id = p.chapter_id WHERE p.id = $1", + ) + .bind(page_id) + .fetch_optional(pool) + .await?; + Ok(row) +} + pub async fn list_for_chapter(pool: &PgPool, chapter_id: Uuid) -> AppResult> { let rows = sqlx::query_as::<_, Page>( r#" diff --git a/backend/tests/analysis_worker.rs b/backend/tests/analysis_worker.rs index 34d5c07..fd2d236 100644 --- a/backend/tests/analysis_worker.rs +++ b/backend/tests/analysis_worker.rs @@ -75,6 +75,9 @@ fn spawn_with( dispatcher, workers: 1, job_timeout: Duration::from_secs(5), + events: std::sync::Arc::new( + mangalord::analysis::events::AnalysisEvents::new(), + ), }, ); (handle, cancel) @@ -206,6 +209,92 @@ async fn worker_isolates_dispatcher_panics(pool: PgPool) { assert_eq!(row.status, AnalysisStatus::Failed); } +#[sqlx::test(migrations = "./migrations")] +async fn worker_publishes_started_and_completed_events(pool: PgPool) { + let page_id = seed_page(&pool).await; + page_analysis::enqueue_for_page(&pool, page_id, false) + .await + .unwrap(); + + let events = Arc::new(mangalord::analysis::events::AnalysisEvents::new()); + let mut rx = events.subscribe(); + let cancel = CancellationToken::new(); + let handle = daemon::spawn( + pool.clone(), + cancel, + AnalysisDaemonConfig { + dispatcher: CountingDispatcher::ok(), + workers: 1, + job_timeout: Duration::from_secs(5), + events: events.clone(), + }, + ); + + let mut kinds = Vec::new(); + let _ = tokio::time::timeout(Duration::from_secs(5), async { + while let Ok(ev) = rx.recv().await { + let v = serde_json::to_value(&ev).unwrap(); + let kind = v["kind"].as_str().unwrap().to_string(); + // Each event must carry the page breadcrumb. + assert_eq!(v["page_id"].as_str().unwrap(), page_id.to_string()); + let done = kind == "completed"; + kinds.push(kind); + if done { + break; + } + } + }) + .await; + handle.shutdown().await; + + assert!(kinds.contains(&"started".to_string()), "expected a started event"); + assert!( + kinds.contains(&"completed".to_string()), + "expected a completed event" + ); +} + +#[sqlx::test(migrations = "./migrations")] +async fn worker_publishes_failed_event_on_dispatch_error(pool: PgPool) { + let page_id = seed_page(&pool).await; + page_analysis::enqueue_for_page(&pool, page_id, false) + .await + .unwrap(); + sqlx::query("UPDATE crawler_jobs SET max_attempts = 1 WHERE payload->>'page_id' = $1") + .bind(page_id.to_string()) + .execute(&pool) + .await + .unwrap(); + + let events = Arc::new(mangalord::analysis::events::AnalysisEvents::new()); + let mut rx = events.subscribe(); + let cancel = CancellationToken::new(); + let handle = daemon::spawn( + pool.clone(), + cancel, + AnalysisDaemonConfig { + dispatcher: CountingDispatcher::failing(), + workers: 1, + job_timeout: Duration::from_secs(5), + events: events.clone(), + }, + ); + + let mut saw_failed = false; + let _ = tokio::time::timeout(Duration::from_secs(5), async { + while let Ok(ev) = rx.recv().await { + let v = serde_json::to_value(&ev).unwrap(); + if v["kind"] == "failed" { + saw_failed = true; + break; + } + } + }) + .await; + handle.shutdown().await; + assert!(saw_failed, "expected a failed event"); +} + #[sqlx::test(migrations = "./migrations")] async fn worker_ignores_non_analyze_jobs(pool: PgPool) { // A crawl job must be left untouched by the analysis worker. diff --git a/backend/tests/api_admin_analysis.rs b/backend/tests/api_admin_analysis.rs index 5fef297..d7acad1 100644 --- a/backend/tests/api_admin_analysis.rs +++ b/backend/tests/api_admin_analysis.rs @@ -548,6 +548,43 @@ async fn page_detail_unknown_page_is_404(pool: PgPool) { assert_eq!(resp.status(), StatusCode::NOT_FOUND); } +#[sqlx::test(migrations = "./migrations")] +async fn status_stream_is_admin_gated_and_event_stream(pool: PgPool) { + let h = common::harness(pool.clone()); + + // Non-admin → 403 (don't consume the streaming body). + let (_u, cookie) = common::register_user(&h.app).await; + let resp = h + .app + .clone() + .oneshot(common::get_with_cookie( + "/api/v1/admin/analysis/status/stream", + &cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::FORBIDDEN); + + // Admin → 200 with an SSE content type. + let admin_cookie = seed_admin(&pool, &h.app).await; + let resp = h + .app + .clone() + .oneshot(common::get_with_cookie( + "/api/v1/admin/analysis/status/stream", + &admin_cookie, + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let ct = resp + .headers() + .get(axum::http::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or_default(); + assert!(ct.starts_with("text/event-stream"), "got content-type {ct:?}"); +} + #[sqlx::test(migrations = "./migrations")] async fn coverage_requires_admin(pool: PgPool) { let h = common::harness(pool.clone()); diff --git a/backend/tests/api_admin_role.rs b/backend/tests/api_admin_role.rs index 5d5dc9b..587fbb2 100644 --- a/backend/tests/api_admin_role.rs +++ b/backend/tests/api_admin_role.rs @@ -53,6 +53,7 @@ fn admin_test_router(pool: PgPool) -> (Router, TempDir) { crawler: None, admin_allowed_origins: Arc::new(Vec::new()), analysis_enabled: false, + analysis_events: Arc::new(mangalord::analysis::events::AnalysisEvents::new()), }; let app = Router::new() .nest("/api/v1", api::routes()) diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index c5fb4ef..7e9e622 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -83,6 +83,7 @@ fn harness_with_auth_config( // harness `harness_with_admin_origins` overrides this. admin_allowed_origins: Arc::new(Vec::new()), analysis_enabled: false, + analysis_events: Arc::new(mangalord::analysis::events::AnalysisEvents::new()), }; Harness { app: router(state), _storage_dir: storage_dir } } @@ -160,6 +161,7 @@ pub fn harness_with_resync( crawler: None, admin_allowed_origins: Arc::new(Vec::new()), analysis_enabled: false, + analysis_events: Arc::new(mangalord::analysis::events::AnalysisEvents::new()), }; Harness { app: router(state), @@ -191,6 +193,7 @@ pub fn harness_with_analysis(pool: PgPool) -> Harness { crawler: None, admin_allowed_origins: Arc::new(Vec::new()), analysis_enabled: true, + analysis_events: Arc::new(mangalord::analysis::events::AnalysisEvents::new()), }; Harness { app: router(state), @@ -221,6 +224,7 @@ pub fn harness_with_admin_origins(pool: PgPool, origins: Vec) -> Harness crawler: None, admin_allowed_origins: Arc::new(origins), analysis_enabled: false, + analysis_events: Arc::new(mangalord::analysis::events::AnalysisEvents::new()), }; Harness { app: router(state), diff --git a/frontend/package.json b/frontend/package.json index 44bb5eb..6b3b491 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.76.0", + "version": "0.77.0", "private": true, "type": "module", "scripts": {