iios/packages/iios-service/src/messaging/message.spec.ts

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<void> {
  await resetDb(prisma);
}

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);
  });
});

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('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('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']);
  });
});