From 90f8fe7e538bb7e52d2558152a0390e693f39b11 Mon Sep 17 00:00:00 2001 From: Akihiko Odaki Date: Thu, 29 Mar 2018 01:20:40 +0900 Subject: Introduce processor --- src/server/api/endpoints/messaging/history.ts | 43 ++++++ src/server/api/endpoints/messaging/messages.ts | 102 ++++++++++++++ .../api/endpoints/messaging/messages/create.ts | 156 +++++++++++++++++++++ src/server/api/endpoints/messaging/unread.ts | 33 +++++ 4 files changed, 334 insertions(+) create mode 100644 src/server/api/endpoints/messaging/history.ts create mode 100644 src/server/api/endpoints/messaging/messages.ts create mode 100644 src/server/api/endpoints/messaging/messages/create.ts create mode 100644 src/server/api/endpoints/messaging/unread.ts (limited to 'src/server/api/endpoints/messaging') diff --git a/src/server/api/endpoints/messaging/history.ts b/src/server/api/endpoints/messaging/history.ts new file mode 100644 index 0000000000..1683ca7a89 --- /dev/null +++ b/src/server/api/endpoints/messaging/history.ts @@ -0,0 +1,43 @@ +/** + * Module dependencies + */ +import $ from 'cafy'; +import History from '../../models/messaging-history'; +import Mute from '../../models/mute'; +import { pack } from '../../models/messaging-message'; + +/** + * Show messaging history + * + * @param {any} params + * @param {any} user + * @return {Promise} + */ +module.exports = (params, user) => new Promise(async (res, rej) => { + // Get 'limit' parameter + const [limit = 10, limitErr] = $(params.limit).optional.number().range(1, 100).$; + if (limitErr) return rej('invalid limit param'); + + const mute = await Mute.find({ + muter_id: user._id, + deleted_at: { $exists: false } + }); + + // Get history + const history = await History + .find({ + user_id: user._id, + partner: { + $nin: mute.map(m => m.mutee_id) + } + }, { + limit: limit, + sort: { + updated_at: -1 + } + }); + + // Serialize + res(await Promise.all(history.map(async h => + await pack(h.message, user)))); +}); diff --git a/src/server/api/endpoints/messaging/messages.ts b/src/server/api/endpoints/messaging/messages.ts new file mode 100644 index 0000000000..67ba5e9d6d --- /dev/null +++ b/src/server/api/endpoints/messaging/messages.ts @@ -0,0 +1,102 @@ +/** + * Module dependencies + */ +import $ from 'cafy'; +import Message from '../../models/messaging-message'; +import User from '../../models/user'; +import { pack } from '../../models/messaging-message'; +import read from '../../common/read-messaging-message'; + +/** + * Get messages + * + * @param {any} params + * @param {any} user + * @return {Promise} + */ +module.exports = (params, user) => new Promise(async (res, rej) => { + // Get 'user_id' parameter + const [recipientId, recipientIdErr] = $(params.user_id).id().$; + if (recipientIdErr) return rej('invalid user_id param'); + + // Fetch recipient + const recipient = await User.findOne({ + _id: recipientId + }, { + fields: { + _id: true + } + }); + + if (recipient === null) { + return rej('user not found'); + } + + // Get 'mark_as_read' parameter + const [markAsRead = true, markAsReadErr] = $(params.mark_as_read).optional.boolean().$; + if (markAsReadErr) return rej('invalid mark_as_read param'); + + // Get 'limit' parameter + const [limit = 10, limitErr] = $(params.limit).optional.number().range(1, 100).$; + if (limitErr) return rej('invalid limit param'); + + // Get 'since_id' parameter + const [sinceId, sinceIdErr] = $(params.since_id).optional.id().$; + if (sinceIdErr) return rej('invalid since_id param'); + + // Get 'until_id' parameter + const [untilId, untilIdErr] = $(params.until_id).optional.id().$; + if (untilIdErr) return rej('invalid until_id param'); + + // Check if both of since_id and until_id is specified + if (sinceId && untilId) { + return rej('cannot set since_id and until_id'); + } + + const query = { + $or: [{ + user_id: user._id, + recipient_id: recipient._id + }, { + user_id: recipient._id, + recipient_id: user._id + }] + } as any; + + const sort = { + _id: -1 + }; + + if (sinceId) { + sort._id = 1; + query._id = { + $gt: sinceId + }; + } else if (untilId) { + query._id = { + $lt: untilId + }; + } + + // Issue query + const messages = await Message + .find(query, { + limit: limit, + sort: sort + }); + + // Serialize + res(await Promise.all(messages.map(async message => + await pack(message, user, { + populateRecipient: false + })))); + + if (messages.length === 0) { + return; + } + + // Mark as read all + if (markAsRead) { + read(user._id, recipient._id, messages); + } +}); diff --git a/src/server/api/endpoints/messaging/messages/create.ts b/src/server/api/endpoints/messaging/messages/create.ts new file mode 100644 index 0000000000..5184b2bd34 --- /dev/null +++ b/src/server/api/endpoints/messaging/messages/create.ts @@ -0,0 +1,156 @@ +/** + * Module dependencies + */ +import $ from 'cafy'; +import Message from '../../../models/messaging-message'; +import { isValidText } from '../../../models/messaging-message'; +import History from '../../../models/messaging-history'; +import User from '../../../models/user'; +import Mute from '../../../models/mute'; +import DriveFile from '../../../models/drive-file'; +import { pack } from '../../../models/messaging-message'; +import publishUserStream from '../../../event'; +import { publishMessagingStream, publishMessagingIndexStream, pushSw } from '../../../event'; +import config from '../../../../../conf'; + +/** + * Create a message + * + * @param {any} params + * @param {any} user + * @return {Promise} + */ +module.exports = (params, user) => new Promise(async (res, rej) => { + // Get 'user_id' parameter + const [recipientId, recipientIdErr] = $(params.user_id).id().$; + if (recipientIdErr) return rej('invalid user_id param'); + + // Myself + if (recipientId.equals(user._id)) { + return rej('cannot send message to myself'); + } + + // Fetch recipient + const recipient = await User.findOne({ + _id: recipientId + }, { + fields: { + _id: true + } + }); + + if (recipient === null) { + return rej('user not found'); + } + + // Get 'text' parameter + const [text, textErr] = $(params.text).optional.string().pipe(isValidText).$; + if (textErr) return rej('invalid text'); + + // Get 'file_id' parameter + const [fileId, fileIdErr] = $(params.file_id).optional.id().$; + if (fileIdErr) return rej('invalid file_id param'); + + let file = null; + if (fileId !== undefined) { + file = await DriveFile.findOne({ + _id: fileId, + 'metadata.user_id': user._id + }); + + if (file === null) { + return rej('file not found'); + } + } + + // テキストが無いかつ添付ファイルも無かったらエラー + if (text === undefined && file === null) { + return rej('text or file is required'); + } + + // メッセージを作成 + const message = await Message.insert({ + created_at: new Date(), + file_id: file ? file._id : undefined, + recipient_id: recipient._id, + text: text ? text : undefined, + user_id: user._id, + is_read: false + }); + + // Serialize + const messageObj = await pack(message); + + // Reponse + res(messageObj); + + // 自分のストリーム + publishMessagingStream(message.user_id, message.recipient_id, 'message', messageObj); + publishMessagingIndexStream(message.user_id, 'message', messageObj); + publishUserStream(message.user_id, 'messaging_message', messageObj); + + // 相手のストリーム + publishMessagingStream(message.recipient_id, message.user_id, 'message', messageObj); + publishMessagingIndexStream(message.recipient_id, 'message', messageObj); + publishUserStream(message.recipient_id, 'messaging_message', messageObj); + + // 3秒経っても(今回作成した)メッセージが既読にならなかったら「未読のメッセージがありますよ」イベントを発行する + setTimeout(async () => { + const freshMessage = await Message.findOne({ _id: message._id }, { is_read: true }); + if (!freshMessage.is_read) { + //#region ただしミュートされているなら発行しない + const mute = await Mute.find({ + muter_id: recipient._id, + deleted_at: { $exists: false } + }); + const mutedUserIds = mute.map(m => m.mutee_id.toString()); + if (mutedUserIds.indexOf(user._id.toString()) != -1) { + return; + } + //#endregion + + publishUserStream(message.recipient_id, 'unread_messaging_message', messageObj); + pushSw(message.recipient_id, 'unread_messaging_message', messageObj); + } + }, 3000); + + // Register to search database + if (message.text && config.elasticsearch.enable) { + const es = require('../../../db/elasticsearch'); + + es.index({ + index: 'misskey', + type: 'messaging_message', + id: message._id.toString(), + body: { + text: message.text + } + }); + } + + // 履歴作成(自分) + History.update({ + user_id: user._id, + partner: recipient._id + }, { + updated_at: new Date(), + user_id: user._id, + partner: recipient._id, + message: message._id + }, { + upsert: true + }); + + // 履歴作成(相手) + History.update({ + user_id: recipient._id, + partner: user._id + }, { + updated_at: new Date(), + user_id: recipient._id, + partner: user._id, + message: message._id + }, { + upsert: true + }); +}); diff --git a/src/server/api/endpoints/messaging/unread.ts b/src/server/api/endpoints/messaging/unread.ts new file mode 100644 index 0000000000..c4326e1d22 --- /dev/null +++ b/src/server/api/endpoints/messaging/unread.ts @@ -0,0 +1,33 @@ +/** + * Module dependencies + */ +import Message from '../../models/messaging-message'; +import Mute from '../../models/mute'; + +/** + * Get count of unread messages + * + * @param {any} params + * @param {any} user + * @return {Promise} + */ +module.exports = (params, user) => new Promise(async (res, rej) => { + const mute = await Mute.find({ + muter_id: user._id, + deleted_at: { $exists: false } + }); + const mutedUserIds = mute.map(m => m.mutee_id); + + const count = await Message + .count({ + user_id: { + $nin: mutedUserIds + }, + recipient_id: user._id, + is_read: false + }); + + res({ + count: count + }); +}); -- cgit v1.2.3-freya