fix(events): correlate service completions to curator tasks

2026-07-14 09:55:52 +05:30
parent 8d43bea58e
commit 41f9f0dfec
2 changed files with 161 additions and 13 deletions
@@ -0,0 +1,61 @@
+import assert from "node:assert/strict";
+import { correlateEventWithServiceSession } from "../src/events/record-grow-event.js";
+import type { GrowEventEnvelope } from "../src/events/envelope.js";
+
+const event: GrowEventEnvelope = {
+  id: "event-1",
+  userId: "user-a",
+  source: "interview-service",
+  type: "interview.session.completed",
+  category: "service",
+  occurredAt: "2026-07-14T00:00:00.000Z",
+  correlation: { sessionId: "session-1" },
+  payload: { status: "completed" },
+};
+
+const link = {
+  userId: "user-a",
+  missionInstanceId: "sprint-1",
+  missionId: "curator-sprint",
+  stageId: "day-1:interview",
+  metadata: { taskId: "curator-task-1" },
+};
+
+{
+  const correlated = correlateEventWithServiceSession(event, "user-a", link);
+  assert.deepEqual(correlated.correlation, {
+    sessionId: "session-1",
+    taskId: "curator-task-1",
+    curatorTaskId: "curator-task-1",
+    missionInstanceId: "sprint-1",
+    missionId: "curator-sprint",
+    stageId: "day-1:interview",
+  });
+  assert.deepEqual(correlated.mission, {
+    instanceId: "sprint-1",
+    missionId: "curator-sprint",
+    stageId: "day-1:interview",
+  });
+}
+
+{
+  const foreign = correlateEventWithServiceSession(event, "user-a", { ...link, userId: "user-b" });
+  assert.strictEqual(foreign, event, "a service session owned by another user must never enrich the event");
+}
+
+{
+  const unresolved = correlateEventWithServiceSession(event, undefined, link);
+  assert.strictEqual(unresolved, event, "an unresolved event must never inherit session context");
+}
+
+{
+  const explicit = correlateEventWithServiceSession({
+    ...event,
+    mission: { missionId: "explicit-mission", stageId: "explicit-stage" },
+    correlation: { sessionId: "session-1", taskId: "explicit-task" },
+  }, "user-a", link);
+  assert.equal(explicit.correlation?.taskId, "explicit-task");
+  assert.deepEqual(explicit.mission, { missionId: "explicit-mission", stageId: "explicit-stage" });
+}
+
+console.log("service session correlation tests passed");
@@ -1,4 +1,4 @@
-import { and, eq } from "drizzle-orm";
+import { and, desc, eq, sql } from "drizzle-orm";
 import { db } from "../db/client.js";
 import { growEvents, missionServiceSessions, users, type GrowEventRow } from "../db/schema.js";
 import { asRecord, getString, type GrowEventEnvelope } from "./envelope.js";
@@ -11,6 +11,39 @@ async function ensureUser(userId: string) {
    .onConflictDoNothing();
 }

+type ServiceSessionLink = Pick<
+  typeof missionServiceSessions.$inferSelect,
+  "userId" | "missionInstanceId" | "missionId" | "stageId" | "metadata"
+>;
+
+export function correlateEventWithServiceSession(
+  event: GrowEventEnvelope,
+  resolvedUserId: string | undefined,
+  link: ServiceSessionLink | undefined,
+): GrowEventEnvelope {
+  if (!resolvedUserId || !link || link.userId !== resolvedUserId) return event;
+  const metadata = asRecord(link.metadata);
+  const taskId = getString(metadata.taskId ?? metadata.curatorTaskId ?? metadata.curator_task_id);
+  const missionInstanceId = getString(link.missionInstanceId ?? metadata.missionInstanceId ?? metadata.mission_instance_id);
+  const missionId = getString(link.missionId ?? metadata.missionId ?? metadata.mission_id);
+  const stageId = getString(link.stageId ?? metadata.stageId ?? metadata.stage_id);
+  const eventCorrelation = event.correlation ?? {};
+  const explicitTaskId = getString(eventCorrelation.taskId ?? eventCorrelation.curatorTaskId ?? eventCorrelation.curator_task_id);
+  const correlation = {
+    ...(taskId && !explicitTaskId ? { taskId, curatorTaskId: taskId } : {}),
+    ...(missionInstanceId && !getString(eventCorrelation.missionInstanceId) ? { missionInstanceId } : {}),
+    ...(missionId && !getString(eventCorrelation.missionId) ? { missionId } : {}),
+    ...(stageId && !getString(eventCorrelation.stageId) ? { stageId } : {}),
+    ...eventCorrelation,
+  };
+  const mission = event.mission ?? (
+    missionInstanceId || missionId || stageId
+      ? { instanceId: missionInstanceId, missionId, stageId }
+      : undefined
+  );
+  return { ...event, correlation, mission };
+}
+
 async function resolveUserId(event: GrowEventEnvelope): Promise<string | undefined> {
  if (event.userId) return event.userId;
  const sessionId = getString(event.correlation?.sessionId ?? event.correlation?.session_id);
@@ -26,6 +59,58 @@ async function resolveUserId(event: GrowEventEnvelope): Promise<string | undefin
  return undefined;
 }

+async function resolveServiceSessionLink(event: GrowEventEnvelope, userId: string | undefined): Promise<ServiceSessionLink | undefined> {
+  if (!userId) return undefined;
+  const sessionId = getString(event.correlation?.sessionId ?? event.correlation?.session_id);
+  if (!sessionId) return undefined;
+  const serviceId = normalizeServiceId(event.source);
+  const [row] = await db
+    .select({
+      userId: missionServiceSessions.userId,
+      missionInstanceId: missionServiceSessions.missionInstanceId,
+      missionId: missionServiceSessions.missionId,
+      stageId: missionServiceSessions.stageId,
+      metadata: missionServiceSessions.metadata,
+    })
+    .from(missionServiceSessions)
+    .where(and(
+      eq(missionServiceSessions.userId, userId),
+      eq(missionServiceSessions.serviceId, serviceId),
+      eq(missionServiceSessions.externalId, sessionId),
+    ))
+    .limit(1);
+
+  const rowMetadata = asRecord(row?.metadata);
+  const rowTaskId = getString(rowMetadata.taskId ?? rowMetadata.curatorTaskId ?? rowMetadata.curator_task_id);
+  if (rowTaskId) return row;
+
+  const [prior] = await db
+    .select({ correlation: growEvents.correlation, payload: growEvents.payload, mission: growEvents.mission })
+    .from(growEvents)
+    .where(and(
+      eq(growEvents.userId, userId),
+      sql`${growEvents.correlation}->>'sessionId' = ${sessionId}`,
+      sql`coalesce(${growEvents.correlation}->>'taskId', ${growEvents.correlation}->>'curatorTaskId', ${growEvents.payload}->>'taskId', ${growEvents.payload}->>'curatorTaskId') is not null`,
+    ))
+    .orderBy(desc(growEvents.occurredAt))
+    .limit(1);
+  if (!prior) return row;
+  const priorCorrelation = asRecord(prior.correlation);
+  const priorPayload = asRecord(prior.payload);
+  const priorMission = asRecord(prior.mission);
+  return {
+    userId,
+    missionInstanceId: row?.missionInstanceId ?? getString(priorMission.instanceId ?? priorCorrelation.missionInstanceId) ?? null,
+    missionId: row?.missionId ?? getString(priorMission.missionId ?? priorCorrelation.missionId) ?? null,
+    stageId: row?.stageId ?? getString(priorMission.stageId ?? priorCorrelation.stageId) ?? null,
+    metadata: {
+      ...priorPayload,
+      ...priorCorrelation,
+      ...rowMetadata,
+    },
+  };
+}
+
 export function normalizeServiceId(source: string): string {
  if (source.includes("interview")) return "interview";
  if (source.includes("roleplay")) return "roleplay";
@@ -44,24 +129,26 @@ export async function recordGrowEvent(input: unknown, overrides: { userId?: stri
 export async function recordGrowEventWithResult(input: unknown, overrides: { userId?: string; source?: string } = {}): Promise<{ event: GrowEventRow; inserted: boolean }> {
  const normalized = normalizeGrowEvent(input, overrides);
  const resolvedUserId = await resolveUserId(normalized);
+  const sessionLink = await resolveServiceSessionLink(normalized, resolvedUserId);
+  const correlated = correlateEventWithServiceSession(normalized, resolvedUserId, sessionLink);
  if (resolvedUserId) await ensureUser(resolvedUserId);

  const processingStatus = resolvedUserId ? "pending" : "unresolved";

  const values = {
-    id: normalized.id,
+    id: correlated.id,
    userId: resolvedUserId,
-    orgId: normalized.orgId,
-    source: normalized.source,
-    type: normalized.type,
-    category: normalized.category,
-    occurredAt: new Date(normalized.occurredAt),
-    mission: normalized.mission as Record<string, unknown> | undefined,
-    subject: normalized.subject as Record<string, unknown> | undefined,
-    correlation: normalized.correlation as Record<string, unknown> | undefined,
-    payload: normalized.payload,
-    raw: normalized.raw ?? asRecord(input),
-    dedupeKey: normalized.dedupeKey,
+    orgId: correlated.orgId,
+    source: correlated.source,
+    type: correlated.type,
+    category: correlated.category,
+    occurredAt: new Date(correlated.occurredAt),
+    mission: correlated.mission as Record<string, unknown> | undefined,
+    subject: correlated.subject as Record<string, unknown> | undefined,
+    correlation: correlated.correlation as Record<string, unknown> | undefined,
+    payload: correlated.payload,
+    raw: correlated.raw ?? asRecord(input),
+    dedupeKey: correlated.dedupeKey,
    processingStatus,
  } satisfies typeof growEvents.$inferInsert;