push notifications: Add support for removing GCM push notifications.

This uses the recently introduced active_mobile_push_notification
flag; messages that have had a mobile push notification sent will have
a removal push notification sent as soon as they are marked as read.

Note that this feature is behind a setting,
SEND_REMOVE_PUSH_NOTIFICATIONS, since the notification format is not
supported by the mobile apps yet, and we want to give a grace period
before we start sending notifications that appear as (null) to
clients.  But the tracking logic to maintain the set of message IDs
with an active push notification runs unconditionally.

This is designed with at-least-once semantics; so mobile clients need
to handle the possibility that they receive duplicat requests to
remove a push notification.

We reuse the existing missedmessage_mobile_notifications queue
processor for the work, to avoid materially impacting the latency of
marking messages as read.

Fixes #7459, though we'll need to open a follow-up issue for
using these data on iOS.

zerver/lib/actions.py

@@ -3549,11 +3549,19 @@ def do_update_pointer(user_profile: UserProfile, client: Client,
         # that will mark as read any messages up until the pointer
         # move; we expect to remove this feature entirely before long,
         # when we drop support for the old Android app entirely.
+        app_message_ids = UserMessage.objects.filter(
+            user_profile=user_profile,
+            flags=UserMessage.flags.active_mobile_push_notification,
+            message__id__gt=prev_pointer,
+            message__id__lte=pointer).extra(where=[
+                UserMessage.where_unread(),
+            ]).values_list("message_id", flat=True)
+
         UserMessage.objects.filter(user_profile=user_profile,
                                    message__id__gt=prev_pointer,
-                                   message__id__lte=pointer).extra(
-                                       where=[UserMessage.where_unread()]).update(
-                                           flags=F('flags').bitor(UserMessage.flags.read))
+                                   message__id__lte=pointer).extra(where=[UserMessage.where_unread()]) \
+                           .update(flags=F('flags').bitor(UserMessage.flags.read))
+        do_clear_mobile_push_notifications_for_ids(user_profile, app_message_ids)
 
     event = dict(type='pointer', pointer=pointer)
     send_event(event, [user_profile.id])
@@ -3581,6 +3589,13 @@ def do_mark_all_as_read(user_profile: UserProfile, client: Client) -> int:
     send_event(event, [user_profile.id])
 
     statsd.incr("mark_all_as_read", count)
+
+    all_push_message_ids = UserMessage.objects.filter(
+        user_profile=user_profile,
+        flags=UserMessage.flags.active_mobile_push_notification,
+    ).values_list("message_id", flat=True)
+    do_clear_mobile_push_notifications_for_ids(user_profile, all_push_message_ids)
+
     return count
 
 def do_mark_stream_messages_as_read(user_profile: UserProfile,
@@ -3617,10 +3632,24 @@ def do_mark_stream_messages_as_read(user_profile: UserProfile,
         all=False,
     )
     send_event(event, [user_profile.id])
+    do_clear_mobile_push_notifications_for_ids(user_profile, message_ids)
 
     statsd.incr("mark_stream_as_read", count)
     return count
 
+def do_clear_mobile_push_notifications_for_ids(user_profile: UserProfile,
+                                               message_ids: List[int]) -> None:
+    for user_message in UserMessage.objects.filter(
+            message_id__in=message_ids,
+            flags=UserMessage.flags.active_mobile_push_notification,
+            user_profile=user_profile):
+        event = {
+            "user_profile_id": user_profile.id,
+            "message_id": user_message.message_id,
+            "type": "remove",
+        }
+        queue_json_publish("missedmessage_mobile_notifications", event)
+
 def do_update_message_flags(user_profile: UserProfile,
                             client: Client,
                             operation: str,
@@ -3665,6 +3694,9 @@ def do_update_message_flags(user_profile: UserProfile,
              'all': False}
     send_event(event, [user_profile.id])
 
+    if flag == "read" and operation == "add":
+        do_clear_mobile_push_notifications_for_ids(user_profile, messages)
+
     statsd.incr("flags.%s.%s" % (flag, operation), count)
     return count
 

zerver/lib/push_notifications.py

@@ -556,7 +556,17 @@ def handle_remove_push_notification(user_profile_id: int, message_id: int) -> No
 
     """
     user_profile = get_user_profile_by_id(user_profile_id)
-    message = access_message(user_profile, message_id)[0]
+    message, user_message = access_message(user_profile, message_id)
+
+    if not settings.SEND_REMOVE_PUSH_NOTIFICATIONS:
+        # It's a little annoying that we duplicate this flag-clearing
+        # code (also present below), but this block is scheduled to be
+        # removed in a few weeks, once the app has supported the
+        # feature for long enough.
+        user_message.flags.active_mobile_push_notification = False
+        user_message.save(update_fields=["flags"])
+        return
+
     gcm_payload = get_common_payload(message)
     gcm_payload.update({
         'event': 'remove',
@@ -581,6 +591,9 @@ def handle_remove_push_notification(user_profile_id: int, message_id: int) -> No
     if android_devices:
         send_android_push_notification(android_devices, gcm_payload)
 
+    user_message.flags.active_mobile_push_notification = False
+    user_message.save(update_fields=["flags"])
+
 @statsd_increment("push_notifications")
 def handle_push_notification(user_profile_id: int, missed_message: Dict[str, Any]) -> None:
     """
@@ -602,6 +615,12 @@ def handle_push_notification(user_profile_id: int, missed_message: Dict[str, Any
         # the `zerver/tornado/event_queue.py` logic?
         if user_message.flags.read:
             return
+
+        # Otherwise, we mark the message as having an active mobile
+        # push notification, so that we can send revocation messages
+        # later.
+        user_message.flags.active_mobile_push_notification = True
+        user_message.save(update_fields=["flags"])
     else:
         # Users should only be getting push notifications into this
         # queue for messages they haven't received if they're

zerver/models.py

@@ -1535,6 +1535,8 @@ class AbstractUserMessage(models.Model):
         # Use this for Django ORM queries where we are getting lots
         # of rows.  This custom SQL plays nice with our partial indexes.
         # Grep the code for example usage.
+        #
+        # This optimization is only helpful for checking a flag being False.
         return 'flags & 1 = 0'
 
     def flags_list(self) -> List[str]:

zerver/tests/test_push_notifications.py

@@ -536,6 +536,7 @@ class HandlePushNotificationTest(PushNotificationTest):
             mock_send_android.assert_called_with(android_devices,
                                                  {'gcm': True})
 
+    @override_settings(SEND_REMOVE_PUSH_NOTIFICATIONS=True)
     def test_send_remove_notifications_to_bouncer(self) -> None:
         user_profile = self.example_user('hamlet')
         message = self.get_message(Recipient.PERSONAL, type_id=1)
@@ -555,6 +556,7 @@ class HandlePushNotificationTest(PushNotificationTest):
                                                   'event': 'remove',
                                                   'zulip_message_id': message.id})
 
+    @override_settings(SEND_REMOVE_PUSH_NOTIFICATIONS=True)
     def test_non_bouncer_push_remove(self) -> None:
         message = self.get_message(Recipient.PERSONAL, type_id=1)
         UserMessage.objects.create(

zerver/tests/test_signup.py

@@ -454,7 +454,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, 77)
+        self.assert_length(queries, 78)
         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_unread.py

@@ -3,6 +3,7 @@
 from typing import Any, Dict, List, Mapping
 
 from django.db import connection
+from django.test import override_settings
 
 from zerver.models import (
     get_realm,
@@ -13,6 +14,7 @@ from zerver.models import (
     Stream,
     Subscription,
     UserMessage,
+    UserProfile,
 )
 
 from zerver.lib.fix_unreads import (
@@ -471,3 +473,60 @@ class FixUnreadTests(ZulipTestCase):
         assert_unread(um_muted_stream_id)
         assert_unread(um_post_pointer_id)
         assert_read(um_unsubscribed_id)
+
+class PushNotificationMarkReadFlowsTest(ZulipTestCase):
+    def get_mobile_push_notification_ids(self, user_profile: UserProfile) -> List[int]:
+        return list(UserMessage.objects.filter(
+            user_profile=user_profile,
+            flags=UserMessage.flags.active_mobile_push_notification).order_by(
+                "message_id").values_list("message_id", flat=True))
+
+    @override_settings(SEND_REMOVE_PUSH_NOTIFICATIONS=True)
+    def test_track_active_mobile_push_notifications(self) -> None:
+        self.login(self.example_email("hamlet"))
+        user_profile = self.example_user('hamlet')
+        stream = self.subscribe(user_profile, "test_stream")
+        second_stream = self.subscribe(user_profile, "second_stream")
+
+        property_name = "push_notifications"
+        result = self.api_post(user_profile.email, "/api/v1/users/me/subscriptions/properties",
+                               {"subscription_data": ujson.dumps([{"property": property_name,
+                                                                   "value": True,
+                                                                   "stream_id": stream.id}])})
+        result = self.api_post(user_profile.email, "/api/v1/users/me/subscriptions/properties",
+                               {"subscription_data": ujson.dumps([{"property": property_name,
+                                                                   "value": True,
+                                                                   "stream_id": second_stream.id}])})
+        self.assert_json_success(result)
+        self.assertEqual(self.get_mobile_push_notification_ids(user_profile), [])
+
+        message_id = self.send_stream_message(self.example_email("cordelia"), "test_stream", "hello", "test_topic")
+        second_message_id = self.send_stream_message(self.example_email("cordelia"), "test_stream", "hello", "other_topic")
+        third_message_id = self.send_stream_message(self.example_email("cordelia"), "second_stream", "hello", "test_topic")
+
+        self.assertEqual(self.get_mobile_push_notification_ids(user_profile),
+                         [message_id, second_message_id, third_message_id])
+
+        result = self.client_post("/json/mark_topic_as_read", {
+            "stream_id": str(stream.id),
+            "topic_name": "test_topic",
+        })
+
+        self.assert_json_success(result)
+        self.assertEqual(self.get_mobile_push_notification_ids(user_profile),
+                         [second_message_id, third_message_id])
+
+        result = self.client_post("/json/mark_stream_as_read", {
+            "stream_id": str(stream.id),
+            "topic_name": "test_topic",
+        })
+        self.assertEqual(self.get_mobile_push_notification_ids(user_profile),
+                         [third_message_id])
+
+        fourth_message_id = self.send_stream_message(self.example_email("cordelia"), "test_stream", "hello", "test_topic")
+        self.assertEqual(self.get_mobile_push_notification_ids(user_profile),
+                         [third_message_id, fourth_message_id])
+
+        result = self.client_post("/json/mark_all_as_read", {})
+        self.assertEqual(self.get_mobile_push_notification_ids(user_profile),
+                         [])

zerver/worker/queue_processors.py

@@ -22,7 +22,7 @@ from zerver.lib.feedback import handle_feedback
 from zerver.lib.queue import SimpleQueueClient, queue_json_publish, retry_event
 from zerver.lib.timestamp import timestamp_to_datetime
 from zerver.lib.notifications import handle_missedmessage_emails
-from zerver.lib.push_notifications import handle_push_notification
+from zerver.lib.push_notifications import handle_push_notification, handle_remove_push_notification
 from zerver.lib.actions import do_send_confirmation_email, \
     do_update_user_activity, do_update_user_activity_interval, do_update_user_presence, \
     internal_send_message, check_send_message, extract_recipients, \
@@ -306,6 +306,8 @@ class MissedMessageSendingWorker(EmailSendingWorker):  # nocoverage
 @assign_queue('missedmessage_mobile_notifications')
 class PushNotificationsWorker(QueueProcessingWorker):  # nocoverage
     def consume(self, data: Mapping[str, Any]) -> None:
+        if data.get("type", "add") == "remove":
+            handle_remove_push_notification(data['user_profile_id'], data['message_id'])
         handle_push_notification(data['user_profile_id'], data)
 
 # We probably could stop running this queue worker at all if ENABLE_FEEDBACK is False

zproject/settings.py

@@ -220,6 +220,11 @@ DEFAULT_SETTINGS = {
     'SEND_LOGIN_EMAILS': True,
     'EMBEDDED_BOTS_ENABLED': False,
 
+    # Temporary setting while we wait for app support for removing
+    # push notifications.  Controls whether the Zulip server sends
+    # cancellation notices for previously sent push notifications.
+    'SEND_REMOVE_PUSH_NOTIFICATIONS': False,
+
     # Two Factor Authentication is not yet implementation-complete
     'TWO_FACTOR_AUTHENTICATION_ENABLED': False,