perf(push): batch guild settings lookups for push eligibility (#2175)

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)

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

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