#!/bin/bash
# =============================================================================
# Load master (seed) data from the central repository.
#
# Shipped in muli-neo-database, installed to /usr/local/muli/bin.
#
#   muli-db-seed                      apply to every database on this host
#   muli-db-seed <database> <port>    apply to one database
#   muli-db-seed --status             report only, change nothing
#
# The data itself is NOT in this package. It is pulled from the seed repository
# (see muli-db-seed-pull) into $MULI_SEED_ROOT, so a tax-table change is a
# commit there, not a release here.
#
# Three rules govern every load, and the third is the important one:
#
#   * `_u` tables are never written. They are the customer's.
#   * `_m` tables may be replaced wholesale. They are Muli's.
#   * base tables are NEVER rebuilt - only a delta derived from the change to
#     `_m` is applied. A live `allocs` carries rows in neither `_m` nor `_u`,
#     inserted at runtime by rtbc_fns.inc as "Missing Reinserted"/seq 998 when
#     the trial balance meets a missing allocation code. Those rows are
#     referenced by financial data. Rebuilding the base table would delete them.
# =============================================================================
set -euo pipefail

SEED_ROOT="${MULI_SEED_ROOT:-/usr/local/muli/dbmaster}"
PG_VERSION="${MULI_PG_VERSION:-14}"
PG_CLUSTER_ROOT="${MULI_PG_CLUSTER_ROOT:-/usr/local/mulipg/${PG_VERSION}}"
SUPERUSER="${MULI_DB_SUPERUSER:-postgres}"
STATUS_ONLY=false

log()  { echo "[muli-db-seed] $*"; }
warn() { echo "[muli-db-seed] WARNING: $*" >&2; }
die()  { echo "[muli-db-seed] ERROR: $*" >&2; exit 1; }

case "${1:-}" in
    --status) STATUS_ONLY=true; shift ;;
    -h|--help) sed -n '3,26p' "$0" | sed 's/^# \{0,1\}//'; exit 0 ;;
esac

MANIFEST="${SEED_ROOT}/manifest.tsv"
[ -f "$MANIFEST" ] || die "no manifest at ${MANIFEST}; has the seed repository been pulled?"

# Revision of the data being applied. Recorded per table so a container that has
# stopped pulling is visible.
REVISION=$(git -C "$SEED_ROOT" rev-parse --short HEAD 2>/dev/null || echo "unversioned")

tier_of() {
    case "$1" in
        login_db) echo login ;;
        cdb*)     echo common ;;
        [0-9][0-9][0-9][0-9][0-9][A-I]) echo contract ;;
        *)        echo "" ;;
    esac
}
port_of() {
    local db="$1" c l i
    case "$(tier_of "$db")" in login) echo 59999; return ;; common) echo 60000; return ;; esac
    c="${db:0:5}"; l="${db:5:1}"
    i=$(awk -v x="$l" 'BEGIN{print index("ABCDEFGHI", toupper(x))}')
    [ "$i" -gt 0 ] || return 1
    echo $(( 50000 + 10#"${c:0:1}${c:2:3}${i}" ))
}
discover() {
    [ -d "$PG_CLUSTER_ROOT" ] || return 0
    local d n
    for d in "${PG_CLUSTER_ROOT}"/*pgsql; do
        [ -d "$d" ] || continue
        n=$(basename "$d"); n="${n%pgsql}"
        [ -n "$(tier_of "$n")" ] && echo "$n"
    done
}

q() { psql -tAq -v ON_ERROR_STOP=1 -U "$SUPERUSER" -p "$2" -d "$1" -c "$3"; }

# Column list of a table, comma separated, in ordinal order.
columns_of() { q "$1" "$2" "SELECT string_agg(quote_ident(attname), ',' ORDER BY attnum)
                           FROM pg_attribute WHERE attrelid='$3'::regclass
                            AND attnum>0 AND NOT attisdropped"; }

ensure_ledger() {
    q "$1" "$2" "SET client_min_messages=warning;
                 CREATE TABLE IF NOT EXISTS muli_seed_state (
                     table_name text PRIMARY KEY,
                     revision   text NOT NULL,
                     row_count  integer,
                     applied_at timestamptz NOT NULL DEFAULT now());
                 GRANT SELECT ON muli_seed_state TO muli;" >/dev/null
}

schema_at_least() { # schema_at_least <db> <port> <version>
    [ "$3" = "-" ] && return 0
    local have
    have=$(q "$1" "$2" "SELECT count(*) FROM schema_migrations WHERE version >= '$3'" 2>/dev/null || echo 0)
    [ "${have:-0}" -gt 0 ]
}

requirements_met() { # requirements_met <db> <port> <comma-separated tables>
    [ "$3" = "-" ] || [ -z "$3" ] && return 0
    local t n
    for t in $(echo "$3" | tr ',' ' '); do
        n=$(q "$1" "$2" "SELECT count(*) FROM ${t}" 2>/dev/null || echo 0)
        [ "${n:-0}" -gt 0 ] || return 1
    done
    return 0
}

seed_one_db() { # seed_one_db <db> <port> <tier>
    local db="$1" port="$2" tier="$3" applied=0 skipped=0 failed=0
    ensure_ledger "$db" "$port"

    local file mtier table keycols mode base minschema requires
    while IFS=$'\t' read -r file mtier table keycols mode base minschema requires; do
        case "$file" in ''|\#*) continue ;; esac
        [ "$mtier" = "$tier" ] || continue
        [ -f "${SEED_ROOT}/${file}" ] || { warn "${file} listed but not present"; continue; }

        if ! q "$db" "$port" "SELECT to_regclass('${table}') IS NOT NULL" | grep -qx t; then
            warn "${db}: table ${table} does not exist, skipping ${file}"; skipped=$((skipped+1)); continue
        fi
        if ! schema_at_least "$db" "$port" "$minschema"; then
            warn "${db}: ${table} needs schema >= ${minschema}, not yet applied; skipping"
            skipped=$((skipped+1)); continue
        fi
        # Some tables reference per-contract provisioning records rather than
        # other seed data - opendoc_tpl and digital have foreign keys to
        # project, rpctemplate_m to rpcs. On an unprovisioned database those
        # loads fail with a foreign key violation, which is noise rather than a
        # fault. Skip them explicitly instead.
        if ! requirements_met "$db" "$port" "${requires:--}"; then
            warn "${db}: ${table} needs provisioning data (${requires}); skipping"
            skipped=$((skipped+1)); continue
        fi

        local current
        current=$(q "$db" "$port" "SELECT revision FROM muli_seed_state WHERE table_name='${table}'" || true)
        if [ "$current" = "$REVISION" ]; then
            [ "$STATUS_ONLY" = true ] && log "  ${table}: up to date (${REVISION})"
            continue
        fi

        if [ "$STATUS_ONLY" = true ]; then
            log "  ${table}: would apply ${REVISION} (currently ${current:-none})"
            applied=$((applied+1)); continue
        fi

        if apply_table "$db" "$port" "$file" "$table" "$keycols" "$mode" "$base"; then
            applied=$((applied+1))
        else
            warn "${db}: ${table} failed; continuing with the remaining tables"
            failed=$((failed+1))
        fi
    done < "$MANIFEST"

    if [ "$failed" -gt 0 ]; then
        log "${db} (${tier}): ${applied} applied, ${skipped} skipped, ${failed} FAILED"
        return 1
    fi
    log "${db} (${tier}): ${applied} applied, ${skipped} skipped"
}

apply_table() { # apply_table <db> <port> <file> <table> <keys> <mode> <base>
    local db="$1" port="$2" file="$3" table="$4" keycols="$5" mode="$6" base="$7"
    local cols keyjoin setlist rowcount
    cols=$(columns_of "$db" "$port" "$table")

    # key equality between two aliases, and the SET list for an upsert
    # Plain `=` for the same reason as in muli-db-seed-delta: key columns are
    # PRIMARY KEY columns and cannot be NULL, and IS NOT DISTINCT FROM cannot
    # use an index.
    keyjoin=$(echo "$keycols" | tr ',' '\n' | awk '{printf "%s a.%s = b.%s", (NR>1?" AND":""), $1, $1}')
    setlist=$(echo "$cols" | tr ',' '\n' | grep -vxF -f <(echo "$keycols" | tr ',' '\n') \
              | awk '{printf "%s%s=EXCLUDED.%s", (NR>1?", ":""), $1, $1}')

    log "  ${table} <- ${file} (${mode})"

    # Everything for one table in one transaction: a partial load of a tax table
    # is worse than none.
    if ! psql -v ON_ERROR_STOP=1 -q -U "$SUPERUSER" -p "$port" -d "$db" <<SQL
BEGIN;
CREATE TEMP TABLE _seed_stage (LIKE ${table} INCLUDING DEFAULTS) ON COMMIT DROP;
\copy _seed_stage (${cols}) FROM '${SEED_ROOT}/${file}' WITH (FORMAT csv, HEADER true)
-- Indexed before anything joins to it: the full-sync DELETE below and the
-- delta both match on the key columns, and without this they degrade to a
-- sequential scan per row - painful on postcd_aus (16k), unspsc (18k) and
-- calendar (93k).
CREATE INDEX ON _seed_stage (${keycols});
ANALYZE _seed_stage;
$( [ "$base" != "-" ] && echo "CREATE TEMP TABLE _seed_old AS SELECT * FROM ${table};" )

INSERT INTO ${table} (${cols}) SELECT ${cols} FROM _seed_stage
  ON CONFLICT (${keycols}) DO UPDATE SET ${setlist};

$( [ "$mode" = "full-sync" ] && cat <<DEL
DELETE FROM ${table} a
 WHERE NOT EXISTS (SELECT 1 FROM _seed_stage b WHERE ${keyjoin});
DEL
)
$( [ "$base" != "-" ] && "$(dirname "$0")/muli-db-seed-delta" "${table}" "${base}" "${keycols}" "${cols}" )
COMMIT;
SQL
    then
        return 1
    fi

    rowcount=$(q "$db" "$port" "SELECT count(*) FROM ${table}")
    q "$db" "$port" "INSERT INTO muli_seed_state (table_name, revision, row_count)
                     VALUES ('${table}', '${REVISION}', ${rowcount})
                     ON CONFLICT (table_name) DO UPDATE
                        SET revision=EXCLUDED.revision, row_count=EXCLUDED.row_count, applied_at=now()" >/dev/null
    log "    ${rowcount} rows"
}

if [ $# -eq 2 ]; then TARGETS="$1"; ONE_PORT="$2"; else TARGETS=$(discover); ONE_PORT=""; fi
[ -n "$TARGETS" ] || { log "no databases found"; exit 0; }

log "seed revision ${REVISION} from ${SEED_ROOT}"
rc=0
for db in $TARGETS; do
    tier=$(tier_of "$db")
    port="${ONE_PORT:-$(port_of "$db")}" || { warn "no port for ${db}"; continue; }
    seed_one_db "$db" "$port" "$tier" || rc=1
done
exit $rc
