package com.skr.mdm.netty;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.skr.mdm.authentication.TokenExpireException;
import com.skr.mdm.authentication.user.IllegalUserRequestException;
import com.skr.mdm.bean.vo.WebSocketResultMessageVO;
import com.skr.mdm.config.MdmWebsocketChannelFactory;
import com.skr.mdm.constant.CmdTypeConstants;
import com.skr.mdm.dao.DeviceEquipmentDao;
import com.skr.mdm.dao.SysSendMsgDao;
import com.skr.mdm.dto.LoginUserDTO;
import com.skr.mdm.dto.WebsocketDTO;
import com.skr.mdm.entity.SysSendMsgDO;
import com.skr.mdm.result.ResponseResult;
import com.skr.mdm.service.CmdHandlerService;
import com.skr.mdm.service.StrategyFactory;
import com.skr.mdm.util.WsAntiReplayUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;


/**
 * @author nfq
 * @date 2020/7/24 14:58
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class WebsocketHandler extends SimpleChannelInboundHandler<Object> {

    @Value("${mdm.encrypt}")
    private String mdmEncrypt;

    private static final String WEBSOCKET = "websocket";
    private static final String UPGRADE = "Upgrade";
    private static final String ANDROID = "android";
    private static final String USER_AGENT = "User-Agent";

    private WebSocketServerHandshaker webSocketServerHandshaker;
    @Autowired
    private WsAntiReplayUtil wsAntiReplayUtil;
    @Autowired
    private AfterUserConnectionHandler userConnectionHandler;
    @Autowired
    private DeviceEquipmentDao equipmentDao;
    @Autowired
    private SysSendMsgDao sysSendMsgDao;


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 握手请求
        if (msg instanceof FullHttpRequest) {
            handHttpRequest(ctx, (FullHttpRequest) msg);
        }
        // websocket消息
        if (msg instanceof WebSocketFrame) {
            handWebsocket(ctx, (WebSocketFrame) msg);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        String channelId = ctx.channel().id().asShortText();
        log.warn("channelId：{} websocket连接异常：{}", channelId, cause.getMessage());
        ctx.close();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {}

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        userConnectionHandler.userOffline(ctx);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    /**
     * 心跳机制
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        super.userEventTriggered(ctx, evt);
        // 心跳机制
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            String channelId = ctx.channel().id().asShortText();
            if (IdleState.READER_IDLE.equals(event.state())) {
                log.info("channelId：{} Websocket 180秒没有收到客户端回复断开连接", channelId);
                ctx.close();
            }
            if (IdleState.WRITER_IDLE.equals(event.state())) {
                log.warn("channelId:{} Websocket 120秒没有发送消息,推送PING" + channelId);
                ctx.channel().writeAndFlush(new BinaryWebSocketFrame(WebsocketEncoder.encoder(CmdTypeConstants.SERVER_HEART, "MDM_PING", mdmEncrypt)));
            }
            if (IdleState.ALL_IDLE.equals(event.state())) {
                if (log.isDebugEnabled()) {
                    log.debug("channelId:{} Websocket 60秒定期PING", channelId);
                }
                ctx.channel().writeAndFlush(new BinaryWebSocketFrame(WebsocketEncoder.encoder(CmdTypeConstants.SERVER_HEART, "MDM_PING", mdmEncrypt)));
            }
        }
    }

    private void handHttpRequest(ChannelHandlerContext context, FullHttpRequest fullHttpRequest) {
        String channelId = context.channel().id().asShortText();
        if (!fullHttpRequest.decoderResult().isSuccess() || !StringUtils.equals(WEBSOCKET, fullHttpRequest.headers().get(UPGRADE)) ||
                !StringUtils.equalsIgnoreCase(ANDROID, fullHttpRequest.headers().get(USER_AGENT))) {
            log.warn("channelId:{} websocket非法请求，拒绝连接", channelId);
            setHttpResponse(context, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
        LoginUserDTO loginUserDTO;
        try {
            WebSocketServerHandshakerFactory webSocketServerHandshakerFactory = new WebSocketServerHandshakerFactory(null, null, false, 65536 * 10);
            webSocketServerHandshaker = webSocketServerHandshakerFactory.newHandshaker(fullHttpRequest);
            if (webSocketServerHandshaker == null) {
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(context.channel());
            } else {
                webSocketServerHandshaker.handshake(context.channel(), fullHttpRequest);
            }
            // http请求过滤
            loginUserDTO = isLegal(fullHttpRequest);
            if (loginUserDTO != null) {
                userConnectionHandler.userOnline(context, loginUserDTO);
            } else {
                log.warn("channelId:{} websocket连接用户认证失败", channelId);
                setHttpResponse(context, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.UNAUTHORIZED));
                context.close();
                return;
            }
        } catch (TokenExpireException | IllegalUserRequestException | NullPointerException e) {
            log.error("channelId:" + channelId + "websocket连接失败：", e);
            setHttpResponse(context, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.UNAUTHORIZED));
            context.close();
            return;
        }
        log.info("channelId:{} websocket连接成功 用户名：{} 设备ID：{}", channelId, loginUserDTO.getUsername(), loginUserDTO.getId());
    }

    private void setHttpResponse(ChannelHandlerContext context, DefaultFullHttpResponse response) {
        if (response.status().code() != HttpStatus.OK.value()) {
            ByteBuf copiedBuffer = Unpooled.copiedBuffer(response.status().toString(), CharsetUtil.UTF_8);
            response.content().writeBytes(copiedBuffer);
            copiedBuffer.release();
        }
        ChannelFuture channelFuture = context.channel().writeAndFlush(response);
        if (response.status().code() != HttpStatus.OK.value()) {
            channelFuture.addListener(ChannelFutureListener.CLOSE);
        }
    }

    private void handWebsocket(ChannelHandlerContext context, WebSocketFrame webSocketFrame) {
        if (webSocketFrame instanceof CloseWebSocketFrame) {
            webSocketServerHandshaker.close(context.channel(), (CloseWebSocketFrame) webSocketFrame.retain());
        }
        // 处理二进制数据
        if (webSocketFrame instanceof BinaryWebSocketFrame) {
            BinaryWebSocketFrame binaryWebSocketFrame = (BinaryWebSocketFrame) webSocketFrame;
            ByteBuf content = binaryWebSocketFrame.content();
            WebsocketDTO websocketDTO = WebsocketDecoder.decoder(content, mdmEncrypt);
            if (websocketDTO == null) {
                log.warn("客户端websocketDTO解析失败");
                return;
            }
            String channelId = context.channel().id().asShortText();
            Integer equipmentId = MdmWebsocketChannelFactory.getEquipmentId(channelId);
            if (null == equipmentId) {
                log.error("channelId :{} 设备不存在", channelId);
                return;
            }
            // 心跳处理，检测未下发成功的指令进行推送
            if (websocketDTO.getType() == CmdTypeConstants.SERVER_HEART) {
                heartbeatProcess(context, channelId, equipmentId, websocketDTO);
                return;
            }
            CmdHandlerService strategy = StrategyFactory.getStrategy(websocketDTO.getType());

            ResponseResult result;
            if (StringUtils.isBlank(websocketDTO.getBody())) {
                log.warn("channelId :{} 消息体为空:{}", channelId, JSONUtil.toJsonStr(websocketDTO));
                result = ResponseResult.success();
            } else {
                result = strategy.cmdHandler(equipmentId, websocketDTO.getBody());
            }
            ByteBuf byteBuf = WebsocketEncoder.encoder(websocketDTO.getType(),
                    JSONUtil.toJsonStr(JSONUtil.parseObj(result)), mdmEncrypt);
            context.writeAndFlush(new BinaryWebSocketFrame(byteBuf));
        }

    }

    /**
     * 检测未下发成功的指令
     * @param context
     * @param channelId
     * @param equipmentId
     * @param websocketDTO
     */
    private void heartbeatProcess(ChannelHandlerContext context, String channelId, Integer equipmentId, WebsocketDTO websocketDTO) {
        if (log.isDebugEnabled()) {
            log.debug("channelId:{} 客户端回复PONG正常 msg:{}", channelId, websocketDTO.getBody());
        }
        // 发送没有成功的指令
        List<SysSendMsgDO> sendMsgInfoList = sysSendMsgDao.getSendMsgInfo(equipmentId, 1);
        if (CollectionUtil.isEmpty(sendMsgInfoList)) {
            return;
        }
        List<WebSocketResultMessageVO> msg = sendMsgInfoList.stream()
                .map(sysSendMsgDO -> WebSocketResultMessageVO.builder()
                        .msgId(sysSendMsgDO.getId())
                        .type(sysSendMsgDO.getCmdType())
                        .data(sysSendMsgDO.getData())
                        .sendTime(sysSendMsgDO.getCreateTime().getTime())
                        .secretKey(sysSendMsgDO.getSecretKey())
                        .build())
                .collect(Collectors.toList());
        List<Long> msgIdList = msg.stream().map(WebSocketResultMessageVO::getMsgId).collect(Collectors.toList());
        if (log.isDebugEnabled()) {
            log.debug("channelId：{} websocket心跳机制推送未下发成功的消息 ID：{}", channelId, JSONUtil.toJsonStr(msgIdList));
        }
        ByteBuf byteBuf = WebsocketEncoder.encoder(CmdTypeConstants.ISSUE, JSONUtil.toJsonStr(JSONUtil.parseArray(msg)), mdmEncrypt);
        context.writeAndFlush(new BinaryWebSocketFrame(byteBuf));
    }

    /**
     * 判断请求是否合法
     *
     * @param fullHttpRequest
     * @return
     * @throws TokenExpireException
     * @throws IllegalUserRequestException
     */
    private LoginUserDTO isLegal(FullHttpRequest fullHttpRequest) throws TokenExpireException, IllegalUserRequestException {
        // 防重放
        return wsAntiReplayUtil.antiReplay(fullHttpRequest);
    }

}
