stream_subscription: Add subscribe_ids_with_stream_history_access.

This new function returns the set of `user_ids` with access to the
stream's full history, for use in send_event calls.

zerver/lib/stream_subscription.py

@@ -246,3 +246,35 @@ def handle_stream_notifications_compatibility(
         stream_dict[notification_type] = (
             False if user_profile is None else getattr(user_profile, target_attr)
         )
+
+
+def subscriber_ids_with_stream_history_access(stream: Stream) -> Set[int]:
+    """Returns the set of active user IDs who can access any message
+    history on this stream (regardless of whether they have a
+    UserMessage) based on the stream's configuration.
+
+    1. if !history_public_to_subscribers:
+          History is not available to anyone
+    2. if history_public_to_subscribers and is_web_public:
+          All subscribers can access the history including guests
+    3. if history_public_to_subscribers and !is_web_public:
+          All subscribers can access the history excluding guests
+
+    """
+
+    if not stream.is_history_public_to_subscribers():
+        return set()
+
+    subscriptions = get_active_subscriptions_for_stream_id(stream.id)
+    if stream.is_web_public:
+        return set(
+            subscriptions.filter(user_profile__is_active=True).values_list(
+                "user_profile__id", flat=True
+            )
+        )
+
+    return set(
+        subscriptions.filter(user_profile__is_active=True)
+        .exclude(user_profile__role=600)
+        .values_list("user_profile__id", flat=True)
+    )

zerver/tests/test_subs.py

@@ -49,6 +49,7 @@ from zerver.lib.response import json_error, json_success
 from zerver.lib.stream_subscription import (
     get_active_subscriptions_for_stream_id,
     num_subscribers_for_stream_id,
+    subscriber_ids_with_stream_history_access,
 )
 from zerver.lib.streams import (
     StreamDict,
@@ -516,6 +517,38 @@ class StreamAdminTest(ZulipTestCase):
         self.assertFalse(stream.invite_only)
         self.assertTrue(stream.history_public_to_subscribers)
 
+    def test_subscriber_ids_with_stream_history_access(self) -> None:
+        hamlet = self.example_user("hamlet")
+        polonius = self.example_user("polonius")
+
+        stream1 = self.make_stream(
+            "history_private_stream", invite_only=True, history_public_to_subscribers=False
+        )
+        self.subscribe(hamlet, stream1.name)
+        self.subscribe(polonius, stream1.name)
+        self.assertEqual(set(), subscriber_ids_with_stream_history_access(stream1))
+
+        stream2 = self.make_stream(
+            "history_public_web_private_stream",
+            invite_only=True,
+            is_web_public=False,
+            history_public_to_subscribers=True,
+        )
+        self.subscribe(hamlet, stream2.name)
+        self.subscribe(polonius, stream2.name)
+        self.assertEqual({hamlet.id}, subscriber_ids_with_stream_history_access(stream2))
+
+        stream3 = self.make_stream(
+            "history_public_web_public_stream",
+            is_web_public=True,
+            history_public_to_subscribers=True,
+        )
+        self.subscribe(hamlet, stream3.name)
+        self.subscribe(polonius, stream3.name)
+        self.assertEqual(
+            {hamlet.id, polonius.id}, subscriber_ids_with_stream_history_access(stream3)
+        )
+
     def test_deactivate_stream_backend(self) -> None:
         user_profile = self.example_user("hamlet")
         self.login_user(user_profile)