無駄なAP deliverをしないように (#5589)

* DeliverManager, note/create

* recipe

* followers delivers

* comment

* rename

* fix

* cleanup

5 files changed, 50 insertions, 152 deletions

diff --git a/src/services/i/pin.ts b/src/services/i/pin.ts
index a6d2dfcdbf..9fd7263ffc 100644
--- a/src/services/i/pin.ts
+++ b/src/services/i/pin.ts
@@ -2,13 +2,13 @@ import config from '../../config';
 import renderAdd from '../../remote/activitypub/renderer/add';
 import renderRemove from '../../remote/activitypub/renderer/remove';
 import { renderActivity } from '../../remote/activitypub/renderer';
-import { deliver } from '../../queue';
 import { IdentifiableError } from '../../misc/identifiable-error';
-import { User, ILocalUser } from '../../models/entities/user';
+import { User } from '../../models/entities/user';
 import { Note } from '../../models/entities/note';
-import { Notes, UserNotePinings, Users, Followings } from '../../models';
+import { Notes, UserNotePinings, Users } from '../../models';
 import { UserNotePining } from '../../models/entities/user-note-pinings';
 import { genId } from '../../misc/gen-id';
+import { deliverToFollowers } from '../../remote/activitypub/deliver-manager';
 
 /**
  * 指定した投稿をピン留めします
@@ -82,36 +82,9 @@ export async function deliverPinnedChange(userId: User['id'], noteId: Note['id']
 
 	if (!Users.isLocalUser(user)) return;
 
-	const queue = await CreateRemoteInboxes(user);
-
-	if (queue.length < 1) return;
-
 	const target = `${config.url}/users/${user.id}/collections/featured`;
-
 	const item = `${config.url}/notes/${noteId}`;
 	const content = renderActivity(isAddition ? renderAdd(user, target, item) : renderRemove(user, target, item));
-	for (const inbox of queue) {
-		deliver(user, content, inbox);
-	}
-}
-
-/**
- * ローカルユーザーのリモートフォロワーのinboxリストを作成する
- * @param user ローカルユーザー
- */
-async function CreateRemoteInboxes(user: ILocalUser): Promise<string[]> {
-	const followers = await Followings.find({
-		followeeId: user.id
-	});
-
-	const queue: string[] = [];
-
-	for (const following of followers) {
-		if (Followings.isRemoteFollower(following)) {
-			const inbox = following.followerSharedInbox || following.followerInbox;
-			if (!queue.includes(inbox)) queue.push(inbox);
-		}
-	}
 
-	return queue;
+	deliverToFollowers(user, content);
 }
diff --git a/src/services/i/update.ts b/src/services/i/update.ts
index ddb6704a03..ae72e91345 100644
--- a/src/services/i/update.ts
+++ b/src/services/i/update.ts
@@ -1,34 +1,17 @@
 import renderUpdate from '../../remote/activitypub/renderer/update';
 import { renderActivity } from '../../remote/activitypub/renderer';
-import { deliver } from '../../queue';
-import { Followings, Users } from '../../models';
+import { Users } from '../../models';
 import { User } from '../../models/entities/user';
 import { renderPerson } from '../../remote/activitypub/renderer/person';
+import { deliverToFollowers } from '../../remote/activitypub/deliver-manager';
 
 export async function publishToFollowers(userId: User['id']) {
 	const user = await Users.findOne(userId);
 	if (user == null) throw new Error('user not found');
 
-	const followers = await Followings.find({
-		followeeId: user.id
-	});
-
-	const queue: string[] = [];
-
 	// フォロワーがリモートユーザーかつ投稿者がローカルユーザーならUpdateを配信
 	if (Users.isLocalUser(user)) {
-		for (const following of followers) {
-			if (Followings.isRemoteFollower(following)) {
-				const inbox = following.followerSharedInbox || following.followerInbox;
-				if (!queue.includes(inbox)) queue.push(inbox);
-			}
-		}
-
-		if (queue.length > 0) {
-			const content = renderActivity(renderUpdate(await renderPerson(user), user));
-			for (const inbox of queue) {
-				deliver(user, content, inbox);
-			}
-		}
+		const content = renderActivity(renderUpdate(await renderPerson(user), user));
+		deliverToFollowers(user, content);
 	}
 }
diff --git a/src/services/note/create.ts b/src/services/note/create.ts
index 5d9492e6ab..09deb0c53a 100644
--- a/src/services/note/create.ts
+++ b/src/services/note/create.ts
@@ -1,6 +1,6 @@
 import es from '../../db/elasticsearch';
 import { publishMainStream, publishNotesStream } from '../stream';
-import { deliver } from '../../queue';
+import DeliverManager from '../../remote/activitypub/deliver-manager';
 import renderNote from '../../remote/activitypub/renderer/note';
 import renderCreate from '../../remote/activitypub/renderer/create';
 import renderAnnounce from '../../remote/activitypub/renderer/announce';
@@ -17,7 +17,7 @@ import extractMentions from '../../misc/extract-mentions';
 import extractEmojis from '../../misc/extract-emojis';
 import extractHashtags from '../../misc/extract-hashtags';
 import { Note, IMentionedRemoteUsers } from '../../models/entities/note';
-import { Mutings, Users, NoteWatchings, Followings, Notes, Instances, UserProfiles } from '../../models';
+import { Mutings, Users, NoteWatchings, Notes, Instances, UserProfiles } from '../../models';
 import { DriveFile } from '../../models/entities/drive-file';
 import { App } from '../../models/entities/app';
 import { Not, getConnection, In } from 'typeorm';
@@ -246,12 +246,6 @@ export default async (user: User, data: Option, silent = false) => new Promise<N
 
 		createMentionedEvents(mentionedUsers, note, nm);
 
-		const noteActivity = await renderNoteOrRenoteActivity(data, note);
-
-		if (Users.isLocalUser(user)) {
-			deliverNoteToMentionedRemoteUsers(mentionedUsers, user, noteActivity);
-		}
-
 		const profile = await UserProfiles.findOne(user.id).then(ensure);
 
 		// If has in reply to note
@@ -294,11 +288,42 @@ export default async (user: User, data: Option, silent = false) => new Promise<N
 			}
 		}
 
-		publish(user, note, data.reply, data.renote, noteActivity);
-
 		Promise.all(nmRelatedPromises).then(() => {
 			nm.deliver();
 		});
+
+		//#region AP deliver
+		if (Users.isLocalUser(user)) {
+			(async () => {
+				const noteActivity = await renderNoteOrRenoteActivity(data, note);
+				const dm = new DeliverManager(user, noteActivity);
+
+				// メンションされたリモートユーザーに配送
+				for (const u of mentionedUsers.filter(u => Users.isRemoteUser(u))) {
+					dm.addDirectRecipe(u as IRemoteUser);
+				}
+
+				// 投稿がリプライかつ投稿者がローカルユーザーかつリプライ先の投稿の投稿者がリモートユーザーなら配送
+				if (data.reply && data.reply.userHost !== null) {
+					const u = await Users.findOne(data.reply.userId);
+					if (u && Users.isRemoteUser(u)) dm.addDirectRecipe(u);
+				}
+
+				// 投稿がRenoteかつ投稿者がローカルユーザーかつRenote元の投稿の投稿者がリモートユーザーなら配送
+				if (data.renote && data.renote.userHost !== null) {
+					const u = await Users.findOne(data.renote.userId);
+					if (u && Users.isRemoteUser(u)) dm.addDirectRecipe(u);
+				}
+
+				// フォロワーに配送
+				if (['public', 'home', 'followers'].includes(note.visibility)) {
+					dm.addFollowersRecipe();
+				}
+
+				dm.execute();
+			})();
+		}
+		//#endregion
 	}
 
 	// Register to search database
@@ -320,29 +345,6 @@ function incRenoteCount(renote: Note) {
 	Notes.increment({ id: renote.id }, 'score', 1);
 }
 
-async function publish(user: User, note: Note, reply: Note | null | undefined, renote: Note | null | undefined, noteActivity: any) {
-	if (Users.isLocalUser(user)) {
-		// 投稿がリプライかつ投稿者がローカルユーザーかつリプライ先の投稿の投稿者がリモートユーザーなら配送
-		if (reply && reply.userHost !== null) {
-			Users.findOne(reply.userId).then(ensure).then(u => {
-				deliver(user, noteActivity, u.inbox);
-			});
-		}
-
-		// 投稿がRenoteかつ投稿者がローカルユーザーかつRenote元の投稿の投稿者がリモートユーザーなら配送
-		if (renote && renote.userHost !== null) {
-			Users.findOne(renote.userId).then(ensure).then(u => {
-				deliver(user, noteActivity, u.inbox);
-			});
-		}
-	}
-
-	if (['public', 'home', 'followers'].includes(note.visibility)) {
-		// フォロワーに配信
-		publishToFollowers(note, user, noteActivity);
-	}
-}
-
 async function insertNote(user: User, data: Option, tags: string[], emojis: string[], mentionedUsers: User[]) {
 	const insert = new Note({
 		id: genId(data.createdAt!),
@@ -472,34 +474,6 @@ async function notifyToWatchersOfReplyee(reply: Note, user: User, nm: Notificati
 	}
 }
 
-async function publishToFollowers(note: Note, user: User, noteActivity: any) {
-	const followers = await Followings.find({
-		followeeId: note.userId
-	});
-
-	const queue: string[] = [];
-
-	for (const following of followers) {
-		if (Followings.isRemoteFollower(following)) {
-			// フォロワーがリモートユーザーかつ投稿者がローカルユーザーなら投稿を配信
-			if (Users.isLocalUser(user)) {
-				const inbox = following.followerSharedInbox || following.followerInbox;
-				if (!queue.includes(inbox)) queue.push(inbox);
-			}
-		}
-	}
-
-	for (const inbox of queue) {
-		deliver(user as any, noteActivity, inbox);
-	}
-}
-
-function deliverNoteToMentionedRemoteUsers(mentionedUsers: User[], user: ILocalUser, noteActivity: any) {
-	for (const u of mentionedUsers.filter(u => Users.isRemoteUser(u))) {
-		deliver(user, noteActivity, (u as IRemoteUser).inbox);
-	}
-}
-
 async function createMentionedEvents(mentionedUsers: User[], note: Note, nm: NotificationManager) {
 	for (const u of mentionedUsers.filter(u => Users.isLocalUser(u))) {
 		const detailPackedNote = await Notes.pack(note, u, {
diff --git a/src/services/note/delete.ts b/src/services/note/delete.ts
index 0f087de420..73cb0f4117 100644
--- a/src/services/note/delete.ts
+++ b/src/services/note/delete.ts
@@ -3,14 +3,14 @@ import renderDelete from '../../remote/activitypub/renderer/delete';
 import renderAnnounce from '../../remote/activitypub/renderer/announce';
 import renderUndo from '../../remote/activitypub/renderer/undo';
 import { renderActivity } from '../../remote/activitypub/renderer';
-import { deliver } from '../../queue';
 import renderTombstone from '../../remote/activitypub/renderer/tombstone';
 import config from '../../config';
 import { registerOrFetchInstanceDoc } from '../register-or-fetch-instance-doc';
 import { User } from '../../models/entities/user';
 import { Note } from '../../models/entities/note';
-import { Notes, Users, Followings, Instances } from '../../models';
+import { Notes, Users, Instances } from '../../models';
 import { notesChart, perUserNotesChart, instanceChart } from '../chart';
+import { deliverToFollowers } from '../../remote/activitypub/deliver-manager';
 
 /**
  * 投稿を削除します。
@@ -49,22 +49,7 @@ export default async function(user: User, note: Note, quiet = false) {
 				? renderUndo(renderAnnounce(renote.uri || `${config.url}/notes/${renote.id}`, note), user)
 				: renderDelete(renderTombstone(`${config.url}/notes/${note.id}`), user));
 
-			const queue: string[] = [];
-
-			const followers = await Followings.find({
-				followeeId: note.userId
-			});
-
-			for (const following of followers) {
-				if (Followings.isRemoteFollower(following)) {
-					const inbox = following.followerSharedInbox || following.followerInbox;
-					if (!queue.includes(inbox)) queue.push(inbox);
-				}
-			}
-
-			for (const inbox of queue) {
-				deliver(user as any, content, inbox);
-			}
+			deliverToFollowers(user, content);
 		}
 		//#endregion
 
diff --git a/src/services/note/polls/update.ts b/src/services/note/polls/update.ts
index f979ef2f0a..c076d13040 100644
--- a/src/services/note/polls/update.ts
+++ b/src/services/note/polls/update.ts
@@ -1,9 +1,9 @@
 import renderUpdate from '../../../remote/activitypub/renderer/update';
 import { renderActivity } from '../../../remote/activitypub/renderer';
-import { deliver } from '../../../queue';
 import renderNote from '../../../remote/activitypub/renderer/note';
-import { Users, Notes, Followings } from '../../../models';
+import { Users, Notes } from '../../../models';
 import { Note } from '../../../models/entities/note';
+import { deliverToFollowers } from '../../../remote/activitypub/deliver-manager';
 
 export async function deliverQuestionUpdate(noteId: Note['id']) {
 	const note = await Notes.findOne(noteId);
@@ -12,26 +12,9 @@ export async function deliverQuestionUpdate(noteId: Note['id']) {
 	const user = await Users.findOne(note.userId);
 	if (user == null) throw new Error('note not found');
 
-	const followers = await Followings.find({
-		followeeId: user.id
-	});
-
-	const queue: string[] = [];
-
-	// フォロワーがリモートユーザーかつ投稿者がローカルユーザーならUpdateを配信
 	if (Users.isLocalUser(user)) {
-		for (const following of followers) {
-			if (Followings.isRemoteFollower(following)) {
-				const inbox = following.followerSharedInbox || following.followerInbox;
-				if (!queue.includes(inbox)) queue.push(inbox);
-			}
-		}
 
-		if (queue.length > 0) {
-			const content = renderActivity(renderUpdate(await renderNote(note, false), user));
-			for (const inbox of queue) {
-				deliver(user, content, inbox);
-			}
-		}
+		const content = renderActivity(renderUpdate(await renderNote(note, false), user));
+		deliverToFollowers(user, content);
 	}
 }