package com.fastbee.mqtt.handler;

import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.redis.RedisCache;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.auth.AuthService;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ResponseManager;
import com.fastbee.mqtt.manager.SessionManger;
import com.fastbee.mqtt.manager.WillMessageManager;
import com.fastbee.mqtt.model.WillMessage;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.net.InetSocketAddress;


@Slf4j
@Process(type = MqttMessageType.CONNECT)
public class MqttConnect implements MqttHandler {

    @Autowired
    private AuthService authService;
    @Resource
    private RedisCache redisCache;

    @Override
    public void handler(ChannelHandlerContext ctx, MqttMessage message) {
        MqttConnectMessage connectMessage = (MqttConnectMessage) message;
        /*获取客户端Id*/
        String clientId = connectMessage.payload().clientIdentifier();
        if (clientId.contains("&")){
            clientId =  clientId.split("&")[1];
        }
        log.debug("=>客户端:{} 连接:{}", clientId, message);
        /*获取session*/
        Session session = new Session();
        /*mqtt版本*/
        MqttVersion version = MqttVersion.fromProtocolNameAndLevel(connectMessage.variableHeader().name(), (byte) connectMessage.variableHeader().version());
        boolean cleanSession = connectMessage.variableHeader().isCleanSession();
        session.setHandlerContext(ctx);
        session.setVersion(version);
        session.setClientId(clientId);
        session.setCleanSession(cleanSession);
        session.setUsername(connectMessage.payload().userName());
        session.setConnected_at(DateUtils.getNowDate());
        InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
        session.setIp(socketAddress.getAddress().getHostAddress());
        session.setServerType(ServerType.MQTT);
        /*设置客户端ping时间*/
        MqttConnectVariableHeader header = connectMessage.variableHeader();
        if (header.keepAliveTimeSeconds() > 0 && session.getKeepAliveMax() >= header.keepAliveTimeSeconds()) {
            session.setKeepAlive(header.keepAliveTimeSeconds());
        }
        /*mqtt客户端登录校验*/
        if (!check(session, connectMessage)) {
            log.error("=>客户端:{},连接异常", clientId);
            session.getHandlerContext().close();
            return;
        }

        /*保存ClientId 和 session 到Attribute*/
        AttributeUtils.setClientId(ctx.channel(), clientId);
        AttributeUtils.setSession(ctx.channel(), session);
        SessionManger.removeClient(clientId);
        session.setConnected(true);

        SessionManger.buildSession(clientId, session);
        handleWill(connectMessage);
        ResponseManager.responseMessage(session,
                MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0x02),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, false),
                        null),
                true
        );
    }

    /**
     * 遗嘱消息处理
     *
     * @param message 连接消息
     */
    private void handleWill(MqttConnectMessage message) {
        /*如果没有设置处理遗嘱消息，返回*/
        if (!message.variableHeader().isWillFlag()) {
            return;
        }
        /*生成客户端model*/
        MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.PUBLISH, false,
                        MqttQoS.valueOf(message.variableHeader().willQos()),
                        message.variableHeader().isWillRetain(), 0),
                new MqttPublishVariableHeader(message.payload().willTopic(), 0),
                Unpooled.buffer().writeBytes(message.payload().willMessageInBytes()));

        WillMessage msg = new WillMessage(message.payload().clientIdentifier(),
                message.variableHeader().isCleanSession(), message.payload().willTopic(), publishMessage);
        WillMessageManager.push(msg);
    }

    /**
     * 客户端连接校验
     *
     * @param session 客户端
     * @param message 连接消息
     * @return 结果
     */
    private boolean check(Session session, MqttConnectMessage message) {
        /*获取客户端连接地址*/
        InetSocketAddress address = (InetSocketAddress) session.getHandlerContext().channel().remoteAddress();
        String host = address.getAddress().getHostAddress();
        /*webSocket客户端 系统内部客户端不校验*/
        String clientId = message.payload().clientIdentifier();
        /*根据用户名，密码校验*/
        String username = message.payload().userName();
        String password = message.payload().passwordInBytes() == null ? null : new String(message.payload().passwordInBytes(), CharsetUtil.UTF_8);
        /*验证失败，应答客户端*/
        if (!authService.auth(clientId, username, password)) {
            MqttConnAckMessage connAckMessage = MqttMessageUtils.buildConntAckMessage(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false);
            ResponseManager.responseMessage(session, connAckMessage, true);
            return false;
        }
        return true;
    }

}