← back to Vendor Agents Viewer
server.js
198 lines
/**
* Vendor Agents Viewer — review / launch / restart the DW vendor-agent fleet.
* Reads the agent registry from local dw_unified (vendor_registry), gets live
* up/down status from Kamatera's listening ports, and proxies launch/restart/scan
* actions to the prod box over SSH. Read-only by default; actions are explicit.
*/
const express = require('express');
const { Pool } = require('pg');
const { execFile } = require('child_process');
const fs = require('fs');
const path = require('path');
const app = express();
const PORT = process.env.PORT || 9789;
const SSH = 'my-server'; // ~/.ssh/config alias for root@45.61.58.125
// Agents run LOCALLY on Mac2 (where the scrapers live). Launch = pm2 start agent-host.
const SCRAPERS_DIR = '/Users/macstudio3/Projects/Designer-Wallcoverings/DW-Programming/ImportNewSkufromURL/lib/scrapers';
const AGENT_HOST = path.join(__dirname, 'agent-host.js');
// vendor_code (underscores) -> scraper id (hyphens); null if no scraper file exists.
// SCRAPER_ALIASES: registry codes whose scraper file lives under a different id
// (renames, private-label sources, brand-of-house mappings). Verified 2026-06-10.
const SCRAPER_ALIASES = {
DWDP: 'dwdp', // exact-case match (APFS is case-insensitive but be precise)
brewster_york: 'brewster',
elitis: 'elitis-wallcovering',
york_contract: 'york',
cole_son: 'cole-and-son',
maya: 'maya-romanoff',
w1838: '1838-wallcoverings',
fabricut: 'fabricut-stroheim',
hygge: 'hygge-west',
mindthegap: 'mind-the-gap',
osborne: 'osborne-little',
wallquest: 'wallquest-residential',
timorous: 'timorous-beasties',
gp_baker: 'gpj-baker',
baker_lifestyle: 'gpj-baker', // Baker Lifestyle is a GP&J Baker house brand
gaston_daniela: 'gaston-y-daniela',
missoni: 'missoni-home',
versace: 'versace-home',
mulberry: 'mulberry-home',
newwall: 'new-wall',
china_seas: 'quadrille', // China Seas is a Quadrille brand (see quadrille-core)
hollywood: 'momentum-textiles', // Hollywood = Momentum private label
jeffrey_stevens: 'york', // registry: "York (Jeffrey Stevens source)"
phillipe_romano: 'york', // registry: "York/Brewster (Phillipe Romano source)"
};
function resolveScraper(vendorCode) {
if (!vendorCode) return null;
const cands = [vendorCode.replace(/_/g, '-'), vendorCode.replace(/_/g, ''), vendorCode];
const alias = SCRAPER_ALIASES[vendorCode];
if (alias) cands.unshift(alias);
for (const c of cands) {
if (fs.existsSync(path.join(SCRAPERS_DIR, `${c}-new-products-scraper.ts`))) return c;
}
return null;
}
function localExec(cmd) {
return new Promise((resolve) => execFile('bash', ['-lc', cmd],
{ timeout: 60000, maxBuffer: 8 * 1024 * 1024 }, (e, out, err) => resolve({ ok: !e, out: (out || '') + (err || '') })));
}
const pool = new Pool({
host: '127.0.0.1', port: 5432, user: 'dw_admin',
password: process.env.PGPASSWORD || '', database: 'dw_unified',
});
app.use(express.json());
app.use(express.static(__dirname + '/public'));
// --- live port -> real pm2 name map on Kamatera (cached 30s) ---
// The registry's pm2_name is unreliable; resolve the ACTUAL pm2 process that
// owns each agent_port so restart/launch hit the right process.
let portCache = { at: 0, map: {} };
// LOCAL port -> pm2 process name (agents run on Mac2). Cached 15s.
async function livePortMap() {
if (Date.now() - portCache.at < 15000) return portCache.map;
const ls = await localExec("lsof -nP -iTCP -sTCP:LISTEN 2>/dev/null | grep -oE 'node .*:(9[0-9]{3}) ' | grep -oE ':(9[0-9]{3})' | tr -d ':' | sort -u");
const ports = new Set(ls.out.split('\n').map(s => s.trim()).filter(Boolean));
// pid->name + port->pid via lsof for node, joined with pm2
const lsofPid = await localExec("lsof -nP -iTCP -sTCP:LISTEN 2>/dev/null | awk '/node/{print $2\" \"$9}'");
const pm2j = await localExec("pm2 jlist 2>/dev/null");
const pidName = {};
try { JSON.parse(pm2j.out).forEach(p => { pidName[String(p.pid)] = p.name; }); } catch {}
const map = {};
lsofPid.out.split('\n').forEach(l => {
const m = l.match(/^(\d+)\s.*:(9\d{3})$/);
if (m) map[m[2]] = pidName[m[1]] || ('pid:' + m[1]);
});
// ensure any listening port present even if pm2 name unknown
ports.forEach(p => { if (!map[p]) map[p] = 'listening'; });
portCache = { at: Date.now(), map };
return map;
}
function describe(a) {
const cat = a.catalog_count || 0, shop = a.shopify_count || 0;
return `${a.agent_name || 'Agent'} maintains the ${a.vendor_name} catalog — scrapes the vendor site, ` +
`normalizes SKUs, and syncs to Shopify. Currently tracking ${cat.toLocaleString()} catalog SKUs` +
(shop ? `, ${shop.toLocaleString()} live on Shopify.` : '.');
}
// --- API: all agents with status + usage ---
app.get('/api/agents', async (req, res) => {
try {
const live = await livePortMap();
const { rows } = await pool.query(`
SELECT vendor_code, vendor_name, agent_name, agent_port, pm2_name, website_url,
catalog_count, shopify_count, catalog_with_images, catalog_on_shopify,
catalog_not_on_shopify, website_score, discontinued_count, updated_at,
skip_catalog_scrape, is_private_label, is_active
FROM vendor_registry
WHERE agent_port IS NOT NULL OR agent_name IS NOT NULL OR pm2_name IS NOT NULL
ORDER BY agent_port NULLS LAST, vendor_name`);
const slug = s => (s || '').toLowerCase().replace(/[^a-z0-9]/g, '');
const agents = rows.map(a => {
const realPm2 = a.agent_port ? live[String(a.agent_port)] : null;
const scraperId = resolveScraper(a.vendor_code);
const myPm2 = 'va-' + slug(a.vendor_code); // name this viewer launches under
// ONLINE only if the live process owning the port is THIS viewer's agent.
const matches = realPm2 && (realPm2 === myPm2 || slug(realPm2) === slug(myPm2));
let status = 'offline';
if (matches) status = 'online';
else if (realPm2) status = 'conflict'; // port reused by an unrelated app
return {
...a,
status,
scraper_id: scraperId,
launchable: !!scraperId,
pm2_name: myPm2,
registry_pm2: a.pm2_name,
conflict_owner: (!matches && realPm2) ? realPm2 : null,
description: describe(a) + (scraperId ? '' :
a.skip_catalog_scrape
? (a.is_private_label
? ' (private label — no external source to scrape; agent serves health only)'
: ' (no scrapeable source — dead/parked/gated site; agent serves health only)')
: ' (no scraper built yet — launch serves health only)'),
url: (matches && a.agent_port) ? `http://127.0.0.1:${a.agent_port}/` : null,
};
});
const summary = {
total: agents.length,
online: agents.filter(x => x.status === 'online').length,
offline: agents.filter(x => x.status === 'offline').length,
conflict: agents.filter(x => x.status === 'conflict').length,
totalCatalog: agents.reduce((s, x) => s + (x.catalog_count || 0), 0),
totalShopify: agents.reduce((s, x) => s + (x.shopify_count || 0), 0),
};
res.json({ ok: true, summary, agents });
} catch (e) { res.status(500).json({ ok: false, error: e.message }); }
});
async function vendorRow(code) {
const { rows } = await pool.query(
'SELECT vendor_code, vendor_name, agent_name, agent_port FROM vendor_registry WHERE vendor_code=$1', [code]);
return rows[0];
}
const sh = s => "'" + String(s || '').replace(/'/g, "'\\''") + "'";
// --- LAUNCH: pm2-start the generic agent host locally for this vendor ---
app.post('/api/launch', async (req, res) => {
const code = String(req.body.vendor_code || '').replace(/[^a-zA-Z0-9_-]/g, '');
const v = await vendorRow(code);
if (!v || !v.agent_port) return res.json({ ok: false, error: 'vendor has no agent_port' });
const scraper = resolveScraper(code) || '';
const name = 'va-' + code.toLowerCase().replace(/[^a-z0-9]/g, '');
portCache.at = 0;
const env = `AGENT_PORT=${v.agent_port} VENDOR_NAME=${sh(v.vendor_name)} VENDOR_CODE=${sh(code)} ` +
`AGENT_NAME=${sh(v.agent_name || 'Agent')} SCRAPER_ID=${sh(scraper)}`;
// --kill-timeout 4000: give the old instance 4s to release the listen socket on a
// restart before SIGKILL, so the new bind doesn't race into EADDRINUSE (pairs with
// the agent-host.js backoff binder).
const cmd = `pm2 delete ${name} >/dev/null 2>&1; ${env} pm2 start ${sh(AGENT_HOST)} --name ${name} --kill-timeout 4000 --update-env 2>&1 | tail -2; pm2 save >/dev/null 2>&1`;
const r = await localExec(cmd);
res.json({ ...r, name, port: v.agent_port, scraper: scraper || '(none — health only)' });
});
// --- RESTART: restart this vendor's local agent ---
app.post('/api/restart', async (req, res) => {
const code = String(req.body.vendor_code || '').replace(/[^a-zA-Z0-9_-]/g, '');
const name = 'va-' + code.toLowerCase().replace(/[^a-z0-9]/g, '');
portCache.at = 0;
const r = await localExec(`pm2 restart ${name} --update-env 2>&1 | tail -2 || echo 'not running — use Launch'`);
res.json({ ...r, name });
});
// --- SCAN: hit the running agent's /api/scan (runs the real scraper) ---
app.post('/api/scan', async (req, res) => {
const code = String(req.body.vendor_code || '').replace(/[^a-zA-Z0-9_-]/g, '');
const v = await vendorRow(code);
if (!v || !v.agent_port) return res.json({ ok: false, error: 'no agent_port' });
const r = await localExec(`curl -s -m 130 -X POST http://127.0.0.1:${v.agent_port}/api/scan -H 'Content-Type: application/json' -d '{"maxResults":8}' 2>&1 | head -c 800`);
res.json(r);
});
app.listen(PORT, '0.0.0.0', () => console.log(`Vendor Agents Viewer → http://127.0.0.1:${PORT}`));