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

feat(analysis): live SSE event stream for the admin dashboard

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-13 21:06:20 +02:00
parent 6bb72b8775
commit d0cd31c9a7
13 changed files with 314 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/Cargo.lock generated
View File

@@ -1517,7 +1517,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
[[package]]
name = "mangalord"
version = "0.76.0"
version = "0.77.0"
dependencies = [
"anyhow",
"argon2",

2
backend/Cargo.toml
View File

@@ -1,6 +1,6 @@
[package]
name = "mangalord"
version = "0.76.0"
version = "0.77.0"
edition = "2021"
default-run = "mangalord"

28
backend/src/analysis/daemon.rs
View File

@@ -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<dyn AnalyzeDispatcher>,
pub workers: usize,
pub job_timeout: Duration,
/// Live-event sink (Started/Completed/Failed) for admin SSE.
pub events: Arc<AnalysisEvents>,
}
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<dyn AnalyzeDispatcher>,
job_timeout: Duration,
events: Arc<AnalysisEvents>,
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;

78
backend/src/analysis/events.rs Normal file
View File

@@ -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<Uuid>,
chapter_id: Option<Uuid>,
},
/// 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<AnalysisEvent>,
}
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<AnalysisEvent> {
self.tx.subscribe()
}
}
impl Default for AnalysisEvents {
fn default() -> Self {
Self::new()
}
}

1
backend/src/analysis/mod.rs
View File

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

46
backend/src/api/admin/analysis.rs
View File

@@ -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<AppState> {
.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)]
@@ -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,

11
backend/src/app.rs
View File

@@ -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<crate::analysis::events::AnalysisEvents>,
}
/// Shared handle the admin crawler endpoints use to observe and control
@@ -137,6 +141,11 @@ pub async fn build(config: Config) -> anyhow::Result<AppHandle> {
(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<AppHandle> {
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<AppHandle> {
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 })

16
backend/src/repo/page.rs
View File

@@ -42,6 +42,22 @@ pub async fn find_by_id(pool: &PgPool, id: Uuid) -> AppResult<Option<Page>> {
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<Option<(Uuid, Uuid, i32)>> {
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<Vec<Page>> {
let rows = sqlx::query_as::<_, Page>(
r#"

89
backend/tests/analysis_worker.rs
View File

@@ -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.

37
backend/tests/api_admin_analysis.rs
View File

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

1
backend/tests/api_admin_role.rs
View File

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

4
backend/tests/common/mod.rs
View File

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

2
frontend/package.json
View File

@@ -1,6 +1,6 @@
{
"name": "mangalord-frontend",
"version": "0.76.0",
"version": "0.77.0",
"private": true,
"type": "module",
"scripts": {