[API] Refactor and Bug fix

src/api/common/read-messaging-message.ts

+import * as mongo from 'mongodb';
+import Message from '../models/messaging-message';
+import { IMessagingMessage as IMessage } from '../models/messaging-message';
+import publishUserStream from '../event';
+import { publishMessagingStream } from '../event';
+
+/**
+ * Mark as read message(s)
+ */
+export default (
+	user: string | mongo.ObjectID,
+	otherparty: string | mongo.ObjectID,
+	message: string | string[] | IMessage | IMessage[] | mongo.ObjectID | mongo.ObjectID[]
+) => new Promise<any>(async (resolve, reject) => {
+
+	const userId = mongo.ObjectID.prototype.isPrototypeOf(user)
+		? user
+		: new mongo.ObjectID(user);
+
+	const otherpartyId = mongo.ObjectID.prototype.isPrototypeOf(otherparty)
+		? otherparty
+		: new mongo.ObjectID(otherparty);
+
+	const ids: mongo.ObjectID[] = Array.isArray(message)
+		? mongo.ObjectID.prototype.isPrototypeOf(message[0])
+			? (message as mongo.ObjectID[])
+			: typeof message[0] === 'string'
+				? (message as string[]).map(m => new mongo.ObjectID(m))
+				: (message as IMessage[]).map(m => m._id)
+		: mongo.ObjectID.prototype.isPrototypeOf(message)
+			? [(message as mongo.ObjectID)]
+			: typeof message === 'string'
+				? [new mongo.ObjectID(message)]
+				: [(message as IMessage)._id];
+
+	// Update documents
+	await Message.update({
+		_id: { $in: ids },
+		user_id: otherpartyId,
+		recipient_id: userId,
+		is_read: false
+	}, {
+		$set: {
+			is_read: true
+		}
+	}, {
+		multi: true
+	});
+
+	// Publish event
+	publishMessagingStream(otherpartyId, userId, 'read', ids.map(id => id.toString()));
+
+	// Calc count of my unread messages
+	const count = await Message
+		.count({
+			recipient_id: userId,
+			is_read: false
+		});
+
+	if (count == 0) {
+		// 全ての(いままで未読だった)自分宛てのメッセージを(これで)読みましたよというイベントを発行
+		publishUserStream(userId, 'read_all_messaging_messages');
+	}
+});

src/api/endpoints/messaging/messages.ts

@@ -5,8 +5,7 @@ import $ from 'cafy';
 import Message from '../../models/messaging-message';
 import User from '../../models/user';
 import serialize from '../../serializers/messaging-message';
-import publishUserStream from '../../event';
-import { publishMessagingStream } from '../../event';
+import read from '../../common/read-messaging-message';
 
 /**
  * Get messages
@@ -98,32 +97,6 @@ module.exports = (params, user) => new Promise(async (res, rej) => {
 
 	// Mark as read all
 	if (markAsRead) {
-		const ids = messages
-			.filter(m => m.is_read == false)
-			.filter(m => m.recipient_id.equals(user._id))
-			.map(m => m._id);
-
-		// Update documents
-		await Message.update({
-			_id: { $in: ids }
-		}, {
-			$set: { is_read: true }
-		}, {
-			multi: true
-		});
-
-		// Publish event
-		publishMessagingStream(recipient._id, user._id, 'read', ids.map(id => id.toString()));
-
-		const count = await Message
-			.count({
-				recipient_id: user._id,
-				is_read: false
-			});
-
-		if (count == 0) {
-			// 全ての(いままで未読だった)メッセージを(これで)読みましたよというイベントを発行
-			publishUserStream(user._id, 'read_all_messaging_messages');
-		}
+		read(user._id, recipient._id, messages);
 	}
 });

src/api/models/messaging-message.ts

+import * as mongo from 'mongodb';
 import db from '../../db/mongodb';
 
 export default db.get('messaging_messages') as any; // fuck type definition
 
+export interface IMessagingMessage {
+	_id: mongo.ObjectID;
+}
+
 export function isValidText(text: string): boolean {
 	return text.length <= 1000 && text.trim() != '';
 }

src/api/stream/messaging.ts

 import * as mongodb from 'mongodb';
 import * as websocket from 'websocket';
 import * as redis from 'redis';
-import Message from '../models/messaging-message';
-import { publishMessagingStream } from '../event';
+import read from '../common/read-messaging-message';
 
 export default function messagingStream(request: websocket.request, connection: websocket.connection, subscriber: redis.RedisClient, user: any): void {
 	const otherparty = request.resourceURL.query.otherparty;
@@ -18,42 +17,8 @@ export default function messagingStream(request: websocket.request, connection:
 
 		switch (msg.type) {
 			case 'read':
-				if (!msg.id) {
-					return;
-				}
-
-				const id = new mongodb.ObjectID(msg.id);
-
-				// Fetch message
-				// SELECT _id, user_id, is_read
-				const message = await Message.findOne({
-					_id: id,
-					recipient_id: user._id
-				}, {
-					fields: {
-						_id: true,
-						user_id: true,
-						is_read: true
-					}
-				});
-
-				if (message == null) {
-					return;
-				}
-
-				if (message.is_read) {
-					return;
-				}
-
-				// Update documents
-				await Message.update({
-					_id: id
-				}, {
-					$set: { is_read: true }
-				});
-
-				// Publish event
-				publishMessagingStream(message.user_id, user._id, 'read', id.toString());
+				if (!msg.id) return;
+				read(user._id, otherparty, msg.id);
 				break;
 		}
 	});