`generate-manifests.js`

166 lines
#!/usr/bin/env node
// CADENCE MANIFEST GENERATOR — fills data/launch-batches/ with small, ready-to-run
// batches for the staggered 1,000-published-SKUs/24h cadence (default 42 rows each,
// 42/hr × 24 ≈ 1,008/day).
//
// WHY small batches: the existing 1000-row manifests are burst-shaped; they trip
// Shopify's daily VARIANT-creation limit when drained fast. The cadence wrapper
// (run-cadence.sh, --limit 42) consumes ONE batch worth per hour. consumer.js
// already dedups across manifests by id, so manifest size is purely a cadence knob.
//
// SELECTION — mirrors consumer.js eligibility EXACTLY so a manifested row will
// actually CREATE, not get skipped at consume time:
//   • net-new (sync_status='new' OR not-on-shopify) with mfr_sku
//   • THREE-SOURCE NEVER-DUPLICATE: mfr_sku not in shopify_products AND not in
//     dw_sku_registry (the same NETNEW_PRED server.js uses)
//   • image-bearing (image_url present) — hard rule, no imageless products
//   • dw_sku + pattern_name present
//   • vendor NOT denylisted (cowtan_tout), NOT on-hold (command54),
//     NOT skip_shopify, and if private-label MUST have a mask name (else the
//     consumer skips it to avoid leaking the real vendor)
//   • OLDEST-FIRST within each vendor (lowest vc.id = scraped earliest)
//   • ROUND-ROBIN across vendors so a single mega-vendor can't monopolize a day
//
// IDEMPOTENT: skips ids already in the ledger as 'created' AND ids already present
// in existing launch-*.json manifests on disk, so re-running tops up rather than
// duplicating. Cadence-tagged manifests are named launch-cad-<n>-<seq>.json.
//
// Usage:
//   node generate-manifests.js                 # default: 24 batches × 42 = 1008 rows
//   node generate-manifests.js --batches 24 --size 42
//   node generate-manifests.js --batches 4 --size 42 --dry   # plan only, write nothing
const fs = require('fs');
const path = require('path');
const { spawnSync } = require('child_process');

const ARGS = process.argv.slice(2);
const opt = (n, d) => { const i = ARGS.indexOf(n); return i >= 0 ? ARGS[i + 1] : d; };
const flag = (n) => ARGS.includes(n);

const PSQL = process.env.PSQL || '/opt/homebrew/opt/postgresql@14/bin/psql';
const PGUSER = process.env.PGUSER || 'stevestudio2';
const PGDB = process.env.PGDATABASE || 'dw_unified';
const US = '\x1f', RS = '\x1e';
function q(sql) {
  const r = spawnSync(PSQL, ['-U', PGUSER, '-d', PGDB, '-tA', '-F', US, '-R', RS, '-c', sql],
    { encoding: 'utf8', timeout: 180000, maxBuffer: 1 << 28 });
  if (r.status !== 0) throw new Error((r.stderr || 'psql failed').slice(0, 600));
  return (r.stdout || '').replace(/\n+$/, '').split(RS).filter(l => l.length).map(l => l.split(US));
}

const BATCHES = parseInt(opt('--batches', '24'), 10) || 24;
const SIZE = parseInt(opt('--size', '42'), 10) || 42;
const DRY = flag('--dry');
const TARGET = BATCHES * SIZE;
const DIR = path.join(__dirname, 'data', 'launch-batches');
const LEDGER = path.join(__dirname, 'data', 'launch-consumer-ledger.jsonl');

// ── exclusion set: ids already created OR already manifested ──
function alreadySeenIds() {
  const seen = new Set();
  // ledger 'created'
  try {
    for (const line of fs.readFileSync(LEDGER, 'utf8').split('\n')) {
      if (!line.trim()) continue;
      try { const e = JSON.parse(line); if (e.id != null && e.status === 'created') seen.add(+e.id); } catch {}
    }
  } catch {}
  // existing manifests (any launch-*.json) — don't re-stage what's queued
  let manifested = 0;
  try {
    for (const f of fs.readdirSync(DIR)) {
      if (!f.startsWith('launch-') || !f.endsWith('.json')) continue;
      try {
        const m = JSON.parse(fs.readFileSync(path.join(DIR, f), 'utf8'));
        if (m && Array.isArray(m.skus)) for (const s of m.skus) { if (s.id != null) { seen.add(+s.id); manifested++; } }
      } catch {}
    }
  } catch {}
  return { seen, manifested };
}

// ── candidate pool, OLDEST-FIRST per vendor, consumer-eligible only ──
// Window-numbered by vendor so we can round-robin in JS. We pull a generous
// multiple of TARGET so the exclusion filter still leaves enough.
function pullPool(excludeIds, want) {
  const excl = excludeIds.size ? `AND vc.id NOT IN (${[...excludeIds].join(',')})` : '';
  // Pull up to want*4 oldest eligible rows per the consumer's full gate set.
  const rows = q(`
    WITH elig AS (
      SELECT vc.id, vc.vendor_code, vc.mfr_sku, vc.dw_sku, vc.pattern_name,
             ROW_NUMBER() OVER (PARTITION BY vc.vendor_code ORDER BY vc.id ASC) AS rn
      FROM vendor_catalog vc
      JOIN vendor_registry vr ON vr.vendor_code = vc.vendor_code
      WHERE (vc.sync_status='new' OR ((vc.on_shopify IS NOT TRUE) AND vc.shopify_product_id IS NULL))
        -- belt: a stale sync_status='new' can coexist with on_shopify=TRUE / a real
        -- product id (seen on rebel_walls). The consumer skips these as 'already
        -- linked' at consume time, but excluding them here keeps cadence slots full.
        AND vc.shopify_product_id IS NULL AND vc.on_shopify IS NOT TRUE
        AND vc.mfr_sku IS NOT NULL AND vc.mfr_sku <> ''
        AND vc.image_url IS NOT NULL AND vc.image_url <> ''
        AND vc.dw_sku IS NOT NULL AND vc.pattern_name IS NOT NULL
        AND vc.vendor_code NOT IN ('cowtan_tout','command54')          -- deny + hold
        AND COALESCE(vr.skip_shopify, FALSE) = FALSE                    -- skip_shopify
        AND NOT (vr.is_private_label = TRUE                             -- PL w/ no mask = leak risk
                 AND (vr.private_label_name IS NULL OR trim(vr.private_label_name)=''))
        AND NOT EXISTS (SELECT 1 FROM shopify_products sx WHERE upper(trim(sx.mfr_sku)) = upper(trim(vc.mfr_sku)))
        AND NOT EXISTS (SELECT 1 FROM dw_sku_registry rx WHERE upper(trim(rx.mfr_sku)) = upper(trim(vc.mfr_sku)))
        ${excl}
    )
    SELECT id, vendor_code, mfr_sku, dw_sku, pattern_name, rn
    FROM elig
    WHERE rn <= ${Math.ceil(want)}            -- cap per-vendor depth so RR is fair
    ORDER BY rn ASC, vendor_code ASC          -- rn-major = round-robin across vendors
  `);
  return rows.map(r => ({
    id: +r[0], vendor: r[1], sku: r[2], dwSku: r[3], pattern: r[4], rn: +r[5],
  }));
}

(function main() {
  if (!fs.existsSync(DIR)) fs.mkdirSync(DIR, { recursive: true });
  const { seen, manifested } = alreadySeenIds();
  console.log(`exclusion set: ${seen.size.toLocaleString()} ids (${manifested.toLocaleString()} already in on-disk manifests, rest ledger-created)`);

  // rn<=ceil(TARGET/vendorCount) would need vendor count; simpler: cap per-vendor
  // depth at TARGET (a single vendor can at most fill the whole day if it's the
  // only one). Pool is already rn-major ordered = round-robin.
  const pool = pullPool(seen, TARGET);
  console.log(`eligible pool (oldest-first, round-robin): ${pool.length.toLocaleString()} rows available`);

  const pick = pool.slice(0, TARGET);
  if (pick.length < TARGET) {
    console.warn(`  note: only ${pick.length} eligible rows after exclusions (wanted ${TARGET}) — generating ${Math.ceil(pick.length / SIZE)} batch(es)`);
  }

  // vendor distribution of the pick
  const dist = {};
  for (const p of pick) dist[p.vendor] = (dist[p.vendor] || 0) + 1;
  console.log(`pick = ${pick.length} rows across ${Object.keys(dist).length} vendors; top:`,
    Object.entries(dist).sort((a, b) => b[1] - a[1]).slice(0, 8).map(([v, n]) => `${v}:${n}`).join(' '));

  const nBatches = Math.ceil(pick.length / SIZE);
  const now = new Date();
  const written = [];
  for (let b = 0; b < nBatches; b++) {
    const slice = pick.slice(b * SIZE, (b + 1) * SIZE);
    if (!slice.length) break;
    // stagger the 'at' timestamps 1 hour apart so consumer's oldest-first manifest
    // ordering drains them in cadence order even if run out of band.
    const at = new Date(now.getTime() + b * 3600 * 1000).toISOString();
    const seq = Math.random().toString(36).slice(2, 10);
    const batchId = `launch-cad-${slice.length}-${seq}`;
    const manifest = {
      batchId, action: 'launch', mode: 'active', cadence: true,
      at, count: slice.length,
      skus: slice.map(s => ({ id: s.id, vendor: s.vendor, sku: s.sku, dwSku: s.dwSku, pattern: s.pattern, from: 'CADENCE' })),
    };
    const file = path.join(DIR, `${batchId}.json`);
    if (DRY) { console.log(`  [dry] would write ${path.basename(file)} (${slice.length} rows, at ${at})`); }
    else { fs.writeFileSync(file, JSON.stringify(manifest, null, 1)); written.push(path.basename(file)); }
  }

  console.log(`\n${DRY ? 'DRY-RUN — wrote nothing' : `wrote ${written.length} cadence manifest(s)`} · ${pick.length} rows · ${SIZE}/batch`);
  if (!DRY && written.length) console.log('  first/last:', written[0], '…', written[written.length - 1]);
})();