import { Inject, Injectable } from '@nestjs/common'; import { Prisma } from '@prisma/client'; import { CloudEvent, IIOS_EVENTS, IiosPlatformPorts, IngestInteractionRequest, IngestInteractionResponse, } from '@insignia/iios-contracts'; import { currentTraceId, newTraceId, toTraceparent } from '../observability/trace-context'; 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', traceId: req.traceId, 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, sizeBytes: p.sizeBytes != null ? BigInt(p.sizeBytes) : undefined, })), }); 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', traceparent: toTraceparent(created.traceId ?? currentTraceId() ?? newTraceId()), insignia: { scopeSnapshotId: scope.id, correlationId: created.traceId ?? currentTraceId() ?? newTraceId(), 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; } } }