`lib/write-staging.js`

54 lines
// 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 };