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.entity.base.ChannelAttributeKey;
import com.ym.im.entity.base.NettyConstant;
import com.ym.im.entity.model.IdModel;
import com.ym.im.factory.PushFactory;
import com.ym.im.handler.ChannelGroupHandler;
import com.ym.im.mq.Queue;
import com.ym.im.service.ChatRecordService;
import com.ym.im.service.ChatService;
import com.ym.im.service.MsgBodyService;
import com.ym.im.util.JsonUtils;
import com.ym.im.validation.group.*;
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.HashMap;
import java.util.Map;
import java.util.Set;

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

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

    @Autowired
    private Queue queue;

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

    @Autowired
    private ChatRecordService chatRecordService;

    @Autowired
    private MsgBodyService msgBodyService;

    @Autowired
    private ChannelGroupHandler channelGroup;

    @Autowired
    private PushFactory pushFactory;

    @Override
    public void init(ChannelHandlerContext ctx) {

        final Long staffId = ctx.channel().attr(ChannelAttributeKey.ROLE_ID).get();
        final Long merchantId = ctx.channel().attr(ChannelAttributeKey.MERCHANT_ID).get();
        StaffSocketInfo staffSocketInfo = new StaffSocketInfo();
        staffSocketInfo.setStaffId(staffId);
        staffSocketInfo.setMerchantId(merchantId);
        staffSocketInfo.setToken(ctx.channel().attr(ChannelAttributeKey.TOKEN_INFO).get());
        staffSocketInfo.setChannel((NioSocketChannel) ctx.channel());
        staffSocketInfo.setUserIds(redisTemplate.opsForSet().members(NettyConstant.STAFF_USERIDS_KEY + staffId));
        channelGroup.putMerchantStaff(merchantId, staffSocketInfo);
        log.info("客服: " + staffId + " 上线:");
    }

    @Override
    public void offline(ChannelHandlerContext ctx) {

        final Long staffId = ctx.channel().attr(ChannelAttributeKey.ROLE_ID).get();
        final Long merchantId = ctx.channel().attr(ChannelAttributeKey.MERCHANT_ID).get();
        final StaffSocketInfo staffSocketInfo = channelGroup.getMerchantStaff(merchantId, staffId);
        if (ctx.channel().attr(ChannelAttributeKey.TOKEN_INFO).get().equals(staffSocketInfo.getToken())) {
            final Set<Long> userIds = staffSocketInfo.getUserIds();
            if (userIds.size() != 0) {
                final String userListKey = NettyConstant.STAFF_USERIDS_KEY + staffId;
                redisTemplate.delete(userListKey);
                redisTemplate.opsForSet().add(userListKey, userIds.toArray());
                queue.staffOfflineQueue(new StaffSocketInfo(staffId, userIds)); //NioSocketChannel无法序列化 所以new StaffSocketInfo
            }
            channelGroup.removeMerchantStaff(staffId);
            ctx.close();
            log.info("客服: " + staffId + " 下线:");
        }
    }

    @Override
    @Validated({MsgBodyGroup.class, ChatRecordSendGroup.class, StaffSendGroup.class})
    public NioSocketChannel distribution(Long id, @Valid MsgBody<ChatRecord> msgBody) {
        final ChatRecord chatRecord = msgBody.getData();
        final Long userId = chatRecord.getUserId();
        final Long merchantId = channelGroup.getMerchantStaff(id).getMerchantId();
        final UserSocketInfo userSocketInfo = channelGroup.USER_GROUP.get(userId);
        if (userSocketInfo == null) {
            //用户不在线,保存最后发送消息的客服ID
            Map<Long, Long> staff = new HashMap<>();
            staff.put(merchantId, id);
            redisTemplate.opsForHash().put(NettyConstant.IM_USERS, userId, staff);
            //推送通知
            pushNotifications(userId);
            return null;
        }
        final Long currentStaffId = userSocketInfo.getStaffId(merchantId);
        if (currentStaffId == null) {
            //通知用户 新的客服
            userSocketInfo.setStaff(merchantId, id);
            userSocketInfo.writeAndFlush(new MsgBody<>().setCode(MsgBody.DISTRIBUTION_STAFF).setData(new IdModel().setStaffId(id).setUserId(userId)));
            channelGroup.getMerchantStaff(id).getUserIds().add(userId);
        }
        if (currentStaffId != null && !currentStaffId.equals(id)) {
            //通知客服 绑定失败 当前用户已绑定客服
            channelGroup.getMerchantStaff(id).writeAndFlush(new MsgBody<>().setCode(MsgBody.BINDINGFAILURE).setData(new IdModel().setStaffId(currentStaffId).setUserId(userId).setMerchantId(merchantId)));
            return null;
        }
        return userSocketInfo.getChannel();
    }


    @Override
    @Validated({MsgBodyGroup.class, ChatRecordSendGroup.class, StaffSendGroup.class})
    public void save(Long id, @Valid MsgBody<ChatRecord> msgBody) {
        // 设置聊天基本信息
        final ChatRecord record = msgBody.getData();
        record.setId(IdWorker.getId())
                .setMerchantId(channelGroup.getMerchantStaff(id).getMerchantId())
                .setSendReceive(RECEIVE)
                .setCreateTime(new Date());
        // 先保存至数据库，再发送消息（若颠倒顺序可能导致数据未保存，更新已读操作先执行导致消息一直是未读状态
        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);
        pushNotifications(msgBody.getData().getUserId());
    }


    /**
     * 通知用户端客服端收到消息
     *
     * @param msgBody 消息对象
     * @throws JsonProcessingException e
     */
    @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 = channelGroup.USER_GROUP.get(record.getUserId());
        if (userSocketInfo != null) {
            userSocketInfo.writeAndFlush(msgBody);
            redisTemplate.opsForHash().put(NettyConstant.MSG_KEY + record.getUserId(), record.getId(), JsonUtils.obj2Json(msgBody));
        }
        log.info("客服 消息回执:" + record.getId());
    }

    /**
     * 推送通知
     *
     * @param userId
     */
    private void pushNotifications(Long userId) {
        final DeviceTokenParam deviceTokenParam = channelGroup.USER_DEVICE_PUSH_TOKEN.get(userId);
        pushFactory.getService(deviceTokenParam.getDeviceType()).unicast(deviceTokenParam.getDeviceToken());
    }


}

