#!/usr/bin/env bash
# Retry failed delta loads. Usage: bash ~/migration/retry_failed_delta.sh
set -uo pipefail
source ~/migration/db-env.sh
cd ~/migration/delta_csv
mkdir -p filtered
LOG=~/migration/retry_failed.log
: > "$LOG"

filter_csv() {
  local csv="$1" schema="$2" table="$3"
  psql -At -c "
    SELECT a.attname
    FROM pg_attribute a
    JOIN pg_class c ON c.oid = a.attrelid
    JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE n.nspname='$schema' AND c.relname='$table'
      AND a.attnum > 0 AND NOT a.attisdropped AND a.attgenerated = ''
    ORDER BY a.attnum;
  " > "/tmp/tgt_cols_${schema}_${table}.txt"

  python3 - "$csv" "filtered/$csv" "/tmp/tgt_cols_${schema}_${table}.txt" <<'PY'
import csv, sys
src, dst, tgt_path = sys.argv[1], sys.argv[2], sys.argv[3]
tgt = {l.strip() for l in open(tgt_path) if l.strip()}
with open(src, newline="", encoding="utf-8-sig") as f:
    r = csv.DictReader(f)
    cols = [c.strip() for c in r.fieldnames if c and c.strip() in tgt]
    rows = list(r)
with open(dst, "w", newline="", encoding="utf-8") as f:
    w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore")
    w.writeheader()
    for row in rows:
        w.writerow({c: row.get(c, "") for c in cols})
print(",".join(cols), end="")
PY
}

quote_cols() {
  local cols="$1"
  if [[ -z "$cols" ]]; then
    echo "ERROR: empty column list" >&2
    return 1
  fi
  psql -At -c "SELECT string_agg(quote_ident(x), ', ') FROM unnest(string_to_array('$cols', ',')) AS t(x);"
}

# args: label csv schema table conflict_clause use_overriding(yes/no) pre_copy_sql post_copy_sql
load_one() {
  local label="$1" csv="$2" schema="$3" table="$4" conflict="$5" use_over="$6" pre_sql="${7:-}" post_sql="${8:-}"
  local tbl="${schema}.${table}"
  local cols quoted over_clause=""
  echo "=== $label ===" | tee -a "$LOG"

  cols=$(filter_csv "$csv" "$schema" "$table") || { echo "FAIL filter $label" | tee -a "$LOG"; return 1; }
  quoted=$(quote_cols "$cols") || { echo "FAIL quote $label" | tee -a "$LOG"; return 1; }
  echo "cols=$cols" | tee -a "$LOG"

  if [[ "$use_over" == "yes" ]]; then
    over_clause="OVERRIDING SYSTEM VALUE"
  fi

  psql --variable ON_ERROR_STOP=1 <<EOSQL 2>&1 | tee -a "$LOG"
SET session_replication_role = replica;
CREATE TEMP TABLE staging_delta (LIKE ${tbl} INCLUDING DEFAULTS);
SELECT format('ALTER TABLE staging_delta ALTER COLUMN %I DROP NOT NULL;', a.attname)
FROM pg_attribute a
JOIN pg_class c ON c.oid = a.attrelid
WHERE c.oid = 'staging_delta'::regclass AND a.attnum > 0 AND NOT a.attisdropped AND a.attnotnull
\\gexec
${pre_sql}
\\copy staging_delta (${quoted}) FROM '${PWD}/filtered/${csv}' CSV HEADER
${post_sql}
INSERT INTO ${tbl} (${quoted})
${over_clause}
SELECT ${quoted} FROM staging_delta
${conflict};
DROP TABLE staging_delta;
EOSQL
  local rc=${PIPESTATUS[0]}
  if [[ $rc -eq 0 ]]; then
    echo "OK $label" | tee -a "$LOG"
  else
    echo "FAILED $label" | tee -a "$LOG"
  fi
  return 0
}

# --- retries ---

load_one "auth.audit_log_entries" "auth.audit_log_entries.csv" auth audit_log_entries \
  "ON CONFLICT (id) DO NOTHING" "no" \
  "ALTER TABLE staging_delta ALTER COLUMN ip_address DROP NOT NULL;" \
  "UPDATE staging_delta SET ip_address = '0.0.0.0'::inet WHERE ip_address IS NULL;"

load_one "public.product" "public.product.csv" public product \
  "ON CONFLICT (id) DO NOTHING" "yes"

load_one "public.product_images" "public.product_images.csv" public product_images \
  "ON CONFLICT (id) DO NOTHING" "yes" \
  "" \
  "DELETE FROM staging_delta WHERE image_path IS NULL OR btrim(image_path::text) = '';"

load_one "public.product_store" "public.product_store.csv" public product_store \
  "ON CONFLICT (product_id, store_code) DO NOTHING" "yes"

load_one "public.product_store_processing_queue" "public.product_store_processing_queue.csv" public product_store_processing_queue \
  "ON CONFLICT (customer_id, store_code, week_start_date) DO NOTHING" "yes"

load_one "public.sjreport" "public.sjreport.csv" public sjreport \
  'ON CONFLICT ("row", week_start_date, file, sheetname) DO NOTHING' "yes"

load_one "public.store_metrics" "public.store_metrics.csv" public store_metrics \
  "ON CONFLICT (id) DO NOTHING" "yes"

load_one "public.store_visibility" "public.store_visibility.csv" public store_visibility \
  "ON CONFLICT (customer_id, store_code, week_start_date) DO NOTHING" "yes"

load_one "public.task" "public.task.csv" public task \
  "ON CONFLICT (chain, store_name, product_id, week_start_date) DO NOTHING" "yes"

load_one "storage.objects" "storage.objects.csv" storage objects \
  "ON CONFLICT (bucket_id, name) DO NOTHING" "no"

echo "=== ALL RETRIES FINISHED — see $LOG ===" | tee -a "$LOG"
grep -E '^(OK|FAILED)' "$LOG" || true