fix(websocket): project session ownership before tickets
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
import assert from "node:assert/strict";
|
||||
import type { GrowEventRow } from "../src/db/schema.js";
|
||||
import { recordGatewayEvent } from "../src/routes/services.js";
|
||||
|
||||
for (const service of [
|
||||
{
|
||||
source: "interview-service",
|
||||
type: "interview.session.configured",
|
||||
serviceId: "interview",
|
||||
},
|
||||
{
|
||||
source: "roleplay-service",
|
||||
type: "roleplay.scenario.configured",
|
||||
serviceId: "roleplay",
|
||||
},
|
||||
] as const) {
|
||||
const event = {
|
||||
id: `event-${service.serviceId}`,
|
||||
userId: "user-1",
|
||||
orgId: null,
|
||||
source: service.source,
|
||||
type: service.type,
|
||||
category: "service",
|
||||
occurredAt: new Date("2026-07-14T00:00:00.000Z"),
|
||||
receivedAt: new Date("2026-07-14T00:00:00.000Z"),
|
||||
mission: null,
|
||||
subject: { serviceId: service.serviceId, externalId: "session-1" },
|
||||
correlation: { sessionId: "session-1" },
|
||||
payload: {},
|
||||
raw: null,
|
||||
dedupeKey: `dedupe-${service.serviceId}`,
|
||||
processingStatus: "pending",
|
||||
processingError: null,
|
||||
processedAt: null,
|
||||
} as GrowEventRow;
|
||||
|
||||
const order: string[] = [];
|
||||
const returned = await recordGatewayEvent(
|
||||
{
|
||||
userId: "user-1",
|
||||
source: service.source,
|
||||
type: service.type,
|
||||
payload: { session_id: "session-1" },
|
||||
},
|
||||
{
|
||||
recordLegacyEvent: async () => {
|
||||
order.push("legacy");
|
||||
},
|
||||
recordGrowEvent: async () => {
|
||||
order.push("record");
|
||||
return event;
|
||||
},
|
||||
applyServiceSessionProjection: async (projectedEvent) => {
|
||||
assert.strictEqual(projectedEvent, event);
|
||||
order.push("project");
|
||||
return null;
|
||||
},
|
||||
routeGrowEventToUserActor: async (routedEvent) => {
|
||||
assert.strictEqual(routedEvent, event);
|
||||
order.push("route");
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
assert.strictEqual(returned, event);
|
||||
assert.deepEqual(order, ["legacy", "record", "project", "route"], `${service.serviceId} must project ownership before actor routing`);
|
||||
}
|
||||
|
||||
console.log("interview and roleplay service session projection ordering tests passed");
|
||||
+31
-10
@@ -10,6 +10,7 @@ import { signWsTicket } from "../services/ws-tickets.js";
|
||||
import { db } from "../db/client.js";
|
||||
import { events, growEvents, missionServiceSessions } from "../db/schema.js";
|
||||
import { recordGrowEvent } from "../events/record-grow-event.js";
|
||||
import { applyServiceSessionProjection } from "../events/projectors/service-session-projector.js";
|
||||
import { routeGrowEventToUserActor } from "../events/route-to-user-actor.js";
|
||||
import { getRequestUserPreferences, getRequestUserProfile } from "../services/user-context.js";
|
||||
import { log } from "../log.js";
|
||||
@@ -237,7 +238,7 @@ function gatewayDedupeKey(input: {
|
||||
return `${input.source}:${input.type}:${input.userId}:${anchor ?? stableDedupeFingerprint(input)}`;
|
||||
}
|
||||
|
||||
async function recordGatewayEvent(input: {
|
||||
export type GatewayEventInput = {
|
||||
userId: string;
|
||||
source: string;
|
||||
type: string;
|
||||
@@ -246,15 +247,34 @@ async function recordGatewayEvent(input: {
|
||||
mission?: Record<string, unknown>;
|
||||
subject?: Record<string, unknown>;
|
||||
dedupeKey?: string;
|
||||
}) {
|
||||
await db.insert(events).values({
|
||||
userId: input.userId,
|
||||
actorId: "service-gateway",
|
||||
type: input.type,
|
||||
payload: input.payload,
|
||||
}).catch((err) => log.warn({ err, type: input.type }, "failed to record legacy service event"));
|
||||
};
|
||||
|
||||
const event = await recordGrowEvent({
|
||||
export type GatewayEventDependencies = {
|
||||
recordLegacyEvent: (input: GatewayEventInput) => Promise<void>;
|
||||
recordGrowEvent: typeof recordGrowEvent;
|
||||
applyServiceSessionProjection: typeof applyServiceSessionProjection;
|
||||
routeGrowEventToUserActor: typeof routeGrowEventToUserActor;
|
||||
};
|
||||
|
||||
export async function recordGatewayEvent(
|
||||
input: GatewayEventInput,
|
||||
dependencies: GatewayEventDependencies = {
|
||||
recordLegacyEvent: async (gatewayInput) => {
|
||||
await db.insert(events).values({
|
||||
userId: gatewayInput.userId,
|
||||
actorId: "service-gateway",
|
||||
type: gatewayInput.type,
|
||||
payload: gatewayInput.payload,
|
||||
}).catch((err) => log.warn({ err, type: gatewayInput.type }, "failed to record legacy service event"));
|
||||
},
|
||||
recordGrowEvent,
|
||||
applyServiceSessionProjection,
|
||||
routeGrowEventToUserActor,
|
||||
},
|
||||
) {
|
||||
await dependencies.recordLegacyEvent(input);
|
||||
|
||||
const event = await dependencies.recordGrowEvent({
|
||||
source: input.source,
|
||||
type: input.type,
|
||||
category: "service",
|
||||
@@ -269,7 +289,8 @@ async function recordGatewayEvent(input: {
|
||||
correlation: input.correlation,
|
||||
payload: input.payload,
|
||||
});
|
||||
await routeGrowEventToUserActor(event);
|
||||
await dependencies.applyServiceSessionProjection(event);
|
||||
await dependencies.routeGrowEventToUserActor(event);
|
||||
return event;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user