// P9 slice-7 projection-cursor smoke: injected events advance a projection's ordered // cursor (last_offset high-water-mark) and expose a determinism checksum + lag on // /metrics (KG-06 observability). Requires the service running with IIOS_DEV_TOKENS=1 // (the relay ticks by default, driving the projectors). import 'dotenv/config'; const SERVICE = process.env.SMOKE_URL ?? 'http://localhost:3200'; const APP_ID = 'portal-demo'; const PROJECTION = 'raw-event-projector'; const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); }; const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); async function devToken(userId, orgId) { const r = await fetch(`${SERVICE}/v1/dev/token`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ appId: APP_ID, userId, name: userId, orgId }), }); if (!r.ok) throw new Error(`devToken ${r.status} (service with IIOS_DEV_TOKENS=1?)`); return (await r.json()).token; } async function metrics(token) { const r = await fetch(`${SERVICE}/metrics`, { headers: { authorization: `Bearer ${token}` } }); if (!r.ok) throw new Error(`GET /metrics ${r.status}: ${await r.text()}`); return r.json(); } async function injectWebhook(token, text) { const r = await fetch(`${SERVICE}/v1/dev/webhook/WEBHOOK`, { method: 'POST', headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` }, body: JSON.stringify({ text }), }); if (!r.ok) throw new Error(`inject webhook ${r.status}: ${await r.text()}`); return r.json(); } // Sum of last_offset across this projection's cursors — advances by ≥1 per injected event. const projOffset = (m) => (m.projections ?? []).filter((p) => p.projectionName === PROJECTION).reduce((s, p) => s + p.lastOffset, 0); const token = await devToken('proj-a', 'org_proj_A'); const base = projOffset(await metrics(token)); console.log(`baseline ${PROJECTION} offset-sum = ${base}`); // Inject one event → the projection cursor must appear/advance. await injectWebhook(token, 'projection smoke one'); let after1 = base; for (let i = 0; i < 40; i++) { const m = await metrics(token); after1 = projOffset(m); if (after1 > base) break; await sleep(250); } assert(after1 > base, `injected event advanced ${PROJECTION}'s cursor (${base} → ${after1})`); // The cursor row carries a 64-hex determinism checksum + non-negative lag. const m1 = await metrics(token); const cur = m1.projections.find((p) => p.projectionName === PROJECTION); assert(/^[0-9a-f]{64}$/.test(cur.checksum), `cursor carries a 64-hex rolling checksum (${cur.checksum.slice(0, 12)}…)`); assert(cur.lastEventId && cur.lastOffset >= 1, `cursor has a last_event_id + offset ≥ 1 (offset ${cur.lastOffset})`); assert(m1.projections.every((p) => p.lagMs >= 0), 'every projection reports a non-negative lagMs'); // A second event advances the ordered high-water-mark further. await injectWebhook(token, 'projection smoke two'); let after2 = after1; for (let i = 0; i < 40; i++) { after2 = projOffset(await metrics(token)); if (after2 > after1) break; await sleep(250); } assert(after2 > after1, `a second event advanced the high-water-mark again (${after1} → ${after2})`); console.log('\nP9 projection-cursor smoke: PASS'); process.exit(0);