diff --git a/src/hooks/useWorkersFeed.js b/src/hooks/useWorkersFeed.js new file mode 100644 index 0000000..ffb2daf --- /dev/null +++ b/src/hooks/useWorkersFeed.js @@ -0,0 +1,111 @@ +import { useCallback, useEffect, useRef, useState } from 'react' +import { createFeedController } from '../lib/workersFeed' +import { createRequestGuard } from '../lib/requestGuard' + +// Spur eines Hustlers wird bei sichtbarem Tab alle 15 s neu geholt +const TRACE_MS = 15000 +const UNKNOWN_ERROR = 'Unbekannter Fehler' + +const isVisible = () => document.visibilityState !== 'hidden' + +// { state, error, loaded, refresh }: Zustand der Hustler (alle 15 s) und Ereignisse (alle 5 s, über onEvents) von der +// Dienst-API. state ist null bis zum ersten Erfolg, loaded wird dann wahr und bleibt es. error ist der deutsche Fehlertext +// (oder null); die Abfrage läuft bei einem Fehler weiter. Im Hintergrund-Tab ruht alles, beim Zurückkehren wird sofort +// abgefragt. refresh() holt den Zustand sofort (z. B. nach dem Pausenschalter). onEvents darf bei jedem Render neu sein. +export function useWorkersFeed(api, { onEvents } = {}) { + const [state, setState] = useState(null) + const [error, setError] = useState(null) + const [loaded, setLoaded] = useState(false) + const onEventsRef = useRef(onEvents) + const controllerRef = useRef(null) + + useEffect(() => { onEventsRef.current = onEvents }, [onEvents]) + + useEffect(() => { + setError(null) + const controller = createFeedController({ + api, + setTimer: (fn, ms) => window.setTimeout(fn, ms), + clearTimer: (id) => window.clearTimeout(id), + isVisible, + onSnapshot: (snapshot) => { + setState(snapshot) + setLoaded(true) + }, + onEvents: (list) => { if (onEventsRef.current) onEventsRef.current(list) }, + onError: (err) => setError(err ? err.message || UNKNOWN_ERROR : null), + }) + controllerRef.current = controller + const onVisibility = () => controller.visibilityChanged() + document.addEventListener('visibilitychange', onVisibility) + controller.start() + return () => { + document.removeEventListener('visibilitychange', onVisibility) + controller.stop() + controllerRef.current = null + } + }, [api]) + + const refresh = useCallback(() => { + if (controllerRef.current) controllerRef.current.refreshState() + }, []) + + return { state, error, loaded, refresh } +} + +// { trace, loading, error }: Spur eines Hustlers (api.trace). workerId null ergibt { trace: null } ohne Abruf. Beim +// Wechsel der workerId wird neu geladen (loading, bis die erste Antwort da ist), danach alle 15 s bei sichtbarem Tab und +// sofort beim Zurückkehren. Eine späte Antwort für einen früheren Hustler oder eine ältere Anfrage wird verworfen. +export function useWorkerTrace(api, workerId) { + const [result, setResult] = useState({ key: null, trace: null, error: null }) + + useEffect(() => { + if (!workerId) return undefined + const guard = createRequestGuard() + let timer = null + + const stopTimer = () => { + if (timer === null) return + window.clearTimeout(timer) + timer = null + } + + const load = async () => { + stopTimer() + const id = guard.next() + try { + const trace = await api.trace(workerId) + if (!guard.isCurrent(id)) return + setResult({ key: workerId, trace, error: null }) + } catch (err) { + if (!guard.isCurrent(id)) return + // die letzte Spur desselben Hustlers bleibt stehen, der Fehler kommt dazu + setResult((prev) => ({ key: workerId, trace: prev.key === workerId ? prev.trace : null, error: err?.message || UNKNOWN_ERROR })) + } + timer = window.setTimeout(tick, TRACE_MS) + } + + // Im Hintergrund kein Abruf, nur neu planen + const tick = () => { + timer = null + if (isVisible()) load() + else timer = window.setTimeout(tick, TRACE_MS) + } + + const onVisibility = () => { if (isVisible()) load() } + document.addEventListener('visibilitychange', onVisibility) + load() + return () => { + guard.next() + stopTimer() + document.removeEventListener('visibilitychange', onVisibility) + } + }, [api, workerId]) + + const current = Boolean(workerId) && result.key === workerId + return { + trace: current ? result.trace : null, + loading: Boolean(workerId) && !current, + error: current ? result.error : null, + } +} diff --git a/src/lib/workersFeed.js b/src/lib/workersFeed.js new file mode 100644 index 0000000..1a75164 --- /dev/null +++ b/src/lib/workersFeed.js @@ -0,0 +1,198 @@ +// Polling-Steuerung der Hustler-Seite: Zustand alle 15 s, Ereignisse alle 5 s (Übersicht §6.2). +// Kein React, kein Appwrite, kein window/document: Timer und Sichtbarkeit werden injiziert (unter Node testbar). +// Je Schleife läuft höchstens ein Abruf und höchstens ein wartender Timer; der nächste Timer wird erst nach +// Abschluss des Abrufs gesetzt (kein Intervall, kein Überlappen). +import { takeNewEvents, latestTs, needsStateRefresh } from './workersEvents.js' + +const EPOCH = '1970-01-01T00:00:00.000Z' +const noop = () => {} + +const isRecord = (v) => v !== null && typeof v === 'object' && !Array.isArray(v) + +/** + * Steuerung für den Feed. Rückrufe: onSnapshot(state) bei jedem geholten Zustand, onEvents(liste) einmal je Runde + * mit allen neuen Ereignissen (aufsteigend), onError(err) einmal je Fehlerfolge und onError(null) beim nächsten Erfolg. + * Der Dienst liefert bei events(after) auch Einträge der letzten 5 s davor; die Deduplizierung übernimmt takeNewEvents. + */ +export function createFeedController({ + api, setTimer, clearTimer, isVisible, + onSnapshot = noop, onEvents = noop, onError = noop, + stateMs = 15000, eventsMs = 5000, pageLimit = 200, maxPages = 5, +}) { + let running = false + let gen = 0 // wird bei start/stop erhöht: Antworten einer früheren Generation ändern nichts + let cursor = null // null bis der erste Zustand da ist; danach der jüngste gesehene Zeitstempel + let historyMs = NaN // lastEventTs des ersten Zustands: Ereignisse bis dahin sind Vorgeschichte + let seen = new Set() + let stateAgain = false // refreshState() kam während eines laufenden Zustand-Abrufs + let reported = null // zuletzt gemeldete Fehlermeldung (null = kein Fehler) + const timer = { state: null, events: null } + const inFlight = { state: false, events: false } + const failure = { state: null, events: null } + + function cancel(loop) { + if (timer[loop] === null) return + clearTimer(timer[loop]) + timer[loop] = null + } + + function schedule(loop) { + if (!running) return + cancel(loop) + timer[loop] = setTimer(() => tick(loop), loop === 'state' ? stateMs : eventsMs) + } + + // Im Hintergrund kein Abruf, nur neu planen; beim Zurückkehren holt visibilityChanged() sofort nach + function tick(loop) { + timer[loop] = null + if (!running) return + if (!isVisible()) { + schedule(loop) + return + } + if (loop === 'state') void loadState() + else void loadEvents() + } + + // Ein Fehler gilt, bis beide Schleifen wieder laufen; gemeldet wird nur, wenn sich der Text ändert + function publishError() { + const err = failure.state || failure.events || null + const message = err ? String(err.message ?? err) : null + if (message === reported) return + reported = message + onError(err) + } + + async function loadState() { + if (!running || inFlight.state) return + cancel('state') + inFlight.state = true + const myGen = gen + let state = null + let failed = null + try { + state = await api.state() + } catch (err) { + failed = err + } + if (myGen !== gen) return + inFlight.state = false + failure.state = failed + let first = false + try { + if (failed === null) { + const snapshot = isRecord(state) ? state : {} + if (cursor === null) { + first = true + // fehlt lastEventTs oder ist es kein Datum, gibt es keine Vorgeschichte und der Cursor beginnt bei 1970 + const lastMs = typeof snapshot.lastEventTs === 'string' ? Date.parse(snapshot.lastEventTs) : NaN + cursor = Number.isNaN(lastMs) ? EPOCH : snapshot.lastEventTs + historyMs = lastMs + } + onSnapshot(snapshot) + } + publishError() + } finally { + if (myGen === gen) { + if (stateAgain) { + stateAgain = false + void loadState() + } else { + schedule('state') + } + if (first) { + if (isVisible()) void loadEvents() + else schedule('events') + } + } + } + } + + // Vorgeschichte: in der ersten Runde nach dem Zustand zählen Ereignisse bis lastEventTs als gesehen, aber nicht neu + function isHistory(e) { + return !Number.isNaN(historyMs) && Date.parse(e.ts) <= historyMs + } + + // Holt Seiten, solange sie voll sind (höchstens maxPages); wirft nie, ein Fehler steht im Ergebnis + async function fetchRound(myGen) { + const fresh = [] + let failed = null + try { + for (let page = 0; page < maxPages; page += 1) { + const res = await api.events(cursor, pageLimit) + if (myGen !== gen) break + const list = isRecord(res) && Array.isArray(res.events) ? res.events : [] + for (const e of takeNewEvents(seen, list)) { + if (!isHistory(e)) fresh.push(e) + } + cursor = latestTs(list, cursor) + if (list.length < pageLimit) break + } + } catch (err) { + failed = err + } + return { fresh, failed } + } + + async function loadEvents() { + if (!running || cursor === null || inFlight.events) return + cancel('events') + inFlight.events = true + const myGen = gen + const { fresh, failed } = await fetchRound(myGen) + if (myGen !== gen) return + inFlight.events = false + failure.events = failed + if (failed === null) historyMs = NaN + try { + // Bei einem Fehler nach einer vollen Seite wird das bereits Geholte trotzdem gemeldet (es gilt schon als gesehen) + if (fresh.length > 0) onEvents(fresh) + publishError() + if (needsStateRefresh(fresh)) refreshState() + } finally { + if (myGen === gen) schedule('events') + } + } + + function refreshState() { + if (!running) return + if (inFlight.state) { + stateAgain = true + return + } + void loadState() + } + + function visibilityChanged() { + if (!running || !isVisible()) return + void loadState() + void loadEvents() + } + + function start() { + if (running) return + running = true + gen += 1 + cursor = null + historyMs = NaN + seen = new Set() + stateAgain = false + reported = null + failure.state = null + failure.events = null + if (isVisible()) void loadState() + else schedule('state') + } + + function stop() { + running = false + gen += 1 + cancel('state') + cancel('events') + inFlight.state = false + inFlight.events = false + stateAgain = false + } + + return { start, stop, visibilityChanged, refreshState } +} diff --git a/tests/unit/workersFeed.test.mjs b/tests/unit/workersFeed.test.mjs new file mode 100644 index 0000000..1276310 --- /dev/null +++ b/tests/unit/workersFeed.test.mjs @@ -0,0 +1,321 @@ +import test from 'node:test' +import assert from 'node:assert/strict' +import { readFileSync } from 'node:fs' +import { loadModule } from './load.mjs' + +// Feed-Steuerung der Hustler-Seite: Zustand 15 s, Ereignisse 5 s, Cursor, Seiten, Sichtbarkeit, Fehler. +// Alles mit Fake-Timern und einer Fake-API (kein Netz, keine echte Zeit). +const { createFeedController } = await loadModule('src/lib/workersFeed.js') + +const nextTick = () => new Promise((resolve) => setImmediate(resolve)) + +function deferred() { + let resolve + const promise = new Promise((r) => { resolve = r }) + return { promise, resolve } +} + +// Fake-Timer + Fake-API. Ereignisse kommen entweder aus der Antwortliste `eventReplies` (ein Error darin wird zur +// Ablehnung) oder, mit `store`, aus einer veränderlichen Liste und verhalten sich wie der Dienst P1: +// events(after) liefert aufsteigend alle Einträge mit ts >= after - 5000 (5 s Überlappung), höchstens `limit`. +// `stateReplies` wie eventReplies für den Zustand; `h.gate` hält state() an, bis das Versprechen erfüllt wird. +function makeFeedHarness({ state, eventReplies = [], eventsHang = false, stateReplies = [], store = null, opts = {} } = {}) { + let now = 0 + let nextId = 1 + const timers = new Map() + const log = { state: 0, events: [], snapshots: [], batches: [], errors: [] } + const visible = { v: true } + const h = { gate: null, timers, log, visible } + const base = state ?? { lastEventTs: '2026-10-17T20:00:00.000Z', workers: [] } + + const setTimer = (fn, ms) => { + const id = nextId + nextId += 1 + timers.set(id, { at: now + ms, fn }) + return id + } + const clearTimer = (id) => { timers.delete(id) } + + const api = { + async state() { + log.state += 1 + if (h.gate) await h.gate.promise + const r = stateReplies.shift() + if (r instanceof Error) throw r + return r ?? base + }, + async events(after, limit) { + log.events.push([after, limit]) + if (eventsHang) return new Promise(() => {}) + if (store) { + const from = Date.parse(after) - 5000 + const list = store.filter((e) => Date.parse(e.ts) >= from).sort((a, b) => Date.parse(a.ts) - Date.parse(b.ts)) + return { events: list.slice(0, limit) } + } + const r = eventReplies.shift() ?? [] + if (r instanceof Error) throw r + return { events: r } + }, + } + + h.c = createFeedController({ + api, + setTimer, + clearTimer, + isVisible: () => visible.v, + onSnapshot: (s) => { log.snapshots.push(s) }, + onEvents: (list) => { log.batches.push(list) }, + onError: (e) => { log.errors.push(e ? e.message : null) }, + ...opts, + }) + // feuert fällige Timer in Zeitreihenfolge; je Timer lässt ein setImmediate alle Antworten durchlaufen + h.advance = async (ms) => { + const target = now + ms + while (true) { + let pick = null + for (const [id, t] of timers) { + if (t.at <= target && (pick === null || t.at < pick.t.at)) pick = { id, t } + } + if (pick === null) break + timers.delete(pick.id) + now = pick.t.at + pick.t.fn() + await nextTick() + } + now = target + } + h.settle = async () => { + await nextTick() + await nextTick() + } + return h +} + +const mk = (i) => ({ id: `e${i}`, ts: new Date(Date.UTC(2026, 9, 17, 20, 0, 1 + i)).toISOString(), workerId: 'kim', kind: 'fund', text: 't' }) +const ids = (batches) => batches.map((b) => b.map((e) => e.id)) + +test('Start: Zustand, Cursor aus lastEventTs, dann Ereignisse (Review Focus 3)', async () => { + const h = makeFeedHarness({}) + h.c.start() + await h.settle() + assert.equal(h.log.snapshots.length, 1) + assert.deepEqual(h.log.events[0], ['2026-10-17T20:00:00.000Z', 200]) + const h2 = makeFeedHarness({ state: { workers: [] } }) + h2.c.start() + await h2.settle() + assert.equal(h2.log.events[0][0], '1970-01-01T00:00:00.000Z') + // ein unlesbares lastEventTs wird nie als Cursor an den Dienst geschickt + const h3 = makeFeedHarness({ state: { lastEventTs: 'kaputt', workers: [] } }) + h3.c.start() + await h3.settle() + assert.equal(h3.log.events[0][0], '1970-01-01T00:00:00.000Z') +}) + +test('Takt: Ereignisse alle 5 s mit weitergeschobenem Cursor, Zustand alle 15 s', async () => { + const h = makeFeedHarness({ eventReplies: [[mk(1)], [], []] }) + h.c.start() + await h.settle() + await h.advance(5000) + assert.equal(h.log.events[1][0], mk(1).ts) + await h.advance(10000) + assert.equal(h.log.state, 2) + assert.equal(h.log.events.length, 4) +}) + +test('Hintergrund: keine Abrufe, beim Zurückkehren sofort beide', async () => { + const h = makeFeedHarness({}) + h.c.start() + await h.settle() + h.visible.v = false + await h.advance(60000) + assert.equal(h.log.state, 1) + assert.equal(h.log.events.length, 1) + h.visible.v = true + h.c.visibilityChanged() + await h.settle() + assert.equal(h.log.state, 2) + assert.equal(h.log.events.length, 2) + await h.advance(5000) + assert.equal(h.log.events.length, 3) + assert.equal(h.timers.size, 2) // eine Kette je Schleife, kein doppelter Timer nach der Rückkehr +}) + +test('Doppelte Ereignisse werden nie zweimal gemeldet (Review Focus 3)', async () => { + const h = makeFeedHarness({ eventReplies: [[mk(1), mk(2)], [mk(1), mk(2), mk(3)]] }) + h.c.start() + await h.settle() + await h.advance(5000) + assert.deepEqual(h.log.batches.map((b) => b.map((e) => e.id)), [['e1', 'e2'], ['e3']]) +}) + +test('Rückstau: volle Seiten werden sofort nachgeladen, höchstens 5 je Runde (Review Focus 2)', async () => { + const page = (n) => Array.from({ length: 200 }, (_, i) => mk(n * 200 + i)) + const h = makeFeedHarness({ eventReplies: [page(0), page(1), page(2), page(3), page(4), page(5)] }) + h.c.start() + await h.settle() + assert.equal(h.log.events.length, 5) + assert.equal(h.log.events[1][0], mk(199).ts) + // alle Seiten einer Runde kommen in einem einzigen onEvents-Aufruf + assert.equal(h.log.batches.length, 1) + assert.equal(h.log.batches[0].length, 1000) +}) + +test('keine überlappenden Abrufe, Fehler laufen weiter, Zustand-Auffrischung bei wichtigen Ereignissen', async () => { + const hang = makeFeedHarness({ eventsHang: true }) + hang.c.start() + await hang.settle() + await hang.advance(60000) + assert.equal(hang.log.events.length, 1) + const err = makeFeedHarness({ eventReplies: [new Error('Der Hustler-Dienst ist gerade nicht erreichbar.'), []] }) + err.c.start() + await err.settle() + await err.advance(5000) + assert.deepEqual(err.log.errors, ['Der Hustler-Dienst ist gerade nicht erreichbar.', null]) + const deal = makeFeedHarness({ eventReplies: [[{ ...mk(1), kind: 'deal' }]] }) + deal.c.start() + await deal.settle() + assert.equal(deal.log.state, 2) +}) + +test('stop räumt auf, start ist idempotent, späte Antworten werden verworfen', async () => { + const h = makeFeedHarness({}) + h.c.start() + h.c.start() + await h.settle() + assert.equal(h.log.state, 1) + h.c.stop() + assert.equal(h.timers.size, 0) + await h.advance(60000) + assert.equal(h.log.state, 1) +}) + +test('Hooks: Sichtbarkeit und Aufräumen im Quelltext', () => { + const src = readFileSync(new URL('../../src/hooks/useWorkersFeed.js', import.meta.url), 'utf8') + assert.match(src, /visibilitychange/) + assert.match(src, /removeEventListener\('visibilitychange'/) + assert.match(src, /\.stop\(\)/) + assert.match(src, /createRequestGuard/) +}) + +// ---- Ergänzungen: Vorgeschichte, Überlappung des Dienstes, Teilrunden, Spätantworten ---- + +test('Vorgeschichte: Ereignisse bis lastEventTs gelten als gesehen und werden nie gemeldet', async () => { + const hist1 = { ...mk(0), id: 'h1', ts: '2026-10-17T19:59:58.000Z' } + const hist2 = { ...mk(0), id: 'h2', ts: '2026-10-17T20:00:00.000Z' } // genau am Cursor + const h = makeFeedHarness({ eventReplies: [[hist1, hist2, mk(1)], [hist1, hist2, mk(1), mk(2)]] }) + h.c.start() + await h.settle() + await h.advance(5000) + assert.deepEqual(ids(h.log.batches), [['e1'], ['e2']]) + // ohne lastEventTs im Zustand (nichts davor) wird alles gemeldet + const none = makeFeedHarness({ state: { workers: [] }, eventReplies: [[hist1, mk(1)]] }) + none.c.start() + await none.settle() + assert.deepEqual(ids(none.log.batches), [['h1', 'e1']]) +}) + +test('wie der Dienst: 5 s Überlappung liefert jedes Ereignis genau einmal', async () => { + const store = [{ ...mk(0), id: 'alt', ts: '2026-10-17T19:59:57.000Z' }, { ...mk(0), id: 'am-cursor', ts: '2026-10-17T20:00:00.000Z' }] + const h = makeFeedHarness({ store }) + h.c.start() + await h.settle() + assert.equal(h.log.batches.length, 0) // die Vorgeschichte aus der Überlappung wird nicht nachgespielt + store.push(mk(1), mk(2)) + await h.advance(5000) + store.push(mk(3)) + await h.advance(5000) + await h.advance(5000) + await h.advance(5000) + assert.deepEqual(ids(h.log.batches), [['e1', 'e2'], ['e3']]) +}) + +test('wie der Dienst: 450 neue Ereignisse über drei Seiten, einmal gemeldet, aufsteigend', async () => { + const store = Array.from({ length: 450 }, (_, i) => mk(i)) + const h = makeFeedHarness({ store }) + h.c.start() + await h.settle() + assert.equal(h.log.events.length, 3) + assert.equal(h.log.batches.length, 1) + assert.equal(h.log.batches[0].length, 450) + assert.deepEqual(h.log.batches[0].map((e) => e.id), store.map((e) => e.id)) + await h.advance(5000) + assert.equal(h.log.batches.length, 1) +}) + +test('Fehler nach einer Seite: bereits Geholtes wird gemeldet, danach der Fehler', async () => { + const page = Array.from({ length: 200 }, (_, i) => mk(i)) + const h = makeFeedHarness({ eventReplies: [page, new Error('Der Hustler-Dienst ist gerade nicht erreichbar.'), []] }) + h.c.start() + await h.settle() + assert.equal(h.log.batches.length, 1) + assert.equal(h.log.batches[0].length, 200) + assert.deepEqual(h.log.errors, ['Der Hustler-Dienst ist gerade nicht erreichbar.']) + await h.advance(5000) + assert.deepEqual(h.log.errors, ['Der Hustler-Dienst ist gerade nicht erreichbar.', null]) + assert.equal(h.log.batches.length, 1) // die Seite wird nicht noch einmal gemeldet +}) + +test('erster Zustand schlägt fehl: kein Abruf der Ereignisse, später Cursor und Fehler zurückgesetzt', async () => { + const h = makeFeedHarness({ stateReplies: [new Error('Der Hustler-Dienst antwortet nicht richtig.')] }) + h.c.start() + await h.settle() + assert.equal(h.log.events.length, 0) + assert.equal(h.log.snapshots.length, 0) + assert.deepEqual(h.log.errors, ['Der Hustler-Dienst antwortet nicht richtig.']) + await h.advance(15000) + assert.equal(h.log.snapshots.length, 1) + assert.equal(h.log.events.length, 1) + assert.equal(h.log.events[0][0], '2026-10-17T20:00:00.000Z') + assert.deepEqual(h.log.errors, ['Der Hustler-Dienst antwortet nicht richtig.', null]) +}) + +test('Zustand-Auffrischung während eines laufenden Abrufs wird danach nachgeholt', async () => { + const h = makeFeedHarness({}) + h.c.start() + await h.settle() + h.gate = deferred() + h.c.refreshState() + await h.settle() + assert.equal(h.log.state, 2) + h.c.refreshState() + h.c.refreshState() + await h.settle() + assert.equal(h.log.state, 2) // kein zweiter Abruf daneben + const gate = h.gate + h.gate = null + gate.resolve() + await h.settle() + assert.equal(h.log.state, 3) + assert.equal(h.log.snapshots.length, 3) + assert.equal(h.timers.size, 2) +}) + +test('Rückkehr in den Vordergrund startet keinen zweiten Abruf neben einem hängenden', async () => { + const h = makeFeedHarness({ eventsHang: true }) + h.c.start() + await h.settle() + h.c.visibilityChanged() + await h.settle() + assert.equal(h.log.events.length, 1) + assert.equal(h.log.state, 2) // der Zustand ist frei und wird geholt +}) + +test('späte Antwort nach stop wird verworfen, ein neuer Start beginnt frisch', async () => { + const h = makeFeedHarness({ eventReplies: [[mk(1)]] }) + h.gate = deferred() + h.c.start() + await h.settle() + h.c.stop() + const gate = h.gate + h.gate = null + gate.resolve() + await h.settle() + assert.equal(h.log.snapshots.length, 0) + assert.equal(h.log.events.length, 0) + assert.equal(h.timers.size, 0) + h.c.start() + await h.settle() + assert.equal(h.log.snapshots.length, 1) + assert.deepEqual(ids(h.log.batches), [['e1']]) + assert.equal(h.timers.size, 2) +})