From 376afd2ad697a448810142172d0b414f3968b07e Mon Sep 17 00:00:00 2001 From: Hampus Date: Thu, 1 Oct 2026 14:03:04 +0200 Subject: [PATCH] fix(voice): keep mic publish state in sync with voice state (#3088) --- .../infrastructure/DisabledLiveKitService.ts | 12 +++ .../src/api/infrastructure/ILiveKitService.ts | 12 +++ .../src/api/infrastructure/LiveKitService.ts | 53 ++++++++++++- .../infrastructure/LiveKitWebhookService.ts | 45 ++++++++++- .../tests/LiveKitService.test.ts | 61 ++++++++++++++- .../tests/LiveKitWebhookService.test.ts | 76 ++++++++++++++++++- .../VoiceEngineV2AppMediaExecutionAdapter.tsx | 12 ++- 7 files changed, 264 insertions(+), 7 deletions(-) diff --git a/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts b/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts index 0959a7c35..57a5341f8 100644 --- a/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts +++ b/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 {} + async muteMicrophoneTrack(_params: MuteMicrophoneTrackParams): Promise {} + async disconnectParticipant(_params: DisconnectParticipantParams): Promise {} async listParticipants(_params: { diff --git a/fluxer_api/src/api/infrastructure/ILiveKitService.ts b/fluxer_api/src/api/infrastructure/ILiveKitService.ts index 16d3109da..3e095cf43 100644 --- a/fluxer_api/src/api/infrastructure/ILiveKitService.ts +++ b/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; + abstract muteMicrophoneTrack(params: MuteMicrophoneTrackParams): Promise; + abstract disconnectParticipant(params: DisconnectParticipantParams): Promise; abstract listParticipants(params: ListParticipantsParams): Promise; diff --git a/fluxer_api/src/api/infrastructure/LiveKitService.ts b/fluxer_api/src/api/infrastructure/LiveKitService.ts index 48099f1de..cb139be40 100644 --- a/fluxer_api/src/api/infrastructure/LiveKitService.ts +++ b/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 { const sources: Array = []; 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 | 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 { + 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 { const {userId, guildId, channelId, connectionId, regionId, serverId} = params; const roomName = this.getRoomName(guildId, channelId); diff --git a/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts b/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts index 22cba442c..81f05a400 100644 --- a/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts +++ b/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 { + 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 { 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'); } diff --git a/fluxer_api/src/api/infrastructure/tests/LiveKitService.test.ts b/fluxer_api/src/api/infrastructure/tests/LiveKitService.test.ts index beb86f03d..5ff99cef0 100644 --- a/fluxer_api/src/api/infrastructure/tests/LiveKitService.test.ts +++ b/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> = []; + const client = { + listParticipants: async () => [{identity: 'user_3_conn-1', tracks: listedTracks}], + updateParticipant: async (...args: Array) => { + calls.push(['updateParticipant', ...args]); + return {identity: 'user_3_conn-1', tracks: updatedTracks}; + }, + mutePublishedTrack: async (...args: Array) => { + 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: ''}}], + ]); + }); +}); diff --git a/fluxer_api/src/api/infrastructure/tests/LiveKitWebhookService.test.ts b/fluxer_api/src/api/infrastructure/tests/LiveKitWebhookService.test.ts index f1baffe74..69b61eed7 100644 --- a/fluxer_api/src/api/infrastructure/tests/LiveKitWebhookService.test.ts +++ b/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): 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(); + }); +}); diff --git a/fluxer_app/src/features/voice/engine/v2/VoiceEngineV2AppMediaExecutionAdapter.tsx b/fluxer_app/src/features/voice/engine/v2/VoiceEngineV2AppMediaExecutionAdapter.tsx index 1244134e0..5cd196969 100644 --- a/fluxer_app/src/features/voice/engine/v2/VoiceEngineV2AppMediaExecutionAdapter.tsx +++ b/fluxer_app/src/features/voice/engine/v2/VoiceEngineV2AppMediaExecutionAdapter.tsx @@ -334,6 +334,12 @@ export class VoiceEngineV2AppMediaExecutionAdapter extends Store { return isVoiceSpeakPermissionDenied(connection?.guildId ?? null, channelId); } + private isMicrophoneHeldByVoiceState(channelId: string | null): boolean { + if (this.isSpeakPermissionDenied(channelId)) return false; + const audioState = this.getEffectiveAudioState(); + return audioState.serverMute || audioState.serverDeaf; + } + private transitionMediaState(event: VoiceMediaEvent): void { this.update(() => { const nextSnapshot = transitionVoiceMediaSnapshot(this.mediaStateSnapshot, event); @@ -563,7 +569,7 @@ export class VoiceEngineV2AppMediaExecutionAdapter extends Store { return; } } - this.setAudioPublicationsMuted(room, false, 'voice state update'); + this.setAudioPublicationsMuted(room, this.shouldMuteMicrophonePublication(), 'voice state update'); this.syncLocalSpeakingOverride(room); } @@ -898,6 +904,10 @@ export class VoiceEngineV2AppMediaExecutionAdapter extends Store { channelId: string | null, options: VoiceEngineV2MicrophoneOptions, ): Promise { + if (this.isMicrophoneHeldByVoiceState(channelId)) { + logger.debug('Skipping microphone enable: muted or deafened by the server'); + return; + } this.transitionMediaState({ type: 'microphone.enable.request', hasPublication: this.hasMicrophonePublication(room),