feat(service): P6.4 RouteProjector — AUTOMATIC forwarding on interaction.normalized
Reacts to channel-bearing (ingested/adapter) interactions, not native DMs; simulates per binding; AUTOMATIC+ALLOW executes to sandbox, MANUAL/REVIEW/DENY stay pending; idempotent. 8 route tests; 63 total. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,56 @@
|
|||||||
|
import { Injectable, OnModuleInit } from '@nestjs/common';
|
||||||
|
import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
|
||||||
|
import { PrismaService } from '../prisma/prisma.service';
|
||||||
|
import { OutboxBus } from '../outbox/outbox.bus';
|
||||||
|
import { RouteService } from './route.service';
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Auto-routing (P6). Reacts to `interaction.normalized` — i.e. messages that
|
||||||
|
* arrived through a channel (adapter ingest). Native DMs have no origin channel,
|
||||||
|
* so they are never routed. For each matching binding it simulates a decision;
|
||||||
|
* only **AUTOMATIC + ALLOW** executes (to the P5 sandbox). MANUAL/REVIEW/DENY
|
||||||
|
* stay pending for an admin; SIMULATION_ONLY never executes. Idempotent.
|
||||||
|
*/
|
||||||
|
@Injectable()
|
||||||
|
export class RouteProjector implements OnModuleInit {
|
||||||
|
private readonly consumer = 'route-projector';
|
||||||
|
|
||||||
|
constructor(
|
||||||
|
private readonly prisma: PrismaService,
|
||||||
|
private readonly bus: OutboxBus,
|
||||||
|
private readonly routes: RouteService,
|
||||||
|
) {}
|
||||||
|
|
||||||
|
onModuleInit(): void {
|
||||||
|
this.bus.on(IIOS_EVENTS.interactionNormalized, (p) => void this.onNormalized(p as CloudEvent).catch(() => {}));
|
||||||
|
}
|
||||||
|
|
||||||
|
async onNormalized(event: CloudEvent): Promise<void> {
|
||||||
|
if (!(await this.claim(event.id))) return;
|
||||||
|
const { interactionId } = event.data as { interactionId: string };
|
||||||
|
const interaction = await this.prisma.iiosInteraction.findUnique({
|
||||||
|
where: { id: interactionId },
|
||||||
|
include: { channel: true },
|
||||||
|
});
|
||||||
|
if (!interaction?.channel) return; // no origin channel → not routable (e.g. native DM)
|
||||||
|
|
||||||
|
const { decisions } = await this.routes.simulate(interactionId, {
|
||||||
|
channelType: interaction.channel.channelType,
|
||||||
|
ref: interaction.channel.externalRef ?? undefined,
|
||||||
|
});
|
||||||
|
|
||||||
|
for (const d of decisions) {
|
||||||
|
if (d.decisionState !== 'ALLOW') continue;
|
||||||
|
const binding = await this.prisma.iiosRouteBinding.findUnique({ where: { id: d.routeBindingId } });
|
||||||
|
if (binding?.mode === 'AUTOMATIC') await this.routes.execute(d, binding);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
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;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -8,8 +8,12 @@ import { ActorResolver } from '../identity/actor.resolver';
|
|||||||
import { RuleEngine } from './rule-engine';
|
import { RuleEngine } from './rule-engine';
|
||||||
import { RouteSimulator } from './route-simulator';
|
import { RouteSimulator } from './route-simulator';
|
||||||
import { RouteService } from './route.service';
|
import { RouteService } from './route.service';
|
||||||
|
import { RouteProjector } from './route.projector';
|
||||||
|
import { IngestService } from '../interactions/ingest.service';
|
||||||
import { OutboundService } from '../adapters/outbound.service';
|
import { OutboundService } from '../adapters/outbound.service';
|
||||||
import { AdapterRegistry } from '../adapters/adapter.registry';
|
import { AdapterRegistry } from '../adapters/adapter.registry';
|
||||||
|
import { OutboxBus } from '../outbox/outbox.bus';
|
||||||
|
import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
|
||||||
import type { PrismaService } from '../prisma/prisma.service';
|
import type { PrismaService } from '../prisma/prisma.service';
|
||||||
|
|
||||||
const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
|
const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
|
||||||
@@ -132,3 +136,70 @@ describe('RouteService approve/deny → execute (P6)', () => {
|
|||||||
expect(await prisma.iiosOutboundCommand.count()).toBe(0);
|
expect(await prisma.iiosOutboundCommand.count()).toBe(0);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe('RouteProjector — AUTOMATIC forwarding (P6)', () => {
|
||||||
|
const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new AdapterRegistry()), actors);
|
||||||
|
const projector = () => new RouteProjector(asService, new OutboxBus(), routeSvc());
|
||||||
|
|
||||||
|
// Ingest an inbound (channel-bearing) interaction — routable, unlike native DMs.
|
||||||
|
async function ingestInbound(text: string): Promise<{ interactionId: string; scopeId: string; event: CloudEvent }> {
|
||||||
|
const ingest = new IngestService(asService, makeFakePorts(), actors);
|
||||||
|
const res = await ingest.ingest(
|
||||||
|
{
|
||||||
|
scope: { orgId: 'org_demo', appId: 'portal-demo' },
|
||||||
|
channel: { type: 'WEBHOOK', externalChannelId: 'webhook' },
|
||||||
|
source: { handleKind: 'EXTERNAL_VISITOR', externalId: 'ext-1' },
|
||||||
|
kind: 'MESSAGE',
|
||||||
|
thread: { externalThreadId: `wh-${randomUUID().slice(0, 6)}` },
|
||||||
|
parts: [{ kind: 'TEXT', bodyText: text }],
|
||||||
|
occurredAt: new Date().toISOString(),
|
||||||
|
providerEventId: `pe-${randomUUID()}`,
|
||||||
|
},
|
||||||
|
`ing-${randomUUID()}`,
|
||||||
|
);
|
||||||
|
const interaction = await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: res.interactionId } });
|
||||||
|
const ev = (await prisma.iiosOutboxEvent.findFirstOrThrow({
|
||||||
|
where: { eventType: IIOS_EVENTS.interactionNormalized, aggregateId: res.interactionId },
|
||||||
|
})).cloudEvent as unknown as CloudEvent;
|
||||||
|
return { interactionId: res.interactionId, scopeId: interaction.scopeId, event: ev };
|
||||||
|
}
|
||||||
|
|
||||||
|
async function automaticBinding(scopeId: string, restriction?: string) {
|
||||||
|
return prisma.iiosRouteBinding.create({
|
||||||
|
data: {
|
||||||
|
scopeId,
|
||||||
|
originChannelType: 'WEBHOOK',
|
||||||
|
originRef: 'webhook',
|
||||||
|
destinationChannelType: 'PORTAL',
|
||||||
|
destinationRef: 'community',
|
||||||
|
restrictionProfile: restriction,
|
||||||
|
mode: 'AUTOMATIC',
|
||||||
|
requiresReview: false,
|
||||||
|
enabled: true,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
it('AUTOMATIC + safe → auto-forwards to sandbox', async () => {
|
||||||
|
const { scopeId, event } = await ingestInbound('good morning everyone');
|
||||||
|
await automaticBinding(scopeId);
|
||||||
|
await projector().onNormalized(event);
|
||||||
|
expect(await prisma.iiosOutboundCommand.count()).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('AUTOMATIC but flagged+protected → NOT forwarded (deny)', async () => {
|
||||||
|
const { scopeId, event } = await ingestInbound('party at 9 PM');
|
||||||
|
await automaticBinding(scopeId, 'CHILD');
|
||||||
|
await projector().onNormalized(event);
|
||||||
|
expect(await prisma.iiosOutboundCommand.count()).toBe(0);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('idempotent: replaying the normalized event forwards once', async () => {
|
||||||
|
const { scopeId, event } = await ingestInbound('hello team');
|
||||||
|
await automaticBinding(scopeId);
|
||||||
|
const p = projector();
|
||||||
|
await p.onNormalized(event);
|
||||||
|
await p.onNormalized(event);
|
||||||
|
expect(await prisma.iiosOutboundCommand.count()).toBe(1);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|||||||
@@ -1,14 +1,16 @@
|
|||||||
import { Module } from '@nestjs/common';
|
import { Module } from '@nestjs/common';
|
||||||
|
import { OutboxModule } from '../outbox/outbox.module';
|
||||||
import { AdaptersModule } from '../adapters/adapters.module';
|
import { AdaptersModule } from '../adapters/adapters.module';
|
||||||
import { RuleEngine } from './rule-engine';
|
import { RuleEngine } from './rule-engine';
|
||||||
import { RouteSimulator } from './route-simulator';
|
import { RouteSimulator } from './route-simulator';
|
||||||
import { RouteService } from './route.service';
|
import { RouteService } from './route.service';
|
||||||
|
import { RouteProjector } from './route.projector';
|
||||||
import { RouteController } from './route.controller';
|
import { RouteController } from './route.controller';
|
||||||
|
|
||||||
@Module({
|
@Module({
|
||||||
imports: [AdaptersModule],
|
imports: [OutboxModule, AdaptersModule],
|
||||||
controllers: [RouteController],
|
controllers: [RouteController],
|
||||||
providers: [RuleEngine, RouteSimulator, RouteService],
|
providers: [RuleEngine, RouteSimulator, RouteService, RouteProjector],
|
||||||
exports: [RuleEngine, RouteSimulator, RouteService],
|
exports: [RuleEngine, RouteSimulator, RouteService],
|
||||||
})
|
})
|
||||||
export class RoutingModule {}
|
export class RoutingModule {}
|
||||||
|
|||||||
Reference in New Issue
Block a user