← back to Mfr Recovery 2026 08 23
durable-push-mfr.mjs
194 lines
#!/usr/bin/env node
/**
* durable-push-mfr.mjs — TK-10827 (follow-up to TK-10677)
* THE durable mfr_sku recovery push: writes BOTH authoritative surfaces so the
* recovery does NOT revert on the next Kamatera→Mac2 sync.
*
* Surface 1 (canonical, remote): Kamatera dw_unified `shopify_products.mfr_sku`
* — via ssh root@45.61.58.125 psql, one transaction, per-row.
* Surface 2 (customer-facing): Shopify product metafields
* custom.manufacturer_sku + dwc.manufacturer_sku (single_line_text_field).
*
* ORDER = Kamatera DB THEN Shopify (PostgreSQL-before-Shopify). The Mac2 mirror
* re-syncs FROM Kamatera, so we deliberately never write the Mac2 mirror here
* (that was the NON-durable bug: a mirror-only write reverts on next sync).
*
* SAFETY / RAILS:
* - DEFAULT is --dry-run: reads live state on BOTH surfaces, prints the plan,
* writes NOTHING. You must pass --apply to write.
* - --apply is a HARD-GATED, customer-facing + Kamatera-canonical write. Do NOT
* run it without Steve's approval (the harness classifier also blocks the ssh/
* Shopify write). See pending-approval/2026-08-25-TK-10827-durable-push-memo.md.
* - Every applied row records an old->new restore-map to
* out/durable-restore-<ts>.jsonl (both surfaces). Companion rollback:
* node rollback-durable-push.mjs out/durable-restore-<ts>.jsonl [--apply]
* - Idempotent: rows already correct on a surface are SKIPped, not rewritten.
* - Bounded: only the enumerated recovered rows (24 Zoffany, verified) are touched.
* - NEVER touches variants, price, status, images — mfr identity only.
*
* Usage:
* node durable-push-mfr.mjs # DRY-RUN (default), all vendors
* node durable-push-mfr.mjs --vendor=zoffany # scope (zoffany|novasuede|all)
* node durable-push-mfr.mjs --surface=shopify # scope (db|shopify|both, default both)
* node durable-push-mfr.mjs --apply # GATED write to both surfaces
*
* Novasuede note: the only "recoverable" Novasuede row resolves to a STRIPPED
* HANDLE (novasuede™-mist), not a real vendor code. It is EXCLUDED by default
* (writing a handle-string into mfr_sku is worse than blank). Pass
* --allow-novasuede-handle to include it anyway (still gated).
*/
import fs from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { execFileSync } from 'node:child_process';
import { gql, SHOP, VER, TOKEN } from '../designerwallcoverings/scripts/lib/shopify.mjs';
const HERE = path.dirname(fileURLToPath(import.meta.url));
const OUT = path.join(HERE, 'out');
fs.mkdirSync(OUT, { recursive: true });
const APPLY = process.argv.includes('--apply');
const ALLOW_NOVA_HANDLE = process.argv.includes('--allow-novasuede-handle');
const arg = (k, d) => (process.argv.find(a => a.startsWith(`--${k}=`)) || `--${k}=${d}`).split('=')[1].toLowerCase();
const vendorArg = arg('vendor', 'all');
const surfaceArg = arg('surface', 'both'); // db | shopify | both
const KAM = 'root@45.61.58.125';
const NS = [
{ namespace: 'custom', key: 'manufacturer_sku', type: 'single_line_text_field' },
{ namespace: 'dwc', key: 'manufacturer_sku', type: 'single_line_text_field' },
];
// ── recovered rows (id, handle, mfr_sku, vendor) ────────────────────────────────
// Zoffany: from the VERIFIED CSV (24 real Z####NNNNNN mfr codes; live-join confirmed 0 drift).
function loadZoffany() {
const csv = path.join(HERE, 'zoffany_24_verified_20260824.csv');
if (!fs.existsSync(csv)) return [];
const [head, ...rows] = fs.readFileSync(csv, 'utf8').trim().split('\n');
const cols = head.split(',');
const ii = cols.indexOf('shopify_id'), hi = cols.indexOf('handle'), mi = cols.indexOf('mfr_sku');
return rows.map(l => l.split(',')).map(c => ({
vendor: 'Zoffany', id: Number(c[ii]), handle: c[hi], mfr_sku: c[mi],
})).filter(r => r.id && r.handle && r.mfr_sku);
}
// Novasuede: derived LIVE from the Mac2 mirror so we never push a stale set.
// Excluded by default because the only match is a stripped handle, not a real code.
function loadNovasuede() {
if (!ALLOW_NOVA_HANDLE) return [];
const sql = `
WITH stripped AS (
SELECT id, handle, regexp_replace(handle, '-luxury-suede.*|(-fabric-wallcovering)', '', 'g') AS sh
FROM shopify_products
WHERE vendor='Novasuede' AND status='ACTIVE' AND (mfr_sku IS NULL OR mfr_sku='')
)
SELECT s.id || '\t' || s.handle || '\t' || nc.mfr_sku
FROM stripped s JOIN novasuede_catalog nc ON nc.mfr_sku = s.sh;`;
let out;
try { out = execFileSync('psql', ['-h', '/tmp', '-d', 'dw_unified', '-tA', '-c', sql], { encoding: 'utf8' }); }
catch { return []; }
return out.trim().split('\n').filter(Boolean).map(l => {
const [id, handle, mfr_sku] = l.split('\t');
return { vendor: 'Novasuede', id: Number(id), handle, mfr_sku };
});
}
// ── Kamatera DB surface (canonical shopify_products.mfr_sku) ─────────────────────
function kamRead(ids) {
// returns {id: mfr_sku_or_empty} straight from Kamatera-canonical shopify_products
if (!ids.length) return {};
const sql = `SELECT id||'\t'||coalesce(mfr_sku,'') FROM shopify_products WHERE id IN (${ids.join(',')});`;
const out = execFileSync('ssh', [KAM, `psql -d dw_unified -tA -c "${sql.replace(/"/g, '\\"')}"`], { encoding: 'utf8' });
const m = {};
out.trim().split('\n').filter(Boolean).forEach(l => { const [id, v] = l.split('\t'); m[id] = v; });
return m;
}
function kamWrite(rows) {
// one transaction; per-row UPDATE keyed by id (id is safe/stable, no regex on Kamatera)
const stmts = rows.map(r =>
`UPDATE shopify_products SET mfr_sku='${r.mfr_sku.replace(/'/g, "''")}' WHERE id=${r.id} AND vendor='${r.vendor}';`
);
const body = `BEGIN; ${stmts.join(' ')} COMMIT;`;
execFileSync('ssh', [KAM, `psql -d dw_unified -c "${body.replace(/"/g, '\\"')}"`], { encoding: 'utf8', stdio: 'pipe' });
}
// ── Shopify surface ─────────────────────────────────────────────────────────────
async function productByHandle(handle) {
const q = `query($h:String!){ productByHandle(handle:$h){ id title status
metafields(first:50){ nodes { namespace key value } } } }`;
const d = await gql(q, { h: handle });
return d?.productByHandle || null;
}
async function main() {
let items = [];
if (vendorArg === 'all' || vendorArg === 'zoffany') items = items.concat(loadZoffany());
if (vendorArg === 'all' || vendorArg === 'novasuede') items = items.concat(loadNovasuede());
const doDB = surfaceArg === 'db' || surfaceArg === 'both';
const doSH = surfaceArg === 'shopify' || surfaceArg === 'both';
console.log(`Store ${SHOP} · API ${VER} · token …${TOKEN.slice(-4)} · Kamatera ${KAM}`);
console.log(`scope: vendor=${vendorArg} surface=${surfaceArg} · ${items.length} recovered rows · ${APPLY ? 'APPLY (GATED)' : 'DRY-RUN (default)'}\n`);
if (!items.length) { console.log('No recovered rows for this scope. Nothing to do.'); return; }
const restore = APPLY ? fs.openSync(path.join(OUT, `durable-restore-${Date.now()}.jsonl`), 'a') : null;
const c = { dbPlan: 0, dbWrote: 0, dbSkip: 0, shPlan: 0, shWrote: 0, shSkip: 0, notFound: 0, errs: 0 };
// ── Surface 1: Kamatera DB (canonical) — do first (PG-before-Shopify) ──────────
if (doDB) {
let live;
try { live = kamRead(items.map(i => i.id)); }
catch (e) { console.error(`Kamatera read FAILED (ssh/psql): ${String(e.message || e).slice(0, 200)}`); process.exit(2); }
const need = [];
for (const it of items) {
const cur = live[String(it.id)];
if (cur === undefined) { console.error(` ✗ DB NOT-FOUND id=${it.id} ${it.handle}`); c.notFound++; continue; }
if (cur === it.mfr_sku) { console.log(` = DB SKIP id=${it.id} ${it.mfr_sku}`); c.dbSkip++; continue; }
console.log(` ${APPLY ? '→ DB SET' : '· DB PLAN'} id=${it.id} ${it.mfr_sku} (was '${cur}')`);
c.dbPlan++;
it._dbOld = cur; need.push(it);
}
if (APPLY && need.length) {
try {
kamWrite(need);
for (const it of need) {
fs.writeSync(restore, JSON.stringify({ ts: new Date().toISOString(), surface: 'kamatera_db',
id: it.id, handle: it.handle, vendor: it.vendor, old: it._dbOld ?? '', new: it.mfr_sku }) + '\n');
c.dbWrote++;
}
} catch (e) { console.error(` ⚠ DB WRITE FAILED (txn rolled back): ${String(e.message || e).slice(0, 200)}`); c.errs++; }
}
}
// ── Surface 2: Shopify metafields ─────────────────────────────────────────────
if (doSH) {
for (const it of items) {
const p = await productByHandle(it.handle);
if (!p) { console.error(` ✗ SH NOT-FOUND ${it.handle}`); c.notFound++; continue; }
const ex = Object.fromEntries((p.metafields?.nodes || []).map(m => [`${m.namespace}.${m.key}`, m.value]));
if (NS.every(t => ex[`${t.namespace}.${t.key}`] === it.mfr_sku)) {
console.log(` = SH SKIP ${it.handle} ${it.mfr_sku}`); c.shSkip++; continue;
}
console.log(` ${APPLY ? '→ SH SET' : '· SH PLAN'} ${it.handle} ${it.mfr_sku} (was custom=${ex['custom.manufacturer_sku'] ?? '∅'} dwc=${ex['dwc.manufacturer_sku'] ?? '∅'})`);
c.shPlan++;
if (!APPLY) continue;
const mf = NS.map(t => ({ ownerId: p.id, namespace: t.namespace, key: t.key, type: t.type, value: it.mfr_sku }));
const r = await gql(`mutation($m:[MetafieldsSetInput!]!){ metafieldsSet(metafields:$m){ userErrors{ field message } } }`, { m: mf });
const ue = r?.metafieldsSet?.userErrors || r?.__err || [];
if (ue.length) { console.error(` ⚠ SH error ${it.handle}: ${JSON.stringify(ue).slice(0, 180)}`); c.errs++; continue; }
fs.writeSync(restore, JSON.stringify({ ts: new Date().toISOString(), surface: 'shopify_metafield',
ownerId: p.id, handle: it.handle, vendor: it.vendor,
set: NS.map(t => ({ ns: t.namespace, key: t.key, old: ex[`${t.namespace}.${t.key}`] ?? null, new: it.mfr_sku })) }) + '\n');
c.shWrote++;
}
}
if (restore) fs.closeSync(restore);
console.log(`\nDone. DB(plan=${c.dbPlan} wrote=${c.dbWrote} skip=${c.dbSkip}) SH(plan=${c.shPlan} wrote=${c.shWrote} skip=${c.shSkip}) not-found=${c.notFound} errors=${c.errs}`);
console.log(APPLY ? `restore-map written to out/ · rollback: node rollback-durable-push.mjs <that-file> --apply`
: `DRY-RUN — add --apply (GATED, needs Steve's go) to write both surfaces.`);
}
main().catch(e => { console.error(e); process.exit(1); });