全sharedInbox配信の最適化

package.json

@@ -122,6 +122,7 @@
 		"parse5": "6.0.1",
 		"parsimmon": "1.18.1",
 		"pg": "8.8.0",
+		"pg-query-stream": "4.5.5",
 		"portscanner": "2.2.0",
 		"postcss": "8.4.31",
 		"postcss-loader": "7.0.2",

src/services/suspend-user.ts

@@ -4,31 +4,64 @@ import { deliver } from '../queue';
 import config from '../config';
 import { User } from '../models/entities/user';
 import { Users, Followings } from '../models';
-import { Not, IsNull } from 'typeorm';
+import { Not, IsNull, Brackets, SelectQueryBuilder } from 'typeorm';
+import { ReadStream } from 'typeorm/platform/PlatformTools';
 
 export async function doPostSuspend(user: User) {
 	if (Users.isLocalUser(user)) {
 		// 知り得る全SharedInboxにDelete配信
 		const content = renderActivity(renderDelete(`${config.url}/users/${user.id}`, user));
 
-		const queue: string[] = [];
+		const query = Followings.createQueryBuilder('following')
+			.select('distinct coalesce(following.followerSharedInbox, following.followeeSharedInbox) as inbox') 
+			.where(new Brackets((qb) =>
+				qb.where({ followerHost: Not(IsNull()) })
+				.orWhere({ followeeHost: Not(IsNull()) })
+			))
+			.andWhere(new Brackets((qb) => 
+				qb.where({ followerSharedInbox: Not(IsNull()) })
+				.orWhere({ followeeSharedInbox: Not(IsNull()) })
+			));
 
-		const followings = await Followings.find({
-			where: [
-				{ followerSharedInbox: Not(IsNull()) },
-				{ followeeSharedInbox: Not(IsNull()) }
-			],
-			select: ['followerSharedInbox', 'followeeSharedInbox']
-		});
+		/*
+		for (const row of await query.getRawMany()) {
+			deliver(user as any, content, row.inbox);
+		}
+		*/
 
-		const inboxes = followings.map(x => x.followerSharedInbox || x.followeeSharedInbox);
+		async function ProcessStreamingRows<T> (query: SelectQueryBuilder<T>, callback: (row: Record<string, unknown>) => Promise<void>) {
+			return new Promise(async (res, rej) => {
+				// query and get stream
+				let stream: ReadStream;
+				try {
+					stream = await query.stream();
+				} catch (e) {
+					return rej(e);
+				}
 
-		for (const inbox of inboxes) {
-			if (inbox != null && !queue.includes(inbox)) queue.push(inbox);
+				stream
+					.on('data', async (data: any) => {	// Buffer | string のはずだけどobjectが返ってくる
+						try {
+							await callback(data);
+						} catch (e) {
+							rej(e);
+						}
+					})
+					.on('end', () => res('end'))
+					.on('error', err => rej(err));
+			});
 		}
 
-		for (const inbox of queue) {
-			deliver(user as any, content, inbox);
-		}
+		await ProcessStreamingRows(query, async (row: Record<string, unknown>) => {
+			if (typeof row.inbox === 'string') {
+				try {
+					await deliver(user as any, content, row.inbox);
+				} catch (e) {
+					console.warn('mmm');
+				}
+			} else {
+				console.warn('nnn');
+			}
+		});
 	}
 }

src/services/unsuspend-user.ts

@@ -5,31 +5,26 @@ import { deliver } from '../queue';
 import config from '../config';
 import { User } from '../models/entities/user';
 import { Users, Followings } from '../models';
-import { Not, IsNull } from 'typeorm';
+import { Not, IsNull, Brackets } from 'typeorm';
 
 export async function doPostUnsuspend(user: User) {
 	if (Users.isLocalUser(user)) {
 		// 知り得る全SharedInboxにUndo Delete配信
 		const content = renderActivity(renderUndo(renderDelete(`${config.url}/users/${user.id}`, user), user));
 
-		const queue: string[] = [];
+		const query = Followings.createQueryBuilder('following')
+			.select('distinct coalesce(following.followerSharedInbox, following.followeeSharedInbox) as inbox') 
+			.where(new Brackets((qb) =>
+				qb.where({ followerHost: Not(IsNull()) })
+				.orWhere({ followeeHost: Not(IsNull()) })
+			))
+			.andWhere(new Brackets((qb) => 
+				qb.where({ followerSharedInbox: Not(IsNull()) })
+				.orWhere({ followeeSharedInbox: Not(IsNull()) })
+			));
 
-		const followings = await Followings.find({
-			where: [
-				{ followerSharedInbox: Not(IsNull()) },
-				{ followeeSharedInbox: Not(IsNull()) }
-			],
-			select: ['followerSharedInbox', 'followeeSharedInbox']
-		});
-
-		const inboxes = followings.map(x => x.followerSharedInbox || x.followeeSharedInbox);
-
-		for (const inbox of inboxes) {
-			if (inbox != null && !queue.includes(inbox)) queue.push(inbox);
-		}
-
-		for (const inbox of queue) {
-			deliver(user as any, content, inbox);
+		for (const row of await query.getRawMany()) {
+			deliver(user as any, content, row.inbox);
 		}
 	}
 }