From 03d1354562e82080f05b52726b19cc9cf0aecefd Mon Sep 17 00:00:00 2001 From: Hampus Date: Wed, 16 Sep 2026 17:09:57 +0200 Subject: [PATCH] chore(voice): remove voice reconciliation leftovers (#2810) --- deploy/self-hosting/docker-compose.yml | 1 - .../infrastructure/DisabledLiveKitService.ts | 10 +- .../src/api/infrastructure/GatewayService.ts | 58 +-- .../src/api/infrastructure/IGatewayService.ts | 24 -- .../src/api/infrastructure/ILiveKitService.ts | 22 -- .../src/api/infrastructure/LiveKitService.ts | 67 +--- .../infrastructure/LiveKitWebhookService.ts | 4 +- fluxer_api/src/api/test/NoopGatewayService.ts | 33 -- .../src/gateway/gateway_rpc_guild.erl | 1 - .../src/gateway/gateway_rpc_guild_voice.erl | 23 -- .../gateway/gateway_rpc_guild_voice_util.erl | 13 - .../src/gateway/gateway_rpc_misc.erl | 10 +- .../src/gateway/gateway_rpc_misc_push.erl | 356 ------------------ .../src/gateway/gateway_rpc_voice.erl | 14 - fluxer_gateway/src/guild/guild.erl | 6 - .../src/guild/guild_voice_lifecycle.erl | 10 - .../src/guild/voice/guild_voice_server.erl | 7 - .../guild/voice/guild_voice_server_state.erl | 109 ------ 18 files changed, 6 insertions(+), 762 deletions(-) delete mode 100644 fluxer_gateway/src/gateway/gateway_rpc_misc_push.erl diff --git a/deploy/self-hosting/docker-compose.yml b/deploy/self-hosting/docker-compose.yml index 4f12d4a96..041aff1b2 100644 --- a/deploy/self-hosting/docker-compose.yml +++ b/deploy/self-hosting/docker-compose.yml @@ -413,7 +413,6 @@ services: NODE_OPTIONS: --enable-source-maps${FLUXER_WORKER_NODE_HEAP_MB:+ --max-old-space-size=$FLUXER_WORKER_NODE_HEAP_MB} FLUXER_API_WORKER_MODE: all_lanes FLUXER_API_WORKER_ENABLE_CRON_SCHEDULER: "true" - FLUXER_API_WORKER_ENABLE_VOICE_RECONCILIATION: "true" FLUXER_POSTGRES_MAX_CONNECTIONS: "25" healthcheck: test: ["CMD", "node", "-e", "const age=Date.now()-require('node:fs').statSync('/tmp/fluxer-worker-heartbeat').mtimeMs;if(age>30000){console.error('worker heartbeat is '+Math.round(age)+'ms old');process.exit(1)}"] diff --git a/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts b/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts index b7a319252..0959a7c35 100644 --- a/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts +++ b/fluxer_api/src/api/infrastructure/DisabledLiveKitService.ts @@ -1,11 +1,7 @@ // SPDX-License-Identifier: AGPL-3.0-or-later import type {ChannelID, GuildID, UserID} from '@app/api/BrandedTypes'; -import type { - ILiveKitService, - ListActiveRoomsResult, - ListParticipantsResult, -} from '@app/api/infrastructure/ILiveKitService'; +import type {ILiveKitService, ListParticipantsResult} from '@app/api/infrastructure/ILiveKitService'; import type {VoiceRegionMetadata, VoiceServerRecord} from '@app/api/voice/VoiceModel'; interface CreateTokenParams { @@ -78,10 +74,6 @@ export class DisabledLiveKitService implements ILiveKitService { return {status: 'ok', participants: []}; } - async listActiveRooms(): Promise { - return {rooms: [], errors: [], searchedServers: 0, completed: true}; - } - getDefaultRegionId(): string | null { return null; } diff --git a/fluxer_api/src/api/infrastructure/GatewayService.ts b/fluxer_api/src/api/infrastructure/GatewayService.ts index a2bdad9e9..14d9e077d 100644 --- a/fluxer_api/src/api/infrastructure/GatewayService.ts +++ b/fluxer_api/src/api/infrastructure/GatewayService.ts @@ -1,14 +1,13 @@ // SPDX-License-Identifier: AGPL-3.0-or-later import type {ChannelID, GuildID, MessageID, RoleID, UserID} from '@app/api/BrandedTypes'; -import {createChannelID, createGuildID, createRoleID, createUserID} from '@app/api/BrandedTypes'; +import {createChannelID, createRoleID, createUserID} from '@app/api/BrandedTypes'; import {SYSTEM_USER_ID} from '@app/api/constants/Core'; import type {GatewayDispatchEvent} from '@app/api/constants/Gateway'; import {GatewayRpcClient} from '@app/api/infrastructure/GatewayRpcClient'; import {GatewayRpcMethodError, GatewayRpcMethodErrorCodes} from '@app/api/infrastructure/GatewayRpcError'; import type { CallData, - GatewayActiveVoiceRooms, GatewayChannelMention, GatewayGuildMemoryStats, GatewayMentionSources, @@ -1066,26 +1065,6 @@ export class GatewayService { }; } - async getActiveVoiceRooms(): Promise { - const result = await this.call<{ - rooms?: Array<{ - guild_id?: string | null; - channel_id: string; - voice_state_count?: number; - }>; - node_count?: number; - }>('process.active_voice_rooms', {}); - return { - nodeCount: result.node_count ?? 0, - rooms: (result.rooms ?? []).map((room) => ({ - guildId: - room.guild_id === undefined || room.guild_id === null ? undefined : createGuildID(BigInt(room.guild_id)), - channelId: createChannelID(BigInt(room.channel_id)), - voiceStateCount: room.voice_state_count ?? 0, - })), - }; - } - async getUserPermissions({guildId, userId, channelId}: UserPermissionsParams): Promise { const result = await this.call<{ permissions: string; @@ -1638,41 +1617,6 @@ export class GatewayService { }; } - async repairVoiceStateFromCache({ - guildId, - channelId, - userId, - connectionId, - }: { - guildId?: GuildID; - channelId: ChannelID; - userId: UserID; - connectionId: string; - }): Promise<{ - success: boolean; - repaired?: boolean; - error?: string; - }> { - const params: Record = { - channel_id: channelId.toString(), - user_id: userId.toString(), - connection_id: connectionId, - }; - if (guildId !== undefined) { - params['guild_id'] = guildId.toString(); - } - const result = await this.call<{ - success: boolean; - repaired?: boolean; - error?: string; - }>('voice.repair_state_from_cache', params); - return { - success: result.success, - repaired: result.repaired, - error: result.error, - }; - } - async getVoiceStatesForChannel({guildId, channelId}: {guildId?: GuildID; channelId: ChannelID}): Promise<{ voiceStates: Array; }> { diff --git a/fluxer_api/src/api/infrastructure/IGatewayService.ts b/fluxer_api/src/api/infrastructure/IGatewayService.ts index 402a6392f..601ad1d43 100644 --- a/fluxer_api/src/api/infrastructure/IGatewayService.ts +++ b/fluxer_api/src/api/infrastructure/IGatewayService.ts @@ -88,17 +88,6 @@ export interface GatewayVoiceStateCounts { servers: Array; } -export interface GatewayActiveVoiceRoom { - guildId?: GuildID; - channelId: ChannelID; - voiceStateCount: number; -} - -export interface GatewayActiveVoiceRooms { - rooms: Array; - nodeCount: number; -} - export interface GatewayVoiceStateEntry { connectionId: string; userId: string; @@ -162,8 +151,6 @@ export abstract class IGatewayService { abstract getVoiceStateCounts(): Promise; - abstract getActiveVoiceRooms(): Promise; - abstract getUsersToMentionByRoles(params: { guildId: GuildID; channelId: ChannelID; @@ -345,17 +332,6 @@ export abstract class IGatewayService { error?: string; }>; - abstract repairVoiceStateFromCache(params: { - guildId?: GuildID; - channelId: ChannelID; - userId: UserID; - connectionId: string; - }): Promise<{ - success: boolean; - repaired?: boolean; - error?: string; - }>; - abstract getVoiceStatesForChannel(params: {guildId?: GuildID; channelId: ChannelID}): Promise<{ voiceStates: Array; }>; diff --git a/fluxer_api/src/api/infrastructure/ILiveKitService.ts b/fluxer_api/src/api/infrastructure/ILiveKitService.ts index f28eec4a9..16d3109da 100644 --- a/fluxer_api/src/api/infrastructure/ILiveKitService.ts +++ b/fluxer_api/src/api/infrastructure/ILiveKitService.ts @@ -61,13 +61,6 @@ interface ListParticipantsParams { serverId: string; } -export interface LiveKitServerError { - regionId: string; - serverId: string; - errorCode: string; - retryable: boolean; -} - export interface ListParticipantsSuccess { status: 'ok'; participants: Array<{ @@ -84,19 +77,6 @@ export interface ListParticipantsError { export type ListParticipantsResult = ListParticipantsSuccess | ListParticipantsError; -export interface LiveKitRoomLocation { - roomName: string; - regionId: string; - serverId: string; -} - -export interface ListActiveRoomsResult { - rooms: Array; - errors: Array; - searchedServers: number; - completed: boolean; -} - export abstract class ILiveKitService { abstract createToken(params: CreateTokenParams): Promise<{ token: string; @@ -111,8 +91,6 @@ export abstract class ILiveKitService { abstract listParticipants(params: ListParticipantsParams): Promise; - abstract listActiveRooms(): Promise; - abstract getDefaultRegionId(): string | null; abstract getRegionMetadata(): Array; diff --git a/fluxer_api/src/api/infrastructure/LiveKitService.ts b/fluxer_api/src/api/infrastructure/LiveKitService.ts index fc8d1f59d..931ea44a3 100644 --- a/fluxer_api/src/api/infrastructure/LiveKitService.ts +++ b/fluxer_api/src/api/infrastructure/LiveKitService.ts @@ -2,11 +2,7 @@ import type {ChannelID, GuildID, UserID} from '@app/api/BrandedTypes'; import {Config} from '@app/api/Config'; -import type { - ListActiveRoomsResult, - ListParticipantsResult, - LiveKitServerError, -} from '@app/api/infrastructure/ILiveKitService'; +import type {ListParticipantsResult} from '@app/api/infrastructure/ILiveKitService'; import {ILiveKitService} from '@app/api/infrastructure/ILiveKitService'; import {Logger} from '@app/api/Logger'; import type {VoiceRegionMetadata, VoiceServerRecord} from '@app/api/voice/VoiceModel'; @@ -339,39 +335,6 @@ export class LiveKitService extends ILiveKitService { } } - async listActiveRooms(): Promise { - const rooms: ListActiveRoomsResult['rooms'] = []; - const errors: Array = []; - const servers = this.getActiveServerClients(); - for (const server of servers) { - try { - const liveRooms = await server.roomServiceClient.listRooms(); - for (const room of liveRooms) { - if (typeof room.name !== 'string' || room.name.length === 0) { - continue; - } - rooms.push({ - roomName: room.name, - regionId: server.regionId, - serverId: server.serverId, - }); - } - } catch (error) { - Logger.warn( - {error, regionId: server.regionId, serverId: server.serverId}, - 'LiveKit listRooms failed during voice reconciliation', - ); - errors.push(this.toServerError(server.regionId, server.serverId, error)); - } - } - return { - rooms, - errors, - searchedServers: servers.length, - completed: errors.length === 0, - }; - } - private static isHttp404(error: unknown): boolean { return LiveKitService.getHttpStatus(error) === 404; } @@ -396,34 +359,6 @@ export class LiveKitService extends ILiveKitService { return region.get(serverId) ?? null; } - private getActiveServerClients(): Array { - const servers: Array = []; - for (const [regionId, region] of this.serverClients.entries()) { - for (const [serverId, server] of region.entries()) { - if (server.isActive) { - servers.push({...server, regionId, serverId}); - } - } - } - return servers.sort((left, right) => { - const regionComparison = left.regionId.localeCompare(right.regionId); - if (regionComparison !== 0) { - return regionComparison; - } - return left.serverId.localeCompare(right.serverId); - }); - } - - private toServerError(regionId: string, serverId: string, error: unknown): LiveKitServerError { - const status = LiveKitService.getHttpStatus(error); - return { - regionId, - serverId, - errorCode: error instanceof Error ? error.message : 'unknown', - retryable: status != null && status >= 500, - }; - } - getDefaultRegionId(): string | null { return this.topology.getDefaultRegionId(); } diff --git a/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts b/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts index a0b60f120..22cba442c 100644 --- a/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts +++ b/fluxer_api/src/api/infrastructure/LiveKitWebhookService.ts @@ -92,7 +92,7 @@ export class LiveKitWebhookService { roomName: data.event.room?.name ?? null, eventType: data.event.event, }, - 'LiveKit webhook reports dropped events - reconciliation may be needed', + 'LiveKit webhook reports dropped events', ); } await this.processEvent(data); @@ -256,7 +256,7 @@ export class LiveKitWebhookService { error: result.error, participantIdentity: participant.identity, }, - 'LiveKit participant_joined did not match gateway state; leaving participant connected for reconciliation', + 'LiveKit participant_joined did not match gateway state; leaving participant connected', ); return; } diff --git a/fluxer_api/src/api/test/NoopGatewayService.ts b/fluxer_api/src/api/test/NoopGatewayService.ts index 5adf2c123..23fc60535 100644 --- a/fluxer_api/src/api/test/NoopGatewayService.ts +++ b/fluxer_api/src/api/test/NoopGatewayService.ts @@ -20,7 +20,6 @@ import {GuildRepository} from '@app/api/guild/repositories/GuildRepository'; import {GuildRoleRepository} from '@app/api/guild/repositories/GuildRoleRepository'; import { type CallData, - type GatewayActiveVoiceRooms, type GatewayChannelMention, type GatewayGuildMemoryStats, type GatewayMentionSources, @@ -168,25 +167,6 @@ export class NoopGatewayService extends IGatewayService { }; } - async getActiveVoiceRooms(): Promise { - return { - nodeCount: 1, - rooms: Array.from(this.voiceStatesByChannel.entries()).flatMap(([key, voiceStates]) => { - if (voiceStates.length === 0) { - return []; - } - const [guildIdText, channelIdText] = key.split(':'); - return [ - { - guildId: guildIdText === 'dm' ? undefined : (BigInt(guildIdText) as GuildID), - channelId: BigInt(channelIdText) as ChannelID, - voiceStateCount: voiceStates.length, - }, - ]; - }), - }; - } - async getUsersToMentionByRoles(_params: { guildId: GuildID; channelId: ChannelID; @@ -882,19 +862,6 @@ export class NoopGatewayService extends IGatewayService { return {success: false}; } - async repairVoiceStateFromCache(_params: { - guildId?: GuildID; - channelId: ChannelID; - userId: UserID; - connectionId: string; - }): Promise<{ - success: boolean; - repaired?: boolean; - error?: string; - }> { - return {success: false}; - } - async getVoiceStatesForChannel(params: {guildId?: GuildID; channelId: ChannelID}): Promise<{ voiceStates: Array; }> { diff --git a/fluxer_gateway/src/gateway/gateway_rpc_guild.erl b/fluxer_gateway/src/gateway/gateway_rpc_guild.erl index bffd36904..57ef1fe6e 100644 --- a/fluxer_gateway/src/gateway/gateway_rpc_guild.erl +++ b/fluxer_gateway/src/gateway/gateway_rpc_guild.erl @@ -75,7 +75,6 @@ route(M, P) -> <<"guild.disconnect_voice_user_if_in_channel">> => fun gateway_rpc_guild_voice:handle/2, <<"guild.disconnect_all_voice_users_in_channel">> => fun gateway_rpc_guild_voice:handle/2, <<"guild.confirm_voice_connection_from_livekit">> => fun gateway_rpc_guild_voice:handle/2, - <<"guild.repair_voice_state_from_cache">> => fun gateway_rpc_guild_voice:handle/2, <<"guild.get_voice_states_for_channel">> => fun gateway_rpc_guild_voice:handle/2, <<"guild.get_pending_joins_for_channel">> => fun gateway_rpc_guild_voice:handle/2, <<"guild.move_member">> => fun gateway_rpc_guild_voice:handle/2, diff --git a/fluxer_gateway/src/gateway/gateway_rpc_guild_voice.erl b/fluxer_gateway/src/gateway/gateway_rpc_guild_voice.erl index 6adb4b6fe..7a4fd07f8 100644 --- a/fluxer_gateway/src/gateway/gateway_rpc_guild_voice.erl +++ b/fluxer_gateway/src/gateway/gateway_rpc_guild_voice.erl @@ -24,8 +24,6 @@ handle(<<"guild.disconnect_all_voice_users_in_channel">>, P) -> handle_disconnect_all_in_channel(P); handle(<<"guild.confirm_voice_connection_from_livekit">>, P) -> handle_confirm_connection(P); -handle(<<"guild.repair_voice_state_from_cache">>, P) -> - handle_repair_state(P); handle(<<"guild.get_voice_states_for_channel">>, P) -> handle_get_voice_states(P); handle(<<"guild.get_pending_joins_for_channel">>, P) -> @@ -167,27 +165,6 @@ format_confirm_result({error, _, EA}) -> format_confirm_result(#{error := E}) -> raise_voice_error(E). --spec handle_repair_state(map()) -> term(). -handle_repair_state(P) -> - GuildId = validation:snowflake_or_throw(<<"guild_id">>, maps:get(<<"guild_id">>, P)), - ChannelId = validation:snowflake_or_throw(<<"channel_id">>, maps:get(<<"channel_id">>, P)), - UserId = validation:snowflake_or_throw(<<"user_id">>, maps:get(<<"user_id">>, P)), - gateway_rpc_guild_infra:with_voice_server(GuildId, fun(VPid, _) -> - repair_voice_state(VPid, P, ChannelId, UserId) - end). - --spec repair_voice_state(pid(), map(), integer(), integer()) -> map(). -repair_voice_state(VPid, P, ChannelId, UserId) -> - Req = #{ - connection_id => maps:get(<<"connection_id">>, P), - channel_id => ChannelId, - user_id => UserId - }, - Result = gen_server:call( - VPid, {repair_voice_state_from_guild_cache, Req}, ?GUILD_CALL_TIMEOUT - ), - gateway_rpc_guild_voice_util:handle_repair_result(Result). - -spec handle_get_voice_states(map()) -> term(). handle_get_voice_states(P) -> case validate_gc_params(P) of diff --git a/fluxer_gateway/src/gateway/gateway_rpc_guild_voice_util.erl b/fluxer_gateway/src/gateway/gateway_rpc_guild_voice_util.erl index ec1896d5e..1848380f6 100644 --- a/fluxer_gateway/src/gateway/gateway_rpc_guild_voice_util.erl +++ b/fluxer_gateway/src/gateway/gateway_rpc_guild_voice_util.erl @@ -6,7 +6,6 @@ -export([ handle_move_member_result/4, - handle_repair_result/1, normalize_voice_rpc_error/1, parse_voice_update/1, process_voice_update/1 @@ -58,18 +57,6 @@ cleanup_virtual_access_on_disconnect(DisconnectedUserId, Pid) -> spawn(fun() -> guild_voice:cleanup_virtual_access_on_disconnect(DisconnectedUserId, Pid) end), ok. --spec handle_repair_result(term()) -> map(). -handle_repair_result(#{success := true, repaired := Repaired}) -> - #{<<"success">> => true, <<"repaired">> => Repaired}; -handle_repair_result(#{success := true}) -> - #{<<"success">> => true}; -handle_repair_result(#{success := false, error := Error}) -> - #{<<"success">> => false, <<"error">> => normalize_voice_rpc_error(Error)}; -handle_repair_result(#{error := Error}) -> - gateway_rpc_error:raise(normalize_voice_rpc_error(Error)); -handle_repair_result(_) -> - #{<<"success">> => false, <<"error">> => <<"repair_voice_state_error">>}. - -spec voice_session_data([term()]) -> [map()]. voice_session_data(SessionData) -> [Entry || Entry <- SessionData, is_map(Entry)]. diff --git a/fluxer_gateway/src/gateway/gateway_rpc_misc.erl b/fluxer_gateway/src/gateway/gateway_rpc_misc.erl index 5f3682fef..2c4bb53c6 100644 --- a/fluxer_gateway/src/gateway/gateway_rpc_misc.erl +++ b/fluxer_gateway/src/gateway/gateway_rpc_misc.erl @@ -8,8 +8,7 @@ get_local_node_id/0, get_local_node_stats/0, get_local_memory_stats/1, - get_local_voice_state_counts/0, - get_local_active_voice_rooms/0 + get_local_voice_state_counts/0 ]). -define(DEFAULT_MEMORY_STATS_LIMIT, 100). @@ -31,9 +30,6 @@ execute_method(<<"process.node_stats">>, _Params) -> execute_method(<<"process.voice_state_counts">>, _Params) -> ActiveNodes = voice_state_count_nodes(), gateway_rpc_misc_presence:collect_and_aggregate_voice_state_counts(ActiveNodes); -execute_method(<<"process.active_voice_rooms">>, _Params) -> - ActiveNodes = gateway_node_router:active_nodes(), - gateway_rpc_misc_push:collect_and_aggregate_active_voice_rooms(ActiveNodes); execute_method(Method, _Params) -> gateway_rpc_error:raise(<<"Unknown method: ", Method/binary>>). @@ -53,10 +49,6 @@ get_local_memory_stats(Limit) -> get_local_voice_state_counts() -> gateway_rpc_misc_presence:get_local_voice_state_counts(). --spec get_local_active_voice_rooms() -> map(). -get_local_active_voice_rooms() -> - gateway_rpc_misc_push:get_local_active_voice_rooms(). - -spec parse_memory_stats_limit(map()) -> pos_integer(). parse_memory_stats_limit(Params) -> Limit = diff --git a/fluxer_gateway/src/gateway/gateway_rpc_misc_push.erl b/fluxer_gateway/src/gateway/gateway_rpc_misc_push.erl deleted file mode 100644 index 5f002b17a..000000000 --- a/fluxer_gateway/src/gateway/gateway_rpc_misc_push.erl +++ /dev/null @@ -1,356 +0,0 @@ -%% SPDX-License-Identifier: AGPL-3.0-or-later - --module(gateway_rpc_misc_push). --typing([eqwalizer]). - --export([ - get_local_active_voice_rooms/0, - collect_and_aggregate_active_voice_rooms/1 -]). - --define(NODE_RPC_TIMEOUT, 10000). - --spec get_local_active_voice_rooms() -> map(). -get_local_active_voice_rooms() -> - #{ - <<"node_id">> => gateway_rpc_misc_session:node_id(node()), - <<"rooms">> => get_local_guild_active_voice_rooms() ++ get_local_dm_active_voice_rooms() - }. - --spec collect_and_aggregate_active_voice_rooms([node()]) -> map(). -collect_and_aggregate_active_voice_rooms(Nodes) -> - NodeRooms = [ - fetch_node_active_voice_rooms(N) - || N <- gateway_rpc_misc_session:normalize_nodes(Nodes) - ], - aggregate_active_voice_rooms(NodeRooms). - --spec get_local_guild_active_voice_rooms() -> [map()]. -get_local_guild_active_voice_rooms() -> - try - ets:foldl( - fun - ({GuildId, Pid}, Acc) when is_integer(GuildId), is_pid(Pid) -> - get_guild_active_voice_rooms(GuildId, Pid) ++ Acc; - (_Row, Acc) -> - Acc - end, - [], - guild_voice_registry - ) - catch - error:badarg -> - [] - end. - --spec get_guild_active_voice_rooms(integer(), pid()) -> [map()]. -get_guild_active_voice_rooms(GuildId, Pid) -> - case process_liveness:is_alive(Pid) of - true -> fetch_guild_voice_states(GuildId, Pid); - false -> [] - end. - --spec fetch_guild_voice_states(integer(), pid()) -> [map()]. -fetch_guild_voice_states(GuildId, Pid) -> - try gen_server:call(Pid, {get_voice_states_map}, 500) of - VoiceStates when is_map(VoiceStates) -> - build_guild_voice_room_entries(GuildId, VoiceStates); - _ -> - [] - catch - _:_ -> [] - end. - --spec build_guild_voice_room_entries(integer(), map()) -> [map()]. -build_guild_voice_room_entries(GuildId, VoiceStates) -> - Counts = maps:fold( - fun(_ConnId, VoiceState, Acc) -> - fold_voice_state(VoiceState, Acc) - end, - #{}, - VoiceStates - ), - [ - #{ - <<"guild_id">> => integer_to_binary(GuildId), - <<"channel_id">> => ChannelId, - <<"voice_state_count">> => Count - } - || {ChannelId, Count} <- lists:sort(maps:to_list(Counts)), - Count > 0 - ]. - --spec fold_voice_state(term(), map()) -> map(). -fold_voice_state(State, Acc) when is_map(State) -> - ChannelId = normalize_id(maps:get(<<"channel_id">>, State, undefined)), - increment_channel_count(ChannelId, Acc); -fold_voice_state(_, Acc) -> - Acc. - --spec increment_channel_count(binary() | undefined, map()) -> map(). -increment_channel_count(undefined, Acc) -> - Acc; -increment_channel_count(ChannelId, Acc) -> - maps:update_with(ChannelId, fun(C) -> C + 1 end, 1, Acc). - --spec get_local_dm_active_voice_rooms() -> [map()]. -get_local_dm_active_voice_rooms() -> - RawIds = shard_utils:safe_apply(fun call_manager:local_call_ids/0, []), - CallIds = integer_entries(RawIds), - lists:flatmap(fun get_dm_active_voice_room/1, CallIds). - --spec integer_entries(term()) -> [integer()]. -integer_entries(Values) when is_list(Values) -> - [Value || Value <- Values, is_integer(Value)]; -integer_entries(_) -> - []. - --spec get_dm_active_voice_room(integer()) -> [map()]. -get_dm_active_voice_room(ChannelId) -> - case lookup_call(ChannelId) of - {ok, Pid} -> get_dm_room_from_pid(ChannelId, Pid); - _ -> [] - end. - --spec lookup_call(integer()) -> {ok, pid()} | error. -lookup_call(ChannelId) -> - try call_manager:lookup(ChannelId) of - {ok, Pid} when is_pid(Pid) -> {ok, Pid}; - _ -> error - catch - _:_ -> error - end. - --spec get_dm_room_from_pid(integer(), pid()) -> [map()]. -get_dm_room_from_pid(ChannelId, Pid) -> - try gen_server:call(Pid, {get_state}, 500) of - {ok, CallData} when is_map(CallData) -> - build_dm_room_entry(ChannelId, CallData); - _ -> - [] - catch - _:_ -> [] - end. - --spec build_dm_room_entry(integer(), map()) -> [map()]. -build_dm_room_entry(ChannelId, CallData) -> - VoiceStates = maps:get( - voice_states, - CallData, - maps:get(<<"voice_states">>, CallData, []) - ), - Count = - case VoiceStates of - Values when is_list(Values) -> length(Values); - _ -> 0 - end, - case Count > 0 of - true -> - [ - #{ - <<"guild_id">> => null, - <<"channel_id">> => integer_to_binary(ChannelId), - <<"voice_state_count">> => Count - } - ]; - false -> - [] - end. - --spec fetch_node_active_voice_rooms(node()) -> map(). -fetch_node_active_voice_rooms(TargetNode) -> - case - gateway_rpc_misc_session:safe_node_call( - TargetNode, get_local_active_voice_rooms, [], ?NODE_RPC_TIMEOUT - ) - of - #{<<"rooms">> := Rooms} = Result when is_list(Rooms) -> - Result#{<<"node_id">> => gateway_rpc_misc_session:node_id(TargetNode)}; - _ -> - #{ - <<"node_id">> => gateway_rpc_misc_session:node_id(TargetNode), - <<"rooms">> => [] - } - end. - --spec aggregate_active_voice_rooms([term()]) -> map(). -aggregate_active_voice_rooms(NodeRooms) -> - {RoomCounts, NodeCount} = lists:foldl( - fun merge_node_rooms/2, - {#{}, 0}, - NodeRooms - ), - Rooms = [ - build_room_entry(RoomKey, Count) - || {RoomKey, Count} <- lists:sort(maps:to_list(RoomCounts)), - Count > 0 - ], - #{<<"node_count">> => NodeCount, <<"rooms">> => Rooms}. - --spec merge_node_rooms(term(), {map(), non_neg_integer()}) -> {map(), non_neg_integer()}. -merge_node_rooms(NodeRooms, {RoomAcc, NodeCount}) when is_map(NodeRooms) -> - Rooms = maps:get(<<"rooms">>, NodeRooms, []), - MergedRooms = - case Rooms of - RoomList when is_list(RoomList) -> - lists:foldl(fun merge_single_room/2, RoomAcc, RoomList); - _ -> - RoomAcc - end, - {MergedRooms, NodeCount + 1}; -merge_node_rooms(_, Acc) -> - Acc. - --spec merge_single_room(term(), map()) -> map(). -merge_single_room(Room, Acc) when is_map(Room) -> - ChannelId = normalize_id(maps:get(<<"channel_id">>, Room, undefined)), - GuildIdResult = normalize_room_guild_id(maps:get(<<"guild_id">>, Room, null)), - Count = gateway_rpc_misc_session:decode_integer( - maps:get(<<"voice_state_count">>, Room, 0) - ), - apply_room_merge(ChannelId, GuildIdResult, Count, Acc); -merge_single_room(_, Acc) -> - Acc. - --spec apply_room_merge( - binary() | undefined, {ok, binary() | null} | error, non_neg_integer(), map() -) -> map(). -apply_room_merge(undefined, _, _, Acc) -> - Acc; -apply_room_merge(_, error, _, Acc) -> - Acc; -apply_room_merge(_, _, Count, Acc) when Count =< 0 -> Acc; -apply_room_merge(ChannelId, {ok, GuildId}, Count, Acc) -> - Key = - case GuildId of - null -> {dm, ChannelId}; - _ -> {guild, GuildId, ChannelId} - end, - maps:update_with(Key, fun(Current) -> Current + Count end, Count, Acc). - --spec build_room_entry(tuple(), non_neg_integer()) -> map(). -build_room_entry({dm, ChannelId}, Count) -> - #{ - <<"guild_id">> => null, - <<"channel_id">> => ChannelId, - <<"voice_state_count">> => Count - }; -build_room_entry({guild, GuildId, ChannelId}, Count) -> - #{ - <<"guild_id">> => GuildId, - <<"channel_id">> => ChannelId, - <<"voice_state_count">> => Count - }. - --spec normalize_id(term()) -> binary() | undefined. -normalize_id(Value) when is_binary(Value), byte_size(Value) > 0 -> - case validation:validate_snowflake(Value) of - {ok, Id} when Id > 0 -> integer_to_binary(Id); - _ -> undefined - end; -normalize_id(Value) when is_integer(Value), Value > 0 -> - integer_to_binary(Value); -normalize_id(Value) when is_list(Value), Value =/= [] -> - normalize_integer_id(type_conv:to_integer(Value)); -normalize_id(_) -> - undefined. - --spec normalize_integer_id(integer() | undefined) -> binary() | undefined. -normalize_integer_id(Id) when is_integer(Id), Id > 0 -> - integer_to_binary(Id); -normalize_integer_id(_) -> - undefined. - --spec normalize_room_guild_id(term()) -> {ok, binary() | null} | error. -normalize_room_guild_id(null) -> - {ok, null}; -normalize_room_guild_id(undefined) -> - {ok, null}; -normalize_room_guild_id(Value) -> - case normalize_id(Value) of - undefined -> error; - Normalized -> {ok, Normalized} - end. - --ifdef(TEST). --include_lib("eunit/include/eunit.hrl"). - -malformed_test_node_rooms() -> - [ - #{ - <<"rooms">> => [ - #{ - <<"guild_id">> => <<"10">>, - <<"channel_id">> => <<"20">>, - <<"voice_state_count">> => <<"2">> - }, - #{ - <<"guild_id">> => 10, - <<"channel_id">> => 20, - <<"voice_state_count">> => 3 - }, - #{ - <<"guild_id">> => null, - <<"channel_id">> => "30", - <<"voice_state_count">> => "4" - }, - #{ - <<"guild_id">> => <<"bad">>, - <<"channel_id">> => <<"40">>, - <<"voice_state_count">> => 5 - }, - #{ - <<"guild_id">> => <<"10">>, - <<"channel_id">> => <<"bad">>, - <<"voice_state_count">> => 5 - }, - #{ - <<"guild_id">> => <<"10">>, - <<"channel_id">> => <<"50">>, - <<"voice_state_count">> => 0 - }, - not_a_room - ] - }, - #{<<"rooms">> => not_a_list}, - not_a_node - ]. - -aggregate_active_voice_rooms_ignores_malformed_nodes_and_rooms_test() -> - Aggregate = aggregate_active_voice_rooms(malformed_test_node_rooms()), - ?assertEqual(2, maps:get(<<"node_count">>, Aggregate)), - ?assertEqual( - [ - #{ - <<"guild_id">> => null, - <<"channel_id">> => <<"30">>, - <<"voice_state_count">> => 4 - }, - #{ - <<"guild_id">> => <<"10">>, - <<"channel_id">> => <<"20">>, - <<"voice_state_count">> => 5 - } - ], - maps:get(<<"rooms">>, Aggregate) - ). - -build_guild_voice_room_entries_ignores_malformed_voice_states_test() -> - VoiceStates = #{ - <<"good">> => #{<<"channel_id">> => <<"20">>}, - <<"same">> => #{<<"channel_id">> => 20}, - <<"bad_channel">> => #{<<"channel_id">> => <<"bad">>}, - <<"not_a_map">> => not_a_map - }, - ?assertEqual( - [ - #{ - <<"guild_id">> => <<"10">>, - <<"channel_id">> => <<"20">>, - <<"voice_state_count">> => 2 - } - ], - build_guild_voice_room_entries(10, VoiceStates) - ). - --endif. diff --git a/fluxer_gateway/src/gateway/gateway_rpc_voice.erl b/fluxer_gateway/src/gateway/gateway_rpc_voice.erl index ad002abfd..3b280911b 100644 --- a/fluxer_gateway/src/gateway/gateway_rpc_voice.erl +++ b/fluxer_gateway/src/gateway/gateway_rpc_voice.erl @@ -13,8 +13,6 @@ -spec execute_method(binary(), map()) -> term(). execute_method(<<"voice.confirm_connection">>, P) -> handle_confirm_connection(P); -execute_method(<<"voice.repair_state_from_cache">>, P) -> - handle_repair_state(P); execute_method(<<"voice.disconnect_user_if_in_channel">>, P) -> handle_disconnect_if_in_channel(P); execute_method(<<"voice.get_voice_states_for_channel">>, P) -> @@ -46,18 +44,6 @@ handle_confirm_connection(Params) -> ) end. --spec handle_repair_state(map()) -> map() | term(). -handle_repair_state(Params) -> - case parse_optional_guild_id(Params) of - undefined -> - #{<<"success">> => false, <<"error">> => <<"voice_not_supported">>}; - GuildId -> - gateway_rpc_guild:execute_method( - <<"guild.repair_voice_state_from_cache">>, - Params#{<<"guild_id">> => integer_to_binary(GuildId)} - ) - end. - -spec handle_disconnect_if_in_channel(map()) -> term(). handle_disconnect_if_in_channel(Params) -> ChannelIdBin = maps:get(<<"channel_id">>, Params), diff --git a/fluxer_gateway/src/guild/guild.erl b/fluxer_gateway/src/guild/guild.erl index 3bf87e5a0..6b7664fc2 100644 --- a/fluxer_gateway/src/guild/guild.erl +++ b/fluxer_gateway/src/guild/guild.erl @@ -45,8 +45,6 @@ handle_call({session_connect, Request}, {CallerPid, _}, State) -> handle_session_connect_call(Request, CallerPid, State); handle_call(export_handoff_state, _From, State) -> {reply, {ok, guild_handoff:export_handoff_state(State)}, State}; -handle_call({get_cached_voice_state_by_connection, ConnectionId}, _From, State) -> - handle_cached_voice_state_call(ConnectionId, State); handle_call({get_guild_id}, _From, State) -> {reply, maps:get(id, State, undefined), State}; handle_call({get_voice_guild_state}, _From, State) -> @@ -239,10 +237,6 @@ session_connect_pid(#{session_pid := Pid}, _CallerPid) -> session_connect_pid(_Request, CallerPid) -> CallerPid. --spec handle_cached_voice_state_call(term(), guild_state()) -> call_reply(). -handle_cached_voice_state_call(ConnectionId, State) when is_binary(ConnectionId) -> - guild_voice_lifecycle:reply_cached_voice_state(ConnectionId, State). - -spec handle_reload_call(term(), guild_state()) -> call_reply(). handle_reload_call(NewData, State) when is_map(NewData) -> guild_init:handle_reload(NewData, State). diff --git a/fluxer_gateway/src/guild/guild_voice_lifecycle.erl b/fluxer_gateway/src/guild/guild_voice_lifecycle.erl index 704019170..99f68b108 100644 --- a/fluxer_gateway/src/guild/guild_voice_lifecycle.erl +++ b/fluxer_gateway/src/guild/guild_voice_lifecycle.erl @@ -7,7 +7,6 @@ ensure_voice_server/1, handle_voice_server_exit/3, reply_voice_server_pid/1, - reply_cached_voice_state/2, clear_stale_cached_voice_states/2 ]). @@ -61,15 +60,6 @@ reply_voice_server_pid(State) -> {{error, Reason}, NewState} -> {reply, {error, Reason}, NewState} end. --spec reply_cached_voice_state(binary(), guild_state()) -> - {reply, {ok, map()} | {error, not_found}, guild_state()}. -reply_cached_voice_state(ConnectionId, State) -> - VoiceStates = maps:get(voice_states, State, #{}), - case maps:find(ConnectionId, VoiceStates) of - {ok, VoiceState} -> {reply, {ok, VoiceState}, State}; - error -> {reply, {error, not_found}, State} - end. - -spec clear_stale_cached_voice_states([binary()], guild_state()) -> guild_state(). clear_stale_cached_voice_states(ConnectionIds, State) -> case read_authoritative_voice_states(State) of diff --git a/fluxer_gateway/src/guild/voice/guild_voice_server.erl b/fluxer_gateway/src/guild/voice/guild_voice_server.erl index 167f984aa..a7bdc35f9 100644 --- a/fluxer_gateway/src/guild/voice/guild_voice_server.erl +++ b/fluxer_gateway/src/guild/voice/guild_voice_server.erl @@ -189,13 +189,6 @@ voice_request_handler(_) -> error. -spec handle_call_local(term(), server_state()) -> {reply, term(), server_state()}. -handle_call_local({repair_voice_state_from_guild_cache, Request}, State) when is_map(Request) -> - {Reply, NewState} = guild_voice_server_state:repair_voice_state_from_guild_cache( - Request, State - ), - {reply, Reply, NewState}; -handle_call_local({repair_voice_state_from_guild_cache, _Request}, State) -> - {reply, #{success => false, error => voice_invalid_state}, State}; handle_call_local({store_pending_connection, ConnectionId, Metadata}, State) -> Pending = maps:get(pending_voice_connections, State, #{}), NewPending = bounded_put(ConnectionId, Metadata, Pending, ?MAX_PENDING_CONNECTIONS), diff --git a/fluxer_gateway/src/guild/voice/guild_voice_server_state.erl b/fluxer_gateway/src/guild/voice/guild_voice_server_state.erl index 4f77dc31f..9fdf0dd10 100644 --- a/fluxer_gateway/src/guild/voice/guild_voice_server_state.erl +++ b/fluxer_gateway/src/guild/voice/guild_voice_server_state.erl @@ -10,7 +10,6 @@ local_voice_states_for_channel/2, local_pending_joins_for_channel/2, parse_voice_channel_id/1, - repair_voice_state_from_guild_cache/2, voice_state_rpc_entries/1, pending_join_rpc_entries/1, fetch_guild_data/1, @@ -117,114 +116,6 @@ parse_voice_channel_id(ChannelId) -> error end. --spec repair_voice_state_from_guild_cache(map(), server_state()) -> {map(), server_state()}. -repair_voice_state_from_guild_cache(Request, State) -> - ConnectionId = maps:get(connection_id, Request, undefined), - UserId = maps:get(user_id, Request, undefined), - ChannelId = maps:get(channel_id, Request, undefined), - case {ConnectionId, UserId, ChannelId} of - {Conn, UId, CId} when is_binary(Conn), is_integer(UId), is_integer(CId) -> - do_repair(Conn, UId, CId, State); - _ -> - {#{success => false, error => voice_invalid_state}, State} - end. - --spec do_repair(binary(), integer(), integer(), server_state()) -> {map(), server_state()}. -do_repair(ConnectionId, UserId, ChannelId, State) -> - VoiceStates = maps:get(voice_states, State, #{}), - case maps:get(ConnectionId, VoiceStates, undefined) of - ExistingVoiceState when is_map(ExistingVoiceState) -> - repair_existing_voice_state( - ExistingVoiceState, ConnectionId, UserId, ChannelId, State - ); - _ -> - repair_from_cached(ConnectionId, UserId, ChannelId, VoiceStates, State) - end. - --spec repair_existing_voice_state( - voice_state(), binary(), integer(), integer(), server_state() -) -> - {map(), server_state()}. -repair_existing_voice_state(ExistingVoiceState, ConnectionId, UserId, ChannelId, State) -> - case voice_state_matches(ExistingVoiceState, ConnectionId, UserId, ChannelId) of - true -> {#{success => true, repaired => false}, State}; - false -> {#{success => false, error => voice_state_mismatch}, State} - end. - --spec repair_from_cached(binary(), integer(), integer(), voice_state_map(), server_state()) -> - {map(), server_state()}. -repair_from_cached(ConnectionId, UserId, ChannelId, VoiceStates, State) -> - case fetch_cached_voice_state(ConnectionId, State) of - {ok, CachedVoiceState} -> - repair_with_cached( - ConnectionId, UserId, ChannelId, CachedVoiceState, VoiceStates, State - ); - {error, _Reason} -> - {#{success => false, error => voice_connection_not_found}, State} - end. - --spec repair_with_cached( - binary(), integer(), integer(), voice_state(), voice_state_map(), server_state() -) -> {map(), server_state()}. -repair_with_cached(ConnectionId, UserId, ChannelId, CachedVoiceState, VoiceStates, State) -> - case voice_state_matches(CachedVoiceState, ConnectionId, UserId, ChannelId) of - false -> - {#{success => false, error => voice_state_mismatch}, State}; - true -> - OldVoiceStates = maps:get(voice_states, State, #{}), - NewVoiceStates = VoiceStates#{ConnectionId => CachedVoiceState}, - _ = guild_voice_server_sync:sync_replaced_voice_states( - OldVoiceStates, NewVoiceStates - ), - PendingConns = maps:remove( - ConnectionId, maps:get(pending_voice_connections, State, #{}) - ), - RecentDisc = maps:remove( - ConnectionId, - maps:get(recently_disconnected_voice_states, State, #{}) - ), - NewState0 = State#{ - voice_states => NewVoiceStates, - pending_voice_connections => PendingConns, - recently_disconnected_voice_states => RecentDisc - }, - GuildState = build_guild_state(NewState0), - ChannelIdBin = maps:get(<<"channel_id">>, CachedVoiceState, null), - guild_voice_broadcast:broadcast_voice_state_update( - CachedVoiceState, GuildState, ChannelIdBin - ), - logger:warning( - voice_state_repaired_log_message(), - [maps:get(guild_id, State), ChannelId, UserId, ConnectionId] - ), - {#{success => true, repaired => true}, NewState0} - end. - --spec voice_state_repaired_log_message() -> string(). -voice_state_repaired_log_message() -> - "guild_voice_state_repaired_from_guild_cache: guild_id=~p " - "channel_id=~p user_id=~p connection_id=~p". - --spec fetch_cached_voice_state(binary(), server_state()) -> - {ok, voice_state()} | {error, not_found}. -fetch_cached_voice_state(ConnectionId, #{guild_pid := GuildPid}) when is_pid(GuildPid) -> - try gen_server:call(GuildPid, {get_cached_voice_state_by_connection, ConnectionId}, 1000) of - {ok, VoiceState} when is_map(VoiceState) -> {ok, VoiceState}; - _ -> {error, not_found} - catch - throw:_ -> {error, not_found}; - error:_ -> {error, not_found}; - exit:_ -> {error, not_found} - end; -fetch_cached_voice_state(_ConnectionId, _State) -> - {error, not_found}. - --spec voice_state_matches(voice_state(), binary(), integer(), integer()) -> boolean(). -voice_state_matches(VoiceState, ConnectionId, UserId, ChannelId) -> - maps:get(<<"connection_id">>, VoiceState, ConnectionId) =:= ConnectionId andalso - voice_state_utils:voice_state_user_id(VoiceState) =:= UserId andalso - voice_state_utils:voice_state_channel_id(VoiceState) =:= ChannelId. - -spec voice_state_rpc_entries(term()) -> [map()]. voice_state_rpc_entries(VoiceStates) when is_list(VoiceStates) -> lists:filtermap(fun voice_state_rpc_entry/1, VoiceStates);