`data/tk11050-item3/backfill-one.sh`

69 lines
#!/usr/bin/env bash
# TK-11050 ITEM 3 — additive Kamatera->Mac2 backfill for ONE *_catalog table.
# Inserts ONLY the genuinely-missing rows (distinct mfr_sku present on Kamatera, absent on Mac2).
# Reversible: snapshots the Mac2 table first. Run AFTER audit-distinct-sku-gap.sh. Steve-gated.
set -euo pipefail
T="${1:?usage: backfill-one.sh <table>}"
case "$T" in (*[!a-z0-9_]*|''|*_catalog_prebackfill_bak) echo "Invalid catalog table: $T" >&2; exit 2;; esac
case "$T" in (*_catalog) :;; (*) echo "Refusing non-catalog table: $T" >&2; exit 2;; esac
case "$T" in
  armani_casa_catalog)
    echo "Refusing excluded catalog: $T (Steve, 2026-09-01)" >&2; exit 2;;
esac
# Carlisle is staging/crawl-only by Steve directive (2026-09-01). This loader writes only the
# local catalog table; it must never be composed with Shopify onboarding or publish actions.
KAM="root@45.61.58.125"
KPSQL="psql -h 127.0.0.1 -U dw_admin -d dw_unified -tA"
WORK=/tmp/tk11050-item3
KON="$WORK/kamonly.$T"
[ -s "$KON" ] || { echo "No Kamatera-only SKUs for $T (nothing to backfill)."; exit 0; }
N=$(wc -l < "$KON" | tr -d ' ')
[ "$N" -le 500 ] || { echo "Refusing $N-row target; bounded auto-exec cap is 500." >&2; exit 2; }
[ -f "$WORK/mac2only.$T" ] || { echo "Missing audit artifact for $T; rerun audit first." >&2; exit 2; }
[ ! -s "$WORK/mac2only.$T" ] || { echo "Refusing two-way divergence for $T; reconcile manually." >&2; exit 2; }
# Shared column set (exclude id so Mac2 assigns fresh serials; avoids PK/serial collision)
psql -h /tmp -d dw_unified -tA -c "SELECT column_name FROM information_schema.columns WHERE table_schema='public' AND table_name='$T' AND column_name<>'id' ORDER BY column_name" | sort -u > "$WORK/m2cols.$T"
ssh -n "$KAM" "$KPSQL -v ON_ERROR_STOP=1 -c \"SELECT column_name FROM information_schema.columns WHERE table_schema='public' AND table_name='$T' AND column_name<>'id' ORDER BY column_name\"" | sort -u > "$WORK/kcols.$T"
comm -12 "$WORK/m2cols.$T" "$WORK/kcols.$T" | paste -sd, - > "$WORK/cols.$T"
COLS=$(cat "$WORK/cols.$T")
echo "$T: backfilling $N missing SKUs on shared cols: $COLS"
# 1) Load kamonly SKU list onto Kamatera into a temp, dump exactly one row per mfr_sku.
# Use a session-scoped approach via a single psql invocation piping the file in:
ssh "$KAM" "psql -q -h 127.0.0.1 -U dw_admin -d dw_unified -v ON_ERROR_STOP=1 <<'REMOTE'
CREATE TEMP TABLE _kon(mfr_sku text);
\copy _kon FROM STDIN
$(cat "$KON")
\.
\copy (SELECT $COLS FROM (SELECT DISTINCT ON (mfr_sku) * FROM $T t WHERE t.mfr_sku IN (SELECT mfr_sku FROM _kon) ORDER BY mfr_sku, id) t) TO STDOUT WITH CSV
REMOTE" > "$WORK/rows.$T.csv"
ROWS=$(wc -l < "$WORK/rows.$T.csv" | tr -d ' ')
echo "$T: dumped CSV payload ($ROWS physical lines; quoted fields may contain newlines)."
# 2/3) Validate CSV RECORD cardinality in PostgreSQL, snapshot, and insert in one transaction.
# wc -l is intentionally not a record count because quoted catalog descriptions can contain newlines.
BAK="${T}_prebackfill_bak_$(date +%Y%m%d%H%M%S)"
psql -h /tmp -d dw_unified -v ON_ERROR_STOP=1 <<LOCAL
BEGIN;
LOCK TABLE $T IN SHARE ROW EXCLUSIVE MODE;
CREATE TEMP TABLE _tk11050_load (LIKE $T INCLUDING DEFAULTS);
\copy _tk11050_load ($COLS) FROM '$WORK/rows.$T.csv' WITH CSV
DO \$\$
DECLARE n_rows bigint; n_skus bigint;
BEGIN
  SELECT count(*), count(DISTINCT mfr_sku) INTO n_rows, n_skus FROM _tk11050_load;
  IF n_rows <> $N OR n_skus <> $N THEN
    RAISE EXCEPTION 'source cardinality mismatch: expected $N rows/SKUs, got % rows / % SKUs', n_rows, n_skus;
  END IF;
  IF EXISTS (SELECT 1 FROM _tk11050_load l JOIN $T t USING (mfr_sku)) THEN
    RAISE EXCEPTION 'target drift: at least one audited SKU now exists on Mac2';
  END IF;
END \$\$;
CREATE TABLE $BAK AS SELECT * FROM $T;
SELECT setval(pg_get_serial_sequence('$T','id'), COALESCE((SELECT max(id) FROM $T), 1), true)
WHERE pg_get_serial_sequence('$T','id') IS NOT NULL;
INSERT INTO $T ($COLS) SELECT $COLS FROM _tk11050_load;
COMMIT;
LOCAL
AFTER=$(psql -h /tmp -d dw_unified -tA -c "SELECT count(*), count(distinct mfr_sku) FROM $T")
echo "$T: Mac2 now -> $AFTER (rows, distinct_mfr_sku)"
echo "Undo: psql -h /tmp -d dw_unified -c 'BEGIN; TRUNCATE $T; INSERT INTO $T SELECT * FROM $BAK; COMMIT;'"