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

65 lines
2 KiB
JavaScript
Raw Normal View History

import { join } from 'path';
import { tmpdir } from 'os';
import { unlink } from 'fs/promises';
import { query } from '../db/client.js';
import { downloadFromS3, uploadToS3 } from '../s3/client.js';
import { trimSegment } from '../ffmpeg/executor.js';
const S3_BUCKET = process.env.S3_BUCKET || 'wild-dragon';
export const trimWorker = async (job) => {
const { clipInstanceId, assetId, sourceInFrames, sourceOutFrames } = job.data;
const jobId = job.id;
const tmpDir = tmpdir();
const downloadPath = join(tmpDir, `trim-${jobId}-src`);
const outputPath = join(tmpDir, `trim-${jobId}.mov`);
try {
const assetRes = await query(
'SELECT original_s3_key FROM assets WHERE id = $1 LIMIT 1',
[assetId]
);
if (assetRes.rows.length === 0) {
throw new Error(`Asset not found: ${assetId}`);
}
const { original_s3_key: sourceKey } = assetRes.rows[0];
await job.updateProgress(10);
console.log(`[trim] Downloading asset ${assetId} from S3 (${sourceKey})`);
await downloadFromS3(S3_BUCKET, sourceKey, downloadPath);
await job.updateProgress(40);
console.log(`[trim] Trimming frames ${sourceInFrames}${sourceOutFrames}`);
await trimSegment(downloadPath, outputPath, sourceInFrames, sourceOutFrames);
await job.updateProgress(70);
const s3Key = `temp-segments/${clipInstanceId}.mov`;
console.log(`[trim] Uploading trimmed segment to ${s3Key}`);
await uploadToS3(S3_BUCKET, s3Key, outputPath);
await job.updateProgress(85);
await query(
`INSERT INTO temp_segments (clip_instance_id, s3_key, expires_at)
VALUES ($1, $2, NOW() + INTERVAL '24 hours')`,
[clipInstanceId, s3Key]
);
await job.updateProgress(100);
console.log(`[trim] Job ${jobId} complete for clip ${clipInstanceId}`);
return { clipInstanceId, s3Key };
} catch (error) {
console.error(`[trim] Error in job ${jobId}:`, error);
throw error;
} finally {
await Promise.all([
unlink(downloadPath).catch(() => {}),
unlink(outputPath).catch(() => {}),
]);
}
};