diff --git a/services/worker/src/workers/thumbnail.js b/services/worker/src/workers/thumbnail.js new file mode 100644 index 0000000..da103e0 --- /dev/null +++ b/services/worker/src/workers/thumbnail.js @@ -0,0 +1,59 @@ +import { join } from 'path'; +import { unlink } from 'fs/promises'; +import { tmpdir } from 'os'; +import { query } from '../db/client.js'; +import { downloadFromS3, uploadToS3 } from '../s3/client.js'; +import { extractFrameAtTime } from '../ffmpeg/executor.js'; + +const S3_BUCKET = process.env.S3_BUCKET || 'wild-dragon'; + +export const thumbnailWorker = async (job) => { + const { assetId, proxyKey, outputKey } = job.data; + + const tmpDir = tmpdir(); + const inputPath = join(tmpDir, `thumb-input-${job.id}.mp4`); + const outputPath = join(tmpDir, `thumb-output-${job.id}.jpg`); + + try { + // Download proxy from S3 + job.updateProgress(10); + console.log(`[thumbnail] Downloading ${proxyKey} for asset ${assetId}`); + await downloadFromS3(S3_BUCKET, proxyKey, inputPath); + + // Extract frame at 5 seconds + job.updateProgress(40); + console.log(`[thumbnail] Extracting frame for asset ${assetId}`); + await extractFrameAtTime(inputPath, outputPath, '00:00:05'); + + // Upload to S3 + job.updateProgress(70); + console.log(`[thumbnail] Uploading to ${outputKey}`); + await uploadToS3(S3_BUCKET, outputKey, outputPath); + + // Update database + job.updateProgress(90); + console.log(`[thumbnail] Updating asset record for ${assetId}`); + await query( + 'UPDATE assets SET thumbnail_s3_key = $1 WHERE id = $2', + [outputKey, assetId] + ); + + job.updateProgress(100); + console.log(`[thumbnail] Asset ${assetId} thumbnail complete`); + + return { assetId, outputKey }; + } catch (error) { + console.error(`[thumbnail] Error processing asset ${assetId}:`, error); + throw error; + } finally { + // Cleanup + try { + await Promise.all([ + unlink(inputPath).catch(() => {}), + unlink(outputPath).catch(() => {}), + ]); + } catch (err) { + console.error(`[thumbnail] Cleanup error for job ${job.id}:`, err); + } + } +};