`shrink_worker.js`

117 lines
// Core worker: atomic-claim a ledger row → GET product snapshot → download the
// master → re-encode (shrink_master.py) → DELETE-first the old image → base64
// upload the smaller one → verify → record. Keeps the ORIGINAL bytes in hand and
// AUTO-RESTORES them if the new upload fails, so a product is never left
// permanently imageless. Resumable + crash-safe (FOR UPDATE SKIP LOCKED claim).
//   node shrink_worker.js [--shard k] [--allshards] [--limit N] [--wid id] [--writedelay 500] [--dry]
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');
const LIMIT=+arg('limit','0')||1e9;
const WID=arg('wid','w'+process.pid);
const WRITEDELAY=+arg('writedelay','500');   // ms between Shopify WRITE calls (2/s at 500)
const DRY=process.argv.includes('--dry');
const sleep=ms=>new Promise(r=>setTimeout(r,ms));

// ---- Shopify REST with leaky-bucket backoff ----
function raw(method,path,body){return new Promise((res,rej)=>{const data=body?JSON.stringify(body):null;const req=https.request({method,host:SHOP,path,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(method,path,body){for(let a=0;a<10;a++){let r;try{r=await raw(method,path,body);}catch(e){await sleep(1500);continue;}if(r.status===429){await sleep((parseFloat(r.retryAfter)||3)*1000);continue;}const [u,m]=(r.limit||'0/40').split('/').map(Number);await sleep(u>(m*0.5)?800:200);return r;}return{status:429,json:{error:'ratelimited'}};}
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 head(url){return new Promise(res=>{try{const u=new URL(url);const req=https.request({method:'HEAD',host:u.host,path:u.pathname+u.search},r=>{r.resume();res(+(r.headers['content-length']||0));});req.on('error',()=>res(0));req.setTimeout(15000,()=>{req.destroy();res(0);});req.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,fields){const keys=Object.keys(fields);const set=keys.map((k,i)=>`${k}=$${i+2}`).join(',');await db.query(`UPDATE img_shrink_ledger SET ${set}, updated_at=now() WHERE id=$1`,[id,...keys.map(k=>fields[k])]);}

let done=0,skipped=0,failed=0,alerts=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);
  try{
    // 1) snapshot current product images
    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 target=imgs.find(im=>im.id===oid);
    if(!target){ await mark(row.id,{state:'skipped',skip_reason:'image_gone'}); skipped++; continue; }
    const origCount=imgs.length, position=target.position, alt=target.alt||null, handle=pr.json.product.handle;
    await mark(row.id,{product_snapshot:JSON.stringify(imgs),orig_position:position,orig_alt:alt});

    // 2) download original master
    const origBuf=await download(target.src);
    if(!origBuf||!origBuf.length){ await mark(row.id,{state:'failed',fail_reason:'download'}); failed++; continue; }
    const origBytes=origBuf.length;
    const fin=`${tmp}/shr_${row.id}_in`, fout=`${tmp}/shr_${row.id}_out.jpg`;
    fs.writeFileSync(fin,origBuf);

    // 3) encode
    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'}; }
    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 newBytesLocal=encBuf.length;
    const ext = v.reason==='PNG'?'png':'jpg';
    const filename=`${(handle||'img').slice(0,60)}-${oid}.${ext}`;

    if(DRY){ await mark(row.id,{state:'skipped',skip_reason:'dry',orig_bytes:origBytes,new_bytes:newBytesLocal}); skipped++; cleanup(fin,fout); continue; }

    // 4) DELETE-FIRST (frees storage so the smaller upload won't 413)
    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,orig_bytes:origBytes}); failed++; cleanup(fin,fout); continue; }

    // 5) UPLOAD smaller (base64 attachment — no external host needed)
    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){
      // RESTORE original — never leave imageless
      await sleep(WRITEDELAY);
      const rs=await api('POST',`/admin/api/2024-10/products/${pid}/images.json`,{image:{attachment:origBuf.toString('base64'),filename:`restore-${oid}.jpg`,position,...(alt?{alt}:{})}});
      if(rs.status>=400||!rs.json.image){ await mark(row.id,{state:'imageless_alert',fail_reason:'upload+restore_failed',orig_bytes:origBytes}); alerts++; }
      else { await mark(row.id,{state:'failed',fail_reason:'upload_'+up.status+'_restored',orig_bytes:origBytes}); failed++; }
      cleanup(fin,fout); continue;
    }
    const ni=up.json.image;

    // 6) verify: new bytes < orig, dims ok, product image count unchanged
    const nb=await head(ni.src)||newBytesLocal;
    const pr2=await api('GET',`/admin/api/2024-10/products/${pid}.json?fields=images`);
    const count2=((pr2.json.product&&pr2.json.product.images)||[]).length;
    const ok = nb>0 && nb<origBytes && (ni.width||0)<=2048 && count2===origCount;
    if(ok){
      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);
    } else {
      // regression → restore original, remove the new one
      await sleep(WRITEDELAY);
      await api('POST',`/admin/api/2024-10/products/${pid}/images.json`,{image:{attachment:origBuf.toString('base64'),filename:`restore-${oid}.jpg`,position,...(alt?{alt}:{})}});
      await sleep(WRITEDELAY);
      await api('DELETE',`/admin/api/2024-10/products/${pid}/images/${ni.id}.json`);
      await mark(row.id,{state:'failed',fail_reason:`verify(nb=${nb},w=${ni.width},cnt=${count2}/${origCount})`,orig_bytes:origBytes});
      failed++;
    }
    cleanup(fin,fout);
  }catch(e){
    await mark(row.id,{state:'failed',fail_reason:('exc:'+String(e.message||e)).slice(0,120)}); failed++;
  }
  if(processed%25===0) console.log(`[${WID}] processed ${processed} | done ${done} skip ${skipped} fail ${failed} alert ${alerts} | 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} imageless_alert=${alerts} reclaimed=${(reclaimed/1048576).toFixed(1)}MB`);
await db.end();