From 1997fe48bb1dbb9abb9ba5bc22bdfa98d6ebf4e1 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Fri, 3 Jul 2026 02:39:31 +0530 Subject: [PATCH] feat(iios): projection-cursor ledger + ProjectionCursorService (P9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the doc-mandated IiosProjectionCursor (ordered high-water-mark + rolling checksum per projection/topic/partition) and a reusable ProjectionCursorService.advance(). The checksum is a deterministic fold over the ordered event stream, so replaying the same events reproduces it — the KG-06 proof that replay reproduces projection outcomes. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../migration.sql | 19 ++++ packages/iios-service/prisma/schema.prisma | 16 ++++ packages/iios-service/src/app.module.ts | 2 + .../projection-cursor.service.spec.ts | 88 +++++++++++++++++++ .../projection/projection-cursor.service.ts | 51 +++++++++++ .../src/projection/projection.module.ts | 10 +++ .../iios-service/src/test-utils/reset-db.ts | 2 +- 7 files changed, 187 insertions(+), 1 deletion(-) create mode 100644 packages/iios-service/prisma/migrations/20260702210802_projection_cursor/migration.sql create mode 100644 packages/iios-service/src/projection/projection-cursor.service.spec.ts create mode 100644 packages/iios-service/src/projection/projection-cursor.service.ts create mode 100644 packages/iios-service/src/projection/projection.module.ts diff --git a/packages/iios-service/prisma/migrations/20260702210802_projection_cursor/migration.sql b/packages/iios-service/prisma/migrations/20260702210802_projection_cursor/migration.sql new file mode 100644 index 0000000..c578fbb --- /dev/null +++ b/packages/iios-service/prisma/migrations/20260702210802_projection_cursor/migration.sql @@ -0,0 +1,19 @@ +-- CreateTable +CREATE TABLE "IiosProjectionCursor" ( + "id" TEXT NOT NULL, + "projectionName" TEXT NOT NULL, + "sourceTopic" TEXT NOT NULL, + "partitionKey" TEXT NOT NULL, + "lastEventId" TEXT NOT NULL, + "lastOffset" INTEGER NOT NULL DEFAULT 0, + "checksum" TEXT NOT NULL, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "IiosProjectionCursor_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "IiosProjectionCursor_partitionKey_idx" ON "IiosProjectionCursor"("partitionKey"); + +-- CreateIndex +CREATE UNIQUE INDEX "IiosProjectionCursor_projectionName_sourceTopic_partitionKe_key" ON "IiosProjectionCursor"("projectionName", "sourceTopic", "partitionKey"); diff --git a/packages/iios-service/prisma/schema.prisma b/packages/iios-service/prisma/schema.prisma index 3c185d0..3bb9f81 100644 --- a/packages/iios-service/prisma/schema.prisma +++ b/packages/iios-service/prisma/schema.prisma @@ -492,6 +492,22 @@ model IiosProcessedEvent { @@id([consumerName, eventId]) } +/// Projection replay cursor (P9, KG-06). Ordered high-water-mark + rolling checksum per +/// (projection, topic, partition) proving replay reproduces the same projection outcome. +model IiosProjectionCursor { + id String @id @default(cuid()) + projectionName String + sourceTopic String + partitionKey String + lastEventId String + lastOffset Int @default(0) + checksum String + updatedAt DateTime @updatedAt + + @@unique([projectionName, sourceTopic, partitionKey]) + @@index([partitionKey]) +} + /// Idempotent-command ledger (P9). Suppresses duplicate externally-visible mutations: /// same (scope, command, key) replays the cached response; a different request → conflict. model IiosIdempotencyCommand { diff --git a/packages/iios-service/src/app.module.ts b/packages/iios-service/src/app.module.ts index 257cf25..5210d1a 100644 --- a/packages/iios-service/src/app.module.ts +++ b/packages/iios-service/src/app.module.ts @@ -4,6 +4,7 @@ import { PlatformModule } from './platform/platform.module'; import { IdentityModule } from './identity/identity.module'; import { InteractionsModule } from './interactions/interactions.module'; import { IdempotencyModule } from './idempotency/idempotency.module'; +import { ProjectionModule } from './projection/projection.module'; import { OutboxModule } from './outbox/outbox.module'; import { ThreadsModule } from './threads/threads.module'; import { MessageModule } from './messaging/message.module'; @@ -25,6 +26,7 @@ import { DevController } from './dev/dev.controller'; IdentityModule, InteractionsModule, IdempotencyModule, + ProjectionModule, OutboxModule, ThreadsModule, MessageModule, diff --git a/packages/iios-service/src/projection/projection-cursor.service.spec.ts b/packages/iios-service/src/projection/projection-cursor.service.spec.ts new file mode 100644 index 0000000..464a957 --- /dev/null +++ b/packages/iios-service/src/projection/projection-cursor.service.spec.ts @@ -0,0 +1,88 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { PrismaClient } from '@prisma/client'; +import type { CloudEvent } from '@insignia/iios-contracts'; +import { resetDb } from '../test-utils/reset-db'; +import { ProjectionCursorService } from './projection-cursor.service'; +import type { PrismaService } from '../prisma/prisma.service'; + +const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public'; +const prisma = new PrismaClient({ datasources: { db: { url } } }); +const asService = prisma as unknown as PrismaService; +const svc = () => new ProjectionCursorService(asService); + +const TOPIC = 'com.insignia.iios.test.v1'; +const ce = (id: string, data: Record = {}, scope = 'scope-1', type = TOPIC): CloudEvent => + ({ + specversion: '1.0', + id, + type, + source: 'iios/test', + time: '2026-01-01T00:00:00.000Z', + insignia: { scopeSnapshotId: scope, idempotencyKey: id, dataClass: 'internal' }, + data, + }) as unknown as CloudEvent; + +beforeAll(async () => { await prisma.$connect(); }); +afterAll(async () => { await prisma.$disconnect(); }); +beforeEach(async () => { await resetDb(prisma); }); + +describe('ProjectionCursorService (P9 — replay cursor, KG-06)', () => { + it('advance creates a cursor at offset 1 with the event id and a 64-hex checksum', async () => { + const row = await svc().advance('p', ce('e1', { n: 1 })); + expect(row.lastOffset).toBe(1); + expect(row.lastEventId).toBe('e1'); + expect(row.checksum).toMatch(/^[0-9a-f]{64}$/); + }); + + it('three distinct events advance the offset to 3 and track the latest event id', async () => { + const s = svc(); + await s.advance('p', ce('e1')); + await s.advance('p', ce('e2')); + const third = await s.advance('p', ce('e3')); + expect(third.lastOffset).toBe(3); + expect(third.lastEventId).toBe('e3'); + }); + + it('replay determinism: re-running the same ordered stream reproduces the checksum (KG-06)', async () => { + const s = svc(); + await s.advance('p', ce('e1', { a: 1 })); + await s.advance('p', ce('e2', { b: 2 })); + const c1 = (await s.advance('p', ce('e3', { c: 3 }))).checksum; + + // Wipe the cursor and replay the identical stream. + await prisma.iiosProjectionCursor.deleteMany({}); + await s.advance('p', ce('e1', { a: 1 })); + await s.advance('p', ce('e2', { b: 2 })); + const c2 = (await s.advance('p', ce('e3', { c: 3 }))).checksum; + + expect(c2).toBe(c1); + }); + + it('order sensitivity: a different event order yields a different checksum', async () => { + const s = svc(); + await s.advance('p', ce('e1')); + const ab = (await s.advance('p', ce('e2'))).checksum; + + await prisma.iiosProjectionCursor.deleteMany({}); + await s.advance('p', ce('e2')); + const ba = (await s.advance('p', ce('e1'))).checksum; + + expect(ba).not.toBe(ab); + }); + + it('a different source_topic for the same projection is a separate cursor row', async () => { + const s = svc(); + await s.advance('p', ce('e1', {}, 'scope-1', 'topic.a')); + await s.advance('p', ce('e2', {}, 'scope-1', 'topic.b')); + expect(await prisma.iiosProjectionCursor.count({ where: { projectionName: 'p' } })).toBe(2); + }); + + it('two partitions (scopes) track independently', async () => { + const s = svc(); + const a = await s.advance('p', ce('e1', {}, 'scope-A')); + const b = await s.advance('p', ce('e2', {}, 'scope-B')); + expect(a.partitionKey).toBe('scope-A'); + expect(b.partitionKey).toBe('scope-B'); + expect(await prisma.iiosProjectionCursor.count()).toBe(2); + }); +}); diff --git a/packages/iios-service/src/projection/projection-cursor.service.ts b/packages/iios-service/src/projection/projection-cursor.service.ts new file mode 100644 index 0000000..ae3b43e --- /dev/null +++ b/packages/iios-service/src/projection/projection-cursor.service.ts @@ -0,0 +1,51 @@ +import { Injectable } from '@nestjs/common'; +import { createHash } from 'node:crypto'; +import type { CloudEvent } from '@insignia/iios-contracts'; +import { PrismaService } from '../prisma/prisma.service'; + +const sha = (s: string) => createHash('sha256').update(s).digest('hex'); +// Stable stringify: sort object keys so {a,b} and {b,a} hash equal. +const stable = (v: unknown): string => + v === null || typeof v !== 'object' + ? JSON.stringify(v) ?? 'null' + : Array.isArray(v) + ? `[${v.map(stable).join(',')}]` + : `{${Object.keys(v as object) + .sort() + .map((k) => `${JSON.stringify(k)}:${stable((v as Record)[k])}`) + .join(',')}}`; + +/** + * Projection replay cursor (P9, KG-06). Each projection worker folds every event it + * consumes into an ordered high-water-mark (lastEventId + monotonic lastOffset) and a + * rolling checksum, keyed per (projection, source_topic, partition). Because the checksum + * is a deterministic function of the ordered (eventId, eventHash) stream, replaying the + * same events reproduces the same checksum — the proof that replay reproduces outcomes. + */ +@Injectable() +export class ProjectionCursorService { + constructor(private readonly prisma: PrismaService) {} + + /** Fold one consumed event into the projection's ordered cursor + rolling checksum. */ + async advance(projectionName: string, event: CloudEvent) { + const sourceTopic = event.type; + const partitionKey = event.insignia?.scopeSnapshotId ?? 'global'; + const eventHash = sha(stable({ id: event.id, type: event.type, data: event.data })); + const key = { projectionName_sourceTopic_partitionKey: { projectionName, sourceTopic, partitionKey } }; + + const prev = await this.prisma.iiosProjectionCursor.findUnique({ where: key }); + const checksum = sha(`${prev?.checksum ?? ''}${event.id}${eventHash}`); + const lastOffset = (prev?.lastOffset ?? 0) + 1; + + return this.prisma.iiosProjectionCursor.upsert({ + where: key, + create: { projectionName, sourceTopic, partitionKey, lastEventId: event.id, lastOffset, checksum }, + update: { lastEventId: event.id, lastOffset, checksum }, + }); + } + + /** All cursors (global ops read for /metrics + replay/repair tooling). */ + async list() { + return this.prisma.iiosProjectionCursor.findMany({ orderBy: [{ projectionName: 'asc' }, { sourceTopic: 'asc' }] }); + } +} diff --git a/packages/iios-service/src/projection/projection.module.ts b/packages/iios-service/src/projection/projection.module.ts new file mode 100644 index 0000000..f831873 --- /dev/null +++ b/packages/iios-service/src/projection/projection.module.ts @@ -0,0 +1,10 @@ +import { Global, Module } from '@nestjs/common'; +import { ProjectionCursorService } from './projection-cursor.service'; + +/** Projection replay-cursor ledger, global so every projector can checkpoint. */ +@Global() +@Module({ + providers: [ProjectionCursorService], + exports: [ProjectionCursorService], +}) +export class ProjectionModule {} diff --git a/packages/iios-service/src/test-utils/reset-db.ts b/packages/iios-service/src/test-utils/reset-db.ts index 2327aab..7dabe6b 100644 --- a/packages/iios-service/src/test-utils/reset-db.ts +++ b/packages/iios-service/src/test-utils/reset-db.ts @@ -8,7 +8,7 @@ import type { PrismaClient } from '@prisma/client'; export async function resetDb(prisma: PrismaClient): Promise { await prisma.$executeRawUnsafe( `TRUNCATE TABLE - "IiosIdempotencyCommand","IiosDlqItem","IiosAuditLink", + "IiosProjectionCursor","IiosIdempotencyCommand","IiosDlqItem","IiosAuditLink", "IiosMeetingActionItem","IiosActionItem","IiosTranscriptSegment","IiosMeetingTranscript","IiosMeetingSummary","IiosMeetingParticipant","IiosMeeting","IiosMeetingRequest", "IiosCalendarSyncCursor","IiosCalendarEvent","IiosCalendarProviderAccount","IiosAvailabilityWindow", "IiosAiEvidenceLink","IiosAiClaim","IiosAiToolCall","IiosAiArtifact","IiosAiModelRun","IiosAiJob","IiosEmbeddingRef",