From 67da42a10344210768a970c335db3b9b3310261c Mon Sep 17 00:00:00 2001 From: maaz519 Date: Fri, 3 Jul 2026 02:48:14 +0530 Subject: [PATCH] feat(iios): projection-cursor observability on /metrics + smoke (P9) Adds a global `projections` section to GET /metrics exposing each cursor's position (last_event_id/last_offset), determinism checksum, and lagMs. smoke-projection.mjs injects events and asserts the projection's ordered high-water-mark advances with a 64-hex rolling checksum (KG-06). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../iios-service/scripts/smoke-projection.mjs | 74 +++++++++++++++++++ .../src/observability/metrics.controller.ts | 14 ++++ 2 files changed, 88 insertions(+) create mode 100644 packages/iios-service/scripts/smoke-projection.mjs diff --git a/packages/iios-service/scripts/smoke-projection.mjs b/packages/iios-service/scripts/smoke-projection.mjs new file mode 100644 index 0000000..6b7d704 --- /dev/null +++ b/packages/iios-service/scripts/smoke-projection.mjs @@ -0,0 +1,74 @@ +// 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); diff --git a/packages/iios-service/src/observability/metrics.controller.ts b/packages/iios-service/src/observability/metrics.controller.ts index d26a5f7..b657904 100644 --- a/packages/iios-service/src/observability/metrics.controller.ts +++ b/packages/iios-service/src/observability/metrics.controller.ts @@ -2,6 +2,7 @@ import { BadRequestException, Controller, Get, Headers } from '@nestjs/common'; import { PrismaService } from '../prisma/prisma.service'; import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver'; import { SessionVerifier } from '../platform/session.verifier'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; /** * Tenant-scoped metrics (P9 observability, JSON). Returns the CALLER's tenant @@ -15,6 +16,7 @@ export class MetricsController { private readonly prisma: PrismaService, private readonly actors: ActorResolver, private readonly session: SessionVerifier, + private readonly cursors: ProjectionCursorService, ) {} @Get() @@ -40,11 +42,23 @@ export class MetricsController { const oldest = await this.prisma.iiosOutboxEvent.findFirst({ where: { status: { not: 'PUBLISHED' } }, orderBy: { createdAt: 'asc' } }); const oldestPendingMs = oldest ? Date.now() - new Date(oldest.createdAt).getTime() : 0; + // Projection cursors (global ops signal — position + determinism checksum + lag). + const projections = (await this.cursors.list()).map((c) => ({ + projectionName: c.projectionName, + sourceTopic: c.sourceTopic, + partitionKey: c.partitionKey, + lastEventId: c.lastEventId, + lastOffset: c.lastOffset, + checksum: c.checksum, + lagMs: Date.now() - new Date(c.updatedAt).getTime(), + })); + return { ts: new Date().toISOString(), scope: scope ? { orgId: scope.orgId, appId: scope.appId, tenantId: scope.tenantId, cellId: scope.cellId } : null, tenant, relay: { outboxPending, oldestPendingMs }, + projections, }; }