perf: 各ストリーミング接続ごとにポーリングしないように

5 files changed, 29 insertions, 9 deletions

diff --git a/src/services/blocking/create.ts b/src/services/blocking/create.ts
index def4f33585..4f0238db91 100644
--- a/src/services/blocking/create.ts
+++ b/src/services/blocking/create.ts
@@ -1,4 +1,4 @@
-import { publishMainStream } from '../stream';
+import { publishMainStream, publishUserEvent } from '../stream';
 import { renderActivity } from '../../remote/activitypub/renderer';
 import renderFollow from '../../remote/activitypub/renderer/follow';
 import renderUndo from '../../remote/activitypub/renderer/undo';
@@ -55,7 +55,10 @@ async function cancelRequest(follower: User, followee: User) {
 	if (Users.isLocalUser(follower)) {
 		Users.pack(followee, follower, {
 			detail: true
-		}).then(packed => publishMainStream(follower.id, 'unfollow', packed));
+		}).then(packed => {
+			publishUserEvent(follower.id, 'unfollow', packed);
+			publishMainStream(follower.id, 'unfollow', packed);
+		});
 	}
 
 	// リモートにフォローリクエストをしていたらUndoFollow送信
@@ -97,7 +100,10 @@ async function unFollow(follower: User, followee: User) {
 	if (Users.isLocalUser(follower)) {
 		Users.pack(followee, follower, {
 			detail: true
-		}).then(packed => publishMainStream(follower.id, 'unfollow', packed));
+		}).then(packed => {
+			publishUserEvent(follower.id, 'unfollow', packed);
+			publishMainStream(follower.id, 'unfollow', packed);
+		});
 	}
 
 	// リモートにフォローをしていたらUndoFollow送信
diff --git a/src/services/following/create.ts b/src/services/following/create.ts
index 6bc98aee87..eb6699b0bf 100644
--- a/src/services/following/create.ts
+++ b/src/services/following/create.ts
@@ -1,4 +1,4 @@
-import { publishMainStream } from '../stream';
+import { publishMainStream, publishUserEvent } from '../stream';
 import { renderActivity } from '../../remote/activitypub/renderer';
 import renderFollow from '../../remote/activitypub/renderer/follow';
 import renderAccept from '../../remote/activitypub/renderer/accept';
@@ -88,7 +88,10 @@ export async function insertFollowingDoc(followee: User, follower: User) {
 	if (Users.isLocalUser(follower)) {
 		Users.pack(followee, follower, {
 			detail: true
-		}).then(packed => publishMainStream(follower.id, 'follow', packed));
+		}).then(packed => {
+			publishUserEvent(follower.id, 'follow', packed);
+			publishMainStream(follower.id, 'follow', packed);
+		});
 	}
 
 	// Publish followed event
diff --git a/src/services/following/delete.ts b/src/services/following/delete.ts
index 8821611515..32c47ea7f4 100644
--- a/src/services/following/delete.ts
+++ b/src/services/following/delete.ts
@@ -1,4 +1,4 @@
-import { publishMainStream } from '../stream';
+import { publishMainStream, publishUserEvent } from '../stream';
 import { renderActivity } from '../../remote/activitypub/renderer';
 import renderFollow from '../../remote/activitypub/renderer/follow';
 import renderUndo from '../../remote/activitypub/renderer/undo';
@@ -30,7 +30,10 @@ export default async function(follower: User, followee: User, silent = false) {
 	if (!silent && Users.isLocalUser(follower)) {
 		Users.pack(followee, follower, {
 			detail: true
-		}).then(packed => publishMainStream(follower.id, 'unfollow', packed));
+		}).then(packed => {
+			publishUserEvent(follower.id, 'unfollow', packed);
+			publishMainStream(follower.id, 'unfollow', packed);
+		});
 	}
 
 	if (Users.isLocalUser(follower) && Users.isRemoteUser(followee)) {
diff --git a/src/services/following/requests/reject.ts b/src/services/following/requests/reject.ts
index 9a8b14bbfd..d8d3788088 100644
--- a/src/services/following/requests/reject.ts
+++ b/src/services/following/requests/reject.ts
@@ -2,7 +2,7 @@ import { renderActivity } from '../../../remote/activitypub/renderer';
 import renderFollow from '../../../remote/activitypub/renderer/follow';
 import renderReject from '../../../remote/activitypub/renderer/reject';
 import { deliver } from '../../../queue';
-import { publishMainStream } from '../../stream';
+import { publishMainStream, publishUserEvent } from '../../stream';
 import { User, ILocalUser } from '../../../models/entities/user';
 import { Users, FollowRequests, Followings } from '../../../models';
 import { decrementFollowing } from '../delete';
@@ -39,5 +39,8 @@ export default async function(followee: User, follower: User) {
 
 	Users.pack(followee, follower, {
 		detail: true
-	}).then(packed => publishMainStream(follower.id, 'unfollow', packed));
+	}).then(packed => {
+		publishUserEvent(follower.id, 'unfollow', packed);
+		publishMainStream(follower.id, 'unfollow', packed);
+	});
 }
diff --git a/src/services/stream.ts b/src/services/stream.ts
index d833d700fe..75385847ce 100644
--- a/src/services/stream.ts
+++ b/src/services/stream.ts
@@ -20,6 +20,10 @@ class Publisher {
 		}));
 	}
 
+	public publishUserEvent = (userId: User['id'], type: string, value?: any): void => {
+		this.publish(`user:${userId}`, type, typeof value === 'undefined' ? null : value);
+	}
+
 	public publishBroadcastStream = (type: string, value?: any): void => {
 		this.publish('broadcast', type, typeof value === 'undefined' ? null : value);
 	}
@@ -84,6 +88,7 @@ const publisher = new Publisher();
 
 export default publisher;
 
+export const publishUserEvent = publisher.publishUserEvent;
 export const publishBroadcastStream = publisher.publishBroadcastStream;
 export const publishMainStream = publisher.publishMainStream;
 export const publishDriveStream = publisher.publishDriveStream;