`shrink_worker_safe.js`

79 lines
// SAFE-ORDER compression worker (runs ONLY once the store is under the cap).
// Per image: claim → download original → encode smaller → UPLOAD new FIRST →
// verify new is READY+smaller → DELETE original → verify count unchanged.
// If the upload 413s (still at cap) the original is left 100% intact and the row
// is marked 'blocked' (retryable) — NEVER an imageless window. Resumable via the
// same ledger + FOR UPDATE SKIP LOCKED claim.
//   node shrink_worker_safe.js [--shard k] [--allshards] [--limit N] [--wid id] [--writedelay 500]
import fs from 'fs';
import os from 'os';
import https from 'https';
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 ALL=process.argv.includes('--allshards');
const SHARD=+arg('shard','0'), LIMIT=+arg('limit','0')||1e9, WID=arg('wid','w'+process.pid), WRITEDELAY=+arg('writedelay','500');
const sleep=ms=>new Promise(r=>setTimeout(r,ms));
function raw(m,p,b){return new Promise((res,rej)=>{const data=b?JSON.stringify(b):null;const req=https.request({method:m,host:SHOP,path:p,headers:{'X-Shopify-Access-Token':ATOK,'Content-Type':'application/json',...(data?{'Content-Length':Buffer.byteLength(data)}:{})}},r=>{let d='';r.on('data',c=>d+=c);r.on('end',()=>res({status:r.statusCode,retryAfter:r.headers['retry-after'],limit:r.headers['x-shopify-shop-api-call-limit']||'',json:(()=>{try{return JSON.parse(d)}catch(e){return d}})()}));});req.on('error',rej);req.setTimeout(60000,()=>req.destroy(new Error('timeout')));if(data)req.write(data);req.end();});}
async function api(m,p,b){for(let a=0;a<10;a++){let r;try{r=await raw(m,p,b);}catch(e){await sleep(1500);continue;}if(r.status===429){await sleep((parseFloat(r.retryAfter)||3)*1000);continue;}const [u,mx]=(r.limit||'0/40').split('/').map(Number);await sleep(u>(mx*0.5)?800:200);return r;}return{status:429,json:{}};}
function download(url){return new Promise(res=>{try{const u=new URL(url);https.get({host:u.host,path:u.pathname+u.search,headers:{'User-Agent':'dw-shrink'}},r=>{if(r.statusCode!==200){r.resume();return res(null);}const ch=[];r.on('data',c=>ch.push(c));r.on('end',()=>res(Buffer.concat(ch)));r.on('error',()=>res(null));}).on('error',()=>res(null));}catch(e){res(null);}});}
function headLen(url){return new Promise(res=>{try{const u=new URL(url);const rq=https.request({method:'HEAD',host:u.host,path:u.pathname+u.search},r=>{r.resume();res(+(r.headers['content-length']||0));});rq.on('error',()=>res(0));rq.setTimeout(15000,()=>{rq.destroy();res(0);});rq.end();}catch(e){res(0);}});}

const db=new pg.Client({host:'/tmp',database:'dw_unified'}); await db.connect();
async function claim(){const where=ALL?"state='pending'":"shard=$1 AND state='pending'";const params=ALL?[]:[SHARD];const q=`UPDATE img_shrink_ledger SET state='claimed', worker_id=$${params.length+1}, claimed_at=now(), attempts=attempts+1, updated_at=now() WHERE id=(SELECT id FROM img_shrink_ledger WHERE ${where} ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *`;const r=await db.query(q,[...params,WID]);return r.rows[0]||null;}
async function mark(id,f){const k=Object.keys(f);await db.query(`UPDATE img_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,skipped=0,failed=0,blocked=0,processed=0,reclaimed=0;
const tmp=os.tmpdir();
while(processed<LIMIT){
  const row=await claim(); if(!row)break; processed++;
  const pid=row.product_id, oid=Number(row.orig_image_id);
  const fin=`${tmp}/ss_${row.id}_in`, fout=`${tmp}/ss_${row.id}_out.jpg`;
  try{
    const pr=await api('GET',`/admin/api/2024-10/products/${pid}.json?fields=id,handle,images`);
    const imgs=(pr.json.product&&pr.json.product.images)||[];
    const t=imgs.find(im=>im.id===oid);
    if(!t){ await mark(row.id,{state:'skipped',skip_reason:'image_gone'}); skipped++; continue; }
    const origCount=imgs.length, position=t.position, alt=t.alt||null, handle=pr.json.product.handle;
    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','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}`;

    // 1) UPLOAD SMALLER FIRST (original still intact)
    await sleep(WRITEDELAY);
    const up=await api('POST',`/admin/api/2024-10/products/${pid}/images.json`,{image:{attachment:encBuf.toString('base64'),filename,position,...(alt?{alt}:{})}});
    if(up.status>=400||!up.json.image){
      const capped=/exceed the file storage/i.test(JSON.stringify(up.json));
      await mark(row.id,{state: capped?'blocked':'failed', fail_reason:(capped?'still_at_cap':'upload_'+up.status), orig_bytes:origBytes});
      if(capped) blocked++; else failed++;
      cleanup(fin,fout); continue;   // original untouched — safe
    }
    const ni=up.json.image;
    const nb=await headLen(ni.src)||encBuf.length;
    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;
    }
    // 2) DELETE ORIGINAL (only after the smaller one is confirmed live)
    await sleep(WRITEDELAY);
    const del=await api('DELETE',`/admin/api/2024-10/products/${pid}/images/${oid}.json`);
    if(del.status>=400){ await mark(row.id,{state:'failed',fail_reason:'delete_'+del.status+'_new_kept',new_image_id:ni.id,new_src:ni.src,new_bytes:nb,orig_bytes:origBytes}); failed++; cleanup(fin,fout); continue; }
    await mark(row.id,{state:'done',new_image_id:ni.id,new_src:ni.src,new_bytes:nb,new_width:ni.width,new_height:ni.height,orig_bytes:origBytes,fail_reason:null});
    done++; reclaimed+=(origBytes-nb);
    cleanup(fin,fout);
  }catch(e){ await mark(row.id,{state:'failed',fail_reason:('exc:'+String(e.message||e)).slice(0,120)}); failed++; cleanup(fin,fout); }
  if(processed%25===0) console.log(`[${WID}] ${processed} | done ${done} skip ${skipped} fail ${failed} blocked ${blocked} | reclaimed ${(reclaimed/1048576).toFixed(1)}MB`);
}
function cleanup(...f){for(const x of f)try{fs.unlinkSync(x)}catch(e){}}
console.log(`[${WID}] FINISHED processed=${processed} done=${done} skipped=${skipped} failed=${failed} blocked=${blocked} reclaimed=${(reclaimed/1048576).toFixed(1)}MB`);
if(blocked>0) console.log(`  NOTE: ${blocked} rows blocked = store still at cap. Free more space, reset blocked→pending, rerun.`);
await db.end();