package com.ym.im.mq;

import com.alibaba.fastjson.JSON;
import com.ym.im.entity.*;
import com.ym.im.entity.base.NettyConstant;
import com.ym.im.entity.model.IdModel;
import com.ym.im.entity.model.OrderModel;
import com.ym.im.service.ChannelGroupService;
import com.ym.im.service.StaffService;
import com.ym.im.util.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Set;

/**
 * @author: JJww
 * @Date:2019-05-30
 */
@Slf4j
@Component
public class Receiver {

    @Autowired
    private Queue queue;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private StaffService staffService;

    /**
     * 禁用用户 队列名称
     */
    public static final String USER_QUEUE_NAME = "disable.user";

    /**
     * 订单队列
     */
    public static final String ORDER_QUEUE_NAME = "push.order";

    @RabbitListener(queues = "#{delayQueue.name}")
    public void delayAckHandler(MsgBody msgBody) throws IOException {
        log.info("接收时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + "  消息内容：" + JsonUtils.obj2Json(msgBody));
        retry(msgBody);
    }


    @RabbitListener(queues = "#{staffOfflineQueue.name}")
    public void offlineHandler(StaffSocketInfo staffSocketInfo) {

        final Long staffId = staffSocketInfo.getStaffId();
        //移除用户列表
        redisTemplate.delete(NettyConstant.STAFF_USERIDS_KEY + staffId);
        //客服真离线后 才转发
        if (ChannelGroupService.STAFF_GROUP.get(staffId) == null) {
            final Set<Long> userIds = staffSocketInfo.getUserIds();
            log.info("客服离线队列: " + "ID: " + "UserIds:" + userIds);
            userIds.forEach((Long userId) -> {
                //用户在线才重新分配和转发
                if (ChannelGroupService.USER_GROUP.get(userId) != null) {
                    final StaffSocketInfo idleStaff = staffService.getIdleStaff(userId);
                    if (idleStaff != null) {
                        idleStaff.writeAndFlush(new MsgBody<>().setStatus(MsgBody.DISTRIBUTION_STAFF).setData(new IdModel().setStaffId(staffId).setUserId(userId)));
                    }
                }
            });
        }
    }


    /**
     * 禁用用户后 关闭socket
     *
     * @param userId
     * @throws IOException
     */
    @RabbitListener(queues = USER_QUEUE_NAME)
    public void disableUserHandler(String userId) {

        final UserSocketInfo userSocketInfo = ChannelGroupService.USER_GROUP.get(Long.valueOf(userId));
        if (userSocketInfo != null) {
            userSocketInfo.writeAndFlush(new MsgBody<>().setStatus(MsgBody.LOGOUT));
            userSocketInfo.getChannel().close();
            log.info("用户: " + userId + "被禁用");
        }
    }

    /**
     * 订单相关处理
     *
     * @param orderModel
     */
    @RabbitListener(queues = ORDER_QUEUE_NAME)
    public void orderHandler(OrderModel orderModel) {

        log.info("Constants.ORDER_QUEUE_NAME: " + JSON.toJSONString(orderModel));

        final UserSocketInfo userSocketInfo = ChannelGroupService.USER_GROUP.get(Long.valueOf(orderModel.getUserId()));
        if (userSocketInfo == null) {
            return;
        }
        StaffSocketInfo staffSocketInfo = ChannelGroupService.STAFF_GROUP.get(userSocketInfo.getStaffId());
        final MsgBody<OrderModel> orderInfo = new MsgBody<OrderModel>().setStatus(MsgBody.ORDER).setData(orderModel);
        /**
         * 绑定客服在线,发送订单信息
         */
        if (staffSocketInfo != null) {
            staffSocketInfo.writeAndFlush(orderInfo);
            log.info("客服订单: " + "给客服(" + staffSocketInfo.getStaffId() + ")发送订单:" + orderInfo.toString());
            return;
        }
        /**
         * 绑定客服不在线,给历史客服发送订单信息
         */
        final Long staffId = (Long) redisTemplate.opsForHash().get(NettyConstant.IM_USERS, orderModel.getUserId());
        if (staffId != null) {
            log.info("客服订单: " + "尝试给历史客服(" + staffId + ")发送订单:" + orderInfo.toString());
            staffSocketInfo = ChannelGroupService.STAFF_GROUP.get(staffId);
            if (staffSocketInfo != null) {
                staffSocketInfo.writeAndFlush(orderInfo);
                log.info("客服订单: " + "给历史客服(" + staffId + ")发送订单:" + orderInfo.toString());
            }
        }
    }


    /**
     * 重发未回执的消息
     *
     * @param msgBody
     * @throws IOException
     */
    public void retry(MsgBody<ChatRecord> msgBody) throws IOException {

        ChatRecord chatRecord = msgBody.getData();
        Long userId = Long.valueOf(chatRecord.getUserId());
        String recordId = String.valueOf(chatRecord.getId());

        if (msgBody != null && chatRecord.getRetryCount().intValue() < NettyConstant.RETRY_COUNT.intValue()) {

            UserSocketInfo userSocketInfo = ChannelGroupService.USER_GROUP.get(userId);
            if (userSocketInfo == null) {
                return;
            }

            MsgBody<ChatRecord> msg = JsonUtils.json2Obj(
                    String.valueOf(redisTemplate.opsForHash().get(NettyConstant.MSG_KEY + userId, recordId)),
                    JsonUtils.getJavaType(MsgBody.class, ChatRecord.class));

            if (msg != null && msg.getStatus().equals(MsgBody.HAVE_READ)) {
                redisTemplate.opsForHash().delete(NettyConstant.MSG_KEY + userId, recordId);
                return;
            }

            switch (chatRecord.getSendReceive()) {

                case NettyConstant.CONNECT_TYPE_USER:
                    Long staffId = userSocketInfo.getStaffId() == null ? chatRecord.getStaffId() : userSocketInfo.getStaffId();
                    StaffSocketInfo staffSocketInfo = ChannelGroupService.STAFF_GROUP.get(staffId);
                    if (staffSocketInfo != null) {
                        staffSocketInfo.writeAndFlush(msgBody);
                    }
                    break;

                case NettyConstant.CONNECT_TYPE_STAFF:
                    userSocketInfo.writeAndFlush(msgBody);
                    break;

                default:
            }
            //重发三次
            chatRecord.setRetryCount(chatRecord.getRetryCount() + 1);
            queue.delaysQueue(msgBody);

        } else {
            //移除失败消息
            redisTemplate.opsForHash().delete(NettyConstant.MSG_KEY + userId, recordId);
        }

    }

}