`scripts/pipe-title-backfill/apply-clean-23.mjs`

210 lines
#!/usr/bin/env node
// Apply the 23 CLEAN pipe-title-backfill rows. PostgreSQL-FIRST, then Shopify.
// Steve APPROVED in-session 2026-07-13 (canonical write on the 23 CLEAN rows ONLY).
// - Reads payload apply[] (the 23). Ignores draft[] (the 31 stay staged/untouched).
// - PG: UPDATE title (+ mfr_sku when payload changes it) per row, verify each before moving on.
// - Shopify: product title update + custom.manufacturer_sku / dwc.manufacturer_sku metafieldsSet
//   ONLY when mfr_sku changes (mfr_sku is NOT the variant SKU). No pricing/status/tag/channel/image writes.
// - Batches >= 95s apart per standing bulk cadence. Rate-limit backoff + retry, never abort mid-run.
// Author: steve@designerwallcoverings.com
import pg from 'pg';
import fs from 'node:fs';

const DB = process.env.DATABASE_URL || 'postgresql://stevestudio2@/dw_unified?host=/tmp';
const SHOP = 'designer-laboratory-sandbox.myshopify.com';
const API = '2024-10';
const GQL = `https://${SHOP}/admin/api/${API}/graphql.json`;

// --- token (avoid sourcing the whole .env; it has a parse-error line) ---
function readToken() {
  const envPath = `${process.env.HOME}/Projects/secrets-manager/.env`;
  const line = fs.readFileSync(envPath, 'utf8').split('\n').find(l => /^SHOPIFY_ADMIN_TOKEN=/.test(l));
  if (!line) throw new Error('SHOPIFY_ADMIN_TOKEN not found');
  return line.replace(/^SHOPIFY_ADMIN_TOKEN=/, '').replace(/["']/g, '').trim();
}
const TOKEN = readToken();

const PAYLOAD = new URL('../../../../.claude/yolo-queue/pending-approval/pipe-title-backfill-payload.json', import.meta.url).pathname;
const APPLY_LOG = new URL('./apply-clean-23.result.json', import.meta.url).pathname;

const sleep = (ms) => new Promise(r => setTimeout(r, ms));

async function gql(query, variables, attempt = 0) {
  const res = await fetch(GQL, {
    method: 'POST',
    headers: { 'X-Shopify-Access-Token': TOKEN, 'Content-Type': 'application/json' },
    body: JSON.stringify({ query, variables }),
  });
  if (res.status === 429 || res.status === 502 || res.status === 503) {
    if (attempt >= 6) throw new Error(`Shopify ${res.status} after ${attempt} retries`);
    const wait = Math.min(30000, 2000 * 2 ** attempt);
    console.log(`   [rate/transient ${res.status}] backing off ${wait}ms (attempt ${attempt + 1})`);
    await sleep(wait);
    return gql(query, variables, attempt + 1);
  }
  const json = await res.json();
  // GraphQL-level THROTTLED
  if (json.errors && json.errors.some(e => (e.extensions?.code || '') === 'THROTTLED')) {
    if (attempt >= 6) throw new Error('THROTTLED after retries');
    const wait = Math.min(30000, 2000 * 2 ** attempt);
    console.log(`   [THROTTLED] backing off ${wait}ms (attempt ${attempt + 1})`);
    await sleep(wait);
    return gql(query, variables, attempt + 1);
  }
  return json;
}

const M_PRODUCT_UPDATE = `
mutation($input: ProductInput!) {
  productUpdate(input: $input) {
    product { id title }
    userErrors { field message }
  }
}`;

const M_METAFIELDS_SET = `
mutation($mf: [MetafieldsSetInput!]!) {
  metafieldsSet(metafields: $mf) {
    metafields { namespace key value }
    userErrors { field message }
  }
}`;

const Q_VERIFY = `
query($id: ID!) {
  product(id: $id) {
    id title
    cm: metafield(namespace: "custom", key: "manufacturer_sku") { value }
    dm: metafield(namespace: "dwc", key: "manufacturer_sku") { value }
  }
}`;

async function main() {
  const dry = process.argv.includes('--dry');
  const payload = JSON.parse(fs.readFileSync(PAYLOAD, 'utf8'));
  const rows = payload.apply;
  if (rows.length !== 23) throw new Error(`expected 23 apply rows, got ${rows.length}`);

  // Safety: private-label leak guard — corrected titles must not expose an upstream true-vendor name.
  const LEAK = /\b(command\s*54|wallquest|chesapeake|nextwall|seabrook|brewster|greenland|momentum|versa|desima|carlsten|nicolette\s*mayer)\b/i;
  const leaks = rows.filter(r => LEAK.test(r.after.title));
  if (leaks.length) {
    console.error('LEAK GUARD tripped — STOPPING these rows:', leaks.map(r => r.dw_sku));
    // Continue with the non-leaking rows only.
  }

  const client = new pg.Client({ connectionString: DB });
  await client.connect();

  const results = [];
  const BATCH = 6;
  let batchIdx = 0;

  for (let i = 0; i < rows.length; i++) {
    const r = rows[i];
    if (LEAK.test(r.after.title)) {
      results.push({ id: r.id, dw_sku: r.dw_sku, status: 'SKIPPED_LEAK' });
      continue;
    }

    const gid = r.shopify_id;
    const newTitle = r.after.title;
    const newMfr = r.after.mfr_sku;
    const oldMfr = r.before.mfr_sku;
    const mfrChanges = newMfr !== oldMfr;

    console.log(`\n[${i + 1}/23] id=${r.id} ${r.dw_sku}`);
    console.log(`   title: ${JSON.stringify(r.before.title)} -> ${JSON.stringify(newTitle)}`);
    console.log(`   mfr  : ${JSON.stringify(oldMfr)} -> ${JSON.stringify(newMfr)} ${mfrChanges ? '(WRITE)' : '(unchanged)'}`);

    const rec = { id: r.id, dw_sku: r.dw_sku, shopify_id: gid,
      before: r.before, after: { title: newTitle, mfr_sku: newMfr },
      pg: null, shopify_title: null, shopify_mfr: null };

    // ---------- 1) PostgreSQL FIRST (source of truth) ----------
    // Idempotent guard: only write rows still showing the malformed title.
    const cur = await client.query(
      'SELECT title, mfr_sku, status FROM shopify_products WHERE id=$1', [r.id]);
    if (!cur.rows[0]) { rec.pg = 'MISSING'; results.push(rec); console.log('   PG: MISSING, skip'); continue; }
    const curr = cur.rows[0];
    if (curr.title === newTitle && curr.mfr_sku === newMfr) {
      rec.pg = 'ALREADY_APPLIED';
    } else if (!/^\| /.test(curr.title) && curr.title !== r.before.title) {
      rec.pg = `DRIFT_SKIP(current=${JSON.stringify(curr.title)})`;
      console.log(`   PG: DRIFT — current title ${JSON.stringify(curr.title)} not the malformed 'before'; SKIP`);
      results.push(rec); continue;
    } else {
      if (!dry) {
        await client.query(
          'UPDATE shopify_products SET title=$1, mfr_sku=$2 WHERE id=$3',
          [newTitle, newMfr, r.id]);
      }
      // verify
      const chk = await client.query('SELECT title, mfr_sku FROM shopify_products WHERE id=$1', [r.id]);
      const v = chk.rows[0];
      rec.pg = (dry) ? 'DRY' : ((v.title === newTitle && v.mfr_sku === newMfr) ? 'OK' : `FAIL(${JSON.stringify(v)})`);
      console.log(`   PG: ${rec.pg}`);
      if (!dry && rec.pg !== 'OK') { results.push(rec); continue; } // do not push to Shopify on a failed PG write
    }

    // ---------- 2) Shopify (title + mfr metafields only) ----------
    if (dry) { rec.shopify_title = 'DRY'; rec.shopify_mfr = 'DRY'; results.push(rec); continue; }

    // 2a) title
    const up = await gql(M_PRODUCT_UPDATE, { input: { id: gid, title: newTitle } });
    const ue = up?.data?.productUpdate?.userErrors || [];
    if (ue.length) { rec.shopify_title = `ERR:${JSON.stringify(ue)}`; }
    else { rec.shopify_title = up?.data?.productUpdate?.product?.title === newTitle ? 'OK' : `MISMATCH:${up?.data?.productUpdate?.product?.title}`; }
    console.log(`   Shopify title: ${rec.shopify_title}`);
    await sleep(600);

    // 2b) mfr metafields (only when it changes)
    if (mfrChanges) {
      const mf = [
        { ownerId: gid, namespace: 'custom', key: 'manufacturer_sku', type: 'single_line_text_field', value: newMfr },
        { ownerId: gid, namespace: 'dwc', key: 'manufacturer_sku', type: 'single_line_text_field', value: newMfr },
      ];
      const ms = await gql(M_METAFIELDS_SET, { mf });
      const me = ms?.data?.metafieldsSet?.userErrors || [];
      rec.shopify_mfr = me.length ? `ERR:${JSON.stringify(me)}` : 'OK';
      console.log(`   Shopify mfr metafields: ${rec.shopify_mfr}`);
      await sleep(600);
    } else {
      rec.shopify_mfr = 'UNCHANGED';
    }

    // 2c) live verify
    const vr = await gql(Q_VERIFY, { id: gid });
    const vp = vr?.data?.product;
    rec.shopify_verify = { title: vp?.title, custom_mfr: vp?.cm?.value, dwc_mfr: vp?.dm?.value };
    console.log(`   Shopify verify: title=${JSON.stringify(vp?.title)} custom=${vp?.cm?.value} dwc=${vp?.dm?.value}`);

    results.push(rec);

    // ---------- batch cadence: >=95s gap between batches of 6 ----------
    const done = i + 1;
    if (done % BATCH === 0 && done < rows.length) {
      batchIdx++;
      console.log(`\n=== batch ${batchIdx} done (${done}/23). Cadence gap 95s before next batch ===`);
      await sleep(95000);
    }
  }

  await client.end();

  const out = { applied_at: new Date().toISOString(), dry, count: results.length, results };
  fs.writeFileSync(APPLY_LOG, JSON.stringify(out, null, 2));
  console.log(`\nwrote ${APPLY_LOG}`);

  // summary
  const pgOk = results.filter(x => x.pg === 'OK' || x.pg === 'ALREADY_APPLIED').length;
  const shTitleOk = results.filter(x => x.shopify_title === 'OK').length;
  const shMfrOk = results.filter(x => x.shopify_mfr === 'OK' || x.shopify_mfr === 'UNCHANGED').length;
  console.log(`SUMMARY: PG ok/already=${pgOk}/23  ShopifyTitle ok=${shTitleOk}  ShopifyMfr ok/unchanged=${shMfrOk}`);
  const bad = results.filter(x => !['OK','ALREADY_APPLIED','DRY'].includes(x.pg) ||
    (!dry && x.shopify_title && x.shopify_title !== 'OK') ||
    (!dry && x.shopify_mfr && !['OK','UNCHANGED'].includes(x.shopify_mfr)));
  if (bad.length) console.log('NEEDS ATTENTION:', bad.map(b => `${b.dw_sku}[pg=${b.pg},t=${b.shopify_title},m=${b.shopify_mfr}]`).join('  '));
}
main().catch(e => { console.error('FATAL', e); process.exit(1); });