feat(vibe): persist durable session plans and events
This commit is contained in:
kami
@@ -10,6 +10,18 @@ import { SearchService } from './search.service.js';
|
||||
/** Anything with a `.query()` — either the shared Pool or a checked-out client. */
|
||||
type Queryable = Pool | PoolClient;
|
||||
|
||||
type VibePlanVersionRow = Omit<VibePlan, 'items'> & {
|
||||
item_plan_version_id: string | null;
|
||||
ordinal: number | null;
|
||||
track_id: string | null;
|
||||
slot_role: string | null;
|
||||
candidate_source: string | null;
|
||||
score: number | null;
|
||||
score_breakdown: Record<string, unknown> | null;
|
||||
explanation: unknown | null;
|
||||
committed: boolean | null;
|
||||
};
|
||||
|
||||
import { MIGRATIONS } from '../db/migrations.js';
|
||||
import { allowedFields } from '../db/updatable-columns.js';
|
||||
|
||||
@@ -33,6 +45,12 @@ import type {
|
||||
SessionState,
|
||||
DiversityBudget,
|
||||
RepetitionRule,
|
||||
VibeSession,
|
||||
VibeSessionStatus,
|
||||
VibeEvent,
|
||||
RecordedVibeEvent,
|
||||
VibePlan,
|
||||
VibePlanItem,
|
||||
} from '../db/types.js';
|
||||
import { AUDIO_PREFERENCE_BUCKETS, FEEDBACK_ACTIONS } from '../db/types.js';
|
||||
export * from '../db/types.js';
|
||||
@@ -1479,6 +1497,254 @@ export class DbService {
|
||||
return (res.rows[0] as SessionState) || null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the authoritative Vibe v2 session record. This intentionally does
|
||||
* not create a legacy session_state row: callers can migrate to the durable
|
||||
* ledger without changing the existing v2 endpoint contract first.
|
||||
*/
|
||||
async createVibeSession(params: {
|
||||
userId: string;
|
||||
policyVersion: string;
|
||||
seedTrackId?: string | null;
|
||||
context?: Record<string, unknown>;
|
||||
}): Promise<VibeSession> {
|
||||
const res = await this.pgClient.query(
|
||||
`INSERT INTO vibe_sessions (user_id, status, seed_track_id, context, policy_version)
|
||||
VALUES ($1, 'active', $2, $3::jsonb, $4)
|
||||
RETURNING *`,
|
||||
[
|
||||
params.userId,
|
||||
params.seedTrackId ?? null,
|
||||
JSON.stringify(params.context ?? {}),
|
||||
params.policyVersion,
|
||||
]
|
||||
);
|
||||
return res.rows[0] as VibeSession;
|
||||
}
|
||||
|
||||
/** Fetch a Vibe session only when it belongs to the requesting user. */
|
||||
async getVibeSession(sessionId: string, userId: string): Promise<VibeSession | null> {
|
||||
const res = await this.pgClient.query(
|
||||
'SELECT * FROM vibe_sessions WHERE id = $1 AND user_id = $2',
|
||||
[sessionId, userId]
|
||||
);
|
||||
return (res.rows[0] as VibeSession) ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* End (or expire/replace) a session without changing its original end time
|
||||
* when a client retries the same request.
|
||||
*/
|
||||
async endVibeSession(
|
||||
sessionId: string,
|
||||
userId: string,
|
||||
status: Extract<VibeSessionStatus, 'ended' | 'expired' | 'replaced'> = 'ended'
|
||||
): Promise<VibeSession | null> {
|
||||
const res = await this.pgClient.query(
|
||||
`UPDATE vibe_sessions
|
||||
SET status = CASE WHEN ended_at IS NULL THEN $3 ELSE status END,
|
||||
ended_at = COALESCE(ended_at, NOW()),
|
||||
last_event_at = CASE WHEN ended_at IS NULL THEN NOW() ELSE last_event_at END
|
||||
WHERE id = $1 AND user_id = $2
|
||||
RETURNING *`,
|
||||
[sessionId, userId, status]
|
||||
);
|
||||
return (res.rows[0] as VibeSession) ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Append an immutable Vibe event. A supplied clientEventId is idempotent per
|
||||
* session: a retry returns the original event and does not advance the
|
||||
* session timestamp a second time. A missing id deliberately means a new,
|
||||
* server-originated event.
|
||||
*/
|
||||
async recordVibeEvent(params: {
|
||||
sessionId: string;
|
||||
userId: string;
|
||||
type: string;
|
||||
clientEventId?: string | null;
|
||||
trackId?: string | null;
|
||||
occurredAt?: Date;
|
||||
positionMs?: number | null;
|
||||
durationMs?: number | null;
|
||||
payload?: Record<string, unknown>;
|
||||
}): Promise<RecordedVibeEvent> {
|
||||
return this.withTransaction(async (client) => {
|
||||
// A session-row lock serializes both event writes and terminal state
|
||||
// transitions. In particular, it avoids the READ COMMITTED CTE snapshot
|
||||
// race where ON CONFLICT observes a concurrent event but a later CTE
|
||||
// cannot yet read it. The duplicate lookup happens after the lock, so an
|
||||
// idempotent retry remains valid even after the session has ended.
|
||||
const sessionRes = await client.query(
|
||||
`SELECT id, status
|
||||
FROM vibe_sessions
|
||||
WHERE id = $1 AND user_id = $2
|
||||
FOR UPDATE`,
|
||||
[params.sessionId, params.userId]
|
||||
);
|
||||
const session = sessionRes.rows[0] as Pick<VibeSession, 'id' | 'status'> | undefined;
|
||||
if (!session) {
|
||||
throw new Error('Vibe session was not found or is not owned by this user');
|
||||
}
|
||||
|
||||
if (params.clientEventId) {
|
||||
const existingRes = await client.query(
|
||||
`SELECT *
|
||||
FROM vibe_events
|
||||
WHERE session_id = $1 AND client_event_id = $2::uuid`,
|
||||
[params.sessionId, params.clientEventId]
|
||||
);
|
||||
const existing = existingRes.rows[0] as VibeEvent | undefined;
|
||||
if (existing) {
|
||||
return { event: existing, inserted: false };
|
||||
}
|
||||
}
|
||||
|
||||
if (session.status === 'ended' || session.status === 'expired' || session.status === 'replaced') {
|
||||
throw new Error(`Cannot record a new event for ${session.status} Vibe session`);
|
||||
}
|
||||
|
||||
const occurredAt = params.occurredAt?.toISOString() ?? null;
|
||||
const insertRes = await client.query(
|
||||
`INSERT INTO vibe_events
|
||||
(client_event_id, session_id, user_id, track_id, type, occurred_at, position_ms, duration_ms, payload)
|
||||
VALUES ($1::uuid, $2, $3, $4::uuid, $5, COALESCE($6::timestamptz, NOW()), $7, $8, $9::jsonb)
|
||||
RETURNING *`,
|
||||
[
|
||||
params.clientEventId ?? null,
|
||||
params.sessionId,
|
||||
params.userId,
|
||||
params.trackId ?? null,
|
||||
params.type,
|
||||
occurredAt,
|
||||
params.positionMs ?? null,
|
||||
params.durationMs ?? null,
|
||||
JSON.stringify(params.payload ?? {}),
|
||||
]
|
||||
);
|
||||
const event = insertRes.rows[0] as VibeEvent | undefined;
|
||||
if (!event) {
|
||||
throw new Error('Vibe event could not be recorded');
|
||||
}
|
||||
|
||||
await client.query(
|
||||
`UPDATE vibe_sessions
|
||||
SET last_event_at = GREATEST(last_event_at, $2::timestamptz)
|
||||
WHERE id = $1`,
|
||||
[params.sessionId, event.occurred_at]
|
||||
);
|
||||
return { event, inserted: true };
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist one complete revision of a session plan atomically. The caller
|
||||
* supplies the monotonically increasing version; session-level scheduling
|
||||
* will own version allocation when the director is migrated to this ledger.
|
||||
*/
|
||||
async persistVibePlan(params: {
|
||||
sessionId: string;
|
||||
userId: string;
|
||||
version: number;
|
||||
reason: string;
|
||||
stateSnapshot: Record<string, unknown>;
|
||||
objectiveSnapshot: Record<string, unknown>;
|
||||
items: Array<Omit<VibePlanItem, 'plan_version_id'>>;
|
||||
}): Promise<VibePlan> {
|
||||
return this.withTransaction(async (client) => {
|
||||
const header = await client.query(
|
||||
`INSERT INTO vibe_plan_versions
|
||||
(session_id, version, reason, state_snapshot, objective_snapshot)
|
||||
SELECT s.id, $3, $4, $5::jsonb, $6::jsonb
|
||||
FROM vibe_sessions s
|
||||
WHERE s.id = $1 AND s.user_id = $2
|
||||
RETURNING *`,
|
||||
[
|
||||
params.sessionId,
|
||||
params.userId,
|
||||
params.version,
|
||||
params.reason,
|
||||
JSON.stringify(params.stateSnapshot),
|
||||
JSON.stringify(params.objectiveSnapshot),
|
||||
]
|
||||
);
|
||||
const planVersion = header.rows[0] as VibePlan | undefined;
|
||||
if (!planVersion) {
|
||||
throw new Error('Vibe session was not found or is not owned by this user');
|
||||
}
|
||||
|
||||
for (const item of params.items) {
|
||||
await client.query(
|
||||
`INSERT INTO vibe_plan_items
|
||||
(plan_version_id, ordinal, track_id, slot_role, candidate_source, score, score_breakdown, explanation, committed)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8::jsonb, $9)`,
|
||||
[
|
||||
planVersion.id,
|
||||
item.ordinal,
|
||||
item.track_id,
|
||||
item.slot_role,
|
||||
item.candidate_source,
|
||||
item.score,
|
||||
JSON.stringify(item.score_breakdown),
|
||||
JSON.stringify(item.explanation),
|
||||
item.committed,
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
return { ...planVersion, items: params.items.map((item) => ({ ...item, plan_version_id: planVersion.id })) };
|
||||
});
|
||||
}
|
||||
|
||||
/** Read a specific plan revision, or the latest revision for a session. */
|
||||
async getVibePlan(sessionId: string, userId: string, version?: number): Promise<VibePlan | null> {
|
||||
const res = await this.pgClient.query(
|
||||
`SELECT p.*, i.plan_version_id AS item_plan_version_id, i.ordinal, i.track_id,
|
||||
i.slot_role, i.candidate_source, i.score, i.score_breakdown,
|
||||
i.explanation, i.committed
|
||||
FROM vibe_plan_versions p
|
||||
JOIN vibe_sessions s ON s.id = p.session_id
|
||||
LEFT JOIN vibe_plan_items i ON i.plan_version_id = p.id
|
||||
WHERE p.session_id = $1 AND s.user_id = $2
|
||||
AND (
|
||||
($3::integer IS NOT NULL AND p.version = $3)
|
||||
OR ($3::integer IS NULL AND p.version = (
|
||||
SELECT MAX(version) FROM vibe_plan_versions WHERE session_id = $1
|
||||
))
|
||||
)
|
||||
ORDER BY i.ordinal ASC`,
|
||||
[sessionId, userId, version ?? null]
|
||||
);
|
||||
if (!res.rows[0]) return null;
|
||||
|
||||
const first = res.rows[0] as VibePlanVersionRow;
|
||||
const plan: VibePlan = {
|
||||
id: first.id,
|
||||
session_id: first.session_id,
|
||||
version: first.version,
|
||||
reason: first.reason,
|
||||
state_snapshot: first.state_snapshot,
|
||||
objective_snapshot: first.objective_snapshot,
|
||||
created_at: first.created_at,
|
||||
items: [],
|
||||
};
|
||||
for (const row of res.rows as VibePlanVersionRow[]) {
|
||||
if (!row.item_plan_version_id) continue;
|
||||
plan.items.push({
|
||||
plan_version_id: row.item_plan_version_id,
|
||||
ordinal: row.ordinal!,
|
||||
track_id: row.track_id!,
|
||||
slot_role: row.slot_role,
|
||||
candidate_source: row.candidate_source!,
|
||||
score: row.score!,
|
||||
score_breakdown: row.score_breakdown!,
|
||||
explanation: row.explanation,
|
||||
committed: row.committed!,
|
||||
});
|
||||
}
|
||||
return plan;
|
||||
}
|
||||
|
||||
/**
|
||||
* Upsert a diversity budget for a user.
|
||||
*/
|
||||
|
||||
Reference in New Issue
Block a user