← 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.');
}