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. if (res != null) return; // 已经缓存过的会话不再缓存
  139. (function (sessionId, userId) {
  140. let business_type = session.business_type;
  141. if (!session.business_type && session.type == SESSION_TYPE.MUC) {
  142. business_type = 2
  143. } else if (!session.business_type && session.type != SESSION_TYPE.MUC) {
  144. business_type = 1
  145. }
  146. let redisSession = [
  147. "id", session.id,
  148. "name", session.name,
  149. "type", session.type,
  150. "business_type", business_type,
  151. "last_sender_id", session.last_sender_id == null ? "" : session.last_sender_id,
  152. "last_sender_name", session.last_sender_name == null ? "" : session.last_sender_name,
  153. "last_content_type", session.last_content_type == null ? "" : session.last_content_type,
  154. "last_content", session.last_content == null ? "" : session.last_content,
  155. "last_message_time", session.last_message_time == null ? "" : session.last_message_time,
  156. "create_date", ObjectUtil.timestampToLong(session.create_date),
  157. ];
  158. // cache sessions
  159. redisConn.multi()
  160. .zadd(REDIS_KEYS.Sessions, lastLoginTime.getTime(), sessionId) // 会话的最后活动时间设置为此用户的登录时间
  161. .zadd(RedisModel.makeRedisKey(REDIS_KEYS.UserSessions, userId), lastLoginTime.getTime(), sessionId) // 会话的最后活动时间设置为此用户的登录时间
  162. .hmset(RedisModel.makeRedisKey(REDIS_KEYS.Session, sessionId), redisSession)
  163. .execAsync()
  164. .then(function (res) {
  165. // cache participants
  166. let sessionParticipantsKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipants, sessionId);
  167. let sessionParticipantsRoleKey = RedisModel.makeRedisKey(REDIS_KEYS.SessionParticipantsRole, sessionId);
  168. ParticipantRepo.findAll(sessionId, function (err, participants) {
  169. if (err) {
  170. ModelUtil.emitError(self.eventEmitter, err.message);
  171. return;
  172. }
  173. let multi = redisConn.multi();
  174. participants.forEach(function (participant) {
  175. let participantId = participant.id;
  176. let participantRole = participant.role;
  177. let score = ObjectUtil.timestampToLong(participant.last_fetch_time);
  178. multi = multi.zadd(sessionParticipantsKey, score, participantId)
  179. .hset(sessionParticipantsRoleKey, participantId, participantRole);
  180. });
  181. multi.execAsync()
  182. .then(function (res) {
  183. })
  184. .catch(function (ex) {
  185. log.error("Login failed while caching participants: ", ex);
  186. });
  187. });
  188. // cache messages
  189. let messagesKey = RedisModel.makeRedisKey(REDIS_KEYS.Messages, sessionId);
  190. let messagesByTimestampKey = RedisModel.makeRedisKey(REDIS_KEYS.MessagesByTimestamp, sessionId);
  191. MessageRepo.findBySessionId(sessionId, 0, config.sessionConfig.maxMessageCount, null, function (err, messages) {
  192. if (err) {
  193. ModelUtil.emitError(self.eventEmitter, err.message);
  194. return;
  195. }
  196. let multi = redisConn.multi();
  197. messages.forEach(function (message) {
  198. let msgJson = {
  199. id: message.id,
  200. sender_id: message.sender_id,
  201. sender_name: message.sender_name,
  202. timestamp: ObjectUtil.timestampToLong(message.timestamp),
  203. content_type: message.content_type,
  204. content: message.content
  205. };
  206. multi = multi.hset(messagesKey, message.id, JSON.stringify(msgJson))
  207. .zadd(messagesByTimestampKey, ObjectUtil.timestampToLong(message.timestamp), message.id);
  208. });
  209. multi.execAsync()
  210. .then(function (res) {
  211. })
  212. .catch(function (ex) {
  213. log.error("Login failed while caching messages: ", ex);
  214. });
  215. });
  216. // cache topics for MUC
  217. let topicsKey = RedisModel.makeRedisKey(REDIS_KEYS.Topics, sessionId);
  218. TopicRepo.findAllBySessionId(sessionId, function (err, topics) {
  219. if (err) {
  220. ModelUtil.emitError(self.eventEmitter, err.message);
  221. return;
  222. }
  223. topics.forEach(function (topic) {
  224. let topicKey = RedisModel.makeRedisKey(REDIS_KEYS.Topic, topic.id);
  225. let topicId = topic.id;
  226. let name = topic.name == null ? "" : topic.name;
  227. let createTime = ObjectUtil.timestampToLong(topic.create_time);
  228. let endBy = topic.end_by == null ? "" : topic.end_by;
  229. let endTime = topic.end_time == null ? 0 : ObjectUtil.timestampToLong(topic.end_time);
  230. let startMessageId = topic.start_message_id == null ? "" : topic.start_message_id;
  231. let endMessageId = topic.end_message_id == null ? "" : topic.end_message_id;
  232. let description = topic.description == null ? "" : topic.description;
  233. let status = topic.status == null ? 0 : topic.status;
  234. redisConn.multi()
  235. .zadd(topicsKey, createTime, topicId)
  236. .hmset(topicKey,
  237. 'name', name,
  238. 'session_id', sessionId,
  239. 'create_time', createTime,
  240. 'end_by', endBy,
  241. 'end_time', endTime,
  242. 'start_message_id', startMessageId,
  243. 'end_message_id', endMessageId,
  244. 'description', description,
  245. 'status', status)
  246. .execAsync()
  247. .catch(function (ex) {
  248. log.error("Login failed while caching topics: ", ex);
  249. });
  250. });
  251. });
  252. })
  253. .catch(function (ex) {
  254. log.error("Login failed while caching sessions: ", ex);
  255. });
  256. })(session.id, userId);
  257. });
  258. });
  259. });
  260. callback(null, null);
  261. }
  262. ],
  263. function (err, res) {
  264. ModelUtil.emitOK(self.eventEmitter, {});
  265. });
  266. }
  267. logout(userId) {
  268. let self = this;
  269. async.waterfall([
  270. function (callback) {
  271. Users.isPatientId(userId, function (err, isPatient) {
  272. callback(null, isPatient)
  273. });
  274. },
  275. function (isPatient, callback) {
  276. let usersKey = REDIS_KEYS.Users;
  277. let userStatusKey = RedisModel.makeRedisKey(isPatient ? REDIS_KEYS.UserWechatStatus : REDIS_KEYS.UserAppStatus, userId);
  278. redisConn.multi()
  279. .zrem(usersKey, userId)
  280. .del(userStatusKey)
  281. .execAsync()
  282. .then(function (res) {
  283. if (res.length > 0 && res[0] === 0) {
  284. ModelUtil.emitDataNotFound(self.eventEmitter, {message: "User not found."});
  285. } else {
  286. ModelUtil.emitOK(self.eventEmitter, {});
  287. }
  288. })
  289. .catch(function (ex) {
  290. log.error("Logout failed: ", ex);
  291. });
  292. }],
  293. function (err, res) {
  294. }
  295. );
  296. }
  297. /**
  298. * 用户ID是否属于患者。
  299. *
  300. * @param userId
  301. * @param callback
  302. */
  303. static isPatientId(userId, callback) {
  304. async.waterfall([
  305. function (callback) {
  306. var sql = "select case when count(*) > 0 then true else false end 'is_patient' from patients where id = ?";
  307. ImDb.execQuery({
  308. "sql": sql,
  309. "args": [userId],
  310. "handler": function (err, res) {
  311. if (err) callback(err, res);
  312. callback(null, res);
  313. }
  314. });
  315. },
  316. function (res, callback) {
  317. if (res.length === 0) return false;
  318. callback(null, res[0].is_patient);
  319. }
  320. ],
  321. function (err, res) {
  322. if (err) {
  323. log.error("User id check failed: ", err);
  324. callback(null, false);
  325. return;
  326. }
  327. callback(null, res !== 0);
  328. });
  329. }
  330. }
  331. let Promises = require('bluebird');
  332. Promises.promisifyAll(Users.prototype);
  333. module.exports = Users;