streams: Add bulk_access_stream_metadata_user_ids.

This function will be useful in sending events for users gaining or
losing metadata access when the members of a user group change in any
way.
This commit is contained in:
Shubham Padia
2025-03-20 08:29:51 +00:00
committed by Tim Abbott
parent 55ea5be022
commit 208ee1b8d9
3 changed files with 258 additions and 10 deletions

View File

@@ -18,7 +18,9 @@ from zerver.lib.exceptions import (
)
from zerver.lib.stream_subscription import (
get_active_subscriptions_for_stream_id,
get_guest_user_ids_for_streams,
get_subscribed_stream_ids_for_user,
get_user_ids_for_streams,
)
from zerver.lib.stream_traffic import get_average_weekly_stream_traffic, get_streams_traffic
from zerver.lib.string_validation import check_stream_name
@@ -993,6 +995,47 @@ def can_access_stream_metadata_user_ids(stream: Stream) -> set[int]:
)
def bulk_can_access_stream_metadata_user_ids(streams: list[Stream]) -> dict[int, set[int]]:
# return user ids of users who can access the attributes of a
# stream, such as its name/description. Useful for sending events
# to all users with access to a stream's attributes.
result: dict[int, set[int]] = {}
public_streams = []
private_streams = []
for stream in streams:
if stream.is_public():
public_streams.append(stream)
else:
private_streams.append(stream)
if len(public_streams) > 0:
guest_subscriptions = get_guest_user_ids_for_streams(
{stream.id for stream in public_streams}
)
active_non_guest_user_id_set = set(active_non_guest_user_ids(public_streams[0].realm_id))
for stream in public_streams:
result[stream.id] = set(active_non_guest_user_id_set | guest_subscriptions[stream.id])
if len(private_streams) > 0:
private_stream_user_ids = get_user_ids_for_streams(
{stream.id for stream in private_streams}
)
admin_users_and_bots = {user.id for user in stream.realm.get_admin_users_and_bots()}
users_dict_with_metadata_access_to_streams_via_permission_groups = (
get_users_dict_with_metadata_access_to_streams_via_permission_groups(
private_streams, private_streams[0].realm_id
)
)
for stream in private_streams:
result[stream.id] = (
private_stream_user_ids[stream.id]
| admin_users_and_bots
| users_dict_with_metadata_access_to_streams_via_permission_groups[stream.id]
)
return result
def can_access_stream_history(user_profile: UserProfile, stream: Stream) -> bool:
"""Determine whether the provided user is allowed to access the
history of the target stream.