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 Participants = require('./Participants');
let SessionRepo = require('../../repository/mysql/session.repo');
let TopicRepo = require('../../repository/mysql/topics.repo');
let ParticipantRepo = require('../../repository/mysql/participant.repo');
let MessageRepo = require('../../repository/mysql/message.repo');

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");


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 SESSION_BUSINESS_TYPE = require('../../include/commons').SESSION_BUSINESS_TYPE;

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){
                logger.error(err);
            }else{
                logger.info("create session to mysql success :" +JSON.stringify(res));
            }
        });

        //创建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();
        var 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;
            }
            if (participantIdArray.length != 2) {
                handler("P2P session only allow 2 participants.",null);
                return false;
            }
            ParticipantRepo.findSessionIdByParticipantIds(participantIdArray[0], participantIdArray[1], function (err, res) {
                sessionId = res;
                callBusinessType(sessionId);
            });
        } else {
            if(!sessionId){
                handler("MUC OR GROUP session sessionId is not  allow null .",null);
                return;
            }
            callBusinessType(sessionId);
        }

        function callBusinessType(sessionId) {
            ParticipantRepo.getBusinessType(participantIdArray.join("','"),function(err,businessType){
                callCreate(sessionId, businessType);
            });
        }

        function callCreate(sessionId, businessType) {
            let createDate = new Date();
            let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
            let message = {
                sender_id: "system",
                sender_name: "system",
                content_type: 11,
                content: "会话创建成功",
                timestamp: createDate,
                id:messageId
            };
            let session = {
                id :sessionId,
                name:name,
                type:type,
                create_date:createDate.getTime(),
                business_type:businessType,
                last_sender_id : message.sender_id,
                last_sender_name : message.sender_name,
                last_message_time: message.timestamp.getTime(),
                last_content  : message.content,
                last_content_type : message.content_type
            }
            redis.hmsetAsync(sessionKey, session).then(function(){
                Participants.saveParticipantsToRedis(sessionId, participantArray, createDate, function (res) {
                    handler(null, session);
                    //messages.saveMessageToRedisFromCreateSession(sessionId, messageId, message);
                });
            })
        }
    }

    /**
     * 创建会话。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过程
        var 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{
                handler("MUC模式和团队模式，不允许sessionId为空！",null);
                return;
            }
        }else{
            callBusinessType();
        }

        //流程2-判断session的业务类型;
        function callBusinessType(){
            ParticipantRepo.getBusinessType(participantIdArray.join("','"),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.length > 0) {//已经存在
                    handler(null,res[0]);
                }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;};
                callBeginTrans(session);
            })
        }

        //流程5-发起session会话
        function callBeginTrans(session) {
            let mesDate = new Date();
            let message = {
                sender_id: "system",
                sender_name: "system",
                content_type: 6,
                content: "会话创建成功",
                timestamp: mesDate,
                id:messageId
            };
            session.last_sender_id = message.sender_id;

            session.last_sender_name = message.sender_name;

            session.last_message_time = mesDate.getTime();

            session.last_content  = message.content;

            session.last_content_type = message.content_type;
            SessionRepo.updateSessionLastStatus(message.sender_id,
                message.sender_name,
                message.timestamp,
                message.content,
                message.content_type,
                sessionId, function (err, res) {
                if (err) {handler(err,null);return;};
                handler(null,session);
            });
        }
    }

    /**
     * 保存session到MySQL
     * @param sessionId
     * @param name
     * @param type
     * @param createDate
     * @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);
    }

    /**
     * 根据用户ID获取用户的session列表
     * @param userId
     * @param page
     * @param size
     */
    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 (sessionIds, callback) {
                let sessionList = [];
                sessionIds.forEach(function (sessionId) {
                    let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
                    let participantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
                    let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
                    let participantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants,sessionId);
                    redis.multi()
                        .hgetall(sessionKey)                       // 会话实体
                        .hget(participantsRoleKey, userId)         // 用户在此会话中的角色
                        .zscore(sessionParticipantsKey, userId)    // 用户在此会话中最后一次获取未读消息的时间
                        .zrange(participantsKey,0,-1)
                        .execAsync()
                        .then(function (res) {
                            let session = res[0];
                            let role = res[1];
                            let lastFetchTime = res[2];
                            let users = res[3];
                            let sessionName="";
                            let otheruserId ="";
                            if(session.type==SESSION_TYPES.P2P){
                                for(var 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, lastFetchTime, 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;
                                            }
                                            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.sender_id,
                                                type: session.type,
                                                sender_name: session.sender_name,
                                                unread_count: count,
                                                business_type: session.business_type,
                                                my_role: role
                                            });
                                            if (sessionId === sessionIds[sessionIds.length - 1]) {
                                                ModelUtil.emitOK(self.eventEmitter, sessionList);
                                            }
                                        })
                                    })
                                }).catch(function (err) {
                                    logger.error("Get sessions failed: ", ex);
                                    ModelUtil.emitError(self.eventEmitter, "Get sessions failed: " + err);
                                });
                });
            }
        ]);
    }




    /**
     * 获取会话消息。全部，不管已读/未读状态。
     *
     * @param sessionId 会话ID
     * @param userId 拉取消息的人
     * @param page 第几页
     * @param pagesize 分页数量
     * @param start_msg_id 消息会话最新的一条消息的ID
     * @param end_msg_id 消息会话刚开始的消息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) {
                if (res.length == 0) {
                    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.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.length == 0) {
                    if(handler){
                        handler(null,res);
                        return;
                    }
                    ModelUtil.emitOK(self.eventEmitter, res);
                    return;
                }
                start_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 (!end_msg_id) {
            redis.zrangeAsync(message_timestamp_key, 0, 0).then(function (res) {
                if (res.length == 0) {
                    ModelUtil.emitOK(self.eventEmitter, res);
                    return;
                }
                end_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 {
            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 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) {
            if (!res) {
                handler(Error("User not found in session " + sessionId), null);
            } else {
                //将消息ID转换成分值
                redis.multi()
                    .zscore(messagesTimestampKey, startMsgId)
                    .zscore(messagesTimestampKey, endMsgId)
                    .execAsync()
                    .then(function (res) {
                        let startMsgScore = res[1];
                        let endMsgScore = res[0];

                        if (startMsgScore == null || endMsgScore == null || startMsgScore == endMsgScore) {
                            handler(null, []);
                            return;
                        }

                        // 从消息时间表中过滤出要获取的消息ID列表，倒序取出消息
                        redis.zrevrangebyscoreAsync(messagesTimestampKey, startMsgScore, endMsgScore, "limit", offset, count)
                            .then(function (res) {
                                if (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());
                                })
                            })
                            .catch(function (err) {
                                logger.error("Get message by page failed: ", err);

                                handler(err, false);
                            })
                    })
            }
        })
    }

    /**
     * 获取所有会话的未读消息数。
     */
    getAllSessionsUnreadMessageCount(userId) {
        let self = this;
        let count = 0;
        SessionRepo.findAll(userId,function(err,res){
            if(err){
                logger.err("getAllSessionsUnreadMessageCount is fail :"+err);
                return;
            }
            if(res.length==0){
                ModelUtil.emitOK(self.eventEmitter, {count: count});
                return;
            }
            for(var j in res){
                if(res[j].type==SESSION_TYPES.SYSTEM){
                    if(j==res.length-1){
                        ModelUtil.emitOK(self.eventEmitter, {count: count});
                    }
                    continue;
                }
                callback(res,j,res[j]);
            }
        })

        function callback(res,j,session){
            self.getSessionUnreadMessageCount(res[j].id,userId,function(err,con){
                if(err){
                    logger.err("getAllSessionsUnreadMessageCount is fail :"+err);
                }
                count = count+con;
                if(j==res.length-1){
                    ModelUtil.emitOK(self.eventEmitter, {count: count});
                }
            })
        }
    }

    /**
     * 获取会话的未读消息数。根据成员最后一次获取消息的时候与当前时间。
     *
     * @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, lastFetchTime, 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();
        message.id = messageId;
        // 检查会话中是否存在此成员
        participants.existsParticipant(sessionId, message.sender_id, function (err, res) {
            if (err) {
                ModelUtil.emitError(self.eventEmitter, "Check session paticipant failed: ", err);
                return;
            }

            if (res) {
                redis.hmgetAsync(sessionKey, ["type", "name"]).then(function (res) {
                    let sessionType = res[0];
                    let sessionName = res[1];
                    if (sessionType == null) {
                        ModelUtil.emitError(self.eventEmitter, "Session with id " + sessionId + " not found.");
                        return;
                    }
                    //消息保存到REDIS
                    messages.saveMessageToRedis(sessionId, sessionType, messageId, message);
                    //更新REDIS最后一条消息
                    Messages.updateLastContent(sessionKey, sessionType, sessionName, message);
                    //更新用户最后一次获取消息的时间
                    Sessions.updateParticipantLastFetchTime(sessionId, message.sender_id, message.timestamp.getTime());
                    //更新最后一条消息到数据库
                    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) {
                                    Sessions.pushNotification(participant.id, message);
                                }
                            });
                        }
                    })
                }).catch(function (err) {
                    ModelUtil.emitError(self.eventEmitter, {message: "Error occurred while save message to session: " + err});
                })
            } else {
                ModelUtil.emitDataNotFound(self.eventEmitter, {message: "当前会话找不到此发送者"});
            }
        });
    }

    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 = "";
        message.id = messageId;
        // 发送成员必须处于会话中
        participants.existsParticipant(sessionId, message.sender_id, function (err, res) {
            if (res) {
                redis.hmgetAsync(session_key, ["type", "name"]).then(function (res) {
                    sessionType = res[0];
                    sessionName = res[1];
                    if (!sessionType || !sessionName) {
                        logger.error("session is error for key " + session_key);
                        throw "session is not found";
                    }
                }).then(function (res) {
                    // 更新消息存储REDIS
                    messages.saveMessageToRedis(sessionId, sessionType, messageId, message);
                    //更新消息存储到mysql
                    messages.saveMessageToMysql(sessionId, sessionType, messageId, message);
                    // 更新会话最新状态及成员最后一次消息获取时间
                    Sessions.updateParticipantLastFetchTime(sessionId, message.sender_id, message.timestamp.getTime());
                    //更新最后一条消息
                    Messages.updateLastContent(session_key, sessionType, sessionName, message);
                    //更新session实体的最后一条消息
                    SessionRepo.updateSessionLastStatus(message.sender_id,message.sender_name,message.timestamp,message.content,message.content_type,sessionId);

                    handler(null, messageId);
                }).then(function (res) {
                    // 推送消息
                    ParticipantRepo.findIds(sessionId, function (err, res) {
                        if (err) {
                            handler(err, messageId)
                        } else {
                            message.session_id = sessionId;
                            res.forEach(function (participant) {
                                if (participant.id !== message.sender_id) {
                                    Sessions.pushNotification(participant.id, message);
                                }
                            });
                        }
                    })
                }).catch(function (err) {
                    handler(err, messageId)
                })
            } else {
                handler("用户不在此会话当中！", messageId);
            }
        })
    }

    /**
     * 置顶操作
     */
    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.unstickSession(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, message) {
        Users.isPatientId(targetUserId, function (err, isPatient) {
            if (isPatient) {
                WechatClient.sendMessage(targetUserId, message);
            }
            else {
                AppClient.sendNotification(targetUserId, message);
            }
        });
    }

}

// Expose class
module.exports = Sessions;