← back to New Import Viewer
review fixes (3-agent panel): consumer atomic ledger-before-UPDATE (no orphaned creates→dup), images.length gate (no imageless products via non-http urls), dedup keyed on mfr-alone + sibling-row check (dup2) + self-sealing mirror insert + NULL-mfr skip, command54 HOLD pending brand ruling, private-label-no-mask skip, payload-wide command54 leak scan, consec-failure circuit breaker, skipped no longer blocks reprocess, per-manifest parse guard + action=launch assert, single-number parsePrice, .env-first precedence; server localhost bind + SSRF redirect:manual host-revalidation + numeric-host block, NETNEW upper(trim()) match, /api/ids real 250k cap; functional indexes idx_sp/vc_mfr_upper_trim (stats 60s→1s)
01ee7be580e71f6695f0235191c5b50f3e72c5fa · 2026-06-10 08:21:02 -0700 · SteveStudio2
Files touched
Diff
commit 01ee7be580e71f6695f0235191c5b50f3e72c5fa
Author: SteveStudio2 <stevestudio2@SteveStacStudio.lan>
Date: Wed Jun 10 08:21:02 2026 -0700
review fixes (3-agent panel): consumer atomic ledger-before-UPDATE (no orphaned creates→dup), images.length gate (no imageless products via non-http urls), dedup keyed on mfr-alone + sibling-row check (dup2) + self-sealing mirror insert + NULL-mfr skip, command54 HOLD pending brand ruling, private-label-no-mask skip, payload-wide command54 leak scan, consec-failure circuit breaker, skipped no longer blocks reprocess, per-manifest parse guard + action=launch assert, single-number parsePrice, .env-first precedence; server localhost bind + SSRF redirect:manual host-revalidation + numeric-host block, NETNEW upper(trim()) match, /api/ids real 250k cap; functional indexes idx_sp/vc_mfr_upper_trim (stats 60s→1s)
---
consumer.js | 93 ++++++++++++++++++++++++++++++++++++++++++++++++-------------
server.js | 40 ++++++++++++++++++++------
2 files changed, 106 insertions(+), 27 deletions(-)
diff --git a/consumer.js b/consumer.js
index 41e23af..0409644 100644
--- a/consumer.js
+++ b/consumer.js
@@ -56,8 +56,10 @@ function loadEnv() {
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 || '';
+// .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 ──
@@ -68,6 +70,10 @@ const HARD_CAP = 450; // per-run ceiling, daily
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());
@@ -77,7 +83,10 @@ function loadLedger() {
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 {}
+ // 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;
@@ -88,7 +97,13 @@ function ledger(e) { fs.appendFileSync(LEDGER, JSON.stringify({ ...e, at: new Da
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')) }))
+ .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) {
@@ -101,8 +116,10 @@ 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);
+ // 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;
@@ -154,8 +171,8 @@ async function shopifyCreate(payload) {
});
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 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 }); };
@@ -167,34 +184,51 @@ async function shopifyCreate(payload) {
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
+ 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] = rows[0];
+ 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; }
- const mfrKey = vendor + '|' + String(mfr || '').trim().toUpperCase();
- if (mfr && runMfr.has(mfrKey)) { skipNote(+id, dwSku, 'duplicate within run'); continue; }
- if (mfr) runMfr.add(mfrKey);
+ 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.
+ // 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());
@@ -215,6 +249,12 @@ async function shopifyCreate(payload) {
],
};
+ // 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 || '—'}`);
@@ -222,16 +262,31 @@ async function shopifyCreate(payload) {
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}`);
+ 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;
+ // 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 });
+ 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++;
+ 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
}
diff --git a/server.js b/server.js
index 77ad704..6b90c18 100644
--- a/server.js
+++ b/server.js
@@ -81,8 +81,11 @@ const NEW_PRED = "(vc.sync_status='new' OR ((vc.on_shopify IS NOT TRUE) AND vc.s
// Index-friendly equality (idx_shopify_products_mfr_sku) → 240ms, not a seq scan.
// vendor_catalog + shopify_products store mfr_sku with consistent casing, so raw
// equality catches the overlap; a trim() guard handles stray whitespace cheaply.
+// Case/trim-folded match: raw equality missed 3,346 rows already on Shopify under
+// whitespace/case-differing mfr_sku, inflating NET NEW. (A functional index on
+// shopify_products(upper(trim(mfr_sku))) would keep this fast — recommend adding.)
const NETNEW_PRED = `(${NEW_PRED} AND vc.mfr_sku IS NOT NULL AND vc.mfr_sku <> '' AND NOT EXISTS (
- SELECT 1 FROM shopify_products sx WHERE sx.mfr_sku = vc.mfr_sku))`;
+ SELECT 1 FROM shopify_products sx WHERE upper(trim(sx.mfr_sku)) = upper(trim(vc.mfr_sku))))`;
// Resolve one shopify_products row per vendor_catalog row. shopify_id is a
// gid://shopify/Product/<n> string (UNIQUE index); vc.shopify_product_id is the
// bare bigint. Construct the gid on the vc side so the unique index on
@@ -179,13 +182,22 @@ function stageAction({ ids, action, mode }) {
const crypto = require('crypto');
const IMG_CACHE = path.join(__dirname, 'data', 'imgcache');
const UA = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126 Safari/537.36';
+// SSRF host guard — http(s) only; reject localhost/.local, IPv6 (colon), dotted
+// IPv4 (the WHATWG URL parser normalizes decimal/hex/octal IP literals to dotted
+// quads, so this regex catches http://2130706433 too), and any all-digit host.
+function hostAllowed(urlObj) {
+ const h = urlObj.hostname;
+ if (!/^https?:$/.test(urlObj.protocol)) return false;
+ if (h === 'localhost' || h.endsWith('.local')) return false;
+ if (h.includes(':')) return false; // IPv6 literal
+ if (/^\d+\.\d+\.\d+\.\d+$/.test(h)) return false; // dotted IPv4
+ if (/^\d+$/.test(h)) return false; // bare numeric edge encodings
+ return true;
+}
async function serveImg(u0, res) {
let t;
try { t = new URL(u0); } catch { return send(res, 400, JSON.stringify({ error: 'bad url' })); }
- const h = t.hostname;
- if (!/^https?:$/.test(t.protocol) || h === 'localhost' || h.endsWith('.local')
- || /^\d+\.\d+\.\d+\.\d+$/.test(h) || h.includes(':'))
- return send(res, 400, JSON.stringify({ error: 'host not allowed' }));
+ if (!hostAllowed(t)) return send(res, 400, JSON.stringify({ error: 'host not allowed' }));
const key = crypto.createHash('sha1').update(u0).digest('hex');
const f = path.join(IMG_CACHE, key), fct = f + '.ct';
try {
@@ -195,7 +207,19 @@ async function serveImg(u0, res) {
return res.end(body);
} catch { /* cache miss → fetch */ }
try {
- const r = await fetch(t, { headers: { 'user-agent': UA }, redirect: 'follow', signal: AbortSignal.timeout(15000) });
+ // Follow redirects MANUALLY so each hop's target is re-validated against the
+ // host guard — redirect:'follow' would let an allowed host 302 us into
+ // 127.0.0.1 / 169.254.169.254 / a LAN host and we'd fetch+cache it (SSRF).
+ let cur = t, r;
+ for (let hop = 0; hop < 4; hop++) {
+ r = await fetch(cur, { headers: { 'user-agent': UA }, redirect: 'manual', signal: AbortSignal.timeout(15000) });
+ if (r.status >= 300 && r.status < 400 && r.headers.get('location')) {
+ let next; try { next = new URL(r.headers.get('location'), cur); } catch { return send(res, 502, JSON.stringify({ error: 'bad redirect' })); }
+ if (!hostAllowed(next)) return send(res, 400, JSON.stringify({ error: 'redirect host not allowed' }));
+ cur = next; continue;
+ }
+ break;
+ }
if (!r.ok) return send(res, 502, JSON.stringify({ error: 'upstream ' + r.status }));
const ct = (r.headers.get('content-type') || 'image/jpeg').split(';')[0].trim();
if (!ct.startsWith('image/')) return send(res, 502, JSON.stringify({ error: 'not an image: ' + ct }));
@@ -296,7 +320,7 @@ const server = http.createServer((req, res) => {
const limit = Math.min(Math.max(parseInt(u.searchParams.get('limit'), 10) || 0, 0), 250000);
const where = buildWhere({ status, vendor, term });
const rows = q(`SELECT vc.id FROM vendor_catalog vc ${SP_JOIN} ${where}
- ORDER BY ${SORTS[sort]}${limit ? ` LIMIT ${limit}` : ''}`);
+ ORDER BY ${SORTS[sort]} LIMIT ${limit || 250000}`);
return send(res, 200, JSON.stringify({ ids: rows.map(r => +r[0]) }));
}
// Generalized action endpoint — stages launch / publish / unpublish / archive.
@@ -411,4 +435,4 @@ const server = http.createServer((req, res) => {
return send(res, 500, JSON.stringify({ error: e.message }));
}
});
-server.listen(PORT, () => console.log(`[new-import-viewer] http://127.0.0.1:${PORT} · store=${SHOP_STORE} · live-actions=[${[...LIVE_ACTIONS].join(',') || 'none'}]${SHOP_TOKEN ? '' : ' (NO TOKEN)'}`));
+server.listen(PORT, '127.0.0.1', () => console.log(`[new-import-viewer] http://127.0.0.1:${PORT} · store=${SHOP_STORE} · live-actions=[${[...LIVE_ACTIONS].join(',') || 'none'}]${SHOP_TOKEN ? '' : ' (NO TOKEN)'}`));
← 24a8d83 launch-queue CONSUMER: manifested batches → Shopify draft pr
·
back to New Import Viewer
·
index.html review fixes: map Shopify lowercase status (activ 7be6207 →