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));