domverse
ce5c8db5bc
Add the reusable browser shell primitives Phase B views build on: - store.js: observable store (get/set/subscribe) - events.js: job activity adapter — SSE preferred, polling fallback, sharing the event `seq` as cursor so a transport switch drops nothing - api.js: cancellable() (AbortController) + job endpoints; aborted requests reject with code "cancelled" Make the SSE stream generically consumable: emit default `message` events with the type in the JSON payload instead of `event: <type>`, so a browser EventSource receives the open-ended type set (state:*, claimed, …) via onmessage without enumerating it. The `event: done` sentinel and resumable `id:` cursors are unchanged. Tests: frontend/js/tests/ in-browser unit suite (api errors + cancellation, store transitions, routing, SSE→polling fallback) served over the app's static mount and driven by tests/e2e/test_frontend_shell.py, which also asserts asset loading, deep-link + reload restore, JSON-only /api/v1, and a clean console/network. Reuses the installed playwright — no JS toolchain added. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
63 lines
2.0 KiB
JavaScript
63 lines
2.0 KiB
JavaScript
// Job activity adapter: prefer SSE (resumable, pushed), fall back to polling the
|
|
// same durable events when EventSource is missing or the stream errors. The cursor
|
|
// is the event `seq`, shared by both transports, so switching mid-stream never
|
|
// drops or repeats an event. subscribeJob returns a stop() handle.
|
|
import { api } from "./api.js";
|
|
|
|
// Mirrors ACTIVE_STATES in services/jobs.py; a job outside this set is terminal
|
|
// and the poller stops.
|
|
const ACTIVE = new Set(["queued", "running", "cancelling", "retry_queued"]);
|
|
|
|
export function subscribeJob(jobId, { onEvent, onDone, onError, after = 0, pollMs = 1000 } = {}) {
|
|
let cursor = after;
|
|
let stopped = false;
|
|
let source = null;
|
|
let timer = null;
|
|
|
|
function stop() {
|
|
if (stopped) return;
|
|
stopped = true;
|
|
if (source) source.close();
|
|
if (timer) clearTimeout(timer);
|
|
if (onDone) onDone(cursor);
|
|
}
|
|
|
|
async function poll() {
|
|
if (stopped) return;
|
|
try {
|
|
const data = await api.jobEvents(jobId, cursor);
|
|
for (const event of data.events) {
|
|
cursor = event.seq;
|
|
if (onEvent) onEvent(event);
|
|
}
|
|
if (!ACTIVE.has(data.state)) return stop();
|
|
} catch (error) {
|
|
if (!stopped && onError) onError(error);
|
|
}
|
|
if (!stopped) timer = setTimeout(poll, pollMs);
|
|
}
|
|
|
|
if (typeof EventSource === "function") {
|
|
source = new EventSource(api.jobEventsStreamUrl(jobId, cursor));
|
|
source.onmessage = (message) => {
|
|
const event = JSON.parse(message.data);
|
|
cursor = Number(message.lastEventId) || cursor;
|
|
if (onEvent) onEvent({ seq: cursor, ...event });
|
|
};
|
|
// Server sends `event: done` when the job reaches a terminal state.
|
|
source.addEventListener("done", stop);
|
|
// Network/stream error: drop SSE and continue from the same cursor via polling
|
|
// so no consumer notices the transport switch.
|
|
source.onerror = () => {
|
|
if (stopped || !source) return;
|
|
source.close();
|
|
source = null;
|
|
poll();
|
|
};
|
|
} else {
|
|
poll();
|
|
}
|
|
|
|
return { stop };
|
|
}
|