import { Worker, Job } from 'bullmq';
import { connection, QUEUE_NAME, queue } from './queue.js';
import { MetadataRefreshJob, AudioAnalysisJob, LibraryScanJob, ArtistSimilarityJob, ArtistImageJob, AlbumCoverJob, ReprocessArtistsJob } from './types.js';
import { Pool } from 'pg';
import { ScannerService } from './scanner.service.js';
import { IntegrityService } from './integrity.service.js';
import { EnrichmentService } from './enrichment.service.js';
import { AudioFeaturesService } from './audio-features.service.js';
import { CleanupSweepService } from './cleanup.service.js';
import { reprocessArtists } from './reprocess-artists.service.js';

// Cron for the periodic integrity sweep (default: daily at 03:00). Configurable
// via INTEGRITY_SWEEP_CRON. MUSIC_DIR (consumed by IntegrityService) controls
// the rescan root.
const INTEGRITY_SWEEP_CRON = process.env.INTEGRITY_SWEEP_CRON || '0 3 * * *';
// Cron for the dislike cleanup sweep (default: every 6 hours). Configurable via CLEANUP_SWEEP_CRON.
const CLEANUP_SWEEP_CRON = process.env.CLEANUP_SWEEP_CRON || '0 */6 * * *';
// Cron for the stale vibe-session reaper (default: hourly). Configurable via
// VIBE_REAP_CRON. Spec §4 / Invariant B: ACTIVE batches with no interaction for
// 24h must transition to RESOLVED so returning users start fresh sessions.
const VIBE_REAP_CRON = process.env.VIBE_REAP_CRON || '0 * * * *';
// Worker concurrency - how many jobs to process in parallel
const WORKER_CONCURRENCY = parseInt(process.env.WORKER_CONCURRENCY || '10', 10);

// A Pool, not a single Client. The worker processes jobs with
// `concurrency: 10` on one event loop, so a shared Client would multiplex every
// job's queries onto one connection: a `BEGIN` issued by one job (see
// cleanup.service.ts) would enrol other jobs' unrelated queries in that
// transaction and discard them on `ROLLBACK`. Pool gives each transaction its
// own connection. Same reasoning as backend/src/app.ts.
//
// Sized at least as large as the job concurrency so concurrent jobs never
// serialise waiting for a connection.
const pgPool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: Math.max(WORKER_CONCURRENCY + 2, 10),
});

pgPool.on('error', (err) => {
  // Idle-client errors would otherwise be an unhandled 'error' event and crash
  // the worker; the pool discards the client itself.
  console.error('[DB] Idle pool client error:', err);
});

async function initWorker() {
  // Fail fast if the database is unreachable at boot (Pool is lazy otherwise).
  await pgPool.query('SELECT 1');
  console.log('Worker connected to PostgreSQL');

  const scannerService = new ScannerService(pgPool, queue);
  const enrichmentService = new EnrichmentService(pgPool);
  const audioFeaturesService = new AudioFeaturesService(pgPool);
  await audioFeaturesService.ensureSchema();

  // Self-provision the enrichment schema additions (artist_similar + reused
  // columns) at startup, alongside the integrity table.
  await enrichmentService.ensureSchema();
  await new IntegrityService(pgPool, queue).ensureSchema();

  const worker = new Worker(QUEUE_NAME, async (job: Job<any>) => {
    // console.log(`Processing job: ${job.name} (ID: ${job.id})`);
    console.log(`Processing job: ${job.name} (ID: ${job.id})`);
    
    switch (job.name) {
      case 'scan_library': {
        const payload = job.data as LibraryScanJob;
        console.log(`[Scanner] Starting library scan in: ${payload.directory}`);
        await scannerService.scanDirectory(payload.directory);
        // Re-sync Typesense after every scan so search reflects new/changed tracks.
        await queue.add('reindex_tracks', {}, { removeOnComplete: { age: 86400, count: 100 }, removeOnFail: { age: 86400, count: 100 } });
        console.log(`[Scanner] Enqueued Typesense reindex after scan`);
        break;
      }
      case 'metadata_refresh': {
        const payload = job.data as MetadataRefreshJob;
        console.log(`[Metadata] Refreshing track: ${payload.trackId} (Mode: ${payload.refreshType})`);

        // Real best-effort enrichment via the external-integration clients.
        // Each provider is isolated inside the service so one failing source
        // never aborts the others or fails the job.
        await enrichmentService.enrichTrack(payload.trackId);
        console.log(`[Metadata] Successfully refreshed track: ${payload.trackId}`);
        break;
      }
      case 'artist_similarity': {
        const payload = job.data as ArtistSimilarityJob;
        console.log(`[Similarity] Refreshing similar artists for: ${payload.artistId}`);
        await enrichmentService.refreshArtistSimilarity(payload.artistId);
        break;
      }
      case 'artist_image': {
        const payload = job.data as ArtistImageJob;
        console.log(`[ArtistImage] Refreshing image for artist: ${payload.artistId}`);
        await enrichmentService.refreshArtistImage(payload.artistId);
        break;
      }
      case 'album_cover': {
        const payload = job.data as AlbumCoverJob;
        console.log(`[AlbumCover] Refreshing cover for album: ${payload.albumId}`);
        await enrichmentService.refreshAlbumCover(payload.albumId);
        break;
      }
      case 'audio_analysis': {
        const payload = job.data as AudioAnalysisJob;
        console.log(`[Audio] Analyzing track: ${payload.trackId}`);
        await audioFeaturesService.extractAndPersist(payload.trackId);
        console.log(`[Audio] Successfully analyzed track: ${payload.trackId}`);
        break;
      }
      case 'integrity_sweep': {
        // Periodic self-healing pass: detect corrupt/missing track metadata,
        // auto-fix via rescan + SQL strip, flag the rest for manual review.
        console.log('[Integrity] Starting integrity sweep');
        const integrityService = new IntegrityService(pgPool, queue);
        const summary = await integrityService.runSweep();
        if (summary.aborted) {
          console.error(`[Integrity] Sweep ABORTED by safety guard: ${summary.abortReason}`);
        } else {
          console.log(
            `[Integrity] Sweep complete: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}`
          );
        }
        break;
      }
      case 'cleanup_sweep': {
        console.log('[Cleanup] Starting dislike cleanup sweep');
        const cleanupService = new CleanupSweepService(pgPool);
        const result = await cleanupService.runSweep();
        console.log(
          `[Cleanup] Sweep complete: warned=${result.warned} eligible=${result.eligible} ` +
            `deleted=${result.deleted} failed=${result.failed} reaped=${result.reaped}` +
            (result.dryRun ? ' (DRY RUN — MUZICK_ALLOW_HARD_DELETE off, nothing deleted)' : '')
        );
        break;
      }
      case 'vibe_reap': {
        // Stale-session reaper (Invariant B). Transitions ACTIVE batches with no
        // interaction for 24h to RESOLVED. Runs hourly; the SQL is idempotent.
        const reaped = await pgPool.query(
          `UPDATE recommendation_batch
              SET status = 'RESOLVED'
            WHERE status = 'ACTIVE'
              AND last_interaction_at < NOW() - INTERVAL '24 hours'`
        );
        const n = reaped.rowCount ?? 0;
        if (n > 0) console.log(`[VibeReap] Reaped ${n} stale ACTIVE session(s)`);
        break;
      }
      case 'reindex_tracks': {
        // Sync all LIBRARY tracks from Postgres into the Typesense 'tracks'
        // collection. Inlined to avoid duplicating a service module.
        const { Client } = await import('typesense');
        const searchHost = process.env.TYPESENSE_HOST || 'localhost';
        const searchPort = parseInt(process.env.TYPESENSE_PORT || '8108', 10);
        const searchApiKey = process.env.TYPESENSE_API_KEY || '';

        const tsClient = new Client({
          nodes: [{ host: searchHost, port: searchPort, protocol: 'http' }],
          apiKey: searchApiKey,
        });

        // Ensure collection schema exists.
        try {
          await tsClient.collections().create({
            name: 'tracks',
            fields: [
              { name: 'id', type: 'string' },
              { name: 'title', type: 'string' },
              { name: 'artist', type: 'string' },
              { name: 'album', type: 'string' },
              { name: 'duration', type: 'int32' },
              { name: 'play_count', type: 'int32' },
              { name: 'genre', type: 'string[]', facet: true },
              { name: 'source_type', type: 'string' },
            ],
          });
        } catch (err: any) {
          if (!err?.message?.includes('already exists')) throw err;
        }

        const tracksRes = await pgPool.query(
          `SELECT t.id, t.title, t.artist, al.title AS album, t.duration, t.play_count, t.source_type
           FROM tracks t
           LEFT JOIN albums al ON al.id = t.album_id
           WHERE t.state = 'LIBRARY'`
        );
        const tracks = tracksRes.rows;

        const genreMap = new Map<string, string[]>();
        const genreRes = await pgPool.query(
          `SELECT tg.track_id, g.name FROM track_genre tg JOIN genre g ON g.id = tg.genre_id`
        );
        for (const row of genreRes.rows) {
          const arr = genreMap.get(row.track_id) || [];
          arr.push(row.name);
          genreMap.set(row.track_id, arr);
        }

        const collection = tsClient.collections('tracks').documents();
        const batchSize = 100;
        let indexed = 0;

        for (let i = 0; i < tracks.length; i += batchSize) {
          const batch = tracks.slice(i, i + batchSize);
          const documents = batch.map((t: any) => ({
            id: String(t.id),
            title: t.title,
            artist: t.artist,
            album: t.album ?? '',
            duration: Math.round(Number(t.duration)),
            play_count: Number(t.play_count),
            genre: genreMap.get(String(t.id)) || [],
            source_type: t.source_type,
          }));
          await collection.import(documents, { action: 'upsert' });
          indexed += documents.length;
        }

        console.log(`[Reindex] Indexed ${indexed} tracks into Typesense`);
        break;
      }
      case 'reprocess_artists': {
        await reprocessArtists(job.data as ReprocessArtistsJob, {
          pgPool,
          queue,
          enrichmentService,
        });
        break;
      }
      default:
        console.log(`Received job of type: ${job.name} with data:`, job.data);
        break;
    }
  }, { connection, concurrency: WORKER_CONCURRENCY });

  worker.on('completed', (job) => {
    console.log(`Job ${job.id} has completed!`);
  });

  worker.on('failed', (job, err) => {
    console.log(`Job ${job?.id} has failed with error: ${err.message}`);
  });

  // Register the repeatable integrity sweep. bullmq's upsertJobScheduler is
  // idempotent: keying on a fixed scheduler id ('integrity-sweep') means worker
  // restarts update the existing schedule in place rather than stacking
  // duplicate repeatable jobs.
  await queue.upsertJobScheduler(
    'integrity-sweep',
    { pattern: INTEGRITY_SWEEP_CRON },
    { name: 'integrity_sweep', data: { reason: 'scheduled' } }
  );
  console.log(`[Integrity] Sweep scheduled with cron: ${INTEGRITY_SWEEP_CRON}`);

  await queue.upsertJobScheduler(
    'cleanup-sweep',
    { pattern: CLEANUP_SWEEP_CRON },
    { name: 'cleanup_sweep', data: { reason: 'scheduled' } }
  );
  console.log(`[Cleanup] Sweep scheduled with cron: ${CLEANUP_SWEEP_CRON}`);

  // Stale vibe-session reaper (Invariant B): hourly sweep that transitions
  // ACTIVE batches idle for 24h to RESOLVED.
  await queue.upsertJobScheduler(
    'vibe-reap',
    { pattern: VIBE_REAP_CRON },
    { name: 'vibe_reap', data: { reason: 'scheduled' } }
  );
  console.log(`[VibeReap] Reaper scheduled with cron: ${VIBE_REAP_CRON}`);

  // Enqueue a startup scan only when the database is empty (fresh volume after
  // deleting data/postgres, or first deploy). On subsequent restarts the library
  // is already populated, so an unconditional scan would be wasteful.
  const startupDir = process.env.MUSIC_DIR || '/music';
  const trackCount = await pgPool.query('SELECT COUNT(*)::int AS n FROM tracks');
  if (trackCount.rows[0].n === 0) {
    await queue.add('scan_library', { directory: startupDir }, {
      removeOnComplete: { age: 86400, count: 100 },
      removeOnFail: { age: 86400, count: 100 },
    });
    console.log(`[Startup] Enqueued library scan (fresh DB): ${startupDir}`);
  } else {
    console.log(`[Startup] Library has ${trackCount.rows[0].n} tracks — skipping startup scan`);
  }

  console.log('Worker is running and listening for jobs...');

  // Graceful shutdown: close the worker, the scheduler queue and the pg client
  // on termination signals. Guarded so a second signal is a no-op.
  let shuttingDown = false;
  const shutdown = async (signal: string) => {
    if (shuttingDown) return;
    shuttingDown = true;
    console.log(`[Shutdown] Received ${signal}, shutting down gracefully...`);
    try {
      await worker.close();
      await queue.close();
      await pgPool.end();
      console.log('[Shutdown] Clean shutdown complete.');
    } catch (err) {
      console.error('[Shutdown] Error during shutdown:', err);
    }
    process.exit(0);
  };

  process.on('SIGTERM', () => void shutdown('SIGTERM'));
  process.on('SIGINT', () => void shutdown('SIGINT'));
}

initWorker().catch(err => {
  console.error('Failed to initialize worker:', err);
  process.exit(1);
});
