mirror of
https://github.com/zulip/zulip.git
synced 2026-07-12 21:04:41 +08:00
events: Stop sending occupy/vacate events.
We used to send occupy/vacate events when either the first person entered a stream or the last person exited. It appears that our two main apps have never looked at these events. Instead, it's generally the case that clients handle events related to stream creation/deactivation and subscribe/unsubscribe. Note that we removed the apply_events code related to these events. This doesn't affect the webapp, because the webapp doesn't care about the "streams" field in do_events_register. There is a theoretical situation where a third party client could be the victim of a race where the "streams" data includes a stream where the last subscriber has left. I suspect in most of those situations it will be harmless, or possibly even helpful to the extent that they'll learn about streams that are in a "quasi" state where they're activated but not occupied. We could try to patch apply_event to detect when subscriptions get added or removed. Or we could just make the "streams" piece of do_events_register not care about occupy/vacate semantics. I favor the latter, since it might actually be what users what, and it will also simplify the code and improve performance.
This commit is contained in:
Steve Howell
Tim Abbott
parent
1bcb8d8ee8
commit
a9356508ca
@ -87,7 +87,7 @@ def create_integration_stream(integration: WebhookIntegration, bot: UserProfile)
|
||||
assert isinstance(bot.bot_owner, UserProfile)
|
||||
realm = bot.bot_owner.realm
|
||||
stream, created = create_stream_if_needed(realm, integration.stream_name)
|
||||
bulk_add_subscriptions(realm, [stream], [bot, bot.bot_owner], from_stream_creation=created)
|
||||
bulk_add_subscriptions(realm, [stream], [bot, bot.bot_owner])
|
||||
|
||||
def get_integration(integration_name: str) -> WebhookIntegration:
|
||||
integration = INTEGRATIONS[integration_name]
|
||||
|
||||
@ -2814,7 +2814,6 @@ def bulk_add_subscriptions(
|
||||
streams: Iterable[Stream],
|
||||
users: Iterable[UserProfile],
|
||||
color_map: Mapping[str, str]={},
|
||||
from_stream_creation: bool=False,
|
||||
acting_user: Optional[UserProfile]=None
|
||||
) -> SubT:
|
||||
users = list(users)
|
||||
@ -2867,21 +2866,13 @@ def bulk_add_subscriptions(
|
||||
color=color, recipient_id=recipient_id)
|
||||
subs_to_add.append((sub_to_add, stream))
|
||||
|
||||
new_occupied_streams = bulk_add_subs_to_db_with_logging(
|
||||
bulk_add_subs_to_db_with_logging(
|
||||
realm=realm,
|
||||
acting_user=acting_user,
|
||||
subs_to_add=subs_to_add,
|
||||
subs_to_activate=subs_to_activate,
|
||||
)
|
||||
|
||||
if new_occupied_streams and not from_stream_creation:
|
||||
event: Dict[str, object] = dict(
|
||||
type="stream",
|
||||
op="occupy",
|
||||
streams=[stream.to_dict() for stream in new_occupied_streams],
|
||||
)
|
||||
send_event(realm, event, active_user_ids(realm.id))
|
||||
|
||||
# Notify all existing users on streams that users have joined
|
||||
|
||||
# First, get all users subscribed to the streams that we care about
|
||||
@ -2926,16 +2917,11 @@ def bulk_add_subs_to_db_with_logging(
|
||||
acting_user: Optional[UserProfile],
|
||||
subs_to_add: List[Tuple[Subscription, Stream]],
|
||||
subs_to_activate: List[Tuple[Subscription, Stream]],
|
||||
) -> List[Stream]:
|
||||
) -> None:
|
||||
|
||||
# TODO: XXX: This transaction really needs to be done at the serializeable
|
||||
# transaction isolation level.
|
||||
with transaction.atomic():
|
||||
occupied_streams_before = list(get_occupied_streams(realm))
|
||||
Subscription.objects.bulk_create(sub for (sub, stream) in subs_to_add)
|
||||
sub_ids = [sub.id for (sub, stream) in subs_to_activate]
|
||||
Subscription.objects.filter(id__in=sub_ids).update(active=True)
|
||||
occupied_streams_after = list(get_occupied_streams(realm))
|
||||
Subscription.objects.bulk_create(sub for (sub, stream) in subs_to_add)
|
||||
sub_ids = [sub.id for (sub, stream) in subs_to_activate]
|
||||
Subscription.objects.filter(id__in=sub_ids).update(active=True)
|
||||
|
||||
# Log Subscription Activities in RealmAuditLog
|
||||
event_time = timezone_now()
|
||||
@ -2961,11 +2947,6 @@ def bulk_add_subs_to_db_with_logging(
|
||||
# Now since we have all log objects generated we can do a bulk insert
|
||||
RealmAuditLog.objects.bulk_create(all_subscription_logs)
|
||||
|
||||
new_occupied_streams = [stream for stream in
|
||||
set(occupied_streams_after) - set(occupied_streams_before)
|
||||
if not stream.invite_only]
|
||||
return new_occupied_streams
|
||||
|
||||
def send_stream_creation_events_for_private_streams(
|
||||
realm: Realm,
|
||||
stream_dict: Dict[int, Stream],
|
||||
@ -3158,13 +3139,7 @@ def bulk_remove_subscriptions(users: Iterable[UserProfile],
|
||||
new_vacant_streams = set(occupied_streams_before) - set(occupied_streams_after)
|
||||
new_vacant_private_streams = [stream for stream in new_vacant_streams
|
||||
if stream.invite_only]
|
||||
new_vacant_public_streams = [stream for stream in new_vacant_streams
|
||||
if not stream.invite_only]
|
||||
if new_vacant_public_streams:
|
||||
event = dict(type="stream", op="vacate",
|
||||
streams=[stream.to_dict()
|
||||
for stream in new_vacant_public_streams])
|
||||
send_event(our_realm, event, active_user_ids(our_realm.id))
|
||||
|
||||
if new_vacant_private_streams:
|
||||
# Deactivate any newly-vacant private streams
|
||||
for stream in new_vacant_private_streams:
|
||||
|
||||
@ -625,11 +625,6 @@ def apply_event(state: Dict[str, Any],
|
||||
stream[prop] = event['value']
|
||||
if prop == 'description':
|
||||
stream['rendered_description'] = event['rendered_description']
|
||||
elif event['op'] == "occupy":
|
||||
state['streams'] += event['streams']
|
||||
elif event['op'] == "vacate":
|
||||
stream_ids = [s["stream_id"] for s in event['streams']]
|
||||
state['streams'] = [s for s in state['streams'] if s["stream_id"] not in stream_ids]
|
||||
elif event['type'] == 'default_streams':
|
||||
state['realm_default_streams'] = event['default_streams']
|
||||
elif event['type'] == 'default_stream_groups':
|
||||
|
||||
@ -872,10 +872,9 @@ Output:
|
||||
realm = user_profile.realm
|
||||
try:
|
||||
stream = get_stream(stream_name, user_profile.realm)
|
||||
from_stream_creation = False
|
||||
except Stream.DoesNotExist:
|
||||
stream, from_stream_creation = create_stream_if_needed(realm, stream_name)
|
||||
bulk_add_subscriptions(realm, [stream], [user_profile], from_stream_creation=from_stream_creation)
|
||||
bulk_add_subscriptions(realm, [stream], [user_profile])
|
||||
return stream
|
||||
|
||||
def unsubscribe(self, user_profile: UserProfile, stream_name: str) -> None:
|
||||
|
||||
@ -974,104 +974,6 @@ paths:
|
||||
],
|
||||
"id": 0,
|
||||
}
|
||||
- type: object
|
||||
description: |
|
||||
Event sent to all users who can see a stream exists when a stream
|
||||
that previously had no subscribers
|
||||
gains its first subscriber (typically when a deactivated stream
|
||||
is reactivated). It works very similarly to `create` events.
|
||||
|
||||
We expect to change how stream deactivation/reactivation is
|
||||
represented in the future.
|
||||
properties:
|
||||
id:
|
||||
type: integer
|
||||
type:
|
||||
type: string
|
||||
enum:
|
||||
- stream
|
||||
op:
|
||||
type: string
|
||||
enum:
|
||||
- occupy
|
||||
streams:
|
||||
type: array
|
||||
description: |
|
||||
Array of stream objects, each contatining
|
||||
details about the newly occupied streams.
|
||||
items:
|
||||
$ref: "#/components/schemas/BasicStream"
|
||||
additionalProperties: false
|
||||
example:
|
||||
{
|
||||
"type": "stream",
|
||||
"op": "occupy",
|
||||
"streams":
|
||||
[
|
||||
{
|
||||
"name": "test_stream",
|
||||
"stream_id": 11,
|
||||
"description": "",
|
||||
"rendered_description": "",
|
||||
"invite_only": false,
|
||||
"is_web_public": false,
|
||||
"stream_post_policy": 1,
|
||||
"history_public_to_subscribers": true,
|
||||
"first_message_id": null,
|
||||
"message_retention_days": null,
|
||||
"is_announcement_only": false,
|
||||
},
|
||||
],
|
||||
"id": 0,
|
||||
}
|
||||
- type: object
|
||||
description: |
|
||||
Event sent to all users who can see a stream exists when a stream no
|
||||
longer has any users (which deactivates the stream).
|
||||
|
||||
We expect to change how stream deactivation/reactivation is
|
||||
represented in the future.
|
||||
properties:
|
||||
id:
|
||||
type: integer
|
||||
type:
|
||||
type: string
|
||||
enum:
|
||||
- stream
|
||||
op:
|
||||
type: string
|
||||
enum:
|
||||
- vacate
|
||||
streams:
|
||||
type: array
|
||||
description: |
|
||||
Array of stream objects, each containing
|
||||
details about a now-empty stream.
|
||||
items:
|
||||
$ref: "#/components/schemas/BasicStream"
|
||||
additionalProperties: false
|
||||
example:
|
||||
{
|
||||
"type": "stream",
|
||||
"op": "vacate",
|
||||
"streams":
|
||||
[
|
||||
{
|
||||
"name": "test_stream",
|
||||
"stream_id": 11,
|
||||
"description": "",
|
||||
"rendered_description": "",
|
||||
"invite_only": false,
|
||||
"is_web_public": false,
|
||||
"stream_post_policy": 1,
|
||||
"history_public_to_subscribers": true,
|
||||
"first_message_id": null,
|
||||
"message_retention_days": null,
|
||||
"is_announcement_only": false,
|
||||
},
|
||||
],
|
||||
"id": 2,
|
||||
}
|
||||
- type: object
|
||||
description: |
|
||||
Event sent to all users who can see a stream when it is deactivated.
|
||||
|
||||
@ -206,6 +206,7 @@ class BaseAction(ZulipTestCase):
|
||||
client_gravatar: bool=True,
|
||||
user_avatar_url_field_optional: bool=False,
|
||||
slim_presence: bool=False,
|
||||
include_streams: bool=True,
|
||||
num_events: int=1,
|
||||
bulk_message_deletion: bool=True) -> List[Dict[str, Any]]:
|
||||
'''
|
||||
@ -241,6 +242,7 @@ class BaseAction(ZulipTestCase):
|
||||
user_avatar_url_field_optional=user_avatar_url_field_optional,
|
||||
slim_presence=slim_presence,
|
||||
include_subscribers=include_subscribers,
|
||||
include_streams=include_streams,
|
||||
realm=self.user_profile.realm,
|
||||
)
|
||||
action()
|
||||
@ -282,6 +284,7 @@ class BaseAction(ZulipTestCase):
|
||||
user_avatar_url_field_optional=user_avatar_url_field_optional,
|
||||
slim_presence=slim_presence,
|
||||
include_subscribers=include_subscribers,
|
||||
include_streams=include_streams,
|
||||
realm=self.user_profile.realm,
|
||||
)
|
||||
post_process_state(self.user_profile, normal_state, notification_settings_null)
|
||||
@ -1862,7 +1865,8 @@ class SubscribeActionTest(BaseAction):
|
||||
events = self.verify_action(
|
||||
action,
|
||||
include_subscribers=include_subscribers,
|
||||
num_events=3)
|
||||
include_streams=False,
|
||||
num_events=2)
|
||||
check_subscription_remove('events[0]', events[0])
|
||||
self.assertEqual(len(events[0]['subscriptions']), 1)
|
||||
self.assertEqual(
|
||||
@ -1875,8 +1879,9 @@ class SubscribeActionTest(BaseAction):
|
||||
events = self.verify_action(
|
||||
action,
|
||||
include_subscribers=include_subscribers,
|
||||
num_events=2)
|
||||
check_subscription_add('events[1]', events[1], include_subscribers)
|
||||
include_streams=False,
|
||||
num_events=1)
|
||||
check_subscription_add('events[0]', events[0], include_subscribers)
|
||||
|
||||
action = lambda: do_change_stream_description(stream, 'new description')
|
||||
events = self.verify_action(
|
||||
|
||||
@ -649,7 +649,7 @@ class LoginTest(ZulipTestCase):
|
||||
with queries_captured() as queries:
|
||||
self.register(self.nonreg_email('test'), "test")
|
||||
# Ensure the number of queries we make is not O(streams)
|
||||
self.assertEqual(len(queries), 78)
|
||||
self.assertEqual(len(queries), 75)
|
||||
user_profile = self.nonreg_user('test')
|
||||
self.assert_logged_in_user_id(user_profile.id)
|
||||
self.assertFalse(user_profile.enable_stream_desktop_notifications)
|
||||
|
||||
@ -2560,7 +2560,7 @@ class SubscriptionAPITest(ZulipTestCase):
|
||||
self.helper_check_subs_before_and_after_add(self.streams + add_streams, {},
|
||||
add_streams, self.streams, self.test_email,
|
||||
self.streams + add_streams, self.test_realm)
|
||||
self.assert_length(events, 8)
|
||||
self.assert_length(events, 7)
|
||||
|
||||
def test_successful_subscriptions_add_with_announce(self) -> None:
|
||||
"""
|
||||
@ -2588,7 +2588,7 @@ class SubscriptionAPITest(ZulipTestCase):
|
||||
self.helper_check_subs_before_and_after_add(self.streams + add_streams, other_params,
|
||||
add_streams, self.streams, self.test_email,
|
||||
self.streams + add_streams, self.test_realm)
|
||||
self.assertEqual(len(events), 9)
|
||||
self.assertEqual(len(events), 8)
|
||||
|
||||
def test_successful_subscriptions_notifies_pm(self) -> None:
|
||||
"""
|
||||
@ -2890,9 +2890,9 @@ class SubscriptionAPITest(ZulipTestCase):
|
||||
streams_to_sub,
|
||||
dict(principals=orjson.dumps([user1.id, user2.id]).decode()),
|
||||
)
|
||||
self.assert_length(queries, 40)
|
||||
self.assert_length(queries, 37)
|
||||
|
||||
self.assert_length(events, 7)
|
||||
self.assert_length(events, 6)
|
||||
for ev in [x for x in events if x['event']['type'] not in ('message', 'stream')]:
|
||||
if ev['event']['op'] == 'add':
|
||||
self.assertEqual(
|
||||
@ -2917,7 +2917,7 @@ class SubscriptionAPITest(ZulipTestCase):
|
||||
streams_to_sub,
|
||||
dict(principals=orjson.dumps([self.test_user.id]).decode()),
|
||||
)
|
||||
self.assert_length(queries, 14)
|
||||
self.assert_length(queries, 12)
|
||||
|
||||
self.assert_length(events, 2)
|
||||
add_event, add_peer_event = events
|
||||
@ -3298,7 +3298,7 @@ class SubscriptionAPITest(ZulipTestCase):
|
||||
|
||||
# The only known O(N) behavior here is that we call
|
||||
# principal_to_user_profile for each of our users.
|
||||
self.assert_length(queries, 22)
|
||||
self.assert_length(queries, 19)
|
||||
|
||||
def test_subscriptions_add_for_principal(self) -> None:
|
||||
"""
|
||||
@ -3719,7 +3719,7 @@ class SubscriptionAPITest(ZulipTestCase):
|
||||
[new_streams[0]],
|
||||
dict(principals=orjson.dumps([user1.id, user2.id]).decode()),
|
||||
)
|
||||
self.assert_length(queries, 40)
|
||||
self.assert_length(queries, 37)
|
||||
|
||||
# Test creating private stream.
|
||||
with queries_captured() as queries:
|
||||
@ -3729,7 +3729,7 @@ class SubscriptionAPITest(ZulipTestCase):
|
||||
dict(principals=orjson.dumps([user1.id, user2.id]).decode()),
|
||||
invite_only=True,
|
||||
)
|
||||
self.assert_length(queries, 40)
|
||||
self.assert_length(queries, 38)
|
||||
|
||||
# Test creating a public stream with announce when realm has a notification stream.
|
||||
notifications_stream = get_stream(self.streams[0], self.test_realm)
|
||||
@ -3744,7 +3744,7 @@ class SubscriptionAPITest(ZulipTestCase):
|
||||
principals=orjson.dumps([user1.id, user2.id]).decode(),
|
||||
),
|
||||
)
|
||||
self.assert_length(queries, 52)
|
||||
self.assert_length(queries, 49)
|
||||
|
||||
class GetStreamsTest(ZulipTestCase):
|
||||
def test_streams_api_for_bot_owners(self) -> None:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user