Merge branch 'develop' into markjh/edu_frequency

5 files changed, 173 insertions, 91 deletions

diff --git a/synapse/handlers/account_data.py b/synapse/handlers/account_data.py
index 1d35d3b7dc..fe773bee9b 100644
--- a/synapse/handlers/account_data.py
+++ b/synapse/handlers/account_data.py
@@ -29,9 +29,10 @@ class AccountDataEventSource(object):
         last_stream_id = from_key
 
         current_stream_id = yield self.store.get_max_account_data_stream_id()
-        tags = yield self.store.get_updated_tags(user_id, last_stream_id)
 
         results = []
+        tags = yield self.store.get_updated_tags(user_id, last_stream_id)
+
         for room_id, room_tags in tags.items():
             results.append({
                 "type": "m.tag",
@@ -39,6 +40,24 @@ class AccountDataEventSource(object):
                 "room_id": room_id,
             })
 
+        account_data, room_account_data = (
+            yield self.store.get_updated_account_data_for_user(user_id, last_stream_id)
+        )
+
+        for account_data_type, content in account_data.items():
+            results.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
+        for room_id, account_data in room_account_data.items():
+            for account_data_type, content in account_data.items():
+                results.append({
+                    "type": account_data_type,
+                    "content": content,
+                    "room_id": room_id,
+                })
+
         defer.returnValue((results, current_stream_id))
 
     @defer.inlineCallbacks
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 64c57375f7..e959ce50be 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -359,6 +359,10 @@ class MessageHandler(BaseHandler):
 
         tags_by_room = yield self.store.get_tags_for_user(user_id)
 
+        account_data, account_data_by_room = (
+            yield self.store.get_account_data_for_user(user_id)
+        )
+
         public_room_ids = yield self.store.get_public_room_ids()
 
         limit = pagin_config.limit
@@ -436,14 +440,22 @@ class MessageHandler(BaseHandler):
                     for c in current_state.values()
                 ]
 
-                account_data = []
+                account_data_events = []
                 tags = tags_by_room.get(event.room_id)
                 if tags:
-                    account_data.append({
+                    account_data_events.append({
                         "type": "m.tag",
                         "content": {"tags": tags},
                     })
-                d["account_data"] = account_data
+
+                account_data = account_data_by_room.get(event.room_id, {})
+                for account_data_type, content in account_data.items():
+                    account_data_events.append({
+                        "type": account_data_type,
+                        "content": content,
+                    })
+
+                d["account_data"] = account_data_events
             except:
                 logger.exception("Failed to get snapshot")
 
@@ -456,9 +468,17 @@ class MessageHandler(BaseHandler):
                 consumeErrors=True
             ).addErrback(unwrapFirstError)
 
+        account_data_events = []
+        for account_data_type, content in account_data.items():
+            account_data_events.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
         ret = {
             "rooms": rooms_ret,
             "presence": presence,
+            "account_data": account_data_events,
             "receipts": receipt,
             "end": now_token.to_string(),
         }
@@ -498,14 +518,22 @@ class MessageHandler(BaseHandler):
                 user_id, room_id, pagin_config, membership, member_event_id, is_guest
             )
 
-        account_data = []
+        account_data_events = []
         tags = yield self.store.get_tags_for_room(user_id, room_id)
         if tags:
-            account_data.append({
+            account_data_events.append({
                 "type": "m.tag",
                 "content": {"tags": tags},
             })
-        result["account_data"] = account_data
+
+        account_data = yield self.store.get_account_data_for_room(user_id, room_id)
+        for account_data_type, content in account_data.items():
+            account_data_events.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
+        result["account_data"] = account_data_events
 
         defer.returnValue(result)
 
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index aca65096fc..e95e821c9a 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -467,7 +467,7 @@ class PresenceHandler(BaseHandler):
             )
 
     @defer.inlineCallbacks
-    def send_invite(self, observer_user, observed_user):
+    def send_presence_invite(self, observer_user, observed_user):
         """Request the presence of a local or remote user for a local user"""
         if not self.hs.is_mine(observer_user):
             raise SynapseError(400, "User is not hosted on this Home Server")
diff --git a/synapse/handlers/search.py b/synapse/handlers/search.py
index 6d2197339e..df6390cf05 100644
--- a/synapse/handlers/search.py
+++ b/synapse/handlers/search.py
@@ -131,6 +131,17 @@ class SearchHandler(BaseHandler):
         if batch_group == "room_id":
             room_ids.intersection_update({batch_group_key})
 
+        if not room_ids:
+            defer.returnValue({
+                "search_categories": {
+                    "room_events": {
+                        "results": {},
+                        "count": 0,
+                        "highlights": [],
+                    }
+                }
+            })
+
         rank_map = {}  # event_id -> rank of event
         allowed_events = []
         room_groups = {}  # Holds result of grouping by room, if applicable
@@ -178,85 +189,67 @@ class SearchHandler(BaseHandler):
                 s["results"].append(e.event_id)
 
         elif order_by == "recent":
-            # In this case we specifically loop through each room as the given
-            # limit applies to each room, rather than a global list.
-            # This is not necessarilly a good idea.
-            for room_id in room_ids:
-                room_events = []
-                if batch_group == "room_id" and batch_group_key == room_id:
-                    pagination_token = batch_token
-                else:
-                    pagination_token = None
-                i = 0
-
-                # We keep looping and we keep filtering until we reach the limit
-                # or we run out of things.
-                # But only go around 5 times since otherwise synapse will be sad.
-                while len(room_events) < search_filter.limit() and i < 5:
-                    i += 1
-                    search_result = yield self.store.search_room(
-                        room_id, search_term, keys, search_filter.limit() * 2,
-                        pagination_token=pagination_token,
-                    )
+            room_events = []
+            i = 0
+
+            pagination_token = batch_token
+
+            # We keep looping and we keep filtering until we reach the limit
+            # or we run out of things.
+            # But only go around 5 times since otherwise synapse will be sad.
+            while len(room_events) < search_filter.limit() and i < 5:
+                i += 1
+                search_result = yield self.store.search_rooms(
+                    room_ids, search_term, keys, search_filter.limit() * 2,
+                    pagination_token=pagination_token,
+                )
 
-                    if search_result["highlights"]:
-                        highlights.update(search_result["highlights"])
+                if search_result["highlights"]:
+                    highlights.update(search_result["highlights"])
 
-                    results = search_result["results"]
+                results = search_result["results"]
 
-                    results_map = {r["event"].event_id: r for r in results}
+                results_map = {r["event"].event_id: r for r in results}
 
-                    rank_map.update({r["event"].event_id: r["rank"] for r in results})
+                rank_map.update({r["event"].event_id: r["rank"] for r in results})
 
-                    filtered_events = search_filter.filter([
-                        r["event"] for r in results
-                    ])
+                filtered_events = search_filter.filter([
+                    r["event"] for r in results
+                ])
 
-                    events = yield self._filter_events_for_client(
-                        user.to_string(), filtered_events
-                    )
+                events = yield self._filter_events_for_client(
+                    user.to_string(), filtered_events
+                )
 
-                    room_events.extend(events)
-                    room_events = room_events[:search_filter.limit()]
+                room_events.extend(events)
+                room_events = room_events[:search_filter.limit()]
 
-                    if len(results) < search_filter.limit() * 2:
-                        pagination_token = None
-                        break
-                    else:
-                        pagination_token = results[-1]["pagination_token"]
-
-                if room_events:
-                    res = results_map[room_events[-1].event_id]
-                    pagination_token = res["pagination_token"]
-
-                    group = room_groups.setdefault(room_id, {})
-                    if pagination_token:
-                        next_batch = encode_base64("%s\n%s\n%s" % (
-                            "room_id", room_id, pagination_token
-                        ))
-                        group["next_batch"] = next_batch
-
-                        if batch_token:
-                            global_next_batch = next_batch
-
-                    group["results"] = [e.event_id for e in room_events]
-                    group["order"] = max(
-                        e.origin_server_ts/1000 for e in room_events
-                        if hasattr(e, "origin_server_ts")
-                    )
+                if len(results) < search_filter.limit() * 2:
+                    pagination_token = None
+                    break
+                else:
+                    pagination_token = results[-1]["pagination_token"]
 
-                allowed_events.extend(room_events)
+            for event in room_events:
+                group = room_groups.setdefault(event.room_id, {
+                    "results": [],
+                })
+                group["results"].append(event.event_id)
 
-            # Normalize the group orders
-            if room_groups:
-                if len(room_groups) > 1:
-                    mx = max(g["order"] for g in room_groups.values())
-                    mn = min(g["order"] for g in room_groups.values())
+            if room_events and len(room_events) >= search_filter.limit():
+                last_event_id = room_events[-1].event_id
+                pagination_token = results_map[last_event_id]["pagination_token"]
 
-                    for g in room_groups.values():
-                        g["order"] = (g["order"] - mn) * 1.0 / (mx - mn)
-                else:
-                    room_groups.values()[0]["order"] = 1
+                global_next_batch = encode_base64("%s\n%s\n%s" % (
+                    "all", "", pagination_token
+                ))
+
+                for room_id, group in room_groups.items():
+                    group["next_batch"] = encode_base64("%s\n%s\n%s" % (
+                        "room_id", room_id, pagination_token
+                    ))
+
+            allowed_events.extend(room_events)
 
         else:
             # We should never get here due to the guard earlier.
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 877328b29e..943ce368ef 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -100,6 +100,7 @@ class InvitedSyncResult(collections.namedtuple("InvitedSyncResult", [
 class SyncResult(collections.namedtuple("SyncResult", [
     "next_batch",  # Token for the next sync
     "presence",  # List of presence events for the user.
+    "account_data",  # List of account_data events for the user.
     "joined",  # JoinedSyncResult for each joined room.
     "invited",  # InvitedSyncResult for each invited room.
     "archived",  # ArchivedSyncResult for each archived room.
@@ -195,6 +196,12 @@ class SyncHandler(BaseHandler):
             )
         )
 
+        account_data, account_data_by_room = (
+            yield self.store.get_account_data_for_user(
+                sync_config.user.to_string()
+            )
+        )
+
         tags_by_room = yield self.store.get_tags_for_user(
             sync_config.user.to_string()
         )
@@ -211,6 +218,7 @@ class SyncHandler(BaseHandler):
                     timeline_since_token=timeline_since_token,
                     ephemeral_by_room=ephemeral_by_room,
                     tags_by_room=tags_by_room,
+                    account_data_by_room=account_data_by_room,
                 )
                 joined.append(room_sync)
             elif event.membership == Membership.INVITE:
@@ -230,11 +238,13 @@ class SyncHandler(BaseHandler):
                     leave_token=leave_token,
                     timeline_since_token=timeline_since_token,
                     tags_by_room=tags_by_room,
+                    account_data_by_room=account_data_by_room,
                 )
                 archived.append(room_sync)
 
         defer.returnValue(SyncResult(
             presence=presence,
+            account_data=self.account_data_for_user(account_data),
             joined=joined,
             invited=invited,
             archived=archived,
@@ -244,7 +254,8 @@ class SyncHandler(BaseHandler):
     @defer.inlineCallbacks
     def full_state_sync_for_joined_room(self, room_id, sync_config,
                                         now_token, timeline_since_token,
-                                        ephemeral_by_room, tags_by_room):
+                                        ephemeral_by_room, tags_by_room,
+                                        account_data_by_room):
         """Sync a room for a client which is starting without any state
         Returns:
             A Deferred JoinedSyncResult.
@@ -262,19 +273,38 @@ class SyncHandler(BaseHandler):
             state=current_state,
             ephemeral=ephemeral_by_room.get(room_id, []),
             account_data=self.account_data_for_room(
-                room_id, tags_by_room
+                room_id, tags_by_room, account_data_by_room
             ),
         ))
 
-    def account_data_for_room(self, room_id, tags_by_room):
-        account_data = []
+    def account_data_for_user(self, account_data):
+        account_data_events = []
+
+        for account_data_type, content in account_data.items():
+            account_data_events.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
+        return account_data_events
+
+    def account_data_for_room(self, room_id, tags_by_room, account_data_by_room):
+        account_data_events = []
         tags = tags_by_room.get(room_id)
         if tags is not None:
-            account_data.append({
+            account_data_events.append({
                 "type": "m.tag",
                 "content": {"tags": tags},
             })
-        return account_data
+
+        account_data = account_data_by_room.get(room_id, {})
+        for account_data_type, content in account_data.items():
+            account_data_events.append({
+                "type": account_data_type,
+                "content": content,
+            })
+
+        return account_data_events
 
     @defer.inlineCallbacks
     def ephemeral_by_room(self, sync_config, now_token, since_token=None):
@@ -341,7 +371,8 @@ class SyncHandler(BaseHandler):
     @defer.inlineCallbacks
     def full_state_sync_for_archived_room(self, room_id, sync_config,
                                           leave_event_id, leave_token,
-                                          timeline_since_token, tags_by_room):
+                                          timeline_since_token, tags_by_room,
+                                          account_data_by_room):
         """Sync a room for a client which is starting without any state
         Returns:
             A Deferred JoinedSyncResult.
@@ -358,7 +389,7 @@ class SyncHandler(BaseHandler):
             timeline=batch,
             state=leave_state,
             account_data=self.account_data_for_room(
-                room_id, tags_by_room
+                room_id, tags_by_room, account_data_by_room
             ),
         ))
 
@@ -415,6 +446,13 @@ class SyncHandler(BaseHandler):
             since_token.account_data_key,
         )
 
+        account_data, account_data_by_room = (
+            yield self.store.get_updated_account_data_for_user(
+                sync_config.user.to_string(),
+                since_token.account_data_key,
+            )
+        )
+
         joined = []
         archived = []
         if len(room_events) <= timeline_limit:
@@ -469,7 +507,7 @@ class SyncHandler(BaseHandler):
                     state=state,
                     ephemeral=ephemeral_by_room.get(room_id, []),
                     account_data=self.account_data_for_room(
-                        room_id, tags_by_room
+                        room_id, tags_by_room, account_data_by_room
                     ),
                 )
                 logger.debug("Result for room %s: %r", room_id, room_sync)
@@ -492,14 +530,15 @@ class SyncHandler(BaseHandler):
             for room_id in joined_room_ids:
                 room_sync = yield self.incremental_sync_with_gap_for_room(
                     room_id, sync_config, since_token, now_token,
-                    ephemeral_by_room, tags_by_room
+                    ephemeral_by_room, tags_by_room, account_data_by_room
                 )
                 if room_sync:
                     joined.append(room_sync)
 
         for leave_event in leave_events:
             room_sync = yield self.incremental_sync_for_archived_room(
-                sync_config, leave_event, since_token, tags_by_room
+                sync_config, leave_event, since_token, tags_by_room,
+                account_data_by_room
             )
             archived.append(room_sync)
 
@@ -510,6 +549,7 @@ class SyncHandler(BaseHandler):
 
         defer.returnValue(SyncResult(
             presence=presence,
+            account_data=self.account_data_for_user(account_data),
             joined=joined,
             invited=invited,
             archived=archived,
@@ -566,7 +606,8 @@ class SyncHandler(BaseHandler):
     @defer.inlineCallbacks
     def incremental_sync_with_gap_for_room(self, room_id, sync_config,
                                            since_token, now_token,
-                                           ephemeral_by_room, tags_by_room):
+                                           ephemeral_by_room, tags_by_room,
+                                           account_data_by_room):
         """ Get the incremental delta needed to bring the client up to date for
         the room. Gives the client the most recent events and the changes to
         state.
@@ -606,7 +647,7 @@ class SyncHandler(BaseHandler):
             state=state,
             ephemeral=ephemeral_by_room.get(room_id, []),
             account_data=self.account_data_for_room(
-                room_id, tags_by_room
+                room_id, tags_by_room, account_data_by_room
             ),
         )
 
@@ -616,7 +657,8 @@ class SyncHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def incremental_sync_for_archived_room(self, sync_config, leave_event,
-                                           since_token, tags_by_room):
+                                           since_token, tags_by_room,
+                                           account_data_by_room):
         """ Get the incremental delta needed to bring the client up to date for
         the archived room.
         Returns:
@@ -654,7 +696,7 @@ class SyncHandler(BaseHandler):
             timeline=batch,
             state=state_events_delta,
             account_data=self.account_data_for_room(
-                leave_event.room_id, tags_by_room
+                leave_event.room_id, tags_by_room, account_data_by_room
             ),
         )