users.js 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367
  1. /**
  2. * 用户集合。管理Redis中的用户列表。
  3. *
  4. * author: Sand
  5. * since: 12/13/2016
  6. */
  7. "use strict";
  8. let RedisClient = require('../../repository/redis/redis.client');
  9. let RedisModel = require('../redis.model');
  10. let ImDb = require('../../repository/mysql/db/im.db');
  11. let ParticipantRepo = require('../../repository/mysql/participant.repo');
  12. let DoctorRepo = require('../../repository/mysql/doctor.repo');
  13. let PatientRepo = require('../../repository/mysql/patient.repo');
  14. let SessionRepo = require('../../repository/mysql/session.repo');
  15. let MessageRepo = require('../../repository/mysql/message.repo');
  16. let TopicRepo = require('../../repository/mysql/topics.repo');
  17. let ModelUtil = require('../../util/model.util');
  18. let ObjectUtil = require("../../util/object.util.js");
  19. let Patient = require('./patient');
  20. let Doctor = require('./doctor');
  21. let redisConn = RedisClient.redisClient().connection;
  22. let async = require('async');
  23. let log = require('../../util/log');
  24. let configFile = require('../../include/commons').CONFIG_FILE;
  25. let config = require('../../resources/config/' + configFile);
  26. const REDIS_KEYS = require('../../include/commons').REDIS_KEYS;
  27. const PLATFORMS = require('../../include/commons').PLATFORM;
  28. const SESSION_TYPE = require('../../include/commons').SESSION_TYPES;
  29. class Users extends RedisModel {
  30. constructor() {
  31. super();
  32. }
  33. /**
  34. * 获取用户,直接从MYSQL获取,缓存是否有在不能确定。
  35. *
  36. * @param userId
  37. * @param outCallback
  38. */
  39. getUserFromMySQL(userId, outCallback) {
  40. let self = this;
  41. async.waterfall([
  42. // determine user type
  43. function (callback) {
  44. Users.isPatientId(userId, function (err, isPatient) {
  45. callback(null, isPatient);
  46. });
  47. },
  48. // get from mysql
  49. function (isPatientId) {
  50. let repoProto = isPatientId ? PatientRepo : DoctorRepo;
  51. repoProto.findOne(userId, function (err, res) {
  52. let user = isPatientId ? new Patient() : new Doctor();
  53. if (res.length > 0) {
  54. user.name = res[0].name;
  55. user.sex = res[0].sex;
  56. user.birthdate = res[0].birthdate;
  57. user.avatar = res[0].avatar;
  58. if(res[0].openid) user.openid = res[0].openid;
  59. }
  60. outCallback(null, user);
  61. });
  62. }
  63. ]);
  64. }
  65. /**
  66. * 用户登录,仅缓存用户客户端状态信息,不缓存用户基本信息。
  67. *
  68. * 用户登录时会加载与之相关的会话列表,会话消息,用户自身信息:App状态与微信状态。
  69. *
  70. * TODO: 如果用户已经登录,但因为异常退出重新登录,是否需要刷新状态信息。
  71. *
  72. * @param userId
  73. * @param platform
  74. * @param deviceToken
  75. * @param clientId
  76. *
  77. * @return 用户token
  78. */
  79. login(userId, platform, deviceToken, clientId) {
  80. let self = this;
  81. let loginFromApp = platform !== PLATFORMS.Wechat;
  82. let usersKey = REDIS_KEYS.Users;
  83. let userKey = RedisModel.makeRedisKey(REDIS_KEYS.User, userId);
  84. let userStatusKey = RedisModel.makeRedisKey(loginFromApp ? REDIS_KEYS.UserAppStatus : REDIS_KEYS.UserWechatStatus, userId);
  85. let lastLoginTime = new Date();
  86. async.waterfall([
  87. // get user info from mysql
  88. function (callback) {
  89. self.getUserFromMySQL(userId, function (err, userInfo) {
  90. if (!userInfo) {
  91. ModelUtil.emitDataNotFound(self, 'User not exists.');
  92. return;
  93. }
  94. callback(null, userInfo);
  95. })
  96. },
  97. // cache user and app/wechat status
  98. function (userInfo, callback) {
  99. let multi = redisConn.multi()
  100. .zadd(usersKey, lastLoginTime.getTime(), userId);
  101. /*.hmset(userKey,
  102. 'avatar', userInfo.avatar ? userInfo.avatar : '',
  103. 'birthdate', userInfo.birthdate ? ObjectUtil.timestampToLong(userInfo.birthdate) : '',
  104. 'name', userInfo.name,
  105. 'role', loginFromApp ? 'doctor' : 'patient');*/
  106. if (loginFromApp) {
  107. // cache app status
  108. multi = multi.hmset(userStatusKey,
  109. 'app_in_bg', 0,
  110. 'client_id', clientId,
  111. 'device_token', deviceToken,
  112. 'last_login_time', lastLoginTime.getTime(),
  113. 'platform', platform);
  114. } else {
  115. // cache wechat status
  116. multi = multi.hmset(userStatusKey,
  117. 'last_login_time', lastLoginTime.getTime(),
  118. 'openid', userInfo.openid,
  119. 'platform', platform);
  120. }
  121. multi.execAsync()
  122. .then(function (res) {
  123. callback(null);
  124. })
  125. .catch(function (ex) {
  126. log.error("Login failed while cache user status: ", ex);
  127. });
  128. },
  129. // cache sessions, participants, topics, messages
  130. function (callback) {
  131. SessionRepo.findAll(userId, function (err, sessions) {
  132. if (err) {
  133. ModelUtil.emitError(self.eventEmitter, err.message);
  134. return;
  135. }
  136. sessions.forEach(function (session) {
  137. redisConn.zscore(REDIS_KEYS.Sessions, session.id, function (err, res) {
  138. // 已经缓存过的会话不再缓存
  139. if (res != null) return;
  140. (function (sessionId, userId) {
  141. var business_type = session.business_type;
  142. if(!session.business_type&&session.type==SESSION_TYPE.MUC){
  143. business_type = 2
  144. }else if(!session.business_type&&session.type!=SESSION_TYPE.MUC){
  145. business_type = 1
  146. }
  147. let redisSession = [
  148. "id", session.id,
  149. "name", session.name,
  150. "type", session.type,
  151. "business_type", business_type,
  152. "last_sender_id", session.last_sender_id == null ? "" : session.last_sender_id,
  153. "last_sender_name", session.last_sender_name == null ? "" : session.last_sender_name,
  154. "last_content_type", session.last_content_type == null ? "" : session.last_content_type,
  155. "last_content", session.last_content == null ? "" : session.last_content,
  156. "last_message_time", session.last_message_time == null ? "" : session.last_message_time,
  157. "create_date", ObjectUtil.timestampToLong(session.create_date),
  158. ];
  159. // cache sessions
  160. redisConn.multi()
  161. .zadd(REDIS_KEYS.Sessions, lastLoginTime.getTime(), sessionId) // 会话的最后活动时间设置为此用户的登录时间
  162. .zadd(RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, userId), lastLoginTime.getTime(), sessionId) // 会话的最后活动时间设置为此用户的登录时间
  163. .hmset(RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId), redisSession)
  164. .execAsync()
  165. .then(function (res) {
  166. // cache participants
  167. let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
  168. let sessionParticipantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
  169. ParticipantRepo.findAll(sessionId, function (err, participants) {
  170. if (err) {
  171. ModelUtil.emitError(self.eventEmitter, err.message);
  172. return;
  173. }
  174. let multi = redisConn.multi();
  175. participants.forEach(function (participant) {
  176. let participantId = participant.id;
  177. let participantRole = participant.role;
  178. let score = ObjectUtil.timestampToLong(participant.last_fetch_time);
  179. multi = multi.zadd(sessionParticipantsKey, score, participantId)
  180. .hset(sessionParticipantsRoleKey, participantId, participantRole);
  181. });
  182. multi.execAsync()
  183. .then(function (res) {
  184. })
  185. .catch(function (ex) {
  186. log.error("Login failed while caching participants: ", ex);
  187. });
  188. });
  189. // cache messages
  190. let messagesKey = RedisModel.makeRedisKey(REDIS_KEYS.Messages, sessionId);
  191. let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
  192. MessageRepo.findBySessionId(sessionId, 0, config.sessionConfig.maxMessageCount, null, function (err, messages) {
  193. if (err) {
  194. ModelUtil.emitError(self.eventEmitter, err.message);
  195. return;
  196. }
  197. let multi = redisConn.multi();
  198. messages.forEach(function (message) {
  199. let msgJson = {
  200. id: message.id,
  201. sender_id: message.sender_id,
  202. sender_name: message.sender_name,
  203. timestamp: ObjectUtil.timestampToLong(message.timestamp),
  204. content_type: message.content_type,
  205. content: message.content
  206. };
  207. multi = multi.hset(messagesKey, message.id, JSON.stringify(msgJson))
  208. .zadd(messagesByTimestampKey, ObjectUtil.timestampToLong(message.timestamp), message.id);
  209. });
  210. multi.execAsync()
  211. .then(function (res) {
  212. })
  213. .catch(function (ex) {
  214. log.error("Login failed while caching messages: ", ex);
  215. });
  216. });
  217. // cache topics for MUC
  218. let topicsKey = RedisModel.makeRedisKey(REDIS_KEYS.Topics, sessionId);
  219. TopicRepo.findAllBySessionId(sessionId, function (err, topics) {
  220. if (err) {
  221. ModelUtil.emitError(self.eventEmitter, err.message);
  222. return;
  223. }
  224. topics.forEach(function (topic) {
  225. let topicKey = RedisModel.makeRedisKey(REDIS_KEYS.Topic, topic.id);
  226. let topicId = topic.id;
  227. let name = topic.name == null ? "" : topic.name;
  228. let createTime = ObjectUtil.timestampToLong(topic.create_time);
  229. let endBy = topic.end_by == null ? "" : topic.end_by;
  230. let endTime = topic.end_time == null ? 0 : ObjectUtil.timestampToLong(topic.end_time);
  231. let startMessageId = topic.start_message_id == null ? "" : topic.start_message_id;
  232. let endMessageId = topic.end_message_id == null ? "" : topic.end_message_id;
  233. let description = topic.description == null ? "" : topic.description;
  234. let status = topic.status == null ? 0 : topic.status;
  235. redisConn.multi()
  236. .zadd(topicsKey, createTime, topicId)
  237. .hmset(topicKey,
  238. 'name', name,
  239. 'session_id', sessionId,
  240. 'create_time', createTime,
  241. 'end_by', endBy,
  242. 'end_time', endTime,
  243. 'start_message_id', startMessageId,
  244. 'end_message_id', endMessageId,
  245. 'description', description,
  246. 'status', status)
  247. .execAsync()
  248. .catch(function (ex) {
  249. log.error("Login failed while caching topics: ", ex);
  250. });
  251. });
  252. });
  253. })
  254. .catch(function (ex) {
  255. log.error("Login failed while caching sessions: ", ex);
  256. });
  257. })(session.id, userId);
  258. });
  259. });
  260. });
  261. callback(null, null);
  262. }
  263. ],
  264. function (err, res) {
  265. ModelUtil.emitOK(self.eventEmitter, {});
  266. });
  267. }
  268. logout(userId) {
  269. let self = this;
  270. async.waterfall([
  271. function (callback) {
  272. Users.isPatientId(userId, function (err, isPatient) {
  273. callback(null, isPatient)
  274. });
  275. },
  276. function (isPatient, callback) {
  277. let usersKey = REDIS_KEYS.Users;
  278. let userStatusKey = RedisModel.makeRedisKey(isPatient ? REDIS_KEYS.UserWechatStatus : REDIS_KEYS.UserAppStatus, userId);
  279. redisConn.multi()
  280. .zrem(usersKey, userId)
  281. .del(userStatusKey)
  282. .execAsync()
  283. .then(function (res) {
  284. if (res.length > 0 && res[0] === 0) {
  285. ModelUtil.emitDataNotFound(self.eventEmitter, {message: "User not found."});
  286. } else {
  287. ModelUtil.emitOK(self.eventEmitter, {});
  288. }
  289. })
  290. .catch(function (ex) {
  291. log.error("Logout failed: ", ex);
  292. });
  293. }],
  294. function (err, res) {
  295. }
  296. );
  297. }
  298. /**
  299. * 用户ID是否属于患者。
  300. *
  301. * @param userId
  302. * @param callback
  303. */
  304. static isPatientId(userId, callback) {
  305. async.waterfall([
  306. function (callback) {
  307. var sql = "select case when count(*) > 0 then true else false end 'is_patient' from patients where id = ?";
  308. ImDb.execQuery({
  309. "sql": sql,
  310. "args": [userId],
  311. "handler": function (err, res) {
  312. if (err) callback(err, res);
  313. callback(null, res);
  314. }
  315. });
  316. },
  317. function (res, callback) {
  318. if (res.length === 0) return false;
  319. callback(null, res[0].is_patient);
  320. }
  321. ],
  322. function (err, res) {
  323. if (err) {
  324. log.error("User id check failed: ", err);
  325. callback(null, false);
  326. return;
  327. }
  328. callback(null, res !== 0);
  329. });
  330. }
  331. }
  332. let Promises = require('bluebird');
  333. Promises.promisifyAll(Users.prototype);
  334. module.exports = Users;