← back to Dw Scrape Scheduler
wire canonical scrape_staging DB writes (dedupe + insert/update/gone change-detection); dispatcher pool cleanup; shopify cohort proven end-to-end
1b2ac9120919ae69263f902c87c5021360295040 · 2026-07-13 00:41:25 -0700 · Steve Abrams
Files touched
M dispatcher.jsA lib/write-staging.jsM run-vendor.js
Diff
commit 1b2ac9120919ae69263f902c87c5021360295040
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Mon Jul 13 00:41:25 2026 -0700
wire canonical scrape_staging DB writes (dedupe + insert/update/gone change-detection); dispatcher pool cleanup; shopify cohort proven end-to-end
---
dispatcher.js | 15 ++++++++++-----
lib/write-staging.js | 53 ++++++++++++++++++++++++++++++++++++++++++++++++++++
run-vendor.js | 14 ++++++++++++--
3 files changed, 75 insertions(+), 7 deletions(-)
diff --git a/dispatcher.js b/dispatcher.js
index d03f473..2135061 100644
--- a/dispatcher.js
+++ b/dispatcher.js
@@ -9,6 +9,7 @@
const fs = require('fs');
const path = require('path');
const { runVendor } = require('./run-vendor');
+const { pool } = require('./lib/db');
const args = process.argv.slice(2);
const flag = (n, d) => { const i = args.indexOf(n); return i >= 0 ? (args[i + 1] && !args[i + 1].startsWith('--') ? args[i + 1] : true) : d; };
@@ -18,7 +19,7 @@ const ALL = args.includes('--all');
const ONE = flag('--vendor', null);
const DAY = Number(flag('--day', new Date().getDate()));
-async function pool(items, cap, fn) {
+async function runPool(items, cap, fn) {
const out = []; let i = 0;
const workers = Array.from({ length: Math.min(cap, items.length) }, async () => {
while (i < items.length) { const idx = i++; out[idx] = await fn(items[idx]); }
@@ -42,11 +43,15 @@ async function pool(items, cap, fn) {
if (todo.length === 0) { line(' nothing scheduled today — done.'); return; }
if (DRY) { todo.forEach(e => line(` would scrape: ${e.vendor} [${e.adapter}] ${e.base_url}`)); return; }
- const results = await pool(todo, CAP, runVendor);
+ const results = await runPool(todo, CAP, runVendor);
let ok = 0, fail = 0;
for (const r of results) {
- if (r.ok) { ok++; line(` ✓ ${r.vendor}: ${r.count} products (${r.ms}ms)`); }
- else { fail++; line(` ✗ ${r.vendor}: ${r.skipped ? 'SKIP ' : 'FAIL '}${r.reason}`); }
+ if (r.ok) {
+ ok++;
+ const s = r.staged && !r.staged.skipped ? ` [+${r.staged.inserted} new, ~${r.staged.updated} upd, ${r.staged.gone} gone]` : (r.staged && r.staged.skipped ? ` [db-skip: ${r.staged.skipped}]` : '');
+ line(` ✓ ${r.vendor}: ${r.count} products (${r.ms}ms)${s}`);
+ } else { fail++; line(` ✗ ${r.vendor}: ${r.skipped ? 'SKIP ' : 'FAIL '}${r.reason}`); }
}
line(`[${new Date().toISOString()}] done: ${ok} ok, ${fail} fail/skip`);
-})().catch(e => { console.error('FATAL', e); process.exit(1); });
+ await pool.end();
+})().catch(async e => { console.error('FATAL', e); try { await pool.end(); } catch {} process.exit(1); });
diff --git a/lib/write-staging.js b/lib/write-staging.js
new file mode 100644
index 0000000..2a17b12
--- /dev/null
+++ b/lib/write-staging.js
@@ -0,0 +1,53 @@
+// Upsert a vendor's scraped products into the shared scrape_staging table.
+// Change detection: rows whose last_seen predates this run's start are no longer
+// in the feed => "gone" (candidate discontinued). We only REPORT gone; we never
+// delete (a bad scrape shouldn't wipe a catalog — same fail-safe as elsewhere).
+const { pool } = require('./db');
+
+const CHUNK = 500;
+
+async function writeStaging(vendorCode, rawProducts) {
+ if (!rawProducts.length) return { inserted: 0, updated: 0, total: 0, gone: 0, dupes: 0 };
+ // Dedupe by mfr_sku (Shopify feeds reuse/blank SKUs across products) — a single
+ // ON CONFLICT batch can't touch the same key twice. Keep the last occurrence.
+ const byKey = new Map();
+ for (const p of rawProducts) if (p.mfr_sku) byKey.set(p.mfr_sku, p);
+ const products = [...byKey.values()];
+ const dupes = rawProducts.length - products.length;
+ const client = await pool.connect();
+ try {
+ const runStart = new Date();
+ let inserted = 0, updated = 0;
+ for (let i = 0; i < products.length; i += CHUNK) {
+ const batch = products.slice(i, i + CHUNK);
+ const vals = [], params = [];
+ batch.forEach((p, j) => {
+ const b = j * 12;
+ vals.push(`($${b+1},$${b+2},$${b+3},$${b+4},$${b+5},$${b+6},$${b+7},$${b+8},$${b+9},$${b+10},$${b+11},$${b+12},now(),now())`);
+ params.push(vendorCode, p.mfr_sku, p.handle || null, p.title || null, p.product_type || null,
+ p.vendor || null, p.price, p.image_url || null, p.product_url || null,
+ p.tags || null, p.all_images || null, p.raw ? JSON.stringify(p.raw) : null);
+ });
+ // xmax=0 in RETURNING distinguishes INSERT (0) from UPDATE (nonzero).
+ const res = await client.query(`
+ INSERT INTO scrape_staging
+ (vendor_code,mfr_sku,handle,title,product_type,vendor,price,image_url,product_url,tags,all_images,raw,last_seen,scraped_at)
+ VALUES ${vals.join(',')}
+ ON CONFLICT (vendor_code,mfr_sku) DO UPDATE SET
+ handle=EXCLUDED.handle, title=EXCLUDED.title, product_type=EXCLUDED.product_type,
+ vendor=EXCLUDED.vendor, price=EXCLUDED.price, image_url=EXCLUDED.image_url,
+ product_url=EXCLUDED.product_url, tags=EXCLUDED.tags, all_images=EXCLUDED.all_images,
+ raw=EXCLUDED.raw, last_seen=now(), scraped_at=now()
+ RETURNING (xmax = 0) AS is_insert`, params);
+ for (const r of res.rows) r.is_insert ? inserted++ : updated++;
+ }
+ const gone = await client.query(
+ `SELECT count(*)::int c FROM scrape_staging WHERE vendor_code=$1 AND last_seen < $2`,
+ [vendorCode, runStart]);
+ return { inserted, updated, total: products.length, gone: gone.rows[0].c, dupes };
+ } finally {
+ client.release();
+ }
+}
+
+module.exports = { writeStaging };
diff --git a/run-vendor.js b/run-vendor.js
index e4468e0..88e7ee2 100644
--- a/run-vendor.js
+++ b/run-vendor.js
@@ -6,9 +6,14 @@ const fs = require('fs');
const path = require('path');
const ADAPTERS = { shopify: require('./adapters/shopify') };
+const { writeStaging } = require('./lib/write-staging');
const SNAP_DIR = path.join(__dirname, 'data', 'snapshots');
-async function runVendor(entry) {
+// Guard against a broken/partial feed wiping a catalog's freshness: if a scrape
+// returns implausibly few products, snapshot it but SKIP the DB write.
+const MIN_PLAUSIBLE = 3;
+
+async function runVendor(entry, { db = true } = {}) {
const adapter = ADAPTERS[entry.adapter];
if (!adapter) return { vendor: entry.vendor, ok: false, skipped: true, reason: `adapter '${entry.adapter}' not yet implemented` };
const t0 = Date.now();
@@ -18,7 +23,12 @@ async function runVendor(entry) {
const stamp = new Date().toISOString().slice(0, 10).replace(/-/g, '');
const file = path.join(SNAP_DIR, `${entry.vendor_code}-${stamp}.json`);
fs.writeFileSync(file, JSON.stringify({ vendor_code: entry.vendor_code, scraped_at: new Date().toISOString(), count: products.length, products }, null, 0));
- return { vendor: entry.vendor, ok: true, count: products.length, ms: Date.now() - t0, file };
+ let staged = null;
+ if (db) {
+ if (products.length < MIN_PLAUSIBLE) staged = { skipped: `only ${products.length} products — DB write skipped (feed suspect)` };
+ else staged = await writeStaging(entry.vendor_code, products);
+ }
+ return { vendor: entry.vendor, ok: true, count: products.length, ms: Date.now() - t0, file, staged };
} catch (e) {
return { vendor: entry.vendor, ok: false, reason: e.message, ms: Date.now() - t0 };
}
← a454b7e scaffold dw-scrape-scheduler (DTD verdict A): manifest + sho
·
back to Dw Scrape Scheduler
·
auto-save: 2026-07-13T00:51:43 (4 files) — data/manifest.jso 755fba7 →