diff options
| -rw-r--r-- | contrib/graph/graph3.py | 151 | ||||
| -rw-r--r-- | synapse/handlers/_base.py | 14 | ||||
| -rw-r--r-- | synapse/handlers/events.py | 2 | ||||
| -rw-r--r-- | synapse/handlers/message.py | 93 | ||||
| -rw-r--r-- | synapse/handlers/register.py | 5 | ||||
| -rw-r--r-- | synapse/handlers/room.py | 2 | ||||
| -rw-r--r-- | synapse/handlers/sync.py | 4 | ||||
| -rw-r--r-- | synapse/notifier.py | 71 | ||||
| -rw-r--r-- | synapse/push/__init__.py | 73 | ||||
| -rw-r--r-- | synapse/push/baserules.py | 36 | ||||
| -rw-r--r-- | synapse/push/bulk_push_rule_evaluator.py | 3 | ||||
| -rw-r--r-- | synapse/push/httppusher.py | 14 | ||||
| -rw-r--r-- | synapse/push/pusherpool.py | 15 | ||||
| -rw-r--r-- | synapse/rest/client/v1/events.py | 1 | ||||
| -rw-r--r-- | synapse/rest/client/v1/room.py | 6 | ||||
| -rw-r--r-- | synapse/rest/client/v2_alpha/account.py | 5 | ||||
| -rw-r--r-- | synapse/rest/client/v2_alpha/sync.py | 7 | ||||
| -rw-r--r-- | synapse/storage/receipts.py | 54 | ||||
| -rw-r--r-- | synapse/storage/schema/delta/28/receipts_user_id_index.sql | 18 |
19 files changed, 400 insertions, 174 deletions
diff --git a/contrib/graph/graph3.py b/contrib/graph/graph3.py new file mode 100644 index 0000000000..88d92c89d7 --- /dev/null +++ b/contrib/graph/graph3.py @@ -0,0 +1,151 @@ +# Copyright 2016 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +import pydot +import cgi +import simplejson as json +import datetime +import argparse + +from synapse.events import FrozenEvent +from synapse.util.frozenutils import unfreeze + + +def make_graph(file_name, room_id, file_prefix, limit): + print "Reading lines" + with open(file_name) as f: + lines = f.readlines() + + print "Read lines" + + events = [FrozenEvent(json.loads(line)) for line in lines] + + print "Loaded events." + + events.sort(key=lambda e: e.depth) + + print "Sorted events" + + if limit: + events = events[-int(limit):] + + node_map = {} + + graph = pydot.Dot(graph_name="Test") + + for event in events: + t = datetime.datetime.fromtimestamp( + float(event.origin_server_ts) / 1000 + ).strftime('%Y-%m-%d %H:%M:%S,%f') + + content = json.dumps(unfreeze(event.get_dict()["content"]), indent=4) + content = content.replace("\n", "<br/>\n") + + print content + content = [] + for key, value in unfreeze(event.get_dict()["content"]).items(): + if value is None: + value = "<null>" + elif isinstance(value, basestring): + pass + else: + value = json.dumps(value) + + content.append( + "<b>%s</b>: %s," % ( + cgi.escape(key, quote=True).encode("ascii", 'xmlcharrefreplace'), + cgi.escape(value, quote=True).encode("ascii", 'xmlcharrefreplace'), + ) + ) + + content = "<br/>\n".join(content) + + print content + + label = ( + "<" + "<b>%(name)s </b><br/>" + "Type: <b>%(type)s </b><br/>" + "State key: <b>%(state_key)s </b><br/>" + "Content: <b>%(content)s </b><br/>" + "Time: <b>%(time)s </b><br/>" + "Depth: <b>%(depth)s </b><br/>" + ">" + ) % { + "name": event.event_id, + "type": event.type, + "state_key": event.get("state_key", None), + "content": content, + "time": t, + "depth": event.depth, + } + + node = pydot.Node( + name=event.event_id, + label=label, + ) + + node_map[event.event_id] = node + graph.add_node(node) + + print "Created Nodes" + + for event in events: + for prev_id, _ in event.prev_events: + try: + end_node = node_map[prev_id] + except: + end_node = pydot.Node( + name=prev_id, + label="<<b>%s</b>>" % (prev_id,), + ) + + node_map[prev_id] = end_node + graph.add_node(end_node) + + edge = pydot.Edge(node_map[event.event_id], end_node) + graph.add_edge(edge) + + print "Created edges" + + graph.write('%s.dot' % file_prefix, format='raw', prog='dot') + + print "Created Dot" + + graph.write_svg("%s.svg" % file_prefix, prog='dot') + + print "Created svg" + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="Generate a PDU graph for a given room by reading " + "from a file with line deliminated events. \n" + "Requires pydot." + ) + parser.add_argument( + "-p", "--prefix", dest="prefix", + help="String to prefix output files with", + default="graph_output" + ) + parser.add_argument( + "-l", "--limit", + help="Only retrieve the last N events.", + ) + parser.add_argument('event_file') + parser.add_argument('room') + + args = parser.parse_args() + + make_graph(args.event_file, args.room, args.prefix, args.limit) diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 46abb8ec51..744a9ee507 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -84,7 +84,7 @@ class BaseHandler(object): row["event_id"] for rows in forgotten for row in rows ) - def allowed(event, user_id, is_guest): + def allowed(event, user_id, is_peeking): state = event_id_to_state[event.event_id] visibility_event = state.get((EventTypes.RoomHistoryVisibility, ""), None) @@ -96,7 +96,7 @@ class BaseHandler(object): if visibility == "world_readable": return True - if is_guest: + if is_peeking: return False membership_event = state.get((EventTypes.Member, user_id), None) @@ -112,7 +112,7 @@ class BaseHandler(object): return True if event.type == EventTypes.RoomHistoryVisibility: - return not is_guest + return not is_peeking if visibility == "shared": return True @@ -127,15 +127,15 @@ class BaseHandler(object): user_id: [ event for event in events - if allowed(event, user_id, is_guest) + if allowed(event, user_id, is_peeking) ] - for user_id, is_guest in user_tuples + for user_id, is_peeking in user_tuples }) @defer.inlineCallbacks - def _filter_events_for_client(self, user_id, events, is_guest=False): + def _filter_events_for_client(self, user_id, events, is_peeking=False): # Assumes that user has at some point joined the room if not is_guest. - res = yield self._filter_events_for_clients([(user_id, is_guest)], events) + res = yield self._filter_events_for_clients([(user_id, is_peeking)], events) defer.returnValue(res.get(user_id, [])) def ratelimit(self, user_id): diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index c73eec2b91..aca4b6754e 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -135,7 +135,7 @@ class EventStreamHandler(BaseHandler): events, tokens = yield self.notifier.get_events_for( auth_user, pagin_config, timeout, only_room_events=only_room_events, - is_guest=is_guest, guest_room_id=room_id + is_guest=is_guest, explicit_room_id=room_id ) time_now = self.clock.time_msec() diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 4c7bf2bef3..ff800f8af1 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -78,21 +78,20 @@ class MessageHandler(BaseHandler): defer.returnValue(None) @defer.inlineCallbacks - def get_messages(self, user_id=None, room_id=None, pagin_config=None, - as_client_event=True, is_guest=False): + def get_messages(self, requester, room_id=None, pagin_config=None, + as_client_event=True): """Get messages in a room. Args: - user_id (str): The user requesting messages. + requester (Requester): The user requesting messages. room_id (str): The room they want messages from. pagin_config (synapse.api.streams.PaginationConfig): The pagination config rules to apply, if any. as_client_event (bool): True to get events in client-server format. - is_guest (bool): Whether the requesting user is a guest (as opposed - to a fully registered user). Returns: dict: Pagination API results """ + user_id = requester.user.to_string() data_source = self.hs.get_event_sources().sources["room"] if pagin_config.from_token: @@ -115,36 +114,33 @@ class MessageHandler(BaseHandler): source_config = pagin_config.get_source_config("room") - if not is_guest: - member_event = yield self.auth.check_user_was_in_room(room_id, user_id) - if member_event.membership == Membership.LEAVE: - # If they have left the room then clamp the token to be before - # they left the room. - # If they're a guest, we'll just 403 them if they're asking for - # events they can't see. - leave_token = yield self.store.get_topological_token_for_event( - member_event.event_id - ) - leave_token = RoomStreamToken.parse(leave_token) - if leave_token.topological < room_token.topological: - source_config.from_key = str(leave_token) - - if source_config.direction == "f": - if source_config.to_key is None: + membership, member_event_id = yield self._check_in_room_or_world_readable( + room_id, user_id + ) + if membership == Membership.LEAVE: + # If they have left the room then clamp the token to be before + # they left the room. + leave_token = yield self.store.get_topological_token_for_event( + member_event_id + ) + leave_token = RoomStreamToken.parse(leave_token) + if leave_token.topological < room_token.topological: + source_config.from_key = str(leave_token) + + if source_config.direction == "f": + if source_config.to_key is None: + source_config.to_key = str(leave_token) + else: + to_token = RoomStreamToken.parse(source_config.to_key) + if leave_token.topological < to_token.topological: source_config.to_key = str(leave_token) - else: - to_token = RoomStreamToken.parse(source_config.to_key) - if leave_token.topological < to_token.topological: - source_config.to_key = str(leave_token) yield self.hs.get_handlers().federation_handler.maybe_backfill( room_id, room_token.topological ) - user = UserID.from_string(user_id) - events, next_key = yield data_source.get_pagination_rows( - user, source_config, room_id + requester.user, source_config, room_id ) next_token = pagin_config.from_token.copy_and_replace( @@ -158,7 +154,11 @@ class MessageHandler(BaseHandler): "end": next_token.to_string(), }) - events = yield self._filter_events_for_client(user_id, events, is_guest=is_guest) + events = yield self._filter_events_for_client( + user_id, + events, + is_peeking=(member_event_id is None), + ) time_now = self.clock.time_msec() @@ -289,7 +289,7 @@ class MessageHandler(BaseHandler): SynapseError if something went wrong. """ membership, membership_event_id = yield self._check_in_room_or_world_readable( - room_id, user_id, is_guest + room_id, user_id ) if membership == Membership.JOIN: @@ -306,7 +306,7 @@ class MessageHandler(BaseHandler): defer.returnValue(data) @defer.inlineCallbacks - def _check_in_room_or_world_readable(self, room_id, user_id, is_guest): + def _check_in_room_or_world_readable(self, room_id, user_id): try: # check_user_was_in_room will return the most recent membership # event for the user if: @@ -316,7 +316,7 @@ class MessageHandler(BaseHandler): member_event = yield self.auth.check_user_was_in_room(room_id, user_id) defer.returnValue((member_event.membership, member_event.event_id)) return - except AuthError, auth_error: + except AuthError: visibility = yield self.state_handler.get_current_state( room_id, EventTypes.RoomHistoryVisibility, "" ) @@ -326,8 +326,6 @@ class MessageHandler(BaseHandler): ): defer.returnValue((Membership.JOIN, None)) return - if not is_guest: - raise auth_error raise AuthError( 403, "Guest access not allowed", errcode=Codes.GUEST_ACCESS_FORBIDDEN ) @@ -345,7 +343,7 @@ class MessageHandler(BaseHandler): A list of dicts representing state events. [{}, {}, {}] """ membership, membership_event_id = yield self._check_in_room_or_world_readable( - room_id, user_id, is_guest + room_id, user_id ) if membership == Membership.JOIN: @@ -556,13 +554,13 @@ class MessageHandler(BaseHandler): defer.returnValue(ret) @defer.inlineCallbacks - def room_initial_sync(self, user_id, room_id, pagin_config=None, is_guest=False): + def room_initial_sync(self, requester, room_id, pagin_config=None): """Capture the a snapshot of a room. If user is currently a member of the room this will be what is currently in the room. If the user left the room this will be what was in the room when they left. Args: - user_id(str): The user to get a snapshot for. + requester(Requester): The user to get a snapshot for. room_id(str): The room to get a snapshot of. pagin_config(synapse.streams.config.PaginationConfig): The pagination config used to determine how many messages to @@ -573,19 +571,20 @@ class MessageHandler(BaseHandler): A JSON serialisable dict with the snapshot of the room. """ + user_id = requester.user.to_string() + membership, member_event_id = yield self._check_in_room_or_world_readable( - room_id, - user_id, - is_guest + room_id, user_id, ) + is_peeking = member_event_id is None if membership == Membership.JOIN: result = yield self._room_initial_sync_joined( - user_id, room_id, pagin_config, membership, is_guest + user_id, room_id, pagin_config, membership, is_peeking ) elif membership == Membership.LEAVE: result = yield self._room_initial_sync_parted( - user_id, room_id, pagin_config, membership, member_event_id, is_guest + user_id, room_id, pagin_config, membership, member_event_id, is_peeking ) account_data_events = [] @@ -609,7 +608,7 @@ class MessageHandler(BaseHandler): @defer.inlineCallbacks def _room_initial_sync_parted(self, user_id, room_id, pagin_config, - membership, member_event_id, is_guest): + membership, member_event_id, is_peeking): room_state = yield self.store.get_state_for_events( [member_event_id], None ) @@ -631,7 +630,7 @@ class MessageHandler(BaseHandler): ) messages = yield self._filter_events_for_client( - user_id, messages, is_guest=is_guest + user_id, messages, is_peeking=is_peeking ) start_token = StreamToken(token[0], 0, 0, 0, 0) @@ -654,7 +653,7 @@ class MessageHandler(BaseHandler): @defer.inlineCallbacks def _room_initial_sync_joined(self, user_id, room_id, pagin_config, - membership, is_guest): + membership, is_peeking): current_state = yield self.state.get_current_state( room_id=room_id, ) @@ -718,7 +717,7 @@ class MessageHandler(BaseHandler): ).addErrback(unwrapFirstError) messages = yield self._filter_events_for_client( - user_id, messages, is_guest=is_guest, + user_id, messages, is_peeking=is_peeking, ) start_token = now_token.copy_and_replace("room_key", token[0]) @@ -737,7 +736,7 @@ class MessageHandler(BaseHandler): "presence": presence, "receipts": receipts, } - if not is_guest: + if not is_peeking: ret["membership"] = membership defer.returnValue(ret) diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py index 8e601b052b..1e99c1303c 100644 --- a/synapse/handlers/register.py +++ b/synapse/handlers/register.py @@ -49,11 +49,10 @@ class RegistrationHandler(BaseHandler): def check_username(self, localpart, guest_access_token=None): yield run_on_reactor() - if urllib.quote(localpart) != localpart: + if urllib.quote(localpart.encode('utf-8')) != localpart: raise SynapseError( 400, - "User ID must only contain characters which do not" - " require URL encoding.", + "User ID can only contain characters a-z, 0-9, or '-./'", Codes.INVALID_USERNAME ) diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index a1baf9d200..58e2d25f97 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -936,7 +936,7 @@ class RoomContextHandler(BaseHandler): return self._filter_events_for_client( user.to_string(), events, - is_guest=is_guest) + is_peeking=is_guest) event = yield self.store.get_event(event_id, get_prev_content=True, allow_none=True) diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index aca200c1e7..53e1eb0508 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -150,7 +150,7 @@ class SyncHandler(BaseHandler): return self.current_sync_for_user(sync_config, since_token) result = yield self.notifier.wait_for_events( - sync_config.user, timeout, current_sync_callback, + sync_config.user.to_string(), timeout, current_sync_callback, from_token=since_token ) defer.returnValue(result) @@ -640,7 +640,7 @@ class SyncHandler(BaseHandler): loaded_recents = yield self._filter_events_for_client( sync_config.user.to_string(), loaded_recents, - is_guest=sync_config.is_guest, + is_peeking=sync_config.is_guest, ) loaded_recents.extend(recents) recents = loaded_recents diff --git a/synapse/notifier.py b/synapse/notifier.py index 0a5653b8d5..3285487551 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -63,9 +63,9 @@ class _NotifierUserStream(object): so that it can remove itself from the indexes in the Notifier class. """ - def __init__(self, user, rooms, current_token, time_now_ms, + def __init__(self, user_id, rooms, current_token, time_now_ms, appservice=None): - self.user = str(user) + self.user_id = user_id self.appservice = appservice self.rooms = set(rooms) self.current_token = current_token @@ -98,7 +98,7 @@ class _NotifierUserStream(object): lst = notifier.room_to_user_streams.get(room, set()) lst.discard(self) - notifier.user_to_user_stream.pop(self.user) + notifier.user_to_user_stream.pop(self.user_id) if self.appservice: notifier.appservice_to_user_streams.get( @@ -271,21 +271,20 @@ class Notifier(object): logger.exception("Failed to notify listener") @defer.inlineCallbacks - def wait_for_events(self, user, timeout, callback, room_ids=None, + def wait_for_events(self, user_id, timeout, callback, room_ids=None, from_token=StreamToken("s0", "0", "0", "0", "0")): """Wait until the callback returns a non empty response or the timeout fires. """ - user = str(user) - user_stream = self.user_to_user_stream.get(user) + user_stream = self.user_to_user_stream.get(user_id) if user_stream is None: - appservice = yield self.store.get_app_service_by_user_id(user) + appservice = yield self.store.get_app_service_by_user_id(user_id) current_token = yield self.event_sources.get_current_token() if room_ids is None: - rooms = yield self.store.get_rooms_for_user(user) + rooms = yield self.store.get_rooms_for_user(user_id) room_ids = [room.room_id for room in rooms] user_stream = _NotifierUserStream( - user=user, + user_id=user_id, rooms=room_ids, appservice=appservice, current_token=current_token, @@ -333,12 +332,17 @@ class Notifier(object): @defer.inlineCallbacks def get_events_for(self, user, pagination_config, timeout, only_room_events=False, - is_guest=False, guest_room_id=None): + is_guest=False, explicit_room_id=None): """ For the given user and rooms, return any new events for them. If there are no new events wait for up to `timeout` milliseconds for any new events to happen before returning. If `only_room_events` is `True` only room events will be returned. + + If explicit_room_id is not set, the user's joined rooms will be polled + for events. + If explicit_room_id is set, that room will be polled for events only if + it is world readable or the user has joined the room. """ from_token = pagination_config.from_token if not from_token: @@ -346,15 +350,8 @@ class Notifier(object): limit = pagination_config.limit - room_ids = [] - if is_guest: - if guest_room_id: - if not (yield self._is_world_readable(guest_room_id)): - raise AuthError(403, "Guest access not allowed") - room_ids = [guest_room_id] - else: - rooms = yield self.store.get_rooms_for_user(user.to_string()) - room_ids = [room.room_id for room in rooms] + room_ids, is_joined = yield self._get_room_ids(user, explicit_room_id) + is_peeking = not is_joined @defer.inlineCallbacks def check_for_updates(before_token, after_token): @@ -376,7 +373,7 @@ class Notifier(object): user=user, from_key=getattr(from_token, keyname), limit=limit, - is_guest=is_guest, + is_guest=is_peeking, room_ids=room_ids, ) @@ -385,7 +382,7 @@ class Notifier(object): new_events = yield room_member_handler._filter_events_for_client( user.to_string(), new_events, - is_guest=is_guest, + is_peeking=is_peeking, ) events.extend(new_events) @@ -396,8 +393,24 @@ class Notifier(object): else: defer.returnValue(None) + user_id_for_stream = user.to_string() + if is_peeking: + # Internally, the notifier keeps an event stream per user_id. + # This is used by both /sync and /events. + # We want /events to be used for peeking independently of /sync, + # without polluting its contents. So we invent an illegal user ID + # (which thus cannot clash with any real users) for keying peeking + # over /events. + # + # I am sorry for what I have done. + user_id_for_stream = "_PEEKING_" + user_id_for_stream + result = yield self.wait_for_events( - user, timeout, check_for_updates, room_ids=room_ids, from_token=from_token + user_id_for_stream, + timeout, + check_for_updates, + room_ids=room_ids, + from_token=from_token, ) if result is None: @@ -406,6 +419,18 @@ class Notifier(object): defer.returnValue(result) @defer.inlineCallbacks + def _get_room_ids(self, user, explicit_room_id): + joined_rooms = yield self.store.get_rooms_for_user(user.to_string()) + joined_room_ids = map(lambda r: r.room_id, joined_rooms) + if explicit_room_id: + if explicit_room_id in joined_room_ids: + defer.returnValue(([explicit_room_id], True)) + if (yield self._is_world_readable(explicit_room_id)): + defer.returnValue(([explicit_room_id], False)) + raise AuthError(403, "Non-joined access not allowed") + defer.returnValue((joined_room_ids, True)) + + @defer.inlineCallbacks def _is_world_readable(self, room_id): state = yield self.hs.get_state_handler().get_current_state( room_id, @@ -432,7 +457,7 @@ class Notifier(object): @log_function def _register_with_keys(self, user_stream): - self.user_to_user_stream[user_stream.user] = user_stream + self.user_to_user_stream[user_stream.user_id] = user_stream for room in user_stream.rooms: s = self.room_to_user_streams.setdefault(room, set()) diff --git a/synapse/push/__init__.py b/synapse/push/__init__.py index a5dc84160c..9a4af2b3ca 100644 --- a/synapse/push/__init__.py +++ b/synapse/push/__init__.py @@ -17,6 +17,7 @@ from twisted.internet import defer from synapse.streams.config import PaginationConfig from synapse.types import StreamToken +from synapse.api.constants import Membership import synapse.util.async import push_rule_evaluator as push_rule_evaluator @@ -55,6 +56,7 @@ class Pusher(object): self.backoff_delay = Pusher.INITIAL_BACKOFF self.failing_since = failing_since self.alive = True + self.badge = None # The last value of last_active_time that we saw self.last_last_active_time = 0 @@ -92,8 +94,7 @@ class Pusher(object): # we fail to dispatch the push) config = PaginationConfig(from_token=None, limit='1') chunk = yield self.evStreamHandler.get_stream( - self.user_id, config, timeout=0, affect_presence=False, - only_room_events=True + self.user_id, config, timeout=0, affect_presence=False ) self.last_token = chunk['end'] self.store.update_pusher_last_token( @@ -124,9 +125,11 @@ class Pusher(object): from_tok = StreamToken.from_string(self.last_token) config = PaginationConfig(from_token=from_tok, limit='1') timeout = (300 + random.randint(-60, 60)) * 1000 + # note that we need to get read receipts down the stream as we need to + # wake up when one arrives. we don't need to explicitly look for + # them though. chunk = yield self.evStreamHandler.get_stream( - self.user_id, config, timeout=timeout, affect_presence=False, - only_room_events=True + self.user_id, config, timeout=timeout, affect_presence=False ) # limiting to 1 may get 1 event plus 1 presence event, so @@ -135,10 +138,10 @@ class Pusher(object): for c in chunk['chunk']: if 'event_id' in c: # Hmmm... single_event = c - break + if not single_event: + yield self.update_badge() self.last_token = chunk['end'] - logger.debug("Event stream timeout for pushkey %s", self.pushkey) yield self.store.update_pusher_last_token( self.app_id, self.pushkey, @@ -161,7 +164,8 @@ class Pusher(object): tweaks = rule_evaluator.tweaks_for_actions(actions) if 'notify' in actions: - rejected = yield self.dispatch_push(single_event, tweaks) + self.badge = yield self._get_badge_count() + rejected = yield self.dispatch_push(single_event, tweaks, self.badge) self.has_unread = True if isinstance(rejected, list) or isinstance(rejected, tuple): processed = True @@ -181,7 +185,6 @@ class Pusher(object): yield self.hs.get_pusherpool().remove_pusher( self.app_id, pk, self.user_id ) - else: processed = True if not self.alive: @@ -254,7 +257,7 @@ class Pusher(object): def stop(self): self.alive = False - def dispatch_push(self, p, tweaks): + def dispatch_push(self, p, tweaks, badge): """ Overridden by implementing classes to actually deliver the notification Args: @@ -266,23 +269,47 @@ class Pusher(object): """ pass - def reset_badge_count(self): - pass + @defer.inlineCallbacks + def update_badge(self): + new_badge = yield self._get_badge_count() + if self.badge != new_badge: + self.badge = new_badge + yield self.send_badge(self.badge) - def presence_changed(self, state): + def send_badge(self, badge): """ - We clear badge counts whenever a user's last_active time is bumped - This is by no means perfect but I think it's the best we can do - without read receipts. + Overridden by implementing classes to send an updated badge count """ - if 'last_active' in state.state: - last_active = state.state['last_active'] - if last_active > self.last_last_active_time: - self.last_last_active_time = last_active - if self.has_unread: - logger.info("Resetting badge count for %s", self.user_id) - self.reset_badge_count() - self.has_unread = False + pass + + @defer.inlineCallbacks + def _get_badge_count(self): + room_list = yield self.store.get_rooms_for_user_where_membership_is( + user_id=self.user_id, + membership_list=(Membership.INVITE, Membership.JOIN) + ) + + my_receipts_by_room = yield self.store.get_receipts_for_user( + self.user_id, + "m.read", + ) + + badge = 0 + + for r in room_list: + if r.membership == Membership.INVITE: + badge += 1 + else: + if r.room_id in my_receipts_by_room: + last_unread_event_id = my_receipts_by_room[r.room_id] + + notifs = yield ( + self.store.get_unread_event_push_actions_by_room_for_user( + r.room_id, self.user_id, last_unread_event_id + ) + ) + badge += len(notifs) + defer.returnValue(badge) class PusherConfigException(Exception): diff --git a/synapse/push/baserules.py b/synapse/push/baserules.py index 3b526c4e33..186281dfa3 100644 --- a/synapse/push/baserules.py +++ b/synapse/push/baserules.py @@ -218,23 +218,25 @@ BASE_APPEND_UNDERRIDE_RULES = [ } ] }, - { - 'rule_id': 'global/underride/.m.rule.member_event', - 'conditions': [ - { - 'kind': 'event_match', - 'key': 'type', - 'pattern': 'm.room.member', - '_id': '_member', - } - ], - 'actions': [ - 'notify', { - 'set_tweak': 'highlight', - 'value': False - } - ] - }, + # This is too simple: https://matrix.org/jira/browse/SYN-607 + # Removing for now + # { + # 'rule_id': 'global/underride/.m.rule.member_event', + # 'conditions': [ + # { + # 'kind': 'event_match', + # 'key': 'type', + # 'pattern': 'm.room.member', + # '_id': '_member', + # } + # ], + # 'actions': [ + # 'notify', { + # 'set_tweak': 'highlight', + # 'value': False + # } + # ] + # }, { 'rule_id': 'global/underride/.m.rule.message', 'conditions': [ diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py index 1000ae6301..b91c165e2b 100644 --- a/synapse/push/bulk_push_rule_evaluator.py +++ b/synapse/push/bulk_push_rule_evaluator.py @@ -107,6 +107,9 @@ class BulkPushRuleEvaluator: if len(filtered) == 0: continue + if filtered[0].sender == uid: + continue + for rule in rules: if 'enabled' in rule and not rule['enabled']: continue diff --git a/synapse/push/httppusher.py b/synapse/push/httppusher.py index 28f1fab0e4..cdc4494928 100644 --- a/synapse/push/httppusher.py +++ b/synapse/push/httppusher.py @@ -51,7 +51,7 @@ class HttpPusher(Pusher): del self.data_minus_url['url'] @defer.inlineCallbacks - def _build_notification_dict(self, event, tweaks): + def _build_notification_dict(self, event, tweaks, badge): # we probably do not want to push for every presence update # (we may want to be able to set up notifications when specific # people sign in, but we'd want to only deliver the pertinent ones) @@ -71,7 +71,7 @@ class HttpPusher(Pusher): 'counts': { # -- we don't mark messages as read yet so # we have no way of knowing # Just set the badge to 1 until we have read receipts - 'unread': 1, + 'unread': badge, # 'missed_calls': 2 }, 'devices': [ @@ -101,8 +101,8 @@ class HttpPusher(Pusher): defer.returnValue(d) @defer.inlineCallbacks - def dispatch_push(self, event, tweaks): - notification_dict = yield self._build_notification_dict(event, tweaks) + def dispatch_push(self, event, tweaks, badge): + notification_dict = yield self._build_notification_dict(event, tweaks, badge) if not notification_dict: defer.returnValue([]) try: @@ -116,15 +116,15 @@ class HttpPusher(Pusher): defer.returnValue(rejected) @defer.inlineCallbacks - def reset_badge_count(self): + def send_badge(self, badge): + logger.info("Sending updated badge count %d to %r", badge, self.user_id) d = { 'notification': { 'id': '', 'type': None, 'sender': '', 'counts': { - 'unread': 0, - 'missed_calls': 0 + 'unread': badge }, 'devices': [ { diff --git a/synapse/push/pusherpool.py b/synapse/push/pusherpool.py index 12c4af14bd..d1b7c0802f 100644 --- a/synapse/push/pusherpool.py +++ b/synapse/push/pusherpool.py @@ -31,21 +31,6 @@ class PusherPool: self.pushers = {} self.last_pusher_started = -1 - distributor = self.hs.get_distributor() - distributor.observe( - "user_presence_changed", self.user_presence_changed - ) - - @defer.inlineCallbacks - def user_presence_changed(self, user, state): - user_id = user.to_string() - - # until we have read receipts, pushers use this to reset a user's - # badge counters to zero - for p in self.pushers.values(): - if p.user_id == user_id: - yield p.presence_changed(state) - @defer.inlineCallbacks def start(self): pushers = yield self.store.get_all_pushers() diff --git a/synapse/rest/client/v1/events.py b/synapse/rest/client/v1/events.py index e89118b37d..d1afa0f0d5 100644 --- a/synapse/rest/client/v1/events.py +++ b/synapse/rest/client/v1/events.py @@ -43,6 +43,7 @@ class EventStreamRestServlet(ClientV1RestServlet): if is_guest: if "room_id" not in request.args: raise SynapseError(400, "Guest users must specify room_id param") + if "room_id" in request.args: room_id = request.args["room_id"][0] try: handler = self.handlers.event_stream_handler diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index 85b9f253e3..c7ea15c624 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -348,8 +348,7 @@ class RoomMessageListRestServlet(ClientV1RestServlet): handler = self.handlers.message_handler msgs = yield handler.get_messages( room_id=room_id, - user_id=requester.user.to_string(), - is_guest=requester.is_guest, + requester=requester, pagin_config=pagination_config, as_client_event=as_client_event ) @@ -384,9 +383,8 @@ class RoomInitialSyncRestServlet(ClientV1RestServlet): pagination_config = PaginationConfig.from_request(request) content = yield self.handlers.message_handler.room_initial_sync( room_id=room_id, - user_id=requester.user.to_string(), + requester=requester, pagin_config=pagination_config, - is_guest=requester.is_guest, ) defer.returnValue((200, content)) diff --git a/synapse/rest/client/v2_alpha/account.py b/synapse/rest/client/v2_alpha/account.py index fa56249a69..d507172704 100644 --- a/synapse/rest/client/v2_alpha/account.py +++ b/synapse/rest/client/v2_alpha/account.py @@ -56,10 +56,9 @@ class PasswordRestServlet(RestServlet): if LoginType.PASSWORD in result: # if using password, they should also be logged in requester = yield self.auth.get_user_by_req(request) - requester_user_id = requester.user.to_string() - if requester_user_id.to_string() != result[LoginType.PASSWORD]: + user_id = requester.user.to_string() + if user_id != result[LoginType.PASSWORD]: raise LoginError(400, "", Codes.UNKNOWN) - user_id = requester_user_id elif LoginType.EMAIL_IDENTITY in result: threepid = result[LoginType.EMAIL_IDENTITY] if 'medium' not in threepid or 'address' not in threepid: diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py index df4b305b49..4114a7e430 100644 --- a/synapse/rest/client/v2_alpha/sync.py +++ b/synapse/rest/client/v2_alpha/sync.py @@ -85,6 +85,13 @@ class SyncRestServlet(RestServlet): @defer.inlineCallbacks def on_GET(self, request): + if "from" in request.args: + # /events used to use 'from', but /sync uses 'since'. + # Lets be helpful and whine if we see a 'from'. + raise SynapseError( + 400, "'from' is not a valid query parameter. Did you mean 'since'?" + ) + requester = yield self.auth.get_user_by_req( request, allow_guest=True ) diff --git a/synapse/storage/receipts.py b/synapse/storage/receipts.py index c80e576620..c4232bdc65 100644 --- a/synapse/storage/receipts.py +++ b/synapse/storage/receipts.py @@ -45,6 +45,21 @@ class ReceiptsStore(SQLBaseStore): desc="get_receipts_for_room", ) + @cachedInlineCallbacks(num_args=2) + def get_receipts_for_user(self, user_id, receipt_type): + def f(txn): + sql = ( + "SELECT room_id,event_id " + "FROM receipts_linearized " + "WHERE user_id = ? AND receipt_type = ? " + ) + txn.execute(sql, (user_id, receipt_type)) + return txn.fetchall() + + defer.returnValue(dict( + (yield self.runInteraction("get_receipts_for_user", f)) + )) + @defer.inlineCallbacks def get_linearized_receipts_for_rooms(self, room_ids, to_key, from_key=None): """Get receipts for multiple rooms for sending to clients. @@ -194,29 +209,16 @@ class ReceiptsStore(SQLBaseStore): def get_max_receipt_stream_id(self): return self._receipts_id_gen.get_max_token(self) - @cachedInlineCallbacks() - def get_graph_receipts_for_room(self, room_id): - """Get receipts for sending to remote servers. - """ - rows = yield self._simple_select_list( - table="receipts_graph", - keyvalues={"room_id": room_id}, - retcols=["receipt_type", "user_id", "event_id"], - desc="get_linearized_receipts_for_room", - ) - - result = {} - for row in rows: - result.setdefault( - row["user_id"], {} - ).setdefault( - row["receipt_type"], [] - ).append(row["event_id"]) - - defer.returnValue(result) - def insert_linearized_receipt_txn(self, txn, room_id, receipt_type, user_id, event_id, data, stream_id): + txn.call_after( + self.get_receipts_for_room.invalidate, (room_id, receipt_type) + ) + txn.call_after( + self.get_receipts_for_user.invalidate, (user_id, receipt_type) + ) + # FIXME: This shouldn't invalidate the whole cache + txn.call_after(self.get_linearized_receipts_for_room.invalidate_all) # We don't want to clobber receipts for more recent events, so we # have to compare orderings of existing receipts @@ -324,6 +326,7 @@ class ReceiptsStore(SQLBaseStore): ) max_persisted_id = yield self._stream_id_gen.get_max_token(self) + defer.returnValue((stream_id, max_persisted_id)) def insert_graph_receipt(self, room_id, receipt_type, user_id, event_ids, @@ -336,6 +339,15 @@ class ReceiptsStore(SQLBaseStore): def insert_graph_receipt_txn(self, txn, room_id, receipt_type, user_id, event_ids, data): + txn.call_after( + self.get_receipts_for_room.invalidate, (room_id, receipt_type) + ) + txn.call_after( + self.get_receipts_for_user.invalidate, (user_id, receipt_type) + ) + # FIXME: This shouldn't invalidate the whole cache + txn.call_after(self.get_linearized_receipts_for_room.invalidate_all) + self._simple_delete_txn( txn, table="receipts_graph", diff --git a/synapse/storage/schema/delta/28/receipts_user_id_index.sql b/synapse/storage/schema/delta/28/receipts_user_id_index.sql new file mode 100644 index 0000000000..452a1b3c6c --- /dev/null +++ b/synapse/storage/schema/delta/28/receipts_user_id_index.sql @@ -0,0 +1,18 @@ +/* Copyright 2015, 2016 OpenMarket Ltd + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +CREATE INDEX receipts_linearized_user ON receipts_linearized( + user_id +); |