From b8b7e0d3e83648c9668c52b90d2665dbf84aa232 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Fri, 3 Jul 2026 02:19:03 +0530 Subject: [PATCH] feat(iios): idempotency-command ledger + IdempotencyService (P9) Adds the doc-mandated IiosIdempotencyCommand table + a reusable IdempotencyService.run() that writes IN_FLIGHT at command start, replays the cached response on a same-key/same-body retry, rejects a same-key/different-body reuse (409), and retries a FAILED command. Tenant-scoped via @@unique([scopeId, commandName, idempotencyKey]) (KG-02/KG-06). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../migration.sql | 23 +++++ packages/iios-service/prisma/schema.prisma | 20 +++++ packages/iios-service/src/app.module.ts | 2 + .../src/idempotency/idempotency.module.ts | 10 +++ .../idempotency/idempotency.service.spec.ts | 89 +++++++++++++++++++ .../src/idempotency/idempotency.service.ts | 86 ++++++++++++++++++ .../iios-service/src/test-utils/reset-db.ts | 2 +- 7 files changed, 231 insertions(+), 1 deletion(-) create mode 100644 packages/iios-service/prisma/migrations/20260702204703_idempotency_command/migration.sql create mode 100644 packages/iios-service/src/idempotency/idempotency.module.ts create mode 100644 packages/iios-service/src/idempotency/idempotency.service.spec.ts create mode 100644 packages/iios-service/src/idempotency/idempotency.service.ts diff --git a/packages/iios-service/prisma/migrations/20260702204703_idempotency_command/migration.sql b/packages/iios-service/prisma/migrations/20260702204703_idempotency_command/migration.sql new file mode 100644 index 0000000..278ab71 --- /dev/null +++ b/packages/iios-service/prisma/migrations/20260702204703_idempotency_command/migration.sql @@ -0,0 +1,23 @@ +-- CreateTable +CREATE TABLE "IiosIdempotencyCommand" ( + "id" TEXT NOT NULL, + "scopeId" TEXT, + "commandName" TEXT NOT NULL, + "idempotencyKey" TEXT NOT NULL, + "schemaVersion" TEXT NOT NULL DEFAULT 'v1', + "actorRefId" TEXT, + "requestHash" TEXT NOT NULL, + "responseHash" TEXT, + "response" JSONB, + "status" TEXT NOT NULL DEFAULT 'IN_FLIGHT', + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "completedAt" TIMESTAMP(3), + + CONSTRAINT "IiosIdempotencyCommand_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "IiosIdempotencyCommand_scopeId_status_idx" ON "IiosIdempotencyCommand"("scopeId", "status"); + +-- CreateIndex +CREATE UNIQUE INDEX "IiosIdempotencyCommand_scopeId_commandName_idempotencyKey_key" ON "IiosIdempotencyCommand"("scopeId", "commandName", "idempotencyKey"); diff --git a/packages/iios-service/prisma/schema.prisma b/packages/iios-service/prisma/schema.prisma index c277e5e..3c185d0 100644 --- a/packages/iios-service/prisma/schema.prisma +++ b/packages/iios-service/prisma/schema.prisma @@ -492,6 +492,26 @@ model IiosProcessedEvent { @@id([consumerName, eventId]) } +/// Idempotent-command ledger (P9). Suppresses duplicate externally-visible mutations: +/// same (scope, command, key) replays the cached response; a different request → conflict. +model IiosIdempotencyCommand { + id String @id @default(cuid()) + scopeId String? + commandName String + idempotencyKey String + schemaVersion String @default("v1") + actorRefId String? + requestHash String + responseHash String? + response Json? + status String @default("IN_FLIGHT") // IN_FLIGHT | COMPLETED | FAILED + createdAt DateTime @default(now()) + completedAt DateTime? + + @@unique([scopeId, commandName, idempotencyKey]) + @@index([scopeId, status]) +} + // ─── P2: messaging — receipts & unread ──────────────────────────── /// Delivery/read receipt for a message interaction, per actor. diff --git a/packages/iios-service/src/app.module.ts b/packages/iios-service/src/app.module.ts index e74c935..257cf25 100644 --- a/packages/iios-service/src/app.module.ts +++ b/packages/iios-service/src/app.module.ts @@ -3,6 +3,7 @@ import { PrismaModule } from './prisma/prisma.module'; 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 { OutboxModule } from './outbox/outbox.module'; import { ThreadsModule } from './threads/threads.module'; import { MessageModule } from './messaging/message.module'; @@ -23,6 +24,7 @@ import { DevController } from './dev/dev.controller'; PlatformModule, IdentityModule, InteractionsModule, + IdempotencyModule, OutboxModule, ThreadsModule, MessageModule, diff --git a/packages/iios-service/src/idempotency/idempotency.module.ts b/packages/iios-service/src/idempotency/idempotency.module.ts new file mode 100644 index 0000000..e60ad8c --- /dev/null +++ b/packages/iios-service/src/idempotency/idempotency.module.ts @@ -0,0 +1,10 @@ +import { Global, Module } from '@nestjs/common'; +import { IdempotencyService } from './idempotency.service'; + +/** Command-idempotency ledger, global so any feature module can wrap its commands. */ +@Global() +@Module({ + providers: [IdempotencyService], + exports: [IdempotencyService], +}) +export class IdempotencyModule {} diff --git a/packages/iios-service/src/idempotency/idempotency.service.spec.ts b/packages/iios-service/src/idempotency/idempotency.service.spec.ts new file mode 100644 index 0000000..6a649e2 --- /dev/null +++ b/packages/iios-service/src/idempotency/idempotency.service.spec.ts @@ -0,0 +1,89 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { ConflictException } from '@nestjs/common'; +import { PrismaClient } from '@prisma/client'; +import { resetDb } from '../test-utils/reset-db'; +import { IdempotencyService } from './idempotency.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 IdempotencyService(asService); +const CMD = 'test.command'; + +beforeAll(async () => { await prisma.$connect(); }); +afterAll(async () => { await prisma.$disconnect(); }); +beforeEach(async () => { await resetDb(prisma); }); + +describe('IdempotencyService (P9 — command ledger)', () => { + it('runs fn once, returns its result, and records a COMPLETED row with the response', async () => { + let calls = 0; + const result = await svc().run({ scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => { + calls++; + return { id: 'r1', ok: true }; + }); + expect(result).toEqual({ id: 'r1', ok: true }); + expect(calls).toBe(1); + const row = await prisma.iiosIdempotencyCommand.findFirstOrThrow({ where: { scopeId: 's1', commandName: CMD, idempotencyKey: 'k1' } }); + expect(row.status).toBe('COMPLETED'); + expect(row.response).toEqual({ id: 'r1', ok: true }); + expect(row.completedAt).toBeTruthy(); + }); + + it('replays the cached response on a duplicate (same request) without re-running fn', async () => { + let calls = 0; + const params = { scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } }; + const first = await svc().run(params, async () => { calls++; return { id: 'r1' }; }); + const second = await svc().run(params, async () => { calls++; return { id: 'r2-should-not-run' }; }); + expect(second).toEqual(first); + expect(calls).toBe(1); // fn never ran the second time + expect(await prisma.iiosIdempotencyCommand.count()).toBe(1); + }); + + it('rejects the same key reused with a different request (fn not called)', async () => { + let calls = 0; + await svc().run({ scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => { calls++; return { id: 'r1' }; }); + await expect( + svc().run({ scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 2 } }, async () => { calls++; return { id: 'r2' }; }), + ).rejects.toBeInstanceOf(ConflictException); + expect(calls).toBe(1); + }); + + it('marks FAILED on throw + propagates, and a later run with the same key retries to COMPLETED', async () => { + let calls = 0; + const params = { scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } }; + await expect( + svc().run(params, async () => { calls++; throw new Error('boom'); }), + ).rejects.toThrow('boom'); + let row = await prisma.iiosIdempotencyCommand.findFirstOrThrow({ where: { scopeId: 's1', commandName: CMD, idempotencyKey: 'k1' } }); + expect(row.status).toBe('FAILED'); + + const retry = await svc().run(params, async () => { calls++; return { id: 'r-ok' }; }); + expect(retry).toEqual({ id: 'r-ok' }); + expect(calls).toBe(2); // failed attempt + successful retry + row = await prisma.iiosIdempotencyCommand.findFirstOrThrow({ where: { scopeId: 's1', commandName: CMD, idempotencyKey: 'k1' } }); + expect(row.status).toBe('COMPLETED'); + }); + + it('a concurrent IN_FLIGHT row → conflict (command in progress)', async () => { + await prisma.iiosIdempotencyCommand.create({ + data: { scopeId: 's1', commandName: CMD, idempotencyKey: 'k1', requestHash: 'irrelevant', status: 'IN_FLIGHT' }, + }); + // Same requestHash so we reach the IN_FLIGHT branch, not the mismatch branch. + const { createHash } = await import('node:crypto'); + const rh = createHash('sha256').update('{"a":1}').digest('hex'); + await prisma.iiosIdempotencyCommand.updateMany({ where: { idempotencyKey: 'k1' }, data: { requestHash: rh } }); + await expect( + svc().run({ scopeId: 's1', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => ({ id: 'r' })), + ).rejects.toThrow(/in progress/); + }); + + it('the same command+key under two different scopes are independent (tenant isolation)', async () => { + const a = await svc().run({ scopeId: 'scope-A', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => ({ id: 'A' })); + const b = await svc().run({ scopeId: 'scope-B', commandName: CMD, key: 'k1', request: { a: 1 } }, async () => ({ id: 'B' })); + expect(a).toEqual({ id: 'A' }); + expect(b).toEqual({ id: 'B' }); // B did not replay A's response + expect(await prisma.iiosIdempotencyCommand.count()).toBe(2); + }); +}); diff --git a/packages/iios-service/src/idempotency/idempotency.service.ts b/packages/iios-service/src/idempotency/idempotency.service.ts new file mode 100644 index 0000000..4ec9471 --- /dev/null +++ b/packages/iios-service/src/idempotency/idempotency.service.ts @@ -0,0 +1,86 @@ +import { ConflictException, Injectable } from '@nestjs/common'; +import { createHash } from 'node:crypto'; +import { PrismaService } from '../prisma/prisma.service'; + +const hash = (v: unknown) => createHash('sha256').update(stable(v)).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(',')}}`; + +export interface IdempotencyParams { + scopeId?: string; + commandName: string; + key: string; + request: unknown; + actorRefId?: string; +} + +/** + * Idempotent-command ledger (P9, KG-06). Wraps an externally-visible mutation so a + * retry with the same (scope, command, key) replays the cached response instead of + * re-executing; the same key with a different request body is rejected as a conflict. + * Write IN_FLIGHT at command start → COMPLETED (+ cached response) or FAILED on throw. + */ +@Injectable() +export class IdempotencyService { + constructor(private readonly prisma: PrismaService) {} + + async run(p: IdempotencyParams, fn: () => Promise): Promise { + const requestHash = hash(p.request); + const scopeId = p.scopeId ?? null; + const where = { scopeId, commandName: p.commandName, idempotencyKey: p.key }; + + const existing = await this.prisma.iiosIdempotencyCommand.findFirst({ where }); + if (existing) return this.resolveExisting(where, existing, requestHash, fn); + + try { + await this.prisma.iiosIdempotencyCommand.create({ + data: { scopeId, commandName: p.commandName, idempotencyKey: p.key, requestHash, actorRefId: p.actorRefId }, + }); + } catch { + // Lost the create race: another caller inserted the same (scope, command, key). + const row = await this.prisma.iiosIdempotencyCommand.findFirst({ where }); + if (row) return this.resolveExisting(where, row, requestHash, fn); + throw new ConflictException('idempotent command conflict'); + } + + return this.executeAndRecord(where, fn); + } + + private async resolveExisting( + where: { scopeId: string | null; commandName: string; idempotencyKey: string }, + row: { status: string; requestHash: string; response: unknown }, + requestHash: string, + fn: () => Promise, + ): Promise { + if (row.requestHash !== requestHash) throw new ConflictException('idempotency key reused with a different request'); + if (row.status === 'COMPLETED') return row.response as T; // replay — no re-execution + if (row.status === 'IN_FLIGHT') throw new ConflictException('command in progress'); + return this.executeAndRecord(where, fn); // FAILED → retry + re-record the outcome + } + + /** Run the command, then record COMPLETED (+ cached response) or FAILED against its ledger row. */ + private async executeAndRecord( + where: { scopeId: string | null; commandName: string; idempotencyKey: string }, + fn: () => Promise, + ): Promise { + try { + const result = await fn(); + await this.prisma.iiosIdempotencyCommand.updateMany({ + where, + data: { status: 'COMPLETED', responseHash: hash(result), response: result as object, completedAt: new Date() }, + }); + return result; + } catch (err) { + await this.prisma.iiosIdempotencyCommand.updateMany({ where, data: { status: 'FAILED', completedAt: new Date() } }); + throw err; + } + } +} diff --git a/packages/iios-service/src/test-utils/reset-db.ts b/packages/iios-service/src/test-utils/reset-db.ts index 5ce8371..2327aab 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 - "IiosDlqItem","IiosAuditLink", + "IiosIdempotencyCommand","IiosDlqItem","IiosAuditLink", "IiosMeetingActionItem","IiosActionItem","IiosTranscriptSegment","IiosMeetingTranscript","IiosMeetingSummary","IiosMeetingParticipant","IiosMeeting","IiosMeetingRequest", "IiosCalendarSyncCursor","IiosCalendarEvent","IiosCalendarProviderAccount","IiosAvailabilityWindow", "IiosAiEvidenceLink","IiosAiClaim","IiosAiToolCall","IiosAiArtifact","IiosAiModelRun","IiosAiJob","IiosEmbeddingRef",