import { randomUUID } from 'node:crypto'; import { Inject, Injectable, NotFoundException } from '@nestjs/common'; import { Prisma } from '@prisma/client'; import { CloudEvent, IIOS_EVENTS, IiosPlatformPorts } from '@insignia/iios-contracts'; import { PrismaService } from '../prisma/prisma.service'; import { PLATFORM_PORTS } from '../platform/platform-ports'; import { decideOrThrow } from '../platform/fail-closed'; import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver'; export type { MessagePrincipal }; /** A generic annotation aggregate on a message (opaque type/value + who applied it). */ export interface AnnotationDto { type: string; value: string; users: string[]; } /** A media attachment on a message (the app renders it by `kind`). */ export interface AttachmentDto { contentRef: string; mimeType: string; sizeBytes: number; kind: 'image' | 'video' | 'audio' | 'file'; } function mediaKind(mime: string): AttachmentDto['kind'] { if (mime.startsWith('image/')) return 'image'; if (mime.startsWith('video/')) return 'video'; if (mime.startsWith('audio/')) return 'audio'; return 'file'; } export interface MessageDto { id: string; threadId: string; senderActorId: string; senderId: string; // the sender's stable externalId (email/username) — reliable "is this mine?" senderName: string; content: string; contentRef?: string; attachment?: AttachmentDto; parentInteractionId?: string; annotations: AnnotationDto[]; traceId: string; createdAt: Date; } export interface ThreadSummary { threadId: string; subject: string | null; membership?: string; participants: string[]; participantCount: number; unread: number; lastMessage?: string; lastAt?: Date; } export interface OpenThreadResult { threadId: string; status: string; history: MessageDto[]; } /** * Native direct messaging (P2) over the P1 kernel. Pure message semantics — no * tickets/agents/SLA. Send writes an interaction + parts + a `message.sent` * outbox event in one tx, bumps every other participant's unread, and stamps a * trace id (DB → event). Reuses the kernel handle/actor resolution from ingest. */ @Injectable() export class MessageService { constructor( private readonly prisma: PrismaService, @Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts, private readonly actors: ActorResolver, ) {} /** * Open an existing thread, or create one when no id is given. On create, an optional * generic `membership` attribute is stamped on the thread's metadata (the app's DM/group * hint — the kernel never branches on it; policy does), and the creator joins as ADMIN * for a group. Opening an existing thread is a GOVERNED join: only an existing member may * re-open it (policy `iios.thread.join`) — new members enter via addParticipant. */ async openThread(threadId: string | null, principal: MessagePrincipal, opts?: { membership?: string; creatorRole?: string; subject?: string }): Promise { if (!threadId) { await decideOrThrow(this.ports, { action: 'iios.thread.create', scope: principal }); const scope = await this.actors.resolveScope(principal); const actor = await this.actors.resolveActor(scope.id, principal); // `membership`/`creatorRole`/`subject` are generic, app-supplied thread attributes — the // kernel stores/echoes them but never branches on their chat meaning (that lives in policy + app). const thread = await this.prisma.iiosThread.create({ data: { scopeId: scope.id, createdByActorId: actor.id, subject: opts?.subject?.trim() || undefined, metadata: opts?.membership ? ({ membership: opts.membership } as Prisma.InputJsonValue) : undefined, }, }); await this.actors.ensureParticipant(thread.id, actor.id, opts?.creatorRole ?? 'MEMBER'); return { threadId: thread.id, status: thread.status, history: [] }; } const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } }); if (!thread) throw new NotFoundException('thread not found'); await decideOrThrow(this.ports, { action: 'iios.thread.read', threadId, scopeId: thread.scopeId }); const actor = await this.actors.resolveActor(thread.scopeId, principal); const alreadyMember = (await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: actor.id } } })) !== null; // Join is governed ONLY for threads that opted into a membership model (chat dm/group); // support/inbox/generic threads (no membership attr) keep open-join, unchanged. const membership = (thread.metadata as { membership?: string } | null)?.membership; await decideOrThrow(this.ports, { action: 'iios.thread.join', threadId, scopeId: thread.scopeId, membership, alreadyMember }); await this.actors.ensureParticipant(threadId, actor.id); return { threadId, status: thread.status, history: await this.history(threadId) }; } /** * Governed thread membership (a member adds another user by userId). Fail-closed via * policy: the DM cap / group-admin rule is enforced by OPA (dev stub now, real later); * the kernel only supplies generic context and adds the participant. The target's actor * is resolved-or-created so you can add someone who hasn't logged in yet. */ async addParticipant(threadId: string, principal: MessagePrincipal, targetUserId: string, role = 'MEMBER'): Promise<{ threadId: string; participantCount: number }> { const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } }); if (!thread) throw new NotFoundException('thread not found'); const caller = await this.actors.resolveActor(thread.scopeId, principal); const callerP = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: caller.id } } }); const participantCount = await this.prisma.iiosThreadParticipant.count({ where: { threadId } }); const membership = (thread.metadata as { membership?: string } | null)?.membership; await decideOrThrow(this.ports, { action: 'iios.thread.participant.add', threadId, scopeId: thread.scopeId, membership, participantCount, callerRole: callerP?.participantRole, targetUserId, role, }); const scope = await this.actors.resolveScope(principal); const target = await this.actors.resolveActor(scope.id, { userId: targetUserId, appId: principal.appId, orgId: principal.orgId, tenantId: principal.tenantId, displayName: targetUserId, }); await this.actors.ensureParticipant(threadId, target.id, role); return { threadId, participantCount: participantCount + 1 }; } /** Generic "my threads": every thread the caller participates in, with last message + unread. */ async listThreads(principal: MessagePrincipal): Promise { const scope = await this.actors.findScope(principal); if (!scope) return []; const actor = await this.actors.resolveActor(scope.id, principal); const memberships = await this.prisma.iiosThreadParticipant.findMany({ where: { actorId: actor.id }, select: { threadId: true } }); const threadIds = memberships.map((m) => m.threadId); if (threadIds.length === 0) return []; const [threads, unreads, allParts] = await Promise.all([ this.prisma.iiosThread.findMany({ where: { id: { in: threadIds } } }), this.prisma.iiosUnreadCounter.findMany({ where: { threadId: { in: threadIds }, actorId: actor.id } }), this.prisma.iiosThreadParticipant.findMany({ where: { threadId: { in: threadIds } }, include: { actor: { include: { sourceHandle: true } } }, }), ]); const unreadBy = new Map(unreads.map((u) => [u.threadId, u.unreadCount])); const membersBy = new Map(); for (const p of allParts) { const name = p.actor?.sourceHandle?.externalId ?? p.actor?.displayName ?? p.actorId; membersBy.set(p.threadId, [...(membersBy.get(p.threadId) ?? []), name]); } const summaries = await Promise.all( threads.map(async (t) => { const last = await this.prisma.iiosInteraction.findFirst({ where: { threadId: t.id }, orderBy: { occurredAt: 'desc' }, include: { parts: { where: { kind: 'TEXT' }, take: 1 } }, }); const members = membersBy.get(t.id) ?? []; return { threadId: t.id, subject: t.subject, membership: (t.metadata as { membership?: string } | null)?.membership, participants: members, participantCount: members.length, unread: unreadBy.get(t.id) ?? 0, lastMessage: last?.parts[0]?.bodyText ?? undefined, lastAt: last?.occurredAt, }; }), ); return summaries.sort((a, b) => (b.lastAt?.getTime() ?? 0) - (a.lastAt?.getTime() ?? 0)); } async send( threadId: string, principal: MessagePrincipal, body: { content: string; contentRef?: string; mimeType?: string; sizeBytes?: number; checksumSha256?: string }, idempotencyKey: string, traceId: string = randomUUID(), parentInteractionId?: string, mentions?: string[], ): Promise { const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } }); if (!thread) throw new NotFoundException('thread not found'); await decideOrThrow(this.ports, { action: 'iios.message.send', threadId, scopeId: thread.scopeId }); const actor = await this.actors.resolveActor(thread.scopeId, principal); await this.actors.ensureParticipant(threadId, actor.id); // A reply links to its parent — but only if the parent is in the same thread (else ignored). const parentRef = parentInteractionId ? (await this.prisma.iiosInteraction.findFirst({ where: { id: parentInteractionId, threadId }, select: { id: true } }))?.id : undefined; // Idempotency: a repeat key returns the existing message (no re-increment). const existing = await this.prisma.iiosInteraction.findUnique({ where: { scopeId_idempotencyKey: { scopeId: thread.scopeId, idempotencyKey } }, include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } }, }); if (existing) return this.toDto(existing, threadId); try { const interaction = await this.prisma.$transaction(async (tx) => { const created = await tx.iiosInteraction.create({ data: { scopeId: thread.scopeId, kind: 'MESSAGE', threadId, actorId: actor.id, idempotencyKey, status: 'NORMALIZED', traceId, parentInteractionId: parentRef, }, }); const parts: Prisma.IiosMessagePartCreateManyInput[] = [ { interactionId: created.id, partIndex: 0, kind: 'TEXT', bodyText: body.content }, ]; if (body.contentRef) { const mime = body.mimeType ?? ''; // Generic part kind from mime — the kernel stores media as opaque parts. const kind = /^(image|video)\//.test(mime) ? 'MEDIA_REF' : /^audio\//.test(mime) ? 'VOICE_REF' : 'FILE_REF'; parts.push({ interactionId: created.id, partIndex: 1, kind, contentRef: body.contentRef, mimeType: body.mimeType, sizeBytes: body.sizeBytes != null ? BigInt(body.sizeBytes) : undefined, checksumSha256: body.checksumSha256, }); } await tx.iiosMessagePart.createMany({ data: parts }); const event: CloudEvent = { specversion: '1.0', id: `evt_${created.id}`, type: IIOS_EVENTS.messageSent, source: `iios/message/${thread.scopeId}`, subject: `interaction/${created.id}`, time: new Date().toISOString(), datacontenttype: 'application/json', traceparent: `00-${traceId.replace(/-/g, '')}-0000000000000000-01`, insignia: { scopeSnapshotId: thread.scopeId, correlationId: traceId, idempotencyKey, dataClass: 'internal' }, // `mentions` is an OPAQUE app-supplied notify-list (userIds) — the kernel never parses // "@"; the inbox projector generically fans out a notification to those actors. data: { interactionId: created.id, threadId, senderActorId: actor.id, mentions: mentions ?? [] }, }; await tx.iiosOutboxEvent.create({ data: { aggregateType: 'interaction', aggregateId: created.id, eventType: IIOS_EVENTS.messageSent, cloudEvent: event as unknown as Prisma.InputJsonValue, partitionKey: `${thread.scopeId}:${threadId}`, }, }); // Bump unread for every participant except the sender. const participants = await tx.iiosThreadParticipant.findMany({ where: { threadId } }); for (const p of participants) { if (p.actorId === actor.id) continue; await tx.iiosUnreadCounter.upsert({ where: { threadId_actorId: { threadId, actorId: p.actorId } }, create: { threadId, actorId: p.actorId, unreadCount: 1 }, update: { unreadCount: { increment: 1 } }, }); } return tx.iiosInteraction.findUniqueOrThrow({ where: { id: created.id }, include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } }, }); }); return this.toDto(interaction, threadId); } catch (err) { if (err instanceof Prisma.PrismaClientKnownRequestError && err.code === 'P2002') { const winner = await this.prisma.iiosInteraction.findUniqueOrThrow({ where: { scopeId_idempotencyKey: { scopeId: thread.scopeId, idempotencyKey } }, include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } }, }); return this.toDto(winner, threadId); } throw err; } } async markRead( threadId: string, principal: MessagePrincipal, interactionId: string, ): Promise<{ interactionId: string; actorId: string }> { const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } }); if (!thread) throw new NotFoundException('thread not found'); const actor = await this.actors.resolveActor(thread.scopeId, principal); const readEvent: CloudEvent = { specversion: '1.0', id: `evt_read_${interactionId}_${actor.id}`, type: IIOS_EVENTS.messageRead, source: `iios/message/${thread.scopeId}`, subject: `interaction/${interactionId}`, time: new Date().toISOString(), datacontenttype: 'application/json', insignia: { scopeSnapshotId: thread.scopeId, idempotencyKey: `read:${interactionId}:${actor.id}`, dataClass: 'internal' }, data: { threadId, actorId: actor.id, interactionId }, }; await this.prisma.$transaction([ this.prisma.iiosMessageReceipt.upsert({ where: { interactionId_actorId_receiptKind: { interactionId, actorId: actor.id, receiptKind: 'READ' } }, create: { interactionId, actorId: actor.id, receiptKind: 'READ' }, update: { occurredAt: new Date() }, }), this.prisma.iiosUnreadCounter.upsert({ where: { threadId_actorId: { threadId, actorId: actor.id } }, create: { threadId, actorId: actor.id, unreadCount: 0, lastReadInteractionId: interactionId }, update: { unreadCount: 0, lastReadInteractionId: interactionId }, }), this.prisma.iiosOutboxEvent.create({ data: { aggregateType: 'interaction', aggregateId: interactionId, eventType: IIOS_EVENTS.messageRead, cloudEvent: readEvent as unknown as Prisma.InputJsonValue, partitionKey: `${thread.scopeId}:${threadId}`, }, }), ]); return { interactionId, actorId: actor.id }; } async markDelivered( threadId: string, principal: MessagePrincipal, interactionId: string, ): Promise<{ interactionId: string; actorId: string }> { const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } }); if (!thread) throw new NotFoundException('thread not found'); const actor = await this.actors.resolveActor(thread.scopeId, principal); await this.prisma.iiosMessageReceipt.upsert({ where: { interactionId_actorId_receiptKind: { interactionId, actorId: actor.id, receiptKind: 'DELIVERED' } }, create: { interactionId, actorId: actor.id, receiptKind: 'DELIVERED' }, update: { occurredAt: new Date() }, }); return { interactionId, actorId: actor.id }; } async getMessageById(id: string, principal?: MessagePrincipal): Promise { const i = await this.prisma.iiosInteraction.findUnique({ where: { id }, include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } }, }); if (!i || !i.threadId) return null; if (principal) await this.actors.assertOwns(principal, i.scopeId); // tenant fence (KG-02) when called on behalf of a caller return this.toDto(i, i.threadId); } async history(threadId: string): Promise { const interactions = await this.prisma.iiosInteraction.findMany({ where: { threadId }, orderBy: { occurredAt: 'asc' }, include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } }, }); const annotations = await this.annotationsByInteraction(interactions.map((i) => i.id)); return interactions.map((i) => this.toDto(i, threadId, annotations.get(i.id) ?? [])); } /** * Toggle a generic annotation (actor attaches/removes an opaque label on an interaction). * `annotationType`/`value` are app-supplied and never interpreted here (the chat app uses * type "reaction" + an emoji value). Governed: only a thread participant may annotate. * Returns the refreshed user list for that (type,value) so callers can broadcast it. */ async toggleAnnotation( interactionId: string, principal: MessagePrincipal, annotationType: string, value: string, ): Promise<{ interactionId: string; threadId: string; type: string; value: string; op: 'add' | 'remove'; users: string[] }> { const target = await this.prisma.iiosInteraction.findUnique({ where: { id: interactionId }, select: { id: true, threadId: true, scopeId: true }, }); if (!target || !target.threadId) throw new NotFoundException('message not found'); const actor = await this.actors.resolveActor(target.scopeId, principal); const isMember = (await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId: target.threadId, actorId: actor.id } } })) !== null; await decideOrThrow(this.ports, { action: 'iios.interaction.annotate', threadId: target.threadId, scopeId: target.scopeId, isMember }); const key = { scopeId_targetInteractionId_actorId_annotationType_value: { scopeId: target.scopeId, targetInteractionId: interactionId, actorId: actor.id, annotationType, value, }, }; const existing = await this.prisma.iiosInteractionAnnotation.findUnique({ where: key }); let op: 'add' | 'remove'; if (existing) { await this.prisma.iiosInteractionAnnotation.delete({ where: { id: existing.id } }); op = 'remove'; } else { await this.prisma.iiosInteractionAnnotation.create({ data: { scopeId: target.scopeId, targetInteractionId: interactionId, actorId: actor.id, annotationType, value }, }); op = 'add'; } const users = (await this.annotationsByInteraction([interactionId])).get(interactionId)?.find((a) => a.type === annotationType && a.value === value)?.users ?? []; return { interactionId, threadId: target.threadId, type: annotationType, value, op, users }; } /** * Generic "my annotated interactions" — every message the caller annotated with a given * type, newest first, with thread context. The app uses type "save" for a personal * bookmarks list (and could use "pin" for a cross-thread pinned view). Generic: the kernel * never interprets the type string. */ async listMyAnnotated( principal: MessagePrincipal, annotationType: string, ): Promise> { const scope = await this.actors.findScope(principal); if (!scope) return []; const actor = await this.actors.resolveActor(scope.id, principal); const anns = await this.prisma.iiosInteractionAnnotation.findMany({ where: { actorId: actor.id, annotationType }, orderBy: { createdAt: 'desc' }, select: { targetInteractionId: true }, }); const ids = anns.map((a) => a.targetInteractionId); if (ids.length === 0) return []; const [interactions, agg] = await Promise.all([ this.prisma.iiosInteraction.findMany({ where: { id: { in: ids } }, include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } }, thread: { select: { subject: true } }, }, }), this.annotationsByInteraction(ids), ]); const byId = new Map(interactions.map((i) => [i.id, i])); return ids .map((id) => byId.get(id)) .filter((i): i is NonNullable => Boolean(i && i.threadId)) .map((i) => ({ message: this.toDto(i, i.threadId as string, agg.get(i.id) ?? []), threadId: i.threadId as string, threadSubject: i.thread?.subject ?? null, })); } // ─── helpers ──────────────────────────────────────────────────── /** Aggregate annotations for the given interactions into { type, value, users[] } groups. */ private async annotationsByInteraction(interactionIds: string[]): Promise> { const map = new Map(); if (interactionIds.length === 0) return map; const rows = await this.prisma.iiosInteractionAnnotation.findMany({ where: { targetInteractionId: { in: interactionIds } }, include: { actor: { include: { sourceHandle: true } } }, orderBy: { createdAt: 'asc' }, }); for (const r of rows) { const user = r.actor?.sourceHandle?.externalId ?? r.actor?.displayName ?? r.actorId; const list = map.get(r.targetInteractionId) ?? []; let entry = list.find((e) => e.type === r.annotationType && e.value === r.value); if (!entry) { entry = { type: r.annotationType, value: r.value, users: [] }; list.push(entry); } entry.users.push(user); map.set(r.targetInteractionId, list); } for (const list of map.values()) for (const e of list) e.users.sort(); // deterministic order return map; } private toDto( interaction: { id: string; actorId: string | null; actor?: { displayName: string | null; sourceHandle: { externalId: string } | null } | null; parentInteractionId?: string | null; traceId: string | null; occurredAt: Date; parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null; mimeType?: string | null; sizeBytes?: bigint | null }>; }, threadId: string, annotations: AnnotationDto[] = [], ): MessageDto { const text = interaction.parts.find((p) => p.kind === 'TEXT'); const file = interaction.parts.find((p) => p.contentRef); const attachment: AttachmentDto | undefined = file?.contentRef ? { contentRef: file.contentRef, mimeType: file.mimeType ?? 'application/octet-stream', sizeBytes: file.sizeBytes != null ? Number(file.sizeBytes) : 0, kind: mediaKind(file.mimeType ?? ''), } : undefined; return { id: interaction.id, threadId, senderActorId: interaction.actorId ?? '', senderId: interaction.actor?.sourceHandle?.externalId ?? '', senderName: interaction.actor?.displayName ?? interaction.actor?.sourceHandle?.externalId ?? 'unknown', content: text?.bodyText ?? '', contentRef: file?.contentRef ?? undefined, attachment, parentInteractionId: interaction.parentInteractionId ?? undefined, annotations, traceId: interaction.traceId ?? '', createdAt: interaction.occurredAt, }; } }