diff options
63 files changed, 1956 insertions, 1386 deletions
diff --git a/CHANGES.rst b/CHANGES.rst new file mode 100644 index 0000000000..055f8bc01b --- /dev/null +++ b/CHANGES.rst @@ -0,0 +1,20 @@ +Changes in synapse 0.0.1 +======================= +Homeserver: + * Completely change the database schema to support generic event types. + * Improve presence reliability. + * Improve reliability of joining remote rooms. + * Fix bug where room join events were duplicated. + * Improve initial sync API to return more information to the client. + * Stop generating fake messages for room membership events. + +Webclient: + * Add tab completion of names. + * Add ability to upload and send images. + * Add profile pages. + * Improve CSS layout of room. + * Disambiguate identical display names. + * Don't get remote users display names and avatars individually. + * Use the new initial sync API to reduce number of round trips to the homeserver. + * Change url scheme to use room aliases instead of room ids where known. + * Increase longpoll timeout. diff --git a/README.rst b/README.rst index 378b460d0b..069f37ec06 100644 --- a/README.rst +++ b/README.rst @@ -24,11 +24,8 @@ To get up and running: - To run your own **private** homeserver on localhost:8080, install synapse with ``python setup.py develop --user`` and then run one with - ``python synapse/app/homeserver.py`` - - - To run your own webclient, add ``-w``: - ``python synapse/app/homeserver.py -w`` and hit http://localhost:8080/matrix/client - in your web browser (a recent Chrome, Safari or Firefox for now, + ``python synapse/app/homeserver.py`` - you will find a webclient running + at http://localhost:8080 (use a recent Chrome, Safari or Firefox for now, please...) - To make the homeserver **public** and let it exchange messages with @@ -36,7 +33,8 @@ To get up and running: up port 8080 and run ``python synapse/app/homeserver.py --host machine.my.domain.name``. Then come join ``#matrix:matrix.org`` and say hi! :) - + + About Matrix ============ @@ -146,6 +144,13 @@ This should end with a 'PASSED' result:: PASSED (successes=143) +Upgrading an existing homeserver +================================ + +Before upgrading an existing homeserver to a new version, please refer to +UPGRADE.rst for any additional instructions. + + Setting up Federation ===================== @@ -201,9 +206,7 @@ http://localhost:8080. Simply run:: Running The Demo Web Client =========================== -You can run the web client when you run the homeserver by adding ``-w`` to the -command to run ``homeserver.py``. The web client can be accessed via -http://localhost:8080/matrix/client +The homeserver runs a web client by default at http://localhost:8080. If this is the first time you have used the client from that browser (it uses HTML5 local storage to remember its config), you will need to log in to your diff --git a/UPGRADE.rst b/UPGRADE.rst new file mode 100644 index 0000000000..2e75d77bca --- /dev/null +++ b/UPGRADE.rst @@ -0,0 +1,24 @@ +Upgrading to v0.0.1 +=================== + +This release completely changes the database schema and so requires upgrading +it before starting the new version of the homeserver. + +The script "database-prepare-for-0.0.1.sh" should be used to upgrade the +database. This will save all user information, such as logins and profiles, +but will otherwise purge the database. This includes messages, which +rooms the home server was a member of and room alias mappings. + +Before running the command the homeserver should be first completely +shutdown. To run it, simply specify the location of the database, e.g.: + + ./database-prepare-for-0.0.1.sh "homeserver.db" + +Once this has successfully completed it will be safe to restart the +homeserver. You may notice that the homeserver takes a few seconds longer to +restart than usual as it reinitializes the database. + +On startup of the new version, users can either rejoin remote rooms using room +aliases or by being reinvited. Alternatively, if any other homeserver sends a +message to a room that the homeserver was previously in the local HS will +automatically rejoin the room. diff --git a/VERSION b/VERSION new file mode 100644 index 0000000000..8acdd82b76 --- /dev/null +++ b/VERSION @@ -0,0 +1 @@ +0.0.1 diff --git a/cmdclient/console.py b/cmdclient/console.py index 6c6e2085b4..64557a4c40 100755 --- a/cmdclient/console.py +++ b/cmdclient/console.py @@ -233,56 +233,68 @@ class SynapseCmd(cmd.Cmd): defer.returnValue(False) defer.returnValue(True) - def do_3pidrequest(self, line): + def do_emailrequest(self, line): """Requests the association of a third party identifier - <medium> The medium of the identifer (currently only 'email') - <address> The address of the identifer (ie. the email address) + <address> The email address) + <clientSecret> A string of characters generated when requesting an email that you'll supply in subsequent calls to identify yourself + <sendAttempt> The number of times the user has requested an email. Leave this the same between requests to retry the request at the transport level. Increment it to request that the email be sent again. """ - args = self._parse(line, ['medium', 'address']) + args = self._parse(line, ['address', 'clientSecret', 'sendAttempt']) - if not args['medium'] == 'email': - print "Only email is supported currently" - return - - postArgs = {'email': args['address'], 'clientSecret': '____'} + postArgs = {'email': args['address'], 'clientSecret': args['clientSecret'], 'sendAttempt': args['sendAttempt']} - reactor.callFromThread(self._do_3pidrequest, postArgs) + reactor.callFromThread(self._do_emailrequest, postArgs) @defer.inlineCallbacks - def _do_3pidrequest(self, args): + def _do_emailrequest(self, args): url = self._identityServerUrl()+"/matrix/identity/api/v1/validate/email/requestToken" json_res = yield self.http_client.do_request("POST", url, data=urllib.urlencode(args), jsonreq=False, headers={'Content-Type': ['application/x-www-form-urlencoded']}) print json_res - if 'tokenId' in json_res: - print "Token ID %s sent" % (json_res['tokenId']) + if 'sid' in json_res: + print "Token sent. Your session ID is %s" % (json_res['sid']) - def do_3pidvalidate(self, line): + def do_emailvalidate(self, line): """Validate and associate a third party ID - <medium> The medium of the identifer (currently only 'email') - <tokenId> The identifier iof the token given in 3pidrequest + <sid> The session ID (sid) given to you in the response to requestToken <token> The token sent to your third party identifier address + <clientSecret> The same clientSecret you supplied in requestToken """ - args = self._parse(line, ['medium', 'tokenId', 'token']) - - if not args['medium'] == 'email': - print "Only email is supported currently" - return + args = self._parse(line, ['sid', 'token', 'clientSecret']) - postArgs = { 'tokenId' : args['tokenId'], 'token' : args['token'] } - postArgs['mxId'] = self.config["user"] + postArgs = { 'sid' : args['sid'], 'token' : args['token'], 'clientSecret': args['clientSecret'] } - reactor.callFromThread(self._do_3pidvalidate, postArgs) + reactor.callFromThread(self._do_emailvalidate, postArgs) @defer.inlineCallbacks - def _do_3pidvalidate(self, args): + def _do_emailvalidate(self, args): url = self._identityServerUrl()+"/matrix/identity/api/v1/validate/email/submitToken" json_res = yield self.http_client.do_request("POST", url, data=urllib.urlencode(args), jsonreq=False, headers={'Content-Type': ['application/x-www-form-urlencoded']}) print json_res + def do_3pidbind(self, line): + """Validate and associate a third party ID + <sid> The session ID (sid) given to you in the response to requestToken + <clientSecret> The same clientSecret you supplied in requestToken + """ + args = self._parse(line, ['sid', 'clientSecret']) + + postArgs = { 'sid' : args['sid'], 'clientSecret': args['clientSecret'] } + postArgs['mxid'] = self.config["user"] + + reactor.callFromThread(self._do_3pidbind, postArgs) + + @defer.inlineCallbacks + def _do_3pidbind(self, args): + url = self._identityServerUrl()+"/matrix/identity/api/v1/3pid/bind" + + json_res = yield self.http_client.do_request("POST", url, data=urllib.urlencode(args), jsonreq=False, + headers={'Content-Type': ['application/x-www-form-urlencoded']}) + print json_res + def do_join(self, line): """Joins a room: "join <roomid>" """ try: diff --git a/database-prepare-for-0.0.1.sh b/database-prepare-for-0.0.1.sh new file mode 100755 index 0000000000..43d759a5cd --- /dev/null +++ b/database-prepare-for-0.0.1.sh @@ -0,0 +1,21 @@ +#!/bin/bash + +# This is will prepare a synapse database for running with v0.0.1 of synapse. +# It will store all the user information, but will *delete* all messages and +# room data. + +set -e + +cp "$1" "$1.bak" + +DUMP=$(sqlite3 "$1" << 'EOF' +.dump users +.dump access_tokens +.dump presence +.dump profiles +EOF +) + +rm "$1" + +sqlite3 "$1" <<< "$DUMP" diff --git a/database-save.sh b/database-save.sh index c80f676f76..040c8a4943 100755 --- a/database-save.sh +++ b/database-save.sh @@ -8,7 +8,7 @@ # # $ sqlite3 homeserver.db < table-save.sql -sqlite3 homeserver.db <<'EOF' >table-save.sql +sqlite3 "$1" <<'EOF' >table-save.sql .dump users .dump access_tokens .dump presence diff --git a/graph/graph.py b/graph/graph.py index da829c388f..220f5eb1d5 100644 --- a/graph/graph.py +++ b/graph/graph.py @@ -113,7 +113,7 @@ def make_graph(pdus, room, filename_prefix): graph.add_edge(state_edge) graph.write('%s.dot' % filename_prefix, format='raw', prog='dot') - graph.write_png("%s.png" % filename_prefix, prog='dot') +# graph.write_png("%s.png" % filename_prefix, prog='dot') graph.write_svg("%s.svg" % filename_prefix, prog='dot') diff --git a/setup.py b/setup.py index fca3c77700..f01eec436f 100644 --- a/setup.py +++ b/setup.py @@ -25,7 +25,7 @@ def read(fname): setup( name="SynapseHomeServer", - version="0.1", + version="0.0.1", packages=find_packages(exclude=["tests"]), description="Reference Synapse Home Server", install_requires=[ diff --git a/synapse/__init__.py b/synapse/__init__.py index 1e7b2ab272..47fc1b2ea4 100644 --- a/synapse/__init__.py +++ b/synapse/__init__.py @@ -15,3 +15,5 @@ """ This is a reference implementation of a synapse home server. """ + +__version__ = "0.0.1" diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py index 921fd08832..aa04dbece7 100644 --- a/synapse/api/events/__init__.py +++ b/synapse/api/events/__init__.py @@ -51,6 +51,7 @@ class SynapseEvent(JsonEncodedObject): "depth", "destinations", "origin", + "outlier", ] required_keys = [ diff --git a/synapse/api/events/factory.py b/synapse/api/events/factory.py index 12aa04fc6e..c2cdcddf41 100644 --- a/synapse/api/events/factory.py +++ b/synapse/api/events/factory.py @@ -15,7 +15,7 @@ from synapse.api.events.room import ( RoomTopicEvent, MessageEvent, RoomMemberEvent, FeedbackEvent, - InviteJoinEvent, RoomConfigEvent + InviteJoinEvent, RoomConfigEvent, RoomNameEvent, GenericEvent, ) from synapse.util.stringutils import random_string @@ -25,6 +25,7 @@ class EventFactory(object): _event_classes = [ RoomTopicEvent, + RoomNameEvent, MessageEvent, RoomMemberEvent, FeedbackEvent, @@ -32,20 +33,24 @@ class EventFactory(object): RoomConfigEvent ] - def __init__(self): + def __init__(self, hs): self._event_list = {} # dict of TYPE to event class for event_class in EventFactory._event_classes: self._event_list[event_class.TYPE] = event_class + self.clock = hs.get_clock() + def create_event(self, etype=None, **kwargs): kwargs["type"] = etype if "event_id" not in kwargs: kwargs["event_id"] = random_string(10) - try: + if "ts" not in kwargs: + kwargs["ts"] = int(self.clock.time_msec()) + + if etype in self._event_list: handler = self._event_list[etype] - except KeyError: # unknown event type - # TODO allow custom event types. - raise NotImplementedError("Unknown etype=%s" % etype) + else: + handler = GenericEvent return handler(**kwargs) diff --git a/synapse/api/events/room.py b/synapse/api/events/room.py index f3df849af2..42459f3f21 100644 --- a/synapse/api/events/room.py +++ b/synapse/api/events/room.py @@ -16,17 +16,45 @@ from . import SynapseEvent +class GenericEvent(SynapseEvent): + def get_content_template(self): + return {} + + class RoomTopicEvent(SynapseEvent): TYPE = "m.room.topic" + internal_keys = SynapseEvent.internal_keys + [ + "topic", + ] + def __init__(self, **kwargs): kwargs["state_key"] = "" + if "topic" in kwargs["content"]: + kwargs["topic"] = kwargs["content"]["topic"] super(RoomTopicEvent, self).__init__(**kwargs) def get_content_template(self): return {"topic": u"string"} +class RoomNameEvent(SynapseEvent): + TYPE = "m.room.name" + + internal_keys = SynapseEvent.internal_keys + [ + "name", + ] + + def __init__(self, **kwargs): + kwargs["state_key"] = "" + if "name" in kwargs["content"]: + kwargs["name"] = kwargs["content"]["name"] + super(RoomNameEvent, self).__init__(**kwargs) + + def get_content_template(self): + return {"name": u"string"} + + class RoomMemberEvent(SynapseEvent): TYPE = "m.room.member" @@ -38,6 +66,8 @@ class RoomMemberEvent(SynapseEvent): def __init__(self, **kwargs): if "target_user_id" in kwargs: kwargs["state_key"] = kwargs["target_user_id"] + if "membership" not in kwargs: + kwargs["membership"] = kwargs.get("content", {}).get("membership") super(RoomMemberEvent, self).__init__(**kwargs) def get_content_template(self): diff --git a/synapse/api/notifier.py b/synapse/api/notifier.py index 65b5a4ebb3..9f622df6bb 100644 --- a/synapse/api/notifier.py +++ b/synapse/api/notifier.py @@ -15,6 +15,7 @@ from synapse.api.constants import Membership from synapse.api.events.room import RoomMemberEvent +from synapse.api.streams.event import EventsStreamData from twisted.internet import defer from twisted.internet import reactor @@ -66,7 +67,7 @@ class Notifier(object): self._notify_and_callback( user_id=user_id, event_data=event.get_dict(), - stream_type=event.type, + stream_type=EventsStreamData.EVENT_TYPE, store_id=store_id) def on_new_user_event(self, user_id, event_data, stream_type, store_id): diff --git a/synapse/api/streams/__init__.py b/synapse/api/streams/__init__.py index 989e63f9ec..d831eafbab 100644 --- a/synapse/api/streams/__init__.py +++ b/synapse/api/streams/__init__.py @@ -20,23 +20,24 @@ class PaginationConfig(object): """A configuration object which stores pagination parameters.""" - def __init__(self, from_tok=None, to_tok=None, limit=0): + def __init__(self, from_tok=None, to_tok=None, direction='f', limit=0): self.from_tok = from_tok self.to_tok = to_tok + self.direction = direction self.limit = limit @classmethod def from_request(cls, request, raise_invalid_params=True): params = { - "from_tok": PaginationStream.TOK_START, - "to_tok": PaginationStream.TOK_END, - "limit": 0 + "from_tok": "END", + "direction": 'f', } query_param_mappings = [ # 3-tuple of qp_key, attribute, rules ("from", "from_tok", lambda x: type(x) == str), ("to", "to_tok", lambda x: type(x) == str), - ("limit", "limit", lambda x: x.isdigit()) + ("limit", "limit", lambda x: x.isdigit()), + ("dir", "direction", lambda x: x == 'f' or x == 'b'), ] for qp, attr, is_valid in query_param_mappings: @@ -48,12 +49,17 @@ class PaginationConfig(object): return PaginationConfig(**params) + def __str__(self): + return ( + "<PaginationConfig from_tok=%s, to_tok=%s, " + "direction=%s, limit=%s>" + ) % (self.from_tok, self.to_tok, self.direction, self.limit) + class PaginationStream(object): """ An interface for streaming data as chunks. """ - TOK_START = "START" TOK_END = "END" def get_chunk(self, config=None): @@ -76,7 +82,7 @@ class StreamData(object): self.hs = hs self.store = hs.get_datastore() - def get_rows(self, user_id, from_pkey, to_pkey, limit): + def get_rows(self, user_id, from_pkey, to_pkey, limit, direction): """ Get event stream data between the specified pkeys. Args: diff --git a/synapse/api/streams/event.py b/synapse/api/streams/event.py index 4b6d739e54..a5c8b2b31f 100644 --- a/synapse/api/streams/event.py +++ b/synapse/api/streams/event.py @@ -18,6 +18,7 @@ from twisted.internet import defer from synapse.api.errors import EventStreamError +from synapse.api.events import SynapseEvent from synapse.api.events.room import ( RoomMemberEvent, MessageEvent, FeedbackEvent, RoomTopicEvent ) @@ -28,17 +29,17 @@ import logging logger = logging.getLogger(__name__) -class MessagesStreamData(StreamData): - EVENT_TYPE = MessageEvent.TYPE +class EventsStreamData(StreamData): + EVENT_TYPE = "EventsStream" def __init__(self, hs, room_id=None, feedback=False): - super(MessagesStreamData, self).__init__(hs) + super(EventsStreamData, self).__init__(hs) self.room_id = room_id self.with_feedback = feedback @defer.inlineCallbacks - def get_rows(self, user_id, from_key, to_key, limit): - (data, latest_ver) = yield self.store.get_message_stream( + def get_rows(self, user_id, from_key, to_key, limit, direction): + data, latest_ver = yield self.store.get_room_events( user_id=user_id, from_key=from_key, to_key=to_key, @@ -50,74 +51,7 @@ class MessagesStreamData(StreamData): @defer.inlineCallbacks def max_token(self): - val = yield self.store.get_max_message_id() - defer.returnValue(val) - - -class RoomMemberStreamData(StreamData): - EVENT_TYPE = RoomMemberEvent.TYPE - - @defer.inlineCallbacks - def get_rows(self, user_id, from_key, to_key, limit): - (data, latest_ver) = yield self.store.get_room_member_stream( - user_id=user_id, - from_key=from_key, - to_key=to_key - ) - - defer.returnValue((data, latest_ver)) - - @defer.inlineCallbacks - def max_token(self): - val = yield self.store.get_max_room_member_id() - defer.returnValue(val) - - -class FeedbackStreamData(StreamData): - EVENT_TYPE = FeedbackEvent.TYPE - - def __init__(self, hs, room_id=None): - super(FeedbackStreamData, self).__init__(hs) - self.room_id = room_id - - @defer.inlineCallbacks - def get_rows(self, user_id, from_key, to_key, limit): - (data, latest_ver) = yield self.store.get_feedback_stream( - user_id=user_id, - from_key=from_key, - to_key=to_key, - limit=limit, - room_id=self.room_id - ) - defer.returnValue((data, latest_ver)) - - @defer.inlineCallbacks - def max_token(self): - val = yield self.store.get_max_feedback_id() - defer.returnValue(val) - - -class RoomDataStreamData(StreamData): - EVENT_TYPE = RoomTopicEvent.TYPE # TODO need multiple event types - - def __init__(self, hs, room_id=None): - super(RoomDataStreamData, self).__init__(hs) - self.room_id = room_id - - @defer.inlineCallbacks - def get_rows(self, user_id, from_key, to_key, limit): - (data, latest_ver) = yield self.store.get_room_data_stream( - user_id=user_id, - from_key=from_key, - to_key=to_key, - limit=limit, - room_id=self.room_id - ) - defer.returnValue((data, latest_ver)) - - @defer.inlineCallbacks - def max_token(self): - val = yield self.store.get_max_room_data_id() + val = yield self.store.get_room_events_max_id() defer.returnValue(val) @@ -136,6 +70,15 @@ class EventStream(PaginationStream): pagination_config.from_tok) pagination_config.to_tok = yield self.fix_token( pagination_config.to_tok) + + if ( + not pagination_config.to_tok + and pagination_config.direction == 'f' + ): + pagination_config.to_tok = yield self.get_current_max_token() + + logger.debug("pagination_config: %s", pagination_config) + defer.returnValue(pagination_config) @defer.inlineCallbacks @@ -147,39 +90,42 @@ class EventStream(PaginationStream): Returns: The fixed-up token, which may == token. """ - # replace TOK_START and TOK_END with 0_0_0 or -1_-1_-1 depending. - replacements = [ - (PaginationStream.TOK_START, "0"), - (PaginationStream.TOK_END, "-1") - ] - for magic_token, key in replacements: - if magic_token == token: - token = EventStream.SEPARATOR.join( - [key] * len(self.stream_data) - ) - - # replace -1 values with an actual pkey - token_segments = self._split_token(token) - for i, tok in enumerate(token_segments): - if tok == -1: - # add 1 to the max token because results are EXCLUSIVE from the - # latest version. - token_segments[i] = 1 + (yield self.stream_data[i].max_token()) - defer.returnValue(EventStream.SEPARATOR.join( - str(x) for x in token_segments - )) + if token == PaginationStream.TOK_END: + new_token = yield self.get_current_max_token() + + logger.debug("fix_token: From %s to %s", token, new_token) + + token = new_token + + defer.returnValue(token) @defer.inlineCallbacks - def get_chunk(self, config=None): + def get_current_max_token(self): + new_token_parts = [] + for s in self.stream_data: + mx = yield s.max_token() + new_token_parts.append(str(mx)) + + new_token = EventStream.SEPARATOR.join(new_token_parts) + + logger.debug("get_current_max_token: %s", new_token) + + defer.returnValue(new_token) + + @defer.inlineCallbacks + def get_chunk(self, config): # no support for limit on >1 streams, makes no sense. if config.limit and len(self.stream_data) > 1: raise EventStreamError( 400, "Limit not supported on multiplexed streams." ) - (chunk_data, next_tok) = yield self._get_chunk_data(config.from_tok, - config.to_tok, - config.limit) + chunk_data, next_tok = yield self._get_chunk_data( + config.from_tok, + config.to_tok, + config.limit, + config.direction, + ) defer.returnValue({ "chunk": chunk_data, @@ -188,7 +134,7 @@ class EventStream(PaginationStream): }) @defer.inlineCallbacks - def _get_chunk_data(self, from_tok, to_tok, limit): + def _get_chunk_data(self, from_tok, to_tok, limit, direction): """ Get event data between the two tokens. Tokens are SEPARATOR separated values representing pkey values of @@ -206,11 +152,12 @@ class EventStream(PaginationStream): EventStreamError if something went wrong. """ # sanity check - if (from_tok.count(EventStream.SEPARATOR) != - to_tok.count(EventStream.SEPARATOR) or - (from_tok.count(EventStream.SEPARATOR) + 1) != - len(self.stream_data)): - raise EventStreamError(400, "Token lengths don't match.") + if to_tok is not None: + if (from_tok.count(EventStream.SEPARATOR) != + to_tok.count(EventStream.SEPARATOR) or + (from_tok.count(EventStream.SEPARATOR) + 1) != + len(self.stream_data)): + raise EventStreamError(400, "Token lengths don't match.") chunk = [] next_ver = [] @@ -224,10 +171,13 @@ class EventStream(PaginationStream): continue (event_chunk, max_pkey) = yield self.stream_data[i].get_rows( - self.user_id, from_pkey, to_pkey, limit + self.user_id, from_pkey, to_pkey, limit, direction, ) - chunk += event_chunk + chunk.extend([ + e.get_dict() if isinstance(e, SynapseEvent) else e + for e in event_chunk + ]) next_ver.append(str(max_pkey)) defer.returnValue((chunk, EventStream.SEPARATOR.join(next_ver))) @@ -240,9 +190,8 @@ class EventStream(PaginationStream): Returns: A list of ints. """ - segments = token.split(EventStream.SEPARATOR) - try: - int_segments = [int(x) for x in segments] - except ValueError: - raise EventStreamError(400, "Bad token: %s" % token) - return int_segments + if token: + segments = token.split(EventStream.SEPARATOR) + else: + segments = [None] * len(self.stream_data) + return segments diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index ca102236cf..495149466c 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -56,7 +56,7 @@ class SynapseHomeServer(HomeServer): return File("webclient") # TODO configurable? def build_resource_for_content_repo(self): - return ContentRepoResource("uploads", self.auth) + return ContentRepoResource(self, self.upload_dir, self.auth) def build_db_pool(self): """ Set up all the dbs. Since all the *.sql have IF NOT EXISTS, so we @@ -235,8 +235,8 @@ def setup(): parser.add_argument('--pid-file', dest="pid", help="When running as a " "daemon, the file to store the pid in", default="hs.pid") - parser.add_argument("-w", "--webclient", dest="webclient", - action="store_true", help="Host the web client.") + parser.add_argument("-W", "--webclient", dest="webclient", default=True, + action="store_false", help="Don't host a web client.") args = parser.parse_args() verbosity = int(args.verbose) if args.verbose else None @@ -257,7 +257,8 @@ def setup(): hs = SynapseHomeServer( args.host, - db_name=db_name + upload_dir=os.path.abspath("uploads"), + db_name=db_name, ) # This object doesn't need to be saved because it's set as the handler for diff --git a/synapse/federation/handler.py b/synapse/federation/handler.py index 580e591aca..984c1558e9 100644 --- a/synapse/federation/handler.py +++ b/synapse/federation/handler.py @@ -63,7 +63,7 @@ class FederationEventHandler(object): Deferred: Resolved when it has successfully been queued for processing. """ - yield self._fill_out_prev_events(event) + yield self.fill_out_prev_events(event) pdu = self.pdu_codec.pdu_from_event(event) @@ -74,10 +74,18 @@ class FederationEventHandler(object): @log_function @defer.inlineCallbacks - def backfill(self, room_id, limit): - # TODO: Work out which destinations to ask for backfill - # self.replication_layer.backfill(dest, room_id, limit) - pass + def backfill(self, dest, room_id, limit): + pdus = yield self.replication_layer.backfill(dest, room_id, limit) + + if not pdus: + defer.returnValue([]) + + events = [ + self.pdu_codec.event_from_pdu(pdu) + for pdu in pdus + ] + + defer.returnValue(events) @log_function def get_state_for_room(self, destination, room_id): @@ -87,7 +95,7 @@ class FederationEventHandler(object): @log_function @defer.inlineCallbacks - def on_receive_pdu(self, pdu): + def on_receive_pdu(self, pdu, backfilled): """ Called by the ReplicationLayer when we have a new pdu. We need to do auth checks and put it throught the StateHandler. """ @@ -95,7 +103,7 @@ class FederationEventHandler(object): try: with (yield self.lock_manager.lock(pdu.context)): - if event.is_state: + if event.is_state and not backfilled: is_new_state = yield self.state_handler.handle_new_state( pdu ) @@ -104,7 +112,7 @@ class FederationEventHandler(object): else: is_new_state = False - yield self.event_handler.on_receive(event, is_new_state) + yield self.event_handler.on_receive(event, is_new_state, backfilled) except AuthError: # TODO: Implement something in federation that allows us to @@ -129,7 +137,7 @@ class FederationEventHandler(object): yield self.event_handler.on_receive(new_state_event) @defer.inlineCallbacks - def _fill_out_prev_events(self, event): + def fill_out_prev_events(self, event): if hasattr(event, "prev_events"): return diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index c9f2e06b7b..8030d0963f 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -209,7 +209,7 @@ class ReplicationLayer(object): pdus = [Pdu(outlier=False, **p) for p in transaction.pdus] for pdu in pdus: - yield self._handle_new_pdu(pdu) + yield self._handle_new_pdu(pdu, backfilled=True) defer.returnValue(pdus) @@ -416,7 +416,7 @@ class ReplicationLayer(object): @defer.inlineCallbacks @log_function - def _handle_new_pdu(self, pdu): + def _handle_new_pdu(self, pdu, backfilled=False): # We reprocess pdus when we have seen them only as outliers existing = yield self._get_persisted_pdu(pdu.pdu_id, pdu.origin) @@ -452,7 +452,10 @@ class ReplicationLayer(object): # Persist the Pdu, but don't mark it as processed yet. yield self.pdu_actions.persist_received(pdu) - ret = yield self.handler.on_receive_pdu(pdu) + if not backfilled: + ret = yield self.handler.on_receive_pdu(pdu, backfilled=backfilled) + else: + ret = None yield self.pdu_actions.mark_as_processed(pdu) diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index c2f4685c92..3f07b5aa4a 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -24,4 +24,5 @@ class BaseHandler(object): self.notifier = hs.get_notifier() self.room_lock = hs.get_room_lock_manager() self.state_handler = hs.get_state_handler() + self.distributor = hs.get_distributor() self.hs = hs diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index 3af7d824a2..6bb797caf2 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -17,8 +17,7 @@ from twisted.internet import defer from ._base import BaseHandler from synapse.api.streams.event import ( - EventStream, MessagesStreamData, RoomMemberStreamData, FeedbackStreamData, - RoomDataStreamData + EventStream, EventsStreamData ) from synapse.handlers.presence import PresenceStreamData @@ -26,10 +25,7 @@ from synapse.handlers.presence import PresenceStreamData class EventStreamHandler(BaseHandler): stream_data_classes = [ - MessagesStreamData, - RoomMemberStreamData, - FeedbackStreamData, - RoomDataStreamData, + EventsStreamData, PresenceStreamData, ] diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 7026df90a2..9cff444779 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -32,10 +32,19 @@ logger = logging.getLogger(__name__) class FederationHandler(BaseHandler): """Handles events that originated from federation.""" + def __init__(self, hs): + super(FederationHandler, self).__init__(hs) + + self.distributor.observe( + "user_joined_room", + self._on_user_joined + ) + + self.waiting_for_join_list = {} @log_function @defer.inlineCallbacks - def on_receive(self, event, is_new_state): + def on_receive(self, event, is_new_state, backfilled): if hasattr(event, "state_key") and not is_new_state: logger.debug("Ignoring old state.") return @@ -70,6 +79,115 @@ class FederationHandler(BaseHandler): else: with (yield self.room_lock.lock(event.room_id)): - store_id = yield self.store.persist_event(event) + store_id = yield self.store.persist_event(event, backfilled) + + room = yield self.store.get_room(event.room_id) + + if not room: + # Huh, let's try and get the current state + try: + federation = self.hs.get_federation() + yield federation.get_state_for_room( + event.origin, event.room_id + ) + + hosts = yield self.store.get_joined_hosts_for_room( + event.room_id + ) + if self.hs.hostname in hosts: + try: + yield self.store.store_room( + event.room_id, + "", + is_public=False + ) + except: + pass + except: + logger.exception( + "Failed to get current state for room %s", + event.room_id + ) + + if not backfilled: + yield self.notifier.on_new_room_event(event, store_id) + + if event.type == RoomMemberEvent.TYPE: + if event.membership == Membership.JOIN: + user = self.hs.parse_userid(event.target_user_id) + self.distributor.fire( + "user_joined_room", user=user, room_id=event.room_id + ) + + + @log_function + @defer.inlineCallbacks + def backfill(self, dest, room_id, limit): + events = yield self.hs.get_federation().backfill(dest, room_id, limit) + + for event in events: + try: + yield self.store.persist_event(event, backfilled=True) + except: + logger.exception("Failed to persist event: %s", event) + + defer.returnValue(events) - yield self.notifier.on_new_room_event(event, store_id) + @log_function + @defer.inlineCallbacks + def do_invite_join(self, target_host, room_id, joinee, content): + federation = self.hs.get_federation() + + hosts = yield self.store.get_joined_hosts_for_room(room_id) + if self.hs.hostname in hosts: + # We are already in the room. + logger.debug("We're already in the room apparently") + defer.returnValue(False) + + # First get current state to see if we are already joined. + try: + yield federation.get_state_for_room(target_host, room_id) + + hosts = yield self.store.get_joined_hosts_for_room(room_id) + if self.hs.hostname in hosts: + # Oh, we were actually in the room already. + logger.debug("We're already in the room apparently") + defer.returnValue(False) + except Exception: + logger.exception("Failed to get current state") + + new_event = self.event_factory.create_event( + etype=InviteJoinEvent.TYPE, + target_host=target_host, + room_id=room_id, + user_id=joinee, + content=content + ) + + new_event.destinations = [target_host] + + yield federation.handle_new_event(new_event) + + # TODO (erikj): Time out here. + d = defer.Deferred() + self.waiting_for_join_list.setdefault((joinee, room_id), []).append(d) + yield d + + try: + yield self.store.store_room( + event.room_id, + "", + is_public=False + ) + except: + pass + + + defer.returnValue(True) + + + @log_function + def _on_user_joined(self, user, room_id): + waiters = self.waiting_for_join_list.get((user.to_string(), room_id), []) + while waiters: + waiters.pop().callback(None) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index a2152c99cf..540e114b82 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -685,7 +685,10 @@ class PresenceStreamData(StreamData): super(PresenceStreamData, self).__init__(hs) self.presence = hs.get_handlers().presence_handler - def get_rows(self, user_id, from_key, to_key, limit): + def get_rows(self, user_id, from_key, to_key, limit, direction): + from_key = int(from_key) + to_key = int(to_key) + cachemap = self.presence._user_cachemap # TODO(paul): limit, and filter by visibility diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 5d0379254b..899b653fb7 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -23,7 +23,8 @@ from synapse.api.events.room import ( RoomTopicEvent, MessageEvent, InviteJoinEvent, RoomMemberEvent, RoomConfigEvent ) -from synapse.api.streams.event import EventStream, MessagesStreamData +from synapse.api.streams.event import EventStream, EventsStreamData +from synapse.handlers.presence import PresenceStreamData from synapse.util import stringutils from ._base import BaseHandler @@ -59,12 +60,14 @@ class MessageHandler(BaseHandler): yield self.auth.check_joined_room(room_id, user_id) # Pull out the message from the db - msg = yield self.store.get_message(room_id=room_id, - msg_id=msg_id, - user_id=sender_id) +# msg = yield self.store.get_message( +# room_id=room_id, +# msg_id=msg_id, +# user_id=sender_id +# ) + + # TODO (erikj): Once we work out the correct c-s api we need to think on how to do this. - if msg: - defer.returnValue(msg) defer.returnValue(None) @defer.inlineCallbacks @@ -114,8 +117,9 @@ class MessageHandler(BaseHandler): """ yield self.auth.check_joined_room(room_id, user_id) - data_source = [MessagesStreamData(self.hs, room_id=room_id, - feedback=feedback)] + data_source = [ + EventsStreamData(self.hs, room_id=room_id, feedback=feedback) + ] event_stream = EventStream(user_id, data_source) pagin_config = yield event_stream.fix_tokens(pagin_config) data_chunk = yield event_stream.get_chunk(config=pagin_config) @@ -141,12 +145,7 @@ class MessageHandler(BaseHandler): yield self.state_handler.handle_new_event(event) # store in db - store_id = yield self.store.store_room_data( - room_id=event.room_id, - etype=event.type, - state_key=event.state_key, - content=json.dumps(event.content) - ) + store_id = yield self.store.persist_event(event) event.destinations = yield self.store.get_joined_hosts_for_room( event.room_id @@ -201,19 +200,17 @@ class MessageHandler(BaseHandler): raise RoomError( 403, "Member does not meet private room rules.") - data = yield self.store.get_room_data(room_id, event_type, state_key) + data = yield self.store.get_current_state( + room_id, event_type, state_key + ) defer.returnValue(data) @defer.inlineCallbacks - def get_feedback(self, room_id=None, msg_sender_id=None, msg_id=None, - user_id=None, fb_sender_id=None, fb_type=None): - yield self.auth.check_joined_room(room_id, user_id) + def get_feedback(self, event_id): + # yield self.auth.check_joined_room(room_id, user_id) # Pull out the feedback from the db - fb = yield self.store.get_feedback( - room_id=room_id, msg_id=msg_id, msg_sender_id=msg_sender_id, - fb_sender_id=fb_sender_id, fb_type=fb_type - ) + fb = yield self.store.get_feedback(event_id) if fb: defer.returnValue(fb) @@ -260,20 +257,59 @@ class MessageHandler(BaseHandler): user_id=user_id, membership_list=[Membership.INVITE, Membership.JOIN] ) - for room_info in room_list: - if room_info["membership"] != Membership.JOIN: + + rooms_ret = [] + + now_rooms_token = yield self.store.get_room_events_max_id() + + # FIXME (erikj): Fix this. + presence_stream = PresenceStreamData(self.hs) + now_presence_token = yield presence_stream.max_token() + presence = yield presence_stream.get_rows( + user_id, 0, now_presence_token, None, None + ) + + # FIXME (erikj): We need to not generate this token, + now_token = "%s_%s" % (now_rooms_token, now_presence_token) + + for event in room_list: + d = { + "room_id": event.room_id, + "membership": event.membership, + } + + if event.membership == Membership.INVITE: + d["inviter"] = event.user_id + + rooms_ret.append(d) + + if event.membership != Membership.JOIN: continue try: - event_chunk = yield self.get_messages( - user_id=user_id, - pagin_config=pagin_config, - feedback=feedback, - room_id=room_info["room_id"] + messages, token = yield self.store.get_recent_events_for_room( + event.room_id, + limit=10, + end_token=now_rooms_token, ) - room_info["messages"] = event_chunk + + d["messages"] = { + "chunk": [m.get_dict() for m in messages], + "start": token[0], + "end": token[1], + } + + current_state = yield self.store.get_current_state(event.room_id) + d["state"] = [c.get_dict() for c in current_state] except: - pass - defer.returnValue(room_list) + logger.exception("Failed to get snapshot") + + user = self.hs.parse_userid(user_id) + + ret = {"rooms": rooms_ret, "presence": presence[0], "end": now_token} + + logger.debug("snapshot_all_rooms returning: %s", ret) + + defer.returnValue(ret) class RoomCreationHandler(BaseHandler): @@ -372,7 +408,6 @@ class RoomCreationHandler(BaseHandler): yield self.hs.get_handlers().room_member_handler.change_membership( join_event, - broadcast_msg=True, do_auth=False ) @@ -451,11 +486,11 @@ class RoomMemberHandler(BaseHandler): member_list = yield self.store.get_room_members(room_id=room_id) event_list = [ - entry.as_event(self.event_factory).get_dict() + entry.get_dict() for entry in member_list ] chunk_data = { - "start": "START", + "start": "START", # FIXME (erikj): START is no longer a valid value "end": "END", "chunk": event_list } @@ -484,29 +519,28 @@ class RoomMemberHandler(BaseHandler): defer.returnValue(member) @defer.inlineCallbacks - def change_membership(self, event=None, broadcast_msg=False, do_auth=True): + def change_membership(self, event=None, do_auth=True): """ Change the membership status of a user in a room. Args: event (SynapseEvent): The membership event - broadcast_msg (bool): True to inject a membership message into this - room on success. Raises: SynapseError if there was a problem changing the membership. """ - #broadcast_msg = False - prev_state = yield self.store.get_room_member( event.target_user_id, event.room_id ) - if prev_state and prev_state.membership == event.membership: - # treat this event as a NOOP. - if do_auth: # This is mainly to fix a unit test. - yield self.auth.check(event, raises=True) - defer.returnValue({}) - return + if prev_state: + event.content["prev"] = prev_state.membership + +# if prev_state and prev_state.membership == event.membership: +# # treat this event as a NOOP. +# if do_auth: # This is mainly to fix a unit test. +# yield self.auth.check(event, raises=True) +# defer.returnValue({}) +# return room_id = event.room_id @@ -514,9 +548,7 @@ class RoomMemberHandler(BaseHandler): # if this HS is not currently in the room, i.e. we have to do the # invite/join dance. if event.membership == Membership.JOIN: - yield self._do_join( - event, do_auth=do_auth, broadcast_msg=broadcast_msg - ) + yield self._do_join(event, do_auth=do_auth) else: # This is not a JOIN, so we can handle it normally. if do_auth: @@ -534,7 +566,6 @@ class RoomMemberHandler(BaseHandler): yield self._do_local_membership_update( event, membership=event.content["membership"], - broadcast_msg=broadcast_msg, ) defer.returnValue({"room_id": room_id}) @@ -569,14 +600,14 @@ class RoomMemberHandler(BaseHandler): defer.returnValue({"room_id": room_id}) @defer.inlineCallbacks - def _do_join(self, event, room_host=None, do_auth=True, broadcast_msg=True): + def _do_join(self, event, room_host=None, do_auth=True): joinee = self.hs.parse_userid(event.target_user_id) # room_id = RoomID.from_string(event.room_id, self.hs) room_id = event.room_id # If event doesn't include a display name, add one. - yield self._fill_out_join_content( - joinee, event.content + yield self.distributor.fire( + "collect_presencelike_data", joinee, event.content ) # XXX: We don't do an auth check if we are doing an invite @@ -584,9 +615,9 @@ class RoomMemberHandler(BaseHandler): # that we are allowed to join when we decide whether or not we # need to do the invite/join dance. - room = yield self.store.get_room(room_id) + hosts = yield self.store.get_joined_hosts_for_room(room_id) - if room: + if self.hs.hostname in hosts: should_do_dance = False elif room_host: should_do_dance = True @@ -598,7 +629,7 @@ class RoomMemberHandler(BaseHandler): if prev_state and prev_state.membership == Membership.INVITE: room = yield self.store.get_room(room_id) inviter = UserID.from_string( - prev_state.sender, self.hs + prev_state.user_id, self.hs ) should_do_dance = not inviter.is_mine and not room @@ -606,8 +637,15 @@ class RoomMemberHandler(BaseHandler): else: should_do_dance = False + have_joined = False + if should_do_dance: + handler = self.hs.get_handlers().federation_handler + have_joined = yield handler.do_invite_join( + room_host, room_id, event.user_id, event.content + ) + # We want to do the _do_update inside the room lock. - if not should_do_dance: + if not have_joined: logger.debug("Doing normal join") if do_auth: @@ -617,16 +655,6 @@ class RoomMemberHandler(BaseHandler): yield self._do_local_membership_update( event, membership=event.content["membership"], - broadcast_msg=broadcast_msg, - ) - - - if should_do_dance: - yield self._do_invite_join_dance( - room_id=room_id, - joinee=event.user_id, - target_host=room_host, - content=event.content, ) user = self.hs.parse_userid(event.user_id) @@ -635,32 +663,6 @@ class RoomMemberHandler(BaseHandler): ) @defer.inlineCallbacks - def _fill_out_join_content(self, user_id, content): - # If event doesn't include a display name, add one. - profile_handler = self.hs.get_handlers().profile_handler - if "displayname" not in content: - try: - display_name = yield profile_handler.get_displayname( - user_id - ) - - if display_name: - content["displayname"] = display_name - except: - logger.exception("Failed to set display_name") - - if "avatar_url" not in content: - try: - avatar_url = yield profile_handler.get_avatar_url( - user_id - ) - - if avatar_url: - content["avatar_url"] = avatar_url - except: - logger.exception("Failed to set display_name") - - @defer.inlineCallbacks def _should_invite_join(self, room_id, prev_state, do_auth): logger.debug("_should_invite_join: room_id: %s", room_id) @@ -694,18 +696,12 @@ class RoomMemberHandler(BaseHandler): user_id=user.to_string(), membership_list=membership_list ) - defer.returnValue([r["room_id"] for r in rooms]) + defer.returnValue([r.room_id for r in rooms]) @defer.inlineCallbacks - def _do_local_membership_update(self, event, membership, broadcast_msg): + def _do_local_membership_update(self, event, membership): # store membership - store_id = yield self.store.store_room_member( - user_id=event.target_user_id, - sender=event.user_id, - room_id=event.room_id, - content=event.content, - membership=membership - ) + store_id = yield self.store.persist_event(event) # Send a PDU to all hosts who have joined the room. destinations = yield self.store.get_joined_hosts_for_room( @@ -732,78 +728,11 @@ class RoomMemberHandler(BaseHandler): yield self.hs.get_federation().handle_new_event(event) self.notifier.on_new_room_event(event, store_id) - if broadcast_msg: - yield self._inject_membership_msg( - source=event.user_id, - target=event.target_user_id, - room_id=event.room_id, - membership=event.content["membership"] - ) - - @defer.inlineCallbacks - def _do_invite_join_dance(self, room_id, joinee, target_host, content): - logger.debug("Doing remote join dance") - - # do invite join dance - federation = self.hs.get_federation() - new_event = self.event_factory.create_event( - etype=InviteJoinEvent.TYPE, - target_host=target_host, - room_id=room_id, - user_id=joinee, - content=content - ) - - new_event.destinations = [target_host] - - yield self.store.store_room( - room_id, "", is_public=False - ) - - #yield self.state_handler.handle_new_event(event) - yield federation.handle_new_event(new_event) - yield federation.get_state_for_room( - target_host, room_id - ) - - @defer.inlineCallbacks - def _inject_membership_msg(self, room_id=None, source=None, target=None, - membership=None): - # TODO this should be a different type of message, not m.text - if membership == Membership.INVITE: - body = "%s invited %s to the room." % (source, target) - elif membership == Membership.JOIN: - body = "%s joined the room." % (target) - elif membership == Membership.LEAVE: - body = "%s left the room." % (target) - else: - raise RoomError(500, "Unknown membership value %s" % membership) - - membership_json = { - "msgtype": u"m.text", - "body": body, - "membership_source": source, - "membership_target": target, - "membership": membership, - } - - msg_id = "m%s" % int(self.clock.time_msec()) - - event = self.event_factory.create_event( - etype=MessageEvent.TYPE, - room_id=room_id, - user_id="_homeserver_", - msg_id=msg_id, - content=membership_json - ) - - handler = self.hs.get_handlers().message_handler - yield handler.send_message(event, suppress_auth=True) - class RoomListHandler(BaseHandler): @defer.inlineCallbacks def get_public_room_list(self): - chunk = yield self.store.get_rooms(is_public=True, with_topics=True) + chunk = yield self.store.get_rooms(is_public=True) + # FIXME (erikj): START is no longer a valid value defer.returnValue({"start": "START", "end": "END", "chunk": chunk}) diff --git a/synapse/http/server.py b/synapse/http/server.py index c28d9a33f9..d1f99460c1 100644 --- a/synapse/http/server.py +++ b/synapse/http/server.py @@ -212,8 +212,9 @@ class ContentRepoResource(resource.Resource): """ isLeaf = True - def __init__(self, directory, auth): + def __init__(self, hs, directory, auth): resource.Resource.__init__(self) + self.hs = hs self.directory = directory self.auth = auth @@ -250,7 +251,8 @@ class ContentRepoResource(resource.Resource): file_ext = re.sub("[^a-z]", "", file_ext) suffix += "." + file_ext - file_path = os.path.join(self.directory, prefix + main_part + suffix) + file_name = prefix + main_part + suffix + file_path = os.path.join(self.directory, file_name) logger.info("User %s is uploading a file to path %s", auth_user.to_string(), file_path) @@ -259,8 +261,8 @@ class ContentRepoResource(resource.Resource): attempts = 0 while os.path.exists(file_path): main_part = random_string(24) - file_path = os.path.join(self.directory, - prefix + main_part + suffix) + file_name = prefix + main_part + suffix + file_path = os.path.join(self.directory, file_name) attempts += 1 if attempts > 25: # really? Really? raise SynapseError(500, "Unable to create file.") @@ -272,11 +274,14 @@ class ContentRepoResource(resource.Resource): # servers. # TODO: A little crude here, we could do this better. - filename = request.path.split(self.directory + "/")[1] + filename = request.path.split('/')[-1] # be paranoid filename = re.sub("[^0-9A-z.-_]", "", filename) file_path = self.directory + "/" + filename + + logger.debug("Searching for %s", file_path) + if os.path.isfile(file_path): # filename has the content type base64_contentype = filename.split(".")[1] @@ -304,6 +309,10 @@ class ContentRepoResource(resource.Resource): self._async_render(request) return server.NOT_DONE_YET + def render_OPTIONS(self, request): + respond_with_json_bytes(request, 200, {}, send_cors=True) + return server.NOT_DONE_YET + @defer.inlineCallbacks def _async_render(self, request): try: @@ -313,8 +322,13 @@ class ContentRepoResource(resource.Resource): with open(fname, "wb") as f: f.write(request.content.read()) + + # FIXME (erikj): These should use constants. + file_name = os.path.basename(fname) + url = "http://%s/matrix/content/%s" % (self.hs.hostname, file_name) + respond_with_json_bytes(request, 200, - json.dumps({"content_token": fname}), + json.dumps({"content_token": url}), send_cors=True) except CodeMessageException as e: diff --git a/synapse/rest/room.py b/synapse/rest/room.py index db8f18e8b3..f5b547b963 100644 --- a/synapse/rest/room.py +++ b/synapse/rest/room.py @@ -115,7 +115,7 @@ class RoomTopicRestServlet(RestServlet): if not data: raise SynapseError(404, "Topic not found.", errcode=Codes.NOT_FOUND) - defer.returnValue((200, json.loads(data.content))) + defer.returnValue((200, data.content)) @defer.inlineCallbacks def on_PUT(self, request, room_id): @@ -177,7 +177,7 @@ class RoomMemberRestServlet(RestServlet): if not member: raise SynapseError(404, "Member not found.", errcode=Codes.NOT_FOUND) - defer.returnValue((200, json.loads(member.content))) + defer.returnValue((200, member.content)) @defer.inlineCallbacks def on_DELETE(self, request, roomid, target_user_id): @@ -193,7 +193,7 @@ class RoomMemberRestServlet(RestServlet): ) handler = self.handlers.room_member_handler - yield handler.change_membership(event, broadcast_msg=True) + yield handler.change_membership(event) defer.returnValue((200, "")) @defer.inlineCallbacks @@ -220,7 +220,7 @@ class RoomMemberRestServlet(RestServlet): ) handler = self.handlers.room_member_handler - yield handler.change_membership(event, broadcast_msg=True) + yield handler.change_membership(event) defer.returnValue((200, "")) @@ -287,25 +287,28 @@ class FeedbackRestServlet(RestServlet): feedback_type): user = yield (self.auth.get_user_by_req(request)) - if feedback_type not in Feedback.LIST: - raise SynapseError(400, "Bad feedback type.", - errcode=Codes.BAD_JSON) - - msg_handler = self.handlers.message_handler - feedback = yield msg_handler.get_feedback( - room_id=urllib.unquote(room_id), - msg_sender_id=msg_sender_id, - msg_id=msg_id, - user_id=user.to_string(), - fb_sender_id=fb_sender_id, - fb_type=feedback_type - ) - - if not feedback: - raise SynapseError(404, "Feedback not found.", - errcode=Codes.NOT_FOUND) + # TODO (erikj): Implement this? + raise NotImplementedError("Getting feedback is not supported") - defer.returnValue((200, json.loads(feedback.content))) +# if feedback_type not in Feedback.LIST: +# raise SynapseError(400, "Bad feedback type.", +# errcode=Codes.BAD_JSON) +# +# msg_handler = self.handlers.message_handler +# feedback = yield msg_handler.get_feedback( +# room_id=urllib.unquote(room_id), +# msg_sender_id=msg_sender_id, +# msg_id=msg_id, +# user_id=user.to_string(), +# fb_sender_id=fb_sender_id, +# fb_type=feedback_type +# ) +# +# if not feedback: +# raise SynapseError(404, "Feedback not found.", +# errcode=Codes.NOT_FOUND) +# +# defer.returnValue((200, json.loads(feedback.content))) @defer.inlineCallbacks def on_PUT(self, request, room_id, sender_id, msg_id, fb_sender_id, @@ -382,6 +385,21 @@ class RoomMessageListRestServlet(RestServlet): defer.returnValue((200, msgs)) +class RoomTriggerBackfill(RestServlet): + PATTERN = client_path_pattern("/rooms/(?P<room_id>[^/]*)/backfill$") + + @defer.inlineCallbacks + def on_GET(self, request, room_id): + remote_server = urllib.unquote(request.args["remote"][0]) + room_id = urllib.unquote(room_id) + limit = int(request.args["limit"][0]) + + handler = self.handlers.federation_handler + events = yield handler.backfill(remote_server, room_id, limit) + + res = [event.get_dict() for event in events] + defer.returnValue((200, res)) + def _parse_json(request): try: content = json.loads(request.content.read()) @@ -402,3 +420,4 @@ def register_servlets(hs, http_server): RoomMemberListRestServlet(hs).register(http_server) RoomMessageListRestServlet(hs).register(http_server) JoinRoomAliasServlet(hs).register(http_server) + RoomTriggerBackfill(hs).register(http_server) diff --git a/synapse/server.py b/synapse/server.py index d4c2481483..c5b0a32757 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -159,7 +159,7 @@ class HomeServer(BaseHomeServer): return DataStore(self) def build_event_factory(self): - return EventFactory() + return EventFactory(self) def build_handlers(self): return Handlers(self) diff --git a/synapse/state.py b/synapse/state.py index 4f8b4d9760..ca8e1ca630 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -86,7 +86,7 @@ class StateHandler(object): else: event.depth = 0 - current_state = yield self.store.get_current_state( + current_state = yield self.store.get_current_state_pdu( key.context, key.type, key.state_key ) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 5d5b5f7c44..d06033b980 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -13,30 +13,35 @@ # See the License for the specific language governing permissions and # limitations under the License. +from twisted.internet import defer from synapse.api.events.room import ( RoomMemberEvent, MessageEvent, RoomTopicEvent, FeedbackEvent, - RoomConfigEvent + RoomConfigEvent, RoomNameEvent, ) +from synapse.util.logutils import log_function + from .directory import DirectoryStore from .feedback import FeedbackStore -from .message import MessageStore from .presence import PresenceStore from .profile import ProfileStore from .registration import RegistrationStore from .room import RoomStore from .roommember import RoomMemberStore -from .roomdata import RoomDataStore from .stream import StreamStore from .pdu import StatePduStore, PduStore from .transactions import TransactionStore import json +import logging import os -class DataStore(RoomDataStore, RoomMemberStore, MessageStore, RoomStore, +logger = logging.getLogger(__name__) + + +class DataStore(RoomMemberStore, RoomStore, RegistrationStore, StreamStore, ProfileStore, FeedbackStore, PresenceStore, PduStore, StatePduStore, TransactionStore, DirectoryStore): @@ -44,51 +49,147 @@ class DataStore(RoomDataStore, RoomMemberStore, MessageStore, RoomStore, def __init__(self, hs): super(DataStore, self).__init__(hs) self.event_factory = hs.get_event_factory() + self.hs = hs - def persist_event(self, event): - if event.type == MessageEvent.TYPE: - return self.store_message( - user_id=event.user_id, - room_id=event.room_id, - msg_id=event.msg_id, - content=json.dumps(event.content) - ) - elif event.type == RoomMemberEvent.TYPE: - return self.store_room_member( - user_id=event.target_user_id, - sender=event.user_id, - room_id=event.room_id, - content=event.content, - membership=event.content["membership"] - ) + self.min_token_deferred = self._get_min_token() + self.min_token = None + + @defer.inlineCallbacks + @log_function + def persist_event(self, event, backfilled=False): + if event.type == RoomMemberEvent.TYPE: + yield self._store_room_member(event) elif event.type == FeedbackEvent.TYPE: - return self.store_feedback( - room_id=event.room_id, - msg_id=event.msg_id, - msg_sender_id=event.msg_sender_id, - fb_sender_id=event.user_id, - fb_type=event.feedback_type, - content=json.dumps(event.content) - ) + yield self._store_feedback(event) +# elif event.type == RoomConfigEvent.TYPE: +# yield self._store_room_config(event) + elif event.type == RoomNameEvent.TYPE: + yield self._store_room_name(event) elif event.type == RoomTopicEvent.TYPE: - return self.store_room_data( - room_id=event.room_id, - etype=event.type, - state_key=event.state_key, - content=json.dumps(event.content) + yield self._store_room_topic(event) + + ret = yield self._store_event(event, backfilled) + defer.returnValue(ret) + + @defer.inlineCallbacks + def get_event(self, event_id): + events_dict = yield self._simple_select_one( + "events", + {"event_id": event_id}, + [ + "event_id", + "type", + "sender", + "room_id", + "content", + "unrecognized_keys" + ], + ) + + event = self._parse_event_from_row(events_dict) + defer.returnValue(event) + + @defer.inlineCallbacks + @log_function + def _store_event(self, event, backfilled): + # FIXME (erikj): This should be removed when we start amalgamating + # event and pdu storage + yield self.hs.get_federation().fill_out_prev_events(event) + + vals = { + "topological_ordering": event.depth, + "event_id": event.event_id, + "type": event.type, + "room_id": event.room_id, + "content": json.dumps(event.content), + "processed": True, + } + + if hasattr(event, "outlier"): + vals["outlier"] = event.outlier + else: + vals["outlier"] = False + + if backfilled: + if not self.min_token_deferred.called: + yield self.min_token_deferred + self.min_token -= 1 + vals["stream_ordering"] = self.min_token + + unrec = { + k: v + for k, v in event.get_full_dict().items() + if k not in vals.keys() + } + vals["unrecognized_keys"] = json.dumps(unrec) + + try: + yield self._simple_insert("events", vals) + except: + logger.exception( + "Failed to persist, probably duplicate: %s", + event.event_id + ) + return + + if not backfilled and hasattr(event, "state_key"): + vals = { + "event_id": event.event_id, + "room_id": event.room_id, + "type": event.type, + "state_key": event.state_key, + } + + if hasattr(event, "prev_state"): + vals["prev_state"] = event.prev_state + + yield self._simple_insert("state_events", vals) + + yield self._simple_insert( + "current_state_events", + { + "event_id": event.event_id, + "room_id": event.room_id, + "type": event.type, + "state_key": event.state_key, + } ) - elif event.type == RoomConfigEvent.TYPE: - if "visibility" in event.content: - visibility = event.content["visibility"] - return self.store_room_config( - room_id=event.room_id, - visibility=visibility - ) + latest = yield self.get_room_events_max_id() + defer.returnValue(latest) + + @defer.inlineCallbacks + def get_current_state(self, room_id, event_type=None, state_key=""): + sql = ( + "SELECT e.* FROM events as e " + "INNER JOIN current_state_events as c ON e.event_id = c.event_id " + "INNER JOIN state_events as s ON e.event_id = s.event_id " + "WHERE c.room_id = ? " + ) + + if event_type: + sql += " AND s.type = ? AND s.state_key = ? " + args = (room_id, event_type, state_key) else: - raise NotImplementedError( - "Don't know how to persist type=%s" % event.type - ) + args = (room_id, ) + + results = yield self._execute_and_decode(sql, *args) + + defer.returnValue([self._parse_event_from_row(r) for r in results]) + + @defer.inlineCallbacks + def _get_min_token(self): + row = yield self._execute( + None, + "SELECT MIN(stream_ordering) FROM events" + ) + + self.min_token = row[0][0] if row and row[0] and row[0][0] else -1 + self.min_token = min(self.min_token, -1) + + logger.debug("min_token is: %s", self.min_token) + + defer.returnValue(self.min_token) def schema_path(schema): diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index bf1800f4bf..75aab2d3b9 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -12,7 +12,6 @@ # 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 logging from twisted.internet import defer @@ -20,6 +19,9 @@ from twisted.internet import defer from synapse.api.errors import StoreError import collections +import copy +import json + logger = logging.getLogger(__name__) @@ -29,6 +31,7 @@ class SQLBaseStore(object): def __init__(self, hs): self.hs = hs self._db_pool = hs.get_db_pool() + self.event_factory = hs.get_event_factory() self._clock = hs.get_clock() def cursor_to_dict(self, cursor): @@ -57,14 +60,22 @@ class SQLBaseStore(object): The result of decoder(results) """ logger.debug( - "[SQL] %s Args=%s Func=%s", query, args, decoder.__name__ + "[SQL] %s Args=%s Func=%s", + query, args, decoder.__name__ if decoder else None ) def interaction(txn): cursor = txn.execute(query, args) - return decoder(cursor) + if decoder: + return decoder(cursor) + else: + return cursor.fetchall() + return self._db_pool.runInteraction(interaction) + def _execute_and_decode(self, query, *args): + return self._execute(self.cursor_to_dict, query, *args) + # "Simple" SQL API methods that operate on a single table with no JOINs, # no complex WHERE clauses, just a dict of values for columns. @@ -281,6 +292,22 @@ class SQLBaseStore(object): return self._db_pool.runInteraction(func) + def _parse_event_from_row(self, row_dict): + d = copy.deepcopy({k: v for k, v in row_dict.items() if v}) + + d.pop("stream_ordering", None) + d.pop("topological_ordering", None) + d.pop("processed", None) + + d.update(json.loads(row_dict["unrecognized_keys"])) + d["content"] = json.loads(d["content"]) + del d["unrecognized_keys"] + + return self.event_factory.create_event( + etype=d["type"], + **d + ) + class Table(object): """ A base class used to store information about a particular table. diff --git a/synapse/storage/feedback.py b/synapse/storage/feedback.py index 9bd562c762..e60f98d1e1 100644 --- a/synapse/storage/feedback.py +++ b/synapse/storage/feedback.py @@ -13,6 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from twisted.internet import defer + from ._base import SQLBaseStore, Table from synapse.api.events.room import FeedbackEvent @@ -22,54 +24,28 @@ import json class FeedbackStore(SQLBaseStore): - def store_feedback(self, room_id, msg_id, msg_sender_id, - fb_sender_id, fb_type, content): - return self._simple_insert(FeedbackTable.table_name, dict( - room_id=room_id, - msg_id=msg_id, - msg_sender_id=msg_sender_id, - fb_sender_id=fb_sender_id, - fb_type=fb_type, - content=content, - )) - - def get_feedback(self, room_id=None, msg_id=None, msg_sender_id=None, - fb_sender_id=None, fb_type=None): - query = FeedbackTable.select_statement( - "msg_sender_id = ? AND room_id = ? AND msg_id = ? " + - "AND fb_sender_id = ? AND feedback_type = ? " + - "ORDER BY id DESC LIMIT 1") - return self._execute( - FeedbackTable.decode_single_result, - query, msg_sender_id, room_id, msg_id, fb_sender_id, fb_type, + def _store_feedback(self, event): + return self._simple_insert("feedback", { + "event_id": event.event_id, + "feedback_type": event.feedback_type, + "room_id": event.room_id, + "target_event_id": event.target_event, + "sender": event.user_id, + }) + + @defer.inlineCallbacks + def get_feedback_for_event(self, event_id): + sql = ( + "SELECT events.* FROM events INNER JOIN feedback " + "ON events.event_id = feedback.event_id " + "WHERE feedback.target_event_id = ? " ) - def get_max_feedback_id(self): - return self._simple_max_id(FeedbackTable.table_name) - - -class FeedbackTable(Table): - table_name = "feedback" + rows = yield self._execute_and_decode(sql, event_id) - fields = [ - "id", - "content", - "feedback_type", - "fb_sender_id", - "msg_id", - "room_id", - "msg_sender_id" - ] - - class EntryType(collections.namedtuple("FeedbackEntry", fields)): - - def as_event(self, event_factory): - return event_factory.create_event( - etype=FeedbackEvent.TYPE, - room_id=self.room_id, - msg_id=self.msg_id, - msg_sender_id=self.msg_sender_id, - user_id=self.fb_sender_id, - feedback_type=self.feedback_type, - content=json.loads(self.content), - ) + defer.returnValue( + [ + self._parse_event_from_row(r) + for r in rows + ] + ) diff --git a/synapse/storage/message.py b/synapse/storage/message.py deleted file mode 100644 index 7bb69c1384..0000000000 --- a/synapse/storage/message.py +++ /dev/null @@ -1,81 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 matrix.org -# -# 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. - -from ._base import SQLBaseStore, Table -from synapse.api.events.room import MessageEvent - -import collections -import json - - -class MessageStore(SQLBaseStore): - - def get_message(self, user_id, room_id, msg_id): - """Get a message from the store. - - Args: - user_id (str): The ID of the user who sent the message. - room_id (str): The room the message was sent in. - msg_id (str): The unique ID for this user/room combo. - """ - query = MessagesTable.select_statement( - "user_id = ? AND room_id = ? AND msg_id = ? " + - "ORDER BY id DESC LIMIT 1") - return self._execute( - MessagesTable.decode_single_result, - query, user_id, room_id, msg_id, - ) - - def store_message(self, user_id, room_id, msg_id, content): - """Store a message in the store. - - Args: - user_id (str): The ID of the user who sent the message. - room_id (str): The room the message was sent in. - msg_id (str): The unique ID for this user/room combo. - content (str): The content of the message (JSON) - """ - return self._simple_insert(MessagesTable.table_name, dict( - user_id=user_id, - room_id=room_id, - msg_id=msg_id, - content=content, - )) - - def get_max_message_id(self): - return self._simple_max_id(MessagesTable.table_name) - - -class MessagesTable(Table): - table_name = "messages" - - fields = [ - "id", - "user_id", - "room_id", - "msg_id", - "content" - ] - - class EntryType(collections.namedtuple("MessageEntry", fields)): - - def as_event(self, event_factory): - return event_factory.create_event( - etype=MessageEvent.TYPE, - room_id=self.room_id, - user_id=self.user_id, - msg_id=self.msg_id, - content=json.loads(self.content), - ) diff --git a/synapse/storage/pdu.py b/synapse/storage/pdu.py index 13adc581e1..7655f43ede 100644 --- a/synapse/storage/pdu.py +++ b/synapse/storage/pdu.py @@ -13,6 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from twisted.internet import defer + from ._base import SQLBaseStore, Table, JoinHelper from synapse.util.logutils import log_function @@ -319,6 +321,7 @@ class PduStore(SQLBaseStore): return [(row[0], row[1], row[2]) for row in results] + @defer.inlineCallbacks def get_oldest_pdus_in_context(self, context): """Get a list of Pdus that we haven't backfilled beyond yet (and haven't seen). This list is used when we want to backfill backwards and is the @@ -331,17 +334,14 @@ class PduStore(SQLBaseStore): Returns: list: A list of PduIdTuple. """ - return self._db_pool.runInteraction( - self._get_oldest_pdus_in_context, context - ) - - def _get_oldest_pdus_in_context(self, txn, context): - txn.execute( + results = yield self._execute( + None, "SELECT pdu_id, origin FROM %(back)s WHERE context = ?" % {"back": PduBackwardExtremitiesTable.table_name, }, - (context,) + context ) - return [PduIdTuple(i, o) for i, o in txn.fetchall()] + + defer.returnValue([PduIdTuple(i, o) for i, o in results]) def is_pdu_new(self, pdu_id, origin, context, depth): """For a given Pdu, try and figure out if it's 'new', i.e., if it's @@ -580,7 +580,7 @@ class StatePduStore(SQLBaseStore): txn.execute(query, query_args) - def get_current_state(self, context, pdu_type, state_key): + def get_current_state_pdu(self, context, pdu_type, state_key): """For a given context, pdu_type, state_key 3-tuple, return what is currently considered the current state. @@ -595,10 +595,10 @@ class StatePduStore(SQLBaseStore): """ return self._db_pool.runInteraction( - self._get_current_state, context, pdu_type, state_key + self._get_current_state_pdu, context, pdu_type, state_key ) - def _get_current_state(self, txn, context, pdu_type, state_key): + def _get_current_state_pdu(self, txn, context, pdu_type, state_key): return self._get_current_interaction(txn, context, pdu_type, state_key) def _get_current_interaction(self, txn, context, pdu_type, state_key): diff --git a/synapse/storage/room.py b/synapse/storage/room.py index a97162831b..22f2dcca45 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -76,49 +76,80 @@ class RoomStore(SQLBaseStore): ) @defer.inlineCallbacks - def get_rooms(self, is_public, with_topics): + def get_rooms(self, is_public): """Retrieve a list of all public rooms. Args: is_public (bool): True if the rooms returned should be public. - with_topics (bool): True to include the current topic for the room - in the response. Returns: - A list of room dicts containing at least a "room_id" key, and a - "topic" key if one is set and with_topic=True. + A list of room dicts containing at least a "room_id" key, a + "topic" key if one is set, and a "name" key if one is set """ - room_data_type = RoomTopicEvent.TYPE - public = 1 if is_public else 0 - - latest_topic = ("SELECT max(room_data.id) FROM room_data WHERE " - + "room_data.type = ? GROUP BY room_id") - - query = ("SELECT rooms.*, room_data.content, room_alias FROM rooms " - + "LEFT JOIN " - + "room_aliases ON room_aliases.room_id = rooms.room_id " - + "LEFT JOIN " - + "room_data ON rooms.room_id = room_data.room_id WHERE " - + "(room_data.id IN (" + latest_topic + ") " - + "OR room_data.id IS NULL) AND rooms.is_public = ?") - - res = yield self._execute( - self.cursor_to_dict, query, room_data_type, public + + topic_subquery = ( + "SELECT topics.event_id as event_id, " + "topics.room_id as room_id, topic " + "FROM topics " + "INNER JOIN current_state_events as c " + "ON c.event_id = topics.event_id " ) - # return only the keys the specification expects - ret_keys = ["room_id", "topic", "room_alias"] + name_subquery = ( + "SELECT room_names.event_id as event_id, " + "room_names.room_id as room_id, name " + "FROM room_names " + "INNER JOIN current_state_events as c " + "ON c.event_id = room_names.event_id " + ) - # extract topic from the json (icky) FIXME - for i, room_row in enumerate(res): - try: - content_json = json.loads(room_row["content"]) - room_row["topic"] = content_json["topic"] - except: - pass # no topic set - # filter the dict based on ret_keys - res[i] = {k: v for k, v in room_row.iteritems() if k in ret_keys} + # We use non printing ascii character US ( ) as a seperator + sql = ( + "SELECT r.room_id, n.name, t.topic, " + "group_concat(a.room_alias, ' ') " + "FROM rooms AS r " + "LEFT JOIN (%(topic)s) AS t ON t.room_id = r.room_id " + "LEFT JOIN (%(name)s) AS n ON n.room_id = r.room_id " + "INNER JOIN room_aliases AS a ON a.room_id = r.room_id " + "WHERE r.is_public = ? " + "GROUP BY r.room_id " + ) % { + "topic": topic_subquery, + "name": name_subquery, + } + + rows = yield self._execute(None, sql, is_public) + + ret = [ + { + "room_id": r[0], + "name": r[1], + "topic": r[2], + "aliases": r[3].split(" "), + } + for r in rows + ] + + defer.returnValue(ret) + + def _store_room_topic(self, event): + return self._simple_insert( + "topics", + { + "event_id": event.event_id, + "room_id": event.room_id, + "topic": event.topic, + } + ) - defer.returnValue(res) + def _store_room_name(self, event): + return self._simple_insert( + "room_names", + { + "event_id": event.event_id, + "room_id": event.room_id, + "name": event.name, + } + ) class RoomsTable(Table): diff --git a/synapse/storage/roomdata.py b/synapse/storage/roomdata.py deleted file mode 100644 index cc04d1ba14..0000000000 --- a/synapse/storage/roomdata.py +++ /dev/null @@ -1,85 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 matrix.org -# -# 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. - -from ._base import SQLBaseStore, Table - -import collections -import json - - -class RoomDataStore(SQLBaseStore): - - """Provides various CRUD operations for Room Events. """ - - def get_room_data(self, room_id, etype, state_key=""): - """Retrieve the data stored under this type and state_key. - - Args: - room_id (str) - etype (str) - state_key (str) - Returns: - namedtuple: Or None if nothing exists at this path. - """ - query = RoomDataTable.select_statement( - "room_id = ? AND type = ? AND state_key = ? " - "ORDER BY id DESC LIMIT 1" - ) - return self._execute( - RoomDataTable.decode_single_result, - query, room_id, etype, state_key, - ) - - def store_room_data(self, room_id, etype, state_key="", content=None): - """Stores room specific data. - - Args: - room_id (str) - etype (str) - state_key (str) - data (str)- The data to store for this path in JSON. - Returns: - The store ID for this data. - """ - return self._simple_insert(RoomDataTable.table_name, dict( - etype=etype, - state_key=state_key, - room_id=room_id, - content=content, - )) - - def get_max_room_data_id(self): - return self._simple_max_id(RoomDataTable.table_name) - - -class RoomDataTable(Table): - table_name = "room_data" - - fields = [ - "id", - "room_id", - "type", - "state_key", - "content" - ] - - class EntryType(collections.namedtuple("RoomDataEntry", fields)): - - def as_event(self, event_factory): - return event_factory.create_event( - etype=self.type, - room_id=self.room_id, - content=json.loads(self.content), - ) diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index c45d128f1b..89c87290cf 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -31,6 +31,38 @@ logger = logging.getLogger(__name__) class RoomMemberStore(SQLBaseStore): + @defer.inlineCallbacks + def _store_room_member(self, event): + """Store a room member in the database. + """ + domain = self.hs.parse_userid(event.target_user_id).domain + + yield self._simple_insert( + "room_memberships", + { + "event_id": event.event_id, + "user_id": event.target_user_id, + "sender": event.user_id, + "room_id": event.room_id, + "membership": event.membership, + } + ) + + # Update room hosts table + if event.membership == Membership.JOIN: + sql = ( + "INSERT OR IGNORE INTO room_hosts (room_id, host) " + "VALUES (?, ?)" + ) + yield self._execute(None, sql, event.room_id, domain) + else: + sql = ( + "DELETE FROM room_hosts WHERE room_id = ? AND host = ?" + ) + + yield self._execute(None, sql, event.room_id, domain) + + @defer.inlineCallbacks def get_room_member(self, user_id, room_id): """Retrieve the current state of a room member. @@ -38,36 +70,15 @@ class RoomMemberStore(SQLBaseStore): user_id (str): The member's user ID. room_id (str): The room the member is in. Returns: - namedtuple: The room member from the database, or None if this - member does not exist. + Deferred: Results in a MembershipEvent or None. """ - query = RoomMemberTable.select_statement( - "room_id = ? AND user_id = ? ORDER BY id DESC LIMIT 1") - return self._execute( - RoomMemberTable.decode_single_result, - query, room_id, user_id, - ) + rows = yield self._get_members_by_dict({ + "e.room_id": room_id, + "m.user_id": user_id, + }) - def store_room_member(self, user_id, sender, room_id, membership, content): - """Store a room member in the database. + defer.returnValue(rows[0] if rows else None) - Args: - user_id (str): The member's user ID. - room_id (str): The room in relation to the member. - membership (synapse.api.constants.Membership): The new membership - state. - content (dict): The content of the membership (JSON). - """ - content_json = json.dumps(content) - return self._simple_insert(RoomMemberTable.table_name, dict( - user_id=user_id, - sender=sender, - room_id=room_id, - membership=membership, - content=content_json, - )) - - @defer.inlineCallbacks def get_room_members(self, room_id, membership=None): """Retrieve the current room member list for a room. @@ -79,17 +90,12 @@ class RoomMemberStore(SQLBaseStore): Returns: list of namedtuples representing the members in this room. """ - query = RoomMemberTable.select_statement( - "id IN (SELECT MAX(id) FROM " + RoomMemberTable.table_name - + " WHERE room_id = ? GROUP BY user_id)" - ) - res = yield self._execute( - RoomMemberTable.decode_results, query, room_id, - ) - # strip memberships which don't match + + where = {"m.room_id": room_id} if membership: - res = [entry for entry in res if entry.membership == membership] - defer.returnValue(res) + where["m.membership"] = membership + + return self._get_members_by_dict(where) def get_rooms_for_user_where_membership_is(self, user_id, membership_list): """ Get all the rooms for this user where the membership for this user @@ -106,70 +112,40 @@ class RoomMemberStore(SQLBaseStore): return defer.succeed(None) args = [user_id] - membership_placeholder = ["membership=?"] * len(membership_list) - where_membership = "(" + " OR ".join(membership_placeholder) + ")" - for membership in membership_list: - args.append(membership) - - # sub-select finds the row ID for the most recent (i.e. current) - # state change of this user per room, then the outer select finds those - query = ("SELECT room_id, membership FROM room_memberships" - + " WHERE id IN (SELECT MAX(id) FROM room_memberships" - + " WHERE user_id=? GROUP BY room_id)" - + " AND " + where_membership) - return self._execute( - self.cursor_to_dict, query, *args - ) + args.extend(membership_list) - @defer.inlineCallbacks - def get_joined_hosts_for_room(self, room_id): - query = RoomMemberTable.select_statement( - "id IN (SELECT MAX(id) FROM " + RoomMemberTable.table_name - + " WHERE room_id = ? GROUP BY user_id)" - ) - - res = yield self._execute( - RoomMemberTable.decode_results, query, room_id, + where_clause = "user_id = ? AND (%s)" % ( + " OR ".join(["membership = ?" for _ in membership_list]), ) - def host_from_user_id_string(user_id): - domain = UserID.from_string(entry.user_id, self.hs).domain - return domain - - # strip memberships which don't match - hosts = [ - host_from_user_id_string(entry.user_id) - for entry in res - if entry.membership == Membership.JOIN - ] + return self._get_members_query(where_clause, args) - logger.debug("Returning hosts: %s from results: %s", hosts, res) - - defer.returnValue(hosts) - - def get_max_room_member_id(self): - return self._simple_max_id(RoomMemberTable.table_name) - - -class RoomMemberTable(Table): - table_name = "room_memberships" - - fields = [ - "id", - "user_id", - "sender", - "room_id", - "membership", - "content" - ] + def get_joined_hosts_for_room(self, room_id): + return self._simple_select_onecol( + "room_hosts", + {"room_id": room_id}, + "host" + ) - class EntryType(collections.namedtuple("RoomMemberEntry", fields)): + def _get_members_by_dict(self, where_dict): + clause = " AND ".join("%s = ?" % k for k in where_dict.keys()) + vals = where_dict.values() + return self._get_members_query(clause, vals) - def as_event(self, event_factory): - return event_factory.create_event( - etype=RoomMemberEvent.TYPE, - room_id=self.room_id, - target_user_id=self.user_id, - user_id=self.sender, - content=json.loads(self.content), - ) + @defer.inlineCallbacks + def _get_members_query(self, where_clause, where_values): + sql = ( + "SELECT e.* FROM events as e " + "INNER JOIN room_memberships as m " + "ON e.event_id = m.event_id " + "INNER JOIN current_state_events as c " + "ON m.event_id = c.event_id " + "WHERE %s " + ) % (where_clause,) + + rows = yield self._execute_and_decode(sql, *where_values) + + logger.debug("_get_members_query Got rows %s", rows) + + results = [self._parse_event_from_row(r) for r in rows] + defer.returnValue(results) diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql index 77096546b2..39a1ed703e 100644 --- a/synapse/storage/schema/im.sql +++ b/synapse/storage/schema/im.sql @@ -12,43 +12,71 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -CREATE TABLE IF NOT EXISTS rooms( - room_id TEXT PRIMARY KEY NOT NULL, - is_public INTEGER, - creator TEXT + +CREATE TABLE IF NOT EXISTS events( + stream_ordering INTEGER PRIMARY KEY AUTOINCREMENT, + topological_ordering INTEGER NOT NULL, + event_id TEXT NOT NULL, + type TEXT NOT NULL, + room_id TEXT NOT NULL, + content TEXT NOT NULL, + unrecognized_keys TEXT, + processed BOOL NOT NULL, + outlier BOOL NOT NULL, + CONSTRAINT ev_uniq UNIQUE (event_id) ); -CREATE TABLE IF NOT EXISTS room_memberships( - id INTEGER PRIMARY KEY AUTOINCREMENT, - user_id TEXT NOT NULL, -- no foreign key to users table, it could be an id belonging to another home server - sender TEXT NOT NULL, +CREATE TABLE IF NOT EXISTS state_events( + event_id TEXT NOT NULL, room_id TEXT NOT NULL, - membership TEXT NOT NULL, - content TEXT NOT NULL + type TEXT NOT NULL, + state_key TEXT NOT NULL, + prev_state TEXT ); -CREATE TABLE IF NOT EXISTS messages( - id INTEGER PRIMARY KEY AUTOINCREMENT, - user_id TEXT, - room_id TEXT, - msg_id TEXT, - content TEXT +CREATE TABLE IF NOT EXISTS current_state_events( + event_id TEXT NOT NULL, + room_id TEXT NOT NULL, + type TEXT NOT NULL, + state_key TEXT NOT NULL, + CONSTRAINT curr_uniq UNIQUE (room_id, type, state_key) ON CONFLICT REPLACE +); + +CREATE TABLE IF NOT EXISTS room_memberships( + event_id TEXT NOT NULL, + user_id TEXT NOT NULL, + sender TEXT NOT NULL, + room_id TEXT NOT NULL, + membership TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS feedback( - id INTEGER PRIMARY KEY AUTOINCREMENT, - content TEXT, + event_id TEXT NOT NULL, feedback_type TEXT, - fb_sender_id TEXT, - msg_id TEXT, - room_id TEXT, - msg_sender_id TEXT + target_event_id TEXT, + sender TEXT, + room_id TEXT ); -CREATE TABLE IF NOT EXISTS room_data( - id INTEGER PRIMARY KEY AUTOINCREMENT, +CREATE TABLE IF NOT EXISTS topics( + event_id TEXT NOT NULL, room_id TEXT NOT NULL, - type TEXT NOT NULL, - state_key TEXT NOT NULL, - content TEXT + topic TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS room_names( + event_id TEXT NOT NULL, + room_id TEXT NOT NULL, + name TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS rooms( + room_id TEXT PRIMARY KEY NOT NULL, + is_public INTEGER, + creator TEXT +); + +CREATE TABLE IF NOT EXISTS room_hosts( + room_id TEXT NOT NULL, + host TEXT NOT NULL ); diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 47a1f2c45a..87ae961ccd 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -13,267 +13,287 @@ # See the License for the specific language governing permissions and # limitations under the License. +""" This module is responsible for getting events from the DB for pagination +and event streaming. + +The order it returns events in depend on whether we are streaming forwards or +are paginating backwards. We do this because we want to handle out of order +messages nicely, while still returning them in the correct order when we +paginate bacwards. + +This is implemented by keeping two ordering columns: stream_ordering and +topological_ordering. Stream ordering is basically insertion/received order +(except for events from backfill requests). The topolgical_ordering is a +weak ordering of events based on the pdu graph. + +This means that we have to have two different types of tokens, depending on +what sort order was used: + - stream tokens are of the form: "s%d", which maps directly to the column + - topological tokems: "t%d-%d", where the integers map to the topological + and stream ordering columns respectively. +""" + +from twisted.internet import defer from ._base import SQLBaseStore -from .message import MessagesTable -from .feedback import FeedbackTable -from .roomdata import RoomDataTable -from .roommember import RoomMemberTable +from synapse.api.errors import SynapseError +from synapse.api.constants import Membership +from synapse.util.logutils import log_function import json import logging + logger = logging.getLogger(__name__) +MAX_STREAM_SIZE = 1000 + + +_STREAM_TOKEN = "stream" +_TOPOLOGICAL_TOKEN = "topological" + + +def _parse_stream_token(string): + try: + if string[0] != 's': + raise + return int(string[1:]) + except: + raise SynapseError(400, "Invalid token") + + +def _parse_topological_token(string): + try: + if string[0] != 't': + raise + parts = string[1:].split('-', 1) + return (int(parts[0]), int(parts[1])) + except: + raise SynapseError(400, "Invalid token") + + +def is_stream_token(string): + try: + _parse_stream_token(string) + return True + except: + return False + + +def is_topological_token(string): + try: + _parse_topological_token(string) + return True + except: + return False + + +def _get_token_bound(token, comparison): + try: + s = _parse_stream_token(token) + return "%s %s %d" % ("stream_ordering", comparison, s) + except: + pass + + try: + top, stream = _parse_topological_token(token) + return "%s %s %d AND %s %s %d" % ( + "topological_ordering", comparison, top, + "stream_ordering", comparison, stream, + ) + except: + pass + + raise SynapseError(400, "Invalid token") + + class StreamStore(SQLBaseStore): + @log_function + def get_room_events(self, user_id, from_key, to_key, room_id, limit=0, + direction='f', with_feedback=False): + # We deal with events request in two different ways depending on if + # this looks like an /events request or a pagination request. + is_events = ( + direction == 'f' + and user_id + and is_stream_token(from_key) + and to_key and is_stream_token(to_key) + ) - def get_message_stream(self, user_id, from_key, to_key, room_id, limit=0, - with_feedback=False): - """Get all messages for this user between the given keys. - - Args: - user_id (str): The user who is requesting messages. - from_key (int): The ID to start returning results from (exclusive). - to_key (int): The ID to stop returning results (exclusive). - room_id (str): Gets messages only for this room. Can be None, in - which case all room messages will be returned. - Returns: - A tuple of rows (list of namedtuples), new_id(int) - """ - if with_feedback and room_id: # with fb MUST specify a room ID - return self._db_pool.runInteraction( - self._get_message_rows_with_feedback, - user_id, from_key, to_key, room_id, limit + if is_events: + return self.get_room_events_stream( + user_id=user_id, + from_key=from_key, + to_key=to_key, + room_id=room_id, + limit=limit, + with_feedback=with_feedback, ) else: - return self._db_pool.runInteraction( - self._get_message_rows, - user_id, from_key, to_key, room_id, limit + return self.paginate_room_events( + from_key=from_key, + to_key=to_key, + room_id=room_id, + limit=limit, + with_feedback=with_feedback, ) - def _get_message_rows(self, txn, user_id, from_pkey, to_pkey, room_id, - limit): - # work out which rooms this user is joined in on and join them with - # the room id on the messages table, bounded by the specified pkeys + @defer.inlineCallbacks + @log_function + def get_room_events_stream(self, user_id, from_key, to_key, room_id, + limit=0, with_feedback=False): + # TODO (erikj): Handle compressed feedback - # get all messages where the *current* membership state is 'join' for - # this user in that room. - query = ("SELECT messages.* FROM messages WHERE ? IN" - + " (SELECT membership from room_memberships WHERE user_id=?" - + " AND room_id = messages.room_id ORDER BY id DESC LIMIT 1)") - query_args = ["join", user_id] + current_room_membership_sql = ( + "SELECT m.room_id FROM room_memberships as m " + "INNER JOIN current_state_events as c ON m.event_id = c.event_id " + "WHERE m.user_id = ?" + ) - if room_id: - query += " AND messages.room_id=?" - query_args.append(room_id) + # We also want to get any membership events about that user, e.g. + # invites or leave notifications. + membership_sql = ( + "SELECT m.event_id FROM room_memberships as m " + "INNER JOIN current_state_events as c ON m.event_id = c.event_id " + "WHERE m.user_id = ? " + ) - (query, query_args) = self._append_stream_operations( - "messages", query, query_args, from_pkey, to_pkey, limit=limit + if limit: + limit = max(limit, MAX_STREAM_SIZE) + else: + limit = MAX_STREAM_SIZE + + # From and to keys should be integers from ordering. + from_id = _parse_stream_token(from_key) + to_id = _parse_stream_token(to_key) + + if from_key == to_key: + defer.returnValue(([], to_key)) + return + + sql = ( + "SELECT * FROM events as e WHERE " + "((room_id IN (%(current)s)) OR " + "(event_id IN (%(invites)s))) " + "AND e.stream_ordering > ? AND e.stream_ordering < ? " + "AND e.outlier = 0 " + "ORDER BY stream_ordering ASC LIMIT %(limit)d " + ) % { + "current": current_room_membership_sql, + "invites": membership_sql, + "limit": limit + } + + rows = yield self._execute_and_decode( + sql, + user_id, user_id, from_id, to_id ) - cursor = txn.execute(query, query_args) - return self._as_events(cursor, MessagesTable, from_pkey) + ret = [self._parse_event_from_row(r) for r in rows] - def _get_message_rows_with_feedback(self, txn, user_id, from_pkey, to_pkey, - room_id, limit): - # this col represents the compressed feedback JSON as per spec - compressed_feedback_col = ( - "'[' || group_concat('{\"sender_id\":\"' || f.fb_sender_id" - + " || '\",\"feedback_type\":\"' || f.feedback_type" - + " || '\",\"content\":' || f.content || '}') || ']'" - ) + if rows: + key = "s%d" % max([r["stream_ordering"] for r in rows]) + else: + # Assume we didn't get anything because there was nothing to get. + key = to_key + + defer.returnValue((ret, key)) - global_msg_id_join = ("f.room_id = messages.room_id" - + " and f.msg_id = messages.msg_id" - + " and messages.user_id = f.msg_sender_id") + @defer.inlineCallbacks + @log_function + def paginate_room_events(self, room_id, from_key, to_key=None, + direction='b', limit=-1, + with_feedback=False): + # TODO (erikj): Handle compressed feedback - select_query = ( - "SELECT messages.*, f.content AS fb_content, f.fb_sender_id" - + ", " + compressed_feedback_col + " AS compressed_fb" - + " FROM messages LEFT JOIN feedback f ON " + global_msg_id_join) + from_comp = '<' if direction =='b' else '>' + to_comp = '>' if direction =='b' else '<' + order = "DESC" if direction == 'b' else "ASC" - current_membership_sub_query = ( - "(SELECT membership from room_memberships rm" - + " WHERE user_id=? AND room_id = rm.room_id" - + " ORDER BY id DESC LIMIT 1)") + args = [room_id] - where = (" WHERE ? IN " + current_membership_sub_query - + " AND messages.room_id=?") + bounds = _get_token_bound(from_key, from_comp) + if to_key: + bounds = "%s AND %s" % (bounds, _get_token_bound(to_key, to_comp)) + + if int(limit) > 0: + args.append(int(limit)) + limit_str = " LIMIT ?" + else: + limit_str = "" - query = select_query + where - query_args = ["join", user_id, room_id] + sql = ( + "SELECT * FROM events " + "WHERE outlier = 0 AND room_id = ? AND %(bounds)s " + "ORDER BY topological_ordering %(order)s, stream_ordering %(order)s %(limit)s " + ) % {"bounds": bounds, "order": order, "limit": limit_str} - (query, query_args) = self._append_stream_operations( - "messages", query, query_args, from_pkey, to_pkey, - limit=limit, group_by=" GROUP BY messages.id " + rows = yield self._execute_and_decode( + sql, + *args ) - cursor = txn.execute(query, query_args) - - # convert the result set into events - entries = self.cursor_to_dict(cursor) - events = [] - for entry in entries: - # TODO we should spec the cursor > event mapping somewhere else. - event = {} - straight_mappings = ["msg_id", "user_id", "room_id"] - for key in straight_mappings: - event[key] = entry[key] - event["content"] = json.loads(entry["content"]) - if entry["compressed_fb"]: - event["feedback"] = json.loads(entry["compressed_fb"]) - events.append(event) - - latest_pkey = from_pkey if len(entries) == 0 else entries[-1]["id"] - - return (events, latest_pkey) - - def get_room_member_stream(self, user_id, from_key, to_key): - """Get all room membership events for this user between the given keys. - - Args: - user_id (str): The user who is requesting membership events. - from_key (int): The ID to start returning results from (exclusive). - to_key (int): The ID to stop returning results (exclusive). - Returns: - A tuple of rows (list of namedtuples), new_id(int) - """ - return self._db_pool.runInteraction( - self._get_room_member_rows, user_id, from_key, to_key + if rows: + topo = rows[-1]["topological_ordering"] + toke = rows[-1]["stream_ordering"] + next_token = "t%s-%s" % (topo, toke) + else: + # TODO (erikj): We should work out what to do here instead. + next_token = to_key if to_key else from_key + + defer.returnValue( + ( + [self._parse_event_from_row(r) for r in rows], + next_token + ) ) - def _get_room_member_rows(self, txn, user_id, from_pkey, to_pkey): - # get all room membership events for rooms which the user is - # *currently* joined in on, or all invite events for this user. - current_membership_sub_query = ( - "(SELECT membership FROM room_memberships" - + " WHERE user_id=? AND room_id = rm.room_id" - + " ORDER BY id DESC LIMIT 1)") - - query = ("SELECT rm.* FROM room_memberships rm " - # all membership events for rooms you've currently joined. - + " WHERE (? IN " + current_membership_sub_query - # all invite membership events for this user - + " OR rm.membership=? AND user_id=?)" - + " AND rm.id > ?") - query_args = ["join", user_id, "invite", user_id, from_pkey] - - if to_pkey != -1: - query += " AND rm.id < ?" - query_args.append(to_pkey) - - cursor = txn.execute(query, query_args) - return self._as_events(cursor, RoomMemberTable, from_pkey) - - def get_feedback_stream(self, user_id, from_key, to_key, room_id, limit=0): - return self._db_pool.runInteraction( - self._get_feedback_rows, - user_id, from_key, to_key, room_id, limit + @defer.inlineCallbacks + def get_recent_events_for_room(self, room_id, limit, end_token, + with_feedback=False): + # TODO (erikj): Handle compressed feedback + + sql = ( + "SELECT * FROM events " + "WHERE room_id = ? AND stream_ordering <= ? " + "ORDER BY topological_ordering, stream_ordering DESC LIMIT ? " ) - def _get_feedback_rows(self, txn, user_id, from_pkey, to_pkey, room_id, - limit): - # work out which rooms this user is joined in on and join them with - # the room id on the feedback table, bounded by the specified pkeys - - # get all messages where the *current* membership state is 'join' for - # this user in that room. - query = ( - "SELECT feedback.* FROM feedback WHERE ? IN " - + "(SELECT membership from room_memberships WHERE user_id=?" - + " AND room_id = feedback.room_id ORDER BY id DESC LIMIT 1)") - query_args = ["join", user_id] - - if room_id: - query += " AND feedback.room_id=?" - query_args.append(room_id) - - (query, query_args) = self._append_stream_operations( - "feedback", query, query_args, from_pkey, to_pkey, limit=limit + rows = yield self._execute_and_decode( + sql, + room_id, end_token, limit ) - cursor = txn.execute(query, query_args) - return self._as_events(cursor, FeedbackTable, from_pkey) + rows.reverse() # As we selected with reverse ordering + + if rows: + topo = rows[0]["topological_ordering"] + toke = rows[0]["stream_ordering"] + start_token = "t%s-%s" % (topo, toke) - def get_room_data_stream(self, user_id, from_key, to_key, room_id, - limit=0): - return self._db_pool.runInteraction( - self._get_room_data_rows, - user_id, from_key, to_key, room_id, limit + token = (start_token, end_token) + else: + token = (end_token, end_token) + + defer.returnValue( + ( + [self._parse_event_from_row(r) for r in rows], + token + ) ) - def _get_room_data_rows(self, txn, user_id, from_pkey, to_pkey, room_id, - limit): - # work out which rooms this user is joined in on and join them with - # the room id on the feedback table, bounded by the specified pkeys - - # get all messages where the *current* membership state is 'join' for - # this user in that room. - query = ( - "SELECT room_data.* FROM room_data WHERE ? IN " - + "(SELECT membership from room_memberships WHERE user_id=?" - + " AND room_id = room_data.room_id ORDER BY id DESC LIMIT 1)") - query_args = ["join", user_id] - - if room_id: - query += " AND room_data.room_id=?" - query_args.append(room_id) - - (query, query_args) = self._append_stream_operations( - "room_data", query, query_args, from_pkey, to_pkey, limit=limit + @defer.inlineCallbacks + def get_room_events_max_id(self): + res = yield self._execute_and_decode( + "SELECT MAX(stream_ordering) as m FROM events" ) - cursor = txn.execute(query, query_args) - return self._as_events(cursor, RoomDataTable, from_pkey) - - def _append_stream_operations(self, table_name, query, query_args, - from_pkey, to_pkey, limit=None, - group_by=""): - LATEST_ROW = -1 - order_by = "" - if to_pkey > from_pkey: - if from_pkey != LATEST_ROW: - # e.g. from=5 to=9 >> from 5 to 9 >> id>5 AND id<9 - query += (" AND %s.id > ? AND %s.id < ?" % - (table_name, table_name)) - query_args.append(from_pkey) - query_args.append(to_pkey) - else: - # e.g. from=-1 to=5 >> from now to 5 >> id>5 ORDER BY id DESC - query += " AND %s.id > ? " % table_name - order_by = "ORDER BY id DESC" - query_args.append(to_pkey) - elif from_pkey > to_pkey: - if to_pkey != LATEST_ROW: - # from=9 to=5 >> from 9 to 5 >> id>5 AND id<9 ORDER BY id DESC - query += (" AND %s.id > ? AND %s.id < ? " % - (table_name, table_name)) - order_by = "ORDER BY id DESC" - query_args.append(to_pkey) - query_args.append(from_pkey) - else: - # from=5 to=-1 >> from 5 to now >> id>5 - query += " AND %s.id > ?" % table_name - query_args.append(from_pkey) - - query += group_by + order_by - - if limit and limit > 0: - query += " LIMIT ?" - query_args.append(str(limit)) - - return (query, query_args) - - def _as_events(self, cursor, table, from_pkey): - data_entries = table.decode_results(cursor) - last_pkey = from_pkey - if data_entries: - last_pkey = data_entries[-1].id - - events = [ - entry.as_event(self.event_factory).get_dict() - for entry in data_entries - ] - - return (events, last_pkey) + logger.debug("get_room_events_max_id: %s", res) + + if not res or not res[0] or not res[0]["m"]: + defer.returnValue("s1") + return + + key = res[0]["m"] + 1 + defer.returnValue("s%d" % (key,)) diff --git a/synapse/types.py b/synapse/types.py index b8e191bb3c..fd6a3d1d72 100644 --- a/synapse/types.py +++ b/synapse/types.py @@ -38,6 +38,14 @@ class DomainSpecificString( def __iter__(self): raise ValueError("Attempted to iterate a %s" % (type(self).__name__)) + # Because this class is a namedtuple of strings and booleans, it is deeply + # immutable. + def __copy__(self): + return self + + def __deepcopy__(self, memo): + return self + @classmethod def from_string(cls, s, hs): """Parse the string given by 's' into a structure object.""" diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index ab9c242579..f4cf54b7ee 100644 --- a/tests/handlers/test_federation.py +++ b/tests/handlers/test_federation.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # Copyright 2014 matrix.org # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -41,6 +40,7 @@ class FederationTestCase(unittest.TestCase): datastore=NonCallableMock(spec_set=[ "persist_event", "store_room", + "get_room", ]), resource_for_federation=NonCallableMock(), http_client=NonCallableMock(spec_set=[]), @@ -69,10 +69,11 @@ class FederationTestCase(unittest.TestCase): store_id = "ASD" self.datastore.persist_event.return_value = defer.succeed(store_id) + self.datastore.get_room.return_value = defer.succeed(True) - yield self.handlers.federation_handler.on_receive(event, False) + yield self.handlers.federation_handler.on_receive(event, False, False) - self.datastore.persist_event.assert_called_once_with(event) + self.datastore.persist_event.assert_called_once_with(event, False) self.notifier.on_new_room_event.assert_called_once_with( event, store_id) @@ -89,7 +90,7 @@ class FederationTestCase(unittest.TestCase): content={}, ) - yield self.handlers.federation_handler.on_receive(event, False) + yield self.handlers.federation_handler.on_receive(event, False, False) mem_handler = self.handlers.room_member_handler self.assertEquals(1, mem_handler.change_membership.call_count) @@ -115,7 +116,7 @@ class FederationTestCase(unittest.TestCase): content={}, ) - yield self.handlers.federation_handler.on_receive(event, False) + yield self.handlers.federation_handler.on_receive(event, False, False) mem_handler = self.handlers.room_member_handler self.assertEquals(0, mem_handler.change_membership.call_count) diff --git a/tests/handlers/test_room.py b/tests/handlers/test_room.py index fd2d66db38..bf71d3be3b 100644 --- a/tests/handlers/test_room.py +++ b/tests/handlers/test_room.py @@ -40,7 +40,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): self.hostname, db_pool=None, datastore=NonCallableMock(spec_set=[ - "store_room_member", + "persist_event", "get_joined_hosts_for_room", "get_room_member", "get_room", @@ -69,6 +69,8 @@ class RoomMemberHandlerTestCase(unittest.TestCase): self.distributor = hs.get_distributor() self.hs = hs + self.distributor.declare("collect_presencelike_data") + self.handlers.room_member_handler = RoomMemberHandler(self.hs) self.handlers.profile_handler = ProfileHandler(self.hs) self.room_member_handler = self.handlers.room_member_handler @@ -97,7 +99,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): ) store_id = "store_id_fooo" - self.datastore.store_room_member.return_value = defer.succeed(store_id) + self.datastore.persist_event.return_value = defer.succeed(store_id) # Actual invocation yield self.room_member_handler.change_membership(event) @@ -110,12 +112,8 @@ class RoomMemberHandlerTestCase(unittest.TestCase): set(event.destinations) ) - self.datastore.store_room_member.assert_called_once_with( - user_id=target_user_id, - sender=user_id, - room_id=room_id, - content=content, - membership=Membership.INVITE, + self.datastore.persist_event.assert_called_once_with( + event ) self.notifier.on_new_room_event.assert_called_once_with( event, store_id) @@ -144,12 +142,14 @@ class RoomMemberHandlerTestCase(unittest.TestCase): joined = ["red", "green"] self.state_handler.handle_new_event.return_value = defer.succeed(True) - self.datastore.get_joined_hosts_for_room.return_value = ( - defer.succeed(joined) - ) + + def get_joined(*args): + return defer.succeed(joined) + + self.datastore.get_joined_hosts_for_room.side_effect = get_joined store_id = "store_id_fooo" - self.datastore.store_room_member.return_value = defer.succeed(store_id) + self.datastore.persist_event.return_value = defer.succeed(store_id) self.datastore.get_room.return_value = defer.succeed(1) # Not None. prev_state = NonCallableMock() @@ -171,12 +171,8 @@ class RoomMemberHandlerTestCase(unittest.TestCase): set(event.destinations) ) - self.datastore.store_room_member.assert_called_once_with( - user_id=target_user_id, - sender=user_id, - room_id=room_id, - content=content, - membership=Membership.JOIN, + self.datastore.persist_event.assert_called_once_with( + event ) self.notifier.on_new_room_event.assert_called_once_with( event, store_id) diff --git a/tests/rest/test_events.py b/tests/rest/test_events.py index 1ab92395f2..4025e14581 100644 --- a/tests/rest/test_events.py +++ b/tests/rest/test_events.py @@ -190,9 +190,7 @@ class EventStreamPermissionsTestCase(RestTestCase): "/events?access_token=%s&timeout=0" % (self.token)) self.assertEquals(200, code, msg=str(response)) - # First message is a reflection of my own presence status change - self.assertEquals(1, len(response["chunk"])) - self.assertEquals("m.presence", response["chunk"][0]["type"]) + self.assertEquals(0, len(response["chunk"])) # joined room (expect all content for room) yield self.join(room=room_id, user=self.user_id, tok=self.token) diff --git a/tests/rest/test_presence.py b/tests/rest/test_presence.py index 0ba72addf6..8ac246b4d5 100644 --- a/tests/rest/test_presence.py +++ b/tests/rest/test_presence.py @@ -287,14 +287,7 @@ class PresenceEventStreamTestCase(unittest.TestCase): # all be ours # I'll already get my own presence state change - self.assertEquals({"start": "0", "end": "1", "chunk": [ - {"type": "m.presence", - "content": { - "user_id": "@apple:test", - "state": ONLINE, - "mtime_age": 0, - }}, - ]}, response) + self.assertEquals({"start": "1", "end": "1", "chunk": []}, response) self.mock_datastore.set_presence_state.return_value = defer.succeed( {"state": ONLINE}) diff --git a/tests/rest/test_profile.py b/tests/rest/test_profile.py index ce64c5253b..9bd8dc9783 100644 --- a/tests/rest/test_profile.py +++ b/tests/rest/test_profile.py @@ -46,6 +46,7 @@ class ProfileTestCase(unittest.TestCase): resource_for_client=self.mock_resource, federation=Mock(), replication_layer=Mock(), + datastore=None, ) def _get_user_by_token(token=None): diff --git a/tests/rest/test_rooms.py b/tests/rest/test_rooms.py index c0ae26dd64..a9b66df912 100644 --- a/tests/rest/test_rooms.py +++ b/tests/rest/test_rooms.py @@ -104,36 +104,36 @@ class RoomPermissionsTestCase(RestTestCase): def tearDown(self): pass - @defer.inlineCallbacks - def test_get_message(self): - # get message in uncreated room, expect 403 - (code, response) = yield self.mock_resource.trigger_get( - "/rooms/noroom/messages/someid/m1") - self.assertEquals(403, code, msg=str(response)) - - # get message in created room not joined (no state), expect 403 - (code, response) = yield self.mock_resource.trigger_get( - self.created_rmid_msg_path) - self.assertEquals(403, code, msg=str(response)) - - # get message in created room and invited, expect 403 - yield self.invite(room=self.created_rmid, src=self.rmcreator_id, - targ=self.user_id) - (code, response) = yield self.mock_resource.trigger_get( - self.created_rmid_msg_path) - self.assertEquals(403, code, msg=str(response)) - - # get message in created room and joined, expect 200 - yield self.join(room=self.created_rmid, user=self.user_id) - (code, response) = yield self.mock_resource.trigger_get( - self.created_rmid_msg_path) - self.assertEquals(200, code, msg=str(response)) - - # get message in created room and left, expect 403 - yield self.leave(room=self.created_rmid, user=self.user_id) - (code, response) = yield self.mock_resource.trigger_get( - self.created_rmid_msg_path) - self.assertEquals(403, code, msg=str(response)) +# @defer.inlineCallbacks +# def test_get_message(self): +# # get message in uncreated room, expect 403 +# (code, response) = yield self.mock_resource.trigger_get( +# "/rooms/noroom/messages/someid/m1") +# self.assertEquals(403, code, msg=str(response)) +# +# # get message in created room not joined (no state), expect 403 +# (code, response) = yield self.mock_resource.trigger_get( +# self.created_rmid_msg_path) +# self.assertEquals(403, code, msg=str(response)) +# +# # get message in created room and invited, expect 403 +# yield self.invite(room=self.created_rmid, src=self.rmcreator_id, +# targ=self.user_id) +# (code, response) = yield self.mock_resource.trigger_get( +# self.created_rmid_msg_path) +# self.assertEquals(403, code, msg=str(response)) +# +# # get message in created room and joined, expect 200 +# yield self.join(room=self.created_rmid, user=self.user_id) +# (code, response) = yield self.mock_resource.trigger_get( +# self.created_rmid_msg_path) +# self.assertEquals(200, code, msg=str(response)) +# +# # get message in created room and left, expect 403 +# yield self.leave(room=self.created_rmid, user=self.user_id) +# (code, response) = yield self.mock_resource.trigger_get( +# self.created_rmid_msg_path) +# self.assertEquals(403, code, msg=str(response)) @defer.inlineCallbacks def test_send_message(self): @@ -794,7 +794,12 @@ class RoomMemberStateTestCase(RestTestCase): (code, response) = yield self.mock_resource.trigger("GET", path, None) self.assertEquals(200, code, msg=str(response)) - self.assertEquals(json.loads(content), response) + + expected_response = { + "membership": Membership.JOIN, + "prev": Membership.JOIN, + } + self.assertEquals(expected_response, response) @defer.inlineCallbacks def test_rooms_members_other(self): @@ -913,9 +918,9 @@ class RoomMessagesTestCase(RestTestCase): (code, response) = yield self.mock_resource.trigger("PUT", path, content) self.assertEquals(200, code, msg=str(response)) - (code, response) = yield self.mock_resource.trigger("GET", path, None) - self.assertEquals(200, code, msg=str(response)) - self.assert_dict(json.loads(content), response) +# (code, response) = yield self.mock_resource.trigger("GET", path, None) +# self.assertEquals(200, code, msg=str(response)) +# self.assert_dict(json.loads(content), response) # m.text message type path = "/rooms/%s/messages/%s/mid2" % ( @@ -925,9 +930,9 @@ class RoomMessagesTestCase(RestTestCase): (code, response) = yield self.mock_resource.trigger("PUT", path, content) self.assertEquals(200, code, msg=str(response)) - (code, response) = yield self.mock_resource.trigger("GET", path, None) - self.assertEquals(200, code, msg=str(response)) - self.assert_dict(json.loads(content), response) +# (code, response) = yield self.mock_resource.trigger("GET", path, None) +# self.assertEquals(200, code, msg=str(response)) +# self.assert_dict(json.loads(content), response) # trying to send message in different user path path = "/rooms/%s/messages/%s/mid2" % ( diff --git a/tests/test_state.py b/tests/test_state.py index aaf873a856..e64d15a3a2 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -36,7 +36,7 @@ class StateTestCase(unittest.TestCase): "get_unresolved_state_tree", "update_current_state", "get_latest_pdus_in_context", - "get_current_state", + "get_current_state_pdu", "get_pdu", ]) self.replication = Mock(spec=["get_pdu"]) @@ -247,7 +247,7 @@ class StateTestCase(unittest.TestCase): pdus = [tup] self.persistence.get_latest_pdus_in_context.return_value = pdus - self.persistence.get_current_state.return_value = state_pdu + self.persistence.get_current_state_pdu.return_value = state_pdu yield self.state.handle_new_event(event) diff --git a/tests/utils.py b/tests/utils.py index 9b0de38a9d..c68b17f7b9 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -112,35 +112,20 @@ class MockClock(object): class MemoryDataStore(object): - class RoomMember(namedtuple( - "RoomMember", - ["room_id", "user_id", "sender", "membership", "content"] - )): - def as_event(self, event_factory): - return event_factory.create_event( - etype=RoomMemberEvent.TYPE, - room_id=self.room_id, - target_user_id=self.user_id, - user_id=self.sender, - content=json.loads(self.content), - ) - - PathData = namedtuple("PathData", - ["room_id", "path", "content"]) - - Message = namedtuple("Message", - ["room_id", "msg_id", "user_id", "content"]) - - Room = namedtuple("Room", - ["room_id", "is_public", "creator"]) + Room = namedtuple( + "Room", + ["room_id", "is_public", "creator"] + ) def __init__(self): self.tokens_to_users = {} self.paths_to_content = {} + self.members = {} - self.messages = {} self.rooms = {} - self.room_members = {} + + self.current_state = {} + self.events = [] def register(self, user_id, token, password_hash): if user_id in self.tokens_to_users.values(): @@ -163,117 +148,60 @@ class MemoryDataStore(object): if room_id in self.rooms: raise StoreError(409, "Conflicting room!") - room = MemoryDataStore.Room(room_id=room_id, is_public=is_public, - creator=room_creator_user_id) + room = MemoryDataStore.Room( + room_id=room_id, + is_public=is_public, + creator=room_creator_user_id + ) self.rooms[room_id] = room - #self.store_room_member(user_id=room_creator_user_id, room_id=room_id, - #membership=Membership.JOIN, - #content={"membership": Membership.JOIN}) - def get_message(self, user_id=None, room_id=None, msg_id=None): - try: - return self.messages[user_id + room_id + msg_id] - except: - return None - - def store_message(self, user_id=None, room_id=None, msg_id=None, - content=None): - msg = MemoryDataStore.Message(room_id=room_id, msg_id=msg_id, - user_id=user_id, content=content) - self.messages[user_id + room_id + msg_id] = msg + def get_room_member(self, user_id, room_id): + return self.members.get(room_id, {}).get(user_id) - def get_room_member(self, user_id=None, room_id=None): - try: - return self.members[user_id + room_id] - except: - return None - - def get_room_members(self, room_id=None, membership=None): - try: - return self.room_members[room_id] - except: - return None + def get_room_members(self, room_id, membership=None): + if membership: + return [ + v for k, v in self.members.get(room_id, {}).items() + if v.membership == membership + ] + else: + return self.members.get(room_id, {}).values() def get_rooms_for_user_where_membership_is(self, user_id, membership_list): - return [r for r in self.room_members - if user_id in self.room_members[r]] - - def store_room_member(self, user_id=None, sender=None, room_id=None, - membership=None, content=None): - member = MemoryDataStore.RoomMember(room_id=room_id, user_id=user_id, - sender=sender, membership=membership, content=json.dumps(content)) - self.members[user_id + room_id] = member - - # TODO should be latest state - if room_id not in self.room_members: - self.room_members[room_id] = [] - self.room_members[room_id].append(member) - - def get_room_data(self, room_id, etype, state_key=""): - path = "%s-%s-%s" % (room_id, etype, state_key) - try: - return self.paths_to_content[path] - except: - return None - - def store_room_data(self, room_id, etype, state_key="", content=None): - path = "%s-%s-%s" % (room_id, etype, state_key) - data = MemoryDataStore.PathData(path=path, room_id=room_id, - content=content) - self.paths_to_content[path] = data + return [ + r for r in self.members + if self.members[r].get(user_id).membership in membership_list + ] - def get_message_stream(self, user_id=None, from_key=None, to_key=None, + def get_room_events_stream(self, user_id=None, from_key=None, to_key=None, room_id=None, limit=0, with_feedback=False): return ([], from_key) # TODO - def get_room_member_stream(self, user_id=None, from_key=None, to_key=None): - return ([], from_key) # TODO - - def get_feedback_stream(self, user_id=None, from_key=None, to_key=None, - room_id=None, limit=0): - return ([], from_key) # TODO - - def get_room_data_stream(self, user_id=None, from_key=None, to_key=None, - room_id=None, limit=0): - return ([], from_key) # TODO - - def to_events(self, data_store_list): - return data_store_list # TODO - - def get_max_message_id(self): - return 0 # TODO - - def get_max_feedback_id(self): - return 0 # TODO - - def get_max_room_member_id(self): - return 0 # TODO - - def get_max_room_data_id(self): - return 0 # TODO - def get_joined_hosts_for_room(self, room_id): return defer.succeed([]) def persist_event(self, event): - if event.type == MessageEvent.TYPE: - return self.store_message( - user_id=event.user_id, - room_id=event.room_id, - msg_id=event.msg_id, - content=json.dumps(event.content) - ) - elif event.type == RoomMemberEvent.TYPE: - return self.store_room_member( - user_id=event.target_user_id, - room_id=event.room_id, - content=event.content, - membership=event.content["membership"] - ) + if event.type == RoomMemberEvent.TYPE: + room_id = event.room_id + user = event.target_user_id + membership = event.membership + self.members.setdefault(room_id, {})[user] = event + + if hasattr(event, "state_key"): + key = (event.room_id, event.type, event.state_key) + self.current_state[key] = event + + self.events.append(event) + + def get_current_state(self, room_id, event_type=None, state_key=""): + if event_type: + key = (room_id, event_type, state_key) + return self.current_state.get(key) else: - raise NotImplementedError( - "Don't know how to persist type=%s" % event.type - ) + return [ + e for e in self.current_state + if e[0] == room_id + ] def set_presence_state(self, user_localpart, state): return defer.succeed({"state": 0}) @@ -281,6 +209,8 @@ class MemoryDataStore(object): def get_presence_list(self, user_localpart, accepted): return [] + def get_room_events_max_id(self): + return 0 # TODO (erikj) def _format_call(args, kwargs): return ", ".join( diff --git a/webclient/app-controller.js b/webclient/app-controller.js index 96656e12c3..92ad01e4f9 100644 --- a/webclient/app-controller.js +++ b/webclient/app-controller.js @@ -53,7 +53,7 @@ angular.module('MatrixWebClientController', ['matrixService']) }; if (matrixService.isUserLoggedIn()) { - eventStreamService.resume(); + // eventStreamService.resume(); } // Logs the user out @@ -66,7 +66,7 @@ angular.module('MatrixWebClientController', ['matrixService']) matrixService.saveConfig(); // And go to the login page - $location.path("login"); + $location.url("login"); }; // Listen to the event indicating that the access token is no longer valid. diff --git a/webclient/app-filter.js b/webclient/app-filter.js index 64c3bb04de..b8f4ed25bc 100644 --- a/webclient/app-filter.js +++ b/webclient/app-filter.js @@ -54,12 +54,15 @@ angular.module('matrixWebClient') }); // FIXME: we shouldn't disambiguate displayNames on every orderMembersList - // invocation but keep track of duplicates incrementally somewhere + // invocation but keep track of duplicates incrementally somewhere angular.forEach(displayNames, function(value, key) { if (value.length > 1) { // console.log(key + ": " + value); - for (i=0; i < value.length; i++) { + for (var i=0; i < value.length; i++) { var v = value[i]; + // FIXME: this permenantly rewrites the displayname for a given + // room member. which means we can't reset their name if it is + // no longer ambiguous! members[v].displayname += " (" + v + ")"; // console.log(v + " " + members[v]); }; diff --git a/webclient/app.css b/webclient/app.css index 869db69cd6..207f35f5f3 100644 --- a/webclient/app.css +++ b/webclient/app.css @@ -66,6 +66,10 @@ h1 { background-color: #faa; } +.mouse-pointer { + cursor: pointer; +} + /*** Participant list ***/ #usersTableWrapper { @@ -89,7 +93,6 @@ h1 { height: 100px; position: relative; background-color: #000; - cursor: pointer; } .userAvatar .userAvatarImage { @@ -142,6 +145,7 @@ h1 { max-width: 1280px; width: 100%; border-collapse: collapse; + table-layout: fixed; } #messageTable td { @@ -149,7 +153,8 @@ h1 { } .leftBlock { - width: 10em; + width: 14em; + word-wrap: break-word; vertical-align: top; background-color: #fff; color: #888; @@ -187,24 +192,13 @@ h1 { object-fit: cover; } -.text { - background-color: #eee; - border: 1px solid #d8d8d8; - height: 31px; - display: inline-table; - max-width: 90%; - font-size: 16px; - /* word-wrap: break-word; */ - word-break: break-all; -} - .emote { - background-color: #fff ! important; + background-color: transparent ! important; border: 0px ! important; } .membership { - background-color: #fff ! important; + background-color: transparent ! important; border: 0px ! important; } @@ -216,33 +210,66 @@ h1 { height: auto; } +.text { + vertical-align: top; +} + .bubble { + background-color: #eee; + border: 1px solid #d8d8d8; + display: inline-block; + margin-bottom: -1px; + max-width: 90%; + font-size: 16px; + word-wrap: break-word; padding-top: 7px; padding-bottom: 5px; padding-left: 1em; padding-right: 1em; vertical-align: middle; + -webkit-text-size-adjust:100% } .differentUser td { - padding-top: 5px ! important; - margin-top: 5px ! important; + padding-bottom: 5px ! important; } .mine { text-align: right; } -.mine .text { +.text.emote .bubble, +.text.membership .bubble, +.mine .text.emote .bubble, +.mine .text.membership .bubble + { + background-color: transparent ! important; + border: 0px ! important; +} + +.mine .text .bubble { background-color: #f8f8ff ! important; + text-align: left ! important; } -.mine .emote { - background-color: #fff ! important; +#room-fullscreen-image { + position: absolute; + top: 0px; + height: 0px; + width: 100%; + height: 100%; } -.mine .text .bubble { - text-align: left ! important; +#room-fullscreen-image img { + max-width: 100%; + max-height: 100%; + bottom: 0; + left: 0; + margin: auto; + overflow: auto; + position: fixed; + right: 0; + top: 0; } /*** Profile ***/ diff --git a/webclient/app.js b/webclient/app.js index f27ebedc6f..944b8ec270 100644 --- a/webclient/app.js +++ b/webclient/app.js @@ -80,6 +80,6 @@ matrixWebClient.run(['$location', 'matrixService', 'eventStreamService', functio $location.path("login"); } else { - eventStreamService.resume(); + // eventStreamService.resume(); } }]); diff --git a/webclient/components/fileUpload/file-upload-service.js b/webclient/components/fileUpload/file-upload-service.js index d620e6a4d0..5f01478fd1 100644 --- a/webclient/components/fileUpload/file-upload-service.js +++ b/webclient/components/fileUpload/file-upload-service.js @@ -20,19 +20,20 @@ /* * Upload an HTML5 file to a server */ -angular.module('mFileUpload', []) -.service('mFileUpload', ['matrixService', '$q', function (matrixService, $q) { +angular.module('mFileUpload', ['matrixService', 'mUtilities']) +.service('mFileUpload', ['$q', 'matrixService', 'mUtilities', function ($q, matrixService, mUtilities) { /* - * Upload an HTML5 file to a server and returned a promise + * Upload an HTML5 file or blob to a server and returned a promise * that will provide the URL of the uploaded file. + * @param {File|Blob} file the file data to send */ this.uploadFile = function(file) { var deferred = $q.defer(); console.log("Uploading " + file.name + "... to /matrix/content"); matrixService.uploadContent(file).then( function(response) { - var content_url = location.origin + "/matrix/content/" + response.data.content_token; + var content_url = response.data.content_token; console.log(" -> Successfully uploaded! Available at " + content_url); deferred.resolve(content_url); }, @@ -44,4 +45,135 @@ angular.module('mFileUpload', []) return deferred.promise; }; + + /* + * Upload an image file plus generate a thumbnail of it and upload it so that + * we will have all information to fulfill an image message request data. + * @param {File} imageFile the imageFile to send + * @param {Integer} thumbnailSize the max side size of the thumbnail to create + * @returns {promise} A promise that will be resolved by a image message object + * ready to be send with the Matrix API + */ + this.uploadImageAndThumbnail = function(imageFile, thumbnailSize) { + var self = this; + var deferred = $q.defer(); + + console.log("uploadImageAndThumbnail " + imageFile.name + " - thumbnailSize: " + thumbnailSize); + + // The message structure that will be returned in the promise + var imageMessage = { + msgtype: "m.image", + url: undefined, + body: { + size: undefined, + w: undefined, + h: undefined, + mimetype: undefined + }, + thumbnail_url: undefined, + thumbnail_info: { + size: undefined, + w: undefined, + h: undefined, + mimetype: undefined + } + }; + + // First, get the image size + mUtilities.getImageSize(imageFile).then( + function(size) { + console.log("image size: " + JSON.stringify(size)); + + // The final operation: send imageFile + var uploadImage = function() { + self.uploadFile(imageFile).then( + function(url) { + // Update message metadata + imageMessage.url = url; + imageMessage.body = { + size: imageFile.size, + w: size.width, + h: size.height, + mimetype: imageFile.type + }; + + // If there is no thumbnail (because the original image is smaller than thumbnailSize), + // reuse the original image info for thumbnail data + if (!imageMessage.thumbnail_url) { + imageMessage.thumbnail_url = imageMessage.url; + imageMessage.thumbnail_info = imageMessage.body; + } + + // We are done + deferred.resolve(imageMessage); + }, + function(error) { + console.log(" -> Can't upload image"); + deferred.reject(error); + } + ); + }; + + // Create a thumbnail if the image size exceeds thumbnailSize + if (Math.max(size.width, size.height) > thumbnailSize) { + console.log(" Creating thumbnail..."); + mUtilities.resizeImage(imageFile, thumbnailSize).then( + function(thumbnailBlob) { + + // Get its size + mUtilities.getImageSize(thumbnailBlob).then( + function(thumbnailSize) { + console.log(" -> Thumbnail size: " + JSON.stringify(thumbnailSize)); + + // Upload it to the server + self.uploadFile(thumbnailBlob).then( + function(thumbnailUrl) { + + // Update image message data + imageMessage.thumbnail_url = thumbnailUrl; + imageMessage.thumbnail_info = { + size: thumbnailBlob.size, + w: thumbnailSize.width, + h: thumbnailSize.height, + mimetype: thumbnailBlob.type + }; + + // Then, upload the original image + uploadImage(); + }, + function(error) { + console.log(" -> Can't upload thumbnail"); + deferred.reject(error); + } + ); + }, + function(error) { + console.log(" -> Failed to get thumbnail size"); + deferred.reject(error); + } + ); + + }, + function(error) { + console.log(" -> Failed to create thumbnail: " + error); + deferred.reject(error); + } + ); + } + else { + // No need of thumbnail + console.log(" Thumbnail is not required"); + uploadImage(); + } + + }, + function(error) { + console.log(" -> Failed to get image size"); + deferred.reject(error); + } + ); + + return deferred.promise; + }; + }]); diff --git a/webclient/components/matrix/event-handler-service.js b/webclient/components/matrix/event-handler-service.js index b8529895fe..b5eb73d92b 100644 --- a/webclient/components/matrix/event-handler-service.js +++ b/webclient/components/matrix/event-handler-service.js @@ -35,6 +35,8 @@ angular.module('eventHandlerService', []) $rootScope.events = { rooms: {}, // will contain roomId: { messages:[], members:{userid1: event} } }; + + $rootScope.presence = {}; var initRoom = function(room_id) { if (!(room_id in $rootScope.events.rooms)) { @@ -44,6 +46,12 @@ angular.module('eventHandlerService', []) $rootScope.events.rooms[room_id].members = {}; } } + + var reInitRoom = function(room_id) { + $rootScope.events.rooms[room_id] = {}; + $rootScope.events.rooms[room_id].messages = []; + $rootScope.events.rooms[room_id].members = {}; + } var handleMessage = function(event, isLiveEvent) { if ("membership_target" in event.content) { @@ -69,11 +77,23 @@ angular.module('eventHandlerService', []) var handleRoomMember = function(event, isLiveEvent) { initRoom(event.room_id); + + // add membership changes as if they were a room message if something interesting changed + if (event.content.prev !== event.content.membership) { + if (isLiveEvent) { + $rootScope.events.rooms[event.room_id].messages.push(event); + } + else { + $rootScope.events.rooms[event.room_id].messages.unshift(event); + } + } + $rootScope.events.rooms[event.room_id].members[event.user_id] = event; $rootScope.$broadcast(MEMBER_EVENT, event, isLiveEvent); }; var handlePresence = function(event, isLiveEvent) { + $rootScope.presence[event.content.user_id] = event; $rootScope.$broadcast(PRESENCE_EVENT, event, isLiveEvent); }; @@ -107,6 +127,10 @@ angular.module('eventHandlerService', []) for (var i=0; i<events.length; i++) { this.handleEvent(events[i], isLiveEvents); } - } + }, + + reInitRoom: function(room_id) { + reInitRoom(room_id); + }, }; }]); diff --git a/webclient/components/matrix/event-stream-service.js b/webclient/components/matrix/event-stream-service.js index c94cf0fe72..a1a98b2a36 100644 --- a/webclient/components/matrix/event-stream-service.js +++ b/webclient/components/matrix/event-stream-service.js @@ -25,7 +25,6 @@ the eventHandlerService. angular.module('eventStreamService', []) .factory('eventStreamService', ['$q', '$timeout', 'matrixService', 'eventHandlerService', function($q, $timeout, matrixService, eventHandlerService) { var END = "END"; - var START = "START"; var TIMEOUT_MS = 30000; var ERR_TIMEOUT_MS = 5000; @@ -49,11 +48,12 @@ angular.module('eventStreamService', []) var saveStreamSettings = function() { localStorage.setItem("streamSettings", JSON.stringify(settings)); }; - - var startEventStream = function() { + + var doEventStream = function(deferred) { settings.shouldPoll = true; settings.isActive = true; - var deferred = $q.defer(); + deferred = deferred || $q.defer(); + // run the stream from the latest token matrixService.getEventStream(settings.from, TIMEOUT_MS).then( function(response) { @@ -64,13 +64,16 @@ angular.module('eventStreamService', []) settings.from = response.data.end; - console.log("[EventStream] Got response from "+settings.from+" to "+response.data.end); + console.log( + "[EventStream] Got response from "+settings.from+ + " to "+response.data.end + ); eventHandlerService.handleEvents(response.data.chunk, true); deferred.resolve(response); if (settings.shouldPoll) { - $timeout(startEventStream, 0); + $timeout(doEventStream, 0); } else { console.log("[EventStream] Stopping poll."); @@ -84,13 +87,48 @@ angular.module('eventStreamService', []) deferred.reject(error); if (settings.shouldPoll) { - $timeout(startEventStream, ERR_TIMEOUT_MS); + $timeout(doEventStream, ERR_TIMEOUT_MS); } else { console.log("[EventStream] Stopping polling."); } } ); + + return deferred.promise; + } + + var startEventStream = function() { + settings.shouldPoll = true; + settings.isActive = true; + var deferred = $q.defer(); + + // FIXME: We are discarding all the messages. + matrixService.rooms().then( + function(response) { + var rooms = response.data.rooms; + for (var i = 0; i < rooms.length; ++i) { + var room = rooms[i]; + if ("state" in room) { + for (var j = 0; j < room.state.length; ++j) { + eventHandlerService.handleEvents(room.state[j], false); + } + } + } + + var presence = response.data.presence; + for (var i = 0; i < presence.length; ++i) { + eventHandlerService.handleEvent(presence[i], false); + } + + settings.from = response.data.end + doEventStream(deferred); + }, + function(error) { + $scope.feedback = "Failure: " + error.data; + } + ); + return deferred.promise; }; diff --git a/webclient/components/matrix/matrix-service.js b/webclient/components/matrix/matrix-service.js index 2463b51203..d5738e01c8 100644 --- a/webclient/components/matrix/matrix-service.js +++ b/webclient/components/matrix/matrix-service.js @@ -61,16 +61,23 @@ angular.module('matrixService', []) return doBaseRequest(config.homeserver, method, path, params, data, undefined); }; - var doBaseRequest = function(baseUrl, method, path, params, data, headers) { - return $http({ + var doBaseRequest = function(baseUrl, method, path, params, data, headers, $httpParams) { + + var request = { method: method, url: baseUrl + path, params: params, data: data, headers: headers - }); - }; + }; + + // Add additional $http parameters + if ($httpParams) { + angular.extend(request, $httpParams); + } + return $http(request); + }; return { /****** Home server API ******/ @@ -204,11 +211,11 @@ angular.module('matrixService', []) }, // Send an image message - sendImageMessage: function(room_id, image_url, image_alt, msg_id) { + sendImageMessage: function(room_id, image_url, image_body, msg_id) { var content = { msgtype: "m.image", url: image_url, - body: image_alt + body: image_body }; return this.sendMessage(room_id, msg_id, content); @@ -239,8 +246,8 @@ angular.module('matrixService', []) path = path.replace("$room_id", room_id); var params = { from: from_token, - to: "START", - limit: limit + limit: limit, + dir: 'b' }; return doRequest("GET", path, params); }, @@ -302,17 +309,25 @@ angular.module('matrixService', []) }, // hit the Identity Server for a 3PID request. - linkEmail: function(email) { + linkEmail: function(email, clientSecret, sendAttempt) { var path = "/matrix/identity/api/v1/validate/email/requestToken" - var data = "clientSecret=abc123&email=" + encodeURIComponent(email); + var data = "clientSecret="+clientSecret+"&email=" + encodeURIComponent(email)+"&sendAttempt="+sendAttempt; var headers = {}; headers["Content-Type"] = "application/x-www-form-urlencoded"; return doBaseRequest(config.identityServer, "POST", path, {}, data, headers); }, - authEmail: function(userId, tokenId, code) { + authEmail: function(clientSecret, tokenId, code) { var path = "/matrix/identity/api/v1/validate/email/submitToken"; - var data = "token="+code+"&mxId="+encodeURIComponent(userId)+"&tokenId="+tokenId; + var data = "token="+code+"&sid="+tokenId+"&clientSecret="+clientSecret; + var headers = {}; + headers["Content-Type"] = "application/x-www-form-urlencoded"; + return doBaseRequest(config.identityServer, "POST", path, {}, data, headers); + }, + + bindEmail: function(userId, tokenId, clientSecret) { + var path = "/matrix/identity/api/v1/3pid/bind"; + var data = "mxid="+encodeURIComponent(userId)+"&sid="+tokenId+"&clientSecret="+clientSecret; var headers = {}; headers["Content-Type"] = "application/x-www-form-urlencoded"; return doBaseRequest(config.identityServer, "POST", path, {}, data, headers); @@ -326,7 +341,17 @@ angular.module('matrixService', []) var params = { access_token: config.access_token }; - return doBaseRequest(config.homeserver, "POST", path, params, file, headers); + + // If the file is actually a Blob object, prevent $http from JSON-stringified it before sending + // (Equivalent to jQuery ajax processData = false) + var $httpParams; + if (file instanceof Blob) { + $httpParams = { + transformRequest: angular.identity + }; + } + + return doBaseRequest(config.homeserver, "POST", path, params, file, headers, $httpParams); }, // start listening on /events @@ -375,6 +400,7 @@ angular.module('matrixService', []) // Set a new config (Use saveConfig to actually store it permanently) setConfig: function(newConfig) { config = newConfig; + console.log("new IS: "+config.identityServer); }, // Commits config into permanent storage diff --git a/webclient/components/utilities/utilities-service.js b/webclient/components/utilities/utilities-service.js new file mode 100644 index 0000000000..3df2f04458 --- /dev/null +++ b/webclient/components/utilities/utilities-service.js @@ -0,0 +1,151 @@ +/* + Copyright 2014 matrix.org + + 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. + */ + +'use strict'; + +/* + * This service contains multipurpose helper functions. + */ +angular.module('mUtilities', []) +.service('mUtilities', ['$q', function ($q) { + /* + * Get the size of an image + * @param {File|Blob} imageFile the file containing the image + * @returns {promise} A promise that will be resolved by an object with 2 members: + * width & height + */ + this.getImageSize = function(imageFile) { + var deferred = $q.defer(); + + // Load the file into an html element + var img = document.createElement("img"); + + var reader = new FileReader(); + reader.onload = function(e) { + img.src = e.target.result; + + // Once ready, returns its size + img.onload = function() { + deferred.resolve({ + width: img.width, + height: img.height + }); + }; + img.onerror = function(e) { + deferred.reject(e); + }; + }; + reader.onerror = function(e) { + deferred.reject(e); + }; + reader.readAsDataURL(imageFile); + + return deferred.promise; + }; + + /* + * Resize the image to fit in a square of the side maxSize. + * The aspect ratio is kept. The returned image data uses JPEG compression. + * Source: http://hacks.mozilla.org/2011/01/how-to-develop-a-html5-image-uploader/ + * @param {File} imageFile the file containing the image + * @param {Integer} maxSize the max side size + * @returns {promise} A promise that will be resolved by a Blob object containing + * the resized image data + */ + this.resizeImage = function(imageFile, maxSize) { + var self = this; + var deferred = $q.defer(); + + var canvas = document.createElement("canvas"); + + var img = document.createElement("img"); + var reader = new FileReader(); + reader.onload = function(e) { + + img.src = e.target.result; + + // Once ready, returns its size + img.onload = function() { + var ctx = canvas.getContext("2d"); + ctx.drawImage(img, 0, 0); + + var MAX_WIDTH = maxSize; + var MAX_HEIGHT = maxSize; + var width = img.width; + var height = img.height; + + if (width > height) { + if (width > MAX_WIDTH) { + height *= MAX_WIDTH / width; + width = MAX_WIDTH; + } + } else { + if (height > MAX_HEIGHT) { + width *= MAX_HEIGHT / height; + height = MAX_HEIGHT; + } + } + canvas.width = width; + canvas.height = height; + var ctx = canvas.getContext("2d"); + ctx.drawImage(img, 0, 0, width, height); + + // Extract image data in the same format as the original one. + // The 0.7 compression value will work with formats that supports it like JPEG. + var dataUrl = canvas.toDataURL(imageFile.type, 0.7); + deferred.resolve(self.dataURItoBlob(dataUrl)); + }; + img.onerror = function(e) { + deferred.reject(e); + }; + }; + reader.onerror = function(e) { + deferred.reject(e); + }; + reader.readAsDataURL(imageFile); + + return deferred.promise; + }; + + /* + * Convert a dataURI string to a blob + * Source: http://stackoverflow.com/a/17682951 + * @param {String} dataURI the dataURI can be a base64 encoded string or an URL encoded string. + * @returns {Blob} the blob + */ + this.dataURItoBlob = function(dataURI) { + // convert base64 to raw binary data held in a string + // doesn't handle URLEncoded DataURIs + var byteString; + if (dataURI.split(',')[0].indexOf('base64') >= 0) + byteString = atob(dataURI.split(',')[1]); + else + byteString = unescape(dataURI.split(',')[1]); + // separate out the mime component + var mimeString = dataURI.split(',')[0].split(':')[1].split(';')[0]; + + // write the bytes of the string to an ArrayBuffer + var ab = new ArrayBuffer(byteString.length); + var ia = new Uint8Array(ab); + for (var i = 0; i < byteString.length; i++) { + ia[i] = byteString.charCodeAt(i); + } + + // write the ArrayBuffer to a blob, and you're done + return new Blob([ab],{type: mimeString}); + }; + +}]); \ No newline at end of file diff --git a/webclient/index.html b/webclient/index.html index a7e9cd9341..27d9208193 100644 --- a/webclient/index.html +++ b/webclient/index.html @@ -25,6 +25,7 @@ <script src="components/matrix/event-handler-service.js"></script> <script src="components/fileInput/file-input-directive.js"></script> <script src="components/fileUpload/file-upload-service.js"></script> + <script src="components/utilities/utilities-service.js"></script> </head> <body> diff --git a/webclient/login/login-controller.js b/webclient/login/login-controller.js index 67d0b7b90c..e3d0eca946 100644 --- a/webclient/login/login-controller.js +++ b/webclient/login/login-controller.js @@ -53,7 +53,7 @@ angular.module('LoginController', ['matrixService']) matrixService.saveConfig(); eventStreamService.resume(); // Go to the user's rooms list page - $location.path("rooms"); + $location.url("rooms"); }, function(error) { if (error.data) { @@ -70,6 +70,7 @@ angular.module('LoginController', ['matrixService']) $scope.login = function() { matrixService.setConfig({ homeserver: $scope.account.homeserver, + identityServer: $scope.account.identityServer, user_id: $scope.account.user_id }); // try to login @@ -79,12 +80,13 @@ angular.module('LoginController', ['matrixService']) $scope.feedback = "Login successful."; matrixService.setConfig({ homeserver: $scope.account.homeserver, + identityServer: $scope.account.identityServer, user_id: response.data.user_id, access_token: response.data.access_token }); matrixService.saveConfig(); eventStreamService.resume(); - $location.path("rooms"); + $location.url("rooms"); } else { $scope.feedback = "Failed to login: " + JSON.stringify(response.data); diff --git a/webclient/room/room-controller.js b/webclient/room/room-controller.js index 364ca41510..58ba432ce5 100644 --- a/webclient/room/room-controller.js +++ b/webclient/room/room-controller.js @@ -14,11 +14,12 @@ See the License for the specific language governing permissions and limitations under the License. */ -angular.module('RoomController', ['ngSanitize']) -.controller('RoomController', ['$scope', '$http', '$timeout', '$routeParams', '$location', 'matrixService', 'eventStreamService', 'eventHandlerService', - function($scope, $http, $timeout, $routeParams, $location, matrixService, eventStreamService, eventHandlerService) { +angular.module('RoomController', ['ngSanitize', 'mUtilities']) +.controller('RoomController', ['$scope', '$http', '$timeout', '$routeParams', '$location', 'matrixService', 'eventStreamService', 'eventHandlerService', 'mFileUpload', 'mUtilities', '$rootScope', + function($scope, $http, $timeout, $routeParams, $location, matrixService, eventStreamService, eventHandlerService, mFileUpload, mUtilities, $rootScope) { 'use strict'; var MESSAGES_PER_PAGINATION = 30; + var THUMBNAIL_SIZE = 320; // Room ids. Computed and resolved in onInit $scope.room_id = undefined; @@ -28,9 +29,12 @@ angular.module('RoomController', ['ngSanitize']) user_id: matrixService.config().user_id, events_from: "END", // when to start the event stream from. earliest_token: "END", // stores how far back we've paginated. + first_pagination: true, // this is toggled off when the first pagination is done can_paginate: true, // this is toggled off when we run out of items paginating: false, // used to avoid concurrent pagination requests pulling in dup contents - stream_failure: undefined // the response when the stream fails + stream_failure: undefined, // the response when the stream fails + // FIXME: sending has been disabled, as surely messages should be sent in the background rather than locking the UI synchronously --Matthew + sending: false // true when a message is being sent. It helps to disable the UI when a process is running }; $scope.members = {}; $scope.autoCompleting = false; @@ -88,7 +92,7 @@ angular.module('RoomController', ['ngSanitize']) var paginate = function(numItems) { // console.log("paginate " + numItems); - if ($scope.state.paginating) { + if ($scope.state.paginating || !$scope.room_id) { return; } else { @@ -98,7 +102,6 @@ angular.module('RoomController', ['ngSanitize']) var originalTopRow = $("#messageTable>tbody>tr:first")[0]; matrixService.paginateBackMessages($scope.room_id, $scope.state.earliest_token, numItems).then( function(response) { - var firstPagination = !$scope.events.rooms[$scope.room_id]; eventHandlerService.handleEvents(response.data.chunk, false); $scope.state.earliest_token = response.data.end; if (response.data.chunk.length < MESSAGES_PER_PAGINATION) { @@ -124,8 +127,9 @@ angular.module('RoomController', ['ngSanitize']) }, 0); } - if (firstPagination) { + if ($scope.state.first_pagination) { scrollToBottom(); + $scope.state.first_pagination = false; } else { // lock the scroll position @@ -144,10 +148,12 @@ angular.module('RoomController', ['ngSanitize']) console.log("Failed to paginateBackMessages: " + JSON.stringify(error)); $scope.state.paginating = false; } - ) + ); }; var updateMemberList = function(chunk) { + if (chunk.room_id != $scope.room_id) return; + var isNewMember = !(chunk.target_user_id in $scope.members); if (isNewMember) { // FIXME: why are we copying these fields around inside chunk? @@ -157,8 +163,7 @@ angular.module('RoomController', ['ngSanitize']) if ("mtime_age" in chunk.content) { chunk.mtime_age = chunk.content.mtime_age; } -/* - // FIXME: once the HS reliably returns the displaynames & avatar_urls for both + // Once the HS reliably returns the displaynames & avatar_urls for both // local and remote users, we should use this rather than the evalAsync block // below if ("displayname" in chunk.content) { @@ -167,9 +172,11 @@ angular.module('RoomController', ['ngSanitize']) if ("avatar_url" in chunk.content) { chunk.avatar_url = chunk.content.avatar_url; } - */ $scope.members[chunk.target_user_id] = chunk; +/* + // Stale code for explicitly hammering the homeserver for every displayname & avatar_url + // get their display name and profile picture and set it to their // member entry in $scope.members. We HAVE to use $timeout with 0 delay // to make this function run AFTER the current digest cycle, else the @@ -193,6 +200,11 @@ angular.module('RoomController', ['ngSanitize']) } ); }); +*/ + + if (chunk.target_user_id in $rootScope.presence) { + updatePresence($rootScope.presence[chunk.target_user_id]); + } } else { // selectively update membership else it will nuke the picture and displayname too :/ @@ -232,7 +244,9 @@ angular.module('RoomController', ['ngSanitize']) if ($scope.textInput == "") { return; } - + + $scope.state.sending = true; + // Send the text message var promise; // FIXME: handle other commands too @@ -247,16 +261,17 @@ angular.module('RoomController', ['ngSanitize']) function() { console.log("Sent message"); $scope.textInput = ""; + $scope.state.sending = false; }, function(error) { $scope.feedback = "Failed to send: " + error.data.error; - }); + $scope.state.sending = false; + }); }; $scope.onInit = function() { - // $timeout(function() { document.getElementById('textInput').focus() }, 0); console.log("onInit"); - + // Does the room ID provided in the URL? var room_id_or_alias; if ($routeParams.room_id_or_alias) { @@ -284,7 +299,7 @@ angular.module('RoomController', ['ngSanitize']) else { // In case of issue, go to the default page console.log("Error: cannot extract room alias"); - $location.path("/"); + $location.url("/"); return; } } @@ -301,12 +316,14 @@ angular.module('RoomController', ['ngSanitize']) function () { // In case of issue, go to the default page console.log("Error: cannot resolve room alias"); - $location.path("/"); + $location.url("/"); }); } }; var onInit2 = function() { + eventHandlerService.reInitRoom($scope.room_id); + // Join the room matrixService.join($scope.room_id).then( function() { @@ -319,6 +336,7 @@ angular.module('RoomController', ['ngSanitize']) var chunk = response.data.chunk[i]; updateMemberList(chunk); } + eventStreamService.resume(); }, function(error) { $scope.feedback = "Failed get member list: " + error.data.error; @@ -354,36 +372,51 @@ angular.module('RoomController', ['ngSanitize']) matrixService.leave($scope.room_id).then( function(response) { console.log("Left room "); - $location.path("rooms"); + $location.url("rooms"); }, function(error) { $scope.feedback = "Failed to leave room: " + error.data.error; }); }; - $scope.sendImage = function(url) { - matrixService.sendImageMessage($scope.room_id, url).then( + $scope.sendImage = function(url, body) { + $scope.state.sending = true; + + matrixService.sendImageMessage($scope.room_id, url, body).then( function() { console.log("Image sent"); + $scope.state.sending = false; }, function(error) { $scope.feedback = "Failed to send image: " + error.data.error; + $scope.state.sending = false; }); }; $scope.imageFileToSend; $scope.$watch("imageFileToSend", function(newValue, oldValue) { if ($scope.imageFileToSend) { - // First download the image to the Internet - console.log("Uploading image..."); - mFileUpload.uploadFile($scope.imageFileToSend).then( - function(url) { - // Then share the URL - $scope.sendImage(url); + + $scope.state.sending = true; + + // Upload this image with its thumbnail to Internet + mFileUpload.uploadImageAndThumbnail($scope.imageFileToSend, THUMBNAIL_SIZE).then( + function(imageMessage) { + // imageMessage is complete message structure, send it as is + matrixService.sendMessage($scope.room_id, undefined, imageMessage).then( + function() { + console.log("Image message sent"); + $scope.state.sending = false; + }, + function(error) { + $scope.feedback = "Failed to send image message: " + error.data.error; + $scope.state.sending = false; + }); }, function(error) { $scope.feedback = "Can't upload image"; - } + $scope.state.sending = false; + } ); } }); diff --git a/webclient/room/room-directive.js b/webclient/room/room-directive.js index 94655336df..1a99a37abb 100644 --- a/webclient/room/room-directive.js +++ b/webclient/room/room-directive.js @@ -17,30 +17,30 @@ 'use strict'; angular.module('RoomController') -.directive('autoComplete', ['$timeout', function ($timeout) { +.directive('tabComplete', ['$timeout', function ($timeout) { return function (scope, element, attrs) { element.bind("keydown keypress", function (event) { // console.log("event: " + event.which); if (event.which === 9) { - if (!scope.autoCompleting) { // cache our starting text + if (!scope.tabCompleting) { // cache our starting text // console.log("caching " + element[0].value); - scope.autoCompleteOriginal = element[0].value; - scope.autoCompleting = true; + scope.tabCompleteOriginal = element[0].value; + scope.tabCompleting = true; } if (event.shiftKey) { - scope.autoCompleteIndex--; - if (scope.autoCompleteIndex < 0) { - scope.autoCompleteIndex = 0; + scope.tabCompleteIndex--; + if (scope.tabCompleteIndex < 0) { + scope.tabCompleteIndex = 0; } } else { - scope.autoCompleteIndex++; + scope.tabCompleteIndex++; } var searchIndex = 0; - var targetIndex = scope.autoCompleteIndex; - var text = scope.autoCompleteOriginal; + var targetIndex = scope.tabCompleteIndex; + var text = scope.tabCompleteOriginal; // console.log("targetIndex: " + targetIndex + ", text=" + text); @@ -90,17 +90,17 @@ angular.module('RoomController') element[0].className = ""; }, 150); element[0].value = text; - scope.autoCompleteIndex = 0; + scope.tabCompleteIndex = 0; } } else { - scope.autoCompleteIndex = 0; + scope.tabCompleteIndex = 0; } event.preventDefault(); } - else if (event.which !== 16 && scope.autoCompleting) { - scope.autoCompleting = false; - scope.autoCompleteIndex = 0; + else if (event.which !== 16 && scope.tabCompleting) { + scope.tabCompleting = false; + scope.tabCompleteIndex = 0; } }); }; diff --git a/webclient/room/room.html b/webclient/room/room.html index 8cae7ee515..95da067714 100644 --- a/webclient/room/room.html +++ b/webclient/room/room.html @@ -10,7 +10,7 @@ <div id="usersTableWrapper"> <table id="usersTable"> <tr ng-repeat="member in members | orderMembersList"> - <td class="userAvatar" ng-click="goToUserPage(member.id)"> + <td class="userAvatar mouse-pointer" ng-click="goToUserPage(member.id)"> <img class="userAvatarImage" ng-src="{{member.avatar_url || 'img/default-profile.jpg'}}" alt="{{ member.displayname || member.id.substr(0, member.id.indexOf(':')) }}" @@ -26,22 +26,36 @@ </div> <div id="messageTableWrapper" keep-scroll> + <!-- FIXME: need to have better timestamp semantics than the (msg.content.hsob_ts || msg.ts) hack below --> <table id="messageTable" infinite-scroll="paginateMore()"> <tr ng-repeat="msg in events.rooms[room_id].messages" - ng-class="(events.rooms[room_id].messages[$index - 1].user_id !== msg.user_id ? 'differentUser' : '') + (msg.user_id === state.user_id ? ' mine' : '')" scroll-item> + ng-class="(events.rooms[room_id].messages[$index + 1].user_id !== msg.user_id ? 'differentUser' : '') + (msg.user_id === state.user_id ? ' mine' : '')" scroll-item> <td class="leftBlock"> <div class="sender" ng-hide="events.rooms[room_id].messages[$index - 1].user_id === msg.user_id">{{ members[msg.user_id].displayname || msg.user_id }}</div> - <div class="timestamp">{{ msg.content.hsob_ts | date:'MMM d HH:mm:ss' }}</div> + <div class="timestamp">{{ (msg.content.hsob_ts || msg.ts) | date:'MMM d HH:mm:ss' }}</div> </td> <td class="avatar"> <img class="avatarImage" ng-src="{{ members[msg.user_id].avatar_url || 'img/default-profile.jpg' }}" width="32" height="32" ng-hide="events.rooms[room_id].messages[$index - 1].user_id === msg.user_id || msg.user_id === state.user_id"/> </td> - <td ng-class="!msg.content.membership_target ? (msg.content.msgtype === 'm.emote' ? 'emote text' : 'text') : 'membership text'"> + <td ng-class="!msg.content.membership ? (msg.content.msgtype === 'm.emote' ? 'emote text' : 'text') : 'membership text'"> <div class="bubble"> + <span ng-hide='msg.type !== "m.room.member"'> + {{ members[msg.user_id].displayname || msg.user_id }} + {{ {"join": "joined", "leave": "left", "invite": "invited"}[msg.content.membership] }} + {{ msg.content.target_id || '' }} + </span> <span ng-hide='msg.content.msgtype !== "m.emote"' ng-bind-html="'* ' + (members[msg.user_id].displayname || msg.user_id) + ' ' + msg.content.body | linky:'_blank'"/> - <span ng-hide='msg.content.msgtype !== "m.text"' ng-bind-html="msg.content.body | linky:'_blank'"/> - <img class="image" ng-hide='msg.content.msgtype !== "m.image"' ng-src="{{ msg.content.url }}" alt="{{ msg.content.body }}"/> + <span ng-hide='msg.content.msgtype !== "m.text"' ng-bind-html="((msg.content.msgtype === 'm.text') ? msg.content.body : '') | linky:'_blank'"/> + <div ng-show='msg.content.msgtype === "m.image"'> + <div ng-hide='msg.content.thumbnail_url' ng-style="msg.content.body.h && { 'height' : (msg.content.body.h < 320) ? msg.content.body.h : 320}"> + <img class="image" ng-src="{{ msg.content.url }}"/> + </div> + <div ng-show='msg.content.thumbnail_url' ng-style="{ 'height' : msg.content.thumbnail_info.h }"> + <img class="image mouse-pointer" ng-src="{{ msg.content.thumbnail_url }}" + ng-click="$parent.fullScreenImageURL = msg.content.url"/> + </div> + </div> </div> </td> <td class="rightBlock"> @@ -63,30 +77,16 @@ {{ state.user_id }} </td> <td width="*" style="min-width: 100px"> - <input id="mainInput" ng-model="textInput" ng-enter="send()" ng-focus="true" auto-complete/> + <input id="mainInput" ng-model="textInput" ng-enter="send()" ng-focus="true" autocomplete="off" tab-complete/> </td> - <td width="1"> + <td width="150px"> <button ng-click="send()">Send</button> + <button m-file-input="imageFileToSend">Send Image</button> </td> <td width="1"> </td> </tr> - <tr> - <td> - </td> - <td> - <input id="mainInput" ng-model="imageURLToSend" ng-enter="sendImage(imageURLToSend)" placeholder="Image URL"/> - </td> - <td width="100px"> - <button ng-click="sendImage(imageURLToSend)">Send URL</button> - </td> - <!-- TODO: To enable once we have an upload server - <td width="100px"> - <button m-file-input="imageFileToSend">Send Image</button> - </td> - --> - </tr> </table> <span> @@ -103,4 +103,8 @@ </div> </div> + <div id="room-fullscreen-image" ng-show="fullScreenImageURL" ng-click="fullScreenImageURL = undefined;"> + <img ng-src="{{ fullScreenImageURL }}"/> + </div> + </div> diff --git a/webclient/rooms/rooms-controller.js b/webclient/rooms/rooms-controller.js index a237b59b4e..d9c8baff47 100644 --- a/webclient/rooms/rooms-controller.js +++ b/webclient/rooms/rooms-controller.js @@ -17,8 +17,8 @@ limitations under the License. 'use strict'; angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', 'eventHandlerService']) -.controller('RoomsController', ['$scope', '$location', 'matrixService', 'mFileUpload', 'eventHandlerService', - function($scope, $location, matrixService, mFileUpload, eventHandlerService) { +.controller('RoomsController', ['$scope', '$location', 'matrixService', 'mFileUpload', 'eventHandlerService', 'eventStreamService', + function($scope, $location, matrixService, mFileUpload, eventHandlerService, eventStreamService) { $scope.rooms = {}; $scope.public_rooms = []; @@ -48,6 +48,8 @@ angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', linkNewEmail: "", // the email entry box emailBeingAuthed: undefined, // to populate verification text authTokenId: undefined, // the token id from the IS + clientSecret: undefined, // our client secret + sendAttempt: 1, emailCode: "", // the code entry box linkedEmailList: matrixService.config().emailList // linked email list }; @@ -59,7 +61,7 @@ angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', // FIXME push membership to top level key to match /im/sync event.membership = event.content.membership; // FIXME bodge a nicer name than the room ID for this invite. - event.room_alias = event.user_id + "'s room"; + event.room_display_name = event.user_id + "'s room"; $scope.rooms[event.room_id] = event; } }); @@ -70,14 +72,20 @@ angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', if (alias) { // use the existing alias from storage data[i].room_alias = alias; + data[i].room_display_name = alias; } - else if (data[i].room_alias) { + else if (data[i].aliases && data[i].aliases[0]) { // save the mapping - matrixService.createRoomIdToAliasMapping(data[i].room_id, data[i].room_alias); + // TODO: select the smarter alias from the array + matrixService.createRoomIdToAliasMapping(data[i].room_id, data[i].aliases[0]); + data[i].room_display_name = data[i].aliases[0]; + } + else if (data[i].membership == "invite" && "inviter" in data[i]) { + data[i].room_display_name = data[i].inviter + "'s room" } else { // last resort use the room id - data[i].room_alias = data[i].room_id; + data[i].room_display_name = data[i].room_id; } } return data; @@ -87,11 +95,16 @@ angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', // List all rooms joined or been invited to matrixService.rooms().then( function(response) { - var data = assignRoomAliases(response.data); + var data = assignRoomAliases(response.data.rooms); $scope.feedback = "Success"; for (var i=0; i<data.length; i++) { $scope.rooms[data[i].room_id] = data[i]; } + + var presence = response.data.presence; + for (var i = 0; i < presence.length; ++i) { + eventHandlerService.handleEvent(presence[i], false); + } }, function(error) { $scope.feedback = "Failure: " + error.data; @@ -102,6 +115,8 @@ angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', $scope.public_rooms = assignRoomAliases(response.data.chunk); } ); + + eventStreamService.resume(); }; $scope.createNewRoom = function(room_id, isPrivate) { @@ -128,17 +143,17 @@ angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', // Go to a room $scope.goToRoom = function(room_id) { // Simply open the room page on this room id - //$location.path("room/" + room_id); + //$location.url("room/" + room_id); matrixService.join(room_id).then( function(response) { if (response.data.hasOwnProperty("room_id")) { if (response.data.room_id != room_id) { - $location.path("room/" + response.data.room_id); + $location.url("room/" + response.data.room_id); return; } } - $location.path("room/" + room_id); + $location.url("room/" + room_id); }, function(error) { $scope.feedback = "Can't join room: " + error.data; @@ -150,7 +165,7 @@ angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', matrixService.joinAlias(room_alias).then( function(response) { // Go to this room - $location.path("room/" + room_alias); + $location.url("room/" + room_alias); }, function(error) { $scope.feedback = "Can't join room: " + error.data; @@ -206,11 +221,27 @@ angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', ); }; + var generateClientSecret = function() { + var ret = ""; + var chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"; + + for (var i = 0; i < 32; i++) { + ret += chars.charAt(Math.floor(Math.random() * chars.length)); + } + + return ret; + }; + + $scope.linkEmail = function(email) { - matrixService.linkEmail(email).then( + if (email != $scope.linkedEmails.emailBeingAuthed) { + $scope.linkedEmails.clientSecret = generateClientSecret(); + $scope.linkedEmails.sendAttempt = 1; + } + matrixService.linkEmail(email, $scope.linkedEmails.clientSecret, $scope.linkedEmails.sendAttempt).then( function(response) { if (response.data.success === true) { - $scope.linkedEmails.authTokenId = response.data.tokenId; + $scope.linkedEmails.authTokenId = response.data.sid; $scope.emailFeedback = "You have been sent an email."; $scope.linkedEmails.emailBeingAuthed = email; } @@ -230,28 +261,34 @@ angular.module('RoomsController', ['matrixService', 'mFileInput', 'mFileUpload', $scope.emailFeedback = "You have not requested a code with this email."; return; } - matrixService.authEmail(matrixService.config().user_id, tokenId, code).then( + matrixService.authEmail(matrixService.config().user_id, tokenId, code, $scope.linkedEmails.clientSecret).then( function(response) { if ("success" in response.data && response.data.success === false) { $scope.emailFeedback = "Failed to authenticate email."; return; } - var config = matrixService.config(); - var emailList = {}; - if ("emailList" in config) { - emailList = config.emailList; - } - emailList[response.address] = response; - // save the new email list - config.emailList = emailList; - matrixService.setConfig(config); - matrixService.saveConfig(); - // invalidate the email being authed and update UI. - $scope.linkedEmails.emailBeingAuthed = undefined; - $scope.emailFeedback = ""; - $scope.linkedEmails.linkedEmailList = emailList; - $scope.linkedEmails.linkNewEmail = ""; - $scope.linkedEmails.emailCode = ""; + matrixService.bindEmail(matrixService.config().user_id, tokenId, $scope.linkedEmails.clientSecret).then( + function(response) { + var config = matrixService.config(); + var emailList = {}; + if ("emailList" in config) { + emailList = config.emailList; + } + emailList[$scope.linkedEmails.emailBeingAuthed] = response; + // save the new email list + config.emailList = emailList; + matrixService.setConfig(config); + matrixService.saveConfig(); + // invalidate the email being authed and update UI. + $scope.linkedEmails.emailBeingAuthed = undefined; + $scope.emailFeedback = ""; + $scope.linkedEmails.linkedEmailList = emailList; + $scope.linkedEmails.linkNewEmail = ""; + $scope.linkedEmails.emailCode = ""; + }, function(reason) { + $scope.emailFeedback = "Failed to link email: " + reason; + } + ); }, function(reason) { $scope.emailFeedback = "Failed to auth email: " + reason; diff --git a/webclient/rooms/rooms.html b/webclient/rooms/rooms.html index 2602209bd3..ba3b7d8bad 100644 --- a/webclient/rooms/rooms.html +++ b/webclient/rooms/rooms.html @@ -65,7 +65,7 @@ <div class="rooms" ng-repeat="(rm_id, room) in rooms"> <div> - <a href="#/room/{{ room.room_alias ? room.room_alias : rm_id }}" >{{ room.room_alias }}</a> {{room.membership === 'invite' ? ' (invited)' : ''}} + <a href="#/room/{{ room.room_alias ? room.room_alias : rm_id }}" >{{ room.room_display_name }}</a> {{room.membership === 'invite' ? ' (invited)' : ''}} </div> </div> <br/> |