Add messaging widget

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

@@ -3,6 +3,7 @@ import Message from '../models/messaging-message';
 import { IMessagingMessage as IMessage } from '../models/messaging-message';
 import publishUserStream from '../event';
 import { publishMessagingStream } from '../event';
+import { publishMessagingIndexStream } from '../event';
 
 /**
  * Mark as read message(s)
@@ -49,6 +50,7 @@ export default (
 
 	// Publish event
 	publishMessagingStream(otherpartyId, userId, 'read', ids.map(id => id.toString()));
+	publishMessagingIndexStream(userId, 'read', ids.map(id => id.toString()));
 
 	// Calc count of my unread messages
 	const count = await Message

src/api/endpoints/messaging/messages/create.ts

@@ -10,6 +10,7 @@ import DriveFile from '../../../models/drive-file';
 import serialize from '../../../serializers/messaging-message';
 import publishUserStream from '../../../event';
 import { publishMessagingStream } from '../../../event';
+import { publishMessagingIndexStream } from '../../../event';
 import config from '../../../../conf';
 
 /**
@@ -85,10 +86,12 @@ module.exports = (params, user) => new Promise(async (res, rej) => {
 
 	// 自分のストリーム
 	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秒経っても(今回作成した)メッセージが既読にならなかったら「未読のメッセージがありますよ」イベントを発行する

src/api/event.ts

@@ -25,6 +25,10 @@ class MisskeyEvent {
 		this.publish(`messaging-stream:${userId}-${otherpartyId}`, type, typeof value === 'undefined' ? null : value);
 	}
 
+	public publishMessagingIndexStream(userId: ID, type: string, value?: any): void {
+		this.publish(`messaging-index-stream:${userId}`, type, typeof value === 'undefined' ? null : value);
+	}
+
 	public publishChannelStream(channelId: ID, type: string, value?: any): void {
 		this.publish(`channel-stream:${channelId}`, type, typeof value === 'undefined' ? null : value);
 	}
@@ -46,4 +50,6 @@ export const publishPostStream = ev.publishPostStream.bind(ev);
 
 export const publishMessagingStream = ev.publishMessagingStream.bind(ev);
 
+export const publishMessagingIndexStream = ev.publishMessagingIndexStream.bind(ev);
+
 export const publishChannelStream = ev.publishChannelStream.bind(ev);

src/api/stream/messaging-index.ts

@@ -0,0 +1,10 @@
+import * as websocket from 'websocket';
+import * as redis from 'redis';
+
+export default function(request: websocket.request, connection: websocket.connection, subscriber: redis.RedisClient, user: any): void {
+	// Subscribe messaging index stream
+	subscriber.subscribe(`misskey:messaging-index-stream:${user._id}`);
+	subscriber.on('message', (_, data) => {
+		connection.send(data);
+	});
+}

src/api/streaming.ts

@@ -8,6 +8,7 @@ import isNativeToken from './common/is-native-token';
 
 import homeStream from './stream/home';
 import messagingStream from './stream/messaging';
+import messagingIndexStream from './stream/messaging-index';
 import serverStream from './stream/server';
 import requestsStream from './stream/requests';
 import channelStream from './stream/channel';
@@ -58,6 +59,7 @@ module.exports = (server: http.Server) => {
 		const channel =
 			request.resourceURL.pathname === '/' ? homeStream :
 			request.resourceURL.pathname === '/messaging' ? messagingStream :
+			request.resourceURL.pathname === '/messaging-index' ? messagingIndexStream :
 			null;
 
 		if (channel !== null) {