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 { withTransaction } from './db.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} deleted=${result.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': {
        const payload = job.data as ReprocessArtistsJob;
        const batchSize = payload.batchSize ?? 100;
        const offset = payload.offset ?? 0;
        console.log(`[ReprocessArtists] Starting artist reprocessing (batch=${batchSize}, offset=${offset})`);
        
        const artistsRes = await pgPool.query(
          `SELECT id, name, canonical_name, mbid FROM artists ORDER BY id LIMIT $1 OFFSET $2`,
          [batchSize, offset]
        );
        
        let processed = 0;
        let updated = 0;
        let merged = 0;
        
        for (const artist of artistsRes.rows) {
          processed++;
          try {
            const result = await enrichmentService.resolveArtistIdentity(artist.name);
            // Merge if resolved to a different artist (duplicate detected)
            if (result.artistId !== artist.id) {
              merged++;
              // Whole merge is one transaction on one dedicated connection: the
              // intermediate states (track links moved but albums not yet, or
              // vice versa) must never be visible, and a failure part-way must
              // not leave an artist half-merged. ON DELETE CASCADE makes a
              // partial merge destructive.
              await withTransaction(pgPool, async (client) => {
                // Move track links to the keeper, skipping any (track, role) the
                // keeper already has, then drop the loser's — a plain UPDATE would
                // violate track_artists_pkey when both are on the same track.
                await client.query(
                  `INSERT INTO track_artists (track_id, artist_id, role)
                   SELECT track_id, $1, role FROM track_artists WHERE artist_id = $2
                   ON CONFLICT (track_id, artist_id, role) DO NOTHING`,
                  [result.artistId, artist.id]
                );
                await client.query(
                  `DELETE FROM track_artists WHERE artist_id = $1`,
                  [artist.id]
                );
                // Fold albums into the keeper. Move tracks of any same-title album
                // to the keeper's matching album first (UNIQUE(artist_id,title) and
                // ON DELETE CASCADE mean a blind UPDATE could collide or, worse,
                // cascade-delete tracks when the loser artist is removed).
                const dupAlbums = await client.query(
                  `SELECT l.id AS loser_id, k.id AS keeper_id
                   FROM albums l JOIN albums k
                     ON k.artist_id = $1 AND lower(k.title) = lower(l.title)
                   WHERE l.artist_id = $2`,
                  [result.artistId, artist.id]
                );
                for (const { loser_id, keeper_id } of dupAlbums.rows) {
                  await client.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]);
                  await client.query(`DELETE FROM albums WHERE id = $1`, [loser_id]);
                }
                // Remaining (non-colliding) albums move over cleanly.
                await client.query(
                  `UPDATE albums SET artist_id = $1 WHERE artist_id = $2`,
                  [result.artistId, artist.id]
                );
                await client.query(
                  `DELETE FROM artists WHERE id = $1`,
                  [artist.id]
                );
              });
              console.log(`[ReprocessArtists] Merged "${artist.name}" (${artist.id}) -> "${result.canonicalName}" (${result.artistId})`);
            } else {
              // Update the existing artist record with new canonical_name and/or mbid
              const updates: string[] = [];
              const params: any[] = [];
              let paramIdx = 1;
              
              if (result.canonicalName && artist.canonical_name !== result.canonicalName) {
                updates.push(`canonical_name = $${paramIdx++}`);
                params.push(result.canonicalName);
              }
              if (result.mbid && artist.mbid !== result.mbid) {
                updates.push(`mbid = $${paramIdx++}`);
                params.push(result.mbid);
              }
              if (result.sortName && artist.sort_name !== result.sortName) {
                updates.push(`sort_name = $${paramIdx++}`);
                params.push(result.sortName);
              }
              if (updates.length > 0) {
                updates.push(`updated_at = CURRENT_TIMESTAMP`);
                params.push(artist.id);
                await pgPool.query(
                  `UPDATE artists SET ${updates.join(', ')} WHERE id = $${paramIdx}`,
                  params
                );
                updated++;
                console.log(`[ReprocessArtists] Updated "${artist.name}" (${artist.id}): ${updates.join(', ')}`);
              }
            }
            
            // Refresh the artist image via the dedicated job rather than inline,
            // so the reprocess batches aren't blocked on image HTTP. Deduped by
            // jobId across the run.
            await queue.add(
              'artist_image',
              { artistId: result.artistId } satisfies ArtistImageJob,
              {
                jobId: `artist-image-${result.artistId}`,
                removeOnComplete: { age: 86400, count: 5000 },
                removeOnFail: { age: 86400, count: 5000 },
              }
            );
          } catch (err) {
            console.error(`[ReprocessArtists] Failed for artist ${artist.id} (${artist.name}):`, err);
          }
        }
        
        console.log(`[ReprocessArtists] Batch complete: processed=${processed}, updated=${updated}, merged=${merged}`);
        
        // If we processed a full batch, enqueue the next one
        if (artistsRes.rows.length === batchSize) {
          await queue.add('reprocess_artists', { batchSize, offset: offset + batchSize }, {
            removeOnComplete: { age: 86400, count: 100 },
            removeOnFail: { age: 86400, count: 100 },
          });
          console.log(`[ReprocessArtists] Enqueued next batch at offset ${offset + batchSize}`);
        } else {
          // Final batch - run deduplication pass to merge artists with same normalized_name
          console.log(`[ReprocessArtists] All batches complete, running deduplication pass...`);
          
          const dupRes = await pgPool.query(
            `SELECT normalized_name, array_agg(id ORDER BY 
              CASE WHEN mbid IS NOT NULL THEN 0 ELSE 1 END,
              CASE WHEN canonical_name IS NOT NULL THEN 0 ELSE 1 END,
              id
            ) as ids
             FROM artists
             GROUP BY normalized_name
             HAVING COUNT(*) > 1`
          );
          
          let dedupMerged = 0;
          for (const row of dupRes.rows) {
            const ids = row.ids;
            const keepId = ids[0]; // First one (prefers MBID, then canonical_name, then lowest id)
            const mergeIds = ids.slice(1);
            
            for (const mergeId of mergeIds) {
              if (mergeId === keepId) continue;
              try {
                // One transaction per merge, on a dedicated connection: a
                // partial merge is destructive (ON DELETE CASCADE).
                await withTransaction(pgPool, async (client) => {
                  // First, handle track_artists conflicts: if both artists are on same track,
                  // keep the 'main' role, or merge roles. Use ON CONFLICT DO NOTHING to skip duplicates.
                  await client.query(
                    `INSERT INTO track_artists (track_id, artist_id, role)
                     SELECT track_id, $1, role FROM track_artists WHERE artist_id = $2
                     ON CONFLICT (track_id, artist_id, role) DO NOTHING`,
                    [keepId, mergeId]
                  );
                  // Then delete the old track_artists entries
                  await client.query(
                    `DELETE FROM track_artists WHERE artist_id = $1`,
                    [mergeId]
                  );

                  // Fold same-title albums (move tracks) before reassigning the rest,
                  // to avoid UNIQUE(artist_id,title) collisions / cascade deletes.
                  const dupAlbums = await client.query(
                    `SELECT l.id AS loser_id, k.id AS keeper_id
                     FROM albums l JOIN albums k
                       ON k.artist_id = $1 AND lower(k.title) = lower(l.title)
                     WHERE l.artist_id = $2`,
                    [keepId, mergeId]
                  );
                  for (const { loser_id, keeper_id } of dupAlbums.rows) {
                    await client.query(`UPDATE tracks SET album_id = $1 WHERE album_id = $2`, [keeper_id, loser_id]);
                    await client.query(`DELETE FROM albums WHERE id = $1`, [loser_id]);
                  }
                  await client.query(
                    `UPDATE albums SET artist_id = $1 WHERE artist_id = $2`,
                    [keepId, mergeId]
                  );
                  await client.query(
                    `DELETE FROM artists WHERE id = $1`,
                    [mergeId]
                  );
                });
                dedupMerged++;
                console.log(`[ReprocessArtists] Dedup merged ${mergeId} -> ${keepId} (normalized: ${row.normalized_name})`);
              } catch (err) {
                console.error(`[ReprocessArtists] Dedup failed for ${mergeId}:`, err);
              }
            }
          }
          
          console.log(`[ReprocessArtists] Deduplication complete: merged=${dedupMerged}`);

          // Album dedup pass: merge duplicate album rows (same title or same
          // MBID) that accumulated before the albumartist scanner fix. Run
          // after artist dedup so album artist_id refs are already resolved.
          try {
            const albumMerged = await enrichmentService.dedupAlbums();
            console.log(`[ReprocessArtists] Album dedup: merged=${albumMerged}`);
          } catch (err) {
            console.error('[ReprocessArtists] Album dedup failed:', err);
          }

          console.log(`[ReprocessArtists] All artists processed!`);
        }
        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);
});