iios/packages/iios-service/src/adapters/outbound.service.ts

import { randomUUID } from 'node:crypto';
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';

/**
 * Outbound command layer (P5). Owns idempotency + per-(channelType,target) rate
 * limiting + the delivery ledger. As of P9 the actual execute step is delegated to
 * the governed CapabilityBroker (policy gate + obligations + provider selection);
 * this layer no longer talks to a provider directly.
 */
@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);
  /** Per-tenant (scope) egress cap — noisy-neighbor protection (P9). */
  tenantLimit = Number(process.env.IIOS_TENANT_OUTBOUND_LIMIT ?? 10_000);
  tenantWindowMs = Number(process.env.IIOS_TENANT_OUTBOUND_WINDOW_MS ?? 60_000);

  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,
    scopeId?: string,
    purpose?: string,
  ) {
    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;
      }

      const cmd = await this.prisma.iiosOutboundCommand.create({
        data: { channelType, target, scopeId, payload: payload as Prisma.InputJsonValue, status: 'PENDING', idempotencyKey: key },
      });

      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;
      }

      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. */
  private async allowTenant(scopeId?: string): Promise<boolean> {
    if (!scopeId) return true; // unscoped sends are not tenant-limited
    const since = new Date(Date.now() - this.tenantWindowMs);
    const count = await this.prisma.iiosOutboundCommand.count({ where: { scopeId, createdAt: { gte: since } } });
    return count < this.tenantLimit;
  }

  /** 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;
  }
}