57 lines
4.6 KiB
TypeScript
57 lines
4.6 KiB
TypeScript
import { mkdtemp,rm,writeFile } from "node:fs/promises";
|
|
import { createWriteStream } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import { pipeline } from "node:stream/promises";
|
|
import type { Readable } from "node:stream";
|
|
import { setTimeout } from "node:timers/promises";
|
|
import { db,transaction } from "./lib/db";
|
|
import { getObject,putFile } from "./lib/storage";
|
|
import { probe,run } from "./lib/probe";
|
|
import {randomUUID} from "node:crypto";
|
|
import {linkDerivedAssets} from "./lib/asset-links";
|
|
let stopping=false;
|
|
process.on('SIGTERM',()=>{stopping=true;});process.on('SIGINT',()=>{stopping=true;});
|
|
async function processOne(){
|
|
const job=await transaction(async client=>(await client.query(`UPDATE media_jobs SET status='processing',attempts=attempts+1,locked_at=now() WHERE id=(SELECT id FROM media_jobs WHERE (status='pending' AND available_at<=now()) OR (status='processing' AND locked_at<now()-interval '15 minutes') ORDER BY available_at FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *`)).rows[0]);
|
|
if(!job)return false;
|
|
const dir=await mkdtemp(join(tmpdir(),'prompt-video-'));
|
|
try{
|
|
await db().query("UPDATE assets SET processing_status='processing',processing_error=NULL WHERE id=$1",[job.asset_id]);
|
|
const asset=(await db().query('SELECT * FROM assets WHERE id=$1',[job.asset_id])).rows[0];
|
|
const input=join(dir,'input');const object=await getObject(asset.storage_key);
|
|
await pipeline(object.Body as Readable,createWriteStream(input));
|
|
const metadata=await probe(input);const hasAudio=metadata.streams.some(s=>s.codec_type==='audio');
|
|
const variants=[{role:'silent_video',mime:'video/mp4',file:'silent.mp4',args:['-map','0:v:0','-an','-c:v','libx264','-preset','veryfast','-crf','22','-pix_fmt','yuv420p','-vf','scale=trunc(iw/2)*2:trunc(ih/2)*2','-movflags','+faststart']},
|
|
{role:'poster',mime:'image/jpeg',file:'poster.jpg',args:['-map','0:v:0','-frames:v','1','-vf','scale=min(1200\\,iw):-2']}];
|
|
if(hasAudio)variants.push({role:'audio_track',mime:'audio/mp4',file:'audio.m4a',args:['-map','0:a:0','-vn','-c:a','aac','-b:a','192k']});
|
|
const outputs:{role:string;mime:string;file:string;args:string[];key:string;size:number}[]=[];
|
|
for(const v of variants){const path=join(dir,v.file);await run(process.env.FFMPEG_PATH||'ffmpeg',['-nostdin','-y','-v','error','-protocol_whitelist','file,pipe','-i',input,...v.args,path],{timeout:10*60*1000,maxBuffer:2*1024*1024});const key=`derived/${asset.id}/${v.file}`;const size=await putFile(key,path,v.mime);outputs.push({...v,key,size});}
|
|
await transaction(async client=>{
|
|
for(const v of outputs)await client.query(`INSERT INTO assets(id,kind,filename,storage_key,mime_type,byte_size,source_asset_id,derivation)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT(source_asset_id,derivation) WHERE source_asset_id IS NOT NULL
|
|
DO UPDATE SET storage_key=excluded.storage_key,mime_type=excluded.mime_type,byte_size=excluded.byte_size`,
|
|
[randomUUID(),v.role==='poster'?'image':v.role==='audio_track'?'audio':'video',v.file,v.key,v.mime,v.size,asset.id,v.role]);
|
|
const posts=(await client.query("SELECT DISTINCT post_id FROM post_assets WHERE asset_id=$1 AND role='content' ORDER BY post_id",[asset.id])).rows;
|
|
for(const post of posts)await linkDerivedAssets(client,post.post_id);
|
|
await client.query("UPDATE assets SET has_audio=$2,processing_status='ready',processing_error=NULL WHERE id=$1",[asset.id,hasAudio]);
|
|
await client.query("UPDATE media_jobs SET status='ready',last_error=NULL WHERE id=$1",[job.id]);
|
|
});
|
|
}catch(error){
|
|
const message=(error instanceof Error?error.message:'Processing failed').slice(0,2000);const final=job.attempts>=3;
|
|
await transaction(async client=>{
|
|
await client.query("UPDATE media_jobs SET status=$2,last_error=$3,available_at=now()+interval '30 seconds' WHERE id=$1",[job.id,final?'failed':'pending',message]);
|
|
await client.query('UPDATE assets SET processing_status=$2,processing_error=$3 WHERE id=$1',[job.asset_id,final?'failed':'pending',message]);
|
|
});
|
|
console.error(`Video processing failed for ${job.asset_id}; retry=${!final}`);
|
|
}finally{await rm(dir,{recursive:true,force:true});}
|
|
return true;
|
|
}
|
|
async function main(){
|
|
const heartbeat=()=>writeFile('/tmp/prompt-worker-health',String(Date.now())).catch(()=>console.error('Worker heartbeat failed'));
|
|
await heartbeat();const timer=setInterval(()=>{void heartbeat();},15000);
|
|
try{while(!stopping){try{if(!await processOne())await setTimeout(2000);}catch{console.error('Worker connection error; retrying');await setTimeout(5000);}}}
|
|
finally{clearInterval(timer);await db().end();}
|
|
}
|
|
main().catch(()=>process.exit(1));
|