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

* DeliverManager, note/create

* recipe

* followers delivers

* comment

* rename

* fix

* cleanup

src/remote/activitypub/deliver-manager.ts

@@ -0,0 +1,131 @@
+import { Users, Followings } from '../../models';
+import { ILocalUser, IRemoteUser } from '../../models/entities/user';
+import { deliver } from '../../queue';
+
+//#region types
+interface IRecipe {
+	type: string;
+}
+
+interface IFollowersRecipe extends IRecipe {
+	type: 'Followers';
+}
+
+interface IDirectRecipe extends IRecipe {
+	type: 'Direct';
+	to: IRemoteUser;
+}
+
+const isFollowers = (recipe: any): recipe is IFollowersRecipe =>
+	recipe.type === 'Followers';
+
+const isDirect = (recipe: any): recipe is IDirectRecipe =>
+	recipe.type === 'Direct';
+//#endregion
+
+export default class DeliverManager {
+	private actor: ILocalUser;
+	private activity: any;
+	private recipes: IRecipe[] = [];
+
+	/**
+	 * Constructor
+	 * @param actor Actor
+	 * @param activity Activity to deliver
+	 */
+	constructor(actor: ILocalUser, activity: any) {
+		this.actor = actor;
+		this.activity = activity;
+	}
+
+	/**
+	 * Add recipe for followers deliver
+	 */
+	public addFollowersRecipe() {
+		const deliver = {
+			type: 'Followers'
+		} as IFollowersRecipe;
+
+		this.addRecipe(deliver);
+	}
+
+	/**
+	 * Add recipe for direct deliver
+	 * @param to To
+	 */
+	public addDirectRecipe(to: IRemoteUser) {
+		const recipe = {
+			type: 'Direct',
+			to
+		} as IDirectRecipe;
+
+		this.addRecipe(recipe);
+	}
+
+	/**
+	 * Add recipe
+	 * @param recipe Recipe
+	 */
+	public addRecipe(recipe: IRecipe) {
+		this.recipes.push(recipe);
+	}
+
+	/**
+	 * Execute delivers
+	 */
+	public async execute() {
+		if (!Users.isLocalUser(this.actor)) return;
+
+		const inboxes: string[] = [];
+
+		// build inbox list
+		for (const recipe of this.recipes) {
+			if (isFollowers(recipe)) {
+				// followers deliver
+				const followers = await Followings.find({
+					followeeId: this.actor.id
+				});
+
+				for (const following of followers) {
+					if (Followings.isRemoteFollower(following)) {
+						const inbox = following.followerSharedInbox || following.followerInbox;
+						if (!inboxes.includes(inbox)) inboxes.push(inbox);
+					}
+				}
+			} else if (isDirect(recipe)) {
+				// direct deliver
+				const inbox = recipe.to.inbox;
+				if (inbox && !inboxes.includes(inbox)) inboxes.push(inbox);
+			}
+		}
+
+		// deliver
+		for (const inbox of inboxes) {
+			deliver(this.actor, this.activity, inbox);
+		}
+	}
+}
+
+//#region Utilities
+/**
+ * Deliver activity to followers
+ * @param activity Activity
+ * @param from Followee
+ */
+export async function deliverToFollowers(actor: ILocalUser, activity: any) {
+	const manager = new DeliverManager(actor, activity);
+	manager.addFollowersRecipe();
+	await manager.execute();
+}
+
+/**
+ * Deliver activity to user
+ * @param activity Activity
+ * @param to Target user
+ */
+export async function deliverToUser(actor: ILocalUser, activity: any, to: IRemoteUser) {
+	const manager = new DeliverManager(actor, activity);
+	manager.addDirectRecipe(to);
+	await manager.execute();
+}
+//#endregion

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);
+	deliverToFollowers(user, content);
-	}
-}
-
-/**
- * ローカルユーザーのリモートフォロワーの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;
 }

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 { Users } from '../../models';
-import { Followings, 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) {
+		const content = renderActivity(renderUpdate(await renderPerson(user), user));
-			if (Followings.isRemoteFollower(following)) {
+		deliverToFollowers(user, content);
-				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);
-			}
-		}
 	}
 }

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, {

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[] = [];
+			deliverToFollowers(user, content);
-
-			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);
-			}
 		}
 		//#endregion
 

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));
-			const content = renderActivity(renderUpdate(await renderNote(note, false), user));
+		deliverToFollowers(user, content);
-			for (const inbox of queue) {
-				deliver(user, content, inbox);
-			}
-		}
 	}
 }