import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; import { PolicyDeniedError, type IiosPlatformPorts } from '@insignia/iios-contracts'; import { resetDb } from '../test-utils/reset-db'; import { makeFakePorts } from '@insignia/iios-testkit'; import { MessageService, type MessagePrincipal } from './message.service'; import { ActorResolver } from '../identity/actor.resolver'; import { DevOpaPort } from '../platform/dev-opa.port'; 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(), new ActorResolver(asService)); // A service whose OPA actually enforces the membership rules (real dev policy plane). const gov = () => new MessageService(asService, { ...makeFakePorts(), opa: new DevOpaPort() } as IiosPlatformPorts, new ActorResolver(asService)); 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 resetDb(prisma); } 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); }); }); describe('Governed membership + replies (v1.1, policy-enforced)', () => { it('a direct message is capped at two people (OPA policy)', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'dm' }); expect((await s.addParticipant(threadId, alice, 'bob')).participantCount).toBe(2); await expect(s.addParticipant(threadId, alice, 'carol')).rejects.toBeInstanceOf(PolicyDeniedError); expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(2); // unchanged }); it('group: the creator is ADMIN and can add; a plain member cannot', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); await s.addParticipant(threadId, alice, 'bob'); // admin adds a member expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(2); await expect(s.addParticipant(threadId, bob, 'carol')).rejects.toBeInstanceOf(PolicyDeniedError); // bob is MEMBER }); it('group settings: admin renames + lists members + removes; a plain member cannot rename/remove', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN', subject: 'Design' }); await s.addParticipant(threadId, alice, 'bob'); // admin renames expect((await s.renameThread(threadId, alice, 'Design Team')).subject).toBe('Design Team'); // a plain member cannot rename await expect(s.renameThread(threadId, bob, 'Hacked')).rejects.toBeInstanceOf(PolicyDeniedError); // member list carries roles const members = await s.listParticipants(threadId, alice); expect(members.map((m) => m.userId).sort()).toEqual(['alice', 'bob']); expect(members.find((m) => m.userId === 'alice')?.role).toBe('ADMIN'); // a plain member cannot remove await expect(s.removeParticipant(threadId, bob, 'alice')).rejects.toBeInstanceOf(PolicyDeniedError); // admin removes bob expect((await s.removeParticipant(threadId, alice, 'bob')).participantCount).toBe(1); expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(1); }); it('self-join is governed: a non-member cannot open a thread by id; after being added, they can', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); await expect(s.openThread(threadId, bob)).rejects.toBeInstanceOf(PolicyDeniedError); await s.addParticipant(threadId, alice, 'bob'); expect((await s.openThread(threadId, bob)).threadId).toBe(threadId); }); it('channels: a PUBLIC channel allows open self-join; a PRIVATE one does not', async () => { const s = gov(); const { threadId: pub } = await s.openThread(null, alice, { membership: 'channel', metadata: { visibility: 'public' }, subject: 'general', creatorRole: 'ADMIN' }); expect((await s.openThread(pub, bob)).threadId).toBe(pub); // bob self-joins a public channel const { threadId: priv } = await s.openThread(null, alice, { membership: 'channel', metadata: { visibility: 'private' }, subject: 'deals', creatorRole: 'ADMIN' }); await expect(s.openThread(priv, bob)).rejects.toBeInstanceOf(PolicyDeniedError); // private = invite-only }); it('discoverThreads browses same-scope channels with a joined flag; leave removes me', async () => { const s = gov(); const { threadId: gen } = await s.openThread(null, alice, { membership: 'channel', metadata: { visibility: 'public' }, subject: 'general', creatorRole: 'ADMIN' }); await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); // a group must NOT appear in a channel browse // bob (non-member, same scope) discovers the public channel as not-joined. const seen = await s.discoverThreads(bob, { metadata: { membership: 'channel', visibility: 'public' } }); expect(seen.map((t) => t.threadId)).toEqual([gen]); expect(seen[0]!.joined).toBe(false); // alice (member) sees it joined. expect((await s.discoverThreads(alice, { metadata: { membership: 'channel', visibility: 'public' } }))[0]!.joined).toBe(true); // bob joins, appears in his list, then leaves. await s.openThread(gen, bob); expect((await s.listThreads(bob)).map((t) => t.threadId)).toContain(gen); await s.leaveThread(gen, bob); expect((await s.listThreads(bob)).map((t) => t.threadId)).not.toContain(gen); }); it('listThreads returns the caller’s threads with membership, count, last message + unread', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); await s.addParticipant(threadId, alice, 'bob'); await s.send(threadId, alice, { content: 'hello team' }, 'k1'); const mine = await s.listThreads(alice); expect(mine).toHaveLength(1); expect(mine[0]).toMatchObject({ membership: 'group', participantCount: 2, lastMessage: 'hello team' }); expect((await s.listThreads(bob))[0]?.unread).toBe(1); }); it('opaque metadata: stored on create, echoed in listThreads, and a metadata filter narrows the list', async () => { const s = svc(); // Two threads with different opaque app attributes β€” the kernel never interprets these values. await s.openThread(null, alice, { metadata: { source: 'crm-support', crmCustomerId: 'cust_1' } }); await s.openThread(null, alice, { metadata: { source: 'other' } }); const all = await s.listThreads(alice); expect(all).toHaveLength(2); const support = all.find((t) => (t.metadata as { source?: string } | null)?.source === 'crm-support'); expect(support?.metadata).toMatchObject({ source: 'crm-support', crmCustomerId: 'cust_1' }); // Equality filter on the opaque bag returns only the matching thread. const filtered = await s.listThreads(alice, { metadata: { source: 'crm-support' } }); expect(filtered).toHaveLength(1); expect(filtered[0]?.metadata).toMatchObject({ crmCustomerId: 'cust_1' }); }); it('a reply stores + returns parentInteractionId; a cross-thread parent is ignored', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'dm' }); const first = await s.send(threadId, alice, { content: 'question?' }, 'k1'); const reply = await s.send(threadId, alice, { content: 'answer' }, 'k2', undefined, first.id); expect(reply.parentInteractionId).toBe(first.id); const { threadId: other } = await s.openThread(null, alice, { membership: 'dm' }); const cross = await s.send(other, alice, { content: 'x' }, 'k3', undefined, first.id); expect(cross.parentInteractionId).toBeUndefined(); // parent not in this thread }); }); describe('Interaction annotations (generic reactions primitive)', () => { it('toggleAnnotation adds then removes (toggle) and aggregates users into history', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); await s.addParticipant(threadId, alice, 'bob'); await s.openThread(threadId, bob); const msg = await s.send(threadId, alice, { content: 'ship it' }, 'k1'); const add = await s.toggleAnnotation(msg.id, bob, 'reaction', 'πŸ‘'); expect(add.op).toBe('add'); expect(add.users).toEqual(['bob']); const add2 = await s.toggleAnnotation(msg.id, alice, 'reaction', 'πŸ‘'); // alice also πŸ‘ expect(add2.op).toBe('add'); expect(add2.users).toEqual(['alice', 'bob']); // sorted, deterministic const rem = await s.toggleAnnotation(msg.id, bob, 'reaction', 'πŸ‘'); // bob toggles off expect(rem.op).toBe('remove'); expect(rem.users).toEqual(['alice']); const hist = await s.history(threadId); const m = hist.find((x) => x.id === msg.id)!; expect(m.annotations).toEqual([{ type: 'reaction', value: 'πŸ‘', users: ['alice'] }]); }); it('different values coexist for the same actor (πŸ‘ and πŸŽ‰ both stick)', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); const msg = await s.send(threadId, alice, { content: 'hi' }, 'k1'); await s.toggleAnnotation(msg.id, alice, 'reaction', 'πŸ‘'); await s.toggleAnnotation(msg.id, alice, 'reaction', 'πŸŽ‰'); const m = (await s.history(threadId)).find((x) => x.id === msg.id)!; expect(m.annotations).toEqual( expect.arrayContaining([ { type: 'reaction', value: 'πŸ‘', users: ['alice'] }, { type: 'reaction', value: 'πŸŽ‰', users: ['alice'] }, ]), ); }); it('a non-member cannot annotate (governed by policy)', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); const msg = await s.send(threadId, alice, { content: 'secret' }, 'k1'); await expect(s.toggleAnnotation(msg.id, bob, 'reaction', 'πŸ‘')).rejects.toBeInstanceOf(PolicyDeniedError); }); it('listMyAnnotated returns the caller’s saved messages with thread context; others see none', async () => { const s = gov(); const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN', subject: 'Design' }); await s.addParticipant(threadId, alice, 'bob'); const msg = await s.send(threadId, alice, { content: 'save this' }, 'k1'); await s.toggleAnnotation(msg.id, alice, 'save', ''); // save is a personal annotation const saved = await s.listMyAnnotated(alice, 'save'); expect(saved).toHaveLength(1); expect(saved[0]).toMatchObject({ threadId, threadSubject: 'Design' }); expect(saved[0]?.message.content).toBe('save this'); expect(await s.listMyAnnotated(bob, 'save')).toHaveLength(0); // bob saved nothing }); it('send carries an OPAQUE mentions[] into the message event (kernel never parses @)', async () => { const s = svc(); const { threadId } = await s.openThread(null, alice); await s.openThread(threadId, bob); await s.send(threadId, alice, { content: 'hi @bob' }, 'k1', undefined, undefined, ['bob']); const ev = await prisma.iiosOutboxEvent.findFirstOrThrow({ where: { eventType: 'com.insignia.iios.message.sent.v1' } }); const ce = ev.cloudEvent as { data: { mentions?: string[] } }; expect(ce.data.mentions).toEqual(['bob']); }); });