Add backend support for handling new 'read' message flag

(imported from commit 6194e9332caa2d279cbc304f0d6a69f969aa9a72)

humbug/urls.py

@@ -75,6 +75,7 @@ urlpatterns = patterns('',
     url(r'^json/change_enter_sends$',       'zephyr.views.json_change_enter_sends'),
     url(r'^json/get_profile$',              'zephyr.views.json_get_profile'),
     url(r'^json/report_error$',             'zephyr.views.json_report_error'),
+    url(r'^json/update_message_flags$',     'zephyr.views.json_update_flags'),
 
     # These are json format views used by the API.  They require an API key.
     url(r'^api/v1/get_messages$',           'zephyr.tornadoviews.api_get_messages'),

zephyr/lib/actions.py

@@ -6,6 +6,7 @@ from zephyr.models import Realm, Stream, UserProfile, UserActivity, \
     DefaultStream, StreamColor, UserPresence, \
     MAX_MESSAGE_LENGTH, get_client, get_stream
 from django.db import transaction, IntegrityError
+from django.db.models import F
 from zephyr.lib.initial_password import initial_password
 from zephyr.lib.timestamp import timestamp_to_datetime, datetime_to_timestamp
 from zephyr.lib.message_cache import cache_save_message
@@ -152,6 +153,9 @@ def do_send_message(message, rendered_content=None, no_log=False):
         ums_to_create = [UserMessage(user_profile=user_profile, message=message)
                          for user_profile in recipients
                          if user_profile.user.is_active]
+        for um in ums_to_create:
+            if um.user_profile == message.sender:
+                um.flags |= UserMessage.flags.read
         batch_bulk_create(UserMessage, ums_to_create)
 
     cache_save_message(message)
@@ -398,9 +402,9 @@ def do_update_user_presence(user_profile, client, log_time, status):
     presence.save()
 
 if settings.USING_RABBITMQ or settings.TEST_SUITE:
-    # RabbitMQ is required for idle functionality
+    # RabbitMQ is required for idle and unread functionality
     if settings.USING_RABBITMQ:
-        presence_queue = SimpleQueueClient()
+        actions_queue = SimpleQueueClient()
 
     def update_user_presence(user_profile, client, log_time, status):
         event={'type': 'user_presence',
@@ -410,11 +414,24 @@ if settings.USING_RABBITMQ or settings.TEST_SUITE:
                'client': client.name}
 
         if settings.USING_RABBITMQ:
-            presence_queue.json_publish("user_activity", event)
+            actions_queue.json_publish("user_activity", event)
         elif settings.TEST_SUITE:
             process_user_presence_event(event)
+
+    def update_message_flags(user_profile, operation, flag, messages, all):
+        event = {'type':            'update_message',
+                 'user_profile_id': user_profile.id,
+                 'operation':       operation,
+                 'flag':            flag,
+                 'messages':        messages,
+                 'all':             all}
+        if settings.USING_RABBITMQ:
+            actions_queue.json_publish("user_activity", event)
+        else:
+            return process_update_message_flags(event)
 else:
     update_user_presence = lambda user_profile, client, log_time, status: None
+    update_message_flags = lambda user_profile, operation, flag, messages, all: None
 
 def process_user_presence_event(event):
     user_profile = UserProfile.objects.get(id=event["user_profile_id"])
@@ -423,6 +440,28 @@ def process_user_presence_event(event):
     status = event["status"]
     return do_update_user_presence(user_profile, client, log_time, status)
 
+def process_update_message_flags(event):
+    user_profile = UserProfile.objects.get(id=event["user_profile_id"])
+    try:
+        msg_ids = event["messages"]
+        flag = getattr(UserMessage.flags, event["flag"])
+        op = event["operation"]
+    except (KeyError, AttributeError):
+        return False
+
+    if event["all"] == True:
+        messages = UserMessage.objects.filter(user_profile=user_profile)
+    else:
+        messages = UserMessage.objects.filter(user_profile=user_profile,
+                                              message__id__in=msg_ids)
+
+    if op == "add":
+        messages.update(flags=F('flags') | flag)
+    elif op == "remove":
+        messages.update(flags=F('flags') & ~flag)
+
+    return True
+
 def subscribed_to_stream(user_profile, stream):
     try:
         if Subscription.objects.get(user_profile=user_profile,

zephyr/management/commands/process_user_activity.py

@@ -2,7 +2,8 @@ from optparse import make_option
 from django.core.management.base import BaseCommand
 import simplejson
 import pika
-from zephyr.lib.actions import process_user_activity_event, process_user_presence_event
+from zephyr.lib.actions import process_user_activity_event, \
+        process_user_presence_event, process_update_message_flags
 from zephyr.lib.queue import SimpleQueueClient
 import sys
 import signal
@@ -21,6 +22,8 @@ class Command(BaseCommand):
                 process_user_activity_event(event)
             elif msg_type == 'user_presence':
                 process_user_presence_event(event)
+            elif msg_type == 'update_message':
+                process_update_message_flags(event)
             else:
                 print("[*] Unknown message type: %s" (msg_type,))
 

zephyr/models.py

@@ -271,10 +271,10 @@ class UserMessage(models.Model):
 
     def __repr__(self):
         display_recipient = get_display_recipient(self.message.recipient)
-        return "<UserMessage: %s / %s>" % (display_recipient, self.user_profile.user.email)
+        return "<UserMessage: %s / %s (%s)>" % (display_recipient, self.user_profile.user.email, self.flags_dict())
 
     def flags_dict(self):
-        return dict(flags = self.flags.keys())
+        return dict(flags = [flag for flag in self.flags.keys() if getattr(self.flags, flag).is_set])
 
 
 class Subscription(models.Model):

zephyr/tests.py

@@ -6,7 +6,7 @@ from django.db.models import Q
 
 from zephyr.models import Message, UserProfile, Stream, Recipient, Subscription, \
     filter_by_subscriptions, get_display_recipient, Realm, Client, \
-    PreregistrationUser
+    PreregistrationUser, UserMessage
 from zephyr.tornadoviews import json_get_updates, api_get_messages
 from zephyr.decorator import RespondAsynchronously, RequestVariableConversionError
 from zephyr.lib.initial_password import initial_password, initial_api_key
@@ -2144,6 +2144,86 @@ class UserPresenceTests(AuthedTestCase):
         for email in json['presences'].keys():
             self.assertEqual(email.split('@')[1], 'humbughq.com')
 
+class UnreadCountTests(AuthedTestCase):
+    fixtures = ['messages.json']
+
+    def get_old_messages(self):
+        post_params = {"anchor": 1, "num_before": 1, "num_after": 1}
+        result = self.client.post("/json/get_old_messages", dict(post_params))
+        data = simplejson.loads(result.content)
+        return data['messages']
+
+    def test_initial_counts(self):
+        # All test users have a pointer at -1, so all messages are read
+        for user in UserProfile.objects.all():
+            for message in UserMessage.objects.filter(user_profile=user):
+                self.assertFalse(message.flags.read)
+
+        self.login('hamlet@humbughq.com')
+        for msg in self.get_old_messages():
+            self.assertEqual(msg['flags'], [])
+
+    def test_new_message(self):
+        # Sending a new message results in unread UserMessages being created
+        self.login("hamlet@humbughq.com")
+        content = "Test message for unset read bit"
+        self.client.post("/json/send_message", {"type": "stream",
+                                                         "to": "Verona",
+                                                         "client": "test suite",
+                                                         "content": content,
+                                                         "subject": "Test subject"})
+        msgs = Message.objects.all().order_by("id")
+        last = msgs[len(msgs) - 1]
+        self.assertEqual(last.content, "Test message for unset read bit")
+        for um in UserMessage.objects.filter(message=last):
+            self.assertEqual(um.message.content, content)
+            if um.user_profile.user.email != "hamlet@humbughq.com":
+                self.assertFalse(um.flags.read)
+
+    def test_update_flags(self):
+        self.login("hamlet@humbughq.com")
+
+        result = self.client.post("/json/update_message_flags", {"messages": simplejson.dumps([1, 2]),
+                                                                 "op": "add",
+                                                                 "flag": "read"})
+        self.assert_json_success(result)
+
+        # Ensure we properly set the flags
+        for msg in self.get_old_messages():
+            if msg['id'] == 1:
+                self.assertEqual(msg['flags'], ['read'])
+            elif msg['id'] == 2:
+                self.assertEqual(msg['flags'], ['read'])
+
+        result = self.client.post("/json/update_message_flags", {"messages": simplejson.dumps([2]),
+                                                                 "op": "remove",
+                                                                 "flag": "read"})
+        self.assert_json_success(result)
+
+        # Ensure we properly remove just one flag
+        for msg in self.get_old_messages():
+            if msg['id'] == 1:
+                self.assertEqual(msg['flags'], ['read'])
+            elif msg['id'] == 2:
+                self.assertEqual(msg['flags'], [])
+
+    def test_update_all_flags(self):
+        self.login("hamlet@humbughq.com")
+
+        result = self.client.post("/json/update_message_flags", {"messages": simplejson.dumps([1, 2]),
+                                                                 "op": "add",
+                                                                 "flag": "read"})
+        self.assert_json_success(result)
+
+        result = self.client.post("/json/update_message_flags", {"messages": simplejson.dumps([]),
+                                                                 "op": "remove",
+                                                                 "flag": "read",
+                                                                 "all": simplejson.dumps(True)})
+        self.assert_json_success(result)
+
+        for msg in self.get_old_messages():
+            self.assertEqual(msg['flags'], [])
+
 class Runner(DjangoTestSuiteRunner):
     option_list = (
         optparse.make_option('--skip-generate',

zephyr/views.py

@@ -22,7 +22,7 @@ from zephyr.lib.actions import do_add_subscription, do_remove_subscription, \
     do_activate_user, add_default_subs, do_create_user, do_send_message, \
     log_subscription_property_change, internal_send_message, \
     create_stream_if_needed, gather_subscriptions, subscribed_to_stream, \
-    update_user_presence, set_stream_color, get_stream_colors
+    update_user_presence, set_stream_color, get_stream_colors, update_message_flags
 from zephyr.forms import RegistrationForm, HomepageForm, ToSForm, is_unique, \
     is_inactive, isnt_mit
 from django.views.decorators.csrf import csrf_exempt
@@ -600,8 +600,8 @@ def get_old_messages_backend(request, anchor = POST(converter=int),
     if stream is not None:
         stream = get_public_stream(request, stream, user_profile.realm)
         recipient = Recipient.objects.get(type_id=stream.id, type=Recipient.STREAM)
-        query = UserMessage.objects.select_related().filter(message__recipient=recipient,
+        query = UserMessage.objects.select_related('message').filter(message__recipient=recipient,
-                                                            user_profile=user_profile) \
+                                                                     user_profile=user_profile) \
                                                     .order_by('id')
     else:
         query = UserMessage.objects.select_related().filter(user_profile=user_profile) \
@@ -655,6 +655,16 @@ def get_profile_backend(request, user_profile):
 
     return json_success(result)
 
+@authenticated_json_post_view
+@has_request_variables
+def json_update_flags(request, user_profile, messages=POST('messages', converter=json_to_list),
+                                            operation=POST('op'),
+                                            flag=POST('flag'),
+                                            all=POST('all', converter=json_to_bool, default=False)):
+    update_message_flags(user_profile, operation, flag, messages, all)
+    return json_success({'result': 'success',
+                         'msg': ''})
+
 @authenticated_api_view
 def api_send_message(request, user_profile):
     return send_message_backend(request, user_profile, request._client)