muzick/workers/src/integrity.service.ts

import fs from 'fs/promises';
import { Client as PgClient } from 'pg';
import { Queue } from 'bullmq';
import { ScannerService } from './scanner.service.js';

const MUSIC_DIR = process.env.MUSIC_DIR || '/mnt/hdd1/media/Music';

// Detection regexes for the known (now-fixed) corruption bug. These are the
// exact patterns used by the original one-off repair script:
//   title  : trailing ' (Enriched)' (possibly stacked)
//   artist : trailing 'Unknown Artist' (concatenated, no separator, possibly stacked)
// Anchored to end-of-string; '+' handles stacked corruption.
const TITLE_CORRUPT_RE = `( \\(Enriched\\))+\\s*$`;
const ARTIST_CORRUPT_RE = `\\s*(Unknown Artist)+\\s*$`;

interface CorruptRow {
  id: string;
  path: string;
  title: string;
  artist: string;
}

interface MissingRow {
  id: string;
  path: string;
  title: string;
  artist: string;
}

export interface SweepSummary {
  detected: number;
  fixed: number;
  needsReview: number;
}

export class IntegrityService {
  private scanner: ScannerService;

  constructor(private pgClient: PgClient, private queue: Queue) {
    this.scanner = new ScannerService(pgClient, queue);
  }

  /**
   * Self-provision the issues table. Idempotent (CREATE TABLE IF NOT EXISTS),
   * mirrors backend/src/db/schema.sql so the worker runs on older DBs.
   */
  async ensureSchema(): Promise<void> {
    await this.pgClient.query(
      `CREATE TABLE IF NOT EXISTS track_integrity_issues (
         id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
         track_id UUID REFERENCES tracks(id) ON DELETE CASCADE,
         issue_type TEXT NOT NULL,
         status TEXT NOT NULL DEFAULT 'OPEN',
         details TEXT,
         detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
         resolved_at TIMESTAMP,
         UNIQUE(track_id, issue_type)
       )`
    );
  }

  /** Tracks whose title or artist still carry the corruption markers. */
  async detectCorruptedMetadata(): Promise<CorruptRow[]> {
    const res = await this.pgClient.query<CorruptRow>(
      `SELECT id, path, title, artist
         FROM tracks
        WHERE title ~ $1 OR artist ~ $2`,
      [TITLE_CORRUPT_RE, ARTIST_CORRUPT_RE]
    );
    return res.rows;
  }

  /** Tracks (excluding ones already flagged DELETED) whose file is gone from disk. */
  async detectMissingFiles(limit = 5000): Promise<MissingRow[]> {
    const res = await this.pgClient.query<MissingRow>(
      `SELECT id, path, title, artist
         FROM tracks
        WHERE state <> 'DELETED'
        LIMIT $1`,
      [limit]
    );

    const missing: MissingRow[] = [];
    for (const row of res.rows) {
      try {
        await fs.access(row.path);
      } catch {
        missing.push(row);
      }
    }
    return missing;
  }

  /** Upsert an issue row keyed on (track_id, issue_type), refreshing on re-detection. */
  private async upsertIssue(
    trackId: string,
    issueType: 'CORRUPT_METADATA' | 'MISSING_FILE',
    status: 'OPEN' | 'FIXED' | 'NEEDS_REVIEW',
    details: string | null
  ): Promise<void> {
    await this.pgClient.query(
      `INSERT INTO track_integrity_issues (track_id, issue_type, status, details, detected_at, resolved_at)
       VALUES ($1, $2, $3, $4, NOW(), CASE WHEN $3 = 'FIXED' THEN NOW() ELSE NULL END)
       ON CONFLICT (track_id, issue_type) DO UPDATE SET
         status = EXCLUDED.status,
         details = EXCLUDED.details,
         detected_at = NOW(),
         resolved_at = CASE WHEN EXCLUDED.status = 'FIXED' THEN NOW() ELSE NULL END`,
      [trackId, issueType, status, details]
    );
  }

  /**
   * The defensive SQL strip (PASS 2 from the original repair script). Anchored,
   * idempotent, only touches rows that actually match — a no-op on clean data.
   */
  private async stripCorruptionMarkers(): Promise<void> {
    await this.pgClient.query(
      `UPDATE tracks
          SET title = btrim(regexp_replace(title, '( \\(Enriched\\))+\\s*$', ''))
        WHERE title ~ '( \\(Enriched\\))+\\s*$' OR title <> btrim(title)`
    );
    await this.pgClient.query(
      `UPDATE tracks
          SET artist = btrim(regexp_replace(artist, '\\s*(Unknown Artist)+\\s*$', ''))
        WHERE artist ~ '\\s*(Unknown Artist)+\\s*$' OR artist <> btrim(artist)`
    );
  }

  /**
   * Orchestrate a full integrity sweep: detect, attempt auto-repair, then
   * reconcile issue rows. Safe to run repeatedly; a clean DB is a no-op and
   * produces no spurious issue rows.
   */
  async runSweep(): Promise<SweepSummary> {
    await this.ensureSchema();

    const summary: SweepSummary = { detected: 0, fixed: 0, needsReview: 0 };

    // --- Corrupted metadata --------------------------------------------------
    const corrupt = await this.detectCorruptedMetadata();
    summary.detected = corrupt.length;

    // Record each corrupt row as an OPEN issue (only real matches get rows).
    for (const row of corrupt) {
      const details = `corrupt title="${row.title}" artist="${row.artist}"`;
      await this.upsertIssue(row.id, 'CORRUPT_METADATA', 'OPEN', details);
    }

    if (corrupt.length > 0) {
      // PASS 1: authoritative re-scan from disk overwrites title/artist by path.
      console.log(`[Integrity] PASS 1: re-scanning library at ${MUSIC_DIR}`);
      try {
        await this.scanner.scanDirectory(MUSIC_DIR);
      } catch (err) {
        console.error('[Integrity] PASS 1 rescan failed (continuing to PASS 2):', err);
      }
      // PASS 2: defensive SQL strip for rows the scanner couldn't fix.
      console.log('[Integrity] PASS 2: defensive SQL strip of corruption markers');
      await this.stripCorruptionMarkers();

      // Reconcile: re-check each previously-corrupt row individually.
      const stillCorrupt = await this.pgClient.query<{ id: string }>(
        `SELECT id FROM tracks
          WHERE id = ANY($1::uuid[]) AND (title ~ $2 OR artist ~ $3)`,
        [corrupt.map((r) => r.id), TITLE_CORRUPT_RE, ARTIST_CORRUPT_RE]
      );
      const stillCorruptIds = new Set(stillCorrupt.rows.map((r) => r.id));

      for (const row of corrupt) {
        if (stillCorruptIds.has(row.id)) {
          // Could not auto-fix (e.g. file missing) -> flag for manual review.
          await this.upsertIssue(
            row.id,
            'CORRUPT_METADATA',
            'NEEDS_REVIEW',
            `could not auto-repair; corrupt title="${row.title}" artist="${row.artist}"`
          );
          summary.needsReview++;
        } else {
          await this.upsertIssue(row.id, 'CORRUPT_METADATA', 'FIXED', null);
          summary.fixed++;
        }
      }
    }

    // --- Missing files -------------------------------------------------------
    const missing = await this.detectMissingFiles();
    for (const row of missing) {
      summary.detected++;
      summary.needsReview++;
      await this.upsertIssue(
        row.id,
        'MISSING_FILE',
        'NEEDS_REVIEW',
        `file not found on disk: ${row.path}`
      );
      // No-ghost rule: mark the track MISSING rather than deleting/keeping it as
      // playable. Guarded UPDATE keeps this idempotent (only flips non-MISSING rows).
      await this.pgClient.query(
        `UPDATE tracks SET state = 'MISSING' WHERE id = $1 AND state <> 'MISSING'`,
        [row.id]
      );
    }

    console.log(
      `[Integrity] Sweep summary: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}`
    );
    return summary;
  }
}