← back to New Import Viewer
consumer.js
370 lines
#!/usr/bin/env node
// launch-queue CONSUMER — turns manifested launch batches (data/launch-batches/)
// into Shopify products. DRY-RUN BY DEFAULT; a real store write requires BOTH
// the env ceiling LAUNCH_CONSUMER_LIVE=1 AND the --live flag (Steve-gated).
//
// Execution-time gates (per product, fresh DB reads — staging-time checks are
// not trusted because the catalog moves):
// 1. image present (ALL PRODUCTS MUST HAVE IMAGES — hard rule 2026-06-10)
// 2. dw_sku + pattern present (catalog discipline)
// 3. NEVER DUPLICATE (mfr_sku not already in shopify_products, fresh;
// row not linked; in-run dedup by vendor+mfr_sku)
// 4. vendor denylist (cowtan_tout: Steve 'never')
// 5. private label (command54 → 'Phillipe Romano'; code never public)
// 6. status forced DRAFT (active only via --allow-active, still Steve-gated)
//
// Throughput: hard cap per run (default 400 live — Shopify daily variant-creation
// limits killed faster pushes before; rebel-walls ran ~495/day). Ledger
// (data/launch-consumer-ledger.jsonl) makes runs idempotent + resumable.
//
// Usage:
// node consumer.js # dry-run, 25 products
// node consumer.js --limit 100 # dry-run, 100
// LAUNCH_CONSUMER_LIVE=1 node consumer.js --live --limit 400 # real draft creates
// node consumer.js --retry-failed # re-attempt ledger 'failed' rows
const fs = require('fs');
const path = require('path');
const { spawnSync } = require('child_process');
const ARGS = process.argv.slice(2);
const flag = (n) => ARGS.includes(n);
const opt = (n, d) => { const i = ARGS.indexOf(n); return i >= 0 ? ARGS[i + 1] : d; };
const PSQL = process.env.PSQL || '/opt/homebrew/opt/postgresql@14/bin/psql';
const PGUSER = process.env.PGUSER || 'stevestudio2';
const PGDB = process.env.PGDATABASE || 'dw_unified';
const US = '\x1f', RS = '\x1e';
function q(sql) {
const r = spawnSync(PSQL, ['-U', PGUSER, '-d', PGDB, '-tA', '-F', US, '-R', RS, '-c', sql],
{ encoding: 'utf8', timeout: 120000, maxBuffer: 1 << 28 });
if (r.status !== 0) throw new Error((r.stderr || 'psql failed').slice(0, 400));
return (r.stdout || '').replace(/\n+$/, '').split(RS).filter(l => l.length).map(l => l.split(US));
}
const esc = (s) => String(s == null ? '' : s).replace(/'/g, "''");
function loadEnv() {
const files = [path.join(__dirname, '.env'), '/Users/macstudio3/Projects/secrets-manager/.env'];
const env = {};
for (const f of files) {
try {
for (const line of fs.readFileSync(f, 'utf8').split('\n')) {
const m = line.match(/^([A-Z0-9_]+)=(.*)$/);
if (m && env[m[1]] === undefined) env[m[1]] = m[2].replace(/^["']|["']$/g, '');
}
} catch {}
}
return env;
}
const ENV = loadEnv();
// .env FILE wins over process.env (matches server.js) — defeats stale-shell drift
// silently retargeting the store or token.
const SHOP_STORE = ENV.SHOPIFY_STORE || process.env.SHOPIFY_STORE || 'designer-laboratory-sandbox.myshopify.com';
const SHOP_TOKEN = ENV.SHOPIFY_ADMIN_TOKEN || process.env.SHOPIFY_ADMIN_TOKEN || '';
const SHOP_API = '2024-10';
// ── the Steve gate ──
const LIVE = flag('--live') && (process.env.LAUNCH_CONSUMER_LIVE || ENV.LAUNCH_CONSUMER_LIVE) === '1';
const ALLOW_ACTIVE = flag('--allow-active'); // otherwise everything is DRAFT
const RETRY_FAILED = flag('--retry-failed');
const HARD_CAP = 450; // per-run ceiling, daily-limit safety
const LIMIT = Math.min(parseInt(opt('--limit', LIVE ? '400' : '25'), 10) || 25, HARD_CAP);
const VENDOR_DENY = new Set(['cowtan_tout']); // Steve: 'never'
// HOLD = staged but blocked from creation pending a Steve ruling. command54's
// private-label brand is disputed: vendor_registry says 'Hollywood Wallcoverings',
// Steve's command54 skill says 'Phillipe Romano'. Until resolved, never create.
const VENDOR_HOLD = new Set(['command54']);
const PRIVATE_LABEL = { command54: 'Phillipe Romano' };// code must NEVER appear publicly
const titleCase = (v) => String(v || '').replace(/_/g, ' ').replace(/\b\w/g, c => c.toUpperCase());
const LEDGER = path.join(__dirname, 'data', 'launch-consumer-ledger.jsonl');
function loadLedger() {
const done = new Map(); // id -> status
try {
for (const line of fs.readFileSync(LEDGER, 'utf8').split('\n')) {
if (!line.trim()) continue;
// Only 'created' and 'failed' block reprocessing. 'dryrun' and 'skipped'
// do NOT block — every gate re-checks fresh from the DB, so a row skipped
// for a transient reason (no image, scrape gap) re-flows once it's fixed.
try { const e = JSON.parse(line); if (e.id != null && (e.status === 'created' || e.status === 'failed')) done.set(e.id, e.status); } catch {}
}
} catch {}
return done;
}
function ledger(e) { fs.appendFileSync(LEDGER, JSON.stringify({ ...e, at: new Date().toISOString() }) + '\n'); }
// Collect candidate ids from manifested launch batches, oldest first, deduped.
function candidates() {
const dir = path.join(__dirname, 'data', 'launch-batches');
const files = fs.readdirSync(dir).filter(f => f.startsWith('launch-') && f.endsWith('.json'))
.map(f => {
// One corrupt/truncated manifest (crash mid-write) must not FATAL the run.
try { return { f, m: JSON.parse(fs.readFileSync(path.join(dir, f), 'utf8')) }; }
catch (e) { console.warn(` skip corrupt manifest ${f}: ${e.message}`); return null; }
})
.filter(Boolean)
.filter(({ m }) => m && m.action === 'launch' && Array.isArray(m.skus)) // creation consumer only
.sort((a, b) => a.m.at.localeCompare(b.m.at));
const seen = new Set(), out = [];
for (const { m } of files) for (const s of m.skus) {
if (!seen.has(s.id)) { seen.add(s.id); out.push(s.id); }
}
return out;
}
function parsePrice(specs) {
try {
const s = typeof specs === 'string' ? JSON.parse(specs) : specs || {};
for (const k of ['retail_price_usd', 'retail_price', 'list_price']) {
// Match the FIRST plain number, don't digit-mash: "$12.99 – $45" → 12.99,
// "1.234,56" → 1.234 (not 1.23456→1.23). Avoids corrupted prices shipping.
const m = String(s[k] ?? '').match(/\d+(?:\.\d{1,2})?/);
if (m && +m[0] > 0) return (+m[0]).toFixed(2);
}
} catch {}
return null;
}
function parseImages(imageUrl, allImages) {
const out = [];
const push = (u) => { u = String(u || '').trim(); if (/^https?:\/\//.test(u) && !out.includes(u)) out.push(u); };
push(imageUrl);
if (allImages) {
let list = [];
try { list = JSON.parse(allImages); } catch { list = String(allImages).split(/[\n,|]+/); }
if (Array.isArray(list)) list.forEach(push);
}
return out.slice(0, 6);
}
async function shopifyCreate(payload) {
const url = `https://${SHOP_STORE}/admin/api/${SHOP_API}/products.json`;
for (let attempt = 1; attempt <= 5; attempt++) {
const r = await fetch(url, {
method: 'POST',
headers: { 'X-Shopify-Access-Token': SHOP_TOKEN, 'content-type': 'application/json' },
body: JSON.stringify({ product: payload }),
});
if (r.status === 429) {
const wait = (parseFloat(r.headers.get('retry-after')) || 2) * 1000;
await new Promise(res => setTimeout(res, wait)); continue;
}
if (!r.ok) return { ok: false, error: `HTTP ${r.status} ${(await r.text()).slice(0, 200)}` };
const j = await r.json().catch(() => ({}));
return { ok: true, id: j.product?.id, handle: j.product?.handle };
}
return { ok: false, error: '429 retries exhausted' };
}
// ── publish gate fix (2026-06-19) ──────────────────────────────────────────
// REST `status:'active'` does NOT publish a product to a sales channel — it only
// flips the admin status. `published_at`/`published_scope` are NOT set by the
// create payload, so every active product this consumer created landed on ZERO
// channels (publishedOnPublication=false, resourcePublicationsCount=0). The
// Online Store channel requires an explicit publish call. We resolve the
// Online Store publication id once, then publishablePublish each active product.
// Drafts are intentionally NEVER published (image/width gate not yet satisfied).
const SHOP_GQL = `https://${SHOP_STORE}/admin/api/${SHOP_API}/graphql.json`;
let ONLINE_STORE_PUB_ID = null;
async function gql(query, variables) {
for (let attempt = 1; attempt <= 5; attempt++) {
const r = await fetch(SHOP_GQL, {
method: 'POST',
headers: { 'X-Shopify-Access-Token': SHOP_TOKEN, 'content-type': 'application/json' },
body: JSON.stringify({ query, variables }),
});
if (r.status === 429) { await new Promise(res => setTimeout(res, 2000 * attempt)); continue; }
const j = await r.json().catch(() => ({}));
if (j.errors && /throttled/i.test(JSON.stringify(j.errors))) { await new Promise(res => setTimeout(res, 2000 * attempt)); continue; }
return j;
}
return { errors: [{ message: 'gql 429 retries exhausted' }] };
}
async function resolveOnlineStorePub() {
if (ONLINE_STORE_PUB_ID) return ONLINE_STORE_PUB_ID;
const j = await gql(`{publications(first:25){edges{node{id name}}}}`);
const edges = j.data?.publications?.edges || [];
const os = edges.find(e => /online store/i.test(e.node.name));
ONLINE_STORE_PUB_ID = os ? os.node.id : null;
return ONLINE_STORE_PUB_ID;
}
// Publish one product to the Online Store channel. Returns {ok} or {ok:false,error}.
async function shopifyPublishOnlineStore(productId) {
const pubId = await resolveOnlineStorePub();
if (!pubId) return { ok: false, error: 'Online Store publication not found' };
const gid = `gid://shopify/Product/${productId}`;
const j = await gql(
`mutation($id:ID!,$pub:ID!){publishablePublish(id:$id,input:[{publicationId:$pub}]){userErrors{field message}}}`,
{ id: gid, pub: pubId });
if (j.errors) return { ok: false, error: JSON.stringify(j.errors).slice(0, 200) };
const ue = (j.data?.publishablePublish?.userErrors || []).filter(e => !/already/i.test(e.message || ''));
if (ue.length) return { ok: false, error: JSON.stringify(ue).slice(0, 200) };
return { ok: true };
}
(async () => {
console.log(`launch-consumer · mode=${LIVE ? '🔴 LIVE (draft creates)' : '🟢 DRY-RUN'} · limit=${LIMIT} · store=${SHOP_STORE}${SHOP_TOKEN ? '' : ' · NO TOKEN'}`);
if (flag('--live') && !LIVE) {
console.error('refused: --live also requires env LAUNCH_CONSUMER_LIVE=1 (the Steve gate). Staying dry.');
process.exit(2);
}
if (LIVE && !SHOP_TOKEN) { console.error('refused: live mode with no SHOPIFY_ADMIN_TOKEN'); process.exit(2); }
const done = loadLedger();
const ids = candidates().filter(id => {
const st = done.get(id);
if (!st) return true;
return RETRY_FAILED && st === 'failed';
});
console.log(`candidates: ${ids.length.toLocaleString()} unprocessed (ledger has ${done.size.toLocaleString()})`);
const runMfr = new Set(); // in-run dedup, keyed on mfr alone
let created = 0, skipped = 0, failed = 0, dry = 0, consecFail = 0;
const counts = {};
const skipNote = (id, dwSku, reason) => { skipped++; counts[reason] = (counts[reason] || 0) + 1; ledger({ id, dwSku, status: 'skipped', reason }); };
for (const id of ids) {
if (created + dry >= LIMIT) break;
const rows = q(`SELECT vc.id, vc.vendor_code, vc.mfr_sku, vc.dw_sku, vc.pattern_name, vc.color_name,
vc.collection, vc.product_type, vc.image_url, vc.all_images, vc.specs::text,
vc.ai_description, vc.width, vc.pattern_repeat, vc.match_type, vc.fire_rating,
vc.composition, vc.sell_unit, vc.original_vendor_name, vc.shopify_product_id,
(SELECT count(*) FROM shopify_products sx
WHERE upper(trim(sx.mfr_sku)) = upper(trim(vc.mfr_sku))) AS dup,
vr.vendor_name, vr.is_private_label, vr.private_label_name, vr.skip_shopify,
(SELECT count(*) FROM vendor_catalog v2
WHERE upper(trim(v2.mfr_sku)) = upper(trim(vc.mfr_sku))
AND v2.shopify_product_id IS NOT NULL AND v2.id <> vc.id) AS dup2
FROM vendor_catalog vc
LEFT JOIN vendor_registry vr ON vr.vendor_code = vc.vendor_code
WHERE vc.id = ${+id}`);
if (!rows.length) { skipNote(id, null, 'row gone'); continue; }
const [, vendor, mfr, dwSku, pattern, color, collection, ptype, imageUrl, allImages, specs,
aiDesc, width, repeat, match, fire, composition, sellUnit, origVendor, linkedPid, dup,
regName, isPL, plName, skipShopify, dup2] = rows[0];
// gates
if (VENDOR_DENY.has(vendor)) { skipNote(+id, dwSku, 'vendor denylisted'); continue; }
if (VENDOR_HOLD.has(vendor)) { skipNote(+id, dwSku, 'vendor on hold (brand ruling pending)'); continue; }
if (skipShopify === 't') { skipNote(+id, dwSku, 'vendor_registry.skip_shopify'); continue; }
if (!imageUrl) { skipNote(+id, dwSku, 'no image'); continue; }
if (!dwSku) { skipNote(+id, dwSku, 'no dw_sku'); continue; }
if (!pattern) { skipNote(+id, dwSku, 'no pattern name'); continue; }
if (linkedPid) { skipNote(+id, dwSku, 'already linked'); continue; }
// NEVER DUPLICATE — blocked if the mfr_sku is already on Shopify (mirror) OR
// a sibling vendor_catalog row is already linked (catches re-scrape cohorts
// before the nightly mirror sync sees them). NULL/empty mfr can't be deduped
// at all → must skip, never create.
const mfrNorm = String(mfr || '').trim().toUpperCase();
if (!mfrNorm) { skipNote(+id, dwSku, 'no mfr_sku (cannot dedup)'); continue; }
if (+dup > 0) { skipNote(+id, dwSku, 'mfr_sku already on Shopify'); continue; }
if (+dup2 > 0) { skipNote(+id, dwSku, 'mfr_sku already linked via sibling row'); continue; }
if (runMfr.has(mfrNorm)) { skipNote(+id, dwSku, 'duplicate within run'); continue; }
runMfr.add(mfrNorm); // key on mfr ALONE — same SKU under 2 vendor codes is 1 product
// Display name resolution: private-label name wins (vendor code must never
// leak publicly), then registry display name, then catalog original, then
// de-snaked code as last resort. A private-label-flagged vendor with NO mask
// name must NOT fall through to its real name → skip rather than leak.
const displayVendor = (isPL === 't' && plName && plName.trim()) || PRIVATE_LABEL[vendor]
|| (regName && regName.trim()) || (origVendor && origVendor.trim()) || titleCase(vendor);
if (isPL === 't' && !((plName && plName.trim()) || PRIVATE_LABEL[vendor])) {
skipNote(+id, dwSku, 'private-label vendor with no mask name (would leak real name)'); continue;
}
const price = parsePrice(specs);
const images = parseImages(imageUrl, allImages);
// The launch gate checks raw image_url, but only http(s) URLs survive into the
// payload — a relative/junk image_url would create an IMAGELESS product. Block.
if (!images.length) { skipNote(+id, dwSku, 'no usable http image'); continue; }
const specRows = [['Width', width], ['Repeat', repeat], ['Match', match], ['Fire Rating', fire],
['Composition', composition], ['Sold By', sellUnit]]
.filter(([, v]) => v && String(v).trim());
// specs live in metafields (rendered by product-description-meta.liquid), NEVER a body list (TK-10039)
void specRows;
const body = aiDesc ? `<p>${aiDesc}</p>` : '';
const payload = {
title: `${pattern}${color ? ' - ' + color : ''} | ${displayVendor}`,
vendor: displayVendor,
product_type: ptype || 'Wallcovering',
status: ALLOW_ACTIVE ? 'active' : 'draft',
tags: [displayVendor, collection].filter(Boolean).join(', '),
body_html: body,
// Main variant + REQUIRED Sample variant (standing rule: every product
// ships a {DW_SKU}-Sample at $4.25 with no inventory tracking).
// REST requires every variant carry the option value when `options` is set,
// so the main variant gets option1:'Default' (Shopify's reserved single-
// option title) and the sample gets option1:'Sample'.
variants: [
{ option1: 'Default', sku: dwSku, ...(price ? { price } : {}), inventory_management: null },
{ option1: 'Sample', sku: `${dwSku}-Sample`, price: '4.25', inventory_management: null },
],
options: [{ name: 'Title' }],
images: images.map(src => ({ src })),
metafields: [
{ namespace: 'custom', key: 'mfr_sku', value: String(mfr || ''), type: 'single_line_text_field' },
...(collection ? [{ namespace: 'custom', key: 'collection', value: String(collection), type: 'single_line_text_field' }] : []),
// spec facts migrated from the old body <ul> → the metafields the theme renders (TK-10039)
...(width ? [{ namespace:'custom', key:'width', value:String(width), type:'single_line_text_field' }, { namespace:'global', key:'width', value:String(width), type:'single_line_text_field' }] : []),
...(repeat ? [{ namespace:'custom', key:'pattern_repeat', value:String(repeat), type:'single_line_text_field' }] : []),
...(match ? [{ namespace:'global', key:'Match', value:String(match), type:'single_line_text_field' }] : []),
...(fire ? [{ namespace:'custom', key:'fire_rating', value:String(fire), type:'single_line_text_field' }] : []),
...(composition ? [{ namespace:'custom', key:'material', value:String(composition), type:'multi_line_text_field' }] : []),
...(sellUnit ? [{ namespace:'global', key:'unit_of_measure', value:String(sellUnit), type:'single_line_text_field' }] : []),
],
};
// Final belt: no vendor text (ai_description, collection, etc.) may leak the
// raw private-label code anywhere in the outgoing payload.
if (/command\s*-?\s*54/i.test(JSON.stringify(payload))) {
skipNote(+id, dwSku, 'command54 string leaked into payload'); continue;
}
if (!LIVE) {
dry++;
if (dry <= 5) console.log(` DRY ${dwSku} "${payload.title}" imgs=${images.length} price=${price || '—'}`);
ledger({ id: +id, dwSku, status: 'dryrun', title: payload.title, imgs: images.length, price });
continue;
}
const w = await shopifyCreate(payload);
if (w.ok && !w.id) {
// 2xx but no product id parsed — treat as failure; do NOT UPDATE (NaN id
// would be a SQL error) and do NOT claim created.
failed++; consecFail++;
ledger({ id: +id, dwSku, status: 'failed', reason: 'created but no id returned' });
} else if (w.ok) {
created++; consecFail = 0;
// PUBLISH GATE FIX: an active product is invisible to shoppers until it's
// published to the Online Store channel. REST create never does this, so
// publish explicitly here. Drafts (no --allow-active) are left unpublished
// by design. A publish failure does NOT fail the create — the product
// exists + is ledgered; we record the publish outcome for re-sweeping.
let published = false, publishErr = null;
if (ALLOW_ACTIVE && payload.status === 'active') {
const pub = await shopifyPublishOnlineStore(w.id);
published = pub.ok; if (!pub.ok) publishErr = pub.error;
}
// LEDGER FIRST (before the fallible UPDATE) so a DB hiccup can never leave a
// created product recorded nowhere → re-created as a duplicate next run.
ledger({ id: +id, dwSku, status: 'created', shopifyId: w.id, handle: w.handle, published, publishErr });
try {
q(`UPDATE vendor_catalog SET shopify_product_id=${+w.id}, on_shopify=TRUE,
sync_status='pushed', shopify_synced_at=now() WHERE id=${+id}`);
// Self-seal the dup gate within this run, before the nightly mirror sync:
// insert the new product into the mirror so later candidates see it.
q(`INSERT INTO shopify_products (shopify_id, mfr_sku, status, vendor)
VALUES ('gid://shopify/Product/${+w.id}', '${esc(mfr)}', 'DRAFT', '${esc(displayVendor)}')
ON CONFLICT (shopify_id) DO NOTHING`);
} catch (e) { console.warn(`\n warn: vc/mirror update failed for ${dwSku} (product ${w.id} created+ledgered): ${e.message}`); }
process.stdout.write(`\r created ${created}/${LIMIT} (skipped ${skipped}, failed ${failed}) `);
} else {
failed++; consecFail++;
ledger({ id: +id, dwSku, status: 'failed', reason: w.error });
}
if (consecFail >= 10) { console.error(`\n circuit breaker: 10 consecutive failures — aborting (check token/store)`); break; }
await new Promise(res => setTimeout(res, 600)); // ~1.6/s, under REST 2/s
}
console.log(`\nSUMMARY · ${LIVE ? 'created ' + created : 'dry-run would create ' + dry} · skipped ${skipped} · failed ${failed}`);
if (Object.keys(counts).length) console.log('skip reasons:', JSON.stringify(counts));
if (!LIVE) console.log(`\nTo go live (Steve only): LAUNCH_CONSUMER_LIVE=1 node consumer.js --live --limit 400`);
})().catch(e => { console.error('FATAL:', e.message); process.exit(1); });