diff --git a/packages/iios-contracts/src/events.ts b/packages/iios-contracts/src/events.ts index 280a6fa..f9e9fd0 100644 --- a/packages/iios-contracts/src/events.ts +++ b/packages/iios-contracts/src/events.ts @@ -34,6 +34,7 @@ export interface CloudEvent { export const IIOS_EVENTS = { rawReceived: 'com.insignia.iios.raw.received.v1', interactionNormalized: 'com.insignia.iios.interaction.normalized.v1', + messageSent: 'com.insignia.iios.message.sent.v1', messagePersisted: 'com.insignia.iios.message.persisted.v1', inboxItemCreated: 'com.insignia.iios.inbox.item.created.v1', ticketCreated: 'com.insignia.iios.support.ticket.created.v1', diff --git a/packages/iios-service/src/app.module.ts b/packages/iios-service/src/app.module.ts index 5e4af9b..fdaad9f 100644 --- a/packages/iios-service/src/app.module.ts +++ b/packages/iios-service/src/app.module.ts @@ -4,10 +4,11 @@ import { PlatformModule } from './platform/platform.module'; import { InteractionsModule } from './interactions/interactions.module'; import { OutboxModule } from './outbox/outbox.module'; import { ThreadsModule } from './threads/threads.module'; +import { MessageModule } from './messaging/message.module'; import { HealthController } from './health.controller'; @Module({ - imports: [PrismaModule, PlatformModule, InteractionsModule, OutboxModule, ThreadsModule], + imports: [PrismaModule, PlatformModule, InteractionsModule, OutboxModule, ThreadsModule, MessageModule], controllers: [HealthController], }) export class AppModule {} diff --git a/packages/iios-service/src/messaging/message.module.ts b/packages/iios-service/src/messaging/message.module.ts new file mode 100644 index 0000000..cdf9467 --- /dev/null +++ b/packages/iios-service/src/messaging/message.module.ts @@ -0,0 +1,8 @@ +import { Module } from '@nestjs/common'; +import { MessageService } from './message.service'; + +@Module({ + providers: [MessageService], + exports: [MessageService], +}) +export class MessageModule {} diff --git a/packages/iios-service/src/messaging/message.service.ts b/packages/iios-service/src/messaging/message.service.ts new file mode 100644 index 0000000..c51b124 --- /dev/null +++ b/packages/iios-service/src/messaging/message.service.ts @@ -0,0 +1,246 @@ +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'; + +/** The authenticated sender, derived from the session port (token). */ +export interface MessagePrincipal { + userId: string; + orgId: string; + appId: string; + tenantId?: string; + displayName?: string; +} + +export interface MessageDto { + id: string; + threadId: string; + senderActorId: string; + content: string; + contentRef?: string; + traceId: string; + createdAt: 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, + ) {} + + /** Open an existing thread, or create one when no id is given. Joins as participant. */ + async openThread(threadId: string | null, principal: MessagePrincipal): Promise { + if (!threadId) { + await decideOrThrow(this.ports, { action: 'iios.thread.create', scope: principal }); + const scope = await this.resolveScope(principal); + const actor = await this.resolveActor(scope.id, principal); + const thread = await this.prisma.iiosThread.create({ + data: { scopeId: scope.id, createdByActorId: actor.id }, + }); + await this.ensureParticipant(thread.id, actor.id); + 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.resolveActor(thread.scopeId, principal); + await this.ensureParticipant(threadId, actor.id); + return { threadId, status: thread.status, history: await this.history(threadId) }; + } + + async send( + threadId: string, + principal: MessagePrincipal, + body: { content: string; contentRef?: string }, + idempotencyKey: string, + traceId: string = randomUUID(), + ): 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.resolveActor(thread.scopeId, principal); + await this.ensureParticipant(threadId, actor.id); + + // 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' } } }, + }); + 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, + }, + }); + + const parts: Prisma.IiosMessagePartCreateManyInput[] = [ + { interactionId: created.id, partIndex: 0, kind: 'TEXT', bodyText: body.content }, + ]; + if (body.contentRef) { + parts.push({ interactionId: created.id, partIndex: 1, kind: 'FILE_REF', contentRef: body.contentRef }); + } + 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' }, + data: { interactionId: created.id, threadId, senderActorId: actor.id }, + }; + 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' } } }, + }); + }); + + 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' } } }, + }); + 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.resolveActor(thread.scopeId, principal); + + await 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() }, + }); + await 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 }, + }); + return { interactionId, actorId: actor.id }; + } + + async history(threadId: string): Promise { + const interactions = await this.prisma.iiosInteraction.findMany({ + where: { threadId }, + orderBy: { occurredAt: 'asc' }, + include: { parts: { orderBy: { partIndex: 'asc' } } }, + }); + return interactions.map((i) => this.toDto(i, threadId)); + } + + // ─── helpers ──────────────────────────────────────────────────── + + private async resolveScope(principal: MessagePrincipal) { + return ( + (await this.prisma.iiosScope.findFirst({ + where: { orgId: principal.orgId, appId: principal.appId, tenantId: principal.tenantId ?? null }, + })) ?? + (await this.prisma.iiosScope.create({ + data: { orgId: principal.orgId, appId: principal.appId, tenantId: principal.tenantId }, + })) + ); + } + + private async resolveActor(scopeId: string, principal: MessagePrincipal) { + const handle = await this.prisma.iiosSourceHandle.upsert({ + where: { scopeId_kind_externalId: { scopeId, kind: 'PORTAL_USER', externalId: principal.userId } }, + create: { scopeId, kind: 'PORTAL_USER', externalId: principal.userId, displayName: principal.displayName }, + update: { lastSeenAt: new Date(), displayName: principal.displayName }, + }); + return ( + (await this.prisma.iiosActorRef.findFirst({ where: { sourceHandleId: handle.id } })) ?? + (await this.prisma.iiosActorRef.create({ + data: { kind: 'HUMAN', sourceHandleId: handle.id, displayName: principal.displayName }, + })) + ); + } + + private async ensureParticipant(threadId: string, actorId: string): Promise { + await this.prisma.iiosThreadParticipant.upsert({ + where: { threadId_actorId: { threadId, actorId } }, + create: { threadId, actorId }, + update: {}, + }); + } + + private toDto( + interaction: { id: string; actorId: string | null; traceId: string | null; occurredAt: Date; parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null }> }, + threadId: string, + ): MessageDto { + const text = interaction.parts.find((p) => p.kind === 'TEXT'); + const file = interaction.parts.find((p) => p.contentRef); + return { + id: interaction.id, + threadId, + senderActorId: interaction.actorId ?? '', + content: text?.bodyText ?? '', + contentRef: file?.contentRef ?? undefined, + traceId: interaction.traceId ?? '', + createdAt: interaction.occurredAt, + }; + } +} diff --git a/packages/iios-service/src/messaging/message.spec.ts b/packages/iios-service/src/messaging/message.spec.ts new file mode 100644 index 0000000..7a40704 --- /dev/null +++ b/packages/iios-service/src/messaging/message.spec.ts @@ -0,0 +1,112 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { PrismaClient } from '@prisma/client'; +import { makeFakePorts } from '@insignia/iios-testkit'; +import { MessageService, type MessagePrincipal } from './message.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 } } }); +const asService = prisma as unknown as PrismaService; +const svc = () => new MessageService(asService, makeFakePorts()); + +const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' }; +const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' }; + +async function clean(): Promise { + await prisma.iiosUnreadCounter.deleteMany(); + await prisma.iiosMessageReceipt.deleteMany(); + 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(); +} + +async function actorIdFor(userId: string): Promise { + const handle = await prisma.iiosSourceHandle.findFirstOrThrow({ where: { externalId: userId } }); + const actor = await prisma.iiosActorRef.findFirstOrThrow({ where: { sourceHandleId: handle.id } }); + return actor.id; +} + +async function unread(threadId: string, userId: string): Promise { + const c = await prisma.iiosUnreadCounter.findUnique({ + where: { threadId_actorId: { threadId, actorId: await actorIdFor(userId) } }, + }); + return c?.unreadCount ?? 0; +} + +beforeAll(async () => { await prisma.$connect(); }); +afterAll(async () => { await prisma.$disconnect(); }); +beforeEach(async () => { await clean(); }); + +describe('MessageService (P2 native messaging)', () => { + it('open_thread with no id creates a thread; send bumps the other participant unread +1', async () => { + const s = svc(); + const { threadId } = await s.openThread(null, alice); // Alice creates + await s.openThread(threadId, bob); // Bob joins + + const msg = await s.send(threadId, alice, { content: 'hi bob' }, 'k1'); + expect(msg.content).toBe('hi bob'); + expect(msg.traceId).not.toBe(''); + expect(await unread(threadId, 'bob')).toBe(1); + expect(await unread(threadId, 'alice')).toBe(0); + expect(await prisma.iiosOutboxEvent.count()).toBe(1); + }); + + it('is idempotent: same key twice → 1 interaction, recipient unread still 1', async () => { + const s = svc(); + const { threadId } = await s.openThread(null, alice); + await s.openThread(threadId, bob); + + const a = await s.send(threadId, alice, { content: 'hi' }, 'k1'); + const b = await s.send(threadId, alice, { content: 'hi' }, 'k1'); + expect(b.id).toBe(a.id); + expect(await prisma.iiosInteraction.count()).toBe(1); + expect(await unread(threadId, 'bob')).toBe(1); + }); + + it('markRead writes a READ receipt, sets lastRead, and resets unread to 0', async () => { + const s = svc(); + const { threadId } = await s.openThread(null, alice); + await s.openThread(threadId, bob); + const msg = await s.send(threadId, alice, { content: 'hi' }, 'k1'); + expect(await unread(threadId, 'bob')).toBe(1); + + await s.markRead(threadId, bob, msg.id); + expect(await unread(threadId, 'bob')).toBe(0); + const receipt = await prisma.iiosMessageReceipt.findFirstOrThrow(); + expect(receipt.receiptKind).toBe('READ'); + const counter = await prisma.iiosUnreadCounter.findUnique({ + where: { threadId_actorId: { threadId, actorId: await actorIdFor('bob') } }, + }); + expect(counter?.lastReadInteractionId).toBe(msg.id); + }); + + it('attachment placeholder: a contentRef part round-trips (no upload)', async () => { + const s = svc(); + const { threadId } = await s.openThread(null, alice); + const msg = await s.send(threadId, alice, { content: 'see file', contentRef: 'sas://obj/123' }, 'k1'); + expect(msg.contentRef).toBe('sas://obj/123'); + const fileParts = await prisma.iiosMessagePart.count({ where: { kind: 'FILE_REF' } }); + expect(fileParts).toBe(1); + }); + + it('traceId flows from interaction into the outbox CloudEvent', async () => { + const s = svc(); + const { threadId } = await s.openThread(null, alice); + const msg = await s.send(threadId, alice, { content: 'hi' }, 'k1'); + + const interaction = await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: msg.id } }); + expect(interaction.traceId).toBe(msg.traceId); + + const event = await prisma.iiosOutboxEvent.findFirstOrThrow(); + const ce = event.cloudEvent as { insignia: { correlationId: string }; type: string }; + expect(ce.type).toBe('com.insignia.iios.message.sent.v1'); + expect(ce.insignia.correlationId).toBe(msg.traceId); + }); +});