← back to Dw Image Shrink
video_worker_safe.js
162 lines
// SAFE-ORDER product-video compression worker (mirrors shrink_worker_safe.js).
// Per video: claim → download original → ffmpeg re-encode smaller → UPLOAD new FIRST
// (stagedUploadsCreate → PUT → productCreateMedia) → poll new until READY + smaller →
// DELETE original. If upload hits the storage cap the original is left 100% intact and
// the row parks 'blocked' (retryable) — NEVER a video-less window. Resumable via ledger.
// node video_worker_safe.js [--limit N] [--wid id] [--dry-run] [--target-h 1080] [--crf 23]
// --dry-run = download + encode + report win ONLY. Zero Shopify writes. Safe at the cap.
import fs from 'fs';
import os from 'os';
import pg from 'pg';
import { execFileSync } from 'child_process';
const DIR='/Users/macstudio3/Projects/dw-image-shrink';
const SHOP='designer-laboratory-sandbox.myshopify.com';
const ATOK=fs.readFileSync('/Users/macstudio3/Projects/secrets-manager/.env','utf8').split('\n').find(l=>l.startsWith('SHOPIFY_ADMIN_TOKEN=')).split('=').slice(1).join('=').replace(/["' ]/g,'');
const arg=(k,d)=>{const a=process.argv.find(x=>x.startsWith('--'+k+'='));return a?a.split('=').slice(1).join('='):d;};
const LIMIT=+arg('limit','0')||1e9, WID=arg('wid','v'+process.pid);
const DRY=!process.argv.includes('--apply');
const TARGET_H=+arg('target-h','1080'), CRF=+arg('crf','23');
const sleep=ms=>new Promise(r=>setTimeout(r,ms));
const MB=b=>(b/1048576).toFixed(1)+'MB';
async function gql(query,variables){
for(let a=0;a<12;a++){
let r; try{ r=await fetch(`https://${SHOP}/admin/api/2024-10/graphql.json`,{method:'POST',
headers:{'X-Shopify-Access-Token':ATOK,'Content-Type':'application/json'},
body:JSON.stringify({query,variables})}); }catch(e){ await sleep(1500); continue; }
if(r.status===429||r.status>=500){await sleep(2000);continue;}
const j=await r.json();
if(j.errors&&JSON.stringify(j.errors).includes('THROTTLED')){await sleep(2500);continue;}
const cost=j.extensions?.cost?.throttleStatus;
if(cost&&cost.currentlyAvailable<400)await sleep(1200); else await sleep(250);
return j;
}
return {errors:[{message:'exhausted'}]};
}
function download(url,dest){return new Promise((res,rej)=>{
import('https').then(({default:https})=>{
const u=new URL(url);
https.get({host:u.host,path:u.pathname+u.search,headers:{'User-Agent':'dw-vshrink'}},r=>{
if(r.statusCode>=300&&r.statusCode<400&&r.headers.location){r.resume();return download(r.headers.location,dest).then(res,rej);}
if(r.statusCode!==200){r.resume();return rej(new Error('dl_'+r.statusCode));}
const ws=fs.createWriteStream(dest); r.pipe(ws);
ws.on('finish',()=>ws.close(()=>res(fs.statSync(dest).size))); ws.on('error',rej);
}).on('error',rej);
});
});}
const capped=s=>/exceed|storage|quota|limit/i.test(s||'');
// ---- Shopify video upload (upload-first) ----
async function stagedTarget(filename,fileSize){
const q=`mutation($input:[StagedUploadInput!]!){stagedUploadsCreate(input:$input){
stagedTargets{url resourceUrl parameters{name value}} userErrors{field message}}}`;
const j=await gql(q,{input:[{resource:'VIDEO',filename,mimeType:'video/mp4',fileSize:String(fileSize),httpMethod:'POST'}]});
const ue=j.data?.stagedUploadsCreate?.userErrors||j.errors||[];
if(ue.length) return {err:JSON.stringify(ue)};
return {t:j.data.stagedUploadsCreate.stagedTargets[0]};
}
async function putStaged(t,buf,filename){
const fd=new FormData();
for(const p of t.parameters) fd.append(p.name,p.value);
fd.append('file',new Blob([buf],{type:'video/mp4'}),filename);
const r=await fetch(t.url,{method:'POST',body:fd});
if(r.status>=400) return {err:'put_'+r.status+':'+(await r.text()).slice(0,200)};
return {ok:true};
}
async function createMedia(pid,resourceUrl,alt){
const q=`mutation($pid:ID!,$media:[CreateMediaInput!]!){productCreateMedia(productId:$pid,media:$media){
media{... on Video{id status}} mediaUserErrors{field message code}}}`;
const j=await gql(q,{pid:`gid://shopify/Product/${pid}`,media:[{originalSource:resourceUrl,mediaContentType:'VIDEO',...(alt?{alt}:{})}]});
const ue=j.data?.productCreateMedia?.mediaUserErrors||j.errors||[];
if(ue.length) return {err:JSON.stringify(ue)};
return {media:j.data.productCreateMedia.media[0]};
}
async function pollReady(pid,mediaId,maxMs=1200000){
const q=`query($pid:ID!){product(id:$pid){media(first:50){nodes{... on Video{id status originalSource{fileSize}}}}}}`;
const t0=Date.now();
while(Date.now()-t0<maxMs){
await sleep(10000);
const j=await gql(q,{pid:`gid://shopify/Product/${pid}`});
const n=(j.data?.product?.media?.nodes||[]).find(x=>x.id===mediaId);
if(!n)continue;
if(n.status==='READY')return {ready:true,bytes:n.originalSource?.fileSize||0};
if(n.status==='FAILED')return {ready:false,failed:true};
}
return {ready:false,timeout:true};
}
async function deleteMedia(pid,mediaId){
const q=`mutation($pid:ID!,$ids:[ID!]!){productDeleteMedia(productId:$pid,mediaIds:$ids){deletedMediaIds mediaUserErrors{field message}}}`;
const j=await gql(q,{pid:`gid://shopify/Product/${pid}`,ids:[mediaId]});
const ue=j.data?.productDeleteMedia?.mediaUserErrors||j.errors||[];
return ue.length?{err:JSON.stringify(ue)}:{ok:true};
}
// ---- encode ----
function encode(fin,fout){
let w=0,h=0;
try{ const p=JSON.parse(execFileSync('ffprobe',['-v','quiet','-print_format','json','-show_streams','-select_streams','v:0',fin],{encoding:'utf8'}));
w=p.streams?.[0]?.width||0; h=p.streams?.[0]?.height||0; }catch(e){}
const args=['-y','-i',fin,'-map_metadata','-1'];
// cap the SHORT edge at TARGET_H (preserve aspect, never upscale):
// landscape → set height; portrait → set width; leave already-small clips alone.
const shortEdge=Math.min(w||1e9,h||1e9);
if(shortEdge>TARGET_H){
if(w>=h) args.push('-vf',`scale=-2:${TARGET_H}`); // landscape: short edge = height
else args.push('-vf',`scale=${TARGET_H}:-2`); // portrait: short edge = width
}
args.push('-c:v','libx264','-preset','medium','-crf',String(CRF),'-pix_fmt','yuv420p',
'-c:a','aac','-b:a','128k','-movflags','+faststart',fout);
execFileSync('ffmpeg',args,{stdio:['ignore','ignore','ignore']});
return {w,h,bytes:fs.statSync(fout).size};
}
const db=new pg.Client({host:'/tmp',database:'dw_unified'}); await db.connect();
async function claim(){const r=await db.query(`UPDATE vid_shrink_ledger SET state='claimed',worker_id=$1,claimed_at=now(),attempts=attempts+1,updated_at=now() WHERE id=(SELECT id FROM vid_shrink_ledger WHERE state='pending' ORDER BY orig_bytes DESC NULLS LAST FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *`,[WID]);return r.rows[0]||null;}
async function mark(id,f){const k=Object.keys(f);await db.query(`UPDATE vid_shrink_ledger SET ${k.map((x,i)=>`${x}=$${i+2}`).join(',')},updated_at=now() WHERE id=$1`,[id,...k.map(x=>f[x])]);}
let done=0,skip=0,fail=0,blocked=0,proc=0,reclaimed=0;
const tmp=os.tmpdir();
while(proc<LIMIT){
const row=await claim(); if(!row)break; proc++;
const fin=`${tmp}/vv_${row.id}.src`, fout=`${tmp}/vv_${row.id}.mp4`;
try{
let ob=row.orig_bytes||0;
try{ ob=await download(row.orig_url,fin); }
catch(e){ await mark(row.id,{state:'failed',fail_reason:'dl:'+String(e.message).slice(0,80)}); fail++; cleanup(fin,fout); continue; }
let enc; try{ enc=encode(fin,fout); }catch(e){ await mark(row.id,{state:'failed',fail_reason:'ffmpeg:'+String(e.message).slice(0,80),orig_bytes:ob}); fail++; cleanup(fin,fout); continue; }
const win = enc.bytes>0 && enc.bytes < ob*0.90;
if(!win){ await mark(row.id,{state:'skipped',skip_reason:`no_win(${MB(enc.bytes)}vs${MB(ob)})`,orig_bytes:ob,new_bytes:enc.bytes}); skip++; cleanup(fin,fout); continue; }
if(DRY){ await mark(row.id,{state:'skipped',skip_reason:`DRYRUN_win ${MB(ob)}→${MB(enc.bytes)}`,orig_bytes:ob,new_bytes:enc.bytes}); done++; reclaimed+=(ob-enc.bytes); console.log(`[DRY] ${row.handle} ${MB(ob)}→${MB(enc.bytes)} (${(100*(1-enc.bytes/ob)).toFixed(0)}% off)`); cleanup(fin,fout); continue; }
// 1) UPLOAD SMALLER FIRST
const fn=`${(row.handle||'video').slice(0,50)}-${row.id}.mp4`;
const buf=fs.readFileSync(fout);
const st=await stagedTarget(fn,buf.length);
if(st.err){ const c=capped(st.err); await mark(row.id,{state:c?'blocked':'failed',fail_reason:(c?'still_at_cap':'staged:'+st.err.slice(0,80)),orig_bytes:ob,new_bytes:enc.bytes}); c?blocked++:fail++; cleanup(fin,fout); continue; }
const put=await putStaged(st.t,buf,fn);
if(put.err){ const c=capped(put.err); await mark(row.id,{state:c?'blocked':'failed',fail_reason:(c?'still_at_cap':put.err.slice(0,80)),orig_bytes:ob}); c?blocked++:fail++; cleanup(fin,fout); continue; }
const cm=await createMedia(row.product_id,st.t.resourceUrl,null);
if(cm.err){ const c=capped(cm.err); await mark(row.id,{state:c?'blocked':'failed',fail_reason:(c?'still_at_cap':'create:'+cm.err.slice(0,80)),orig_bytes:ob}); c?blocked++:fail++; cleanup(fin,fout); continue; }
const newId=cm.media.id;
// 2) wait for the new video to finish processing
const pr=await pollReady(row.product_id,newId);
if(!pr.ready){ // original still intact; drop the bad/half new one, keep original
await deleteMedia(row.product_id,newId);
await mark(row.id,{state:'failed',fail_reason:pr.failed?'new_FAILED_orig_kept':'new_timeout_orig_kept',new_media_id:newId,orig_bytes:ob,new_bytes:enc.bytes}); fail++; cleanup(fin,fout); continue; }
const nb=pr.bytes||enc.bytes;
if(!(nb>0 && nb<ob)){ await deleteMedia(row.product_id,newId); await mark(row.id,{state:'skipped',skip_reason:`no_win_ready(${MB(nb)})`,new_media_id:newId,orig_bytes:ob,new_bytes:nb}); skip++; cleanup(fin,fout); continue; }
// 3) DELETE ORIGINAL (only after new is READY + smaller)
const del=await deleteMedia(row.product_id,row.orig_media_id);
if(del.err){ await mark(row.id,{state:'failed',fail_reason:'delete_orig_new_kept:'+del.err.slice(0,60),new_media_id:newId,new_bytes:nb,orig_bytes:ob}); fail++; cleanup(fin,fout); continue; }
await mark(row.id,{state:'done',new_media_id:newId,new_bytes:nb,orig_bytes:ob,fail_reason:null});
done++; reclaimed+=(ob-nb);
console.log(`[${WID}] ✓ ${row.handle} ${MB(ob)}→${MB(nb)}`);
}catch(e){ await mark(row.id,{state:'failed',fail_reason:('exc:'+String(e.message||e)).slice(0,120)}); fail++; cleanup(fin,fout); }
}
function cleanup(...f){for(const x of f)try{fs.unlinkSync(x)}catch(e){}}
console.log(`[${WID}] FINISHED proc=${proc} done=${done} skip=${skip} fail=${fail} blocked=${blocked} reclaimed=${MB(reclaimed)}`);
if(blocked>0)console.log(` NOTE: ${blocked} blocked = store still at cap. Free space, reset blocked→pending, rerun.`);
await db.end();