← back to Gmc Titlefix
TK-10862: scaled primary-fix batch runner (capped, reversible, verify)
b76a6a6fb6fb2a57135bfe0b15e616c7bc611945 · 2026-08-27 19:50:15 -0700 · Steve Abrams
Files touched
Diff
commit b76a6a6fb6fb2a57135bfe0b15e616c7bc611945
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Thu Aug 27 19:50:15 2026 -0700
TK-10862: scaled primary-fix batch runner (capped, reversible, verify)
---
primary-fix-batch.mjs | 280 ++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 280 insertions(+)
diff --git a/primary-fix-batch.mjs b/primary-fix-batch.mjs
new file mode 100644
index 0000000..64f89e2
--- /dev/null
+++ b/primary-fix-batch.mjs
@@ -0,0 +1,280 @@
+// TK-10862 — Scaled PRIMARY-SOURCE roll-price fix (batch, capped).
+// Scales the PROVEN TK-10451 primary-fix-canary.mjs mechanism to the full roll-leaker
+// population, one capped batch at a time.
+//
+// MECHANISM (verified serving on the 3 canary offers as of 2026-08-27):
+// For each active roll-leaker, the $4.25 SAMPLE-variant offer lives in PRIMARY DS 180695450
+// (Content API input=API — the Shopify Google&YouTube channel's offers). We overwrite it
+// in place via Merchant API productInputs:insert on the SAME sample offerId with the REAL
+// ROLL variant's full attributes (title/link/sku/price/weight). Same offerId => insert
+// REPLACES the input on the same DS (no duplicate, no merge to lose) => serves real roll price.
+//
+// LEAK signature (matches tk10862-pilot-select): active product with
+// minVariantPrice <= 4.25 AND onGoogle AND maxVariantPrice > 4.26.
+// Sample offer id = shopify_US_<productId>_<sampleVariantId>, where sampleVid = the
+// lowest-priced (<=4.25) variant (the memo sample). Roll = highest-priced variant > 4.26.
+//
+// SAFETY / REVERSIBILITY (identical spirit to the canary):
+// - Enumerate -> for each, GET the processed sample offer from primary DS.
+// - SKIP-OWNERSHIP: if the offer is NOT owned by primary DS 180695450, refuse (would transfer ownership).
+// - SKIP-ALREADY-FIXED: if the offer already serves > $4.26 (roll price landed), no-op skip.
+// - SKIP no-image / no-sample-offer / bad-roll.
+// - Capture EXACT current processed body to data/primary-fix-rollback/<offerId>.json BEFORE writing
+// (re-insert reverts). Append every action to data/primary-fix-batch-log.jsonl (resumable).
+// - FULL productAttributes on insert (insert replaces the whole input).
+// - CAP guard (BATCH_CAP, default 4000). Refuses to exceed.
+// - Token refresh every 50 min. 90s+ nothing here — per-insert ~120ms pace.
+//
+// UNDO for a batch: node primary-fix-batch.mjs --rollback --apply
+// (re-inserts every captured pre-state body recorded in this batch's log).
+//
+// node primary-fix-batch.mjs # DRY-RUN: enumerate + plan (writes candidate + restore-map, NO writes to GMC)
+// node primary-fix-batch.mjs --apply # GATED: write the batch to primary DS 180695450
+// node primary-fix-batch.mjs --verify [N] # read back served price for N (default 30) landed offers spread across vendors/prices
+// node primary-fix-batch.mjs --rollback --apply # REVERT this batch: re-insert captured $4.25 bodies for every landed offerId
+// BATCH_CAP=4000 node primary-fix-batch.mjs --apply
+import fs from 'fs';
+import crypto from 'crypto';
+import { createRequire } from 'module';
+const require = createRequire(import.meta.url);
+const { token, MERCHANT } = require('./_auth.js');
+
+const HOME = process.env.HOME;
+const PRIMARY_DS = `accounts/${MERCHANT}/dataSources/180695450`;
+const ROLLBACK_DIR = new URL('./data/primary-fix-rollback/', import.meta.url).pathname;
+const DATA = new URL('./data/', import.meta.url).pathname;
+const LOG = DATA + 'primary-fix-batch-log.jsonl';
+const CAND = DATA + 'primary-fix-batch-candidates.json';
+const RESTORE = DATA + 'primary-fix-batch-restore-map.json';
+fs.mkdirSync(ROLLBACK_DIR, { recursive: true });
+
+const args = process.argv.slice(2);
+const has = f => args.includes(f);
+const APPLY = has('--apply');
+const ROLLBACK = has('--rollback');
+const VERIFY = has('--verify');
+const BATCH_CAP = parseInt(process.env.BATCH_CAP || '4000', 10);
+const sleep = ms => new Promise(r => setTimeout(r, ms));
+
+const SHOP = 'designer-laboratory-sandbox.myshopify.com';
+const ENDPOINT = `https://${SHOP}/admin/api/2024-10/graphql.json`;
+const GOOG = 'gid://shopify/Publication/29646651457';
+const envTxt = fs.readFileSync(HOME + '/Projects/secrets-manager/.env', 'utf8');
+const SHOP_TOKEN = (envTxt.match(/^SHOPIFY_ADMIN_TOKEN=(.+)$/m) || [])[1].replace(/['"\s]/g, '');
+const vid = g => g.split('/').pop();
+
+async function gql(query, variables) {
+ for (let a = 0; a < 8; a++) {
+ let r;
+ try { r = await fetch(ENDPOINT, { method: 'POST', headers: { 'X-Shopify-Access-Token': SHOP_TOKEN, 'Content-Type': 'application/json' }, body: JSON.stringify({ query, variables }) }); }
+ catch (e) { await sleep(1500 * (a + 1)); continue; }
+ if (r.status === 429) { await sleep(1500 * (a + 1)); continue; }
+ const j = await r.json();
+ if (j.errors) { if (JSON.stringify(j.errors).includes('THROTTLED')) { await sleep(1500 * (a + 1)); continue; } throw new Error(JSON.stringify(j.errors)); }
+ if ((j.extensions?.cost?.throttleStatus?.currentlyAvailable ?? 2000) < 500) await sleep(700);
+ return j.data;
+ }
+ throw new Error('gql exhausted');
+}
+
+async function getProcessed(tok, offerId) {
+ const name = `accounts/${MERCHANT}/products/online~en~US~${offerId}`;
+ for (let a = 0; a < 5; a++) {
+ const r = await fetch(`https://merchantapi.googleapis.com/products/v1/${name}`, { headers: { Authorization: 'Bearer ' + tok, Accept: 'application/json' } });
+ if (r.status === 429 || r.status >= 500) { await sleep(1000 * (a + 1)); continue; }
+ return { status: r.status, body: await r.json() };
+ }
+ return { status: 429, body: {} };
+}
+
+async function getServedV21(tok, offerId) {
+ const rid = `online:en:US:${offerId}`;
+ const r = await fetch(`https://shoppingcontent.googleapis.com/content/v2.1/${MERCHANT}/products/${encodeURIComponent(rid)}`, { headers: { Authorization: 'Bearer ' + tok } });
+ return r.json();
+}
+
+async function insertPrimary(tok, body) {
+ const url = `https://merchantapi.googleapis.com/products/v1/accounts/${MERCHANT}/productInputs:insert?dataSource=${encodeURIComponent(PRIMARY_DS)}`;
+ for (let a = 0; a < 6; a++) {
+ const r = await fetch(url, { method: 'POST', headers: { Authorization: 'Bearer ' + tok, 'Content-Type': 'application/json' }, body: JSON.stringify(body) });
+ if (r.status === 429 || r.status >= 500) { await sleep(1500 * (a + 1)); continue; }
+ return { status: r.status, text: await r.text() };
+ }
+ return { status: 429, text: 'insert backoff exhausted' };
+}
+
+// Build the FULL roll-variant productAttributes from the current (sample) processed offer + Shopify roll variant. (mirror of canary)
+function buildRollBody(processed, rollVariant, productTitle, handle) {
+ const pa = processed.productAttributes || {};
+ const rollTitle = (productTitle || pa.title || '').replace(/^Sample:\s*/i, '').slice(0, 150);
+ const rollLink = (pa.link || '').replace(/([?&]variant=)\d+/, `$1${rollVariant.id}`) ||
+ `https://www.designerwallcoverings.com/products/${handle}?variant=${rollVariant.id}&country=US¤cy=USD`;
+ const rollWeightLb = /wallcovering/i.test((pa.productTypes || []).join(' ')) ? 4 : 1;
+ return {
+ offerId: processed.offerId,
+ contentLanguage: 'en',
+ feedLabel: 'US',
+ productAttributes: {
+ title: rollTitle,
+ description: pa.description,
+ link: rollLink,
+ imageLink: pa.imageLink,
+ additionalImageLinks: pa.additionalImageLinks || [],
+ availability: pa.availability || 'in_stock',
+ condition: 'new',
+ price: { amountMicros: String(Math.round(parseFloat(rollVariant.price) * 1e6)), currencyCode: 'USD' },
+ brand: pa.brand,
+ mpn: rollVariant.sku || pa.mpn,
+ gtin: pa.gtin,
+ googleProductCategory: pa.googleProductCategory,
+ productTypes: pa.productTypes || [],
+ shippingWeight: { value: rollWeightLb, unit: 'lb' },
+ },
+ _meta: { rollSku: rollVariant.sku, rollPrice: rollVariant.price, oldPrice: pa.price?.amountMicros, rollTitle, imageLink: pa.imageLink },
+ };
+}
+
+// --- ROLLBACK: re-insert captured pre-state bodies for every landed offer in this batch's log ---
+async function doRollback() {
+ const landed = new Set();
+ if (fs.existsSync(LOG)) for (const ln of fs.readFileSync(LOG, 'utf8').split('\n')) { if (!ln) continue; try { const o = JSON.parse(ln); if (o.action === 'insert' && o.ok) landed.add(o.offerId); } catch {} }
+ console.log(`ROLLBACK — ${landed.size} landed offers to revert to captured pre-state.`);
+ let tok = await token(), tokAt = Date.now(), ok = 0, fail = 0;
+ for (const offerId of landed) {
+ if (Date.now() - tokAt > 50 * 60 * 1000) { tok = await token(); tokAt = Date.now(); }
+ const f = ROLLBACK_DIR + offerId + '.json';
+ if (!fs.existsSync(f)) { console.log(` [skip] no capture for ${offerId}`); continue; }
+ const orig = JSON.parse(fs.readFileSync(f, 'utf8'));
+ const body = { offerId, contentLanguage: 'en', feedLabel: 'US', productAttributes: orig.productAttributes };
+ if (!APPLY) { console.log(`[DRY rollback] would restore ${offerId} price=${orig.productAttributes?.price?.amountMicros}`); continue; }
+ const r = await insertPrimary(tok, body);
+ if (r.status >= 200 && r.status < 300) ok++; else fail++;
+ if ((ok + fail) % 100 === 0) console.log(` reverted ${ok} fail ${fail}`);
+ await sleep(140);
+ }
+ console.log(`ROLLBACK DONE: reverted ${ok} fail ${fail}`);
+}
+
+// --- VERIFY: read back served price for a spread of landed offers ---
+async function doVerify(n) {
+ const landed = [];
+ if (fs.existsSync(LOG)) for (const ln of fs.readFileSync(LOG, 'utf8').split('\n')) { if (!ln) continue; try { const o = JSON.parse(ln); if (o.action === 'insert' && o.ok) landed.push(o); } catch {} }
+ if (!landed.length) { console.log('No landed offers in log to verify.'); return; }
+ // spread across vendors + price bands
+ landed.sort((a, b) => (a.vendor || '').localeCompare(b.vendor || '') || (a.rollPrice - b.rollPrice));
+ const step = Math.max(1, Math.floor(landed.length / n));
+ const sample = [];
+ for (let i = 0; i < landed.length && sample.length < n; i += step) sample.push(landed[i]);
+ const tok = await token();
+ console.log(`=== VERIFY served price for ${sample.length} landed offers (spread across ${new Set(landed.map(x=>x.vendor)).size} vendors) ===`);
+ const rows = [];
+ let stillSample = 0;
+ for (const o of sample) {
+ const pr = await getServedV21(tok, o.offerId);
+ const served = pr.error ? `ERR ${pr.error.code}` : (pr.price ? parseFloat(pr.price.value) : null);
+ const link = pr.link || '';
+ const linkOk = /[?&]variant=/.test(link) ? '✓' : '—';
+ if (served !== null && !pr.error && served <= 4.26) stillSample++;
+ rows.push({ offerId: o.offerId, vendor: o.vendor, before: '$4.25', after_served: served === null ? 'n/a' : `$${served}`, roll_expected: `$${o.rollPrice}`, link: linkOk, title: (pr.title || '').slice(0, 40) });
+ await sleep(120);
+ }
+ console.table(rows.map(r => ({ offer: r.offerId.slice(0, 30), vendor: (r.vendor || '').slice(0, 16), before: r.before, after_served: r.after_served, roll: r.roll_expected, link: r.link })));
+ fs.writeFileSync(DATA + 'primary-fix-batch-verify.json', JSON.stringify({ ts: new Date().toISOString(), sampled: rows.length, still_at_4_25: stillSample, rows }, null, 2));
+ console.log(`\nVERIFY: ${rows.length} sampled, ${stillSample} still serving <=$4.26.`);
+ if (stillSample > 0) console.log('!! WARNING: some offers still serve the sample price — override may not have taken. Investigate before continuing cadence.');
+ return { sampled: rows.length, stillSample };
+}
+
+async function main() {
+ if (ROLLBACK) return doRollback();
+ if (VERIFY) { const n = parseInt(args[args.indexOf('--verify') + 1] || '30', 10) || 30; return doVerify(n); }
+
+ let tok = await token(); let tokAt = Date.now();
+ // resume: which offerIds already landed this batch
+ const done = new Set();
+ if (fs.existsSync(LOG)) for (const ln of fs.readFileSync(LOG, 'utf8').split('\n')) { if (!ln) continue; try { const o = JSON.parse(ln); if (o.action === 'insert' && o.ok) done.add(o.offerId); } catch {} }
+
+ const skip = { notPrimaryDs: 0, alreadyFixed: 0, noImage: 0, badRoll: 0, noSample: 0, processedGetFail: 0, resumeSkip: 0 };
+ const plans = [];
+ let scanned = 0, leaksSeen = 0, pages = 0, cur = null;
+ process.stderr.write(`Enumerating active roll-leakers (cap ${BATCH_CAP})...\n`);
+ do {
+ if (Date.now() - tokAt > 50 * 60 * 1000) { tok = await token(); tokAt = Date.now(); }
+ const d = await gql(`query($c:String){products(first:80,after:$c,query:"status:active"){pageInfo{hasNextPage endCursor} nodes{id handle vendor title onGoogle:publishedOnPublication(publicationId:"${GOOG}") featuredImage{url} variants(first:100){nodes{id sku title price}}}}}`, { c: cur });
+ const pg = d.products;
+ for (const p of pg.nodes) {
+ if (plans.length >= BATCH_CAP) break;
+ scanned++;
+ const vs = p.variants.nodes.map(v => ({ id: vid(v.id), sku: v.sku || '', title: v.title || '', price: parseFloat(v.price) }));
+ const prices = vs.map(v => v.price).filter(x => !isNaN(x));
+ if (!prices.length) continue;
+ const mn = Math.min(...prices), mx = Math.max(...prices);
+ if (!(mn <= 4.25 && p.onGoogle && mx > 4.26)) continue; // LEAK signature
+ leaksSeen++;
+ if (!p.featuredImage?.url) { skip.noImage++; continue; }
+ const rollV = vs.filter(v => v.price > 4.26).sort((a, b) => b.price - a.price)[0];
+ if (!rollV) { skip.badRoll++; continue; }
+ // sample variant = lowest-priced <=4.25 (the memo sample carrying the leaking offer id)
+ const sampleV = vs.filter(v => v.price <= 4.25).sort((a, b) => a.price - b.price)[0];
+ if (!sampleV) { skip.noSample++; continue; }
+ const pid = vid(p.id);
+ const offerId = `shopify_US_${pid}_${sampleV.id}`;
+ if (done.has(offerId)) { skip.resumeSkip++; continue; }
+ // GET processed sample offer from primary DS
+ const { status, body: processed } = await getProcessed(tok, offerId);
+ if (status !== 200) { skip.processedGetFail++; continue; }
+ if (processed.dataSource !== PRIMARY_DS) { skip.notPrimaryDs++; continue; } // refuse ownership transfer
+ // already fixed? served roll price already
+ const curPrice = parseFloat(processed.productAttributes?.price?.amountMicros || '0') / 1e6;
+ if (curPrice > 4.26) { skip.alreadyFixed++; continue; }
+ // capture prestate BEFORE any write
+ fs.writeFileSync(ROLLBACK_DIR + offerId + '.json', JSON.stringify({ capturedAt: new Date().toISOString(), dataSource: processed.dataSource, offerId, productAttributes: processed.productAttributes }, null, 2));
+ const plan = buildRollBody(processed, rollV, p.title, p.handle);
+ if (!plan.productAttributes.imageLink) { skip.noImage++; continue; }
+ plans.push({ offerId, vendor: p.vendor || '(none)', handle: p.handle, plan, rollPrice: +rollV.price.toFixed(2), oldPrice: curPrice });
+ await sleep(50);
+ }
+ pages++; if (pages % 10 === 0) process.stderr.write(` ...scanned ${scanned}, leaksSeen ${leaksSeen}, planned ${plans.length}/${BATCH_CAP}\n`);
+ cur = pg.pageInfo.hasNextPage ? pg.pageInfo.endCursor : null;
+ } while (cur && plans.length < BATCH_CAP);
+
+ const summary = {
+ generated_at: new Date().toISOString(), ticket: 'TK-10862', mechanism: 'TK-10451 primary-fix (productInputs:insert -> primary DS 180695450, sample-offerId supersede)',
+ mode: APPLY ? 'APPLY' : 'DRY-RUN', cap: BATCH_CAP, scanned_active: scanned, leaks_seen: leaksSeen, planned: plans.length, skipped: skip,
+ roll_price_dist: (() => { const a = plans.map(x => x.rollPrice).sort((x, y) => x - y); return a.length ? { min: a[0], median: a[a.length >> 1], max: a[a.length - 1] } : {}; })(),
+ vendors: [...new Set(plans.map(x => x.vendor))].length, cost: '$0 (Content API free)'
+ };
+ fs.writeFileSync(CAND, JSON.stringify({ summary, plans: plans.map(p => ({ offerId: p.offerId, vendor: p.vendor, handle: p.handle, rollPrice: p.rollPrice, oldPrice: p.oldPrice, rollSku: p.plan._meta.rollSku, rollTitle: p.plan._meta.rollTitle })) }, null, 2));
+ fs.writeFileSync(RESTORE, JSON.stringify({ ticket: 'TK-10862', generated_at: new Date().toISOString(), merchant: MERCHANT, primary_ds: PRIMARY_DS, undo: 'node primary-fix-batch.mjs --rollback --apply', count: plans.length, offers: plans.map(p => ({ offerId: p.offerId, pre_price: `$${p.oldPrice}`, will_set: `$${p.rollPrice}`, capture: 'data/primary-fix-rollback/' + p.offerId + '.json' })) }, null, 2));
+
+ console.log(JSON.stringify(summary, null, 2));
+
+ if (!APPLY) { console.log(`\nDRY-RUN — ${plans.length} planned. Candidate+restore-map written. Re-run with --apply to write to primary DS ${PRIMARY_DS}.`); return; }
+
+ console.log(`\n=== APPLYING ${plans.length} to primary DS ${PRIMARY_DS} ===`);
+ let ok = 0, fail = 0; const failSamples = [];
+ for (let i = 0; i < plans.length; i++) {
+ if (Date.now() - tokAt > 50 * 60 * 1000) { tok = await token(); tokAt = Date.now(); }
+ const pln = plans[i];
+ const body = { offerId: pln.plan.offerId, contentLanguage: pln.plan.contentLanguage, feedLabel: pln.plan.feedLabel, productAttributes: pln.plan.productAttributes };
+ const r = await insertPrimary(tok, body);
+ if (r.status >= 200 && r.status < 300) {
+ ok++;
+ fs.appendFileSync(LOG, JSON.stringify({ ts: new Date().toISOString(), offerId: pln.offerId, action: 'insert', ok: true, vendor: pln.vendor, rollPrice: pln.rollPrice, oldPrice: pln.oldPrice }) + '\n');
+ } else {
+ fail++;
+ const msg = r.text.slice(0, 160);
+ if (failSamples.length < 15) failSamples.push({ offerId: pln.offerId, status: r.status, msg });
+ fs.appendFileSync(LOG, JSON.stringify({ ts: new Date().toISOString(), offerId: pln.offerId, action: 'inserr', status: r.status, error: msg }) + '\n');
+ }
+ if (i % 50 === 0) process.stdout.write(` ${i}/${plans.length} ok ${ok} fail ${fail}\n`);
+ await sleep(140);
+ }
+ const result = { ts: new Date().toISOString(), ticket: 'TK-10862', mode: 'APPLY', cap: BATCH_CAP, planned: plans.length, applied_ok: ok, failed: fail, failSamples, cost: '$0' };
+ fs.writeFileSync(DATA + 'primary-fix-batch-result.json', JSON.stringify(result, null, 2));
+ console.log(`\nBATCH DONE: ok ${ok} / fail ${fail} of ${plans.length}`);
+ if (failSamples.length) { console.log('--- first failures ---'); failSamples.forEach(f => console.log(` ${f.offerId} ${f.status} ${f.msg}`)); }
+ console.log('\nWait ~2-4 min, then: node primary-fix-batch.mjs --verify 30');
+}
+
+main().catch(e => { console.error('FATAL', e.message); process.exit(1); });
← 0ab1c50 TK-10451: primary-source GMC $4.25 leak fix — canary proves
·
back to Gmc Titlefix
·
TK-10862: two-phase enumeration (fast Shopify scan + concurr 95d0cb6 →