feat(threads): govern roster reads + add a bulk participant primitive

Phase A of channel-roster import (add everyone from another channel).

SECURITY: listParticipants was gated by iios.thread.read, which has no case in
DevOpaPort and so fell through to default-allow — any caller in a scope could
enumerate ANY thread's members, including private channels they aren't in. Since
PlatformModule binds LocalDevPorts unconditionally (no real OPA adapter exists),
that was live. Adds iios.thread.participant.list: members only, public channels
exempt, ungoverned threads unchanged. This is also Zoom's rule for this feature
('you must be a member of the channel to invite all of its members').

Adds MessageService.addParticipants(): many users in ONE governed call, capped at
MAX_BULK_PARTICIPANTS (200), idempotent (already-members are 'skipped'), and
deliberately non-atomic so one unresolvable user can't sink an import — outcome
is reported per user as {added, skipped, failed}. The dm two-person cap now reads
targetCount so it holds for a batch, not just one add at a time. New REST route
POST /v1/threads/:id/participants/bulk.

The kernel stays generic: it takes an explicit userId list and never learns where
that list came from — chat meaning lives only in the policy plane.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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

@@ -1,5 +1,5 @@
 import { randomUUID } from 'node:crypto';
-import { Inject, Injectable, NotFoundException } from '@nestjs/common';
+import { BadRequestException, 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';
@@ -74,6 +74,10 @@ export interface OpenThreadResult {
  */
 @Injectable()
 export class MessageService {
+  /** Ceiling on one bulk participant import — bounds the work per request and the blast radius of a
+   *  mistaken import. Raise here if a larger roster copy is ever needed. */
+  static readonly MAX_BULK_PARTICIPANTS = 200;
+
   constructor(
     private readonly prisma: PrismaService,
     @Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts,
@@ -167,6 +171,76 @@ export class MessageService {
     return { threadId, participantCount: participantCount + 1 };
   }
 
+  /**
+   * Bulk sibling of {@link addParticipant}: add many users in ONE governed operation. The app uses
+   * this to import a roster (e.g. "add everyone from that channel"); the kernel stays generic — it
+   * receives an explicit list of userIds and never learns where the list came from.
+   *
+   * Deliberately NOT atomic: a single unresolvable user must not sink the whole import, so each is
+   * attempted independently and the outcome is reported per user. Already-members are `skipped`,
+   * which makes a re-run a no-op.
+   */
+  async addParticipants(
+    threadId: string,
+    principal: MessagePrincipal,
+    targetUserIds: string[],
+    role = 'MEMBER',
+  ): Promise<{ threadId: string; added: string[]; skipped: string[]; failed: string[]; participantCount: number }> {
+    const unique = [...new Set(targetUserIds.map((u) => u.trim()).filter(Boolean))];
+    if (unique.length === 0) throw new BadRequestException('at least one userId is required');
+    if (unique.length > MessageService.MAX_BULK_PARTICIPANTS) {
+      throw new BadRequestException(`at most ${MessageService.MAX_BULK_PARTICIPANTS} participants can be added at once`);
+    }
+
+    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;
+
+    // ONE decision for the whole batch — `targetCount` lets policy reason about the resulting size
+    // (e.g. the dm two-person cap) instead of being asked the same question N times.
+    await decideOrThrow(this.ports, {
+      action: 'iios.thread.participant.add',
+      threadId,
+      scopeId: thread.scopeId,
+      membership,
+      participantCount,
+      callerRole: callerP?.participantRole,
+      targetCount: unique.length,
+      role,
+    });
+
+    const scope = await this.actors.resolveScope(principal);
+    const added: string[] = [];
+    const skipped: string[] = [];
+    const failed: string[] = [];
+    for (const targetUserId of unique) {
+      try {
+        const target = await this.actors.resolveActor(scope.id, {
+          userId: targetUserId,
+          appId: principal.appId,
+          orgId: principal.orgId,
+          tenantId: principal.tenantId,
+          displayName: targetUserId,
+        });
+        const existing = await this.prisma.iiosThreadParticipant.findUnique({
+          where: { threadId_actorId: { threadId, actorId: target.id } },
+        });
+        if (existing) {
+          skipped.push(targetUserId);
+          continue;
+        }
+        await this.actors.ensureParticipant(threadId, target.id, role);
+        added.push(targetUserId);
+      } catch {
+        failed.push(targetUserId);
+      }
+    }
+    return { threadId, added, skipped, failed, participantCount: participantCount + added.length };
+  }
+
   /**
    * Governed thread rename (a generic subject update). Policy decides who may rename — for a
    * membership thread the dev OPA requires the caller be a group ADMIN. The kernel only writes
@@ -224,7 +298,19 @@ export class MessageService {
   async listParticipants(threadId: string, principal: MessagePrincipal): Promise<Array<{ userId: string; displayName: string; role: string }>> {
     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 });
+    // Roster reads are membership-governed (a private channel's members must not be enumerable by
+    // a non-member), so the decision carries the caller's role + the thread's opaque attributes.
+    const caller = await this.actors.resolveActor(thread.scopeId, principal);
+    const callerP = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: caller.id } } });
+    const meta = thread.metadata as { membership?: string; visibility?: string } | null;
+    await decideOrThrow(this.ports, {
+      action: 'iios.thread.participant.list',
+      threadId,
+      scopeId: thread.scopeId,
+      membership: meta?.membership,
+      visibility: meta?.visibility,
+      callerRole: callerP?.participantRole,
+    });
     const parts = await this.prisma.iiosThreadParticipant.findMany({
       where: { threadId },
       include: { actor: { include: { sourceHandle: true } } },

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

@@ -146,6 +146,61 @@ describe('Governed membership + replies (v1.1, policy-enforced)', () => {
     expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(1);
   });
 
+  it('roster reads are membership-governed: a non-member cannot list a private channel’s members', async () => {
+    const s = gov();
+    const { threadId: priv } = await s.openThread(null, alice, { membership: 'channel', metadata: { visibility: 'private' }, subject: 'deals', creatorRole: 'ADMIN' });
+    // bob is not a member — he must not be able to enumerate who is.
+    await expect(s.listParticipants(priv, bob)).rejects.toBeInstanceOf(PolicyDeniedError);
+    await s.addParticipant(priv, alice, 'bob');
+    expect((await s.listParticipants(priv, bob)).map((m) => m.userId).sort()).toEqual(['alice', 'bob']);
+
+    // a PUBLIC channel's roster stays open (it is discoverable anyway)
+    const { threadId: pub } = await s.openThread(null, alice, { membership: 'channel', metadata: { visibility: 'public' }, subject: 'general', creatorRole: 'ADMIN' });
+    expect((await s.listParticipants(pub, bob)).map((m) => m.userId)).toEqual(['alice']);
+  });
+
+  it('addParticipants: bulk-adds in one governed call, skipping existing members (re-run is a no-op)', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'channel', metadata: { visibility: 'private' }, subject: 'ops', creatorRole: 'ADMIN' });
+    await s.addParticipant(threadId, alice, 'bob'); // bob is already in
+
+    const res = await s.addParticipants(threadId, alice, ['bob', 'carol', 'dave']);
+    expect(res.added.sort()).toEqual(['carol', 'dave']);
+    expect(res.skipped).toEqual(['bob']);
+    expect(res.failed).toEqual([]);
+    expect(res.participantCount).toBe(4); // alice, bob, carol, dave
+    expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(4);
+
+    // idempotent: a second identical import adds nobody
+    const again = await s.addParticipants(threadId, alice, ['bob', 'carol', 'dave']);
+    expect(again.added).toEqual([]);
+    expect(again.skipped.sort()).toEqual(['bob', 'carol', 'dave']);
+    expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(4);
+  });
+
+  it('addParticipants is governed by the same policy as a single add (a plain member is denied)', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await s.addParticipant(threadId, alice, 'bob'); // bob joins as a plain MEMBER
+    await expect(s.addParticipants(threadId, bob, ['carol', 'dave'])).rejects.toBeInstanceOf(PolicyDeniedError);
+  });
+
+  it('addParticipants rejects an empty list and anything over the batch cap', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'channel', metadata: { visibility: 'private' }, creatorRole: 'ADMIN' });
+    await expect(s.addParticipants(threadId, alice, [])).rejects.toThrow(/at least one/i);
+    const tooMany = Array.from({ length: MessageService.MAX_BULK_PARTICIPANTS + 1 }, (_, n) => `user_${n}`);
+    await expect(s.addParticipants(threadId, alice, tooMany)).rejects.toThrow(/at most/i);
+  });
+
+  it('addParticipants respects the dm two-person cap for the whole batch', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'dm' });
+    // alice is alone; adding two at once would make three — policy must deny the batch.
+    await expect(s.addParticipants(threadId, alice, ['bob', 'carol'])).rejects.toBeInstanceOf(PolicyDeniedError);
+    expect((await s.addParticipants(threadId, alice, ['bob'])).added).toEqual(['bob']);
+  });
+
   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' });

packages/iios-service/src/platform/dev-opa.port.spec.ts

@@ -49,6 +49,29 @@ describe('DevOpaPort (dev policy plane — membership rules)', () => {
     expect((await opa.decide({ action: 'iios.thread.participant.remove', callerRole: 'MEMBER' })).allow).toBe(true);
   });
 
+  it('the dm cap counts the whole batch, not one add at a time', async () => {
+    const add = (participantCount: number, targetCount: number) =>
+      opa.decide({ action: 'iios.thread.participant.add', membership: 'dm', callerRole: 'MEMBER', participantCount, targetCount });
+    expect((await add(1, 1)).allow).toBe(true); // 1 + 1 = 2, fine
+    const batch = await add(1, 2); // 1 + 2 = 3 — must be denied even though count is only 1
+    expect(batch.allow).toBe(false);
+    expect(batch.obligations[0]?.reason).toMatch(/two people/);
+  });
+
+  it('listing a roster requires membership, except on a public channel', async () => {
+    const list = (extra: Record<string, unknown>) => opa.decide({ action: 'iios.thread.participant.list', ...extra });
+    // a private channel / group / dm: members only
+    const stranger = await list({ membership: 'channel', visibility: 'private', callerRole: undefined });
+    expect(stranger.allow).toBe(false);
+    expect(stranger.obligations[0]?.reason).toMatch(/not a member/);
+    expect((await list({ membership: 'channel', visibility: 'private', callerRole: 'MEMBER' })).allow).toBe(true);
+    expect((await list({ membership: 'group', callerRole: 'ADMIN' })).allow).toBe(true);
+    expect((await list({ membership: 'group', callerRole: undefined })).allow).toBe(false);
+    // a public channel's roster is open, and ungoverned threads are unchanged
+    expect((await list({ membership: 'channel', visibility: 'public', callerRole: undefined })).allow).toBe(true);
+    expect((await list({})).allow).toBe(true);
+  });
+
   it('media upload allows images + docs (incl. html/markdown/csv), denies unknown types and oversize', async () => {
     const up = (mime: string, sizeBytes = 1024) => opa.decide({ action: 'iios.media.upload', mime, sizeBytes });
     for (const mime of ['image/png', 'video/mp4', 'audio/mpeg', 'application/pdf', 'text/plain', 'text/markdown', 'text/html', 'text/csv']) {

packages/iios-service/src/platform/dev-opa.port.ts

@@ -12,6 +12,7 @@ export interface OpaInput {
   membership?: string; // app-set generic thread attribute: 'dm' | 'group' | 'channel'
   visibility?: string; // 'public' | 'private' (channels)
   participantCount?: number;
+  targetCount?: number; // how many participants a single add call is inviting (1 for a single add)
   callerRole?: string; // 'MEMBER' | 'ADMIN'
   alreadyMember?: boolean;
   isMember?: boolean;
@@ -38,11 +39,23 @@ export class DevOpaPort {
     switch (i.action) {
       case 'iios.thread.participant.add': {
         const count = i.participantCount ?? 0;
-        if (i.membership === 'dm' && count >= 2) return deny('a direct message is limited to two people');
+        // `targetCount` is how many are being added in this call (1 for a single add), so the dm cap
+        // holds for a bulk import too instead of being evaluated one-at-a-time.
+        if (i.membership === 'dm' && count + (i.targetCount ?? 1) > 2) return deny('a direct message is limited to two people');
         if (i.callerRole !== 'MEMBER' && i.callerRole !== 'ADMIN') return deny('only a member can add participants');
         if (i.membership === 'group' && i.callerRole !== 'ADMIN') return deny('only a group admin can add or remove members');
         return allow();
       }
+      case 'iios.thread.participant.list': {
+        // Reading a thread's ROSTER. A public channel is open (it is discoverable anyway), but every
+        // other membership thread — dm, group, private channel — requires you to be a member, or any
+        // caller in the scope could enumerate a private channel's members by id. Ungoverned threads
+        // (no membership attribute, e.g. support) keep the previous open behaviour.
+        if (!i.membership) return allow();
+        if (i.membership === 'channel' && i.visibility === 'public') return allow();
+        if (i.callerRole === 'MEMBER' || i.callerRole === 'ADMIN') return allow();
+        return deny('you are not a member of this thread');
+      }
       case 'iios.thread.participant.remove': {
         // Same governance as add: for a group, only an ADMIN removes members. Ungoverned
         // (no membership attr) threads allow removal by any member.

packages/iios-service/src/threads/threads.controller.ts

@@ -70,6 +70,17 @@ export class ThreadsController {
     return this.messages.addParticipant(id, this.principal(auth), body.userId, body.role);
   }
 
+  /**
+   * Governed bulk membership: add many users in one call (e.g. importing another channel's roster).
+   * Reports per-user outcome — `skipped` are already members — so a partial result is never silent.
+   */
+  @Post(':id/participants/bulk')
+  @HttpCode(200)
+  async addParticipants(@Param('id') id: string, @Body() body: { userIds?: string[]; role?: string }, @Headers('authorization') auth?: string) {
+    if (!Array.isArray(body?.userIds)) throw new BadRequestException('userIds must be an array');
+    return this.messages.addParticipants(id, this.principal(auth), body.userIds, body.role);
+  }
+
   /** Members of a thread with their role — drives the group settings member list. */
   @Get(':id/participants')
   async listParticipants(@Param('id') id: string, @Headers('authorization') auth?: string) {