package com.ym.im.service.impl;

import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.ym.im.entity.*;
import com.ym.im.service.*;
import com.ym.im.util.JsonUtils;
import com.ym.im.validation.group.ChatRecordReceiveGroup;
import com.ym.im.validation.group.ChatRecordSaveGroup;
import com.ym.im.validation.group.ChatRecordSendGroup;
import com.ym.im.validation.group.MsgBodyGroup;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;

import javax.annotation.Resource;
import javax.validation.Valid;
import java.util.Date;
import java.util.Set;

import static com.ym.im.entity.ChatRecord.SEND;

/**
 * @author: JJww
 * @Date:2019-05-21
 */
@Slf4j
@Service
@Validated
public class UserSingleChatServiceImpl implements ChatService {

    @Resource(name = "myRedisTemplate")
    private RedisTemplate redisTemplate;

    @Autowired
    private StaffService staffService;

    @Autowired
    private ChatRecordService chatRecordService;

    @Autowired
    private MsgBodyService msgBodyService;

    @Override
    public void init(ChannelHandlerContext ctx) {

        ctx.channel().attr(NettyConstant.TYPE).set(NettyConstant.CONNECT_TYPE_USER);
        final Long userId = ctx.channel().attr(NettyConstant.ID).get();
        UserSocketInfo userSocketInfo = new UserSocketInfo();
        userSocketInfo.setUserId(userId);
        userSocketInfo.setChannel((NioSocketChannel) ctx.channel());
        userSocketInfo.setCol(ctx.channel().attr(NettyConstant.COL_INFO).get());
        userSocketInfo.setToken(ctx.channel().attr(NettyConstant.TOKEN_INFO).get());
        userSocketInfo.setPushToken(null);
        ChannelGroupService.USER_GROUP.put(userId, userSocketInfo);
        //恢复历史绑定关系
        restoreBindingRelationship(userId);
        //通知客服 用户上线
        broadcastUserOnline(userId);
        log.info("用户: " + userId + " 上线");

    }


    @Override
    public void offline(ChannelHandlerContext ctx) {

        final Long userId = ctx.channel().attr(NettyConstant.ID).get();
        final UserSocketInfo userSocketInfo = ChannelGroupService.USER_GROUP.get(userId);

        if (userSocketInfo != null && ctx.channel().attr(NettyConstant.TOKEN_INFO).get().equals(userSocketInfo.getToken())) {
            final Long staffId = userSocketInfo.getStaffId();
            ChannelGroupService.USER_GROUP.remove(userId);
            ctx.channel().close();
            if (staffId != null && ChannelGroupService.STAFF_GROUP.get(staffId) != null) {
                // 保存最后的客服
                redisTemplate.opsForHash().put(NettyConstant.IM_USERS, userId, staffId);
            }
            log.info("用户: " + userId + " 下线");
        }
    }


    @Override
    @Validated({MsgBodyGroup.class, ChatRecordSendGroup.class})
    public NioSocketChannel distribution(Long id, @Valid MsgBody<ChatRecord> msgBody) {

        // 获取服务用户的客服Id
        Long staffId = ChannelGroupService.USER_GROUP.get(id).getStaffId();
        // 客服SocketInfo对象
        StaffSocketInfo staffSocketInfo;
        // 若客服Id为空，分配客服，不为空则获取客服SocketInfo
        if (staffId == null) {
            staffSocketInfo = staffService.getIdleStaff(id);
            if (staffSocketInfo == null) {
                return null;
            }
        } else {
            // 根据客服id获取SocketInfo
            staffSocketInfo = ChannelGroupService.STAFF_GROUP.get(staffId);
            // 服务用户的客服不在线
            if (staffSocketInfo == null) {
                // Redis是否存在当前客服服务用户Set对象（TTL=60秒）
                Set members = redisTemplate.opsForSet().members(NettyConstant.STAFF_USERIDS_KEY + staffId);
                // 客服下线超过60秒，当前客服服务用户Set对象已过期，重新分配客服
                // 过期后set对象不会为空而是大小为0
                if (members.size() == 0) {
                    staffSocketInfo = staffService.getIdleStaff(id);
                }
                return null;
            } else {
                /*
                 * 解决客服下线超过一分钟后再上线收到下线前服务的用户（用户没有再分配新客服情况）发来的消息，
                 * 客服无法回复用户的问题，
                 * 无论该用户Id是否存在于客服服务用户Set内，都向Set添加用户ID
                 */
                staffSocketInfo.getUserIds().add(id);
            }
        }
        return staffSocketInfo.getChannel();
    }


    @Override
    @Validated({MsgBodyGroup.class, ChatRecordSendGroup.class})
    public void save(Long id, @Valid MsgBody<ChatRecord> msgBody) {

        // 设置聊天基本信息
        final ChatRecord record = msgBody.getData()
                .setId(IdWorker.getId())
                .setUserId(id)
                .setSendReceive(SEND)
                .setCreateTime(new Date());
        /*
         * 设置客服Id：
         *  情况1：当前用户无客服服务，且无客服可分配，ChatRecord.staffId = null;
         *  情况2：当前用户无客服服务，且有客服可分配，ChatRecord.staffId = StaffSocketInfo.id;
         *  情况3：当前用户有客服服务，且客服在线，ChatRecord.staffId = StaffSocketInfo.id;
         *  情况3：当前用户有客服服务，且客服已下线并不超过60秒，CharRecord.staffId = staffId;
         *  情况4：当前用户有客服服务，且客服已下线并超过60秒，重新分配客服且有客服可分配，ChatRecord.staffId = StaffSocketInfo.id;
         *  情况5（特殊：需求未定暂时这样处理）：当前用户有客服服务，且客服已下线并超过60秒，重新分配客服且无客服可分配，ChatRecord.staffId = staffId;
         */
        record.setStaffId(ChannelGroupService.USER_GROUP.get(id).getStaffId());
        // 先保存至数据库，再发送消息（若颠倒顺序可能导致数据未保存，更新已读操作先执行导致消息一直是未读状态）
        chatRecordService.insertSelective(record);
        log.info("用户 消息保存:" + record.getId());
    }

    /**
     * 发送消息至客服端
     *
     * @param msgBody 消息对象
     * @throws JsonProcessingException e
     */
    @Override
    public void send(NioSocketChannel channel, MsgBody<ChatRecord> msgBody) throws JsonProcessingException {
        msgBodyService.sendAndAck(channel, msgBody);
    }


    @Override
    @Validated({MsgBodyGroup.class, ChatRecordReceiveGroup.class, ChatRecordSaveGroup.class})
    public void ack(@Valid MsgBody<ChatRecord> msgBody) throws JsonProcessingException {
        final ChatRecord record = msgBody.getData();
        record.setModifyTime(new Date());
        chatRecordService.updateReceiveTime(record);
        UserSocketInfo userSocketInfo = ChannelGroupService.USER_GROUP.get(record.getUserId());
        if (userSocketInfo != null) {
            userSocketInfo.setStaffId(userSocketInfo.getStaffId() == null ? record.getStaffId() : userSocketInfo.getStaffId());
            StaffSocketInfo staffSocketInfo = ChannelGroupService.STAFF_GROUP.get(userSocketInfo.getStaffId());
            if (staffSocketInfo != null) {
                staffSocketInfo.writeAndFlush(msgBody);
            }
            redisTemplate.opsForHash().put(NettyConstant.MSG_KEY + record.getUserId(), record.getId(), JsonUtils.obj2Json(msgBody));
        }
        log.info("用户 回执:" + record.getId());
    }

    /**
     * 恢复历史绑定关系
     *
     * @param userId
     */
    private void restoreBindingRelationship(Long userId) {
        if (redisTemplate.opsForHash().hasKey(NettyConstant.IM_USERS, userId)) {
            final Long staffId = (Long) redisTemplate.opsForHash().get(NettyConstant.IM_USERS, userId);
            final StaffSocketInfo staffSocketInfo = ChannelGroupService.STAFF_GROUP.get(staffId);
            if (staffSocketInfo != null) {
                ChannelGroupService.USER_GROUP.get(userId).setStaffId(staffId);
                staffSocketInfo.getUserIds().add(userId);
            }
        }
    }

    /**
     * 广播用户上线信息 并移除对应Set
     *
     * @param userId
     */
    private void broadcastUserOnline(Long userId) {
        final Long staffId = ChannelGroupService.USER_GROUP.get(userId).getStaffId();
        ChannelGroupService.STAFF_GROUP.forEach((key, staffSocketInfo) -> {
            staffSocketInfo.getUserIds().remove(userId);
            staffSocketInfo.writeAndFlush(new MsgBody<>().setStatus(MsgBody.USERS_ONLINE).setData(new IdModel().setStaffId(staffId).setUserId(userId)));
        });
    }
}
