`dwsku-backlog-scan.mjs`

138 lines
#!/usr/bin/env node
// dwsku-backlog-scan.mjs — READ-ONLY scanner for the canonical dw_sku backlog.
//
// Runs the deterministic classifier (classify.mjs) over every ACTIVE product
// whose dw_sku is blank, and reports the A-E / job segmentation plus a plan
// JSONL of (row -> class -> recovered candidate -> collision flag).
//
// SAFETY: this tool issues ONLY SELECT statements. It never UPDATE/INSERT/
// DELETEs, never mints a code, never writes to the database. The single
// artifact it writes is a local plan/summary file under --out.
//
// PORTABILITY: the psql prefix is parameterized so the SAME tool produces the
// canonical number on Kamatera and the mirror number on Mac2:
//   Mac2 mirror (default):  node dwsku-backlog-scan.mjs
//   Kamatera (canonical):   DWSKU_PSQL='ssh <kam> psql' node dwsku-backlog-scan.mjs
//                           (or set DWSKU_PSQL to any psql-compatible prefix)
//
// Parent ticket: TK-10896.

import { execFileSync } from 'node:child_process';
import { writeFileSync, mkdirSync } from 'node:fs';
import { dirname } from 'node:path';
import { classifyRow, RECOVERY_GROUP } from './classify.mjs';

const US = '\x1f'; // ASCII unit separator — safe field delimiter for code columns
const RS = '\x1e'; // ASCII record separator

// psql prefix (everything up to the -c). Default = local Mac2 mirror socket.
const PSQL = (process.env.DWSKU_PSQL || 'psql -h /tmp -d dw_unified').split(/\s+/);

const argv = process.argv.slice(2);
const outArg = argv.indexOf('--out');
const OUT = outArg >= 0 ? argv[outArg + 1] : null;
const emitPlan = argv.includes('--plan'); // also emit per-row plan JSONL next to --out

function psql(sql) {
  // Feed SQL via STDIN, not -c: over `ssh <host> psql ...` the -c argument would be
  // re-parsed by the REMOTE shell and choke on SQL parens/quotes. psql reads stdin
  // when given no -c/-f, which works identically local and over ssh.
  const args = [...PSQL.slice(1), '-tA', '-F', US, '-R', RS];
  const raw = execFileSync(PSQL[0], args, { input: sql, maxBuffer: 1 << 30, encoding: 'utf8' });
  return raw.split(RS).map((r) => r.replace(/\n$/, '')).filter((r) => r.length).map((r) => r.split(US));
}

const BLANK_ACTIVE = `lower(coalesce(status,''))='active' and (dw_sku is null or btrim(dw_sku)='')`;

console.error(`[dwsku-scan] psql prefix: ${PSQL.join(' ')}`);
console.error('[dwsku-scan] reading active dw_sku code set...');
const activeCodeSet = new Set(
  psql(`select distinct dw_sku from shopify_products where lower(coalesce(status,''))='active' and dw_sku is not null and btrim(dw_sku)<>''`)
    .map((r) => r[0])
);
console.error(`[dwsku-scan] active codes: ${activeCodeSet.size}`);

console.error('[dwsku-scan] reading blank-dw_sku active rows...');
const rows = psql(
  `select coalesce(vendor,''), coalesce(sku,''), coalesce(mfr_sku,''), coalesce(dw_sku,''), coalesce(status,'') ` +
  `from shopify_products where ${BLANK_ACTIVE}`
);
console.error(`[dwsku-scan] blank rows: ${rows.length}`);

const byClass = {};
const byGroup = {};
const byVendorRescrape = {};
const planLines = [];
// Provenance cross-tab: among self-copy-recoverable rows, how many ALSO carry a
// real mfr_sku (so the recovered code can be CROSS-VERIFIED against the vendor's
// staging catalog) vs sku-only (no independent check). Higher mfr coverage =
// stronger evidence the code is scraper-assigned, not mint residue.
let selfCopyWithMfr = 0;
let selfCopyNoMfr = 0;

// First pass: count how many blank rows resolve to each self-copy candidate, so
// we can distinguish a candidate shared by a pattern's own sellable+sample rows
// (BENIGN — dw_sku is intentionally non-unique across a pattern's variants) from
// a true cross-product collision (caught separately by the distinct-title test /
// activeCodeSet check). A downstream apply MUST NOT treat dw_sku as a per-row
// unique key; these shared groups are surfaced so it doesn't choke on them.
const candCount = new Map();
for (const [vendor, sku, mfr_sku, dw_sku, status] of rows) {
  const v = classifyRow({ vendor, sku, mfr_sku, dw_sku, status }, activeCodeSet);
  if (v.candidate) candCount.set(v.candidate, (candCount.get(v.candidate) || 0) + 1);
}
let sharedGroupRows = 0;
let sharedGroups = 0;
for (const n of candCount.values()) if (n > 1) { sharedGroups++; sharedGroupRows += n; }

for (const [vendor, sku, mfr_sku, dw_sku, status] of rows) {
  const v = classifyRow({ vendor, sku, mfr_sku, dw_sku, status }, activeCodeSet);
  const sharedCandidate = !!(v.candidate && candCount.get(v.candidate) > 1);
  byClass[v.class] = (byClass[v.class] || 0) + 1;
  const grp = RECOVERY_GROUP[v.class] || 'unknown';
  byGroup[grp] = (byGroup[grp] || 0) + 1;
  if (grp === 'recoverable_now_self_copy') {
    if (mfr_sku && mfr_sku.trim()) selfCopyWithMfr++;
    else selfCopyNoMfr++;
  }
  if (grp === 'rescrape_program_TK10900') {
    byVendorRescrape[vendor] = (byVendorRescrape[vendor] || 0) + 1;
  }
  if (emitPlan) planLines.push(JSON.stringify({ vendor, sku, mfr_sku, class: v.class, candidate: v.candidate, collides: v.collides, shared_pattern_candidate: sharedCandidate, group: grp }));
}

const summary = {
  generated_by: 'dwsku-backlog-scan.mjs',
  ticket: 'TK-10896',
  psql_prefix: PSQL.join(' '),
  source_note: process.env.DWSKU_PSQL ? 'CUSTOM (verify which DB)' : 'Mac2 dw_unified mirror (NOT canonical — Kamatera owns shopify_products)',
  active_code_count: activeCodeSet.size,
  active_blank_dwsku: rows.length,
  by_class: Object.fromEntries(Object.entries(byClass).sort((a, b) => b[1] - a[1])),
  by_recovery_group: Object.fromEntries(Object.entries(byGroup).sort((a, b) => b[1] - a[1])),
  self_copy_provenance: {
    with_real_mfr_sku_cross_verifiable: selfCopyWithMfr,
    sku_only_no_independent_check: selfCopyNoMfr,
  },
  within_batch_shared_candidates: {
    note: 'BENIGN: same-pattern sellable+sample rows resolving to one shared dw_sku (dw_sku is intentionally non-unique per pattern). These are NOT collisions; an apply step must upsert, not assume per-row uniqueness. True cross-product collisions are counted under *_COLLISION classes.',
    shared_candidate_groups: sharedGroups,
    rows_in_shared_groups: sharedGroupRows,
  },
  rescrape_cohorts_top: Object.fromEntries(Object.entries(byVendorRescrape).sort((a, b) => b[1] - a[1]).slice(0, 15)),
};

const out = JSON.stringify(summary, null, 2);
console.log(out);

if (OUT) {
  mkdirSync(dirname(OUT), { recursive: true });
  writeFileSync(OUT, out + '\n');
  console.error(`[dwsku-scan] wrote summary -> ${OUT}`);
  if (emitPlan) {
    const planPath = OUT.replace(/\.json$/, '') + '.plan.jsonl';
    writeFileSync(planPath, planLines.join('\n') + '\n');
    console.error(`[dwsku-scan] wrote plan (${planLines.length} rows) -> ${planPath}`);
  }
}