sessions.js 69 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542
  1. /**
  2. * 会话模型。
  3. */
  4. "use strict";
  5. let RedisClient = require('../../repository/redis/redis.client.js');
  6. let RedisModel = require('./../redis.model.js');
  7. let ModelUtil = require('../../util/model.util');
  8. let Messages = require('../messages/messages');
  9. let Users = require('../user/users');
  10. let Participants = require('./participants');
  11. let SessionRepo = require('../../repository/mysql/session.repo');
  12. let TopicRepo = require('../../repository/mysql/topics.repo');
  13. let MessageRepo = require('../../repository/mysql/message.repo');
  14. let ParticipantRepo = require('../../repository/mysql/participant.repo');
  15. let ImDb = require('../../repository/mysql/db/im.db');
  16. let WlyySDK = require("../../util/wlyy.sdk");
  17. let WechatClient = require("../client/wechat.client.js");
  18. let AppClient = require("../client/app.client.js");
  19. let configFile = require('../../include/commons').CONFIG_FILE;
  20. let config = require('../../resources/config/' + configFile);
  21. let redis = RedisClient.redisClient().connection;
  22. let logger = require('../../util/log.js');
  23. let mongoose = require('mongoose');
  24. let async = require("async");
  25. let log = require("../../util/log.js");
  26. let pubSub = require("../redis/pubSub.js");
  27. const REDIS_KEYS = require('../../include/commons').REDIS_KEYS;
  28. const SESSION_TYPES = require('../../include/commons').SESSION_TYPES;
  29. const STICKY_SESSION_BASE_SCORE = require('../../include/commons').STICKY_SESSION_BASE_SCORE;
  30. const PARTICIPANT_ROLES = require('../../include/commons').PARTICIPANT_ROLES;
  31. const CONTENT_TYPES = require('../../include/commons').CONTENT_TYPES;
  32. const SESSION_BUSINESS_TYPE = require('../../include/commons').SESSION_BUSINESS_TYPE;
  33. const SESSION_STATUS = require('../../include/commons').SESSION_STATUS;
  34. class Sessions extends RedisModel {
  35. constructor() {
  36. super();
  37. }
  38. /**
  39. * 创建会话。会话的ID来源:
  40. * MUC:患者的ID
  41. * P2P:对成员的ID排序后,取hash值
  42. * GROUP:团队的ID
  43. *
  44. * @param sessionId
  45. * @param name 会话名称
  46. * @param type 会话类型
  47. * @param participantArray 会话成员
  48. * @param handler 回调,仅MUC模式使用
  49. */
  50. createSession(sessionId, name, type, participantArray, handler) {
  51. let self = this;
  52. let messageId = mongoose.Types.ObjectId().toString();
  53. //创建session到mysql
  54. self.createSessionToMysql(sessionId, name, type, participantArray, messageId, function (err, res) {
  55. if (err) {
  56. if (handler) {
  57. handler(err, null);
  58. return;
  59. }
  60. ModelUtil.emitError(self.eventEmitter, {message: err, status: -1}, null);
  61. } else {
  62. //创建session到redis
  63. self.createSessionToRedis(sessionId, name, type, participantArray, messageId, function (err, res) {
  64. if (err) {
  65. if (handler) {
  66. handler(err, null);
  67. return;
  68. }
  69. ModelUtil.emitError(self.eventEmitter, {message: err, status: -1}, null);
  70. } else {
  71. if (handler) {
  72. handler(null, res);
  73. return;
  74. }
  75. ModelUtil.emitOK(self.eventEmitter, {status: 200, data: res});
  76. }
  77. });
  78. }
  79. });
  80. }
  81. /**
  82. * 创建会话。REDIS
  83. * @param sessionId
  84. * @param name
  85. * @param type
  86. * @param participantArray
  87. * @param messageId
  88. * @param handler
  89. * @returns {boolean}
  90. */
  91. createSessionToRedis(sessionId, name, type, participantArray, messageId, handler) {
  92. let self = this;
  93. let messages = new Messages();
  94. let participantIdArray = [];
  95. for (let i in participantArray) {
  96. participantIdArray.push(participantArray[i].split(":")[0]);
  97. }
  98. if (type == SESSION_TYPES.P2P || type == SESSION_TYPES.SYSTEM) {
  99. if (sessionId) {
  100. callBusinessType(sessionId);
  101. return;
  102. }
  103. else if (participantIdArray.length != 2) {
  104. handler("P2P session only allow 2 participants.", null);
  105. return false;
  106. }else{
  107. ParticipantRepo.findSessionIdByParticipantIds(participantIdArray[0], participantIdArray[1], function (err, res) {
  108. sessionId = res;
  109. callBusinessType(sessionId);
  110. return;
  111. });
  112. }
  113. } else {
  114. if (!sessionId) {
  115. handler("MUC OR GROUP session sessionId is not allow null .", null);
  116. return;
  117. }
  118. callBusinessType(sessionId);
  119. }
  120. function callBusinessType(sessionId) {
  121. if(type == SESSION_TYPES.MUC||type == SESSION_TYPES.PRESCRIPTION){
  122. callCreate(sessionId, SESSION_BUSINESS_TYPE.PATIENT);
  123. }else {
  124. ParticipantRepo.getBusinessType(participantIdArray, function (err, businessType) {
  125. callCreate(sessionId, businessType);
  126. });
  127. }
  128. }
  129. function callCreate(sessionId, businessType) {
  130. let createDate = new Date();
  131. Participants.saveParticipantsToRedis(sessionId, participantArray, createDate, function (res) {
  132. let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
  133. if (type == SESSION_TYPES.MUC||type == SESSION_TYPES.PRESCRIPTION) {
  134. businessType = 2;
  135. }
  136. let session = {
  137. id: sessionId,
  138. name: name,
  139. type: type,
  140. create_date: createDate.getTime(),
  141. business_type: businessType
  142. };
  143. //如果会话已经存在的就不需要发送会话成功的消息不更新最后一条消息
  144. redis.hexistsAsync(sessionKey, sessionId).then(function(res){
  145. if(res==0){
  146. redis.hmsetAsync(sessionKey, session).then(function () {
  147. handler(null, session);
  148. })
  149. }
  150. })
  151. });
  152. }
  153. }
  154. /**
  155. * 创建会话。mysql
  156. * @param sessionId
  157. * @param name
  158. * @param type
  159. * @param participantArray
  160. * @param messageId
  161. * @param handler
  162. */
  163. createSessionToMysql(sessionId, name, type, participantArray, messageId, handler) {
  164. let self = this;
  165. //如果sessionId不存在则执行创建sessionId过程
  166. let participantIdArray = [];
  167. for (let i in participantArray) {
  168. participantIdArray.push(participantArray[i].split(":")[0]);
  169. }
  170. //流程1-判断是否存在sessionId不存在则创建对应的sessionId;
  171. if (!sessionId) {
  172. if (type == SESSION_TYPES.P2P || type == SESSION_TYPES.SYSTEM) {
  173. ParticipantRepo.findSessionIdByParticipantIds(participantIdArray[0], participantIdArray[1], function (err, res) {
  174. sessionId = res;
  175. callBusinessType();
  176. });
  177. } else {
  178. return handler("MUC模式和团队模式,不允许sessionId为空!", null);
  179. }
  180. } else {
  181. callBusinessType();
  182. }
  183. //流程2-判断session的业务类型;
  184. function callBusinessType() {
  185. if(type==SESSION_TYPES.MUC||type==SESSION_TYPES.PRESCRIPTION){
  186. callCreateSession(SESSION_BUSINESS_TYPE.PATIENT);
  187. }else{
  188. ParticipantRepo.getBusinessType(participantIdArray, function (err, businessType) {
  189. if (err) {
  190. handler(err, null);
  191. return;
  192. }
  193. callCreateSession(businessType);
  194. });
  195. }
  196. }
  197. //流程3-发起session创建 返回session实例
  198. function callCreateSession(businessType) {
  199. //查找该sessionId是否存在存在则直接返回实例
  200. SessionRepo.findOne(sessionId, function (err, res) {
  201. if (res.length > 0) {//已经存在
  202. //更新成员
  203. Participants.saveParticipantsToMysql(sessionId, participantArray, function (err, update) {
  204. handler(null, res[0]);
  205. return;
  206. })
  207. let createDate = new Date();
  208. self.saveSessionToMysql(sessionId, name, type, createDate, businessType, function (err, res) {
  209. logger.info("update session status is true");
  210. })
  211. } else {
  212. let createDate = new Date();
  213. let session = {
  214. id: sessionId,
  215. name: name,
  216. type: type,
  217. create_date: createDate.getTime(),
  218. business_type: businessType
  219. };
  220. //将session写入数据库
  221. self.saveSessionToMysql(sessionId, name, type, createDate, businessType, function (err, res) {
  222. if (err) {
  223. handler(err, null);
  224. return;
  225. }
  226. callCreateParticipants(session);
  227. })
  228. }
  229. });
  230. }
  231. //流程4-发起session成员创建
  232. function callCreateParticipants(session) {
  233. Participants.saveParticipantsToMysql(sessionId, participantArray, function (err, res) {
  234. if (err) {
  235. handler(err, null);
  236. return;
  237. } else {
  238. handler(null, session);
  239. return;
  240. }
  241. })
  242. }
  243. }
  244. /**
  245. * 最近会话列表,7天内。
  246. *
  247. * @param userId
  248. * @param dateSpan
  249. */
  250. getRecentSessions(userId, dateSpan) {
  251. let self = this;
  252. SessionRepo.findAllByTimestampAndType(userId, dateSpan, function (err, res) {
  253. if (err) {
  254. ModelUtil.emitError(self.eventEmitter, "Get recent sessions failed", err);
  255. return;
  256. }
  257. let sessions = [];
  258. res.forEach(function (session) {
  259. //最近列表用于转发,过滤不可用的咨询
  260. if(session.last_content_type != CONTENT_TYPES.TopicEnd){
  261. sessions.push({
  262. id: session.id,
  263. name: session.name,
  264. type: session.type,
  265. business_type: session.business_type,
  266. create_date: session.create_date
  267. })
  268. }
  269. });
  270. ModelUtil.emitOK(self.eventEmitter, sessions);
  271. });
  272. }
  273. /**
  274. * 保存session到MySQL
  275. * @param sessionId
  276. * @param name
  277. * @param type
  278. * @param createDate
  279. * @param businessType
  280. * @param handler
  281. */
  282. saveSessionToMysql(sessionId, name, type, createDate, businessType, handler) {
  283. SessionRepo.saveSession(sessionId, name, type, createDate, businessType, handler);
  284. }
  285. /**
  286. * 获取某个用户的全部session列表
  287. * @param userId
  288. * @param handler
  289. */
  290. getUserSessionsFromMysql(userId, handler) {
  291. SessionRepo.findAll(userId, handler);
  292. }
  293. /**
  294. * 获取session单个对象
  295. * @param sessionId
  296. * @param handler
  297. */
  298. getSessions(sessionId, handler) {
  299. SessionRepo.findOne(sessionId, handler);
  300. }
  301. /**
  302. * 判断会话是否存在
  303. * @param sessionId
  304. * @param handler
  305. */
  306. isExist(sessionId) {
  307. let self = this;
  308. SessionRepo.findOne(sessionId, function (err, res) {
  309. if(err){
  310. log.error(err);
  311. ModelUtil.emitError(self.eventEmitter, {message: err, status: -1}, null);
  312. } else if(res&&res.length>0){
  313. ModelUtil.emitOK(self.eventEmitter, {sessionId: res[0].id, status: 200});
  314. }else {
  315. ModelUtil.emitOK(self.eventEmitter, {sessionId: '', status: 200});
  316. }
  317. });
  318. }
  319. getSession(sessionId,userId,handler){
  320. let self = this;
  321. let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
  322. redis.hgetallAsync(sessionKey).then(function(session){
  323. if(session.type==SESSION_TYPES.P2P){
  324. ParticipantRepo.findNameById(userId, function (err, res) {
  325. session.name = res[0].name;
  326. if(handler){
  327. handler(null,session);
  328. return;
  329. }
  330. ModelUtil.emitOK(self.eventEmitter, session);
  331. })
  332. }else{
  333. if(handler){
  334. handler(null,session);
  335. return;
  336. }
  337. ModelUtil.emitOK(self.eventEmitter, session);
  338. }
  339. }).catch(function (err) {
  340. logger.error("Get session failed: ", err);
  341. if(handler){
  342. handler(null,session);
  343. return;
  344. }
  345. ModelUtil.emitError(self.eventEmitter, {message: err, status: -1}, null);
  346. })
  347. }
  348. /**
  349. * 根据用户ID获取用户的session列表
  350. * @param userId
  351. * @param page
  352. * @param size
  353. * @param businessType
  354. */
  355. getUserSessions(userId, page, size, businessType) {
  356. let userSessionKey = RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, userId);
  357. let self = this;
  358. if (page > 0) {
  359. if (page == 1) {
  360. page = 0;
  361. }
  362. page = page + page * size;
  363. }
  364. async.waterfall([
  365. // 获取会话ID列表
  366. function (callback) {
  367. redis.zrevrangeAsync(userSessionKey, page, size)
  368. .then(function (sessionIds) {
  369. if (sessionIds.length == 0) {
  370. ModelUtil.emitOK(self.eventEmitter, []);
  371. return;
  372. }
  373. callback(null, sessionIds);
  374. })
  375. },
  376. // 遍历会话
  377. function (sessionIds) {
  378. let sessionList = [];
  379. let functionList = [];
  380. for (let j = 0; j < sessionIds.length; j++) {
  381. let fun = function (index, callback) {
  382. if (!callback) {
  383. callback = index, index = 0
  384. }
  385. let sessionId = sessionIds[index];
  386. let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
  387. let participantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
  388. let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
  389. redis.multi()
  390. .hgetall(sessionKey) // 会话实体
  391. .hget(participantsRoleKey, userId) // 用户在此会话中的角色
  392. .zscore(sessionParticipantsKey, userId) // 用户在此会话中最后一次获取未读消息的时间
  393. .zrange(sessionParticipantsKey, 0, -1)
  394. .zrange(sessionParticipantsKey, 0,-1,'withscores') // 所有用户在此会话中最后一次获取未读消息的时间
  395. .hgetall(participantsRoleKey) // 所有用户在此会话中角色
  396. .execAsync()
  397. .then(function (res) {
  398. let session = res[0];
  399. let role = res[1];
  400. let lastFetchTime = res[2];
  401. let users = res[3];
  402. let participantsTimeArray = res[4];
  403. let userRoles = res[5];
  404. let participantsTime = [];
  405. let isInvite = true;
  406. for(var j in userRoles){
  407. if(userRoles[j]==1){
  408. isInvite = false;
  409. break;
  410. }
  411. }
  412. for(var j = 0 ;j<participantsTimeArray.length;j++){
  413. if(j%2!=0)continue;
  414. let participantsTimeJson = {};
  415. participantsTimeJson[participantsTimeArray[j]] = participantsTimeArray[j+1];
  416. participantsTime.push(participantsTimeJson);
  417. }
  418. let sessionName = "";
  419. let otherUserId = "";
  420. if (session.type == SESSION_TYPES.P2P) {
  421. for (let j in users) {
  422. if (users[j] != userId) {
  423. otherUserId = users[j];
  424. }
  425. }
  426. }
  427. if (!role) role = 0;
  428. if (!lastFetchTime) lastFetchTime = new Date().getTime();
  429. // 计算未读消息数
  430. let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
  431. redis.zcountAsync(messagesByTimestampKey, parseInt(lastFetchTime)+1, new Date().getTime())
  432. .then(function (count) {
  433. if (!otherUserId) otherUserId = userId;
  434. ParticipantRepo.findNameById(otherUserId, function (err, res) {
  435. if ((res && res.length == 0) || session.type != SESSION_TYPES.P2P) {
  436. sessionName = session.name;
  437. } else {
  438. sessionName = res[0].name;
  439. }
  440. var bir = new Date().getTime();
  441. if (res.length != 0 && res[0].birthdate) {
  442. bir = res[0].birthdate.getTime();
  443. }
  444. var sex = 1;
  445. if (res.length != 0 && res[0].sex) {
  446. sex = res[0].sex;
  447. }
  448. //end
  449. sessionList.push({
  450. id: sessionId,
  451. name: sessionName,
  452. create_date: session.create_date,
  453. last_content_type: session.last_content_type||"",
  454. last_content: session.last_content||"",
  455. sender_id: session.last_sender_id||"",
  456. type: session.type,
  457. sender_name: session.last_sender_name||"",
  458. unread_count: count,
  459. business_type: session.business_type,
  460. my_role: role,
  461. sender_sex: sex,
  462. sender_birthday: bir,
  463. participantsTimeArray:participantsTime,
  464. status:session.status,
  465. is_invite:isInvite
  466. });
  467. index = (parseInt(index) + 1);
  468. if (index == sessionIds.length) {
  469. ModelUtil.emitOK(self.eventEmitter, sessionList);
  470. } else {
  471. callback(null, index);
  472. }
  473. })
  474. })
  475. })
  476. .catch(function (err) {
  477. logger.error("Get sessions failed: ", err);
  478. });
  479. };
  480. functionList.push(fun);
  481. }
  482. async.waterfall(functionList);
  483. }
  484. ]);
  485. }
  486. /**
  487. * 根据用户ID获取用户已经结束咨询的session列表
  488. * @param userId
  489. * @param page
  490. * @param size
  491. * @param businessType
  492. */
  493. getUserStatusSessions(userId,status,businessType,page, size) {
  494. let userSessionKey = RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, userId);
  495. let self = this;
  496. async.waterfall([
  497. // 获取会话ID列表
  498. function (callback) {
  499. SessionRepo.findAllByTypeAndStatus(userId,businessType,status,page,size,function(err,res){
  500. if (res.length == 0) {
  501. ModelUtil.emitOK(self.eventEmitter, []);
  502. return;
  503. }
  504. var sessionIds=[];
  505. for(var j in res){
  506. sessionIds.push(res[j].id);
  507. }
  508. callback(null,sessionIds);
  509. })
  510. },
  511. // 遍历会话
  512. function (sessionIds) {
  513. let sessionList = [];
  514. let functionList = [];
  515. for (let j = 0; j < sessionIds.length; j++) {
  516. let fun = function (index, callback) {
  517. if (!callback) {
  518. callback = index, index = 0
  519. }
  520. let sessionId = sessionIds[index];
  521. let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
  522. let participantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
  523. let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
  524. redis.multi()
  525. .hgetall(sessionKey) // 会话实体
  526. .hget(participantsRoleKey, userId) // 用户在此会话中的角色
  527. .zscore(sessionParticipantsKey, userId) // 用户在此会话中最后一次获取未读消息的时间
  528. .zrange(sessionParticipantsKey, 0, -1)
  529. .zrange(sessionParticipantsKey, 0,-1,'withscores') // 所有用户在此会话中最后一次获取未读消息的时间
  530. .hgetall(participantsRoleKey) // 所有用户在此会话中角色
  531. .execAsync()
  532. .then(function (res) {
  533. let session = res[0];
  534. let role = res[1];
  535. let lastFetchTime = res[2];
  536. let users = res[3];
  537. let participantsTimeArray = res[4];
  538. let userRoles = res[5];
  539. let participantsTime = [];
  540. let isInvite = true;
  541. for(var j in userRoles){
  542. if(userRoles[j]==1){
  543. isInvite = false;
  544. break;
  545. }
  546. }
  547. for(var j = 0 ;j<participantsTimeArray.length;j++){
  548. if(j%2!=0)continue;
  549. let participantsTimeJson = {};
  550. participantsTimeJson[participantsTimeArray[j]] = participantsTimeArray[j+1];
  551. participantsTime.push(participantsTimeJson);
  552. }
  553. let sessionName = "";
  554. let otherUserId = "";
  555. if (session.type == SESSION_TYPES.P2P) {
  556. for (let j in users) {
  557. if (users[j] != userId) {
  558. otherUserId = users[j];
  559. }
  560. }
  561. }
  562. if (!role) role = 0;
  563. if (!lastFetchTime) lastFetchTime = new Date().getTime();
  564. // 计算未读消息数
  565. let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
  566. redis.zcountAsync(messagesByTimestampKey, parseInt(lastFetchTime)+1, new Date().getTime())
  567. .then(function (count) {
  568. if (!otherUserId) otherUserId = userId;
  569. ParticipantRepo.findNameById(otherUserId, function (err, res) {
  570. if ((res && res.length == 0) || session.type != SESSION_TYPES.P2P) {
  571. sessionName = session.name;
  572. } else {
  573. sessionName = res[0].name;
  574. }
  575. var bir = new Date().getTime();
  576. if (res.length != 0 && res[0].birthdate) {
  577. bir = res[0].birthdate.getTime();
  578. }
  579. var sex = 1;
  580. if (res.length != 0 && res[0].sex) {
  581. sex = res[0].sex;
  582. }
  583. sessionList.push({
  584. id: sessionId,
  585. name: sessionName,
  586. create_date: session.create_date,
  587. last_content_type: session.last_content_type,
  588. last_content: session.last_content,
  589. sender_id: session.last_sender_id,
  590. type: session.type,
  591. sender_name: session.last_sender_name,
  592. unread_count: count,
  593. business_type: session.business_type,
  594. my_role: role,
  595. sender_sex: sex,
  596. sender_birthday: bir,
  597. participantsTimeArray:participantsTime,
  598. status:session.status,
  599. is_invite:isInvite
  600. });
  601. index = (parseInt(index) + 1);
  602. if (index == sessionIds.length) {
  603. ModelUtil.emitOK(self.eventEmitter, sessionList);
  604. } else {
  605. callback(null, index);
  606. }
  607. })
  608. })
  609. })
  610. .catch(function (err) {
  611. logger.error("Get sessions failed: ", err);
  612. });
  613. };
  614. functionList.push(fun);
  615. }
  616. async.waterfall(functionList);
  617. }
  618. ]);
  619. }
  620. /**
  621. * 获取会话消息。全部,不管已读/未读状态。
  622. *
  623. * @param sessionId 会话ID
  624. * @param userId 拉取消息的人
  625. * @param page 第几页
  626. * @param pagesize 分页数量
  627. * @param start_msg_id 消息会话最新的一条消息的ID
  628. * @param end_msg_id 消息会话刚开始的消息ID
  629. * @remark
  630. * start_msg_id end_msg_id 为空取会话倒序的根据分页的消息数据
  631. * start_msg_id 为空 end_msg_id 不为空根据end_msg_id取旧的消息(offset=1可以排除本身)返回指定分页的数据
  632. * end_msg_id 为空 start_msg_id 不为空 根据 start_msg_id 取出新消息(offset=1可以排除本身)返回指定分页的数据
  633. * end_msg_id start_msg_id 都不为空返回指定区间的消息
  634. */
  635. getMessages(sessionId, user, start_msg_id, end_msg_id, page, pagesize, isoffset, handler) {
  636. let self = this;
  637. let message_timestamp_key = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
  638. if (!start_msg_id && !end_msg_id) {
  639. redis.zrevrangeAsync(message_timestamp_key, 0, 0).then(function (res) {
  640. if (res.length == 0) {
  641. if (handler) {
  642. handler(null, res);
  643. return;
  644. }
  645. ModelUtil.emitOK(self.eventEmitter, res);
  646. return;
  647. }
  648. start_msg_id = res[0];
  649. redis.zrangeAsync(message_timestamp_key, 0, 0).then(function (res) {
  650. if (res.length == 0) {
  651. if (handler) {
  652. handler(null, res);
  653. return;
  654. }
  655. ModelUtil.emitOK(self.eventEmitter, res);
  656. return;
  657. }
  658. end_msg_id = res[0];
  659. self.getMessagesByPage(sessionId, user, end_msg_id, start_msg_id, page, pagesize, isoffset, function (err, res) {
  660. if (err) {
  661. if (handler) {
  662. handler(err, null);
  663. return;
  664. }
  665. logger.error("getMessagesByPage error" + err);
  666. ModelUtil.emitError(self.eventEmitter, err, err);
  667. } else {
  668. if (handler) {
  669. handler(null, res);
  670. return;
  671. }
  672. ModelUtil.emitOK(self.eventEmitter, res);
  673. }
  674. })
  675. })
  676. })
  677. } else if (!start_msg_id) {
  678. redis.zrevrangeAsync(message_timestamp_key, 0, 0).then(function (res) {
  679. if (res.length == 0) {
  680. if (handler) {
  681. handler(null, res);
  682. return;
  683. }
  684. ModelUtil.emitOK(self.eventEmitter, res);
  685. return;
  686. }
  687. start_msg_id = res[0];
  688. self.getMessagesByPage(sessionId, user, start_msg_id,end_msg_id , page, pagesize, isoffset, function (err, res) {
  689. if (err) {
  690. if (handler) {
  691. handler(err, null);
  692. return;
  693. }
  694. logger.error("getMessagesByPage error" + err);
  695. ModelUtil.emitError(self.eventEmitter, err, err);
  696. } else {
  697. if (handler) {
  698. handler(null, res);
  699. return;
  700. }
  701. ModelUtil.emitOK(self.eventEmitter, res);
  702. }
  703. })
  704. })
  705. } else if (!end_msg_id) {
  706. redis.zrangeAsync(message_timestamp_key, 0, 0).then(function (res) {
  707. if (res.length == 0) {
  708. ModelUtil.emitOK(self.eventEmitter, res);
  709. return;
  710. }
  711. end_msg_id = res[0];
  712. self.getMessagesByPage(sessionId, user, end_msg_id,start_msg_id, page, pagesize, isoffset, function (err, res) {
  713. if (err) {
  714. if (handler) {
  715. handler(err, null);
  716. return;
  717. }
  718. logger.error("getMessagesByPage error" + err);
  719. ModelUtil.emitError(self.eventEmitter, err, err);
  720. } else {
  721. if (handler) {
  722. handler(null, res);
  723. return;
  724. }
  725. ModelUtil.emitOK(self.eventEmitter, res);
  726. }
  727. })
  728. })
  729. } else {
  730. self.getMessagesByPage(sessionId, user, end_msg_id, start_msg_id, page, pagesize, isoffset, function (err, res) {
  731. if (err) {
  732. if (handler) {
  733. handler(err, null);
  734. return;
  735. }
  736. logger.error("getMessagesByPage error" + err);
  737. ModelUtil.emitError(self.eventEmitter, err, err);
  738. } else {
  739. if (handler) {
  740. handler(null, res);
  741. return;
  742. }
  743. ModelUtil.emitOK(self.eventEmitter, res);
  744. }
  745. })
  746. }
  747. }
  748. /**
  749. * 分页获取会话消息。
  750. *
  751. * @param sessionId 必选。会话ID
  752. * @param userId 必选。用户ID
  753. * @param startMsgId 必选。会话的的起始消息ID,作为检索的起始依据
  754. * @param endMsgId 必选。会话中的结束消息ID
  755. * @param page 必选。页码
  756. * @param size 必选。页面大小
  757. * @param handler 必选。回调
  758. */
  759. getMessagesByPage(sessionId, userId, startMsgId, endMsgId, page, size, isoffset, handler) {
  760. let messagesTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
  761. let messagesKey = RedisModel.makeRedisKey(REDIS_KEYS.Messages, sessionId);
  762. let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
  763. let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
  764. let participants = new Participants();
  765. let offset = (page - 1 < 0 ? 0 : page - 1) * size;
  766. let count = size;
  767. if (page > 1 || isoffset == 1) {
  768. offset += 1; // 翻页由于闭区间,需跳过本身数据
  769. }
  770. participants.existsParticipant(sessionId, userId, function (err, res) {
  771. if (!res) {
  772. handler(Error("User not found in session " + sessionId), null);
  773. } else {
  774. //将消息ID转换成分值
  775. redis.multi()
  776. .zscore(messagesTimestampKey, startMsgId)
  777. .zscore(messagesTimestampKey, endMsgId)
  778. .hgetall(sessionKey)
  779. .zrange(sessionParticipantsKey, 0, -1)
  780. .execAsync()
  781. .then(function (res) {
  782. let startMsgScore = res[1];
  783. let endMsgScore = res[0];
  784. let session = res[2];
  785. let users = res[3];
  786. if (startMsgScore == null || endMsgScore == null || (startMsgScore == endMsgScore && isoffset == 1)) {
  787. handler(null, []);
  788. return;
  789. }
  790. //结束大于开始,正序取数据,返回的数据顺序也是逆序的,反向拉取数据,
  791. // 当end>start取出来都是空的,为了给前端获取新数据使用,一般不出现这种情况
  792. if(endMsgScore>startMsgScore){
  793. redis.zrangebyscoreAsync(messagesTimestampKey, startMsgScore, endMsgScore, "limit", offset, count)
  794. .then(function (res) {
  795. if (res.length == 0) {
  796. handler(null, []);
  797. return;
  798. }
  799. redis.hmgetAsync(messagesKey, res).then(function (messages) {
  800. messages.reverse();
  801. handler(null, messages);
  802. }).then(function () {
  803. Sessions.updateParticipantLastFetchTime(sessionId, userId, new Date().getTime());
  804. if(session.type == SESSION_TYPES.P2P){
  805. for(var j in users){
  806. if(users[j]==userId)continue;
  807. WechatClient.sendAllRead(users[j],sessionId);
  808. }
  809. }else if(session.type == SESSION_TYPES.MUC || session.type == SESSION_TYPES.PRESCRIPTION){
  810. for(var j in users){
  811. if(users[j]==userId)continue;
  812. WechatClient.sendMucAllRead(users[j],userId,sessionId);
  813. }
  814. }
  815. })
  816. })
  817. .catch(function (err) {
  818. logger.error("Get message by page failed: ", err);
  819. handler(err, false);
  820. })
  821. }else{
  822. // 从消息时间表中过滤出要获取的消息ID列表,倒序取出消息
  823. redis.zrevrangebyscoreAsync(messagesTimestampKey, startMsgScore, endMsgScore, "limit", offset, count)
  824. .then(function (res) {
  825. if (res.length == 0) {
  826. handler(null, []);
  827. return;
  828. }
  829. redis.hmgetAsync(messagesKey, res).then(function (messages) {
  830. handler(null, messages);
  831. }).then(function () {
  832. Sessions.updateParticipantLastFetchTime(sessionId, userId, new Date().getTime());
  833. if(session.type == SESSION_TYPES.P2P){
  834. for(var j in users){
  835. if(users[j]==userId)continue;
  836. //通知对方自己已经读取数据
  837. WechatClient.sendAllRead(users[j],sessionId);
  838. }
  839. }else if(session.type == SESSION_TYPES.MUC || session.type == SESSION_TYPES.PRESCRIPTION){
  840. for(var j in users){
  841. if(users[j]==userId)continue;
  842. //如果是患者拉取数据告诉在线的医生患者已经读取数据
  843. WechatClient.sendMucAllRead(users[j],userId,sessionId);
  844. }
  845. }
  846. })
  847. })
  848. .catch(function (err) {
  849. logger.error("Get message by page failed: ", err);
  850. handler(err, false);
  851. })
  852. }
  853. })
  854. }
  855. })
  856. }
  857. /**
  858. * 获取所有会话的未读消息数。
  859. */
  860. getAllSessionsUnreadMessageCount(userId,handler) {
  861. let self = this;
  862. let count = 0;
  863. let patientCount = 0;
  864. let doctorCount = 0;
  865. let patientEndCount = 0;
  866. SessionRepo.findAll(userId, function (err, res) {
  867. // SessionRepo.findUnEndAll(userId, function (err, res) {
  868. if (err) {
  869. if(handler)
  870. {
  871. handler(err,res);
  872. return;
  873. }
  874. ModelUtil.logError("getAllSessionsUnreadMessageCount is failed", err);
  875. return;
  876. }
  877. if (res.length == 0) {
  878. if(handler)
  879. {
  880. handler(err,count);
  881. return;
  882. }
  883. ModelUtil.emitOK(self.eventEmitter, {count: count});
  884. return;
  885. }
  886. for (let j in res) {
  887. if (res[j].type == SESSION_TYPES.SYSTEM) {
  888. if (j == res.length - 1) {
  889. if(handler){
  890. handler(err,count);
  891. return;
  892. }
  893. ModelUtil.emitOK(self.eventEmitter, {count: count, patient: patientCount, doctor: doctorCount});
  894. }
  895. continue;
  896. }
  897. callback(res, j, res[j]);
  898. }
  899. });
  900. function callback(res, j, session) {
  901. self.getSessionUnreadMessageCount(res[j].id, userId, function (err, con) {
  902. if (err) {
  903. if(handler)
  904. {
  905. handler(err,count);
  906. return;
  907. }
  908. ModelUtil.logError("getAllSessionsUnreadMessageCount is failed", err);
  909. }
  910. count = count + con;
  911. if (session.business_type == SESSION_BUSINESS_TYPE.PATIENT) {
  912. if(session.status == SESSION_STATUS.ENDED){//新增判断是否咨询结束
  913. patientEndCount = patientEndCount + con;
  914. }else{
  915. patientCount = patientCount + con;
  916. }
  917. } else {
  918. doctorCount = doctorCount + con;
  919. }
  920. if (j == res.length - 1) {
  921. if(handler)
  922. {
  923. handler(err,count)
  924. return;
  925. }
  926. ModelUtil.emitOK(self.eventEmitter, {count: count, patient: patientCount,patientEnd: patientEndCount, doctor: doctorCount});
  927. }
  928. })
  929. }
  930. }
  931. /**
  932. * 获取会话的未读消息数。根据成员最后一次获取消息的时候与当前时间。
  933. *
  934. * @param sessionId
  935. * @param userId
  936. */
  937. getSessionUnreadMessageCount(sessionId, userId, handler) {
  938. let self = this;
  939. let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
  940. let participantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
  941. async.waterfall([
  942. // 此成员最后获取消息的时间
  943. function (callback) {
  944. redis.zscoreAsync(participantsKey, userId)
  945. .then(function (lastFetchTime) {
  946. callback(null, lastFetchTime);
  947. })
  948. },
  949. // 计算最后获取消息的时间之后到现在有多少条消息
  950. function (lastFetchTime, callback) {
  951. if (!lastFetchTime) lastFetchTime = 0;
  952. let now = new Date().getTime();
  953. redis.zcountAsync(messagesByTimestampKey, parseInt(lastFetchTime)+1, now)
  954. .then(function (count) {
  955. if (handler) {
  956. handler(null, count);
  957. } else {
  958. ModelUtil.emitOK(self.eventEmitter, {count: count});
  959. }
  960. })
  961. }
  962. ], function (err, res) {
  963. if (err) {
  964. if (handler) {
  965. handler(err, 0);
  966. } else {
  967. ModelUtil.emitError(self.eventEmitter, "Get session unread message count failed.")
  968. }
  969. }
  970. });
  971. }
  972. /**
  973. * 获取会话未读消息数。根据成员最后一次获取消息的时候与当前时间。
  974. */
  975. getSessionUnreadMessages(sessionId, userId) {
  976. let self = this;
  977. let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
  978. let participantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
  979. async.waterfall([
  980. // 此成员最后获取消息的时间
  981. function (callback) {
  982. redis.zscoreAsync(participantsKey, userId)
  983. .then(function (lastFetchTime) {
  984. callback(null, lastFetchTime);
  985. })
  986. },
  987. // 最后获取消息的时间之后到现在的消息ID列表
  988. function (lastFetchTime, callback) {
  989. if (!lastFetchTime) lastFetchTime = 0;
  990. let now = new Date().getTime();
  991. redis.zrangebyscoreAsync(messagesByTimestampKey, lastFetchTime, now)
  992. .then(function (messageIds) {
  993. callback(null, messageIds);
  994. })
  995. },
  996. // 获取消息
  997. function (messageIds, callback) {
  998. if (messageIds.length == 0) {
  999. ModelUtil.emitOK(self.eventEmitter, []);
  1000. return;
  1001. }
  1002. let startMsgId = messageIds[0];
  1003. let endMsgId = messageIds[messageIds.length - 1];
  1004. self.getMessagesByPage(sessionId, userId, startMsgId, endMsgId, 0, messageIds.length, 0, function (err, res) {
  1005. if (err) {
  1006. ModelUtil.emitError(self.eventEmitter, err.message);
  1007. return;
  1008. }
  1009. ModelUtil.emitOK(self.eventEmitter, res);
  1010. });
  1011. }
  1012. ], function (err, res) {
  1013. if (err) {
  1014. ModelUtil.emitError(self.eventEmitter, "Get session unread message count failed.")
  1015. }
  1016. });
  1017. }
  1018. /**
  1019. * 保存消息。
  1020. *
  1021. * 也可以根据议题保存消息,但最终还是保存到与会话对象。
  1022. *
  1023. * see also: saveMessageByTopic
  1024. *
  1025. * @param message
  1026. * @param sessionId
  1027. */
  1028. saveMessageBySession(sessionId, message) {
  1029. let self = this;
  1030. let messages = new Messages();
  1031. let participants = new Participants();
  1032. let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
  1033. let messageId = mongoose.Types.ObjectId().toString();
  1034. let sessionType =0;
  1035. message.id = messageId;
  1036. // 检查会话中是否存在此成员
  1037. participants.existsParticipant(sessionId, message.sender_id, function (err, res) {
  1038. if (err) {
  1039. ModelUtil.emitError(self.eventEmitter, "Check session participant failed: ", err);
  1040. return;
  1041. }
  1042. if (res) {
  1043. redis.hmgetAsync(sessionKey, ["type", "name"]).then(function (res) {
  1044. sessionType = res[0];
  1045. let sessionName = res[1];
  1046. if (sessionType == null) {
  1047. ModelUtil.emitError(self.eventEmitter, "Session " + sessionId + " is not found.");
  1048. return;
  1049. }
  1050. if(sessionType == SESSION_TYPES.MUC || sessionType == SESSION_TYPES.PRESCRIPTION){
  1051. if(message.content_type == CONTENT_TYPES.PlainText ||
  1052. message.content_type == CONTENT_TYPES.Image ||
  1053. message.content_type == CONTENT_TYPES.Audio||
  1054. message.content_type == CONTENT_TYPES.Video){
  1055. TopicRepo.findLastBySessionId(sessionId,function(err,res){
  1056. if(res&&res.length>0&&res[0].reply==0){
  1057. TopicRepo.replyTopic(message.sender_id,message.id,res[0].id,function(err,res){
  1058. if(err){
  1059. logger.error("update topic reply error");
  1060. }else{
  1061. logger.warn("update topic reply success");
  1062. }
  1063. });
  1064. }
  1065. })
  1066. }
  1067. }
  1068. // 消息保存到Redis,并更新会话最后状态、用户最后消息获取时间
  1069. messages.saveMessageToRedis(sessionId, sessionType, messageId, message);
  1070. Messages.updateLastContent(sessionKey, sessionType, sessionName, message);
  1071. Sessions.updateParticipantLastFetchTime(sessionId, message.sender_id, message.timestamp.getTime());
  1072. // 更新MYSQL中会话的最新状态,并保存消息
  1073. SessionRepo.updateSessionLastStatus(message.sender_id, message.sender_name, message.timestamp, message.content, message.content_type, sessionId);
  1074. messages.saveMessageToMysql(sessionId, sessionType, messageId, message, function (err, res) {
  1075. if (err) {
  1076. ModelUtil.emitError(self.eventEmitter, {message: "Failed to save message to mysql: " + err});
  1077. } else {
  1078. message.timestamp = message.timestamp.getTime();
  1079. ModelUtil.emitOK(self.eventEmitter, {count: 1, messages: [message]});
  1080. }
  1081. });
  1082. }).then(function (res) {
  1083. // 推送消息
  1084. ParticipantRepo.findIds(sessionId, function (err, res) {
  1085. if (err) {
  1086. ModelUtil.logError("Push message from session: get participant's id list failed: ", err);
  1087. } else {
  1088. message.session_id = sessionId;
  1089. res.forEach(function (participant) {
  1090. if (participant.id == message.sender_id||sessionType==SESSION_TYPES.SYSTEM){
  1091. message.sender_img = participant.avatar;
  1092. callPush(res,message);
  1093. }
  1094. })
  1095. }
  1096. })
  1097. }).catch(function (err) {
  1098. ModelUtil.emitError(self.eventEmitter, {message: "Error occurred while save message to session: " + err});
  1099. })
  1100. } else {
  1101. ModelUtil.emitDataNotFound(self.eventEmitter, {message: "当前会话找不到此发送者"});
  1102. }
  1103. });
  1104. function callPush(participants,message){
  1105. participants.forEach(function (participant) {
  1106. if (participant.id !== message.sender_id &&
  1107. participant.participant_role == PARTICIPANT_ROLES.HOST) {
  1108. Sessions.pushNotification(participant.id, participant.name, message,sessionType);
  1109. }
  1110. });
  1111. }
  1112. }
  1113. sendTopicMessages(topicId, message) {
  1114. let self = this;
  1115. TopicRepo.findAllByTopicId(topicId, function (err, res) {
  1116. if (err || res.length == 0) {
  1117. ModelUtil.emitOK(self.eventEmitter, {status: -1, "message": "议题获取失败"});
  1118. return;
  1119. }
  1120. self.saveMessageByTopic(message, res[0].session_id, function (err, messageId) {
  1121. if (err) {
  1122. ModelUtil.emitOK(self.eventEmitter, {status: -1, "message": err});
  1123. } else {
  1124. message.id = messageId;
  1125. ModelUtil.emitOK(self.eventEmitter, {status: 200, "message": "发送成功", data: message});
  1126. }
  1127. });
  1128. });
  1129. }
  1130. /**
  1131. * 保存消息
  1132. *
  1133. * @param message
  1134. * @param sessionId
  1135. * @param handler
  1136. */
  1137. saveMessageByTopic(message, sessionId, handler) {
  1138. let messages = new Messages();
  1139. let participants = new Participants();
  1140. let session_key = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
  1141. let messageId = mongoose.Types.ObjectId().toString();
  1142. let self = this;
  1143. let sessionType = 0;
  1144. let sessionName = "";
  1145. let agent = message.agent;//代理人
  1146. message.id = messageId;
  1147. if(!message.timestamp){
  1148. message.timestamp = new Date();
  1149. }
  1150. // 发送成员必须处于会话中
  1151. participants.existsParticipant(sessionId, message.sender_id, function (err, res) {
  1152. if (res) {
  1153. redis.hmgetAsync(session_key, ["type", "name"]).then(function (res) {
  1154. sessionType = res[0];
  1155. sessionName = res[1];
  1156. if (!sessionType || !sessionName) {
  1157. logger.error("Unknown session key " + session_key);
  1158. if (handler) {
  1159. handler(new Error("Unknown session key " + session_key));return;
  1160. };
  1161. }
  1162. }).then(function (res) {
  1163. // 消息数据双写,并更新用户最后消息获取时间,会话新状态等
  1164. messages.saveMessageToRedis(sessionId, sessionType, messageId, message);
  1165. messages.saveMessageToMysql(sessionId, sessionType, messageId, message);
  1166. // 更新会话最新状态及成员最后一次消息获取时间
  1167. Sessions.updateParticipantLastFetchTime(sessionId, message.sender_id, message.timestamp.getTime());
  1168. Messages.updateLastContent(session_key, sessionType, sessionName, message);
  1169. SessionRepo.updateSessionLastStatus(message.sender_id, message.sender_name, message.timestamp, message.content, message.content_type, sessionId);
  1170. if (handler) {
  1171. handler(null, messageId);
  1172. return;
  1173. }
  1174. }).then(function (res) {
  1175. // 推送消息
  1176. ParticipantRepo.findIds(sessionId, function (err, res) {
  1177. if (err) {
  1178. logger.error(err);
  1179. } else {
  1180. message.session_id = sessionId;
  1181. res.forEach(function (participant) {
  1182. if (participant.id == message.sender_id){
  1183. message.sender_img = participant.avatar;
  1184. callPush(res,message);
  1185. }
  1186. })
  1187. }
  1188. })
  1189. }).catch(function (err) {
  1190. log.error(err);
  1191. return;
  1192. })
  1193. } else {
  1194. if (handler){ handler("用户不在此会话当中!", messageId);return;}
  1195. }
  1196. });
  1197. function callPush(participants,message){
  1198. participants.forEach(function (participant) {
  1199. if (participant.id !== message.sender_id &&
  1200. participant.participant_role == PARTICIPANT_ROLES.HOST) {
  1201. Sessions.pushNotification(participant.id, participant.name, message,sessionType);
  1202. }
  1203. });
  1204. }
  1205. }
  1206. /**
  1207. * 保存代理人进入的消息
  1208. *
  1209. * @param message
  1210. * @param sessionId
  1211. * @param handler
  1212. */
  1213. saveIntoMessageByTopic(message, sessionId, handler) {
  1214. let messages = new Messages();
  1215. let participants = new Participants();
  1216. let session_key = RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId);
  1217. let messageId = mongoose.Types.ObjectId().toString();
  1218. let self = this;
  1219. let sessionType = 0;
  1220. let sessionName = "";
  1221. let agent = message.agent;//代理人
  1222. message.id = messageId;
  1223. if(!message.timestamp){
  1224. message.timestamp = new Date();
  1225. }
  1226. // 发送成员必须处于会话中
  1227. participants.existsParticipant(sessionId, message.sender_id, function (err, res) {
  1228. if (res) {
  1229. redis.hmgetAsync(session_key, ["type", "name"]).then(function (res) {
  1230. sessionType = res[0];
  1231. sessionName = res[1];
  1232. if (!sessionType || !sessionName) {
  1233. logger.error("Unknown session key " + session_key);
  1234. if (handler) {
  1235. handler(new Error("Unknown session key " + session_key));return;
  1236. };
  1237. }
  1238. }).then(function (res) {
  1239. //查找最后一条数据,如果一致就不保存
  1240. MessageRepo.findLastMessage(sessionId,sessionType,function (err, res) {
  1241. if (err) {
  1242. logger.error(err);
  1243. } else {
  1244. res.forEach(function (mes) {
  1245. // if(mes.content==message.content&&message.content_type==mes.content_type){
  1246. if(mes.agent==message.agent){
  1247. handler(null, messageId);
  1248. return;
  1249. }else {
  1250. // 消息数据双写,并更新用户最后消息获取时间,会话新状态等
  1251. messages.saveMessageToRedis(sessionId, sessionType, messageId, message);
  1252. messages.saveMessageToMysql(sessionId, sessionType, messageId, message);
  1253. // 更新会话最新状态及成员最后一次消息获取时间
  1254. Sessions.updateParticipantLastFetchTime(sessionId, message.sender_id, message.timestamp.getTime());
  1255. Messages.updateLastContent(session_key, sessionType, sessionName, message);
  1256. SessionRepo.updateSessionLastStatus(message.sender_id, message.sender_name, message.timestamp, message.content, message.content_type, sessionId);
  1257. if (handler) {
  1258. handler(null, messageId);
  1259. return;
  1260. }
  1261. }
  1262. })
  1263. }
  1264. });
  1265. }).then(function (res) {
  1266. // 推送消息
  1267. ParticipantRepo.findDoctorIds(sessionId, function (err, res) {
  1268. if (err) {
  1269. logger.error(err);
  1270. } else {
  1271. message.session_id = sessionId;
  1272. res.forEach(function (participant) {
  1273. Sessions.pushIntoNotification(participant.id, participant.name, message,sessionType);
  1274. })
  1275. }
  1276. })
  1277. }).catch(function (err) {
  1278. log.error(err);
  1279. return;
  1280. })
  1281. } else {
  1282. if (handler){ handler("用户不在此会话当中!", messageId);return;}
  1283. }
  1284. });
  1285. }
  1286. /**
  1287. * 置顶操作
  1288. */
  1289. stickSession(sessionId, user) {
  1290. let user_session_key = RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, user);
  1291. let self = this;
  1292. //取出最大的session
  1293. redis.zrevrangeAsync(user_session_key, 0, 0).then(function (res) {
  1294. //获取该session的时间搓
  1295. redis.zscoreAsync(user_session_key, res).then(function (scoreres) {
  1296. let nowtime = new Date().getTime();
  1297. //当前时间搓比redis的时间搓更早证明没有置顶过
  1298. if (scoreres <= nowtime) {
  1299. //初始化置顶
  1300. redis.zaddAsync(user_session_key, STICKY_SESSION_BASE_SCORE, sessionId).then(function (res) {
  1301. logger.info("stickSession:" + sessionId + ",res:" + res);
  1302. ModelUtil.emitOK(self.eventEmitter, {});
  1303. }).then(function () {
  1304. SessionRepo.saveStickySession(sessionId, user, STICKY_SESSION_BASE_SCORE);
  1305. })
  1306. } else {
  1307. //已有置顶的数据,取出来加1保存回去
  1308. scoreres = Number(scoreres) + 1;
  1309. redis.zaddAsync(user_session_key, scoreres, sessionId).then(function () {
  1310. logger.info("stickSession:" + sessionId + ",res:" + res);
  1311. ModelUtil.emitOK(self.eventEmitter, {});
  1312. }).then(function () {
  1313. SessionRepo.saveStickySession(sessionId, user, scoreres);
  1314. })
  1315. }
  1316. })
  1317. })
  1318. }
  1319. /**
  1320. * 取消会话置顶
  1321. */
  1322. cancelStickSession(sessionId, user) {
  1323. let user_session_key = RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, user);
  1324. let participants_key = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
  1325. let self = this;
  1326. redis.zscoreAsync(participants_key, user).then(function (res) {
  1327. if (!res) {
  1328. res = new Date().getTime();
  1329. }
  1330. redis.zaddAsync(user_session_key, res, sessionId).then(function (res) {
  1331. logger.info("cancelStickSession:" + sessionId);
  1332. ModelUtil.emitOK(self.eventEmitter, {});
  1333. }).then(function () {
  1334. SessionRepo.unStickySession(sessionId, user);
  1335. });
  1336. })
  1337. }
  1338. /**
  1339. * 更新会话参与者的最后消息获取时间。
  1340. *
  1341. * @param sessionId
  1342. * @param userId
  1343. */
  1344. static updateParticipantLastFetchTime(sessionId, userId, score) {
  1345. score = score + 1;
  1346. let participantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
  1347. redis.zaddAsync(participantsKey, score, userId)
  1348. .then(function (res) {
  1349. ParticipantRepo.updateLastFetchTime(new Date(score), sessionId, userId, function (err, res) {
  1350. if (err) {
  1351. logger.error("Update participant last fetch time failed: ", err);
  1352. }
  1353. });
  1354. })
  1355. .catch(function (err) {
  1356. logger.error("Update participant last fetch time failed: ", err);
  1357. });
  1358. }
  1359. /**
  1360. * 向用户推送通知,微信端用户直接推送消息,APP端通过个推发送通知消息。
  1361. *
  1362. * @param targetUserId
  1363. * @param message
  1364. */
  1365. static pushNotification(targetUserId, targetUserName, message,sessionType) {
  1366. let self = this;
  1367. Users.isPatientId(targetUserId, function (err, isPatient) {
  1368. if (isPatient) {
  1369. if(config.environment!='local'){//pc版不直接发送给居民,通过redis的publish
  1370. WechatClient.sendMessage(targetUserId, targetUserName, message);
  1371. }
  1372. message.targetUserId = targetUserId;
  1373. message.targetUserName = targetUserName;
  1374. message.sessionType = sessionType;
  1375. message.targetType = 'patient';
  1376. } else {
  1377. if(sessionType==SESSION_TYPES.P2P){
  1378. WechatClient.sendReadDoctorByDoctorId(targetUserId, message);
  1379. }
  1380. //告知医生新消息
  1381. WechatClient.sendSocketMessageToDoctor(targetUserId,message);
  1382. if(config.environment!='local'){//pc版不推送个推,通过redis的publish
  1383. WlyySDK.request(targetUserId, '', '', '', '/im/common/message/messages', 'POST', function (err, res) {
  1384. let count = 0;
  1385. res = JSON.parse(res)
  1386. if (res.status == 200) {
  1387. let data = res.data;
  1388. count = parseInt(JSON.parse(data.imMsgCount).count) + parseInt(data.system.amount) + parseInt(data.healthIndex.amount) + parseInt(data.sign.amount);
  1389. }
  1390. AppClient.sendNotification(targetUserId, message,sessionType,count);
  1391. });
  1392. }
  1393. message.targetUserId = targetUserId;
  1394. message.targetUserName = targetUserName;
  1395. message.sessionType = sessionType;
  1396. message.targetType = 'doctor';
  1397. }
  1398. //redis发布消息
  1399. if(config.pubSubSwitch) {//接收订阅消息处理开关,本地运行和测试库单独运行时防止用户接收消息2次
  1400. pubSub.publish(config.pubChannel,JSON.stringify(message));
  1401. }
  1402. });
  1403. }
  1404. /**
  1405. * 获取redis订阅消息,并处理
  1406. * @param targetUserId
  1407. * @param targetUserName
  1408. * @param message
  1409. * @param sessionType
  1410. */
  1411. static getRedisPushNotification(message) {
  1412. if (message.targetType=='patient') {
  1413. if(config.environment!='local'){//pc版接收要发给居民的消息不做处理
  1414. WechatClient.sendMessage(message.targetUserId, message.targetUserName, message);
  1415. }
  1416. } else {
  1417. if(message.sessionType==SESSION_TYPES.P2P){
  1418. WechatClient.sendReadDoctorByDoctorId(message.targetUserId, message);
  1419. }
  1420. //告知医生新消息
  1421. WechatClient.sendSocketMessageToDoctor(message.targetUserId,message);
  1422. if(config.environment!='local'){//pc版不推送个推
  1423. WlyySDK.request(message.targetUserId, '', '', '', '/im/common/message/messages', 'POST', function (err, res) {
  1424. let count = 0;
  1425. res = JSON.parse(res)
  1426. if (res.status == 200) {
  1427. let data = res.data;
  1428. count = parseInt(JSON.parse(data.imMsgCount).count) + parseInt(data.system.amount) + parseInt(data.healthIndex.amount) + parseInt(data.sign.amount);
  1429. }
  1430. AppClient.sendNotification(message.targetUserId, message,message.sessionType,count);
  1431. });
  1432. }
  1433. }
  1434. }
  1435. /**
  1436. * 向APP端通过socket发送通知消息。
  1437. *
  1438. * @param targetUserId
  1439. * @param message
  1440. */
  1441. static pushIntoNotification(targetUserId, targetUserName, message,sessionType) {
  1442. if(sessionType==SESSION_TYPES.P2P){
  1443. WechatClient.sendReadDoctorByDoctorId(targetUserId, message);
  1444. }
  1445. //告知医生新消息
  1446. WechatClient.sendSocketMessageToDoctor(targetUserId,message);
  1447. }
  1448. /**
  1449. * 针对MUC模式更新会话的当前状态
  1450. * @param sessionId
  1451. */
  1452. updateSessionStatus(sessionId,status,handler){
  1453. let self = this;
  1454. let sessionKey = RedisModel.makeRedisKey(REDIS_KEYS.Session,sessionId);
  1455. redis.hsetAsync(sessionKey,"status",status).then(function(res){
  1456. SessionRepo.updateSessionStatus(sessionId,status,function(err,sqlResult){
  1457. if(handler){
  1458. handler(err,sqlResult);
  1459. return;
  1460. }
  1461. if(err){
  1462. logger.error("set session status to mysql is error !");
  1463. }else{
  1464. logger.info("set session status is success");
  1465. ModelUtil.emitOK(self.eventEmitter, []);
  1466. }
  1467. });
  1468. });
  1469. }
  1470. }
  1471. // Expose class
  1472. module.exports = Sessions;