`sync.js`

144 lines
#!/usr/bin/env node
/**
 * dw-vendor-registry-sync — Option C (DTD verdict, 2026-06-30)
 *
 * Sanctioned, parameterized, NO-shell-out write-back of vendor_registry metadata
 * from the LOCAL Mac2 dw_unified mirror to the CANONICAL Kamatera dw_unified.
 *
 * Why this exists: Mac2 can EDIT vendor_registry but had no sanctioned write-back
 * to canonical Kamatera, so every local edit was a latent split-brain (a
 * Kamatera->Mac2 sync would silently revert it). Raw `ssh my-server psql` is
 * (correctly) hard-blocked by the auto-mode classifier because psql can run
 * arbitrary code (\!, COPY TO PROGRAM). This tool removes that vector:
 *   - reads LOCAL only on a dry run (no remote connection at all)
 *   - on --apply, connects REMOTE and runs ONLY parameterized UPDATEs
 *   - restricts writes to a hard COLUMN WHITELIST (purchasing/pricing metadata)
 *   - never INSERTs vendors, never DROP/TRUNCATE/DELETE, never DDL
 *
 * Usage:
 *   node sync.js --vendors DWCQ-,DWXL-,DWAL-,DWOI-                # dry run (default cols)
 *   node sync.js --vendors DWCQ- --columns buy_from,buy_from_confirmed,buy_from_notes
 *   node sync.js --vendors DWCQ-,DWXL-,DWAL-,DWOI- --apply       # writes to Kamatera
 *
 * Env (.env, gitignored):
 *   LOCAL_DATABASE_URL   default postgresql:///dw_unified?host=/tmp
 *   REMOTE_DATABASE_URL  the canonical Kamatera dw_unified DSN (required for --apply)
 */
const { Client } = require('pg');
require('node:fs');
try { require('dotenv').config(); } catch (_) { /* dotenv optional; env may be preset */ }

// Hard whitelist — the ONLY columns this tool may ever write to canonical prod.
const ALLOWED_COLUMNS = new Set([
  'buy_from', 'buy_from_confirmed', 'buy_from_notes',
  'vendor_discount_pct', 'discount_confirmed', 'discount_confirmed_at', 'discount_notes',
  'pricing_model', 'do_not_price', 'sample_price', 'display_prices',
  'tariff_pct', 'tariff_notes', 'tariff_source', 'tariff_confirmed', 'country_of_origin', 'hts_code',
  'parent_brand', 'is_active',
]);
const DEFAULT_COLUMNS = ['buy_from', 'buy_from_confirmed', 'buy_from_notes'];

function parseArgs(argv) {
  const out = { vendors: [], columns: DEFAULT_COLUMNS, apply: false };
  for (let i = 2; i < argv.length; i++) {
    const a = argv[i];
    if (a === '--apply') out.apply = true;
    else if (a === '--vendors') out.vendors = (argv[++i] || '').split(',').map(s => s.trim()).filter(Boolean);
    else if (a === '--columns') out.columns = (argv[++i] || '').split(',').map(s => s.trim()).filter(Boolean);
    else { console.error(`Unknown arg: ${a}`); process.exit(2); }
  }
  return out;
}

function quoteIdent(id) {
  // identifiers come ONLY from the whitelist, but double-quote defensively anyway
  if (!ALLOWED_COLUMNS.has(id)) throw new Error(`Refusing non-whitelisted column: ${id}`);
  return '"' + id.replace(/"/g, '""') + '"';
}

async function main() {
  const args = parseArgs(process.argv);
  if (!args.vendors.length) { console.error('Need --vendors <sku_prefix-or-vendor_code,...>'); process.exit(2); }

  // validate every requested column against the whitelist BEFORE touching any DB
  const bad = args.columns.filter(c => !ALLOWED_COLUMNS.has(c));
  if (bad.length) { console.error(`Refusing non-whitelisted column(s): ${bad.join(', ')}`); process.exit(2); }

  const localUrl = process.env.LOCAL_DATABASE_URL || 'postgresql:///dw_unified?host=/tmp';
  const selCols = ['vendor_name', 'vendor_code', 'sku_prefix', ...args.columns];
  const local = new Client({ connectionString: localUrl });
  await local.connect();
  const rows = (await local.query(
    `SELECT ${selCols.map(quoteIdentSafe).join(', ')}
       FROM vendor_registry
      WHERE sku_prefix = ANY($1) OR vendor_code = ANY($1) OR sku_prefix = ANY($2)`,
    [args.vendors, args.vendors.map(v => v.endsWith('-') ? v : v + '-')]
  )).rows;
  await local.end();

  if (!rows.length) { console.error('No matching vendor_registry rows for: ' + args.vendors.join(', ')); process.exit(1); }

  console.log(`\nLOCAL source: ${localUrl}`);
  console.log(`Columns to sync: ${args.columns.join(', ')}`);
  console.log(`Matched ${rows.length} vendor(s):\n`);
  for (const r of rows) {
    console.log(`  • ${r.vendor_name}  [${r.sku_prefix} / ${r.vendor_code}]`);
    for (const c of args.columns) console.log(`      ${c} = ${JSON.stringify(r[c])}`);
  }

  if (!args.apply) {
    console.log(`\nDRY RUN — no remote connection made. Re-run with --apply to write these values`);
    console.log(`to the canonical Kamatera dw_unified (REMOTE_DATABASE_URL).`);
    return;
  }

  const remoteUrl = process.env.REMOTE_DATABASE_URL;
  if (!remoteUrl) { console.error('\n--apply needs REMOTE_DATABASE_URL (set it in .env). Aborting.'); process.exit(2); }
  const remote = new Client({ connectionString: remoteUrl, statement_timeout: 15000 });
  await remote.connect();
  console.log(`\nREMOTE target: ${remoteUrl.replace(/:[^:@/]+@/, ':***@')}`);
  let updated = 0;
  try {
    await remote.query('BEGIN');
    for (const r of rows) {
      const sets = args.columns.map((c, i) => `${quoteIdent(c)} = $${i + 1}`);
      const vals = args.columns.map(c => r[c]);
      vals.push(r.vendor_code); // WHERE param
      const res = await remote.query(
        `UPDATE vendor_registry SET ${sets.join(', ')}, updated_at = now()
          WHERE vendor_code = $${vals.length}`,
        vals
      );
      updated += res.rowCount;
      console.log(`  ✓ ${r.vendor_code}: ${res.rowCount} row updated`);
    }
    await remote.query('COMMIT');
  } catch (e) {
    await remote.query('ROLLBACK').catch(() => {});
    await remote.end();
    console.error('\nFAILED — rolled back, no changes written. ' + e.message);
    process.exit(1);
  }
  // verify
  const check = (await remote.query(
    `SELECT vendor_name, ${args.columns.map(quoteIdent).join(', ')}
       FROM vendor_registry WHERE vendor_code = ANY($1) ORDER BY vendor_name`,
    [rows.map(r => r.vendor_code)]
  )).rows;
  await remote.end();
  console.log(`\nApplied ${updated} update(s) to canonical Kamatera. Verified remote state:`);
  for (const r of check) {
    console.log(`  • ${r.vendor_name}`);
    for (const c of args.columns) console.log(`      ${c} = ${JSON.stringify(r[c])}`);
  }
}

// selCols may include the 3 fixed identity cols (not in the write-whitelist) for SELECT only
function quoteIdentSafe(id) {
  const ok = ALLOWED_COLUMNS.has(id) || ['vendor_name', 'vendor_code', 'sku_prefix'].includes(id);
  if (!ok) throw new Error(`Refusing column in SELECT: ${id}`);
  return '"' + id.replace(/"/g, '""') + '"';
}

main().catch(e => { console.error(e); process.exit(1); });