perf(guild): add a lean channel auth context rpc (#2226)

fluxer_api/src/api/channel/services/BaseChannelAuthService.ts

@@ -9,12 +9,11 @@ import {UnknownGuildError} from '@fluxer/errors/src/domains/guild/UnknownGuildEr
 import {NsfwContentRequiresAgeVerificationError} from '@fluxer/errors/src/domains/moderation/NsfwContentRequiresAgeVerificationError';
 import {UnknownUserError} from '@fluxer/errors/src/domains/user/UnknownUserError';
 import type {GuildMemberResponse} from '@fluxer/schema/src/domains/guild/GuildMemberSchemas';
-import type {GuildResponse} from '@fluxer/schema/src/domains/guild/GuildResponseSchemas';
 import type {ChannelID, GuildID, UserID} from '../../BrandedTypes';
 import {SYSTEM_USER_ID} from '../../constants/Core';
 import type {IGuildRepositoryAggregate} from '../../guild/repositories/IGuildRepositoryAggregate';
 import {createGuildMfaEnforcer} from '../../guild/services/GuildMfaEnforcement';
-import type {IGatewayService} from '../../infrastructure/IGatewayService';
+import type {GuildChannelAuthContext, IGatewayService} from '../../infrastructure/IGatewayService';
 import type {Channel} from '../../models/Channel';
 import type {GuildMember} from '../../models/GuildMember';
 import type {User} from '../../models/User';
@@ -173,13 +172,14 @@ export abstract class BaseChannelAuthService {
 		skipNsfwValidation?: boolean;
 	}): Promise<AuthenticatedChannel> {
 		const guildId = channel.guildId!;
-		const [guildDataResult, guildMemberResult] = await Promise.all([
-			this.fetchGuildDataOrThrow({guildId, userId}),
+		const [authContextResult, guildMemberResult] = await Promise.all([
+			this.fetchGuildAuthContextOrThrow({guildId, userId, channelId: this.parentLookupChannelId(channel)}),
 			this.gatewayService.getGuildMember({guildId, userId}),
 		]);
-		if (!guildDataResult) {
+		if (!authContextResult) {
 			this.throwGuildAccessError();
 		}
+		const guildDataResult = authContextResult.guild;
 		if (!guildMemberResult.success || !guildMemberResult.memberData) {
 			this.throwGuildAccessError();
 		}
@@ -190,7 +190,7 @@ export abstract class BaseChannelAuthService {
 		});
 		const enforceGuildMfa = await createGuildMfaEnforcer({
 			userRepository: this.userRepository,
-			guildData: guildDataResult!,
+			guildData: guildDataResult,
 			userId,
 		});
 		const channelPermissions = await this.gatewayService.getUserPermissions({
@@ -210,12 +210,12 @@ export abstract class BaseChannelAuthService {
 		await checkPermission(Permissions.VIEW_CHANNEL);
 		const parentCategory = await this.getParentCategoryContentWarningView({
 			channel,
-			guild: guildDataResult!,
+			parentChannel: authContextResult.parentChannel,
 		});
 		const requiresAgeVerification = computeEffectiveChannelNsfw(
 			channelToContentWarningView(channel),
 			parentCategory,
-			guildResponseToContentWarningView(guildDataResult!),
+			guildResponseToContentWarningView(guildDataResult),
 		);
 		if (
 			this.options.validateNsfw &&
@@ -233,27 +233,32 @@ export abstract class BaseChannelAuthService {
 		}
 		return {
 			channel,
-			guild: guildDataResult!,
+			guild: guildDataResult,
 			member,
 			hasPermission,
 			checkPermission,
 		};
 	}
 
+	private parentLookupChannelId(channel: Channel): ChannelID | undefined {
+		if (!channel.parentId || channel.type === ChannelTypes.GUILD_CATEGORY) {
+			return undefined;
+		}
+		return channel.parentId;
+	}
+
 	private async getParentCategoryContentWarningView({
 		channel,
-		guild,
+		parentChannel,
 	}: {
 		channel: Channel;
-		guild: GuildResponse;
+		parentChannel: GuildChannelAuthContext['parentChannel'];
 	}): Promise<ContentWarningChannelLike | null> {
 		if (!channel.parentId || channel.type === ChannelTypes.GUILD_CATEGORY) {
 			return null;
 		}
-		const parentId = channel.parentId.toString();
-		const parentFromGateway = guild.channels?.find((guildChannel) => guildChannel.id === parentId);
-		if (parentFromGateway) {
-			return channelResponseToContentWarningView(parentFromGateway);
+		if (parentChannel) {
+			return channelResponseToContentWarningView(parentChannel);
 		}
 		const parentCategory = await this.channelRepository.channelData.findUnique(channel.parentId);
 		return parentCategory ? channelToContentWarningView(parentCategory) : null;
@@ -266,10 +271,14 @@ export abstract class BaseChannelAuthService {
 		throw new UnknownChannelError();
 	}
 
-	private async fetchGuildDataOrThrow(params: {guildId: GuildID; userId: UserID}): Promise<GuildResponse | null> {
-		const {guildId, userId} = params;
+	private async fetchGuildAuthContextOrThrow(params: {
+		guildId: GuildID;
+		userId: UserID;
+		channelId?: ChannelID;
+	}): Promise<GuildChannelAuthContext | null> {
+		const {guildId, userId, channelId} = params;
 		try {
-			return await this.gatewayService.getGuildData({guildId, userId});
+			return await this.gatewayService.getGuildAuthContext({guildId, userId, channelId});
 		} catch (error) {
 			await this.handleGuildAccessError(error, guildId);
 			return null;

fluxer_api/src/api/channel/tests/MessageSendPermissions.test.ts

@@ -111,7 +111,7 @@ describe('Message send permissions', () => {
 		});
 		await ensureSessionStarted(harness, member.token);
 		const gatewayService = getGatewayService();
-		const getGuildData = vi.spyOn(gatewayService, 'getGuildData');
+		const getGuildAuthContext = vi.spyOn(gatewayService, 'getGuildAuthContext');
 		const getGuildMember = vi.spyOn(gatewayService, 'getGuildMember');
 		const getUserPermissions = vi.spyOn(gatewayService, 'getUserPermissions');
 
@@ -121,11 +121,11 @@ describe('Message send permissions', () => {
 				.body({content: 'authenticate me once'})
 				.execute();
 			const sendCounts = {
-				guildData: getGuildData.mock.calls.length,
+				authContext: getGuildAuthContext.mock.calls.length,
 				guildMember: getGuildMember.mock.calls.length,
 				userPermissions: getUserPermissions.mock.calls.length,
 			};
-			getGuildData.mockClear();
+			getGuildAuthContext.mockClear();
 			getGuildMember.mockClear();
 			getUserPermissions.mockClear();
 			await createBuilder<MessageResponse>(harness, member.token)
@@ -133,15 +133,15 @@ describe('Message send permissions', () => {
 				.body({content: 'authenticate me once again'})
 				.execute();
 			const editCounts = {
-				guildData: getGuildData.mock.calls.length,
+				authContext: getGuildAuthContext.mock.calls.length,
 				guildMember: getGuildMember.mock.calls.length,
 				userPermissions: getUserPermissions.mock.calls.length,
 			};
 
-			expect(sendCounts).toEqual({guildData: 1, guildMember: 1, userPermissions: 1});
-			expect(editCounts).toEqual({guildData: 1, guildMember: 1, userPermissions: 1});
+			expect(sendCounts).toEqual({authContext: 1, guildMember: 1, userPermissions: 1});
+			expect(editCounts).toEqual({authContext: 1, guildMember: 1, userPermissions: 1});
 		} finally {
-			getGuildData.mockRestore();
+			getGuildAuthContext.mockRestore();
 			getGuildMember.mockRestore();
 			getUserPermissions.mockRestore();
 		}

fluxer_api/src/api/infrastructure/GatewayService.ts

@@ -10,6 +10,7 @@ import {MissingPermissionsError} from '@fluxer/errors/src/domains/core/MissingPe
 import {ServiceUnavailableError} from '@fluxer/errors/src/domains/core/ServiceUnavailableError';
 import {UnknownGuildError} from '@fluxer/errors/src/domains/guild/UnknownGuildError';
 import {UserNotInVoiceError} from '@fluxer/errors/src/domains/user/UserNotInVoiceError';
+import type {ChannelResponse} from '@fluxer/schema/src/domains/channel/ChannelSchemas';
 import type {GuildMemberResponse} from '@fluxer/schema/src/domains/guild/GuildMemberSchemas';
 import type {GuildResponse} from '@fluxer/schema/src/domains/guild/GuildResponseSchemas';
 import {ms} from 'itty-time';
@@ -30,6 +31,7 @@ import type {
 	GatewayNodeStats,
 	GatewayVoiceStateCounts,
 	GatewayVoiceStateEntry,
+	GuildChannelAuthContext,
 } from './IGatewayService';
 
 const PUSH_BADGE_COUNT_BATCH_SIZE = 100;
@@ -82,6 +84,12 @@ interface GuildDataParams {
 	userId: UserID;
 }
 
+interface GuildAuthContextParams {
+	guildId: GuildID;
+	userId: UserID;
+	channelId?: ChannelID;
+}
+
 interface GuildMemberParams {
 	guildId: GuildID;
 	userId: UserID;
@@ -232,6 +240,11 @@ interface GuildMemberRpcResponse {
 	member_data?: GuildMemberResponse;
 }
 
+interface GuildAuthContextRpcResponse {
+	guild: GuildResponse;
+	parent_channel?: ChannelResponse | null;
+}
+
 type PendingRequest<T> = {
 	resolve: (value: T) => void;
 	reject: (error: Error) => void;
@@ -252,17 +265,22 @@ export class GatewayService {
 		>
 	>();
 	private pendingPermissionRequests = new Map<string, Array<PendingRequest<boolean>>>();
+	private pendingAuthContextRequests = new Map<string, Array<PendingRequest<GuildChannelAuthContext>>>();
 	private pendingBatchRequestCount = 0;
 	private guildDataBatchTimeout: NodeJS.Timeout | null = null;
 	private guildMemberBatchTimeout: NodeJS.Timeout | null = null;
 	private permissionBatchTimeout: NodeJS.Timeout | null = null;
+	private authContextBatchTimeout: NodeJS.Timeout | null = null;
 	private activeGuildDataRequests = 0;
 	private activeGuildMemberRequests = 0;
 	private activePermissionRequests = 0;
+	private activeAuthContextRequests = 0;
+	private authContextUnsupportedUntil = 0;
 	private readonly BATCH_DELAY_MS = ms('5 milliseconds');
 	private readonly MAX_PENDING_BATCH_REQUESTS = 2000;
 	private readonly MAX_BATCH_CONCURRENCY = 50;
 	private readonly PENDING_REQUEST_TIMEOUT_MS = ms('30 seconds');
+	private readonly AUTH_CONTEXT_FALLBACK_MS = ms('5 minutes');
 
 	constructor() {
 		this.rpcClient = GatewayRpcClient.getInstance();
@@ -376,6 +394,16 @@ export class GatewayService {
 		}, this.BATCH_DELAY_MS);
 	}
 
+	private scheduleAuthContextBatch(): void {
+		if (this.authContextBatchTimeout || this.activeAuthContextRequests >= this.MAX_BATCH_CONCURRENCY) {
+			return;
+		}
+		this.authContextBatchTimeout = setTimeout(() => {
+			this.authContextBatchTimeout = null;
+			this.processAuthContextQueue();
+		}, this.BATCH_DELAY_MS);
+	}
+
 	private takePendingEntries<T>(
 		requests: Map<string, Array<PendingRequest<T>>>,
 		limit: number,
@@ -482,12 +510,43 @@ export class GatewayService {
 		for (const pendingRequests of this.pendingPermissionRequests.values()) {
 			this.rejectPendingRequests(pendingRequests, error);
 		}
+		for (const pendingRequests of this.pendingAuthContextRequests.values()) {
+			this.rejectPendingRequests(pendingRequests, error);
+		}
 		this.pendingGuildDataRequests.clear();
 		this.pendingGuildMemberRequests.clear();
 		this.pendingPermissionRequests.clear();
+		this.pendingAuthContextRequests.clear();
 		this.pendingBatchRequestCount = 0;
 	}
 
+	private processAuthContextQueue(): void {
+		const availableSlots = this.MAX_BATCH_CONCURRENCY - this.activeAuthContextRequests;
+		if (availableSlots <= 0) {
+			return;
+		}
+		const entries = this.takePendingEntries(this.pendingAuthContextRequests, availableSlots);
+		if (entries.length === 0) {
+			return;
+		}
+		const totalAuthContextRequests = entries.reduce((sum, [, pending]) => sum + pending.length, 0);
+		Logger.debug(
+			`[gateway-batch] Processing guild.get_auth_context batch: ${entries.length} unique requests (${totalAuthContextRequests} total)`,
+		);
+		for (const entry of entries) {
+			this.activeAuthContextRequests += 1;
+			void this.processAuthContextEntry(entry).finally(() => {
+				this.activeAuthContextRequests = Math.max(0, this.activeAuthContextRequests - 1);
+				if (this.pendingAuthContextRequests.size > 0) {
+					this.scheduleAuthContextBatch();
+				}
+			});
+		}
+		if (this.pendingAuthContextRequests.size > 0) {
+			this.scheduleAuthContextBatch();
+		}
+	}
+
 	private async processGuildDataEntry([key, pending]: [string, Array<PendingRequest<GuildResponse>>]): Promise<void> {
 		try {
 			const [guildIdStr, userIdStr, skipCheck] = key.split('-');
@@ -559,6 +618,65 @@ export class GatewayService {
 		}
 	}
 
+	private async processAuthContextEntry([key, pending]: [
+		string,
+		Array<PendingRequest<GuildChannelAuthContext>>,
+	]): Promise<void> {
+		const [guildIdStr, userIdStr, channelIdStr] = key.split('-');
+		const guildId = BigInt(guildIdStr) as GuildID;
+		const userId = BigInt(userIdStr) as UserID;
+		const channelId = channelIdStr !== '0' ? (BigInt(channelIdStr) as ChannelID) : undefined;
+		try {
+			const result = await this.call<GuildAuthContextRpcResponse>('guild.get_auth_context', {
+				guild_id: guildId.toString(),
+				user_id: userId.toString(),
+				channel_id: channelId ? channelId.toString() : null,
+			});
+			this.resolvePendingRequests(pending, {
+				guild: result.guild,
+				parentChannel: result.parent_channel ?? null,
+			});
+		} catch (error) {
+			const transformedError = this.transformGatewayError(error);
+			if (!this.isAuthContextUnsupportedError(transformedError)) {
+				this.rejectPendingRequests(pending, transformedError);
+				this.logBatchFailures('guild.get_auth_context', [{status: 'rejected', reason: error}]);
+				return;
+			}
+			this.authContextUnsupportedUntil = Date.now() + this.AUTH_CONTEXT_FALLBACK_MS;
+			Logger.warn({error}, '[gateway-rpc] guild.get_auth_context unavailable, falling back to guild.get_data');
+			await this.settleAuthContextFromGuildData({guildId, userId, channelId}, pending);
+		}
+	}
+
+	private isAuthContextUnsupportedError(error: Error): boolean {
+		return error instanceof BadGatewayError || error instanceof GatewayTimeoutError;
+	}
+
+	private async settleAuthContextFromGuildData(
+		params: GuildAuthContextParams,
+		pending: Array<PendingRequest<GuildChannelAuthContext>>,
+	): Promise<void> {
+		try {
+			this.resolvePendingRequests(pending, await this.getGuildAuthContextFromGuildData(params));
+		} catch (error) {
+			const transformedError = this.transformGatewayError(error);
+			this.rejectPendingRequests(pending, transformedError);
+			this.logBatchFailures('guild.get_data', [{status: 'rejected', reason: error}]);
+		}
+	}
+
+	private async getGuildAuthContextFromGuildData({
+		guildId,
+		userId,
+		channelId,
+	}: GuildAuthContextParams): Promise<GuildChannelAuthContext> {
+		const guild = await this.getGuildData({guildId, userId});
+		const parentId = channelId?.toString();
+		const parentChannel = parentId ? (guild.channels?.find((channel) => channel.id === parentId) ?? null) : null;
+		return {guild, parentChannel};
+	}
+
 	async dispatchGuild({guildId, event, data}: DispatchGuildParams): Promise<void> {
 		await this.call('guild.dispatch', {
 			guild_id: guildId.toString(),
@@ -713,6 +831,48 @@ export class GatewayService {
 		});
 	}
 
+	async getGuildAuthContext({guildId, userId, channelId}: GuildAuthContextParams): Promise<GuildChannelAuthContext> {
+		if (Date.now() < this.authContextUnsupportedUntil) {
+			return await this.getGuildAuthContextFromGuildData({guildId, userId, channelId});
+		}
+		const key = `${guildId.toString()}-${userId.toString()}-${channelId ? channelId.toString() : '0'}`;
+		return new Promise<GuildChannelAuthContext>((resolve, reject) => {
+			if (this.pendingBatchRequestCount >= this.MAX_PENDING_BATCH_REQUESTS) {
+				reject(new ServiceUnavailableError());
+				return;
+			}
+			const pendingRequest: PendingRequest<GuildChannelAuthContext> = {
+				resolve,
+				reject,
+				settled: false,
+				timeoutId: null,
+			};
+			pendingRequest.timeoutId = setTimeout(() => {
+				const error = new GatewayTimeoutError();
+				this.rejectPendingRequests([pendingRequest], error);
+				this.removePendingAuthContextRequest(key, pendingRequest);
+			}, this.PENDING_REQUEST_TIMEOUT_MS);
+			const pending = this.pendingAuthContextRequests.get(key) || [];
+			pending.push(pendingRequest);
+			this.pendingAuthContextRequests.set(key, pending);
+			this.pendingBatchRequestCount += 1;
+			this.scheduleAuthContextBatch();
+		});
+	}
+
+	private removePendingAuthContextRequest(key: string, request: PendingRequest<GuildChannelAuthContext>): void {
+		const pending = this.pendingAuthContextRequests.get(key);
+		if (pending) {
+			const index = pending.indexOf(request);
+			if (index >= 0) {
+				pending.splice(index, 1);
+				if (pending.length === 0) {
+					this.pendingAuthContextRequests.delete(key);
+				}
+			}
+		}
+	}
+
 	private removePendingGuildDataRequest(key: string, request: PendingRequest<GuildResponse>): void {
 		const pending = this.pendingGuildDataRequests.get(key);
 		if (pending) {
@@ -1649,6 +1809,10 @@ export class GatewayService {
 			clearTimeout(this.permissionBatchTimeout);
 			this.permissionBatchTimeout = null;
 		}
+		if (this.authContextBatchTimeout) {
+			clearTimeout(this.authContextBatchTimeout);
+			this.authContextBatchTimeout = null;
+		}
 		this.rejectAllPendingBatchRequests(new ServiceUnavailableError());
 	}
 }

fluxer_api/src/api/infrastructure/IGatewayService.ts

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import type {ChannelResponse} from '@fluxer/schema/src/domains/channel/ChannelSchemas';
 import type {GuildMemberResponse} from '@fluxer/schema/src/domains/guild/GuildMemberSchemas';
 import type {GuildResponse} from '@fluxer/schema/src/domains/guild/GuildResponseSchemas';
 import type {ChannelID, GuildID, MessageID, RoleID, UserID} from '../BrandedTypes';
@@ -106,6 +107,11 @@ export interface GatewayVoiceStateEntry {
 	serverId?: string;
 }
 
+export interface GuildChannelAuthContext {
+	guild: GuildResponse;
+	parentChannel: ChannelResponse | null;
+}
+
 export interface GatewayChannelMention {
 	id: string;
 	name: string;
@@ -250,6 +256,12 @@ export abstract class IGatewayService {
 		skipMembershipCheck?: boolean;
 	}): Promise<GuildResponse>;
 
+	abstract getGuildAuthContext(params: {
+		guildId: GuildID;
+		userId: UserID;
+		channelId?: ChannelID;
+	}): Promise<GuildChannelAuthContext>;
+
 	abstract getGuildMember(params: {guildId: GuildID; userId: UserID}): Promise<{
 		success: boolean;
 		memberData?: GuildMemberResponse;

fluxer_api/src/api/test/NoopGatewayService.ts

@@ -25,6 +25,7 @@ import {
 	type GatewayNodeStats,
 	type GatewayVoiceStateCounts,
 	type GatewayVoiceStateEntry,
+	type GuildChannelAuthContext,
 	IGatewayService,
 } from '../infrastructure/IGatewayService';
 import {UserRepository} from '../user/repositories/UserRepository';
@@ -562,6 +563,17 @@ export class NoopGatewayService extends IGatewayService {
 		return createDummyGuildResponse({guildId: params.guildId, userId: params.userId});
 	}
 
+	async getGuildAuthContext(params: {
+		guildId: GuildID;
+		userId: UserID;
+		channelId?: ChannelID;
+	}): Promise<GuildChannelAuthContext> {
+		const guild = await this.getGuildData({guildId: params.guildId, userId: params.userId});
+		const parentId = params.channelId?.toString();
+		const parentChannel = parentId ? (guild.channels?.find((channel) => channel.id === parentId) ?? null) : null;
+		return {guild, parentChannel};
+	}
+
 	async getGuildMember(params: {guildId: GuildID; userId: UserID}): Promise<{
 		success: boolean;
 		memberData?: GuildMemberResponse;