package com.czg.mqtt.handler;

import com.czg.mqtt.manager.DeviceStatusManager;
import com.czg.mqtt.manager.MqttSubscriptionManager;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.*;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 实现 MQTT 消息处理器
 * 自定义 ChannelInboundHandlerAdapter，处理 MQTT 协议的核心消息类型：
 * CONNECT：客户端连接请求
 * PUBLISH：客户端发布消息
 * SUBSCRIBE：客户端订阅主题
 * UNSUBSCRIBE：客户端取消订阅
 * DISCONNECT：客户端断开连接
 * PINGREQ：客户端心跳请求
 * PUBREL：QoS2 第二步确认请求
 *
 * @author yjjie
 */
@Slf4j
public class MqttServerHandler extends ChannelInboundHandlerAdapter {
    // 注入设备状态管理器（单例）
    private final DeviceStatusManager deviceStatusManager;

    // 并发安全的 Map：缓存 Channel（连接）→ clientId（设备唯一标识）的映射
    private final Map<Channel, String> channelClientIdMap = new ConcurrentHashMap<>();
    private final Map<String, Channel> clientIdChannelMap = new ConcurrentHashMap<>();

    // 注入共享的订阅关系管理器（单例）
    private final MqttSubscriptionManager subscriptionManager;
    // JSON 序列化工具（用于构造回复消息格式）
    private final ObjectMapper objectMapper = new ObjectMapper();

    // 创建一个调度线程池（可以全局复用）
    private static final ScheduledExecutorService SCHEDULER = Executors.newScheduledThreadPool(4);


    // 构造函数（接收共享组件）
    public MqttServerHandler(MqttSubscriptionManager subscriptionManager, DeviceStatusManager deviceStatusManager) {
        this.subscriptionManager = subscriptionManager;
        this.deviceStatusManager = deviceStatusManager;
    }

    // -------------------------- 连接相关 --------------------------
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("客户端连接成功：{}", ctx.channel().remoteAddress());
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        String clientId = channelClientIdMap.get(channel);

        if (clientId != null) {
            // 移除两个映射
            channelClientIdMap.remove(channel);
            clientIdChannelMap.remove(clientId);

            // 标记设备离线
            deviceStatusManager.markDeviceOffline(clientId);
            log.info("设备断开连接，已标记离线：clientId={}, 地址={}", clientId, channel.remoteAddress());
        } else {
            log.warn("未知 Channel 断开：{}", channel.remoteAddress());
        }

        super.channelInactive(ctx);
    }

    // -------------------------- 消息处理 --------------------------
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof MqttMessage mqttMsg) {
            MqttFixedHeader fixedHeader = mqttMsg.fixedHeader();
            MqttMessageType msgType = fixedHeader.messageType();

            // 新增：处理 QoS2 第二步 - 客户端发送的 PUBREL 消息
            if (msgType == MqttMessageType.PUBREL) {
                handlePubRel(ctx, (MqttMessageIdVariableHeader) mqttMsg.variableHeader());
                return;
            }

            // 根据消息类型分发处理
            switch (msgType) {
                case CONNECT:
                    handleConnect(ctx, (MqttConnectMessage) mqttMsg);
                    break;
                case PUBLISH:
                    handlePublish(ctx, (MqttPublishMessage) mqttMsg);
                    break;
                case SUBSCRIBE:
                    handleSubscribe(ctx, (MqttSubscribeMessage) mqttMsg);
                    break;
                case UNSUBSCRIBE:
                    handleUnsubscribe(ctx, (MqttUnsubscribeMessage) mqttMsg);
                    break;
                case DISCONNECT:
                    handleDisconnect(ctx);
                    break;
                case PINGREQ:
                    handlePingReq(ctx);
                    break;
                default:
                    log.warn("不支持的MQTT消息类型：{}", msgType);
                    break;
            }
        }
    }

    // -------------------------- 具体消息处理逻辑 --------------------------

    /**
     * 处理 PINGREQ 消息（客户端心跳请求）
     * 响应 PINGRESP 消息以维持连接
     * 更新设备在线状态
     */
    private void handlePingReq(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        // 关键：通过 Channel 关联 clientId
        String clientId = channelClientIdMap.get(channel);

        if (clientId != null) {
            // 更新设备最后 Ping 时间（维持在线）
            deviceStatusManager.updateDevicePing(clientId);
            log.info("收到设备心跳：clientId={}, 地址={}", clientId, channel.remoteAddress());
        } else {
            log.warn("收到未知设备心跳：地址={}（未找到 clientId，可能连接未完成）", channel.remoteAddress());
        }

        // 构造 PINGRESP 消息（固定头：PINGRESP，无可变头和负载）
        MqttFixedHeader pingRespHeader = new MqttFixedHeader(
                // 消息类型：PINGRESP
                MqttMessageType.PINGRESP,
                // 是否为重复消息
                false,
                // QoS
                MqttQoS.AT_MOST_ONCE,
                // 是否保留
                false,
                // 剩余长度（PINGRESP 无可变头和负载，剩余长度为0）
                0
        );

        MqttMessage pingRespMsg = new MqttMessage(pingRespHeader);

        // 回复 PINGRESP
        ctx.writeAndFlush(pingRespMsg);
        log.info("已向客户端 {} 发送 PINGRESP 心跳响应", ctx.channel().remoteAddress());
    }

    /**
     * 处理 CONNECT 消息（设备连接请求）：缓存 clientId ↔ Channel 映射，注册设备状态
     */
    private void handleConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
        MqttConnectPayload payload = msg.payload();
        String clientId = payload.clientIdentifier();
        String remoteAddress = ctx.channel().remoteAddress().toString();

        if (clientId == null || clientId.isEmpty()) {
            sendConnectAck(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_CLIENT_IDENTIFIER_NOT_VALID);
            log.error("设备连接拒绝：clientId 为空，地址={}", remoteAddress);
            return;
        }

        // 🔒 检查是否已经存在该 clientId 的连接
        if (clientIdChannelMap.containsKey(clientId)) {
            Channel existingChannel = clientIdChannelMap.get(clientId);
            log.warn("设备重复连接，已拒绝：clientId={}, 原连接={}, 新连接={}",
                    clientId, existingChannel.remoteAddress(), ctx.channel().remoteAddress());

            // 返回 "服务器不可用" 或 "已存在连接" 错误码，拒绝新连接
            sendConnectAck(ctx, MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE);
            ctx.close(); // 主动关闭多余的连接
            return;
        }

        // ✅ 当前连接是合法的、唯一的，继续处理
        Channel channel = ctx.channel();
        channelClientIdMap.put(channel, clientId);
        // 维护反向映射
        clientIdChannelMap.put(clientId, channel);

        // 注册设备状态
        deviceStatusManager.registerDevice(clientId, remoteAddress);

        // 回复 CONNACK
        sendConnectAck(ctx, MqttConnectReturnCode.CONNECTION_ACCEPTED);
        log.info("设备连接成功：clientId={}, 地址={}, 协议版本={}", clientId, remoteAddress, msg.variableHeader().version());
    }

    private void handlePublish(ChannelHandlerContext ctx, MqttPublishMessage msg) {
        String topic = msg.variableHeader().topicName();
        ByteBuf payloadBuf = msg.payload();
        byte[] payload = new byte[payloadBuf.readableBytes()];
        payloadBuf.readBytes(payload);
        // 释放ByteBuf，避免内存泄漏
        payloadBuf.release();

        String content = new String(payload, StandardCharsets.UTF_8);
        // 获取固定头（存储QoS）
        MqttFixedHeader fixedHeader = msg.fixedHeader();
        MqttPublishVariableHeader variableHeader = msg.variableHeader();
        MqttQoS qosLevel = fixedHeader.qosLevel();
        int messageId = variableHeader.packetId();

        // 日志输出（包含QoS等级和消息ID）
        log.info("收到发布消息：topic={}, QoS={}, 消息ID={}, 内容={}, 客户端={}",
                topic, qosLevel.value(), messageId, content, ctx.channel().remoteAddress());

        // -------------------------- 收到 con 主题消息，回复  --------------------------
        if ("con".equals(topic)) {
            // 延迟 2 秒执行
            SCHEDULER.schedule(() -> sendResponseToClient(ctx, "处理成功"), 2, TimeUnit.SECONDS);
            try {
                deviceStatusManager.sendMqMsg(content);
            } catch (Exception e) {
                log.error("发送消息异常：{}", e.getMessage());
            }
        }

        // 按QoS等级回复确认（重点实现QoS2）
        switch (qosLevel) {
            // QoS0：自定义主题回复确认
            case AT_MOST_ONCE:
                sendQos0Ack(ctx, topic, messageId, content);
                break;
            // QoS1：回复 PUBACK（协议要求）
            case AT_LEAST_ONCE:
                sendPubAck(ctx, messageId);
                break;
            // QoS2：第一步回复 PUBREC（协议要求）
            case EXACTLY_ONCE:
                sendPubRec(ctx, messageId);
                log.info("已处理QoS2消息，发送PUBREC确认：消息ID={}", messageId);
                break;
            default:
                log.warn("不支持的QoS等级：{}", qosLevel);
        }

        // 转发消息逻辑（不变）
        Set<Channel> subscribers = subscriptionManager.getSubscribers(topic);
        if (subscribers.isEmpty()) {
            log.debug("主题 {} 无订阅者，消息丢弃", topic);
            return;
        }

        MqttPublishMessage forwardMsg = buildForwardPublishMessage(msg, payload);
        for (Channel subscriber : subscribers) {
            if (subscriber.isActive() && subscriber != ctx.channel()) {
                subscriber.writeAndFlush(forwardMsg.retain());
                log.debug("转发消息到客户端 {}：topic={}, 内容={}",
                        subscriber.remoteAddress(), topic, content);
            }
        }
    }

    /**
     * 向设备指定格式的回复消息
     */
    private void sendResponseToClient(ChannelHandlerContext ctx, String msg) {
        try {
            Channel channel = ctx.channel();
            // 关键：通过 Channel 关联 clientId
            String clientId = channelClientIdMap.get(channel);

            // 1. 构造回复消息内容（按要求格式）
            ResponseMessage responseMsg = new ResponseMessage();
            responseMsg.setMsg(msg);
            responseMsg.setSn("超管");
            responseMsg.setName("超管");
            responseMsg.setPly(51);

            // 2. 序列化为 JSON 字符串
            String jsonContent = objectMapper.writeValueAsString(responseMsg);
            log.info("向主题 {} 发布回复消息：{}", clientId, jsonContent);

            // 3. 构造 MQTT 发布消息（QoS2，与客户端发送的等级一致，确保可靠送达）
            MqttFixedHeader fixedHeader = new MqttFixedHeader(
                    // 消息类型：PUBLISH
                    MqttMessageType.PUBLISH,
                    // 不重复
                    false,
                    // QoS2（与客户端一致，确保可靠）
                    MqttQoS.EXACTLY_ONCE,
                    // 不保留
                    false,
                    // 剩余长度（Netty自动计算）
                    0
            );
            // 消息ID：生成随机不重复的ID（1-65535之间）
            int responseMsgId = (int) (Math.random() * 65535) + 1;
            MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(clientId, responseMsgId);
            ByteBuf payload = Unpooled.wrappedBuffer(jsonContent.getBytes(StandardCharsets.UTF_8));

            // 4. 发布消息（所有订阅 7941610A 主题的客户端都会收到）
            MqttPublishMessage publishMsg = new MqttPublishMessage(fixedHeader, variableHeader, payload);
            ctx.writeAndFlush(publishMsg);

            log.info("已向主题 {} 发送回复消息，消息ID：{}", clientId, responseMsgId);
        } catch (JsonProcessingException e) {
            log.error("构造回复消息失败：", e);
        }
    }

    /**
     * 回复消息实体类（用于构造指定JSON格式）
     */
    @Data
    private static class ResponseMessage {
        private String msg;
        private String sn;
        private String name;
        private int ply;

    }

    /**
     * 处理 QoS2 第二步：客户端发送的 PUBREL 消息，回复 PUBCOMP 完成最终确认
     * 用 MqttMessage 构造，避免依赖 MqttPubCompMessage
     */
    private void handlePubRel(ChannelHandlerContext ctx, MqttMessageIdVariableHeader variableHeader) {
        int messageId = variableHeader.messageId();
        log.info("收到客户端 PUBREL 消息（QoS2第二步）：客户端={}, 消息ID={}",
                ctx.channel().remoteAddress(), messageId);

        // 构造 PUBCOMP 固定头（消息类型为 PUBCOMP）
        MqttFixedHeader pubCompFixedHeader = new MqttFixedHeader(
                // 关键：消息类型指定为 PUBCOMP
                MqttMessageType.PUBCOMP,
                // 不重复
                false,
                // QoS0
                MqttQoS.AT_MOST_ONCE,
                // 不保留
                false,
                // 可变头长度（消息ID占2字节，Netty可自动计算，填2更明确）
                2
        );

        // 构造 PUBCOMP 消息（固定头 + 消息ID可变头）
        MqttMessage pubCompMsg = new MqttMessage(pubCompFixedHeader, variableHeader);

        ctx.writeAndFlush(pubCompMsg);
        log.info("发送PUBCOMP（QoS2最终确认）到客户端 {}：消息ID={}",
                ctx.channel().remoteAddress(), messageId);
    }

    private void handleSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) {
        Channel clientChannel = ctx.channel();
        msg.payload().topicSubscriptions().forEach(sub -> {
            String topic = sub.topicFilter();
            log.info("客户端订阅主题：client={}, topic={}, 请求QoS={}",
                    clientChannel.remoteAddress(), topic, sub.qualityOfService().value());

            // 调用共享管理器添加订阅
            subscriptionManager.addSubscription(topic, clientChannel);
        });

        // 响应订阅成功（SUBACK）
        MqttFixedHeader fixedHeader = new MqttFixedHeader(
                MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0
        );
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(msg.variableHeader().messageId());
        MqttSubAckPayload payload = new MqttSubAckPayload(
                msg.payload().topicSubscriptions().stream()
                        // 按客户端请求QoS返回
                        .map(sub -> sub.qualityOfService().value())
                        .toList()
        );

        MqttSubAckMessage subAckMsg = new MqttSubAckMessage(fixedHeader, variableHeader, payload);
        ctx.writeAndFlush(subAckMsg);
        log.info("订阅响应已发送：client={}, 消息ID={}", clientChannel.remoteAddress(), variableHeader.messageId());
    }

    private void handleUnsubscribe(ChannelHandlerContext ctx, MqttUnsubscribeMessage msg) {
        Channel clientChannel = ctx.channel();
        msg.payload().topics().forEach(topic -> {
            log.info("客户端取消订阅：client={}, topic={}", clientChannel.remoteAddress(), topic);
            // 调用共享管理器移除订阅
            subscriptionManager.removeSubscription(topic, clientChannel);
        });

        // 响应取消订阅成功（UNSUBACK）
        MqttFixedHeader fixedHeader = new MqttFixedHeader(
                MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0
        );
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(msg.variableHeader().messageId());
        MqttUnsubAckMessage unsubAckMsg = new MqttUnsubAckMessage(fixedHeader, variableHeader);
        ctx.writeAndFlush(unsubAckMsg);
    }

    private void handleDisconnect(ChannelHandlerContext ctx) {
        log.info("客户端主动断开连接：{}", ctx.channel().remoteAddress());
        ctx.close();
    }

    // -------------------------- 工具方法 --------------------------
    private void sendConnectAck(ChannelHandlerContext ctx, MqttConnectReturnCode returnCode) {
        MqttFixedHeader fixedHeader = new MqttFixedHeader(
                MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0
        );
        MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(returnCode, false);
        ctx.writeAndFlush(new MqttConnAckMessage(fixedHeader, variableHeader));
        if (returnCode != MqttConnectReturnCode.CONNECTION_ACCEPTED) {
            ctx.close();
        }
    }

    private MqttPublishMessage buildForwardPublishMessage(MqttPublishMessage originalMsg, byte[] payload) {
        MqttFixedHeader fixedHeader = originalMsg.fixedHeader();
        MqttPublishVariableHeader variableHeader = originalMsg.variableHeader();
        ByteBuf forwardPayload = Unpooled.wrappedBuffer(payload);
        return new MqttPublishMessage(fixedHeader, variableHeader, forwardPayload);
    }

    // -------------------------- 异常处理 --------------------------
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("客户端处理异常：{}，原因：{}", ctx.channel().remoteAddress(), cause.getMessage(), cause);
        ctx.close();
    }

    // -------------------------- 不同QoS的回复实现 --------------------------

    /**
     * QoS0 自定义确认：发布一个「原主题/ack」的消息，告知客户端已收到
     */
    private void sendQos0Ack(ChannelHandlerContext ctx, String originalTopic, int messageId, String content) {
        String ackTopic = originalTopic + "/ack";
        String ackContent = String.format("服务端已收到消息：ID=%d, 内容=%s", messageId, content);

        MqttFixedHeader fixedHeader = new MqttFixedHeader(
                MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0
        );
        // QoS0 消息ID为0
        MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(ackTopic, 0);
        ByteBuf ackPayload = Unpooled.wrappedBuffer(ackContent.getBytes(StandardCharsets.UTF_8));

        MqttPublishMessage ackMsg = new MqttPublishMessage(fixedHeader, variableHeader, ackPayload);
        ctx.writeAndFlush(ackMsg);

        log.debug("发送QoS0确认消息到客户端 {}：topic={}, 内容={}",
                ctx.channel().remoteAddress(), ackTopic, ackContent);
    }

    /**
     * QoS1 协议确认：回复 PUBACK 消息（带原消息ID）
     */
    private void sendPubAck(ChannelHandlerContext ctx, int messageId) {
        MqttFixedHeader fixedHeader = new MqttFixedHeader(
                MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2
        );
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
        MqttPubAckMessage pubAckMsg = new MqttPubAckMessage(fixedHeader, variableHeader);

        ctx.writeAndFlush(pubAckMsg);
        log.debug("发送PUBACK确认（QoS1）到客户端 {}：消息ID={}",
                ctx.channel().remoteAddress(), messageId);
    }

    /**
     * QoS2 第一步确认：回复 PUBREC 消息（带原消息ID）
     * 用 MqttMessage 构造，避免依赖 MqttPubRecMessage
     */
    private void sendPubRec(ChannelHandlerContext ctx, int messageId) {
        // 构造 PUBREC 固定头（消息类型为 PUBREC）
        MqttFixedHeader pubRecFixedHeader = new MqttFixedHeader(
                // 关键：消息类型指定为 PUBREC
                MqttMessageType.PUBREC,
                // 不重复
                false,
                // QoS0
                MqttQoS.AT_MOST_ONCE,
                // 不保留
                false,
                // 可变头长度（消息ID占2字节）
                2
        );

        // 构造 PUBREC 消息（固定头 + 消息ID可变头）
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
        MqttMessage pubRecMsg = new MqttMessage(pubRecFixedHeader, variableHeader);

        ctx.writeAndFlush(pubRecMsg);
        log.info("发送PUBREC确认（QoS2第一步）到客户端 {}：消息ID={}",
                ctx.channel().remoteAddress(), messageId);
    }
}