package com.njcn.system.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import com.njcn.common.pojo.constant.LogInfo;
import com.njcn.common.pojo.constant.PatternRegex;
import com.njcn.common.pojo.dto.LogInfoDTO;
import com.njcn.common.utils.PubUtils;
import com.njcn.redis.pojo.enums.RedisKeyEnum;
import com.njcn.redis.utils.RedisMessageQueueUtil;
import com.njcn.redis.utils.RedisUtil;
import com.njcn.system.service.IUserLogService;
import com.njcn.user.api.UserFeignClient;
import com.njcn.user.pojo.po.User;
import com.njcn.web.utils.EmailUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * Redis 日志消费者
 * 替代原 MQTT 订阅方式，从 Redis List 中消费日志消息
 *
 * 核心优化：
 * 1. 多线程并发消费（3个用户日志线程 + 1个邮件线程）
 * 2. 智能批量入库（满50条或超时2秒自动刷新）
 * 3. 实时性保证（BRPOP阻塞式消费，延迟<2秒）
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisLogConsumer {

    private final RedisMessageQueueUtil redisMessageQueueUtil;
    private final IUserLogService userLogService;
    private final UserFeignClient userFeignClient;
    private final RedisUtil redisUtil;
    private final EmailUtil emailUtil;

    private ExecutorService executorService;
    private volatile boolean running = true;

    /**
     * 用户角色类型缓存（用于判断日志类型）
     */
    private JSONObject userRoleTypeCache = null;

    /**
     * 批量阈值：每批最多50条
     */
    private static final int BATCH_SIZE = 50;

    /**
     * 超时阈值：2秒未满50条也强制刷新
     */
    private static final long FLUSH_TIMEOUT_MS = 2000;

    @PostConstruct
    public void startConsume() {
        log.info("启动 Redis 日志消费者...");

        // 创建消费线程（3个用户日志 + 1个邮件推送 + 1个设备日志，根据需要调整）
        executorService = Executors.newFixedThreadPool(5, r -> {
            Thread thread = new Thread(r);
            thread.setDaemon(false); // 非守护线程，确保消息消费完
            return thread;
        });

        // 启动3个用户日志消费线程
        for (int i = 0; i < 3; i++) {
            final int threadId = i;
            executorService.submit(() -> consumeUserLogWithBatch(threadId));
        }

        // 启动1个邮件推送消费线程
        executorService.submit(this::consumeEmailPush);

        // 启动1个设备日志消费线程（可选）
        // executorService.submit(this::consumeDeviceLog);

        log.info("Redis 日志消费者启动完成（3个用户日志线程 + 1个邮件线程）");
    }

    /**
     * 消费用户日志（带批量优化）
     */
    private void consumeUserLogWithBatch(int threadId) {
        log.info("用户日志消费线程-{} 启动", threadId);

        List<String> buffer = new ArrayList<>(BATCH_SIZE);
        long lastFlushTime = System.currentTimeMillis();

        while (running) {
            try {
                // BRPOP 阻塞1秒（缩短超时时间，用于定期检查批量刷新条件）
                String logJson = redisMessageQueueUtil.popMessage(
                        RedisKeyEnum.USER_LOG_QUEUE.getKey(),
                        1
                );

                // 收集到消息，加入缓冲区
                if (logJson != null) {
                    buffer.add(logJson);
                }

                // 批量刷新条件：满50条 OR 距离上次刷新超过2秒
                boolean shouldFlush = buffer.size() >= BATCH_SIZE ||
                        (System.currentTimeMillis() - lastFlushTime > FLUSH_TIMEOUT_MS && !buffer.isEmpty());

                if (shouldFlush) {
                    batchProcessUserLogs(buffer, threadId);
                    buffer.clear();
                    lastFlushTime = System.currentTimeMillis();
                }

            } catch (Exception e) {
                log.error("线程-{} 消费用户日志失败", threadId, e);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }

        // 线程退出前，刷新剩余缓冲区
        if (!buffer.isEmpty()) {
            batchProcessUserLogs(buffer, threadId);
        }

        log.info("用户日志消费线程-{} 退出", threadId);
    }

    /**
     * 批量处理用户日志
     */
    private void batchProcessUserLogs(List<String> logJsonList, int threadId) {
        if (logJsonList.isEmpty()) {
            return;
        }

        try {
            // 确保用户角色类型缓存已加载
            loadUserRoleTypeCache();

            // 批量解析并处理
            for (String logJson : logJsonList) {
                try {
                    LogInfoDTO logInfoDTO = PubUtils.json2obj(logJson, LogInfoDTO.class);
                    // 过滤无效日志
                    if (logInfoDTO.getLoginName().equals(LogInfo.UNKNOWN_USER)
                            || logInfoDTO.getOperate().equals(LogInfo.UNKNOWN_OPERATE)
                            || logInfoDTO.getIp().equals(LogInfo.UNKNOWN_IP)) {
                        continue;
                    }
                    // 设置日志类型（业务事件=0，系统事件=1）
                    if ("注销".equals(logInfoDTO.getOperateType()) || "认证".equals(logInfoDTO.getOperateType())) {
                        logInfoDTO.setType(1); // 系统事件
                    } else {
                        String loginName = logInfoDTO.getLoginName();
                        if (StrUtil.isNotBlank(loginName) && userRoleTypeCache.containsKey(loginName)) {
                            if ((Integer) userRoleTypeCache.get(loginName) == 2) {
                                logInfoDTO.setType(0); // 业务事件
                            } else {
                                logInfoDTO.setType(1); // 系统事件
                            }
                        } else {
                            logInfoDTO.setType(1); // 默认系统事件
                        }
                    }
                    // 入库
                    userLogService.addUserLog(logInfoDTO);
                } catch (Exception e) {
                    log.error("解析日志失败: {}", logJson, e);
                }
            }

            log.info("线程-{} 批量入库 {} 条日志", threadId, logJsonList.size());

        } catch (Exception e) {
            log.error("线程-{} 批量处理日志失败，涉及 {} 条", threadId, logJsonList.size(), e);
        }
    }

    /**
     * 加载用户角色类型缓存
     */
    private void loadUserRoleTypeCache() {
        if (Objects.isNull(userRoleTypeCache)) {
            userRoleTypeCache = (JSONObject) redisUtil.getObjectByKey(RedisKeyEnum.USER_ROLE_TYPE_KEY.getKey());
            if (Objects.isNull(userRoleTypeCache)) {
                // 调用 Feign 接口获取用户角色列表（会自动写入 Redis）
                userFeignClient.userRoleList();
                userRoleTypeCache = (JSONObject) redisUtil.getObjectByKey(RedisKeyEnum.USER_ROLE_TYPE_KEY.getKey());
            }
        }
    }

    /**
     * 降级处理：批量失败时逐条处理（已移除，直接在批量方法中逐条调用）
     */
    private void fallbackProcessOneByOne(List<String> logJsonList, int threadId) {
        log.warn("线程-{} 批量入库失败，降级为逐条处理", threadId);

        // 确保用户角色类型缓存已加载
        loadUserRoleTypeCache();

        for (String logJson : logJsonList) {
            try {
                LogInfoDTO logInfoDTO = PubUtils.json2obj(logJson, LogInfoDTO.class);

                // 过滤无效日志
                if (logInfoDTO.getLoginName().equals(LogInfo.UNKNOWN_USER)
                        || logInfoDTO.getOperate().equals(LogInfo.UNKNOWN_OPERATE)
                        || logInfoDTO.getIp().equals(LogInfo.UNKNOWN_IP)) {
                    continue;
                }

                // 设置日志类型
                if ("注销".equals(logInfoDTO.getOperateType()) || "认证".equals(logInfoDTO.getOperateType())) {
                    logInfoDTO.setType(1);
                } else {
                    String loginName = logInfoDTO.getLoginName();
                    if (StrUtil.isNotBlank(loginName) && userRoleTypeCache.containsKey(loginName)) {
                        if ((Integer) userRoleTypeCache.get(loginName) == 2) {
                            logInfoDTO.setType(0);
                        } else {
                            logInfoDTO.setType(1);
                        }
                    } else {
                        logInfoDTO.setType(1);
                    }
                }

                userLogService.addUserLog(logInfoDTO);

            } catch (Exception e) {
                log.error("线程-{} 单条处理也失败: {}", threadId, logJson, e);
            }
        }
    }

    /**
     * 消费邮件推送（无需批量优化，邮件量少）
     */
    private void consumeEmailPush() {
        log.info("邮件推送消费线程启动");

        while (running) {
            try {
                String pushJson = redisMessageQueueUtil.popMessage(
                        RedisKeyEnum.USER_LOG_EMAIL_QUEUE.getKey(),
                        3 // 邮件推送不频繁，5秒超时即可
                );
                if (pushJson != null) {
                    log.debug("消费邮件推送: {}", pushJson);
                    // 解析日志DTO
                    LogInfoDTO auditLogVO = PubUtils.json2obj(pushJson, LogInfoDTO.class);
                    // 获取审计管理员的邮箱列表
                    List<User> data = userFeignClient.getUserListByRoleCode("audit_manager").getData();
                    List<String> emails = data.stream()
                            .filter(x -> Pattern.matches(PatternRegex.EMAIL_REGEX, x.getEmail()))
                            .filter(x -> StrUtil.isNotBlank(x.getEmail()) && x.getEmailNotice() == 1)
                            .map(User::getEmail)
                            .collect(Collectors.toList());
                    // 发送邮件
                    if (CollUtil.isNotEmpty(emails)) {
                        StringBuilder describe = new StringBuilder();
                        describe.append(auditLogVO.getType() == 0 ? "业务事件 -> " : "系统事件 -> ");
                        describe.append(auditLogVO.getUserName()).append("在")
                                .append(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(auditLogVO.getCreateTime()))
                                .append("在")
                                .append(auditLogVO.getIp())
                                .append("执行了")
                                .append(auditLogVO.getOperate())
                                .append("，结果为");
                        if (auditLogVO.getResult() == 1) {
                            describe.append("成功");
                        }
                        if (auditLogVO.getResult() == 0) {
                            describe.append("失败").append("，失败原因为").append(auditLogVO.getFailReason());
                        }
                        emailUtil.sendMultiple(emails, "告警消息", describe.toString(), false);
                    }
                }

            } catch (Exception e) {
                log.error("消费邮件推送失败", e);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }

        log.info("邮件推送消费线程退出");
    }

    /**
     * 消费设备日志（可选，如果需要处理设备日志）
     */
    private void consumeDeviceLog() {
        log.info("设备日志消费线程启动");

        while (running) {
            try {
                String deviceLogJson = redisMessageQueueUtil.popMessage(
                        RedisKeyEnum.DEVICE_LOG_QUEUE.getKey(),
                        5 // 设备日志不频繁，5秒超时即可
                );

                if (deviceLogJson != null) {
                    log.debug("消费设备日志: {}", deviceLogJson);
                    // TODO: 解析并保存设备日志
                    // DeviceLog deviceLog = JSONUtil.toBean(deviceLogJson, DeviceLog.class);
                    // deviceLogService.save(deviceLog);
                }

            } catch (Exception e) {
                log.error("消费设备日志失败", e);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }

        log.info("设备日志消费线程退出");
    }

    @PreDestroy
    public void shutdown() {
        log.info("开始停止 Redis 日志消费者...");
        running = false;

        if (executorService != null) {
            executorService.shutdown();
            try {
                // 等待60秒，让队列中的消息处理完
                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                    log.warn("消费线程未在60秒内完成，强制关闭");
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                log.error("等待消费线程关闭被中断", e);
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }

        log.info("Redis 日志消费者已停止");
    }
}