#!/bin/bash
#
# ch_merge_partitions.sh
#
# Переносит партиции ClickHouse-таблицы operational_events из "старой" БД (source, --db-from)
# в "новую" (destination, --db-to), без остановки сервисов evi-* и без удаления/перезаписи
# данных в целевой БД.
#
# Для каждой партиции:
#   1. делает FREEZE (дешёвый хардлинк-снапшот, без рестарта сервера) партиции в источнике —
#      резервную копию на случай, если что-то пойдёт не так;
#   2. выгружает партицию в CSV (SELECT * ... WHERE _partition_id = ...);
#   3. прогоняет CSV через gen_operational_event_dump.py — он чинит содержимое JSON внутри
#      entity-полей (например, добавляет nameDi в di/prevDi, если отсутствует) и добавляет
#      колонки, которых не было в схеме 1.5 (targetUser, role — всегда пустые для старых событий);
#   4. вставляет результат в целевую таблицу;
#   5. дедуплицирует партицию в целевой таблице (OPTIMIZE ... FINAL) — движок ReplacingMergeTree,
#      полные дубликаты строк схлопнутся сами, полагаться только на фоновый мерж не стоит.
#
# db-from и db-to не обязаны быть на одном инстансе ClickHouse — перенос идёт через CSV
# по сети, а не через ATTACH PARTITION FROM.
#
# После успешного переноса и проверки всех партиций резервные FREEZE-снапшоты источника
# удаляются напрямую с файловой системы (sudo rm -rf shadow/<name>) — ALTER TABLE ... UNFREEZE
# не везде физически чистит файлы (наблюдалось на ClickHouse 26.5.2). Из-за этого скрипт нужно
# запускать на том же хосте, где физически лежат данные --db-from, от пользователя с sudo.
#
# Идемпотентность: прогресс по партициям пишется в .ch_merge_state_<db_from>_<db_to>_<table>.done
# рядом со скриптом. Повторный запуск пропускает уже перенесённые партиции.
#
# Таблица всегда operational_events. Все параметры подключения запрашиваются интерактивно —
# никаких аргументов командной строки, кроме -h/--help, скрипт не принимает.
#
# Usage:
#   ./ch_merge_partitions.sh
#   ./ch_merge_partitions.sh -h | --help

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
LOG_FILE="$SCRIPT_DIR/merge_cache/ch_merge_partitions.log"
STATE_FILE_PREFIX="$SCRIPT_DIR/merge_cache/.ch_merge_state"
PY_SCRIPT="$SCRIPT_DIR/src/py_utils/gen_operational_event_dump.py"
TMP_DIR="$SCRIPT_DIR/merge_cache/.ch_merge_tmp"

table="operational_events"

usage() {
    echo "Usage: $0"
    echo
    echo "Аргументы командной строки не принимаются — все параметры подключения (хост, порт,"
    echo "пользователь, пароль, имя БД для источника и назначения) запрашиваются интерактивно."
    echo "Таблица всегда operational_events."
    exit 0
}

if [ "$#" -gt 0 ]; then
    case "$1" in
        -h|--help) usage ;;
        *) echo "Неизвестный аргумент: $1"; echo; usage ;;
    esac
fi

if [ ! -f "$PY_SCRIPT" ]; then
    echo "Не найден $PY_SCRIPT"
    exit 1
fi

# запрашивает значение, повторяет вопрос пока не введут непустое (если нет default)
prompt() {
    local var_name="$1" label="$2" default="${3:-}"
    local value
    if [ -n "$default" ]; then
        read -r -p "$label [$default]: " value
        value="${value:-$default}"
    else
        value=""
        while [ -z "$value" ]; do
            read -r -p "$label: " value
        done
    fi
    printf -v "$var_name" '%s' "$value"
}

prompt_password() {
    local var_name="$1" label="$2"
    local value
    read -r -s -p "$label: " value
    echo
    printf -v "$var_name" '%s' "$value"
}

echo "=== Источник (откуда переносим) ==="
prompt host_from "Host" "localhost"
prompt port_from "Port" "9000"
prompt user_from "User" "default"
prompt_password password_from "Password"
prompt db_from "Database"

echo
echo "=== Назначение (куда переносим) ==="
prompt host_to "Host" "localhost"
prompt port_to "Port" "9000"
prompt user_to "User" "default"
prompt_password password_to "Password"
prompt db_to "Database"
echo

max_parallel=""
while ! [[ "$max_parallel" =~ ^[1-9][0-9]*$ ]]; do
    prompt max_parallel "Сколько партиций переносить параллельно" "4"
done
echo

log() {
    local line="[$(date '+%Y-%m-%d %H:%M:%S')] $1"
    echo "$line"
    echo "$line" >> "$LOG_FILE"
}

ch_from() { clickhouse-client --host="$host_from" --port="$port_from" --user="$user_from" --password="$password_from" "$@"; }
ch_to()   { clickhouse-client --host="$host_to"   --port="$port_to"   --user="$user_to"   --password="$password_to"   "$@"; }

mkdir -p "$TMP_DIR"
state_file="${STATE_FILE_PREFIX}_${db_from}_${db_to}_${table}.done"
frozen_state_file="${STATE_FILE_PREFIX}_${db_from}_${db_to}_${table}.frozen"
touch "$state_file" "$frozen_state_file"

log "=== Запуск переноса $db_from.$table -> $db_to.$table ==="
log "from: $host_from:$port_from   to: $host_to:$port_to"

log "создаю базу $db_to (если ещё не существует)..."
ch_to --query="CREATE DATABASE IF NOT EXISTS $db_to"

log "создаю таблицу $db_to.$table (если ещё не существует) — схема как в 1.6..."
ch_to --query="CREATE TABLE IF NOT EXISTS $db_to.$table
(
    \`id\` UUID,
    \`timestamp\` DateTime64(6),
    \`code\` Int16,
    \`level\` Int16,
    \`controller\` String,
    \`location\` String,
    \`user\` String,
    \`relay\` String,
    \`di\` String,
    \`camera\` String,
    \`cameraStream\` String,
    \`card\` String,
    \`car\` String,
    \`worker\` String,
    \`guest\` String,
    \`prevCard\` String,
    \`prevWorker\` String,
    \`prevGuest\` String,
    \`prevDi\` String,
    \`service\` String,
    \`updateTask\` String,
    \`faceList\` String,
    \`carList\` String,
    \`face\` String,
    \`object\` String,
    \`directionZone\` String,
    \`crossLine\` String,
    \`targetUser\` String,
    \`role\` String
)
ENGINE = ReplacingMergeTree
PARTITION BY toYYYYMM(timestamp)
ORDER BY (timestamp, code, level, controller, location, user, relay, di, camera, cameraStream, card, car, worker, guest, prevCard, prevWorker, prevGuest, prevDi, service, updateTask, faceList, carList, face, object, directionZone, crossLine, targetUser, role)
SETTINGS index_granularity = 8192"

# Обрабатывает одну партицию целиком (freeze -> csv -> python-фикс -> insert -> optimize).
# Вызывается в фоне (в параллельных батчах), поэтому пишет прогресс только через append
# в файлы (лог, .done, .frozen) — эти записи короткие (одна строка), такие append на POSIX
# атомарны, параллельные вызовы друг друга не портят.
process_partition() {
    local p="$1"
    local freeze_name="chmerge_${db_from}_${db_to}_${table}_${p}"

    if grep -qx "$p" "$frozen_state_file"; then
        log "[$p] резервный снапшот уже был сделан в прошлой (незавершённой) попытке, повторно не делаю"
    else
        log "[$p] резервный снапшот источника (FREEZE, имя '$freeze_name')..."
        ch_from --query="ALTER TABLE $db_from.$table FREEZE PARTITION ID '$p' WITH NAME '$freeze_name'"
        echo "$p" >> "$frozen_state_file"
        log "[$p] снапшот лежит в shadow/$freeze_name/ на диске сервера, где живёт $db_from (обычно /var/lib/clickhouse/shadow/$freeze_name/)"
    fi

    local raw_csv="$TMP_DIR/${db_from}_${p}_raw.csv"
    local fixed_csv="$TMP_DIR/${db_from}_${p}_fixed.csv"

    log "[$p] выгружаю партицию в CSV..."
    ch_from --query="SELECT * FROM $db_from.$table WHERE _partition_id = '$p' FORMAT CSVWithNames" > "$raw_csv"

    log "[$p] прогоняю через gen_operational_event_dump.py..."
    python3 "$PY_SCRIPT" "$raw_csv" "$fixed_csv"

    log "[$p] вставляю в $db_to.$table..."
    ch_to --query="INSERT INTO $db_to.$table FORMAT CSV" < "$fixed_csv"

    log "[$p] дедуплицирую (OPTIMIZE ... FINAL)..."
    ch_to --query="OPTIMIZE TABLE $db_to.$table PARTITION ID '$p' FINAL"

    rm -f "$raw_csv" "$fixed_csv"
    echo "$p" >> "$state_file"
    log "[$p] готово"
}

partitions=$(ch_from --query="SELECT DISTINCT partition_id FROM system.parts WHERE database='$db_from' AND table='$table' AND active ORDER BY partition_id FORMAT TSV")

pending=()
for p in $partitions; do
    if grep -qx "$p" "$state_file"; then
        log "[$p] уже перенесена ранее, пропускаю"
        continue
    fi
    pending+=("$p")
done

failed_partitions=()
for ((i = 0; i < ${#pending[@]}; i += max_parallel)); do
    batch=("${pending[@]:i:max_parallel}")
    log "запускаю параллельно (до $max_parallel одновременно): ${batch[*]}"

    batch_pids=()
    for p in "${batch[@]}"; do
        process_partition "$p" &
        batch_pids+=("$!")
    done

    for idx in "${!batch_pids[@]}"; do
        if ! wait "${batch_pids[$idx]}"; then
            failed_partitions+=("${batch[$idx]}")
            log "[${batch[$idx]}] ОШИБКА при переносе, смотрите сообщения выше"
        fi
    done
done

if [ "${#failed_partitions[@]}" -gt 0 ]; then
    log "ОШИБКА: не удалось перенести партиции: ${failed_partitions[*]}"
    log "Повторный запуск скрипта продолжит с этого места (снапшоты и уже перенесённые партиции сохранены)."
    exit 1
fi

log "Проверка после переноса:"
result=$(ch_to --query="SELECT _partition_id AS partition_id, count() FROM $db_to.$table FINAL GROUP BY partition_id ORDER BY partition_id FORMAT PrettyCompact")
echo "$result"
echo "$result" >> "$LOG_FILE"

echo
remove_backups=""
read -r -p "Хотите удалить бэкапы мигрированных партиций? [Y/n]: " remove_backups
remove_backups="${remove_backups:-Y}"

if [[ "$remove_backups" =~ ^[Yy] ]]; then
    # ALTER TABLE ... UNFREEZE не везде физически удаляет файлы (наблюдалось на 26.5.2 — команда
    # отрабатывает без ошибки, но каталог в shadow/ остаётся). Поэтому чистим напрямую по файловой
    # системе. Это требует, чтобы скрипт выполнялся на том же хосте, где физически лежат данные
    # db_from (--host-from), и чтобы у пользователя, запускающего скрипт, был sudo — права
    # запрашиваются явно прямо сейчас, а не неожиданно посреди цикла удаления.
    echo "Для удаления снапшотов нужны права sudo — сейчас будет запрошен пароль."
    if sudo -v; then
        log "Удаляю резервные снапшоты источника..."
        ch_data_path=$(ch_from --query="SELECT path FROM system.disks WHERE name = 'default'" | tr -d '[:space:]')
        ch_data_path="${ch_data_path%/}"

        if [ -z "$ch_data_path" ]; then
            log "ПРЕДУПРЕЖДЕНИЕ: не удалось определить путь данных ClickHouse (system.disks). Снапшоты нужно будет удалить вручную из shadow/: $(cat "$frozen_state_file" | tr '\n' ' ')"
        else
            while read -r p; do
                [ -z "$p" ] && continue
                freeze_name="chmerge_${db_from}_${db_to}_${table}_${p}"
                snapshot_dir="${ch_data_path}/shadow/${freeze_name}"
                log "[$p] удаляю снапшот '$freeze_name' ($snapshot_dir)..."
                sudo rm -rf "$snapshot_dir"
            done < "$frozen_state_file"
        fi
    else
        log "ОШИБКА: не удалось получить права sudo. Снапшоты НЕ удалены, остались в shadow/ (chmerge_${db_from}_${db_to}_${table}_*)."
    fi
else
    log "Резервные снапшоты оставлены как есть в shadow/ (chmerge_${db_from}_${db_to}_${table}_*), sudo не запрашивался."
fi

rm -f "$state_file" "$frozen_state_file"

log "=== Готово ==="
