tickte-system/tests/unit/workersFeed.test.mjs

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 { createWorkersApi } = await loadModule('src/lib/workersApi.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)
})

// ---- Dichte Ereignisse: mehr als pageLimit Einträge im 5-s-Fenster des Dienstes (Fix-Runde 1) ----
// Der Dienst liefert bei events(after) die ÄLTESTEN pageLimit Einträge ab after - 5 s. Mit after = jüngster ts käme bei
// mehr als 200 Einträgen im Fenster immer dieselbe, schon bekannte Seite: der Feed stünde still, ohne Fehler.

// n Ereignisse im Abstand von step ms, Standard ab 20:00:01 (nach dem Standard-Zustand lastEventTs 20:00:00)
const dense = (n, { from = Date.UTC(2026, 9, 17, 20, 0, 1), step = 10, prefix = 'd' } = {}) => Array.from({ length: n }, (_, i) => ({
  id: `${prefix}${i}`,
  ts: new Date(from + i * step).toISOString(),
  workerId: 'kim',
  kind: 'fund',
  text: 't',
}))
const flat = (batches) => batches.flat().map((e) => e.id)

test('dichter Speicher wie der Dienst: 450 Ereignisse im Abstand von 10 ms, kein Stillstand', async () => {
  const store = dense(450)
  const h = makeFeedHarness({ store })
  h.c.start()
  await h.settle()
  // alles in einer Runde, ein onEvents, aufsteigend, höchstens maxPages Abrufe
  assert.equal(h.log.batches.length, 1)
  assert.deepEqual(h.log.batches[0].map((e) => e.id), store.map((e) => e.id))
  assert.ok(h.log.events.length <= 5)
  // volle Seite ohne Neues: der nächste Abruf fragt hinter die Spitze (Dienst liefert dann ab dem jüngsten ts)
  assert.equal(h.log.events[1][0], store[199].ts)
  assert.equal(h.log.events[2][0], new Date(Date.parse(store[199].ts) + 5000).toISOString())
  for (let i = 0; i < 4; i += 1) await h.advance(5000)
  assert.equal(h.log.batches.length, 1)   // nichts doppelt
  // ein Ereignis nach dem Stau kommt an
  store.push({ ...mk(0), id: 'spaet', ts: '2026-10-17T20:00:30.000Z' })
  await h.advance(5000)
  assert.deepEqual(ids(h.log.batches).slice(1), [['spaet']])
  assert.deepEqual(h.log.errors, [])
})

for (const maxPages of [1, 2]) {
  test(`dichter Speicher mit maxPages ${maxPages}: der Sprung gilt über Runden hinweg, alles kommt genau einmal an`, async () => {
    const store = dense(450)
    const h = makeFeedHarness({ store, opts: { maxPages } })
    h.c.start()
    await h.settle()
    for (let i = 0; i < 8; i += 1) await h.advance(5000)
    assert.deepEqual(flat(h.log.batches), store.map((e) => e.id))
    assert.deepEqual(h.log.errors, [])
  })
}

test('dichte Vorgeschichte: über 200 Ereignisse kurz vor lastEventTs werden nie gemeldet, das Neue danach schon', async () => {
  const history = dense(300, { from: Date.UTC(2026, 9, 17, 19, 59, 57), prefix: 'alt' })   // 19:59:57.000 bis 19:59:59.990
  const store = [...history]
  const h = makeFeedHarness({ store, state: { lastEventTs: history[299].ts, workers: [] } })
  h.c.start()
  await h.settle()
  assert.equal(h.log.batches.length, 0)
  store.push({ ...mk(0), id: 'neu1', ts: '2026-10-17T20:00:01.000Z' })
  await h.advance(5000)
  store.push({ ...mk(0), id: 'neu2', ts: '2026-10-17T20:00:07.000Z' })
  await h.advance(5000)
  assert.deepEqual(ids(h.log.batches), [['neu1'], ['neu2']])
})

test('200 Ereignisse mit demselben ts halten den Feed nicht an (Fremd-API ohne eindeutige ts)', async () => {
  // P1 vergibt strikt aufsteigende ts; hier geht es nur darum, dass auch dann nichts dauerhaft hängen bleibt
  const store = Array.from({ length: 250 }, (_, i) => ({ ...mk(0), id: `g${i}`, ts: '2026-10-17T20:00:01.000Z' }))
  const h = makeFeedHarness({ store })
  h.c.start()
  await h.settle()
  assert.ok(h.log.events.length <= 5)
  assert.equal(flat(h.log.batches).length, 200)   // mehr als pageLimit Einträge mit einem ts sind über die API nicht erreichbar
  store.push({ ...mk(0), id: 'spaet', ts: '2026-10-17T20:00:30.000Z' })
  await h.advance(5000)
  const all = flat(h.log.batches)
  assert.equal(all[all.length - 1], 'spaet')
  assert.equal(new Set(all).size, all.length)
  assert.deepEqual(h.log.errors, [])
})

test('hängender Abruf (tote Verbindung): die Zeitgrenze des API-Clients gibt die Schleife frei, Fehler erscheint und verschwindet wieder (Endprüfung P2)', async () => {
  // Echter createWorkersApi mit kurzer Zeitgrenze, echte Timer mit kurzen Takten: der erste Zustands-Abruf kommt nie an
  const state = { lastEventTs: '2026-10-17T20:00:00.000Z', workers: [] }
  const urls = []
  let hungOnce = false
  const fetchImpl = async (url) => {
    urls.push(url)
    if (url.endsWith('/state') && !hungOnce) {
      hungOnce = true
      return new Promise(() => {})
    }
    return { ok: true, status: 200, json: async () => (url.endsWith('/state') ? state : { events: [] }) }
  }
  const api = createWorkersApi({ getJwt: async () => 'jwt', fetchImpl, timeoutMs: 40 })
  const snapshots = []
  const errors = []
  const c = createFeedController({
    api,
    setTimer: (fn, ms) => setTimeout(fn, ms),
    clearTimer: (id) => clearTimeout(id),
    isVisible: () => true,
    onSnapshot: (x) => snapshots.push(x),
    onError: (e) => errors.push(e ? e.message : null),
    stateMs: 20,
    eventsMs: 20,
  })
  c.start()
  const until = async (ok) => {
    for (let i = 0; i < 100 && !ok(); i += 1) await new Promise((r) => setTimeout(r, 20))
  }
  await until(() => snapshots.length > 0)
  c.stop()
  assert.ok(snapshots.length >= 1, 'nach der Zeitgrenze kommt der Zustand doch noch')
  assert.deepEqual(errors, ['Der Hustler-Dienst ist gerade nicht erreichbar.', null])
})