fix(voice): keep mic publish state in sync with voice state (#3088)

fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts

@@ -38,6 +38,16 @@ interface DisconnectParticipantParams {
 	serverId: string;
 }
 
+interface MuteMicrophoneTrackParams {
+	userId: UserID;
+	guildId?: GuildID;
+	channelId: ChannelID;
+	connectionId: string;
+	regionId: string;
+	serverId: string;
+	trackSid: string;
+}
+
 interface UpdateParticipantPermissionsParams {
 	userId: UserID;
 	guildId?: GuildID;
@@ -63,6 +73,8 @@ export class DisabledLiveKitService implements ILiveKitService {
 
 	async updateParticipantPermissions(_params: UpdateParticipantPermissionsParams): Promise<void> {}
 
+	async muteMicrophoneTrack(_params: MuteMicrophoneTrackParams): Promise<void> {}
+
 	async disconnectParticipant(_params: DisconnectParticipantParams): Promise<void> {}
 
 	async listParticipants(_params: {

fluxer_api/src/api/infrastructure/ILiveKitService.ts

@@ -32,6 +32,16 @@ interface UpdateParticipantParams {
 	canVideo?: boolean;
 }
 
+interface MuteMicrophoneTrackParams {
+	userId: UserID;
+	guildId?: GuildID;
+	channelId: ChannelID;
+	connectionId: string;
+	regionId: string;
+	serverId: string;
+	trackSid: string;
+}
+
 interface UpdateParticipantPermissionsParams {
 	userId: UserID;
 	guildId?: GuildID;
@@ -87,6 +97,8 @@ export abstract class ILiveKitService {
 
 	abstract updateParticipantPermissions(params: UpdateParticipantPermissionsParams): Promise<void>;
 
+	abstract muteMicrophoneTrack(params: MuteMicrophoneTrackParams): Promise<void>;
+
 	abstract disconnectParticipant(params: DisconnectParticipantParams): Promise<void>;
 
 	abstract listParticipants(params: ListParticipantsParams): Promise<ListParticipantsResult>;

fluxer_api/src/api/infrastructure/LiveKitService.ts

@@ -7,7 +7,7 @@ import {ILiveKitService} from '@app/api/infrastructure/ILiveKitService';
 import {Logger} from '@app/api/Logger';
 import type {VoiceRegionMetadata, VoiceServerRecord} from '@app/api/voice/VoiceModel';
 import type {VoiceTopology} from '@app/api/voice/VoiceTopology';
-import {AccessToken, RoomServiceClient, TrackSource} from 'livekit-server-sdk';
+import {AccessToken, RoomServiceClient, type TrackInfo, TrackSource} from 'livekit-server-sdk';
 
 interface CreateTokenParams {
 	userId: UserID;
@@ -47,6 +47,16 @@ interface DisconnectParticipantParams {
 	serverId: string;
 }
 
+interface MuteMicrophoneTrackParams {
+	userId: UserID;
+	guildId?: GuildID;
+	channelId: ChannelID;
+	connectionId: string;
+	regionId: string;
+	serverId: string;
+	trackSid: string;
+}
+
 interface UpdateParticipantPermissionsParams {
 	userId: UserID;
 	guildId?: GuildID;
@@ -76,6 +86,8 @@ interface LiveKitPublishPermissions {
 
 export const VOICE_TOKEN_TTL_SECONDS = 60 * 10;
 
+export const SERVER_MUTE_ATTRIBUTE = 'server_mute';
+
 export function computeLiveKitPublishSources(permissions: LiveKitPublishPermissions): Array<TrackSource> {
 	const sources: Array<TrackSource> = [];
 	if (permissions.canSpeak) {
@@ -202,8 +214,10 @@ export class LiveKitService extends ILiveKitService {
 			if (!participant) {
 				return;
 			}
-			if (mute !== undefined && participant.tracks) {
-				for (const track of participant.tracks) {
+			if (mute !== undefined) {
+				const tracks =
+					(await this.setServerMuteAttribute(server, roomName, participantIdentity, mute)) ?? participant.tracks;
+				for (const track of tracks) {
 					if (track.source === TrackSource.MICROPHONE && track.sid) {
 						await server.roomServiceClient.mutePublishedTrack(roomName, participantIdentity, track.sid, mute);
 					}
@@ -277,6 +291,39 @@ export class LiveKitService extends ILiveKitService {
 		}
 	}
 
+	private async setServerMuteAttribute(
+		server: ServerClientConfig,
+		roomName: string,
+		participantIdentity: string,
+		mute: boolean,
+	): Promise<Array<TrackInfo> | null> {
+		try {
+			const updated = await server.roomServiceClient.updateParticipant(roomName, participantIdentity, {
+				attributes: {[SERVER_MUTE_ATTRIBUTE]: mute ? 'true' : ''},
+			});
+			return updated.tracks;
+		} catch (error) {
+			Logger.warn({error, participantIdentity, roomName}, 'Failed to record server mute on LiveKit participant');
+			return null;
+		}
+	}
+
+	async muteMicrophoneTrack(params: MuteMicrophoneTrackParams): Promise<void> {
+		const {userId, guildId, channelId, connectionId, regionId, serverId, trackSid} = 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}, 'LiveKit track mute skipped: unknown server');
+			return;
+		}
+		try {
+			await server.roomServiceClient.mutePublishedTrack(roomName, participantIdentity, trackSid, true);
+		} catch (error) {
+			Logger.error({error, participantIdentity, roomName, trackSid}, 'Error muting LiveKit microphone track');
+		}
+	}
+
 	async disconnectParticipant(params: DisconnectParticipantParams): Promise<void> {
 		const {userId, guildId, channelId, connectionId, regionId, serverId} = params;
 		const roomName = this.getRoomName(guildId, channelId);

fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts

@@ -4,11 +4,12 @@ import type {ChannelID, GuildID} from '@app/api/BrandedTypes';
 import type {IGatewayService} from '@app/api/infrastructure/IGatewayService';
 import type {ILiveKitService} from '@app/api/infrastructure/ILiveKitService';
 import type {IVoiceRoomStore} from '@app/api/infrastructure/IVoiceRoomStore';
+import {SERVER_MUTE_ATTRIBUTE} from '@app/api/infrastructure/LiveKitService';
 import {isDMRoom, parseParticipantMetadataWithRaw, parseRoomName} from '@app/api/infrastructure/VoiceRoomContext';
 import {Logger} from '@app/api/Logger';
 import type {VoiceTopology} from '@app/api/voice/VoiceTopology';
 import type {WebhookEvent} from 'livekit-server-sdk';
-import {WebhookReceiver} from 'livekit-server-sdk';
+import {TrackSource, WebhookReceiver} from 'livekit-server-sdk';
 
 interface VoiceWebhookParticipantContext {
 	readonly type: 'dm' | 'guild';
@@ -440,6 +441,45 @@ export class LiveKitWebhookService {
 		}
 	}
 
+	async handleTrackPublished(event: WebhookEvent): Promise<void> {
+		if (event.event !== 'track_published') {
+			return;
+		}
+		const {participant, track} = event;
+		if (!participant?.metadata || track?.source !== TrackSource.MICROPHONE || !track.sid) {
+			return;
+		}
+		if (participant.attributes[SERVER_MUTE_ATTRIBUTE] !== 'true') {
+			return;
+		}
+		const parsed = parseParticipantMetadataWithRaw(participant.metadata);
+		if (!parsed) {
+			Logger.warn({metadata: participant.metadata}, 'Failed to parse participant metadata');
+			return;
+		}
+		const {context, raw} = parsed;
+		if (!raw.region_id || !raw.server_id) {
+			Logger.warn(
+				{participantIdentity: participant.identity, trackSid: track.sid},
+				'Cannot mute published microphone without a region and server',
+			);
+			return;
+		}
+		Logger.debug(
+			{participantIdentity: participant.identity, trackSid: track.sid},
+			'Muting microphone published while server muted',
+		);
+		await this.liveKitService.muteMicrophoneTrack({
+			guildId: context.type === 'guild' ? context.guildId : undefined,
+			channelId: context.channelId,
+			userId: context.userId,
+			connectionId: context.connectionId,
+			regionId: raw.region_id,
+			serverId: raw.server_id,
+			trackSid: track.sid,
+		});
+	}
+
 	async processEvent(data: {event: WebhookEvent; apiKey: string}): Promise<void> {
 		const {event, apiKey} = data;
 		Logger.debug({event: event.event, apiKey}, 'Dispatching LiveKit webhook event');
@@ -454,6 +494,9 @@ export class LiveKitWebhookService {
 			case 'room_finished':
 				await this.handleRoomFinished(event, apiKey);
 				break;
+			case 'track_published':
+				await this.handleTrackPublished(event);
+				break;
 			default:
 				Logger.debug({event: event.event}, 'Ignoring LiveKit webhook event');
 		}

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

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-import {createChannelID, createGuildID} from '@app/api/BrandedTypes';
+import {createChannelID, createGuildID, createUserID} from '@app/api/BrandedTypes';
 import {
 	computeLiveKitPublishSources,
 	LiveKitService,
@@ -137,3 +137,62 @@ describe('LiveKitService listParticipants', () => {
 		expect(result).toEqual({status: 'ok', participants: []});
 	});
 });
+
+describe('LiveKitService updateParticipant mute', () => {
+	const params = {
+		userId: createUserID(3n),
+		guildId: createGuildID(1n),
+		channelId: createChannelID(2n),
+		connectionId: 'conn-1',
+		regionId: 'region-1',
+		serverId: 'region-1-server-1',
+	};
+
+	function roomServiceClient(
+		listedTracks: Array<{sid: string; source: TrackSource}>,
+		updatedTracks: typeof listedTracks,
+	) {
+		const calls: Array<Array<unknown>> = [];
+		const client = {
+			listParticipants: async () => [{identity: 'user_3_conn-1', tracks: listedTracks}],
+			updateParticipant: async (...args: Array<unknown>) => {
+				calls.push(['updateParticipant', ...args]);
+				return {identity: 'user_3_conn-1', tracks: updatedTracks};
+			},
+			mutePublishedTrack: async (...args: Array<unknown>) => {
+				calls.push(['mutePublishedTrack', ...args]);
+			},
+		};
+		return {client, calls};
+	}
+
+	it('records the mute on the participant and mutes every current microphone track', async () => {
+		const {client, calls} = roomServiceClient(
+			[{sid: 'TR_old', source: TrackSource.MICROPHONE}],
+			[
+				{sid: 'TR_old', source: TrackSource.MICROPHONE},
+				{sid: 'TR_new', source: TrackSource.MICROPHONE},
+			],
+		);
+		const service = createServiceWithRoomServiceClient(client);
+
+		await service.updateParticipant({...params, mute: true});
+
+		expect(calls).toEqual([
+			['updateParticipant', 'guild_1_channel_2', 'user_3_conn-1', {attributes: {server_mute: 'true'}}],
+			['mutePublishedTrack', 'guild_1_channel_2', 'user_3_conn-1', 'TR_old', true],
+			['mutePublishedTrack', 'guild_1_channel_2', 'user_3_conn-1', 'TR_new', true],
+		]);
+	});
+
+	it('clears the recorded mute on unmute', async () => {
+		const {client, calls} = roomServiceClient([], []);
+		const service = createServiceWithRoomServiceClient(client);
+
+		await service.updateParticipant({...params, mute: false});
+
+		expect(calls).toEqual([
+			['updateParticipant', 'guild_1_channel_2', 'user_3_conn-1', {attributes: {server_mute: ''}}],
+		]);
+	});
+});

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

@@ -4,10 +4,11 @@ import {createChannelID, createGuildID} from '@app/api/BrandedTypes';
 import type {IGatewayService} from '@app/api/infrastructure/IGatewayService';
 import type {ILiveKitService} from '@app/api/infrastructure/ILiveKitService';
 import type {IVoiceRoomStore} from '@app/api/infrastructure/IVoiceRoomStore';
+import {SERVER_MUTE_ATTRIBUTE} from '@app/api/infrastructure/LiveKitService';
 import {LiveKitWebhookService} from '@app/api/infrastructure/LiveKitWebhookService';
 import type {IVoiceRepository} from '@app/api/voice/IVoiceRepository';
 import {VoiceTopology} from '@app/api/voice/VoiceTopology';
-import type {WebhookEvent} from 'livekit-server-sdk';
+import {TrackSource, type WebhookEvent} from 'livekit-server-sdk';
 import {describe, expect, it, vi} from 'vitest';
 
 const GUILD_ID = createGuildID(1n);
@@ -66,3 +67,76 @@ describe('LiveKitWebhookService room_finished', () => {
 		expect(disconnectAllVoiceUsersInChannel).not.toHaveBeenCalled();
 	});
 });
+
+function trackPublished(source: TrackSource, attributes: Record<string, string>): WebhookEvent {
+	return {
+		event: 'track_published',
+		room: {name: `guild_${GUILD_ID}_channel_${CHANNEL_ID}`},
+		participant: {
+			identity: 'user_3_conn-1',
+			attributes,
+			metadata: JSON.stringify({
+				user_id: '3',
+				channel_id: CHANNEL_ID.toString(),
+				guild_id: GUILD_ID.toString(),
+				connection_id: 'conn-1',
+				region_id: 'eu',
+				server_id: 'eu-1',
+				token_nonce: 'nonce',
+				issued_at: '0',
+			}),
+		},
+		track: {sid: 'TR_new', source},
+	} as unknown as WebhookEvent;
+}
+
+function trackHarness() {
+	const muteMicrophoneTrack = vi.fn(async () => {});
+	const service = new LiveKitWebhookService(
+		{} as unknown as IVoiceRoomStore,
+		{} as unknown as IGatewayService,
+		{muteMicrophoneTrack} as unknown as ILiveKitService,
+		new VoiceTopology({} as unknown as IVoiceRepository, null),
+	);
+	return {service, muteMicrophoneTrack};
+}
+
+describe('LiveKitWebhookService track_published', () => {
+	it('mutes a microphone published by a participant the guild has muted', async () => {
+		const {service, muteMicrophoneTrack} = trackHarness();
+
+		await service.processEvent({
+			event: trackPublished(TrackSource.MICROPHONE, {[SERVER_MUTE_ATTRIBUTE]: 'true'}),
+			apiKey: 'key',
+		});
+
+		expect(muteMicrophoneTrack).toHaveBeenCalledWith({
+			guildId: GUILD_ID,
+			channelId: CHANNEL_ID,
+			userId: 3n,
+			connectionId: 'conn-1',
+			regionId: 'eu',
+			serverId: 'eu-1',
+			trackSid: 'TR_new',
+		});
+	});
+
+	it('leaves a microphone alone when the participant is not muted by the guild', async () => {
+		const {service, muteMicrophoneTrack} = trackHarness();
+
+		await service.processEvent({event: trackPublished(TrackSource.MICROPHONE, {}), apiKey: 'key'});
+
+		expect(muteMicrophoneTrack).not.toHaveBeenCalled();
+	});
+
+	it('leaves other track sources alone', async () => {
+		const {service, muteMicrophoneTrack} = trackHarness();
+
+		await service.processEvent({
+			event: trackPublished(TrackSource.CAMERA, {[SERVER_MUTE_ATTRIBUTE]: 'true'}),
+			apiKey: 'key',
+		});
+
+		expect(muteMicrophoneTrack).not.toHaveBeenCalled();
+	});
+});