#!/bin/bash

# ==============================================================================
# 脚本名称: clean_redis_cluster_keys.sh
# 脚本描述: 用于批量且安全地删除 Redis Cluster 中指定 Pattern 的 Key（使用 UNLINK 避免阻塞）
# 优化点: 引入 Bash 数组安全防注入参数传递、安全转义 key、多节点不健康状态过滤、浮点 sleep 兼容降级、SCAN 网络抖动重试机制
# 使用限制: 适用于 Redis 4.0+ (支持 UNLINK 命令)
# ==============================================================================

# ----------------- 1. 配置部分 -----------------
# redis-cli 客户端路径配置 (如果为空，脚本将尝试从系统的 PATH 中寻找)
REDIS_CLI="redis-cli"

# Redis 集群连接信息 (输入集群中的任意一个可用节点作为种子节点即可)
SEED_HOST="127.0.0.1"
SEED_PORT="6379"
REDIS_PASS="" # 如果有密码，在此填写。若无密码，请保持为空 ""

# 要匹配并删除的 Key 的 Pattern
PATTERN="iuap-apcom-bipuser_yht-ml-redis*"

# 生产安全限速配置
BATCH_SIZE=1000        # 单次 SCAN 返回的最大 Key 数量 (降低单次处理大小以减小网络和 CPU 峰值)
SLEEP_INTERVAL="0.1"   # 每次 UNLINK 后的休眠延迟时间（秒），设为 0 表示不休眠。建议生产环境设为 0.1 ~ 0.5 秒

# 验证与限制配置
MAX_DELETE_LIMIT=100000 # 最大删除 Key 数量限制。达到此数量后脚本自动安全终止。设为 0 表示不限制（全量删除）

# 日志配置
LOG_FILE="/home/lin/download/redis_clean.log" # 日志文件保存路径

# ----------------- 2. 初始化与日志函数 -----------------

# 创建日志目录（若不存在）
log_dir=$(dirname "$LOG_FILE")
mkdir -p "$log_dir"

# 日志输出函数，同时输出到终端和日志文件
log_msg() {
    local level=$1
    local msg=$2
    local time_str
    time_str=$(date "+%Y-%m-%d %H:%M:%S")
    echo "[$time_str] [$level] $msg" | tee -a "$LOG_FILE"
}

log_msg "INFO" "=================== 脚本启动 ==================="

# 检查 redis-cli 可执行程序
if ! command -v "$REDIS_CLI" &> /dev/null; then
    if command -v redis-cli &> /dev/null; then
        log_msg "WARNING" "指定的 redis-cli 路径不可用，将使用系统 PATH 中的 redis-cli。"
        REDIS_CLI="redis-cli"
    else
        log_msg "ERROR" "未找到 redis-cli 客户端，请检查 REDIS_CLI 配置路径。"
        exit 1
    fi
fi

# 密码参数组装 (使用 Bash 数组避免密码含有特殊字符或空格时发生命令参数拆分及注入)
REDIS_AUTH_ARGS=()
if [ -n "$REDIS_PASS" ]; then
    REDIS_AUTH_ARGS=(-a "$REDIS_PASS")
fi

# 校验种子节点是否可用
if ! $REDIS_CLI -h "$SEED_HOST" -p "$SEED_PORT" "${REDIS_AUTH_ARGS[@]}" PING &>/dev/null; then
    log_msg "ERROR" "无法连接到种子节点 ${SEED_HOST}:${SEED_PORT}，请检查网络配置或密码是否正确。"
    exit 1
fi

# 检查是否为集群模式
IS_CLUSTER=$($REDIS_CLI -h "$SEED_HOST" -p "$SEED_PORT" "${REDIS_AUTH_ARGS[@]}" info cluster 2>/dev/null | grep "cluster_enabled:1")
if [ -z "$IS_CLUSTER" ]; then
    log_msg "ERROR" "该节点不是集群模式 (cluster_enabled:1)。本脚本专为 Redis Cluster 设计。"
    exit 1
fi

# 检查系统是否支持浮点数 sleep 探针，防止在 Alpine/busybox 等只支持整数休眠的环境中脚本崩溃
if [ -n "$SLEEP_INTERVAL" ] && [ "$SLEEP_INTERVAL" != "0" ]; then
    if ! sleep 0.01 2>/dev/null; then
        log_msg "WARNING" "当前系统环境的 sleep 命令不支持浮点数，已将休眠间隔自动降级调整为 1 秒。"
        SLEEP_INTERVAL=1
    fi
fi

# ----------------- 3. 获取所有 Master 节点 -----------------
log_msg "INFO" "正在获取集群 Master 节点列表..."
# 获取集群节点列表，过滤出正常的 master 节点，排除 fail、handshake 和 noaddr 等异常状态节点
master_nodes=$($REDIS_CLI -h "$SEED_HOST" -p "$SEED_PORT" "${REDIS_AUTH_ARGS[@]}" cluster nodes 2>/dev/null | grep master | grep -v -E "fail|handshake|noaddr" | awk '{print $2}' | cut -d'@' -f1)

if [ -z "$master_nodes" ]; then
    log_msg "ERROR" "无法获取集群正常的 Master 节点列表，请确认集群状态。"
    exit 1
fi

log_msg "INFO" "发现以下健康 Master 节点:"
for node in $master_nodes; do
    log_msg "INFO" "  - $node"
done

log_msg "INFO" "--------------------------------------------------"
log_msg "INFO" "准备删除匹配规则  : '$PATTERN'"
log_msg "INFO" "使用客户端路径    : '$REDIS_CLI'"
log_msg "INFO" "批量删除命令      : 'UNLINK' (非阻塞删除)"
log_msg "INFO" "每次扫描批次大小  : $BATCH_SIZE"
log_msg "INFO" "安全休眠间隔时间  : ${SLEEP_INTERVAL} 秒"
if [ "$MAX_DELETE_LIMIT" -gt 0 ]; then
    log_msg "INFO" "最大删除数量限制  : $MAX_DELETE_LIMIT 个 (达到上限后将自动停止以做验证)"
else
    log_msg "INFO" "最大删除数量限制  : 无限制 (将全量删除所有匹配 Key)"
fi
log_msg "INFO" "日志记录文件      : $LOG_FILE"
log_msg "INFO" "--------------------------------------------------"

# ----------------- 3.5. 统计 Key 数量函数与执行前统计 -----------------
# 全局变量，用于接收统计函数的结果，避免子 Shell 捕获时污染日志输出
COUNT_RESULT=0

# 统计集群中匹配指定 Pattern 的 Key 的总量 (使用 SCAN 避免阻塞)
count_keys_by_pattern() {
    local pattern=$1
    local total_count=0
    local node node_ip node_port cursor node_count retry_count max_retries scan_res keys key_count
    
    for node in $master_nodes; do
        node_ip=$(echo "$node" | cut -d':' -f1)
        node_port=$(echo "$node" | cut -d':' -f2)
        if [ -z "$node_ip" ]; then
            node_ip="$SEED_HOST"
        fi

        cursor="0"
        node_count=0

        while :; do
            retry_count=0
            max_retries=3
            scan_res=""
            
            while [ $retry_count -lt $max_retries ]; do
                scan_res=$($REDIS_CLI -h "$node_ip" -p "$node_port" "${REDIS_AUTH_ARGS[@]}" --raw SCAN "$cursor" MATCH "$pattern" COUNT "$BATCH_SIZE" 2>/dev/null)
                if [ $? -eq 0 ]; then
                    break
                fi
                retry_count=$((retry_count + 1))
                
                # 兼容不支持浮点 sleep 的环境（如 Alpine Linux / BusyBox）
                if [ "$SLEEP_INTERVAL" = "1" ]; then
                    sleep 1
                else
                    sleep 0.5
                fi
            done

            if [ $retry_count -eq $max_retries ]; then
                log_msg "WARNING" "统计时连接节点 ${node_ip}:${node_port} SCAN 失败，跳过该节点统计。"
                break
            fi

            cursor=$(echo "$scan_res" | head -n 1)
            keys=$(echo "$scan_res" | tail -n +2)

            if [ -n "$keys" ]; then
                key_count=$(echo "$keys" | wc -l | awk '{print $1}')
                node_count=$((node_count + key_count))
            fi

            if [ "$cursor" = "0" ] || [ -z "$cursor" ]; then
                break
            fi
        done
        total_count=$((total_count + node_count))
    done
    COUNT_RESULT=$total_count
}

log_msg "INFO" "正在统计当前匹配模式 '$PATTERN' 的 Key 总量，请稍候 (若 Key 数量巨大，此过程可能较慢)..."
count_keys_by_pattern "$PATTERN"
PRE_DELETE_COUNT=$COUNT_RESULT
log_msg "INFO" "当前匹配的 Key 总量为: $PRE_DELETE_COUNT"
log_msg "INFO" "--------------------------------------------------"

# 支持非交互模式下的批量运行
read -p "确定要开始删除操作吗？这可能会影响生产数据！(y/N): " confirm

if [[ "$confirm" != "y" && "$confirm" != "Y" ]]; then
    log_msg "INFO" "用户取消了删除操作。"
    exit 0
fi

# ----------------- 4. 遍历 Master 节点执行删除 -----------------
total_all_deleted=0

# 标志是否因达到删除上限而终止
limit_reached=false

for node in $master_nodes; do
    # 解析 IP 和 Port
    node_ip=$(echo "$node" | cut -d':' -f1)
    node_port=$(echo "$node" | cut -d':' -f2)

    if [ -z "$node_ip" ]; then
        node_ip="$SEED_HOST"
    fi

    log_msg "INFO" "开始扫描并删除节点 ${node_ip}:${node_port} 上的匹配 Key..."

    cursor="0"
    node_deleted=0

    while :; do
        # 1. 使用 SCAN 命令扫描 Key，增加抖动重试机制，避免临时网卡丢包导致数据漏删
        retry_count=0
        max_retries=3
        scan_res=""
        
        while [ $retry_count -lt $max_retries ]; do
            scan_res=$($REDIS_CLI -h "$node_ip" -p "$node_port" "${REDIS_AUTH_ARGS[@]}" --raw SCAN "$cursor" MATCH "$PATTERN" COUNT "$BATCH_SIZE" 2>/dev/null)
            if [ $? -eq 0 ]; then
                break
            fi
            retry_count=$((retry_count + 1))
            log_msg "WARNING" "连接节点 ${node_ip}:${node_port} 执行 SCAN 失败，正在进行第 ${retry_count} 次重试..."
            sleep 1
        done

        if [ $retry_count -eq $max_retries ]; then
            log_msg "ERROR" "连接节点 ${node_ip}:${node_port} 连续重试 ${max_retries} 次均失败，跳过此节点后续扫描。"
            break
        fi

        # 2. 提取下一游标及 keys
        cursor=$(echo "$scan_res" | head -n 1)
        keys=$(echo "$scan_res" | tail -n +2)

        if [ -n "$keys" ]; then
            # 统计当前批次扫描出来的 key 数量
            key_count=$(echo "$keys" | wc -l | awk '{print $1}')
            
            if [ "$key_count" -gt 0 ]; then
                # 3. 检查是否超出最大删除数量限制
                to_delete_keys="$keys"
                to_delete_count=$key_count
                
                if [ "$MAX_DELETE_LIMIT" -gt 0 ]; then
                    temp_total=$((total_all_deleted + key_count))
                    if [ "$temp_total" -ge "$MAX_DELETE_LIMIT" ]; then
                        to_delete_count=$((MAX_DELETE_LIMIT - total_all_deleted))
                        to_delete_keys=$(echo "$keys" | head -n "$to_delete_count")
                        limit_reached=true
                    fi
                fi

                if [ "$to_delete_count" -gt 0 ]; then
                    # 4. 执行 UNLINK
                    # 使用 awk 进行高强度防御性转义：将 key 中的 \ 替换为 \\，" 替换为 \"
                    # 避免键名中带有空格、引号、反斜线或换行导致命令截断或注入漏洞
                    echo "$to_delete_keys" | awk '{ gsub(/\\/, "\\\\"); gsub(/"/, "\\\""); print "UNLINK", "\""$0"\"" }' | $REDIS_CLI -h "$node_ip" -p "$node_port" "${REDIS_AUTH_ARGS[@]}" --raw >/dev/null 2>&1
                    
                    node_deleted=$((node_deleted + to_delete_count))
                    total_all_deleted=$((total_all_deleted + to_delete_count))
                    
                    log_msg "INFO" "  节点 ${node_ip}:${node_port} 本次删除 ${to_delete_count} 个 Key (已累计删除: ${total_all_deleted}，当前游标: ${cursor})"
                fi
                
                if [ "$limit_reached" = true ]; then
                    log_msg "INFO" "已达到设定的最大删除限制数量 (${MAX_DELETE_LIMIT} 条)，正在安全终止操作..."
                    break 2
                fi

                # 5. 限速休眠机制
                if [ -n "$SLEEP_INTERVAL" ] && [ "$SLEEP_INTERVAL" != "0" ]; then
                    sleep "$SLEEP_INTERVAL"
                fi
            fi
        fi

        # 游标为 0 代表该节点 SCAN 结束
        if [ "$cursor" = "0" ] || [ -z "$cursor" ]; then
            break
        fi
    done

    log_msg "INFO" "节点 ${node_ip}:${node_port} 处理完毕，该节点共删除: ${node_deleted} 个 keys。"
done

log_msg "INFO" "正在统计删除后剩余匹配模式 '$PATTERN' 的 Key 数量，请稍候..."
count_keys_by_pattern "$PATTERN"
POST_DELETE_COUNT=$COUNT_RESULT

log_msg "INFO" "=================== 任务总结 ==================="
log_msg "INFO" "所有 Master 节点处理完成！"
log_msg "INFO" "删除前 Key 总数  : ${PRE_DELETE_COUNT}"
log_msg "INFO" "累计删除 Key 总数  : ${total_all_deleted}"
log_msg "INFO" "删除后剩余 Key 总数: ${POST_DELETE_COUNT}"
if [ "$limit_reached" = true ]; then
    log_msg "INFO" "状态: 因达到删除上限限制 (${MAX_DELETE_LIMIT} 条) 自动终止，验证成功。"
else
    log_msg "INFO" "状态: 全量删除已完成。"
fi
log_msg "INFO" "日志保存在: $LOG_FILE"
log_msg "INFO" "=================================================="