feat(analysis): live SSE updates in the admin coverage dashboard
The Analysis section now reflects worker progress in real time: - Opens an EventSource on the live stream while mounted; a "Live" pill reflects connection state and reconnects on drop (probes after repeated failures so a lost session logs out). - Applies events incrementally: enqueued marks in-scope loaded pages queued; started flips a chip to a pulsing "analyzing"; completed flips it green and live-bumps the manga + chapter coverage badges (guarded so no double count); failed flips it red. A compact activity ticker shows the last few events. - Server numbers stay authoritative: re-loading the overview or a page grid clears the matching live overrides. API client: analysisStatusStreamUrl() + AnalysisEvent type. Tests: vitest for the stream URL; Playwright asserts SSE frames flow into the ticker (existing coverage/drill/queue specs get a default quiet stream). svelte-check + build clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
@@ -155,6 +155,14 @@ async function mockAdmin(page: Page, cap: Captured) {
|
||||
})
|
||||
);
|
||||
|
||||
// Default: keep the SSE connection pending (no events) so tests that
|
||||
// don't care about live updates don't trigger reconnect churn. The
|
||||
// live-updates test overrides this with a fulfilling stream.
|
||||
await page.route(
|
||||
'**/api/v1/admin/analysis/status/stream',
|
||||
() => new Promise(() => {})
|
||||
);
|
||||
|
||||
await page.route('**/api/v1/admin/analysis/reenqueue', (r) => {
|
||||
cap.reenqueue = JSON.parse(r.request().postData() ?? '{}');
|
||||
return r.fulfill({
|
||||
@@ -221,6 +229,49 @@ test.describe('/admin/analysis', () => {
|
||||
);
|
||||
});
|
||||
|
||||
test('live SSE events drive the indicator and activity ticker', async ({
|
||||
page
|
||||
}) => {
|
||||
const cap: Captured = { reenqueue: null, analyzeCalls: 0 };
|
||||
await mockAdmin(page, cap);
|
||||
// Override the default hanging stream with one that emits two frames.
|
||||
const frames =
|
||||
`event: analysis\ndata: ${JSON.stringify({
|
||||
kind: 'started',
|
||||
page_id: pageNone,
|
||||
manga_id: mangaId,
|
||||
chapter_id: chapterId,
|
||||
page_number: 2
|
||||
})}\n\n` +
|
||||
`event: analysis\ndata: ${JSON.stringify({
|
||||
kind: 'completed',
|
||||
page_id: pageNone,
|
||||
manga_id: mangaId,
|
||||
chapter_id: chapterId,
|
||||
page_number: 2
|
||||
})}\n\n`;
|
||||
let served = false;
|
||||
await page.route('**/api/v1/admin/analysis/status/stream', (r) => {
|
||||
if (served) return new Promise(() => {}); // hang on reconnect
|
||||
served = true;
|
||||
return r.fulfill({
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream' },
|
||||
body: frames
|
||||
});
|
||||
});
|
||||
|
||||
await page.setViewportSize(DESKTOP);
|
||||
await page.goto('/admin/analysis');
|
||||
|
||||
// The two SSE frames are parsed and applied to the activity ticker.
|
||||
// (The mocked stream closes after its body, so the live pill flips
|
||||
// back to "Reconnecting…" — the ticker is the durable signal.)
|
||||
const tick = page.getByTestId('admin-analysis-ticker');
|
||||
await expect(tick).toContainText('Analyzing page 2');
|
||||
await expect(tick).toContainText('Analyzed page 2');
|
||||
});
|
||||
|
||||
test('queue an unanalyzed page from its detail modal', async ({ page }) => {
|
||||
const cap: Captured = { reenqueue: null, analyzeCalls: 0 };
|
||||
await mockAdmin(page, cap);
|
||||
|
||||
@@ -32,7 +32,8 @@ import {
|
||||
getAnalysisMangaCoverage,
|
||||
getAnalysisChapterCoverage,
|
||||
getAnalysisChapterPages,
|
||||
getAnalysisPageDetail
|
||||
getAnalysisPageDetail,
|
||||
analysisStatusStreamUrl
|
||||
} from './admin';
|
||||
|
||||
function ok(body: unknown, status = 200): Response {
|
||||
@@ -580,4 +581,10 @@ describe('admin crawler api client', () => {
|
||||
expect(url).toMatch(/\/v1\/admin\/analysis\/pages\/p1$/);
|
||||
expect(d.status).toBe('none');
|
||||
});
|
||||
|
||||
it('analysisStatusStreamUrl points at the SSE endpoint', () => {
|
||||
expect(analysisStatusStreamUrl()).toMatch(
|
||||
/\/v1\/admin\/analysis\/status\/stream$/
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -532,3 +532,27 @@ export async function getAnalysisPageDetail(
|
||||
`/v1/admin/analysis/pages/${encodeURIComponent(pageId)}`
|
||||
);
|
||||
}
|
||||
|
||||
/** One live analysis event from the SSE stream. */
|
||||
export type AnalysisEvent =
|
||||
| {
|
||||
kind: 'enqueued';
|
||||
count: number;
|
||||
manga_id: string | null;
|
||||
chapter_id: string | null;
|
||||
}
|
||||
| {
|
||||
kind: 'started' | 'completed' | 'failed';
|
||||
page_id: string;
|
||||
manga_id: string;
|
||||
chapter_id: string;
|
||||
page_number: number;
|
||||
};
|
||||
|
||||
/** URL of the live analysis SSE stream. Open with `new EventSource(...)`
|
||||
* while the admin Analysis page is mounted and close it on navigate-away.
|
||||
* Each message is a named `analysis` event whose `data` is an
|
||||
* {@link AnalysisEvent}; a `lagged` event signals dropped frames. */
|
||||
export function analysisStatusStreamUrl(): string {
|
||||
return apiUrl('/v1/admin/analysis/status/stream');
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
<script lang="ts">
|
||||
import { onMount } from 'svelte';
|
||||
import { onMount, onDestroy } from 'svelte';
|
||||
import { ApiError } from '$lib/api/client';
|
||||
import {
|
||||
reenqueueAnalysis,
|
||||
@@ -8,11 +8,13 @@
|
||||
getAnalysisChapterCoverage,
|
||||
getAnalysisChapterPages,
|
||||
getAnalysisPageDetail,
|
||||
analysisStatusStreamUrl,
|
||||
type MangaCoverage,
|
||||
type ChapterCoverage,
|
||||
type PageStatusItem,
|
||||
type PageAnalysisDetail,
|
||||
type ReenqueueAnalysisOptions
|
||||
type ReenqueueAnalysisOptions,
|
||||
type AnalysisEvent
|
||||
} from '$lib/api/admin';
|
||||
import { chapterLabel } from '$lib/api/chapters';
|
||||
import CoverageBadge from '$lib/components/CoverageBadge.svelte';
|
||||
@@ -50,10 +52,154 @@
|
||||
let notice = $state<string | null>(null);
|
||||
let error = $state<string | null>(null);
|
||||
|
||||
// --- Live status (SSE) ---
|
||||
type LiveStatus = 'analyzing' | 'done' | 'failed' | 'queued';
|
||||
// Per-page override applied on top of the fetched grid status.
|
||||
let liveStatus = $state<Record<string, LiveStatus>>({});
|
||||
// Pages we've already live-counted into coverage, so a repeat
|
||||
// `completed` can't double-bump a badge.
|
||||
let counted = new Set<string>();
|
||||
let ticker = $state<{ id: number; text: string }[]>([]);
|
||||
let tickerSeq = 0;
|
||||
let live = $state(false);
|
||||
let source: EventSource | null = null;
|
||||
let sseErrors = 0;
|
||||
|
||||
onMount(() => {
|
||||
void loadCoverage(true);
|
||||
openStream();
|
||||
});
|
||||
|
||||
onDestroy(() => {
|
||||
source?.close();
|
||||
source = null;
|
||||
});
|
||||
|
||||
function pushTicker(text: string) {
|
||||
ticker = [{ id: ++tickerSeq, text }, ...ticker].slice(0, 8);
|
||||
}
|
||||
|
||||
/** Effective chip status: live override wins over the fetched value. */
|
||||
function chipStatus(p: PageStatusItem): string {
|
||||
return liveStatus[p.page_id] ?? p.status;
|
||||
}
|
||||
|
||||
/** mangaId that owns a loaded chapter, via the chapters cache. */
|
||||
function mangaOfChapter(chapterId: string): string | null {
|
||||
for (const [mid, val] of Object.entries(chapters)) {
|
||||
if (Array.isArray(val) && val.some((c) => c.chapter_id === chapterId)) {
|
||||
return mid;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function bumpCoverage(mangaId: string, chapterId: string) {
|
||||
mangas = mangas.map((m) =>
|
||||
m.manga_id === mangaId
|
||||
? {
|
||||
...m,
|
||||
analyzed_pages: Math.min(m.total_pages, m.analyzed_pages + 1)
|
||||
}
|
||||
: m
|
||||
);
|
||||
const chs = chapters[mangaId];
|
||||
if (Array.isArray(chs)) {
|
||||
chapters = {
|
||||
...chapters,
|
||||
[mangaId]: chs.map((c) =>
|
||||
c.chapter_id === chapterId
|
||||
? {
|
||||
...c,
|
||||
analyzed_pages: Math.min(
|
||||
c.total_pages,
|
||||
c.analyzed_pages + 1
|
||||
)
|
||||
}
|
||||
: c
|
||||
)
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/** Mark loaded, in-scope, not-yet-done pages as queued. */
|
||||
function markQueued(ev: Extract<AnalysisEvent, { kind: 'enqueued' }>) {
|
||||
const next = { ...liveStatus };
|
||||
for (const [chapterId, val] of Object.entries(pages)) {
|
||||
if (!Array.isArray(val)) continue;
|
||||
const inScope =
|
||||
ev.chapter_id != null
|
||||
? chapterId === ev.chapter_id
|
||||
: ev.manga_id != null
|
||||
? mangaOfChapter(chapterId) === ev.manga_id
|
||||
: true; // whole library
|
||||
if (!inScope) continue;
|
||||
for (const p of val) {
|
||||
const cur: string = next[p.page_id] ?? p.status;
|
||||
if (cur === 'none' || cur === 'failed') next[p.page_id] = 'queued';
|
||||
}
|
||||
}
|
||||
liveStatus = next;
|
||||
}
|
||||
|
||||
function applyEvent(ev: AnalysisEvent) {
|
||||
if (ev.kind === 'enqueued') {
|
||||
const scope = ev.chapter_id
|
||||
? ' (chapter)'
|
||||
: ev.manga_id
|
||||
? ' (manga)'
|
||||
: ' (library)';
|
||||
pushTicker(`Queued ${ev.count} page${ev.count === 1 ? '' : 's'}${scope}`);
|
||||
markQueued(ev);
|
||||
return;
|
||||
}
|
||||
if (ev.kind === 'started') {
|
||||
liveStatus = { ...liveStatus, [ev.page_id]: 'analyzing' };
|
||||
pushTicker(`Analyzing page ${ev.page_number}`);
|
||||
} else if (ev.kind === 'completed') {
|
||||
liveStatus = { ...liveStatus, [ev.page_id]: 'done' };
|
||||
if (!counted.has(ev.page_id)) {
|
||||
counted.add(ev.page_id);
|
||||
bumpCoverage(ev.manga_id, ev.chapter_id);
|
||||
}
|
||||
pushTicker(`✓ Analyzed page ${ev.page_number}`);
|
||||
} else if (ev.kind === 'failed') {
|
||||
liveStatus = { ...liveStatus, [ev.page_id]: 'failed' };
|
||||
pushTicker(`✗ Failed page ${ev.page_number}`);
|
||||
}
|
||||
}
|
||||
|
||||
function openStream() {
|
||||
if (source) return;
|
||||
const es = new EventSource(analysisStatusStreamUrl(), { withCredentials: true });
|
||||
es.addEventListener('analysis', (e) => {
|
||||
try {
|
||||
applyEvent(JSON.parse((e as MessageEvent).data) as AnalysisEvent);
|
||||
live = true;
|
||||
sseErrors = 0;
|
||||
} catch {
|
||||
// ignore a malformed frame
|
||||
}
|
||||
});
|
||||
// Dropped frames — resync the overview from the server.
|
||||
es.addEventListener('lagged', () => void loadCoverage(true));
|
||||
es.onopen = () => {
|
||||
live = true;
|
||||
sseErrors = 0;
|
||||
};
|
||||
es.onerror = () => {
|
||||
live = false;
|
||||
sseErrors += 1;
|
||||
// EventSource hides the status code; after several failures,
|
||||
// probe a normal admin call so the global on401 hook can fire.
|
||||
if (sseErrors >= 4) {
|
||||
sseErrors = 0;
|
||||
getAnalysisMangaCoverage({ limit: 1 }).catch(() => {});
|
||||
}
|
||||
};
|
||||
source = es;
|
||||
}
|
||||
|
||||
function onSearchInput() {
|
||||
if (searchTimer) clearTimeout(searchTimer);
|
||||
searchTimer = setTimeout(() => loadCoverage(true), 300);
|
||||
@@ -66,6 +212,9 @@
|
||||
// Collapse open rows so stale chapter/page state doesn't linger.
|
||||
expandedManga = null;
|
||||
expandedChapter = null;
|
||||
// Server numbers are authoritative again — drop live overrides.
|
||||
liveStatus = {};
|
||||
counted = new Set();
|
||||
} else {
|
||||
loadingMore = true;
|
||||
}
|
||||
@@ -122,7 +271,13 @@
|
||||
async function loadPages(id: string) {
|
||||
pages[id] = 'loading';
|
||||
try {
|
||||
pages[id] = await getAnalysisChapterPages(id);
|
||||
const fresh = await getAnalysisChapterPages(id);
|
||||
// Backend status is authoritative for just-fetched pages; drop
|
||||
// any live overrides for them so the grid reflects the server.
|
||||
const next = { ...liveStatus };
|
||||
for (const p of fresh) delete next[p.page_id];
|
||||
liveStatus = next;
|
||||
pages[id] = fresh;
|
||||
} catch (e) {
|
||||
delete pages[id];
|
||||
error = e instanceof ApiError ? e.message : 'Failed to load pages.';
|
||||
@@ -220,13 +375,32 @@
|
||||
<title>Mangalord | Admin — Analysis</title>
|
||||
</svelte:head>
|
||||
|
||||
<h1 class="heading">Content analysis</h1>
|
||||
<div class="title-row">
|
||||
<h1 class="heading">Content analysis</h1>
|
||||
<span
|
||||
class="live-pill"
|
||||
class:on={live}
|
||||
data-testid="admin-analysis-live"
|
||||
title={live ? 'Live updates connected' : 'Reconnecting…'}
|
||||
>
|
||||
<span class="live-dot"></span>
|
||||
{live ? 'Live' : 'Reconnecting…'}
|
||||
</span>
|
||||
</div>
|
||||
<p class="lede">
|
||||
Coverage of the AI analysis worker (OCR, auto-tags, scene description,
|
||||
NSFW moderation). Browse what's analyzed, queue more, and inspect any
|
||||
page's result.
|
||||
page's result. Updates stream live as pages are queued and processed.
|
||||
</p>
|
||||
|
||||
{#if ticker.length > 0}
|
||||
<ul class="ticker" data-testid="admin-analysis-ticker" aria-live="polite">
|
||||
{#each ticker as t (t.id)}
|
||||
<li>{t.text}</li>
|
||||
{/each}
|
||||
</ul>
|
||||
{/if}
|
||||
|
||||
<label class="checkbox">
|
||||
<input
|
||||
type="checkbox"
|
||||
@@ -365,11 +539,11 @@
|
||||
{#each ps as p (p.page_id)}
|
||||
<button
|
||||
type="button"
|
||||
class="page-chip {p.status}"
|
||||
title={`Page ${p.page_number} — ${p.status}`}
|
||||
class="page-chip {chipStatus(p)}"
|
||||
title={`Page ${p.page_number} — ${chipStatus(p)}`}
|
||||
onclick={() => openDetail(p.page_id)}
|
||||
data-testid={`admin-analysis-page-${p.page_id}`}
|
||||
data-status={p.status}
|
||||
data-status={chipStatus(p)}
|
||||
>
|
||||
{p.page_number}
|
||||
</button>
|
||||
@@ -377,6 +551,7 @@
|
||||
</div>
|
||||
<p class="legend">
|
||||
<span class="dot done"></span> analyzed
|
||||
<span class="dot analyzing"></span> analyzing
|
||||
<span class="dot queued"></span> queued
|
||||
<span class="dot failed"></span> failed
|
||||
<span class="dot none"></span> not analyzed
|
||||
@@ -496,9 +671,68 @@
|
||||
</Modal>
|
||||
|
||||
<style>
|
||||
.title-row {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
gap: var(--space-3);
|
||||
margin-bottom: var(--space-2);
|
||||
}
|
||||
.heading {
|
||||
margin-bottom: var(--space-2);
|
||||
}
|
||||
.title-row .heading {
|
||||
margin-bottom: 0;
|
||||
}
|
||||
.live-pill {
|
||||
display: inline-flex;
|
||||
align-items: center;
|
||||
gap: 6px;
|
||||
font-size: var(--font-xs);
|
||||
font-weight: var(--weight-semibold);
|
||||
padding: 2px var(--space-2);
|
||||
border-radius: var(--radius-pill);
|
||||
background: var(--surface-elevated);
|
||||
color: var(--text-muted);
|
||||
border: 1px solid var(--border);
|
||||
}
|
||||
.live-pill.on {
|
||||
color: #2e7d32;
|
||||
border-color: color-mix(in srgb, #2e7d32 40%, transparent);
|
||||
}
|
||||
.live-dot {
|
||||
width: 8px;
|
||||
height: 8px;
|
||||
border-radius: 50%;
|
||||
background: currentColor;
|
||||
}
|
||||
.live-pill.on .live-dot {
|
||||
animation: pulse 1.6s ease-in-out infinite;
|
||||
}
|
||||
.ticker {
|
||||
list-style: none;
|
||||
margin: 0 0 var(--space-3);
|
||||
padding: var(--space-2) var(--space-3);
|
||||
max-width: 44rem;
|
||||
border: 1px solid var(--border);
|
||||
border-radius: var(--radius-md);
|
||||
background: var(--surface);
|
||||
font-size: var(--font-sm);
|
||||
color: var(--text-muted);
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 2px;
|
||||
max-height: 9rem;
|
||||
overflow: hidden;
|
||||
}
|
||||
@keyframes pulse {
|
||||
0%,
|
||||
100% {
|
||||
opacity: 1;
|
||||
}
|
||||
50% {
|
||||
opacity: 0.35;
|
||||
}
|
||||
}
|
||||
.lede {
|
||||
color: var(--text-muted);
|
||||
margin-bottom: var(--space-4);
|
||||
@@ -620,6 +854,12 @@
|
||||
border-color: color-mix(in srgb, #2563eb 45%, transparent);
|
||||
color: #2563eb;
|
||||
}
|
||||
.page-chip.analyzing {
|
||||
background: color-mix(in srgb, #d4762a 22%, transparent);
|
||||
border-color: color-mix(in srgb, #d4762a 55%, transparent);
|
||||
color: #b85e1a;
|
||||
animation: pulse 1.2s ease-in-out infinite;
|
||||
}
|
||||
.page-chip.failed {
|
||||
background: color-mix(in srgb, var(--danger) 18%, transparent);
|
||||
border-color: color-mix(in srgb, var(--danger) 45%, transparent);
|
||||
@@ -647,6 +887,9 @@
|
||||
.dot.queued {
|
||||
background: #2563eb;
|
||||
}
|
||||
.dot.analyzing {
|
||||
background: #d4762a;
|
||||
}
|
||||
.dot.failed {
|
||||
background: var(--danger);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user