From c41316ee995d306ef9ef05584de0918c9f1e21df Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 17 Jul 2026 13:22:06 +0400 Subject: [PATCH] fix vibe engine audit findings: pg.Pool, plan replan, dead exclusions, legacy engine removal Backend: - app.ts: switch shared pg.Client to pg.Pool with per-transaction clients (#205) - v2.routes.ts: replace plan instead of appending on replan, fixing self-duplication (#206) - session-director: populate recentExclusions, per-candidate ranking, batch repetition checks (#209/#211/#213/#215 + minor) - db.service.ts: claim-fusion watermark, legacy recommendation_batch engine removed (#216/#219/#232) - app.ts: drop test enqueue-job endpoint (#234) Frontend: - AudioEngine/Vibe/usePlaybackStore: dedupe completed feedback, gate feedback to vibe sessions, End Vibe stops playback, Keep toast, shuffle played-set (#207/#236/#237/#238/#239/#240) Co-Authored-By: Claude Fable 5 --- backend/src/app.ts | 43 +- backend/src/routes/v2.routes.ts | 4 +- backend/src/routes/vibe.routes.ts | 76 -- backend/src/services/db.service.ts | 829 ++++-------------- backend/src/services/generators.service.ts | 151 ++-- .../src/services/session-director.service.ts | 177 +++- backend/src/services/session-director.test.ts | 3 +- frontend/src/components/AudioEngine.tsx | 48 +- frontend/src/pages/Vibe.tsx | 27 +- frontend/src/store/usePlaybackStore.ts | 29 +- 10 files changed, 465 insertions(+), 922 deletions(-) delete mode 100644 backend/src/routes/vibe.routes.ts diff --git a/backend/src/app.ts b/backend/src/app.ts index fcf19f5..fd97d79 100644 --- a/backend/src/app.ts +++ b/backend/src/app.ts @@ -1,6 +1,6 @@ import Fastify from 'fastify'; import cors from '@fastify/cors'; -import { Client as PgClient } from 'pg'; +import { Pool } from 'pg'; import { createClient as createRedisClient } from 'redis'; import { DbService } from './services/db.service.js'; import { JobService } from './services/job.service.js'; @@ -8,7 +8,6 @@ import { SearchService } from './services/search.service.js'; import libraryRoutes from './routes/library.routes.js'; import searchRoutes from './routes/search.routes.js'; import adminRoutes from './routes/admin.routes.js'; -import vibeRoutes from './routes/vibe.routes.js'; import historyRoutes from './routes/history.routes.js'; import streamRoutes from './routes/stream.routes.js'; import quarantineRoutes from './routes/quarantine.routes.js'; @@ -33,10 +32,14 @@ export async function buildApp(config: AppConfig) { origin: process.env.CORS_ORIGIN?.split(',') || ['http://localhost:5173', 'http://localhost:5174'], }); - const pgClient = new PgClient({ + // A Pool (not a single Client) so concurrent requests get independent + // connections. Critical for correctness: recordPlay/recordSkip run + // BEGIN...COMMIT transactions, and other requests' queries must not be + // able to land inside another request's open transaction on a shared + // connection (see db.service.ts withTransaction()). + const pgPool = new Pool({ connectionString: process.env.DATABASE_URL, }); - await pgClient.connect(); const redisClient = createRedisClient({ url: process.env.REDIS_URL, @@ -50,7 +53,7 @@ export async function buildApp(config: AppConfig) { protocol: 'http', apiKey: config.searchApiKey, }); - const dbService = new DbService(pgClient, searchService); + const dbService = new DbService(pgPool, searchService); // Apply the idempotent schema on boot so tables added after the initial DB // volume was created (e.g. play_history, feedback) exist. The init-time @@ -118,7 +121,7 @@ export async function buildApp(config: AppConfig) { }; try { - await pgClient.query('SELECT 1'); + await pgPool.query('SELECT 1'); status.postgres = 'ok'; } catch (err) { status.postgres = 'error'; @@ -136,7 +139,7 @@ export async function buildApp(config: AppConfig) { } const isHealthy = status.postgres === 'ok' && status.redis === 'ok'; - + if (isHealthy) { return reply.code(200).send(status); } else { @@ -148,7 +151,6 @@ export async function buildApp(config: AppConfig) { fastify.register(libraryRoutes, { prefix: '/api', dbService }); fastify.register(searchRoutes, { prefix: '/api', dbService }); fastify.register(adminRoutes, { prefix: '/api/admin', jobService, dbService }); - fastify.register(vibeRoutes, { prefix: '/api/vibe', dbService }); fastify.register(historyRoutes, { prefix: '/api', dbService }); fastify.register(streamRoutes, { prefix: '/api', dbService }); fastify.register(quarantineRoutes, { prefix: '/api', dbService }); @@ -159,24 +161,9 @@ export async function buildApp(config: AppConfig) { fastify.register(v2Routes, { prefix: '/api', dbService, sessionDirector }); fastify.register(discoveryRoutes, { prefix: '/api', dbService }); - fastify.post('/api/test/enqueue-job', async (request, reply) => { - const { jobType, trackId, payload } = request.body as any; - try { - if (jobType === 'metadataRefresh') { - await jobService.enqueueMetadataRefresh(trackId, payload.type); - } else if (jobType === 'audioAnalysis') { - await jobService.enqueueAudioAnalysis(trackId, payload.features); - } else if (jobType === 'cleanup') { - await jobService.enqueueCleanup(payload.reason, payload.targetFiles); - } else { - return reply.code(400).send({ error: 'Unknown job type' }); - } - await reply.send({ message: 'Job enqueued' }); - } catch (error) { - request.log.error(error); - await reply.status(500).send({ error: 'Internal server error' }); - } - }); + // ponytail: /api/test/enqueue-job (manual job-enqueue test endpoint) removed — + // nothing in the deployed app or its tests called it, and deployment never + // sets NODE_ENV so an env gate would've stayed live in prod anyway. // Register hooks to close connections on shutdown fastify.addHook('onClose', async () => { @@ -188,7 +175,7 @@ export async function buildApp(config: AppConfig) { fastify.log.error(err); } try { - await pgClient.end(); + await pgPool.end(); } catch (err) { fastify.log.error(err); } @@ -204,5 +191,5 @@ export async function buildApp(config: AppConfig) { } }); - return { fastify, pgClient, redisClient }; + return { fastify, pgPool, redisClient }; } diff --git a/backend/src/routes/v2.routes.ts b/backend/src/routes/v2.routes.ts index 78b5f9f..dfbd0ec 100644 --- a/backend/src/routes/v2.routes.ts +++ b/backend/src/routes/v2.routes.ts @@ -74,7 +74,7 @@ export default async function v2Routes(fastify: FastifyInstance, options: { dbSe // Replan if running low if (active.plan.length < 5) { const refill = await director.replan(userId, active.sessionId, active.plan, [next.trackId], active.seedTrackId ?? undefined); - active.plan.push(...refill); + active.plan = refill; } await setActivePlan(userId, active); @@ -111,7 +111,7 @@ export default async function v2Routes(fastify: FastifyInstance, options: { dbSe if (active) { const playedTrackIds = [trackId]; const refill = await director.replan(userId, active.sessionId, active.plan, playedTrackIds, active.seedTrackId ?? undefined); - active.plan.push(...refill); + active.plan = refill; await setActivePlan(userId, active); } diff --git a/backend/src/routes/vibe.routes.ts b/backend/src/routes/vibe.routes.ts deleted file mode 100644 index 50632fd..0000000 --- a/backend/src/routes/vibe.routes.ts +++ /dev/null @@ -1,76 +0,0 @@ -import { FastifyInstance } from 'fastify'; -import { DbService } from '../services/db.service.js'; - -export default async function vibeRoutes(fastify: FastifyInstance, options: { dbService: DbService }) { - const { dbService } = options; - - // Start a new vibe session - fastify.post('/start', async (request, reply) => { - const { seedTrackId } = request.body as { seedTrackId: string }; - if (!seedTrackId) { - return reply.code(400).send({ error: 'seedTrackId is required' }); - } - const userId = (request.headers['x-user-id'] as string) || '00000000-0000-0000-0000-000000000000'; - const batchId = await dbService.createVibeSession(userId, seedTrackId); - return reply.send({ batchId }); - }); - - // Get the next chunk of tracks for the active session - fastify.get('/next', async (request, reply) => { - const userId = (request.headers['x-user-id'] as string) || '00000000-0000-0000-0000-000000000000'; - const activeSession = await dbService.getActiveVibeSession(userId); - - if (!activeSession) { - return reply.code(404).send({ error: 'No active vibe session found' }); - } - - const tracks = await dbService.getNextVibeChunk(activeSession.batchId); - - // Update the session timestamp to keep it alive - await dbService.updateVibeSession(activeSession.batchId); - - return tracks; - }); - - // Start/return a chunk seeded by a genre (id or name) — no active session required. - fastify.get('/from-genre', async (request, reply) => { - const { genre } = request.query as { genre?: string }; - if (!genre) { - return reply.code(400).send({ error: 'genre query param is required' }); - } - const userId = (request.headers['x-user-id'] as string) || '00000000-0000-0000-0000-000000000000'; - const tracks = await dbService.getVibeChunkFromGenre(genre, userId); - return tracks; - }); - - // Current ACTIVE batch metadata for the user, or 404. - fastify.get('/current', async (request, reply) => { - const userId = (request.headers['x-user-id'] as string) || '00000000-0000-0000-0000-000000000000'; - const session = await dbService.getCurrentVibeSession(userId); - if (!session) { - return reply.code(404).send({ error: 'No active vibe session found' }); - } - return reply.send(session); - }); - - // Heartbeat: keep the active batch alive by bumping last_interaction_at. - fastify.post('/heartbeat', async (request, reply) => { - const userId = (request.headers['x-user-id'] as string) || '00000000-0000-0000-0000-000000000000'; - const updated = await dbService.heartbeatVibeSession(userId); - if (!updated) { - return reply.code(404).send({ error: 'No active vibe session found' }); - } - return reply.send({ status: 'ok' }); - }); - - // End the vibe session - fastify.post('/end', async (request, reply) => { - const userId = (request.headers['x-user-id'] as string) || '00000000-0000-0000-0000-000000000000'; - const activeSession = await dbService.getActiveVibeSession(userId); - - if (activeSession) { - await dbService.endVibeSession(activeSession.batchId); - } - return reply.send({ status: 'session_ended' }); - }); -} diff --git a/backend/src/services/db.service.ts b/backend/src/services/db.service.ts index 98cc8e0..160ad11 100644 --- a/backend/src/services/db.service.ts +++ b/backend/src/services/db.service.ts @@ -1,9 +1,12 @@ import { readFile, unlink } from 'fs/promises'; import { fileURLToPath } from 'url'; import { dirname, join } from 'path'; -import { Client as PgClient } from 'pg'; +import { Pool, PoolClient } from 'pg'; import { SearchService } from './search.service.js'; +/** Anything with a `.query()` — either the shared Pool or a checked-out client. */ +type Queryable = Pool | PoolClient; + export interface Artist { id: string; name: string; @@ -444,15 +447,39 @@ const MIGRATIONS: { id: string; sql: string }[] = [ export class DbService { /** Exposed so route handlers (e.g. settings) can query the database directly. */ - readonly pgClient: PgClient; + readonly pgClient: Pool; constructor( - pgClient: PgClient, + pgClient: Pool, private searchService?: SearchService ) { this.pgClient = pgClient; } + /** + * Check out a dedicated client for a transaction and BEGIN/COMMIT/ROLLBACK + * on it, then release it back to the pool. Never run BEGIN...COMMIT on + * `this.pgClient` directly — the Pool hands out a different connection to + * every query, so a shared-connection transaction would let concurrent + * requests' queries land inside it (a ROLLBACK could discard another + * request's writes). Pass `client` through to any nested calls that must + * participate in the same transaction (e.g. recordEvidence, upsertClaim). + */ + private async withTransaction(fn: (client: PoolClient) => Promise): Promise { + const client = await this.pgClient.connect(); + try { + await client.query('BEGIN'); + const result = await fn(client); + await client.query('COMMIT'); + return result; + } catch (err) { + await client.query('ROLLBACK'); + throw err; + } finally { + client.release(); + } + } + /** * Apply the canonical schema (backend/src/db/schema.sql) on boot. The file is * fully idempotent — enums are guarded with DO/EXCEPTION blocks and every @@ -496,17 +523,13 @@ export class DbService { for (const migration of MIGRATIONS) { if (appliedIds.has(migration.id)) continue; console.log(`[DB] Running migration: ${migration.id}`); - await this.pgClient.query('BEGIN'); try { - await this.pgClient.query(migration.sql); - await this.pgClient.query( - 'INSERT INTO schema_migrations (id) VALUES ($1)', - [migration.id] - ); - await this.pgClient.query('COMMIT'); + await this.withTransaction(async (client) => { + await client.query(migration.sql); + await client.query('INSERT INTO schema_migrations (id) VALUES ($1)', [migration.id]); + }); console.log(`[DB] Migration applied: ${migration.id}`); } catch (err) { - await this.pgClient.query('ROLLBACK'); console.error(`[DB] Migration failed: ${migration.id}`, err); throw err; } @@ -737,32 +760,25 @@ export class DbService { * LIBRARY -> HIDDEN and is removed from all active views immediately. */ async dislikeTrack(userId: string, trackId: string): Promise { - try { - await this.pgClient.query('BEGIN'); - + await this.withTransaction(async (client) => { // Phase 1: hide the track in all active views - await this.pgClient.query( + await client.query( "UPDATE tracks SET state = 'HIDDEN' WHERE id = $1 AND state = 'LIBRARY'", [trackId] ); // Phase 1: insert dislike row (idempotent — won't create duplicate) - await this.pgClient.query( + await client.query( 'INSERT INTO dislikes (track_id) VALUES ($1) ON CONFLICT (track_id) DO NOTHING', [trackId] ); // Phase 1: log feedback signal for the Vibe learning loop - await this.pgClient.query( + await client.query( "INSERT INTO feedback (user_id, track_id, action) VALUES ($1, $2, 'disliked')", [userId, trackId] ); - - await this.pgClient.query('COMMIT'); - } catch (err) { - await this.pgClient.query('ROLLBACK'); - throw err; - } + }); // Write evidence: hidden → negative profile (only on success) await this.recordEvidence({ @@ -781,20 +797,10 @@ export class DbService { * reversal of the dislike action. */ async restoreDislike(trackId: string): Promise { - try { - await this.pgClient.query('BEGIN'); - - await this.pgClient.query('DELETE FROM dislikes WHERE track_id = $1', [trackId]); - await this.pgClient.query( - "UPDATE tracks SET state = 'LIBRARY' WHERE id = $1", - [trackId] - ); - - await this.pgClient.query('COMMIT'); - } catch (err) { - await this.pgClient.query('ROLLBACK'); - throw err; - } + await this.withTransaction(async (client) => { + await client.query('DELETE FROM dislikes WHERE track_id = $1', [trackId]); + await client.query("UPDATE tracks SET state = 'LIBRARY' WHERE id = $1", [trackId]); + }); } /** @@ -856,7 +862,7 @@ export class DbService { }); // Delete DB record (ON DELETE CASCADE handles track_genre, play_history, - // feedback, track_audio_features, track_lyrics, recommendation_batch_track) + // feedback, track_audio_features, track_lyrics) await this.pgClient.query('DELETE FROM tracks WHERE id = $1', [trackId]); // Delete physical file from disk @@ -878,36 +884,21 @@ export class DbService { // Completed play: history insert + play_count bump + Success-Driven Center // + evidence writing + listener-behavior claims. All atomic. - try { - await this.pgClient.query('BEGIN'); - + return this.withTransaction(async (client) => { // 1. Record play history - const insertRes = await this.pgClient.query( + const insertRes = await client.query( 'INSERT INTO play_history (user_id, track_id, batch_id, completed) VALUES ($1, $2, $3, $4) RETURNING id', [userId, trackId, batchId ?? null, true] ); const historyId = insertRes.rows[0].id as string; // 2. Bump play count + last_played_at - await this.pgClient.query( + await client.query( 'UPDATE tracks SET play_count = play_count + 1, last_played_at = NOW() WHERE id = $1', [trackId] ); - // 3. Success-Driven Center: move the user's most-recent ACTIVE batch seed - await this.pgClient.query( - `UPDATE recommendation_batch - SET seed_track_id = $2, last_interaction_at = NOW() - WHERE id = ( - SELECT id FROM recommendation_batch - WHERE user_id = $1 AND status = 'ACTIVE' - ORDER BY last_interaction_at DESC - LIMIT 1 - )`, - [userId, trackId] - ); - - // 4. Write evidence: playback_completed → longterm affinity + // 3. Write evidence: playback_completed → longterm affinity await this.recordEvidence({ user_id: userId, entity_type: 'track', @@ -916,10 +907,10 @@ export class DbService { profile: 'longterm', weight: 0.10, context: batchId ? { batch_id: batchId } : undefined, - }); + }, client); - // 5. Check for replay within 24h → strengthens longterm + obsession - const recentPlays = await this.pgClient.query( + // 4. Check for replay within 24h → strengthens longterm + obsession + const recentPlays = await client.query( `SELECT COUNT(*)::int AS cnt FROM play_history WHERE user_id = $1 AND track_id = $2 AND completed = true AND played_at > NOW() - INTERVAL '24 hours'`, @@ -933,7 +924,7 @@ export class DbService { signal: 'replay_within_24h', profile: 'longterm', weight: 0.25, - }); + }, client); await this.recordEvidence({ user_id: userId, entity_type: 'track', @@ -941,111 +932,98 @@ export class DbService { signal: 'replay_within_24h', profile: 'obsession', weight: 0.40, - }); + }, client); } - // 6. Listener-behavior writer: back-to-back play within 30 min → weak edges + // 5. Listener-behavior writer: back-to-back play within 30 min → weak edges // Resolve artist IDs for current track and previous track, then write // alias_of (same artist, different name) or same_scene_as (different artists). - const currentArtist = await this.pgClient.query( - `SELECT a.id AS artist_id, a.normalized_name - FROM track_artists ta - JOIN artists a ON a.id = ta.artist_id - WHERE ta.track_id = $1 AND ta.role = 'main' + // Reads track_artists_v2 (claim-fusion-backed), NOT the legacy + // track_artists table — the vibe engine reads claim_fusion, so writing + // claims keyed off a possibly-stale legacy table silently stops these + // edges from being written for newly-enriched tracks. + const currentArtist = await client.query( + `SELECT tv.artist_id, a.normalized_name + FROM track_artists_v2 tv + JOIN artists a ON a.id = tv.artist_id + WHERE tv.track_id = $1 AND tv.role = 'main' + ORDER BY tv.confidence DESC LIMIT 1`, [trackId] ); const currentArtistRow = currentArtist.rows[0] as { artist_id: string; normalized_name: string } | undefined; if (currentArtistRow) { - // Get the previous completed play's track + artist - const prevPlay = await this.pgClient.query( - `SELECT ph_prev.track_id AS prev_track_id - FROM play_history ph_this - JOIN play_history ph_prev ON ph_prev.user_id = ph_this.user_id AND ph_prev.completed = true - WHERE ph_this.track_id = $1 AND ph_this.user_id = $2 AND ph_this.completed = true - AND ph_prev.played_at < ph_this.played_at - ORDER BY ph_prev.played_at DESC + // Previous completed play's track + the gap in minutes, in one query + // via LAG() instead of a separate prev-track lookup + gap lookup. + const prevPlay = await client.query( + `SELECT prev_track_id, EXTRACT(EPOCH FROM (played_at - prev_played_at)) / 60 AS min_gap + FROM ( + SELECT track_id, played_at, + LAG(track_id) OVER (ORDER BY played_at) AS prev_track_id, + LAG(played_at) OVER (ORDER BY played_at) AS prev_played_at + FROM play_history + WHERE user_id = $1 AND completed = true + ) w + WHERE w.track_id = $2 + ORDER BY w.played_at DESC LIMIT 1`, - [trackId, userId] + [userId, trackId] ); const prevTrackId = prevPlay.rows[0]?.prev_track_id as string | undefined; + const gapMinutes = prevPlay.rows[0]?.min_gap as number | undefined; - if (prevTrackId) { - // Check time gap - const times = await this.pgClient.query( - `SELECT EXTRACT(EPOCH FROM (ph_this.played_at - ph_prev.played_at)) / 60 AS min_gap - FROM play_history ph_this - JOIN play_history ph_prev ON ph_prev.id = ( - SELECT id FROM play_history - WHERE user_id = $1 AND track_id = $2 AND completed = true - ORDER BY played_at DESC LIMIT 1 - ) - WHERE ph_this.id = ( - SELECT id FROM play_history - WHERE user_id = $1 AND track_id = $3 AND completed = true - ORDER BY played_at DESC LIMIT 1 - )`, - [userId, prevTrackId, trackId] + if (prevTrackId && gapMinutes !== undefined && gapMinutes <= 30) { + const prevArtist = await client.query( + `SELECT tv.artist_id, a.normalized_name + FROM track_artists_v2 tv + JOIN artists a ON a.id = tv.artist_id + WHERE tv.track_id = $1 AND tv.role = 'main' + ORDER BY tv.confidence DESC + LIMIT 1`, + [prevTrackId] ); - const gapMinutes = times.rows[0]?.min_gap as number | undefined; + const prevArtistRow = prevArtist.rows[0] as { artist_id: string; normalized_name: string } | undefined; - if (gapMinutes !== undefined && gapMinutes <= 30) { - const prevArtist = await this.pgClient.query( - `SELECT a.id AS artist_id, a.normalized_name - FROM track_artists ta - JOIN artists a ON a.id = ta.artist_id - WHERE ta.track_id = $1 AND ta.role = 'main' - LIMIT 1`, - [prevTrackId] - ); - const prevArtistRow = prevArtist.rows[0] as { artist_id: string; normalized_name: string } | undefined; - - if (prevArtistRow) { - if (prevArtistRow.normalized_name === currentArtistRow.normalized_name) { - // Same normalized artist name → weak alias_of - await this.upsertClaim({ - user_id: userId, - subject_type: 'artist', - subject_id: prevArtistRow.artist_id, - predicate: 'alias_of', - object_type: 'artist', - object_id: currentArtistRow.artist_id, - source: 'listener_behavior', - confidence: 0.2, - }); - } else { - // Different artists played back-to-back → weak same_scene_as - await this.upsertClaim({ - user_id: userId, - subject_type: 'artist', - subject_id: prevArtistRow.artist_id, - predicate: 'same_scene_as', - object_type: 'artist', - object_id: currentArtistRow.artist_id, - source: 'listener_behavior', - confidence: 0.3, - }); - } + if (prevArtistRow) { + if (prevArtistRow.normalized_name === currentArtistRow.normalized_name) { + // Same normalized artist name → weak alias_of + await this.upsertClaim({ + user_id: userId, + subject_type: 'artist', + subject_id: prevArtistRow.artist_id, + predicate: 'alias_of', + object_type: 'artist', + object_id: currentArtistRow.artist_id, + source: 'listener_behavior', + confidence: 0.2, + }, client); + } else { + // Different artists played back-to-back → weak same_scene_as + await this.upsertClaim({ + user_id: userId, + subject_type: 'artist', + subject_id: prevArtistRow.artist_id, + predicate: 'same_scene_as', + object_type: 'artist', + object_id: currentArtistRow.artist_id, + source: 'listener_behavior', + confidence: 0.3, + }, client); } } } } - await this.pgClient.query('COMMIT'); return historyId; - } catch (err) { - await this.pgClient.query('ROLLBACK'); - throw err; - } + }); } async recordSkip(userId: string, trackId: string): Promise { // Skips do NOT move the center (transient per spec). Also writes negative evidence. - try { - await this.pgClient.query('BEGIN'); - await this.pgClient.query('UPDATE tracks SET skip_count = skip_count + 1 WHERE id = $1', [trackId]); - await this.pgClient.query( + await this.withTransaction(async (client) => { + await client.query('UPDATE tracks SET skip_count = skip_count + 1 WHERE id = $1', [trackId]); + await client.query( "INSERT INTO feedback (user_id, track_id, action) VALUES ($1, $2, 'skipped')", [userId, trackId] ); @@ -1057,12 +1035,8 @@ export class DbService { signal: 'skip_quick', profile: 'negative', weight: -0.20, - }); - await this.pgClient.query('COMMIT'); - } catch (err) { - await this.pgClient.query('ROLLBACK'); - throw err; - } + }, client); + }); } async recordFeedback(userId: string, trackId: string, action: FeedbackAction): Promise { @@ -1110,478 +1084,6 @@ export class DbService { return res.rows as HistoryEntry[]; } - async createVibeSession(userId: string, seedTrackId: string): Promise { - const res = await this.pgClient.query( - 'INSERT INTO recommendation_batch (user_id, seed_track_id, status) VALUES ($1, $2, \'ACTIVE\') RETURNING id', - [userId, seedTrackId] - ); - return res.rows[0].id; - } - - async getActiveVibeSession(userId: string): Promise<{ batchId: string, seedTrackId: string } | null> { - const res = await this.pgClient.query( - `SELECT b.id as "batchId", b.seed_track_id as "seedTrackId" - FROM recommendation_batch b - WHERE b.user_id = $1 AND b.status = 'ACTIVE' - ORDER BY b.last_interaction_at DESC LIMIT 1`, - [userId] - ); - return res.rows[0] || null; - } - - /** - * "Rolling Vibe" recommendation engine (full spec implementation). - * - * This is a DB-only, single-round-trip scorer: NO external API calls happen on - * the request path. All enrichment (artist_similar, track_genre, audio - * features) is populated out-of-band by the workers; here we only read it. - * - * ============================ SCORING MODEL ============================ - * Every LIBRARY candidate gets a weighted score: - * - * score = W_GENRE * genre_overlap (dominant signal) - * + W_ARTSIM * artist_similarity (Last.fm similar-artist match) - * + W_SAMEART * same_artist (mild "more of this artist") - * + W_FEEDBCK * feedback_affinity (per-user promoted/disliked genres) - * + W_AUDIO * audio_closeness (NULL-safe, 0 when data missing) - * + W_RANDOM * jitter (tie-break / exploration) - * - * - genre_overlap: SUM(track_genre.weight) over genres shared with the seed. - * Dominant because genre is the most reliable similarity signal we have. - * - artist_similarity: if the candidate's artist name is listed in - * artist_similar for the SEED's artist, add the stored Last.fm `match` - * (0..1). This is the "discovery within library" nudge. - * - same_artist: small flat bonus when candidate shares the seed's artist. - * - feedback_affinity: bounded per-user term. Genres the user has 'promoted' - * push a candidate up; genres they've 'disliked' push it down. Clamped to - * [-1, 1] so a noisy feedback history can never dominate genre matching. - * - audio_closeness: see audioClosenessSQL() — NULL-safe, contributes 0 when - * either side lacks features. Low weight until real Essentia data lands. - * - jitter: RANDOM() in [0,1), scaled small, purely for variety / tie-breaks. - * - * ===================== 80/20 LOCAL vs PROBATION ======================== - * The 20-track chunk is split into two pools UNIONed with a `source` tag: - * - LOCAL (~16): the full scoring model above. "More of what you know." - * - PROBATION (~4): discovery-leaning. Scores LIBRARY tracks by - * artist_similar.match + least-recently-played recency, with genre - * overlap down-weighted, so it feels exploratory. If the probation pool - * is empty (no similarity data yet) the local pool simply fills the full - * 20 (see the genre-cap fill step), so the chunk is never short. - * - * ===================== DIVERSITY CONSTRAINTS ========================== - * Two caps are enforced AFTER scoring, over a generously-sized candidate set: - * 1. Max 1 track per ARTIST: ROW_NUMBER() OVER (PARTITION BY artist ...) = 1. - * 2. Max 2 tracks per GENRE: we attribute each track a single "primary genre" - * (highest-weight track_genre row) and apply a running - * COUNT() <= 2 window over that primary genre, ordered by the merged - * pool priority. Tracks with no genre are never capped. We over-fetch - * (LIMIT 60) before the caps so the caps don't starve the final 20 when - * more diverse tracks are actually available. - * - * ============================ FALLBACK =============================== - * If the seed has no genres AND no artist-similar data, every structured term - * is 0 and ordering collapses to (same_artist + feedback + jitter) — i.e. a - * graceful, mostly-random ordering over LIBRARY tracks rather than empties. - * - * After selection we record the chunk into recommendation_batch_track - * (ON CONFLICT DO NOTHING) so subsequent chunks for this batch exclude them. - */ - async getNextVibeChunk(batchId: string): Promise { - // --- Named scoring weights (see scoring-model comment block above) --- - const W_GENRE = 1.0; // dominant: shared-genre weight sum - const W_ARTSIM = 0.8; // Last.fm similar-artist match (0..1) - const W_SAMEART = 0.4; // flat bonus for same artist as the seed - const W_FEEDBCK = 0.6; // per-user promoted/disliked genre affinity, clamped - const W_AUDIO = 0.4; // NULL-safe audio closeness (energy + bpm + danceability) - const W_RANDOM = 0.3; // jitter for tie-breaking / exploration - - // Probation (discovery) pool weights: lean on artist similarity + recency, - // de-emphasise direct genre overlap so it feels like exploration. - const P_ARTSIM = 1.0; // similar-artist match is the primary discovery signal - const P_GENRE = 0.25; // genre overlap matters less in discovery - const P_RECENCY = 0.5; // least-recently-played gets surfaced - const P_RANDOM = 0.4; - - const LOCAL_TARGET = 16; // ~80% of the 20-track chunk - const PROBATION_TARGET = 4; // ~20% of the 20-track chunk - const CHUNK_SIZE = 20; - const OVERFETCH = 60; // fetch extra so diversity caps don't starve the 20 - const GENRE_CAP = 2; // max tracks per primary genre per chunk - - const trackCols = ` - id, path, hash, title, artist, album_id, duration, state, - play_count, skip_count, dislike_count, last_played_at, mtime, source_type`; - - const query = ` - WITH seed AS ( - SELECT t.id, t.artist, t.normalized_artist - FROM recommendation_batch rb - JOIN tracks t ON t.id = rb.seed_track_id - WHERE rb.id = $1 - ), - seed_user AS ( - SELECT user_id FROM recommendation_batch WHERE id = $1 - ), - -- seed artist UUID (tracks.artist is a name; artist_similar keys on artists.id) - -- Join on normalized identity, NOT raw name, so "The Beatles" (track) still - -- matches "the beatles" / "Beatles" (artist row). A raw-name join silently - -- fails on any case/feature difference and zeroes the artist_sim signal. - seed_artist AS ( - SELECT a.id AS artist_id - FROM seed s - JOIN artists a ON a.normalized_name = s.normalized_artist - ), - seed_genres AS ( - SELECT tg.genre_id, tg.weight - FROM seed s - JOIN track_genre tg ON tg.track_id = s.id - ), - -- artists Last.fm-similar to the seed's artist (by name, for candidate join) - -- normalize_artist() on similar_name so the join to tracks.normalized_artist - -- matches case- and feature-insensitively. Handles both old rows (raw - -- Last.fm names) and new rows (already normalized by the worker). - similar_artists AS ( - SELECT normalize_artist(asim.similar_name) AS similar_name, asim.match - FROM seed_artist sa - JOIN artist_similar asim ON asim.artist_id = sa.artist_id - ), - -- per-user genre affinity from feedback: +promoted, -disliked, clamped [-1,1] - feedback_genre AS ( - SELECT tg.genre_id, - GREATEST(-1.0, LEAST(1.0, - SUM(CASE f.action - WHEN 'promoted' THEN 0.5 - WHEN 'disliked' THEN -0.5 - ELSE 0 END) - )) AS affinity - FROM feedback f - JOIN seed_user su ON su.user_id = f.user_id - JOIN track_genre tg ON tg.track_id = f.track_id - WHERE f.action IN ('promoted', 'disliked') - GROUP BY tg.genre_id - ), - -- primary genre per track = its highest-weight track_genre row (for genre cap) - primary_genre AS ( - SELECT track_id, genre_id FROM ( - SELECT tg.track_id, tg.genre_id, - ROW_NUMBER() OVER (PARTITION BY tg.track_id ORDER BY tg.weight DESC) AS rn - FROM track_genre tg - ) pg WHERE rn = 1 - ), - base AS ( - SELECT - t.*, - pg.genre_id AS primary_genre_id, - -- genre overlap: sum of shared-genre weights with the seed - COALESCE(( - SELECT SUM(tg.weight) - FROM track_genre tg - JOIN seed_genres sg ON sg.genre_id = tg.genre_id - WHERE tg.track_id = t.id - ), 0) AS genre_overlap, - -- best similar-artist match for this candidate's normalized artist (0 if none) - COALESCE(( - SELECT MAX(sa.match) FROM similar_artists sa - WHERE sa.similar_name = t.normalized_artist - ), 0) AS artist_sim, - (CASE WHEN t.normalized_artist = s.normalized_artist THEN 1 ELSE 0 END) AS same_artist, - -- bounded feedback affinity: sum candidate's genre affinities, clamp - GREATEST(-1.0, LEAST(1.0, COALESCE(( - SELECT SUM(fg.affinity) - FROM track_genre tg - JOIN feedback_genre fg ON fg.genre_id = tg.genre_id - WHERE tg.track_id = t.id - ), 0))) AS feedback_affinity, - -- NULL-safe audio closeness vs seed (0 when either side has no features) - ${this.audioClosenessSQL('t.id')} AS audio_closeness, - -- recency: oldest last_played_at scores highest (NULLs = never played = max) - COALESCE(EXTRACT(EPOCH FROM (NOW() - t.last_played_at)) / 2592000.0, 1.0) - AS recency, - -- session-level artist play count: how many times this normalized artist - -- has already been recommended in the current batch (for decay scoring). - -- Using normalized_artist so "Artist feat. Guest" and "Artist" share a count. - (SELECT COUNT(*) FROM recommendation_batch_track rbt - JOIN tracks tr ON tr.id = rbt.track_id - WHERE rbt.batch_id = $1 AND tr.normalized_artist = t.normalized_artist) AS artist_play_count - FROM tracks t - CROSS JOIN seed s - LEFT JOIN primary_genre pg ON pg.track_id = t.id - WHERE t.state = 'LIBRARY' - AND t.id != s.id - AND NOT EXISTS ( - SELECT 1 FROM recommendation_batch_track rbt - WHERE rbt.batch_id = $1 AND rbt.track_id = t.id - ) - ), - -- LOCAL pool: full scoring model - local_pool AS ( - SELECT b.*, - 'local'::text AS source, - ( ( ${W_GENRE} * b.genre_overlap - + ${W_ARTSIM} * b.artist_sim - + ${W_SAMEART} * b.same_artist - + ${W_FEEDBCK} * b.feedback_affinity - + ${W_AUDIO} * b.audio_closeness - + ${W_RANDOM} * RANDOM() ) - * GREATEST(0.2, 1.0 - (b.artist_play_count - 1) * 0.20) ) AS score - FROM base b - ), - -- PROBATION pool: discovery-leaning, only candidates with a similarity signal - probation_pool AS ( - SELECT b.*, - 'probation'::text AS source, - ( ( ${P_ARTSIM} * b.artist_sim - + ${P_GENRE} * b.genre_overlap - + ${P_RECENCY} * LEAST(b.recency, 2.0) - + ${P_RANDOM} * RANDOM() ) - * GREATEST(0.2, 1.0 - (b.artist_play_count - 1) * 0.20) ) AS score - FROM base b - WHERE b.artist_sim > 0 - ), - local_ranked AS ( - SELECT lp.*, ROW_NUMBER() OVER (ORDER BY lp.score DESC) AS rn - FROM local_pool lp - ), - probation_ranked AS ( - SELECT pp.*, ROW_NUMBER() OVER (ORDER BY pp.score DESC) AS rn - FROM probation_pool pp - ), - -- merge: take top probation candidates first, then top local, dedup by id - merged AS ( - SELECT * FROM ( - SELECT pr.*, 0 AS pool_order FROM probation_ranked pr WHERE pr.rn <= ${PROBATION_TARGET} - UNION ALL - SELECT lr.*, 1 AS pool_order FROM local_ranked lr WHERE lr.rn <= ${OVERFETCH} - ) u - ), - -- dedup (a track can appear in both pools): keep its best (probation-first) row - deduped AS ( - SELECT m.* FROM ( - SELECT *, ROW_NUMBER() OVER ( - PARTITION BY id ORDER BY pool_order ASC, score DESC - ) AS dedup_rn - FROM merged - ) m WHERE m.dedup_rn = 1 - ), - -- DIVERSITY CAP 1: max 1 per normalized artist (keep best-scoring row per - -- normalized artist). Uses normalized_artist so "Artist feat. Guest" and - -- "Artist" are treated as the same artist for dedup. - artist_capped AS ( - SELECT d.* FROM ( - SELECT *, ROW_NUMBER() OVER ( - PARTITION BY normalized_artist ORDER BY pool_order ASC, score DESC - ) AS artist_rn - FROM deduped - ) d WHERE d.artist_rn = 1 - ), - -- DIVERSITY CAP 2: max ${GENRE_CAP} per primary genre. NULL-genre tracks - -- are never capped (assigned rank 1). Order by pool/score so the best - -- representatives of each genre survive. - genre_capped AS ( - SELECT g.* FROM ( - SELECT *, - CASE WHEN primary_genre_id IS NULL THEN 1 - ELSE ROW_NUMBER() OVER ( - PARTITION BY primary_genre_id ORDER BY pool_order ASC, score DESC - ) END AS genre_rn - FROM artist_capped - ) g WHERE g.genre_rn <= ${GENRE_CAP} - ), - chosen AS ( - SELECT * FROM genre_capped - ORDER BY pool_order ASC, score DESC - LIMIT ${CHUNK_SIZE} - ), - recorded AS ( - INSERT INTO recommendation_batch_track (batch_id, track_id) - SELECT $1, id FROM chosen - ON CONFLICT DO NOTHING - ) - SELECT ${trackCols}, al.artwork_id - FROM chosen - LEFT JOIN albums al ON al.id = chosen.album_id - ORDER BY score DESC; - `; - - void LOCAL_TARGET; // documented split target; LOCAL fills remainder via OVERFETCH - const res = await this.pgClient.query(query, [batchId]); - return res.rows as Track[]; - } - - /** - * NULL-safe audio-feature closeness term. - * - * Returns a SQL scalar expression (0..1, higher = more similar) comparing the - * candidate track (`candidateIdExpr`) against the seed's audio features via a - * LEFT JOIN-style correlated lookup. It is COALESCE/NULL-safe: if EITHER the - * candidate OR the seed lacks a track_audio_features row (or the compared - * columns are NULL), the term evaluates to 0 so missing audio data never zeroes - * a track out of the running — it simply doesn't contribute. - * - * Closeness = average normalized closeness across energy, bpm and danceability. - * energy/danceability are 0..1; bpm is normalised over a 200 BPM span. - * Each dimension is optional: only dimensions where BOTH seed and candidate - * have a non-NULL value contribute, and the divisor shrinks accordingly so a - * track missing one feature isn't unfairly penalised. - */ - private audioClosenessSQL(candidateIdExpr: string): string { - return ` - COALESCE(( - SELECT - ( CASE WHEN sf.energy IS NOT NULL AND cf.energy IS NOT NULL - THEN (1 - LEAST(ABS(cf.energy - sf.energy), 1)) ELSE NULL END - + CASE WHEN sf.bpm IS NOT NULL AND cf.bpm IS NOT NULL - THEN (1 - LEAST(ABS(cf.bpm - sf.bpm) / 200.0, 1)) ELSE NULL END - + CASE WHEN sf.danceability IS NOT NULL AND cf.danceability IS NOT NULL - THEN (1 - LEAST(ABS(cf.danceability - sf.danceability), 1)) ELSE NULL END - ) / NULLIF( - (CASE WHEN sf.energy IS NOT NULL AND cf.energy IS NOT NULL THEN 1 ELSE 0 END - + CASE WHEN sf.bpm IS NOT NULL AND cf.bpm IS NOT NULL THEN 1 ELSE 0 END - + CASE WHEN sf.danceability IS NOT NULL AND cf.danceability IS NOT NULL THEN 1 ELSE 0 END - ), 0) - FROM track_audio_features cf - JOIN track_audio_features sf ON sf.track_id = (SELECT id FROM seed) - WHERE cf.track_id = ${candidateIdExpr} - ), 0)`; - } - - /** - * GET /api/vibe/from-genre — start a chunk seeded by a genre rather than a track. - * - * Accepts a genre id (UUID) or a genre name. Scores LIBRARY tracks by their - * membership weight in that genre (+ jitter), applies the same diversity caps - * (max 1 per artist, max 2 per primary genre) and returns up to 20 Track[]. - * - * Design decision: this does NOT create a recommendation_batch and does NOT - * require an active session. It is a lightweight, stateless "play this genre" - * entry point; the caller can subsequently POST /start to roll a real session. - * Because there's no batch, returned tracks are not recorded anywhere. - */ - async getVibeChunkFromGenre(genreIdOrName: string, _userId: string): Promise { - const isUuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test( - genreIdOrName - ); - const GENRE_CAP = 2; - // Note: no table prefix — these cols flow through CTEs (genre_capped) - // where the `t.` / `al.` aliases no longer apply. - const trackCols = ` - id, path, hash, title, artist, album_id, duration, state, - play_count, skip_count, dislike_count, last_played_at, mtime, source_type, artwork_id`; - - const query = ` - WITH target_genre AS ( - SELECT id FROM genre WHERE ${isUuid ? 'id = $1::uuid' : 'name = $1'} - ), - base AS ( - SELECT t.*, al.artwork_id, - tg.weight AS genre_weight, - pg.genre_id AS primary_genre_id, - (tg.weight * 1.0 + RANDOM() * 0.3) AS score - FROM tracks t - JOIN track_genre tg ON tg.track_id = t.id - JOIN target_genre g ON g.id = tg.genre_id - LEFT JOIN albums al ON al.id = t.album_id - LEFT JOIN ( - SELECT track_id, genre_id FROM ( - SELECT tg2.track_id, tg2.genre_id, - ROW_NUMBER() OVER (PARTITION BY tg2.track_id ORDER BY tg2.weight DESC) AS rn - FROM track_genre tg2 - ) p WHERE rn = 1 - ) pg ON pg.track_id = t.id - WHERE t.state = 'LIBRARY' - ), - artist_capped AS ( - SELECT b.* FROM ( - SELECT *, ROW_NUMBER() OVER (PARTITION BY normalize_artist(artist) ORDER BY score DESC) AS artist_rn - FROM base - ) b WHERE b.artist_rn = 1 - ), - genre_capped AS ( - SELECT g.* FROM ( - SELECT *, - CASE WHEN primary_genre_id IS NULL THEN 1 - ELSE ROW_NUMBER() OVER (PARTITION BY primary_genre_id ORDER BY score DESC) END - AS genre_rn - FROM artist_capped - ) g WHERE g.genre_rn <= ${GENRE_CAP} - ) - SELECT ${trackCols} FROM genre_capped - ORDER BY score DESC - LIMIT 20; - `; - const res = await this.pgClient.query(query, [genreIdOrName]); - return res.rows as Track[]; - } - - /** - * GET /api/vibe/current — active batch metadata for a user, or null. - */ - async getCurrentVibeSession(userId: string): Promise< - { batchId: string; seedTrackId: string | null; lastInteractionAt: Date } | null - > { - const res = await this.pgClient.query( - `SELECT id AS "batchId", seed_track_id AS "seedTrackId", - last_interaction_at AS "lastInteractionAt" - FROM recommendation_batch - WHERE user_id = $1 AND status = 'ACTIVE' - ORDER BY last_interaction_at DESC - LIMIT 1`, - [userId] - ); - return res.rows[0] || null; - } - - /** - * POST /api/vibe/heartbeat — bump last_interaction_at on the user's ACTIVE batch. - * Returns true if a session was found and updated. - */ - async heartbeatVibeSession(userId: string): Promise { - const res = await this.pgClient.query( - `UPDATE recommendation_batch - SET last_interaction_at = CURRENT_TIMESTAMP - WHERE id = ( - SELECT id FROM recommendation_batch - WHERE user_id = $1 AND status = 'ACTIVE' - ORDER BY last_interaction_at DESC - LIMIT 1 - )`, - [userId] - ); - return (res.rowCount ?? 0) > 0; - } - - async updateVibeSession(batchId: string): Promise { - await this.pgClient.query( - 'UPDATE recommendation_batch SET last_interaction_at = CURRENT_TIMESTAMP WHERE id = $1', - [batchId] - ); - } - - async endVibeSession(batchId: string): Promise { - await this.pgClient.query( - "UPDATE recommendation_batch SET status = 'RESOLVED' WHERE id = $1", - [batchId] - ); - } - - /** - * Reap stale ACTIVE vibe sessions — Invariant B ("No Deadlocks"): every - * ACTIVE batch must eventually reach a terminal state. Sessions with no - * interaction for `staleHours` (default 24, per spec §4) are transitioned to - * RESOLVED so a returning user starts a fresh session instead of resuming a - * frozen one. Returns the number of sessions reaped. - */ - async reapStaleVibeSessions(staleHours = 24): Promise { - const res = await this.pgClient.query( - `UPDATE recommendation_batch - SET status = 'RESOLVED' - WHERE status = 'ACTIVE' - AND last_interaction_at < NOW() - ($1 || ' hours')::INTERVAL`, - [String(staleHours)] - ); - return res.rowCount ?? 0; - } - async createArtist(data: Artist): Promise { const res = await this.pgClient.query( `INSERT INTO artists (name, mbid, discogs_id, image_path) @@ -1698,19 +1200,14 @@ export class DbService { if (err.code !== 'ENOENT') throw err; } - try { - await this.pgClient.query('BEGIN'); - await this.pgClient.query( + await this.withTransaction(async (client) => { + await client.query( "INSERT INTO feedback (user_id, track_id, action) VALUES ($1, $2, 'deleted_permanent')", [userId, trackId] ); // CASCADE deletes dislikes, play_history, feedback, etc. - await this.pgClient.query('DELETE FROM tracks WHERE id = $1', [trackId]); - await this.pgClient.query('COMMIT'); - } catch (err) { - await this.pgClient.query('ROLLBACK'); - throw err; - } + await client.query('DELETE FROM tracks WHERE id = $1', [trackId]); + }); } /** @@ -1816,24 +1313,19 @@ export class DbService { } // 3. DB transaction: re-parent history + delete rows. - await this.pgClient.query('BEGIN'); - try { + await this.withTransaction(async (client) => { for (const id of deleteIds) { - await this.pgClient.query( + await client.query( `UPDATE play_history SET track_id = $1 WHERE track_id = $2`, [keepId, id] ); - await this.pgClient.query( + await client.query( `UPDATE feedback SET track_id = $1 WHERE track_id = $2`, [keepId, id] ); - await this.pgClient.query(`DELETE FROM tracks WHERE id = $1`, [id]); + await client.query(`DELETE FROM tracks WHERE id = $1`, [id]); } - await this.pgClient.query('COMMIT'); - } catch (err) { - await this.pgClient.query('ROLLBACK'); - throw err; - } + }); } /** @@ -1865,8 +1357,8 @@ export class DbService { source: string; confidence?: number; raw?: unknown; - }): Promise { - const res = await this.pgClient.query( + }, client?: Queryable): Promise { + const res = await (client ?? this.pgClient).query( `INSERT INTO claims (user_id, subject_type, subject_id, predicate, object_type, object_id, source, confidence, raw) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id) @@ -1891,19 +1383,13 @@ export class DbService { * Batch UPSERT claims. Wraps multiple upsertClaim calls in a transaction. */ async upsertClaims(claims: Parameters[0][]): Promise { - const ids: string[] = []; - await this.pgClient.query('BEGIN'); - try { + return this.withTransaction(async (client) => { + const ids: string[] = []; for (const claim of claims) { - const id = await this.upsertClaim(claim); - ids.push(id); + ids.push(await this.upsertClaim(claim, client)); } - await this.pgClient.query('COMMIT'); return ids; - } catch (err) { - await this.pgClient.query('ROLLBACK'); - throw err; - } + }); } /** @@ -2011,8 +1497,8 @@ export class DbService { profile: string; weight: number; context?: unknown; - }): Promise { - const res = await this.pgClient.query( + }, client?: Queryable): Promise { + const res = await (client ?? this.pgClient).query( `INSERT INTO evidence (user_id, entity_type, entity_id, signal, profile, weight, context) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id`, [ @@ -2041,7 +1527,7 @@ export class DbService { dimension, value_delta: evidence.weight, confidence_delta: 0.05, - }); + }, client); return id; } @@ -2052,19 +1538,13 @@ export class DbService { async recordEvidenceBatch( evidenceList: Parameters[0][] ): Promise { - const ids: string[] = []; - await this.pgClient.query('BEGIN'); - try { + return this.withTransaction(async (client) => { + const ids: string[] = []; for (const ev of evidenceList) { - const id = await this.recordEvidence(ev); - ids.push(id); + ids.push(await this.recordEvidence(ev, client)); } - await this.pgClient.query('COMMIT'); return ids; - } catch (err) { - await this.pgClient.query('ROLLBACK'); - throw err; - } + }); } /** @@ -2116,10 +1596,10 @@ export class DbService { dimension: string; value_delta: number; confidence_delta?: number; - }): Promise { + }, client?: Queryable): Promise { const { user_id, profile, entity_type, entity_id, dimension, value_delta, confidence_delta = 0.05 } = params; - await this.pgClient.query( + await (client ?? this.pgClient).query( `INSERT INTO listener_beliefs (user_id, profile, entity_type, entity_id, dimension, value, confidence, evidence_count, last_reinforced_at) VALUES ($1, $2, $3, $4, $5, $6, $7, 1, NOW()) ON CONFLICT (user_id, profile, entity_type, entity_id, dimension) @@ -2199,15 +1679,32 @@ export class DbService { } } + /** Watermark for refreshClaimFusion() — last-seen MAX(claims.last_reinforced_at). */ + private lastClaimsWatermark: string | null = null; + /** - * Refresh the claim_fusion materialised view. Called on a periodic - * timer so the graph's read path stays current with new claims. + * Refresh the claim_fusion materialised view. Called on a periodic timer so + * the graph's read path stays current with new claims. Skips the (fairly + * expensive, shared-connection-stalling) REFRESH when claims haven't + * changed since the last tick — a NOTIFY trigger exists on the claims + * table but nothing LISTENs for it yet, so this cheap watermark check + * stands in for that. last_reinforced_at is bumped by both inserts + * (DEFAULT NOW()) and upsertClaim's ON CONFLICT UPDATE, so it tracks all + * claim mutations that would change the view's output. * CONCURRENTLY requires the unique index (idx_claim_fusion_pk), * which the 20260708_materialize_claim_fusion migration creates. */ async refreshClaimFusion(): Promise { try { + const watermarkRes = await this.pgClient.query<{ max: string | null }>( + 'SELECT MAX(last_reinforced_at) AS max FROM claims' + ); + const watermark = watermarkRes.rows[0]?.max ?? null; + if (watermark !== null && watermark === this.lastClaimsWatermark) { + return; // no claim changes since the last refresh — skip it + } await this.pgClient.query('SELECT refresh_claim_fusion()'); + this.lastClaimsWatermark = watermark; } catch (err) { // Non-fatal: the MV may not exist yet on first boot before // migrations run. Log and move on; the next tick will retry. diff --git a/backend/src/services/generators.service.ts b/backend/src/services/generators.service.ts index fef14a4..1f9c815 100644 --- a/backend/src/services/generators.service.ts +++ b/backend/src/services/generators.service.ts @@ -178,7 +178,7 @@ async function discoveryGenerator(db: DbService, ctx: GeneratorContext): Promise const maxCandidates = Math.max(1, Math.floor(10 * noveltyTolerance)); const unfamiliarRes = await db.pgClient.query( - `SELECT DISTINCT cf.object_id AS artist_id + `SELECT cf.object_id AS artist_id, MAX(cf.fused_value) AS edge_strength FROM claim_fusion cf WHERE cf.subject_type = 'artist' AND cf.subject_id = ANY($1::uuid[]) @@ -191,16 +191,19 @@ async function discoveryGenerator(db: DbService, ctx: GeneratorContext): Promise AND lb.entity_id = cf.object_id AND lb.profile IN ('longterm', 'obsession') ) + GROUP BY cf.object_id LIMIT 30`, [trustedIds, ctx.userId] ); - const unfamiliarArtistIds = (unfamiliarRes.rows as { artist_id: string }[]).map(r => r.artist_id); + const unfamiliarRows = unfamiliarRes.rows as { artist_id: string; edge_strength: number }[]; + const unfamiliarArtistIds = unfamiliarRows.map(r => r.artist_id); + const edgeStrengthMap = new Map(unfamiliarRows.map(r => [r.artist_id, r.edge_strength])); if (unfamiliarArtistIds.length === 0) return []; const trackRes = await db.pgClient.query( - `SELECT id FROM ( - SELECT DISTINCT t.id + `SELECT id, artist_id FROM ( + SELECT DISTINCT ON (t.id) t.id, cf.object_id AS artist_id FROM tracks t JOIN claim_fusion cf ON cf.subject_type = 'track' AND cf.subject_id = t.id AND cf.predicate IN ('credited_main_on', 'featured_on') @@ -208,25 +211,29 @@ async function discoveryGenerator(db: DbService, ctx: GeneratorContext): Promise AND cf.object_id = ANY($1::uuid[]) WHERE t.state = 'LIBRARY' AND NOT (t.id = ANY($2::uuid[])) + ORDER BY t.id, cf.fused_value DESC NULLS LAST ) sub ORDER BY RANDOM() LIMIT $3`, [unfamiliarArtistIds, ctx.recentExclusions, maxCandidates] ); - return (trackRes.rows as { id: string }[]).map(row => ({ - trackId: row.id, - generatorId: 'discovery', - explanation: [{ - subjectType: 'artist', - subjectId: unfamiliarArtistIds[0], - predicate: 'credited_main_on', - objectType: 'track', - objectId: row.id, - fusedValue: 0.4, - }], - relevance: 0.4, - })); + return (trackRes.rows as { id: string; artist_id: string }[]).map(row => { + const relevance = edgeStrengthMap.get(row.artist_id) ?? 0.4; + return { + trackId: row.id, + generatorId: 'discovery', + explanation: [{ + subjectType: 'artist', + subjectId: row.artist_id, + predicate: 'credited_main_on', + objectType: 'track', + objectId: row.id, + fusedValue: relevance, + }], + relevance, + }; + }); } // --------------------------------------------------------------------------- @@ -253,6 +260,11 @@ async function deepDiveGenerator(db: DbService, ctx: GeneratorContext): Promise< const albumIds = albumRows.map(a => a.album_id); const albumArtistMap = new Map(albumRows.map(a => [a.album_id, a.artist_id])); + const obsessionValueMap = new Map( + ctx.beliefs + .filter(b => b.profile === 'obsession' && b.entity_type === 'artist' && b.dimension === 'affinity' && b.value > 0.3) + .map(b => [b.entity_id, b.value]) + ); const trackRes = await db.pgClient.query( `SELECT sub.id, sub.album_id @@ -269,19 +281,23 @@ async function deepDiveGenerator(db: DbService, ctx: GeneratorContext): Promise< [albumIds, ctx.recentExclusions] ); - return (trackRes.rows as { id: string; album_id: string }[]).map(row => ({ - trackId: row.id, - generatorId: 'deep-dive', - explanation: [{ - subjectType: 'artist', - subjectId: albumArtistMap.get(row.album_id) ?? 'unknown', - predicate: 'credited_main_on', - objectType: 'track', - objectId: row.id, - fusedValue: 0.7, - }], - relevance: 0.7, - })); + return (trackRes.rows as { id: string; album_id: string }[]).map(row => { + const artistId = albumArtistMap.get(row.album_id) ?? 'unknown'; + const relevance = obsessionValueMap.get(artistId) ?? 0.7; + return { + trackId: row.id, + generatorId: 'deep-dive', + explanation: [{ + subjectType: 'artist', + subjectId: artistId, + predicate: 'credited_main_on', + objectType: 'track', + objectId: row.id, + fusedValue: relevance, + }], + relevance, + }; + }); } // --------------------------------------------------------------------------- @@ -306,10 +322,11 @@ async function revivalGenerator(db: DbService, ctx: GeneratorContext): Promise a.artist_id); + const affinityMap = new Map(staleArtists.map(a => [a.artist_id, a.affinity])); const trackRes = await db.pgClient.query( - `SELECT id FROM ( - SELECT DISTINCT t.id + `SELECT id, artist_id FROM ( + SELECT DISTINCT ON (t.id) t.id, cf.object_id AS artist_id FROM tracks t JOIN claim_fusion cf ON cf.subject_type = 'track' AND cf.subject_id = t.id AND cf.predicate IN ('credited_main_on', 'featured_on') @@ -318,25 +335,29 @@ async function revivalGenerator(db: DbService, ctx: GeneratorContext): Promise ({ - trackId: row.id, - generatorId: 'revival', - explanation: [{ - subjectType: 'artist', - subjectId: staleArtistIds[0], - predicate: 'credited_main_on', - objectType: 'track', - objectId: row.id, - fusedValue: 0.6, - }], - relevance: 0.6, - })); + return (trackRes.rows as { id: string; artist_id: string }[]).map(row => { + const relevance = affinityMap.get(row.artist_id) ?? 0.6; + return { + trackId: row.id, + generatorId: 'revival', + explanation: [{ + subjectType: 'artist', + subjectId: row.artist_id, + predicate: 'credited_main_on', + objectType: 'track', + objectId: row.id, + fusedValue: relevance, + }], + relevance, + }; + }); } // --------------------------------------------------------------------------- @@ -455,15 +476,15 @@ async function contextualGenerator(db: DbService, ctx: GeneratorContext): Promis order: 'DESC', }); - const targetArtistIds = contextualBeliefs - .filter(b => b.entity_type === 'artist' && b.value > 0.2) - .map(b => b.entity_id); + const targetBeliefs = contextualBeliefs.filter(b => b.entity_type === 'artist' && b.value > 0.2); + const targetArtistIds = targetBeliefs.map(b => b.entity_id); + const targetValueMap = new Map(targetBeliefs.map(b => [b.entity_id, b.value])); if (targetArtistIds.length === 0) return []; const trackRes = await db.pgClient.query( - `SELECT id FROM ( - SELECT DISTINCT t.id + `SELECT id, artist_id FROM ( + SELECT DISTINCT ON (t.id) t.id, cf.object_id AS artist_id FROM tracks t JOIN claim_fusion cf ON cf.subject_type = 'track' AND cf.subject_id = t.id AND cf.predicate IN ('credited_main_on', 'featured_on') @@ -472,25 +493,29 @@ async function contextualGenerator(db: DbService, ctx: GeneratorContext): Promis AND (cf.user_id = $2 OR cf.user_id = $3) WHERE t.state = 'LIBRARY' AND NOT (t.id = ANY($4::uuid[])) + ORDER BY t.id, cf.fused_value DESC NULLS LAST ) sub ORDER BY RANDOM() LIMIT 15`, [targetArtistIds, OBJECTIVE_USER, ctx.userId, ctx.recentExclusions] ); - return (trackRes.rows as { id: string }[]).map(row => ({ - trackId: row.id, - generatorId: 'contextual', - explanation: [{ - subjectType: 'artist', - subjectId: targetArtistIds[0], - predicate: 'credited_main_on', - objectType: 'track', - objectId: row.id, - fusedValue: 0.5, - }], - relevance: 0.5, - })); + return (trackRes.rows as { id: string; artist_id: string }[]).map(row => { + const relevance = targetValueMap.get(row.artist_id) ?? 0.5; + return { + trackId: row.id, + generatorId: 'contextual', + explanation: [{ + subjectType: 'artist', + subjectId: row.artist_id, + predicate: 'credited_main_on', + objectType: 'track', + objectId: row.id, + fusedValue: relevance, + }], + relevance, + }; + }); } // --------------------------------------------------------------------------- diff --git a/backend/src/services/session-director.service.ts b/backend/src/services/session-director.service.ts index 722954e..ce66ef8 100644 --- a/backend/src/services/session-director.service.ts +++ b/backend/src/services/session-director.service.ts @@ -28,6 +28,11 @@ export interface DiversityBudget { spent: number; } +export interface RepetitionState { + recentTrackIds: Set; + recentArtistIds: Set; +} + const W_ENJOY = 1.0; const W_FATIGUE = 0.4; const W_DIVERSITY = 0.3; @@ -152,7 +157,7 @@ export class SessionDirector { // D.2 — Fatigue model // --------------------------------------------------------------- async computeFatigue(userId: string): Promise { - // Track fatigue: last 7 days, decay half-life 30d (2592000 seconds) + // Track fatigue: last 7 days, decay time constant 30d (e-folding; half-life ≈ 20.8d) const TRACK_DECAY_SEC = 30 * 24 * 3600; const trackRes = await this.db.pgClient.query( `SELECT ph.track_id, @@ -167,7 +172,8 @@ export class SessionDirector { track.set(row.track_id, row.fatigue); } - // Artist fatigue: last 24h, decay half-life 8h (28800 seconds) + // Artist fatigue: last 24h, decay time constant 8h (28800s) — this is an e-folding + // time (EXP(-t/tau)), not a half-life; the actual half-life is tau*ln(2) ≈ 5.5h const ARTIST_DECAY_SEC = 8 * 3600; const artistRes = await this.db.pgClient.query( `SELECT ta.artist_id, @@ -183,7 +189,7 @@ export class SessionDirector { artist.set(row.artist_id, row.fatigue); } - // Genre fatigue: last 24h, decay half-life 8h + // Genre fatigue: last 24h, decay time constant 8h (e-folding, not half-life; half-life ≈ 5.5h) const genreRes = await this.db.pgClient.query( `SELECT tg.genre_id, LEAST(1.0, SUM(EXP(-EXTRACT(EPOCH FROM (NOW() - ph.played_at)) / $2::float8))) AS fatigue @@ -198,7 +204,7 @@ export class SessionDirector { genre.set(row.genre_id, row.fatigue); } - // Language fatigue: last 2h, decay half-life 1h (3600 seconds) + // Language fatigue: last 2h, decay time constant 1h (3600s, e-folding; half-life ≈ 0.7h) const LANG_DECAY_SEC = 3600; const langRes = await this.db.pgClient.query( `SELECT tl.language, @@ -427,14 +433,25 @@ export class SessionDirector { // --------------------------------------------------------------- // D.5 — Entropy, anti-loop // --------------------------------------------------------------- - computeEntropy(candidates: Candidate[]): number { + // NOTE: despite the name, this computes the Herfindahl-Hirschman Index (artist + // concentration, 0 = maximally diverse, 1 = single artist) — not entropy. + // `artistIdOf` should resolve the candidate's actual attributed artist; without it + // this falls back to guessing from the first artist-typed explanation edge, which + // for some generators (discovery/revival/contextual) isn't the real artist — + // pass a resolver when a real artist map is available (see rankCandidates). + computeEntropy(candidates: Candidate[], artistIdOf?: (c: Candidate) => string): number { if (candidates.length === 0) return 0; const artistCounts = new Map(); for (const c of candidates) { - const mainEdge = c.explanation.find( - e => e.subjectType === 'artist' || e.objectType === 'artist' - ); - const key = mainEdge?.subjectId ?? mainEdge?.objectId ?? 'unknown'; + let key: string; + if (artistIdOf) { + key = artistIdOf(c); + } else { + const mainEdge = c.explanation.find( + e => e.subjectType === 'artist' || e.objectType === 'artist' + ); + key = mainEdge?.subjectId ?? mainEdge?.objectId ?? 'unknown'; + } artistCounts.set(key, (artistCounts.get(key) ?? 0) + 1); } const n = candidates.length; @@ -599,6 +616,61 @@ export class SessionDirector { return false; } + // Batched version of checkRepetition for ranking a whole candidate pool: + // loads repetition_rules once, then one query for recently-played + // tracks/artists within the max window, and checks membership in JS + // instead of 2-3 sequential queries per candidate. + async buildRepetitionState(userId: string): Promise { + const rulesRes = await this.db.pgClient.query( + 'SELECT dimension, min_distance FROM repetition_rules WHERE user_id = $1', + [userId] + ); + const ruleMap = new Map(); + for (const row of rulesRes.rows as { dimension: string; min_distance: number }[]) { + ruleMap.set(row.dimension, row.min_distance); + } + const trackMin = ruleMap.get('track') ?? 120; + const artistMin = ruleMap.get('artist') ?? 20; + + const recentTrackIds = new Set(); + const recentArtistIds = new Set(); + const maxMin = Math.max(trackMin, artistMin); + if (maxMin > 0) { + const res = await this.db.pgClient.query( + `SELECT ph.track_id, ta.artist_id, ph.played_at + FROM play_history ph + LEFT JOIN track_artists_v2 ta ON ta.track_id = ph.track_id AND ta.role = 'main' + WHERE ph.user_id = $1 AND ph.completed = true + AND ph.played_at > NOW() - ($2 || ' minutes')::interval`, + [userId, String(maxMin)] + ); + const now = Date.now(); + for (const row of res.rows as { track_id: string; artist_id: string | null; played_at: Date }[]) { + const ageMin = (now - new Date(row.played_at).getTime()) / 60000; + if (trackMin > 0 && ageMin <= trackMin) recentTrackIds.add(row.track_id); + if (artistMin > 0 && row.artist_id && ageMin <= artistMin) recentArtistIds.add(row.artist_id); + } + } + + return { recentTrackIds, recentArtistIds }; + } + + // Shared track_id -> main artist_id lookup, used by rankCandidates and replan. + private async loadArtistMap(trackIds: string[]): Promise> { + const artistMap = new Map(); + if (trackIds.length === 0) return artistMap; + const artRes = await this.db.pgClient.query( + `SELECT DISTINCT ON (ta.track_id) ta.track_id, ta.artist_id + FROM track_artists_v2 ta + WHERE ta.track_id = ANY($1::uuid[]) AND ta.role = 'main'`, + [trackIds] + ); + for (const row of artRes.rows as { track_id: string; artist_id: string }[]) { + artistMap.set(row.track_id, row.artist_id); + } + return artistMap; + } + // --------------------------------------------------------------- // D.8 — Multi-objective ranking // --------------------------------------------------------------- @@ -607,23 +679,12 @@ export class SessionDirector { fatigue: FatigueState, budgets: DiversityBudget[], state: GeneratorContext['state'], - repetitionCheck: (trackId: string, artistId: string) => Promise + repetitionState: RepetitionState ): Promise { if (candidates.length === 0) return []; const trackIds = [...new Set(candidates.map(c => c.trackId))]; - const artistMap = new Map(); - if (trackIds.length > 0) { - const artRes = await this.db.pgClient.query( - `SELECT DISTINCT ON (ta.track_id) ta.track_id, ta.artist_id - FROM track_artists_v2 ta - WHERE ta.track_id = ANY($1::uuid[]) AND ta.role = 'main'`, - [trackIds] - ); - for (const row of artRes.rows as { track_id: string; artist_id: string }[]) { - artistMap.set(row.track_id, row.artist_id); - } - } + const artistMap = await this.loadArtistMap(trackIds); const genreMap = new Map(); if (trackIds.length > 0) { @@ -639,10 +700,17 @@ export class SessionDirector { } } - const artistBudget = budgets.find(b => b.dimension === 'artist'); - const currentEntropy = this.computeEntropy(candidates); + const currentEntropy = this.computeEntropy(candidates, c => artistMap.get(c.trackId) ?? 'unknown'); const targetEntropy = 0.55; + // Per-candidate artist share within this batch, for a real per-candidate + // entropy contribution instead of the batch-wide constant. + const artistBatchCounts = new Map(); + for (const c of candidates) { + const aid = artistMap.get(c.trackId) ?? ''; + artistBatchCounts.set(aid, (artistBatchCounts.get(aid) ?? 0) + 1); + } + const scored: { candidate: Candidate; score: number }[] = []; for (const c of candidates) { const artistId = artistMap.get(c.trackId) ?? ''; @@ -653,10 +721,14 @@ export class SessionDirector { const genreFatigue = fatigue.genre.get(genreId) ?? 0; const avgFatigue = (trackFatigue + artistFatigue + genreFatigue) / 3; - const artistSpendRatio = artistBudget ? artistBudget.spent : 0; - const diversityBonus = 1 - artistSpendRatio; - const entropyBonus = 1 - Math.abs(currentEntropy - targetEntropy); - const wouldRepeat = await repetitionCheck(c.trackId, artistId); + // diversityBonus: this artist's own fatigue-weighted share — varies per candidate. + const diversityBonus = 1 - artistFatigue; + // entropyBonus: reward candidates whose artist is underrepresented in this batch. + const artistShare = (artistBatchCounts.get(artistId) ?? 0) / candidates.length; + const entropyBonus = 1 - artistShare; + const wouldRepeat = + repetitionState.recentTrackIds.has(c.trackId) || + (!!artistId && repetitionState.recentArtistIds.has(artistId)); let score = W_ENJOY * c.relevance - W_FATIGUE * avgFatigue @@ -687,6 +759,27 @@ export class SessionDirector { return scored.map(s => s.candidate); } + // session_state is otherwise only written once at /v2/vibe/start — persist the + // freshly-computed state vector here so it evolves across the session instead of + // buildState always reading back the boot defaults. + async persistState(sessionId: string, userId: string, state: GeneratorContext['state']): Promise { + await this.db.pgClient.query( + `UPDATE session_state + SET state_vector = $3::jsonb, last_interaction = NOW() + WHERE session_id = $1 AND user_id = $2`, + [ + sessionId, + userId, + JSON.stringify({ + energy: state.energy, + lastArtistIds: state.lastArtistIds, + lastGenreIds: state.lastGenreIds, + noveltyHunger: state.noveltyHunger, + }), + ] + ); + } + // --------------------------------------------------------------- // D.9 — Plan + replan loop // --------------------------------------------------------------- @@ -725,6 +818,7 @@ export class SessionDirector { })); const state = await this.buildState(userId, sessionId); + await this.persistState(sessionId, userId, state); const fatigue = await this.computeFatigue(userId); const budgets = await this.getBudgets(userId); @@ -737,7 +831,7 @@ export class SessionDirector { seedArtistId = await this.resolveSeedArtistId(seedTrackId) ?? null; } - const recentExclusions: string[] = []; + const recentExclusions: string[] = recentPlays.map(p => p.trackId); const toleranceMap: Record = {}; const discoveryBeliefs = allBeliefs.filter(b => b.profile === 'discovery'); for (const b of discoveryBeliefs) { @@ -764,28 +858,17 @@ export class SessionDirector { return []; } - const repetitionCheckFn = (tid: string, aid: string) => - this.checkRepetition(tid, aid, userId); + const repetitionState = await this.buildRepetitionState(userId); const ranked = await this.rankCandidates( - allCandidates, fatigue, budgets, state, repetitionCheckFn + allCandidates, fatigue, budgets, state, repetitionState ); const loopDim = await this.detectAntiLoop(state, fatigue, budgets, recentPlays); let forcedExperimental = false; if (loopDim && ranked.length > 0) { - const expCtx: GeneratorContext = { - ...ctx, - recentExclusions: ctx.recentExclusions.slice(0, Math.min(ctx.recentExclusions.length, 50)), - }; - const extraCandidates: Candidate[] = []; - for (const gen of ALL_GENERATORS) { - const result = await gen(this.db, expCtx); - extraCandidates.push(...result); - } - const expRanked = await this.rankCandidates( - extraCandidates, fatigue, budgets, state, repetitionCheckFn - ); - const injected = expRanked.filter( + // Anti-loop candidates are already present in `ranked` — just pull them to the + // front instead of re-running all generators and re-ranking from scratch. + const injected = ranked.filter( c => c.generatorId === 'experimental' || c.generatorId === 'discovery' ); ranked.unshift(...injected); @@ -855,6 +938,7 @@ export class SessionDirector { const fatigue = await this.computeFatigue(userId); const budgets = await this.getBudgets(userId); const state = await this.buildState(userId, sessionId); + await this.persistState(sessionId, userId, state); // Fetch recent plays for anti-loop const recentPlaysRes = await this.db.pgClient.query( @@ -892,7 +976,8 @@ export class SessionDirector { return this.buildPlan(userId, sessionId, seedTrackId); } - const entropy = this.computeEntropy(currentPlan); + const planArtistMap = await this.loadArtistMap([...new Set(currentPlan.map(c => c.trackId))]); + const entropy = this.computeEntropy(currentPlan, c => planArtistMap.get(c.trackId) ?? 'unknown'); if (Math.abs(entropy - 0.55) > 0.2) { return this.buildPlan(userId, sessionId, seedTrackId); } diff --git a/backend/src/services/session-director.test.ts b/backend/src/services/session-director.test.ts index d2c8897..6d99d97 100644 --- a/backend/src/services/session-director.test.ts +++ b/backend/src/services/session-director.test.ts @@ -95,7 +95,8 @@ describe('SessionDirector', () => { const budgets = [{ dimension: 'artist', budgetShare: 0.2, horizonMin: 30, spent: 0 }]; const state = { energy: 0.5, noveltyHunger: 0.3, sessionAgeMin: 10, lastArtistIds: [], lastGenreIds: [], context: null }; - const ranked = await director.rankCandidates(candidates, fatigue, budgets, state, async () => false); + const repetitionState = { recentTrackIds: new Set(), recentArtistIds: new Set() }; + const ranked = await director.rankCandidates(candidates, fatigue, budgets, state, repetitionState); expect(ranked[0].relevance).toBeGreaterThanOrEqual(ranked[ranked.length - 1].relevance); }); }); diff --git a/frontend/src/components/AudioEngine.tsx b/frontend/src/components/AudioEngine.tsx index 43c1c6b..ddef83a 100644 --- a/frontend/src/components/AudioEngine.tsx +++ b/frontend/src/components/AudioEngine.tsx @@ -1,9 +1,18 @@ import { useEffect, useRef } from 'react'; import { usePlaybackStore } from '../store/usePlaybackStore'; +import { useVibeStore } from '../store/useVibeStore'; import { trackService } from '../services/trackService'; import { vibeService } from '../services/vibeService'; import type { Track } from '../types'; +// Track ids whose next natural feedback transition should be skipped because +// the caller (e.g. Vibe.tsx's dislike button) already recorded feedback for +// them explicitly. Consumed once, then cleared. +const suppressedFeedbackIds = new Set(); +export function suppressAutoFeedback(trackId: string): void { + suppressedFeedbackIds.add(trackId); +} + // Threshold (seconds) above which a store position change is treated as a user // scrub and applied to the audio element. Keeps the timeupdate -> setPosition -> // effect loop from fighting itself. @@ -39,9 +48,6 @@ export const AudioEngine = () => { const endedNaturallyRef = useRef(false); // Track whether the current track has crossed the completion threshold. const crossedThresholdRef = useRef(false); - // Track whether we've already recorded a completed play for the current track - // (to avoid double-recording when both threshold crossed AND ended fires). - const recordedCompletedRef = useRef(false); // --- DOM -> store: media events ----------------------------------------- useEffect(() => { @@ -73,16 +79,10 @@ export const AudioEngine = () => { if (store().isPlaying) store().pause(); }; const onEnded = () => { - const trackId = loadedIdRef.current; - if (trackId && !recordedCompletedRef.current) { - endedNaturallyRef.current = true; - recordedCompletedRef.current = true; - try { - void vibeService.feedback(trackId, 'completed').catch(() => {}); - } catch { - /* best-effort */ - } - } + // Just flag it — applyTrack (below) is the single place that sends + // feedback, on the resulting track-change, so completion is recorded + // exactly once per track. + endedNaturallyRef.current = true; store().next(); }; @@ -114,21 +114,23 @@ export const AudioEngine = () => { // If it crossed the threshold OR ended naturally, record as completed. const prevId = loadedIdRef.current; const completed = endedNaturallyRef.current || crossedThresholdRef.current; - if (prevId) { - try { - if (completed) { - recordedCompletedRef.current = true; - void vibeService.feedback(prevId, 'completed').catch(() => {}); - } else { - void vibeService.feedback(prevId, 'skipped').catch(() => {}); + // Only vibe sessions want this feedback — plain library browsing + // shouldn't write skip/completed evidence for tracks merely sampled. + const inVibeSession = !!useVibeStore.getState().activeSessionId; + if (prevId && inVibeSession) { + if (suppressedFeedbackIds.delete(prevId)) { + // Caller already recorded explicit feedback (e.g. dislike) for + // this track — don't also record the implicit transition. + } else { + try { + void vibeService.feedback(prevId, completed ? 'completed' : 'skipped').catch(() => {}); + } catch { + /* best-effort */ } - } catch { - /* best-effort */ } } endedNaturallyRef.current = false; crossedThresholdRef.current = false; - recordedCompletedRef.current = false; loadedIdRef.current = id; if (!id) { diff --git a/frontend/src/pages/Vibe.tsx b/frontend/src/pages/Vibe.tsx index d307497..0f2d9d7 100644 --- a/frontend/src/pages/Vibe.tsx +++ b/frontend/src/pages/Vibe.tsx @@ -9,6 +9,8 @@ import { TrackRow } from '../components/TrackRow'; import { PageContainer } from '../components/PageContainer'; import type { Track } from '../types'; import { VibeTimeline } from '../components/VibeTimeline'; +import { suppressAutoFeedback } from '../components/AudioEngine'; +import { toast } from '../store/useToastStore'; const INITIAL_BATCH_SIZE = 5; const PREFETCH_THRESHOLD = 3; @@ -19,7 +21,7 @@ function bestEffort(p: Promise): void { } export default function Vibe() { - const { currentTrack, queue, setQueue, playTrack, next: playNext } = usePlaybackStore(); + const { currentTrack, queue, setQueue, playTrack, next: playNext, pause, setCurrentTrack } = usePlaybackStore(); const { activeSessionId, buffer, @@ -117,23 +119,34 @@ export default function Vibe() { if (idx > 0) { setBuffer(buffer.slice(idx)); } - }, [activeSessionId, currentTrack]); + }, [activeSessionId, currentTrack, buffer, setBuffer]); const handleKeep = useCallback(() => { - if (currentTrack) bestEffort(vibeService.feedback(currentTrack.id, 'promoted')); + if (currentTrack) { + bestEffort(vibeService.feedback(currentTrack.id, 'promoted')); + toast.success(`Kept "${currentTrack.title}"`); + } }, [currentTrack]); const handleDislike = useCallback(() => { - if (currentTrack) bestEffort(vibeService.feedback(currentTrack.id, 'disliked')); + if (currentTrack) { + bestEffort(vibeService.feedback(currentTrack.id, 'disliked')); + // AudioEngine would otherwise also record a 'skipped' on the track + // change caused by playNext() below — suppress that duplicate. + suppressAutoFeedback(currentTrack.id); + } playNext(); }, [currentTrack, playNext]); const handleEnd = useCallback(() => { // V2 plan expires via Redis TTL (2h). No explicit end endpoint. + pause(); + setQueue([]); + setCurrentTrack(null); reset(); setEmpty(false); setError(null); - }, [reset]); + }, [reset, pause, setQueue, setCurrentTrack]); const upcoming = currentTrack ? (() => { @@ -208,12 +221,12 @@ export default function Vibe() {

Or pick a seed track

    - {libraryTracks.map((track) => ( + {libraryTracks.map((track, index) => (
  • t.id === track.id)} + index={index} showActions={false} />
    • diff --git a/frontend/src/store/usePlaybackStore.ts b/frontend/src/store/usePlaybackStore.ts index 608928f..3fa2738 100644 --- a/frontend/src/store/usePlaybackStore.ts +++ b/frontend/src/store/usePlaybackStore.ts @@ -12,6 +12,8 @@ interface PlaybackState { volume: number; shuffle: boolean; repeat: RepeatMode; + /** Ids already played this shuffle "lap" (repeat-all), to avoid bouncing between the same few tracks. */ + shufflePlayed: Set; setQueue: (queue: Track[]) => void; playTrack: (track: Track) => void; @@ -36,8 +38,9 @@ export const usePlaybackStore = create((set, get) => ({ volume: 1, shuffle: false, repeat: 'none', + shufflePlayed: new Set(), - setQueue: (queue) => set({ queue }), + setQueue: (queue) => set({ queue, shufflePlayed: new Set() }), playTrack: (track) => set({ @@ -45,13 +48,14 @@ export const usePlaybackStore = create((set, get) => ({ isPlaying: true, position: 0, duration: track.duration ?? 0, + shufflePlayed: new Set(), }), play: () => set({ isPlaying: true }), pause: () => set({ isPlaying: false }), next: () => { - const { queue, currentTrack, shuffle, repeat } = get(); + const { queue, currentTrack, shuffle, repeat, shufflePlayed } = get(); if (queue.length === 0) return; const idx = currentTrack ? queue.findIndex((t) => t.id === currentTrack.id) : -1; @@ -63,17 +67,22 @@ export const usePlaybackStore = create((set, get) => ({ } if (shuffle) { - // Shuffle: pick a random track from the remaining queue (excluding current) - const remaining = queue.filter((t) => t.id !== currentTrack?.id); + // Shuffle: pick a random track not yet played this lap (excludes current), + // so repeat-all doesn't bounce between the same few tracks. + const played = new Set(shufflePlayed); + if (currentTrack) played.add(currentTrack.id); + let remaining = queue.filter((t) => !played.has(t.id)); if (remaining.length === 0) { - if (repeat === 'all') { - const pick = queue[Math.floor(Math.random() * queue.length)]; - set({ currentTrack: pick, position: 0, duration: pick.duration ?? 0, isPlaying: true }); - } - return; + if (repeat !== 'all') return; + // Lap complete — start a fresh one. + played.clear(); + if (currentTrack) played.add(currentTrack.id); + remaining = queue.filter((t) => t.id !== currentTrack?.id); + if (remaining.length === 0) return; } const pick = remaining[Math.floor(Math.random() * remaining.length)]; - set({ currentTrack: pick, position: 0, duration: pick.duration ?? 0, isPlaying: true }); + played.add(pick.id); + set({ currentTrack: pick, shufflePlayed: played, position: 0, duration: pick.duration ?? 0, isPlaying: true }); return; }