← back to Dw Image Shrink
TK-10048: add video shrink pipeline (enum/enqueue/worker) + retune image worker to 2400px/q85
95ce90552b1d88ed83c2143ee0e9769bcfec8a58 · 2026-07-30 07:24:48 -0700 · Steve Abrams
Files touched
A enqueue-videos.jsA enum-videos.jsM shrink_worker_safe.jsA video-ledger.sqlA video_worker_safe.js
Diff
commit 95ce90552b1d88ed83c2143ee0e9769bcfec8a58
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Thu Jul 30 07:24:48 2026 -0700
TK-10048: add video shrink pipeline (enum/enqueue/worker) + retune image worker to 2400px/q85
---
enqueue-videos.js | 24 ++++++++
enum-videos.js | 68 +++++++++++++++++++++
shrink_worker_safe.js | 4 +-
video-ledger.sql | 27 +++++++++
video_worker_safe.js | 161 ++++++++++++++++++++++++++++++++++++++++++++++++++
5 files changed, 282 insertions(+), 2 deletions(-)
diff --git a/enqueue-videos.js b/enqueue-videos.js
new file mode 100644
index 0000000..f809521
--- /dev/null
+++ b/enqueue-videos.js
@@ -0,0 +1,24 @@
+// Populate vid_shrink_ledger from videos-inventory.jsonl (built by enum-videos.js).
+// Scope = active+draft product videos (archived → delete track). Idempotent via
+// ON CONFLICT(orig_media_id) DO NOTHING. Resumable.
+// node enqueue-videos.js [--status active,draft]
+import fs from 'fs';
+import pg from 'pg';
+const DIR='/Users/macstudio3/Projects/dw-image-shrink';
+const arg=(k,d)=>{const a=process.argv.find(x=>x.startsWith('--'+k+'='));return a?a.split('=').slice(1).join('='):d;};
+const STATUS=arg('status','active,draft').split(',').map(s=>s.toUpperCase());
+const rows=fs.readFileSync(DIR+'/videos-inventory.jsonl','utf8').split('\n').filter(Boolean).map(l=>JSON.parse(l))
+ .filter(r=>STATUS.includes(String(r.prod_status||'').toUpperCase()));
+console.log(`enqueue-videos: status[${STATUS}] → ${rows.length} product-videos`);
+const c=new pg.Client({host:'/tmp',database:'dw_unified'}); await c.connect();
+let ins=0;
+for(const r of rows){
+ const res=await c.query(`INSERT INTO vid_shrink_ledger
+ (product_id,handle,prod_status,shard,orig_media_id,orig_url,orig_bytes,orig_format,orig_width,orig_height)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT (orig_media_id) DO NOTHING`,
+ [r.product_id,r.handle,r.prod_status,r.product_id%8,r.media_id,r.orig_url,r.orig_bytes,r.orig_format,r.orig_width,r.orig_height]);
+ ins+=res.rowCount;
+}
+const {rows:[cnt]}=await c.query("SELECT count(*) n, count(*) FILTER(WHERE state='pending') p, pg_size_pretty(sum(orig_bytes)::numeric) gb FROM vid_shrink_ledger");
+console.log(`inserted ${ins} | ledger total ${cnt.n} (${cnt.p} pending, ${cnt.gb} original)`);
+await c.end();
diff --git a/enum-videos.js b/enum-videos.js
new file mode 100644
index 0000000..da9f7e8
--- /dev/null
+++ b/enum-videos.js
@@ -0,0 +1,68 @@
+// READ-ONLY. Enumerate every product-attached Video → videos-inventory.jsonl.
+// Paginates products (250/page) and pulls their Video media (first:10). Writes one
+// JSONL line per Video with product linkage + original source size, so the worker
+// knows what to re-encode and where to re-attach it. Safe to run any time (no writes).
+// node enum-videos.js [--limit-pages N]
+import fs from 'fs';
+const OUT='/Users/macstudio3/Projects/dw-image-shrink/videos-inventory.jsonl';
+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 MAXPAGES=+arg('limit-pages','0')||1e9;
+const sleep=ms=>new Promise(r=>setTimeout(r,ms));
+
+async function gql(query,variables){
+ for(let a=0;a<12;a++){
+ const 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})});
+ 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;}
+ if(j.errors)throw new Error(JSON.stringify(j.errors).slice(0,300));
+ // gentle pacing on the leaky bucket
+ const cost=j.extensions&&j.extensions.cost&&j.extensions.cost.throttleStatus;
+ if(cost&&cost.currentlyAvailable<400)await sleep(1200); else await sleep(300);
+ return j.data;
+ }
+ throw new Error('gql: exhausted retries');
+}
+
+const Q=`query($cursor:String){
+ products(first:250, after:$cursor){
+ pageInfo{hasNextPage endCursor}
+ nodes{
+ id handle status
+ media(first:10){nodes{
+ ... on Video { id filename status
+ originalSource{ url fileSize format mimeType width height } }
+ }}
+ }
+ }
+}`;
+
+let cursor=null, pages=0, prods=0, vids=0, bytes=0;
+const ws=fs.createWriteStream(OUT);
+while(pages<MAXPAGES){
+ const d=await gql(Q,{cursor});
+ const p=d.products;
+ for(const n of p.nodes){
+ prods++;
+ const pid=Number(n.id.split('/').pop());
+ for(const m of (n.media?.nodes||[])){
+ if(!m||!m.id||!m.id.includes('/Video/'))continue;
+ const os=m.originalSource||{};
+ if(!os.url)continue; // no downloadable original yet (still processing) → skip
+ vids++; bytes+=(os.fileSize||0);
+ ws.write(JSON.stringify({product_id:pid,handle:n.handle,prod_status:n.status,
+ media_id:m.id,media_status:m.status,orig_url:os.url,orig_bytes:os.fileSize||0,
+ orig_format:os.format||'',orig_width:os.width||0,orig_height:os.height||0})+'\n');
+ }
+ }
+ pages++;
+ if(pages%10===0)console.log(` ...${pages} pages | ${prods} products | ${vids} videos | ${(bytes/1073741824).toFixed(2)} GB`);
+ if(!p.pageInfo.hasNextPage)break;
+ cursor=p.pageInfo.endCursor;
+}
+ws.end();
+console.log(`DONE: ${pages} pages, ${prods} products scanned, ${vids} product-videos → ${OUT} | total original bytes ${(bytes/1073741824).toFixed(2)} GB`);
diff --git a/shrink_worker_safe.js b/shrink_worker_safe.js
index 73840ad..e152993 100644
--- a/shrink_worker_safe.js
+++ b/shrink_worker_safe.js
@@ -41,7 +41,7 @@ while(processed<LIMIT){
const origBuf=await download(t.src);
if(!origBuf||!origBuf.length){ await mark(row.id,{state:'failed',fail_reason:'download'}); failed++; continue; }
const origBytes=origBuf.length; fs.writeFileSync(fin,origBuf);
- let v; try{ v=JSON.parse(execFileSync('python3',[`${DIR}/shrink_master.py`,'--in',fin,'--out',fout,'--max-edge','2048','--quality','82'],{encoding:'utf8'})); }catch(e){ v={action:'skip',reason:'py_err'}; }
+ let v; try{ v=JSON.parse(execFileSync('python3',[`${DIR}/shrink_master.py`,'--in',fin,'--out',fout,'--max-edge','2400','--quality','85'],{encoding:'utf8'})); }catch(e){ v={action:'skip',reason:'py_err'}; }
if(v.action!=='encoded'){ await mark(row.id,{state:'skipped',skip_reason:v.reason||'no_encode',orig_bytes:origBytes}); skipped++; cleanup(fin,fout); continue; }
const encBuf=fs.readFileSync(fout); const ext=v.reason==='PNG'?'png':'jpg';
const filename=`${(handle||'img').slice(0,60)}-${oid}.${ext}`;
@@ -57,7 +57,7 @@ while(processed<LIMIT){
}
const ni=up.json.image;
const nb=await headLen(ni.src)||encBuf.length;
- if(!(nb>0 && nb<origBytes && (ni.width||0)<=2048)){
+ if(!(nb>0 && nb<origBytes && (ni.width||0)<=2400)){
// new isn't actually better → remove the new one, keep original
await sleep(WRITEDELAY); await api('DELETE',`/admin/api/2024-10/products/${pid}/images/${ni.id}.json`);
await mark(row.id,{state:'skipped',skip_reason:`no_win(nb=${nb})`,orig_bytes:origBytes,new_bytes:nb}); skipped++; cleanup(fin,fout); continue;
diff --git a/video-ledger.sql b/video-ledger.sql
new file mode 100644
index 0000000..4a17766
--- /dev/null
+++ b/video-ledger.sql
@@ -0,0 +1,27 @@
+-- Ledger for the product-video shrink job (mirrors img_shrink_ledger).
+-- One row per attached product Video. Upload-first-safe: never leaves a product
+-- video-less; at the cap the row parks 'blocked' with the original intact.
+CREATE TABLE IF NOT EXISTS vid_shrink_ledger (
+ id bigserial PRIMARY KEY,
+ product_id bigint NOT NULL,
+ handle text,
+ prod_status text,
+ shard int NOT NULL DEFAULT 0,
+ orig_media_id text NOT NULL UNIQUE, -- gid://shopify/Video/...
+ orig_url text NOT NULL, -- originalSource.url (downloadable)
+ orig_bytes bigint,
+ orig_format text,
+ orig_width int,
+ orig_height int,
+ state text NOT NULL DEFAULT 'pending', -- pending|claimed|done|skipped|blocked|failed
+ worker_id text,
+ attempts int NOT NULL DEFAULT 0,
+ new_media_id text,
+ new_url text,
+ new_bytes bigint,
+ skip_reason text,
+ fail_reason text,
+ claimed_at timestamptz,
+ updated_at timestamptz NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS vid_shrink_state_idx ON vid_shrink_ledger(state);
diff --git a/video_worker_safe.js b/video_worker_safe.js
new file mode 100644
index 0000000..c65f321
--- /dev/null
+++ b/video_worker_safe.js
@@ -0,0 +1,161 @@
+// 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('--dry-run');
+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();
← e73a454 chore: untrack audit-log artifacts, version bump (session cl
·
back to Dw Image Shrink
·
auto-save: 2026-07-30T07:45:45 (1 files) — videos-inventory. bae9d65 →