Blame for apps/api/dist/queue/clip-scoring.processor.js

"use strict";
var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
    if (k2 === undefined) k2 = k;
    var desc = Object.getOwnPropertyDescriptor(m, k);
    if (!desc || ("get" in desc ? !m.__esModule : desc.writable || desc.configurable)) {
      desc = { enumerable: true, get: function() { return m[k]; } };
    }
    Object.defineProperty(o, k2, desc);
}) : (function(o, m, k, k2) {
    if (k2 === undefined) k2 = k;
    o[k2] = m[k];
}));
var __setModuleDefault = (this && this.__setModuleDefault) || (Object.create ? (function(o, v) {
    Object.defineProperty(o, "default", { enumerable: true, value: v });
}) : function(o, v) {
    o["default"] = v;
});
var __decorate = (this && this.__decorate) || function (decorators, target, key, desc) {
    var c = arguments.length, r = c < 3 ? target : desc === null ? desc = Object.getOwnPropertyDescriptor(target, key) : desc, d;
    if (typeof Reflect === "object" && typeof Reflect.decorate === "function") r = Reflect.decorate(decorators, target, key, desc);
    else for (var i = decorators.length - 1; i >= 0; i--) if (d = decorators[i]) r = (c < 3 ? d(r) : c > 3 ? d(target, key, r) : d(target, key)) || r;
    return c > 3 && r && Object.defineProperty(target, key, r), r;
};
var __importStar = (this && this.__importStar) || (function () {
    var ownKeys = function(o) {
        ownKeys = Object.getOwnPropertyNames || function (o) {
            var ar = [];
            for (var k in o) if (Object.prototype.hasOwnProperty.call(o, k)) ar[ar.length] = k;
            return ar;
        };
        return ownKeys(o);
    };
    return function (mod) {
        if (mod && mod.__esModule) return mod;
        var result = {};
        if (mod != null) for (var k = ownKeys(mod), i = 0; i < k.length; i++) if (k[i] !== "default") __createBinding(result, mod, k[i]);
        __setModuleDefault(result, mod);
        return result;
    };
})();
var __metadata = (this && this.__metadata) || function (k, v) {
    if (typeof Reflect === "object" && typeof Reflect.metadata === "function") return Reflect.metadata(k, v);
};
var __param = (this && this.__param) || function (paramIndex, decorator) {
    return function (target, key) { decorator(target, key, paramIndex); }
};
var ClipScoringProcessor_1;
Object.defineProperty(exports, "__esModule", { value: true });
exports.ClipScoringProcessor = void 0;
const bullmq_1 = require("@nestjs/bullmq");
const common_1 = require("@nestjs/common");
const child_process_1 = require("child_process");
const util_1 = require("util");
const path = __importStar(require("path"));
const fs = __importStar(require("fs/promises"));
const os = __importStar(require("os"));
const prisma_service_1 = require("../prisma/prisma.service");
const r2_storage_service_1 = require("../common/storage/r2-storage.service");
const redis_constants_1 = require("../common/redis.constants");
const safe_fetch_1 = require("../common/safe-fetch");
const execAsync = (0, util_1.promisify)(child_process_1.execFile);
const RELEASE_LOCK_SCRIPT = `
if redis.call("get", KEYS[1]) == ARGV[1] then
  return redis.call("del", KEYS[1])
else
  return 0
end
`;
const EXTEND_LOCK_SCRIPT = `
if redis.call("get", KEYS[1]) == ARGV[1] then
  return redis.call("pexpire", KEYS[1], ARGV[2])
else
  return 0
end
`;
let ClipScoringProcessor = ClipScoringProcessor_1 = class ClipScoringProcessor extends bullmq_1.WorkerHost {
    prisma;
    r2Storage;
    logger = new common_1.Logger(ClipScoringProcessor_1.name);
    redis;
    releaseLockSha = null;
    extendLockSha = null;
    constructor(prisma, r2Storage, redis) {
        super();
        this.prisma = prisma;
        this.r2Storage = r2Storage;
        this.redis = redis;
    }
    async onModuleInit() {
        try {
            this.releaseLockSha = await this.redis.scriptLoad(RELEASE_LOCK_SCRIPT);
            this.extendLockSha = await this.redis.scriptLoad(EXTEND_LOCK_SCRIPT);
            this.logger.log("Lua scripts compiled for clip-scoring locks");
        }
        catch (err) {
            this.logger.warn("Failed to pre-compile Lua scripts, will use EVAL fallback");
        }
    }
    async process(job) {
        const { clipId, episodeId, videoUrl, transcriptText } = job.data;
        this.logger.log(`Processing clip scoring for ${clipId}`);
        const lockKey = `lock:clip:${clipId}`;
        const lockValue = `worker:${job.id}:${Date.now()}`;
        const lockTTL = 300000;
        const acquired = await this.redis.set(lockKey, lockValue, {
            NX: true,
            PX: lockTTL,
        });
        if (!acquired) {
            await job.log(`Skipping job ${job.id}: clip ${clipId} is already being scored`);
            return { clipId, status: "skipped", reason: "locked" };
        }
        const heartbeat = setInterval(async () => {
            try {
                const extended = await this.evalLua(this.extendLockSha, EXTEND_LOCK_SCRIPT, [lockKey], [lockValue, String(lockTTL)]);
                if (!extended) {
                    this.logger.warn(`Lock lost for clip ${clipId}, aborting heartbeat`);
                    clearInterval(heartbeat);
                }
            }
            catch (err) {
                this.logger.error(`Heartbeat extend failed: ${err}`);
            }
        }, 60000);
        const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), `clip-score-${clipId}-`));
        try {
            await job.updateProgress(5);
            const sourcePath = path.join(tmpDir, "source.mp4");
            await this.downloadVideo(videoUrl, sourcePath);
            await job.updateProgress(15);
            const clipBounds = await this.prisma.clip.findUnique({
                where: { id: clipId },
                select: { startSeconds: true, endSeconds: true },
            });
            const clipStart = clipBounds?.startSeconds ?? 0;
            const clipEnd = clipBounds?.endSeconds ?? 0;
            const clipDuration = Math.max(0, clipEnd - clipStart);
            const trimmedPath = path.join(tmpDir, "trimmed.mp4");
            await this.trimSource(sourcePath, trimmedPath, clipStart, clipDuration);
            await job.updateProgress(25);
            const framesDir = path.join(tmpDir, "frames");
            await fs.mkdir(framesDir, { recursive: true });
            await this.extractFrames(trimmedPath, framesDir);
            await job.updateProgress(40);
            const audioPath = path.join(tmpDir, "audio.wav");
            await this.extractAudio(trimmedPath, audioPath);
            await job.updateProgress(50);
            const pythonResult = await this.runPythonPipeline(framesDir, audioPath, transcriptText, tmpDir);
            await job.updateProgress(85);
            if (pythonResult.status === "error") {
                throw new Error(`Python scoring failed: ${pythonResult.message}`);
            }
            await this.prisma.clip.update({
                where: { id: clipId },
                data: {
                    visualScore: Math.round(pythonResult.breakdown.visualAesthetic),
                    audioScore: Math.round(pythonResult.breakdown.audioCompliance),
                    transcriptScore: Math.round(pythonResult.breakdown.narrativeHook),
                    scoringMetadata: pythonResult.metadata,
                    scoredAt: new Date(),
                },
            });
            await job.updateProgress(100);
            this.logger.log(`Clip ${clipId} scored: overall=${pythonResult.overallScore}, ` +
                `visual=${pythonResult.breakdown.visualAesthetic}, ` +
                `audio=${pythonResult.breakdown.audioCompliance}, ` +
                `hook=${pythonResult.breakdown.narrativeHook}`);
            return {
                clipId,
                status: "completed",
                overallScore: pythonResult.overallScore,
                breakdown: pythonResult.breakdown,
            };
        }
        catch (error) {
            const msg = error instanceof Error ? error.message : String(error);
            this.logger.error(`Clip scoring failed for ${clipId}: ${msg}`);
            throw error;
        }
        finally {
            clearInterval(heartbeat);
            await this.releaseLock(lockKey, lockValue);
            await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => undefined);
        }
    }
    async evalLua(sha, script, keys, args) {
        if (sha) {
            try {
                return await this.redis.evalSha(sha, { keys, arguments: args });
            }
            catch (err) {
                if (err?.message?.includes("NOSCRIPT")) {
                    return await this.redis.eval(script, { keys, arguments: args });
                }
                throw err;
            }
        }
        return await this.redis.eval(script, { keys, arguments: args });
    }
    async releaseLock(lockKey, lockValue) {
        try {
            await this.evalLua(this.releaseLockSha, RELEASE_LOCK_SCRIPT, [lockKey], [lockValue]);
        }
        catch (err) {
            this.logger.error(`Failed to release lock ${lockKey}: ${err}`);
        }
    }
    async downloadVideo(url, destPath) {
        let downloadUrl = url;
        if (!url.startsWith("http")) {
            downloadUrl = await this.r2Storage.getPresignedUrl(url, 3600);
        }
        const r2PublicHost = process.env.R2_PUBLIC_HOST || "r2.cloudflarestorage.com";
        const allowedHostSuffixes = [
            `.${r2PublicHost}`,
            ".s3.amazonaws.com",
            ".storage.googleapis.com",
            ".blob.core.windows.net",
            ".digitaloceanspaces.com",
        ];
        try {
            (0, safe_fetch_1.assertSafeUrl)(downloadUrl, { allowedHostSuffixes });
        }
        catch (err) {
            try {
                const u = new URL(downloadUrl);
                (0, safe_fetch_1.assertSafeUrl)(downloadUrl, { allowedHosts: [u.hostname] });
            }
            catch {
                throw err;
            }
        }
        const response = await fetch(downloadUrl);
        if (!response.ok) {
            throw new Error(`Failed to download video: ${response.status} ${response.statusText}`);
        }
        const MAX_VIDEO_BYTES = 10 * 1024 * 1024 * 1024;
        const contentLength = Number(response.headers.get("content-length"));
        if (Number.isFinite(contentLength) && contentLength > MAX_VIDEO_BYTES) {
            throw new Error(`Video too large (${contentLength} bytes exceeds the ${MAX_VIDEO_BYTES}-byte cap)`);
        }
        if (!response.body) {
            throw new Error("Downloaded video response has no body stream");
        }
        let totalBytes = 0;
        const reader = response.body.getReader();
        try {
            const fileHandle = await fs.open(destPath, "w");
            try {
                for (;;) {
                    const { done, value } = await reader.read();
                    if (done)
                        break;
                    totalBytes += value.byteLength;
                    if (totalBytes > MAX_VIDEO_BYTES) {
                        throw new Error(`Video exceeds the ${MAX_VIDEO_BYTES}-byte cap; aborting download`);
                    }
                    await fileHandle.write(value);
                }
            }
            finally {
                await fileHandle.close().catch(() => undefined);
            }
        }
        finally {
            reader.cancel().catch(() => undefined);
        }
    }
    async trimSource(sourcePath, outputPath, startSeconds, durationSeconds) {
        const args = [
            "-y",
            "-ss", String(Math.max(0, startSeconds)),
            "-i", sourcePath,
            "-t", String(Math.max(0, durationSeconds)),
            "-c", "copy",
            "-avoid_negative_ts", "make_zero",
            outputPath,
        ];
        try {
            await execAsync("ffmpeg", args, { timeout: 60000 });
        }
        catch (err) {
            const fallbackArgs = [
                "-y",
                "-ss", String(Math.max(0, startSeconds)),
                "-i", sourcePath,
                "-t", String(Math.max(0, durationSeconds)),
                "-c:v", "libx264",
                "-c:a", "aac",
                "-preset", "ultrafast",
                outputPath,
            ];
            await execAsync("ffmpeg", fallbackArgs, { timeout: 180000 });
        }
    }
    async extractFrames(sourcePath, framesDir) {
        const outPattern = path.join(framesDir, "frame_%04d.jpg");
        const args = [
            "-y",
            "-i", sourcePath,
            "-vf", "fps=1,scale=224:224:force_original_aspect_ratio=decrease,pad=224:224:(ow-iw)/2:(oh-ih)/2",
            "-q:v", "2",
            outPattern,
        ];
        try {
            await execAsync("ffmpeg", args, { timeout: 120000 });
        }
        catch (err) {
            const message = err?.message || String(err);
            if (message.includes("No such file or directory")) {
                throw new Error("FFmpeg not found. Ensure FFmpeg is installed in the worker container.");
            }
            throw new Error(`FFmpeg frame extraction failed: ${message}`);
        }
    }
    async extractAudio(sourcePath, audioPath) {
        const args = [
            "-y",
            "-i", sourcePath,
            "-vn",
            "-acodec", "pcm_s16le",
            "-ar", "44100",
            audioPath,
        ];
        try {
            await execAsync("ffmpeg", args, { timeout: 60000 });
        }
        catch (err) {
            throw new Error(`Audio extraction failed: ${err?.message || String(err)}`);
        }
    }
    async runPythonPipeline(framesDir, audioPath, transcript, tmpDir) {
        const possiblePaths = [
            path.resolve(process.cwd(), "services/clip-scorer/pipeline.py"),
            path.resolve("/app/scripts/pipeline.py"),
            path.resolve(__dirname, "../../scripts/pipeline.py"),
        ];
        let scriptPath = "";
        for (const p of possiblePaths) {
            try {
                await fs.access(p);
                scriptPath = p;
                break;
            }
            catch {
                continue;
            }
        }
        if (!scriptPath) {
            throw new Error("Python pipeline script not found. Searched: " + possiblePaths.join(", "));
        }
        const transcriptPath = path.join(tmpDir, "transcript.txt");
        await fs.writeFile(transcriptPath, transcript, "utf8");
        const args = [
            scriptPath,
            "--frames_dir", framesDir,
            "--audio_path", audioPath,
            "--transcript_path", transcriptPath,
        ];
        const { stdout, stderr } = await execAsync("python3", args, {
            timeout: 300000,
            maxBuffer: 1024 * 1024,
        });
        if (!stdout.trim()) {
            throw new Error(`Python script produced no output. stderr: ${stderr}`);
        }
        return JSON.parse(stdout.trim());
    }
};
exports.ClipScoringProcessor = ClipScoringProcessor;
exports.ClipScoringProcessor = ClipScoringProcessor = ClipScoringProcessor_1 = __decorate([
    (0, bullmq_1.Processor)("clip-scoring", {
        concurrency: 2,
    }),
    __param(2, (0, common_1.Inject)(redis_constants_1.REDIS_CLIENT)),
    __metadata("design:paramtypes", [prisma_service_1.PrismaService,
        r2_storage_service_1.R2StorageService, Object])
], ClipScoringProcessor);
//# sourceMappingURL=clip-scoring.processor.js.map