Calculate idle users more efficiently when sending messages.

Usually a small minority of users are eligible to receive missed
message emails or mobile notifications.

We now filter users first before hitting UserPresence to find idle
users.  We also simply check for the existence of recent activity
rather than borrowing the more complicated data structures that we
use for the buddy list.

zerver/lib/actions.py

@@ -889,18 +889,22 @@ def do_send_messages(messages_maybe_none):
 
         user_flags = user_message_flags.get(message['message'].id, {})
         sender = message['message'].sender
-        user_presences = get_status_dict(sender)
-        presences = {}
-        for user_profile in message['active_recipients']:
-            if user_profile.email in user_presences:
-                presences[user_profile.id] = user_presences[user_profile.email]
+        message_type = message_dict_no_markdown['type']
+
+        missed_message_userids = get_userids_for_missed_messages(
+            realm=sender.realm,
+            sender_id=sender.id,
+            message_type=message_type,
+            active_recipients=message['active_recipients'],
+            user_flags=user_flags,
+        )
 
         event = dict(
             type='message',
             message=message['message'].id,
             message_dict_markdown=message_dict_markdown,
             message_dict_no_markdown=message_dict_no_markdown,
-            presences=presences,
+            missed_message_userids=missed_message_userids,
         )
 
         users = [{'id': user.id,
@@ -3140,6 +3144,34 @@ def gather_subscriptions(user_profile):
 
     return (subscribed, unsubscribed)
 
+def get_userids_for_missed_messages(realm, sender_id, message_type, active_recipients, user_flags):
+    # type: (Realm, int, str, List[UserProfile], Dict[int, List[str]]) -> List[int]
+    if realm.presence_disabled:
+        return []
+
+    is_pm = message_type == 'private'
+
+    user_ids = set()
+    for user in active_recipients:
+        flags = user_flags.get(user.id, [])  # type: Iterable[str]
+        mentioned = 'mentioned' in flags
+        received_pm = is_pm and user.id != sender_id
+        if mentioned or received_pm:
+            user_ids.add(user.id)
+
+    if not user_ids:
+        return []
+
+    # 140 seconds is consistent with presence.js:OFFLINE_THRESHOLD_SECS
+    recent = timezone_now() - datetime.timedelta(seconds=140)
+    rows = UserPresence.objects.filter(
+        user_profile_id__in=user_ids,
+        timestamp__gte=recent
+    ).distinct('user_profile_id').values('user_profile_id')
+    active_user_ids = {row['user_profile_id'] for row in rows}
+    idle_user_ids = user_ids - active_user_ids
+    return sorted(list(idle_user_ids))
+
 def get_status_dict(requesting_user_profile):
     # type: (UserProfile) -> Dict[Text, Dict[Text, Dict[str, Any]]]
     if requesting_user_profile.realm.presence_disabled:

zerver/tests/test_messages.py

@@ -15,6 +15,7 @@ from zilencer.models import Deployment
 from zerver.lib.addressee import Addressee
 
 from zerver.lib.actions import (
+    get_userids_for_missed_messages,
     internal_send_private_message,
 )
 
@@ -42,7 +43,7 @@ from zerver.lib.soft_deactivation import add_missing_messages, do_soft_deactivat
 from zerver.models import (
     MAX_MESSAGE_LENGTH, MAX_SUBJECT_LENGTH,
     Message, Realm, Recipient, Stream, UserMessage, UserProfile, Attachment,
-    RealmAuditLog, RealmDomain, get_realm,
+    RealmAuditLog, RealmDomain, get_realm, UserPresence,
     get_stream, get_recipient, get_system_bot, get_user, Reaction,
     sew_messages_and_reactions, flush_per_request_caches
 )
@@ -68,7 +69,7 @@ import mock
 import time
 import ujson
 from six.moves import range
-from typing import Any, List, Optional, Text
+from typing import Any, Dict, List, Optional, Text
 
 from collections import namedtuple
 
@@ -510,7 +511,7 @@ class StreamMessagesTest(ZulipTestCase):
         with queries_captured() as queries:
             send_message()
 
-        self.assert_length(queries, 16)
+        self.assert_length(queries, 14)
 
     def test_stream_message_dict(self):
         # type: () -> None
@@ -2034,6 +2035,55 @@ class AttachmentTest(ZulipTestCase):
             attachment = Attachment.objects.get(path_id=path_id)
             self.assertTrue(attachment.is_claimed())
 
+class MissedMessageTest(ZulipTestCase):
+    def test_missed_message_userids(self):
+        # type: () -> None
+        UserPresence.objects.all().delete()
+
+        sender = self.example_user('cordelia')
+        realm = sender.realm
+        hamlet = self.example_user('hamlet')
+        othello = self.example_user('othello')
+        recipients = [hamlet, othello]
+        message_type = 'stream'
+        user_flags = {}  # type: Dict[int, List[str]]
+
+        def assert_missing(user_ids):
+            # type: (List[int]) -> None
+            missed_message_userids = get_userids_for_missed_messages(
+                realm=realm,
+                sender_id=sender.id,
+                message_type=message_type,
+                active_recipients=recipients,
+                user_flags=user_flags,
+            )
+            self.assertEqual(sorted(user_ids), sorted(missed_message_userids))
+
+        def set_presence(user_id, client_name, ago):
+            # type: (int, Text, int) -> None
+            when = timezone_now() - datetime.timedelta(seconds=ago)
+            UserPresence.objects.create(
+                user_profile_id=user_id,
+                client=get_client(client_name),
+                timestamp=when,
+            )
+
+        message_type = 'private'
+        assert_missing([hamlet.id, othello.id])
+
+        message_type = 'stream'
+        user_flags[hamlet.id] = ['mentioned']
+        assert_missing([hamlet.id])
+
+        set_presence(hamlet.id, 'iPhone', ago=5000)
+        assert_missing([hamlet.id])
+
+        set_presence(hamlet.id, 'webapp', ago=15)
+        assert_missing([])
+
+        message_type = 'private'
+        assert_missing([othello.id])
+
 class LogDictTest(ZulipTestCase):
     def test_to_log_dict(self):
         # type: () -> None

zerver/tests/test_signup.py

@@ -314,7 +314,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.assert_length(queries, 67)
+        self.assert_length(queries, 66)
         user_profile = self.nonreg_user('test')
         self.assertEqual(get_session_dict_user(self.client.session), user_profile.id)
         self.assertFalse(user_profile.enable_stream_desktop_notifications)

zerver/tests/test_subs.py

@@ -1618,7 +1618,7 @@ class SubscriptionAPITest(ZulipTestCase):
                     streams_to_sub,
                     dict(principals=ujson.dumps([email1, email2])),
                 )
-        self.assert_length(queries, 42)
+        self.assert_length(queries, 40)
 
         self.assert_length(events, 7)
         for ev in [x for x in events if x['event']['type'] not in ('message', 'stream')]:

zerver/tornado/event_queue.py

@@ -654,39 +654,9 @@ def receiver_is_off_zulip(user_profile_id):
     off_zulip = len(message_event_queues) == 0
     return off_zulip
 
-def receiver_is_idle(user_profile_id, realm_presences):
-    # type: (int, Optional[Dict[int, Dict[Text, Dict[str, Any]]]]) -> bool
-
-    # It's possible a recipient is not in the realm of a sender. We don't have
-    # presence information in this case (and it's hard to get without an additional
-    # db query) so we simply don't try to guess if this cross-realm recipient
-    # has been idle for too long
-    if realm_presences is None or user_profile_id not in realm_presences:
-        return False
-
-    # We want to find the newest "active" presence entity and compare that to the
-    # activity expiry threshold.
-    user_presence = realm_presences[user_profile_id]
-    latest_active_timestamp = None
-    idle = False
-
-    for client, status in six.iteritems(user_presence):
-        if (latest_active_timestamp is None or status['timestamp'] > latest_active_timestamp) and \
-                status['status'] == 'active':
-            latest_active_timestamp = status['timestamp']
-
-    if latest_active_timestamp is None:
-        idle = True
-    else:
-        active_datetime = timestamp_to_datetime(latest_active_timestamp)
-        # 140 seconds is consistent with presence.js:OFFLINE_THRESHOLD_SECS
-        idle = timezone_now() - active_datetime > datetime.timedelta(seconds=140)
-
-    return idle
-
 def process_message_event(event_template, users):
     # type: (Mapping[str, Any], Iterable[Mapping[str, Any]]) -> None
-    realm_presences = {int(k): v for k, v in event_template['presences'].items()}  # type: Dict[int, Dict[Text, Dict[str, Any]]]
+    missed_message_userids = set(event_template.get('missed_message_userids', []))
     sender_queue_id = event_template.get('sender_queue_id', None)  # type: Optional[str]
     message_dict_markdown = event_template['message_dict_markdown']  # type: Dict[str, Any]
     message_dict_no_markdown = event_template['message_dict_no_markdown']  # type: Dict[str, Any]
@@ -722,7 +692,7 @@ def process_message_event(event_template, users):
         mentioned = 'mentioned' in flags
 
         if (received_pm or mentioned):
-            idle = receiver_is_off_zulip(user_profile_id) or receiver_is_idle(user_profile_id, realm_presences)
+            idle = receiver_is_off_zulip(user_profile_id) or (user_profile_id in missed_message_userids)
             always_push_notify = user_data.get('always_push_notify', False)
 
             if (idle or always_push_notify):