muzick/workers/src/acquisition.service.ts

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 = {
  /** A vetted, allow-listed HTTPS URL. Mutually exclusive with `query`. */
  url?: string;
  /** A search phrase to resolve into such a URL at download time. */
  query?: 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);
}

/**
 * Every URL that reaches the downloader passes through here: HTTPS only, no
 * embedded credentials, host on the allow-list. Applied to operator-supplied
 * URLs and to search-resolved ones alike, so a search cannot widen the hosts a
 * download may come from.
 */
export function validateAcquisitionUrl(raw: unknown, allowedHosts: Set<string>): string {
  if (typeof raw !== 'string') throw new Error('acquisition source URL is required');
  let parsed: URL;
  try {
    parsed = new URL(raw);
  } 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 parsed.toString();
}

/**
 * Parse either a vetted HTTPS URL or a search phrase to resolve into one.
 *
 * The search form is what automated discovery emits: a graph walk or a
 * similarity lookup names an artist and a title, never a downloadable file. A
 * resolved search result is trusted no further than an operator-supplied URL —
 * same host allow-list, and `matchesExpected` still rejects the download if the
 * file's tags disagree with the candidate.
 */
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, query, expectedTitle, expectedArtist } = candidate as Record<string, unknown>;
  const expected = {
    expectedTitle: typeof expectedTitle === 'string' ? expectedTitle.slice(0, 500) : undefined,
    expectedArtist: typeof expectedArtist === 'string' ? expectedArtist.slice(0, 500) : undefined,
  };

  if (url !== undefined) {
    return { url: validateAcquisitionUrl(url, allowedHosts), ...expected };
  }
  if (typeof query === 'string' && query.trim() !== '') {
    // Newlines would let a crafted candidate forge extra --print output lines
    // when the resolver parses stdout.
    const cleaned = query.replace(/[\r\n]+/g, ' ').trim().slice(0, 300);
    if (cleaned === '') throw new Error('acquisition search query is empty');
    return { query: cleaned, ...expected };
  }
  throw new Error('acquisition source needs either a url or a query');
}

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]
    );
  }

  /**
   * Turn a search phrase into one allow-listed URL, without downloading.
   *
   * ponytail: first search result only. Ranking alternatives needs a quality
   * signal this system does not have, and the tag check downstream already
   * rejects a wrong hit. Revisit if mismatches become common.
   */
  private async resolveQueryToUrl(query: string): Promise<string> {
    const stdout = await runDownloader(this.config.ytDlpPath, [
      '--no-playlist', '--no-progress', '--skip-download',
      '--print', 'webpage_url',
      '--', `ytsearch1:${query}`,
    ], this.config.timeoutMs);
    const lines = stdout.split(/\r?\n/).map((line) => line.trim()).filter(Boolean);
    if (lines.length !== 1) throw new Error('search did not resolve to exactly one result');
    return validateAcquisitionUrl(lines[0], this.config.allowedHosts);
  }

  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 sourceUrl = spec.url ?? await this.resolveQueryToUrl(spec.query as string);

      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',
        // Without this the mp3 carries no tags at all, the scanner falls back to
        // "<video id>.mp3" / "Unknown Artist", and the tag check below rejects
        // every download.
        '--embed-metadata',
        '--output', outputTemplate,
        '--print', 'after_move:filepath',
        '--', sourceUrl,
      ], 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,
        // A source that embeds no tags leaves nothing to verify against. The
        // candidate's own vetted names are then both the display names and
        // what the check below compares, so the download is accepted and the
        // track stays on probation, where listening decides its fate.
        fallbackTitle: spec.expectedTitle, fallbackArtist: spec.expectedArtist,
      });
      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,
            // Which URL a search actually landed on is the only way to audit a
            // bad acquisition after the fact.
            sourceUrl,
            resolvedFromQuery: spec.query ?? null,
          })]
        );
        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 };
    }
  }
}