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

umask 077

CONFIG_FILE="@@RECEIVER_CONFIG_FILE@@"
LIB_DIR="@@RESTORE_LIB_DIR@@"
SELF_PATH="@@SELF_PATH@@"
SUDOERS_PATH="@@SUDOERS_PATH@@"
SUDOERS_MAIN_FILE="@@SUDOERS_MAIN_FILE@@"
WORK_ROOT="@@WORK_ROOT@@"
WORK_PREFIX="@@WORK_PREFIX@@"
CONTAINER_PREFIX="@@CONTAINER_PREFIX@@"
SHADOW_DB_USER="@@SHADOW_DB_USER@@"
SHADOW_DB_NAME="@@SHADOW_DB_NAME@@"
POSTGRES_IMAGE="@@POSTGRES_IMAGE@@"
DEADMAN_TIMER_UNIT="@@DEADMAN_TIMER_UNIT@@"
HEALTH_TIMER_UNIT="@@HEALTH_TIMER_UNIT@@"
ARCHIVE_BATCH_PREFIX="@@ARCHIVE_BATCH_PREFIX@@"
RUN_ID_PREFIX="@@RUN_ID_PREFIX@@"
EXPECTED_DB_RESTORE_SHA="@@EXPECTED_DB_RESTORE_SHA@@"
EXPECTED_STORAGE_RESTORE_SHA="@@EXPECTED_STORAGE_RESTORE_SHA@@"
EXPECTED_ATTACHMENT_VALIDATOR_SHA="@@EXPECTED_ATTACHMENT_VALIDATOR_SHA@@"
EXPECTED_BUILD_PREFIX="@@EXPECTED_BUILD_PREFIX@@"
EXPECTED_MIGRATION_SET_SHA="@@EXPECTED_MIGRATION_SET_SHA@@"
PHASE="bootstrap"
WORK_DIR=""
CONTAINER_NAME=""
RUNTIME_CLEANED=0
RESULT_STATUS="NO_GO"
RPO_SECONDS=""
RTO_SECONDS=""
MIGRATION_SUCCESS=""
MIGRATION_OTHER=""
MIGRATION_SET_MATCH=""
ARCHIVE_FILE_COUNT=""
RESTORED_FILE_COUNT=""
ATTACHMENT_TOTAL=""
ATTACHMENT_MISMATCH=""
TABLE_COUNTS=""
SOURCE_RUN_ID=""
DUMP_LIST_STATUS="NOT_RUN"
TARGET_CONFIRMATION_STATUS="NOT_RUN"
RESTORE_APPLY_STATUS="NOT_RUN"
RESTORE_EXECUTION_PATH="NOT_RUN"
DUMP_LIST_EXIT_CODE="NOT_RUN"
TARGET_CONFIRMATION_EXIT_CODE="NOT_RUN"
TARGET_NAME_MATCH="NOT_RUN"
TARGET_CONFIRMATION_ATTEMPT_COUNT="0"
TARGET_CONFIRMATION_SUCCESS_STREAK="0"
TARGET_CONFIRMATION_FAILURE_REASON="NOT_APPLICABLE"
RESTORE_SCRIPT_EXIT_CODE="NOT_RUN"
PG_RESTORE_EXIT_CODE="NOT_RUN"
PG_RESTORE_ERROR_COUNT="NOT_RUN"
RESTORE_ERROR_CATEGORY="NOT_APPLICABLE"
FAILURE_CATEGORY="none"
DIAGNOSTICS_EMITTED=0

fail() {
  FAILURE_CATEGORY="$1"
  echo "[restore-drill][FAIL] phase=$PHASE category=$1" >&2
  exit 1
}

require_command() {
  command -v "$1" >/dev/null 2>&1 || fail "required_command_missing"
}

confirm_postgres_target() {
  local actual_database=""
  local query_exit_code=0
  local attempt=0
  TARGET_CONFIRMATION_EXIT_CODE="NOT_RUN"
  TARGET_NAME_MATCH="NOT_RUN"
  TARGET_CONFIRMATION_ATTEMPT_COUNT="0"
  TARGET_CONFIRMATION_SUCCESS_STREAK="0"
  TARGET_CONFIRMATION_FAILURE_REASON="TARGET_QUERY_NONZERO"
  for attempt in $(seq 1 60); do
    TARGET_CONFIRMATION_ATTEMPT_COUNT="$attempt"
    set +e
    actual_database="$(PGPASSWORD="$postgres_password" docker exec -e PGPASSWORD \
      "$CONTAINER_NAME" psql -h 127.0.0.1 -v ON_ERROR_STOP=1 \
      -U "$SHADOW_DB_USER" -d "$SHADOW_DB_NAME" \
      -Atqc 'SELECT current_database()' 2>>"$private_log")"
    query_exit_code="$?"
    set -e
    TARGET_CONFIRMATION_EXIT_CODE="$query_exit_code"
    if [[ "$query_exit_code" -ne 0 ]]; then
      TARGET_NAME_MATCH="NOT_RUN"
      TARGET_CONFIRMATION_SUCCESS_STREAK="0"
      [[ "$attempt" -eq 60 ]] || sleep 1
      continue
    fi
    if [[ "$actual_database" != "$SHADOW_DB_NAME" ]]; then
      TARGET_NAME_MATCH="no"
      TARGET_CONFIRMATION_FAILURE_REASON="TARGET_NAME_MISMATCH"
      return 1
    fi
    TARGET_NAME_MATCH="yes"
    TARGET_CONFIRMATION_SUCCESS_STREAK="$((TARGET_CONFIRMATION_SUCCESS_STREAK + 1))"
    if [[ "$TARGET_CONFIRMATION_SUCCESS_STREAK" -ge 2 ]]; then
      TARGET_CONFIRMATION_FAILURE_REASON="NONE"
      return 0
    fi
    sleep 1
  done
  return 1
}

cleanup_runtime() {
  local cleanup_ok=1
  local container_suffix=""
  local work_suffix=""
  set +e
  if [[ -n "$CONTAINER_NAME" ]]; then
    container_suffix="${CONTAINER_NAME#"$CONTAINER_PREFIX"}"
    if [[ "$CONTAINER_NAME" == "$CONTAINER_PREFIX"* \
      && "$container_suffix" =~ ^[0-9TZ-]+$ ]]; then
      docker rm -f "$CONTAINER_NAME" >/dev/null 2>&1 || true
      docker inspect "$CONTAINER_NAME" >/dev/null 2>&1 && cleanup_ok=0
    else
      cleanup_ok=0
    fi
  fi
  if [[ -n "$WORK_DIR" ]]; then
    work_suffix="${WORK_DIR#"$WORK_ROOT/$WORK_PREFIX"}"
    if [[ "$WORK_DIR" == "$WORK_ROOT/$WORK_PREFIX"* \
      && "$work_suffix" =~ ^[0-9TZ-]+$ \
      && -d "$WORK_DIR" && ! -L "$WORK_DIR" ]]; then
      rm -rf -- "$WORK_DIR"
      [[ ! -e "$WORK_DIR" ]] || cleanup_ok=0
    elif [[ -e "$WORK_DIR" ]]; then
      cleanup_ok=0
    fi
  fi
  set -e
  [[ "$cleanup_ok" -eq 1 ]]
}

cleanup_access() {
  local cleanup_ok=1
  set +e
  if [[ -f "$SUDOERS_PATH" && ! -L "$SUDOERS_PATH" \
    && "$(stat -c %u "$SUDOERS_PATH" 2>/dev/null)" == "0" ]]; then
    rm -f -- "$SUDOERS_PATH" || cleanup_ok=0
  elif [[ -e "$SUDOERS_PATH" ]]; then
    cleanup_ok=0
  fi
  visudo -cf "$SUDOERS_MAIN_FILE" >/dev/null 2>&1 || cleanup_ok=0
  if [[ -d "$LIB_DIR" && ! -L "$LIB_DIR" \
    && "$(stat -c %u "$LIB_DIR" 2>/dev/null)" == "0" ]]; then
    rm -rf -- "$LIB_DIR" || cleanup_ok=0
  elif [[ -e "$LIB_DIR" ]]; then
    cleanup_ok=0
  fi
  if [[ -f "$SELF_PATH" && ! -L "$SELF_PATH" \
    && "$(stat -c %u "$SELF_PATH" 2>/dev/null)" == "0" ]]; then
    rm -f -- "$SELF_PATH" || cleanup_ok=0
  elif [[ -e "$SELF_PATH" ]]; then
    cleanup_ok=0
  fi
  set -e
  [[ "$cleanup_ok" -eq 1 ]]
}

emit_restore_diagnostics() {
  if [[ "$DIAGNOSTICS_EMITTED" -eq 1 ]]; then
    return 0
  fi
  DIAGNOSTICS_EMITTED=1
  echo "failure_category=$FAILURE_CATEGORY"
  echo "dump_list=$DUMP_LIST_STATUS"
  echo "dump_list_exit_code=$DUMP_LIST_EXIT_CODE"
  echo "target_confirmation=$TARGET_CONFIRMATION_STATUS"
  echo "target_confirmation_exit_code=$TARGET_CONFIRMATION_EXIT_CODE"
  echo "target_name_match=$TARGET_NAME_MATCH"
  echo "target_confirmation_attempt_count=$TARGET_CONFIRMATION_ATTEMPT_COUNT"
  echo "target_confirmation_success_streak=$TARGET_CONFIRMATION_SUCCESS_STREAK"
  echo "restore_execution_path=$RESTORE_EXECUTION_PATH"
  echo "restore_apply=$RESTORE_APPLY_STATUS"
  echo "restore_script_exit_code=$RESTORE_SCRIPT_EXIT_CODE"
  echo "pg_restore_exit_code=$PG_RESTORE_EXIT_CODE"
  echo "pg_restore_error_count=$PG_RESTORE_ERROR_COUNT"
  echo "restore_error_category=$RESTORE_ERROR_CATEGORY"
}

finish() {
  local exit_code="$?"
  local runtime_cleanup="PASS"
  local access_cleanup="PASS"
  trap - EXIT HUP INT TERM
  # Emit only allowlisted stage diagnostics while the private runtime still
  # exists. Raw stderr remains mode 0600 and is removed by cleanup_runtime.
  emit_restore_diagnostics
  if [[ "$RUNTIME_CLEANED" -ne 1 ]]; then
    cleanup_runtime || runtime_cleanup="FAIL"
  fi
  cleanup_access || access_cleanup="FAIL"
  if [[ "$runtime_cleanup" != "PASS" || "$access_cleanup" != "PASS" ]]; then
    exit_code=1
  fi
  echo "runtime_cleanup=$runtime_cleanup"
  echo "temporary_access_cleanup=$access_cleanup"
  if [[ "$exit_code" -eq 0 ]]; then
    echo "result=$RESULT_STATUS"
    echo "rpo_seconds=$RPO_SECONDS"
    echo "data_restore_rto_seconds=$RTO_SECONDS"
    echo "migration_success=$MIGRATION_SUCCESS"
    echo "migration_other=$MIGRATION_OTHER"
    echo "migration_set_match=$MIGRATION_SET_MATCH"
    echo "archive_regular_files=$ARCHIVE_FILE_COUNT"
    echo "restored_regular_files=$RESTORED_FILE_COUNT"
    echo "attachment_reference_count=$ATTACHMENT_TOTAL"
    echo "attachment_mismatch_count=$ATTACHMENT_MISMATCH"
    printf '%s\n' "$TABLE_COUNTS"
  else
    echo "result=NO_GO"
  fi
  exit "$exit_code"
}
trap finish EXIT HUP INT TERM

for command_name in docker python3 sha256sum tar find stat df cp openssl visudo systemctl seq; do
  require_command "$command_name"
done

PHASE="entry_contract"
[[ "$(id -u)" -eq 0 ]] || fail "root_required"
for absolute_path in "$CONFIG_FILE" "$LIB_DIR" "$SELF_PATH" "$SUDOERS_PATH" \
  "$SUDOERS_MAIN_FILE" "$WORK_ROOT"; do
  [[ "$absolute_path" == /* && "$absolute_path" != "/" ]] \
    || fail "rendered_path_invalid"
done
[[ "$WORK_PREFIX" =~ ^[a-z0-9][a-z0-9-]*-$ \
  && "$CONTAINER_PREFIX" =~ ^[a-z0-9][a-z0-9-]*-$ \
  && "$SHADOW_DB_USER" =~ ^[a-z_][a-z0-9_]*$ \
  && "$SHADOW_DB_NAME" =~ (^|_)(shadow|drill|restore_test)(_|$) \
  && "$POSTGRES_IMAGE" =~ ^[a-z0-9._/-]+:[a-zA-Z0-9._-]+$ \
  && "$DEADMAN_TIMER_UNIT" =~ ^[a-zA-Z0-9_.@-]+[.]timer$ \
  && "$HEALTH_TIMER_UNIT" =~ ^[a-zA-Z0-9_.@-]+[.]timer$ \
  && "$ARCHIVE_BATCH_PREFIX" =~ ^[a-z0-9][a-z0-9-]*-$ \
  && "$RUN_ID_PREFIX" =~ ^[a-z0-9][a-z0-9-]*-$ ]] \
  || fail "rendered_identity_invalid"
[[ "$EXPECTED_DB_RESTORE_SHA" =~ ^[0-9a-f]{64}$ \
  && "$EXPECTED_STORAGE_RESTORE_SHA" =~ ^[0-9a-f]{64}$ \
  && "$EXPECTED_ATTACHMENT_VALIDATOR_SHA" =~ ^[0-9a-f]{64}$ \
  && "$EXPECTED_BUILD_PREFIX" =~ ^[0-9a-f]{12}$ \
  && "$EXPECTED_MIGRATION_SET_SHA" =~ ^[0-9a-f]{64}$ ]] \
  || fail "rendered_digest_invalid"
[[ -f "$CONFIG_FILE" && ! -L "$CONFIG_FILE" \
  && "$(stat -c %u "$CONFIG_FILE")" == "0" ]] || fail "receiver_config_invalid"
[[ -f "$LIB_DIR/restore-db.sh" && ! -L "$LIB_DIR/restore-db.sh" \
  && -f "$LIB_DIR/restore-storage.sh" && ! -L "$LIB_DIR/restore-storage.sh" \
  && -f "$LIB_DIR/validate-attachment-restore.py" \
  && ! -L "$LIB_DIR/validate-attachment-restore.py" ]] \
  || fail "restore_library_missing"
[[ "$(sha256sum "$LIB_DIR/restore-db.sh" | awk '{print $1}')" == "$EXPECTED_DB_RESTORE_SHA" ]] \
  || fail "db_restore_script_hash_mismatch"
[[ "$(sha256sum "$LIB_DIR/restore-storage.sh" | awk '{print $1}')" == "$EXPECTED_STORAGE_RESTORE_SHA" ]] \
  || fail "storage_restore_script_hash_mismatch"
[[ "$(sha256sum "$LIB_DIR/validate-attachment-restore.py" | awk '{print $1}')" == "$EXPECTED_ATTACHMENT_VALIDATOR_SHA" ]] \
  || fail "attachment_validator_hash_mismatch"

# Root-owned receiver configuration contains path-only assignments.
# shellcheck disable=SC1090
source "$CONFIG_FILE"
BACKUP_ARCHIVE_DIR="${BACKUP_ARCHIVE_DIR:-}"
[[ "$BACKUP_ARCHIVE_DIR" == /* && "$BACKUP_ARCHIVE_DIR" != "/" \
  && -d "$BACKUP_ARCHIVE_DIR" && ! -L "$BACKUP_ARCHIVE_DIR" ]] \
  || fail "archive_root_invalid"

PHASE="monitor_preflight"
deadman_enabled_before="$(systemctl is-enabled "$DEADMAN_TIMER_UNIT" 2>/dev/null || true)"
deadman_active_before="$(systemctl is-active "$DEADMAN_TIMER_UNIT" 2>/dev/null || true)"
health_enabled_before="$(systemctl is-enabled "$HEALTH_TIMER_UNIT" 2>/dev/null || true)"
health_active_before="$(systemctl is-active "$HEALTH_TIMER_UNIT" 2>/dev/null || true)"
[[ "$deadman_enabled_before" == "enabled" && "$deadman_active_before" == "active" \
  && "$health_enabled_before" == "enabled" && "$health_active_before" == "active" ]] \
  || fail "monitor_preflight_not_ready"

validate_batch() {
  local batch_dir="$1"
  local identity_mode="$2"
  local expected_run_id="${3:-}"
  python3 - "$batch_dir" "$EXPECTED_BUILD_PREFIX" "$identity_mode" \
    "$expected_run_id" "$ARCHIVE_BATCH_PREFIX" "$RUN_ID_PREFIX" <<'PY'
import datetime as dt
import hashlib
import json
import os
import re
import stat
import sys

(
    batch_dir,
    build_prefix,
    identity_mode,
    expected_run_id,
    archive_batch_prefix,
    run_id_prefix,
) = sys.argv[1:]
if identity_mode not in {"source", "copy"}:
    raise SystemExit("identity_mode")
if not os.path.isdir(batch_dir) or os.path.islink(batch_dir):
    raise SystemExit("batch_type")
entries = set(os.listdir(batch_dir))
expected = {"database.dump", "storage.tar.gz", "manifest.json"}
if entries != expected:
    raise SystemExit("batch_files")
for name in expected:
    path = os.path.join(batch_dir, name)
    mode = os.lstat(path).st_mode
    if not stat.S_ISREG(mode) or stat.S_ISLNK(mode):
        raise SystemExit("artifact_type")
with open(os.path.join(batch_dir, "manifest.json"), encoding="utf-8") as handle:
    manifest = json.load(handle)
run_id = manifest.get("runId")
if manifest.get("schemaVersion") != 1 or manifest.get("status") != "complete":
    raise SystemExit("manifest_contract")
run_id_pattern = re.escape(run_id_prefix) + r"[0-9]{8}T[0-9]{6}Z"
if not isinstance(run_id, str) or not re.fullmatch(run_id_pattern, run_id):
    raise SystemExit("run_id")
if identity_mode == "source":
    if expected_run_id:
        raise SystemExit("unexpected_expected_run_id")
    if os.path.basename(batch_dir) != archive_batch_prefix + run_id:
        raise SystemExit("source_batch_identity")
else:
    if not re.fullmatch(run_id_pattern, expected_run_id):
        raise SystemExit("expected_run_id")
    if run_id != expected_run_id:
        raise SystemExit("copied_batch_source_identity")
build = manifest.get("buildSha")
if not isinstance(build, str) or not re.fullmatch(r"[0-9a-f]{40,64}", build):
    raise SystemExit("build_identity")
if not build.startswith(build_prefix):
    raise SystemExit("build_mismatch")
created = manifest.get("createdAt")
try:
    created_at = dt.datetime.strptime(created, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=dt.timezone.utc)
except Exception as exc:
    raise SystemExit("created_at") from exc
artifacts = manifest.get("artifacts") or {}
total_bytes = 0
for key, name in (("database", "database.dump"), ("storage", "storage.tar.gz")):
    meta = artifacts.get(key) or {}
    if meta.get("file") != name or os.path.isabs(str(meta.get("file") or "")):
        raise SystemExit("artifact_name")
    path = os.path.join(batch_dir, name)
    size = os.path.getsize(path)
    if size != meta.get("bytes"):
        raise SystemExit("artifact_size")
    digest = hashlib.sha256()
    with open(path, "rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    if digest.hexdigest() != meta.get("sha256"):
        raise SystemExit("artifact_hash")
    total_bytes += size
print(
    int(created_at.timestamp()),
    total_bytes,
    build[:8],
    int(bool(manifest.get("sourceImageDigest"))),
    run_id,
)
PY
}

PHASE="select_latest_batch"
selected_batch="$(python3 - "$BACKUP_ARCHIVE_DIR" \
  "$ARCHIVE_BATCH_PREFIX" "$RUN_ID_PREFIX" <<'PY'
import os
import re
import sys

root, archive_batch_prefix, run_id_prefix = sys.argv[1:]
batch_pattern = re.escape(archive_batch_prefix + run_id_prefix) + r"[0-9]{8}T[0-9]{6}Z"
candidates = []
for entry in os.scandir(root):
    if entry.is_symlink() or not entry.is_dir(follow_symlinks=False):
        continue
    if not re.fullmatch(batch_pattern, entry.name):
        continue
    candidates.append((entry.name, entry.path))
if not candidates:
    raise SystemExit(1)
candidates.sort(key=lambda item: item[0])
print(candidates[-1][1])
PY
)" || fail "latest_batch_not_found"

batch_metadata="$(validate_batch "$selected_batch" source "")" || fail "latest_batch_invalid"
read -r created_epoch batch_total_bytes build_short source_digest_present SOURCE_RUN_ID <<<"$batch_metadata"
drill_started_epoch="$(date -u +%s)"
RPO_SECONDS="$((drill_started_epoch - created_epoch))"
[[ "$RPO_SECONDS" -ge 0 && "$RPO_SECONDS" -le 86400 ]] || fail "latest_batch_outside_rpo"
required_bytes="$((batch_total_bytes * 5 + 2147483648))"
available_kib="$(df -Pk "$WORK_ROOT" | awk 'NR==2 {print $4}')"
[[ "$available_kib" =~ ^[0-9]+$ ]] || fail "disk_capacity_unavailable"
available_bytes="$((available_kib * 1024))"
[[ "$available_bytes" -ge "$required_bytes" ]] || fail "disk_capacity_insufficient"

PHASE="create_isolated_runtime"
run_id="$(date -u +%Y%m%dT%H%M%SZ)-$$-$RANDOM"
[[ "$run_id" =~ ^[0-9TZ-]+$ ]] || fail "run_id_invalid"
WORK_DIR="$WORK_ROOT/$WORK_PREFIX$run_id"
CONTAINER_NAME="$CONTAINER_PREFIX$run_id"
[[ ! -e "$WORK_DIR" ]] || fail "work_dir_exists"
mkdir -m 0700 "$WORK_DIR"
private_log="$WORK_DIR/private.log"
: >"$private_log"
chmod 0600 "$private_log"
rto_started_epoch="$(date -u +%s)"
cp -a -- "$selected_batch" "$WORK_DIR/batch"
copied_metadata="$(validate_batch "$WORK_DIR/batch" copy "$SOURCE_RUN_ID")" \
  || fail "copied_batch_invalid"
[[ "$copied_metadata" == "$batch_metadata" ]] || fail "copied_batch_identity_mismatch"

env_file="$WORK_DIR/postgres.env"
postgres_password="$(openssl rand -hex 24)"
{
  echo "POSTGRES_USER=$SHADOW_DB_USER"
  echo "POSTGRES_PASSWORD=$postgres_password"
  echo "POSTGRES_DB=$SHADOW_DB_NAME"
} >"$env_file"
chmod 0600 "$env_file"
printf '%s\n' '[stage=postgres_container_start]' >>"$private_log"
docker run -d --rm --name "$CONTAINER_NAME" --network none --env-file "$env_file" \
  "$POSTGRES_IMAGE" >>"$private_log" 2>&1 || fail "postgres_container_start_failed"

PHASE="restore_database"
DUMP_LIST_STATUS="FAIL"
RESTORE_EXECUTION_PATH="DOCKER_STDIN"
printf '%s\n' '[stage=dump_list]' >>"$private_log"
set +e
docker exec -i "$CONTAINER_NAME" pg_restore --list \
  <"$WORK_DIR/batch/database.dump" >>"$private_log" 2>&1
DUMP_LIST_EXIT_CODE="$?"
set -e
if [[ "$DUMP_LIST_EXIT_CODE" -ne 0 ]]; then
  RESTORE_ERROR_CATEGORY="DUMP_LIST_NONZERO"
  fail "dump_list_failed"
fi
DUMP_LIST_STATUS="PASS"

TARGET_CONFIRMATION_STATUS="FAIL"
printf '%s\n' '[stage=target_confirmation]' >>"$private_log"
if ! confirm_postgres_target; then
  RESTORE_ERROR_CATEGORY="$TARGET_CONFIRMATION_FAILURE_REASON"
  fail "restore_target_confirmation_failed"
fi
TARGET_CONFIRMATION_STATUS="PASS"

real_docker_path="$(command -v docker)"
docker_proxy_dir="$WORK_DIR/docker-proxy"
docker_proxy="$docker_proxy_dir/docker"
restore_diag_file="$WORK_DIR/restore-diag"
mkdir -m 0700 "$docker_proxy_dir"
cat >"$docker_proxy" <<'PROXY'
#!/usr/bin/env bash
set +e
"$RESTORE_REAL_DOCKER" "$@"
command_exit="$?"
is_pg_restore=0
is_list=0
for argument in "$@"; do
  [[ "$argument" == "pg_restore" ]] && is_pg_restore=1
  [[ "$argument" == "--list" ]] && is_list=1
done
if [[ "$is_pg_restore" -eq 1 ]]; then
  if [[ "$is_list" -eq 1 ]]; then
    printf 'script_dump_list_exit_code=%s\n' "$command_exit" >>"$RESTORE_DIAG_FILE"
  else
    printf 'restore_apply_exit_code=%s\n' "$command_exit" >>"$RESTORE_DIAG_FILE"
  fi
fi
exit "$command_exit"
PROXY
chmod 0700 "$docker_proxy"
: >"$restore_diag_file"
chmod 0600 "$restore_diag_file"

RESTORE_APPLY_STATUS="FAIL"
printf '%s\n' '[stage=restore_apply]' >>"$private_log"
set +e
RESTORE_REAL_DOCKER="$real_docker_path" \
RESTORE_DIAG_FILE="$restore_diag_file" \
PATH="$docker_proxy_dir:$PATH" \
RESTORE_SCOPE=shadow \
  RESTORE_CONFIRM_DATABASE="$SHADOW_DB_NAME" \
  DB_TOOL_MODE=docker \
  DB_CONTAINER="$CONTAINER_NAME" \
  DATABASE_URL="postgresql://shadow.invalid/$SHADOW_DB_NAME" \
  POSTGRES_DB="$SHADOW_DB_NAME" \
  POSTGRES_USER="$SHADOW_DB_USER" \
  POSTGRES_PASSWORD="$postgres_password" \
  "$LIB_DIR/restore-db.sh" "$WORK_DIR/batch/database.dump" \
  >>"$private_log" 2>&1
RESTORE_SCRIPT_EXIT_CODE="$?"
set -e

script_dump_list_exit_code="$(awk -F= \
  '$1 == "script_dump_list_exit_code" {print $2}' "$restore_diag_file")"
apply_exit_code="$(awk -F= \
  '$1 == "restore_apply_exit_code" {print $2}' "$restore_diag_file")"
if [[ ! "$script_dump_list_exit_code" =~ ^[0-9]+$ \
  || "$script_dump_list_exit_code" -ne 0 ]]; then
  RESTORE_ERROR_CATEGORY="SCRIPT_DUMP_LIST_NONZERO"
  fail "restore_script_list_failed"
fi
if [[ ! "$apply_exit_code" =~ ^[0-9]+$ ]]; then
  RESTORE_ERROR_CATEGORY="APPLY_EXIT_CODE_MISSING"
  fail "restore_diagnostic_missing"
fi
PG_RESTORE_EXIT_CODE="$apply_exit_code"
if [[ "$PG_RESTORE_EXIT_CODE" -eq 0 ]]; then
  PG_RESTORE_ERROR_COUNT="0"
else
  PG_RESTORE_ERROR_COUNT="$(sed -nE \
    's/^pg_restore: (warning: )?errors ignored on restore: ([0-9]+)$/\2/p' \
    "$private_log" | tail -n 1)"
  [[ "$PG_RESTORE_ERROR_COUNT" =~ ^[0-9]+$ ]] \
    || PG_RESTORE_ERROR_COUNT="UNKNOWN"
fi
if [[ "$RESTORE_SCRIPT_EXIT_CODE" -ne 0 || "$PG_RESTORE_EXIT_CODE" -ne 0 ]]; then
  RESTORE_ERROR_CATEGORY="RESTORE_APPLY_NONZERO"
  fail "restore_apply_failed_after_list_pass"
fi
RESTORE_APPLY_STATUS="PASS"
RESTORE_ERROR_CATEGORY="NONE"

PHASE="verify_database"
migration_names="$(PGPASSWORD="$postgres_password" docker exec -i -e PGPASSWORD \
  "$CONTAINER_NAME" psql -v ON_ERROR_STOP=1 -U "$SHADOW_DB_USER" -d "$SHADOW_DB_NAME" \
  -At <<'SQL' | LC_ALL=C sort
SELECT migration_name
FROM "_prisma_migrations"
WHERE finished_at IS NOT NULL AND rolled_back_at IS NULL
ORDER BY migration_name;
SQL
)" || fail "migration_name_query_failed"
[[ -n "$migration_names" ]] || fail "migration_set_empty"
if printf '%s\n' "$migration_names" | grep -Ev '^[0-9]{14}_[a-z0-9_]+$' >/dev/null; then
  fail "migration_name_invalid"
fi
MIGRATION_SUCCESS="$(printf '%s\n' "$migration_names" | wc -l | tr -d ' ')"
MIGRATION_OTHER="$(PGPASSWORD="$postgres_password" docker exec -i -e PGPASSWORD \
  "$CONTAINER_NAME" psql -v ON_ERROR_STOP=1 -U "$SHADOW_DB_USER" -d "$SHADOW_DB_NAME" \
  -At <<'SQL'
SELECT count(*)
FROM "_prisma_migrations"
WHERE finished_at IS NULL OR rolled_back_at IS NOT NULL;
SQL
)" || fail "migration_state_query_failed"
[[ "$MIGRATION_OTHER" == "0" ]] || fail "migration_state_mismatch"
migration_set_sha="$(printf '%s\n' "$migration_names" | sha256sum | awk '{print $1}')"
[[ "$migration_set_sha" == "$EXPECTED_MIGRATION_SET_SHA" ]] \
  || fail "migration_set_mismatch"
MIGRATION_SET_MATCH="PASS"

TABLE_COUNTS="$(PGPASSWORD="$postgres_password" docker exec -i -e PGPASSWORD \
  "$CONTAINER_NAME" psql -v ON_ERROR_STOP=1 -U "$SHADOW_DB_USER" -d "$SHADOW_DB_NAME" \
  -At -F = <<'SQL' | LC_ALL=C sort
SELECT key, value FROM (VALUES
  ('departments', (SELECT count(*)::bigint FROM departments)),
  ('employees', (SELECT count(*)::bigint FROM employees)),
  ('employee_roles', (SELECT count(*)::bigint FROM employee_roles)),
  ('purchase_requests', (SELECT count(*)::bigint FROM purchase_requests)),
  ('reimbursement_forms', (SELECT count(*)::bigint FROM reimbursement_forms)),
  ('invoice_pool', (SELECT count(*)::bigint FROM invoice_pool)),
  ('payment_batches', (SELECT count(*)::bigint FROM payment_batches)),
  ('contract_registry', (SELECT count(*)::bigint FROM contract_registry)),
  ('projects', (SELECT count(*)::bigint FROM projects)),
  ('purchase_attachments', (SELECT count(*)::bigint FROM purchase_attachments)),
  ('reimbursement_attachments', (SELECT count(*)::bigint FROM reimbursement_attachments)),
  ('payment_batch_receipt_attachments', (SELECT count(*)::bigint FROM payment_batch_receipt_attachments))
) AS counts(key, value);
SQL
)" || fail "table_count_query_failed"
[[ "$(printf '%s\n' "$TABLE_COUNTS" | sed '/^$/d' | wc -l | tr -d ' ')" == "12" ]] \
  || fail "table_count_key_mismatch"

PHASE="restore_storage"
restored_storage="$WORK_DIR/restored-storage"
printf '%s\n' '[stage=restore_storage]' >>"$private_log"
if ! RESTORE_SCOPE=shadow "$LIB_DIR/restore-storage.sh" \
  "$WORK_DIR/batch/storage.tar.gz" "$restored_storage" \
  >>"$private_log" 2>&1; then
  fail "storage_restore_failed"
fi
ARCHIVE_FILE_COUNT="$(tar -tvzf "$WORK_DIR/batch/storage.tar.gz" \
  | awk 'substr($0,1,1)=="-" {count++} END {print count+0}')"
RESTORED_FILE_COUNT="$(find "$restored_storage" -type f -printf '.' | wc -c | tr -d ' ')"
[[ "$ARCHIVE_FILE_COUNT" == "$RESTORED_FILE_COUNT" ]] || fail "storage_file_count_mismatch"

PHASE="verify_attachment_references"
attachment_result="$(PGPASSWORD="$postgres_password" docker exec -i -e PGPASSWORD \
  "$CONTAINER_NAME" psql -v ON_ERROR_STOP=1 -U "$SHADOW_DB_USER" -d "$SHADOW_DB_NAME" \
  -At -F $'\t' <<'SQL' | python3 "$LIB_DIR/validate-attachment-restore.py" "$restored_storage"
SELECT encode(convert_to(storage_path, 'UTF8'), 'base64'), file_size::text, coalesce(file_sha256, '') FROM invoice_pool
UNION ALL
SELECT encode(convert_to(storage_path, 'UTF8'), 'base64'), file_size::text, coalesce(file_sha256, '') FROM purchase_attachments
UNION ALL
SELECT encode(convert_to(storage_path, 'UTF8'), 'base64'), file_size::text, coalesce(file_sha256, '') FROM reimbursement_attachments
UNION ALL
SELECT encode(convert_to(storage_path, 'UTF8'), 'base64'), file_size::text, coalesce(file_sha256, '') FROM payment_batch_receipt_attachments
UNION ALL
SELECT encode(convert_to(attachment_storage_path, 'UTF8'), 'base64'), '', '' FROM leave_requests WHERE attachment_storage_path IS NOT NULL;
SQL
)" || fail "attachment_reference_query_failed"
read -r ATTACHMENT_TOTAL ATTACHMENT_MISMATCH attachment_s3_count <<<"$attachment_result"
[[ "$ATTACHMENT_TOTAL" =~ ^[0-9]+$ && "$ATTACHMENT_MISMATCH" == "0" \
  && "$attachment_s3_count" == "0" ]] || fail "attachment_reference_mismatch"

PHASE="post_validation"
copied_metadata_after="$(validate_batch "$WORK_DIR/batch" copy "$SOURCE_RUN_ID")" \
  || fail "copied_batch_postcheck_failed"
source_metadata_after="$(validate_batch "$selected_batch" source "")" \
  || fail "source_archive_postcheck_failed"
[[ "$copied_metadata_after" == "$batch_metadata" \
  && "$source_metadata_after" == "$batch_metadata" ]] || fail "archive_identity_changed"
rto_finished_epoch="$(date -u +%s)"
RTO_SECONDS="$((rto_finished_epoch - rto_started_epoch))"

deadman_enabled_after="$(systemctl is-enabled "$DEADMAN_TIMER_UNIT" 2>/dev/null || true)"
deadman_active_after="$(systemctl is-active "$DEADMAN_TIMER_UNIT" 2>/dev/null || true)"
health_enabled_after="$(systemctl is-enabled "$HEALTH_TIMER_UNIT" 2>/dev/null || true)"
health_active_after="$(systemctl is-active "$HEALTH_TIMER_UNIT" 2>/dev/null || true)"
[[ "$deadman_enabled_after" == "$deadman_enabled_before" \
  && "$deadman_active_after" == "$deadman_active_before" \
  && "$health_enabled_after" == "$health_enabled_before" \
  && "$health_active_after" == "$health_active_before" ]] \
  || fail "monitor_state_changed"

PHASE="runtime_cleanup"
cleanup_runtime || fail "runtime_cleanup_failed"
RUNTIME_CLEANED=1
RESULT_STATUS="GO"
PHASE="complete"
exit 0