muzick/workers/src/scanner.service.ts

import fs from 'fs/promises';
import { createReadStream } from 'fs';
import { createHash } from 'crypto';
import path from 'path';
import mm from 'music-metadata';
import type { Queryable } from './db.js';
import { Queue } from 'bullmq';
import { MetadataRefreshJob, ArtistSimilarityJob, ArtistImageJob, AlbumCoverJob, AudioAnalysisJob } from './types.js';
import { AUDIO_ANALYSIS_JOB_OPTIONS, audioAnalysisJobId, AUDIO_ANALYSIS_VERSION } from './audio-analysis.js';
import { splitArtistNames, parseArtists } from './utils/artist-names.js';

/**
 * Scanner provenance is supplied by the acquisition worker, not inferred from
 * tags. The optional candidate id lets the acquisition service make an exact
 * candidate -> scanned-track association after a successful scan.
 */
export interface ScanContext {
  sourceType?: 'MANUAL' | 'RECOMMENDATION';
  probationStatus?: 'probation' | 'retained' | 'retired';
  candidateId?: string;
  /**
   * Names to use when the file carries no title/artist tags. An acquired
   * download often has none, and the filename ("VTKqlmCpTmQ.mp3") plus
   * "Unknown Artist" is worse than the vetted candidate's own names.
   */
  fallbackTitle?: string;
  fallbackArtist?: string;
}

export interface ScanResult {
  trackIds: string[];
}

/**
 * Parse main + featured artists from music-metadata. Prefers the structured
 * `artists[]` array when the tag provides it (each entry already one artist),
 * otherwise falls back to splitting the single `artist` string. Feature markers
 * in the title ("Song (feat. X)") are folded in either way. First artist is the
 * main artist; the rest are featured. See ./utils/artist-names for the rules.
 */
function parseArtistsFromMetadata(common: any, fallbackArtist = 'Unknown Artist'): { main: string; featured: string[] } {
  const rawTitle = common.title || '';

  // Structured array present: still split each entry (tags sometimes put a whole
  // "A & B" collaboration in one slot) and fold in any title feature.
  if (common.artists && Array.isArray(common.artists) && common.artists.length > 0) {
    const fromArray = common.artists.flatMap((a: string) => splitArtistNames(a));
    const fromTitleParse = parseArtists('', rawTitle);
    const all: string[] = [];
    const seen = new Set<string>();
    for (const name of [...fromArray, ...fromTitleParse.featured]) {
      const n = name.trim();
      if (!n || seen.has(n.toLowerCase())) continue;
      seen.add(n.toLowerCase());
      all.push(n);
    }
    if (all.length > 0) {
      return { main: all[0], featured: all.slice(1) };
    }
  }

  return parseArtists(common.artist || fallbackArtist, rawTitle);
}

function hashFile(filePath: string): Promise<string> {
  return new Promise((resolve, reject) => {
    const hash = createHash('md5');
    const stream = createReadStream(filePath);
    stream.on('data', (d) => hash.update(d));
    stream.on('end', () => resolve(hash.digest('hex')));
    stream.on('error', reject);
  });
}

/**
 * Resolve the album artist from metadata tags. The `albumartist` tag is the
 * standard way to identify who "owns" the album as a whole — "Various Artists"
 * for compilations, "Soundtrack" for OSTs, or the primary artist for normal
 * albums. Without it, compilations where each track has a different artist
 * would create duplicate album rows (one per track artist).
 *
 * Falls back to the track's main artist when the tag is absent (single-artist
 * albums where the tag is redundant).
 *
 * Also splits the albumartist string via splitArtistNames so that "A & B" as
 * albumartist uses just the first-billed artist as the album's identity —
 * consistent with how normalize_artist treats track artists.
 */
function resolveAlbumArtist(common: any, fallback: string): string {
  // music-metadata exposes albumartist as either a string or an array.
  const raw = common.albumartist ?? common.albumArtist;
  if (!raw) return fallback;
  const str = Array.isArray(raw) ? raw[0] : String(raw);
  if (str.trim() === '') return fallback;
  const parts = splitArtistNames(str);
  return parts.length > 0 ? parts[0] : fallback;
}

export class ScannerService {
  // Per-scan dedupe sets for per-artist / per-album jobs. Belt-and-braces
  // alongside the BullMQ jobId dedupe (jobId dedupes across overlapping scans /
  // pending jobs; these avoid even issuing the redundant add() within one scan).
  private enqueuedArtists = new Set<string>();
  private enqueuedAlbums = new Set<string>();
  private audioAnalysisEnabled = false;

  constructor(private pgClient: Queryable, private queue: Queue) {}

  async scanDirectory(directory: string, context: ScanContext = {}): Promise<ScanResult> {
    console.log(`[Scanner] Starting scan in: ${directory}`);
    this.enqueuedArtists.clear();
    this.enqueuedAlbums.clear();
    this.audioAnalysisEnabled = await this.loadAudioAnalysisSetting();
    const trackIds: string[] = [];
    await this.walk(directory, context, trackIds);
    console.log(`[Scanner] Scan completed.`);
    return { trackIds };
  }

  private async walk(dir: string, context: ScanContext, trackIds: string[]) {
    const entries = await fs.readdir(dir, { withFileTypes: true });

    for (const entry of entries) {
      const fullPath = path.join(dir, entry.name);
      if (entry.isDirectory()) {
        await this.walk(fullPath, context, trackIds);
      } else if (this.isMusicFile(entry.name)) {
        const trackId = await this.processFile(fullPath, context);
        if (trackId) trackIds.push(trackId);
      }
    }
  }

  private isMusicFile(fileName: string): boolean {
    const extensions = ['.mp3', '.flac', '.m4a', '.wav', '.ogg'];
    return extensions.includes(path.extname(fileName).toLowerCase());
  }

  private async processFile(filePath: string, context: ScanContext): Promise<string | null> {
    try {
      console.log(`[Scanner] Processing: ${filePath}`);
      const metadata = await mm.parseFile(filePath);
      const { common, format } = metadata;

      // 1. Ensure Artist(s) exist.
      const trackTitle = common.title || context.fallbackTitle || path.basename(filePath);
      const { main: mainArtistRaw, featured: featuredArtistNames } =
        parseArtistsFromMetadata(common, context.fallbackArtist || 'Unknown Artist');

      const { id: artistId, name: resolvedArtist } = await this.resolveOrCreateArtist(mainArtistRaw);

      // Upsert featured artists into artists table and collect their ids.
      const featuredIds: string[] = [];
      for (const featName of featuredArtistNames) {
        const feat = await this.resolveOrCreateArtist(featName);
        featuredIds.push(feat.id);
      }

      // 2. Ensure Album exists.
      //    The album is keyed by (album_artist_id, title), NOT (track_artist_id,
      //    title). This prevents duplicate album rows for compilations where
      //    each track has a different artist but they all belong to one album.
      //    We resolve the ALBUM artist from the `albumartist` tag when present
      //    ("Various Artists", "Soundtrack", or the primary artist), falling
      //    back to the track's main artist when the tag is absent.
      const albumTitle = common.album || 'Unknown Album';
      const albumArtistRaw = resolveAlbumArtist(common, mainArtistRaw);
      const albumArtistId = albumArtistRaw === mainArtistRaw
        ? artistId
        : (await this.resolveOrCreateArtist(albumArtistRaw)).id;

      const existingAlbum = await this.pgClient.query(
        'SELECT id FROM albums WHERE artist_id = $1 AND lower(title) = lower($2) LIMIT 1',
        [albumArtistId, albumTitle]
      );
      let albumId: string;
      if (existingAlbum.rows.length > 0) {
        albumId = String(existingAlbum.rows[0].id);
      } else {
        const albumRes = await this.pgClient.query(
          'INSERT INTO albums (artist_id, title) VALUES ($1, $2) ON CONFLICT (artist_id, title) DO UPDATE SET title = EXCLUDED.title RETURNING id',
          [albumArtistId, albumTitle]
        );
        albumId = String(albumRes.rows[0].id);
      }

      // 3. Upsert Track
      const duration = format.duration || 0;
      const fileHash = await hashFile(filePath);

      // A recommendation scan sets provenance at creation time. A routine
      // library rescan must never erase that provenance or reset probation.
      const sourceType = context.sourceType ?? 'MANUAL';
      const state = sourceType === 'RECOMMENDATION' ? 'RECOMMENDED' : 'LIBRARY';
      const probationStatus = sourceType === 'RECOMMENDATION'
        ? (context.probationStatus ?? 'probation')
        : 'retained';
      const trackRes = await this.pgClient.query(
        `INSERT INTO tracks (
           path, hash, title, artist, album_id, duration, state, source_type,
           probation_status, probation_entered_at
         )
         VALUES (
           $1, $2, $3, $4, $5, $6, $7::track_state, $8::track_source_type,
           $9, CASE WHEN $8::track_source_type = 'RECOMMENDATION' THEN NOW() ELSE NULL END
         )
         ON CONFLICT (path) DO UPDATE SET
            hash = EXCLUDED.hash,
            title = EXCLUDED.title,
            artist = EXCLUDED.artist,
            album_id = EXCLUDED.album_id,
            duration = EXCLUDED.duration,
            mtime = EXTRACT(EPOCH FROM NOW()),
            -- Existing recommendation rows stay recommendations during every
            -- ordinary scan. This is load-bearing for probation and Vibe.
            -- A dislike is a listener verdict on a file that is still on disk,
            -- so finding that file again says nothing new: HIDDEN and DELETED
            -- survive a rescan. Without this every scan restored every disliked
            -- track to LIBRARY and the Vibe served it again.
            state = CASE WHEN tracks.source_type = 'RECOMMENDATION'
                         THEN tracks.state
                         WHEN tracks.state IN ('HIDDEN', 'DELETED')
                         THEN tracks.state
                         ELSE EXCLUDED.state END,
            source_type = CASE WHEN tracks.source_type = 'RECOMMENDATION'
                               THEN tracks.source_type ELSE $8::track_source_type END,
            probation_status = CASE WHEN tracks.source_type = 'RECOMMENDATION'
                                    THEN tracks.probation_status ELSE $9 END,
            probation_entered_at = CASE WHEN tracks.source_type = 'RECOMMENDATION'
                                        THEN tracks.probation_entered_at
                                        WHEN $8::track_source_type = 'RECOMMENDATION' THEN NOW()
                                        ELSE tracks.probation_entered_at END
         RETURNING id
         `,
        [filePath, fileHash, trackTitle, resolvedArtist, albumId, duration, state, sourceType, probationStatus]
      );
      const trackId = String(trackRes.rows[0].id);

      // 4. Populate track_artists junction table.
      await this.pgClient.query(
        `INSERT INTO track_artists (track_id, artist_id, role)
         VALUES ($1, $2, 'main')
         ON CONFLICT (track_id, artist_id, role) DO NOTHING`,
        [trackId, artistId]
      );
      for (const featId of featuredIds) {
        await this.pgClient.query(
          `INSERT INTO track_artists (track_id, artist_id, role)
           VALUES ($1, $2, 'featured')
           ON CONFLICT (track_id, artist_id, role) DO NOTHING`,
          [trackId, featId]
        );
      }

      // 5. Put the same credit into the claim graph. Every Vibe generator but
      // the fallback reaches tracks through claim_fusion, and only MusicBrainz
      // was writing those edges — so a track MusicBrainz has never heard of was
      // invisible to the recommender no matter how well tagged the file was.
      // `tag` is the lowest-trust source in the spine, so a later MusicBrainz
      // claim about the same edge still outranks this one.
      await this.writeTagCredits(trackId, String(artistId), featuredIds.map(String));

      console.log(`[Scanner] Successfully processed: ${trackTitle}`);

      // Trigger external-API enrichment for this track + artist + album.
      // Best-effort: an enqueue failure must never abort the scan of remaining files.
      await this.enqueueEnrichment(trackId, String(artistId), String(albumId));
      return trackId;
    } catch (err) {
      console.error(`[Scanner] Error processing ${filePath}:`, err);
      return null;
    }
  }

  /**
   * Resolve an artist by its case-insensitive normalized identity, creating the
   * row only when no existing artist matches. This is what prevents case /
   * punctuation variants ("Booker" vs "BOOKER", "Acryl madness" vs "Acryl
   * Madness") from becoming separate rows (and duplicate albums). Returns the
   * artist id and the stored display name.
   */
  private async resolveOrCreateArtist(rawName: string): Promise<{ id: string; name: string }> {
    const norm = await this.pgClient.query('SELECT normalize_artist($1) AS name', [rawName]);
    let name: string = norm.rows[0].name;

    // Comma heuristic: prefer the pre-comma part if it already exists.
    if (name.includes(',')) {
      const before = name.split(',')[0].trim();
      const existing = await this.pgClient.query(
        'SELECT id FROM artists WHERE lower(normalized_name) = lower(normalize_artist($1)) LIMIT 1',
        [before]
      );
      if (existing.rows.length > 0) name = before;
    }

    // Case-insensitive lookup by normalized identity; prefer the most "complete"
    // row when several somehow match.
    const found = await this.pgClient.query(
      `SELECT id, name FROM artists
       WHERE lower(normalized_name) = lower(normalize_artist($1))
       ORDER BY CASE WHEN mbid IS NOT NULL THEN 0 ELSE 1 END,
                CASE WHEN image_path IS NOT NULL AND image_path <> '' THEN 0 ELSE 1 END,
                created_at
       LIMIT 1`,
      [name]
    );
    if (found.rows.length > 0) {
      return { id: String(found.rows[0].id), name: String(found.rows[0].name) };
    }

    // `canonical_name` is NOT NULL in schema.sql with no default, so it MUST be
    // supplied here — omitting it makes every artist insert fail on a fresh
    // volume (and processFile swallows the error, so the scan silently yields an
    // empty library). It holds the DISPLAY name: we store the raw tag name, not
    // the normalize_artist() output, because that function truncates on `/` and
    // a standalone `x` ("AC/DC" -> "AC", "Felix Mendelssohn" -> "Feli"). The
    // enrichment path later overwrites canonical_name with the MusicBrainz name;
    // until then the raw tag is the most faithful display value we have.
    const inserted = await this.pgClient.query(
      'INSERT INTO artists (name, canonical_name) VALUES ($1, $2) RETURNING id, name',
      [name, rawName.trim() || name]
    );
    return { id: String(inserted.rows[0].id), name: String(inserted.rows[0].name) };
  }

  /**
   * Record the file's own artist credits as claims. Confidence is 1.0 because
   * the tag says this without ambiguity; how much that is worth is decided by
   * the trust attached to the `tag` source, not here.
   */
  private async writeTagCredits(trackId: string, artistId: string, featuredIds: string[]): Promise<void> {
    const credits: Array<[string, string]> = [
      [artistId, 'credited_main_on'],
      ...featuredIds.map((id): [string, string] => [id, 'featured_on']),
    ];
    for (const [objectId, predicate] of credits) {
      try {
        await this.pgClient.query(
          `INSERT INTO claims (
             subject_type, subject_id, predicate, object_type, object_id, source, confidence
           ) VALUES ('track', $1::uuid, $2, 'artist', $3::uuid, 'tag', 1.0)
           ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id)
           DO UPDATE SET last_reinforced_at = NOW()`,
          [trackId, predicate, objectId]
        );
      } catch (err) {
        // A missing edge costs this track its place in the graph; it must not
        // cost the whole scan the file.
        console.error(`[Scanner] Failed to write ${predicate} claim for track ${trackId}:`, err);
      }
    }
  }

  private async enqueueEnrichment(trackId: string, artistId: string, albumId: string) {
    const keep = {
      removeOnComplete: { age: 86400, count: 5000 },
      removeOnFail: { age: 86400, count: 5000 },
    } as const;

    // Audio decoding is CPU/memory intensive. It must never be performed inline
    // with metadata refreshes: enqueue one deduplicated, retryable job only when
    // the user has enabled it and the file has changed or lacks the current
    // analysis version.
    if (this.audioAnalysisEnabled) {
      try {
        const current = await this.pgClient.query<{ current: boolean }>(
          `SELECT EXISTS (
             SELECT 1
             FROM tracks t
             JOIN track_audio_features af ON af.track_id = t.id
             WHERE t.id = $1
               AND af.analysis_version >= $2
               AND af.source_hash = t.hash
           ) AS current`,
          [trackId, AUDIO_ANALYSIS_VERSION]
        );
        if (!current.rows[0]?.current) {
          const payload: AudioAnalysisJob = { trackId };
          await this.queue.add('audio_analysis', payload, {
            jobId: audioAnalysisJobId(trackId),
            ...AUDIO_ANALYSIS_JOB_OPTIONS,
          });
        }
      } catch (err) {
        console.error(`[Scanner] Failed to enqueue audio_analysis for track ${trackId}:`, err);
      }
    }

    // metadata_refresh per track. jobId `meta-<trackId>` collapses duplicate
    // pending jobs across re-scans; the handler (enrichTrack) is idempotent so
    // re-enqueues are always safe. BullMQ 5.x rejects colons in custom ids.
    try {
      const payload: MetadataRefreshJob = { trackId, refreshType: 'full' };
      await this.queue.add('metadata_refresh', payload, { jobId: `meta-${trackId}`, ...keep });
    } catch (err) {
      console.error(`[Scanner] Failed to enqueue metadata_refresh for track ${trackId}:`, err);
    }

    // album_cover per album, deduped per scan (Set) + jobId across scans.
    if (!this.enqueuedAlbums.has(albumId)) {
      this.enqueuedAlbums.add(albumId);
      try {
        const payload: AlbumCoverJob = { albumId };
        await this.queue.add('album_cover', payload, { jobId: `album-cover-${albumId}`, ...keep });
      } catch (err) {
        console.error(`[Scanner] Failed to enqueue album_cover for album ${albumId}:`, err);
      }
    }

    // artist_similarity + artist_image per artist, deduped per scan (Set) AND
    // across overlapping scans / pending jobs via their jobIds.
    if (this.enqueuedArtists.has(artistId)) {
      return;
    }
    this.enqueuedArtists.add(artistId);
    try {
      const payload: ArtistSimilarityJob = { artistId };
      await this.queue.add('artist_similarity', payload, { jobId: `artist-sim-${artistId}`, ...keep });
    } catch (err) {
      console.error(`[Scanner] Failed to enqueue artist_similarity for artist ${artistId}:`, err);
    }
    try {
      const payload: ArtistImageJob = { artistId };
      await this.queue.add('artist_image', payload, { jobId: `artist-image-${artistId}`, ...keep });
    } catch (err) {
      console.error(`[Scanner] Failed to enqueue artist_image for artist ${artistId}:`, err);
    }
  }

  private async loadAudioAnalysisSetting(): Promise<boolean> {
    try {
      const result = await this.pgClient.query<{ value: string }>(
        "SELECT value FROM settings WHERE key = 'enrich_audio_analysis'"
      );
      return result.rows[0]?.value === 'true';
    } catch (err) {
      // Safe default: a schema/startup problem must not fan out expensive DSP
      // work across a scan.
      console.warn('[Scanner] Audio analysis disabled: unable to read setting:', (err as Error).message);
      return false;
    }
  }
}