feat(service): P2.2 MessageService — native send + receipts/unread + traceId

openThread (create-or-join), send (idempotent, bumps other participants' unread, message.sent outbox event, traceId DB->event), markRead (READ receipt + unread reset), contentRef attachment placeholder. Adds messageSent to contracts events. 5 DB-backed tests green; 24 total. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 01:09:19 +05:30
parent 7197157058
commit 827fb52f5f
5 changed files with 369 additions and 1 deletions
@@ -0,0 +1,8 @@
+import { Module } from '@nestjs/common';
+import { MessageService } from './message.service';
+
+@Module({
+  providers: [MessageService],
+  exports: [MessageService],
+})
+export class MessageModule {}
@@ -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<OpenThreadResult> {
+    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<MessageDto> {
+    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<MessageDto[]> {
+    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<void> {
+    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,
+    };
+  }
+}
@@ -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<void> {
+  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<string> {
+  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<number> {
+  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);
+  });
+});