dragonflight/services/worker/src/workers/conform.js

import { join } from 'path';
import { unlink, writeFile, mkdir, rm } from 'fs/promises';
import { tmpdir } from 'os';
import { query } from '../db/client.js';
import { downloadFromS3, uploadToS3 } from '../s3/client.js';
import { trimSegment, concatSegments, runFFmpeg } from '../ffmpeg/executor.js';
import { parseEDL } from '../edl/parser.js';
import { XMLParser } from 'fast-xml-parser';

const S3_BUCKET = process.env.S3_BUCKET || 'wild-dragon';

const xmlParser = new XMLParser({
  ignoreAttributes: false,
  attributeNamePrefix: '@_',
});

function parseFcpXml(xmlContent) {
  const doc = xmlParser.parse(xmlContent);
  const sequence = doc?.xmeml?.sequence;
  if (!sequence) throw new Error('Invalid FCP XML: no sequence element');

  const name = sequence.name || 'Untitled';
  const rate = sequence?.rate?.timebase ? parseInt(sequence.rate.timebase, 10) : 29.97;
  const width = parseInt(sequence?.media?.video?.format?.samplecharacteristics?.width || 1920, 10);
  const height = parseInt(sequence?.media?.video?.format?.samplecharacteristics?.height || 1080, 10);

  const clips = [];
  const videoTracks = sequence?.media?.video?.track || [];
  const tracks = Array.isArray(videoTracks) ? videoTracks : [videoTracks];

  for (const track of tracks) {
    const trackNum = parseInt(track?.['@_currentExplodedTrackIndex'] || 0, 10);
    const trackItems = track?.clipitem || [];
    const items = Array.isArray(trackItems) ? trackItems : [trackItems];

    for (const item of items) {
      if (!item) continue;
      const fileUrl = item?.file?.name || item?.file?.pathurl || '';
      const fileName = fileUrl.split('/').pop() || fileUrl.split('\\').pop() || 'unknown';
      const srcIn = parseFrame(item?.in?.toString() || '0', rate);
      const srcOut = parseFrame(item?.out?.toString() || '0', rate);
      const recIn = parseFrame(item?.start?.toString() || '0', rate);
      const recOut = parseFrame(item?.end?.toString() || '0', rate);
      const duration = parseFrame(item?.duration?.toString() || '0', rate);

      if (srcOut <= srcIn || recOut <= recIn) continue;

      clips.push({
        trackIndex: trackNum,
        fileName,
        fileUrl,
        sourceInFrames: srcIn,
        sourceOutFrames: srcOut,
        timelineInFrames: recIn,
        timelineOutFrames: recOut,
        duration,
      });
    }
  }

  return { name, frameRate: rate, width, height, clips };
}

function parseFrame(value, fps) {
  // FCP XML stores timecode or frame count
  const trimmed = value.trim();
  // If it's a plain number, return as-is
  if (/^\d+$/.test(trimmed)) return parseInt(trimmed, 10);
  // HH:MM:SS:FF or HH:MM:SS;FF
  const parts = trimmed.split(/[:;]/);
  if (parts.length === 4) {
    const hh = parseInt(parts[0], 10);
    const mm = parseInt(parts[1], 10);
    const ss = parseInt(parts[2], 10);
    const ff = parseInt(parts[3], 10);
    return hh * 3600 * fps + mm * 60 * fps + ss * fps + ff;
  }
  return 0;
}

export const conformWorker = async (job) => {
  const { edl, fcpXml, projectId, sequenceId, sequenceName, frameRate, codec, quality, resolution, audio } = job.data;
  const jobId = job.id;

  const tmpDir = tmpdir();
  const segmentsDir = join(tmpDir, `segments-${jobId}`);
  const segmentListPath = join(tmpDir, `segments-${jobId}.txt`);
  const outputPath = join(tmpDir, `output-${jobId}.mp4`);

  try {
    let edits = [];
    let seqName = sequenceName || 'Conformed';
    let seqFps = parseFloat(frameRate) || 29.97;

    // ── Resolve edits ────────────────────────────────────────────────
    //
    // Preference order:
    //   1) sequenceId — read sequence_clips, which the Premiere panel
    //      populated with authoritative asset_id mappings on push. This
    //      avoids any filename matching, which is brittle because the
    //      panel's local Premiere file paths (e.g. "dragonflight-<name>"
    //      with sanitised characters) do not match the original MAM
    //      filenames in the assets table.
    //   2) edl — legacy EDL input, filename-resolved.
    //   3) fcpXml — parse the XML for clipitems, filename-resolved.
    //
    // The XML is still parsed when sequenceId is also provided, because
    // we want its sequence name + frame rate metadata even when the
    // authoritative clip list comes from the DB.
    if (sequenceId) {
      await job.updateProgress(5);
      console.log(`[conform] Resolving edits from sequence_clips for sequence ${sequenceId}`);
      const clipRows = await query(
        `SELECT sc.asset_id, sc.source_in_frames, sc.source_out_frames,
                sc.timeline_in_frames, sc.timeline_out_frames, sc.track,
                a.original_s3_key, a.filename
           FROM sequence_clips sc
           JOIN assets a ON a.id = sc.asset_id
          WHERE sc.sequence_id = $1
          ORDER BY sc.timeline_in_frames ASC, sc.track ASC`,
        [sequenceId]
      );
      if (!clipRows.rows.length) {
        throw new Error('Sequence has no clips. Push the timeline from Premiere first.');
      }
      edits = clipRows.rows.map((r, i) => ({
        editNumber: i + 1,
        reelName: r.filename,
        asset_id: r.asset_id,
        original_s3_key: r.original_s3_key,
        sourceIn: r.source_in_frames,
        sourceOut: r.source_out_frames,
      }));
      // Parse XML for sequence-level metadata if it's also provided.
      if (fcpXml) {
        try {
          const parsed = parseFcpXml(fcpXml);
          seqName = parsed.name || seqName;
          seqFps  = parsed.frameRate || seqFps;
        } catch (e) {
          console.warn(`[conform] XML metadata parse skipped: ${e.message}`);
        }
      }
    } else if (edl) {
      await job.updateProgress(5);
      console.log(`[conform] Parsing EDL for job ${jobId}`);
      edits = parseEDL(edl).map((e, i) => ({
        editNumber: e.editNumber || i + 1,
        reelName: e.reelName,
        sourceIn: e.sourceIn,
        sourceOut: e.sourceOut,
      }));
    } else if (fcpXml) {
      await job.updateProgress(5);
      console.log(`[conform] Parsing FCP XML for job ${jobId}`);
      const parsed = parseFcpXml(fcpXml);
      seqName = parsed.name || seqName;
      seqFps = parsed.frameRate || seqFps;
      edits = parsed.clips.map((c, i) => ({
        editNumber: i + 1,
        reelName: c.fileName,
        sourceIn: c.sourceInFrames,
        sourceOut: c.sourceOutFrames,
      }));
    } else {
      throw new Error('No input provided — expected edl or fcpXml in job data');
    }

    await mkdir(segmentsDir, { recursive: true });

    let processedEdits = 0;
    const concatList = [];

    for (const edit of edits) {
      await job.updateProgress(Math.min(5 + (processedEdits / edits.length) * 50, 55));
      console.log(`[conform] Processing edit ${edit.editNumber}: ${edit.reelName}`);

      // If the edit was resolved from sequence_clips above, the asset's
      // original_s3_key is already attached — skip the filename lookup
      // entirely (it would 0-match anyway because the panel's reelName
      // is the local Premiere file path with "dragonflight-" prefix).
      let sourceKey = edit.original_s3_key || null;

      if (!sourceKey) {
        // Legacy path (EDL or fcpXml without sequenceId): match by filename,
        // preferring same-project assets to avoid cross-project collisions.
        let assetRes;
        if (projectId) {
          assetRes = await query(
            `SELECT id, original_s3_key FROM assets
             WHERE filename = $1 AND project_id = $2
             LIMIT 1`,
            [edit.reelName, projectId]
          );
          if (assetRes.rows.length === 0) {
            assetRes = await query(
              'SELECT id, original_s3_key FROM assets WHERE filename = $1 LIMIT 1',
              [edit.reelName]
            );
          }
        } else {
          assetRes = await query(
            'SELECT id, original_s3_key FROM assets WHERE filename = $1 LIMIT 1',
            [edit.reelName]
          );
        }

        if (assetRes.rows.length === 0) {
          throw new Error(`Asset not found for reel: ${edit.reelName}`);
        }
        sourceKey = assetRes.rows[0].original_s3_key;
      }

      const segmentInputPath = join(segmentsDir, `segment-${edit.editNumber}-src`);
      const segmentOutputPath = join(segmentsDir, `segment-${edit.editNumber}.mov`);

      console.log(`[conform] Downloading segment ${edit.editNumber} from S3 (${sourceKey})`);
      await downloadFromS3(S3_BUCKET, sourceKey, segmentInputPath);

      console.log(`[conform] Trimming ${edit.editNumber}: ${edit.sourceIn} → ${edit.sourceOut}`);
      await trimSegment(segmentInputPath, segmentOutputPath, edit.sourceIn, edit.sourceOut);

      concatList.push(segmentOutputPath);
      await unlink(segmentInputPath).catch(() => {});
      processedEdits++;
    }

    await job.updateProgress(60);
    console.log(`[conform] Writing concat list for ${concatList.length} segments`);
    const concatContent = concatList.map(p => `file '${p}'`).join('\n');
    await writeFile(segmentListPath, concatContent, 'utf-8');

    await job.updateProgress(70);
    console.log(`[conform] Concatenating segments for job ${jobId}`);

    // Audio: be permissive. Anything that isn't an explicit 'none' should
    // get encoded — the panel sends 'broadcast' (default), 'include' is the
    // legacy value, and there's no reason to silently drop audio for any
    // other label. 320k AAC is a safe broadcast-quality default in mp4.
    const audioFlag = (audio === 'none' || audio === 'off')
      ? ['-an']
      : ['-c:a', 'aac', '-b:a', '320k', '-ar', '48000'];

    // Codec map. The panel sends 'prores_hq' / 'prores_4444' / 'h264' / 'h265'
    // / 'dnxhr_hq'; old EDL callers send 'prores' / 'h265' / 'h264'. Match
    // both. prores_ks profiles: 0=proxy 1=lt 2=std 3=hq 4=4444.
    let videoCodec, profileFlag = [];
    if (codec === 'prores_hq' || codec === 'prores') {
      videoCodec = 'prores_ks'; profileFlag = ['-profile:v', '3'];
    } else if (codec === 'prores_4444') {
      videoCodec = 'prores_ks'; profileFlag = ['-profile:v', '4'];
    } else if (codec === 'h265' || codec === 'hevc') {
      videoCodec = 'libx265';
    } else if (codec === 'dnxhr_hq') {
      videoCodec = 'dnxhd'; profileFlag = ['-profile:v', 'dnxhr_hq'];
    } else {
      videoCodec = 'libx264';
    }

    // prores_ks ignores -crf and uses -preset differently; libx264/x265 use
    // crf-based quality. Branch the encode args.
    const isProRes = videoCodec === 'prores_ks';
    const qualityArgs = isProRes
      ? []  // ProRes profile already encodes the quality target
      : [
          '-preset', quality === 'high' ? 'slow' : quality === 'broadcast' ? 'veryslow' : 'fast',
          '-crf',    quality === 'broadcast' ? '18' : quality === 'high' ? '23' : '28',
        ];

    // Concat strategy: the demuxer is fast but requires identical specs
    // (codec, pixel format, framerate, resolution) across all segments.
    // When sources differ (AV1 + H.264, mixed framerates, etc.) it dies
    // with "Error sending frames to consumers: Invalid argument". Use the
    // concat *filter* instead — it normalises each input through a filter
    // graph before joining, so disparate sources work.
    //
    // Normalise every input to the target resolution + 1/SAR + yuv420p so
    // the encode is deterministic. Output frame rate is unset; ffmpeg
    // will use whatever the final filter graph emits, which the target
    // codec accepts.
    const wantAudio = !(audio === 'none' || audio === 'off');
    const inputArgs = [];
    concatList.forEach(p => { inputArgs.push('-i', p); });

    // Build the filter graph. The concat filter in ffmpeg 8.x requires
    // identical resolution, pixel format, SAR, FRAME RATE and audio
    // SAMPLE RATE / CHANNEL LAYOUT across all inputs. Different-spec
    // sources (e.g. a 23.98 fps clip + a 60 fps clip) trip
    //   [fc#0] Error sending frames to consumers: Invalid argument
    // even though our earlier scale+pad+format pass took care of the
    // pixel side. Force the time-domain axes too:
    //   fps=<target>       — resample video to a constant rate
    //   setpts=PTS-STARTPTS — re-zero PTS so concat's per-input clock
    //                        resets cleanly
    //   aresample=48000    — force a single audio sample rate
    //   asetpts=PTS-STARTPTS — same for audio
    const targetW = 1920;
    const targetH = 1080;
    const targetFps = Math.round(seqFps) || 30;
    const targetSampleRate = 48000;

    const vLabels = [];
    const aLabels = [];
    let normalize = '';
    for (let i = 0; i < concatList.length; i++) {
      normalize +=
        `[${i}:v:0]fps=${targetFps},` +
        `scale=${targetW}:${targetH}:force_original_aspect_ratio=decrease,` +
        `pad=${targetW}:${targetH}:(ow-iw)/2:(oh-ih)/2,` +
        `setsar=1,format=yuv420p,setpts=PTS-STARTPTS[v${i}];`;
      vLabels.push(`[v${i}]`);
      if (wantAudio) {
        normalize +=
          `[${i}:a:0]aresample=${targetSampleRate},` +
          `aformat=channel_layouts=stereo:sample_fmts=fltp,` +
          `asetpts=PTS-STARTPTS[a${i}];`;
        aLabels.push(`[a${i}]`);
      }
    }
    const n = concatList.length;
    let concatExpr;
    if (wantAudio) {
      const interleaved = [];
      for (let i = 0; i < n; i++) { interleaved.push(vLabels[i], aLabels[i]); }
      concatExpr = `${interleaved.join('')}concat=n=${n}:v=1:a=1[outv][outa]`;
    } else {
      concatExpr = `${vLabels.join('')}concat=n=${n}:v=1:a=0[outv]`;
    }
    const filterComplex = normalize + concatExpr;

    const mapArgs = wantAudio ? ['-map', '[outv]', '-map', '[outa]'] : ['-map', '[outv]'];
    const encodeAudio = wantAudio ? ['-c:a', 'aac', '-b:a', '320k', '-ar', '48000'] : ['-an'];

    await runFFmpeg([
      ...inputArgs,
      '-filter_complex', filterComplex,
      ...mapArgs,
      '-c:v', videoCodec,
      ...profileFlag,
      ...qualityArgs,
      ...encodeAudio,
      '-y', outputPath,
    ]);

    await job.updateProgress(85);
    const outputKey = `jobs/${jobId}/conformed.mp4`;
    console.log(`[conform] Uploading output to ${outputKey}`);
    await uploadToS3(S3_BUCKET, outputKey, outputPath);

    // Register the conformed output as a new asset
    const assetRes = await query(
      `INSERT INTO assets (project_id, filename, display_name, media_type, status, original_s3_key, codec, resolution, fps, duration_ms, conform_source_sequence_id)
       VALUES ($1, $2, $3, 'video', 'ready', $4, $5, $6, $7, $8, $9) RETURNING id`,
      [
        projectId || null,
        `conformed-${seqName.replace(/[^a-z0-9]/gi, '_')}.mp4`,
        `Conformed: ${seqName}`,
        outputKey,
        // Normalise the panel's codec id into the canonical name we store on
        // the asset row. Keep aligned with the encode branch above.
        (codec === 'prores_hq' || codec === 'prores_4444' || codec === 'prores') ? 'prores'
          : (codec === 'h265' || codec === 'hevc') ? 'hevc'
          : (codec === 'dnxhr_hq') ? 'dnxhd'
          : 'h264',
        resolution !== 'match' ? resolution : '1920x1080',
        seqFps,
        null,
        job.data.sequenceId || null,
      ]
    );

    await job.updateProgress(100);
    console.log(`[conform] Job ${jobId} complete → asset ${assetRes.rows[0].id}`);

    return { jobId, outputKey, assetId: assetRes.rows[0].id };

  } catch (error) {
    console.error(`[conform] Error in job ${jobId}:`, error);
    // BUG FIX #1: Mark the output asset (if any) as 'error' so the UI doesn't
    // show a perpetually-spinning 'processing' state when the conform fails.
    // We don't have an assetId until the INSERT succeeds, so target by job key.
    await query(
      `UPDATE assets
       SET status = 'error', updated_at = NOW()
       WHERE original_s3_key = $1`,
      [`jobs/${jobId}/conformed.mp4`]
    ).catch(e => console.error('[conform] Failed to mark asset error:', e.message));
    throw error;
  } finally {
    await Promise.all([
      unlink(segmentListPath).catch(() => {}),
      unlink(outputPath).catch(() => {}),
      rm(segmentsDir, { recursive: true, force: true }).catch(() => {}),
    ]);
  }
};