diff --git a/packages/iios-service/src/app.module.ts b/packages/iios-service/src/app.module.ts index 1212e62..d9ff377 100644 --- a/packages/iios-service/src/app.module.ts +++ b/packages/iios-service/src/app.module.ts @@ -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 {} diff --git a/packages/iios-service/src/interactions/ingest.dto.ts b/packages/iios-service/src/interactions/ingest.dto.ts new file mode 100644 index 0000000..c092fb4 --- /dev/null +++ b/packages/iios-service/src/interactions/ingest.dto.ts @@ -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; +} + +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; +} diff --git a/packages/iios-service/src/interactions/ingest.service.ts b/packages/iios-service/src/interactions/ingest.service.ts new file mode 100644 index 0000000..0641b26 --- /dev/null +++ b/packages/iios-service/src/interactions/ingest.service.ts @@ -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 { + // 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; + } + } +} diff --git a/packages/iios-service/src/interactions/ingest.spec.ts b/packages/iios-service/src/interactions/ingest.spec.ts new file mode 100644 index 0000000..a34dd5e --- /dev/null +++ b/packages/iios-service/src/interactions/ingest.spec.ts @@ -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 { + 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'); + }); +}); diff --git a/packages/iios-service/src/interactions/interactions.controller.ts b/packages/iios-service/src/interactions/interactions.controller.ts new file mode 100644 index 0000000..155b0ad --- /dev/null +++ b/packages/iios-service/src/interactions/interactions.controller.ts @@ -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 { + if (!idempotencyKey) throw new BadRequestException('Idempotency-Key header is required'); + return this.ingestService.ingest(body as unknown as IngestInteractionRequest, idempotencyKey); + } +} diff --git a/packages/iios-service/src/interactions/interactions.module.ts b/packages/iios-service/src/interactions/interactions.module.ts new file mode 100644 index 0000000..80f1a9a --- /dev/null +++ b/packages/iios-service/src/interactions/interactions.module.ts @@ -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 {}