diff --git a/packages/iios-service/src/adapters/adapters.controller.ts b/packages/iios-service/src/adapters/adapters.controller.ts index 1a24b83..e23a58d 100644 --- a/packages/iios-service/src/adapters/adapters.controller.ts +++ b/packages/iios-service/src/adapters/adapters.controller.ts @@ -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; 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); + } } diff --git a/packages/iios-service/src/adapters/adapters.dto.ts b/packages/iios-service/src/adapters/adapters.dto.ts new file mode 100644 index 0000000..c82bab7 --- /dev/null +++ b/packages/iios-service/src/adapters/adapters.dto.ts @@ -0,0 +1,7 @@ +import { IsObject, IsOptional, IsString } from 'class-validator'; + +export class SendDto { + @IsString() target!: string; + @IsOptional() @IsObject() payload?: Record; + @IsOptional() @IsString() idempotencyKey?: string; +} diff --git a/packages/iios-service/src/adapters/adapters.module.ts b/packages/iios-service/src/adapters/adapters.module.ts index bf5e0c0..71b3900 100644 --- a/packages/iios-service/src/adapters/adapters.module.ts +++ b/packages/iios-service/src/adapters/adapters.module.ts @@ -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 {} diff --git a/packages/iios-service/src/adapters/adapters.spec.ts b/packages/iios-service/src/adapters/adapters.spec.ts index 0130881..3abe49b 100644 --- a/packages/iios-service/src/adapters/adapters.spec.ts +++ b/packages/iios-service/src/adapters/adapters.spec.ts @@ -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); + }); +}); diff --git a/packages/iios-service/src/adapters/outbound.service.ts b/packages/iios-service/src/adapters/outbound.service.ts new file mode 100644 index 0000000..c5f319e --- /dev/null +++ b/packages/iios-service/src/adapters/outbound.service.ts @@ -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, + 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 { + 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; + } +}