refactor: Extract peer_data.js.

This de-clutters stream_data a bit.  Since our
peer data is our biggest performance concern,
I want to contain any optimizations to a fairly
well-focused module.

The name `peer_data` is a bit of a compromise,
since we already have `subs.js` and we use
`sub` as a variable name for stream records
throughout our code, but it's consistent with
our event nomenclature (peer/add, peer/remove)
and it's short while still being fairly easy
to find with grep.

frontend_tests/node_tests/compose.js

@@ -73,6 +73,7 @@ document.location.host = "foo.com";
 
 zrequire("zcommand");
 zrequire("compose_ui");
+const peer_data = zrequire("peer_data");
 const util = zrequire("util");
 zrequire("rtl");
 zrequire("common");
@@ -389,7 +390,7 @@ run_test("validate_stream_message", () => {
     assert(!$("#compose-all-everyone").visible());
     assert(!$("#compose-send-status").visible());
 
-    stream_data.get_subscriber_count = function (stream_id) {
+    peer_data.get_subscriber_count = function (stream_id) {
         assert.equal(stream_id, 101);
         return 16;
     };
@@ -965,13 +966,13 @@ run_test("warn_if_private_stream_is_linked", () => {
     };
 
     stream_data.add_sub(test_sub);
-    stream_data.set_subscribers(test_sub, [1, 2]);
+    peer_data.set_subscribers(test_sub, [1, 2]);
 
     let denmark = {
         stream_id: 100,
         name: "Denmark",
     };
-    stream_data.set_subscribers(denmark, [1, 2, 3]);
+    peer_data.set_subscribers(denmark, [1, 2, 3]);
 
     function test_noop_case(invite_only) {
         compose_state.set_message_type("stream");
@@ -1211,7 +1212,7 @@ run_test("needs_subscribe_warning", () => {
     };
 
     stream_data.add_sub(sub);
-    stream_data.set_subscribers(sub, [bob.user_id, me.user_id]);
+    peer_data.set_subscribers(sub, [bob.user_id, me.user_id]);
 
     blueslip.expect("error", "Unknown user_id in get_by_user_id: 999");
     // Test with an invalid user id.
@@ -1223,7 +1224,7 @@ run_test("needs_subscribe_warning", () => {
     // Test when user is subscribed to the stream.
     assert.equal(compose.needs_subscribe_warning(bob.user_id, sub.stream_id), false);
 
-    stream_data.remove_subscriber(sub.stream_id, bob.user_id);
+    peer_data.remove_subscriber(sub.stream_id, bob.user_id);
     // Test when the user is not subscribed.
     assert.equal(compose.needs_subscribe_warning(bob.user_id, sub.stream_id), true);
 });

frontend_tests/node_tests/compose_fade.js

@@ -6,6 +6,7 @@ const {set_global, zrequire} = require("../zjsunit/namespace");
 const {run_test} = require("../zjsunit/test");
 
 zrequire("stream_data");
+const peer_data = zrequire("peer_data");
 const people = zrequire("people");
 zrequire("compose_fade");
 
@@ -41,7 +42,7 @@ run_test("set_focused_recipient", () => {
         can_access_subscribers: true,
     };
     stream_data.add_sub(sub);
-    stream_data.set_subscribers(sub, [me.user_id, alice.user_id]);
+    peer_data.set_subscribers(sub, [me.user_id, alice.user_id]);
 
     set_global("$", (selector) => {
         switch (selector) {

frontend_tests/node_tests/dispatch_subs.js

@@ -2,7 +2,7 @@
 
 const {strict: assert} = require("assert");
 
-const {set_global, zrequire} = require("../zjsunit/namespace");
+const {set_global, with_field, zrequire} = require("../zjsunit/namespace");
 const {make_stub, with_stub} = require("../zjsunit/stub");
 const {run_test} = require("../zjsunit/test");
 
@@ -15,7 +15,9 @@ set_global("compose_fade", {});
 set_global("stream_events", {});
 set_global("subs", {});
 
+const peer_data = zrequire("peer_data");
 const people = zrequire("people");
+
 zrequire("stream_data");
 zrequire("server_events_dispatch");
 
@@ -140,7 +142,7 @@ test("peer event error handling (bad stream_ids)", (override) => {
     dispatch(remove_event);
 });
 
-test("peer event error handling (add_subscriber)", (override) => {
+test("peer event error handling (add/remove subscriber)", (override) => {
     override("compose_fade.update_faded_users", () => {});
     override("subs.update_subscribers_ui", () => {});
 
@@ -149,28 +151,38 @@ test("peer event error handling (add_subscriber)", (override) => {
         stream_id: 1,
     });
 
-    override("stream_data.add_subscriber", () => false);
+    with_field(
+        peer_data,
+        "add_subscriber",
+        () => false,
+        () => {
+            const add_event = {
+                type: "subscription",
+                op: "peer_add",
+                stream_ids: [1],
+                user_ids: [99999], // id is irrelevant
+            };
 
-    const add_event = {
+            blueslip.expect("warn", "Cannot process peer_add event");
-        type: "subscription",
+            dispatch(add_event);
-        op: "peer_add",
+            blueslip.reset();
-        stream_ids: [1],
+        },
-        user_ids: [99999], // id is irrelevant
+    );
-    };
 
-    blueslip.expect("warn", "Cannot process peer_add event");
+    with_field(
-    dispatch(add_event);
+        peer_data,
-    blueslip.reset();
+        "remove_subscriber",
+        () => false,
+        () => {
+            const remove_event = {
+                type: "subscription",
+                op: "peer_remove",
+                stream_ids: [1],
+                user_ids: [99999], // id is irrelevant
+            };
 
-    override("stream_data.remove_subscriber", () => false);
+            blueslip.expect("warn", "Cannot process peer_remove event.");
-
+            dispatch(remove_event);
-    const remove_event = {
+        },
-        type: "subscription",
+    );
-        op: "peer_remove",
-        stream_ids: [1],
-        user_ids: [99999], // id is irrelevant
-    };
-
-    blueslip.expect("warn", "Cannot process peer_remove event.");
-    dispatch(remove_event);
 });

frontend_tests/node_tests/stream_data.js

@@ -19,6 +19,7 @@ stub_out_jquery();
 zrequire("color_data");
 zrequire("hash_util");
 zrequire("stream_topic_history");
+const peer_data = zrequire("peer_data");
 const people = zrequire("people");
 zrequire("stream_color");
 zrequire("stream_data");
@@ -160,7 +161,7 @@ run_test("unsubscribe", () => {
     // set up our subscription
     stream_data.add_sub(sub);
     sub.subscribed = true;
-    stream_data.set_subscribers(sub, [me.user_id]);
+    peer_data.set_subscribers(sub, [me.user_id]);
 
     // ensure our setup is accurate
     assert(stream_data.is_subscribed("devel"));
@@ -203,7 +204,7 @@ run_test("subscribers", () => {
     people.add_active_user(george);
 
     function potential_subscriber_ids() {
-        const users = stream_data.potential_subscribers(sub.stream_id);
+        const users = peer_data.potential_subscribers(sub.stream_id);
         return users.map((u) => u.user_id).sort();
     }
 
@@ -214,7 +215,7 @@ run_test("subscribers", () => {
         george.user_id,
     ]);
 
-    stream_data.set_subscribers(sub, [me.user_id, fred.user_id, george.user_id]);
+    peer_data.set_subscribers(sub, [me.user_id, fred.user_id, george.user_id]);
     stream_data.update_calculated_fields(sub);
     assert(stream_data.is_user_subscribed(sub.stream_id, me.user_id));
     assert(stream_data.is_user_subscribed(sub.stream_id, fred.user_id));
@@ -223,7 +224,7 @@ run_test("subscribers", () => {
 
     assert.deepEqual(potential_subscriber_ids(), [not_fred.user_id]);
 
-    stream_data.set_subscribers(sub, []);
+    peer_data.set_subscribers(sub, []);
 
     const brutus = {
         email: "brutus@zulip.com",
@@ -234,7 +235,7 @@ run_test("subscribers", () => {
     assert(!stream_data.is_user_subscribed(sub.stream_id, brutus.user_id));
 
     // add
-    let ok = stream_data.add_subscriber(sub.stream_id, brutus.user_id);
+    let ok = peer_data.add_subscriber(sub.stream_id, brutus.user_id);
     assert(ok);
     assert(stream_data.is_user_subscribed(sub.stream_id, brutus.user_id));
     sub = stream_data.get_sub("Rome");
@@ -245,14 +246,14 @@ run_test("subscribers", () => {
     assert.equal(sub.email_address, sub_email);
 
     // verify that adding an already-added subscriber is a noop
-    stream_data.add_subscriber(sub.stream_id, brutus.user_id);
+    peer_data.add_subscriber(sub.stream_id, brutus.user_id);
     assert(stream_data.is_user_subscribed(sub.stream_id, brutus.user_id));
     sub = stream_data.get_sub("Rome");
     stream_data.update_subscribers_count(sub);
     assert.equal(sub.subscriber_count, 1);
 
     // remove
-    ok = stream_data.remove_subscriber(sub.stream_id, brutus.user_id);
+    ok = peer_data.remove_subscriber(sub.stream_id, brutus.user_id);
     assert(ok);
     assert(!stream_data.is_user_subscribed(sub.stream_id, brutus.user_id));
     sub = stream_data.get_sub("Rome");
@@ -270,12 +271,12 @@ run_test("subscribers", () => {
         "warn",
         "We got a remove_subscriber call for an untracked stream " + bad_stream_id,
     );
-    ok = stream_data.remove_subscriber(bad_stream_id, brutus.user_id);
+    ok = peer_data.remove_subscriber(bad_stream_id, brutus.user_id);
     assert(!ok);
 
     // verify that removing an already-removed subscriber is a noop
     blueslip.expect("warn", "We tried to remove invalid subscriber: 104");
-    ok = stream_data.remove_subscriber(sub.stream_id, brutus.user_id);
+    ok = peer_data.remove_subscriber(sub.stream_id, brutus.user_id);
     assert(!ok);
     assert(!stream_data.is_user_subscribed(sub.stream_id, brutus.user_id));
     sub = stream_data.get_sub("Rome");
@@ -284,21 +285,21 @@ run_test("subscribers", () => {
 
     // Verify defensive code in set_subscribers, where the second parameter
     // can be undefined.
-    stream_data.set_subscribers(sub);
+    peer_data.set_subscribers(sub);
     stream_data.add_sub(sub);
-    stream_data.add_subscriber(sub.stream_id, brutus.user_id);
+    peer_data.add_subscriber(sub.stream_id, brutus.user_id);
     sub.subscribed = true;
     assert(stream_data.is_user_subscribed(sub.stream_id, brutus.user_id));
 
     // Verify that we noop and don't crash when unsubscribed.
     sub.subscribed = false;
     stream_data.update_calculated_fields(sub);
-    ok = stream_data.add_subscriber(sub.stream_id, brutus.user_id);
+    ok = peer_data.add_subscriber(sub.stream_id, brutus.user_id);
     assert(ok);
     assert.equal(stream_data.is_user_subscribed(sub.stream_id, brutus.user_id), true);
-    stream_data.remove_subscriber(sub.stream_id, brutus.user_id);
+    peer_data.remove_subscriber(sub.stream_id, brutus.user_id);
     assert.equal(stream_data.is_user_subscribed(sub.stream_id, brutus.user_id), false);
-    stream_data.add_subscriber(sub.stream_id, brutus.user_id);
+    peer_data.add_subscriber(sub.stream_id, brutus.user_id);
     assert.equal(stream_data.is_user_subscribed(sub.stream_id, brutus.user_id), true);
 
     blueslip.expect(
@@ -309,18 +310,18 @@ run_test("subscribers", () => {
     sub.invite_only = true;
     stream_data.update_calculated_fields(sub);
     assert.equal(stream_data.is_user_subscribed(sub.stream_id, brutus.user_id), undefined);
-    stream_data.remove_subscriber(sub.stream_id, brutus.user_id);
+    peer_data.remove_subscriber(sub.stream_id, brutus.user_id);
     assert.equal(stream_data.is_user_subscribed(sub.stream_id, brutus.user_id), undefined);
 
     // Verify that we don't crash and return false for a bad stream.
     blueslip.expect("warn", "We got an add_subscriber call for an untracked stream: 9999999");
-    ok = stream_data.add_subscriber(9999999, brutus.user_id);
+    ok = peer_data.add_subscriber(9999999, brutus.user_id);
     assert(!ok);
 
     // Verify that we don't crash and return false for a bad user id.
     blueslip.expect("error", "Unknown user_id in get_by_user_id: 9999999");
     blueslip.expect("error", "We tried to add invalid subscriber: 9999999");
-    ok = stream_data.add_subscriber(sub.stream_id, 9999999);
+    ok = peer_data.add_subscriber(sub.stream_id, 9999999);
     assert(!ok);
 });
 
@@ -608,10 +609,10 @@ run_test("get_subscriber_count", () => {
     stream_data.clear_subscriptions();
 
     blueslip.expect("warn", "We got a get_subscriber_count call for an untracked stream: 102");
-    assert.equal(stream_data.get_subscriber_count(india.stream_id), undefined);
+    assert.equal(peer_data.get_subscriber_count(india.stream_id), undefined);
 
     stream_data.add_sub(india);
-    assert.equal(stream_data.get_subscriber_count(india.stream_id), 0);
+    assert.equal(peer_data.get_subscriber_count(india.stream_id), 0);
 
     const fred = {
         email: "fred@zulip.com",
@@ -619,19 +620,19 @@ run_test("get_subscriber_count", () => {
         user_id: 101,
     };
     people.add_active_user(fred);
-    stream_data.add_subscriber(india.stream_id, 102);
+    peer_data.add_subscriber(india.stream_id, 102);
-    assert.equal(stream_data.get_subscriber_count(india.stream_id), 1);
+    assert.equal(peer_data.get_subscriber_count(india.stream_id), 1);
     const george = {
         email: "george@zulip.com",
         full_name: "George",
         user_id: 103,
     };
     people.add_active_user(george);
-    stream_data.add_subscriber(india.stream_id, 103);
+    peer_data.add_subscriber(india.stream_id, 103);
-    assert.equal(stream_data.get_subscriber_count(india.stream_id), 2);
+    assert.equal(peer_data.get_subscriber_count(india.stream_id), 2);
 
-    stream_data.remove_subscriber(india.stream_id, 103);
+    peer_data.remove_subscriber(india.stream_id, 103);
-    assert.deepStrictEqual(stream_data.get_subscriber_count(india.stream_id), 1);
+    assert.deepStrictEqual(peer_data.get_subscriber_count(india.stream_id), 1);
 });
 
 run_test("notifications", () => {
@@ -977,7 +978,7 @@ run_test("filter inactives", () => {
 run_test("is_subscriber_subset", () => {
     function make_sub(stream_id, user_ids) {
         const sub = {stream_id};
-        stream_data.set_subscribers(sub, user_ids);
+        peer_data.set_subscribers(sub, user_ids);
         return sub;
     }
 
@@ -1006,7 +1007,7 @@ run_test("is_subscriber_subset", () => {
     ];
 
     for (const row of matrix) {
-        assert.equal(stream_data.is_subscriber_subset(row[0], row[1]), row[2]);
+        assert.equal(peer_data.is_subscriber_subset(row[0], row[1]), row[2]);
     }
 });
 
@@ -1106,7 +1107,7 @@ run_test("warn if subscribers are missing", () => {
             stream_data.is_user_subscribed(sub.stream_id, me.user_id);
 
             blueslip.expect("warn", "We called get_subscribers for an untracked stream: 3");
-            assert.deepEqual(stream_data.get_subscribers(sub.stream_id), []);
+            assert.deepEqual(peer_data.get_subscribers(sub.stream_id), []);
         },
     );
 });

frontend_tests/node_tests/stream_edit.js

@@ -36,6 +36,7 @@ set_global("ui", {
 set_global("$", make_zjquery());
 
 zrequire("input_pill");
+const peer_data = zrequire("peer_data");
 const people = zrequire("people");
 zrequire("pill_typeahead");
 zrequire("subs");
@@ -80,14 +81,14 @@ const denmark = {
     render_subscribers: true,
     should_display_subscription_button: true,
 };
-stream_data.set_subscribers(denmark, [me.user_id, mark.user_id]);
+peer_data.set_subscribers(denmark, [me.user_id, mark.user_id]);
 
 const sweden = {
     stream_id: 2,
     name: "Sweden",
     subscribed: false,
 };
-stream_data.set_subscribers(sweden, [mark.user_id, jill.user_id]);
+peer_data.set_subscribers(sweden, [mark.user_id, jill.user_id]);
 
 const subs = [denmark, sweden];
 for (const sub of subs) {
@@ -232,7 +233,7 @@ run_test("subscriber_pills", () => {
     // We cannot subscribe ourselves (`me`) as
     // we are already subscribed to denmark stream.
     const potential_denmark_stream_subscribers = Array.from(
-        stream_data.get_subscribers(denmark.stream_id),
+        peer_data.get_subscribers(denmark.stream_id),
     ).filter((id) => id !== me.user_id);
 
     // denmark.stream_id is stubbed. Thus request is
@@ -268,7 +269,7 @@ run_test("subscriber_pills", () => {
     // But only one request for mark is sent even though a mark user
     // pill is created and mark is also a subscriber of Denmark stream.
     user_pill.get_user_ids = () => [mark.user_id, fred.user_id];
-    stream_pill.get_user_ids = () => stream_data.get_subscribers(denmark.stream_id);
+    stream_pill.get_user_ids = () => peer_data.get_subscribers(denmark.stream_id);
     expected_user_ids = potential_denmark_stream_subscribers.concat(fred.user_id);
     add_subscribers_handler(event);
 });

frontend_tests/node_tests/stream_events.js

@@ -25,6 +25,7 @@ set_global("stream_list", {});
 set_global("stream_muting", {});
 set_global("subs", {});
 
+const peer_data = zrequire("peer_data");
 const people = zrequire("people");
 zrequire("stream_data");
 zrequire("stream_events");
@@ -303,10 +304,7 @@ run_test("marked_subscribed", (override) => {
 
         const user_ids = [15, 20, 25];
         stream_events.mark_subscribed(frontend, user_ids, "");
-        assert.deepEqual(
+        assert.deepEqual(new Set(peer_data.get_subscribers(frontend.stream_id)), new Set(user_ids));
-            new Set(stream_data.get_subscribers(frontend.stream_id)),
-            new Set(user_ids),
-        );
 
         // assign self as well
         with_stub((stub) => {
@@ -422,7 +420,7 @@ run_test("remove_deactivated_user_from_all_streams", () => {
     assert(!stream_data.is_user_subscribed(dev_help.stream_id, george.user_id));
 
     // verify that deactivating user should unsubscribe user from all streams
-    assert(stream_data.add_subscriber(dev_help.stream_id, george.user_id));
+    assert(peer_data.add_subscriber(dev_help.stream_id, george.user_id));
     assert(stream_data.is_user_subscribed(dev_help.stream_id, george.user_id));
 
     stream_events.remove_deactivated_user_from_all_streams(george.user_id);

frontend_tests/node_tests/typeahead_helper.js

@@ -15,6 +15,7 @@ page_params.realm_email_address_visibility =
     settings_config.email_address_visibility_values.admins_only.code;
 
 zrequire("recent_senders");
+const peer_data = zrequire("peer_data");
 const people = zrequire("people");
 zrequire("stream_data");
 zrequire("narrow");
@@ -55,7 +56,7 @@ run_test("sort_streams", () => {
 
     function process_test_streams() {
         for (const test_stream of test_streams) {
-            stream_data.set_subscribers(test_stream, test_stream.subscribers);
+            peer_data.set_subscribers(test_stream, test_stream.subscribers);
             delete test_stream.subscribers;
             stream_data.update_calculated_fields(test_stream);
         }
@@ -320,9 +321,9 @@ run_test("sort_recipients", () => {
     const subscriber_email_1 = "b_user_2@zulip.net";
     const subscriber_email_2 = "b_user_3@zulip.net";
     const subscriber_email_3 = "b_bot@example.com";
-    stream_data.add_subscriber(1, people.get_user_id(subscriber_email_1));
+    peer_data.add_subscriber(1, people.get_user_id(subscriber_email_1));
-    stream_data.add_subscriber(1, people.get_user_id(subscriber_email_2));
+    peer_data.add_subscriber(1, people.get_user_id(subscriber_email_2));
-    stream_data.add_subscriber(1, people.get_user_id(subscriber_email_3));
+    peer_data.add_subscriber(1, people.get_user_id(subscriber_email_3));
 
     const dev_sub = stream_data.get_sub("Dev");
     const linux_sub = stream_data.get_sub("Linux");

static/js/compose.js

@@ -8,6 +8,7 @@ const render_compose_invite_users = require("../templates/compose_invite_users.h
 const render_compose_not_subscribed = require("../templates/compose_not_subscribed.hbs");
 const render_compose_private_stream_alert = require("../templates/compose_private_stream_alert.hbs");
 
+const peer_data = require("./peer_data");
 const people = require("./people");
 const rendered_markdown = require("./rendered_markdown");
 const settings_config = require("./settings_config");
@@ -44,7 +45,7 @@ function make_uploads_relative(content) {
 }
 
 function show_all_everyone_warnings(stream_id) {
-    const stream_count = stream_data.get_subscriber_count(stream_id) || 0;
+    const stream_count = peer_data.get_subscriber_count(stream_id) || 0;
 
     const all_everyone_template = render_compose_all_everyone({
         count: stream_count,
@@ -95,7 +96,7 @@ function show_sending_indicator(whats_happening) {
 }
 
 function show_announce_warnings(stream_id) {
-    const stream_count = stream_data.get_subscriber_count(stream_id) || 0;
+    const stream_count = peer_data.get_subscriber_count(stream_id) || 0;
 
     const announce_template = render_compose_announce({count: stream_count});
     const error_area_announce = $("#compose-announce");
@@ -528,7 +529,7 @@ exports.wildcard_mention_allowed = function () {
 };
 
 function validate_stream_message_mentions(stream_id) {
-    const stream_count = stream_data.get_subscriber_count(stream_id) || 0;
+    const stream_count = peer_data.get_subscriber_count(stream_id) || 0;
 
     // If the user is attempting to do a wildcard mention in a large
     // stream, check if they permission to do so.
@@ -565,7 +566,7 @@ function validate_stream_message_mentions(stream_id) {
 }
 
 function validate_stream_message_announce(sub) {
-    const stream_count = stream_data.get_subscriber_count(sub.stream_id) || 0;
+    const stream_count = peer_data.get_subscriber_count(sub.stream_id) || 0;
 
     if (sub.name === "announce" && stream_count > exports.announce_warn_threshold) {
         if (user_acknowledged_announce === undefined || user_acknowledged_announce === false) {
@@ -980,7 +981,7 @@ exports.warn_if_private_stream_is_linked = function (linked_stream) {
         return;
     }
 
-    if (stream_data.is_subscriber_subset(compose_stream, linked_stream)) {
+    if (peer_data.is_subscriber_subset(compose_stream, linked_stream)) {
         // Don't warn if subscribers list of current compose_stream is
         // a subset of linked_stream's subscribers list, because
         // everyone will be subscribed to the linked stream and so

static/js/peer_data.js

@@ -0,0 +1,145 @@
+const {LazySet} = require("./lazy_set");
+const people = require("./people");
+
+/*
+
+For legacy reasons this module is mostly tested
+by frontend_tests/node_tests/stream_data.js.
+
+*/
+
+// This maps a stream_id to a LazySet of user_ids who are subscribed.
+// We maintain the invariant that this has keys for all all stream_ids
+// that we track in the other data structures.  We intialize it during
+// clear_subscriptions.
+let stream_subscribers;
+
+export function clear() {
+    stream_subscribers = new Map();
+}
+
+export function maybe_clear_subscribers(sub) {
+    if (!stream_subscribers.has(sub.stream_id)) {
+        set_subscribers(sub, []);
+    }
+}
+
+export function is_subscriber_subset(sub1, sub2) {
+    const stream_id1 = sub1.stream_id;
+    const stream_id2 = sub2.stream_id;
+
+    const sub1_set = stream_subscribers.get(stream_id1);
+    const sub2_set = stream_subscribers.get(stream_id2);
+
+    if (sub1_set && sub2_set) {
+        return Array.from(sub1_set.keys()).every((key) => sub2_set.has(key));
+    }
+
+    return false;
+}
+
+export function potential_subscribers(stream_id) {
+    /*
+        This is a list of unsubscribed users
+        for the current stream, who the current
+        user could potentially subscribe to the
+        stream.  This may include some bots.
+
+        We currently use it for typeahead in
+        stream_edit.js.
+
+        This may be a superset of the actual
+        subscribers that you can change in some cases
+        (like if you're a guest?); we should refine this
+        going forward, especially if we use it for something
+        other than typeahead.  (The guest use case
+        may be moot now for other reasons.)
+    */
+
+    const subscribers = stream_subscribers.get(stream_id);
+
+    function is_potential_subscriber(person) {
+        // Use verbose style to force better test
+        // coverage, plus we may add more conditions over
+        // time.
+        if (subscribers.has(person.user_id)) {
+            return false;
+        }
+
+        return true;
+    }
+
+    return people.filter_all_users(is_potential_subscriber);
+}
+
+export function get_subscriber_count(stream_id) {
+    const subscribers = stream_subscribers.get(stream_id);
+
+    if (!subscribers) {
+        blueslip.warn("We got a get_subscriber_count call for an untracked stream: " + stream_id);
+        return undefined;
+    }
+
+    return subscribers.size;
+}
+
+export function get_subscribers(stream_id) {
+    const subscribers = stream_subscribers.get(stream_id);
+
+    if (typeof subscribers === "undefined") {
+        blueslip.warn("We called get_subscribers for an untracked stream: " + stream_id);
+        return [];
+    }
+
+    return Array.from(subscribers.keys());
+}
+
+export function set_subscribers(sub, user_ids) {
+    const subscribers = new LazySet(user_ids || []);
+    stream_subscribers.set(sub.stream_id, subscribers);
+}
+
+export function add_subscriber(stream_id, user_id) {
+    const subscribers = stream_subscribers.get(stream_id);
+    if (typeof subscribers === "undefined") {
+        blueslip.warn("We got an add_subscriber call for an untracked stream: " + stream_id);
+        return false;
+    }
+    const person = people.get_by_user_id(user_id);
+    if (person === undefined) {
+        blueslip.error("We tried to add invalid subscriber: " + user_id);
+        return false;
+    }
+    subscribers.add(user_id);
+
+    return true;
+}
+
+export function remove_subscriber(stream_id, user_id) {
+    const subscribers = stream_subscribers.get(stream_id);
+    if (typeof subscribers === "undefined") {
+        blueslip.warn("We got a remove_subscriber call for an untracked stream " + stream_id);
+        return false;
+    }
+    if (!subscribers.has(user_id)) {
+        blueslip.warn("We tried to remove invalid subscriber: " + user_id);
+        return false;
+    }
+
+    subscribers.delete(user_id);
+
+    return true;
+}
+
+export function is_user_subscribed(stream_id, user_id) {
+    // Most callers should call stream_data.is_user_subscribed,
+    // which does additional checks.
+
+    const subscribers = stream_subscribers.get(stream_id);
+    if (typeof subscribers === "undefined") {
+        blueslip.warn("We called is_user_subscribed for an untracked stream: " + stream_id);
+        return false;
+    }
+
+    return subscribers.has(user_id);
+}

static/js/server_events_dispatch.js

@@ -2,6 +2,7 @@
 
 const emoji = require("../shared/js/emoji");
 
+const peer_data = require("./peer_data");
 const people = require("./people");
 const settings_config = require("./settings_config");
 
@@ -353,7 +354,7 @@ exports.dispatch_normal_event = function dispatch_normal_event(event) {
                     }
 
                     event.user_ids.forEach((user_id) => {
-                        if (!stream_data.add_subscriber(stream_id, user_id)) {
+                        if (!peer_data.add_subscriber(stream_id, user_id)) {
                             blueslip.warn("Cannot process peer_add event");
                             return;
                         }
@@ -372,7 +373,7 @@ exports.dispatch_normal_event = function dispatch_normal_event(event) {
                     }
 
                     event.user_ids.forEach((user_id) => {
-                        if (!stream_data.remove_subscriber(sub.stream_id, user_id)) {
+                        if (!peer_data.remove_subscriber(sub.stream_id, user_id)) {
                             blueslip.warn("Cannot process peer_remove event.");
                             return;
                         }

static/js/stream_create.js

@@ -4,6 +4,7 @@ const render_announce_stream_docs = require("../templates/announce_stream_docs.h
 const render_new_stream_users = require("../templates/new_stream_users.hbs");
 const render_subscription_invites_warning_modal = require("../templates/subscription_invites_warning_modal.hbs");
 
+const peer_data = require("./peer_data");
 const people = require("./people");
 
 let created_stream;
@@ -289,7 +290,7 @@ exports.show_new_stream_modal = function () {
         const elem = $(this);
         const stream_id = Number.parseInt(elem.attr("data-stream-id"), 10);
         const checked = elem.find("input").prop("checked");
-        const subscriber_ids = new Set(stream_data.get_subscribers(stream_id));
+        const subscriber_ids = new Set(peer_data.get_subscribers(stream_id));
 
         $("#user-checkboxes label.checkbox").each(function () {
             const user_elem = $(this);

static/js/stream_data.js

@@ -1,11 +1,14 @@
 "use strict";
 
 const {FoldDict} = require("./fold_dict");
-const {LazySet} = require("./lazy_set");
+const peer_data = require("./peer_data");
 const people = require("./people");
 const settings_config = require("./settings_config");
 const util = require("./util");
 
+// Expose get_subscriber_count for our automated puppeteer tests.
+exports.get_subscriber_count = peer_data.get_subscriber_count;
+
 class BinaryDict {
     /*
       A dictionary that keeps track of which objects had the predicate
@@ -93,12 +96,6 @@ let filter_out_inactives = false;
 const stream_ids_by_name = new FoldDict();
 const default_stream_ids = new Set();
 
-// This maps a stream_id to a LazySet of user_ids who are subscribed.
-// We maintain the invariant that this has keys for all all stream_ids
-// that we track in the other data structures.  We intialize it during
-// clear_subscriptions.
-let stream_subscribers;
-
 exports.stream_privacy_policy_values = {
     public: {
         code: "public",
@@ -143,7 +140,7 @@ exports.clear_subscriptions = function () {
     // it should only be used in tests.
     stream_info = new BinaryDict((sub) => sub.subscribed);
     subs_by_stream_id = new Map();
-    stream_subscribers = new Map();
+    peer_data.clear();
 };
 
 exports.clear_subscriptions();
@@ -195,30 +192,16 @@ exports.rename_sub = function (sub, new_name) {
 
 exports.subscribe_myself = function (sub) {
     const user_id = people.my_current_user_id();
-    exports.add_subscriber(sub.stream_id, user_id);
+    peer_data.add_subscriber(sub.stream_id, user_id);
     sub.subscribed = true;
     sub.newly_subscribed = true;
     stream_info.set_true(sub.name, sub);
 };
 
-exports.is_subscriber_subset = function (sub1, sub2) {
-    const stream_id1 = sub1.stream_id;
-    const stream_id2 = sub2.stream_id;
-
-    const sub1_set = stream_subscribers.get(stream_id1);
-    const sub2_set = stream_subscribers.get(stream_id2);
-
-    if (sub1_set && sub2_set) {
-        return Array.from(sub1_set.keys()).every((key) => sub2_set.has(key));
-    }
-
-    return false;
-};
-
 exports.unsubscribe_myself = function (sub) {
     // Remove user from subscriber's list
     const user_id = people.my_current_user_id();
-    exports.remove_subscriber(sub.stream_id, user_id);
+    peer_data.remove_subscriber(sub.stream_id, user_id);
     sub.subscribed = false;
     sub.newly_subscribed = false;
     stream_info.set_false(sub.name, sub);
@@ -229,10 +212,7 @@ exports.add_sub = function (sub) {
     // We use create_sub_from_server_data at page load.
     // We use create_streams for new streams in live-update events.
 
-    if (!stream_subscribers.has(sub.stream_id)) {
+    peer_data.maybe_clear_subscribers(sub);
-        exports.set_subscribers(sub, []);
-    }
-
     stream_info.set(sub.name, sub);
     subs_by_stream_id.set(sub.stream_id, sub);
 };
@@ -447,58 +427,13 @@ exports.update_subscribers_count = function (sub) {
     // This is part of an unfortunate legacy hack, where we
     // put calculated fields onto the sub object instead of
     // letting callers build their own objects.
-    sub.subscriber_count = exports.get_subscriber_count(sub.stream_id);
+    sub.subscriber_count = peer_data.get_subscriber_count(sub.stream_id);
-};
-
-exports.potential_subscribers = function (stream_id) {
-    /*
-        This is a list of unsubscribed users
-        for the current stream, who the current
-        user could potentially subscribe to the
-        stream.  This may include some bots.
-
-        We currently use it for typeahead in
-        stream_edit.js.
-
-        This may be a superset of the actual
-        subscribers that you can change in some cases
-        (like if you're a guest?); we should refine this
-        going forward, especially if we use it for something
-        other than typeahead.  (The guest use case
-        may be moot now for other reasons.)
-    */
-
-    const subscribers = stream_subscribers.get(stream_id);
-
-    function is_potential_subscriber(person) {
-        // Use verbose style to force better test
-        // coverage, plus we may add more conditions over
-        // time.
-        if (subscribers.has(person.user_id)) {
-            return false;
-        }
-
-        return true;
-    }
-
-    return people.filter_all_users(is_potential_subscriber);
 };
 
 exports.update_stream_email_address = function (sub, email) {
     sub.email_address = email;
 };
 
-exports.get_subscriber_count = function (stream_id) {
-    const subscribers = stream_subscribers.get(stream_id);
-
-    if (!subscribers) {
-        blueslip.warn("We got a get_subscriber_count call for an untracked stream: " + stream_id);
-        return undefined;
-    }
-
-    return subscribers.size;
-};
-
 exports.update_stream_post_policy = function (sub, stream_post_policy) {
     sub.stream_post_policy = stream_post_policy;
 };
@@ -709,54 +644,6 @@ exports.maybe_get_stream_name = function (stream_id) {
     return stream.name;
 };
 
-exports.get_subscribers = (stream_id) => {
-    const subscribers = stream_subscribers.get(stream_id);
-
-    if (typeof subscribers === "undefined") {
-        blueslip.warn("We called get_subscribers for an untracked stream: " + stream_id);
-        return [];
-    }
-
-    return Array.from(subscribers.keys());
-};
-
-exports.set_subscribers = function (sub, user_ids) {
-    const subscribers = new LazySet(user_ids || []);
-    stream_subscribers.set(sub.stream_id, subscribers);
-};
-
-exports.add_subscriber = function (stream_id, user_id) {
-    const subscribers = stream_subscribers.get(stream_id);
-    if (typeof subscribers === "undefined") {
-        blueslip.warn("We got an add_subscriber call for an untracked stream: " + stream_id);
-        return false;
-    }
-    const person = people.get_by_user_id(user_id);
-    if (person === undefined) {
-        blueslip.error("We tried to add invalid subscriber: " + user_id);
-        return false;
-    }
-    subscribers.add(user_id);
-
-    return true;
-};
-
-exports.remove_subscriber = function (stream_id, user_id) {
-    const subscribers = stream_subscribers.get(stream_id);
-    if (typeof subscribers === "undefined") {
-        blueslip.warn("We got a remove_subscriber call for an untracked stream " + stream_id);
-        return false;
-    }
-    if (!subscribers.has(user_id)) {
-        blueslip.warn("We tried to remove invalid subscriber: " + user_id);
-        return false;
-    }
-
-    subscribers.delete(user_id);
-
-    return true;
-};
-
 exports.is_user_subscribed = function (stream_id, user_id) {
     const sub = exports.get_sub_by_id(stream_id);
     if (typeof sub === "undefined" || !sub.can_access_subscribers) {
@@ -772,13 +659,7 @@ exports.is_user_subscribed = function (stream_id, user_id) {
         return undefined;
     }
 
-    const subscribers = stream_subscribers.get(stream_id);
+    return peer_data.is_user_subscribed(stream_id, user_id);
-    if (typeof subscribers === "undefined") {
-        blueslip.warn("We called is_user_subscribed for an untracked stream: " + stream_id);
-        return false;
-    }
-
-    return subscribers.has(user_id);
 };
 
 exports.create_streams = function (streams) {
@@ -835,7 +716,7 @@ exports.create_sub_from_server_data = function (attrs) {
         ...attrs,
     };
 
-    exports.set_subscribers(sub, subscriber_user_ids);
+    peer_data.set_subscribers(sub, subscriber_user_ids);
 
     if (!sub.color) {
         sub.color = color_data.pick_color();
@@ -936,7 +817,7 @@ exports.sort_for_stream_settings = function (stream_ids, order) {
     }
 
     function by_subscriber_count(id_a, id_b) {
-        const out = exports.get_subscriber_count(id_b) - exports.get_subscriber_count(id_a);
+        const out = peer_data.get_subscriber_count(id_b) - peer_data.get_subscriber_count(id_a);
         if (out === 0) {
             return by_stream_name(id_a, id_b);
         }

static/js/stream_edit.js

@@ -6,6 +6,7 @@ const render_stream_subscription_info = require("../templates/stream_subscriptio
 const render_subscription_settings = require("../templates/subscription_settings.hbs");
 const render_subscription_stream_privacy_modal = require("../templates/subscription_stream_privacy_modal.hbs");
 
+const peer_data = require("./peer_data");
 const people = require("./people");
 const settings_config = require("./settings_config");
 const settings_data = require("./settings_data");
@@ -327,12 +328,12 @@ function show_subscription_settings(sub) {
     const list = get_subscriber_list(sub_settings);
     list.empty();
 
-    const user_ids = stream_data.get_subscribers(sub.stream_id);
+    const user_ids = peer_data.get_subscribers(sub.stream_id);
     const users = exports.get_users_from_subscribers(user_ids);
     exports.sort_but_pin_current_user_on_top(users);
 
     function get_users_for_subscriber_typeahead() {
-        const potential_subscribers = stream_data.potential_subscribers(stream_id);
+        const potential_subscribers = peer_data.potential_subscribers(stream_id);
         return user_pill.filter_taken_users(potential_subscribers, exports.pill_widget);
     }
 

static/js/stream_events.js

@@ -1,5 +1,7 @@
 "use strict";
 
+const peer_data = require("./peer_data");
+
 // In theory, this function should apply the account-level defaults,
 // however, they are only called after a manual override, so
 // doing so is unnecessary with the current code.  Ideally, we'd do a
@@ -97,7 +99,7 @@ exports.mark_subscribed = function (sub, subscribers, color) {
     }
     stream_data.subscribe_myself(sub);
     if (subscribers) {
-        stream_data.set_subscribers(sub, subscribers);
+        peer_data.set_subscribers(sub, subscribers);
     }
     stream_data.update_calculated_fields(sub);
 
@@ -148,7 +150,7 @@ exports.remove_deactivated_user_from_all_streams = function (user_id) {
 
     for (const sub of all_subs) {
         if (stream_data.is_user_subscribed(sub.stream_id, user_id)) {
-            stream_data.remove_subscriber(sub.stream_id, user_id);
+            peer_data.remove_subscriber(sub.stream_id, user_id);
             subs.update_subscribers_ui(sub);
         }
     }

static/js/stream_pill.js

@@ -1,5 +1,7 @@
 "use strict";
 
+const peer_data = require("./peer_data");
+
 function display_pill(sub) {
     return "#" + sub.name + ": " + sub.subscriber_count + " users";
 }
@@ -39,7 +41,7 @@ function get_user_ids_from_subs(items) {
     for (const item of items) {
         // only some of our items have streams (for copy-from-stream)
         if (item.stream_id !== undefined) {
-            user_ids = user_ids.concat(stream_data.get_subscribers(item.stream_id));
+            user_ids = user_ids.concat(peer_data.get_subscribers(item.stream_id));
         }
     }
     return user_ids;

static/js/stream_ui_updates.js

@@ -4,6 +4,8 @@ const render_subscription_count = require("../templates/subscription_count.hbs")
 const render_subscription_setting_icon = require("../templates/subscription_setting_icon.hbs");
 const render_subscription_type = require("../templates/subscription_type.hbs");
 
+const peer_data = require("./peer_data");
+
 exports.update_check_button_for_sub = function (sub) {
     const button = subs.check_button_for_sub(sub);
     if (sub.subscribed) {
@@ -194,7 +196,7 @@ exports.update_subscribers_list = function (sub) {
     if (!sub.can_access_subscribers) {
         $(".subscriber_list_settings_container").hide();
     } else {
-        const subscribers = stream_data.get_subscribers(sub.stream_id);
+        const subscribers = peer_data.get_subscribers(sub.stream_id);
         const users = stream_edit.get_users_from_subscribers(subscribers);
 
         /*

static/js/typeahead_helper.js

@@ -8,6 +8,7 @@ const emoji = require("../shared/js/emoji");
 const typeahead = require("../shared/js/typeahead");
 const render_typeahead_list_item = require("../templates/typeahead_list_item.hbs");
 
+const peer_data = require("./peer_data");
 const people = require("./people");
 const pm_conversations = require("./pm_conversations");
 const settings_data = require("./settings_data");
@@ -390,8 +391,8 @@ exports.compare_by_activity = function (stream_a, stream_b) {
         return diff;
     }
     diff =
-        stream_data.get_subscriber_count(stream_b.stream_id) -
+        peer_data.get_subscriber_count(stream_b.stream_id) -
-        stream_data.get_subscriber_count(stream_a.stream_id);
+        peer_data.get_subscriber_count(stream_a.stream_id);
     if (diff !== 0) {
         return diff;
     }