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-" // 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: scale each video stream to a consistent // resolution + pixel format, then concat them. The audio leg only // runs if audio is being kept. const targetW = isProRes ? 1920 : 1920; const targetH = 1080; const vLabels = []; const aLabels = []; let normalize = ''; for (let i = 0; i < concatList.length; i++) { // scale=W:H force_original_aspect_ratio=decrease + pad to box keeps // mixed-aspect sources inside the frame without distortion. normalize += `[${i}:v:0]scale=${targetW}:${targetH}:force_original_aspect_ratio=decrease,pad=${targetW}:${targetH}:(ow-iw)/2:(oh-ih)/2,setsar=1,format=yuv420p[v${i}];`; vLabels.push(`[v${i}]`); if (wantAudio) { // anullsrc as a fallback so missing audio doesn't blow up concat. normalize += `[${i}:a:0]aresample=async=1:first_pts=0[a${i}];`; aLabels.push(`[a${i}]`); } } const n = concatList.length; let concatExpr; if (wantAudio) { // interleaved [v0][a0][v1][a1]… 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(() => {}), ]); } };