feat(service): P5.4 outbound sandbox sink + rate limit + inspector read endpoints
OutboundService: idempotent commands, per-(channelType,target) rate-limit bucket, sandbox send (no network) + delivery attempts. POST /v1/adapters/:type/send + GET inbound/outbound (session-auth). 55 tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,10 +1,19 @@
|
||||
import { Controller, HttpCode, Param, Post, Req } from '@nestjs/common';
|
||||
import { BadRequestException, Body, Controller, Get, Headers, HttpCode, Param, Post, Req } from '@nestjs/common';
|
||||
import type { Request } from 'express';
|
||||
import { InboundService } from './inbound.service';
|
||||
import { OutboundService } from './outbound.service';
|
||||
import { SessionVerifier } from '../platform/session.verifier';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { SendDto } from './adapters.dto';
|
||||
|
||||
@Controller('v1/adapters')
|
||||
export class AdaptersController {
|
||||
constructor(private readonly inbound: InboundService) {}
|
||||
constructor(
|
||||
private readonly inbound: InboundService,
|
||||
private readonly outbound: OutboundService,
|
||||
private readonly session: SessionVerifier,
|
||||
private readonly prisma: PrismaService,
|
||||
) {}
|
||||
|
||||
/** Provider webhook. Authenticated by HMAC signature (not a session token). */
|
||||
@Post(':channelType/webhook')
|
||||
@@ -14,4 +23,30 @@ export class AdaptersController {
|
||||
const headers = req.headers as Record<string, string | undefined>;
|
||||
return this.inbound.receive(channelType, rawBody, headers);
|
||||
}
|
||||
|
||||
/** Outbound to the sandbox sink (session-auth). No real network. */
|
||||
@Post(':channelType/send')
|
||||
async send(@Param('channelType') channelType: string, @Body() body: SendDto, @Headers('authorization') auth?: string) {
|
||||
this.requireSession(auth);
|
||||
return this.outbound.send(channelType, body.target, body.payload ?? {}, body.idempotencyKey);
|
||||
}
|
||||
|
||||
// ─── inspector reads (session-auth) ───────────────────────────────
|
||||
@Get('inbound')
|
||||
async listInbound(@Headers('authorization') auth?: string) {
|
||||
this.requireSession(auth);
|
||||
return this.prisma.iiosInboundRawEvent.findMany({ orderBy: { createdAt: 'desc' }, take: 50 });
|
||||
}
|
||||
|
||||
@Get('outbound')
|
||||
async listOutbound(@Headers('authorization') auth?: string) {
|
||||
this.requireSession(auth);
|
||||
return this.prisma.iiosOutboundCommand.findMany({ orderBy: { createdAt: 'desc' }, take: 50, include: { attempts: true } });
|
||||
}
|
||||
|
||||
private requireSession(authorization?: string): void {
|
||||
const token = (authorization ?? '').replace(/^Bearer\s+/i, '');
|
||||
if (!token) throw new BadRequestException('Authorization bearer token is required');
|
||||
this.session.verify(token);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
import { IsObject, IsOptional, IsString } from 'class-validator';
|
||||
|
||||
export class SendDto {
|
||||
@IsString() target!: string;
|
||||
@IsOptional() @IsObject() payload?: Record<string, unknown>;
|
||||
@IsOptional() @IsString() idempotencyKey?: string;
|
||||
}
|
||||
@@ -3,13 +3,14 @@ import { OutboxModule } from '../outbox/outbox.module';
|
||||
import { InteractionsModule } from '../interactions/interactions.module';
|
||||
import { AdapterRegistry } from './adapter.registry';
|
||||
import { InboundService } from './inbound.service';
|
||||
import { OutboundService } from './outbound.service';
|
||||
import { RawEventProjector } from './raw-event.projector';
|
||||
import { AdaptersController } from './adapters.controller';
|
||||
|
||||
@Module({
|
||||
imports: [OutboxModule, InteractionsModule],
|
||||
controllers: [AdaptersController],
|
||||
providers: [AdapterRegistry, InboundService, RawEventProjector],
|
||||
exports: [AdapterRegistry, InboundService],
|
||||
providers: [AdapterRegistry, InboundService, OutboundService, RawEventProjector],
|
||||
exports: [AdapterRegistry, InboundService, OutboundService],
|
||||
})
|
||||
export class AdaptersModule {}
|
||||
|
||||
@@ -6,6 +6,7 @@ import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
|
||||
import { signedFixture, emailFixture } from '@insignia/iios-adapter-sdk';
|
||||
import { AdapterRegistry } from './adapter.registry';
|
||||
import { InboundService } from './inbound.service';
|
||||
import { OutboundService } from './outbound.service';
|
||||
import { RawEventProjector } from './raw-event.projector';
|
||||
import { IngestService } from '../interactions/ingest.service';
|
||||
import { ActorResolver } from '../identity/actor.resolver';
|
||||
@@ -73,3 +74,34 @@ describe('Adapter inbound (P5)', () => {
|
||||
expect(await prisma.iiosInteraction.count()).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('Adapter outbound (P5)', () => {
|
||||
const outbound = (limit?: number): OutboundService => {
|
||||
const s = new OutboundService(asService, registry());
|
||||
if (limit) s.limit = limit;
|
||||
return s;
|
||||
};
|
||||
|
||||
it('send → command SENT + one delivery attempt (sandbox, no network)', async () => {
|
||||
const cmd = await outbound().send('WEBHOOK', 'user@x', { text: 'hi' }, 'ob-1');
|
||||
expect(cmd.status).toBe('SENT');
|
||||
expect(cmd.providerRef).toContain('sim-');
|
||||
expect(await prisma.iiosDeliveryAttempt.count({ where: { commandId: cmd.id } })).toBe(1);
|
||||
});
|
||||
|
||||
it('rate limit: exceeding the window → RATE_LIMITED', async () => {
|
||||
const svc = outbound(2);
|
||||
await svc.send('WEBHOOK', 't', {}, 'a');
|
||||
await svc.send('WEBHOOK', 't', {}, 'b');
|
||||
const third = await svc.send('WEBHOOK', 't', {}, 'c');
|
||||
expect(third.status).toBe('RATE_LIMITED');
|
||||
});
|
||||
|
||||
it('idempotent: same key twice → one command', async () => {
|
||||
const svc = outbound();
|
||||
const a = await svc.send('WEBHOOK', 't2', {}, 'k');
|
||||
const b = await svc.send('WEBHOOK', 't2', {}, 'k');
|
||||
expect(b.id).toBe(a.id);
|
||||
expect(await prisma.iiosOutboundCommand.count()).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { Injectable, NotFoundException } from '@nestjs/common';
|
||||
import { Prisma } from '@prisma/client';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { AdapterRegistry } from './adapter.registry';
|
||||
|
||||
/**
|
||||
* Outbound to a provider — in P5 the adapter's `send` is a SANDBOX sink (no real
|
||||
* network). Per-(channelType,target) rate limiting; idempotent per command key;
|
||||
* every attempt recorded. Real delivery is gated to P6+.
|
||||
*/
|
||||
@Injectable()
|
||||
export class OutboundService {
|
||||
/** Public so tests can lower them; default from env. */
|
||||
limit = Number(process.env.IIOS_OUTBOUND_LIMIT ?? 5);
|
||||
windowMs = Number(process.env.IIOS_OUTBOUND_WINDOW_MS ?? 60_000);
|
||||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly registry: AdapterRegistry,
|
||||
) {}
|
||||
|
||||
async send(
|
||||
channelType: string,
|
||||
target: string,
|
||||
payload: Record<string, unknown>,
|
||||
idempotencyKey: string = randomUUID(),
|
||||
) {
|
||||
const reg = this.registry.get(channelType);
|
||||
if (!reg) throw new NotFoundException(`unknown channel type: ${channelType}`);
|
||||
|
||||
const existing = await this.prisma.iiosOutboundCommand.findUnique({ where: { idempotencyKey } });
|
||||
if (existing) return existing;
|
||||
|
||||
if (!(await this.allow(channelType, target))) {
|
||||
const cmd = await this.prisma.iiosOutboundCommand.create({
|
||||
data: { channelType, target, payload: payload as Prisma.InputJsonValue, status: 'RATE_LIMITED', idempotencyKey },
|
||||
});
|
||||
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, payload: payload as Prisma.InputJsonValue, status: 'PENDING', idempotencyKey },
|
||||
});
|
||||
const result = await reg.adapter.send({ channelType, target, payload }); // sandbox — no network
|
||||
const [updated] = await this.prisma.$transaction([
|
||||
this.prisma.iiosOutboundCommand.update({ where: { id: cmd.id }, data: { status: 'SENT', providerRef: result.providerRef } }),
|
||||
this.prisma.iiosDeliveryAttempt.create({
|
||||
data: { commandId: cmd.id, attemptNo: 1, status: 'SENT', providerRef: result.providerRef, latencyMs: result.latencyMs },
|
||||
}),
|
||||
]);
|
||||
return updated;
|
||||
}
|
||||
|
||||
/** Token/window bucket keyed by (channelType, target). Resets when the window elapses. */
|
||||
private async allow(channelType: string, target: string): Promise<boolean> {
|
||||
const now = new Date();
|
||||
const where = { channelType_bucketKey: { channelType, bucketKey: target } };
|
||||
const bucket = await this.prisma.iiosRateLimitBucket.findUnique({ where });
|
||||
if (!bucket || bucket.resetAt <= now) {
|
||||
await this.prisma.iiosRateLimitBucket.upsert({
|
||||
where,
|
||||
create: { channelType, bucketKey: target, windowStart: now, count: 1, resetAt: new Date(now.getTime() + this.windowMs) },
|
||||
update: { windowStart: now, count: 1, resetAt: new Date(now.getTime() + this.windowMs) },
|
||||
});
|
||||
return true;
|
||||
}
|
||||
if (bucket.count >= this.limit) return false;
|
||||
await this.prisma.iiosRateLimitBucket.update({ where, data: { count: { increment: 1 } } });
|
||||
return true;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user