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

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';
import { ProjectionCursorService } from '../projection/projection-cursor.service';
import { makeFakePorts } from '@insignia/iios-testkit';
import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
import { signedFixture, emailFixture, emailInboundFixture, emailReplyFixture } 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';
import { IngestService } from '../interactions/ingest.service';
import { ActorResolver } from '../identity/actor.resolver';
import { OutboxBus } from '../outbox/outbox.bus';
import type { PrismaService } from '../prisma/prisma.service';

const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
const prisma = new PrismaClient({ datasources: { db: { url } } });
const asService = prisma as unknown as PrismaService;
const SECRET = 'dev-adapter-secret'; // AdapterRegistry default

const registry = () => new AdapterRegistry();
const inbound = () => new InboundService(asService, registry());
const projector = () =>
  new RawEventProjector(asService, new OutboxBus(), new DlqService(asService), registry(), new IngestService(asService, makeFakePorts(), new ActorResolver(asService)), new ProjectionCursorService(asService));

async function rawReceivedEvents(): Promise<CloudEvent[]> {
  const rows = await prisma.iiosOutboxEvent.findMany({ where: { eventType: IIOS_EVENTS.rawReceived }, orderBy: { createdAt: 'asc' } });
  return rows.map((r) => r.cloudEvent as unknown as CloudEvent);
}

beforeAll(async () => { await prisma.$connect(); });
afterAll(async () => { await prisma.$disconnect(); });
beforeEach(async () => { await resetDb(prisma); });

describe('Adapter inbound (P5)', () => {
  it('signed webhook → raw stored, then projector normalizes → interaction with the text + traceId', async () => {
    const { body, headers } = signedFixture(emailFixture, SECRET);
    const ack = await inbound().receive('WEBHOOK', body, headers);
    expect(ack.status).toBe('RECEIVED');

    const raw = await prisma.iiosInboundRawEvent.findUniqueOrThrow({ where: { id: ack.rawEventId } });
    expect(raw.signatureStatus).toBe('VERIFIED');

    await projector().onRawReceived((await rawReceivedEvents())[0]!);

    const after = await prisma.iiosInboundRawEvent.findUniqueOrThrow({ where: { id: ack.rawEventId } });
    expect(after.status).toBe('NORMALIZED');
    expect(after.interactionId).toBeTruthy();

    const interaction = await prisma.iiosInteraction.findUniqueOrThrow({
      where: { id: after.interactionId! },
      include: { parts: true },
    });
    expect(interaction.parts[0]?.bodyText).toBe(emailFixture.text);
    expect(interaction.traceId).toBe(raw.traceId); // trace flows raw → interaction
  });

  it('bad signature → REJECTED, nothing ingested (fail-closed)', async () => {
    const { body } = signedFixture(emailFixture, SECRET);
    await expect(inbound().receive('WEBHOOK', body, { 'x-iios-signature': 'sha256=bad' })).rejects.toThrow();
    expect(await prisma.iiosInboundRawEvent.count({ where: { status: 'REJECTED' } })).toBe(1);
    expect(await prisma.iiosInteraction.count()).toBe(0);
  });

  it('replays: same provider eventId twice → one raw event, one interaction', async () => {
    const { body, headers } = signedFixture(emailFixture, SECRET);
    const svc = inbound();
    await svc.receive('WEBHOOK', body, headers);
    const second = await svc.receive('WEBHOOK', body, headers);
    expect(second.deduped).toBe(true);

    await projector().onRawReceived((await rawReceivedEvents())[0]!);
    expect(await prisma.iiosInboundRawEvent.count()).toBe(1);
    expect(await prisma.iiosInteraction.count()).toBe(1);
  });
});

describe('Email adapter — inbound + threading', () => {
  const receiveEmail = async (payload: object) => {
    const { body, headers } = signedFixture(payload, SECRET);
    return inbound().receive('EMAIL', body, headers);
  };
  const normalizeAll = async () => {
    for (const ev of await rawReceivedEvents()) await projector().onRawReceived(ev);
  };

  it('a signed inbound email → a normalized interaction with the body on the email thread', async () => {
    const ack = await receiveEmail(emailInboundFixture);
    expect(ack.status).toBe('RECEIVED');
    await normalizeAll();

    const raw = await prisma.iiosInboundRawEvent.findUniqueOrThrow({ where: { id: ack.rawEventId } });
    expect(raw.status).toBe('NORMALIZED');
    const interaction = await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: raw.interactionId! }, include: { parts: true, thread: true } });
    expect(interaction.parts[0]?.bodyText).toBe(emailInboundFixture.text);
    expect(interaction.thread?.externalThreadRef).toBe('email:<m1@example.com>');
  });

  it('a reply email (In-Reply-To) lands on the SAME email thread as the original', async () => {
    await receiveEmail(emailInboundFixture);
    await receiveEmail(emailReplyFixture);
    await normalizeAll();

    const interactions = await prisma.iiosInteraction.findMany({ orderBy: { occurredAt: 'asc' } });
    expect(interactions).toHaveLength(2);
    expect(interactions[0]!.threadId).toBe(interactions[1]!.threadId); // reply joined the thread
    expect(await prisma.iiosThread.count()).toBe(1); // one email conversation
  });

  it('redelivered email (same Message-ID) → one raw event, one interaction (dedup)', async () => {
    const svc = inbound();
    const { body, headers } = signedFixture(emailInboundFixture, SECRET);
    await svc.receive('EMAIL', body, headers);
    const second = await svc.receive('EMAIL', body, headers);
    expect(second.deduped).toBe(true);
    await normalizeAll();
    expect(await prisma.iiosInboundRawEvent.count()).toBe(1);
    expect(await prisma.iiosInteraction.count()).toBe(1);
  });

  it('bad signature on EMAIL → REJECTED, nothing ingested (fail-closed)', async () => {
    const { body } = signedFixture(emailInboundFixture, SECRET);
    await expect(inbound().receive('EMAIL', body, { 'x-iios-signature': 'sha256=bad' })).rejects.toThrow();
    expect(await prisma.iiosInboundRawEvent.count({ where: { status: 'REJECTED' } })).toBe(1);
    expect(await prisma.iiosInteraction.count()).toBe(0);
  });
});

describe('Adapter outbound (P5)', () => {
  const outbound = (limit?: number): OutboundService => {
    const s = new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService));
    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);
  });

  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()), 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');
  });

  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()), 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');
  });

  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()), 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);
  });

  it('per-tenant quota: a noisy tenant is capped while another still sends (P9 slice 3)', async () => {
    const svc = outbound();
    svc.tenantLimit = 1;
    const a1 = await svc.send('WEBHOOK', 't1', {}, 'ta-1', 'scope-A');
    const a2 = await svc.send('WEBHOOK', 't2', {}, 'ta-2', 'scope-A'); // A over its cap
    const b1 = await svc.send('WEBHOOK', 't3', {}, 'tb-1', 'scope-B'); // B unaffected
    expect(a1.status).toBe('SENT');
    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);
  });
});