#302

src/api/endpoints/posts/polls/vote.ts

@@ -5,6 +5,7 @@ import $ from 'cafy';
 import Vote from '../../../models/poll-vote';
 import Post from '../../../models/post';
 import notify from '../../../common/notify';
+import { publishPostStream } from '../../../event';
 
 /**
  * Vote poll of a post
@@ -62,11 +63,13 @@ module.exports = (params, user) => new Promise(async (res, rej) => {
 	const inc = {};
 	inc[`poll.choices.${findWithAttr(post.poll.choices, 'id', choice)}.votes`] = 1;
 
-	// Increment likes count
-	Post.update({ _id: post._id }, {
+	// Increment votes count
+	await Post.update({ _id: post._id }, {
 		$inc: inc
 	});
 
+	publishPostStream(post._id, 'poll_voted');
+
 	// Notify
 	notify(post.user_id, user._id, 'poll_vote', {
 		post_id: post._id,

src/api/endpoints/posts/reactions/create.ts

@@ -5,6 +5,7 @@ import $ from 'cafy';
 import Reaction from '../../../models/post-reaction';
 import Post from '../../../models/post';
 import notify from '../../../common/notify';
+import { publishPostStream } from '../../../event';
 
 /**
  * React to a post
@@ -69,10 +70,12 @@ module.exports = (params, user) => new Promise(async (res, rej) => {
 	inc['reaction_counts.' + reaction] = 1;
 
 	// Increment reactions count
-	Post.update({ _id: post._id }, {
+	await Post.update({ _id: post._id }, {
 		$inc: inc
 	});
 
+	publishPostStream(post._id, 'reacted');
+
 	// Notify
 	notify(post.user_id, user._id, 'reaction', {
 		post_id: post._id,

src/api/event.ts

@@ -25,6 +25,10 @@ class MisskeyEvent {
 		this.publish(`user-stream:${userId}`, type, typeof value === 'undefined' ? null : value);
 	}
 
+	public publishPostStream(postId: ID, type: string, value?: any): void {
+		this.publish(`post-stream:${postId}`, type, typeof value === 'undefined' ? null : value);
+	}
+
 	public publishMessagingStream(userId: ID, otherpartyId: ID, type: string, value?: any): void {
 		this.publish(`messaging-stream:${userId}-${otherpartyId}`, type, typeof value === 'undefined' ? null : value);
 	}
@@ -34,4 +38,6 @@ const ev = new MisskeyEvent();
 
 export default ev.publishUserStream.bind(ev);
 
+export const publishPostStream = ev.publishPostStream.bind(ev);
+
 export const publishMessagingStream = ev.publishMessagingStream.bind(ev);

src/api/stream/home.ts

@@ -1,10 +1,46 @@
 import * as websocket from 'websocket';
 import * as redis from 'redis';
+import * as debug from 'debug';
+
+import serializePost from '../serializers/post';
+
+const log = debug('misskey');
 
 export default function homeStream(request: websocket.request, connection: websocket.connection, subscriber: redis.RedisClient, user: any): void {
 	// Subscribe Home stream channel
 	subscriber.subscribe(`misskey:user-stream:${user._id}`);
-	subscriber.on('message', (_, data) => {
-		connection.send(data);
+
+	subscriber.on('message', async (channel, data) => {
+		switch (channel.split(':')[1]) {
+			case 'user-stream':
+				connection.send(data);
+				break;
+			case 'post-stream':
+				const postId = channel.split(':')[2];
+				log(`RECEIVED: ${postId} ${data} by @${user.username}`);
+				const post = await serializePost(postId, user, {
+					detail: true
+				});
+				connection.send(JSON.stringify({
+					type: 'post-updated',
+					body: {
+						post: post
+					}
+				}));
+				break;
+		}
+	});
+
+	connection.on('message', data => {
+		const msg = JSON.parse(data.utf8Data);
+
+		switch (msg.type) {
+			case 'capture':
+					if (!msg.id) return;
+					const postId = msg.id;
+					log(`CAPTURE: ${postId} by @${user.username}`);
+					subscriber.subscribe(`misskey:post-stream:${postId}`);
+				break;
+		}
 	});
 }

src/api/streaming.ts

@@ -62,10 +62,6 @@ function authenticate(connection: websocket.connection, token: string): Promise<
 			const user = await User
 				.findOne({
 					token: token
-				}, {
-					fields: {
-						_id: true
-					}
 				});
 
 			resolve(user);