package com.fs.common.utils.redis;

import com.fs.common.constant.CourseWatchKeysConstant;
import com.fs.common.constant.LiveKeysConstant;
import com.fs.common.constant.VideoCommentKeysConstant;
import com.fs.common.core.redis.RedisCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 使用 SCAN 将 Redis 中真实存在的活跃 key 与 Set 索引对账（兜底，避免漏 track）。
 */
@Service
public class RedisActiveKeyIndexRepairService {

    private static final Logger log = LoggerFactory.getLogger(RedisActiveKeyIndexRepairService.class);

    private static final String REPAIR_LOCK_KEY = "redis:active_key_index:repair:lock";
    private static final int SCAN_COUNT = 500;
    private static final int LOCK_MINUTES = 45;

    @Autowired
    private RedisCache redisCache;

    /**
     * 修补全部已注册的索引（多节点仅一个实例执行）。
     */
    public void repairAllIndexes() {
        if (!tryAcquireRepairLock()) {
            log.info("Redis 活跃 key 索引修补跳过：其他节点正在执行");
            return;
        }
        try {
            log.info("Redis 活跃 key 索引修补开始");
            List<RepairResult> results = new ArrayList<>();
            for (IndexRepairTarget target : repairTargets()) {
                results.add(repairIndex(target));
            }
            log.info("Redis 活跃 key 索引修补完成: {}", results);
        } catch (Exception e) {
            log.error("Redis 活跃 key 索引修补失败", e);
        } finally {
            redisCache.deleteObject(REPAIR_LOCK_KEY);
        }
    }

    public RepairResult repairIndex(IndexRepairTarget target) {
        Set<String> scannedKeys = scanKeys(target.scanPattern);
        int added = 0;
        for (String cacheKey : scannedKeys) {
            if (Boolean.TRUE.equals(redisCache.redisTemplate.hasKey(cacheKey))) {
                Long addedCount = redisCache.redisTemplate.opsForSet().add(target.indexKey, cacheKey);
                if (addedCount != null && addedCount > 0) {
                    added++;
                }
            }
        }

        int removed = 0;
        if (target.isRemoveExpiredFromIndex()) {
            Set<Object> members = redisCache.redisTemplate.opsForSet().members(target.indexKey);
            if (members != null) {
                for (Object member : members) {
                    String cacheKey = member.toString();
                    if (!Boolean.TRUE.equals(redisCache.redisTemplate.hasKey(cacheKey))) {
                        redisCache.redisTemplate.opsForSet().remove(target.indexKey, cacheKey);
                        removed++;
                    }
                }
            }
        } else {
            log.info("索引 {} 跳过过期成员清理，交由 checkFsUserWatchStatus 写库后 untrack", target.name);
        }
        return new RepairResult(target.name, target.scanPattern, target.indexKey, scannedKeys.size(), added, removed);
    }

    private List<IndexRepairTarget> repairTargets() {
        List<IndexRepairTarget> targets = new ArrayList<>();
        targets.add(new IndexRepairTarget("liveAutoTask",
                LiveKeysConstant.LIVE_AUTO_TASK_PREFIX + "*",
                LiveKeysConstant.LIVE_AUTO_TASK_INDEX));
        targets.add(new IndexRepairTarget("liveLotteryTask",
                LiveKeysConstant.LIVE_LOTTERY_TASK_PREFIX + "*",
                LiveKeysConstant.LIVE_LOTTERY_TASK_INDEX));
        targets.add(new IndexRepairTarget("liveRedTask",
                LiveKeysConstant.LIVE_RED_TASK_PREFIX + "*",
                LiveKeysConstant.LIVE_RED_TASK_INDEX));
        targets.add(new IndexRepairTarget("liveTagMark",
                LiveKeysConstant.LIVE_TAG_MARK_CACHE.replace("%s", "*"),
                LiveKeysConstant.LIVE_TAG_MARK_INDEX));
        targets.add(new IndexRepairTarget("liveCouponNum",
                LiveKeysConstant.LIVE_COUPON_NUM_PREFIX + "*",
                LiveKeysConstant.LIVE_COUPON_NUM_INDEX));
        targets.add(new IndexRepairTarget("liveUserWatchLog",
                "live:user:watch:log:*",
                LiveKeysConstant.LIVE_USER_WATCH_LOG_INDEX));
        // 心跳索引只补不删：过期 key 由 checkFsUserWatchStatus 写 log_type=4 后再 untrack
        targets.add(new IndexRepairTarget("h5wxHeartbeat",
                CourseWatchKeysConstant.H5_WX_HEARTBEAT_PREFIX + "*",
                CourseWatchKeysConstant.H5_WX_HEARTBEAT_INDEX,
                false));
        targets.add(new IndexRepairTarget("h5wxDuration",
                CourseWatchKeysConstant.H5_WX_DURATION_PREFIX + "*",
                CourseWatchKeysConstant.H5_WX_DURATION_INDEX));
        targets.add(new IndexRepairTarget("videoCommentCount",
                VideoCommentKeysConstant.VIDEO_COMMENT_COUNT_PREFIX + "*",
                VideoCommentKeysConstant.VIDEO_COMMENT_COUNT_INDEX));
        return targets;
    }

    private Set<String> scanKeys(String pattern) {
        Set<String> keys = new HashSet<>();
        ScanOptions options = ScanOptions.scanOptions().match(pattern).count(SCAN_COUNT).build();
        redisCache.redisTemplate.execute((RedisCallback<Void>) connection -> {
            RedisSerializer<String> keySerializer = redisCache.redisTemplate.getStringSerializer();
            try (Cursor<byte[]> cursor = connection.scan(options)) {
                while (cursor.hasNext()) {
                    String key = keySerializer.deserialize(cursor.next());
                    if (key != null) {
                        keys.add(key);
                    }
                }
            } catch (IOException e) {
                throw new RuntimeException("Redis SCAN failed, pattern=" + pattern, e);
            }
            return null;
        });
        return keys;
    }

    private boolean tryAcquireRepairLock() {
        return redisCache.setIfAbsent(REPAIR_LOCK_KEY, "1", LOCK_MINUTES, TimeUnit.MINUTES);
    }

    public static final class IndexRepairTarget {
        private final String name;
        private final String scanPattern;
        private final String indexKey;
        /** false 时仅 SCAN 补登记，不清理索引中已 TTL 过期的成员 */
        private final boolean removeExpiredFromIndex;

        public IndexRepairTarget(String name, String scanPattern, String indexKey) {
            this(name, scanPattern, indexKey, true);
        }

        public IndexRepairTarget(String name, String scanPattern, String indexKey, boolean removeExpiredFromIndex) {
            this.name = name;
            this.scanPattern = scanPattern;
            this.indexKey = indexKey;
            this.removeExpiredFromIndex = removeExpiredFromIndex;
        }

        public boolean isRemoveExpiredFromIndex() {
            return removeExpiredFromIndex;
        }
    }

    public static final class RepairResult {
        private final String name;
        private final String scanPattern;
        private final String indexKey;
        private final int scannedCount;
        private final int addedToIndex;
        private final int removedFromIndex;

        public RepairResult(String name, String scanPattern, String indexKey,
                            int scannedCount, int addedToIndex, int removedFromIndex) {
            this.name = name;
            this.scanPattern = scanPattern;
            this.indexKey = indexKey;
            this.scannedCount = scannedCount;
            this.addedToIndex = addedToIndex;
            this.removedFromIndex = removedFromIndex;
        }

        @Override
        public String toString() {
            return name + "{scanned=" + scannedCount + ", added=" + addedToIndex + ", removed=" + removedFromIndex + "}";
        }
    }
}
