← back to New Import Viewer
server.js
760 lines
#!/usr/bin/env node
// new-import-viewer — the "Products New to Import" agent + web viewer.
// Surfaces vendor_catalog rows by Shopify state (NEW / PUBLISHED / UNPUBLISHED /
// ARCHIVED) so Steve can review the catalog and stage actions before a Phase-8
// push. Read-only against dw_unified EXCEPT the staged launch-queue. Live Shopify
// writes are GATED behind SHOPIFY_LIVE_WRITES=1 (default off → stage-only).
// Zero npm deps — shells out to psql (same pattern as wallco-live-viewer).
const http = require('http');
const https = require('https');
const fs = require('fs');
const path = require('path');
const { spawnSync, spawn } = require('child_process');
// catalog similarity — proxy to all.designerwallcoverings.com /api/similar.
// Credentials live server-side only (ALLDW_BASIC_AUTH env or default) — NEVER sent to the browser.
const ALLDW_AUTH = 'Basic ' + Buffer.from(process.env.ALLDW_BASIC_AUTH || 'admin:DW2024!').toString('base64');
function fetchSimilar(sku, limit) {
return new Promise((resolve) => {
const qs = new URLSearchParams({ sku });
if (limit) qs.set('limit', limit);
const r = https.request({
hostname: 'all.designerwallcoverings.com', path: '/api/similar?' + qs.toString(),
method: 'GET', timeout: 25000,
headers: { Authorization: ALLDW_AUTH, 'User-Agent': 'new-import-viewer' },
}, (resp) => { let body = ''; resp.on('data', d => body += d); resp.on('end', () => resolve({ status: resp.statusCode || 502, body })); });
r.on('timeout', () => r.destroy(new Error('timeout')));
r.on('error', () => resolve(null));
r.end();
});
}
const PORT = process.env.PORT || 9790;
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'; // unit separator — safe field delim for text columns
const RS = '\x1e'; // record separator
// ── Secrets / live-write gate ──────────────────────────────────────────────
// Token is read from this project's .env first, else Steve's secrets master.
// Never logged, never written into the repo (.env is gitignored).
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 { /* file may not exist */ }
}
return env;
}
const ENV = loadEnv();
// .env FILE is the authoritative control surface for the gate — it's read FIRST,
// before process.env. (The pm2 God daemon can carry a stale SHOPIFY_LIVE_WRITES
// in its inherited environment that children pick up; reading the file first
// makes the on-disk .env the single source of truth, which is how it's edited.)
// Basic-auth gate (internal-line-viewer spec): admin / DW2024! default,
// BASIC_AUTH=user:pass env/.env override. Local-only server, but every internal
// line viewer carries the same gate so pages behave identically (incl. the
// Safari creds-in-URL fetch quirk the location.origin rule exists for).
const BASIC_AUTH = ENV.BASIC_AUTH || process.env.BASIC_AUTH || 'admin:DW2024!';
const AUTH_HDR = 'Basic ' + Buffer.from(BASIC_AUTH).toString('base64');
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';
// ── HARD master gate ──────────────────────────────────────────────────────
// Live Shopify writes fire ONLY when SHOPIFY_LIVE_WRITES=1 in the .env file.
// Default is OFF → every action stages to the launch queue and NOTHING reaches
// the store. .env file wins over the environment so daemon-env drift can't flip it.
const LIVE = (ENV.SHOPIFY_LIVE_WRITES ?? process.env.SHOPIFY_LIVE_WRITES) === '1';
// Secondary per-action allowlist — only consulted WHEN the master gate is ON.
// .env file first, same reason. An action writes only if master ON AND allow-listed.
const LIVE_ACTIONS = new Set(((ENV.LIVE_ACTIONS ?? process.env.LIVE_ACTIONS) || '')
.split(',').map(s => s.trim()).filter(Boolean));
function isLiveAction(action) { return LIVE && !!SHOP_TOKEN && LIVE_ACTIONS.has(action); }
// Live Shopify Admin write for one product. Returns {ok,status,error}.
// unpublish → set product status=draft (pulls it off the storefront, reversible).
async function shopifyWrite(action, productId) {
if (!SHOP_TOKEN) return { ok: false, error: 'no token' };
const map = { unpublish: 'draft', publish: 'active', archive: 'archived' };
const status = map[action];
if (!status) return { ok: false, error: `live action '${action}' not supported` };
const url = `https://${SHOP_STORE}/admin/api/${SHOP_API}/products/${productId}.json`;
const r = await fetch(url, {
method: 'PUT',
headers: { 'X-Shopify-Access-Token': SHOP_TOKEN, 'content-type': 'application/json' },
body: JSON.stringify({ product: { id: Number(productId), status } }),
});
if (!r.ok) return { ok: false, error: `HTTP ${r.status} ${(await r.text()).slice(0, 160)}` };
const j = await r.json().catch(() => ({}));
return { ok: true, status: j.product?.status || status };
}
// NEW-to-import predicate — the single source of truth for "new vs Shopify".
const NEW_PRED = "(vc.sync_status='new' OR ((vc.on_shopify IS NOT TRUE) AND vc.shopify_product_id IS NULL))";
// NET-NEW = flagged-new MINUS anything whose mfr pattern number is ALREADY on
// Shopify (Steve 2026-06-09: "new - on shopify = new not on shopify"). The plain
// NEW count is inflated — rows can carry sync_status='new' while the SAME mfr_sku
// is already live on Shopify under a different/severed link. This subtracts those
// by matching the manufacturer pattern number (normalized) against shopify_products,
// so NET-NEW is the genuinely-importable set. NEVER DUPLICATE.
// 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.)
//
// THREE-SOURCE NEVER-DUPLICATE (Steve HARD RULE, /dtd-A 2026-06-10): a scraped
// mfr pattern number is a DUP — and must NOT count as net-new — if it appears in
// ANY of (1) shopify_products (160k), (2) dw_sku_registry (84k, the assigned-SKU
// ledger), (3) the vendor's own catalog table. Source (3) is vendor_catalog ITSELF
// here (this viewer reads vendor_catalog), so the rows we'd stage already live in
// it; the meaningful *additional* dedup source the old shopify-only predicate
// missed is dw_sku_registry. Measured 2026-06-10: 49,386 distinct mfr_skus were
// shopify-clean yet ALREADY in dw_sku_registry — they'd have imported as dups.
// Match normalized upper(trim()), hyphens preserved. idx_sp_mfr_upper_trim +
// idx_vc_mfr_upper_trim cover the shopify side; a functional registry index on
// upper(trim(mfr_sku)) would speed the registry NOT EXISTS (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 upper(trim(sx.mfr_sku)) = upper(trim(vc.mfr_sku)))
AND NOT EXISTS (SELECT 1 FROM dw_sku_registry rx WHERE upper(trim(rx.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
// shopify_id does the lookup (don't wrap shopify_id in a function — that defeats
// the index and seq-scans per row). NULL pid → NULL gid → no match, cheap.
const SP_JOIN = `LEFT JOIN shopify_products sp
ON sp.shopify_id = 'gid://shopify/Product/' || vc.shopify_product_id::text`;
// Derived, normalized Shopify state for a row. GROUND TRUTH wins: if a live
// shopify_products row matches, show its real status even when the pipeline
// still flags the row sync_status='new' (that mismatch is itself a useful
// dedup signal — "flagged new but actually already live as X").
const SHOP_STATUS = `CASE
WHEN upper(sp.status)='ACTIVE' THEN 'PUBLISHED'
WHEN upper(sp.status)='DRAFT' THEN 'UNPUBLISHED'
WHEN upper(sp.status)='ARCHIVED' THEN 'ARCHIVED'
WHEN upper(sp.status) LIKE 'DELETED%' THEN 'DELETED'
WHEN vc.shopify_product_id IS NOT NULL OR vc.on_shopify IS TRUE THEN 'ON_SHOPIFY'
ELSE 'NEW' END`;
// Status filters the UI can request. Keys map to a WHERE fragment.
const STATUS_FILTERS = {
netnew: NETNEW_PRED,
new: NEW_PRED,
all: 'TRUE',
onshopify: '(vc.shopify_product_id IS NOT NULL OR vc.on_shopify IS TRUE)',
published: "upper(sp.status)='ACTIVE'",
unpublished: "upper(sp.status)='DRAFT'",
archived: "upper(sp.status)='ARCHIVED'",
noimage: "(vc.image_url IS NULL OR vc.image_url='')",
};
// PGOPTIONS pins a server-side statement_timeout on EVERY query this process
// fires. Without it, one pathological query (e.g. an unindexed netnew NOT EXISTS)
// runs for tens of minutes; the warmer + every page-load pile more on top, and the
// concurrent load saturates Postgres into a death spiral (the 2026-06-11 9790
// meltdown — 13 stats queries stacked 18-43 min each). 30s ceiling = fail fast,
// never pile up. (The stats query itself now runs ~1.6s once idx_registry_mfr_upper_trim exists.)
const PG_ENV = { ...process.env, PGOPTIONS: '-c statement_timeout=30000' };
function q(sql) {
const r = spawnSync(PSQL, ['-U', PGUSER, '-d', PGDB, '-tA', '-F', US, '-R', RS, '-c', sql],
{ encoding: 'utf8', timeout: 60000, maxBuffer: 1 << 28, env: PG_ENV });
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));
}
// Non-blocking async twin of q() — spawns psql (SAME args) and resolves to the
// same parsed `rows` shape WITHOUT freezing the single-threaded event loop. Used
// by the background stats warmer so /api/stats never blocks every other request.
function qAsync(sql) {
return new Promise((resolve, reject) => {
const child = spawn(PSQL, ['-U', PGUSER, '-d', PGDB, '-tA', '-F', US, '-R', RS, '-c', sql],
{ encoding: 'utf8', env: PG_ENV });
let out = '', err = '';
child.stdout.on('data', d => (out += d));
child.stderr.on('data', d => (err += d));
child.on('error', e => reject(e));
child.on('close', code => {
if (code !== 0) return reject(new Error((err || 'psql failed').slice(0, 400)));
resolve(out.replace(/\n+$/, '').split(RS).filter(l => l.length).map(l => l.split(US)));
});
});
}
// The EXACT whole-catalog status-breakdown query /api/stats serves. Heavy (~15s
// cold over ~200k vendor_catalog rows joined to shopify_products), so it's run by
// the background warmer into STATS_CACHE rather than inline on the request path.
const STATS_SQL = `SELECT
count(*) FILTER (WHERE ${NEW_PRED}),
count(*) FILTER (WHERE vc.shopify_product_id IS NOT NULL OR vc.on_shopify IS TRUE),
count(*) FILTER (WHERE upper(sp.status)='ACTIVE'),
count(*) FILTER (WHERE upper(sp.status)='DRAFT'),
count(*) FILTER (WHERE upper(sp.status)='ARCHIVED'),
count(*),
to_char(max(vc.last_scraped_at),'YYYY-MM-DD HH24:MI'),
count(*) FILTER (WHERE vc.image_url IS NULL OR vc.image_url=''),
count(*) FILTER (WHERE ${NETNEW_PRED})
FROM vendor_catalog vc ${SP_JOIN}`;
// Map one STATS_SQL result row to the {counts, newestCrawl} shape /api/stats emits.
function rowToCounts(r) {
r = r || [];
return {
counts: {
netnew: +r[8] || 0, new: +r[0] || 0, onshopify: +r[1] || 0, published: +r[2] || 0,
unpublished: +r[3] || 0, archived: +r[4] || 0, all: +r[5] || 0,
noimage: +r[7] || 0,
// how many "new" rows are actually already on Shopify (the dup overlap)
newButOnShopify: (+r[0] || 0) - (+r[8] || 0),
},
newestCrawl: r[6] || null,
};
}
// Background stats cache + warmer. STATS_CACHE holds the last computed snapshot;
// refreshStats() recomputes it off the event-loop critical path (via qAsync) and
// keeps the prior cache on error. statsRefreshing guards against piling up
// overlapping warmers while one is in flight.
let STATS_CACHE = null;
let statsRefreshing = false;
async function refreshStats() {
if (statsRefreshing) return;
statsRefreshing = true;
try {
const rows = await qAsync(STATS_SQL);
STATS_CACHE = { ...rowToCounts(rows[0]), computedAt: Date.now() };
} catch (e) {
console.error('[new-import-viewer] stats refresh failed:', e && e.message || e);
} finally {
statsRefreshing = false;
}
}
const esc = (s) => String(s == null ? '' : s).replace(/'/g, "''");
// Whitelisted ORDER BY fragments — the ONLY sort SQL that can reach a query
// (SORTS[param] lookup; unknown keys fall back to 'newest'). Each list-view
// column header maps to an asc/desc pair here, so header-click sorting is
// server-side (correct across pagination) and injection-safe by construction.
// Every fragment ends in vc.id for a stable, deterministic page order.
const SORTS = {
newest: 'vc.last_scraped_at DESC NULLS LAST, vc.id DESC',
oldest: 'vc.last_scraped_at ASC NULLS LAST, vc.id ASC',
firstseen: 'vc.first_seen_at DESC NULLS LAST, vc.id DESC',
firstseen_asc: 'vc.first_seen_at ASC NULLS LAST, vc.id ASC',
vendor: "vc.vendor_code ASC, vc.pattern_name ASC",
vendor_desc: "vc.vendor_code DESC, vc.pattern_name DESC, vc.id DESC",
sku: "vc.mfr_sku ASC",
sku_desc: "vc.mfr_sku DESC NULLS LAST, vc.id DESC",
dwsku: "vc.dw_sku ASC NULLS LAST, vc.id ASC",
dwsku_desc:"vc.dw_sku DESC NULLS LAST, vc.id DESC",
pattern: "vc.pattern_name ASC, vc.color_name ASC",
pattern_desc: "vc.pattern_name DESC NULLS LAST, vc.color_name DESC, vc.id DESC",
color: "vc.color_name ASC, vc.pattern_name ASC",
color_desc:"vc.color_name DESC NULLS LAST, vc.pattern_name DESC, vc.id DESC",
collection:"vc.collection ASC NULLS LAST, vc.id ASC",
collection_desc:"vc.collection DESC NULLS LAST, vc.id DESC",
type: "vc.product_type ASC NULLS LAST, vc.id ASC",
type_desc: "vc.product_type DESC NULLS LAST, vc.id DESC",
sync: "vc.sync_status ASC NULLS LAST, vc.id ASC",
sync_desc: "vc.sync_status DESC NULLS LAST, vc.id DESC",
shopid: "vc.shopify_product_id ASC NULLS LAST, vc.id ASC",
shopid_desc:"vc.shopify_product_id DESC NULLS LAST, vc.id DESC",
// derived Shopify state — reuse the same CASE the SELECT emits (sp is joined
// in both /api/new-products and /api/ids, so the fragment is valid in both)
status: `${SHOP_STATUS} ASC, vc.id ASC`,
status_desc:`${SHOP_STATUS} DESC, vc.id DESC`,
// Style = first AI style tag; Color (group) = color_primary bucket then colorway
style: "vc.ai_styles->>0 ASC NULLS LAST, vc.pattern_name ASC, vc.id ASC",
style_desc:"vc.ai_styles->>0 DESC NULLS LAST, vc.pattern_name DESC, vc.id DESC",
colorgroup:"vc.color_primary ASC NULLS LAST, vc.color_name ASC, vc.id ASC",
colorgroup_desc:"vc.color_primary DESC NULLS LAST, vc.color_name DESC, vc.id DESC",
};
// Facet-field → column map (equality filters). Style is jsonb-contains, pattern
// is a substring match (type-ahead) — both handled separately in buildWhere.
const FACET_COLS = {
vendor: 'vendor_code', color: 'color_primary', colorway: 'color_name',
collection: 'collection', ptype: 'product_type', width: 'width',
match: 'match_type', sync: 'sync_status',
};
// Parse the facet-filter params off a request URL (shared by listing/ids/patterns).
function filParams(u) {
const g = k => (u.searchParams.get(k) || '').slice(0, 120);
const fil = {};
for (const k of Object.keys(FACET_COLS)) fil[k] = g(k);
fil.vendor = fil.vendor.replace(/[^a-z0-9_]/gi, '').slice(0, 64);
fil.style = g('style');
fil.pattern = g('pattern');
return fil;
}
// Build the WHERE clause shared by listing + counts.
function buildWhere({ status, vendor, term, fil }) {
const parts = [STATUS_FILTERS[status] || NEW_PRED];
if (vendor) parts.push(`vc.vendor_code='${esc(vendor)}'`);
if (fil) {
for (const [k, col] of Object.entries(FACET_COLS)) {
if (fil[k]) parts.push(`vc.${col}='${esc(fil[k])}'`);
}
if (fil.style) parts.push(`vc.ai_styles @> '${esc(JSON.stringify([fil.style]))}'::jsonb`);
if (fil.pattern) parts.push(`vc.pattern_name ILIKE '%${esc(fil.pattern.replace(/[%_]/g, ''))}%'`);
}
if (term) {
const t = `%${esc(term)}%`;
parts.push(`(vc.pattern_name ILIKE '${t}' OR vc.mfr_sku ILIKE '${t}' OR vc.color_name ILIKE '${t}' OR vc.dw_sku ILIKE '${t}' OR vc.collection ILIKE '${t}')`);
}
return 'WHERE ' + parts.join(' AND ');
}
// ── Facet tables (left panel) ──────────────────────────────────────────────
// Per-field value+count tables for the CURRENT status view (other filters are
// intentionally not applied — same convention as /api/vendors, keeps the cache
// small and the counts stable while drilling). 10-min TTL per status.
const FACET_CACHE = {};
async function computeFacets(status) {
const where = buildWhere({ status, vendor: '', term: '' });
const jobs = {};
// vendor rows also carry image coverage — the old #vbreak feature folded in
jobs.vendor = qAsync(`SELECT vc.vendor_code, count(*),
count(*) FILTER (WHERE vc.image_url IS NOT NULL AND vc.image_url<>'')
FROM vendor_catalog vc ${SP_JOIN} ${where}
GROUP BY 1 ORDER BY 2 DESC LIMIT 160`)
.then(r => r.map(x => [x[0], +x[1], +x[2]])).catch(() => []);
for (const [k, col] of Object.entries(FACET_COLS)) {
if (k === 'vendor') continue;
jobs[k] = qAsync(`SELECT vc.${col}, count(*) FROM vendor_catalog vc ${SP_JOIN} ${where}
AND vc.${col} IS NOT NULL AND vc.${col}<>'' GROUP BY 1 ORDER BY 2 DESC LIMIT 60`)
.then(r => r.map(x => [x[0], +x[1]])).catch(() => []);
}
jobs.style = qAsync(`SELECT s.v, count(*) FROM vendor_catalog vc ${SP_JOIN},
LATERAL jsonb_array_elements_text(CASE WHEN jsonb_typeof(vc.ai_styles)='array'
THEN vc.ai_styles ELSE '[]'::jsonb END) s(v)
${where} GROUP BY 1 ORDER BY 2 DESC LIMIT 60`)
.then(r => r.map(x => [x[0], +x[1]])).catch(() => []);
const keys = Object.keys(jobs);
const vals = await Promise.all(keys.map(k => jobs[k]));
const data = {};
keys.forEach((k, i) => (data[k] = vals[i]));
return data;
}
function send(res, code, body, type) {
res.writeHead(code, { 'content-type': type || 'application/json', 'cache-control': 'no-store' });
res.end(body);
}
// Stage an action to the append-only launch queue. NEVER writes to the store.
// Each batch persists a FULL manifest (ids + skus + from-state) to
// data/launch-batches/<batchId>.json — queue lines before 2026-06-10 carried
// only counts, which made them un-actionable (a consumer had nothing to act
// on). Manifested batches are the authoritative, image-gated staged set.
// HARD NEVER-DUPLICATE GATE for staging (Steve HARD RULE, /dtd-A 2026-06-10):
// the /api/action launch path stages by raw row IDs — a client (UI bug, stale
// id list, or raw curl) could pass IDs that are NOT net-new (already on Shopify
// or already in dw_sku_registry). The NETNEW_PRED filter only governs what the UI
// *lists*; it does NOT bind what gets staged. So re-validate every candidate row
// here against the two external dedup sources (shopify_products + dw_sku_registry)
// on normalized upper(trim(mfr_sku)) and DROP any row whose mfr_sku already exists.
// Fail-safe: if the dedup query throws, we BLOCK the whole batch (let the error
// propagate) rather than stage an unverified set. Returns {keep:Set<id>, blocked}.
function dedupFilterIds(ids) {
const idList = ids.join(',');
// Candidate (id, normalized mfr_sku) for the requested rows.
const cand = q(`SELECT vc.id, upper(trim(vc.mfr_sku)) AS k
FROM vendor_catalog vc
WHERE vc.id IN (${idList})
AND vc.mfr_sku IS NOT NULL AND vc.mfr_sku <> ''`);
// Rows with a non-empty mfr_sku that ALSO exist on Shopify or in the registry.
// Set-based (DISTINCT CTE + IN) so it stays index/hash-join fast at 5k ids.
const dup = q(`WITH cand AS (
SELECT vc.id, upper(trim(vc.mfr_sku)) AS k FROM vendor_catalog vc
WHERE vc.id IN (${idList}) AND vc.mfr_sku IS NOT NULL AND vc.mfr_sku <> ''),
shop AS (SELECT DISTINCT upper(trim(mfr_sku)) k FROM shopify_products WHERE mfr_sku IS NOT NULL),
reg AS (SELECT DISTINCT upper(trim(mfr_sku)) k FROM dw_sku_registry WHERE mfr_sku IS NOT NULL)
SELECT c.id FROM cand c
WHERE c.k IN (SELECT k FROM shop) OR c.k IN (SELECT k FROM reg)`);
const dupIds = new Set(dup.map(r => +r[0]));
// Rows missing an mfr_sku entirely can't be dedup-verified → BLOCK them too
// (never stage an un-dedupable row; fix the scrape, don't import around it).
const haveSku = new Set(cand.map(r => +r[0]));
const keep = ids.filter(id => haveSku.has(id) && !dupIds.has(id));
const blocked = ids.length - keep.length;
return { keep, blocked, dupBlocked: dupIds.size, noSkuBlocked: ids.length - haveSku.size };
}
function stageAction({ ids, action, mode }) {
// NEVER-DUPLICATE: drop any candidate already on Shopify / in the registry, or
// lacking a dedup-able mfr_sku, BEFORE building the staged manifest. Fail-safe —
// a throw here aborts the whole stage rather than staging an unverified batch.
const ded = dedupFilterIds(ids);
ids = ded.keep;
if (!ids.length) {
return { ok: false, error: `all ${ded.blocked} selected row(s) BLOCKED by NEVER-DUPLICATE gate (already on Shopify / in dw_sku_registry, or missing mfr_sku)`, staged: 0, action, mode, dupBlocked: ded.dupBlocked, noSkuBlocked: ded.noSkuBlocked, live: false };
}
const idList = ids.join(',');
const rows = q(`SELECT vc.id, vc.vendor_code, vc.mfr_sku, vc.dw_sku, vc.pattern_name, ${SHOP_STATUS} AS st
FROM vendor_catalog vc ${SP_JOIN}
WHERE vc.id IN (${idList})`);
const batchId = action + '-' + rows.length + '-' + (ids[0] || 0) + '-' + Date.now().toString(36);
const skus = rows.map(r => ({ id: +r[0], vendor: r[1], sku: r[2], dwSku: r[3], pattern: r[4], from: r[5] }));
const byVendor = {};
rows.forEach(r => (byVendor[r[1]] = (byVendor[r[1]] || 0) + 1));
const bdir = path.join(__dirname, 'data', 'launch-batches');
fs.mkdirSync(bdir, { recursive: true });
const manifest = path.join(bdir, batchId + '.json');
fs.writeFileSync(manifest, JSON.stringify({ batchId, action, mode, at: new Date().toISOString(), count: rows.length, skus }));
const qf = path.join(__dirname, 'data', 'launch-queue.jsonl');
fs.appendFileSync(qf, JSON.stringify({ batchId, action, mode, count: rows.length, skus: rows.length, manifest: 'data/launch-batches/' + batchId + '.json', at: new Date().toISOString() }) + '\n');
const out = { ok: true, staged: rows.length, action, mode, batchId, byVendor, live: false };
if (ded.blocked) {
out.dedupBlocked = ded.blocked;
out.dupBlocked = ded.dupBlocked;
out.noSkuBlocked = ded.noSkuBlocked;
out.dedupNote = `${ded.blocked} row(s) BLOCKED by NEVER-DUPLICATE gate (${ded.dupBlocked} already on Shopify/registry, ${ded.noSkuBlocked} missing mfr_sku)`;
}
return out;
}
// ── Local thumbnail proxy + disk cache ─────────────────────────────────────
// Grid images were direct hotlinks to 100+ vendor hosts — one referrer block,
// content-blocker, Safari tracking-prevention, or vendor change away from a
// blank grid. /img?u=<vendor URL> fetches server-side once and caches to disk
// (data/imgcache/<sha1>), so the browser only ever talks to 127.0.0.1 and each
// remote image is fetched a single time. Lazy-loading means only viewed images
// are cached. SSRF-guarded: http(s) only, no localhost / raw-IP / .local hosts.
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' })); }
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 {
const body = fs.readFileSync(f);
let ct = 'image/jpeg'; try { ct = fs.readFileSync(fct, 'utf8'); } catch {}
res.writeHead(200, { 'content-type': ct, 'cache-control': 'public, max-age=604800', 'x-img-cache': 'hit' });
return res.end(body);
} catch { /* cache miss → fetch */ }
try {
// 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 }));
const buf = Buffer.from(await r.arrayBuffer());
if (buf.length > (12 << 20)) return send(res, 502, JSON.stringify({ error: 'too large' }));
try { fs.mkdirSync(IMG_CACHE, { recursive: true }); fs.writeFileSync(f, buf); fs.writeFileSync(fct, ct); } catch {}
res.writeHead(200, { 'content-type': ct, 'cache-control': 'public, max-age=604800', 'x-img-cache': 'miss' });
return res.end(buf);
} catch (e) { return send(res, 502, JSON.stringify({ error: 'fetch failed: ' + (e && e.message || e) })); }
}
const server = http.createServer((req, res) => {
try {
const u = new URL(req.url, 'http://x');
// basic-auth gate on everything (internal-line-viewer spec)
if ((req.headers.authorization || '') !== AUTH_HDR) {
res.writeHead(401, { 'WWW-Authenticate': 'Basic realm="new-import-viewer"', 'content-type': 'text/plain' });
return res.end('auth required');
}
if (u.pathname === '/' || u.pathname === '/index.html') {
return send(res, 200, fs.readFileSync(path.join(__dirname, 'public', 'index.html')), 'text/html; charset=utf-8');
}
if (u.pathname === '/img') { serveImg(u.searchParams.get('u') || '', res); return; }
if (u.pathname === '/api/similar') {
const sku = u.searchParams.get('sku');
if (!sku) { res.writeHead(400, { 'content-type': 'application/json' }); return res.end('{"error":"no sku"}'); }
fetchSimilar(sku, u.searchParams.get('limit')).then((r) => {
if (!r) { res.writeHead(502, { 'content-type': 'application/json' }); return res.end('{"error":"catalog similarity unavailable"}'); }
res.writeHead(r.status, { 'content-type': 'application/json' }); res.end(r.body);
});
return;
}
if (u.pathname === '/api/stats') {
// Whole-catalog status breakdown for the header pills (vendor/term ignored here).
// Served from STATS_CACHE so this endpoint NEVER blocks the single-threaded
// server on the ~15s cold whole-catalog FILTER-count. First-ever hit (null
// cache) does ONE synchronous compute to return real numbers; thereafter a
// stale (>45s) cache triggers a non-blocking background refresh and we serve
// the prior snapshot instantly.
const armable = LIVE && !!SHOP_TOKEN && LIVE_ACTIONS.size > 0;
if (!STATS_CACHE) {
const r = q(STATS_SQL)[0] || [];
STATS_CACHE = { ...rowToCounts(r), computedAt: Date.now() };
} else if (Date.now() - STATS_CACHE.computedAt > 45000) {
refreshStats(); // non-blocking — serve the current snapshot now, refresh in background
}
return send(res, 200, JSON.stringify({
// `armable` = the .env ceiling permits live writes at all. The actual arm
// is a per-session UI toggle (defaults OFF each load) the client tracks and
// sends as arm:true per action. `live` kept for back-compat = armable.
live: armable,
armable,
liveActions: armable ? [...LIVE_ACTIONS] : [],
store: SHOP_STORE,
counts: STATS_CACHE.counts,
newestCrawl: STATS_CACHE.newestCrawl,
cachedAt: STATS_CACHE.computedAt,
}));
}
if (u.pathname === '/api/facets') {
const status = STATUS_FILTERS[u.searchParams.get('status')] ? u.searchParams.get('status') : 'new';
const cached = FACET_CACHE[status];
if (cached && Date.now() - cached.at < 600000) return send(res, 200, JSON.stringify(cached.data));
(async () => {
try {
const data = await computeFacets(status);
FACET_CACHE[status] = { data, at: Date.now() };
send(res, 200, JSON.stringify(data));
} catch (e) { send(res, 500, JSON.stringify({ error: e.message })); }
})();
return;
}
// Pattern type-ahead — server-backed (126k distinct patterns is too many for
// a static datalist); returns up to 200 matches for the current filter.
if (u.pathname === '/api/patterns') {
const status = STATUS_FILTERS[u.searchParams.get('status')] ? u.searchParams.get('status') : 'new';
const fil = filParams(u); fil.pattern = '';
const qs = (u.searchParams.get('q') || '').slice(0, 80).replace(/[%_]/g, '');
const where = buildWhere({ status, vendor: '', term: '', fil });
const extra = qs ? ` AND vc.pattern_name ILIKE '%${esc(qs)}%'` : '';
(async () => {
try {
const rows = await qAsync(`SELECT DISTINCT vc.pattern_name FROM vendor_catalog vc ${SP_JOIN} ${where}${extra}
AND vc.pattern_name IS NOT NULL AND vc.pattern_name<>'' ORDER BY 1 LIMIT 200`);
send(res, 200, JSON.stringify({ patterns: rows.map(r => r[0]) }));
} catch (e) { send(res, 500, JSON.stringify({ error: e.message })); }
})();
return;
}
if (u.pathname === '/api/vendors') {
const status = STATUS_FILTERS[u.searchParams.get('status')] ? u.searchParams.get('status') : 'new';
const where = buildWhere({ status, vendor: '', term: '' });
// per-vendor count + image coverage (withImg / noImg) for the selected status
const rows = q(`SELECT vc.vendor_code, count(*),
count(*) FILTER (WHERE vc.image_url IS NOT NULL AND vc.image_url<>''),
count(*) FILTER (WHERE vc.image_url IS NULL OR vc.image_url=''),
to_char(max(vc.last_scraped_at),'YYYY-MM-DD HH24:MI')
FROM vendor_catalog vc ${SP_JOIN} ${where}
GROUP BY vc.vendor_code ORDER BY count(*) DESC`);
return send(res, 200, JSON.stringify(rows.map(r => ({
vendor: r[0], count: +r[1], withImg: +r[2], noImg: +r[3], crawled: r[4],
}))));
}
if (u.pathname === '/api/new-products') {
const status = STATUS_FILTERS[u.searchParams.get('status')] ? u.searchParams.get('status') : 'new';
const fil = filParams(u);
const sort = SORTS[u.searchParams.get('sort')] ? u.searchParams.get('sort') : 'newest';
const term = (u.searchParams.get('q') || '').slice(0, 80);
const limit = Math.min(parseInt(u.searchParams.get('limit'), 10) || 120, 500);
const offset = Math.max(parseInt(u.searchParams.get('offset'), 10) || 0, 0);
const where = buildWhere({ status, vendor: '', term, fil });
// image = vendor image, else the Shopify featured image, else the first gallery image
// (fills the "newly imported, no vendor photo yet, but Shopify has it" case). gallery =
// the full Shopify image set for this product (dw_unified.product_images), '|~|'-joined.
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,
COALESCE(NULLIF(vc.image_url,''), sp.image_url,
(SELECT pi.image_url FROM product_images pi WHERE pi.product_id = vc.shopify_product_id::text ORDER BY pi.position LIMIT 1)) AS image,
vc.product_url,
to_char(vc.first_seen_at,'YYYY-MM-DD HH24:MI') AS first_seen,
to_char(vc.last_scraped_at,'YYYY-MM-DD HH24:MI') AS crawled,
vc.sync_status, ${SHOP_STATUS} AS shop_status,
vc.shopify_product_id, sp.handle,
to_char(vc.shopify_synced_at,'YYYY-MM-DD HH24:MI') AS shop_synced,
vc.color_primary, vc.width, vc.match_type, vc.ai_styles->>0 AS style1,
(SELECT string_agg(pi.image_url, '|~|' ORDER BY pi.position) FROM product_images pi WHERE pi.product_id = vc.shopify_product_id::text) AS gallery
FROM vendor_catalog vc ${SP_JOIN} ${where}
ORDER BY ${SORTS[sort]} LIMIT ${limit} OFFSET ${offset}`);
const cnt = +(q(`SELECT count(*) FROM vendor_catalog vc ${SP_JOIN} ${where}`)[0] || [0])[0];
const items = rows.map(r => ({
id: +r[0], vendor: r[1], sku: r[2], dwSku: r[3], pattern: r[4], color: r[5],
collection: r[6], type: r[7], image: r[8], url: r[9],
firstSeen: r[10], crawled: r[11], syncStatus: r[12], shopStatus: r[13],
shopifyId: r[14] || null, handle: r[15] || null, shopSynced: r[16] || null,
colorGroup: r[17] || null, width: r[18] || null, match: r[19] || null, style: r[20] || null,
images: r[21] ? r[21].split('|~|').filter(Boolean) : [],
}));
return send(res, 200, JSON.stringify({ count: cnt, offset, limit, items }));
}
// Bare id list for the current filter — powers "select first N / ALL matching"
// without rendering the rows. Same WHERE/ORDER as the listing so "first N"
// means the first N the user would see. ids only (ints), so even the full
// 197k catalog is a small payload. limit=0/absent → all (hard cap 250k).
if (u.pathname === '/api/ids') {
const status = STATUS_FILTERS[u.searchParams.get('status')] ? u.searchParams.get('status') : 'new';
const fil = filParams(u);
const sort = SORTS[u.searchParams.get('sort')] ? u.searchParams.get('sort') : 'newest';
const term = (u.searchParams.get('q') || '').slice(0, 80);
const limit = Math.min(Math.max(parseInt(u.searchParams.get('limit'), 10) || 0, 0), 250000);
const where = buildWhere({ status, vendor: '', term, fil });
const rows = q(`SELECT vc.id FROM vendor_catalog vc ${SP_JOIN} ${where}
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.
// Actions in LIVE_ACTIONS additionally fire a real Shopify write.
// /api/launch kept as a back-compat alias (action='launch').
if ((u.pathname === '/api/action' || u.pathname === '/api/launch') && req.method === 'POST') {
let body = '';
req.on('data', c => (body += c));
req.on('end', async () => {
try {
const d = JSON.parse(body || '{}');
let ids = Array.isArray(d.ids) ? d.ids.filter(n => Number.isInteger(n)).slice(0, 5000) : [];
const ACTIONS = ['launch', 'publish', 'unpublish', 'archive'];
const action = u.pathname === '/api/launch' ? 'launch'
: (ACTIONS.includes(d.action) ? d.action : null);
if (!action) return send(res, 400, JSON.stringify({ error: 'unknown action' }));
const mode = d.mode === 'active' ? 'active' : 'draft'; // only meaningful for launch
if (!ids.length) return send(res, 400, JSON.stringify({ error: 'no products selected' }));
// HARD GATE (Steve 2026-06-10): ALL PRODUCTS MUST HAVE IMAGES.
// A launch may never stage a row without an image — filter server-side
// so neither single-card nor bulk (nor a raw curl) can sneak one in.
let noImageBlocked = 0;
if (action === 'launch') {
const withImg = new Set(q(`SELECT vc.id FROM vendor_catalog vc
WHERE vc.id IN (${ids.join(',')})
AND vc.image_url IS NOT NULL AND vc.image_url <> ''`).map(r => +r[0]));
noImageBlocked = ids.length - withImg.size;
ids = ids.filter(id => withImg.has(id));
if (!ids.length) return send(res, 400, JSON.stringify({
error: `all ${noImageBlocked} selected row(s) lack images — ALL PRODUCTS MUST HAVE IMAGES; fix the scrape, don't import around it`, noImageBlocked }));
}
// NEVER-DUPLICATE end-to-end (Steve HARD RULE, /dtd-A 2026-06-10): drop
// any selected row already on Shopify / in dw_sku_registry (or missing a
// dedup-able mfr_sku) BEFORE either staging OR a live write — so a stale
// id list / raw curl can't push a known dup down either path. Fail-safe:
// a query error throws and the outer catch returns 500 (nothing staged).
const ded = dedupFilterIds(ids);
const dedupBlocked = ids.length - ded.keep.length;
ids = ded.keep;
if (!ids.length) return send(res, 400, JSON.stringify({
error: `all ${dedupBlocked} selected row(s) BLOCKED by NEVER-DUPLICATE gate (already on Shopify / in dw_sku_registry, or missing mfr_sku)`,
dedupBlocked, dupBlocked: ded.dupBlocked, noSkuBlocked: ded.noSkuBlocked }));
const out = stageAction({ ids, action, mode });
if (!out.ok) return send(res, 400, JSON.stringify(out));
if (noImageBlocked) {
out.noImageBlocked = noImageBlocked;
out.note = `${noImageBlocked} no-image row(s) BLOCKED from launch (all products must have images)`;
}
if (dedupBlocked) {
out.dedupBlocked = dedupBlocked;
out.dupBlocked = ded.dupBlocked;
out.noSkuBlocked = ded.noSkuBlocked;
out.dedupNote = `${dedupBlocked} row(s) BLOCKED by NEVER-DUPLICATE gate before ${action} (${ded.dupBlocked} on Shopify/registry, ${ded.noSkuBlocked} missing mfr_sku)`;
}
// SAFETY: bulk requests (stageOnly) NEVER fire live writes — live store
// changes only ever happen one card at a time. The fat-finger blast
// radius of a bulk live write isn't worth it. Bulk always stages.
const stageOnly = d.stageOnly === true || ids.length > 1;
// SAFETY: the per-session UI arm toggle defaults OFF every page load.
// A live write needs BOTH the .env ceiling (isLiveAction) AND the client
// to be session-armed (d.arm===true). .env stays the hard ceiling — the
// toggle can never arm beyond what SHOPIFY_LIVE_WRITES/LIVE_ACTIONS allow.
const sessionArmed = d.arm === true;
if (isLiveAction(action) && !stageOnly && sessionArmed) {
// LIVE PATH — only reachable when SHOPIFY_LIVE_WRITES=1 AND the action
// is in LIVE_ACTIONS AND a token is present AND it's a single-card act.
// Fetch the row's identity + the LINKED product's identity so we can
// verify they're the same physical product before any live write.
const rows = q(`SELECT vc.id, vc.shopify_product_id, vc.mfr_sku, sp.mfr_sku, vc.pattern_name, sp.title
FROM vendor_catalog vc
LEFT JOIN shopify_products sp ON sp.shopify_id='gid://shopify/Product/'||vc.shopify_product_id::text
WHERE vc.id IN (${ids.join(',')})`);
const results = []; let ok = 0, skipped = 0, failed = 0, blocked = 0;
// ≥1 significant (len≥4, non-noise) pattern token must appear in the title.
const NOISE = new Set(['wallcovering','wallcoverings','wallpaper','mural','grasscloth','vinyl','sample','collection','metallic','embossed','textile','fabric']);
const titleConfirms = (pat, title) => {
const toks = (pat || '').toLowerCase().replace(/[^a-z0-9 ]/g, ' ').split(/\s+/).filter(t => t.length >= 4 && !NOISE.has(t));
const t = (title || '').toLowerCase();
return toks.length > 0 && toks.some(tok => t.includes(tok));
};
for (const r of rows) {
const vcId = +r[0], pid = r[1], vcMfr = (r[2] || '').trim(), prodMfr = (r[3] || '').trim(), pat = r[4], title = r[5];
if (!pid) { skipped++; results.push({ id: vcId, skipped: 'not on Shopify' }); continue; }
// CROSS-LINK GUARD: a live write only fires when BOTH identity signals
// agree the linked product is the same one — mfr_sku matches AND the
// pattern actually appears in the linked title. Either disagreeing →
// refuse (stage for manual review). Catches the Swan-River bug (mfr
// corrupted-matches but title is a different pattern) AND fails safe on
// the multi-SKU-scheme cases. SKU data is unreliable, so we require
// positive confirmation, not absence-of-proof.
const mfrAgrees = !!vcMfr && !!prodMfr && vcMfr === prodMfr;
const titleAgrees = titleConfirms(pat, title);
if (!(mfrAgrees && titleAgrees)) {
blocked++;
results.push({ id: vcId, blocked: `unverified link — row "${pat}" (${vcMfr}) vs linked product "${(title||'').slice(0,40)}" (${prodMfr}); mfr_match=${mfrAgrees} title_match=${titleAgrees}. Refusing live ${action} (would risk wrong product). Verify + relink, then retry.` });
continue;
}
const w = await shopifyWrite(action, pid);
if (w.ok) {
ok++;
// reflect new state locally so the chip updates immediately
try { q(`UPDATE shopify_products SET status='${esc(w.status)}'
WHERE shopify_id='gid://shopify/Product/${esc(String(pid))}'`); } catch {}
results.push({ id: vcId, ok: true, status: w.status });
} else { failed++; results.push({ id: vcId, error: w.error }); }
}
out.live = true;
out.applied = { ok, skipped, failed, blocked };
out.results = results;
out.store = SHOP_STORE;
out.note = `LIVE ${action} on ${SHOP_STORE}: ${ok} ok, ${skipped} skipped (not on Shopify), ${blocked} BLOCKED (mfr_sku cross-link), ${failed} failed`;
} else if (isLiveAction(action) && stageOnly) {
out.note = `staged only — bulk/${ids.length}-item ${action} never fires live (single-card only)`;
} else if (isLiveAction(action) && !sessionArmed) {
out.note = `staged only — session not armed (flip the Live-writes toggle to arm '${action}')`;
} else {
out.note = !LIVE
? 'staged only — live Shopify writes are OFF (SHOPIFY_LIVE_WRITES!=1)'
: `staged only — '${action}' is not in the live-write allowlist (${[...LIVE_ACTIONS].join(',')||'none'})`;
}
return send(res, 200, JSON.stringify(out));
} catch (e) { return send(res, 500, JSON.stringify({ error: e.message })); }
});
return;
}
return send(res, 404, JSON.stringify({ error: 'not found' }));
} catch (e) {
return send(res, 500, JSON.stringify({ error: e.message }));
}
});
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)'}`);
refreshStats(); // warm the stats cache on boot so the first /api/stats hit is fast
// warm the default-view facet tables too (netnew group-bys are the slowest)
computeFacets('netnew').then(d => (FACET_CACHE.netnew = { data: d, at: Date.now() })).catch(() => {});
});