users.js 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370
  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 AppStatusRepo = require('../../repository/mysql/app.status.repo');
  18. let ModelUtil = require('../../util/model.util');
  19. let ObjectUtil = require("../../util/object.util.js");
  20. let Patient = require('./patient');
  21. let Doctor = require('./doctor');
  22. let redisConn = RedisClient.redisClient().connection;
  23. let async = require('async');
  24. let log = require('../../util/log');
  25. let configFile = require('../../include/commons').CONFIG_FILE;
  26. let config = require('../../resources/config/' + configFile);
  27. const REDIS_KEYS = require('../../include/commons').REDIS_KEYS;
  28. const PLATFORMS = require('../../include/commons').PLATFORM;
  29. const SESSION_TYPE = require('../../include/commons').SESSION_TYPES;
  30. class Users extends RedisModel {
  31. constructor() {
  32. super();
  33. }
  34. /**
  35. * 获取用户,直接从MYSQL获取,缓存是否有在不能确定。
  36. *
  37. * @param userId
  38. * @param outCallback
  39. */
  40. getUserFromMySQL(userId, outCallback) {
  41. let self = this;
  42. async.waterfall([
  43. // determine user type
  44. function (callback) {
  45. Users.isPatientId(userId, function (err, isPatient) {
  46. callback(null, isPatient);
  47. });
  48. },
  49. // get from mysql
  50. function (isPatientId) {
  51. let repoProto = isPatientId ? PatientRepo : DoctorRepo;
  52. repoProto.findOne(userId, function (err, res) {
  53. let user = isPatientId ? new Patient() : new Doctor();
  54. if (res.length > 0) {
  55. user.name = res[0].name;
  56. user.sex = res[0].sex;
  57. user.birthdate = res[0].birthdate;
  58. user.avatar = res[0].avatar;
  59. if (res[0].openid) user.openid = res[0].openid;
  60. }
  61. outCallback(null, user);
  62. });
  63. }
  64. ]);
  65. }
  66. /**
  67. * 用户登录,仅缓存用户客户端状态信息,不缓存用户基本信息。
  68. *
  69. * 用户登录时会加载与之相关的会话列表,会话消息,用户自身信息:App状态与微信状态。
  70. *
  71. * TODO: 如果用户已经登录,但因为异常退出重新登录,是否需要刷新状态信息。
  72. *
  73. * @param userId
  74. * @param platform
  75. * @param deviceToken
  76. * @param clientId
  77. *
  78. * @return 用户token
  79. */
  80. login(userId, platform, deviceToken, clientId) {
  81. let self = this;
  82. let loginFromApp = platform !== PLATFORMS.Wechat;
  83. log.error(userId+" "+ platform+" "+deviceToken+" "+clientId);
  84. let usersKey = REDIS_KEYS.Users;
  85. let userKey = RedisModel.makeRedisKey(REDIS_KEYS.User, userId);
  86. let userStatusKey = RedisModel.makeRedisKey(loginFromApp ? REDIS_KEYS.UserAppStatus : REDIS_KEYS.UserWechatStatus, userId);
  87. let lastLoginTime = new Date();
  88. async.waterfall([
  89. // get user info from mysql
  90. function (callback) {
  91. self.getUserFromMySQL(userId, function (err, userInfo) {
  92. if (!userInfo) {
  93. ModelUtil.emitDataNotFound(self, 'User not exists.');
  94. return;
  95. }
  96. callback(null, userInfo);
  97. })
  98. },
  99. // cache user and app/wechat status
  100. function (userInfo, callback) {
  101. let multi = redisConn.multi()
  102. .zadd(usersKey, lastLoginTime.getTime(), userId);
  103. /*.hmset(userKey,
  104. 'avatar', userInfo.avatar ? userInfo.avatar : '',
  105. 'birthdate', userInfo.birthdate ? ObjectUtil.timestampToLong(userInfo.birthdate) : '',
  106. 'name', userInfo.name,
  107. 'role', loginFromApp ? 'doctor' : 'patient');*/
  108. if (loginFromApp) {
  109. AppStatusRepo.save(userId, deviceToken, clientId, platform, function (err, res) {
  110. if (err) log.error(err);
  111. });
  112. // cache app status
  113. multi = multi.hmset(userStatusKey,
  114. 'app_in_bg', 0,
  115. 'client_id', clientId,
  116. 'device_token', deviceToken,
  117. 'last_login_time', lastLoginTime.getTime(),
  118. 'platform', platform);
  119. } else {
  120. // cache wechat status
  121. multi = multi.hmset(userStatusKey,
  122. 'last_login_time', lastLoginTime.getTime(),
  123. 'openid', userInfo.openid,
  124. 'platform', platform);
  125. }
  126. multi.execAsync()
  127. .then(function (res) {
  128. callback(null);
  129. })
  130. .catch(function (ex) {
  131. log.error("Login failed while cache user status: ", ex);
  132. });
  133. },
  134. // cache sessions, participants, topics, messages
  135. function (callback) {
  136. SessionRepo.findAllIgnoreRole(userId, function (err, sessions) {
  137. if (err) {
  138. ModelUtil.emitError(self.eventEmitter, err.message);
  139. return;
  140. }
  141. sessions.forEach(function (session) {
  142. redisConn.zscore(REDIS_KEYS.Sessions, session.id, function (err, res) {
  143. if (res != null) return; // 已经缓存过的会话不再缓存
  144. (function (sessionId, userId) {
  145. let redisSession = [
  146. "id", session.id,
  147. "name", session.name,
  148. "type", session.type,
  149. "business_type", session.business_type,
  150. "last_sender_id", session.last_sender_id||"",
  151. "last_sender_name", session.last_sender_name||"",
  152. "last_content_type", session.last_content_type||"",
  153. "last_content", session.last_content||"",
  154. "last_message_time", session.last_message_time||"",
  155. "create_date", ObjectUtil.timestampToLong(session.create_date),
  156. "status",session.status==null?0:session.status
  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||(new Date()));
  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. AppStatusRepo.destroy(userId, function (err, res) {
  293. if(err) log.error("Delete user status failed: " + err);
  294. });
  295. callback(null, null);
  296. }],
  297. function (err, res) {
  298. }
  299. );
  300. }
  301. /**
  302. * 用户ID是否属于患者。
  303. *
  304. * @param userId
  305. * @param callback
  306. */
  307. static isPatientId(userId, callback) {
  308. async.waterfall([
  309. function (callback) {
  310. ImDb.execQuery({
  311. "sql": "select case when count(*) > 0 then true else false end 'is_patient' from patients where id = ?",
  312. "args": [userId],
  313. "handler": function (err, res) {
  314. if (err) {
  315. callback(err, res);
  316. return;
  317. }
  318. callback(null, res);
  319. }
  320. });
  321. },
  322. function (res, callback) {
  323. if (res.length === 0) return false;
  324. callback(null, res[0].is_patient);
  325. }
  326. ],
  327. function (err, res) {
  328. if (err) {
  329. log.error("User id check failed: ", err);
  330. callback(null, false);
  331. return;
  332. }
  333. callback(null, res !== 0);
  334. });
  335. }
  336. }
  337. let Promises = require('bluebird');
  338. Promises.promisifyAll(Users.prototype);
  339. module.exports = Users;