fix_unreads: Mark pre-pointer messages as unread (unless muted).

zerver/management/commands/fix_unreads.py

@@ -12,6 +12,7 @@
 
 from zerver.lib.management import ZulipBaseCommand
 from zerver.models import (
+    Stream,
     UserProfile
 )
 
@@ -117,6 +118,120 @@ def fix():
 
     cursor.close()
 
+def build_topic_mute_checker(user_profile):
+    # type: (UserProfile) -> Callable[[int, Text], bool]
+    rows = ujson.loads(user_profile.muted_topics)
+    stream_names = {row[0] for row in rows}
+    stream_dict = dict()
+    for name in stream_names:
+        stream_id = Stream.objects.get(
+            name__iexact=name.strip(),
+            realm_id=user_profile.realm_id,
+        ).id
+        stream_dict[name] = stream_id
+    tups = set()
+    for row in rows:
+        stream_name = row[0]
+        topic = row[1]
+        stream_id = stream_dict[stream_name]
+        tups.add((stream_id, topic.lower()))
+
+    def is_muted(stream_id, topic):
+        # type: (int, Text) -> bool
+        return (stream_id, topic.lower()) in tups
+
+    return is_muted
+
+def fix_pre_pointer(user_profile):
+    # type: (UserProfile) -> None
+
+    cursor = connection.cursor()
+
+    pointer = user_profile.pointer
+
+    if not pointer:
+        return
+
+    is_topic_muted = build_topic_mute_checker(user_profile)
+
+    recipient_ids = []
+
+    def find_non_muted_recipients():
+        # type: () -> None
+        query = '''
+            SELECT
+                zerver_subscription.recipient_id
+            FROM
+                zerver_subscription
+            INNER JOIN zerver_recipient ON (
+                zerver_recipient.id = zerver_subscription.recipient_id
+            )
+            WHERE (
+                zerver_subscription.user_profile_id = '%s' AND
+                zerver_recipient.type = 2 AND
+                zerver_subscription.in_home_view AND
+                zerver_subscription.active
+            )
+        '''
+        cursor.execute(query, [user_profile.id])
+        rows = cursor.fetchall()
+        for row in rows:
+            recipient_ids.append(row[0])
+        print(recipient_ids)
+
+    get_timing(
+        'find_non_muted_recipients',
+        find_non_muted_recipients
+    )
+
+    user_message_ids = []
+
+    def find_old_ids():
+        # type: () -> None
+        recips = ', '.join(str(id) for id in recipient_ids)
+
+        query = '''
+            SELECT
+                zerver_usermessage.id,
+                zerver_recipient.type_id,
+                subject
+            FROM
+                zerver_usermessage
+            INNER JOIN zerver_message ON (
+                zerver_message.id = zerver_usermessage.message_id
+            )
+            INNER JOIN zerver_recipient ON (
+                zerver_recipient.id = zerver_message.recipient_id
+            )
+            WHERE (
+                zerver_usermessage.user_profile_id = %s AND
+                zerver_message.id <= %s AND
+                (zerver_usermessage.flags & 1) = 0 AND
+                zerver_message.recipient_id in (%s)
+            )
+        ''' % (user_profile.id, pointer, recips)
+
+        print('''
+            EXPLAIN analyze''' + query.rstrip() + ';')
+
+        cursor.execute(query)
+        rows = cursor.fetchall()
+        for (um_id, stream_id, topic) in rows:
+            if not is_topic_muted(stream_id, topic):
+                user_message_ids.append(um_id)
+        print('rows found: %d' % (len(user_message_ids),))
+
+    get_timing(
+        'finding pre-pointer messages that are not muted',
+        find_old_ids
+    )
+    cursor.close()
+
+def fix(user_profile):
+    # type: (UserProfile) -> None
+    fix_unsubscribed(user_profile)
+    fix_pre_pointer(user_profile)
+
 class Command(ZulipBaseCommand):
     help = """Fix problems related to unread counts."""
 
@@ -136,4 +251,4 @@ def handle(self, *args, **options):
             print("e-mail %s doesn't exist in the realm %s, skipping" % (email, realm))
             return
 
-        fix_unsubscribed(user_profile)
+        fix(user_profile)