feat(service): ingest endpoint with kernel gates (P1.4)

POST /v1/interactions/ingest: fail-closed OPA gate -> resolve scope/handle(MDM, no silent merge)/actor/channel/thread -> idempotent write of interaction+parts+ outbox CloudEvent in one tx. 4 DB-backed gate tests green (idempotent / fail-closed / unresolved-handle / outbox event). 16 tests total. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 21:24:09 +05:30
parent 8cd5d9b248
commit 4b7dc98c6b
6 changed files with 391 additions and 1 deletions
@@ -1,10 +1,11 @@
 import { Module } from '@nestjs/common';
 import { PrismaModule } from './prisma/prisma.module';
 import { PlatformModule } from './platform/platform.module';
+import { InteractionsModule } from './interactions/interactions.module';
 import { HealthController } from './health.controller';

@Module({
-  imports: [PrismaModule, PlatformModule],
+  imports: [PrismaModule, PlatformModule, InteractionsModule],
  controllers: [HealthController],
 })
 export class AppModule {}
@@ -0,0 +1,56 @@
+import { Type } from 'class-transformer';
+import {
+  IsArray,
+  IsISO8601,
+  IsObject,
+  IsOptional,
+  IsString,
+  ValidateNested,
+} from 'class-validator';
+
+class ScopeDto {
+  @IsString() orgId!: string;
+  @IsString() appId!: string;
+  @IsOptional() @IsString() buId?: string;
+  @IsOptional() @IsString() tenantId?: string;
+  @IsOptional() @IsString() workspaceId?: string;
+  @IsOptional() @IsString() projectId?: string;
+  @IsOptional() @IsString() userScopeId?: string;
+}
+
+class ChannelDto {
+  @IsString() type!: string;
+  @IsOptional() @IsString() externalChannelId?: string;
+  @IsOptional() @IsObject() capabilityContract?: Record<string, unknown>;
+}
+
+class SourceDto {
+  @IsString() handleKind!: string;
+  @IsString() externalId!: string;
+  @IsOptional() @IsString() displayName?: string;
+}
+
+class ThreadDto {
+  @IsOptional() @IsString() externalThreadId?: string;
+  @IsOptional() @IsString() subject?: string;
+}
+
+class PartDto {
+  @IsString() kind!: string;
+  @IsOptional() @IsString() bodyText?: string;
+  @IsOptional() @IsString() contentRef?: string;
+  @IsOptional() @IsString() mimeType?: string;
+}
+
+/** Validates the POST /v1/interactions/ingest body (conforms to IngestInteractionRequest). */
+export class IngestRequestDto {
+  @ValidateNested() @Type(() => ScopeDto) scope!: ScopeDto;
+  @ValidateNested() @Type(() => ChannelDto) channel!: ChannelDto;
+  @ValidateNested() @Type(() => SourceDto) source!: SourceDto;
+  @IsOptional() @IsString() kind?: string;
+  @IsOptional() @ValidateNested() @Type(() => ThreadDto) thread?: ThreadDto;
+  @IsArray() @ValidateNested({ each: true }) @Type(() => PartDto) parts!: PartDto[];
+  @IsISO8601() occurredAt!: string;
+  @IsOptional() @IsString() providerEventId?: string;
+  @IsOptional() @IsObject() metadata?: Record<string, unknown>;
+}
@@ -0,0 +1,231 @@
+import { Inject, Injectable } from '@nestjs/common';
+import { Prisma } from '@prisma/client';
+import {
+  CloudEvent,
+  IIOS_EVENTS,
+  IiosPlatformPorts,
+  IngestInteractionRequest,
+  IngestInteractionResponse,
+} from '@insignia/iios-contracts';
+import { PrismaService } from '../prisma/prisma.service';
+import { PLATFORM_PORTS } from '../platform/platform-ports';
+import { decideOrThrow } from '../platform/fail-closed';
+
+/**
+ * Normalizes and stores one inbound interaction (P1: native portal only).
+ *
+ * Order: fail-closed OPA gate → resolve scope → resolve handle (MDM, no silent
+ * merge) → actor → channel → thread → (idempotent) write interaction + parts +
+ * outbox event in ONE transaction. Idempotency key is unique per (scope, key).
+ */
+@Injectable()
+export class IngestService {
+  constructor(
+    private readonly prisma: PrismaService,
+    @Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts,
+  ) {}
+
+  async ingest(req: IngestInteractionRequest, idempotencyKey: string): Promise<IngestInteractionResponse> {
+    // 1. Fail-closed authorization. Throws PolicyDeniedError → nothing written.
+    const decision = await decideOrThrow(this.ports, {
+      action: 'iios.interaction.ingest',
+      scope: req.scope,
+      channel: req.channel.type,
+    });
+
+    // 2. Resolve (or create) the scope snapshot.
+    const scope =
+      (await this.prisma.iiosScope.findFirst({
+        where: {
+          orgId: req.scope.orgId,
+          appId: req.scope.appId,
+          tenantId: req.scope.tenantId ?? null,
+          workspaceId: req.scope.workspaceId ?? null,
+        },
+      })) ??
+      (await this.prisma.iiosScope.create({
+        data: {
+          orgId: req.scope.orgId,
+          appId: req.scope.appId,
+          buId: req.scope.buId,
+          tenantId: req.scope.tenantId,
+          workspaceId: req.scope.workspaceId,
+          projectId: req.scope.projectId,
+          userScopeId: req.scope.userScopeId,
+        },
+      }));
+
+    // 3. Resolve the source handle via MDM. Unknown/ambiguous stays UNVERIFIED
+    //    with a null canonical id — no silent identity merge (Critics KG-01).
+    const resolution = await this.ports.mdm.resolveSourceHandle({
+      scope: req.scope,
+      handleKind: req.source.handleKind,
+      externalId: req.source.externalId,
+    });
+    const isResolved = resolution.status === 'RESOLVED' && !!resolution.canonicalEntityId;
+
+    const handle = await this.prisma.iiosSourceHandle.upsert({
+      where: {
+        scopeId_kind_externalId: {
+          scopeId: scope.id,
+          kind: req.source.handleKind,
+          externalId: req.source.externalId,
+        },
+      },
+      create: {
+        scopeId: scope.id,
+        kind: req.source.handleKind,
+        externalId: req.source.externalId,
+        displayName: req.source.displayName,
+        canonicalEntityId: isResolved ? resolution.canonicalEntityId : null,
+        confidence: resolution.confidence,
+        verificationState: isResolved ? 'VERIFIED' : 'UNVERIFIED',
+      },
+      update: {
+        lastSeenAt: new Date(),
+        displayName: req.source.displayName,
+        confidence: resolution.confidence,
+        ...(isResolved
+          ? { canonicalEntityId: resolution.canonicalEntityId, verificationState: 'VERIFIED' }
+          : {}),
+      },
+    });
+
+    // 4. Resolve (or create) the actor behind the handle.
+    const actor =
+      (await this.prisma.iiosActorRef.findFirst({ where: { sourceHandleId: handle.id } })) ??
+      (await this.prisma.iiosActorRef.create({
+        data: {
+          kind: 'HUMAN',
+          sourceHandleId: handle.id,
+          displayName: req.source.displayName,
+          canonicalEntityId: isResolved ? resolution.canonicalEntityId : null,
+        },
+      }));
+
+    // 5. Resolve (or create) the channel.
+    const channel =
+      (await this.prisma.iiosChannel.findFirst({
+        where: { scopeId: scope.id, channelType: req.channel.type, externalRef: req.channel.externalChannelId ?? null },
+      })) ??
+      (await this.prisma.iiosChannel.create({
+        data: {
+          scopeId: scope.id,
+          channelType: req.channel.type,
+          channelName: req.channel.externalChannelId ?? req.channel.type,
+          externalRef: req.channel.externalChannelId,
+          capabilityContract: (req.channel.capabilityContract ?? undefined) as Prisma.InputJsonValue | undefined,
+        },
+      }));
+
+    // 6. Resolve (or create) the thread.
+    const thread =
+      (req.thread?.externalThreadId
+        ? await this.prisma.iiosThread.findFirst({
+            where: { scopeId: scope.id, externalThreadRef: req.thread.externalThreadId },
+          })
+        : null) ??
+      (await this.prisma.iiosThread.create({
+        data: {
+          scopeId: scope.id,
+          channelId: channel.id,
+          externalThreadRef: req.thread?.externalThreadId,
+          subject: req.thread?.subject,
+          createdByActorId: actor.id,
+        },
+      }));
+
+    // 7. Idempotency: if this (scope, key) already produced an interaction, return it.
+    const existing = await this.prisma.iiosInteraction.findUnique({
+      where: { scopeId_idempotencyKey: { scopeId: scope.id, idempotencyKey } },
+    });
+    if (existing) {
+      return {
+        interactionId: existing.id,
+        sourceHandleId: handle.id,
+        threadId: thread.id,
+        state: 'DUPLICATE',
+        next: [],
+      };
+    }
+
+    // 8. Write interaction + parts + outbox event atomically.
+    try {
+      const interaction = await this.prisma.$transaction(async (tx) => {
+        const created = await tx.iiosInteraction.create({
+          data: {
+            scopeId: scope.id,
+            kind: req.kind ?? 'MESSAGE',
+            threadId: thread.id,
+            channelId: channel.id,
+            actorId: actor.id,
+            idempotencyKey,
+            externalId: req.providerEventId,
+            occurredAt: new Date(req.occurredAt),
+            status: 'NORMALIZED',
+            policyDecisionRef: decision.decisionRef,
+            metadata: (req.metadata ?? undefined) as Prisma.InputJsonValue | undefined,
+          },
+        });
+
+        await tx.iiosMessagePart.createMany({
+          data: req.parts.map((p, i) => ({
+            interactionId: created.id,
+            partIndex: i,
+            kind: p.kind,
+            bodyText: p.bodyText,
+            contentRef: p.contentRef,
+            mimeType: p.mimeType,
+          })),
+        });
+
+        const event: CloudEvent = {
+          specversion: '1.0',
+          id: `evt_${created.id}`,
+          type: IIOS_EVENTS.interactionNormalized,
+          source: `iios/ingest/${scope.appId}`,
+          subject: `interaction/${created.id}`,
+          time: new Date().toISOString(),
+          datacontenttype: 'application/json',
+          insignia: {
+            scopeSnapshotId: scope.id,
+            policyDecisionId: decision.decisionRef,
+            idempotencyKey,
+            dataClass: 'internal',
+          },
+          data: { interactionId: created.id, threadId: thread.id, kind: req.kind ?? 'MESSAGE' },
+        };
+
+        await tx.iiosOutboxEvent.create({
+          data: {
+            aggregateType: 'interaction',
+            aggregateId: created.id,
+            eventType: IIOS_EVENTS.interactionNormalized,
+            cloudEvent: event as unknown as Prisma.InputJsonValue,
+            partitionKey: `${scope.orgId}:${scope.appId}:${thread.id}`,
+          },
+        });
+
+        return created;
+      });
+
+      return {
+        interactionId: interaction.id,
+        sourceHandleId: handle.id,
+        threadId: thread.id,
+        state: 'NORMALIZED',
+        next: ['message.persisted'],
+      };
+    } catch (err) {
+      // Idempotency race: a concurrent ingest with the same key won the unique
+      // constraint. Return the winner rather than duplicating.
+      if (err instanceof Prisma.PrismaClientKnownRequestError && err.code === 'P2002') {
+        const winner = await this.prisma.iiosInteraction.findUniqueOrThrow({
+          where: { scopeId_idempotencyKey: { scopeId: scope.id, idempotencyKey } },
+        });
+        return { interactionId: winner.id, sourceHandleId: handle.id, threadId: thread.id, state: 'DUPLICATE', next: [] };
+      }
+      throw err;
+    }
+  }
+}
@@ -0,0 +1,73 @@
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { PrismaClient } from '@prisma/client';
+import { makeFakePorts, portalMessageBasic } from '@insignia/iios-testkit';
+import { PolicyDeniedError } from '@insignia/iios-contracts';
+import { IngestService } from './ingest.service';
+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 } } });
+
+async function clean(): Promise<void> {
+  await prisma.iiosOutboxEvent.deleteMany();
+  await prisma.iiosProcessedEvent.deleteMany();
+  await prisma.iiosMessagePart.deleteMany();
+  await prisma.iiosInteraction.deleteMany();
+  await prisma.iiosThreadParticipant.deleteMany();
+  await prisma.iiosThread.deleteMany();
+  await prisma.iiosActorRef.deleteMany();
+  await prisma.iiosSourceHandle.deleteMany();
+  await prisma.iiosChannel.deleteMany();
+  await prisma.iiosScope.deleteMany();
+}
+
+const svc = (ports = makeFakePorts()) => new IngestService(prisma as unknown as PrismaService, ports);
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await clean(); });
+
+describe('IngestService — kernel gates', () => {
+  it('idempotent: same idempotency key twice → exactly 1 interaction + 1 outbox event', async () => {
+    const s = svc();
+    const first = await s.ingest(portalMessageBasic, 'idem-1');
+    const second = await s.ingest(portalMessageBasic, 'idem-1');
+
+    expect(second.interactionId).toBe(first.interactionId);
+    expect(second.state).toBe('DUPLICATE');
+    expect(await prisma.iiosInteraction.count()).toBe(1);
+    expect(await prisma.iiosOutboxEvent.count()).toBe(1);
+    expect(await prisma.iiosMessagePart.count()).toBe(1);
+  });
+
+  it('fail-closed: OPA deny → PolicyDeniedError and nothing written', async () => {
+    const ports = makeFakePorts();
+    ports.opa.deny('blocked by policy');
+    const s = svc(ports);
+
+    await expect(s.ingest(portalMessageBasic, 'idem-2')).rejects.toBeInstanceOf(PolicyDeniedError);
+    expect(await prisma.iiosInteraction.count()).toBe(0);
+    expect(await prisma.iiosOutboxEvent.count()).toBe(0);
+    expect(await prisma.iiosSourceHandle.count()).toBe(0);
+  });
+
+  it('unresolved handle: MDM ambiguous → SourceHandle stays UNVERIFIED, null canonical id', async () => {
+    const ports = makeFakePorts();
+    ports.mdm.ambiguous(0.42);
+    const s = svc(ports);
+
+    await s.ingest(portalMessageBasic, 'idem-3');
+    const handle = await prisma.iiosSourceHandle.findFirstOrThrow();
+    expect(handle.canonicalEntityId).toBeNull();
+    expect(handle.verificationState).toBe('UNVERIFIED');
+    expect(handle.confidence).toBe(0.42);
+  });
+
+  it('writes the normalized CloudEvent into the outbox', async () => {
+    const s = svc();
+    await s.ingest(portalMessageBasic, 'idem-4');
+    const event = await prisma.iiosOutboxEvent.findFirstOrThrow();
+    expect(event.eventType).toBe('com.insignia.iios.interaction.normalized.v1');
+    expect(event.status).toBe('PENDING');
+  });
+});
@@ -0,0 +1,19 @@
+import { BadRequestException, Body, Controller, Headers, HttpCode, Post } from '@nestjs/common';
+import type { IngestInteractionRequest, IngestInteractionResponse } from '@insignia/iios-contracts';
+import { IngestService } from './ingest.service';
+import { IngestRequestDto } from './ingest.dto';
+
+@Controller('v1/interactions')
+export class InteractionsController {
+  constructor(private readonly ingestService: IngestService) {}
+
+  @Post('ingest')
+  @HttpCode(202)
+  async ingest(
+    @Body() body: IngestRequestDto,
+    @Headers('idempotency-key') idempotencyKey?: string,
+  ): Promise<IngestInteractionResponse> {
+    if (!idempotencyKey) throw new BadRequestException('Idempotency-Key header is required');
+    return this.ingestService.ingest(body as unknown as IngestInteractionRequest, idempotencyKey);
+  }
+}
@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { InteractionsController } from './interactions.controller';
+import { IngestService } from './ingest.service';
+
+@Module({
+  controllers: [InteractionsController],
+  providers: [IngestService],
+  exports: [IngestService],
+})
+export class InteractionsModule {}