package cn.com.ty.lift.ud.neety.process;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import cn.com.ty.lift.ud.chat.service.IChatSessionService;
import cn.com.ty.lift.ud.chat.service.impl.ChatSessionServiceImpl;
import cn.com.ty.lift.ud.utils.ApplicationContextUtil;
import cn.com.ty.lift.ud.redis.ChatDo;
import cn.com.ty.lift.ud.redis.RedisUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.DependsOn;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import cn.com.ty.lift.ud.neety.protocol.ChatMsgDTO;
import cn.com.ty.lift.ud.neety.protocol.IMDecoder;
import cn.com.ty.lift.ud.neety.protocol.IMEncoder;
import cn.com.ty.lift.ud.neety.protocol.IMMessage;
import cn.com.ty.lift.ud.neety.protocol.IMP;
import cn.com.ty.lift.ud.neety.protocol.RoomVo;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.GlobalEventExecutor;

/** 消息处理类 */

@Deprecated

@Lazy
@Component
@DependsOn("redisUtil")
public class MsgProcessor {
    /** 记录在线用户 */
    private static ChannelGroup onlineUsers = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    private static final Logger logger = LoggerFactory.getLogger(MsgProcessor.class);

    /** 用户名 */
    private final AttributeKey<String> USERNAME = AttributeKey.valueOf("username");
    /** 头像 */
    private final AttributeKey<String> HEAD_PIC = AttributeKey.valueOf("headPic");
    /** ip */
    private final AttributeKey<String> IP_ADDR = AttributeKey.valueOf("ipAddr");

    private final AttributeKey<JSONObject> ATTRS = AttributeKey.valueOf("attrs");
    /** 房间号 */
    private final AttributeKey<String> ROOM = AttributeKey.valueOf("ROOM");
    /** 当前用户id */
    private final AttributeKey<String> SENDID = AttributeKey.valueOf("SENDID");
    /** 小房间用户关系对应 */
    public static Map<String, Set<Channel>> roomUser = new HashMap<>();
    /** redis缓存 */
    private static RedisUtil redis = ApplicationContextUtil.getBean(RedisUtil.class);

    private static IChatSessionService sessionService = ApplicationContextUtil.getBean(ChatSessionServiceImpl.class);
    /** 解析模式（本地测试为：false） */
    private final static boolean testFlag = true;

    /**
     * websoket消息处理
     * 
     * @param client
     * @param msg
     */
    public void process(Channel client, String msg) {
        // 将字符串解析为自定义格式
        IMMessage request = IMDecoder.decode(msg);
        if (null == request) {
            return;
        }

        // 获取消息发送者
        String username = request.getName();
        String room = request.getSessionid();
        String sendId = request.getFromUser();
        String avatarUrl = request.getAvatarUrl();

        client.attr(IP_ADDR).getAndSet(getAddress(client));
        client.attr(USERNAME).getAndSet(username);
        client.attr(HEAD_PIC).getAndSet(avatarUrl);
        client.attr(ROOM).getAndSet(room);
        client.attr(SENDID).getAndSet(sendId);

        // room为总得大房间，map中的key为小房间的房间号，值为房间内对应每个用户的对象
        Map<String, List<RoomVo>> redisRoom = redis.get("ROOM", HashMap.class);
        String txt = redis.getKey("ROOM");
        Object ob = (txt == null || "null".equals(txt)) ? null : JSONObject.parseObject(txt, HashMap.class).get(room);

        // 获取当前用户的小房间持有用户
        Set<Channel> minRoom = null;
        if (null != roomUser.get(room)) {
            minRoom = roomUser.get(room);
        } else {
            minRoom = new HashSet<>();
        }
        boolean flag = true;
        // 确保单个用户服务端channel是唯一的,避免重复发送同一条消息
        for (Channel channel : minRoom) {
            if (channel == client) {
                flag = false;
                continue;
            }
        }
        if (flag) {
            logger.info("初始化channel .");
            minRoom.add(client);
        } else {
            logger.warn("重复初始化channel .");
        }

        // 初始化登录操作
        if (IMP.LOGINALL.getName().equals(request.getCmd())) {

            /** 当前用户有效的房间号 */
            List<String> DBsessionList = sessionService.findListByUserId(sendId);
            for (String dbSession : DBsessionList) {
                roomUser.put(dbSession, minRoom);
            }

            Map allMsg = JSONObject.parseObject(txt, HashMap.class);
            if (null != DBsessionList && null != allMsg) {
                /** 房间号所对应的用户信息 */
                List<RoomVo> voList = new ArrayList<>();
                /** 未读消息列表 */
                Map<String, List<RoomVo>> unreadMap = new HashMap<>();
                /** 登录对象集合 */
                Map<String, List<RoomVo>> loginMap = new HashMap<>();
                List<RoomVo> UNreadList = new ArrayList<>();
                for (String sessionKey : DBsessionList) {
                    Object keyValue = allMsg.get(sessionKey);

                    if (null != keyValue) {

                        // 对应房间的用户信息
                        voList = JSONArray.parseArray(keyValue.toString(), RoomVo.class);
                        UNreadList.addAll(voList);
                        // 组装在线消息
                        String loginList = JSONObject.toJSONString(voList);

                        // 设置该用户持有房间的在线状态
                        for (RoomVo item : voList) {
                            if (sendId.equals(item.getUserId())) {
                                item.setAvatarUrl(avatarUrl);
                                item.setFlag(true);
                                item.setName(username);
                            }
                        }
                        unreadMap.put(sessionKey, voList);
                        loginMap.put(sessionKey, voList);

                        // 向我的所有房间发送上线消息
                        for (Channel item : roomUser.get(sessionKey)) {

                            if (item != client) {
                                // 自定义系统消息格式
                                IMMessage requestSystem = new IMMessage(IMP.SYSTEM.getName(), sysTime(), 0, loginList);
                                sendMessage(item, requestSystem);
                            }
                        }
                        // 更新缓存中所有房间的未读消息列表,将未读消息置空
                        for (RoomVo item : voList) {
                            if (item.getMsgList().size() > 0 && item.getUserId().equals(sendId)) {
                                item.setMsgList(null);// 将未读消息置空
                            }
                        }

                        // 写入缓存对象
                        redisRoom.put(sessionKey, voList);
                    }

                }

                /** 向自己发送消 */
                // 组装 UNREAD消息
                request = new IMMessage(IMP.UNREAD.getName(), sysTime(), 0, JSONObject.toJSONString(unreadMap));
                // sendOnline(client, JSONObject.toJSONString(loginMap));
                sendMessage(client, request);

            }
            // 如果诊单是登录动作
        } else if (IMP.INITROOM.getName().equals(request.getCmd())) {

            // TODO 完成诊单之后将从缓存中删除房间
            roomUser.put(room, minRoom);

            List<RoomVo> voList = new ArrayList<>();
            RoomVo vo = new RoomVo(sendId, 0, username, avatarUrl, true, room);

            if (null != redisRoom) {// 缓存中获取全部房间信息

                if (null == ob) {// 拿到当前房间号
                    voList.add(vo);
                    redisRoom.put(room, voList);
                } else {

                    voList = JSONArray.parseArray(ob.toString(), RoomVo.class);
                    voList.add(vo);
                    voList = voList.stream().filter(distinctByKey(RoomVo::getUserId)).collect(Collectors.toList()); // 去重

                    for (RoomVo item : voList) {
                        if (sendId.equals(item.getUserId())) {
                            item.setAvatarUrl(avatarUrl);
                            item.setFlag(true);
                            item.setName(username);
                        }
                    }

                    redisRoom.put(room, voList);
                }

            } else {
                redisRoom = new HashMap<>();
                voList.add(vo);
                redisRoom.put(room, voList);
            }
            // 获取房间内的用户对象信息
            String loginList = JSONObject.toJSONString(voList);

            /** 向所有用户发送系统消息 */
            for (Channel channel : minRoom) {
                if (channel != client) {
                    // 自定义系统消息格式
                    IMMessage requestSystem = new IMMessage(IMP.SYSTEM.getName(), sysTime(), 0, loginList);
                    sendMessage(channel, requestSystem);
                } else {/** 向自己发送消 */

                    /**
                     * 1、提取该用户所有房间的未读消息列表返回给用户 2、此处为登录初始化调用，不会频繁触发，所以直接查询mysql用的会话列表。来配合缓存读取用户的所有未读消息
                     */

                    if (null != ob) {
                        voList = JSONArray.parseArray(ob.toString(), RoomVo.class);
                        for (RoomVo item : voList) {
                            /** 提取未读消息列表返回给用户 */
                            if (item.getMsgList().size() > 0 && item.getUserId().equals(sendId)) {
                                request = new IMMessage(IMP.UNREAD.getName(), sysTime(), 0,
                                    JSONObject.toJSONString(item.getMsgList()));
                                item.setMsgList(null);// 将未读消息置空

                                // 写入缓存对象
                                redisRoom.put(room, voList);
                            }
                        }
                    } else {

                        request = new IMMessage(IMP.SYSTEM.getName(), sysTime(), 0, loginList);
                    }

                    logger.info("发送消息~~~~~");
                    sendOnline(channel, loginList);
                    sendMessage(channel, request);

                    /*List<String> DBsessionList = sessionService.findListByUserId(sendId);
                    Map allMsg = JSONObject.parseObject(txt, HashMap.class);
                    if (null != DBsessionList && null != allMsg) {
                    
                        // 拿到该用户的所有房间的会话列表以及未读消息列表
                        List<RoomVo> UNreadList = new ArrayList<>();
                        Map<String, List<RoomVo>> unreadMap = new HashMap<>();
                        for (String sessionKey : DBsessionList) {
                            Object keyValue = allMsg.get(sessionKey);
                    
                            if (null != keyValue) {
                                voList = JSONArray.parseArray(keyValue.toString(), RoomVo.class);
                                unreadMap.put(sessionKey, voList);
                                UNreadList.addAll(voList);
                            }
                    
                        }
                    
                        // 组装 UNREAD消息
                        request = new IMMessage(IMP.UNREAD.getName(), sysTime(), 0, JSONObject.toJSONString(unreadMap));
                    
                        // 更新缓存中所有房间的未读消息列表
                        for (RoomVo item : UNreadList) {
                            if (item.getMsgList().size() > 0 && item.getUserId().equals(sendId)) {
                                item.setMsgList(null);// 将未读消息置空
                                // 写入缓存对象
                                redisRoom.put(room, voList);
                            }
                        }
                    } else {
                    
                        request = new IMMessage(IMP.SYSTEM.getName(), sysTime(), 0, loginList);
                    }
                    
                    logger.info("发送消息~~~~~");
                    sendOnline(channel, loginList);
                    sendMessage(channel, request);*/

                }

            }

        } else if (IMP.LOGOUT.getName().equals(request.getCmd())) {/** 如果是登出 */
            logout(client);
        } else if (IMP.CHAT.getName().equals(request.getCmd())) {/** 如果是聊天信息 */

            ChatMsgDTO msgDb = null;
            for (Channel channel : minRoom) {// 向房间内在线用户发送消息
                if (channel != client) {
                    request.setName(username);
                    if (null == msgDb) {
                        msgDb = saveMsg(request);
                    }

                } else { /** 向自己发送消息 */
                    request.setName("MY_SELF");
                }

                sendMessage(channel, request);
            }

            if (null != ob) {/** 向不在线的用户发送未读消息 */

                List<RoomVo> voList = JSONArray.parseArray(ob.toString(), RoomVo.class);
                for (RoomVo item : voList) {
                    if (!item.getFlag()) {
                        // 用户不在线记录未读消息同时向缓存中写入聊天记录，供后期同步到数据库中
                        if (null == msgDb) {
                            msgDb = saveMsg(request);
                        }
                        item.getMsgList().add(msgDb);
                        // 写入缓存对象
                        redisRoom.put(room, voList);
                    }
                }
            }
        }

        // 更新缓存
        redis.setValue("ROOM", redisRoom);

    }

    /**
     * 向房间内在线用户发送其它人的在线状况
     * 
     * @param channel
     * @param loginList
     */
    private void sendOnline(Channel channel, String loginList) {
        // 自定义IM协议解码
        IMMessage request = new IMMessage(IMP.SYSTEM.getName(), sysTime(), 0, loginList);
        String text = IMEncoder.toJsonString(request, testFlag);
        // 发送消息
        channel.writeAndFlush(new TextWebSocketFrame(text));
    }

    /**
     * 发送文本消息
     * 
     * @param channel
     * @param request
     */
    private void sendMessage(Channel channel, IMMessage request) {
        // 自定义IM协议解码
        String text = IMEncoder.toJsonString(request, testFlag);
        // 发送消息
        channel.writeAndFlush(new TextWebSocketFrame(text));
    }

    private static <T> Predicate<T> distinctByKey(Function<? super T, ?> keyExtractor) {
        Set<Object> seen = ConcurrentHashMap.newKeySet();
        return t -> seen.add(keyExtractor.apply(t));
    }

    /**
     * 提取缓存中还未同步到数据库的聊天记录，追加上新的聊天记录 如果缓存中没有历史聊天记录创建新的缓存对象
     * 
     * @param request
     *            请求对象
     * @return 消息对象
     */
    private ChatMsgDTO saveMsg(IMMessage request) {
        if (redis.exists(request.getSessionid())) {/** 提取缓存中还未同步到数据库的聊天记录，追加上新的聊天记录 */
            ChatDo chatDo = redis.get(request.getSessionid(), ChatDo.class);
            return this.saveMsg(chatDo, request.getFromUser(), request.getMsg(), 1, request.getType(),
                request.getSessionid(), request.getName(), request.getAvatarUrl());

        } else {/** 如果缓存中没有历史聊天记录创建新的缓存对象 */
            return this.saveMsg(new ChatDo(), request.getFromUser(), request.getMsg(), 1, request.getType(),
                request.getSessionid(), request.getName(), request.getAvatarUrl());
        }
    }

    /**
     * 获取用户昵称
     * 
     * @param client
     * @return
     */
    public String getNickName(Channel client) {
        return client.attr(USERNAME).get();
    }

    /**
     * 获取用户远程IP地址
     * 
     * @param client
     * @return
     */
    public String getAddress(Channel client) {
        return client.remoteAddress().toString().replaceFirst("/", "");
    }

    /**
     * 获取扩展属性
     * 
     * @param client
     * @return
     */
    public JSONObject getAttrs(Channel client) {
        try {
            return client.attr(ATTRS).get();
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 获取扩展属性
     * 
     * @param client
     * @return
     */
    private void setAttrs(Channel client, String key, Object value) {
        try {
            JSONObject json = client.attr(ATTRS).get();
            json.put(key, value);
            client.attr(ATTRS).set(json);
        } catch (Exception e) {
            JSONObject json = new JSONObject();
            json.put(key, value);
            client.attr(ATTRS).set(json);
        }
    }

    /**
     * 登出通知
     * 
     * @param client
     */
    public void logout(Channel client) {
        String room = client.attr(ROOM).get();
        String sendId = client.attr(SENDID).get();
        Map<String, List<RoomVo>> redisRoom = redis.get("ROOM", HashMap.class);
        String txt = redis.getKey("ROOM");

        IMMessage request = new IMMessage();
        request.setName(client.attr(USERNAME).get());
        request.setCmd(IMP.SYSTEM.getName());
        request.setOnline(0);
        request.setSessionid(client.attr(ROOM).get());

        Set<Channel> minRoom = null;
        if (null != roomUser.get(request.getSessionid())) {
            minRoom = roomUser.get(request.getSessionid());
        } else {
            minRoom = new HashSet<>();
        }

        List<RoomVo> voList = new ArrayList<>();
        List<String> DBsessionList = sessionService.findListByUserId(sendId);
        if (null != DBsessionList) {

            // 拿到该用户的所有房间的会话列表以及未读消息列表
            Map allMsg = JSONObject.parseObject(txt, HashMap.class);
            for (String sessionKey : DBsessionList) {
                Object keyValue = allMsg.get(sessionKey);

                if (null != keyValue) {
                    voList = JSONArray.parseArray(keyValue.toString(), RoomVo.class);
                    for (RoomVo item : voList) {
                        if (item.getUserId().equals(sendId)) {
                            item.setFlag(false);
                        }
                    }
                    redisRoom.put(sessionKey, voList);

                    String loginList = JSONObject.toJSONString(voList);
                    request.setMsg(loginList);
                    // 向小房间内的所有用户发送系统消息
                    for (Channel channel : minRoom) {
                        if (channel != client) {
                            // 发送消息
                            sendMessage(channel, request);
                        }
                    }
                }

            }

        }

        // 更新缓存数据
        redis.setValue("ROOM", redisRoom);

        // 删除用户信息
        minRoom.remove(client);
        // 当前小房间没有用户了。删除小房间
        if (minRoom.size() <= 0) {
            roomUser.remove(room);
        }
    }

    /**
     * 系统当前时间
     * 
     * @return
     */
    private long sysTime() {
        return System.currentTimeMillis();
    }

    /**
     * 将聊天记录存入缓存中
     * 
     * @param chatDo
     *            聊天记录
     * @param sendUserId
     *            发送人id
     * @param msg
     *            文本信息
     * @param status
     *            状态 0:正常1:屏蔽
     * @param type
     *            消息类型：1文本、2图片、3视频、4语音'
     * @param sessionid
     *            房间号号'
     */
    private ChatMsgDTO saveMsg(ChatDo chatDo, String sendUserId, String msg, Integer status, Integer type,
        String sessionid, String name, String avatarUrl) {
        ChatMsgDTO entity = new ChatMsgDTO();
        entity.setMsg(msg);
        entity.setStatuz(status);
        entity.setName(name);
        entity.setSessionid(sessionid);
        entity.setType(type);
        entity.setFromUser(sendUserId);
        entity.setAvatarUrl(avatarUrl);
        entity.setCreateTime(LocalDateTime.now().toString());
        chatDo.getList().add(entity);

        redis.setValue(sessionid, chatDo);
        return entity;
    }
}
