feat(service): P4.3 AssignmentService — event-driven least-loaded assignment

Consumes support.ticket.created -> picks least-loaded available agent within
capacity -> ticket OPEN, activeCount++, agent joins linked thread(s), base
thread untouched; idempotent (processed-event + assignedActorId guard). Queue/
team admin (createQueue/addMember/setAvailability, re-scan on available). 45 tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/iios-service/src/app.module.ts

@@ -7,6 +7,7 @@ import { OutboxModule } from './outbox/outbox.module';
 import { ThreadsModule } from './threads/threads.module';
 import { MessageModule } from './messaging/message.module';
 import { InboxModule } from './inbox/inbox.module';
+import { SupportModule } from './support/support.module';
 import { HealthController } from './health.controller';
 import { DevController } from './dev/dev.controller';
 
@@ -20,6 +21,7 @@ import { DevController } from './dev/dev.controller';
     ThreadsModule,
     MessageModule,
     InboxModule,
+    SupportModule,
   ],
   controllers: [HealthController, DevController],
 })

packages/iios-service/src/support/assignment.service.ts

@@ -0,0 +1,135 @@
+import { Injectable, OnModuleInit } from '@nestjs/common';
+import { Prisma } from '@prisma/client';
+import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
+import { PrismaService } from '../prisma/prisma.service';
+import { OutboxBus } from '../outbox/outbox.bus';
+import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
+
+/**
+ * Event-driven ticket assignment (P4). Consumes support.ticket.created and picks
+ * the least-loaded available agent within capacity, sets the ticket OPEN, and
+ * adds the agent as a participant on the linked thread(s) so they reply over the
+ * P2 socket. The base thread is never mutated. Idempotent via the processed-event
+ * ledger; the assignedActorId guard also prevents double-assignment.
+ */
+@Injectable()
+export class AssignmentService implements OnModuleInit {
+  private readonly consumer = 'assignment';
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly bus: OutboxBus,
+    private readonly actors: ActorResolver,
+  ) {}
+
+  onModuleInit(): void {
+    this.bus.on(IIOS_EVENTS.ticketCreated, (p) => void this.onTicketCreated(p as CloudEvent).catch(() => {}));
+  }
+
+  async onTicketCreated(event: CloudEvent): Promise<void> {
+    if (!(await this.claim(event.id))) return;
+    const data = event.data as { ticketId: string };
+    await this.assignTicket(data.ticketId);
+  }
+
+  /** Assign one ticket to the least-loaded available agent. Returns true if assigned. */
+  async assignTicket(ticketId: string): Promise<boolean> {
+    const ticket = await this.prisma.iiosTicket.findUnique({
+      where: { id: ticketId },
+      include: { threadLinks: true },
+    });
+    if (!ticket || ticket.assignedActorId || !ticket.queueId) return false;
+
+    const candidates = await this.prisma.iiosSupportTeamMember.findMany({
+      where: { queueId: ticket.queueId, availabilityState: 'AVAILABLE' },
+      orderBy: { activeCount: 'asc' },
+    });
+    const member = candidates.find((m) => m.activeCount < m.maxActive);
+    if (!member) return false;
+
+    const event: CloudEvent = {
+      specversion: '1.0',
+      id: `evt_ticketstate_${ticketId}_OPEN`,
+      type: IIOS_EVENTS.ticketStateChanged,
+      source: `iios/support/${ticket.scopeId}`,
+      subject: `ticket/${ticketId}`,
+      time: new Date().toISOString(),
+      datacontenttype: 'application/json',
+      insignia: { scopeSnapshotId: ticket.scopeId, idempotencyKey: `assign:${ticketId}`, dataClass: 'internal' },
+      data: { ticketId, fromState: ticket.state, toState: 'OPEN', assignedActorId: member.actorId },
+    };
+
+    await this.prisma.$transaction([
+      this.prisma.iiosTicket.update({
+        where: { id: ticketId },
+        data: { assignedActorId: member.actorId, state: 'OPEN' },
+      }),
+      this.prisma.iiosSupportTeamMember.update({
+        where: { queueId_actorId: { queueId: ticket.queueId, actorId: member.actorId } },
+        data: { activeCount: { increment: 1 } },
+      }),
+      this.prisma.iiosTicketStateHistory.create({
+        data: { ticketId, fromState: ticket.state, toState: 'OPEN', actorId: member.actorId, reasonCode: 'assigned' },
+      }),
+      this.prisma.iiosOutboxEvent.create({
+        data: {
+          aggregateType: 'ticket',
+          aggregateId: ticketId,
+          eventType: IIOS_EVENTS.ticketStateChanged,
+          cloudEvent: event as unknown as Prisma.InputJsonValue,
+          partitionKey: `${ticket.scopeId}:${ticketId}`,
+        },
+      }),
+    ]);
+
+    // Agent joins the linked thread(s) so they can reply over the message socket.
+    for (const link of ticket.threadLinks) {
+      await this.actors.ensureParticipant(link.threadId, member.actorId);
+    }
+    return true;
+  }
+
+  /** Re-scan: assign all unassigned NEW tickets (optionally in one queue). */
+  async assignPending(queueId?: string): Promise<number> {
+    const tickets = await this.prisma.iiosTicket.findMany({
+      where: { assignedActorId: null, state: 'NEW', ...(queueId ? { queueId } : {}) },
+    });
+    let assigned = 0;
+    for (const t of tickets) if (await this.assignTicket(t.id)) assigned++;
+    return assigned;
+  }
+
+  // ─── queue / team admin ───────────────────────────────────────────
+
+  async createQueue(principal: MessagePrincipal, name: string) {
+    const scope = await this.actors.resolveScope(principal);
+    return this.prisma.iiosSupportQueue.create({ data: { scopeId: scope.id, name } });
+  }
+
+  async addMember(queueId: string, agent: MessagePrincipal, opts?: { maxActive?: number }) {
+    const queue = await this.prisma.iiosSupportQueue.findUniqueOrThrow({ where: { id: queueId } });
+    const actor = await this.actors.resolveActor(queue.scopeId, agent);
+    return this.prisma.iiosSupportTeamMember.upsert({
+      where: { queueId_actorId: { queueId, actorId: actor.id } },
+      create: { queueId, actorId: actor.id, maxActive: opts?.maxActive ?? 3, availabilityState: 'OFFLINE' },
+      update: {},
+    });
+  }
+
+  async setAvailability(agent: MessagePrincipal, state: string): Promise<{ actorId: string; availabilityState: string }> {
+    const scope = await this.actors.resolveScope(agent);
+    const actor = await this.actors.resolveActor(scope.id, agent);
+    await this.prisma.iiosSupportTeamMember.updateMany({ where: { actorId: actor.id }, data: { availabilityState: state } });
+    // Newly-available agents may pick up tickets that were waiting.
+    if (state === 'AVAILABLE') await this.assignPending();
+    return { actorId: actor.id, availabilityState: state };
+  }
+
+  private async claim(eventId: string): Promise<boolean> {
+    const res = await this.prisma.iiosProcessedEvent.createMany({
+      data: [{ consumerName: this.consumer, eventId }],
+      skipDuplicates: true,
+    });
+    return res.count > 0;
+  }
+}

packages/iios-service/src/support/assignment.spec.ts

@@ -0,0 +1,103 @@
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
+import { makeFakePorts } from '@insignia/iios-testkit';
+import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
+import { MessageService, type MessagePrincipal } from '../messaging/message.service';
+import { SupportService } from './support.service';
+import { AssignmentService } from './assignment.service';
+import { ActorResolver } from '../identity/actor.resolver';
+import { OutboxBus } from '../outbox/outbox.bus';
+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 actors = new ActorResolver(asService);
+const support = () => new SupportService(asService, makeFakePorts(), actors);
+const assignment = () => new AssignmentService(asService, new OutboxBus(), actors);
+const msg = () => new MessageService(asService, makeFakePorts(), actors);
+
+const cust: MessagePrincipal = { userId: 'cust', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Customer' };
+const agentA: MessagePrincipal = { userId: 'agentA', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Agent A' };
+const agentB: MessagePrincipal = { userId: 'agentB', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Agent B' };
+
+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;
+}
+
+/** Create a queue and add the given agents as AVAILABLE members. */
+async function seedQueue(agents: MessagePrincipal[]): Promise<string> {
+  const asg = assignment();
+  const queue = await asg.createQueue(cust, 'Support'); // any principal in the scope
+  for (const a of agents) {
+    await asg.addMember(queue.id, a);
+    await asg.setAvailability(a, 'AVAILABLE');
+  }
+  return queue.id;
+}
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await resetDb(prisma); });
+
+describe('AssignmentService (P4)', () => {
+  it('assigns an escalated ticket to an available agent; agent joins the thread; base thread unchanged', async () => {
+    await seedQueue([agentA]);
+    const { threadId } = await msg().openThread(null, cust);
+    const ticket = await support().escalate(threadId, cust);
+
+    const asg = assignment();
+    expect(await asg.assignTicket(ticket.id)).toBe(true);
+
+    const updated = await prisma.iiosTicket.findUniqueOrThrow({ where: { id: ticket.id } });
+    expect(updated.state).toBe('OPEN');
+    expect(updated.assignedActorId).toBe(await actorIdFor('agentA'));
+
+    // agent is now a participant on the (still-generic) thread
+    const part = await prisma.iiosThreadParticipant.findUnique({
+      where: { threadId_actorId: { threadId, actorId: await actorIdFor('agentA') } },
+    });
+    expect(part).not.toBeNull();
+    const thread = await prisma.iiosThread.findUniqueOrThrow({ where: { id: threadId } });
+    expect(thread.status).toBe('OPEN'); // base thread never mutated by support
+  });
+
+  it('picks the least-loaded agent within capacity', async () => {
+    const queueId = await seedQueue([agentA, agentB]);
+    // Pre-load agentA with one active ticket so agentB is least-loaded.
+    await prisma.iiosSupportTeamMember.update({
+      where: { queueId_actorId: { queueId, actorId: await actorIdFor('agentA') } },
+      data: { activeCount: 1 },
+    });
+    const t = await support().createTicket(cust, { subject: 's' });
+    await assignment().assignTicket(t.id);
+    const updated = await prisma.iiosTicket.findUniqueOrThrow({ where: { id: t.id } });
+    expect(updated.assignedActorId).toBe(await actorIdFor('agentB'));
+  });
+
+  it('leaves the ticket NEW when no agent is available', async () => {
+    await seedQueue([agentA]);
+    await assignment().setAvailability(agentA, 'OFFLINE');
+    const t = await support().createTicket(cust, { subject: 's' });
+    expect(await assignment().assignTicket(t.id)).toBe(false);
+    expect((await prisma.iiosTicket.findUniqueOrThrow({ where: { id: t.id } })).state).toBe('NEW');
+  });
+
+  it('is idempotent: replaying ticket.created assigns once (no double activeCount)', async () => {
+    const queueId = await seedQueue([agentA]);
+    const t = await support().createTicket(cust, { subject: 's' });
+    const [event] = (await prisma.iiosOutboxEvent.findMany({ where: { eventType: IIOS_EVENTS.ticketCreated } })).map(
+      (r) => r.cloudEvent as unknown as CloudEvent,
+    );
+    const asg = assignment();
+    await asg.onTicketCreated(event);
+    await asg.onTicketCreated(event); // replay
+    const member = await prisma.iiosSupportTeamMember.findFirstOrThrow({
+      where: { queueId, actorId: await actorIdFor('agentA') },
+    });
+    expect(member.activeCount).toBe(1);
+  });
+});

packages/iios-service/src/support/support.module.ts

@@ -1,8 +1,11 @@
 import { Module } from '@nestjs/common';
+import { OutboxModule } from '../outbox/outbox.module';
 import { SupportService } from './support.service';
+import { AssignmentService } from './assignment.service';
 
 @Module({
-  providers: [SupportService],
-  exports: [SupportService],
+  imports: [OutboxModule],
+  providers: [SupportService, AssignmentService],
+  exports: [SupportService, AssignmentService],
 })
 export class SupportModule {}