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) <noreply@anthropic.com>

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);

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,
     };
   }