From f03de1ea58b90280dc3cf12fb32fdd1644993626 Mon Sep 17 00:00:00 2001 From: -Puter <22245429+puterhimself@users.noreply.github.com> Date: Thu, 4 Jun 2026 16:12:32 +0530 Subject: [PATCH] feat: add grow event backbone --- drizzle/0007_grow_event_backbone.sql | 123 ++++++++++ drizzle/meta/_journal.json | 7 + package.json | 1 + pnpm-lock.yaml | 87 +++++++ src/actors/events/index.ts | 1 + src/actors/events/user-event-actor.ts | 214 ++++++++++++++++++ src/actors/registry.ts | 2 + src/config.ts | 6 + src/db/schema.ts | 136 +++++++++++ src/events/envelope.ts | 93 ++++++++ src/events/normalize.ts | 79 +++++++ src/events/projectors/projection-agent.ts | 104 +++++++++ src/events/projectors/qscore-projector.ts | 189 ++++++++++++++++ .../projectors/service-session-projector.ts | 76 +++++++ src/events/record-grow-event.ts | 114 ++++++++++ src/events/redis-consumer.ts | 102 +++++++++ src/events/route-to-user-actor.ts | 15 ++ src/index.ts | 4 + src/missions/event-reducers.ts | 8 + src/missions/interview-to-offer/reducer.ts | 99 ++++++++ src/missions/reducer-types.ts | 40 ++++ src/routes/events.ts | 61 +++++ src/routes/services.ts | 110 ++++++++- 23 files changed, 1661 insertions(+), 10 deletions(-) create mode 100644 drizzle/0007_grow_event_backbone.sql create mode 100644 src/actors/events/index.ts create mode 100644 src/actors/events/user-event-actor.ts create mode 100644 src/events/envelope.ts create mode 100644 src/events/normalize.ts create mode 100644 src/events/projectors/projection-agent.ts create mode 100644 src/events/projectors/qscore-projector.ts create mode 100644 src/events/projectors/service-session-projector.ts create mode 100644 src/events/record-grow-event.ts create mode 100644 src/events/redis-consumer.ts create mode 100644 src/events/route-to-user-actor.ts create mode 100644 src/missions/event-reducers.ts create mode 100644 src/missions/interview-to-offer/reducer.ts create mode 100644 src/missions/reducer-types.ts create mode 100644 src/routes/events.ts diff --git a/drizzle/0007_grow_event_backbone.sql b/drizzle/0007_grow_event_backbone.sql new file mode 100644 index 0000000..058dec1 --- /dev/null +++ b/drizzle/0007_grow_event_backbone.sql @@ -0,0 +1,123 @@ +-- Grow event backbone: raw event inbox, service session links, artifacts, and QScore projections. +-- Postgres stores every raw GrowEvent first. Actors/projectors can rebuild from this stream. + +CREATE TABLE IF NOT EXISTS grow_events ( + id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text, + user_id TEXT REFERENCES users(id) ON DELETE CASCADE, + org_id TEXT, + source TEXT NOT NULL, + type TEXT NOT NULL, + category TEXT NOT NULL DEFAULT 'service' CHECK (category IN ('mission', 'service', 'artifact', 'usage', 'qscore', 'entitlement', 'system')), + occurred_at TIMESTAMPTZ NOT NULL, + received_at TIMESTAMPTZ NOT NULL DEFAULT now(), + mission JSONB, + subject JSONB, + correlation JSONB, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + raw JSONB, + dedupe_key TEXT, + processing_status TEXT NOT NULL DEFAULT 'pending' CHECK (processing_status IN ('pending', 'processing', 'processed', 'failed', 'unresolved')), + processing_error TEXT, + processed_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS grow_events_user_idx + ON grow_events(user_id, occurred_at DESC); +CREATE INDEX IF NOT EXISTS grow_events_status_idx + ON grow_events(processing_status, received_at ASC); +CREATE INDEX IF NOT EXISTS grow_events_source_idx + ON grow_events(source, type, occurred_at DESC); +CREATE UNIQUE INDEX IF NOT EXISTS grow_events_dedupe_idx + ON grow_events(dedupe_key); + +CREATE TABLE IF NOT EXISTS mission_service_sessions ( + id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text, + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + mission_instance_id TEXT REFERENCES grow_active_missions(instance_id) ON DELETE SET NULL, + mission_id TEXT, + stage_id TEXT, + service_id TEXT NOT NULL, + external_id TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'active', + metadata JSONB, + last_event_id TEXT REFERENCES grow_events(id) ON DELETE SET NULL, + last_checked_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS mission_service_sessions_user_idx + ON mission_service_sessions(user_id, updated_at DESC); +CREATE UNIQUE INDEX IF NOT EXISTS mission_service_sessions_external_idx + ON mission_service_sessions(service_id, external_id); +CREATE INDEX IF NOT EXISTS mission_service_sessions_mission_idx + ON mission_service_sessions(mission_instance_id, stage_id); + +CREATE TABLE IF NOT EXISTS mission_artifacts ( + id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text, + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + mission_instance_id TEXT REFERENCES grow_active_missions(instance_id) ON DELETE CASCADE, + mission_id TEXT, + stage_id TEXT, + source_event_id TEXT REFERENCES grow_events(id) ON DELETE SET NULL, + service_id TEXT, + external_id TEXT, + type TEXT NOT NULL, + title TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'ready', + summary TEXT, + content_md TEXT, + metadata JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS mission_artifacts_user_idx + ON mission_artifacts(user_id, created_at DESC); +CREATE INDEX IF NOT EXISTS mission_artifacts_mission_idx + ON mission_artifacts(mission_instance_id, created_at DESC); +CREATE INDEX IF NOT EXISTS mission_artifacts_external_idx + ON mission_artifacts(service_id, external_id); + +CREATE TABLE IF NOT EXISTS grow_qscore_signals ( + id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text, + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + source_event_id TEXT REFERENCES grow_events(id) ON DELETE SET NULL, + signal_id TEXT NOT NULL, + score DOUBLE PRECISION NOT NULL, + present BOOLEAN NOT NULL DEFAULT TRUE, + source TEXT, + raw JSONB, + occurred_at TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS grow_qscore_signals_user_idx + ON grow_qscore_signals(user_id, occurred_at DESC); +CREATE INDEX IF NOT EXISTS grow_qscore_signals_signal_idx + ON grow_qscore_signals(signal_id, occurred_at DESC); + +CREATE TABLE IF NOT EXISTS grow_qscore_latest ( + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + signal_id TEXT NOT NULL, + score DOUBLE PRECISION NOT NULL, + present BOOLEAN NOT NULL DEFAULT TRUE, + source TEXT, + source_event_id TEXT REFERENCES grow_events(id) ON DELETE SET NULL, + raw JSONB, + occurred_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (user_id, signal_id) +); + +CREATE INDEX IF NOT EXISTS grow_qscore_latest_user_idx + ON grow_qscore_latest(user_id, updated_at DESC); + +CREATE TABLE IF NOT EXISTS grow_qscore_projection_state ( + user_id TEXT PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE, + score INTEGER NOT NULL DEFAULT 0, + signal_count INTEGER NOT NULL DEFAULT 0, + dimensions JSONB, + summary TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index 570b5a8..0bbb5f4 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -50,6 +50,13 @@ "when": 1780481100000, "tag": "0006_conversations_active_missions", "breakpoints": true + }, + { + "idx": 7, + "version": "7", + "when": 1780481200000, + "tag": "0007_grow_event_backbone", + "breakpoints": true } ] } \ No newline at end of file diff --git a/package.json b/package.json index 2e169ea..7cab4c2 100644 --- a/package.json +++ b/package.json @@ -32,6 +32,7 @@ "pino": "^9.5.0", "pino-pretty": "^13.0.0", "postgres": "^3.4.5", + "redis": "^4.7.0", "rivetkit": "^2.2.1", "zod": "^3.24.1" }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 2626cf6..127e9a9 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -41,6 +41,9 @@ importers: postgres: specifier: ^3.4.5 version: 3.4.9 + redis: + specifier: ^4.7.0 + version: 4.7.1 rivetkit: specifier: ^2.2.1 version: 2.2.1(@standard-schema/spec@1.0.0)(dockerode@4.0.12)(drizzle-kit@0.31.10)(ws@8.21.0) @@ -858,6 +861,35 @@ packages: '@protobufjs/utf8@1.1.1': resolution: {integrity: sha512-oOAWABowe8EAbMyWKM0tYDKi8Yaox52D+HWZhAIJqQXbqe0xI/GV7FhLWqlEKreMkfDjshR5FKgi3mnle0h6Eg==} + '@redis/bloom@1.2.0': + resolution: {integrity: sha512-HG2DFjYKbpNmVXsa0keLHp/3leGJz1mjh09f2RLGGLQZzSHpkmZWuwJbAvo3QcRY8p80m5+ZdXZdYOSBLlp7Cg==} + peerDependencies: + '@redis/client': ^1.0.0 + + '@redis/client@1.6.1': + resolution: {integrity: sha512-/KCsg3xSlR+nCK8/8ZYSknYxvXHwubJrU82F3Lm1Fp6789VQ0/3RJKfsmRXjqfaTA++23CvC3hqmqe/2GEt6Kw==} + engines: {node: '>=14'} + + '@redis/graph@1.1.1': + resolution: {integrity: sha512-FEMTcTHZozZciLRl6GiiIB4zGm5z5F3F6a6FZCyrfxdKOhFlGkiAqlexWMBzCi4DcRoyiOsuLfW+cjlGWyExOw==} + peerDependencies: + '@redis/client': ^1.0.0 + + '@redis/json@1.0.7': + resolution: {integrity: sha512-6UyXfjVaTBTJtKNG4/9Z8PSpKE6XgSyEb8iwaqDcy+uKrd/DGYHTWkUdnQDyzm727V7p21WUMhsqz5oy65kPcQ==} + peerDependencies: + '@redis/client': ^1.0.0 + + '@redis/search@1.2.0': + resolution: {integrity: sha512-tYoDBbtqOVigEDMAcTGsRlMycIIjwMCgD8eR2t0NANeQmgK/lvxNAvYyb6bZDD4frHRhIHkJu2TBRvB0ERkOmw==} + peerDependencies: + '@redis/client': ^1.0.0 + + '@redis/time-series@1.1.0': + resolution: {integrity: sha512-c1Q99M5ljsIuc4YdaCwfUEXsofakb9c8+Zse2qxTadu8TalLXuAESzLvFAvNVbkmSlvlzIQOLpBCmWI9wTOt+g==} + peerDependencies: + '@redis/client': ^1.0.0 + '@rivet-dev/agent-os-core@0.1.1': resolution: {integrity: sha512-Uw5jr+gUXDY7TDUFqlypjGe1BD2KL9kTHPNo/f1iNS1R+l9IWuvT+FF/MXsLOEkc3fB06OPVu2ZvUuNwp9MpLQ==} @@ -1131,6 +1163,10 @@ packages: resolution: {integrity: sha512-BSeNnyus75C4//NQ9gQt1/csTXyo/8Sb+afLAkzAptFuMsod9HFokGNudZpi/oQV73hnVK+sR+5PVRMd+Dr7YQ==} engines: {node: '>=12'} + cluster-key-slot@1.1.2: + resolution: {integrity: sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==} + engines: {node: '>=0.10.0'} + color-convert@2.0.1: resolution: {integrity: sha512-RRECPsj7iu/xb5oKYcsFHSppFNnsj/52OVTRKb4zP5onXwVF3zVmmToNcOfGC+CRDpfK/U584fMg38ZHCaElKQ==} engines: {node: '>=7.0.0'} @@ -1440,6 +1476,10 @@ packages: resolution: {integrity: sha512-SFdFmIJi+ybC0vjlHN0ZGVGHc3lgE0DxPAT0djjVg+kjOnSqclqmj0KQ7ykTOLP6YxoqOvuAODGdcHJn+43q3g==} engines: {node: '>= 0.4'} + generic-pool@3.9.0: + resolution: {integrity: sha512-hymDOu5B53XvN4QT9dBmZxPX4CWhBPPLguTZ9MMFeFa/Kg0xWVfylOVNlJji/E7yTZWFd/q9GO5TxDLq156D7g==} + engines: {node: '>= 4'} + get-caller-file@2.0.5: resolution: {integrity: sha512-DyFP3BM/3YHTQOCUL/w0OZHR0lpKeGrxotcHWcqNEdnltqFwXVfhEBQ94eIo34AfQpo0rGki4cyIiftY06h2Fg==} engines: {node: 6.* || 8.* || >= 10.*} @@ -1823,6 +1863,9 @@ packages: resolution: {integrity: sha512-57frrGM/OCTLqLOAh0mhVA9VBMHd+9U7Zb2THMGdBUoZVOtGbJzjxsYGDJ3A9AYYCP4hn6y1TVbaOfzWtm5GFg==} engines: {node: '>= 12.13.0'} + redis@4.7.1: + resolution: {integrity: sha512-S1bJDnqLftzHXHP8JsT5II/CtHWQrASX5K96REjWjlmWKrviSOLWmM7QnRLstAWsu1VBBV1ffV6DzCvxNP0UJQ==} + require-directory@2.1.1: resolution: {integrity: sha512-fGxEI7+wsG9xrvdjsrlmL22OMTTiHRwAMroiEeMgq8gzoLC/PQr7RsRDSTLUg/bZAZtF+TVIkHc6/4RIKrui+Q==} engines: {node: '>=0.10.0'} @@ -2165,6 +2208,9 @@ packages: resolution: {integrity: sha512-0pfFzegeDWJHJIAmTLRP2DwHjdF5s7jo9tuztdQxAhINCdvS+3nGINqPd00AphqJR/0LhANUS6/+7SCb98YOfA==} engines: {node: '>=10'} + yallist@4.0.0: + resolution: {integrity: sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==} + yallist@5.0.0: resolution: {integrity: sha512-YgvUTfwqyc7UXVMrB+SImsVYSmTS8X/tSrtdNZMImM+n7+QTriRXyXim0mBrTXNeqzVF0KWGgHPeiyViFFrNDw==} engines: {node: '>=18'} @@ -2670,6 +2716,32 @@ snapshots: '@protobufjs/utf8@1.1.1': {} + '@redis/bloom@1.2.0(@redis/client@1.6.1)': + dependencies: + '@redis/client': 1.6.1 + + '@redis/client@1.6.1': + dependencies: + cluster-key-slot: 1.1.2 + generic-pool: 3.9.0 + yallist: 4.0.0 + + '@redis/graph@1.1.1(@redis/client@1.6.1)': + dependencies: + '@redis/client': 1.6.1 + + '@redis/json@1.0.7(@redis/client@1.6.1)': + dependencies: + '@redis/client': 1.6.1 + + '@redis/search@1.2.0(@redis/client@1.6.1)': + dependencies: + '@redis/client': 1.6.1 + + '@redis/time-series@1.1.0(@redis/client@1.6.1)': + dependencies: + '@redis/client': 1.6.1 + '@rivet-dev/agent-os-core@0.1.1': dependencies: '@rivet-dev/agent-os-posix': 0.1.0 @@ -3016,6 +3088,8 @@ snapshots: strip-ansi: 6.0.1 wrap-ansi: 7.0.0 + cluster-key-slot@1.1.2: {} + color-convert@2.0.1: dependencies: color-name: 1.1.4 @@ -3346,6 +3420,8 @@ snapshots: generator-function@2.0.1: {} + generic-pool@3.9.0: {} + get-caller-file@2.0.5: {} get-intrinsic@1.3.0: @@ -3799,6 +3875,15 @@ snapshots: real-require@0.2.0: {} + redis@4.7.1: + dependencies: + '@redis/bloom': 1.2.0(@redis/client@1.6.1) + '@redis/client': 1.6.1 + '@redis/graph': 1.1.1(@redis/client@1.6.1) + '@redis/json': 1.0.7(@redis/client@1.6.1) + '@redis/search': 1.2.0(@redis/client@1.6.1) + '@redis/time-series': 1.1.0(@redis/client@1.6.1) + require-directory@2.1.1: {} resolve-pkg-maps@1.0.0: {} @@ -4141,6 +4226,8 @@ snapshots: y18n@5.0.8: {} + yallist@4.0.0: {} + yallist@5.0.0: {} yaml@2.9.0: {} diff --git a/src/actors/events/index.ts b/src/actors/events/index.ts new file mode 100644 index 0000000..defca4c --- /dev/null +++ b/src/actors/events/index.ts @@ -0,0 +1 @@ +export { userEventActor } from "./user-event-actor.js"; diff --git a/src/actors/events/user-event-actor.ts b/src/actors/events/user-event-actor.ts new file mode 100644 index 0000000..5e29f16 --- /dev/null +++ b/src/actors/events/user-event-actor.ts @@ -0,0 +1,214 @@ +import { actor, event, queue } from "rivetkit"; +import { workflow } from "rivetkit/workflow"; +import { and, eq } from "drizzle-orm"; +import { db } from "../../db/client.js"; +import { growEvents, missionArtifacts } from "../../db/schema.js"; +import { listActiveMissionsPg, upsertActiveMissionPg } from "../../grow/persistence.js"; +import type { GrowActiveMission, MissionActorType, MissionSnapshot } from "../missions/types.js"; +import { applyQscoreProjection } from "../../events/projectors/qscore-projector.js"; +import { applyServiceSessionProjection } from "../../events/projectors/service-session-projector.js"; +import { getProjectionInsight } from "../../events/projectors/projection-agent.js"; +import { markGrowEventFailed, markGrowEventProcessed, markGrowEventProcessing } from "../../events/record-grow-event.js"; +import { reducersForMission } from "../../missions/event-reducers.js"; +import type { MissionArtifactPatch, MissionStagePatch } from "../../missions/reducer-types.js"; + +export type UserEventCommand = { + userId: string; + eventId: string; +}; + +type UserEventActorState = { + userId: string; + processedCount: number; + lastEventId?: string; + lastError?: string; + processedEventIds: string[]; + updatedAt?: string; +}; + +function summarizeMissionSnapshot(snapshot: MissionSnapshot): GrowActiveMission { + return { + instanceId: snapshot.instanceId, + missionId: snapshot.missionId, + workflowId: snapshot.workflowId, + title: snapshot.title, + shortTitle: snapshot.shortTitle, + status: snapshot.status, + progressPercent: snapshot.progressPercent, + currentStageId: snapshot.currentStageId, + goal: snapshot.goal, + actorType: actorTypeFor(snapshot.missionId), + createdAt: new Date(snapshot.createdAt).getTime(), + updatedAt: new Date(snapshot.updatedAt).getTime(), + }; +} + +function actorTypeFor(missionId: string): MissionActorType | undefined { + if (missionId === "interview-to-offer") return "interviewToOfferMissionActor"; + if (missionId === "career-transition") return "careerTransitionMissionActor"; + if (missionId === "salary-negotiation-war-room") return "salaryNegotiationWarRoomMissionActor"; + if (missionId === "promotion-readiness") return "promotionReadinessMissionActor"; + if (missionId === "personal-brand-opportunity-engine") return "personalBrandOpportunityEngineMissionActor"; + return undefined; +} + +function missionActorHandle(client: any, userId: string, mission: GrowActiveMission) { + const key = [userId, mission.instanceId]; + switch (mission.actorType) { + case "interviewToOfferMissionActor": return client.interviewToOfferMissionActor.getOrCreate(key); + case "careerTransitionMissionActor": return client.careerTransitionMissionActor.getOrCreate(key); + case "salaryNegotiationWarRoomMissionActor": return client.salaryNegotiationWarRoomMissionActor.getOrCreate(key); + case "promotionReadinessMissionActor": return client.promotionReadinessMissionActor.getOrCreate(key); + case "personalBrandOpportunityEngineMissionActor": return client.personalBrandOpportunityEngineMissionActor.getOrCreate(key); + default: return null; + } +} + +async function applyStagePatches(actorHandle: any, patches: MissionStagePatch[]) { + let snapshot: MissionSnapshot | null = null; + const deduped = new Map(); + for (const patch of patches) deduped.set(`${patch.stageId}:${patch.status ?? ""}:${patch.progressPercent ?? ""}`, patch); + for (const patch of deduped.values()) { + snapshot = await actorHandle.updateStage(patch); + } + return snapshot; +} + +async function applyArtifactPatches(input: { + actorHandle: any; + userId: string; + mission: GrowActiveMission; + eventId: string; + serviceId: string; + externalId?: string; + patches: MissionArtifactPatch[]; +}) { + const created = []; + for (const patch of input.patches) { + const artifact = await input.actorHandle.addArtifact({ + type: patch.type, + title: patch.title, + status: "ready", + summary: patch.summary, + contentMd: patch.contentMd, + metadata: { ...(patch.metadata ?? {}), sourceEventId: input.eventId }, + }); + created.push(artifact); + await db.insert(missionArtifacts).values({ + id: artifact.id, + userId: input.userId, + missionInstanceId: input.mission.instanceId, + missionId: input.mission.missionId, + stageId: patch.stageId, + sourceEventId: input.eventId, + serviceId: input.serviceId, + externalId: input.externalId, + type: patch.type, + title: patch.title, + status: "ready", + summary: patch.summary, + contentMd: patch.contentMd, + metadata: patch.metadata, + createdAt: new Date(artifact.createdAt), + updatedAt: new Date(artifact.updatedAt), + }).onConflictDoNothing(); + } + return created; +} + +export const userEventActor = actor({ + options: { name: "User Event Parser", icon: "route", noSleep: true, actionTimeout: 300_000 }, + state: { userId: "", processedCount: 0, processedEventIds: [] } as UserEventActorState, + events: { + updated: event(), + eventProcessed: event<{ eventId: string; userId: string }>(), + }, + queues: { + events: queue(), + }, + actions: { + enqueueEvent: async (c, input: UserEventCommand) => { + if (c.state.userId && c.state.userId !== input.userId) throw new Error("userEventActor initialized for a different user"); + c.state.userId = input.userId; + c.state.updatedAt = new Date().toISOString(); + await c.queue.send("events", input); + c.broadcast("updated", c.state); + return { queued: true }; + }, + getState: (c) => c.state, + }, + run: workflow(async (ctx) => { + await ctx.loop("user-event-loop", async (loopCtx) => { + const message = await loopCtx.queue.next("wait-event", { names: ["events"] }); + const cmd = message.body as UserEventCommand; + + await loopCtx.step(`process-event:${cmd.eventId}`, async () => { + if (loopCtx.state.processedEventIds.includes(cmd.eventId)) return; + await markGrowEventProcessing(cmd.eventId); + const [row] = await db + .select() + .from(growEvents) + .where(and(eq(growEvents.id, cmd.eventId), eq(growEvents.userId, cmd.userId))) + .limit(1); + if (!row) throw new Error(`grow event not found for user: ${cmd.eventId}`); + + try { + await applyServiceSessionProjection(row); + const qscoreResult = await applyQscoreProjection(row); + const activeRows = await listActiveMissionsPg(cmd.userId); + const insight = await getProjectionInsight({ + event: row, + qscoreSignals: qscoreResult.signals, + activeMissionIds: activeRows.map((item) => item.mission.missionId), + }); + + const client = loopCtx.client(); + for (const active of activeRows) { + const mission = active.mission; + const reducers = reducersForMission(mission.missionId); + if (!reducers.length) continue; + for (const reducer of reducers) { + const reduceCtx = { userId: cmd.userId, activeMission: mission, event: row, qscoreSignals: qscoreResult.signals, insight }; + if (!reducer.accepts(reduceCtx)) continue; + const reduction = reducer.reduce(reduceCtx); + if (!reduction.stagePatches.length && !reduction.artifacts.length && !reduction.eventMessage) continue; + + const actorHandle = missionActorHandle(client, cmd.userId, mission); + if (!actorHandle) continue; + if (reduction.eventMessage) { + await actorHandle.recordEvent({ type: row.type, message: reduction.eventMessage, payload: { sourceEventId: row.id } }); + } + let snapshot: MissionSnapshot | undefined = reduction.stagePatches.length ? (await applyStagePatches(actorHandle, reduction.stagePatches) ?? undefined) : undefined; + await applyArtifactPatches({ + actorHandle, + userId: cmd.userId, + mission, + eventId: row.id, + serviceId: row.source, + externalId: typeof row.correlation?.sessionId === "string" ? row.correlation.sessionId : undefined, + patches: reduction.artifacts, + }); + const finalSnapshot = (await actorHandle.getState()) as MissionSnapshot; + await upsertActiveMissionPg(cmd.userId, summarizeMissionSnapshot(finalSnapshot), finalSnapshot); + } + } + + await markGrowEventProcessed(cmd.eventId); + loopCtx.state.processedCount += 1; + loopCtx.state.lastEventId = cmd.eventId; + loopCtx.state.lastError = undefined; + loopCtx.state.updatedAt = new Date().toISOString(); + loopCtx.state.processedEventIds = [cmd.eventId, ...loopCtx.state.processedEventIds.filter((id) => id !== cmd.eventId)].slice(0, 500); + loopCtx.broadcast("eventProcessed", { eventId: cmd.eventId, userId: cmd.userId }); + loopCtx.broadcast("updated", loopCtx.state); + } catch (err) { + await markGrowEventFailed(cmd.eventId, err); + loopCtx.state.lastError = err instanceof Error ? err.message : String(err); + loopCtx.state.updatedAt = new Date().toISOString(); + loopCtx.broadcast("updated", loopCtx.state); + throw err; + } + }); + }); + }), +}); diff --git a/src/actors/registry.ts b/src/actors/registry.ts index 7b28804..88c9196 100644 --- a/src/actors/registry.ts +++ b/src/actors/registry.ts @@ -5,6 +5,7 @@ import { interviewServiceActor, resumeServiceActor, roleplayServiceActor } from import { conversationActor } from "./conversation/index.js"; import { memoryActor } from "./memory/index.js"; import { growActor } from "./grow/index.js"; +import { userEventActor } from "./events/index.js"; import { careerTransitionMissionActor, interviewToOfferMissionActor, @@ -16,6 +17,7 @@ import { export const registry = setup({ use: { growActor, + userEventActor, conversationActor, memoryActor, interviewToOfferMissionActor, diff --git a/src/config.ts b/src/config.ts index 8a6283e..eb4fcb0 100644 --- a/src/config.ts +++ b/src/config.ts @@ -25,6 +25,12 @@ export const config = { serviceToken: process.env.SERVICE_TOKEN ?? "", a2aAllowedKey: process.env.A2A_ALLOWED_KEY ?? "dev-a2a-key", + // Service → backend event stream. Redis is optional; HTTP /events/ingest/service is always available. + growEventsRedisUrl: process.env.GROW_EVENTS_REDIS_URL ?? process.env.REDIS_URL ?? "", + growEventsStream: process.env.GROW_EVENTS_STREAM ?? "grow.events.raw", + growEventsConsumerGroup: process.env.GROW_EVENTS_CONSUMER_GROUP ?? "growqr-backend", + growEventsConsumerName: process.env.GROW_EVENTS_CONSUMER_NAME ?? `backend-${process.pid}`, + // LLM gateway for the unified user agent. llmProvider: process.env.LLM_PROVIDER ?? "opencode", llmApiKey: diff --git a/src/db/schema.ts b/src/db/schema.ts index b231aef..0592679 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -6,6 +6,7 @@ import { integer, boolean, jsonb, + doublePrecision, uniqueIndex, index, primaryKey, @@ -324,3 +325,138 @@ export const growActiveMissions = pgTable( export type OpencodeSessionRow = typeof opencodeSessions.$inferSelect; export type WorkflowRunRow = typeof workflowRuns.$inferSelect; + +export const growEvents = pgTable( + "grow_events", + { + id: text("id").primaryKey().default(sql`gen_random_uuid()::text`), + userId: text("user_id").references(() => users.id, { onDelete: "cascade" }), + orgId: text("org_id"), + source: text("source").notNull(), + type: text("type").notNull(), + category: text("category", { + enum: ["mission", "service", "artifact", "usage", "qscore", "entitlement", "system"], + }).notNull().default("service"), + occurredAt: timestamp("occurred_at", { withTimezone: true }).notNull(), + receivedAt: timestamp("received_at", { withTimezone: true }).defaultNow().notNull(), + mission: jsonb("mission").$type>(), + subject: jsonb("subject").$type>(), + correlation: jsonb("correlation").$type>(), + payload: jsonb("payload").$type>().notNull().default(sql`'{}'::jsonb`), + raw: jsonb("raw").$type>(), + dedupeKey: text("dedupe_key"), + processingStatus: text("processing_status", { + enum: ["pending", "processing", "processed", "failed", "unresolved"], + }).notNull().default("pending"), + processingError: text("processing_error"), + processedAt: timestamp("processed_at", { withTimezone: true }), + }, + (t) => ({ + userIdx: index("grow_events_user_idx").on(t.userId, t.occurredAt), + statusIdx: index("grow_events_status_idx").on(t.processingStatus, t.receivedAt), + sourceIdx: index("grow_events_source_idx").on(t.source, t.type, t.occurredAt), + dedupeIdx: uniqueIndex("grow_events_dedupe_idx").on(t.dedupeKey), + }), +); + +export const missionServiceSessions = pgTable( + "mission_service_sessions", + { + id: text("id").primaryKey().default(sql`gen_random_uuid()::text`), + userId: text("user_id").notNull().references(() => users.id, { onDelete: "cascade" }), + missionInstanceId: text("mission_instance_id").references(() => growActiveMissions.instanceId, { onDelete: "set null" }), + missionId: text("mission_id"), + stageId: text("stage_id"), + serviceId: text("service_id").notNull(), + externalId: text("external_id").notNull(), + status: text("status").notNull().default("active"), + metadata: jsonb("metadata").$type>(), + lastEventId: text("last_event_id").references(() => growEvents.id, { onDelete: "set null" }), + lastCheckedAt: timestamp("last_checked_at", { withTimezone: true }), + createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(), + updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(), + }, + (t) => ({ + userIdx: index("mission_service_sessions_user_idx").on(t.userId, t.updatedAt), + externalIdx: uniqueIndex("mission_service_sessions_external_idx").on(t.serviceId, t.externalId), + missionIdx: index("mission_service_sessions_mission_idx").on(t.missionInstanceId, t.stageId), + }), +); + +export const missionArtifacts = pgTable( + "mission_artifacts", + { + id: text("id").primaryKey().default(sql`gen_random_uuid()::text`), + userId: text("user_id").notNull().references(() => users.id, { onDelete: "cascade" }), + missionInstanceId: text("mission_instance_id").references(() => growActiveMissions.instanceId, { onDelete: "cascade" }), + missionId: text("mission_id"), + stageId: text("stage_id"), + sourceEventId: text("source_event_id").references(() => growEvents.id, { onDelete: "set null" }), + serviceId: text("service_id"), + externalId: text("external_id"), + type: text("type").notNull(), + title: text("title").notNull(), + status: text("status").notNull().default("ready"), + summary: text("summary"), + contentMd: text("content_md"), + metadata: jsonb("metadata").$type>(), + createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(), + updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(), + }, + (t) => ({ + userIdx: index("mission_artifacts_user_idx").on(t.userId, t.createdAt), + missionIdx: index("mission_artifacts_mission_idx").on(t.missionInstanceId, t.createdAt), + externalIdx: index("mission_artifacts_external_idx").on(t.serviceId, t.externalId), + }), +); + +export const growQscoreSignals = pgTable( + "grow_qscore_signals", + { + id: text("id").primaryKey().default(sql`gen_random_uuid()::text`), + userId: text("user_id").notNull().references(() => users.id, { onDelete: "cascade" }), + sourceEventId: text("source_event_id").references(() => growEvents.id, { onDelete: "set null" }), + signalId: text("signal_id").notNull(), + score: doublePrecision("score").notNull(), + present: boolean("present").notNull().default(true), + source: text("source"), + raw: jsonb("raw").$type>(), + occurredAt: timestamp("occurred_at", { withTimezone: true }).notNull(), + createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(), + }, + (t) => ({ + userIdx: index("grow_qscore_signals_user_idx").on(t.userId, t.occurredAt), + signalIdx: index("grow_qscore_signals_signal_idx").on(t.signalId, t.occurredAt), + }), +); + +export const growQscoreLatest = pgTable( + "grow_qscore_latest", + { + userId: text("user_id").notNull().references(() => users.id, { onDelete: "cascade" }), + signalId: text("signal_id").notNull(), + score: doublePrecision("score").notNull(), + present: boolean("present").notNull().default(true), + source: text("source"), + sourceEventId: text("source_event_id").references(() => growEvents.id, { onDelete: "set null" }), + raw: jsonb("raw").$type>(), + occurredAt: timestamp("occurred_at", { withTimezone: true }).notNull(), + updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(), + }, + (t) => ({ + pk: primaryKey({ columns: [t.userId, t.signalId] }), + userIdx: index("grow_qscore_latest_user_idx").on(t.userId, t.updatedAt), + }), +); + +export const growQscoreProjectionState = pgTable("grow_qscore_projection_state", { + userId: text("user_id").primaryKey().references(() => users.id, { onDelete: "cascade" }), + score: integer("score").notNull().default(0), + signalCount: integer("signal_count").notNull().default(0), + dimensions: jsonb("dimensions").$type>(), + summary: text("summary"), + updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(), +}); + +export type GrowEventRow = typeof growEvents.$inferSelect; +export type NewGrowEvent = typeof growEvents.$inferInsert; diff --git a/src/events/envelope.ts b/src/events/envelope.ts new file mode 100644 index 0000000..7ddaf81 --- /dev/null +++ b/src/events/envelope.ts @@ -0,0 +1,93 @@ +export type GrowEventCategory = + | "mission" + | "service" + | "artifact" + | "usage" + | "qscore" + | "entitlement" + | "system"; + +export type GrowEventSubject = { + kind: string; + id: string; +}; + +export type GrowEventMissionRef = { + instanceId?: string; + missionId?: string; + stageId?: string; +}; + +export type GrowEventCorrelation = { + requestId?: string; + sessionId?: string; + resumeId?: string; + externalId?: string; + [key: string]: unknown; +}; + +export type GrowEventEnvelope = { + id?: string; + userId?: string; + orgId?: string; + source: string; + type: string; + category: GrowEventCategory; + occurredAt: string; + mission?: GrowEventMissionRef; + subject?: GrowEventSubject; + correlation?: GrowEventCorrelation; + payload: Record; + raw?: Record; + dedupeKey?: string; +}; + +export type StoredGrowEvent = GrowEventEnvelope & { + id: string; + userId?: string; + receivedAt?: string; + processingStatus?: "pending" | "processing" | "processed" | "failed" | "unresolved"; +}; + +export type QscoreSignal = { + signalId: string; + score: number; + present: boolean; + source?: string; + raw?: Record; +}; + +export function isGrowEventCategory(value: unknown): value is GrowEventCategory { + return ( + value === "mission" || + value === "service" || + value === "artifact" || + value === "usage" || + value === "qscore" || + value === "entitlement" || + value === "system" + ); +} + +export function asRecord(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : {}; +} + +export function getString(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} + +export function getNumber(value: unknown): number | undefined { + if (typeof value === "number" && Number.isFinite(value)) return value; + if (typeof value === "string" && value.trim()) { + const n = Number(value); + if (Number.isFinite(n)) return n; + } + return undefined; +} + +export function clampScore(score: number): number { + return Math.max(0, Math.min(100, Math.round(score * 100) / 100)); +} diff --git a/src/events/normalize.ts b/src/events/normalize.ts new file mode 100644 index 0000000..6cd0b36 --- /dev/null +++ b/src/events/normalize.ts @@ -0,0 +1,79 @@ +import { randomUUID } from "node:crypto"; +import { asRecord, getString, isGrowEventCategory, type GrowEventEnvelope } from "./envelope.js"; + +function parseDate(value: unknown): string { + const raw = getString(value); + if (!raw) return new Date().toISOString(); + const date = new Date(raw); + return Number.isFinite(date.getTime()) ? date.toISOString() : new Date().toISOString(); +} + +function normalizeSubject(value: unknown) { + const record = asRecord(value); + const kind = getString(record.kind); + const id = getString(record.id); + return kind && id ? { kind, id } : undefined; +} + +function normalizeMission(value: unknown) { + const record = asRecord(value); + const instanceId = getString(record.instanceId ?? record.instance_id ?? record.mission_instance_id); + const missionId = getString(record.missionId ?? record.mission_id); + const stageId = getString(record.stageId ?? record.stage_id); + if (!instanceId && !missionId && !stageId) return undefined; + return { instanceId, missionId, stageId }; +} + +function normalizeCorrelation(value: unknown) { + const record = asRecord(value); + const correlation: Record = { ...record }; + const requestId = getString(record.requestId ?? record.request_id); + const sessionId = getString(record.sessionId ?? record.session_id); + const resumeId = getString(record.resumeId ?? record.resume_id); + const externalId = getString(record.externalId ?? record.external_id); + if (requestId) correlation.requestId = requestId; + if (sessionId) correlation.sessionId = sessionId; + if (resumeId) correlation.resumeId = resumeId; + if (externalId) correlation.externalId = externalId; + return Object.keys(correlation).length ? correlation : undefined; +} + +export function normalizeGrowEvent(input: unknown, overrides: { userId?: string; source?: string } = {}): GrowEventEnvelope { + const raw = asRecord(input); + const payload = asRecord(raw.payload ?? raw.data ?? raw.event_payload); + const correlation = normalizeCorrelation(raw.correlation ?? { + session_id: raw.session_id ?? payload.session_id, + resume_id: raw.resume_id ?? payload.resume_id, + external_id: raw.external_id ?? payload.external_id, + request_id: raw.request_id ?? payload.request_id, + }); + const subject = normalizeSubject(raw.subject) ?? (() => { + const kind = getString(raw.subject_kind ?? payload.subject_kind); + const id = getString(raw.subject_id ?? payload.subject_id); + return kind && id ? { kind, id } : undefined; + })(); + + const source = overrides.source ?? getString(raw.source) ?? "unknown-service"; + const type = getString(raw.type ?? raw.event_type ?? raw.action) ?? "service.event"; + const categoryRaw = raw.category ?? payload.category; + const category = isGrowEventCategory(categoryRaw) ? categoryRaw : "service"; + const userId = overrides.userId ?? getString(raw.userId ?? raw.user_id ?? payload.userId ?? payload.user_id); + const orgId = getString(raw.orgId ?? raw.org_id ?? payload.orgId ?? payload.org_id); + const dedupeKey = getString(raw.dedupeKey ?? raw.dedupe_key) ?? getString(raw.id) ?? undefined; + + return { + id: getString(raw.id) ?? randomUUID(), + userId, + orgId, + source, + type, + category, + occurredAt: parseDate(raw.occurredAt ?? raw.occurred_at ?? raw.timestamp ?? raw.created_at), + mission: normalizeMission(raw.mission), + subject, + correlation, + payload, + raw, + dedupeKey, + }; +} diff --git a/src/events/projectors/projection-agent.ts b/src/events/projectors/projection-agent.ts new file mode 100644 index 0000000..410d594 --- /dev/null +++ b/src/events/projectors/projection-agent.ts @@ -0,0 +1,104 @@ +import { config } from "../../config.js"; +import { createChatCompletion, type LlmMessage } from "../../lib/llm.js"; +import type { GrowEventRow } from "../../db/schema.js"; +import type { QscoreSignal } from "../envelope.js"; + +export type ProjectionInsight = { + summary: string; + confidence: "low" | "medium" | "high"; + recommendedActions: string[]; + missionStageHints: Array<{ + stageId: string; + status?: "locked" | "ready" | "in_progress" | "blocked" | "done"; + progressPercent?: number; + reason?: string; + }>; +}; + +function fallbackInsight(event: GrowEventRow, qscoreSignals: QscoreSignal[]): ProjectionInsight { + return { + summary: `${event.type} received from ${event.source}.`, + confidence: qscoreSignals.length ? "medium" : "low", + recommendedActions: [], + missionStageHints: [], + }; +} + +function parseJsonObject(text: string): Record | null { + const trimmed = text.trim(); + const firstBrace = trimmed.indexOf("{"); + const lastBrace = trimmed.lastIndexOf("}"); + if (firstBrace < 0 || lastBrace < firstBrace) return null; + try { + const parsed = JSON.parse(trimmed.slice(firstBrace, lastBrace + 1)); + return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? (parsed as Record) : null; + } catch { + return null; + } +} + +function normalizeInsight(value: Record | null, fallback: ProjectionInsight): ProjectionInsight { + if (!value) return fallback; + const summary = typeof value.summary === "string" && value.summary.trim() ? value.summary.trim() : fallback.summary; + const confidence = value.confidence === "high" || value.confidence === "medium" || value.confidence === "low" ? value.confidence : fallback.confidence; + const recommendedActions = Array.isArray(value.recommendedActions) + ? value.recommendedActions.filter((item): item is string => typeof item === "string") + : []; + const missionStageHints = Array.isArray(value.missionStageHints) + ? value.missionStageHints.flatMap((item) => { + if (!item || typeof item !== "object" || Array.isArray(item)) return []; + const row = item as Record; + if (typeof row.stageId !== "string") return []; + const status: ProjectionInsight["missionStageHints"][number]["status"] = row.status === "locked" || row.status === "ready" || row.status === "in_progress" || row.status === "blocked" || row.status === "done" ? row.status : undefined; + return [{ + stageId: row.stageId, + status, + progressPercent: typeof row.progressPercent === "number" ? row.progressPercent : undefined, + reason: typeof row.reason === "string" ? row.reason : undefined, + }]; + }) + : []; + return { summary, confidence, recommendedActions, missionStageHints }; +} + +export async function getProjectionInsight(input: { + event: GrowEventRow; + qscoreSignals: QscoreSignal[]; + activeMissionIds: string[]; +}): Promise { + const fallback = fallbackInsight(input.event, input.qscoreSignals); + if (!config.llmApiKey) return fallback; + + const system = `You are GrowQR's deterministic event projection assistant. You do not call tools. You read one raw product event and return only compact JSON. Your job is to help backend reducers interpret business meaning without inventing facts. If evidence is missing, use low confidence and empty hints.`; + const user = { + instruction: "Return JSON only with keys: summary, confidence, recommendedActions, missionStageHints. missionStageHints may include interview-to-offer stage ids: baseline-qscore, resume-fit-scan, interview-prep-plan, behavioral-story-bank, mock-interview, final-readiness-report. Only mark a stage done when the event clearly proves completion.", + event: { + id: input.event.id, + source: input.event.source, + type: input.event.type, + category: input.event.category, + payload: input.event.payload, + correlation: input.event.correlation, + subject: input.event.subject, + }, + qscoreSignals: input.qscoreSignals, + activeMissionIds: input.activeMissionIds, + }; + + const messages: LlmMessage[] = [ + { role: "system", content: system }, + { role: "user", content: JSON.stringify(user) }, + ]; + + try { + const completion = await createChatCompletion({ + model: process.env.PROJECTION_AGENT_MODEL ?? config.agentModel, + messages, + tools: [], + maxTokens: 700, + }); + return normalizeInsight(parseJsonObject(completion.content), fallback); + } catch { + return fallback; + } +} diff --git a/src/events/projectors/qscore-projector.ts b/src/events/projectors/qscore-projector.ts new file mode 100644 index 0000000..48c5f86 --- /dev/null +++ b/src/events/projectors/qscore-projector.ts @@ -0,0 +1,189 @@ +import { and, eq, sql } from "drizzle-orm"; +import { db } from "../../db/client.js"; +import { growQscoreLatest, growQscoreProjectionState, growQscoreSignals, type GrowEventRow } from "../../db/schema.js"; +import { asRecord, clampScore, getNumber, type QscoreSignal } from "../envelope.js"; + +function signal(signalId: string, score: number, raw?: Record, present = true): QscoreSignal { + return { signalId, score: clampScore(score), present, raw }; +} + +function nestedNumber(record: Record, keys: string[]): number | undefined { + for (const key of keys) { + const direct = getNumber(record[key]); + if (direct !== undefined) return direct; + } + return undefined; +} + +function extractResumeSignals(event: GrowEventRow): QscoreSignal[] { + const payload = event.payload ?? {}; + const analysis = asRecord(payload.analysis ?? payload.result ?? payload); + const scoreBreakdown = Array.isArray(analysis.score_breakdown) ? analysis.score_breakdown : []; + const dimensions = Array.isArray(analysis.dimensional_scores) ? analysis.dimensional_scores : []; + + const byCategory = new Map(); + for (const item of scoreBreakdown) { + const row = asRecord(item); + const category = typeof row.category === "string" ? row.category : undefined; + const score = getNumber(row.score); + if (category && score !== undefined) byCategory.set(category, score); + } + + const byDimension = new Map(); + for (const item of dimensions) { + const row = asRecord(item); + const dimension = typeof row.dimension === "string" ? row.dimension : undefined; + const score = getNumber(row.score); + if (dimension && score !== undefined) byDimension.set(dimension, score); + } + + const signals: QscoreSignal[] = []; + if (event.type.includes("uploaded") || event.type.includes("created") || event.type.includes("analysis")) { + signals.push(signal("resume.uploaded", 100, { eventId: event.id })); + } + const ats = byCategory.get("ATS Compatibility") ?? nestedNumber(analysis, ["ats_score", "ats_compatibility", "atsCompatibility"]); + if (ats !== undefined) signals.push(signal("resume.ats_compatibility", ats, { eventId: event.id })); + const keywords = byDimension.get("Keywords") ?? nestedNumber(analysis, ["keyword_score", "keyword_relevance", "keywords"]); + if (keywords !== undefined) { + signals.push(signal("resume.keyword_relevance", keywords, { eventId: event.id })); + signals.push(signal("resume.technical_keywords", keywords, { eventId: event.id })); + } + const quantification = byDimension.get("Quantification") ?? nestedNumber(analysis, ["quantification_score"]); + if (quantification !== undefined) signals.push(signal("resume.quantified_achievements", quantification, { eventId: event.id })); + const contentQuality = byCategory.get("Content Quality") ?? nestedNumber(analysis, ["content_quality", "clarity_score"]); + if (contentQuality !== undefined) signals.push(signal("resume.grammar_clarity", contentQuality, { eventId: event.id })); + const formatting = byCategory.get("Formatting") ?? nestedNumber(analysis, ["formatting", "format_score"]); + if (formatting !== undefined) signals.push(signal("resume.format_structure", formatting, { eventId: event.id })); + const impact = byCategory.get("Impact Demonstration") ?? nestedNumber(analysis, ["impact_score"]); + if (impact !== undefined) { + signals.push(signal("resume.leadership_indicators", impact, { eventId: event.id })); + signals.push(signal("resume.impact_statements", impact, { eventId: event.id })); + } + return signals; +} + +function extractInterviewSignals(event: GrowEventRow): QscoreSignal[] { + const payload = event.payload ?? {}; + const review = asRecord(payload.review ?? payload.result ?? payload); + const status = String(review.status ?? payload.status ?? ""); + if (!event.type.includes("review") && !event.type.includes("completed") && status !== "completed") return []; + + const signals: QscoreSignal[] = []; + signals.push(signal("interview.completed", 100, { eventId: event.id })); + const overall = getNumber(review.overall_score ?? review.overallScore ?? payload.overall_score); + if (overall !== undefined) signals.push(signal("interview.overall_score", overall, { eventId: event.id })); + const rubric = asRecord(review.rubric_scores ?? review.rubricScores); + const content = getNumber(rubric.content_quality ?? rubric.content ?? rubric.communication ?? review.communication_score); + if (content !== undefined) signals.push(signal("interview.response_clarity", content, { eventId: event.id })); + const roleAlignment = getNumber(rubric.role_alignment ?? rubric.technical_accuracy ?? review.role_alignment_score); + if (roleAlignment !== undefined) signals.push(signal("interview.technical_accuracy", roleAlignment, { eventId: event.id })); + const language = getNumber(rubric.language ?? review.language_score); + if (language !== undefined) signals.push(signal("interview.behavioral_quality", language, { eventId: event.id })); + const historical = asRecord(review.historical_comparison ?? review.historicalComparison); + const delta = getNumber(historical.overall_delta ?? historical.overallDelta); + if (delta !== undefined) signals.push(signal("interview.improvement_over_time", 50 + delta * 2.5, { eventId: event.id, delta })); + return signals; +} + +function extractRoleplaySignals(event: GrowEventRow): QscoreSignal[] { + const payload = event.payload ?? {}; + const review = asRecord(payload.review ?? payload.result ?? payload); + const status = String(review.status ?? payload.status ?? ""); + if (!event.type.includes("review") && !event.type.includes("completed") && status !== "completed") return []; + + const signals: QscoreSignal[] = []; + signals.push(signal("roleplay.completed", 100, { eventId: event.id })); + const rubric = asRecord(review.rubric_scores ?? review.rubricScores); + const scenario = getNumber(rubric.scenario_adherence ?? review.scenario_adherence_score); + if (scenario !== undefined) signals.push(signal("roleplay.situational_judgment", scenario, { eventId: event.id })); + const empathy = getNumber(rubric.emotional_intelligence ?? review.emotional_intelligence_score); + if (empathy !== undefined) signals.push(signal("roleplay.empathy_demonstrated", empathy, { eventId: event.id })); + const adaptability = getNumber(rubric.adaptability ?? review.adaptability_score); + if (adaptability !== undefined) signals.push(signal("roleplay.problem_resolution", adaptability, { eventId: event.id })); + const communication = getNumber(rubric.content ?? rubric.communication ?? review.communication_score); + if (communication !== undefined) signals.push(signal("roleplay.communication_effectiveness", communication, { eventId: event.id })); + const historical = asRecord(review.historical_comparison ?? review.historicalComparison); + const delta = getNumber(historical.overall_delta ?? historical.overallDelta); + if (delta !== undefined) signals.push(signal("roleplay.improvement_over_time", 50 + delta * 2.5, { eventId: event.id, delta })); + return signals; +} + +export function extractQscoreSignals(event: GrowEventRow): QscoreSignal[] { + const source = event.source.toLowerCase(); + if (source.includes("resume") || event.type.startsWith("resume.")) return extractResumeSignals(event); + if (source.includes("interview") || event.type.startsWith("interview.")) return extractInterviewSignals(event); + if (source.includes("roleplay") || event.type.startsWith("roleplay.")) return extractRoleplaySignals(event); + if (event.type === "mission.interview_to_offer.started") { + return [signal("goals.goals_set", 100, { eventId: event.id })]; + } + return []; +} + +export async function applyQscoreProjection(event: GrowEventRow) { + if (!event.userId) return { signals: [], score: undefined }; + const signals = extractQscoreSignals(event); + if (!signals.length) return { signals, score: undefined }; + + for (const item of signals) { + await db.insert(growQscoreSignals).values({ + userId: event.userId, + sourceEventId: event.id, + signalId: item.signalId, + score: item.score, + present: item.present, + source: event.source, + raw: item.raw, + occurredAt: event.occurredAt, + }); + + await db.insert(growQscoreLatest).values({ + userId: event.userId, + signalId: item.signalId, + score: item.score, + present: item.present, + source: event.source, + sourceEventId: event.id, + raw: item.raw, + occurredAt: event.occurredAt, + updatedAt: new Date(), + }).onConflictDoUpdate({ + target: [growQscoreLatest.userId, growQscoreLatest.signalId], + set: { + score: item.score, + present: item.present, + source: event.source, + sourceEventId: event.id, + raw: item.raw, + occurredAt: event.occurredAt, + updatedAt: new Date(), + }, + }); + } + + const [aggregate] = await db + .select({ score: sql`round(avg(${growQscoreLatest.score}))::int`, count: sql`count(*)::int` }) + .from(growQscoreLatest) + .where(and(eq(growQscoreLatest.userId, event.userId), eq(growQscoreLatest.present, true))); + + const score = aggregate?.score ?? 0; + const signalCount = aggregate?.count ?? 0; + await db.insert(growQscoreProjectionState).values({ + userId: event.userId, + score, + signalCount, + dimensions: { latestSignalIds: signals.map((s) => s.signalId) }, + summary: `Estimated readiness score from ${signalCount} current signal${signalCount === 1 ? "" : "s"}.`, + updatedAt: new Date(), + }).onConflictDoUpdate({ + target: growQscoreProjectionState.userId, + set: { + score, + signalCount, + dimensions: { latestSignalIds: signals.map((s) => s.signalId) }, + summary: `Estimated readiness score from ${signalCount} current signal${signalCount === 1 ? "" : "s"}.`, + updatedAt: new Date(), + }, + }); + + return { signals, score }; +} diff --git a/src/events/projectors/service-session-projector.ts b/src/events/projectors/service-session-projector.ts new file mode 100644 index 0000000..5f7556d --- /dev/null +++ b/src/events/projectors/service-session-projector.ts @@ -0,0 +1,76 @@ +import { eq } from "drizzle-orm"; +import { db } from "../../db/client.js"; +import { missionServiceSessions, type GrowEventRow } from "../../db/schema.js"; +import { asRecord, getString } from "../envelope.js"; +import { normalizeServiceId } from "../record-grow-event.js"; + +function extractExternalId(event: GrowEventRow): string | undefined { + const correlation = asRecord(event.correlation); + const payload = event.payload ?? {}; + return getString( + correlation.sessionId ?? + correlation.session_id ?? + correlation.externalId ?? + correlation.external_id ?? + payload.session_id ?? + payload.sessionId ?? + payload.id, + ); +} + +function statusFor(event: GrowEventRow): string { + const payload = event.payload ?? {}; + const explicit = getString(payload.status); + if (explicit) return explicit; + if (event.type.includes("review_completed") || event.type.includes("completed")) return "completed"; + if (event.type.includes("failed")) return "failed"; + if (event.type.includes("configured") || event.type.includes("created")) return "active"; + return "active"; +} + +export async function applyServiceSessionProjection(event: GrowEventRow) { + if (!event.userId) return null; + const externalId = extractExternalId(event); + if (!externalId) return null; + const serviceId = normalizeServiceId(event.source); + if (!["interview", "roleplay", "resume"].includes(serviceId)) return null; + + const mission = asRecord(event.mission); + const metadata = { + lastType: event.type, + subject: event.subject, + payloadStatus: event.payload?.status, + }; + + const [row] = await db + .insert(missionServiceSessions) + .values({ + userId: event.userId, + missionInstanceId: getString(mission.instanceId ?? mission.instance_id), + missionId: getString(mission.missionId ?? mission.mission_id), + stageId: getString(mission.stageId ?? mission.stage_id), + serviceId, + externalId, + status: statusFor(event), + metadata, + lastEventId: event.id, + updatedAt: new Date(), + }) + .onConflictDoUpdate({ + target: [missionServiceSessions.serviceId, missionServiceSessions.externalId], + set: { + status: statusFor(event), + metadata, + lastEventId: event.id, + lastCheckedAt: event.type.includes("review") ? new Date() : undefined, + updatedAt: new Date(), + }, + }) + .returning(); + + // Touch the row if Drizzle ever returns no row for an upsert variant. + if (!row) { + await db.update(missionServiceSessions).set({ updatedAt: new Date() }).where(eq(missionServiceSessions.externalId, externalId)); + } + return row ?? null; +} diff --git a/src/events/record-grow-event.ts b/src/events/record-grow-event.ts new file mode 100644 index 0000000..25174f5 --- /dev/null +++ b/src/events/record-grow-event.ts @@ -0,0 +1,114 @@ +import { and, eq } 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"; +import { normalizeGrowEvent } from "./normalize.js"; + +async function ensureUser(userId: string) { + await db + .insert(users) + .values({ id: userId, email: `${userId}@service.local`, displayName: userId }) + .onConflictDoNothing(); +} + +async function resolveUserId(event: GrowEventEnvelope): Promise { + if (event.userId) return event.userId; + const sessionId = getString(event.correlation?.sessionId ?? event.correlation?.session_id); + const serviceId = normalizeServiceId(event.source); + if (sessionId) { + const [row] = await db + .select({ userId: missionServiceSessions.userId }) + .from(missionServiceSessions) + .where(and(eq(missionServiceSessions.serviceId, serviceId), eq(missionServiceSessions.externalId, sessionId))) + .limit(1); + if (row?.userId) return row.userId; + } + return undefined; +} + +export function normalizeServiceId(source: string): string { + if (source.includes("interview")) return "interview"; + if (source.includes("roleplay")) return "roleplay"; + if (source.includes("resume")) return "resume"; + if (source.includes("qscore")) return "qscore"; + return source; +} + +export async function recordGrowEvent(input: unknown, overrides: { userId?: string; source?: string } = {}): Promise { + const normalized = normalizeGrowEvent(input, overrides); + const resolvedUserId = await resolveUserId(normalized); + if (resolvedUserId) await ensureUser(resolvedUserId); + + const processingStatus = resolvedUserId ? "pending" : "unresolved"; + if (normalized.dedupeKey) { + const [existing] = await db.select().from(growEvents).where(eq(growEvents.dedupeKey, normalized.dedupeKey)).limit(1); + if (existing) return existing; + } + + const values = { + id: normalized.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 | undefined, + subject: normalized.subject as Record | undefined, + correlation: normalized.correlation as Record | undefined, + payload: normalized.payload, + raw: normalized.raw ?? asRecord(input), + dedupeKey: normalized.dedupeKey, + processingStatus, + } satisfies typeof growEvents.$inferInsert; + + const [inserted] = await db + .insert(growEvents) + .values(values) + .onConflictDoUpdate({ + target: growEvents.id, + set: { + userId: values.userId, + orgId: values.orgId, + source: values.source, + type: values.type, + category: values.category, + occurredAt: values.occurredAt, + mission: values.mission, + subject: values.subject, + correlation: values.correlation, + payload: values.payload, + raw: values.raw, + processingStatus: values.processingStatus, + }, + }) + .returning(); + + if (!inserted) throw new Error("failed to record grow event"); + return inserted; +} + +export async function markGrowEventProcessing(eventId: string) { + await db + .update(growEvents) + .set({ processingStatus: "processing", processingError: null }) + .where(eq(growEvents.id, eventId)); +} + +export async function markGrowEventProcessed(eventId: string) { + await db + .update(growEvents) + .set({ processingStatus: "processed", processingError: null, processedAt: new Date() }) + .where(eq(growEvents.id, eventId)); +} + +export async function markGrowEventFailed(eventId: string, error: unknown) { + await db + .update(growEvents) + .set({ + processingStatus: "failed", + processingError: error instanceof Error ? error.message : String(error), + processedAt: new Date(), + }) + .where(eq(growEvents.id, eventId)); +} diff --git a/src/events/redis-consumer.ts b/src/events/redis-consumer.ts new file mode 100644 index 0000000..f97b94f --- /dev/null +++ b/src/events/redis-consumer.ts @@ -0,0 +1,102 @@ +import { config } from "../config.js"; +import { log } from "../log.js"; +import { recordGrowEvent } from "./record-grow-event.js"; +import { routeGrowEventToUserActor } from "./route-to-user-actor.js"; + +type RedisMessage = { id: string; message: Record }; +type RedisStreamResponse = Array<{ name: string; messages: RedisMessage[] }>; + +type RedisClientLike = { + connect: () => Promise; + on: (event: string, listener: (err: unknown) => void) => void; + xGroupCreate: (stream: string, group: string, id: string, opts?: Record) => Promise; + xReadGroup: (group: string, consumer: string, streams: { key: string; id: string }, opts?: Record) => Promise; + xAck: (stream: string, group: string, id: string) => Promise; +}; + +async function loadRedisCreateClient(): Promise<(opts: { url: string }) => RedisClientLike> { + const dynamicImport = Function("specifier", "return import(specifier)") as (specifier: string) => Promise; + const mod = (await dynamicImport("redis")) as { createClient?: (opts: { url: string }) => RedisClientLike }; + if (!mod.createClient) throw new Error("redis package did not expose createClient"); + return mod.createClient; +} + +function parseFieldValue(value: string): unknown { + if (!value) return value; + if ((value.startsWith("{") && value.endsWith("}")) || (value.startsWith("[") && value.endsWith("]"))) { + try { return JSON.parse(value); } catch { return value; } + } + return value; +} + +function fieldsToEvent(fields: Record, stream: string) { + const payload = fields.payload ? parseFieldValue(fields.payload) : parseFieldValue(fields.data ?? "{}"); + const correlation = fields.correlation ? parseFieldValue(fields.correlation) : undefined; + const subject = fields.subject ? parseFieldValue(fields.subject) : undefined; + const mission = fields.mission ? parseFieldValue(fields.mission) : undefined; + return { + id: fields.id, + source: fields.source ?? stream, + type: fields.type ?? fields.event_type ?? "service.event", + category: fields.category ?? "service", + userId: fields.userId ?? fields.user_id, + orgId: fields.orgId ?? fields.org_id, + occurredAt: fields.occurredAt ?? fields.occurred_at ?? fields.timestamp, + mission, + subject, + correlation, + payload: payload && typeof payload === "object" && !Array.isArray(payload) ? payload : { value: payload }, + raw: fields, + dedupeKey: fields.dedupeKey ?? fields.dedupe_key, + }; +} + +export async function startGrowEventsRedisConsumer() { + if (!config.growEventsRedisUrl) { + log.info("grow events Redis consumer disabled (GROW_EVENTS_REDIS_URL/REDIS_URL not set)"); + return; + } + + const createClient = await loadRedisCreateClient(); + const redis = createClient({ url: config.growEventsRedisUrl }); + redis.on("error", (err) => log.warn({ err }, "grow events redis client error")); + await redis.connect(); + + try { + await redis.xGroupCreate(config.growEventsStream, config.growEventsConsumerGroup, "0", { MKSTREAM: true }); + log.info({ stream: config.growEventsStream, group: config.growEventsConsumerGroup }, "created grow events consumer group"); + } catch (err) { + if (!String(err).includes("BUSYGROUP")) throw err; + } + + log.info({ stream: config.growEventsStream, group: config.growEventsConsumerGroup }, "grow events redis consumer started"); + + void (async () => { + while (true) { + try { + const response = await redis.xReadGroup( + config.growEventsConsumerGroup, + config.growEventsConsumerName, + { key: config.growEventsStream, id: ">" }, + { COUNT: 50, BLOCK: 5000 }, + ); + if (!response) continue; + for (const stream of response) { + for (const message of stream.messages) { + try { + const event = await recordGrowEvent(fieldsToEvent(message.message, stream.name)); + await routeGrowEventToUserActor(event); + } catch (err) { + log.error({ err, messageId: message.id }, "failed to ingest grow event stream message"); + } finally { + await redis.xAck(stream.name, config.growEventsConsumerGroup, message.id); + } + } + } + } catch (err) { + log.error({ err }, "grow events redis consumer loop error"); + await new Promise((resolve) => setTimeout(resolve, 2000)); + } + } + })(); +} diff --git a/src/events/route-to-user-actor.ts b/src/events/route-to-user-actor.ts new file mode 100644 index 0000000..7c54264 --- /dev/null +++ b/src/events/route-to-user-actor.ts @@ -0,0 +1,15 @@ +import { createClient, type Client } from "rivetkit/client"; +import { config } from "../config.js"; +import type { Registry } from "../actors/registry.js"; +import type { GrowEventRow } from "../db/schema.js"; + +let _client: Client | null = null; +function getClient(): Client { + return (_client ??= createClient(config.rivetClientEndpoint)); +} + +export async function routeGrowEventToUserActor(event: Pick) { + if (!event.userId) return { routed: false, reason: "unresolved_user" } as const; + await getClient().userEventActor.getOrCreate([event.userId]).enqueueEvent({ userId: event.userId, eventId: event.id }); + return { routed: true } as const; +} diff --git a/src/index.ts b/src/index.ts index 2a2418b..6953061 100644 --- a/src/index.ts +++ b/src/index.ts @@ -16,6 +16,8 @@ import { serviceRoutes } from "./routes/services.js"; import { conversationRoutes } from "./routes/conversations.js"; import { growRoutes } from "./routes/grow.js"; import { missionRoutes } from "./routes/missions.js"; +import { eventRoutes } from "./routes/events.js"; +import { startGrowEventsRedisConsumer } from "./events/redis-consumer.js"; import { db } from "./db/client.js"; import { hydratePortAllocator, reconcileOnBoot, ensureCentralGiteaReady } from "./docker/manager.js"; import { initCatalog } from "./agents/catalog.js"; @@ -37,6 +39,7 @@ async function main() { await initCatalog(); await reconcileOnBoot(); + startGrowEventsRedisConsumer().catch((err) => log.error({ err }, "failed to start grow events redis consumer")); const app = new Hono(); @@ -81,6 +84,7 @@ async function main() { app.route("/actors", actorRoutes()); app.route("/grow", growRoutes()); app.route("/missions", missionRoutes()); + app.route("/events", eventRoutes()); app.route("/conversations", conversationRoutes()); app.route("/opencode", opencodeRoutes()); app.route("/git", gitRoutes()); diff --git a/src/missions/event-reducers.ts b/src/missions/event-reducers.ts new file mode 100644 index 0000000..74becfe --- /dev/null +++ b/src/missions/event-reducers.ts @@ -0,0 +1,8 @@ +import type { MissionReducer } from "./reducer-types.js"; +import { interviewToOfferReducer } from "./interview-to-offer/reducer.js"; + +export const missionEventReducers: MissionReducer[] = [interviewToOfferReducer]; + +export function reducersForMission(missionId: string) { + return missionEventReducers.filter((reducer) => reducer.missionId === missionId); +} diff --git a/src/missions/interview-to-offer/reducer.ts b/src/missions/interview-to-offer/reducer.ts new file mode 100644 index 0000000..600ff0e --- /dev/null +++ b/src/missions/interview-to-offer/reducer.ts @@ -0,0 +1,99 @@ +import { asRecord, getNumber, getString } from "../../events/envelope.js"; +import type { MissionArtifactPatch, MissionReducer, MissionReduction, MissionStagePatch } from "../reducer-types.js"; + +function eventMatchesSource(source: string, type: string) { + return ( + source.includes("resume") || + source.includes("interview") || + source.includes("qscore") || + type.startsWith("resume.") || + type.startsWith("interview.") || + type.startsWith("qscore.") || + type.startsWith("mission.interview_to_offer") + ); +} + +function reviewSummary(payload: Record) { + const review = asRecord(payload.review ?? payload.result ?? payload); + const overall = getNumber(review.overall_score ?? review.overallScore); + const summary = getString(review.summary ?? review.feedback_summary ?? review.overall_feedback); + if (overall !== undefined && summary) return `Mock interview review completed with score ${overall}. ${summary}`; + if (overall !== undefined) return `Mock interview review completed with score ${overall}.`; + return summary ?? "Mock interview review completed."; +} + +export const interviewToOfferReducer: MissionReducer = { + missionId: "interview-to-offer", + + accepts(ctx) { + if (ctx.activeMission.missionId !== "interview-to-offer") return false; + const mission = asRecord(ctx.event.mission); + const explicitMissionId = getString(mission.missionId ?? mission.mission_id); + return explicitMissionId === "interview-to-offer" || eventMatchesSource(ctx.event.source.toLowerCase(), ctx.event.type); + }, + + reduce(ctx): MissionReduction { + const type = ctx.event.type; + const payload = ctx.event.payload ?? {}; + const stagePatches: MissionStagePatch[] = [...ctx.insight.missionStageHints.map((hint) => ({ + stageId: hint.stageId, + status: hint.status, + progressPercent: hint.progressPercent, + outputSummary: hint.reason, + }))]; + const artifacts: MissionArtifactPatch[] = []; + let eventMessage = ctx.insight.summary; + + if (type.startsWith("resume.") && (type.includes("analysis_completed") || type.includes("analysis.complete") || type.includes("analyzed"))) { + stagePatches.push({ stageId: "resume", status: "done", progressPercent: 100, outputSummary: "Resume fit scan completed." }); + stagePatches.push({ stageId: "interview-plan", status: "ready", progressPercent: 0 }); + artifacts.push({ + type: "resume_fit_scan", + title: "Resume Fit Scan", + stageId: "resume", + summary: getString(payload.summary) ?? "Resume analysis completed and readiness signals were updated.", + metadata: { sourceEventId: ctx.event.id, payload }, + }); + eventMessage = "Resume fit scan completed for Interview-to-Offer."; + } + + if (type.startsWith("interview.") && (type.includes("configured") || type.includes("created"))) { + stagePatches.push({ stageId: "interview", status: "in_progress", progressPercent: 35, outputSummary: "Mock interview session configured." }); + eventMessage = "Mock interview session configured."; + } + + if (type.startsWith("interview.") && (type.includes("session_completed") || type.includes("session.completed"))) { + stagePatches.push({ stageId: "interview", status: "in_progress", progressPercent: 75, outputSummary: "Interview completed; review is being prepared." }); + eventMessage = "Mock interview completed; waiting for review."; + } + + if (type.startsWith("interview.") && (type.includes("review_completed") || type.includes("review.completed"))) { + stagePatches.push({ stageId: "interview", status: "done", progressPercent: 100, outputSummary: reviewSummary(payload) }); + stagePatches.push({ stageId: "qscore", status: "in_progress", progressPercent: 60, outputSummary: "Readiness signals updated from interview review." }); + artifacts.push({ + type: "mock_interview_review", + title: "Mock Interview Review", + stageId: "interview", + summary: reviewSummary(payload), + metadata: { sourceEventId: ctx.event.id, payload }, + }); + eventMessage = "Mock interview review completed and mission readiness was updated."; + } + + if (ctx.qscoreSignals.length > 0) { + stagePatches.push({ + stageId: "qscore", + status: "in_progress", + progressPercent: Math.max(40, Math.min(90, ctx.qscoreSignals.length * 15)), + outputSummary: `${ctx.qscoreSignals.length} readiness signal${ctx.qscoreSignals.length === 1 ? "" : "s"} updated.`, + }); + } + + if (type.startsWith("qscore.") && (type.includes("snapshot") || type.includes("computed") || type.includes("updated"))) { + stagePatches.push({ stageId: "qscore", status: "done", progressPercent: 100, outputSummary: "Readiness Q Score updated." }); + eventMessage = "Readiness Q Score updated."; + } + + return { stagePatches, artifacts, eventMessage }; + }, +}; diff --git a/src/missions/reducer-types.ts b/src/missions/reducer-types.ts new file mode 100644 index 0000000..2f74ac5 --- /dev/null +++ b/src/missions/reducer-types.ts @@ -0,0 +1,40 @@ +import type { GrowEventRow } from "../db/schema.js"; +import type { ProjectionInsight } from "../events/projectors/projection-agent.js"; +import type { QscoreSignal } from "../events/envelope.js"; +import type { GrowActiveMission, MissionStageStatus } from "../actors/missions/types.js"; + +export type MissionReducerContext = { + userId: string; + activeMission: GrowActiveMission; + event: GrowEventRow; + qscoreSignals: QscoreSignal[]; + insight: ProjectionInsight; +}; + +export type MissionArtifactPatch = { + type: string; + title: string; + stageId?: string; + summary?: string; + contentMd?: string; + metadata?: Record; +}; + +export type MissionStagePatch = { + stageId: string; + status?: MissionStageStatus; + progressPercent?: number; + outputSummary?: string; +}; + +export type MissionReduction = { + stagePatches: MissionStagePatch[]; + artifacts: MissionArtifactPatch[]; + eventMessage?: string; +}; + +export type MissionReducer = { + missionId: string; + accepts(ctx: MissionReducerContext): boolean; + reduce(ctx: MissionReducerContext): MissionReduction; +}; diff --git a/src/routes/events.ts b/src/routes/events.ts new file mode 100644 index 0000000..36fbde8 --- /dev/null +++ b/src/routes/events.ts @@ -0,0 +1,61 @@ +import { Hono } from "hono"; +import { desc, eq } from "drizzle-orm"; +import { config } from "../config.js"; +import { db } from "../db/client.js"; +import { growEvents } from "../db/schema.js"; +import { requireUser, type AuthContext } from "../auth/clerk.js"; +import { recordGrowEvent } from "../events/record-grow-event.js"; +import { routeGrowEventToUserActor } from "../events/route-to-user-actor.js"; + +function serviceAuthorized(auth: string | undefined) { + const token = (auth ?? "").replace(/^Bearer\s+/i, "").trim(); + if (config.serviceToken && token === config.serviceToken) return true; + if (config.a2aAllowedKey && token === config.a2aAllowedKey) return true; + return config.nodeEnv !== "production" && !config.serviceToken; +} + +async function ingest(body: unknown, userId?: string, source?: string) { + const event = await recordGrowEvent(body, { userId, source }); + const route = await routeGrowEventToUserActor(event).catch((err) => ({ + routed: false as const, + reason: err instanceof Error ? err.message : String(err), + })); + return { event, route }; +} + +export function eventRoutes() { + const app = new Hono(); + + // Browser/dashboard fallback. Clerk-scoped userId wins over body userId. + app.post("/ingest", requireUser, async (c) => { + const userId = c.get("userId"); + const body = await c.req.json().catch(() => ({})); + const { event, route } = await ingest(body, userId); + return c.json({ eventId: event.id, processingStatus: event.processingStatus, route }, 202); + }); + + // Service-to-service ingress. Services may include userId directly, or we resolve it from session correlation. + app.post("/ingest/service", async (c) => { + if (!serviceAuthorized(c.req.header("authorization"))) { + return c.json({ error: "unauthorized_service_event_ingest" }, 401); + } + const body = await c.req.json().catch(() => ({})); + const source = c.req.header("x-growqr-source") ?? undefined; + const { event, route } = await ingest(body, undefined, source); + return c.json({ eventId: event.id, processingStatus: event.processingStatus, route }, 202); + }); + + app.get("/", requireUser, async (c) => { + const userId = c.get("userId"); + const limit = Math.min(100, Math.max(1, Number(c.req.query("limit") ?? 50))); + const rows = await db + .select() + .from(growEvents) + .where(eq(growEvents.userId, userId)) + .orderBy(desc(growEvents.occurredAt)) + .limit(limit); + return c.json({ events: rows }); + }); + + return app; +} diff --git a/src/routes/services.ts b/src/routes/services.ts index 23f5878..4991099 100644 --- a/src/routes/services.ts +++ b/src/routes/services.ts @@ -4,6 +4,9 @@ import { listServiceCapabilities } from "../workflows/service-capabilities.js"; import { interviewService, resumeService, roleplayService, type JsonObject } from "../services/product-service-clients.js"; import { db } from "../db/client.js"; import { events } from "../db/schema.js"; +import { recordGrowEvent } from "../events/record-grow-event.js"; +import { routeGrowEventToUserActor } from "../events/route-to-user-actor.js"; +import { log } from "../log.js"; const DEFAULT_QSCORE = { q_score: 76, @@ -18,8 +21,53 @@ const DEFAULT_QSCORE = { }, }; -async function recordServiceEvent(userId: string, type: string, payload: Record) { - await db.insert(events).values({ userId, actorId: "service-gateway", type, payload }); +function getString(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} + +function getSessionId(result: Record, fallback?: string): string | undefined { + return getString(result.session_id ?? result.sessionId ?? result.id) ?? fallback; +} + +function missionFromBody(body: JsonObject): Record | undefined { + const mission = body.mission; + return mission && typeof mission === "object" && !Array.isArray(mission) ? (mission as Record) : undefined; +} + +async function recordGatewayEvent(input: { + userId: string; + source: string; + type: string; + payload: Record; + correlation?: Record; + mission?: Record; +}) { + 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({ + source: input.source, + type: input.type, + category: "service", + userId: input.userId, + occurredAt: new Date().toISOString(), + mission: input.mission, + correlation: input.correlation, + payload: input.payload, + }); + await routeGrowEventToUserActor(event); + return event; +} + +function eventTypeForReview(prefix: "interview" | "roleplay", result: Record) { + const status = getString(result.status); + if (status === "completed") return `${prefix}.review_completed`; + if (status === "failed") return `${prefix}.review_failed`; + return `${prefix}.review_processing`; } export function serviceRoutes() { @@ -38,14 +86,23 @@ export function serviceRoutes() { app.get("/interview/page-state", async (c) => c.json(await interviewService.pageState(c.get("userId")))); app.post("/interview/configure", async (c) => { + const userId = c.get("userId"); const body = await c.req.json(); const payload = { ...body, - user_id: String(body.user_id ?? c.get("userId")), + user_id: String(body.user_id ?? userId), org_id: String(body.org_id ?? "growqr"), } satisfies JsonObject; const result = await interviewService.configure(payload); - await recordServiceEvent(c.get("userId"), "interview.configured", { request: payload, result }); + const resultObj = result as Record; + await recordGatewayEvent({ + userId, + source: "interview-service", + type: "interview.configured", + payload: { request: payload, result: resultObj }, + correlation: { sessionId: getSessionId(resultObj), requestId: getString(body.request_id) }, + mission: missionFromBody(body), + }).catch((err) => log.warn({ err }, "failed to record interview configured event")); return c.json(result); }); app.post("/interview/preview", async (c) => c.json(await interviewService.preview(await c.req.json()))); @@ -59,9 +116,17 @@ export function serviceRoutes() { app.post("/interview/assignments/unassign", async (c) => c.json(await interviewService.unassign(await c.req.json()))); app.post("/interview/results:bulk", async (c) => c.json(await interviewService.resultsBulk(await c.req.json()))); app.get("/interview/review/:sessionId", async (c) => { + const userId = c.get("userId"); const sessionId = c.req.param("sessionId"); const result = await interviewService.review(sessionId); - await recordServiceEvent(c.get("userId"), "interview.review_checked", { sessionId, result }); + const resultObj = result as Record; + await recordGatewayEvent({ + userId, + source: "interview-service", + type: eventTypeForReview("interview", resultObj), + payload: resultObj, + correlation: { sessionId }, + }).catch((err) => log.warn({ err }, "failed to record interview review event")); return c.json(result); }); app.get("/interview/leaderboard", async (c) => c.json(await interviewService.leaderboard())); @@ -71,15 +136,24 @@ export function serviceRoutes() { app.get("/roleplay/page-state", async (c) => c.json(await roleplayService.pageState(c.get("userId")))); app.post("/roleplay/configure", async (c) => { + const userId = c.get("userId"); const body = await c.req.json(); const payload = { ...body, - user_id: String(body.user_id ?? c.get("userId")), + user_id: String(body.user_id ?? userId), org_id: String(body.org_id ?? "growqr"), qscore: (body.qscore as JsonObject | undefined) ?? DEFAULT_QSCORE, } satisfies JsonObject; const result = await roleplayService.configure(payload); - await recordServiceEvent(c.get("userId"), "roleplay.configured", { request: payload, result }); + const resultObj = result as Record; + await recordGatewayEvent({ + userId, + source: "roleplay-service", + type: "roleplay.configured", + payload: { request: payload, result: resultObj }, + correlation: { sessionId: getSessionId(resultObj), requestId: getString(body.request_id) }, + mission: missionFromBody(body), + }).catch((err) => log.warn({ err }, "failed to record roleplay configured event")); return c.json(result); }); app.post("/roleplay/preview", async (c) => c.json(await roleplayService.preview(await c.req.json()))); @@ -93,9 +167,17 @@ export function serviceRoutes() { app.post("/roleplay/assignments/unassign", async (c) => c.json(await roleplayService.unassign(await c.req.json()))); app.post("/roleplay/results:bulk", async (c) => c.json(await roleplayService.resultsBulk(await c.req.json()))); app.get("/roleplay/review/:sessionId", async (c) => { + const userId = c.get("userId"); const sessionId = c.req.param("sessionId"); const result = await roleplayService.review(sessionId); - await recordServiceEvent(c.get("userId"), "roleplay.review_checked", { sessionId, result }); + const resultObj = result as Record; + await recordGatewayEvent({ + userId, + source: "roleplay-service", + type: eventTypeForReview("roleplay", resultObj), + payload: resultObj, + correlation: { sessionId }, + }).catch((err) => log.warn({ err }, "failed to record roleplay review event")); return c.json(result); }); app.get("/roleplay/leaderboard", async (c) => c.json(await roleplayService.leaderboard())); @@ -114,10 +196,18 @@ export function serviceRoutes() { app.get("/resume/resumes/:resumeId", async (c) => c.json(await resumeService.getResume(c.req.param("resumeId")))); app.put("/resume/resumes/:resumeId", async (c) => c.json(await resumeService.updateResume(c.req.param("resumeId"), await c.req.json()))); app.post("/resume/resumes/:resumeId/analyze", async (c) => { + const userId = c.get("userId"); const resumeId = c.req.param("resumeId"); - const payload = await c.req.json().catch(() => ({})); + const payload = (await c.req.json().catch(() => ({}))) as JsonObject; const result = await resumeService.analyzeResume(resumeId, payload); - await recordServiceEvent(c.get("userId"), "resume.analysis_completed", { resumeId, request: payload, result }); + await recordGatewayEvent({ + userId, + source: "resume-service", + type: "resume.analysis_completed", + payload: { resumeId, request: payload, result: result as Record }, + correlation: { resumeId }, + mission: missionFromBody(payload), + }).catch((err) => log.warn({ err }, "failed to record resume analysis event")); return c.json(result); }); app.get("/resume/resumes/:resumeId/suggestions", async (c) => c.json(await resumeService.suggestions(c.req.param("resumeId"))));