feat(discovery): acquire recommendations that keep their names
Acquisition ran yt-dlp without --embed-metadata, so every download arrived untagged. The scanner then stored the video id as the title and "Unknown Artist" as the artist, the vetted-candidate tag check rejected the mismatch, and all 18 acquired tracks were hidden and retired. - Pass --embed-metadata so downloads carry real tags. - Let a scan take fallback title/artist from the candidate, for sources that still ship untagged files. - Install Deno alongside yt-dlp: YouTube guards some formats with a JS challenge yt-dlp must execute, and no other runtime is enabled. - Dedupe candidates by artist and title. The (source, external_id) key misses the same song reaching us under two Deezer release ids. Also carries the in-flight discovery work this builds on: the Recommendations page replacing Discover, the discovery source service, and the acquisition spec tests. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
kami
@@ -18,6 +18,11 @@ import vibeSessionsRoutes from './routes/vibe-sessions.routes.js';
|
||||
import discoveryRoutes from './routes/discovery.routes.js';
|
||||
import imagesRoutes from './routes/images.routes.js';
|
||||
import { VibeSessionCoordinator } from './services/vibe-session-coordinator.service.js';
|
||||
import { DiscoveryService } from './services/discovery.service.js';
|
||||
|
||||
// Single-listener deployment: the same placeholder the vibe-session routes use
|
||||
// when no x-user-id header is supplied.
|
||||
const DEFAULT_USER_ID = '00000000-0000-0000-0000-000000000000';
|
||||
|
||||
export interface AppConfig {
|
||||
port: number;
|
||||
@@ -91,6 +96,34 @@ export async function buildApp(config: AppConfig) {
|
||||
dbService.decayBeliefs().catch(() => {});
|
||||
dbService.deriveForgottenProfile().catch(() => {});
|
||||
|
||||
// Close the discovery loop without an operator in it. The worker's cron jobs
|
||||
// generate candidates; this timer is what decides which of them earn a
|
||||
// probation slot and enqueues the download. Evaluation is cheap SQL against a
|
||||
// capped batch, and every acquisition gate (relevance, novelty tolerance,
|
||||
// backlog, per-artist diversity) lives inside evalCandidates.
|
||||
const discoveryService = new DiscoveryService(dbService);
|
||||
const DISCOVERY_EVAL_INTERVAL_MS = 30 * 60 * 1000;
|
||||
const runDiscoveryEval = async () => {
|
||||
const results = await discoveryService.evalCandidates(DEFAULT_USER_ID);
|
||||
let enqueued = 0;
|
||||
for (const result of results) {
|
||||
if (!result.shouldAcquire) continue;
|
||||
try {
|
||||
await jobService.enqueueDiscoveryAcquisition(result.candidateId);
|
||||
enqueued++;
|
||||
} catch (err) {
|
||||
const reason = err instanceof Error ? err.message : 'failed to enqueue acquisition';
|
||||
await discoveryService.markEnqueueFailed(result.candidateId, reason);
|
||||
}
|
||||
}
|
||||
if (results.length > 0) {
|
||||
console.log(`[Discovery] Evaluated ${results.length} candidates, enqueued ${enqueued}`);
|
||||
}
|
||||
};
|
||||
const discoveryEvalTimer = setInterval(() => {
|
||||
runDiscoveryEval().catch((e) => console.error('[Discovery] eval failed:', e));
|
||||
}, DISCOVERY_EVAL_INTERVAL_MS);
|
||||
|
||||
// Ensure the Typesense 'tracks' collection schema exists on boot so that
|
||||
// the first search request doesn't hit a 404.
|
||||
await searchService.ensureCollection();
|
||||
@@ -176,6 +209,7 @@ export async function buildApp(config: AppConfig) {
|
||||
clearInterval(fusionTimer);
|
||||
clearInterval(decayTimer);
|
||||
clearInterval(forgottenTimer);
|
||||
clearInterval(discoveryEvalTimer);
|
||||
} catch (err) {
|
||||
fastify.log.error(err);
|
||||
}
|
||||
|
||||
@@ -759,4 +759,64 @@ export const MIGRATIONS: Migration[] = [
|
||||
ON CONFLICT (session_id) DO NOTHING;
|
||||
`,
|
||||
},
|
||||
{
|
||||
// Two external candidate strategies, kept as separate source_trust keys so
|
||||
// meta-learning can compare their retention independently: "artists you
|
||||
// already play just released something" is a much stronger prior than
|
||||
// "Last.fm thinks this sounds similar", and the trust values say so.
|
||||
id: '20260806_external_discovery_sources',
|
||||
sql: `
|
||||
INSERT INTO source_trust (key, trust, description) VALUES
|
||||
('new_release', 0.60,
|
||||
'New release by an artist already played from the local library.'),
|
||||
('similar_recommendation', 0.45,
|
||||
'External similarity (Last.fm) seeded from local play history.')
|
||||
ON CONFLICT (key) DO NOTHING;
|
||||
`,
|
||||
},
|
||||
{
|
||||
// play_history is the only durable record of what was listened to and when,
|
||||
// and it had two holes that only show up when you try to read a year back:
|
||||
//
|
||||
// 1. ON DELETE CASCADE meant the gated cleanup sweep silently erased the
|
||||
// plays of every file it removed. A play happened; deleting the file
|
||||
// later does not un-happen it. The FK becomes SET NULL and the track's
|
||||
// identity is denormalised onto the row so it stays readable.
|
||||
// 2. No duration, so listening time was only ever inferable from the
|
||||
// track's current duration — itself gone once the file is.
|
||||
id: '20260806_play_history_durable_facts',
|
||||
sql: `
|
||||
ALTER TABLE play_history ADD COLUMN IF NOT EXISTS listened_ms INTEGER;
|
||||
ALTER TABLE play_history ADD COLUMN IF NOT EXISTS track_title TEXT;
|
||||
ALTER TABLE play_history ADD COLUMN IF NOT EXISTS track_artist TEXT;
|
||||
|
||||
DO $$
|
||||
DECLARE fk_name TEXT;
|
||||
BEGIN
|
||||
SELECT con.conname INTO fk_name
|
||||
FROM pg_constraint con
|
||||
JOIN pg_class rel ON rel.oid = con.conrelid
|
||||
JOIN pg_attribute att ON att.attrelid = rel.oid AND att.attnum = con.conkey[1]
|
||||
WHERE rel.relname = 'play_history'
|
||||
AND con.contype = 'f'
|
||||
AND att.attname = 'track_id'
|
||||
AND con.confdeltype = 'c'
|
||||
LIMIT 1;
|
||||
IF fk_name IS NOT NULL THEN
|
||||
EXECUTE format('ALTER TABLE play_history DROP CONSTRAINT %I', fk_name);
|
||||
ALTER TABLE play_history
|
||||
ADD CONSTRAINT play_history_track_id_fkey
|
||||
FOREIGN KEY (track_id) REFERENCES tracks(id) ON DELETE SET NULL;
|
||||
END IF;
|
||||
END $$;
|
||||
|
||||
-- Backfill identity for rows written before the columns existed. Rows whose
|
||||
-- track was already cascade-deleted are unrecoverable; this at least stops
|
||||
-- the bleeding from here on.
|
||||
UPDATE play_history ph
|
||||
SET track_title = t.title, track_artist = t.artist
|
||||
FROM tracks t
|
||||
WHERE t.id = ph.track_id AND ph.track_title IS NULL;
|
||||
`,
|
||||
},
|
||||
];
|
||||
|
||||
@@ -302,12 +302,20 @@ CREATE TABLE IF NOT EXISTS recommendation_batch_track (
|
||||
-- "Success-Driven Center" rule (a completed play moves the active batch's center)
|
||||
-- and the feedback learning loop. Created with IF NOT EXISTS so it can be
|
||||
-- self-provisioned on databases that predate this schema change.
|
||||
-- A play is a historical fact: it stays true after the file is gone. Hence
|
||||
-- ON DELETE SET NULL rather than CASCADE, plus the denormalised title/artist so
|
||||
-- a row whose track was hard-deleted is still readable. listened_ms is what was
|
||||
-- actually heard, recorded at play time; tracks.duration is not a substitute
|
||||
-- because it disappears with the track.
|
||||
CREATE TABLE IF NOT EXISTS play_history (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
user_id UUID NOT NULL,
|
||||
track_id UUID REFERENCES tracks(id) ON DELETE CASCADE,
|
||||
track_id UUID REFERENCES tracks(id) ON DELETE SET NULL,
|
||||
batch_id UUID REFERENCES recommendation_batch(id) ON DELETE SET NULL,
|
||||
completed BOOLEAN NOT NULL DEFAULT false,
|
||||
listened_ms INTEGER,
|
||||
track_title TEXT,
|
||||
track_artist TEXT,
|
||||
played_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
|
||||
|
||||
@@ -35,6 +35,54 @@ export default async function discoveryRoutes(fastify: FastifyInstance, options:
|
||||
return reply.send({ candidates: res.rows });
|
||||
});
|
||||
|
||||
/**
|
||||
* GET /api/discovery/overview — every recommendation and what became of it.
|
||||
*
|
||||
* One row per candidate, acquired or not, so a stalled candidate is as
|
||||
* visible as a retained track. The play/skip counts come from `evidence`
|
||||
* rather than tracks.play_count because the probation sweep decides on
|
||||
* exactly those two signals: showing anything else would be showing a number
|
||||
* that does not drive the outcome.
|
||||
*/
|
||||
fastify.get('/discovery/overview', async (request, reply) => {
|
||||
const query = request.query as { limit?: string };
|
||||
const limit = Math.min(Math.max(parseInt(query.limit || '200', 10) || 200, 1), 500);
|
||||
|
||||
const rows = await dbService.pgClient.query(
|
||||
`SELECT dc.id, dc.source, dc.status, dc.title, dc.artist_credit, dc.notes,
|
||||
dc.first_seen_at, dc.acquired_at, dc.last_error, dc.acquisition_attempts,
|
||||
t.id AS track_id, t.title AS track_title, t.artist AS track_artist,
|
||||
t.probation_status, t.probation_entered_at,
|
||||
(SELECT COUNT(*)::int FROM evidence e
|
||||
WHERE e.entity_type = 'track' AND e.entity_id = t.id
|
||||
AND e.signal = 'playback_completed') AS completed_plays,
|
||||
(SELECT COUNT(*)::int FROM evidence e
|
||||
WHERE e.entity_type = 'track' AND e.entity_id = t.id
|
||||
AND e.signal = 'skip_quick') AS quick_skips
|
||||
FROM discovery_candidates dc
|
||||
LEFT JOIN tracks t ON t.id = dc.acquired_track_id
|
||||
ORDER BY COALESCE(dc.acquired_at, dc.first_seen_at) DESC
|
||||
LIMIT $1`,
|
||||
[limit]
|
||||
);
|
||||
|
||||
const summary = await dbService.pgClient.query(
|
||||
`SELECT dc.source,
|
||||
COUNT(*)::int AS candidates,
|
||||
COUNT(t.id) FILTER (WHERE t.probation_status = 'probation')::int AS probation,
|
||||
COUNT(t.id) FILTER (WHERE t.probation_status = 'retained')::int AS retained,
|
||||
COUNT(t.id) FILTER (WHERE t.probation_status = 'retired')::int AS retired,
|
||||
COUNT(*) FILTER (WHERE dc.acquired_track_id IS NULL
|
||||
AND dc.status <> 'candidate')::int AS stalled
|
||||
FROM discovery_candidates dc
|
||||
LEFT JOIN tracks t ON t.id = dc.acquired_track_id
|
||||
GROUP BY dc.source
|
||||
ORDER BY dc.source`
|
||||
);
|
||||
|
||||
return reply.send({ rows: rows.rows, summary: summary.rows });
|
||||
});
|
||||
|
||||
/**
|
||||
* POST /api/discovery/eval — evaluate pending candidates for acquisition
|
||||
*/
|
||||
|
||||
@@ -6,16 +6,23 @@ export default async function historyRoutes(fastify: FastifyInstance, options: {
|
||||
|
||||
// Record a playback event. completed defaults to false.
|
||||
fastify.post('/history', async (request, reply) => {
|
||||
const { trackId, completed, batchId } = request.body as {
|
||||
const { trackId, completed, batchId, listenedMs } = request.body as {
|
||||
trackId: string;
|
||||
completed?: boolean;
|
||||
batchId?: string;
|
||||
listenedMs?: number;
|
||||
};
|
||||
if (!trackId) {
|
||||
return reply.code(400).send({ error: 'trackId is required' });
|
||||
}
|
||||
// Clamped rather than rejected: a bogus duration must not cost the caller a
|
||||
// play record, and a 24h ceiling keeps one bad client from dominating any
|
||||
// listening-time total.
|
||||
const listened = typeof listenedMs === 'number' && Number.isFinite(listenedMs)
|
||||
? Math.max(0, Math.min(Math.round(listenedMs), 24 * 60 * 60 * 1000))
|
||||
: undefined;
|
||||
const userId = (request.headers['x-user-id'] as string) || '00000000-0000-0000-0000-000000000000';
|
||||
const historyId = await dbService.recordPlay(userId, trackId, completed === true, batchId);
|
||||
const historyId = await dbService.recordPlay(userId, trackId, completed === true, batchId, listened);
|
||||
return reply.send({ historyId });
|
||||
});
|
||||
|
||||
|
||||
@@ -606,6 +606,32 @@ describe('DbService v2 methods', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('recordPlay durability', () => {
|
||||
it('copies the track identity onto the row and records what was heard', async () => {
|
||||
const { service, mockQuery } = makeService();
|
||||
mockQuery.mockResolvedValue({ rows: [{ id: 'history-1' }] });
|
||||
|
||||
await service.recordPlay('user-1', 'track-1', false, undefined, 91_000);
|
||||
|
||||
const [sql, params] = mockQuery.mock.calls[0];
|
||||
// A LEFT JOIN, not a plain VALUES: the insert must still happen when the
|
||||
// track id resolves to nothing.
|
||||
expect(sql).toContain('LEFT JOIN tracks');
|
||||
expect(sql).toContain('t.title');
|
||||
expect(sql).toContain('t.artist');
|
||||
expect(params).toEqual(['user-1', 'track-1', null, false, 91_000]);
|
||||
});
|
||||
|
||||
it('leaves listened_ms null when the caller has no duration to report', async () => {
|
||||
const { service, mockQuery } = makeService();
|
||||
mockQuery.mockResolvedValue({ rows: [{ id: 'history-1' }] });
|
||||
|
||||
await service.recordPlay('user-1', 'track-1', false);
|
||||
|
||||
expect(mockQuery.mock.calls[0][1][4]).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('upsertClaim', () => {
|
||||
it('calls INSERT ... ON CONFLICT with correct parameters', async () => {
|
||||
const { service, mockQuery } = makeService();
|
||||
|
||||
@@ -69,6 +69,22 @@ import type {
|
||||
import { AUDIO_PREFERENCE_BUCKETS, FEEDBACK_ACTIONS } from '../db/types.js';
|
||||
export * from '../db/types.js';
|
||||
|
||||
/**
|
||||
* Every play_history write goes through this statement.
|
||||
*
|
||||
* The title/artist copy is taken in the same round trip via LEFT JOIN, so a
|
||||
* play stays readable after its track is hard-deleted, and the join never
|
||||
* suppresses the insert when the track id is unknown.
|
||||
*/
|
||||
const PLAY_HISTORY_INSERT = `
|
||||
INSERT INTO play_history (
|
||||
user_id, track_id, batch_id, completed, listened_ms, track_title, track_artist
|
||||
)
|
||||
SELECT $1::uuid, $2::uuid, $3::uuid, $4::boolean, $5::int, t.title, t.artist
|
||||
FROM (SELECT 1) AS always
|
||||
LEFT JOIN tracks t ON t.id = $2::uuid
|
||||
RETURNING id`;
|
||||
|
||||
export class DbService {
|
||||
/** Exposed so route handlers (e.g. settings) can query the database directly. */
|
||||
readonly pgClient: Pool;
|
||||
@@ -483,11 +499,17 @@ export class DbService {
|
||||
// (permanentlyDeleteTrack, below) and exactly one unlink site in the whole
|
||||
// system (workers/src/cleanup.service.ts).
|
||||
|
||||
async recordPlay(userId: string, trackId: string, completed: boolean, batchId?: string): Promise<string> {
|
||||
async recordPlay(
|
||||
userId: string,
|
||||
trackId: string,
|
||||
completed: boolean,
|
||||
batchId?: string,
|
||||
listenedMs?: number,
|
||||
): Promise<string> {
|
||||
if (!completed) {
|
||||
const res = await this.pgClient.query(
|
||||
'INSERT INTO play_history (user_id, track_id, batch_id, completed) VALUES ($1, $2, $3, $4) RETURNING id',
|
||||
[userId, trackId, batchId ?? null, false]
|
||||
PLAY_HISTORY_INSERT,
|
||||
[userId, trackId, batchId ?? null, false, listenedMs ?? null]
|
||||
);
|
||||
return res.rows[0].id as string;
|
||||
}
|
||||
@@ -497,8 +519,8 @@ export class DbService {
|
||||
return this.withTransaction(async (client) => {
|
||||
// 1. Record play history
|
||||
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]
|
||||
PLAY_HISTORY_INSERT,
|
||||
[userId, trackId, batchId ?? null, true, listenedMs ?? null]
|
||||
);
|
||||
const historyId = insertRes.rows[0].id as string;
|
||||
|
||||
@@ -1936,10 +1958,17 @@ export class DbService {
|
||||
const occurredAt = event.occurred_at?.toISOString?.() ?? new Date().toISOString();
|
||||
switch (event.type) {
|
||||
case 'completed':
|
||||
// The Vibe player is the one caller that knows how much was actually
|
||||
// heard. position_ms is where playback reached; duration_ms is the
|
||||
// fallback for a client that only reported the track length.
|
||||
await client.query(
|
||||
`INSERT INTO play_history (user_id, track_id, completed, played_at)
|
||||
VALUES ($1, $2, true, $3::timestamptz)`,
|
||||
[event.user_id, event.track_id, occurredAt],
|
||||
`INSERT INTO play_history (
|
||||
user_id, track_id, completed, played_at, listened_ms, track_title, track_artist
|
||||
)
|
||||
SELECT $1::uuid, $2::uuid, true, $3::timestamptz, $4::int, t.title, t.artist
|
||||
FROM (SELECT 1) AS always
|
||||
LEFT JOIN tracks t ON t.id = $2::uuid`,
|
||||
[event.user_id, event.track_id, occurredAt, event.position_ms ?? event.duration_ms ?? null],
|
||||
);
|
||||
await client.query(
|
||||
`UPDATE tracks
|
||||
|
||||
@@ -148,17 +148,20 @@ export class DiscoveryService {
|
||||
const relevance = claimRes.rows[0]?.fused_value ?? 0;
|
||||
const candidateArtistId = claimRes.rows[0]?.object_id;
|
||||
|
||||
// Graph walks identify artists, not a legal/downloadable recording. A
|
||||
// resolver (or a human) must attach a vetted HTTPS source before the
|
||||
// worker can acquire anything. Do not guess a search query and download
|
||||
// an arbitrary track under an artist's name.
|
||||
// A candidate must name what to fetch: either a vetted HTTPS URL, or a
|
||||
// track-level search phrase the worker resolves against its allow-listed
|
||||
// hosts. An artist-only candidate (what a bare graph walk produces) names
|
||||
// neither, and parks here rather than downloading something arbitrary
|
||||
// under that artist's name.
|
||||
const notes = typeof row.notes === 'string' ? safeJson(row.notes) : row.notes;
|
||||
const resolvedUrl = (notes as { acquisition?: { url?: unknown } } | null)?.acquisition?.url;
|
||||
if (typeof resolvedUrl !== 'string' || resolvedUrl.trim() === '') {
|
||||
const acquisition = (notes as { acquisition?: { url?: unknown; query?: unknown } } | null)?.acquisition;
|
||||
const hasUrl = typeof acquisition?.url === 'string' && acquisition.url.trim() !== '';
|
||||
const hasQuery = typeof acquisition?.query === 'string' && acquisition.query.trim() !== '';
|
||||
if (!hasUrl && !hasQuery) {
|
||||
await this.db.pgClient.query(
|
||||
`UPDATE discovery_candidates
|
||||
SET status = 'awaiting_resolution', last_eval_at = NOW(),
|
||||
last_error = 'no vetted acquisition source attached'
|
||||
last_error = 'no acquisition url or search query attached'
|
||||
WHERE id = $1`,
|
||||
[row.id]
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user