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

import { randomUUID } from 'node:crypto';
import { Injectable, NotFoundException, UnauthorizedException } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
import { PrismaService } from '../prisma/prisma.service';
import { AdapterRegistry } from './adapter.registry';

/**
 * Inbound edge (ACK-fast). Verifies the provider signature (fail-closed), stores
 * the raw envelope, emits `raw.received`, and returns immediately (202). The
 * RawEventProjector does the async normalize→ingest. Deduped on
 * (channelType, externalEventId) for provider replay/retries.
 */
@Injectable()
export class InboundService {
  constructor(
    private readonly prisma: PrismaService,
    private readonly registry: AdapterRegistry,
  ) {}

  async receive(
    channelType: string,
    rawBody: string,
    headers: Record<string, string | undefined>,
  ): Promise<{ rawEventId: string; status: string; deduped?: boolean }> {
    const reg = this.registry.get(channelType);
    if (!reg) throw new NotFoundException(`unknown channel type: ${channelType}`);
    const { adapter, config } = reg;

    const payload = this.safeParse(rawBody);
    // The adapter extracts its provider-native id (eventId, Message-ID, …) for replay dedup.
    const externalEventId = adapter.externalEventId?.(payload) ?? (typeof payload?.eventId === 'string' ? payload.eventId : undefined);

    // Fail-closed: a bad/absent signature is recorded and rejected — never ingested.
    if (!adapter.verifySignature(rawBody, headers, config.secret)) {
      const rejected = await this.prisma.iiosInboundRawEvent.create({
        data: {
          channelType,
          externalEventId,
          signatureStatus: 'INVALID',
          status: 'REJECTED',
          headers: headers as unknown as Prisma.InputJsonValue,
          payload: (payload ?? {}) as Prisma.InputJsonValue,
        },
      });
      throw new UnauthorizedException(`invalid signature (raw ${rejected.id})`);
    }

    // Replay: a repeated provider event id returns the existing raw event.
    if (externalEventId) {
      const existing = await this.prisma.iiosInboundRawEvent.findUnique({
        where: { channelType_externalEventId: { channelType, externalEventId } },
      });
      if (existing) return { rawEventId: existing.id, status: existing.status, deduped: true };
    }

    const traceId = randomUUID();
    const raw = await this.prisma.iiosInboundRawEvent.create({
      data: {
        channelType,
        externalEventId,
        signatureStatus: 'VERIFIED',
        status: 'RECEIVED',
        headers: headers as unknown as Prisma.InputJsonValue,
        payload: payload as Prisma.InputJsonValue,
        traceId,
      },
    });

    const event: CloudEvent = {
      specversion: '1.0',
      id: `evt_raw_${raw.id}`,
      type: IIOS_EVENTS.rawReceived,
      source: `iios/adapter/${channelType}`,
      subject: `raw_event/${raw.id}`,
      time: new Date().toISOString(),
      datacontenttype: 'application/json',
      insignia: { correlationId: traceId, idempotencyKey: `raw:${raw.id}`, dataClass: 'internal' },
      data: { rawEventId: raw.id, channelType },
    };
    await this.prisma.iiosOutboxEvent.create({
      data: {
        aggregateType: 'raw_event',
        aggregateId: raw.id,
        eventType: IIOS_EVENTS.rawReceived,
        cloudEvent: event as unknown as Prisma.InputJsonValue,
        partitionKey: `${channelType}:${raw.id}`,
      },
    });

    return { rawEventId: raw.id, status: 'RECEIVED' };
  }

  private safeParse(body: string): Record<string, unknown> | undefined {
    try {
      return JSON.parse(body) as Record<string, unknown>;
    } catch {
      return undefined;
    }
  }
}