diff --git a/packages/iios-service/src/adapters/adapters.spec.ts b/packages/iios-service/src/adapters/adapters.spec.ts index 7bd68d4..9c793b5 100644 --- a/packages/iios-service/src/adapters/adapters.spec.ts +++ b/packages/iios-service/src/adapters/adapters.spec.ts @@ -1,4 +1,5 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { ConflictException } from '@nestjs/common'; import { PrismaClient } from '@prisma/client'; import { DlqService } from '../outbox/dlq.service'; import { resetDb } from '../test-utils/reset-db'; @@ -9,6 +10,7 @@ import { signedFixture, emailFixture } from '@insignia/iios-adapter-sdk'; import { AdapterRegistry } from './adapter.registry'; import { InboundService } from './inbound.service'; import { OutboundService } from './outbound.service'; +import { IdempotencyService } from '../idempotency/idempotency.service'; import { CapabilityBroker } from '../capability/capability.broker'; import { CapabilityProviderRegistry } from '../capability/capability.registry'; import { RawEventProjector } from './raw-event.projector'; @@ -81,7 +83,7 @@ describe('Adapter inbound (P5)', () => { describe('Adapter outbound (P5)', () => { const outbound = (limit?: number): OutboundService => { - const s = new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())); + const s = new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)); if (limit) s.limit = limit; return s; }; @@ -112,7 +114,7 @@ describe('Adapter outbound (P5)', () => { it('broker obligation DENY_OUTBOUND → command BLOCKED, no send (P9)', async () => { const ports = makeFakePorts(); ports.opa.setDecision({ allow: true, obligations: [{ kind: 'DENY_OUTBOUND', reason: 'recipient opted out' }] }); - const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry())); + const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()), new IdempotencyService(asService)); const cmd = await svc.send('WEBHOOK', 'user@x', { text: 'hi' }, 'blk-1'); expect(cmd.status).toBe('BLOCKED'); expect(cmd.providerRef).toBe('blocked'); @@ -121,7 +123,7 @@ describe('Adapter outbound (P5)', () => { it('CMP allow → command records the consent receipt (P9 slice 2)', async () => { const ports = makeFakePorts(); ports.cmp.setStatus('ALLOW'); - const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry())); + const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()), new IdempotencyService(asService)); const cmd = await svc.send('WEBHOOK', 'user@x', { text: 'hi' }, 'consent-1', 'scope1', 'support'); expect(cmd.status).toBe('SENT'); expect(cmd.consentReceiptRef).toBe('fake-receipt'); @@ -130,7 +132,7 @@ describe('Adapter outbound (P5)', () => { it('CMP DENY for the purpose → command BLOCKED, no send (P9 slice 2)', async () => { const ports = makeFakePorts(); ports.cmp.denyPurpose('marketing'); - const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry())); + const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()), new IdempotencyService(asService)); const cmd = await svc.send('WEBHOOK', 'user@x', { text: 'promo' }, 'consent-2', 'scope1', 'marketing'); expect(cmd.status).toBe('BLOCKED'); expect(await prisma.iiosDeliveryAttempt.count({ where: { commandId: cmd.id, status: 'SENT' } })).toBe(0); @@ -146,4 +148,18 @@ describe('Adapter outbound (P5)', () => { expect(a2.status).toBe('RATE_LIMITED'); expect(b1.status).toBe('SENT'); }); + + it('idempotency ledger: same key + a different target → conflict, no second command (P9 slice 10)', async () => { + const svc = outbound(); + const first = await svc.send('WEBHOOK', 'user@a', { text: 'hi' }, 'dup-key'); + expect(first.status).toBe('SENT'); + await expect(svc.send('WEBHOOK', 'user@b', { text: 'hi' }, 'dup-key')).rejects.toBeInstanceOf(ConflictException); + expect(await prisma.iiosOutboundCommand.count({ where: { idempotencyKey: 'dup-key' } })).toBe(1); + }); + + it('opt-in: a keyless send creates a command but writes no idempotency-ledger row (P9 slice 10)', async () => { + const cmd = await outbound().send('WEBHOOK', 'user@x', { text: 'hi' }); + expect(cmd.status).toBe('SENT'); + expect(await prisma.iiosIdempotencyCommand.count()).toBe(0); + }); }); diff --git a/packages/iios-service/src/adapters/outbound.service.ts b/packages/iios-service/src/adapters/outbound.service.ts index 21f7e78..c0370c8 100644 --- a/packages/iios-service/src/adapters/outbound.service.ts +++ b/packages/iios-service/src/adapters/outbound.service.ts @@ -3,6 +3,7 @@ import { Injectable } from '@nestjs/common'; import { Prisma } from '@prisma/client'; import { PrismaService } from '../prisma/prisma.service'; import { CapabilityBroker } from '../capability/capability.broker'; +import { IdempotencyService } from '../idempotency/idempotency.service'; import { recordAudit } from '../observability/audit'; /** @@ -23,55 +24,67 @@ export class OutboundService { constructor( private readonly prisma: PrismaService, private readonly broker: CapabilityBroker, + private readonly idempotency: IdempotencyService, ) {} async send( channelType: string, target: string, payload: Record, - idempotencyKey: string = randomUUID(), + idempotencyKey?: string, scopeId?: string, purpose?: string, ) { - const existing = await this.prisma.iiosOutboundCommand.findUnique({ where: { idempotencyKey } }); - if (existing) return existing; + const key = idempotencyKey ?? randomUUID(); + + // The actual send. The inner findUnique stays as a backstop so a FAILED-then-retried + // command returns the existing row instead of colliding on idempotencyKey @unique. + const body = async () => { + const existing = await this.prisma.iiosOutboundCommand.findUnique({ where: { idempotencyKey: key } }); + if (existing) return existing; + + // Per-target rate limit AND per-tenant quota (a noisy tenant can't starve shared egress). + if (!(await this.allow(channelType, target)) || !(await this.allowTenant(scopeId))) { + const cmd = await this.prisma.iiosOutboundCommand.create({ + data: { channelType, target, scopeId, payload: payload as Prisma.InputJsonValue, status: 'RATE_LIMITED', idempotencyKey: key }, + }); + await this.prisma.iiosDeliveryAttempt.create({ data: { commandId: cmd.id, attemptNo: 1, status: 'RATE_LIMITED' } }); + return cmd; + } - // Per-target rate limit AND per-tenant quota (a noisy tenant can't starve shared egress). - if (!(await this.allow(channelType, target)) || !(await this.allowTenant(scopeId))) { const cmd = await this.prisma.iiosOutboundCommand.create({ - data: { channelType, target, scopeId, payload: payload as Prisma.InputJsonValue, status: 'RATE_LIMITED', idempotencyKey }, + data: { channelType, target, scopeId, payload: payload as Prisma.InputJsonValue, status: 'PENDING', idempotencyKey: key }, }); - await this.prisma.iiosDeliveryAttempt.create({ data: { commandId: cmd.id, attemptNo: 1, status: 'RATE_LIMITED' } }); - return cmd; - } - const cmd = await this.prisma.iiosOutboundCommand.create({ - data: { channelType, target, scopeId, payload: payload as Prisma.InputJsonValue, status: 'PENDING', idempotencyKey }, - }); + let result; + try { + result = await this.broker.invoke({ capability: 'channel.send', channelType, target, payload, idempotencyKey: key, scopeId, purpose }); + } catch (err) { + // Fail-closed (e.g. PolicyDeniedError): mark the command FAILED, record the attempt, propagate. + await this.prisma.$transaction([ + this.prisma.iiosOutboundCommand.update({ where: { id: cmd.id }, data: { status: 'FAILED' } }), + this.prisma.iiosDeliveryAttempt.create({ data: { commandId: cmd.id, attemptNo: 1, status: 'FAILED', errorCode: (err as Error).name } }), + ]); + throw err; + } - let result; - try { - result = await this.broker.invoke({ capability: 'channel.send', channelType, target, payload, idempotencyKey, scopeId, purpose }); - } catch (err) { - // Fail-closed (e.g. PolicyDeniedError): mark the command FAILED, record the attempt, propagate. - await this.prisma.$transaction([ - this.prisma.iiosOutboundCommand.update({ where: { id: cmd.id }, data: { status: 'FAILED' } }), - this.prisma.iiosDeliveryAttempt.create({ data: { commandId: cmd.id, attemptNo: 1, status: 'FAILED', errorCode: (err as Error).name } }), + const [updated] = await this.prisma.$transaction([ + this.prisma.iiosOutboundCommand.update({ + where: { id: cmd.id }, + data: { status: result.outcome, providerRef: result.providerRef, consentReceiptRef: result.consentReceiptRef }, + }), + this.prisma.iiosDeliveryAttempt.create({ + data: { commandId: cmd.id, attemptNo: 1, status: result.outcome, providerRef: result.providerRef, latencyMs: result.latencyMs, errorCode: result.errorCode }, + }), ]); - throw err; - } + await recordAudit(this.prisma, { action: 'channel.send', resourceType: 'outbound_command', resourceId: cmd.id, scopeId }); + return updated; + }; - const [updated] = await this.prisma.$transaction([ - this.prisma.iiosOutboundCommand.update({ - where: { id: cmd.id }, - data: { status: result.outcome, providerRef: result.providerRef, consentReceiptRef: result.consentReceiptRef }, - }), - this.prisma.iiosDeliveryAttempt.create({ - data: { commandId: cmd.id, attemptNo: 1, status: result.outcome, providerRef: result.providerRef, latencyMs: result.latencyMs, errorCode: result.errorCode }, - }), - ]); - await recordAudit(this.prisma, { action: 'channel.send', resourceType: 'outbound_command', resourceId: cmd.id, scopeId }); - return updated; + // Opt-in idempotency: an explicit key routes through the ledger (request-hash conflict → 409, + // response replay); a keyless send just runs with a fresh uuid (old behavior). + if (!idempotencyKey) return body(); + return this.idempotency.run({ scopeId, commandName: 'channel.send', key: idempotencyKey, request: { channelType, target, payload, purpose } }, body); } /** Per-tenant egress quota (P9): count this scope's commands in the window vs the cap. */ diff --git a/packages/iios-service/src/calendar/calendar-consent.spec.ts b/packages/iios-service/src/calendar/calendar-consent.spec.ts index 9620496..5b52d1b 100644 --- a/packages/iios-service/src/calendar/calendar-consent.spec.ts +++ b/packages/iios-service/src/calendar/calendar-consent.spec.ts @@ -7,6 +7,7 @@ import { ActorResolver } from '../identity/actor.resolver'; import { AiJobService } from '../ai/ai.service'; import { AiBudgetGuard } from '../ai/ai-budget.guard'; import { OutboundService } from '../adapters/outbound.service'; +import { IdempotencyService } from '../idempotency/idempotency.service'; import { CapabilityBroker } from '../capability/capability.broker'; import { CapabilityProviderRegistry } from '../capability/capability.registry'; import { CalendarService } from './calendar.service'; @@ -22,7 +23,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po const START = '2026-07-02T10:00:00.000Z'; const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors); -const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()))); +const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService))); async function scheduledMeeting(attendees: { userId: string; visibility?: 'FULL' | 'NONE' }[]) { const m = await cal().scheduleDirect(alice, { meetingType: 'INTERNAL', title: 'Board sync', startAt: START, attendees }); diff --git a/packages/iios-service/src/calendar/calendar-projector.spec.ts b/packages/iios-service/src/calendar/calendar-projector.spec.ts index ffe2b05..8496410 100644 --- a/packages/iios-service/src/calendar/calendar-projector.spec.ts +++ b/packages/iios-service/src/calendar/calendar-projector.spec.ts @@ -11,6 +11,7 @@ import { ActorResolver } from '../identity/actor.resolver'; import { AiJobService } from '../ai/ai.service'; import { AiBudgetGuard } from '../ai/ai-budget.guard'; import { OutboundService } from '../adapters/outbound.service'; +import { IdempotencyService } from '../idempotency/idempotency.service'; import { CapabilityBroker } from '../capability/capability.broker'; import { CapabilityProviderRegistry } from '../capability/capability.registry'; import { CalendarService } from './calendar.service'; @@ -27,7 +28,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po const START = '2026-07-02T10:00:00.000Z'; const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors); -const cal = () => new CalendarService(asService, makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()))); +const cal = () => new CalendarService(asService, makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService))); async function makeInteraction(text: string): Promise { const m = new MessageService(asService, makeFakePorts(), actors); diff --git a/packages/iios-service/src/calendar/calendar.spec.ts b/packages/iios-service/src/calendar/calendar.spec.ts index 1b43692..2f15c99 100644 --- a/packages/iios-service/src/calendar/calendar.spec.ts +++ b/packages/iios-service/src/calendar/calendar.spec.ts @@ -9,6 +9,7 @@ import { ActorResolver } from '../identity/actor.resolver'; import { AiJobService } from '../ai/ai.service'; import { AiBudgetGuard } from '../ai/ai-budget.guard'; import { OutboundService } from '../adapters/outbound.service'; +import { IdempotencyService } from '../idempotency/idempotency.service'; import { CapabilityBroker } from '../capability/capability.broker'; import { CapabilityProviderRegistry } from '../capability/capability.registry'; import { CalendarService } from './calendar.service'; @@ -24,7 +25,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po const START = '2026-07-02T10:00:00.000Z'; const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors); -const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())); +const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)); const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), outbound()); async function makeInteraction(text: string): Promise { diff --git a/packages/iios-service/src/isolation.spec.ts b/packages/iios-service/src/isolation.spec.ts index f5ad4c3..546f32f 100644 --- a/packages/iios-service/src/isolation.spec.ts +++ b/packages/iios-service/src/isolation.spec.ts @@ -12,6 +12,7 @@ import { AiJobService } from './ai/ai.service'; import { AiBudgetGuard } from './ai/ai-budget.guard'; import { CalendarService } from './calendar/calendar.service'; import { OutboundService } from './adapters/outbound.service'; +import { IdempotencyService } from './idempotency/idempotency.service'; import { CapabilityBroker } from './capability/capability.broker'; import { CapabilityProviderRegistry } from './capability/capability.registry'; import type { PrismaService } from './prisma/prisma.service'; @@ -27,7 +28,7 @@ const B: MessagePrincipal = { userId: 'b1', orgId: 'org_B', appId: 'portal-demo' const START = '2026-07-02T10:00:00.000Z'; const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors); -const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())); +const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)); const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), outbound(), actors); const cal = () => new CalendarService(asService, makeFakePorts(), actors, ai(), outbound()); diff --git a/packages/iios-service/src/observability/audit.spec.ts b/packages/iios-service/src/observability/audit.spec.ts index 636cbea..826de35 100644 --- a/packages/iios-service/src/observability/audit.spec.ts +++ b/packages/iios-service/src/observability/audit.spec.ts @@ -11,6 +11,7 @@ import { RouteService } from '../routing/route.service'; import { AiJobService } from '../ai/ai.service'; import { AiBudgetGuard } from '../ai/ai-budget.guard'; import { OutboundService } from '../adapters/outbound.service'; +import { IdempotencyService } from '../idempotency/idempotency.service'; import { CapabilityBroker } from '../capability/capability.broker'; import { CapabilityProviderRegistry } from '../capability/capability.registry'; import { runWithTrace } from './trace-context'; @@ -23,7 +24,7 @@ const actors = new ActorResolver(asService); const user: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo' }; const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors); -const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())); +const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)); const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), outbound(), actors); async function makeInteraction(text: string): Promise<{ interactionId: string; scopeId: string }> { diff --git a/packages/iios-service/src/routing/route.spec.ts b/packages/iios-service/src/routing/route.spec.ts index c156ffd..6faecae 100644 --- a/packages/iios-service/src/routing/route.spec.ts +++ b/packages/iios-service/src/routing/route.spec.ts @@ -13,6 +13,7 @@ import { RouteService } from './route.service'; import { RouteProjector } from './route.projector'; import { IngestService } from '../interactions/ingest.service'; import { OutboundService } from '../adapters/outbound.service'; +import { IdempotencyService } from '../idempotency/idempotency.service'; import { CapabilityBroker } from '../capability/capability.broker'; import { CapabilityProviderRegistry } from '../capability/capability.registry'; import { OutboxBus } from '../outbox/outbox.bus'; @@ -106,7 +107,7 @@ describe('Route simulator (P6, preview-first)', () => { describe('RouteService approve/deny → execute (P6)', () => { const admin: MessagePrincipal = { userId: 'admin', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Admin' }; - const svc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())), actors); + const svc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)), actors); it('approving a REVIEW decision executes the forward to the sandbox', async () => { const { interactionId, scopeId } = await makeInteraction('party at 9 PM'); @@ -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 routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)), actors); const projector = () => new RouteProjector(asService, new OutboxBus(), new DlqService(asService), routeSvc(), new ProjectionCursorService(asService)); // Ingest an inbound (channel-bearing) interaction — routable, unlike native DMs.