package com.njcn.gather.detection.util.socket.cilent;

import com.alibaba.fastjson.JSON;
import com.njcn.gather.detection.pojo.enums.SourceOperateCodeEnum;
import com.njcn.gather.detection.pojo.param.PreDetectionParam;
import com.njcn.gather.detection.pojo.vo.SocketDataMsg;
import com.njcn.gather.detection.pojo.vo.SocketMsg;
import com.njcn.gather.detection.util.socket.CnSocketUtil;
import com.njcn.gather.detection.util.socket.FormalTestManager;
import com.njcn.gather.detection.util.socket.MsgUtil;
import com.njcn.gather.detection.util.socket.SocketManager;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.concurrent.*;

/**
 * Netty心跳处理器
 * <p>
 * 负责维护Socket长连接的心跳检测机制，通过定期发送心跳包来检测连接状态，
 * 当连续多次未收到心跳响应时自动断开连接并清理相关资源。
 * </p>
 *
 * <h3>核心功能:</h3>
 * <ul>
 *   <li>定时发送心跳包 (默认10秒间隔，3秒后开始)</li>
 *   <li>监听心跳响应，重置失败计数器</li>
 *   <li>连续失败超过阈值时触发断开逻辑 (默认3次)</li>
 *   <li>异步处理断开操作，避免阻塞心跳线程</li>
 *   <li>优雅关闭资源，防止内存泄漏</li>
 * </ul>
 *
 * <h3>心跳机制流程:</h3>
 * <pre>
 * 连接建立 → 启动心跳定时任务(3秒后开始，每10秒执行)
 *           ↓
 * 发送心跳包 → 等待响应 → 收到响应(重置计数器) / 未收到响应(递增计数器)
 *           ↓
 * 连续3次失败 → 异步执行断开逻辑 → 发送退出指令 → 延迟清理连接
 *           ↓
 * 连接断开 → 优雅关闭定时任务和线程池
 * </pre>
 *
 * <h3>线程安全设计:</h3>
 * <p>
 * 使用单线程的ScheduledExecutorService处理心跳发送，避免并发问题。
 * 超时处理使用CompletableFuture异步执行，不阻塞心跳发送线程。
 * Future引用使用volatile修饰，确保多线程环境下的可见性。
 * </p>
 *
 * <h3>设备类型支持:</h3>
 * <ul>
 *   <li>程控源设备 (CnSocketUtil.SOURCE_TAG): "_Source"</li>
 *   <li>被检设备 (CnSocketUtil.DEV_TAG): "_Dev"</li>
 * </ul>
 *
 * <h3>使用示例:</h3>
 * <pre>{@code
 * // 创建心跳处理器
 * HeartbeatHandler handler = new HeartbeatHandler(param, CnSocketUtil.SOURCE_TAG);
 *
 * // 添加到Netty管道中
 * pipeline.addLast(handler);
 * }</pre>
 *
 * @author cdf
 * @version 1.2
 * @since 2025-02-11
 */
@Slf4j
public class HeartbeatHandler extends SimpleChannelInboundHandler<String> {

    /**
     * 心跳定时任务执行器，使用单线程池避免并发问题
     */
    private final ScheduledExecutorService heartbeatExecutor = Executors.newScheduledThreadPool(1);

    /**
     * 检测参数，包含用户页面ID等信息
     */
    private final PreDetectionParam param;

    /**
     * 处理器类型标识（"_Source" 或 "_Dev"）
     */
    private final String handlerType;

    /**
     * 保存定时任务的Future引用，便于取消和管理
     */
    private ScheduledFuture<?> heartbeatFuture;

    /**
     * 允许连续未收到心跳响应的最大次数
     */
    private static final int MAX_HEARTBEAT_MISSES = 3;

    /**
     * 连续未收到心跳响应的次数
     */
    private int consecutiveHeartbeatMisses = 0;


    /**
     * 构造函数
     *
     * @param param 检测参数，包含用户页面ID等信息
     * @param type  处理器类型（CnSocketUtil.SOURCE_TAG 或 CnSocketUtil.DEV_TAG）
     */
    public HeartbeatHandler(PreDetectionParam param, String type) {
        this.param = param;
        this.handlerType = type;
    }


    /**
     * 通道激活时的回调方法
     * 在Socket连接建立成功后被调用，启动心跳机制
     *
     * @param ctx Netty的通道上下文对象
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        log.info("心跳处理器启动 - 设备类型: {}", handlerType);
        // 启动心跳定时任务
        scheduleHeartbeat(ctx);
        // 传播事件给管道中的后续处理器
        ctx.fireChannelActive();
    }

    /**
     * 通道断开时的回调方法
     * 在Socket连接断开时被调用，负责清理相关资源
     *
     * @param ctx Netty的通道上下文对象
     * @throws Exception 异常情况
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("心跳处理器开始清理资源 - 设备类型: {}", handlerType);
        shutdownExecutorGracefully();
        super.channelInactive(ctx);
    }

    /**
     * 启动心跳定时任务
     * 每10秒发送一次心跳包，3秒后开始执行
     *
     * @param ctx Netty的通道上下文对象
     */
    private void scheduleHeartbeat(ChannelHandlerContext ctx) {
        heartbeatFuture = heartbeatExecutor.scheduleAtFixedRate(() -> {
            if (ctx.channel().isActive()) {
                // 发送心跳包
                SocketMsg<String> msg = new SocketMsg<>();
                msg.setRequestId("yxt");
                msg.setOperateCode(SourceOperateCodeEnum.HEARTBEAT.getValue());
                msg.setData("");
                ctx.channel().writeAndFlush(JSON.toJSONString(msg) + "\n");

                log.debug("心跳发送 - 设备类型: {}, 时间: {}", handlerType, LocalDateTime.now());
                consecutiveHeartbeatMisses++;
                if (consecutiveHeartbeatMisses >= MAX_HEARTBEAT_MISSES) {
                    // 连续三次未收到心跳响应，异步处理断开逻辑，避免阻塞心跳线程
                    log.warn("心跳响应超时 - 设备类型: {}, 连续失败次数: {}/{}, 执行断开连接",
                            handlerType, consecutiveHeartbeatMisses, MAX_HEARTBEAT_MISSES);
                    handleHeartbeatTimeoutAsync();
                    consecutiveHeartbeatMisses = 0; // 重置连续心跳丢失次数
                }
            }
        }, 3, 10, TimeUnit.SECONDS);
    }

    /**
     * 异步处理心跳超时断开逻辑
     * <p>
     * 使用CompletableFuture避免阻塞心跳发送线程，确保心跳机制不受影响。
     * 处理流程：
     * 1. 异步发送退出指令
     * 2. 延迟3秒后清理Socket连接
     * 3. 记录处理过程和异常
     * </p>
     */
    private void handleHeartbeatTimeoutAsync() {
        CompletableFuture.runAsync(() -> {
            try {
                log.info("开始执行心跳超时断开处理 - 设备类型: {}", handlerType);
                // 根据设备类型发送对应的退出指令
                if (CnSocketUtil.DEV_TAG.equals(handlerType)) {
                    CnSocketUtil.quitSend(param);
                } else if (CnSocketUtil.SOURCE_TAG.equals(handlerType)) {
                    CnSocketUtil.quitSendSource(param);
                } else {
                    if (FormalTestManager.currentStep == SourceOperateCodeEnum.RECORD_WAVE_STEP1) {
                        CnSocketUtil.contrastSendquit(param.getUserPageId(), SourceOperateCodeEnum.QUIT_INIT_03, true);
                    } else {
                        CnSocketUtil.contrastSendquit(param.getUserPageId(), SourceOperateCodeEnum.QUIT_INIT_02, true);
                    }
                }
                log.debug("退出指令已发送，等待3秒后清理连接 - 设备类型: {}", handlerType);
            } catch (Exception e) {
                log.error("心跳超时处理发送退出指令异常 - 设备类型: {}", handlerType, e);
            }
        }).thenRunAsync(() -> {
            try {
                // 延迟3秒后清理连接，给退出指令留出处理时间
                Thread.sleep(3000);
                // 构建连接Key并从SocketManager中移除
                String key = CnSocketUtil.DEV_TAG.equals(handlerType) ?
                        param.getUserPageId() + CnSocketUtil.DEV_TAG :
                        param.getUserPageId() + CnSocketUtil.SOURCE_TAG;
                SocketManager.removeUser(key);
                log.info("心跳超时断开处理完成 - 设备类型: {}, 连接已清理", handlerType);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error("心跳超时处理等待过程中被中断 - 设备类型: {}", handlerType);
            } catch (Exception e) {
                log.error("心跳超时处理清理连接异常 - 设备类型: {}", handlerType, e);
            }
        });
    }

    /**
     * 优雅关闭线程池执行器
     * <p>
     * 确保资源的完全清理，避免内存泄漏：
     * 1. 取消当前的心跳定时任务
     * 2. 关闭线程池并等待正在执行的任务完成
     * 3. 如果等待超时则强制关闭
     * 4. 处理中断异常并恢复中断状态
     * </p>
     */
    private void shutdownExecutorGracefully() {
        try {
            // 1. 取消心跳定时任务
            if (heartbeatFuture != null && !heartbeatFuture.isCancelled()) {
                boolean cancelled = heartbeatFuture.cancel(false);
                log.debug("心跳定时任务取消结果: {} - 设备类型: {}", cancelled, handlerType);
            }
            // 2. 关闭线程池，不再接收新任务
            heartbeatExecutor.shutdown();
            // 3. 等待已提交的任务完成，最多等待5秒
            if (!heartbeatExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                log.warn("心跳线程池未能在5秒内正常关闭，执行强制关闭 - 设备类型: {}", handlerType);
                heartbeatExecutor.shutdownNow();
                // 再次等待强制关闭完成，最多等待2秒
                if (!heartbeatExecutor.awaitTermination(2, TimeUnit.SECONDS)) {
                    log.error("心跳线程池强制关闭失败 - 设备类型: {}", handlerType);
                }
            } else {
                log.debug("心跳线程池已优雅关闭 - 设备类型: {}", handlerType);
            }
        } catch (InterruptedException e) {
            // 如果等待过程中被中断，立即强制关闭
            log.warn("心跳线程池关闭过程中被中断，执行强制关闭 - 设备类型: {}", handlerType);
            heartbeatExecutor.shutdownNow();
            // 恢复中断状态，遵循Java并发编程最佳实践
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 消息接收处理方法
     * <p>
     * 负责处理从服务端接收到的消息：
     * 1. 过滤心跳响应包，重置失败计数器
     * 2. 业务消息传递给后续处理器
     * </p>
     *
     * @param ctx Netty的通道上下文对象
     * @param msg 接收到的消息内容
     * @throws Exception 异常情况
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        // 过滤心跳包，避免进入业务逻辑
        if (isHeartbeatPacket(msg)) {
            log.debug("心跳响应 - 设备类型: {}, 时间: {}", handlerType, LocalDateTime.now());
            // 重置连续失败计数器，表示连接正常
            consecutiveHeartbeatMisses = 0;
            return;
        }
        // 业务消息传递给管道中的后续处理器
        ctx.fireChannelRead(msg);
    }

    /**
     * 判断是否为心跳数据包
     * <p>
     * 通过解析消息的operateCode字段来判断是否为心跳响应。
     * 心跳包的操作码为SourceOperateCodeEnum.HEARTBEAT。
     * </p>
     *
     * @param msg 需要判断的消息内容
     * @return true:心跳包, false:业务消息
     */
    private boolean isHeartbeatPacket(String msg) {
        try {
            // 解析消息为SocketDataMsg对象
            SocketDataMsg socketDataMsg = MsgUtil.socketDataMsg(msg);
            // 检查操作码是否为心跳类型
            return socketDataMsg != null &&
                    socketDataMsg.getOperateCode() != null &&
                    socketDataMsg.getOperateCode().equals(SourceOperateCodeEnum.HEARTBEAT.getValue());
        } catch (Exception e) {
            // 消息解析失败，可能不是标准格式的心跳包
            log.debug("消息解析失败，可能不是心跳包: {}", msg, e);
            return false;
        }
    }
}
