fix(api): downgrade oversized video instead of ending the call (#2419)

fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import type {TrackSource} from 'livekit-server-sdk';
 import type {ChannelID, GuildID, UserID} from '../BrandedTypes';
 import type {VoiceRegionMetadata, VoiceServerRecord} from '../voice/VoiceModel';
 import type {ILiveKitService, ListActiveRoomsResult, ListParticipantsResult} from './ILiveKitService';
@@ -51,6 +52,27 @@ interface UpdateParticipantPermissionsParams {
 	deaf?: boolean;
 }
 
+interface MuteParticipantTrackParams {
+	userId: UserID;
+	guildId?: GuildID;
+	channelId: ChannelID;
+	connectionId: string;
+	regionId: string;
+	serverId: string;
+	trackSid: string;
+	muted: boolean;
+}
+
+interface RevokeParticipantPublishSourceParams {
+	userId: UserID;
+	guildId?: GuildID;
+	channelId: ChannelID;
+	connectionId: string;
+	regionId: string;
+	serverId: string;
+	source: TrackSource;
+}
+
 export class DisabledLiveKitService implements ILiveKitService {
 	async createToken(_params: CreateTokenParams): Promise<{
 		token: string;
@@ -65,6 +87,14 @@ export class DisabledLiveKitService implements ILiveKitService {
 
 	async disconnectParticipant(_params: DisconnectParticipantParams): Promise<void> {}
 
+	async muteParticipantTrack(_params: MuteParticipantTrackParams): Promise<boolean> {
+		return false;
+	}
+
+	async revokeParticipantPublishSource(_params: RevokeParticipantPublishSourceParams): Promise<boolean> {
+		return false;
+	}
+
 	async listParticipants(_params: {
 		guildId?: GuildID;
 		channelId: ChannelID;

fluxer_api/src/api/infrastructure/ILiveKitService.ts

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import type {TrackSource} from 'livekit-server-sdk';
 import type {ChannelID, GuildID, UserID} from '../BrandedTypes';
 import type {VoiceRegionMetadata, VoiceServerRecord} from '../voice/VoiceModel';
 
@@ -54,6 +55,27 @@ interface DisconnectParticipantParams {
 	serverId: string;
 }
 
+interface MuteParticipantTrackParams {
+	userId: UserID;
+	guildId?: GuildID;
+	channelId: ChannelID;
+	connectionId: string;
+	regionId: string;
+	serverId: string;
+	trackSid: string;
+	muted: boolean;
+}
+
+interface RevokeParticipantPublishSourceParams {
+	userId: UserID;
+	guildId?: GuildID;
+	channelId: ChannelID;
+	connectionId: string;
+	regionId: string;
+	serverId: string;
+	source: TrackSource;
+}
+
 interface ListParticipantsParams {
 	guildId?: GuildID;
 	channelId: ChannelID;
@@ -109,6 +131,10 @@ export abstract class ILiveKitService {
 
 	abstract disconnectParticipant(params: DisconnectParticipantParams): Promise<void>;
 
+	abstract muteParticipantTrack(params: MuteParticipantTrackParams): Promise<boolean>;
+
+	abstract revokeParticipantPublishSource(params: RevokeParticipantPublishSourceParams): Promise<boolean>;
+
 	abstract listParticipants(params: ListParticipantsParams): Promise<ListParticipantsResult>;
 
 	abstract listActiveRooms(): Promise<ListActiveRoomsResult>;

fluxer_api/src/api/infrastructure/LiveKitService.ts

@@ -60,6 +60,27 @@ interface UpdateParticipantPermissionsParams {
 	deaf?: boolean;
 }
 
+interface MuteParticipantTrackParams {
+	userId: UserID;
+	guildId?: GuildID;
+	channelId: ChannelID;
+	connectionId: string;
+	regionId: string;
+	serverId: string;
+	trackSid: string;
+	muted: boolean;
+}
+
+interface RevokeParticipantPublishSourceParams {
+	userId: UserID;
+	guildId?: GuildID;
+	channelId: ChannelID;
+	connectionId: string;
+	regionId: string;
+	serverId: string;
+	source: TrackSource;
+}
+
 interface ServerClientConfig {
 	endpoint: string;
 	apiKey: string;
@@ -76,6 +97,26 @@ interface LiveKitPublishPermissions {
 
 export const VOICE_TOKEN_TTL_SECONDS = 60 * 10;
 
+const ALL_PUBLISH_SOURCES: ReadonlyArray<TrackSource> = [
+	TrackSource.MICROPHONE,
+	TrackSource.CAMERA,
+	TrackSource.SCREEN_SHARE,
+	TrackSource.SCREEN_SHARE_AUDIO,
+];
+
+interface LiveKitPublishGrant {
+	canPublish: boolean;
+	canPublishSources: Array<TrackSource>;
+}
+
+export function computeRevokedPublishGrant(current: LiveKitPublishGrant, source: TrackSource): LiveKitPublishGrant {
+	const allowed = current.canPublishSources.length > 0 ? current.canPublishSources : ALL_PUBLISH_SOURCES;
+	const revoked =
+		source === TrackSource.SCREEN_SHARE ? [TrackSource.SCREEN_SHARE, TrackSource.SCREEN_SHARE_AUDIO] : [source];
+	const canPublishSources = allowed.filter((allowedSource) => !revoked.includes(allowedSource));
+	return {canPublish: current.canPublish && canPublishSources.length > 0, canPublishSources};
+}
+
 export function computeLiveKitPublishSources(permissions: LiveKitPublishPermissions): Array<TrackSource> {
 	const sources: Array<TrackSource> = [];
 	if (permissions.canSpeak) {
@@ -300,6 +341,72 @@ export class LiveKitService extends ILiveKitService {
 		}
 	}
 
+	async muteParticipantTrack(params: MuteParticipantTrackParams): Promise<boolean> {
+		const {userId, guildId, channelId, connectionId, regionId, serverId, trackSid, muted} = params;
+		const roomName = this.getRoomName(guildId, channelId);
+		const participantIdentity = this.getParticipantIdentity(userId, connectionId);
+		const server = this.tryResolveServerClient(regionId, serverId);
+		if (server === null) {
+			Logger.debug(
+				{regionId, serverId, participantIdentity, roomName, trackSid},
+				'LiveKit track mute skipped, pinned server no longer exists in topology',
+			);
+			return false;
+		}
+		try {
+			await server.roomServiceClient.mutePublishedTrack(roomName, participantIdentity, trackSid, muted);
+			return true;
+		} catch (error) {
+			if (LiveKitService.isHttp404(error)) {
+				Logger.debug({participantIdentity, roomName, trackSid}, 'LiveKit track no longer published, nothing to mute');
+				return false;
+			}
+			Logger.error({error, participantIdentity, roomName, trackSid, muted}, 'Error muting LiveKit published track');
+			return false;
+		}
+	}
+
+	async revokeParticipantPublishSource(params: RevokeParticipantPublishSourceParams): Promise<boolean> {
+		const {userId, guildId, channelId, connectionId, regionId, serverId, source} = params;
+		const roomName = this.getRoomName(guildId, channelId);
+		const participantIdentity = this.getParticipantIdentity(userId, connectionId);
+		const server = this.tryResolveServerClient(regionId, serverId);
+		if (server === null) {
+			Logger.debug(
+				{regionId, serverId, participantIdentity, roomName, source},
+				'LiveKit publish source revoke skipped, pinned server no longer exists in topology',
+			);
+			return false;
+		}
+		try {
+			const participants = await server.roomServiceClient.listParticipants(roomName);
+			const participant = participants.find((p) => p.identity === participantIdentity);
+			if (!participant?.permission) {
+				Logger.debug(
+					{participantIdentity, roomName, source},
+					'LiveKit participant no longer in room, nothing to revoke',
+				);
+				return false;
+			}
+			const grant = computeRevokedPublishGrant(participant.permission, source);
+			await server.roomServiceClient.updateParticipant(roomName, participantIdentity, undefined, {
+				...participant.permission,
+				...grant,
+			});
+			return true;
+		} catch (error) {
+			if (LiveKitService.isHttp404(error)) {
+				Logger.debug(
+					{participantIdentity, roomName, source},
+					'LiveKit participant no longer in room, nothing to revoke',
+				);
+				return false;
+			}
+			Logger.error({error, participantIdentity, roomName, source}, 'Error revoking LiveKit publish source');
+			return false;
+		}
+	}
+
 	async listParticipants(params: {
 		guildId?: GuildID;
 		channelId: ChannelID;

fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts

@@ -533,11 +533,18 @@ export class LiveKitWebhookService {
 					height: track.height,
 					trackSource: trackSourceLabel,
 				},
-				'User without higher video quality entitlement attempted to publish video exceeding free tier limits - disconnecting',
+				'User without higher video quality entitlement published video exceeding free tier limits - muting track and revoking source',
 			);
+			if (!track.sid) {
+				Logger.warn(
+					{userId: userId.toString(), roomName: room.name, trackSource: trackSourceLabel},
+					'Track published without a sid, cannot enforce free tier video limits',
+				);
+				return;
+			}
 			const roomContext = parseRoomName(room.name);
 			if (!roomContext) {
-				Logger.warn({roomName: room.name}, 'Unknown room name format, cannot disconnect');
+				Logger.warn({roomName: room.name}, 'Unknown room name format, cannot enforce free tier video limits');
 				return;
 			}
 			let regionId: string | undefined;
@@ -567,7 +574,7 @@ export class LiveKitWebhookService {
 			if (!regionId || !serverId) {
 				Logger.warn(
 					{participantId: participant.identity, roomName: room.name, apiKey},
-					'Missing region or server info, cannot disconnect',
+					'Missing region or server info, cannot enforce free tier video limits',
 				);
 				return;
 			}
@@ -585,21 +592,26 @@ export class LiveKitWebhookService {
 					height: track.height,
 					trackSource: trackSourceLabel,
 				},
-				'Disconnecting user without higher video quality entitlement for exceeding video quality limits',
+				'Muting oversized track and revoking its publish source for user without higher video quality entitlement',
 			);
-			await this.liveKitService.disconnectParticipant({
+			const muted = await this.liveKitService.muteParticipantTrack({
 				userId,
 				guildId,
 				channelId: roomContext.channelId,
 				connectionId,
 				regionId,
 				serverId,
+				trackSid: track.sid,
+				muted: true,
 			});
-			await this.gatewayService.disconnectVoiceUserIfInChannel({
+			const revoked = await this.liveKitService.revokeParticipantPublishSource({
+				userId,
 				guildId,
 				channelId: roomContext.channelId,
-				userId,
 				connectionId,
+				regionId,
+				serverId,
+				source: track.source,
 			});
 			Logger.info(
 				{
@@ -611,8 +623,13 @@ export class LiveKitWebhookService {
 					width: track.width,
 					height: track.height,
 					trackSource: trackSourceLabel,
+					trackSid: track.sid,
+					muted,
+					revoked,
 				},
-				'Disconnected user without higher video quality entitlement for exceeding video quality limits',
+				muted || revoked
+					? 'Enforced free tier video limits on user without higher video quality entitlement'
+					: 'Failed to enforce free tier video limits on user without higher video quality entitlement',
 			);
 		} catch (error) {
 			Logger.error({error}, 'Error processing track_published event');

fluxer_api/src/api/infrastructure/tests/LiveKitService.test.ts

@@ -1,8 +1,19 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import type {WebhookEvent} from 'livekit-server-sdk';
 import {AccessToken, TrackSource} from 'livekit-server-sdk';
-import {describe, expect, it} from 'vitest';
-import {computeLiveKitPublishSources, VOICE_TOKEN_TTL_SECONDS} from '../LiveKitService';
+import {describe, expect, it, vi} from 'vitest';
+import {createUserID} from '../../BrandedTypes';
+import {getConfig} from '../../Config';
+import type {LimitConfigService} from '../../limits/LimitConfigService';
+import type {User} from '../../models/User';
+import type {IUserRepository} from '../../user/IUserRepository';
+import type {VoiceTopology} from '../../voice/VoiceTopology';
+import type {IGatewayService} from '../IGatewayService';
+import type {ILiveKitService} from '../ILiveKitService';
+import type {IVoiceRoomStore} from '../IVoiceRoomStore';
+import {computeLiveKitPublishSources, computeRevokedPublishGrant, VOICE_TOKEN_TTL_SECONDS} from '../LiveKitService';
+import {LiveKitWebhookService} from '../LiveKitWebhookService';
 
 function decodeJwtPayload(token: string): Record<string, unknown> {
 	const [, payload] = token.split('.');
@@ -44,6 +55,45 @@ describe('LiveKitService publish permissions', () => {
 			canPublishSources: ['microphone', 'screen_share', 'screen_share_audio'],
 		});
 	});
+	it('revokes only the offending source and keeps the rest of the grant', () => {
+		expect(
+			computeRevokedPublishGrant(
+				{
+					canPublish: true,
+					canPublishSources: [
+						TrackSource.MICROPHONE,
+						TrackSource.CAMERA,
+						TrackSource.SCREEN_SHARE,
+						TrackSource.SCREEN_SHARE_AUDIO,
+					],
+				},
+				TrackSource.CAMERA,
+			),
+		).toEqual({
+			canPublish: true,
+			canPublishSources: [TrackSource.MICROPHONE, TrackSource.SCREEN_SHARE, TrackSource.SCREEN_SHARE_AUDIO],
+		});
+	});
+	it('revokes screen share audio together with screen share video', () => {
+		expect(
+			computeRevokedPublishGrant(
+				{
+					canPublish: true,
+					canPublishSources: [TrackSource.MICROPHONE, TrackSource.SCREEN_SHARE, TrackSource.SCREEN_SHARE_AUDIO],
+				},
+				TrackSource.SCREEN_SHARE,
+			),
+		).toEqual({canPublish: true, canPublishSources: [TrackSource.MICROPHONE]});
+	});
+	it('treats an empty source list as every source and never leaves it empty', () => {
+		expect(computeRevokedPublishGrant({canPublish: true, canPublishSources: []}, TrackSource.CAMERA)).toEqual({
+			canPublish: true,
+			canPublishSources: [TrackSource.MICROPHONE, TrackSource.SCREEN_SHARE, TrackSource.SCREEN_SHARE_AUDIO],
+		});
+		expect(
+			computeRevokedPublishGrant({canPublish: true, canPublishSources: [TrackSource.CAMERA]}, TrackSource.CAMERA),
+		).toEqual({canPublish: false, canPublishSources: []});
+	});
 	it('bounds voice token lifetime to the configured TTL', async () => {
 		const token = new AccessToken('test-key', 'test-secret', {
 			identity: 'user_1_conn',
@@ -57,3 +107,108 @@ describe('LiveKitService publish permissions', () => {
 		expect(exp - nowSeconds).toBeGreaterThan(0);
 	});
 });
+
+function createFreeUser(): User {
+	return {
+		id: createUserID(1n),
+		isBot: false,
+		premiumType: null,
+		premiumUntil: null,
+		premiumGiftExtensionEndsAt: null,
+		premiumWillCancel: false,
+		premiumGraceEndsAt: null,
+		flags: 0n,
+		premiumFlags: 0,
+		traits: new Set<string>(),
+	} as unknown as User;
+}
+
+function createTrackPublishedEvent(width: number, height: number): WebhookEvent {
+	return {
+		event: 'track_published',
+		room: {name: 'guild_2_channel_3'},
+		participant: {identity: 'user_1_conn'},
+		track: {type: 1, source: TrackSource.CAMERA, sid: 'TR_oversized', width, height},
+	} as unknown as WebhookEvent;
+}
+
+function createWebhookHarness() {
+	const muteParticipantTrack = vi.fn().mockResolvedValue(true);
+	const revokeParticipantPublishSource = vi.fn().mockResolvedValue(true);
+	const disconnectParticipant = vi.fn().mockResolvedValue(undefined);
+	const disconnectVoiceUserIfInChannel = vi.fn().mockResolvedValue(undefined);
+	const service = new LiveKitWebhookService(
+		{
+			getPinnedRoomServer: vi.fn().mockResolvedValue({regionId: 'region-1', serverId: 'region-1-server-1'}),
+		} as unknown as IVoiceRoomStore,
+		{disconnectVoiceUserIfInChannel} as unknown as IGatewayService,
+		{findUnique: vi.fn().mockResolvedValue(createFreeUser())} as unknown as IUserRepository,
+		{muteParticipantTrack, revokeParticipantPublishSource, disconnectParticipant} as unknown as ILiveKitService,
+		{
+			getAllRegions: () => [],
+			getServersForRegion: () => [],
+			registerSubscriber: () => {},
+		} as unknown as VoiceTopology,
+		{getConfigSnapshot: () => null} as unknown as LimitConfigService,
+	);
+	return {
+		service,
+		muteParticipantTrack,
+		revokeParticipantPublishSource,
+		disconnectParticipant,
+		disconnectVoiceUserIfInChannel,
+	};
+}
+
+describe('LiveKit free tier video resolution enforcement', () => {
+	it('mutes the oversized track and revokes its source instead of ending the call', async () => {
+		const {
+			service,
+			muteParticipantTrack,
+			revokeParticipantPublishSource,
+			disconnectParticipant,
+			disconnectVoiceUserIfInChannel,
+		} = createWebhookHarness();
+
+		await service.handleTrackPublished(createTrackPublishedEvent(1920, 1080), 'api-key');
+
+		expect(muteParticipantTrack).toHaveBeenCalledTimes(1);
+		expect(muteParticipantTrack).toHaveBeenCalledWith(
+			expect.objectContaining({trackSid: 'TR_oversized', muted: true, regionId: 'region-1'}),
+		);
+		expect(revokeParticipantPublishSource).toHaveBeenCalledTimes(1);
+		expect(revokeParticipantPublishSource).toHaveBeenCalledWith(
+			expect.objectContaining({source: TrackSource.CAMERA, connectionId: 'conn', regionId: 'region-1'}),
+		);
+		expect(disconnectParticipant).not.toHaveBeenCalled();
+		expect(disconnectVoiceUserIfInChannel).not.toHaveBeenCalled();
+	});
+
+	it('leaves tracks within the free tier limits alone', async () => {
+		const {service, muteParticipantTrack, revokeParticipantPublishSource, disconnectParticipant} =
+			createWebhookHarness();
+
+		await service.handleTrackPublished(createTrackPublishedEvent(1280, 720), 'api-key');
+
+		expect(muteParticipantTrack).not.toHaveBeenCalled();
+		expect(revokeParticipantPublishSource).not.toHaveBeenCalled();
+		expect(disconnectParticipant).not.toHaveBeenCalled();
+	});
+
+	it('does not enforce resolution limits on self-hosted instances', async () => {
+		const {service, muteParticipantTrack, revokeParticipantPublishSource, disconnectParticipant} =
+			createWebhookHarness();
+		const config = getConfig();
+		const originalSelfHosted = config.instance.selfHosted;
+		config.instance.selfHosted = true;
+		try {
+			await service.handleTrackPublished(createTrackPublishedEvent(3840, 2160), 'api-key');
+		} finally {
+			config.instance.selfHosted = originalSelfHosted;
+		}
+
+		expect(muteParticipantTrack).not.toHaveBeenCalled();
+		expect(revokeParticipantPublishSource).not.toHaveBeenCalled();
+		expect(disconnectParticipant).not.toHaveBeenCalled();
+	});
+});