chore(voice): remove voice reconciliation leftovers (#2810)

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)}"]

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 {
+import type {ILiveKitService, ListParticipantsResult} from '@app/api/infrastructure/ILiveKitService';
-	ILiveKitService,
-	ListActiveRoomsResult,
-	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<ListActiveRoomsResult> {
-		return {rooms: [], errors: [], searchedServers: 0, completed: true};
-	}
-
 	getDefaultRegionId(): string | null {
 		return null;
 	}

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<GatewayActiveVoiceRooms> {
-		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<bigint> {
 		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<string, string> = {
-			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<GatewayVoiceStateEntry>;
 	}> {

fluxer_api/src/api/infrastructure/IGatewayService.ts

@@ -88,17 +88,6 @@ export interface GatewayVoiceStateCounts {
 	servers: Array<GatewayVoiceStateServerCount>;
 }
 
-export interface GatewayActiveVoiceRoom {
-	guildId?: GuildID;
-	channelId: ChannelID;
-	voiceStateCount: number;
-}
-
-export interface GatewayActiveVoiceRooms {
-	rooms: Array<GatewayActiveVoiceRoom>;
-	nodeCount: number;
-}
-
 export interface GatewayVoiceStateEntry {
 	connectionId: string;
 	userId: string;
@@ -162,8 +151,6 @@ export abstract class IGatewayService {
 
 	abstract getVoiceStateCounts(): Promise<GatewayVoiceStateCounts>;
 
-	abstract getActiveVoiceRooms(): Promise<GatewayActiveVoiceRooms>;
-
 	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<GatewayVoiceStateEntry>;
 	}>;

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<LiveKitRoomLocation>;
-	errors: Array<LiveKitServerError>;
-	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<ListParticipantsResult>;
 
-	abstract listActiveRooms(): Promise<ListActiveRoomsResult>;
-
 	abstract getDefaultRegionId(): string | null;
 
 	abstract getRegionMetadata(): Array<VoiceRegionMetadata>;

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 {
+import type {ListParticipantsResult} from '@app/api/infrastructure/ILiveKitService';
-	ListActiveRoomsResult,
-	ListParticipantsResult,
-	LiveKitServerError,
-} 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<ListActiveRoomsResult> {
-		const rooms: ListActiveRoomsResult['rooms'] = [];
-		const errors: Array<LiveKitServerError> = [];
-		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<ServerClientConfig & {regionId: string; serverId: string}> {
-		const servers: Array<ServerClientConfig & {regionId: string; serverId: string}> = [];
-		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();
 	}

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;
 				}

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<GatewayActiveVoiceRooms> {
-		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<GatewayVoiceStateEntry>;
 	}> {

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,

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

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)].

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_voice_state_counts/0
-    get_local_active_voice_rooms/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 =

fluxer_gateway/src/gateway/gateway_rpc_misc_push.erl

@@ -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.

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),

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).

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

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),

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);