package com.fs.live.utils.redis;

import com.alibaba.fastjson.JSON;
import com.fs.live.domain.LiveTrafficLog;
import com.fs.live.mapper.LiveTrafficLogMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

@Slf4j
@Component
public class RedisBatchHandler {
    @Autowired
    public RedisTemplate redisTemplate;
    @Autowired
    public LiveTrafficLogMapper liveTrafficLogMapper;
    // Redis队列key
    private static final String BATCH_QUEUE_KEY = "live_traffic:batch:queue";
    // 批量拉取条数
    private static final int BATCH_PULL_SIZE = 500;
    // 定时消费间隔
    public static final long CONSUME_INTERVAL = 500;

    // 1. 入队：多服务实例写入Redis List
    public void addDataToRedis(LiveTrafficLog data) {
        try {
            // 序列化数据（用JSON）
            String json = JSON.toJSONString(data);
            redisTemplate.opsForList().rightPush(BATCH_QUEUE_KEY, json);
        } catch (Exception e) {
            // 降级为本地队列/单条插入
            log.error("Redis入队失败，降级处理", e);
        }
    }

    // 2. 定时消费：独立线程拉取批量数据写入MySQL
    public void consumeBatchData() {
        // 批量拉取Redis List中的数据（非阻塞）
        List<String> jsonList = redisTemplate.opsForList().range(BATCH_QUEUE_KEY, 0, BATCH_PULL_SIZE - 1);
        if (jsonList == null || jsonList.isEmpty()) {
            return;
        }
        // 转换为实体类
        List<LiveTrafficLog> batchList = jsonList.stream()
                .map(json -> JSON.parseObject(json, LiveTrafficLog.class))
                .collect(Collectors.toList());
        int size = batchList.size();

        Map<String, LiveTrafficLog> uniqueMap = batchList.stream()
                .collect(Collectors.toMap(
                        // 分组key：uuId（为null的话用一个特殊值，避免分组异常）
                        log -> log.getUuId()== null ? "NULL_UUID_" + System.currentTimeMillis() : log.getUuId(),
                        // value：当前实体
                        log -> log,
                        // 合并规则：如果key重复，保留internetTraffic更大的那个
                        (log1, log2) -> {
                            // 处理internetTraffic为null的情况，视为0
                            long traffic1 = log1.getInternetTraffic() == null ? 0L : log1.getInternetTraffic();
                            long traffic2 = log2.getInternetTraffic() == null ? 0L : log2.getInternetTraffic();
                            // 比较流量，返回更大的那个；相等时返回任意一个（这里返回log2）
                            return traffic2 > traffic1 ? log2 : log1;
                        }
                ));
        batchList = new ArrayList<>(uniqueMap.values());
        try {
            // 批量写入数据库
            liveTrafficLogMapper.batchInsert(batchList);
            // 删除已消费的数据
            redisTemplate.opsForList().trim(BATCH_QUEUE_KEY, size, -1);
        } catch (Exception e) {
            // 消费失败：记录日志，不删除Redis数据（重试）
            log.error("Redis批量消费失败，将重试", e);
        }
    }
}