im.doctor/src/server/models/sessions/sessions.js

/**
 * 会话模型。
 */
"use strict";

let RedisClient = require('../../repository/redis/redis.client.js');
let RedisModel = require('./../redis.model.js');
let ModelUtil = require('../../util/model.util');
let Messages = require('../messages/messages');
let Users = require('../user/users');
let DoctorRepo = require('../../repository/oracle/doctor.repo');
let Participants = require('./participants');
let SessionRepo = require('../../repository/oracle/session.repo');
let TopicRepo = require('../../repository/oracle/topics.repo');
let MessageRepo = require('../../repository/oracle/message.repo');
let ParticipantRepo = require('../../repository/oracle/participant.repo');
let ImDb = require('../../repository/oracle/db/im.db');
let WlyySDK = require("../../util/wlyy.sdk");
let ObjectUtil = require("../../util/object.util.js");

let WechatClient = require("../client/wechat.client.js");
let AppClient = require("../client/app.client.js");
let configFile = require('../../include/commons').CONFIG_FILE;
let config = require('../../resources/config/' + configFile);
let redis = RedisClient.redisClient().connection;
let logger = require('../../util/log.js');
let mongoose = require('mongoose');
let async = require("async");
let log = require("../../util/log.js");
let pubSub = require("../redis/pubSub.js");

const REDIS_KEYS = require('../../include/commons').REDIS_KEYS;
const SESSION_TYPES = require('../../include/commons').SESSION_TYPES;
const STICKY_SESSION_BASE_SCORE = require('../../include/commons').STICKY_SESSION_BASE_SCORE;
const PARTICIPANT_ROLES = require('../../include/commons').PARTICIPANT_ROLES;
const CONTENT_TYPES = require('../../include/commons').CONTENT_TYPES;
const SESSION_BUSINESS_TYPE = require('../../include/commons').SESSION_BUSINESS_TYPE;
const SESSION_STATUS = require('../../include/commons').SESSION_STATUS;


class Sessions extends RedisModel {
    constructor() {
        super();
    }

    /**
     * 创建会话。会话的ID来源：
     * MUC：患者的ID
     * P2P：对成员的ID排序后，取hash值
     * GROUP：团队的ID
     *
     * @param sessionId
     * @param name 会话名称
     * @param type 会话类型
     * @param participantArray 会话成员
     * @param handler 回调，仅MUC模式使用
     */
    createSession(sessionId, name, type, participantArray, handler) {
        let self = this;
        let messageId = mongoose.Types.ObjectId().toString();
        //创建session到mysql
        self.createSessionToMysql(sessionId, name, type, participantArray, messageId, function (err, res) {
            if (err) {
                if (handler) {
                    handler(err, null);
                    return;
                }
                ModelUtil.emitError(self.eventEmitter, {message: err, status: -1}, null);
            } else {
                name = res.name;
                //创建session到redis
                self.createSessionToRedis(sessionId, name, type, participantArray, messageId, function (err, res) {
                    if (err) {
                        if (handler) {
                            handler(err, null);
                            return;
                        }
                        ModelUtil.emitError(self.eventEmitter, {message: err, status: -1}, null);
                    } else {
                        if (handler) {
                            handler(null, res);
                            return;
                        }
                        ModelUtil.emitOK(self.eventEmitter, {status: 200, data: res});
                    }
                });
            }
        });
    }


    /**
     * 创建会话。REDIS
     * @param sessionId
     * @param name
     * @param type
     * @param participantArray
     * @param messageId
     * @param handler
     * @returns {boolean}
     */
    createSessionToRedis(sessionId, name, type, participantArray, messageId, handler) {
        let self = this;
        let messages = new Messages();
        let participantIdArray = [];
        for (let i in participantArray) {
            participantIdArray.push(participantArray[i].split(":")[0]);
        }
        if (type == SESSION_TYPES.P2P || type == SESSION_TYPES.SYSTEM) {
            if (sessionId) {
                callBusinessType(sessionId);
                return;
            }
            else if (participantIdArray.length != 2) {
                handler("P2P session only allow 2 participants.", null);
                return false;
            }else{
                ParticipantRepo.findSessionIdByParticipantIds(participantIdArray[0], participantIdArray[1], function (err, res) {
                    sessionId = res;
                    callBusinessType(sessionId);
                    return;
                });
            }
        } else {
            if (!sessionId) {
                if(type == SESSION_TYPES.DISCUSSION){
                    sessionId = messageId;
                }else {
                    handler("MUC OR GROUP session sessionId is not  allow null .", null);
                    return;
                }
            }
            callBusinessType(sessionId);
        }

        function callBusinessType(sessionId) {
            if(type == SESSION_TYPES.MUC||
                type == SESSION_TYPES.PRESCRIPTION ||
                SESSION_TYPES.PRESCRIPTION_HOSPITAL ||
                SESSION_TYPES.COLLABORATION_HOSPITAL ||
                SESSION_TYPES.GUIDANCE_HOSPITAL ||
                SESSION_TYPES.GENERAL_EXPERT ||
                SESSION_TYPES.PRESCRIPTION_HOSPITAL_VIDEO ||
                SESSION_TYPES.MUC_VIDEO ||
                SESSION_TYPES.ONDOOR_NURSING){
                callCreate(sessionId, SESSION_BUSINESS_TYPE.PATIENT);
            }else if(type==SESSION_TYPES.DISCUSSION||type==SESSION_TYPES.GROUP){
                callCreate(sessionId, SESSION_BUSINESS_TYPE.DOCTOR);
            }else {
                ParticipantRepo.getBusinessType(participantIdArray, function (err, businessType) {
                    callCreate(sessionId, businessType);
                });
            }
        }

        function callCreate(sessionId, businessType) {
            let createDate = new Date();
            Participants.saveParticipantsToRedis(sessionId, participantArray, createDate, function (res) {
                let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
                if (type == SESSION_TYPES.MUC||
                    type == SESSION_TYPES.PRESCRIPTION ||
                    SESSION_TYPES.PRESCRIPTION_HOSPITAL ||
                    SESSION_TYPES.COLLABORATION_HOSPITAL ||
                    SESSION_TYPES.GUIDANCE_HOSPITAL ||
                    SESSION_TYPES.GENERAL_EXPERT ||
                    SESSION_TYPES.PRESCRIPTION_HOSPITAL_VIDEO ||
                    SESSION_TYPES.MUC_VIDEO ||
                    SESSION_TYPES.ONDOOR_NURSING) {
                    businessType = 2;
                }
                let session = {
                    id: sessionId,
                    name: name,
                    type: type,
                    create_date: createDate.getTime(),
                    business_type: businessType
                };
                //如果会话已经存在的就不需要发送会话成功的消息不更新最后一条消息
                redis.hexistsAsync(sessionKey, sessionId).then(function(res){
                    if(res==0){
                        redis.hmsetAsync(sessionKey, session).then(function () {
                            handler(null, session);
                        })
                    }
                })
            });
        }
    }

    /**
     * 创建会话。mysql
     * @param sessionId
     * @param name
     * @param type
     * @param participantArray
     * @param messageId
     * @param handler
     */
    createSessionToMysql(sessionId, name, type, participantArray, messageId, handler) {
        let self = this;
        //如果sessionId不存在则执行创建sessionId过程
        let participantIdArray = [];
        for (let i in participantArray) {
            participantIdArray.push(participantArray[i].split(":")[0]);
        }
        //流程1-判断是否存在sessionId不存在则创建对应的sessionId;
        if (!sessionId) {
            if (type == SESSION_TYPES.P2P || type == SESSION_TYPES.SYSTEM) {
                ParticipantRepo.findSessionIdByParticipantIds(participantIdArray[0], participantIdArray[1], function (err, res) {
                    sessionId = res;
                    callBusinessType();
                });
            } else if(type == SESSION_TYPES.DISCUSSION){
                sessionId = messageId;
                callBusinessType();
            } else {
                return handler("MUC模式和团队模式，不允许sessionId为空！", null);
            }
        } else {
            callBusinessType();
        }

        //流程2-判断session的业务类型;
        function callBusinessType() {
            if(type==SESSION_TYPES.MUC||
                type==SESSION_TYPES.PRESCRIPTION ||
                SESSION_TYPES.PRESCRIPTION_HOSPITAL ||
                SESSION_TYPES.COLLABORATION_HOSPITAL ||
                SESSION_TYPES.GUIDANCE_HOSPITAL ||
                SESSION_TYPES.GENERAL_EXPERT ||
                SESSION_TYPES.PRESCRIPTION_HOSPITAL_VIDEO ||
                SESSION_TYPES.MUC_VIDEO ||
                SESSION_TYPES.ONDOOR_NURSING){
                callCreateSession(SESSION_BUSINESS_TYPE.PATIENT);
            }else if(type==SESSION_TYPES.SPECIALISTGROUP||type==SESSION_TYPES.DISCUSSION){
                callCreateSession(SESSION_BUSINESS_TYPE.DOCTOR);
            }else{
                ParticipantRepo.getBusinessType(participantIdArray, function (err, businessType) {
                    if (err) {
                        handler(err, null);
                        return;
                    }
                    callCreateSession(businessType);
                });
            }
        }

        //流程3-发起session创建 返回session实例
        function callCreateSession(businessType) {
            //查找该sessionId是否存在存在则直接返回实例
            SessionRepo.findOne(sessionId, function (err, res) {
                if (res && res.length > 0) {//已经存在
                    //已存在的会话不修改名称
                    name = res[0].name;
                    //更新成员
                    Participants.saveParticipantsToMysql(sessionId, participantArray, function (err, update) {
                        handler(null, res[0]);
                        return;
                    })
                    let createDate = new Date();
                    self.saveSessionToMysql(sessionId, name, type, createDate, businessType, function (err, res) {
                        logger.info("update session status is true");
                    })
                } else {
                    let createDate = new Date();
                    let session = {
                        id: sessionId,
                        name: name,
                        type: type,
                        create_date: createDate.getTime(),
                        business_type: businessType
                    };

                    //将session写入数据库
                    self.saveSessionToMysql(sessionId, name, type, createDate, businessType, function (err, res) {
                        if (err) {
                            handler(err, null);
                            return;
                        }
                        callCreateParticipants(session);
                    })

                }
            });
        }

        //流程4-发起session成员创建
        function callCreateParticipants(session) {
            Participants.saveParticipantsToMysql(sessionId, participantArray, function (err, res) {
                if (err) {
                    handler(err, null);
                    return;
                } else {
                    handler(null, session);
                    return;
                }
            })
        }
    }

    /**
     * 最近会话列表，7天内。
     *
     * @param userId
     * @param dateSpan
     */
    getRecentSessions(userId, dateSpan) {
        let self = this;
        SessionRepo.findAllByTimestampAndType(userId, dateSpan, function (err, res) {
            if (err) {
                ModelUtil.emitError(self.eventEmitter, "Get recent sessions failed", err);
                return;
            }

            let sessions = [];
            res.forEach(function (session) {
                //最近列表用于转发，过滤不可用的咨询
                if(session.last_content_type != CONTENT_TYPES.TopicEnd){
                    sessions.push({
                        id: session.id,
                        name: session.name,
                        type: session.type,
                        business_type: session.business_type,
                        create_date: session.create_date
                    })
                }
            });

            ModelUtil.emitOK(self.eventEmitter, sessions);
        });
    }

    /**
     * 保存session到MySQL
     * @param sessionId
     * @param name
     * @param type
     * @param createDate
     * @param businessType
     * @param handler
     */
    saveSessionToMysql(sessionId, name, type, createDate, businessType, handler) {
        SessionRepo.saveSession(sessionId, name, type, createDate, businessType, handler);
    }

    /**
     * 获取某个用户的全部session列表
     * @param userId
     * @param handler
     */
    getUserSessionsFromMysql(userId, handler) {
        SessionRepo.findAll(userId, handler);
    }

    /**
     * 获取session单个对象
     * @param sessionId
     * @param handler
     */
    getSessions(sessionId, handler) {
        SessionRepo.findOne(sessionId, handler);
    }

    /**
     * 判断会话是否存在
     * @param sessionId
     * @param handler
     */
    isExist(sessionId) {
        let self = this;
        SessionRepo.findOne(sessionId, function (err, res) {
            if(err){
                log.error(err);
                ModelUtil.emitError(self.eventEmitter, {message: err, status: -1}, null);
            } else if(res&&res.length>0){
                ModelUtil.emitOK(self.eventEmitter, {sessionId: res[0].id, status: 200});
            }else {
                ModelUtil.emitOK(self.eventEmitter, {sessionId: '', status: 200});
            }
        });
    }

    getSession(sessionId,userId,handler){
        let self = this;
        let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
        redis.hgetallAsync(sessionKey).then(function(session){
            if(session.type==SESSION_TYPES.P2P){
                ParticipantRepo.findNameById(userId, function (err, res) {
                    session.name = res[0].name;
                    if(handler){
                        handler(null,session);
                        return;
                    }
                    ModelUtil.emitOK(self.eventEmitter, session);
                })
            }else{
                if(handler){
                    handler(null,session);
                    return;
                }
                ModelUtil.emitOK(self.eventEmitter, session);
            }
        }).catch(function (err) {
            logger.error("Get session failed: ", err);
            if(handler){
                handler(null,session);
                return;
            }
            ModelUtil.emitError(self.eventEmitter, {message: err, status: -1}, null);
        })
    }

    /**
     *MDT 医生数据转换，根据手机号查询医生CODE
     */
    conversionAarticipant(doctorMobile, callback) {
        let doctorProto = DoctorRepo;
            doctorProto.findByMobile(doctorMobile, function (err, oDoctor) {
                if(oDoctor.length>0 && oDoctor[0].id){
                    if(callback){
                        callback(oDoctor[0].id)
                    }
                }else {
                    if(callback){
                        logger.error("Get doctoc code failed: mobile:", doctorMobile);
                        callback(doctorMobile)
                    }
                }
            });
    }

     /**
     * 根据用户ID获取用户的session列表
     * @param userId
     * @param page
     * @param size
     * @param businessType
     */
    getUserSessions(userId, page, size, businessType) {
        let userSessionKey = RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, userId);
        let self = this;
        if (page > 0) {
            if (page == 1) {
                page = 0;
            }
            page = page + page * size;
        }
        async.waterfall([
            // 获取会话ID列表
            // function (callback) {
            //     redis.zrevrangeAsync(userSessionKey, page, size)
            //         .then(function (sessionIds) {
            //             if (sessionIds.length == 0) {
            //                 ModelUtil.emitOK(self.eventEmitter, []);
            //                 return;
            //             }
            //             callback(null, sessionIds);
            //         })
            // },
            function (callback) {
                SessionRepo.findAllByType(userId,businessType,page,size,function(err,res){
                    if (res.length == 0) {
                        ModelUtil.emitOK(self.eventEmitter, []);
                        return;
                    }
                    var sessionIds=[];
                    for(var j in res){
                        sessionIds.push(res[j].id);
                    }
                    callback(null,sessionIds);
                })
            },
            // 遍历会话
            function (sessionIds) {
                let sessionList = [];
                let functionList = [];
                log.info("sessionIds.length:" + sessionIds.length);
                for (let j = 0; j < sessionIds.length; j++) {
                    let fun = function (index, callback) {
                        log.info("!callback:" + !callback);
                        if (!callback) {
                            callback = index, index = 0
                        }
                        let sessionId = sessionIds[index];
                        let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);

                        log.info("sessionKey:" + sessionKey);

                        let participantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
                        let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
                        redis.multi()
                            .hgetall(sessionKey)                       // 会话实体
                            .hget(participantsRoleKey, userId)         // 用户在此会话中的角色
                            .zscore(sessionParticipantsKey, userId)    // 用户在此会话中最后一次获取未读消息的时间
                            .zrange(sessionParticipantsKey, 0, -1)
                            .zrange(sessionParticipantsKey, 0,-1,'withscores')  // 所有用户在此会话中最后一次获取未读消息的时间
                            .hgetall(participantsRoleKey)  // 所有用户在此会话中角色
                            .execAsync()
                            .then(function (res) {
                                let session = res[0];
                                log.info("top.session.id:" + session.id);
                                log.info("top.session.name:" + session.name);
                                log.info("tope.session.business_type:" + session.business_type);
                                let role = res[1];
                                let lastFetchTime = res[2];
                                let users = res[3];
                                let participantsTimeArray = res[4];
                                let userRoles = res[5];
                                let participantsTime = [];
                                let isInvite = true;

                                //处理session未加入redis的bug

                                log.info("1.session==null:" + session==null);

                                if(session==null){
                                    let lastLoginTime = new Date();
                                    SessionRepo.findOne(sessionId, function (err, res) {
                                        if(res){
                                            session = res;

                                            log.info("1.session.id:" + session.id);
                                            log.info("1.session.name:" + session.name);
                                            log.info("1.session.business_type:" + session.business_type);

                                            let redisSession = [
                                                "id", session.id,
                                                "name", session.name,
                                                "type", session.type,
                                                "business_type", session.business_type,
                                                "last_sender_id", session.last_sender_id||"",
                                                "last_sender_name", session.last_sender_name||"",
                                                "last_content_type", session.last_content_type||"",
                                                "last_content", session.last_content||"",
                                                "last_message_time", session.last_message_time||"",
                                                "create_date", ObjectUtil.timestampToLong(session.create_date),
                                                "status",session.status==null?0:session.status
                                            ];
                                            // cache sessions
                                            redis.multi()
                                                .zadd(REDIS_KEYS.Sessions, lastLoginTime.getTime(), sessionId)                                           // 会话的最后活动时间设置为此用户的登录时间
                                                .zadd(RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, userId), lastLoginTime.getTime(), sessionId)      // 会话的最后活动时间设置为此用户的登录时间
                                                .hmset(RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId), redisSession)
                                                .execAsync()
                                                .then(function (res) {
                                                    // cache participants
                                                    let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
                                                    let sessionParticipantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
                                                    ParticipantRepo.findAll(sessionId, function (err, participants) {
                                                        if (err) {
                                                            ModelUtil.emitError(self.eventEmitter, err.message);
                                                            return;
                                                        }

                                                        let multi = redis.multi();
                                                        participants.forEach(function (participant) {
                                                            let participantId = participant.id;
                                                            let participantRole = participant.role;
                                                            let score = ObjectUtil.timestampToLong(participant.last_fetch_time||(new Date()));

                                                            multi = multi.zadd(sessionParticipantsKey, score, participantId)
                                                                .hset(sessionParticipantsRoleKey, participantId, participantRole);
                                                        });

                                                        multi.execAsync()
                                                            .then(function (res) {
                                                            })
                                                            .catch(function (ex) {
                                                                log.error("Login failed while caching participants: ", ex);
                                                            });
                                                    });

                                                    // cache messages
                                                    let messagesKey = RedisModel.makeRedisKey(REDIS_KEYS.Messages, sessionId);
                                                    let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
                                                    MessageRepo.findBySessionId(sessionId, 0, config.sessionConfig.maxMessageCount, null, function (err, messages) {
                                                        if (err) {
                                                            ModelUtil.emitError(self.eventEmitter, err.message);
                                                            return;
                                                        }

                                                        let multi = redis.multi();
                                                        messages.forEach(function (message) {
                                                            let msgJson = {
                                                                id: message.id,
                                                                sender_id: message.sender_id,
                                                                sender_name: message.sender_name,
                                                                timestamp: ObjectUtil.timestampToLong(message.timestamp),
                                                                content_type: message.content_type,
                                                                content: message.content
                                                            };

                                                            multi = multi.hset(messagesKey, message.id, JSON.stringify(msgJson))
                                                                .zadd(messagesByTimestampKey, ObjectUtil.timestampToLong(message.timestamp), message.id);
                                                        });

                                                        multi.execAsync()
                                                            .then(function (res) {
                                                            })
                                                            .catch(function (ex) {
                                                                log.error("Login failed while caching messages: ", ex);
                                                            });
                                                    });

                                                    // cache topics for MUC
                                                    let topicsKey = RedisModel.makeRedisKey(REDIS_KEYS.Topics, sessionId);
                                                    TopicRepo.findAllBySessionId(sessionId, function (err, topics) {
                                                        if (err) {
                                                            ModelUtil.emitError(self.eventEmitter, err.message);
                                                            return;
                                                        }

                                                        topics.forEach(function (topic) {
                                                            let topicKey = RedisModel.makeRedisKey(REDIS_KEYS.Topic, topic.id);
                                                            let topicId = topic.id;
                                                            let name = topic.name == null ? "" : topic.name;
                                                            let createTime = ObjectUtil.timestampToLong(topic.create_time);
                                                            let endBy = topic.end_by == null ? "" : topic.end_by;
                                                            let endTime = topic.end_time == null ? 0 : ObjectUtil.timestampToLong(topic.end_time);
                                                            let startMessageId = topic.start_message_id == null ? "" : topic.start_message_id;
                                                            let endMessageId = topic.end_message_id == null ? "" : topic.end_message_id;
                                                            let description = topic.description == null ? "" : topic.description;
                                                            let status = topic.status == null ? 0 : topic.status;
                                                            redisConn.multi()
                                                                .zadd(topicsKey, createTime, topicId)
                                                                .hmset(topicKey,
                                                                    'name', name,
                                                                    'session_id', sessionId,
                                                                    'create_time', createTime,
                                                                    'end_by', endBy,
                                                                    'end_time', endTime,
                                                                    'start_message_id', startMessageId,
                                                                    'end_message_id', endMessageId,
                                                                    'description', description,
                                                                    'status', status)
                                                                .execAsync()
                                                                .catch(function (ex) {
                                                                    log.error("Login failed while caching topics: ", ex);
                                                                });
                                                        });
                                                    });
                                                })
                                                .catch(function (ex) {
                                                    log.error("Login failed while caching sessions: ", ex);
                                                });
                                        }
                                    });
                                }

                                for(var j in userRoles){
                                    if(userRoles[j]==1){
                                        isInvite = false;
                                        break;
                                    }
                                }
                                for(var j = 0 ;j<participantsTimeArray.length;j++){
                                    if(j%2!=0)continue;
                                    let participantsTimeJson = {};
                                    participantsTimeJson[participantsTimeArray[j]] = participantsTimeArray[j+1];
                                    participantsTime.push(participantsTimeJson);
                                }
                                let sessionName = "";
                                let otherUserId = "";
                                if (session.type == SESSION_TYPES.P2P) {
                                    for (let j in users) {
                                        if (users[j] != userId) {
                                            otherUserId = users[j];
                                        }
                                    }
                                }

                                if (!role) role = 0;
                                if (!lastFetchTime) lastFetchTime = new Date().getTime();

                                // 计算未读消息数
                                let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
                                redis.zcountAsync(messagesByTimestampKey, parseInt(lastFetchTime)+1, new Date().getTime())
                                    .then(function (count) {
                                        if (!otherUserId) otherUserId = userId;
                                        ParticipantRepo.findNameById(otherUserId, function (err, res) {
                                            if ((res && res.length == 0) || session.type != SESSION_TYPES.P2P) {
                                                sessionName = session.name;
                                            } else {
                                                sessionName = res[0].name;
                                            }
                                            var bir = new Date().getTime();
                                            if (res.length != 0 && res[0].birthdate) {
                                                bir = res[0].birthdate.getTime();
                                            }
                                            var sex = 1;
                                            if (res.length != 0 && res[0].sex) {
                                                sex = res[0].sex;
                                            }
                                            //end

                                            log.info("2.session.id:" + sessionId);
                                            log.info("2.session.name:" + sessionName);
                                            log.info("2.session.business_type:" + session.business_type);

                                            sessionList.push({
                                                id: sessionId,
                                                name: sessionName,
                                                create_date: session.create_date,
                                                last_content_type: session.last_content_type||"",
                                                last_content: session.last_content||"",
                                                sender_id: session.last_sender_id||"",
                                                type: session.type,
                                                sender_name: session.last_sender_name||"",
                                                unread_count: count,
                                                // business_type: session.business_type,
                                                business_type: businessType,
                                                my_role: role,
                                                sender_sex: sex,
                                                sender_birthday: bir,
                                                participantsTimeArray:participantsTime,
                                                status:session.status,
                                                is_invite:isInvite
                                            });

                                            index = (parseInt(index) + 1);
                                            if (index == sessionIds.length) {
                                                ModelUtil.emitOK(self.eventEmitter, sessionList);
                                            } else {
                                                callback(null, index);
                                            }
                                        })
                                    })
                            })
                            .catch(function (err) {
                                logger.error("Get sessions failed: ", err);
                            });
                    };
                    functionList.push(fun);
                }
                async.waterfall(functionList);
            }
        ]);
    }

    /**
     * 查找会话数
     * @param userId
     * @param type
     * @param status
     * @param handler
     */
    getSessionCountByType(userId,type,status,handler){
        let self = this;
        SessionRepo.findSessionCountByType(userId,type,status,function (err,res) {
            if(res){
                ModelUtil.emitOK(self.eventEmitter,{status:200,count:res[0].count});
                return;
            }else {
                ModelUtil.emitOK(self.eventEmitter,{status:200,count:0});
                return;
            }
        })
    }

    /**
     * 根据用户类型获取用户的session列表
     * @param userId
     * @param page
     * @param size
     * @param businessType
     * @param status 0：进行中的会话1：结束的会话
     */
    getUserSessionsByType(userId,type,page, size,status) {
        logger.info("根据用户类型获取用户的session列表1: ");
        let userSessionKey = RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, userId);
        let self = this;
        logger.info("根据用户类型获取用户的session列表2: ");
        async.waterfall([
            // 获取会话ID列表
            function (callback) {
                SessionRepo.findListByType(userId,type,page,size,status,function(err,res){
                    logger.info("根据用户类型获取用户的session列表: res :"+res.toString());
                    logger.info("根据用户类型获取用户的session列表: res.length :"+res.length);
                    if (res == null || res.length == 0) {
                        ModelUtil.emitOK(self.eventEmitter, []);
                        return;
                    }
                    callback(null,res);
                })
            },
            // 遍历会话
            function (sessionIds) {
                let sessionList = [];
                let functionList = [];


                for (let j = 0; j < sessionIds.length; j++) {
                    let fun = function (index, callback) {
                        if (!callback) {
                            callback = index, index = 0
                        }

                        let mysqlSession = sessionIds[index];
                        let sessionId = mysqlSession.id;
                        let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
                        let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
                        redis.multi()
                            .hgetall(sessionKey)                       // 会话实体
                            .zscore(sessionParticipantsKey, userId)    // 用户在此会话中最后一次获取未读消息的时间
                            .zrange(sessionParticipantsKey, 0, -1)
                            .zrange(sessionParticipantsKey, 0,-1,'withscores')  // 所有用户在此会话中最后一次获取未读消息的时间
                            .execAsync()
                            .then(function (res) {
                                let session = res[0];
                                let lastFetchTime = res[1];
                                let users = res[2];
                                let participantsTimeArray = res[3];
                                let participantsTime = [];
                                for(var j = 0 ;j<participantsTimeArray.length;j++){
                                    if(j%2!=0)continue;
                                    let participantsTimeJson = {};
                                    participantsTimeJson[participantsTimeArray[j]] = participantsTimeArray[j+1];
                                    participantsTime.push(participantsTimeJson);
                                }
                                let sessionName = "";
                                let otherUserId = "";
                                if (session.type == SESSION_TYPES.P2P) {
                                    for (let j in users) {
                                        if (users[j] != userId) {
                                            otherUserId = users[j];
                                        }
                                    }
                                }

                                if (!lastFetchTime) lastFetchTime = new Date().getTime();
                                // 计算未读消息数
                                let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
                                redis.zcountAsync(messagesByTimestampKey, parseInt(lastFetchTime)+1, new Date().getTime())
                                    .then(function (count) {
                                        if (!otherUserId) otherUserId = userId;
                                        ParticipantRepo.findNameById(otherUserId, function (err, res) {
                                            if ((res && res.length == 0) || session.type != SESSION_TYPES.P2P) {
                                                sessionName = session.name;
                                            } else {
                                                sessionName = res[0].name;
                                            }
                                            var bir = new Date().getTime();
                                            if (res && res.length != 0 && res[0].birthdate) {
                                                bir = res[0].birthdate.getTime();
                                            }
                                            var sex = 1;
                                            if (res && res.length != 0 && res[0].sex) {
                                                sex = res[0].sex;
                                            }

                                            sessionList.push({
                                                id: sessionId,
                                                name: sessionName,
                                                create_date: new Date(mysqlSession.last_message_time).getTime(),
                                                last_content_type: session.last_content_type,
                                                last_content: session.last_content,
                                                sender_id: session.last_sender_id,
                                                type: session.type,
                                                sender_name: session.last_sender_name,
                                                unread_count: count,
                                                business_type: session.business_type,
                                                sender_sex: sex,
                                                sender_birthday: bir,
                                                participantsTimeArray:participantsTime,
                                                status:session.status,
                                            });

                                            index = (parseInt(index) + 1);
                                            if (index == sessionIds.length) {
                                                ModelUtil.emitOK(self.eventEmitter, sessionList);
                                            } else {
                                                callback(null, index);
                                            }
                                        })
                                    })
                            })
                            .catch(function (err) {
                                logger.error("Get sessions:"+sessionId+" failed: ", err);
                            });
                    };

                    functionList.push(fun);
                }

                async.waterfall(functionList);
            }
        ]);
    }

    /**
     * 根据用户ID获取用户已经结束咨询的session列表
     * @param userId
     * @param page
     * @param size
     * @param businessType
     */
    getUserStatusSessions(userId,status,businessType,page, size) {
        let userSessionKey = RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, userId);
        log.info(userId);
        let self = this;
        async.waterfall([
            // 获取会话ID列表
            function (callback) {
                SessionRepo.findAllByTypeAndStatus(userId,businessType,status,page,size,function(err,res){
                    if (res && res.length == 0) {
                        ModelUtil.emitOK(self.eventEmitter, []);
                        return;
                    }
                    // var sessionIds=[];
                    // for(var j in res){
                    //     sessionIds.push(res[j].id);
                    // }
                    callback(null,res);
                })
            },
            // 遍历会话
            function (sessionIds) {
                let sessionList = [];
                let functionList = [];
                for (let j = 0; j < sessionIds.length; j++) {
                    let fun = function (index, callback) {
                        if (!callback) {
                            callback = index, index = 0
                        }

                        let mysqlSession = sessionIds[index];
                        let sessionId = mysqlSession.id;
                        let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
                        let participantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
                        let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
                        redis.multi()
                            .hgetall(sessionKey)                       // 会话实体
                            .hget(participantsRoleKey, userId)         // 用户在此会话中的角色
                            .zscore(sessionParticipantsKey, userId)    // 用户在此会话中最后一次获取未读消息的时间
                            .zrange(sessionParticipantsKey, 0, -1)
                            .zrange(sessionParticipantsKey, 0,-1,'withscores')  // 所有用户在此会话中最后一次获取未读消息的时间
                            .hgetall(participantsRoleKey)  // 所有用户在此会话中角色
                            .execAsync()
                            .then(function (res) {
                                let session = res[0];
                                let role = res[1];
                                let lastFetchTime = res[2];
                                let users = res[3];
                                let participantsTimeArray = res[4];
                                let userRoles = res[5];
                                let participantsTime = [];
                                let isInvite = true;
                                for(var j in userRoles){
                                    if(userRoles[j]==1){
                                        isInvite = false;
                                        break;
                                    }
                                }
                                for(var j = 0 ;j<participantsTimeArray.length;j++){
                                    if(j%2!=0)continue;
                                    let participantsTimeJson = {};
                                    participantsTimeJson[participantsTimeArray[j]] = participantsTimeArray[j+1];
                                    participantsTime.push(participantsTimeJson);
                                }
                                let sessionName = "";
                                let otherUserId = "";
                                if (session.type == SESSION_TYPES.P2P) {
                                    for (let j in users) {
                                        if (users[j] != userId) {
                                            otherUserId = users[j];
                                        }
                                    }
                                }

                                if (!role) role = 0;
                                if (!lastFetchTime) lastFetchTime = new Date().getTime();

                                // 计算未读消息数
                                let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
                                redis.zcountAsync(messagesByTimestampKey, parseInt(lastFetchTime)+1, new Date().getTime())
                                    .then(function (count) {
                                        if (!otherUserId) otherUserId = userId;
                                        ParticipantRepo.findNameById(otherUserId, function (err, res) {
                                            if ((res && res.length == 0) || session.type != SESSION_TYPES.P2P) {
                                                sessionName = session.name;
                                            } else {
                                                sessionName = res[0].name;
                                            }
                                            var bir = new Date().getTime();
                                            if (res && res.length != 0 && res[0].birthdate) {
                                                bir = res[0].birthdate.getTime();
                                            }
                                            var sex = 1;
                                            if (res.length != 0 && res[0].sex) {
                                                sex = res[0].sex;
                                            }

                                            sessionList.push({
                                                id: sessionId,
                                                name: sessionName,
                                                create_date: new Date(mysqlSession.last_message_time).getTime(),
                                                last_content_type: session.last_content_type,
                                                last_content: session.last_content,
                                                sender_id: session.last_sender_id,
                                                type: session.type,
                                                sender_name: session.last_sender_name,
                                                unread_count: count,
                                                business_type: session.business_type,
                                                my_role: role,
                                                sender_sex: sex,
                                                sender_birthday: bir,
                                                participantsTimeArray:participantsTime,
                                                status:session.status,
                                                is_invite:isInvite
                                            });

                                            index = (parseInt(index) + 1);
                                            if (index == sessionIds.length) {
                                                ModelUtil.emitOK(self.eventEmitter, sessionList);
                                            } else {
                                                callback(null, index);
                                            }
                                        })
                                    })
                            })
                            .catch(function (err) {
                                logger.error("Get sessions:"+sessionId+" failed: ", err);
                            });
                    };

                    functionList.push(fun);
                }

                async.waterfall(functionList);
            }
        ]);
    }

    /**
     * 获取会话消息。全部，不管已读/未读状态。
     *
     * @param sessionId 会话ID
     * @param userId 拉取消息的人
     * @param page 第几页
     * @param pagesize 分页数量
     * @param start_msg_id 消息会话最新的一条消息的ID
     * @param end_msg_id 消息会话刚开始的消息ID
     * @remark
     * start_msg_id end_msg_id 为空取会话倒序的根据分页的消息数据
     * start_msg_id 为空 end_msg_id 不为空根据end_msg_id取旧的消息（offset=1可以排除本身）返回指定分页的数据
     * end_msg_id 为空 start_msg_id 不为空 根据 start_msg_id 取出新消息（offset=1可以排除本身）返回指定分页的数据
     * end_msg_id start_msg_id 都不为空返回指定区间的消息
     */
    getMessages(sessionId, user, start_msg_id, end_msg_id, page, pagesize, isoffset, handler) {
        let self = this;
        let message_timestamp_key = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
        if (!start_msg_id && !end_msg_id) {
            redis.zrevrangeAsync(message_timestamp_key, 0, 0).then(function (res) {
                logger.info("redis return res-----"+res);
                // log.info("session.js--830---getMessages  res.length: " + res.length);
                if (res && res.length == 0) {
                    //修复应redis没有缓冲聊天记录导致会话列表加载不出来
                    // cache messages
                    let messagesKey = RedisModel.makeRedisKey(REDIS_KEYS.Messages, sessionId);
                    MessageRepo.findBySessionId(sessionId, 0, config.sessionConfig.maxMessageCount, null, function (err, messages) {
                        if (err) {
                            ModelUtil.emitError(self.eventEmitter, err.message);
                            return;
                        }

                        let multi = redis.multi();
                        messages.forEach(function (message) {
                            let msgJson = {
                                id: message.id,
                                sender_id: message.sender_id,
                                sender_name: message.sender_name,
                                timestamp: ObjectUtil.timestampToLong(message.timestamp),
                                content_type: message.content_type,
                                content: message.content
                            };

                            multi = multi.hset(messagesKey, message.id, JSON.stringify(msgJson))
                                .zadd(message_timestamp_key, ObjectUtil.timestampToLong(message.timestamp), message.id);
                        });

                        multi.execAsync()
                            .then(function (res) {
                            })
                            .catch(function (ex) {
                                log.error("Login failed while caching messages: ", ex);
                                ModelUtil.emitOK(self.eventEmitter, res);
                                return;
                            });
                    });

                    // if (handler) {
                    //     handler(null, res);
                    //     return;
                    // }
                    // ModelUtil.emitOK(self.eventEmitter, res);
                    // return;
                }
                start_msg_id = res[0];
                redis.zrangeAsync(message_timestamp_key, 0, 0).then(function (res) {
                    if (res && res.length == 0) {
                        if (handler) {
                            handler(null, res);
                            return;
                        }
                        ModelUtil.emitOK(self.eventEmitter, res);
                        return;
                    }
                    end_msg_id = res[0];
                    self.getMessagesByPage(sessionId, user, end_msg_id, start_msg_id, page, pagesize, isoffset, function (err, res) {
                        if (err) {
                            if (handler) {
                                handler(err, null);
                                return;
                            }
                            logger.error("getMessagesByPage error" + err);
                            ModelUtil.emitError(self.eventEmitter, err, err);
                        } else {
                            if (handler) {
                                handler(null, res);
                                return;
                            }
                            ModelUtil.emitOK(self.eventEmitter, res);
                        }
                    })
                })
            })
        } else if (!start_msg_id) {
            redis.zrevrangeAsync(message_timestamp_key, 0, 0).then(function (res) {
                if (res && res.length == 0) {
                    if (handler) {
                        handler(null, res);
                        return;
                    }
                    ModelUtil.emitOK(self.eventEmitter, res);
                    return;
                }
                start_msg_id = res[0];
                self.getMessagesByPage(sessionId, user, start_msg_id,end_msg_id , page, pagesize, isoffset, function (err, res) {
                    if (err) {
                        if (handler) {
                            handler(err, null);
                            return;
                        }
                        logger.error("getMessagesByPage error" + err);
                        ModelUtil.emitError(self.eventEmitter, err, err);
                    } else {
                        if (handler) {
                            handler(null, res);
                            return;
                        }
                        ModelUtil.emitOK(self.eventEmitter, res);
                    }
                })
            })
        } else if (!end_msg_id) {
            redis.zrangeAsync(message_timestamp_key, 0, 0).then(function (res) {
                if (res && res.length == 0) {
                    ModelUtil.emitOK(self.eventEmitter, res);
                    return;
                }
                end_msg_id = res[0];
                self.getMessagesByPage(sessionId, user, end_msg_id,start_msg_id, page, pagesize, isoffset, function (err, res) {
                    if (err) {
                        if (handler) {
                            handler(err, null);
                            return;
                        }
                        logger.error("getMessagesByPage error" + err);
                        ModelUtil.emitError(self.eventEmitter, err, err);
                    } else {
                        if (handler) {
                            handler(null, res);
                            return;
                        }
                        ModelUtil.emitOK(self.eventEmitter, res);
                    }
                })
            })
        } else {
            self.getMessagesByPage(sessionId, user, end_msg_id, start_msg_id, page, pagesize, isoffset, function (err, res) {
                if (err) {
                    if (handler) {
                        handler(err, null);
                        return;
                    }
                    logger.error("getMessagesByPage error" + err);
                    ModelUtil.emitError(self.eventEmitter, err, err);
                } else {
                    if (handler) {
                        handler(null, res);
                        return;
                    }
                    ModelUtil.emitOK(self.eventEmitter, res);
                }
            })
        }
    }

    /**
     * 分页获取会话消息。
     *
     * @param sessionId     必选。会话ID
     * @param userId        必选。用户ID
     * @param startMsgId    必选。会话的的起始消息ID，作为检索的起始依据
     * @param endMsgId      必选。会话中的结束消息ID
     * @param page          必选。页码
     * @param size          必选。页面大小
     * @param handler       必选。回调
     */
    getMessagesByPage(sessionId, userId, startMsgId, endMsgId, page, size, isoffset, handler) {
        let messagesTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
        let messagesKey = RedisModel.makeRedisKey(REDIS_KEYS.Messages, sessionId);
        let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
        let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
        let participants = new Participants();
        let offset = (page - 1 < 0 ? 0 : page - 1) * size;
        let count = size;
        if (page > 1 || isoffset == 1) {
            offset += 1;    // 翻页由于闭区间，需跳过本身数据
        }

        participants.existsParticipant(sessionId, userId, function (err, res) {
            log.info("1151-userId=" + userId);
            log.info(" res[0].exist=" + res[0].exist);
            log.info(!res[0].exist && userId != "system");
            if (! res[0].exist && userId != "system") {
                handler(Error("User not found in session " + sessionId), null);
            } else {
                //将消息ID转换成分值
                redis.multi()
                    .zscore(messagesTimestampKey, startMsgId)
                    .zscore(messagesTimestampKey, endMsgId)
                    .hgetall(sessionKey)
                    .zrange(sessionParticipantsKey, 0, -1)
                    .execAsync()
                    .then(function (res) {
                        let startMsgScore = res[1];
                        let endMsgScore = res[0];
                        let session = res[2];
                        let users = res[3];
                        if (startMsgScore == null || endMsgScore == null || (startMsgScore == endMsgScore && isoffset == 1)) {
                            handler(null, []);
                            return;
                        }
                        //结束大于开始，正序取数据,返回的数据顺序也是逆序的，反向拉取数据，
                        // 当end>start取出来都是空的，为了给前端获取新数据使用，一般不出现这种情况
                        if(endMsgScore>startMsgScore){
                            redis.zrangebyscoreAsync(messagesTimestampKey, startMsgScore, endMsgScore, "limit", offset, count)
                                .then(function (res) {
                                    if (res && res.length == 0) {
                                        handler(null, []);
                                        return;
                                    }
                                    redis.hmgetAsync(messagesKey, res).then(function (messages) {
                                        messages.reverse();
                                        handler(null, messages);
                                    }).then(function () {
                                        Sessions.updateParticipantLastFetchTime(sessionId, userId, new Date().getTime());
                                        if(session.type == SESSION_TYPES.P2P){
                                            for(var j in users){
                                                if(users[j]==userId)continue;
                                                 WechatClient.sendAllRead(users[j],sessionId);
                                            }
                                        }else if(session.type == SESSION_TYPES.MUC ||
                                            session.type == SESSION_TYPES.PRESCRIPTION ||
                                            SESSION_TYPES.PRESCRIPTION_HOSPITAL ||
                                            SESSION_TYPES.COLLABORATION_HOSPITAL ||
                                            SESSION_TYPES.GUIDANCE_HOSPITAL ||
                                            SESSION_TYPES.GENERAL_EXPERT ||
                                            SESSION_TYPES.PRESCRIPTION_HOSPITAL_VIDEO ||
                                            SESSION_TYPES.MUC_VIDEO ||
                                            SESSION_TYPES.ONDOOR_NURSING){
                                            for(var j in users){
                                                if(users[j]==userId)continue;
                                                WechatClient.sendMucAllRead(users[j],userId,sessionId);
                                            }
                                        }

                                    })
                                })
                                .catch(function (err) {
                                    logger.error("Get message by page failed: ", err);

                                    handler(err, false);
                                })
                        }else{
                            // 从消息时间表中过滤出要获取的消息ID列表，倒序取出消息
                            redis.zrevrangebyscoreAsync(messagesTimestampKey, startMsgScore, endMsgScore, "limit", offset, count)
                                .then(function (res) {
                                    if (res && res.length == 0) {
                                        handler(null, []);
                                        return;
                                    }

                                    redis.hmgetAsync(messagesKey, res).then(function (messages) {
                                        handler(null, messages);
                                    }).then(function () {
                                        Sessions.updateParticipantLastFetchTime(sessionId, userId, new Date().getTime());
                                        if(session.type == SESSION_TYPES.P2P){
                                            for(var j in users){
                                                if(users[j]==userId)continue;
                                                //通知对方自己已经读取数据
                                                WechatClient.sendAllRead(users[j],sessionId);
                                            }
                                        }else if(session.type == SESSION_TYPES.MUC ||
                                            session.type == SESSION_TYPES.PRESCRIPTION ||
                                            SESSION_TYPES.PRESCRIPTION_HOSPITAL ||
                                            SESSION_TYPES.COLLABORATION_HOSPITAL ||
                                            SESSION_TYPES.GUIDANCE_HOSPITAL ||
                                            SESSION_TYPES.GENERAL_EXPERT ||
                                            SESSION_TYPES.PRESCRIPTION_HOSPITAL_VIDEO ||
                                            SESSION_TYPES.MUC_VIDEO ||
                                            SESSION_TYPES.ONDOOR_NURSING){
                                            for(var j in users){
                                                if(users[j]==userId)continue;
                                                //如果是患者拉取数据告诉在线的医生患者已经读取数据
                                                WechatClient.sendMucAllRead(users[j],userId,sessionId);
                                            }
                                        }
                                    })
                                })
                                .catch(function (err) {
                                    logger.error("Get message by page failed: ", err);
                                    handler(err, false);
                                })
                        }
                    })
            }
        })
    }

    /**
     * 获取所有会话的未读消息数。
     */
    getAllSessionsUnreadMessageCount(userId,type,handler) {
        let self = this;
        let count = 0;
        let patientCount = 0;
        let doctorCount = 0;
        let patientEndCount = 0;
        SessionRepo.findAll(userId, function (err, res) {
        // SessionRepo.findUnEndAll(userId, function (err, res) {
            if (err) {
                if(handler)
                {
                    handler(err,res);
                    return;
                }
                ModelUtil.logError("getAllSessionsUnreadMessageCount is failed", err);
                return;
            }

            if (res && res.length == 0) {
                if(handler)
                {
                    handler(err,count);
                    return;
                }
                ModelUtil.emitOK(self.eventEmitter, {count: count});
                return;
            }

            for (let j in res) {

                //是否过滤指定类型
                if( type != null){

                    if(type != res[j].type){
                        continue;
                    }
                }

                if (res[j].type == SESSION_TYPES.SYSTEM) {

                    if (j == res.length - 1) {
                        if(handler){
                            handler(err,count);
                            return;
                        }
                        ModelUtil.emitOK(self.eventEmitter, {count: count, patient: patientCount, doctor: doctorCount});
                    }
                    continue;
                }


                callback(res, j, res[j]);
            }
        });

        function callback(res, j, session) {
            self.getSessionUnreadMessageCount(res[j].id, userId, function (err, con) {
                if (err) {
                    if(handler)
                    {
                        handler(err,count);
                        return;
                    }
                    ModelUtil.logError("getAllSessionsUnreadMessageCount is failed", err);
                }

                count = count + con;
                if (session.business_type == SESSION_BUSINESS_TYPE.PATIENT) {
                    if(session.status == SESSION_STATUS.ENDED){//新增判断是否咨询结束
                        patientEndCount = patientEndCount + con;
                    }else{
                        patientCount = patientCount + con;
                    }
                } else {
                    doctorCount = doctorCount + con;
                }
                if (j == res.length - 1) {
                    if(handler)
                    {
                        handler(err,count)
                        return;
                    }
                    ModelUtil.emitOK(self.eventEmitter, {count: count, patient: patientCount,patientEnd: patientEndCount, doctor: doctorCount});
                }
            })
        }
    }

    /**
     * 获取会话的未读消息数。根据成员最后一次获取消息的时候与当前时间。
     *
     * @param sessionId
     * @param userId
     */
    getSessionUnreadMessageCount(sessionId, userId, handler) {
        let self = this;
        let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
        let participantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);

        async.waterfall([
            // 此成员最后获取消息的时间
            function (callback) {
                redis.zscoreAsync(participantsKey, userId)
                    .then(function (lastFetchTime) {
                        callback(null, lastFetchTime);
                    })
            },
            // 计算最后获取消息的时间之后到现在有多少条消息
            function (lastFetchTime, callback) {
                if (!lastFetchTime) lastFetchTime = 0;
                let now = new Date().getTime();
                redis.zcountAsync(messagesByTimestampKey, parseInt(lastFetchTime)+1, now)
                    .then(function (count) {
                        if (handler) {
                            handler(null, count);
                        } else {
                            ModelUtil.emitOK(self.eventEmitter, {count: count});
                        }
                    })
            }
        ], function (err, res) {
            if (err) {
                if (handler) {
                    handler(err, 0);
                } else {
                    ModelUtil.emitError(self.eventEmitter, "Get session unread message count failed.")
                }
            }
        });
    }

    /**
     * 获取会话未读消息数。根据成员最后一次获取消息的时候与当前时间。
     */
    getSessionUnreadMessages(sessionId, userId) {
        let self = this;
        let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
        let participantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);

        async.waterfall([
            // 此成员最后获取消息的时间
            function (callback) {
                redis.zscoreAsync(participantsKey, userId)
                    .then(function (lastFetchTime) {
                        callback(null, lastFetchTime);
                    })
            },
            // 最后获取消息的时间之后到现在的消息ID列表
            function (lastFetchTime, callback) {
                if (!lastFetchTime) lastFetchTime = 0;
                let now = new Date().getTime();
                redis.zrangebyscoreAsync(messagesByTimestampKey, lastFetchTime, now)
                    .then(function (messageIds) {
                        callback(null, messageIds);
                    })
            },
            // 获取消息
            function (messageIds, callback) {
                if (messageIds.length == 0) {
                    ModelUtil.emitOK(self.eventEmitter, []);
                    return;
                }

                let startMsgId = messageIds[0];
                let endMsgId = messageIds[messageIds.length - 1];
                self.getMessagesByPage(sessionId, userId, startMsgId, endMsgId, 0, messageIds.length, 0, function (err, res) {
                    if (err) {
                        ModelUtil.emitError(self.eventEmitter, err.message);
                        return;
                    }

                    ModelUtil.emitOK(self.eventEmitter, res);
                });
            }
        ], function (err, res) {
            if (err) {
                ModelUtil.emitError(self.eventEmitter, "Get session unread message count failed.")
            }
        });
    }

    /**
     * 保存消息。
     *
     * 也可以根据议题保存消息，但最终还是保存到与会话对象。
     *
     * see also: saveMessageByTopic
     *
     * @param message
     * @param sessionId
     */
    saveMessageBySession(sessionId, message) {
        let self = this;
        let messages = new Messages();
        let participants = new Participants();
        let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
        let messageId = mongoose.Types.ObjectId().toString();
        let sessionType =0;
        let sessionName;
        message.id = messageId;

        // 检查会话中是否存在此成员
        participants.existsParticipant(sessionId, message.sender_id, function (err, res) {
            if (err) {
                ModelUtil.emitError(self.eventEmitter, "Check session participant failed: ", err);
                return;
            }
            logger.info("session.js--1443--res[0].exist" + res[0].exist);
            log.info("message.sender_id=" + message.sender_id);
            if ( res[0].exist || message.sender_id == "system") {
                redis.hmgetAsync(sessionKey, ["type", "name"]).then(function (res) {
                    sessionType = res[0];
                    let sessionName = res[1];
                    if (sessionType == null) {
                        self.getSessions(sessionId,function(err,res){
                            if (err){
                                logger.error("session data is error");
                            } else {
                                sessionName=res[0].name;
                                sessionType = res[0].type;
                            }
                        });
                        let participantArray = [];
                        let participantsStr="{";
                        ParticipantRepo.findAll(sessionId, function (err, participants) {
                            if (err) {
                                ModelUtil.emitError(self.eventEmitter, err.message);
                                return;
                            }

                            participants.forEach(function (participant) {
                                let participantId = participant.id;
                                let participantRole = participant.role;
                                let score = ObjectUtil.timestampToLong(participant.last_fetch_time||(new Date()));
                                participantsStr +="\""+participantId+"\":\""+participantRole+"\",";
                            });

                            participantsStr = participantsStr.substring(0,participantsStr.length-1)+'}';
                            participantsStr = JSON.parse(participantsStr);
                            for (let j in participantsStr) {
                                participantArray.push(j + ":" + participantsStr[j]);
                            }
                            //创建session到redis
                            self.createSessionToRedis(sessionId, sessionName, sessionType, participantArray, messageId, function (err, res) {
                                if (err) {
                                    if (handler) {
                                        handler(err, null);
                                    }
                                    ModelUtil.emitError(self.eventEmitter, {message: err, status: -1}, null);
                                } else {
                                    if (handler) {
                                        handler(null, res);
                                    }
                                    ModelUtil.emitOK(self.eventEmitter, {status: 200, data: res});
                                }
                            });
                        });
                    }
                    if(sessionType == SESSION_TYPES.MUC ||
                        sessionType == SESSION_TYPES.PRESCRIPTION ||
                        SESSION_TYPES.PRESCRIPTION_HOSPITAL ||
                        SESSION_TYPES.COLLABORATION_HOSPITAL ||
                        SESSION_TYPES.GUIDANCE_HOSPITAL ||
                        SESSION_TYPES.GENERAL_EXPERT ||
                        SESSION_TYPES.PRESCRIPTION_HOSPITAL_VIDEO ||
                        SESSION_TYPES.MUC_VIDEO ||
                        SESSION_TYPES.ONDOOR_NURSING){
                        if(message.content_type == CONTENT_TYPES.PlainText ||
                           message.content_type == CONTENT_TYPES.Image ||
                           message.content_type == CONTENT_TYPES.Audio||
                           message.content_type == CONTENT_TYPES.Video){
                            TopicRepo.findLastBySessionId(sessionId,function(err,res){
                                if(res&&res.length>0&&res[0].reply==0){
                                    TopicRepo.replyTopic(message.sender_id,message.id,res[0].id,function(err,res){
                                        if(err){
                                            logger.error("update topic reply error");
                                        }else{
                                            logger.warn("update topic reply success");
                                        }
                                    });
                                }
                            })
                        }
                    }

                    // 消息保存到Redis，并更新会话最后状态、用户最后消息获取时间
                    messages.saveMessageToRedis(sessionId, sessionType, messageId, message);
                    Messages.updateLastContent(sessionKey, sessionType, sessionName, message);
                    Sessions.updateParticipantLastFetchTime(sessionId, message.sender_id, message.timestamp.getTime());
                    // 更新MYSQL中会话的最新状态，并保存消息
                    SessionRepo.updateSessionLastStatus(message.sender_id, message.sender_name, message.timestamp, message.content, message.content_type, sessionId);
                    messages.saveMessageToMysql(sessionId, sessionType, messageId, message, function (err, res) {
                        if (err) {
                            ModelUtil.emitError(self.eventEmitter, {message: "Failed to save message to mysql: " + err});
                        } else {
                            message.timestamp = message.timestamp.getTime();
                            ModelUtil.emitOK(self.eventEmitter, {count: 1, messages: [message]});
                        }
                    });
                }).then(function (res) {
                    // 推送消息
                    ParticipantRepo.findIds(sessionId, function (err, res) {
                        if (err) {
                            ModelUtil.logError("Push message from session: get participant's id list failed: ", err);
                        } else {
                            message.session_id = sessionId;
                            res.forEach(function (participant) {
                                if (participant.id == message.sender_id||sessionType==SESSION_TYPES.SYSTEM){
                                    message.sender_img = participant.avatar;
                                    callPush(res,message);
                                }
                            })
                        }
                    })
                }).catch(function (err) {
                    ModelUtil.emitError(self.eventEmitter, {message: "Error occurred while save message to session: " + err});
                })
            } else {
                ModelUtil.emitDataNotFound(self.eventEmitter, {message: "当前会话找不到此发送者"});
            }
        });
        function callPush(participants,message){
            participants.forEach(function (participant) {
                if (participant.id !== message.sender_id &&
                    participant.participant_role == PARTICIPANT_ROLES.HOST) {
                    Sessions.pushNotification(participant.id, participant.name, message,sessionType);
                }
            });
        }
    }


    sendTopicMessages(topicId, message) {
        let self = this;
        TopicRepo.findAllByTopicId(topicId, function (err, res) {
            if (err || res.length == 0) {
                ModelUtil.emitOK(self.eventEmitter, {status: -1, "message": "议题获取失败"});
                return;
            }
            self.saveMessageByTopic(message, res[0].session_id, function (err, messageId) {
                if (err) {
                    ModelUtil.emitOK(self.eventEmitter, {status: -1, "message": err});
                } else {
                    message.id = messageId;
                    ModelUtil.emitOK(self.eventEmitter, {status: 200, "message": "发送成功", data: message});
                }
            });
        });
    }

    /**
     * 保存消息
     *
     * @param message
     * @param sessionId
     * @param handler
     */
    saveMessageByTopic(message, sessionId, handler) {
        let messages = new Messages();
        let participants = new Participants();
        let session_key = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
        let messageId = mongoose.Types.ObjectId().toString();
        let self = this;
        let sessionType = 0;
        let sessionName = "";
        let agent = message.agent;//代理人
        message.id = messageId;
        if(!message.timestamp){
            message.timestamp = new Date();
        }
        // 发送成员必须处于会话中
        participants.existsParticipant(sessionId, message.sender_id, function (err, res) {
            log.info("1599-message.sender_id=" + message.sender_id);
            log.info("res[0].exist=" + res[0].exist);
            if ( res[0].exist || message.sender_id == "system") {
                redis.hmgetAsync(session_key, ["type", "name"]).then(function (res) {
                    sessionType = res[0];
                    sessionName = res[1];
                    if (!sessionType || !sessionName) {
                        logger.error("Unknown session key " + session_key);
                        if (handler) {
                            handler(new Error("Unknown session key " + session_key));return;
                        };
                    }
                }).then(function (res) {
                    // 消息数据双写，并更新用户最后消息获取时间，会话新状态等
                    messages.saveMessageToRedis(sessionId, sessionType, messageId, message);
                    messages.saveMessageToMysql(sessionId, sessionType, messageId, message);
                    // 更新会话最新状态及成员最后一次消息获取时间
                    Sessions.updateParticipantLastFetchTime(sessionId, message.sender_id, message.timestamp.getTime());
                    Messages.updateLastContent(session_key, sessionType, sessionName, message);
                    SessionRepo.updateSessionLastStatus(message.sender_id, message.sender_name, message.timestamp, message.content, message.content_type, sessionId);
                    if (handler) {
                        handler(null, messageId);
                        return;
                    }
                }).then(function (res) {
                    // 推送消息
                    ParticipantRepo.findIds(sessionId, function (err, res) {
                        if (err) {
                            logger.error(err);
                        } else {
                            message.session_id = sessionId;
                            res.forEach(function (participant) {
                                if (participant.id == message.sender_id){
                                    message.sender_img = participant.avatar;
                                    callPush(res,message);
                                }
                            })
                        }
                    })
                }).catch(function (err) {
                    log.error(err);
                    return;
                })
            } else {
                if (handler){ handler("用户不在此会话当中！", messageId);return;}
            }
        });
        function callPush(participants,message){
            participants.forEach(function (participant) {
                if ((participant.id !== message.sender_id||message.content_type == CONTENT_TYPES.PrescriptionBloodStatus ||
                    message.content_type == CONTENT_TYPES.PrescriptionFollowupContent) &&
                    participant.participant_role == PARTICIPANT_ROLES.HOST) {
                    Sessions.pushNotification(participant.id, participant.name, message,sessionType);
                }
            });
        }
    }

    /**
     * 保存代理人进入的消息
     *
     * @param message
     * @param sessionId
     * @param handler
     */
    saveIntoMessageByTopic(message, sessionId, handler) {
        log.info("2--保存代理人进入的消息:");
        let messages = new Messages();
        let participants = new Participants();
        let session_key = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
        let messageId = mongoose.Types.ObjectId().toString();
        let self = this;
        let sessionType = 0;
        let sessionName = "";
        let agent = message.agent;//代理人
        message.id = messageId;
        if(!message.timestamp){
            message.timestamp = new Date();
        }
        // 发送成员必须处于会话中
        participants.existsParticipant(sessionId, message.sender_id, function (err, res) {
            log.info("session.js--1477--res[0].exist" + res[0].exist);
            log.info("message.sender_id=" + message.sender_id);
            if (res[0].exist || message.sender_id == "system") {
                redis.hmgetAsync(session_key, ["type", "name"]).then(function (res) {
                    sessionType = res[0];
                    sessionName = res[1];
                    if (!sessionType || !sessionName) {
                        logger.error("Unknown session key " + session_key);
                        if (handler) {
                            handler(new Error("Unknown session key " + session_key));return;
                        };
                    }
                }).then(function (res) {
                    //查找最后一条数据，如果一致就不保存
                    MessageRepo.findLastMessage(sessionId,sessionType,function (err, res) {
                        if (err) {
                            logger.error(err);
                        } else {
                            res.forEach(function (mes) {
                                // if(mes.content==message.content&&message.content_type==mes.content_type){
                                if(mes.agent==message.agent){
                                    handler(null, messageId);
                                    return;
                                }else {
                                    // 消息数据双写，并更新用户最后消息获取时间，会话新状态等
                                    messages.saveMessageToRedis(sessionId, sessionType, messageId, message);
                                    messages.saveMessageToMysql(sessionId, sessionType, messageId, message);
                                    // 更新会话最新状态及成员最后一次消息获取时间
                                    Sessions.updateParticipantLastFetchTime(sessionId, message.sender_id, message.timestamp.getTime());
                                    Messages.updateLastContent(session_key, sessionType, sessionName, message);
                                    SessionRepo.updateSessionLastStatus(message.sender_id, message.sender_name, message.timestamp, message.content, message.content_type, sessionId);
                                    if (handler) {
                                        handler(null, messageId);
                                        return;
                                    }
                                }
                            })
                        }
                    });

                }).then(function (res) {
                    // 推送消息
                    ParticipantRepo.findDoctorIds(sessionId, function (err, res) {
                        if (err) {
                            logger.error(err);
                        } else {
                            message.session_id = sessionId;
                            res.forEach(function (participant) {
                                Sessions.pushIntoNotification(participant.id, participant.name, message,sessionType);
                            })
                        }
                    })
                }).catch(function (err) {
                    log.error(err);
                    return;
                })
            } else {
                log.info("4--用户不在此会话当中！");
                if (handler){ handler("用户不在此会话当中！", messageId);return;}
            }
        });
    }

    /**
     * 置顶操作
     */
    stickSession(sessionId, user) {
        let user_session_key = RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, user);
        let self = this;
        //取出最大的session
        redis.zrevrangeAsync(user_session_key, 0, 0).then(function (res) {
            //获取该session的时间搓
            redis.zscoreAsync(user_session_key, res).then(function (scoreres) {
                let nowtime = new Date().getTime();
                //当前时间搓比redis的时间搓更早证明没有置顶过
                if (scoreres <= nowtime) {
                    //初始化置顶
                    redis.zaddAsync(user_session_key, STICKY_SESSION_BASE_SCORE, sessionId).then(function (res) {
                        logger.info("stickSession:" + sessionId + ",res:" + res);
                        ModelUtil.emitOK(self.eventEmitter, {});
                    }).then(function () {
                        SessionRepo.saveStickySession(sessionId, user, STICKY_SESSION_BASE_SCORE);
                    })
                } else {
                    //已有置顶的数据，取出来加1保存回去
                    scoreres = Number(scoreres) + 1;
                    redis.zaddAsync(user_session_key, scoreres, sessionId).then(function () {
                        logger.info("stickSession:" + sessionId + ",res:" + res);
                        ModelUtil.emitOK(self.eventEmitter, {});
                    }).then(function () {
                        SessionRepo.saveStickySession(sessionId, user, scoreres);
                    })
                }
            })
        })
    }

    /**
     * 取消会话置顶
     */
    cancelStickSession(sessionId, user) {
        let user_session_key = RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, user);
        let participants_key = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
        let self = this;
        redis.zscoreAsync(participants_key, user).then(function (res) {
            if (!res) {
                res = new Date().getTime();
            }
            redis.zaddAsync(user_session_key, res, sessionId).then(function (res) {
                logger.info("cancelStickSession:" + sessionId);
                ModelUtil.emitOK(self.eventEmitter, {});
            }).then(function () {
                SessionRepo.unStickySession(sessionId, user);
            });
        })
    }

    /**
     * 更新会话参与者的最后消息获取时间。
     *
     * @param sessionId
     * @param userId
     */
    static updateParticipantLastFetchTime(sessionId, userId, score) {
        score = score + 1;
        let participantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
        redis.zaddAsync(participantsKey, score, userId)
            .then(function (res) {
                ParticipantRepo.updateLastFetchTime(new Date(score), sessionId, userId, function (err, res) {
                    if (err) {
                        logger.error("Update participant last fetch time failed: ", err);
                }
                });
            })
            .catch(function (err) {
                logger.error("Update participant last fetch time failed: ", err);
            });
    }

    /**
     * 向用户推送通知，微信端用户直接推送消息，APP端通过个推发送通知消息。
     *
     * @param targetUserId
     * @param message
     */
    static pushNotification(targetUserId, targetUserName, message,sessionType) {
        let self = this;
        Users.isPatientId(targetUserId, function (err, isPatient) {
            if (isPatient) {
                if(config.environment!='local'){//pc版不直接发送给居民，通过redis的publish
                    WechatClient.sendMessage(targetUserId, targetUserName, message);
                }
                message.targetUserId = targetUserId;
                message.targetUserName = targetUserName;
                message.sessionType = sessionType;
                message.targetType = 'patient';
            } else {
                if(sessionType==SESSION_TYPES.P2P){
                     WechatClient.sendReadDoctorByDoctorId(targetUserId, message);
                }
                //告知医生新消息
                WechatClient.sendSocketMessageToDoctor(targetUserId,message);

                let count = 0;
                //系统消息
                // MessageRepo.getWlyyMessageCount(targetUserId,function (err,res) {
                //     if(res){
                //         count = res[0].count;
                //     }
                // });
                //im消息
                let sessions = new Sessions();
                sessions.getAllSessionsUnreadMessageCount(targetUserId,null,function (err,res) {
                    if(res){
                        count += res;
                    }
                });

                if(config.environment!='local'){//pc版不推送个推，通过redis的publish
                    log.info("推送医生智能助手")
                    AppClient.sendNotification(targetUserId, message,sessionType,count);
                }
                //外网pcim通过socket推送
                WechatClient.sendPcImSocket(targetUserId,message,sessionType);

                // WlyySDK.request(targetUserId, '', '', '', '/im/common/message/messages', 'POST', function (err, res) {
                //     let count = 0;
                //     if(err){
                //         logger.error(err);
                //     }else {
                //         logger.error(res);
                //         res =  JSON.parse(res)
                //         if (res.status == 200) {
                //             let data = res.data;
                //             count = parseInt(JSON.parse(data.imMsgCount).count) + parseInt(data.system.amount) + parseInt(data.healthIndex.amount) + parseInt(data.sign.amount);
                //         }
                //     }
                //     if(config.environment!='local'){//pc版不推送个推，通过redis的publish
                //         AppClient.sendNotification(targetUserId, message,sessionType,count);
                //     }
                //     //外网pcim通过socket推送
                //     WechatClient.sendPcImSocket(targetUserId,message,sessionType);
                // });

                message.targetUserId = targetUserId;
                message.targetUserName = targetUserName;
                message.sessionType = sessionType;
                message.targetType = 'doctor';

                //推送MDT，医生外层新消息
                WechatClient.sendMDTSocketMessageToDoctor(targetUserId,message);
            }
            //redis发布消息
            if(config.pubSubSwitch) {//接收订阅消息处理开关，本地运行和测试库单独运行时防止用户接收消息2次
                pubSub.publish(config.pubChannel,JSON.stringify(message));
            }
        });
    }

    /**
     * 获取redis订阅消息，并处理
     * @param targetUserId
     * @param targetUserName
     * @param message
     * @param sessionType
     */
    static getRedisPushNotification(message) {
        if (message.targetType=='patient') {
            if(config.environment!='local'){//pc版接收要发给居民的消息不做处理
                WechatClient.sendMessage(message.targetUserId, message.targetUserName, message);
            }
        } else {
            if(message.sessionType==SESSION_TYPES.P2P){
                WechatClient.sendReadDoctorByDoctorId(message.targetUserId, message);
            }
            //告知医生新消息
            WechatClient.sendSocketMessageToDoctor(message.targetUserId,message);
            if(config.environment!='local'){//pc版不推送个推
                WlyySDK.request(message.targetUserId, '', '', '', '/im/common/message/messages', 'POST', function (err, res) {
                    let count = 0;
                    res =  JSON.parse(res)
                    if (res.status == 200) {
                        let data = res.data;
                        count = parseInt(JSON.parse(data.imMsgCount).count) + parseInt(data.system.amount) + parseInt(data.healthIndex.amount) + parseInt(data.sign.amount);
                    }
                    AppClient.sendNotification(message.targetUserId, message,message.sessionType,count);
                });
            }
        }
    }

    /**
     * 向APP端通过socket发送通知消息。
     *
     * @param targetUserId
     * @param message
     */
    static pushIntoNotification(targetUserId, targetUserName, message,sessionType) {

        if(sessionType==SESSION_TYPES.P2P){
            WechatClient.sendReadDoctorByDoctorId(targetUserId, message);
        }
        //告知医生新消息
        WechatClient.sendSocketMessageToDoctor(targetUserId,message);
    }

    /**
     * 针对MUC模式更新会话的当前状态
     * @param sessionId
     */
    updateSessionStatus(sessionId,status,handler){
        let self = this;
        let sessionKey  =  RedisModel.makeRedisKey(REDIS_KEYS.Session,sessionId);
        redis.hsetAsync(sessionKey,"status",status).then(function(res){
                SessionRepo.updateSessionStatus(sessionId,status,function(err,sqlResult){
                    if(handler){
                        handler(err,sqlResult);
                        return;
                    }
                    if(err){
                        logger.error("set session status to mysql is error !");
                    }else{
                        logger.info("set session status is success");
                        ModelUtil.emitOK(self.eventEmitter, []);
                    }
                });
        });

    }

    /**
     * 针对专科医生和家医模式更新会话的名称
     * @param sessionId
     * @param name
     */
    updateSessionName(sessionId,name,handler){
        let self = this;
        let sessionKey  =  RedisModel.makeRedisKey(REDIS_KEYS.Session,sessionId);
        redis.hsetAsync(sessionKey,"name",name).then(function(res){
            SessionRepo.updateSessionName(sessionId,name,function(err,sqlResult){
                if(handler){
                    handler(err,sqlResult);
                    return;
                }
                if(err){
                    logger.error("set session name to mysql is error !");
                    ModelUtil.emitError(self.eventEmitter,"set session name to mysql is error !",err)
                }else{
                    logger.info("set session name is success");
                    ModelUtil.emitOK(self.eventEmitter, []);
                }
            });
        });

    }


}

// Expose class
module.exports = Sessions;