#!/usr/bin/env bash
set -Eeuo pipefail

die() { echo "azcopy exporter: $*" >&2; exit 1; }

normalize_retention() {
    local variable="$1" value
    value="${!variable}"
    case "${value,,}" in
        none|unlimited) printf -v "$variable" '%s' unlimited ;;
        *) [[ "$value" =~ ^[0-9]+$ ]] || \
            die "$variable must be a non-negative integer, unlimited, or none" ;;
    esac
}

configure_work_directories() {
    local default_root directory variable
    default_root="${BACKMASTER_STATE_DIRECTORY:-${STAGING_ROOT:-/var/lib/backmaster}/${INSTANCE_NAME:-default}}/azcopy"
    AZCOPY_LOG_LOCATION="${AZCOPY_LOG_LOCATION:-$default_root/logs}"
    AZCOPY_JOB_PLAN_LOCATION="${AZCOPY_JOB_PLAN_LOCATION:-$default_root/plans}"

    for variable in AZCOPY_LOG_LOCATION AZCOPY_JOB_PLAN_LOCATION; do
        directory="${!variable}"
        [[ "$directory" == /* ]] || die "$variable must be an absolute path"
        if ! mkdir -p -- "$directory"; then
            die "cannot create $variable: $directory"
        fi
        [[ -d "$directory" && -w "$directory" ]] || \
            die "$variable is not a writable directory: $directory"
    done
    export AZCOPY_LOG_LOCATION AZCOPY_JOB_PLAN_LOCATION
}

load_config() {
    : "${EXPORTER_CONFIG:?EXPORTER_CONFIG is required}"
    [[ -r "$EXPORTER_CONFIG" ]] || die "cannot read $EXPORTER_CONFIG"
    set -a
    # shellcheck disable=SC1090
    source "$EXPORTER_CONFIG"
    if [[ -n "${EXPORTER_SECRET_FILE:-}" ]]; then
        [[ -r "$EXPORTER_SECRET_FILE" ]] || die "cannot read $EXPORTER_SECRET_FILE"
        # shellcheck disable=SC1090
        source "$EXPORTER_SECRET_FILE"
    fi
    set +a

    : "${AZCOPY_DESTINATION:?AZCOPY_DESTINATION is required}"
    AZCOPY_DESTINATION="${AZCOPY_DESTINATION%/}"
    [[ "$AZCOPY_DESTINATION" =~ ^https://[^/]+/[^/?]+ ]] || \
        die "AZCOPY_DESTINATION must be an HTTPS container URL"
    if [[ -n "${AZCOPY_SAS_TOKEN:-}" && "$AZCOPY_DESTINATION" == *\?* ]]; then
        die "set a SAS in either AZCOPY_DESTINATION or AZCOPY_SAS_TOKEN, not both"
    fi

    RETENTION_DAYS="${RETENTION_DAYS:-14}"
    MINIMUM_REDUNDANCY="${MINIMUM_REDUNDANCY:-2}"
    HEALTHCHECK_MAX_AGE_SECONDS="${HEALTHCHECK_MAX_AGE_SECONDS:-129600}"
    WAL_RETENTION_DAYS="${WAL_RETENTION_DAYS:-15}"
    normalize_retention RETENTION_DAYS
    normalize_retention WAL_RETENTION_DAYS
    for value in MINIMUM_REDUNDANCY HEALTHCHECK_MAX_AGE_SECONDS; do
        [[ "${!value}" =~ ^[0-9]+$ ]] || die "$value must be a non-negative integer"
    done
    configure_work_directories
}

encode_path() {
    local path="$1" segment encoded="" separator=""
    local -a segments
    IFS=/ read -r -a segments <<<"$path"
    for segment in "${segments[@]}"; do
        segment="$(jq -nr --arg value "$segment" '$value | @uri')"
        encoded+="${separator}${segment}"
        separator=/
    done
    printf '%s\n' "$encoded"
}

object_url() {
    local key="$1" base query="" encoded
    base="${AZCOPY_DESTINATION%%\?*}"
    if [[ "$AZCOPY_DESTINATION" == *\?* ]]; then
        query="?${AZCOPY_DESTINATION#*\?}"
    elif [[ -n "${AZCOPY_SAS_TOKEN:-}" ]]; then
        query="?${AZCOPY_SAS_TOKEN#\?}"
    fi
    encoded="$(encode_path "$key")"
    printf '%s/%s%s\n' "${base%/}" "$encoded" "$query"
}

azcopy_upload() {
    local -a tier=()
    # Catalogue records must be readable even when the account defaults to a
    # colder tier. Lifecycle policies must also exclude catalogue/.
    [[ "$2" != catalogue/* ]] || tier=(--block-blob-tier=Hot)
    azcopy copy "$1" "$(object_url "$2")" --from-to=LocalBlob \
        --overwrite=true --check-length=true --put-md5 \
        --output-level=essential --log-level=ERROR "${tier[@]}" </dev/null
}

azcopy_download() {
    azcopy copy "$(object_url "$1")" "$2" --from-to=BlobLocal \
        --overwrite=true --check-length=true \
        --output-level=essential --log-level=ERROR </dev/null
}

list_paths() {
    local prefix="$1" output
    # Capture before parsing: a failed listing is not an empty catalogue.
    output="$(azcopy list "$(object_url "$prefix")" --machine-readable \
        --output-level=essential --log-level=ERROR </dev/null)" || \
        die "remote listing failed: $prefix"
    sed -nE 's/^(INFO: )?([^;]+); Content Length: .*/\2/p' <<<"$output"
}

validate_backup_name() {
    [[ "$1" =~ ^[A-Za-z0-9._-]+$ && "$1" != . && "$1" != *..* ]] || \
        die "invalid backup name: $1"
}

manifest_names() {
    local paths path name
    paths="$(list_paths basebackups)" || return 1
    while IFS= read -r path; do
        [[ "$path" =~ ^([^/]+)/manifest\.json$ ]] || continue
        name="${BASH_REMATCH[1]}"
        validate_backup_name "$name"
        printf '%s\n' "$name"
    done <<<"$paths"
}

catalogue_rows() {
    local paths path epoch name rows="" duplicates
    paths="$(list_paths catalogue)" || return 1
    while IFS= read -r path; do
        [[ -n "$path" ]] || continue
        [[ "$path" =~ ^(0|[1-9][0-9]{0,14})/([^/]+)\.json$ ]] || \
            die "invalid catalogue path: $path"
        epoch="${BASH_REMATCH[1]}"; name="${BASH_REMATCH[2]}"
        validate_backup_name "$name"
        rows+="${epoch}"$'\t'"${name}"$'\n'
    done <<<"$paths"
    [[ -n "$rows" ]] || return 0
    duplicates="$(printf '%s' "$rows" | cut -f2 | sort | uniq -d)"
    [[ -z "$duplicates" ]] || die "conflicting catalogue records for: $duplicates"
    printf '%s' "$rows" | LC_ALL=C sort -t$'\t' -k1,1nr -k2,2
}

download_manifest() {
    local key="$1" destination="$2"
    rm -f -- "$destination" || return 1
    azcopy_download "$key" "$destination" >/dev/null || \
        die "remote manifest download failed: $key (check access and archive tier)"
    [[ -s "$destination" ]] || die "remote manifest download produced no file: $key"
}

manifest_epoch() {
    local file="$1" name="$2"
    jq -ers --arg name "$name" '
        select(length == 1) | .[0] |
        select(type == "object" and .backup_name == $name) |
        .created_epoch | select(type == "number") |
        select(. >= 0 and . <= 999999999999999 and . == floor)
    ' "$file" || die "invalid manifest name or creation epoch: $name"
}

latest_epoch() (
    local rows names epoch name temporary actual
    rows="$(catalogue_rows)" || exit 1
    if [[ -z "$rows" ]]; then
        names="$(manifest_names)" || exit 1
        [[ -z "$names" ]] || die "legacy backups need catalogue migration: import the newest verified backup with catalogue-import NAME, or run one backup with --force and a unique name"
        exit 3
    fi
    IFS=$'\t' read -r epoch name <<<"$rows"
    names="$(manifest_names)" || exit 1
    grep -Fxq -- "$name" <<<"$names" || \
        die "catalogue references a missing backup manifest: $name"
    temporary="$(mktemp -d)" || exit 1
    trap 'rm -rf -- "$temporary"' EXIT
    download_manifest "catalogue/$epoch/$name.json" "$temporary/manifest.json" || exit 1
    actual="$(manifest_epoch "$temporary/manifest.json" "$name")" || exit 1
    [[ "$actual" == "$epoch" ]] || die "catalogue epoch mismatch: $name"
    printf '%s\n' "$epoch"
)

next_serial() {
    local day="$1" latest names
    [[ "$day" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ ]] || \
        die "invalid serial date: $day"
    names="$(manifest_names)" || return 1
    latest="$(printf '%s\n' "$names" | \
        sed -nE "s/^${day}-([0-9]+)(-|$).*/\\1/p" | sort -n | tail -1)"
    # Explicit decimal arithmetic handles serials 008 and 009 correctly.
    [[ ${#latest} -le 15 ]] || die "serial out of range: $day"
    printf '%d\n' "$(( 10#${latest:-0} + 1 ))"
}

# A backup name identifies one generation. Replacing it with a different epoch
# would leave an old catalogue record referring to the new payload.
check_catalogue_identity() {
    local name="$1" epoch="$2" rows existing_epoch existing_name
    rows="$(catalogue_rows)" || return 1
    while IFS=$'\t' read -r existing_epoch existing_name; do
        [[ "$existing_name" == "$name" ]] || continue
        [[ "$existing_epoch" == "$epoch" ]] || \
            die "backup name already committed: $name; use daily-time or daily-serial"
    done <<<"$rows"
}

catalogue_import() (
    local name="$1" temporary epoch
    validate_backup_name "$name"
    temporary="$(mktemp -d)" || exit 1
    trap 'rm -rf -- "$temporary"' EXIT
    download_manifest "basebackups/$name/manifest.json" "$temporary/manifest.json" || exit 1
    epoch="$(manifest_epoch "$temporary/manifest.json" "$name")" || exit 1
    check_catalogue_identity "$name" "$epoch" || exit 1
    azcopy_upload "$temporary/manifest.json" "catalogue/$epoch/$name.json" || \
        die "catalogue import failed: $name"
)

publish() (
    local stage="$1" name epoch source relative layout archive expected actual names temporary
    [[ -r "$stage/manifest.json" ]] || die "invalid stage: $stage"
    layout="$(jq -er '.artifact.layout // "files"' "$stage/manifest.json")" || \
        die "invalid stage manifest: $stage"
    case "$layout" in
        files) [[ -r "$stage/checksums.sha256" ]] || die "invalid files stage: $stage" ;;
        archive)
            archive="$(jq -er '.artifact.file' "$stage/manifest.json")" || \
                die "invalid archive manifest: $stage"
            expected="$(jq -er '.artifact.sha256' "$stage/manifest.json")" || \
                die "invalid archive manifest: $stage"
            [[ "$archive" =~ ^backup\.(zip|tar\.(gz|xz|zst))$ && -r "$stage/$archive" ]] || \
                die "invalid archive stage: $stage"
            [[ "$expected" =~ ^[0-9a-f]{64}$ ]] || die "invalid archive checksum: $stage"
            actual="$(sha256sum "$stage/$archive")"; actual="${actual%% *}"
            [[ "$actual" == "$expected" ]] || \
                die "archive checksum mismatch: $stage/$archive"
            ;;
        *) die "unsupported artifact layout: $layout" ;;
    esac
    name="$(jq -er '.backup_name' "$stage/manifest.json")"
    validate_backup_name "$name"
    epoch="$(manifest_epoch "$stage/manifest.json" "$name")" || exit 1
    check_catalogue_identity "$name" "$epoch" || exit 1
    names="$(manifest_names)" || exit 1
    if grep -Fxq -- "$name" <<<"$names"; then
        temporary="$(mktemp -d)" || exit 1
        trap 'rm -rf -- "$temporary"' EXIT
        download_manifest "basebackups/$name/manifest.json" "$temporary/manifest.json" || exit 1
        # Retries of the same sealed stage are allowed; other name collisions
        # must not overwrite a committed backup (including a legacy backup).
        jq -e -s 'length == 2 and .[0] == .[1]' \
            "$stage/manifest.json" "$temporary/manifest.json" >/dev/null || \
            die "backup name already committed with a different manifest: $name"
    fi

    while IFS= read -r -d '' source; do
        relative="${source#"$stage"/}"
        [[ "$relative" != manifest.json ]] || continue
        azcopy_upload "$source" "basebackups/$name/$relative" || \
            die "backup upload failed: $name/$relative"
    done < <(find "$stage" -type f -print0 | sort -z)
    azcopy_upload "$stage/manifest.json" "basebackups/$name/manifest.json" || \
        die "backup manifest upload failed: $name"
    azcopy_upload "$stage/manifest.json" "catalogue/$epoch/$name.json" || \
        die "catalogue publication failed: $name; ready stage must be retried"
)

retain() {
    local cutoff rows keep epoch name wal_cutoff names indexed unindexed
    if [[ "$RETENTION_DAYS" != unlimited ]]; then
        cutoff=$(( $(date +%s) - RETENTION_DAYS * 86400 ))
        rows="$(catalogue_rows)" || return 1
        names="$(manifest_names)" || return 1
        # Stale records must not count toward minimum redundancy and cause
        # deletion of the remaining real backups. Validate before any removal.
        while IFS=$'\t' read -r epoch name; do
            [[ -n "$epoch" ]] || continue
            grep -Fxq -- "$name" <<<"$names" || \
                die "catalogue references a missing backup manifest: $name"
        done <<<"$rows"
        indexed="$(cut -f2 <<<"$rows")"
        unindexed=0
        while IFS= read -r name; do
            [[ -n "$name" ]] || continue
            if ! grep -Fxq -- "$name" <<<"$indexed"; then
                unindexed=$((unindexed + 1))
            fi
        done <<<"$names"
        if (( unindexed > 0 )); then
            echo "azcopy exporter: preserving $unindexed unindexed backup(s); use catalogue-import NAME to include them in retention" >&2
        fi
        keep=0
        while IFS=$'\t' read -r epoch name; do
            [[ -n "$epoch" ]] || continue
            keep=$((keep + 1))
            if (( keep > MINIMUM_REDUNDANCY && epoch < cutoff )); then
                # Remove discoverability first. If payload deletion then fails,
                # it is preserved as an unindexed backup, never a false success.
                azcopy remove "$(object_url "catalogue/$epoch/$name.json")" \
                    --output-level=essential --log-level=ERROR </dev/null || \
                    die "catalogue removal failed: $name"
                azcopy remove "$(object_url "basebackups/$name")" \
                    --recursive=true --output-level=essential --log-level=ERROR \
                    </dev/null || die "backup removal failed (now unindexed): $name"
            fi
        done <<<"$rows"
    fi

    if [[ "$WAL_RETENTION_DAYS" != unlimited ]]; then
        wal_cutoff="$(date -u \
            -d "@$(($(date +%s) - WAL_RETENTION_DAYS * 86400))" \
            '+%Y-%m-%dT%H:%M:%SZ')"
        azcopy remove "$(object_url objects/wal)" --recursive=true \
            --include-before="$wal_cutoff" --output-level=essential \
            --log-level=ERROR </dev/null || true
    fi
}

validate_key() {
    [[ "$1" =~ ^[A-Za-z0-9._/-]+$ && "$1" != /* && "$1" != *..* ]] || \
        die "invalid object key"
}

put_file() {
    local source="$1" key="$2"
    validate_key "$key"
    azcopy_upload "$source" "objects/$key"
}

get_file() {
    local key="$1" destination="$2"
    validate_key "$key"
    azcopy_download "objects/$key" "$destination"
}

connectivitycheck() {
    azcopy list "$(object_url '')" --machine-readable \
        --output-level=essential --log-level=ERROR </dev/null >/dev/null
}

healthcheck() {
    local latest age status
    latest="$(latest_epoch)" || {
        status=$?
        if (( status == 3 )); then
            echo "CRITICAL: no completed exported backup"
        else
            echo "CRITICAL: unable to determine latest exported backup"
        fi
        exit 2
    }
    age=$(( $(date +%s) - latest ))
    if (( age < 0 || age > HEALTHCHECK_MAX_AGE_SECONDS )); then
        echo "CRITICAL: latest exported backup is ${age}s old"
        exit 2
    fi
    echo "OK: latest exported backup is ${age}s old"
}

main() {
    load_config
    case "${1:-}" in
        latest-epoch) latest_epoch ;;
        next-serial) [[ $# -eq 2 ]] || exit 64; next_serial "$2" ;;
        publish) [[ $# -eq 2 ]] || exit 64; publish "$2" ;;
        catalogue-import) [[ $# -eq 2 ]] || exit 64; catalogue_import "$2" ;;
        retain) retain ;;
        connectivitycheck) connectivitycheck ;;
        healthcheck) healthcheck ;;
        put-file) [[ $# -eq 3 ]] || exit 64; put_file "$2" "$3" ;;
        get-file) [[ $# -eq 3 ]] || exit 64; get_file "$2" "$3" ;;
        *) die "unsupported command: ${1:-none}" ;;
    esac
}

main "$@"
