maaz519/iios
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(iios): idempotency ledger on outbound egress (409 on key reuse) (P9)

Browse Source 
OutboundService.send is now opt-in idempotent: an explicit idempotencyKey routes
the send through IdempotencyService.run (Slice 6), so reusing a key with a
different target/payload returns a 409 conflict and a same-key/same-request retry
replays the cached command. The inner findUnique early-return stays as a backstop
so a FAILED-then-retried send never collides on the command's unique key. Keyless
sends keep the old behavior (no ledger row).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Maaz Ahamed
2026-07-03 18:44:22 +05:30
parent cdba0f0179
commit a933e2e3af
8 changed files with 79 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/iios-service/src/adapters/adapters.spec.ts
+20 -4
View File
@@ -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);
});
});
packages/iios-service/src/adapters/outbound.service.ts
+46 -33
View File
@@ -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<string, unknown>,
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. */
packages/iios-service/src/calendar/calendar-consent.spec.ts
+2 -1
View File
@@ -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 });
packages/iios-service/src/calendar/calendar-projector.spec.ts
+2 -1
View File
@@ -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<string> {
const m = new MessageService(asService, makeFakePorts(), actors);
packages/iios-service/src/calendar/calendar.spec.ts
+2 -1
View File
@@ -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<string> {
packages/iios-service/src/isolation.spec.ts
+2 -1
View File
@@ -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());
packages/iios-service/src/observability/audit.spec.ts
+2 -1
View File
@@ -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 }> {
packages/iios-service/src/routing/route.spec.ts
+3 -2
View File
@@ -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.