diff --git a/fluxer_gateway/src/push/push.erl b/fluxer_gateway/src/push/push.erl index a3af565fa..08cfafd00 100644 --- a/fluxer_gateway/src/push/push.erl +++ b/fluxer_gateway/src/push/push.erl @@ -323,6 +323,7 @@ filter_eligible_users( ConnectedUsers ) -> LargeGuildMetadata = large_guild_metadata(GuildId), + push_eligibility:prefetch_user_guild_settings(UserIds, AuthorId, GuildId), lists:filter( fun(UserId) -> push_eligibility:is_eligible_for_push( @@ -588,6 +589,36 @@ filter_eligible_users_fetches_large_metadata_once_test() -> ) end. +filter_eligible_users_batches_missing_settings_lookups_test() -> + push_ets_cache:init(), + Self = self(), + ok = meck:new(rpc_client, [passthrough, no_link]), + try + ok = meck:expect(rpc_client, call, fun(Request) -> + Self ! {rpc_request, Request}, + {ok, #{<<"user_guild_settings">> => [#{}, #{}, #{}]}} + end), + ?assertEqual( + [1, 2, 3], + filter_eligible_users([1, 2, 3], 999, 43, 10, #{}, 0, #{}, #{}) + ), + ?assertEqual(1, drain_settings_request_count(0)) + after + meck:unload(rpc_client), + lists:foreach( + fun(UserId) -> push_ets_cache:delete_user_guild_settings(UserId, 43) end, + [1, 2, 3] + ) + end. + +drain_settings_request_count(Count) -> + receive + {rpc_request, #{<<"type">> := <<"get_user_guild_settings">>}} -> + drain_settings_request_count(Count + 1) + after 0 -> + Count + end. + drain_metadata_lookup_count(Count) -> receive metadata_lookup -> drain_metadata_lookup_count(Count + 1) diff --git a/fluxer_gateway/src/push/push_eligibility.erl b/fluxer_gateway/src/push/push_eligibility.erl index a22d4d333..c309fcaf1 100644 --- a/fluxer_gateway/src/push/push_eligibility.erl +++ b/fluxer_gateway/src/push/push_eligibility.erl @@ -6,6 +6,7 @@ -export([is_user_blocked/2]). -export([check_user_guild_settings/7]). -export([check_user_guild_settings/8]). +-export([prefetch_user_guild_settings/3]). -export([should_allow_notification/6]). -export([is_user_mentioned/5]). -export([is_eligible_for_push/8]). @@ -14,6 +15,7 @@ -define(MESSAGE_NOTIFICATIONS_NO_MESSAGES, 2). -define(MESSAGE_NOTIFICATIONS_ONLY_MENTIONS, 1). +-define(SETTINGS_PREFETCH_CHUNK_SIZE, 200). -spec is_eligible_for_push( integer(), integer(), integer(), integer(), map(), integer(), map(), map() @@ -226,6 +228,112 @@ fetch_settings_rpc(UserId, GuildId) -> exit:_ -> #{} end. +-spec prefetch_user_guild_settings([integer()], integer(), integer()) -> ok. +prefetch_user_guild_settings(_UserIds, _AuthorId, 0) -> + ok; +prefetch_user_guild_settings(UserIds, AuthorId, GuildId) -> + prefetch_settings_chunks(uncached_settings_user_ids(UserIds, AuthorId, GuildId), GuildId). + +-spec uncached_settings_user_ids([integer()], integer(), integer()) -> [integer()]. +uncached_settings_user_ids(UserIds, AuthorId, GuildId) -> + lists:usort( + lists:filter( + fun(UserId) -> is_settings_uncached(UserId, AuthorId, GuildId) end, + UserIds + ) + ). + +-spec is_settings_uncached(integer(), integer(), integer()) -> boolean(). +is_settings_uncached(AuthorId, AuthorId, _GuildId) -> + false; +is_settings_uncached(UserId, _AuthorId, GuildId) -> + push_ets_cache:get_user_guild_settings(UserId, GuildId) =:= undefined. + +-spec prefetch_settings_chunks([integer()], integer()) -> ok. +prefetch_settings_chunks([], _GuildId) -> + ok; +prefetch_settings_chunks(UserIds, GuildId) -> + {Chunk, Rest} = take_settings_chunk(UserIds, ?SETTINGS_PREFETCH_CHUNK_SIZE, []), + prefetch_settings_chunk(Chunk, GuildId), + prefetch_settings_chunks(Rest, GuildId). + +-spec take_settings_chunk([integer()], non_neg_integer(), [integer()]) -> + {[integer()], [integer()]}. +take_settings_chunk(Rest, 0, Acc) -> + {lists:reverse(Acc), Rest}; +take_settings_chunk([], _Remaining, Acc) -> + {lists:reverse(Acc), []}; +take_settings_chunk([UserId | Rest], Remaining, Acc) -> + take_settings_chunk(Rest, Remaining - 1, [UserId | Acc]). + +-spec prefetch_settings_chunk([integer()], integer()) -> ok. +prefetch_settings_chunk(UserIds, GuildId) -> + try prefetch_settings_chunk_rpc(UserIds, GuildId) of + ok -> ok + catch + throw:_ -> ok; + error:_ -> ok; + exit:_ -> ok + end. + +-spec prefetch_settings_chunk_rpc([integer()], integer()) -> ok. +prefetch_settings_chunk_rpc(UserIds, GuildId) -> + Req = #{ + <<"type">> => <<"get_user_guild_settings">>, + <<"user_ids">> => [integer_to_binary(UserId) || UserId <- UserIds], + <<"guild_id">> => integer_to_binary(GuildId) + }, + logger:debug( + "Push: prefetching user guild settings via RPC", + #{user_count => length(UserIds), guild_id => GuildId} + ), + case rpc_client:call(Req) of + {ok, Data} -> + cache_prefetched_settings(UserIds, GuildId, settings_list(Data)); + {error, Reason} -> + logger:debug( + "Push: RPC failed to prefetch user guild settings", + #{user_count => length(UserIds), guild_id => GuildId, reason => Reason} + ), + ok + end. + +-spec settings_list(map()) -> [term()]. +settings_list(Data) -> + case maps:get(<<"user_guild_settings">>, Data, []) of + Settings when is_list(Settings) -> Settings; + _ -> [] + end. + +-spec cache_prefetched_settings([integer()], integer(), [term()]) -> ok. +cache_prefetched_settings(UserIds, GuildId, Settings) when + length(UserIds) =:= length(Settings) +-> + lists:foreach( + fun({UserId, UserSettings}) -> + push_ets_cache:put_user_guild_settings( + UserId, GuildId, settings_map(UserSettings) + ) + end, + lists:zip(UserIds, Settings) + ); +cache_prefetched_settings(UserIds, GuildId, Settings) -> + logger:debug( + "Push: prefetched user guild settings did not match requested users", + #{ + user_count => length(UserIds), + guild_id => GuildId, + settings_count => length(Settings) + } + ), + ok. + +-spec settings_map(term()) -> map(). +settings_map(Settings) when is_map(Settings) -> + Settings; +settings_map(_Settings) -> + #{}. + -spec should_allow_notification( integer(), map(), integer(), map(), map(), map() ) -> boolean(). diff --git a/fluxer_gateway/test/push_tests.erl b/fluxer_gateway/test/push_tests.erl index 07de7f195..ea4bc1fd0 100644 --- a/fluxer_gateway/test/push_tests.erl +++ b/fluxer_gateway/test/push_tests.erl @@ -114,6 +114,76 @@ message_params_context_requires_explicit_guild_id_test() -> {ok, Context} = push_message_params:context(DmParams), ?assertEqual(0, maps:get(guild_id, Context)). +prefetch_user_guild_settings_batches_missing_users_into_one_rpc_test() -> + push_ets_cache:init(), + ok = push_ets_cache:put_user_guild_settings(9004, 4242, #{<<"muted">> => true}), + Response = #{<<"user_guild_settings">> => [#{<<"mobile_push">> => false}, null]}, + with_rpc_client_stub({ok, Response}, fun() -> + ok = push_eligibility:prefetch_user_guild_settings( + [9002, 9001, 9004, 9999, 9001], 9999, 4242 + ) + end), + ?assertEqual([{<<"4242">>, [<<"9001">>, <<"9002">>]}], settings_requests()), + ?assertEqual( + #{<<"mobile_push">> => false}, push_ets_cache:get_user_guild_settings(9001, 4242) + ), + ?assertEqual(#{}, push_ets_cache:get_user_guild_settings(9002, 4242)), + ?assertEqual( + #{<<"muted">> => true}, push_ets_cache:get_user_guild_settings(9004, 4242) + ), + lists:foreach( + fun(UserId) -> push_ets_cache:delete_user_guild_settings(UserId, 4242) end, + [9001, 9002, 9004] + ). + +prefetch_user_guild_settings_chunks_large_batches_test() -> + push_ets_cache:init(), + UserIds = lists:seq(90000, 90200), + with_rpc_client_stub({ok, #{<<"user_guild_settings">> => []}}, fun() -> + ok = push_eligibility:prefetch_user_guild_settings(UserIds, 9999, 4243) + end), + ?assertEqual([200, 1], [length(Ids) || {_GuildId, Ids} <- settings_requests()]). + +prefetch_user_guild_settings_leaves_cache_cold_on_rpc_failure_test() -> + push_ets_cache:init(), + with_rpc_client_stub({error, timeout}, fun() -> + ok = push_eligibility:prefetch_user_guild_settings([9007, 9008], 9999, 4244) + end), + ?assertEqual([{<<"4244">>, [<<"9007">>, <<"9008">>]}], settings_requests()), + ?assertEqual(undefined, push_ets_cache:get_user_guild_settings(9007, 4244)), + ?assertEqual(undefined, push_ets_cache:get_user_guild_settings(9008, 4244)). + +prefetch_user_guild_settings_skips_direct_messages_test() -> + push_ets_cache:init(), + with_rpc_client_stub({error, timeout}, fun() -> + ok = push_eligibility:prefetch_user_guild_settings([9007, 9008], 9999, 0) + end), + ?assertEqual([], settings_requests()). + +with_rpc_client_stub(Result, Fun) -> + Self = self(), + ok = meck:new(rpc_client, [passthrough, no_link]), + try + ok = meck:expect(rpc_client, call, fun(Request) -> + Self ! {rpc_request, Request}, + Result + end), + Fun() + after + meck:unload(rpc_client) + end. + +settings_requests() -> + receive + {rpc_request, #{<<"type">> := <<"get_user_guild_settings">>} = Request} -> + [ + {maps:get(<<"guild_id">>, Request), maps:get(<<"user_ids">>, Request)} + | settings_requests() + ] + after 0 -> + [] + end. + erase_persistent_term(Key) -> try persistent_term:erase(Key) of _ -> ok