streams: Send user creation events on adding subscribers.

This commit adds code to send user creation events to
guests who gain access to new subscribers and to the
new guest subscribers who gain access to existing
stream subscribers.

api_docs/changelog.md

@@ -38,6 +38,9 @@ format used by the Zulip server that they are interacting with.
   `CAN_ACCESS_ALL_USERS_GROUP_LIMITS_PRESENCE` server setting is set
   to `true`.
 
+* [`GET /events`](/api/get-events): `realm_user` events with `op: "add"`
+  are now also sent when a guest user gains access to a user.
+
 **Feature level 227**
 
 * [`PATCH /realm/user_settings_defaults`](/api/update-realm-user-settings-defaults),

zerver/actions/create_user.py

@@ -303,7 +303,7 @@ def process_new_human_user(
     send_initial_direct_message(user_profile)
 
 
-def notify_created_user(user_profile: UserProfile) -> None:
+def notify_created_user(user_profile: UserProfile, notify_user_ids: List[int]) -> None:
     user_row = user_profile_to_user_row(user_profile)
 
     format_user_row_kwargs: Dict[str, Any] = {
@@ -324,6 +324,15 @@ def notify_created_user(user_profile: UserProfile) -> None:
 
     user_ids_without_access_to_created_user: List[int] = []
     users_with_access_to_created_users: List[UserProfile] = []
+
+    if notify_user_ids:
+        # This is currently used to send creation event when a guest
+        # gains access to a user, so we depend on the caller to make
+        # sure that only accessible users receive the user data.
+        users_with_access_to_created_users = list(
+            user_profile.realm.get_active_users().filter(id__in=notify_user_ids)
+        )
+    else:
         active_realm_users = list(user_profile.realm.get_active_users())
 
         # This call to user_access_restricted_in_realm results in
@@ -538,7 +547,7 @@ def do_create_user(
 
     # Note that for bots, the caller will send an additional event
     # with bot-specific info like services.
-    notify_created_user(user_profile)
+    notify_created_user(user_profile, [])
 
     do_send_user_group_members_update_event("add_members", system_user_group, [user_profile.id])
     if user_profile.role == UserProfile.ROLE_MEMBER and not user_profile.is_provisional_member:
@@ -617,7 +626,7 @@ def do_activate_mirror_dummy_user(
         if settings.BILLING_ENABLED:
             update_license_ledger_if_needed(user_profile.realm, event_time)
 
-    notify_created_user(user_profile)
+    notify_created_user(user_profile, [])
 
 
 @transaction.atomic(savepoint=False)

zerver/actions/streams.py

@@ -34,6 +34,7 @@ from zerver.lib.stream_subscription import (
     get_active_subscriptions_for_stream_id,
     get_bulk_stream_subscriber_info,
     get_used_colors_for_user_ids,
+    get_users_for_streams,
 )
 from zerver.lib.stream_traffic import get_streams_traffic
 from zerver.lib.streams import (
@@ -47,6 +48,10 @@ from zerver.lib.streams import (
 )
 from zerver.lib.subscription_info import get_subscribers_query
 from zerver.lib.types import APISubscriptionDict
+from zerver.lib.users import (
+    get_subscribers_of_target_user_subscriptions,
+    get_users_involved_in_dms_with_target_users,
+)
 from zerver.models import (
     ArchivedAttachment,
     Attachment,
@@ -58,6 +63,7 @@ from zerver.models import (
     Recipient,
     Stream,
     Subscription,
+    SystemGroups,
     UserGroup,
     UserProfile,
     active_non_guest_user_ids,
@@ -592,6 +598,65 @@ def send_peer_subscriber_events(
             send_event_on_commit(realm, event, peer_user_ids)
 
 
+def send_user_creation_events_on_adding_subscriptions(
+    realm: Realm,
+    altered_user_dict: Dict[int, Set[int]],
+    altered_streams_dict: Dict[UserProfile, Set[int]],
+    subscribers_of_altered_user_subscriptions: Dict[int, Set[int]],
+) -> None:
+    altered_users = list(altered_streams_dict.keys())
+    non_guest_user_ids = active_non_guest_user_ids(realm.id)
+
+    users_involved_in_dms = get_users_involved_in_dms_with_target_users(altered_users, realm)
+
+    altered_stream_ids = altered_user_dict.keys()
+    subscribers_dict = get_users_for_streams(set(altered_stream_ids))
+
+    subscribers_user_id_map: Dict[int, UserProfile] = {}
+    subscriber_ids_dict: Dict[int, Set[int]] = defaultdict(set)
+    for stream_id, subscribers in subscribers_dict.items():
+        for user in subscribers:
+            subscriber_ids_dict[stream_id].add(user.id)
+            subscribers_user_id_map[user.id] = user
+
+    from zerver.actions.create_user import notify_created_user
+
+    for user in altered_users:
+        streams_for_user = altered_streams_dict[user]
+        subscribers_in_altered_streams: Set[int] = set()
+        for stream_id in streams_for_user:
+            subscribers_in_altered_streams |= subscriber_ids_dict[stream_id]
+
+        users_already_with_access_to_altered_user = (
+            set(non_guest_user_ids)
+            | subscribers_of_altered_user_subscriptions[user.id]
+            | users_involved_in_dms[user.id]
+            | {user.id}
+        )
+
+        users_to_receive_creation_event = (
+            subscribers_in_altered_streams - users_already_with_access_to_altered_user
+        )
+        if users_to_receive_creation_event:
+            notify_created_user(user, list(users_to_receive_creation_event))
+
+        if user.is_guest:
+            # If the altered user is a guest, then the user may receive
+            # user creation events for subscribers of the new stream.
+            users_already_accessible_to_altered_user = (
+                subscribers_of_altered_user_subscriptions[user.id]
+                | users_involved_in_dms[user.id]
+                | {user.id}
+            )
+
+            new_accessible_user_ids = (
+                subscribers_in_altered_streams - users_already_accessible_to_altered_user
+            )
+            for accessible_user_id in new_accessible_user_ids:
+                accessible_user = subscribers_user_id_map[accessible_user_id]
+                notify_created_user(accessible_user, [user.id])
+
+
 SubT: TypeAlias = Tuple[List[SubInfo], List[SubInfo]]
 
 
@@ -680,6 +745,21 @@ def bulk_add_subscriptions(
         # We can return early if users are already subscribed to all the streams.
         return ([], already_subscribed)
 
+    altered_user_dict: Dict[int, Set[int]] = defaultdict(set)
+    altered_guests: Set[int] = set()
+    altered_streams_dict: Dict[UserProfile, Set[int]] = defaultdict(set)
+    for sub_info in subs_to_add + subs_to_activate:
+        altered_user_dict[sub_info.stream.id].add(sub_info.user.id)
+        altered_streams_dict[sub_info.user].add(sub_info.stream.id)
+        if sub_info.user.is_guest:
+            altered_guests.add(sub_info.user.id)
+
+    if realm.can_access_all_users_group.name != SystemGroups.EVERYONE:
+        altered_users = list(altered_streams_dict.keys())
+        subscribers_of_altered_user_subscriptions = get_subscribers_of_target_user_subscriptions(
+            altered_users
+        )
+
     bulk_add_subs_to_db_with_logging(
         realm=realm,
         acting_user=acting_user,
@@ -687,13 +767,6 @@ def bulk_add_subscriptions(
         subs_to_activate=subs_to_activate,
     )
 
-    altered_user_dict: Dict[int, Set[int]] = defaultdict(set)
-    altered_guests: Set[int] = set()
-    for sub_info in subs_to_add + subs_to_activate:
-        altered_user_dict[sub_info.stream.id].add(sub_info.user.id)
-        if sub_info.user.is_guest:
-            altered_guests.add(sub_info.user.id)
-
     stream_dict = {stream.id: stream for stream in streams}
 
     new_streams = [stream_dict[stream_id] for stream_id in altered_user_dict]
@@ -721,6 +794,14 @@ def bulk_add_subscriptions(
             subscriber_dict=subscriber_peer_info.subscribed_ids,
         )
 
+    if realm.can_access_all_users_group.name != SystemGroups.EVERYONE:
+        send_user_creation_events_on_adding_subscriptions(
+            realm,
+            altered_user_dict,
+            altered_streams_dict,
+            subscribers_of_altered_user_subscriptions,
+        )
+
     send_peer_subscriber_events(
         op="peer_add",
         realm=realm,

zerver/lib/stream_subscription.py

@@ -163,6 +163,21 @@ def get_user_ids_for_streams(stream_ids: Set[int]) -> Dict[int, Set[int]]:
     return result
 
 
+def get_users_for_streams(stream_ids: Set[int]) -> Dict[int, Set[UserProfile]]:
+    all_subs = (
+        get_active_subscriptions_for_stream_ids(stream_ids)
+        .select_related("user_profile", "recipient")
+        .order_by("recipient__type_id")
+    )
+
+    result: Dict[int, Set[UserProfile]] = defaultdict(set)
+    for stream_id, rows in itertools.groupby(all_subs, key=lambda obj: obj.recipient.type_id):
+        users = {row.user_profile for row in rows}
+        result[stream_id] = users
+
+    return result
+
+
 def bulk_get_subscriber_peer_info(
     realm: Realm,
     streams: Collection[Stream],

zerver/openapi/zulip.yaml

@@ -1036,8 +1036,13 @@ paths:
                             - type: object
                               description: |
                                 Event sent to all users in a Zulip organization when a new
-                                user joins. Processing this event is important to being able
-                                to display basic details on other users given only their ID.
+                                user joins or when a guest user gains access to a user.
+                                Processing this event is important to being able to display
+                                basic details on other users given only their ID.
+
+                                **Changes**: Starting with Zulip 8.0 (feature level 228),
+                                this event is also sent when a guest user gains access to
+                                a user.
                               properties:
                                 id:
                                   $ref: "#/components/schemas/EventIdSchema"

zerver/tests/test_events.py

@@ -4061,6 +4061,39 @@ class SubscribeActionTest(BaseAction):
         events = self.verify_action(action, include_subscribers=include_subscribers, num_events=1)
         check_subscription_remove("events[0]", events[0])
 
+    def test_user_access_events_on_changing_subscriptions(self) -> None:
+        self.set_up_db_for_testing_user_access()
+        self.user_profile = self.example_user("polonius")
+        realm = self.user_profile.realm
+        stream = get_stream("test_stream1", realm)
+        othello = self.example_user("othello")
+        iago = self.example_user("iago")
+
+        subscribe_action = lambda: bulk_add_subscriptions(
+            realm, [stream], [othello, iago], acting_user=None
+        )
+        events = self.verify_action(subscribe_action, num_events=2)
+        check_realm_user_add("events[0]", events[0])
+        self.assertEqual(events[0]["person"]["user_id"], othello.id)
+        check_subscription_peer_add("events[1]", events[1])
+        self.assertEqual(set(events[1]["user_ids"]), {iago.id, othello.id})
+
+    def test_user_access_events_on_changing_subscriptions_for_guests(self) -> None:
+        self.set_up_db_for_testing_user_access()
+        polonius = self.example_user("polonius")
+        othello = self.example_user("othello")
+        self.user_profile = polonius
+        realm = self.user_profile.realm
+        stream = self.subscribe(self.example_user("othello"), "new_stream")
+        subscribe_action = lambda: bulk_add_subscriptions(
+            realm, [stream], [polonius, self.example_user("iago")], acting_user=None
+        )
+        events = self.verify_action(subscribe_action, num_events=3)
+        check_stream_create("events[0]", events[0])
+        check_subscription_add("events[1]", events[1])
+        check_realm_user_add("events[2]", events[2])
+        self.assertEqual(events[2]["person"]["user_id"], othello.id)
+
 
 class DraftActionTest(BaseAction):
     def do_enable_drafts_synchronization(self, user_profile: UserProfile) -> None:

zerver/tests/test_invite.py

@@ -108,7 +108,7 @@ class StreamSetupTest(ZulipTestCase):
 
         new_user = self.create_simple_new_user(realm, "alice@zulip.com")
 
-        with self.assert_database_query_count(11):
+        with self.assert_database_query_count(12):
             set_up_streams_for_new_human_user(
                 user_profile=new_user,
                 prereg_user=None,