← back to Archive Agent
server.js
393 lines
#!/usr/bin/env node
// Archive Agent — watches Shopify archives, flags wrongful ones, reports to PG.
// • Scans every 60 min by default (ARCHIVE_SCAN_MINUTES to override)
// • Writes alerts to PG archive_alerts (unique on shopify_id)
// • Dashboard at GET /
// • Manual restore via POST /restore { ids: [...] }
// • No auto-restore.
const express = require('express');
const helmet = require('helmet');
const { Pool } = require('pg');
const fs = require('fs');
const PORT = process.env.PORT || 9648;
const SHOP = process.env.SHOPIFY_STORE || 'designer-laboratory-sandbox.myshopify.com';
const TOKEN = process.env.SHOPIFY_ADMIN_TOKEN;
if (!TOKEN) { console.error('FATAL: SHOPIFY_ADMIN_TOKEN env required'); process.exit(1); }
const SCAN_MINUTES = parseInt(process.env.ARCHIVE_SCAN_MINUTES || '60', 10);
const DB_URL = process.env.DATABASE_URL;
if (!DB_URL) { console.error('FATAL: DATABASE_URL env required'); process.exit(1); }
const pool = new Pool({ connectionString: DB_URL });
// Vendor-to-responsible-agent routing (for alerts).
const VENDOR_OWNER = {
'Hollywood Wallcoverings': 'hollywood-agent',
'Wolf Gordon': 'wolfgordon-agent',
'Nina Campbell': 'nina-campbell-agent',
'Phillipe Romano': 'phillipe-romano-agent',
'Sandberg': 'sandberg-agent',
'Brunschwig & Fils': 'kravet-agent',
'Lee Jofa': 'kravet-agent',
'Kravet': 'kravet-agent',
'GP & J Baker': 'kravet-agent',
'Mulberry': 'kravet-agent',
'Gaston Y Daniela': 'kravet-agent',
'Threads': 'kravet-agent',
'Versace': 'dwvs-agent',
'Jeffrey Stevens': 'york-bye-bye',
'Ronald Redding': 'york-bye-bye',
'Malibu Walls Type 2 Vinyls': 'york-bye-bye',
'York Contract': 'york-bye-bye',
};
function ownerFor(vendor) { return VENDOR_OWNER[vendor] || 'plumbing-agent'; }
let lastScan = null;
let scanRunning = false;
let scanResult = { total:0, newAlerts:0, byClass:{}, error:null };
// ---------- Shopify GQL with retry ----------
async function gql(query, variables = {}, attempt = 0) {
const r = await fetch(`https://${SHOP}/admin/api/2024-01/graphql.json`, {
method: 'POST',
headers: { 'Content-Type': 'application/json', 'X-Shopify-Access-Token': TOKEN },
body: JSON.stringify({ query, variables }),
});
const txt = await r.text();
if (r.status === 429 || r.status >= 500 || txt.startsWith('<')) {
if (attempt >= 8) throw new Error(`retry limit; ${r.status}`);
await new Promise(res => setTimeout(res, 1000 * 2 ** attempt));
return gql(query, variables, attempt + 1);
}
const j = JSON.parse(txt);
if (j.errors) {
if ((j.errors[0]?.extensions?.code === 'THROTTLED') && attempt < 8) {
await new Promise(res => setTimeout(res, 2000 * (attempt + 1)));
return gql(query, variables, attempt + 1);
}
throw new Error(JSON.stringify(j.errors));
}
return j.data;
}
// ---------- Classifier ----------
// Inputs: product node from Shopify, PG helpers.
async function classify(p, activeInCache, brewsterYorkSkus) {
const tags = p.tags || [];
const hasYB = tags.some(t => /^YB-/i.test(t));
const hasDisco = tags.some(t => /discontinued/i.test(t));
const variants = p.variants?.nodes || [];
const hadPricedBolt = variants.some(v => !/sample/i.test(v.title || '') && parseFloat(v.price) > 0);
let signal = null;
let cls;
if (hasYB) {
signal = tags.find(t => /^YB-/i.test(t));
cls = 'OK_YB';
} else if (hasDisco) {
signal = 'Discontinued';
cls = 'OK_DISCO';
} else {
// Brewster/York brand check: if vendor is in Brewster catalog + MFR SKU absent → legit disco
const mfr = (p.mfrMf?.value || '').trim();
const brewsterBrands = new Set([
'York Wallcoverings','Ronald Redding','A-Street Prints','Chesapeake','Advantage','Warner',
'Antonina Vella','Candice Olson','Magnolia Home','Rifle Paper Co.','Brewster','Kenneth James',
'York Contract','York','Madcap Cottage','Lemieux et Cie','Carol Benson-Cobb',
]);
const isByBrand = brewsterBrands.has(p.vendor);
if (isByBrand && mfr && !brewsterYorkSkus.has(mfr.toUpperCase())) {
signal = 'NOT_IN_BREWSTER_CATALOG';
cls = 'OK_NOT_IN_CATALOG';
} else if (hadPricedBolt) {
cls = 'WRONG_PRICED';
} else if (activeInCache.has(p.id)) {
cls = 'DRIFT';
} else if (variants.length === 0) {
cls = 'UNKNOWN';
} else {
// Sample-only, no disco signal — wrongful per corrected rule
cls = 'WRONG_SAMPLE';
}
}
return { cls, signal, hadPricedBolt };
}
// ---------- Scanner ----------
async function runScan({ lookbackDays = 7 } = {}) {
if (scanRunning) return { alreadyRunning: true };
scanRunning = true;
const started = Date.now();
try {
const since = new Date(Date.now() - lookbackDays * 24 * 60 * 60 * 1000).toISOString();
// Pre-load PG helpers
const [{ rows: cacheActive }, { rows: brewster }] = await Promise.all([
pool.query(`SELECT DISTINCT shopify_id FROM shopify_products WHERE status='ACTIVE'`),
pool.query(`SELECT DISTINCT mfr_sku FROM brewster_york_master`),
]);
const activeInCache = new Set(cacheActive.map(r => r.shopify_id));
const brewsterYorkSkus = new Set(brewster.map(r => String(r.mfr_sku).trim().toUpperCase()));
const q = `status:archived AND updated_at:>=${since}`;
const query = `
query($q: String!, $after: String) {
products(first: 250, query: $q, after: $after, sortKey: UPDATED_AT, reverse: true) {
pageInfo { hasNextPage endCursor }
nodes {
id handle title vendor updatedAt createdAt tags
mfrMf: metafield(namespace:"custom", key:"manufacturer_sku") { value }
variants(first: 20) { nodes { title price } }
}
}
}`;
let cursor = null, scanned = 0, newAlerts = 0;
const byClass = {};
while (true) {
const d = await gql(query, { q, after: cursor });
for (const p of d.products.nodes) {
scanned++;
const { cls, signal, hadPricedBolt } = await classify(p, activeInCache, brewsterYorkSkus);
byClass[cls] = (byClass[cls] || 0) + 1;
// Upsert into archive_alerts
const res = await pool.query(`
INSERT INTO archive_alerts
(shopify_id, vendor, title, handle, classification, archive_signal,
had_priced_bolt, tags, shopify_updated, shopify_created)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
ON CONFLICT (shopify_id) DO UPDATE SET
classification = EXCLUDED.classification,
archive_signal = EXCLUDED.archive_signal,
had_priced_bolt = EXCLUDED.had_priced_bolt,
tags = EXCLUDED.tags,
shopify_updated = EXCLUDED.shopify_updated
RETURNING (xmax = 0) AS inserted
`, [
p.id, p.vendor || null, p.title, p.handle, cls, signal,
hadPricedBolt, p.tags || [], p.updatedAt, p.createdAt,
]);
if (res.rows[0]?.inserted) newAlerts++;
}
if (!d.products.pageInfo.hasNextPage) break;
cursor = d.products.pageInfo.endCursor;
}
lastScan = new Date().toISOString();
scanResult = { total: scanned, newAlerts, byClass, error: null, tookMs: Date.now() - started, at: lastScan };
return scanResult;
} catch (e) {
scanResult = { total:0, newAlerts:0, byClass:{}, error: e.message, at: new Date().toISOString() };
return scanResult;
} finally {
scanRunning = false;
}
}
// ---------- Restore helper (manual) ----------
async function restoreIds(ids) {
const out = { ok: 0, fail: 0, errors: [] };
for (const id of ids) {
try {
const d = await gql(`query { product(id:"${id}") { images(first:1){nodes{id}} } }`);
const target = d.product?.images?.nodes?.length ? 'ACTIVE' : 'DRAFT';
const r = await gql(`mutation { productUpdate(input:{ id:"${id}", status: ${target} }) { userErrors { message } } }`);
const ue = r.productUpdate?.userErrors || [];
if (ue.length) { out.fail++; out.errors.push({ id, error: ue.map(e=>e.message).join('; ') }); continue; }
await pool.query(`UPDATE archive_alerts SET restored_at = now(), acknowledged_at = now(), acknowledged_by='manual' WHERE shopify_id=$1`, [id]);
out.ok++;
} catch (e) { out.fail++; out.errors.push({ id, error: e.message }); }
await new Promise(r => setTimeout(r, 80));
}
return out;
}
// ---------- HTTP ----------
const app = express();
// Security headers via helmet (added 2026-05-04 overnight YOLO loop)
app.use(helmet({ contentSecurityPolicy: false }));
app.use(express.json({ limit: '5mb' }));
// Defense-in-depth: 404 any request that hits a snapshot/backup-file extension.
// Runs BEFORE any static middleware that might be added later, so leftover
// *.bak / *.pre-* files in the project root never leak via HTTP.
app.use((req, res, next) => {
if (/\.(bak|orig|swp)(\.|$)|\.pre-/i.test(req.path)) {
return res.status(404).send('not found');
}
next();
});
// LAN/tailnet auth gate (env-only, no source fallback)
const BASIC_AUTH = process.env.BASIC_AUTH;
if (!BASIC_AUTH) {
console.error('[fatal] BASIC_AUTH env var unset — refusing to start');
process.exit(1);
}
const AUTH_HEADER = 'Basic ' + Buffer.from(BASIC_AUTH).toString('base64');
app.use((req, res, next) => {
if (req.path === '/health' || req.path === '/healthz') return next();
if (req.get('authorization') === AUTH_HEADER) return next();
res.set('WWW-Authenticate', 'Basic realm="mac2-pm2"');
res.status(401).send('auth required');
});
app.get('/health', (req, res) => res.json({ ok: true, port: PORT, lastScan, scanRunning }));
app.post('/scan', async (req, res) => {
const days = parseInt(req.body?.days || req.query.days || '7', 10);
const result = await runScan({ lookbackDays: days });
res.json(result);
});
app.get('/alerts', async (req, res) => {
try {
const cls = req.query.class || null;
const vendor = req.query.vendor || null;
const acknowledged = req.query.ack === '1';
const limit = Math.min(parseInt(req.query.limit || '500', 10), 5000);
const params = [];
const where = [];
if (cls) { params.push(cls); where.push(`classification = $${params.length}`); }
if (vendor) { params.push(vendor); where.push(`vendor = $${params.length}`); }
where.push(`acknowledged_at IS ${acknowledged ? 'NOT NULL' : 'NULL'}`);
params.push(limit);
const { rows } = await pool.query(
`SELECT * FROM archive_alerts WHERE ${where.join(' AND ')} ORDER BY detected_at DESC LIMIT $${params.length}`,
params
);
res.json({ count: rows.length, rows });
} catch (e) {
res.status(500).json({ error: e.message });
}
});
app.get('/summary', async (req, res) => {
try {
const { rows: byClass } = await pool.query(`
SELECT classification, COUNT(*)::int AS n,
COUNT(*) FILTER (WHERE acknowledged_at IS NULL)::int AS unacked
FROM archive_alerts GROUP BY 1 ORDER BY n DESC`);
const { rows: byVendor } = await pool.query(`
SELECT vendor, COUNT(*)::int AS n,
COUNT(*) FILTER (WHERE classification IN ('WRONG_PRICED','WRONG_SAMPLE','DRIFT'))::int AS wrongful
FROM archive_alerts WHERE acknowledged_at IS NULL
GROUP BY 1 ORDER BY wrongful DESC, n DESC LIMIT 30`);
const { rows: byDay } = await pool.query(`
SELECT DATE(shopify_updated) AS day, COUNT(*)::int AS n,
COUNT(*) FILTER (WHERE classification IN ('WRONG_PRICED','WRONG_SAMPLE','DRIFT'))::int AS wrongful
FROM archive_alerts WHERE shopify_updated > now() - interval '30 days'
GROUP BY 1 ORDER BY 1 DESC`);
res.json({ lastScan, scanResult, byClass, byVendor, byDay });
} catch (e) {
res.status(500).json({ error: e.message });
}
});
app.post('/restore', async (req, res) => {
const ids = req.body?.ids;
if (!Array.isArray(ids) || !ids.length) return res.status(400).json({ error: 'ids[] required' });
if (ids.length > 2000) return res.status(400).json({ error: 'max 2000 per call' });
const result = await restoreIds(ids);
res.json(result);
});
app.post('/ack', async (req, res) => {
try {
const ids = req.body?.ids;
const by = req.body?.by || 'manual';
if (!Array.isArray(ids) || !ids.length) return res.status(400).json({ error: 'ids[] required' });
await pool.query(`UPDATE archive_alerts SET acknowledged_at = now(), acknowledged_by=$1 WHERE shopify_id = ANY($2::text[])`, [by, ids]);
res.json({ ok: true, acked: ids.length });
} catch (e) {
res.status(500).json({ error: e.message });
}
});
app.get('/owner/:vendor', (req, res) => res.json({ vendor: req.params.vendor, owner: ownerFor(req.params.vendor) }));
app.get('/', async (req, res) => {
try {
const { rows: byClass } = await pool.query(`
SELECT classification, COUNT(*)::int AS n,
COUNT(*) FILTER (WHERE acknowledged_at IS NULL)::int AS unacked
FROM archive_alerts GROUP BY 1 ORDER BY n DESC`);
const { rows: byVendor } = await pool.query(`
SELECT vendor, COUNT(*)::int AS n,
COUNT(*) FILTER (WHERE classification IN ('WRONG_PRICED','WRONG_SAMPLE','DRIFT') AND acknowledged_at IS NULL)::int AS wrongful_unacked
FROM archive_alerts WHERE vendor IS NOT NULL
GROUP BY 1 ORDER BY wrongful_unacked DESC, n DESC LIMIT 30`);
const { rows: recent } = await pool.query(`
SELECT shopify_id, vendor, title, classification, archive_signal, had_priced_bolt,
shopify_updated, detected_at, acknowledged_at
FROM archive_alerts
WHERE classification IN ('WRONG_PRICED','WRONG_SAMPLE','DRIFT') AND acknowledged_at IS NULL
ORDER BY shopify_updated DESC LIMIT 100`);
const classColors = {
OK_YB:'#198754', OK_DISCO:'#198754', OK_NOT_IN_CATALOG:'#198754',
WRONG_PRICED:'#dc3545', WRONG_SAMPLE:'#fd7e14', DRIFT:'#6f42c1', UNKNOWN:'#6c757d',
};
function pill(c) { return `<span style="background:${classColors[c]||'#6c757d'};color:#fff;padding:2px 8px;border-radius:10px;font-size:11px">${c}</span>`; }
res.set('Content-Type','text/html').send(`<!doctype html>
<html><head><meta charset="utf-8"><title>Archive Agent</title>
<style>
body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;margin:20px;background:#f8f9fa;color:#212529}
h1{margin-top:0}
table{border-collapse:collapse;width:100%;background:#fff;box-shadow:0 1px 3px rgba(0,0,0,.05);margin-bottom:24px}
th,td{padding:6px 10px;border-bottom:1px solid #eee;font-size:13px;text-align:left;vertical-align:top}
th{background:#e9ecef;font-weight:600}
.meta{color:#6c757d;font-size:12px}
.kpi{display:inline-block;margin-right:24px;padding:14px 20px;background:#fff;border-radius:8px;box-shadow:0 1px 3px rgba(0,0,0,.05)}
.kpi .n{font-size:28px;font-weight:600}
button{padding:6px 12px;background:#0d6efd;color:#fff;border:none;border-radius:4px;cursor:pointer}
button:hover{background:#0b5ed7}
code{background:#f1f3f5;padding:2px 5px;border-radius:3px;font-size:12px}
</style></head><body>
<h1>Archive Agent</h1>
<p class="meta">port ${PORT} · last scan: ${lastScan || '(never)'} · scan in progress: ${scanRunning}</p>
<div>
<span class="kpi"><span class="n">${byClass.reduce((s,r)=>s+r.n,0)}</span><br><span class="meta">total alerts</span></span>
${byClass.map(r => `<span class="kpi"><span class="n" style="color:${classColors[r.classification]||'#000'}">${r.n}</span><br><span class="meta">${r.classification} (${r.unacked} unacked)</span></span>`).join('')}
</div>
<p><button onclick="fetch('/scan',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({days:7})}).then(r=>r.json()).then(j=>{alert(JSON.stringify(j,null,2));location.reload()})">Run scan now (7 days)</button></p>
<h2>Top vendors (unacked wrongful)</h2>
<table><tr><th>Vendor</th><th>Owner</th><th>Wrongful Unacked</th><th>Total</th></tr>
${byVendor.map(r => `<tr><td>${r.vendor}</td><td><code>${ownerFor(r.vendor)}</code></td><td>${r.wrongful_unacked}</td><td>${r.n}</td></tr>`).join('')}
</table>
<h2>Recent wrongful archives (top 100 unacked)</h2>
<table><tr><th>When archived</th><th>Vendor</th><th>Title</th><th>Class</th><th>Had Priced Bolt?</th><th>Actions</th></tr>
${recent.map(r => `<tr>
<td>${(r.shopify_updated||'').toString().slice(0,19)}</td>
<td>${r.vendor||''}</td>
<td>${(r.title||'').slice(0,60)}</td>
<td>${pill(r.classification)}</td>
<td>${r.had_priced_bolt ? '✅' : '—'}</td>
<td><button onclick="fetch('/restore',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({ids:['${r.shopify_id}']})}).then(r=>r.json()).then(j=>{alert(JSON.stringify(j));location.reload()})">Restore</button>
<button onclick="fetch('/ack',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({ids:['${r.shopify_id}']})}).then(r=>r.json()).then(_=>location.reload())">Ack</button></td>
</tr>`).join('')}
</table>
<p class="meta">API: <code>GET /summary</code> · <code>GET /alerts?class=WRONG_PRICED</code> · <code>POST /scan {days:N}</code> · <code>POST /restore {ids:[...]}</code> · <code>POST /ack {ids:[...]}</code></p>
</body></html>`);
} catch (e) {
res.status(500).send(`<pre>DB error: ${e.message}</pre>`);
}
});
// ---------- Bootstrap ----------
app.listen(PORT, '0.0.0.0', () => {
console.log(`[archive-agent] listening on :${PORT}`);
// First scan on startup (non-blocking)
setTimeout(() => runScan({ lookbackDays: 30 }).then(r => console.log('[scan]', JSON.stringify(r))).catch(e => console.error(e)), 5000);
// Recurring scan
setInterval(() => runScan({ lookbackDays: 7 }).catch(e => console.error('[scan-err]', e.message)), SCAN_MINUTES * 60 * 1000);
});