← back to New Import Viewer
launch-queue CONSUMER: manifested batches → Shopify draft products. DRY-RUN default; live needs LAUNCH_CONSUMER_LIVE=1 + --live (Steve gate). Per-product execution gates: image required, dw_sku+pattern required, fresh NEVER-DUPLICATE (mfr_sku vs shopify_products + in-run dedup + linked check), vendor_registry skip_shopify + private-label display names, cowtan denylist, status forced draft, 450/run hard cap (daily variant limit), 600ms throttle + 429 retry, idempotent ledger resume; writes linkage back to vendor_catalog on create
24a8d839546f38bf584414b0afd3947745d7535a · 2026-06-10 07:49:03 -0700 · SteveStudio2
Files touched
M .gitignoreA consumer.js
Diff
commit 24a8d839546f38bf584414b0afd3947745d7535a
Author: SteveStudio2 <stevestudio2@SteveStacStudio.lan>
Date: Wed Jun 10 07:49:03 2026 -0700
launch-queue CONSUMER: manifested batches → Shopify draft products. DRY-RUN default; live needs LAUNCH_CONSUMER_LIVE=1 + --live (Steve gate). Per-product execution gates: image required, dw_sku+pattern required, fresh NEVER-DUPLICATE (mfr_sku vs shopify_products + in-run dedup + linked check), vendor_registry skip_shopify + private-label display names, cowtan denylist, status forced draft, 450/run hard cap (daily variant limit), 600ms throttle + 429 retry, idempotent ledger resume; writes linkage back to vendor_catalog on create
---
.gitignore | 1 +
consumer.js | 241 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 242 insertions(+)
diff --git a/.gitignore b/.gitignore
index 9d12bb6..7b641d5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -8,3 +8,4 @@ build/
.next/
data/imgcache/
data/launch-batches/
+data/launch-consumer-ledger.jsonl
diff --git a/consumer.js b/consumer.js
new file mode 100644
index 0000000..41e23af
--- /dev/null
+++ b/consumer.js
@@ -0,0 +1,241 @@
+#!/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/stevestudio2/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();
+const SHOP_STORE = process.env.SHOPIFY_STORE || ENV.SHOPIFY_STORE || 'designer-laboratory-sandbox.myshopify.com';
+const SHOP_TOKEN = process.env.SHOPIFY_ADMIN_TOKEN || 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'
+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;
+ try { const e = JSON.parse(line); if (e.id != null && e.status !== 'dryrun') 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 => ({ f, m: JSON.parse(fs.readFileSync(path.join(dir, f), 'utf8')) }))
+ .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']) {
+ const v = String(s[k] ?? '').replace(/[^0-9.]/g, '');
+ if (v && +v > 0) return (+v).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 <= 3; 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' };
+}
+
+(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 vendor+mfr
+ let created = 0, skipped = 0, failed = 0, dry = 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
+ 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] = rows[0];
+
+ // gates
+ if (VENDOR_DENY.has(vendor)) { skipNote(+id, dwSku, 'vendor denylisted'); 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; }
+ if (+dup > 0) { skipNote(+id, dwSku, 'mfr_sku already on Shopify'); continue; }
+ const mfrKey = vendor + '|' + String(mfr || '').trim().toUpperCase();
+ if (mfr && runMfr.has(mfrKey)) { skipNote(+id, dwSku, 'duplicate within run'); continue; }
+ if (mfr) runMfr.add(mfrKey);
+
+ // 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.
+ const displayVendor = (isPL === 't' && plName && plName.trim()) || PRIVATE_LABEL[vendor]
+ || (regName && regName.trim()) || (origVendor && origVendor.trim()) || titleCase(vendor);
+ const price = parsePrice(specs);
+ const images = parseImages(imageUrl, allImages);
+ const specRows = [['Width', width], ['Repeat', repeat], ['Match', match], ['Fire Rating', fire],
+ ['Composition', composition], ['Sold By', sellUnit]]
+ .filter(([, v]) => v && String(v).trim());
+ const body = (aiDesc ? `<p>${aiDesc}</p>` : '') +
+ (specRows.length ? '<ul>' + specRows.map(([k, v]) => `<li><strong>${k}:</strong> ${v}</li>`).join('') + '</ul>' : '');
+ 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,
+ variants: [{ sku: dwSku, ...(price ? { price } : {}), inventory_management: null }],
+ 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' }] : []),
+ ],
+ };
+
+ 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) {
+ created++;
+ q(`UPDATE vendor_catalog SET shopify_product_id=${+w.id}, on_shopify=TRUE,
+ sync_status='pushed', shopify_synced_at=now() WHERE id=${+id}`);
+ ledger({ id: +id, dwSku, status: 'created', shopifyId: w.id, handle: w.handle });
+ process.stdout.write(`\r created ${created}/${LIMIT} (skipped ${skipped}, failed ${failed}) `);
+ } else {
+ failed++;
+ ledger({ id: +id, dwSku, status: 'failed', reason: w.error });
+ }
+ 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); });
← 863ce65 staging manifests: every batch persists full id/SKU manifest
·
back to New Import Viewer
·
review fixes (3-agent panel): consumer atomic ledger-before- 01ee7be →