← back to Carnegie Reprice

apply-tk11434.mjs

276 lines

// apply-tk11434.mjs — TK-11434 Carnegie at-cost reprice EXECUTOR (spec-complete).
// ============================================================================
// GATED: --apply / --rollback are customer-facing LIVE Shopify money-writes.
// DRY-RUN by default (touches nothing on Shopify).
//
// CONTEXT (read before firing):
//   The live Carnegie v2 product set was CREATED at vendor cost (rollout*.mjs wrote
//   price:r.price with no markup — now fixed, commit ced1de9) and PUBLISHED by the
//   Sep-10 cutover, so 5,914 ACTIVE non-sample variants (99.76% of the line) sell at
//   cost (ratio ~1.0) vs the 1.5x floor. This fixer lifts each to proposed_retail.
//
//   >>> A SIBLING EXECUTOR (reprice-tk11434.mjs) IS THE ORIGINAL and was Steve-fired
//   >>> live. THIS file is the spec-complete successor: it adds a pre-flight >2%
//   >>> drift-abort, prefers the FULL-access token, prints a Σ-delta plan + an
//   >>> end-of-run re-sweep, and SHARES the same ledger + undo so state and rollback
//   >>> are unified across both. Because it shares the ledger and uses
//   >>> LIVE-RE-READ + COMPARE-AND-SET, running it after (or even alongside) the
//   >>> original is SAFE and idempotent: variants already at proposed_retail are
//   >>> logged NOOP and never re-written.
//
// SOURCE OF TRUTH: dw_unified.carnegie_reprice_staging_tk11434 (needs_reprice rows,
//   staged + join-verified 2026-09-10). NEVER re-derive from the buggy reprice.mjs
//   (that pulled REST ?vendor=Carnegie with no status filter -> 11,822 rows + whole-
//   dollar rounding). proposed_retail = round(cost/0.65/0.85, 2)  (~1.81x, TK-10686).
//
// SAFETY MODEL (mirrors the Steve-approved TK-11403 pattern):
//   * DRY-RUN default: prints count, sample rows, Σ delta (total uplift). Writes NOTHING.
//   * PRE-FLIGHT DRIFT GATE (--apply): live-reads a sample of the plan and classifies
//       each live price as at-staged-cost (repriceable) / at-proposed (already done)
//       / OTHER (real drift). Aborts the WHOLE run before any write if OTHER > 2%.
//   * PER-VARIANT COMPARE-AND-SET: only writes proposed_retail if the live price still
//       equals the staged cost baseline. If live == proposed -> NOOP. If live is
//       anything else -> SKIP (never clobber a value we didn't stage).
//   * BATCHED by product via productVariantsBulkUpdate; THROTTLED backoff; --limit=N
//       daily cap; RESUME-SAFE via the shared ledger (applied vids are skipped).
//   * SAMPLE ($4.25) variants are never in the plan and never touched.
//   * LEDGER: every intent/applied/noop/skip/failed row appended (reversible).
//   * ROLLBACK: node apply-tk11434.mjs --rollback [--apply]   (restores prePrice via
//       compare-and-set) — equivalent to the sibling `node undo-tk11434.mjs`.
//
// Usage:
//   node apply-tk11434.mjs                     # DRY-RUN plan (count, samples, Σ delta) — writes nothing
//   node apply-tk11434.mjs --apply             # GATED live reprice (Steve-approved only)
//   node apply-tk11434.mjs --apply --limit=1000 # first daily batch (Shopify variant cap)
//   node apply-tk11434.mjs verify              # read-only re-sweep: how many still <1.5x
//   node apply-tk11434.mjs --rollback          # DRY-RUN rollback plan
//   node apply-tk11434.mjs --rollback --apply  # GATED restore of prePrice for applied variants
import fs from 'node:fs';
import { execFileSync } from 'node:child_process';

const HOME = process.env.HOME;
const ENV  = `${HOME}/Projects/secrets-manager/.env`;
const env  = k => { const m = fs.readFileSync(ENV,'utf8').split('\n').find(l=>l.startsWith(k+'=')); return m ? m.slice(k.length+1).trim().replace(/^["']|["']$/g,'') : ''; };
// Prefer the FULL-access token (write_products); fall back to the narrow admin token.
const TOKEN_KEY = env('SHOPIFY_FULL_ACCESS_TOKEN') ? 'SHOPIFY_FULL_ACCESS_TOKEN' : 'SHOPIFY_ADMIN_TOKEN';
const TOKEN = env(TOKEN_KEY);
if (!TOKEN) { console.error('FATAL: no Shopify token in', ENV); process.exit(2); }
const DOMAIN = 'designer-laboratory-sandbox.myshopify.com';   // the LIVE DW store (legacy "sandbox" name)
const API = `https://${DOMAIN}/admin/api/2024-10/graphql.json`;
const sleep = ms => new Promise(r=>setTimeout(r,ms));

const ARGV     = process.argv.slice(2);
const APPLY    = ARGV.includes('--apply');
const VERIFY   = ARGV[0] === 'verify';
const ROLLBACK = ARGV.includes('--rollback');
const LIMIT    = Number((ARGV.find(a=>a.startsWith('--limit='))||'=0').split('=')[1]) || 0;
const DRIFT_ABORT_PCT = 0.02;       // >2% real-drift in the sample aborts the whole run
const DRIFT_SAMPLE_MAX = 300;       // pre-flight live-read sample size

const DIR = new URL('.', import.meta.url).pathname;
// SHARED ledger with reprice-tk11434.mjs / undo-tk11434.mjs so resume + rollback are unified.
const LED = `${DIR}reprice-tk11434-ledger.jsonl`;
const log = o => fs.appendFileSync(LED, JSON.stringify({ ts:new Date().toISOString(), tk:'TK-11434', src:'apply-tk11434', ...o }) + '\n');

async function gql(query, variables={}) {
  for (let a=0;a<6;a++){
    const r = await fetch(API,{ method:'POST', headers:{'X-Shopify-Access-Token':TOKEN,'Content-Type':'application/json'},
      body: JSON.stringify({query,variables}), signal: AbortSignal.timeout(45000) });
    if (r.status===429 || r.status>=500){ await sleep(1500*(a+1)); continue; }
    const j = await r.json();
    if (j.errors && JSON.stringify(j.errors).includes('THROTTLED')){ await sleep(3000*(a+1)); continue; }
    return j;
  }
  throw new Error('gql retries exhausted');
}
function psql(sql){
  const out = execFileSync('psql', ['postgresql:///dw_unified?host=/tmp','-At','-F','|','-c',sql], { encoding:'utf8', maxBuffer:256*1024*1024 });
  return out.trim() ? out.trim().split('\n').map(r=>r.split('|')) : [];
}
const money = n => Number(n).toFixed(2);

// ---- plan from the staged, join-verified table (SOURCE OF TRUTH) ----
const rows = psql(
  `SELECT product_id, variant_id, variant_gid, dw_sku, handle, current_live_price, cost, proposed_retail
     FROM carnegie_reprice_staging_tk11434
    WHERE needs_reprice
    ORDER BY product_id, variant_id`
).map(([product_id,variant_id,variant_gid,dw_sku,handle,old,cost,prop]) => ({
  product_id, variant_id, variant_gid: variant_gid || `gid://shopify/ProductVariant/${variant_id}`,
  dw_sku, handle, old:+old, cost:+cost, prop:+prop
}));
const sumDelta = rows.reduce((s,r)=>s+(r.prop - r.old), 0);
console.log(`[TK-11434] token=${TOKEN_KEY}  store=${DOMAIN}`);
console.log(`[plan] ${rows.length} needs_reprice variants  |  formula round(cost/0.65/0.85,2) (~1.81x)`);
console.log(`[plan] Σ current(live)=$${money(rows.reduce((s,r)=>s+r.old,0))}  Σ proposed=$${money(rows.reduce((s,r)=>s+r.prop,0))}  Σ delta(uplift)=$${money(sumDelta)}`);
console.log('[plan] sample:');
for (const r of rows.slice(0,5)) console.log(`   ${r.dw_sku}  ${money(r.old)} -> ${money(r.prop)}  (cost ${money(r.cost)}, +$${money(r.prop-r.old)})`);

// ---- ledger state: applied (and not later undone) vids ----
function ledgerApplied(){
  const done = new Map();   // vid -> {prePrice,newPrice}
  if (fs.existsSync(LED)) for (const l of fs.readFileSync(LED,'utf8').split('\n')){
    if(!l.trim())continue; try{ const r=JSON.parse(l);
      if (r.phase==='applied') done.set(String(r.vid), { prePrice:r.prePrice, newPrice:r.newPrice });
      if (r.phase==='undone')  done.delete(String(r.vid));
    }catch{}
  }
  return done;
}

// ---- live price reader (batched by 100) ----
async function livePrices(vids){
  const map = {};
  for (let i=0;i<vids.length;i+=100){
    const ids = vids.slice(i,i+100).map(v=>`gid://shopify/ProductVariant/${v}`);
    const j = await gql(`query($ids:[ID!]!){ nodes(ids:$ids){ ... on ProductVariant { id price product{ status } } } }`, { ids });
    for (const n of (j.data?.nodes||[]).filter(Boolean)) map[n.id.split('/').pop()] = { price:+n.price, status:n.product?.status };
    await sleep(250);
  }
  return map;
}

// ========================= VERIFY (read-only re-sweep) =========================
async function reSweep(){
  const live = await livePrices(rows.map(r=>r.variant_id));
  let ge15=0, atCost=0, other=0, missing=0;
  for (const r of rows){
    const lv = live[r.variant_id];
    if (!lv){ missing++; continue; }
    if (r.cost>0 && lv.price/r.cost >= 1.5) ge15++;
    else if (r.cost>0 && lv.price/r.cost < 1.5) atCost++;
    else other++;
  }
  return { ge15, atCost, other, missing, total:rows.length };
}
if (VERIFY){
  const s = await reSweep();
  console.log(`[verify] ratio>=1.5x=${s.ge15}  still<1.5x(at-cost)=${s.atCost}  other=${s.other}  missing=${s.missing}  of ${s.total}`);
  process.exit(s.atCost>0 ? 1 : 0);
}

// ============================= ROLLBACK ========================================
if (ROLLBACK){
  const done = ledgerApplied();
  const targets = [...done.entries()].map(([vid,v])=>({ vid, ...v }));
  console.log(`[rollback] ${targets.length} applied variant(s) in ledger to restore to prePrice`);
  if (!targets.length) process.exit(0);
  let restored=0, skip=0, failed=0;
  const byV = await livePrices(targets.map(t=>t.vid));
  const M = `mutation($pid:ID!,$id:ID!,$price:Money!){ productVariantsBulkUpdate(productId:$pid, variants:[{id:$id,price:$price}]){ productVariants{ id price } userErrors{ message } } }`;
  for (const t of targets){
    const gvid = `gid://shopify/ProductVariant/${t.vid}`;
    const j = await gql(`query($id:ID!){ productVariant(id:$id){ id price product{ id } } }`, { id: gvid });
    const v = j.data?.productVariant;
    if (!v){ console.log('SKIP not-found', t.vid); skip++; continue; }
    const cur = +v.price;
    if (Math.abs(cur - Number(t.prePrice)) < 0.005){ console.log('ALREADY-restored', t.vid, cur); skip++; continue; }
    if (Math.abs(cur - Number(t.newPrice)) >= 0.005){ console.log('SKIP moved-since', t.vid, 'live',cur,'expected',t.newPrice); log({phase:'undo-skip',vid:t.vid,liveNow:cur,expected:t.newPrice}); skip++; continue; }
    if (!APPLY){ console.log(`DRY  restore ${t.vid}  ${money(cur)} -> ${money(t.prePrice)}`); continue; }
    const r = await gql(M, { pid: v.product.id, id: gvid, price: money(t.prePrice) });
    const ue = r.data?.productVariantsBulkUpdate?.userErrors || [];
    if (ue.length || r.errors){ console.log('FAIL', t.vid, JSON.stringify(ue.length?ue:r.errors).slice(0,160)); log({phase:'undo-failed',vid:t.vid,err:ue.length?ue:r.errors}); failed++; await sleep(500); continue; }
    log({ phase:'undone', vid:t.vid, restoredTo:t.prePrice, fromPrice:cur }); console.log(`OK   restore ${t.vid}  ${money(cur)} -> ${money(t.prePrice)}`);
    restored++; await sleep(400);
  }
  console.log(`\n== ${APPLY?'RESTORED':'DRY-RUN would-restore'}=${restored} skip=${skip} failed=${failed} ==`);
  if (!APPLY) console.log('DRY-RUN — nothing written. Add --apply to restore.');
  process.exit(0);
}

// ============================= APPLY / DRY-RUN =================================
const done = ledgerApplied();

// ---- PRE-FLIGHT DRIFT GATE (only when about to write) ----
if (APPLY){
  const sampleVids = rows.filter(r=>!done.has(String(r.variant_id))).slice(0, DRIFT_SAMPLE_MAX).map(r=>r.variant_id);
  if (sampleVids.length){
    console.log(`[drift-gate] live-reading ${sampleVids.length}-variant sample...`);
    const live = await livePrices(sampleVids);
    let atCost=0, atProp=0, other=0, missing=0;
    const byId = Object.fromEntries(rows.map(r=>[String(r.variant_id), r]));
    for (const vid of sampleVids){
      const lv = live[String(vid)]; const r = byId[String(vid)];
      if (!lv){ missing++; continue; }
      if (Math.abs(lv.price - r.old)  < 0.005) atCost++;
      else if (Math.abs(lv.price - r.prop) < 0.005) atProp++;
      else other++;
    }
    const checked = sampleVids.length - missing;
    const driftFrac = checked ? other/checked : 0;
    console.log(`[drift-gate] at-staged-cost=${atCost} at-proposed(done)=${atProp} OTHER(drift)=${other} missing=${missing}  drift=${(driftFrac*100).toFixed(2)}%`);
    if (driftFrac > DRIFT_ABORT_PCT){
      console.error(`[drift-gate] ABORT — ${(driftFrac*100).toFixed(2)}% of sample drifted off the staged baseline (> ${(DRIFT_ABORT_PCT*100)}%). Staging is stale; re-stage before writing. Nothing written.`);
      log({ phase:'aborted', reason:'drift-gate', driftFrac, atCost, atProp, other, missing });
      process.exit(3);
    }
  }
}

// ---- group plan by product for bulk update ----
const byProduct = new Map();
for (const r of rows){ if(!byProduct.has(r.product_id)) byProduct.set(r.product_id,[]); byProduct.get(r.product_id).push(r); }

const M = `mutation($pid:ID!,$vars:[ProductVariantsBulkInput!]!){
  productVariantsBulkUpdate(productId:$pid, variants:$vars){
    productVariants{ id sku price } userErrors{ field message } } }`;

let applied=0, noop=0, skip=0, failed=0, variantsSeen=0;
for (const [pid, plan] of byProduct){
  if (LIMIT && variantsSeen >= LIMIT) break;
  const gpid = `gid://shopify/Product/${pid}`;
  const j = await gql(`query($id:ID!){ product(id:$id){ id status variants(first:100){ nodes{ id sku price } } } }`, { id: gpid });
  const p = j.data?.product;
  if (!p){ for(const r of plan){ log({phase:'skip',pid,vid:r.variant_id,sku:r.dw_sku,why:'product not found'}); skip++; } continue; }
  if (p.status !== 'ACTIVE'){ for(const r of plan){ log({phase:'skip',pid,vid:r.variant_id,sku:r.dw_sku,why:'status '+p.status}); skip++; } continue; }
  const liveById = Object.fromEntries(p.variants.nodes.map(n=>[n.id.split('/').pop(), n]));

  const vars = [];
  for (const r of plan){
    if (LIMIT && variantsSeen >= LIMIT) break;
    variantsSeen++;
    const vid = String(r.variant_id);
    if (done.has(vid)){ skip++; continue; }                       // resume: already applied in ledger
    const lv = liveById[vid];
    if (!lv){ log({phase:'skip',pid,vid,sku:r.dw_sku,why:'variant not on product'}); skip++; continue; }
    const cur = +lv.price;
    if (Math.abs(cur - r.prop) < 0.005){ log({phase:'noop',pid,vid,sku:r.dw_sku,price:cur,why:'already at proposed'}); noop++; continue; }
    if (Math.abs(cur - r.old)  >= 0.005){ log({phase:'skip',pid,vid,sku:r.dw_sku,liveNow:cur,stagedOld:r.old,why:'live moved off staged baseline — not clobbering'}); skip++; continue; }
    vars.push({ id: lv.id, price: money(r.prop), _r:r, _pre:cur });
  }
  if (!vars.length) continue;

  for (const v of vars) log({ phase:'intent', pid, vid:v._r.variant_id, sku:v._r.dw_sku, prePrice:v._pre, toPrice:v.price, cost:v._r.cost, undo:'node apply-tk11434.mjs --rollback --apply' });

  if (!APPLY){
    for (const v of vars) console.log(`DRY  ${v._r.dw_sku}  ${money(v._pre)} -> ${v.price}  (cost ${money(v._r.cost)})`);
    applied += vars.length;   // "would-apply" in dry mode
    continue;
  }

  const r = await gql(M, { pid: gpid, vars: vars.map(v=>({ id:v.id, price:v.price })) });
  const ue = r.data?.productVariantsBulkUpdate?.userErrors || [];
  if (ue.length || r.errors){
    for (const v of vars){ log({phase:'failed',pid,vid:v._r.variant_id,sku:v._r.dw_sku,err:ue.length?ue:r.errors}); failed++; }
    console.log(`FAIL product ${pid}: ${JSON.stringify(ue.length?ue:r.errors).slice(0,200)}`);
    await sleep(600); continue;
  }
  const got = Object.fromEntries((r.data.productVariantsBulkUpdate.productVariants||[]).map(n=>[n.id.split('/').pop(), n.price]));
  for (const v of vars){
    const np = got[v._r.variant_id];
    log({ phase:'applied', pid, vid:v._r.variant_id, sku:v._r.dw_sku, prePrice:v._pre, newPrice:np });
    console.log(`OK   ${v._r.dw_sku}  ${money(v._pre)} -> ${np}`);
    applied++;
  }
  await sleep(500);
}
console.log(`\n== ${APPLY?'APPLIED':'DRY-RUN would-apply'}=${applied} noop=${noop} skip=${skip} failed=${failed} (variants scanned ${variantsSeen}${LIMIT?`/limit ${LIMIT}`:''}) ==`);

if (APPLY){
  console.log('[re-sweep] confirming ratio>=1.5x across the full plan...');
  try { const s = await reSweep(); console.log(`[re-sweep] ratio>=1.5x=${s.ge15}/${s.total}  still<1.5x=${s.atCost}  other=${s.other}  missing=${s.missing}`); }
  catch(e){ console.log('[re-sweep] skipped:', e.message); }
} else {
  console.log('DRY-RUN — nothing written to Shopify. Add --apply (Steve-approved) to write.');
}