#!/usr/bin/env bash
set -euo pipefail
umask 077

# 分钟级 binlog 归档。每分钟检查活动 binlog 的 Position，仅在有新增已提交事务时
# 轮转；每个对象以 server_uuid、原始文件名和原始 sha256 命名，避免实例重建后的编号碰撞。

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=backup-common.sh
source "${SCRIPT_DIR}/backup-common.sh"

MYSQL_DATA_DIR='/var/lib/mysql'
work_dir=''
cleanup() {
  if [[ -n "${work_dir}" && -d "${work_dir}" ]]; then
    rm -rf -- "${work_dir}"
  fi
}
trap cleanup EXIT

require_archive_config() {
  backup_require_env BACKUP_DIR >/dev/null
  backup_require_env BACKUP_OSS_URI >/dev/null
  backup_require_env BACKUP_PASSPHRASE >/dev/null
}

mysql_binlog_query() {
  local mysql_cid="$1" query="$2" output code
  output="$(docker exec "${mysql_cid}" sh -c \
    'MYSQL_PWD="$MYSQL_ROOT_PASSWORD" mysql -uroot --protocol=socket -N -B -e "$1"' \
    sh "${query}" 2>&1)" || {
    code=$?
    backup_error "MySQL binlog 查询失败（${query}）：${output}"
    return "${code}"
  }
  printf '%s\n' "${output}"
}

master_status() {
  local mysql_cid="$1"
  mysql_binlog_query "${mysql_cid}" 'SHOW BINARY LOG STATUS;' | awk 'NR == 1 { print $1 "\t" $2 }'
}

state_active_binlog() {
  local state_dir="$1" key="$2" value
  [[ -f "${state_dir}/active-binlog.env" ]] || return
  value="$(awk -F= -v key="${key}" '$1 == key { print substr($0, index($0, "=") + 1); exit }' "${state_dir}/active-binlog.env")"
  printf '%s' "${value}"
}

write_active_binlog_state() {
  local state_dir="$1" filename="$2" position="$3"
  {
    printf 'filename=%s\n' "${filename}"
    printf 'position=%s\n' "${position}"
    printf 'observed_at=%s\n' "$(date -Iseconds)"
  } > "${state_dir}/active-binlog.env"
}

state_last_index() {
  local state_dir="$1"
  local value
  [[ -f "${state_dir}/last-archived.env" ]] || return
  value="$(awk -F= '$1 == "binlog_index" { print $2; exit }' "${state_dir}/last-archived.env")"
  [[ "${value}" =~ ^[0-9]+$ ]] && printf '%s' "${value}"
}

write_state() {
  local state_dir="$1" filename="$2" index="$3" raw_sha="$4" cipher_sha="$5" remote_dir="$6"
  {
    printf 'filename=%s\n' "${filename}"
    printf 'binlog_index=%s\n' "${index}"
    printf 'raw_sha256=%s\n' "${raw_sha}"
    printf 'cipher_sha256=%s\n' "${cipher_sha}"
    printf 'remote_dir=%s\n' "${remote_dir}"
  } > "${state_dir}/${filename}.${raw_sha}.state"
  cp "${state_dir}/${filename}.${raw_sha}.state" "${state_dir}/last-archived.env"
}

archive_one() {
  local mysql_cid="$1" state_dir="$2" server_uuid="$3" oss_uri="$4" filename="$5"
  local source_path source_sha raw_file raw_sha cipher_file cipher_sha cipher_size binlog_index object_id remote_dir
  source_path="${MYSQL_DATA_DIR}/${filename}"
  if ! docker exec "${mysql_cid}" test -f "${source_path}" >/dev/null 2>&1; then
    backup_error "待归档 binlog 已从 MySQL 数据目录消失：${filename}；PITR 链路已断裂"
    exit 1
  fi
  raw_file="${work_dir}/${filename}"
  docker cp "${mysql_cid}:${source_path}" "${raw_file}"
  raw_sha="$(backup_sha256 "${raw_file}")"
  source_sha="$(docker exec "${mysql_cid}" sha256sum "${source_path}" | awk '{print $1}')"
  [[ "${source_sha}" == "${raw_sha}" ]] || {
    backup_error "binlog 拷贝校验失败：${filename}"
    exit 1
  }
  if compgen -G "${state_dir}/${filename}.${raw_sha}.state" >/dev/null; then
    return
  fi
  binlog_index="$(backup_binlog_index "${filename}")"
  object_id="${filename}.${raw_sha}"
  remote_dir="${oss_uri}/mysql/binlog/${server_uuid}/${object_id}"
  cipher_file="${work_dir}/${object_id}.enc"
  backup_encrypt_file "${raw_file}" "${cipher_file}"
  cipher_sha="$(backup_sha256 "${cipher_file}")"
  cipher_size="$(backup_file_size "${cipher_file}")"
  printf '%s  %s\n' "${cipher_sha}" 'binlog.enc' > "${work_dir}/${object_id}.enc.sha256"
  {
    printf 'format_version=1\n'
    printf 'server_uuid=%s\n' "${server_uuid}"
    printf 'source_binlog=%s\n' "${filename}"
    printf 'binlog_index=%s\n' "${binlog_index}"
    printf 'raw_sha256=%s\n' "${raw_sha}"
    printf 'payload=binlog.enc\n'
    printf 'payload_sha256=%s\n' "${cipher_sha}"
    printf 'payload_size_bytes=%s\n' "${cipher_size}"
    printf 'created_at=%s\n' "$(date -Iseconds)"
  } > "${work_dir}/${object_id}.complete.env"

  backup_oss_upload "${cipher_file}" "${remote_dir}/binlog.enc"
  backup_oss_upload "${work_dir}/${object_id}.enc.sha256" "${remote_dir}/binlog.enc.sha256"
  # complete.env 最后上传，恢复程序只使用有该文件的 binlog。
  backup_oss_upload "${work_dir}/${object_id}.complete.env" "${remote_dir}/complete.env"
  write_state "${state_dir}" "${filename}" "${binlog_index}" "${raw_sha}" "${cipher_sha}" "${remote_dir}"
  rm -f -- "${raw_file}" "${cipher_file}" "${work_dir}/${object_id}.enc.sha256" "${work_dir}/${object_id}.complete.env"
}

archive() {
  backup_require_cmd docker
  backup_require_cmd ossutil
  backup_require_cmd openssl
  require_archive_config
  local force="${1:-}" backup_dir oss_uri mysql_cid server_uuid state_dir first_binlog last_index first_index expected_index filename index
  local current_status current_file current_pos previous_file previous_pos final_status final_file final_pos binlog_list closed_binlog_list
  if [[ -n "${force}" && "${force}" != '--force' ]]; then
    printf '用法: %s archive [--force]\n' "$0" >&2
    exit 1
  fi
  backup_dir="$(backup_require_env BACKUP_DIR)"
  backup_validate_dir "${backup_dir}"
  oss_uri="$(backup_require_env BACKUP_OSS_URI)"
  backup_validate_oss_uri "${oss_uri}"
  backup_lock_or_exit "${backup_dir}/.binlog.lock"
  mysql_cid="$(backup_mysql_container_id)"
  server_uuid="$(backup_server_uuid)"
  state_dir="${backup_dir}/binlog-state/${server_uuid}"
  work_dir="${backup_dir}/.tmp-binlog-$$"
  mkdir -p "${state_dir}" "${work_dir}"

  current_status="$(master_status "${mysql_cid}")"
  IFS=$'\t' read -r current_file current_pos <<< "${current_status}"
  [[ -n "${current_file}" && "${current_pos}" =~ ^[0-9]+$ ]] || { backup_error '无法读取当前 binlog 位点'; exit 1; }
  backup_binlog_index "${current_file}" >/dev/null
  previous_file="$(state_active_binlog "${state_dir}" filename || true)"
  previous_pos="$(state_active_binlog "${state_dir}" position || true)"

  if [[ "${force}" == '--force' || -z "${previous_file}" || -z "${previous_pos}" || "${current_file}" != "${previous_file}" || "${current_pos}" != "${previous_pos}" ]]; then
    docker exec "${mysql_cid}" sh -c \
      'MYSQL_PWD="$MYSQL_ROOT_PASSWORD" mysql -uroot --protocol=socket -e "FLUSH BINARY LOGS;"' >/dev/null
  else
    backup_log "活动 binlog 无新增已提交事务，跳过轮转"
  fi
  binlog_list="$(mysql_binlog_query "${mysql_cid}" 'SHOW BINARY LOGS;')"
  first_binlog="$(awk 'NR == 1 { print $1; exit }' <<< "${binlog_list}")"
  [[ -n "${first_binlog}" ]] || { backup_error '无法读取 binlog 列表'; exit 1; }
  last_index="$(state_last_index "${state_dir}" || true)"
  first_index="$(backup_binlog_index "${first_binlog}")"
  if [[ -n "${last_index}" && "${first_index}" -gt $((last_index + 1)) ]]; then
    backup_error "发现未归档 binlog 缺口：已归档至序号 ${last_index}，当前最早为 ${first_index}"
    exit 1
  fi

  closed_binlog_list="$(awk 'NR > 1 { print previous } { previous = $1 }' <<< "${binlog_list}")"
  expected_index="${last_index}"
  while IFS= read -r filename; do
    [[ -n "${filename}" ]] || continue
    index="$(backup_binlog_index "${filename}")"
    if [[ -n "${expected_index}" && "${index}" -le "${expected_index}" ]]; then
      continue
    fi
    if [[ -n "${expected_index}" && "${index}" -ne $((expected_index + 1)) ]]; then
      backup_error "发现未归档 binlog 缺口：期望序号 $((expected_index + 1))，实际为 ${index}"
      exit 1
    fi
    archive_one "${mysql_cid}" "${state_dir}" "${server_uuid}" "${oss_uri}" "${filename}"
    expected_index="${index}"
  done <<< "${closed_binlog_list}"
  final_status="$(master_status "${mysql_cid}")"
  IFS=$'\t' read -r final_file final_pos <<< "${final_status}"
  [[ -n "${final_file}" && "${final_pos}" =~ ^[0-9]+$ ]] || { backup_error '无法读取归档后的 binlog 位点'; exit 1; }
  write_active_binlog_state "${state_dir}" "${final_file}" "${final_pos}"
  backup_log "binlog 归档完成，实例 ${server_uuid}"
}

verify() {
  backup_require_cmd ossutil
  require_archive_config
  local backup_dir server_uuid state_dir state_file remote_dir
  backup_dir="$(backup_require_env BACKUP_DIR)"
  backup_validate_dir "${backup_dir}"
  backup_lock_or_exit "${backup_dir}/.binlog.lock"
  server_uuid="$(backup_server_uuid)"
  state_dir="${backup_dir}/binlog-state/${server_uuid}"
  [[ -d "${state_dir}" ]] || { backup_error '尚未产生 binlog 归档状态'; exit 1; }
  while IFS= read -r -d '' state_file; do
    remote_dir="$(awk -F= '$1 == "remote_dir" { print substr($0, index($0, "=") + 1); exit }' "${state_file}")"
    [[ -n "${remote_dir}" ]] || { backup_error "归档状态不完整：${state_file}"; exit 1; }
    ossutil stat "${remote_dir}/binlog.enc" >/dev/null
    ossutil stat "${remote_dir}/binlog.enc.sha256" >/dev/null
    ossutil stat "${remote_dir}/complete.env" >/dev/null
  done < <(find "${state_dir}" -maxdepth 1 -name '*.state' -print0)
  backup_log '全部本地 binlog 归档状态均有远端完整对象'
}

case "${1:-archive}" in
  archive) archive "${2:-}" ;;
  verify) verify ;;
  *)
    printf '用法: %s {archive|verify}\n' "$0" >&2
    exit 1
    ;;
esac