← back to Dw Staged Active Viewer
server.js
431 lines
#!/usr/bin/env node
/**
* DW Staged-for-Active Viewer (READ-ONLY)
* ----------------------------------------
* Serves a local admin product grid listing EVERY SKU staged to go ACTIVE via the
* metered cadence (com.steve.dw-cadence-hourly). "Staged for active" is defined by
* reusing the cadence's OWN gate logic so counts match `cadence-import.js --gate`:
*
* vendor in vendors.js AND vendor_registry.discount_confirmed = true
* AND vendor NOT in cadence DENY_VENDORS
* per SKU: shopify_product_id empty AND (costExpr)::numeric > 0 AND dedup_skip IS NOT TRUE
*
* Pricing per DW rules: retail = round(cost/0.65/0.85, 2) unless the vendor cfg has a
* sellExpr (MAP — Kravet family), in which case sell = sellExpr. Sample = $4.25.
*
* HARD read-only: this process only ever runs SELECT. No INSERT/UPDATE/DELETE, no Shopify.
*/
const http = require('http');
const path = require('path');
const fs = require('fs');
const { execFileSync } = require('child_process');
const PORT = process.env.PORT || 9846;
const DB = 'dw_unified';
// SNAPSHOT_ONLY=1 → serve from data/staged-snapshot.json instead of querying psql.
// This is the mode the Kamatera-hosted copy runs in: the snapshot is generated on
// Mac2 (where the canonical dedup_skip flags live) and rsynced up hourly. The SAME
// server.js runs in both places; only this flag differs.
const SNAPSHOT_ONLY = process.env.SNAPSHOT_ONLY === '1';
const SNAPSHOT_FILE = path.join(__dirname, 'data', 'staged-snapshot.json');
let _snap = null, _snapMtime = 0;
function loadSnapshot() {
// Reload if the file changed on disk (hourly rsync overwrites it).
let st;
try { st = fs.statSync(SNAPSHOT_FILE); } catch { throw new Error('snapshot file missing: ' + SNAPSHOT_FILE); }
if (!_snap || st.mtimeMs !== _snapMtime) {
_snap = JSON.parse(fs.readFileSync(SNAPSHOT_FILE, 'utf8'));
_snapMtime = st.mtimeMs;
}
return _snap;
}
// Reuse the cadence's verified vendor config + denylist verbatim so the staged set
// is identical to what the cadence will actually create.
// LAZY/GUARDED: in SNAPSHOT_ONLY mode (Kamatera) this Mac2-only file is absent and never
// needed — the snapshot already holds everything. Only the live-DB code paths touch VENDORS.
let VENDORS = {};
if (!SNAPSHOT_ONLY) {
VENDORS = require(path.join(
process.env.HOME,
'Projects/Designer-Wallcoverings/shopify/scripts/cadence/vendors.js'
));
}
// DENY_VENDORS mirrors cadence-import.js (kept in sync; see that file's comment block).
const DENY_VENDORS = new Set([
'Nicolette Mayer','Phillip Jeffries','Et Cie','Spoonflower','Colefax & Fowler','Carlisle',
'Paul Montgomery','Cowtan & Tout','Rebel Walls','Koroseal','Newmor','Wolf Gordon',
'Scalamandre','Desima','York Wallcoverings',
]);
const RETAIL = c => Math.round((c / 0.65 / 0.85) * 100) / 100;
const SAMPLE_PRICE = '4.25';
// ---- cadence rotation constants (mirror run-cadence-hourly.sh + cadence-import.js) ----
// run-cadence-hourly.sh: REQ=36 variants → MAXP=18 products/slot; VPS=6 vendors;
// PER=ceil(MAXP/VPS)=3 skus/vendor; fires :40 every hour → 24 slots/day.
// cadence-import.js rotate(): start=cur.idx%len; picks=min(VPS, len-1, len);
// picked = ready[(start+i)%len] for i<picks; each vendor takes up to PER (capped at MAXP total);
// cursor advances by picked.length mod len.
const CAD = {
MAXP: 18, // products per slot (budget MAXP, 36 variants / 2)
VPS: 6, // vendors led per slot
PER: 3, // skus per vendor per slot (ceil(MAXP/VPS))
SLOTS_PER_DAY: 24,
FIRE_MINUTE: 40, // fires :40 each hour
};
const CURSOR_FILE = path.join(
process.env.HOME,
'Projects/Designer-Wallcoverings/shopify/scripts/cadence/data/cadence-cursor.json'
);
function loadCadenceCursor() {
try { return JSON.parse(fs.readFileSync(CURSOR_FILE, 'utf8')); } catch { return { idx: 0 }; }
}
// ---- read-only psql (same invocation the cadence uses; -d dw_unified, no DSN) ----
function psql(sql) {
return execFileSync('psql', ['-At', '-F', '\t', '-d', DB, '-c', sql],
{ encoding: 'utf8', maxBuffer: 1 << 28 }).trim();
}
// ---- catalog-column detection (cached) ----
const _colCache = {};
function hasCol(table, col) {
const key = `${table}.${col}`;
if (key in _colCache) return _colCache[key];
const t = String(table).replace(/[^a-z0-9_]/gi, '');
const c = String(col).replace(/[^a-z0-9_]/gi, '');
let has = false;
try { has = (psql(`SELECT 1 FROM information_schema.columns WHERE table_name='${t}' AND column_name='${c}' LIMIT 1;`) || '').trim() === '1'; } catch { has = false; }
return (_colCache[key] = has);
}
function dedupSkipClause(table) {
return hasCol(table, 'dedup_skip') ? ' AND dedup_skip IS NOT TRUE' : '';
}
// Per-table "staged date" = first available of created_at / imported_at / updated_at / last_scraped.
function stagedDateExpr(table) {
const cands = ['created_at', 'imported_at', 'updated_at', 'last_scraped'];
const present = cands.filter(c => hasCol(table, c));
if (!present.length) return 'NULL::timestamp';
return 'COALESCE(' + present.map(c => `${c}::timestamp`).join(', ') + ')';
}
let _confirmed = null;
function confirmedDiscountVendors() {
if (_confirmed) return _confirmed;
const out = psql(`SELECT vendor_name FROM vendor_registry WHERE discount_confirmed IS TRUE;`);
return (_confirmed = new Set(out ? out.split('\n') : []));
}
// Vendors that are actually READY (config + confirmed discount + not denied).
function readyVendors() {
const confirmed = confirmedDiscountVendors();
const out = [];
for (const [vendor, cfg] of Object.entries(VENDORS)) {
if (DENY_VENDORS.has(vendor)) continue;
if (!confirmed.has(vendor)) continue;
out.push({ vendor, cfg });
}
return out;
}
const WHERE = cfg => `coalesce(shopify_product_id::text,'')='' AND (${cfg.costExpr})::numeric > 0${dedupSkipClause(cfg.table)}`;
// ---- per-vendor counts (matches `--gate`) ----
function vendorCounts() {
const rows = [];
for (const { vendor, cfg } of readyVendors()) {
let n = 0;
try { n = parseInt(psql(`SELECT count(*) FROM ${cfg.table} WHERE ${WHERE(cfg)};`) || '0', 10); } catch (e) { n = 0; }
rows.push({ vendor, table: cfg.table, count: n });
}
rows.sort((a, b) => b.count - a.count);
return rows;
}
// Newwall true sub-lines (e.g. Tres Tintas) when the catalog carries vendor_name.
function subLines(vendor, cfg) {
if (!hasCol(cfg.table, 'vendor_name')) return null;
try {
const out = psql(`SELECT vendor_name, count(*) FROM ${cfg.table} WHERE ${WHERE(cfg)} GROUP BY vendor_name ORDER BY 2 DESC;`);
if (!out) return null;
return out.split('\n').map(l => { const [name, c] = l.split('\t'); return { name: name || vendor, count: parseInt(c, 10) }; });
} catch { return null; }
}
// ---- paged SKU rows for one vendor ----
function vendorRows(vendor, cfg, limit, offset) {
const t = cfg.table;
const lineSel = hasCol(t, 'vendor_name') ? 'vendor_name' : `'${vendor.replace(/'/g, "''")}'`;
const ov = cfg.colOverrides || {};
const col = name => ov[name] ? ov[name] : (hasCol(t, name) ? name : `NULL`);
const dateExpr = stagedDateExpr(t);
const sellSel = cfg.sellExpr ? `(${cfg.sellExpr})::numeric` : 'NULL::numeric';
const sql = `SELECT
dw_sku,
${lineSel} AS line,
${col('pattern_name')} AS pattern_name,
${col('color_name')} AS color_name,
${col('collection')} AS collection,
${col('image_url')} AS image_url,
(${cfg.costExpr})::numeric AS cost,
${sellSel} AS sell,
${dateExpr} AS staged_at
FROM ${t}
WHERE ${WHERE(cfg)}
ORDER BY ${dateExpr} DESC NULLS LAST, dw_sku
LIMIT ${parseInt(limit, 10)} OFFSET ${parseInt(offset, 10)};`;
const out = psql(sql);
if (!out) return [];
return out.split('\n').map(line => {
const [dw_sku, lineName, pattern_name, color_name, collection, image_url, cost, sell, staged_at] = line.split('\t');
const c = parseFloat(cost) || 0;
const sellNum = sell && sell !== '' ? parseFloat(sell) : null;
const retail = sellNum && sellNum > 0 ? Math.round(sellNum * 100) / 100 : RETAIL(c);
return {
dw_sku,
vendor,
line: lineName && lineName !== '' ? lineName : vendor,
pattern_name: pattern_name === '' ? null : pattern_name,
color_name: color_name === '' ? null : color_name,
collection: collection === '' ? null : collection,
image_url: image_url === '' ? null : image_url,
cost: c ? Math.round(c * 100) / 100 : null,
retail,
sample_price: SAMPLE_PRICE,
priced_at_map: !!(sellNum && sellNum > 0),
staged_at: staged_at && staged_at !== '' ? staged_at : null,
};
});
}
// ---- CADENCE PROJECTION (mirrors run-cadence-hourly.sh + cadence-import.js rotate) ----
// Simulate forward, one slot per hour, 24 slots/day. Each slot: lead = ready[idx%len], pick up to
// VPS vendors round-robin from there, each contributes up to PER skus capped at MAXP total; decrement
// each vendor's remaining staged count; a vendor at 0 drops out of the ready list; cursor advances by
// picked.length % len. Continue until the whole staged backlog drains. Returns the per-slot schedule
// plus its day/week/month rollups. PROJECTION ONLY — assumes a steady 18/hr once the variant cap clears.
function buildProjection(opts) {
const startEpochMs = opts.startEpochMs; // ms of the FIRST projected slot (next :40 from now)
const blockedUntilMs = opts.blockedUntilMs; // slots whose fire-time is < this create 0 (cap blocking)
const maxSlots = opts.maxSlots || 24 * 200; // hard cap so the sim always terminates
// Build the working READY list with per-vendor remaining + sub-lines, in rotation order.
let ready = readyVendors().map(({ vendor, cfg }) => {
let n = 0;
try { n = parseInt(psql(`SELECT count(*) FROM ${cfg.table} WHERE ${WHERE(cfg)};`) || '0', 10); } catch { n = 0; }
return { vendor, cfg, remaining: n, soldBy: cfg.soldBy, mapPriced: !!cfg.sellExpr,
subLines: subLines(vendor, cfg) };
}).filter(v => v.remaining > 0);
const grandTotal = ready.reduce((s, v) => s + v.remaining, 0);
let idx = (loadCadenceCursor().idx || 0);
const slots = [];
let drainedAtSlot = null;
const HOUR = 3600 * 1000;
for (let s = 0; s < maxSlots; s++) {
if (!ready.length) { drainedAtSlot = s; break; }
const fireMs = startEpochMs + s * HOUR;
const blocked = fireMs < blockedUntilMs;
const len = ready.length;
const start = ((idx % len) + len) % len;
const picks = Math.max(1, Math.min(CAD.VPS, Math.max(1, len - 1), len));
// assign skus this slot, capped at MAXP total, PER per vendor, in rotation order from `start`
const assigned = []; // { vendor, count, soldBy, mapPriced, subLines }
let budget = CAD.MAXP;
for (let i = 0; i < picks && budget > 0; i++) {
const v = ready[(start + i) % len];
if (v.remaining <= 0) continue;
const take = blocked ? 0 : Math.min(CAD.PER, v.remaining, budget);
assigned.push({ vendor: v.vendor, count: take, soldBy: v.soldBy, mapPriced: v.mapPriced,
subLines: v.subLines });
if (!blocked) { v.remaining -= take; budget -= take; }
}
const created = assigned.reduce((a, b) => a + b.count, 0);
slots.push({ slotIndex: s, fireMs, blocked, leadVendor: ready[start].vendor,
vendors: assigned.filter(a => a.count > 0 || blocked), created });
// advance cursor exactly as cadence does (by vendors LED, regardless of how many took skus)
idx = (start + picks) % len;
// drop drained vendors AFTER advancing; rebuild ready preserving rotation order
if (!blocked) {
const before = ready.length;
ready = ready.filter(v => v.remaining > 0);
// keep idx pointing at a valid lead by re-modding against new length next iteration
if (ready.length !== before && ready.length) idx = idx % ready.length;
}
if (!ready.length) { drainedAtSlot = s + 1; break; }
}
// ---- rollups keyed by local calendar day ----
const dayKey = ms => { const d = new Date(ms);
return `${d.getFullYear()}-${String(d.getMonth()+1).padStart(2,'0')}-${String(d.getDate()).padStart(2,'0')}`; };
const days = {}; // key -> { date, total, vendors:{name->{count,soldBy,mapPriced,subLines}}, slots:[] }
for (const sl of slots) {
const k = dayKey(sl.fireMs);
if (!days[k]) days[k] = { date: k, total: 0, vendors: {}, slotCount: 0, blockedSlots: 0 };
days[k].slotCount++;
if (sl.blocked) days[k].blockedSlots++;
days[k].total += sl.created;
for (const v of sl.vendors) {
if (v.count <= 0) continue;
if (!days[k].vendors[v.vendor]) days[k].vendors[v.vendor] =
{ count: 0, soldBy: v.soldBy, mapPriced: v.mapPriced, subLines: v.subLines };
days[k].vendors[v.vendor].count += v.count;
}
}
const dayList = Object.values(days).map(d => ({
...d,
vendors: Object.entries(d.vendors).map(([vendor, info]) => ({ vendor, ...info }))
.sort((a, b) => b.count - a.count),
})).sort((a, b) => a.date.localeCompare(b.date));
return {
constants: CAD,
cursorIdx: loadCadenceCursor().idx || 0,
grandTotal,
readyVendorCount: readyVendors().filter(({ vendor, cfg }) => {
try { return parseInt(psql(`SELECT count(*) FROM ${cfg.table} WHERE ${WHERE(cfg)};`) || '0', 10) > 0; } catch { return false; }
}).length,
slots,
days: dayList,
drainSlots: drainedAtSlot,
drainDays: drainedAtSlot != null ? Math.ceil(drainedAtSlot / CAD.SLOTS_PER_DAY) : null,
truncated: drainedAtSlot == null,
};
}
// ---- shared payload builders (used by both live psql and snapshot-generator) ----
function buildSummary() {
const counts = vendorCounts();
const grand = counts.reduce((s, r) => s + r.count, 0);
const out = counts.map(r => {
const cfg = VENDORS[r.vendor];
const subs = subLines(r.vendor, cfg);
return { ...r, mapPriced: !!cfg.sellExpr, soldBy: cfg.soldBy, subLines: subs };
});
return { grandTotal: grand, vendors: out, generatedAt: new Date().toISOString() };
}
function buildCalendarPayload(searchParams) {
const params = searchParams || new URLSearchParams();
const now = new Date();
const next = new Date(now);
next.setMinutes(CAD.FIRE_MINUTE, 0, 0);
if (next.getTime() <= now.getTime()) next.setHours(next.getHours() + 1);
const todayFirst = new Date(now); todayFirst.setHours(0, CAD.FIRE_MINUTE, 0, 0);
const capClearHour = parseInt(params.get('capClearHour') || '12', 10);
const capClear = new Date(now); capClear.setHours(capClearHour, CAD.FIRE_MINUTE, 0, 0);
const blockedUntilMs = Math.max(capClear.getTime(), now.getTime());
const proj = buildProjection({ startEpochMs: todayFirst.getTime(), blockedUntilMs });
return {
generatedAt: now.toISOString(),
now: now.getTime(),
nextFireMs: next.getTime(),
capClearMs: capClear.getTime(),
timezoneOffsetMin: now.getTimezoneOffset(),
...proj,
};
}
// When required as a module (build-snapshot.js), export the read-only builders and stop.
if (require.main !== module) {
module.exports = { buildSummary, buildCalendarPayload, readyVendors, vendorRows, VENDORS };
return;
}
// ---- HTTP ----
function send(res, code, body, type = 'application/json') {
res.writeHead(code, { 'Content-Type': type, 'Cache-Control': 'no-store' });
res.end(body);
}
const server = http.createServer((req, res) => {
const u = new URL(req.url, `http://127.0.0.1:${PORT}`);
// Hard read-only: refuse anything but GET.
if (req.method !== 'GET') return send(res, 405, JSON.stringify({ error: 'read-only viewer; GET only' }));
if (u.pathname === '/api/summary') {
try {
if (SNAPSHOT_ONLY) {
const snap = loadSnapshot();
return send(res, 200, JSON.stringify({ ...snap.summary, snapshot: true, snapshotAt: snap.generatedAt }));
}
return send(res, 200, JSON.stringify(buildSummary()));
} catch (e) {
return send(res, 500, JSON.stringify({ error: String(e.message).slice(0, 300) }));
}
}
if (u.pathname === '/api/calendar') {
try {
if (SNAPSHOT_ONLY) {
const snap = loadSnapshot();
return send(res, 200, JSON.stringify({ ...snap.calendar, snapshot: true, snapshotAt: snap.generatedAt }));
}
return send(res, 200, JSON.stringify(buildCalendarPayload(u.searchParams)));
} catch (e) {
return send(res, 500, JSON.stringify({ error: String(e.message).slice(0, 300) }));
}
}
if (u.pathname === '/api/skus') {
try {
const vendor = u.searchParams.get('vendor') || '';
const limit = Math.min(parseInt(u.searchParams.get('limit') || '200', 10) || 200, 500);
const offset = Math.max(parseInt(u.searchParams.get('offset') || '0', 10) || 0, 0);
// ---- snapshot path: page/filter the in-memory rows exactly like the DB path ----
if (SNAPSHOT_ONLY) {
const snap = loadSnapshot();
let rows = snap.rows;
if (vendor) {
rows = rows.filter(r => r.vendor === vendor);
if (!rows.length) return send(res, 400, JSON.stringify({ error: `unknown/non-ready vendor: ${vendor}` }));
}
// rows are already globally newest-first in the snapshot; vendor-filtered keeps that order.
const page = rows.slice(offset, offset + limit);
return send(res, 200, JSON.stringify({ vendor: vendor || null, limit, offset, count: page.length, rows: page, snapshot: true, snapshotAt: snap.generatedAt }));
}
const ready = readyVendors();
const targets = vendor ? ready.filter(v => v.vendor === vendor) : ready;
if (vendor && !targets.length) return send(res, 400, JSON.stringify({ error: `unknown/non-ready vendor: ${vendor}` }));
// For "all vendors" we page across the union ordered by vendor then date; simplest correct
// approach for an admin tool is to require a vendor filter for deep paging, but still support
// a blended first-page view. We fetch limit rows from each target then globally sort+slice.
let rows = [];
for (const { vendor: v, cfg } of targets) {
rows = rows.concat(vendorRows(v, cfg, vendor ? limit : Math.min(limit, 60), vendor ? offset : 0));
}
if (!vendor) {
rows.sort((a, b) => String(b.staged_at || '').localeCompare(String(a.staged_at || '')));
rows = rows.slice(offset, offset + limit);
}
return send(res, 200, JSON.stringify({ vendor: vendor || null, limit, offset, count: rows.length, rows }));
} catch (e) {
return send(res, 500, JSON.stringify({ error: String(e.message).slice(0, 300) }));
}
}
// static
let p = u.pathname === '/' ? '/index.html' : u.pathname;
const file = path.join(__dirname, 'public', p);
if (!file.startsWith(path.join(__dirname, 'public'))) return send(res, 403, 'forbidden', 'text/plain');
fs.readFile(file, (err, data) => {
if (err) return send(res, 404, 'not found', 'text/plain');
const ext = path.extname(file);
const ct = ext === '.html' ? 'text/html' : ext === '.js' ? 'text/javascript' : ext === '.css' ? 'text/css' : 'application/octet-stream';
send(res, 200, data, ct);
});
});
server.listen(PORT, '127.0.0.1', () => {
console.log(`DW Staged-for-Active viewer (READ-ONLY) → http://127.0.0.1:${PORT}`);
});