feat(api): allow bots to search messages within a single community (#1660)

fluxer_api/src/api/guild/services/GuildSearchService.ts

@@ -24,8 +24,13 @@ import {MessageSearchResponseMapper} from '../../search/MessageSearchResponseMap
 import {searchExistingMessages} from '../../search/MessageSearchResultReconciler';
 import type {IUserRepository} from '../../user/IUserRepository';
 import {canUserAccessNsfwContent} from '../../utils/AgeUtils';
+import {mapWithConcurrency} from '../../utils/ConcurrencyUtils';
 import type {WorkerTaskName} from '../../worker/WorkerLaneConfig';
 
+const GUILD_FANOUT_CONCURRENCY = 16;
+const PERMISSION_CHECK_CONCURRENCY = 64;
+const CHANNEL_INDEX_JOB_ENQUEUE_CONCURRENCY = 16;
+
 export class GuildSearchService {
 	private readonly responseMapper: MessageSearchResponseMapper;
 
@@ -95,15 +100,13 @@ export class GuildSearchService {
 			const channel = channelMap.get(id.toString())!;
 			return !(channel.isNsfw && !canIncludeNsfw);
 		});
-		const permissionResults = await Promise.all(
-			nsfwFilteredIds.map((channelId) =>
-				this.gatewayService.checkPermission({
-					guildId,
-					userId,
-					channelId,
-					permission: Permissions.VIEW_CHANNEL | Permissions.READ_MESSAGE_HISTORY,
-				}),
-			),
+		const permissionResults = await mapWithConcurrency(nsfwFilteredIds, PERMISSION_CHECK_CONCURRENCY, (channelId) =>
+			this.gatewayService.checkPermission({
+				guildId,
+				userId,
+				channelId,
+				permission: Permissions.VIEW_CHANNEL | Permissions.READ_MESSAGE_HISTORY,
+			}),
 		);
 		const validChannelIds: Array<ChannelID> = [];
 		for (let i = 0; i < nsfwFilteredIds.length; i++) {
@@ -133,18 +136,16 @@ export class GuildSearchService {
 			})
 			.map((id) => id.toString());
 		if (channelsToIndex.length > 0) {
-			await Promise.all(
-				channelsToIndex.map((channelId) =>
-					this.workerService.addJob(
-						'indexChannelMessages',
-						{
-							channelId,
-						},
-						{
-							jobKey: `indexChannelMessages-${channelId}`,
-							maxAttempts: 3,
-						},
-					),
+			await mapWithConcurrency(channelsToIndex, CHANNEL_INDEX_JOB_ENQUEUE_CONCURRENCY, (channelId) =>
+				this.workerService.addJob(
+					'indexChannelMessages',
+					{
+						channelId,
+					},
+					{
+						jobKey: `indexChannelMessages-${channelId}`,
+						maxAttempts: 3,
+					},
 				),
 			);
 			return {indexing: true};
@@ -267,16 +268,14 @@ export class GuildSearchService {
 	}
 
 	private async queueIndexingChannels(channelIds: Iterable<string>): Promise<void> {
-		await Promise.all(
-			Array.from(channelIds).map((channelId) =>
-				this.workerService.addJob(
-					'indexChannelMessages',
-					{channelId},
-					{
-						jobKey: `indexChannelMessages-${channelId}`,
-						maxAttempts: 3,
-					},
-				),
+		await mapWithConcurrency(Array.from(channelIds), CHANNEL_INDEX_JOB_ENQUEUE_CONCURRENCY, (channelId) =>
+			this.workerService.addJob(
+				'indexChannelMessages',
+				{channelId},
+				{
+					jobKey: `indexChannelMessages-${channelId}`,
+					maxAttempts: 3,
+				},
 			),
 		);
 	}
@@ -290,42 +289,36 @@ export class GuildSearchService {
 		const accessibleChannels = new Map<string, Channel>();
 		const unindexedChannelIds = new Set<string>();
 		const guildNsfwLevels = new Map<string, number>();
-		const [guildDataResults, guildChannelsResults, viewableChannelsResults] = await Promise.all([
-			Promise.all(guildIds.map((guildId) => this.gatewayService.getGuildData({guildId, userId}))),
-			Promise.all(guildIds.map((guildId) => this.channelRepository.listGuildChannels(guildId))),
-			Promise.all(guildIds.map((guildId) => this.gatewayService.getViewableChannels({guildId, userId}))),
-		]);
-		for (let i = 0; i < guildIds.length; i++) {
-			const guildData = guildDataResults[i];
-			if (guildData) {
-				guildNsfwLevels.set(guildIds[i]!.toString(), guildData.nsfw_level);
-			}
-		}
 		const permissionChecks: Array<{
 			channel: Channel;
 			guildId: GuildID;
 		}> = [];
-		for (let i = 0; i < guildIds.length; i++) {
-			const guildChannels = guildChannelsResults[i]!;
-			if (guildChannels.length === 0) {
-				continue;
+		await mapWithConcurrency(guildIds, GUILD_FANOUT_CONCURRENCY, async (guildId) => {
+			const [guildData, guildChannels, viewableChannels] = await Promise.all([
+				this.gatewayService.getGuildData({guildId, userId}),
+				this.channelRepository.listGuildChannels(guildId),
+				this.gatewayService.getViewableChannels({guildId, userId}),
+			]);
+			if (guildData) {
+				guildNsfwLevels.set(guildId.toString(), guildData.nsfw_level);
 			}
-			const viewableChannelIds = new Set(viewableChannelsResults[i]!.map((channelId) => channelId.toString()));
+			const viewableChannelIds = new Set(viewableChannels.map((channelId) => channelId.toString()));
 			for (const channel of guildChannels) {
 				if (viewableChannelIds.has(channel.id.toString())) {
-					permissionChecks.push({channel, guildId: guildIds[i]!});
+					permissionChecks.push({channel, guildId});
 				}
 			}
-		}
-		const permissionResults = await Promise.all(
-			permissionChecks.map(({channel, guildId}) =>
+		});
+		const permissionResults = await mapWithConcurrency(
+			permissionChecks,
+			PERMISSION_CHECK_CONCURRENCY,
+			({channel, guildId}) =>
 				this.gatewayService.checkPermission({
 					guildId,
 					userId,
 					channelId: channel.id,
 					permission: Permissions.VIEW_CHANNEL | Permissions.READ_MESSAGE_HISTORY,
 				}),
-			),
 		);
 		for (let i = 0; i < permissionChecks.length; i++) {
 			if (!permissionResults[i]) {

fluxer_api/src/api/openapi/openapi.json

@@ -14978,7 +14978,7 @@
 					}
 				},
 				"x-mint": {"metadata": {"title": "Search messages"}},
-				"description": "Searches for messages across guilds and channels accessible to the authenticated user.",
+				"description": "Searches for messages across guilds and channels accessible to the authenticated user. Bots must supply a context guild or channel and may only use the current scope.",
 				"security": [{"sessionToken": []}],
 				"requestBody": {
 					"required": true,

fluxer_api/src/api/search/GlobalSearchService.ts

@@ -23,6 +23,7 @@ import type {Channel} from '../models/Channel';
 import {getMessageSearchService} from '../SearchFactory';
 import type {IUserRepository} from '../user/IUserRepository';
 import {canUserAccessNsfwContent} from '../utils/AgeUtils';
+import {mapWithConcurrency} from '../utils/ConcurrencyUtils';
 import type {WorkerTaskName} from '../worker/WorkerLaneConfig';
 import {buildMessageSearchFilters} from './BuildMessageSearchFilters';
 import {channelNeedsReindexing} from './ChannelIndexingUtils';
@@ -270,21 +271,3 @@ export class GlobalSearchService {
 		return canUserAccessNsfwContent(user);
 	}
 }
-
-async function mapWithConcurrency<T, TResult>(
-	items: ReadonlyArray<T>,
-	concurrency: number,
-	mapper: (item: T, index: number) => Promise<TResult>,
-): Promise<Array<TResult>> {
-	const results = new Array<TResult>(items.length);
-	let nextIndex = 0;
-	async function worker(): Promise<void> {
-		for (;;) {
-			const index = nextIndex++;
-			if (index >= items.length) return;
-			results[index] = await mapper(items[index]!, index);
-		}
-	}
-	await Promise.all(Array.from({length: Math.min(concurrency, items.length)}, () => worker()));
-	return results;
-}

fluxer_api/src/api/search/controllers/SearchController.ts

@@ -1,8 +1,10 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import {ValidationErrorCodes} from '@fluxer/constants/src/ValidationErrorCodes';
+import {InputValidationError} from '@fluxer/errors/src/domains/core/InputValidationError';
 import {GlobalSearchMessagesRequest} from '@fluxer/schema/src/domains/message/MessageRequestSchemas';
 import {MessageSearchResponse} from '@fluxer/schema/src/domains/message/MessageResponseSchemas';
-import {DefaultUserOnly, LoginRequired} from '../../middleware/AuthMiddleware';
+import {LoginRequired} from '../../middleware/AuthMiddleware';
 import {RateLimitMiddleware} from '../../middleware/RateLimitMiddleware';
 import {OpenAPI} from '../../middleware/ResponseTypeMiddleware';
 import {RateLimitConfigs} from '../../RateLimitConfig';
@@ -14,11 +16,11 @@ export function SearchController(app: HonoApp) {
 		'/search/messages',
 		RateLimitMiddleware(RateLimitConfigs.SEARCH_MESSAGES),
 		LoginRequired,
-		DefaultUserOnly,
 		OpenAPI({
 			operationId: 'search_messages',
 			summary: 'Search messages',
-			description: 'Searches for messages across guilds and channels accessible to the authenticated user.',
+			description:
+				'Searches for messages across guilds and channels accessible to the authenticated user. Bots must supply a context guild or channel and may only use the current scope.',
 			responseSchema: MessageSearchResponse,
 			statusCode: 200,
 			security: ['bearerToken', 'sessionToken'],
@@ -27,7 +29,11 @@ export function SearchController(app: HonoApp) {
 		Validator('json', GlobalSearchMessagesRequest),
 		async (ctx) => {
 			const params = ctx.req.valid('json');
-			const userId = ctx.get('user').id;
+			const user = ctx.get('user');
+			if (user.isBot && (params.scope ?? 'current') !== 'current') {
+				throw InputValidationError.fromCode('scope', ValidationErrorCodes.BOT_SEARCH_SCOPE_UNAVAILABLE);
+			}
+			const userId = user.id;
 			const requestCache = ctx.get('requestCache');
 			const result = await ctx.get('searchService').searchMessages({userId, requestCache, data: params});
 			return ctx.json(result);

fluxer_api/src/api/utils/ConcurrencyUtils.ts

@@ -0,0 +1,19 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+export async function mapWithConcurrency<T, TResult>(
+	items: ReadonlyArray<T>,
+	concurrency: number,
+	mapper: (item: T, index: number) => Promise<TResult>,
+): Promise<Array<TResult>> {
+	const results = new Array<TResult>(items.length);
+	let nextIndex = 0;
+	async function worker(): Promise<void> {
+		for (;;) {
+			const index = nextIndex++;
+			if (index >= items.length) return;
+			results[index] = await mapper(items[index]!, index);
+		}
+	}
+	await Promise.all(Array.from({length: Math.min(concurrency, items.length)}, () => worker()));
+	return results;
+}