feat: enhance discovery, vibe sessions, and library enrichment
This commit is contained in:
kami
@@ -0,0 +1,289 @@
|
||||
import { access, mkdir, realpath, stat } from 'node:fs/promises';
|
||||
import { constants as fsConstants } from 'node:fs';
|
||||
import { spawn } from 'node:child_process';
|
||||
import path from 'node:path';
|
||||
import type { Pool } from 'pg';
|
||||
import { ScannerService } from './scanner.service.js';
|
||||
|
||||
type CandidateRow = {
|
||||
id: string;
|
||||
source: string;
|
||||
notes: unknown;
|
||||
status: string;
|
||||
};
|
||||
|
||||
type AcquisitionSpec = {
|
||||
url: string;
|
||||
expectedTitle?: string;
|
||||
expectedArtist?: string;
|
||||
};
|
||||
|
||||
export type AcquisitionResult =
|
||||
| { status: 'acquired'; trackId: string }
|
||||
| { status: 'disabled'; reason: string }
|
||||
| { status: 'failed'; reason: string };
|
||||
|
||||
export interface AcquisitionConfig {
|
||||
enabled: boolean;
|
||||
ytDlpPath: string;
|
||||
musicDir: string;
|
||||
destinationDir: string;
|
||||
allowedHosts: Set<string>;
|
||||
timeoutMs: number;
|
||||
maxFileBytes: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* System E is intentionally off unless every gate is configured. In
|
||||
* particular, a bare `yt-dlp` command is not accepted: an absolute executable
|
||||
* path avoids PATH surprises in a long-running, network-enabled worker.
|
||||
*/
|
||||
export function acquisitionConfigFromEnv(env = process.env): AcquisitionConfig {
|
||||
const musicDir = path.resolve(env.MUSIC_DIR || '/music');
|
||||
const requestedDestination = env.MUZICK_ACQUISITION_DIR || '.recommendations';
|
||||
const destinationDir = path.resolve(musicDir, requestedDestination);
|
||||
return {
|
||||
enabled: env.MUZICK_ACQUISITION_ENABLED === 'true',
|
||||
ytDlpPath: env.MUZICK_ACQUISITION_YTDLP_PATH || '',
|
||||
musicDir,
|
||||
destinationDir,
|
||||
allowedHosts: new Set(
|
||||
(env.MUZICK_ACQUISITION_ALLOWED_HOSTS || '')
|
||||
.split(',').map((value) => value.trim().toLowerCase()).filter(Boolean)
|
||||
),
|
||||
timeoutMs: Math.max(10_000, Math.min(Number(env.MUZICK_ACQUISITION_TIMEOUT_MS || 120_000), 15 * 60_000)),
|
||||
maxFileBytes: Math.max(1_000_000, Math.min(Number(env.MUZICK_ACQUISITION_MAX_FILE_BYTES || 250 * 1024 * 1024), 2 * 1024 * 1024 * 1024)),
|
||||
};
|
||||
}
|
||||
|
||||
function isWithin(parent: string, child: string): boolean {
|
||||
const relative = path.relative(parent, child);
|
||||
return relative === '' || (!relative.startsWith(`..${path.sep}`) && relative !== '..' && !path.isAbsolute(relative));
|
||||
}
|
||||
|
||||
function normalizedMetadata(value: string): string {
|
||||
return value.toLowerCase().normalize('NFKD').replace(/[^\p{L}\p{N}]+/gu, ' ').trim();
|
||||
}
|
||||
|
||||
function matchesExpected(actual: string, expected: string | undefined): boolean {
|
||||
if (!expected) return true;
|
||||
const left = normalizedMetadata(actual);
|
||||
const right = normalizedMetadata(expected);
|
||||
return left === right || left.includes(right) || right.includes(left);
|
||||
}
|
||||
|
||||
/** Parse only a deliberately supplied HTTPS source URL; never accept argv/query strings. */
|
||||
export function parseAcquisitionSpec(notes: unknown, allowedHosts: Set<string>): AcquisitionSpec {
|
||||
const notesValue = typeof notes === 'string' ? JSON.parse(notes) : notes;
|
||||
const candidate = (notesValue as { acquisition?: unknown } | null)?.acquisition;
|
||||
if (!candidate || typeof candidate !== 'object') {
|
||||
throw new Error('candidate has no resolved acquisition source');
|
||||
}
|
||||
const { url, expectedTitle, expectedArtist } = candidate as Record<string, unknown>;
|
||||
if (typeof url !== 'string') throw new Error('acquisition source URL is required');
|
||||
let parsed: URL;
|
||||
try {
|
||||
parsed = new URL(url);
|
||||
} catch {
|
||||
throw new Error('acquisition source URL is invalid');
|
||||
}
|
||||
if (parsed.protocol !== 'https:') throw new Error('acquisition source URL must use HTTPS');
|
||||
if (parsed.username || parsed.password) throw new Error('acquisition source URL must not contain credentials');
|
||||
if (!allowedHosts.has(parsed.hostname.toLowerCase())) {
|
||||
throw new Error(`acquisition host is not allow-listed: ${parsed.hostname}`);
|
||||
}
|
||||
return {
|
||||
url: parsed.toString(),
|
||||
expectedTitle: typeof expectedTitle === 'string' ? expectedTitle.slice(0, 500) : undefined,
|
||||
expectedArtist: typeof expectedArtist === 'string' ? expectedArtist.slice(0, 500) : undefined,
|
||||
};
|
||||
}
|
||||
|
||||
async function runDownloader(executable: string, args: string[], timeoutMs: number): Promise<string> {
|
||||
return await new Promise((resolve, reject) => {
|
||||
const child = spawn(executable, args, { shell: false, stdio: ['ignore', 'pipe', 'pipe'] });
|
||||
let stdout = '';
|
||||
let stderr = '';
|
||||
const timer = setTimeout(() => child.kill('SIGTERM'), timeoutMs);
|
||||
child.stdout.on('data', (chunk: Buffer) => { stdout += chunk.toString(); });
|
||||
child.stderr.on('data', (chunk: Buffer) => { stderr += chunk.toString(); });
|
||||
child.once('error', (err) => {
|
||||
clearTimeout(timer);
|
||||
reject(err);
|
||||
});
|
||||
child.once('close', (code, signal) => {
|
||||
clearTimeout(timer);
|
||||
if (code === 0) return resolve(stdout);
|
||||
const detail = (stderr || `exit=${code ?? 'null'} signal=${signal ?? 'none'}`).trim().slice(0, 1000);
|
||||
reject(new Error(`downloader failed: ${detail}`));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export class AcquisitionService {
|
||||
private readonly config: AcquisitionConfig;
|
||||
|
||||
constructor(
|
||||
private readonly pgPool: Pool,
|
||||
private readonly scanner: ScannerService,
|
||||
config = acquisitionConfigFromEnv(),
|
||||
) {
|
||||
this.config = config;
|
||||
}
|
||||
|
||||
private async disabledReason(): Promise<string | null> {
|
||||
if (!this.config.enabled) return 'MUZICK_ACQUISITION_ENABLED is not true';
|
||||
if (!path.isAbsolute(this.config.ytDlpPath)) return 'MUZICK_ACQUISITION_YTDLP_PATH must be an absolute executable path';
|
||||
if (this.config.allowedHosts.size === 0) return 'MUZICK_ACQUISITION_ALLOWED_HOSTS is empty';
|
||||
if (!isWithin(this.config.musicDir, this.config.destinationDir)) return 'MUZICK_ACQUISITION_DIR must stay inside MUSIC_DIR';
|
||||
try {
|
||||
await access(this.config.ytDlpPath, fsConstants.X_OK);
|
||||
} catch {
|
||||
return `downloader is not executable: ${this.config.ytDlpPath}`;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private async setStatus(candidateId: string, status: string, error: string | null): Promise<void> {
|
||||
await this.pgPool.query(
|
||||
`UPDATE discovery_candidates
|
||||
SET status = $2, last_eval_at = NOW(), last_error = $3
|
||||
WHERE id = $1`,
|
||||
[candidateId, status, error?.slice(0, 2000) ?? null]
|
||||
);
|
||||
}
|
||||
|
||||
async acquire(candidateId: string): Promise<AcquisitionResult> {
|
||||
const rowResult = await this.pgPool.query<CandidateRow>(
|
||||
`SELECT id, source, notes, status FROM discovery_candidates WHERE id = $1`, [candidateId]
|
||||
);
|
||||
const candidate = rowResult.rows[0];
|
||||
if (!candidate) return { status: 'failed', reason: 'candidate does not exist' };
|
||||
if (candidate.status !== 'acquiring') {
|
||||
return { status: 'failed', reason: `candidate is not acquiring (status=${candidate.status})` };
|
||||
}
|
||||
|
||||
const disabled = await this.disabledReason();
|
||||
if (disabled) {
|
||||
await this.setStatus(candidate.id, 'acquisition_disabled', disabled);
|
||||
return { status: 'disabled', reason: disabled };
|
||||
}
|
||||
|
||||
let spec: AcquisitionSpec;
|
||||
try {
|
||||
spec = parseAcquisitionSpec(candidate.notes, this.config.allowedHosts);
|
||||
} catch (err) {
|
||||
const reason = err instanceof Error ? err.message : 'invalid acquisition source';
|
||||
await this.setStatus(candidate.id, 'awaiting_resolution', reason);
|
||||
return { status: 'failed', reason };
|
||||
}
|
||||
|
||||
const sourceTrust = await this.pgPool.query<{ key: string }>(
|
||||
'SELECT key FROM source_trust WHERE key = $1', [candidate.source]
|
||||
);
|
||||
if (sourceTrust.rows.length === 0) {
|
||||
const reason = `candidate source is not registered in source_trust: ${candidate.source}`;
|
||||
await this.setStatus(candidate.id, 'failed', reason);
|
||||
return { status: 'failed', reason };
|
||||
}
|
||||
|
||||
const candidateDir = path.join(this.config.destinationDir, candidate.id);
|
||||
if (!isWithin(this.config.destinationDir, candidateDir)) {
|
||||
const reason = 'candidate destination escaped acquisition directory';
|
||||
await this.setStatus(candidate.id, 'failed', reason);
|
||||
return { status: 'failed', reason };
|
||||
}
|
||||
|
||||
try {
|
||||
await mkdir(candidateDir, { recursive: true });
|
||||
const resolvedDestination = await realpath(this.config.destinationDir);
|
||||
const resolvedCandidateDir = await realpath(candidateDir);
|
||||
if (!isWithin(resolvedDestination, resolvedCandidateDir)) {
|
||||
throw new Error('resolved candidate destination escaped acquisition directory');
|
||||
}
|
||||
|
||||
await this.pgPool.query(
|
||||
`UPDATE discovery_candidates
|
||||
SET status = 'downloading', acquisition_attempts = acquisition_attempts + 1,
|
||||
last_eval_at = NOW(), last_error = NULL
|
||||
WHERE id = $1`, [candidate.id]
|
||||
);
|
||||
|
||||
// Arguments are fixed by us. The sole untrusted value is the validated URL
|
||||
// and spawn() is invoked with shell:false, so no command interpolation is
|
||||
// possible. One URL / one item is deliberate: playlists are out of scope.
|
||||
const outputTemplate = path.join(candidateDir, '%(id)s.%(ext)s');
|
||||
const stdout = await runDownloader(this.config.ytDlpPath, [
|
||||
'--no-playlist', '--no-progress', '--restrict-filenames',
|
||||
'--extract-audio', '--audio-format', 'mp3', '--audio-quality', '5',
|
||||
'--output', outputTemplate,
|
||||
'--print', 'after_move:filepath',
|
||||
'--', spec.url,
|
||||
], this.config.timeoutMs);
|
||||
const reportedPaths = stdout.split(/\r?\n/).map((line) => line.trim()).filter(Boolean);
|
||||
if (reportedPaths.length !== 1) throw new Error('downloader did not report exactly one output file');
|
||||
const outputPath = path.resolve(reportedPaths[0]);
|
||||
if (!isWithin(resolvedCandidateDir, outputPath) || path.extname(outputPath).toLowerCase() !== '.mp3') {
|
||||
throw new Error('downloader reported an unsafe or unsupported output path');
|
||||
}
|
||||
await access(outputPath, fsConstants.R_OK);
|
||||
const outputStat = await stat(outputPath);
|
||||
if (!outputStat.isFile() || outputStat.size <= 0 || outputStat.size > this.config.maxFileBytes) {
|
||||
throw new Error(`downloaded file violates size limit (${this.config.maxFileBytes} bytes)`);
|
||||
}
|
||||
|
||||
await this.setStatus(candidate.id, 'scanning', null);
|
||||
const scan = await this.scanner.scanDirectory(candidateDir, {
|
||||
sourceType: 'RECOMMENDATION', probationStatus: 'probation', candidateId: candidate.id,
|
||||
});
|
||||
if (scan.trackIds.length !== 1) {
|
||||
throw new Error(`scanner created ${scan.trackIds.length} tracks; expected exactly one`);
|
||||
}
|
||||
const trackId = scan.trackIds[0];
|
||||
const metadata = await this.pgPool.query<{ title: string; artist: string }>(
|
||||
'SELECT title, artist FROM tracks WHERE id = $1', [trackId]
|
||||
);
|
||||
const scanned = metadata.rows[0];
|
||||
if (!scanned || !matchesExpected(scanned.title, spec.expectedTitle) || !matchesExpected(scanned.artist, spec.expectedArtist)) {
|
||||
await this.pgPool.query(
|
||||
`UPDATE tracks SET state = 'HIDDEN', probation_status = 'retired'
|
||||
WHERE id = $1`, [trackId]
|
||||
);
|
||||
throw new Error('downloaded metadata does not match the vetted candidate');
|
||||
}
|
||||
|
||||
const client = await this.pgPool.connect();
|
||||
try {
|
||||
await client.query('BEGIN');
|
||||
await client.query(
|
||||
`UPDATE discovery_candidates
|
||||
SET status = 'acquired', acquired_track_id = $2, acquired_at = NOW(),
|
||||
last_eval_at = NOW(), last_error = NULL
|
||||
WHERE id = $1`, [candidate.id, trackId]
|
||||
);
|
||||
await client.query(
|
||||
`INSERT INTO claims (
|
||||
subject_type, subject_id, predicate, object_type, object_id,
|
||||
source, confidence, raw
|
||||
) VALUES ('track', $1::uuid, 'acquired_from', 'discovery_candidate', $2::uuid,
|
||||
$3, 1.0, $4::jsonb)
|
||||
ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id)
|
||||
DO UPDATE SET confidence = EXCLUDED.confidence,
|
||||
last_reinforced_at = NOW(), raw = EXCLUDED.raw`,
|
||||
[trackId, candidate.id, candidate.source, JSON.stringify({ expectedTitle: spec.expectedTitle, expectedArtist: spec.expectedArtist })]
|
||||
);
|
||||
await client.query('COMMIT');
|
||||
} catch (err) {
|
||||
await client.query('ROLLBACK');
|
||||
throw err;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
return { status: 'acquired', trackId };
|
||||
} catch (err) {
|
||||
const reason = err instanceof Error ? err.message : 'acquisition failed';
|
||||
await this.setStatus(candidate.id, 'failed', reason);
|
||||
return { status: 'failed', reason };
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user