`apply-plan-gen.mjs`

286 lines
#!/usr/bin/env node
// apply-plan-gen.mjs -- READ-ONLY generator of GATED apply-plan artifacts for the
// canonical dw_sku self-copy backlog (TK-10896).
//
// HARD CONSTRAINTS (do not relax):
//   * NO database writes. NO minting. NO executing any SQL.
//   * Opens ONLY a read-only connection to attach the stable DB key.
//   * Emits per-vendor apply.sql / undo.sql / restore-map.json into apply-plans/.
//     These are DRAFTS a human fires later -- the dw_sku write is Kamatera-canonical
//     dw_unified and is HARD-GATED; this tool NEVER runs apply.sql.
//
// Recovers an EXISTING code already present in `sku` (self-copy). Never mints.

import { readFileSync, writeFileSync, mkdirSync, rmSync, existsSync } from 'node:fs';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { execFileSync } from 'node:child_process';
import { vendorAllowsMfrSelfCopy } from './classify.mjs';

// mfr_sku-sourced recovery classes. These recover dw_sku from mfr_sku and are
// permitted ONLY for verified-real-mfr vendors (Carnegie, Stout). This sku-keyed
// generator does NOT emit SQL for them (they have their own generator), but it
// HARD-ERRORS if a fabricated-mfr self-copy is ever hand-fed in — see the
// allowlist assertion in selectSelfCopyRows. DTD verdict A, 2026-09-02, TK-10900.
export const MFR_SELF_COPY_CLASSES = new Set(['SELF_COPY_MFR', 'SELF_COPY_MFR_COLLISION']);

const HERE = dirname(fileURLToPath(import.meta.url));

// Field separator for the (sku, vendor) map key AND the psql read. A single named
// constant is the only source of truth so load-bearing keys can never drift on an
// invisible-character re-encode. String.fromCharCode(1) = SOH: never in a SKU,
// vendor name, or handle.
const SEP = String.fromCharCode(1);
export const keyOf = (sku, vendor) => sku + SEP + vendor;

// psql prefix (parameterized like the scanner/preflight so this can target the
// canonical Kamatera DB, not just the local mirror):
//   Mac2 mirror (default):  node apply-plan-gen.mjs
//   Kamatera (canonical):   DWSKU_PSQL='ssh root@<kam> psql dw_unified' node apply-plan-gen.mjs
const PSQL = (process.env.DWSKU_PSQL || 'psql -h /tmp -d dw_unified').split(/\s+/);

// Self-copy classes that recover an existing code from `sku`. These + a non-null
// candidate are the ONLY rows that produce SQL. Everything else is excluded.
export const SELF_COPY_CLASSES = new Set([
  'SELF_COPY_DW',
  'SELF_COPY_DW_PROVEN_NATIVE',
  'SELF_COPY_SOURCE',
  'SELF_COPY_CORK',
]);

// Classes that must NEVER leak into an apply plan (asserted below). Note the
// collision classes literally start with "SELF_COPY", so a naive prefix filter
// would leak them -- the allow-set above is exact-match, and this set is the
// belt-and-suspenders guard.
export const FORBIDDEN_CLASSES = new Set([
  'MINT_RESIDUE_RESCRAPE',
  'RESCRAPE',
  'MFR_FABRICATED_RESCRAPE',
  'SELF_COPY_COLLISION',
  'SELF_COPY_DW_COLLISION',
  'SELF_COPY_MFR',            // mfr recovery has its own generator (mfr-selfcopy-gen)
  'SELF_COPY_MFR_COLLISION',
  'STAGING_LINK',
  'PROVENANCE_REVIEW',
]);

// The two sets must be disjoint (an allow class is never a forbidden class).
for (const c of SELF_COPY_CLASSES) {
  if (FORBIDDEN_CLASSES.has(c)) throw new Error(`class ${c} is both allowed and forbidden`);
}

// ---- helpers ---------------------------------------------------------------

function readJsonLines(buffer, label) {
  const text = buffer.toString('utf8').trim();
  if (!text) return [];
  return text.split('\n').map((line, index) => {
    try { return JSON.parse(line); }
    catch (error) { throw new Error(`${label}:${index + 1}: invalid JSON: ${error.message}`); }
  });
}

// SQL single-quote escape (double any apostrophe). Candidates are DW SKU codes;
// this keeps a stray apostrophe from breaking (or injecting into) the statement.
export function sqlEscape(value) {
  return String(value).replace(/'/g, "''");
}

export function vendorSlug(vendor) {
  const slug = String(vendor || 'unknown')
    .toLowerCase()
    .replace(/[^a-z0-9]+/g, '-')
    .replace(/^-+|-+$/g, '');
  return slug || 'unknown';
}

// A candidate must LOOK like a code, not a product title or a garbage-suffixed
// string. classify.mjs's SELF_COPY_SOURCE path admits any non-DW/non-Cork sku
// verbatim, so a leaked title ("Bouncing Bubbles Mural - Cream") or a malformed
// code ("DWKK-152210-Sold Per Bolt (20.5in x 33ft)") could otherwise be written
// into the canonical dw_sku identity field. Reject anything with whitespace,
// parentheses, or over 40 chars — these need re-scrape/manual, never a self-copy.
export const CODE_SHAPE = /^[A-Za-z0-9][A-Za-z0-9._/-]{0,39}$/;

// Keep ONLY self-copy classes with a non-null, code-shaped candidate. Assert no
// forbidden class ever slips through. Returns { kept, rejectedShape }.
export function selectSelfCopyRows(planRows) {
  const kept = [];
  const rejectedShape = [];
  for (const row of planRows) {
    // HARD MACHINE GATE (DTD verdict A, TK-10900): an mfr_sku-sourced self-copy
    // for a vendor NOT on the verified-real allowlist is a fabricated-code
    // laundering attempt. Refuse to proceed at all — even a hand-fed plan
    // physically cannot slip a fabricated mfr_sku into a canonical write.
    if (MFR_SELF_COPY_CLASSES.has(row.class) && !vendorAllowsMfrSelfCopy(row.vendor)) {
      throw new Error(
        `mfr self-copy blocked: vendor '${row.vendor}' is NOT on MFR_SKU_REAL_ALLOWLIST ` +
        `(class ${row.class}, candidate ${row.candidate}). Route to re-scrape; never self-copy a fabricated mfr_sku.`,
      );
    }
    if (FORBIDDEN_CLASSES.has(row.class)) continue; // rescrape / collision / staging / residue / mfr -> never
    if (!SELF_COPY_CLASSES.has(row.class)) continue; // any other class -> no SQL
    const cand = row.candidate == null ? '' : String(row.candidate).trim();
    if (cand === '') continue; // require an existing code
    if (!CODE_SHAPE.test(cand)) { rejectedShape.push(row); continue; } // title/garbage -> not a self-copy
    kept.push(row);
  }
  kept.rejectedShape = rejectedShape;
  return kept;
}

// ---- read-only DB key attachment ------------------------------------------
// One bulk psql read of all still-blank rows. Builds Map key(sku,vendor) ->
// [{ id, shopify_id, handle }]. NEVER opens a write connection.
export function loadBlankKeyMap(psql = defaultPsqlReader) {
  const rows = psql();
  const map = new Map();
  for (const r of rows) {
    const key = keyOf(r.sku, r.vendor);
    if (!map.has(key)) map.set(key, []);
    map.get(key).push({ id: r.id, shopify_id: r.shopify_id, handle: r.handle });
  }
  return map;
}

function defaultPsqlReader() {
  // Read-only SELECT. No write clause anywhere. Emits SEP-delimited rows for
  // every product whose dw_sku is STILL blank (idempotence at read time).
  const sql =
    "SELECT id, coalesce(shopify_id,''), coalesce(handle,''), coalesce(sku,''), coalesce(vendor,'') " +
    "FROM shopify_products WHERE dw_sku IS NULL OR btrim(dw_sku)='';";
  // Uses DWSKU_PSQL (default local mirror) + feeds SQL via STDIN, not -c, so it
  // works identically local and over `ssh <host> psql ...` (no remote-shell quoting).
  const out = execFileSync(PSQL[0], [...PSQL.slice(1), '-tA', '-F', SEP], {
    input: sql,
    maxBuffer: 1024 * 1024 * 512,
    encoding: 'utf8',
  });
  const rows = [];
  for (const line of out.split('\n')) {
    if (!line) continue;
    const [id, shopify_id, handle, sku, vendor] = line.split(SEP);
    rows.push({ id: Number(id), shopify_id, handle, sku, vendor });
  }
  return rows;
}

// ---- core plan build -------------------------------------------------------

export function buildPlans(planRows, blankKeyMap) {
  const selfCopy = selectSelfCopyRows(planRows);
  const byVendor = new Map(); // vendor -> [{ id, shopify_id, handle, candidate }]
  let unmatched = 0;
  let noShopifyId = 0; // rows we cannot safely target on Kamatera (missing stable key)
  const unmatchedRows = [];

  for (const row of selfCopy) {
    const dbRows = blankKeyMap.get(keyOf(row.sku, row.vendor));
    if (!dbRows || dbRows.length === 0) { unmatched += 1; unmatchedRows.push(row); continue; }
    if (!byVendor.has(row.vendor)) byVendor.set(row.vendor, []);
    const bucket = byVendor.get(row.vendor);
    for (const db of dbRows) {
      // shopify_id is the stable, machine-independent key. A row missing it cannot
      // be safely targeted on the canonical Kamatera DB -> exclude + report.
      if (!db.shopify_id || String(db.shopify_id).trim() === '') { noShopifyId += 1; continue; }
      bucket.push({ id: db.id, shopify_id: db.shopify_id, handle: db.handle, candidate: String(row.candidate) });
    }
  }
  return { byVendor, unmatched, unmatchedRows, noShopifyId, selfCopyCount: selfCopy.length, rejectedShape: selfCopy.rejectedShape || [] };
}

const HEADER =
  '-- GATED -- canonical Kamatera dw_unified write. Do NOT run automatically.\n' +
  '-- Recovers existing code (no mint). Undo via undo.sql / restore-map.json. TK-10896.\n';

// Key on shopify_id (the stable Shopify GID, identical on Mac2 + Kamatera), NOT
// the local `id` (a per-machine nextval serial that maps to a DIFFERENT product
// on the canonical Kamatera DB). The blank-guard keeps a re-run / concurrent fill
// from clobbering.
function applyStmt(shopify_id, candidate) {
  const c = sqlEscape(candidate);
  const sid = sqlEscape(shopify_id);
  return `UPDATE shopify_products SET dw_sku='${c}' WHERE shopify_id='${sid}' AND (dw_sku IS NULL OR btrim(dw_sku)='');`;
}

function undoStmt(shopify_id, candidate) {
  const c = sqlEscape(candidate);
  const sid = sqlEscape(shopify_id);
  return `UPDATE shopify_products SET dw_sku=NULL WHERE shopify_id='${sid}' AND dw_sku='${c}';`;
}

export function renderApplySql(entries) {
  return HEADER + entries.map((e) => applyStmt(e.shopify_id, e.candidate)).join('\n') + '\n';
}

export function renderUndoSql(entries) {
  return HEADER + entries.map((e) => undoStmt(e.shopify_id, e.candidate)).join('\n') + '\n';
}

export function renderRestoreMap(entries) {
  return entries.map((e) => ({
    id: e.id,
    shopify_id: e.shopify_id,
    handle: e.handle,
    column: 'dw_sku',
    // old is the pre-apply canonical value: blank (NULL or ''); undo restores to
    // NULL. null (not '') is the honest record — undo targets `dw_sku='<new>'` so
    // it only ever reverts what this apply set.
    old: null,
    new: e.candidate,
  }));
}

// ---- writer ----------------------------------------------------------------

export function writePlans({ byVendor, unmatched, selfCopyCount, noShopifyId = 0, rejectedShape = [] }, outDir) {
  if (existsSync(outDir)) rmSync(outDir, { recursive: true, force: true });
  mkdirSync(outDir, { recursive: true });

  const perVendor = {};
  let grandTotal = 0;
  for (const [vendor, entries] of [...byVendor.entries()].sort((a, b) => a[0].localeCompare(b[0]))) {
    const slug = vendorSlug(vendor);
    const dir = join(outDir, slug);
    mkdirSync(dir, { recursive: true });
    writeFileSync(join(dir, 'restore-map.json'), JSON.stringify(renderRestoreMap(entries), null, 2) + '\n');
    writeFileSync(join(dir, 'apply.sql'), renderApplySql(entries));
    writeFileSync(join(dir, 'undo.sql'), renderUndoSql(entries));
    perVendor[vendor] = { slug, statements: entries.length };
    grandTotal += entries.length;
  }

  const summary = {
    ticket: 'TK-10896',
    note: 'NOTHING was executed. apply.sql files are GATED drafts for a human to run. dw_sku is Kamatera-canonical dw_unified -- hard-gated.',
    generated_at: new Date().toISOString(),
    key_column: 'shopify_id',
    self_copy_plan_rows: selfCopyCount,
    vendors: Object.keys(perVendor).length,
    grand_total_statements: grandTotal,
    unmatched_plan_rows: unmatched,
    excluded_missing_shopify_id: noShopifyId,
    excluded_bad_candidate_shape: rejectedShape.length,
    per_vendor: perVendor,
  };
  writeFileSync(join(outDir, 'SUMMARY.json'), JSON.stringify(summary, null, 2) + '\n');
  return summary;
}

// ---- CLI -------------------------------------------------------------------

function arg(name, fallback = null) {
  const index = process.argv.indexOf(name);
  return index >= 0 ? process.argv[index + 1] : fallback;
}

if (process.argv[1] && import.meta.url === new URL(`file://${process.argv[1]}`).href) {
  const planPath = arg('--plan', '/tmp/joined3.jsonl');
  const outDir = arg('--out', join(HERE, 'apply-plans'));
  const planRows = readJsonLines(readFileSync(planPath), 'plan');
  const blankKeyMap = loadBlankKeyMap();
  const plans = buildPlans(planRows, blankKeyMap);
  const summary = writePlans(plans, outDir);
  process.stdout.write(JSON.stringify({ ok: true, out: outDir, ...summary }, null, 2) + '\n');
}