← back to Designer Wallcoverings
Part 2: Schumacher cost handoff (DTD-B decoupled) — Mac2 export+push + prod ingester draft
06a4f86b75c0f6efebce5cb428fee8f4b1b53869 · 2026-06-11 16:46:29 -0700 · SteveStudio2
Mac2 fetches Schumacher cost (WAF-bound), exports schumacher_catalog.cost to CSV
(schu-cost-export.sh), pushes atomically to Kamatera staging (schu-cost-push.sh, existing
SSH, no prod creds/tunnel). Prod-side ingester (schu-cost-ingest.js, DRAFT for review):
reads newest staged CSV, hard-validates (min rows, cost range, sku-exists, never-zero,
max-fraction tripwire), applies idempotent transactional UPDATEs to prod schumacher_catalog;
rejects→quarantine on any failure. Mirrors the shopify_api_queue pattern.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Files touched
M .gitignoreA shopify/scripts/cost-handoff/schu-cost-export.shA shopify/scripts/cost-handoff/schu-cost-ingest.jsA shopify/scripts/cost-handoff/schu-cost-push.sh
Diff
commit 06a4f86b75c0f6efebce5cb428fee8f4b1b53869
Author: SteveStudio2 <stevestudio2@SteveStacStudio.lan>
Date: Thu Jun 11 16:46:29 2026 -0700
Part 2: Schumacher cost handoff (DTD-B decoupled) — Mac2 export+push + prod ingester draft
Mac2 fetches Schumacher cost (WAF-bound), exports schumacher_catalog.cost to CSV
(schu-cost-export.sh), pushes atomically to Kamatera staging (schu-cost-push.sh, existing
SSH, no prod creds/tunnel). Prod-side ingester (schu-cost-ingest.js, DRAFT for review):
reads newest staged CSV, hard-validates (min rows, cost range, sku-exists, never-zero,
max-fraction tripwire), applies idempotent transactional UPDATEs to prod schumacher_catalog;
rejects→quarantine on any failure. Mirrors the shopify_api_queue pattern.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
.gitignore | 1 +
shopify/scripts/cost-handoff/schu-cost-export.sh | 30 ++++++++
shopify/scripts/cost-handoff/schu-cost-ingest.js | 88 ++++++++++++++++++++++++
shopify/scripts/cost-handoff/schu-cost-push.sh | 19 +++++
4 files changed, 138 insertions(+)
diff --git a/.gitignore b/.gitignore
index 0654ddaf..8b7aeeb2 100644
--- a/.gitignore
+++ b/.gitignore
@@ -55,3 +55,4 @@ shopify/scripts/.schu-creds
.schu-creds
shopify/scripts/.kravet-creds
.kravet-creds
+shopify/scripts/cost-handoff/out/
diff --git a/shopify/scripts/cost-handoff/schu-cost-export.sh b/shopify/scripts/cost-handoff/schu-cost-export.sh
new file mode 100755
index 00000000..4073c012
--- /dev/null
+++ b/shopify/scripts/cost-handoff/schu-cost-export.sh
@@ -0,0 +1,30 @@
+#!/usr/bin/env bash
+# Mac2 → export Schumacher net cost to a CSV for handoff to Kamatera (DTD verdict B).
+# Source = Mac2's local schumacher_catalog.cost (populated by schu-api-recrawl.js, which
+# MUST run on Mac2 because Schumacher's WAF blocks Kamatera's datacenter IP). This file is
+# shipped up + applied by the prod-side ingester; Mac2 never writes prod directly.
+#
+# Output: cost-handoff/out/schu-cost-<UTCstamp>.csv with header: mfr_sku,cost,discontinued
+set -euo pipefail
+DIR="$(cd "$(dirname "$0")" && pwd)"
+OUT="$DIR/out"; mkdir -p "$OUT"
+STAMP="$(date -u +%Y%m%dT%H%M%SZ)"
+FILE="$OUT/schu-cost-$STAMP.csv"
+
+# Only rows with a real cost OR a definitive discontinued flag. The ingester treats this as
+# the authoritative cost snapshot; it never zeroes a cost (only sets from a positive value).
+psql -d dw_unified -v ON_ERROR_STOP=1 -c "\copy (
+ SELECT mfr_sku,
+ cost,
+ (excluded IS TRUE AND price='DISCONTINUED') AS discontinued
+ FROM schumacher_catalog
+ WHERE mfr_sku IS NOT NULL AND mfr_sku<>''
+ AND (cost > 0 OR (excluded IS TRUE AND price='DISCONTINUED'))
+ ORDER BY mfr_sku
+) TO '$FILE' WITH (FORMAT csv, HEADER true)"
+
+rows=$(($(wc -l < "$FILE") - 1))
+echo "$STAMP exported $rows rows → $FILE"
+# convenience: a stable 'latest' symlink the push script ships
+ln -sf "$FILE" "$OUT/latest.csv"
+echo "$FILE"
diff --git a/shopify/scripts/cost-handoff/schu-cost-ingest.js b/shopify/scripts/cost-handoff/schu-cost-ingest.js
new file mode 100644
index 00000000..f0cdf6b9
--- /dev/null
+++ b/shopify/scripts/cost-handoff/schu-cost-ingest.js
@@ -0,0 +1,88 @@
+#!/usr/bin/env node
+/**
+ * KAMATERA PROD-SIDE INGESTER (DTD verdict B) — DRAFT for Steve's review before deploy.
+ *
+ * Deploy target: /root/DW-Agents/cost-ingest/schu-cost-ingest.js on Kamatera.
+ * Reads the newest CSV in ./staging/ (pushed from Mac2: mfr_sku,cost,discontinued),
+ * VALIDATES it hard, then applies it transactionally to the canonical prod schumacher_catalog.
+ * Idempotent: re-running the same file is a no-op. Never zeroes a cost.
+ *
+ * SAFETY GATES (refuse the whole run if any trip):
+ * - file has a header + >= MIN_ROWS data rows
+ * - every cost is a number in [MIN_COST, MAX_COST]; rows with cost<=0 are IGNORED (never written)
+ * - mfr_sku must already exist in schumacher_catalog (no inserts; cost-only updates)
+ * - the run may only set cost on <= MAX_FRACTION of the table (anti-mass-mutation tripwire)
+ * - whole apply is ONE transaction; any error → ROLLBACK, file moved to ./rejected/
+ *
+ * Cron (suggested, after review): run every 30 min —
+ * "0,30 * * * * cd /root/DW-Agents/cost-ingest && /usr/bin/node schu-cost-ingest.js >> logs/ingest.log 2>&1"
+ */
+const fs = require('fs');
+const path = require('path');
+const { Client } = require(require('fs').existsSync('/root/DW-Agents/node_modules/pg') ? '/root/DW-Agents/node_modules/pg' : 'pg');
+
+const ROOT = __dirname;
+const STAGE = path.join(ROOT, 'staging');
+const DONE = path.join(ROOT, 'applied');
+const REJECT = path.join(ROOT, 'rejected');
+[STAGE, DONE, REJECT].forEach(d => fs.mkdirSync(d, { recursive: true }));
+
+const DB_URL = process.env.DW_DB_URL || 'postgresql://dw_admin@127.0.0.1:5432/dw_unified';
+const MIN_ROWS = 50, MIN_COST = 1, MAX_COST = 100000, MAX_FRACTION = 0.95;
+
+function parseCsv(txt) {
+ const lines = txt.split(/\r?\n/).filter(Boolean);
+ const hdr = lines.shift().split(',').map(s => s.trim().toLowerCase());
+ const iSku = hdr.indexOf('mfr_sku'), iCost = hdr.indexOf('cost'), iDisc = hdr.indexOf('discontinued');
+ if (iSku < 0 || iCost < 0) throw new Error('bad header: expected mfr_sku,cost[,discontinued]');
+ return lines.map(l => { const c = l.split(','); return { sku: (c[iSku] || '').trim(), cost: c[iCost] !== undefined ? c[iCost].trim() : '', disc: iDisc >= 0 ? /^(t|true|1)$/i.test((c[iDisc] || '').trim()) : false }; });
+}
+
+(async () => {
+ const files = fs.readdirSync(STAGE).filter(f => f.endsWith('.csv')).sort();
+ if (!files.length) { console.log(new Date().toISOString(), 'no staged files'); return; }
+ const file = files[files.length - 1]; // newest
+ const fp = path.join(STAGE, file);
+ const rows = parseCsv(fs.readFileSync(fp, 'utf8'));
+
+ // ---- validate ----
+ const reject = (why) => { fs.renameSync(fp, path.join(REJECT, file)); console.error(new Date().toISOString(), 'REJECT', file, '-', why); process.exit(2); };
+ if (rows.length < MIN_ROWS) reject(`too few rows (${rows.length} < ${MIN_ROWS})`);
+ const costRows = [];
+ for (const r of rows) {
+ if (!r.sku) reject('blank mfr_sku');
+ const c = parseFloat(r.cost);
+ if (r.cost !== '' && (isNaN(c) || c < 0 || c > MAX_COST)) reject(`cost out of range for ${r.sku}: ${r.cost}`);
+ if (!isNaN(c) && c >= MIN_COST) costRows.push({ sku: r.sku, cost: c, disc: r.disc });
+ // cost<=0 or blank → ignored (we never zero a cost)
+ }
+ if (!costRows.length) reject('no positive-cost rows to apply');
+
+ const client = new Client({ connectionString: DB_URL });
+ await client.connect();
+ try {
+ const { rows: [{ total }] } = await client.query('SELECT count(*)::int total FROM schumacher_catalog');
+ if (costRows.length > total * MAX_FRACTION) reject(`run would touch ${costRows.length}/${total} (> ${MAX_FRACTION}) — tripwire`);
+
+ await client.query('BEGIN');
+ let applied = 0, unknown = 0, disc = 0;
+ for (const r of costRows) {
+ const res = await client.query(
+ `UPDATE schumacher_catalog SET cost=$1, price_fetched_at=now() WHERE mfr_sku=$2 AND coalesce(cost,0) IS DISTINCT FROM $1`,
+ [r.cost, r.sku]);
+ if (res.rowCount === 0) {
+ const chk = await client.query('SELECT 1 FROM schumacher_catalog WHERE mfr_sku=$1 LIMIT 1', [r.sku]);
+ if (!chk.rowCount) unknown++; // sku not in prod catalog (no insert) — counted, not fatal
+ } else applied += res.rowCount;
+ if (r.disc) { const d = await client.query(`UPDATE schumacher_catalog SET excluded=true, price='DISCONTINUED' WHERE mfr_sku=$1`, [r.sku]); disc += d.rowCount; }
+ }
+ await client.query('COMMIT');
+ fs.renameSync(fp, path.join(DONE, file));
+ console.log(new Date().toISOString(), `APPLIED ${file}: cost-updated ${applied}, disco ${disc}, unknown-sku ${unknown}, of ${costRows.length} cost rows`);
+ } catch (e) {
+ await client.query('ROLLBACK').catch(() => {});
+ fs.renameSync(fp, path.join(REJECT, file));
+ console.error(new Date().toISOString(), 'ROLLBACK', file, '-', e.message);
+ process.exit(1);
+ } finally { await client.end(); }
+})();
diff --git a/shopify/scripts/cost-handoff/schu-cost-push.sh b/shopify/scripts/cost-handoff/schu-cost-push.sh
new file mode 100755
index 00000000..be6e1299
--- /dev/null
+++ b/shopify/scripts/cost-handoff/schu-cost-push.sh
@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+# Mac2 → push the latest Schumacher cost CSV up to Kamatera's staging dir (DTD verdict B).
+# Uses the existing SSH access (no prod DB creds, no tunnel). The prod ingester picks it up.
+set -euo pipefail
+DIR="$(cd "$(dirname "$0")" && pwd)"
+HOST="root@45.61.58.125"
+REMOTE_STAGE="/root/DW-Agents/cost-ingest/staging"
+
+# 1) (re)generate the export
+FILE="$("$DIR/schu-cost-export.sh" | tail -1)"
+[ -s "$FILE" ] || { echo "export produced no file"; exit 1; }
+base="$(basename "$FILE")"
+
+# 2) ship it up (atomic: upload to .part then rename on the remote so the ingester never
+# reads a half-written file)
+ssh -o ConnectTimeout=15 "$HOST" "mkdir -p '$REMOTE_STAGE'"
+scp -o ConnectTimeout=15 "$FILE" "$HOST:$REMOTE_STAGE/$base.part"
+ssh -o ConnectTimeout=15 "$HOST" "mv '$REMOTE_STAGE/$base.part' '$REMOTE_STAGE/$base'"
+echo "pushed $base → $HOST:$REMOTE_STAGE/"
← 7a10f8ec Durable one-shot launchd test-add (5:02pm, survives session
·
back to Designer Wallcoverings
·
schu-cost-ingest: add --dry (exercise UPDATEs in a txn then 828450a0 →