diff --git a/packages/iios-service/src/adapters/adapters.spec.ts b/packages/iios-service/src/adapters/adapters.spec.ts index 8c41bbc..7bd68d4 100644 --- a/packages/iios-service/src/adapters/adapters.spec.ts +++ b/packages/iios-service/src/adapters/adapters.spec.ts @@ -2,6 +2,7 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { makeFakePorts } from '@insignia/iios-testkit'; import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; import { signedFixture, emailFixture } from '@insignia/iios-adapter-sdk'; @@ -24,7 +25,7 @@ const SECRET = 'dev-adapter-secret'; // AdapterRegistry default const registry = () => new AdapterRegistry(); const inbound = () => new InboundService(asService, registry()); const projector = () => - new RawEventProjector(asService, new OutboxBus(), new DlqService(asService), registry(), new IngestService(asService, makeFakePorts(), new ActorResolver(asService))); + new RawEventProjector(asService, new OutboxBus(), new DlqService(asService), registry(), new IngestService(asService, makeFakePorts(), new ActorResolver(asService)), new ProjectionCursorService(asService)); async function rawReceivedEvents(): Promise { const rows = await prisma.iiosOutboxEvent.findMany({ where: { eventType: IIOS_EVENTS.rawReceived }, orderBy: { createdAt: 'asc' } }); diff --git a/packages/iios-service/src/adapters/raw-event.projector.ts b/packages/iios-service/src/adapters/raw-event.projector.ts index b6ae133..1808c2a 100644 --- a/packages/iios-service/src/adapters/raw-event.projector.ts +++ b/packages/iios-service/src/adapters/raw-event.projector.ts @@ -3,6 +3,7 @@ import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; import { DlqService } from '../outbox/dlq.service'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { AdapterRegistry } from './adapter.registry'; import { IngestService } from '../interactions/ingest.service'; @@ -21,6 +22,7 @@ export class RawEventProjector implements OnModuleInit { private readonly dlq: DlqService, private readonly registry: AdapterRegistry, private readonly ingest: IngestService, + private readonly cursor: ProjectionCursorService, ) {} onModuleInit(): void { @@ -29,7 +31,12 @@ export class RawEventProjector implements OnModuleInit { } async onRawReceived(event: CloudEvent): Promise { - if (!(await this.claim(event.id))) return; + if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance + await this.applyRawReceived(event); + await this.cursor.advance(this.consumer, event); + } + + private async applyRawReceived(event: CloudEvent): Promise { const { rawEventId } = event.data as { rawEventId: string }; const raw = await this.prisma.iiosInboundRawEvent.findUnique({ where: { id: rawEventId } }); if (!raw || raw.status !== 'RECEIVED') return; diff --git a/packages/iios-service/src/ai/ai-routing.spec.ts b/packages/iios-service/src/ai/ai-routing.spec.ts index adca2ee..1f16f91 100644 --- a/packages/iios-service/src/ai/ai-routing.spec.ts +++ b/packages/iios-service/src/ai/ai-routing.spec.ts @@ -3,6 +3,7 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { makeFakePorts, makeFakeInference } from '@insignia/iios-testkit'; import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; import { ActorResolver } from '../identity/actor.resolver'; @@ -105,7 +106,7 @@ describe('AI × routing (P7, KG-05: AI proposes, never approves)', () => { it('AiProjector auto-classifies a channel-bearing interaction, idempotently', async () => { const { interactionId, event } = await ingestInbound('There is a party at 9 PM tonight!'); - const p = new AiProjector(asService, new OutboxBus(), new DlqService(asService), ai()); + const p = new AiProjector(asService, new OutboxBus(), new DlqService(asService), ai(), new ProjectionCursorService(asService)); await p.onNormalized(event); await p.onNormalized(event); // replay expect(await prisma.iiosAiJob.count({ where: { interactionId, jobType: 'CLASSIFY' } })).toBe(1); diff --git a/packages/iios-service/src/ai/ai.projector.ts b/packages/iios-service/src/ai/ai.projector.ts index 62ff634..6a22b2d 100644 --- a/packages/iios-service/src/ai/ai.projector.ts +++ b/packages/iios-service/src/ai/ai.projector.ts @@ -3,6 +3,7 @@ import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; import { DlqService } from '../outbox/dlq.service'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { AiJobService } from './ai.service'; /** @@ -21,6 +22,7 @@ export class AiProjector implements OnModuleInit { private readonly bus: OutboxBus, private readonly dlq: DlqService, private readonly ai: AiJobService, + private readonly cursor: ProjectionCursorService, ) {} onModuleInit(): void { @@ -29,7 +31,12 @@ export class AiProjector implements OnModuleInit { } async onNormalized(event: CloudEvent): Promise { - if (!(await this.claim(event.id))) return; + if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance + await this.applyNormalized(event); + await this.cursor.advance(this.consumer, event); + } + + private async applyNormalized(event: CloudEvent): Promise { const { interactionId } = event.data as { interactionId: string }; const interaction = await this.prisma.iiosInteraction.findUnique({ where: { id: interactionId }, diff --git a/packages/iios-service/src/calendar/calendar-projector.spec.ts b/packages/iios-service/src/calendar/calendar-projector.spec.ts index d0e49e9..ffe2b05 100644 --- a/packages/iios-service/src/calendar/calendar-projector.spec.ts +++ b/packages/iios-service/src/calendar/calendar-projector.spec.ts @@ -3,6 +3,7 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { makeFakePorts, makeFakeInference } from '@insignia/iios-testkit'; import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; import { MessageService, type MessagePrincipal } from '../messaging/message.service'; @@ -58,7 +59,7 @@ describe('Calendar reminders + projector (P8)', () => { where: { eventType: IIOS_EVENTS.aiArtifactAccepted, aggregateId: artifact!.id }, })).cloudEvent as unknown as CloudEvent; - const p = new CalendarProjector(asService, new OutboxBus(), new DlqService(asService)); + const p = new CalendarProjector(asService, new OutboxBus(), new DlqService(asService), new ProjectionCursorService(asService)); await p.onArtifactAccepted(event); await p.onArtifactAccepted(event); // replay const claim = await prisma.iiosAiClaim.findFirstOrThrow({ where: { artifactId: artifact!.id, claimType: 'EVENT' } }); diff --git a/packages/iios-service/src/calendar/calendar.projector.ts b/packages/iios-service/src/calendar/calendar.projector.ts index 9dc875c..2e91bf9 100644 --- a/packages/iios-service/src/calendar/calendar.projector.ts +++ b/packages/iios-service/src/calendar/calendar.projector.ts @@ -3,6 +3,7 @@ import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; import { DlqService } from '../outbox/dlq.service'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; /** * P7→P8 auto-bridge (P8). When a P7 AI artifact is accepted and it carries an @@ -18,6 +19,7 @@ export class CalendarProjector implements OnModuleInit { private readonly prisma: PrismaService, private readonly bus: OutboxBus, private readonly dlq: DlqService, + private readonly cursor: ProjectionCursorService, ) {} onModuleInit(): void { @@ -26,7 +28,12 @@ export class CalendarProjector implements OnModuleInit { } async onArtifactAccepted(event: CloudEvent): Promise { - if (!(await this.claim(event.id))) return; + if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance + await this.applyArtifactAccepted(event); + await this.cursor.advance(this.consumer, event); + } + + private async applyArtifactAccepted(event: CloudEvent): Promise { const { artifactId } = event.data as { artifactId: string }; const artifact = await this.prisma.iiosAiArtifact.findUnique({ where: { id: artifactId }, include: { job: true, claims: true } }); if (!artifact?.acceptedByActorId) return; diff --git a/packages/iios-service/src/inbox/inbox.projector.ts b/packages/iios-service/src/inbox/inbox.projector.ts index 819cdc6..0e3a212 100644 --- a/packages/iios-service/src/inbox/inbox.projector.ts +++ b/packages/iios-service/src/inbox/inbox.projector.ts @@ -4,6 +4,7 @@ import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; import { DlqService } from '../outbox/dlq.service'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; interface MessageSentData { interactionId: string; @@ -30,6 +31,7 @@ export class InboxProjector implements OnModuleInit { private readonly prisma: PrismaService, private readonly bus: OutboxBus, private readonly dlq: DlqService, + private readonly cursor: ProjectionCursorService, ) {} onModuleInit(): void { @@ -39,7 +41,12 @@ export class InboxProjector implements OnModuleInit { } async onMessageSent(event: CloudEvent): Promise { - if (!(await this.claim(event.id))) return; + if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance + await this.applyMessageSent(event); + await this.cursor.advance(this.consumer, event); + } + + private async applyMessageSent(event: CloudEvent): Promise { const data = event.data as MessageSentData; const traceId = event.insignia?.correlationId; const thread = await this.prisma.iiosThread.findUnique({ where: { id: data.threadId } }); @@ -53,7 +60,12 @@ export class InboxProjector implements OnModuleInit { } async onMessageRead(event: CloudEvent): Promise { - if (!(await this.claim(event.id))) return; + if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance + await this.applyMessageRead(event); + await this.cursor.advance(this.consumer, event); + } + + private async applyMessageRead(event: CloudEvent): Promise { const data = event.data as MessageReadData; const item = await this.prisma.iiosInboxItem.findFirst({ where: { diff --git a/packages/iios-service/src/inbox/inbox.spec.ts b/packages/iios-service/src/inbox/inbox.spec.ts index 3569459..9b631a6 100644 --- a/packages/iios-service/src/inbox/inbox.spec.ts +++ b/packages/iios-service/src/inbox/inbox.spec.ts @@ -2,6 +2,7 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { makeFakePorts } from '@insignia/iios-testkit'; import { IIOS_EVENTS, PolicyDeniedError, type CloudEvent } from '@insignia/iios-contracts'; import { MessageService, type MessagePrincipal } from '../messaging/message.service'; @@ -16,7 +17,7 @@ const prisma = new PrismaClient({ datasources: { db: { url } } }); const asService = prisma as unknown as PrismaService; const actors = new ActorResolver(asService); const ms = () => new MessageService(asService, makeFakePorts(), actors); -const projector = () => new InboxProjector(asService, new OutboxBus(), new DlqService(asService)); +const projector = () => new InboxProjector(asService, new OutboxBus(), new DlqService(asService), new ProjectionCursorService(asService)); const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' }; const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' }; diff --git a/packages/iios-service/src/outbox/chaos.spec.ts b/packages/iios-service/src/outbox/chaos.spec.ts index 9b41103..3c18d1d 100644 --- a/packages/iios-service/src/outbox/chaos.spec.ts +++ b/packages/iios-service/src/outbox/chaos.spec.ts @@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto'; import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; import { resetDb } from '../test-utils/reset-db'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { makeFakePorts, makeFakeInference } from '@insignia/iios-testkit'; import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; import { ActorResolver } from '../identity/actor.resolver'; @@ -48,7 +49,7 @@ beforeEach(async () => { await resetDb(prisma); }); describe('chaos: poison consumer → DLQ → replay (P9)', () => { it('a consumer failure is dead-lettered (not swallowed), the lane keeps flowing, and replay resolves it', async () => { const dlq = new DlqService(asService); - const projector = new AiProjector(asService, new OutboxBus(), dlq, ai()); + const projector = new AiProjector(asService, new OutboxBus(), dlq, ai(), new ProjectionCursorService(asService)); dlq.registerHandler('ai-projector', (e) => projector.onNormalized(e)); // healthy handler for replay const poison = await ingestInbound('There is a party at 9 PM tonight!'); diff --git a/packages/iios-service/src/routing/route.projector.ts b/packages/iios-service/src/routing/route.projector.ts index d9e1845..ca6c7ed 100644 --- a/packages/iios-service/src/routing/route.projector.ts +++ b/packages/iios-service/src/routing/route.projector.ts @@ -3,6 +3,7 @@ import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { OutboxBus } from '../outbox/outbox.bus'; import { DlqService } from '../outbox/dlq.service'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { RouteService } from './route.service'; /** @@ -21,6 +22,7 @@ export class RouteProjector implements OnModuleInit { private readonly bus: OutboxBus, private readonly dlq: DlqService, private readonly routes: RouteService, + private readonly cursor: ProjectionCursorService, ) {} onModuleInit(): void { @@ -29,7 +31,12 @@ export class RouteProjector implements OnModuleInit { } async onNormalized(event: CloudEvent): Promise { - if (!(await this.claim(event.id))) return; + if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance + await this.applyNormalized(event); + await this.cursor.advance(this.consumer, event); + } + + private async applyNormalized(event: CloudEvent): Promise { const { interactionId } = event.data as { interactionId: string }; const interaction = await this.prisma.iiosInteraction.findUnique({ where: { id: interactionId }, diff --git a/packages/iios-service/src/routing/route.spec.ts b/packages/iios-service/src/routing/route.spec.ts index d86e09a..c156ffd 100644 --- a/packages/iios-service/src/routing/route.spec.ts +++ b/packages/iios-service/src/routing/route.spec.ts @@ -3,6 +3,7 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient, IiosRouteMode } from '@prisma/client'; import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; +import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { makeFakePorts } from '@insignia/iios-testkit'; import { MessageService, type MessagePrincipal } from '../messaging/message.service'; import { ActorResolver } from '../identity/actor.resolver'; @@ -160,7 +161,7 @@ describe('RouteService approve/deny → execute (P6)', () => { describe('RouteProjector — AUTOMATIC forwarding (P6)', () => { const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())), actors); - const projector = () => new RouteProjector(asService, new OutboxBus(), new DlqService(asService), routeSvc()); + const projector = () => new RouteProjector(asService, new OutboxBus(), new DlqService(asService), routeSvc(), new ProjectionCursorService(asService)); // Ingest an inbound (channel-bearing) interaction — routable, unlike native DMs. async function ingestInbound(text: string): Promise<{ interactionId: string; scopeId: string; event: CloudEvent }> { @@ -223,4 +224,30 @@ describe('RouteProjector — AUTOMATIC forwarding (P6)', () => { await p.onNormalized(event); expect(await prisma.iiosOutboundCommand.count()).toBe(1); }); + + it('checkpoints the projection cursor once per event, advancing the offset (P9 slice 7)', async () => { + const { scopeId, event } = await ingestInbound('good morning'); + await automaticBinding(scopeId); + const p = projector(); + + await p.onNormalized(event); + let cursor = await prisma.iiosProjectionCursor.findFirstOrThrow({ where: { projectionName: 'route-projector' } }); + expect(cursor.sourceTopic).toBe(event.type); + expect(cursor.lastOffset).toBe(1); + expect(cursor.lastEventId).toBe(event.id); + + // Replaying the same event does NOT double-advance (claim gate). + await p.onNormalized(event); + cursor = await prisma.iiosProjectionCursor.findFirstOrThrow({ where: { projectionName: 'route-projector' } }); + expect(cursor.lastOffset).toBe(1); + + // A second, distinct normalized event advances the high-water-mark to 2. + const second = await ingestInbound('afternoon all'); + await automaticBinding(second.scopeId); + await p.onNormalized(second.event); + cursor = await prisma.iiosProjectionCursor.findFirstOrThrow({ + where: { projectionName: 'route-projector', partitionKey: second.scopeId }, + }); + expect(cursor.lastOffset).toBe(2); + }); });