From 859124914cccd6b901be602b1b2ce0b6b5b17857 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Wed, 1 Jul 2026 11:55:31 +0530 Subject: [PATCH] =?UTF-8?q?feat(service):=20P4.3=20AssignmentService=20?= =?UTF-8?q?=E2=80=94=20event-driven=20least-loaded=20assignment?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- packages/iios-service/src/app.module.ts | 2 + .../src/support/assignment.service.ts | 135 ++++++++++++++++++ .../src/support/assignment.spec.ts | 103 +++++++++++++ .../src/support/support.module.ts | 7 +- 4 files changed, 245 insertions(+), 2 deletions(-) create mode 100644 packages/iios-service/src/support/assignment.service.ts create mode 100644 packages/iios-service/src/support/assignment.spec.ts diff --git a/packages/iios-service/src/app.module.ts b/packages/iios-service/src/app.module.ts index ceaa390..cb6feed 100644 --- a/packages/iios-service/src/app.module.ts +++ b/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], }) diff --git a/packages/iios-service/src/support/assignment.service.ts b/packages/iios-service/src/support/assignment.service.ts new file mode 100644 index 0000000..a19adf3 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + const res = await this.prisma.iiosProcessedEvent.createMany({ + data: [{ consumerName: this.consumer, eventId }], + skipDuplicates: true, + }); + return res.count > 0; + } +} diff --git a/packages/iios-service/src/support/assignment.spec.ts b/packages/iios-service/src/support/assignment.spec.ts new file mode 100644 index 0000000..5f08262 --- /dev/null +++ b/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 { + 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 { + 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); + }); +}); diff --git a/packages/iios-service/src/support/support.module.ts b/packages/iios-service/src/support/support.module.ts index 8da8f18..7257017 100644 --- a/packages/iios-service/src/support/support.module.ts +++ b/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 {}