diff options
| -rw-r--r-- | synapse/handlers/register.py | 5 | ||||
| -rw-r--r-- | synapse/storage/registration.py | 40 | ||||
| -rw-r--r-- | synapse/storage/schema/delta/30/as_users.py | 23 | ||||
| -rw-r--r-- | synapse/storage/stream.py | 9 |
4 files changed, 58 insertions, 19 deletions
diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py index e2ace6a4e5..6ffb8c0da6 100644 --- a/synapse/handlers/register.py +++ b/synapse/handlers/register.py @@ -182,6 +182,8 @@ class RegistrationHandler(BaseHandler): errcode=Codes.EXCLUSIVE ) + service_id = service.id if service.is_exclusive_user(user_id) else None + yield self.check_user_id_not_appservice_exclusive( user_id, allowed_appservice=service ) @@ -190,7 +192,8 @@ class RegistrationHandler(BaseHandler): yield self.store.register( user_id=user_id, token=token, - password_hash="" + password_hash="", + appservice_id=service_id, ) yield registered_user(self.distributor, user) defer.returnValue((user_id, token)) diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py index ad1157f979..aa49f53458 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -76,7 +76,7 @@ class RegistrationStore(SQLBaseStore): @defer.inlineCallbacks def register(self, user_id, token, password_hash, - was_guest=False, make_guest=False): + was_guest=False, make_guest=False, appservice_id=None): """Attempts to register an account. Args: @@ -87,16 +87,32 @@ class RegistrationStore(SQLBaseStore): upgraded to a non-guest account. make_guest (boolean): True if the the new user should be guest, false to add a regular user account. + appservice_id (str): The ID of the appservice registering the user. Raises: StoreError if the user_id could not be registered. """ yield self.runInteraction( "register", - self._register, user_id, token, password_hash, was_guest, make_guest + self._register, + user_id, + token, + password_hash, + was_guest, + make_guest, + appservice_id ) self.is_guest.invalidate((user_id,)) - def _register(self, txn, user_id, token, password_hash, was_guest, make_guest): + def _register( + self, + txn, + user_id, + token, + password_hash, + was_guest, + make_guest, + appservice_id + ): now = int(self.clock.time()) next_id = self._access_tokens_id_gen.get_next() @@ -111,9 +127,21 @@ class RegistrationStore(SQLBaseStore): [password_hash, now, 1 if make_guest else 0, user_id]) else: txn.execute("INSERT INTO users " - "(name, password_hash, creation_ts, is_guest) " - "VALUES (?,?,?,?)", - [user_id, password_hash, now, 1 if make_guest else 0]) + "(" + " name," + " password_hash," + " creation_ts," + " is_guest," + " appservice_id" + ") " + "VALUES (?,?,?,?,?)", + [ + user_id, + password_hash, + now, + 1 if make_guest else 0, + appservice_id, + ]) except self.database_engine.module.IntegrityError: raise StoreError( 400, "User ID already taken.", errcode=Codes.USER_IN_USE diff --git a/synapse/storage/schema/delta/30/as_users.py b/synapse/storage/schema/delta/30/as_users.py index 4da3c59de2..4f6e9dd540 100644 --- a/synapse/storage/schema/delta/30/as_users.py +++ b/synapse/storage/schema/delta/30/as_users.py @@ -52,12 +52,17 @@ def run_upgrade(cur, database_engine, config, *args, **kwargs): " service (IDs %s and %s); assigning arbitrarily to %s" % (user_id, owned[user_id], appservice.id, owned[user_id]) ) - owned[user_id] = appservice.id - - for user_id, as_id in owned.items(): - cur.execute( - database_engine.convert_param_style( - "UPDATE users SET appservice_id = ? WHERE name = ?" - ), - (as_id, user_id) - ) + owned.setdefault(appservice.id, []).append(user_id) + + for as_id, user_ids in owned.items(): + n = 100 + user_chunks = (user_ids[i:i + 100] for i in xrange(0, len(user_ids), n)) + for chunk in user_chunks: + cur.execute( + database_engine.convert_param_style( + "UPDATE users SET appservice_id = ? WHERE name IN (%s)" % ( + ",".join("?" for _ in chunk), + ) + ), + [as_id] + chunk + ) diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 8908d5b5da..7f4a827528 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -184,6 +184,9 @@ class StreamStore(SQLBaseStore): @defer.inlineCallbacks def get_room_events_stream_for_room(self, room_id, from_key, to_key, limit=0, order='DESC'): + # Note: If from_key is None then we return in topological order. This + # is because in that case we're using this as a "get the last few messages + # in a room" function, rather than "get new messages since last sync" if from_key is not None: from_id = RoomStreamToken.parse_stream_token(from_key).stream else: @@ -217,8 +220,8 @@ class StreamStore(SQLBaseStore): " room_id = ?" " AND not outlier" " AND stream_ordering <= ?" - " ORDER BY stream_ordering %s LIMIT ?" - ) % (order,) + " ORDER BY topological_ordering %s, stream_ordering %s LIMIT ?" + ) % (order, order,) txn.execute(sql, (room_id, to_id, limit)) rows = self.cursor_to_dict(txn) @@ -232,7 +235,7 @@ class StreamStore(SQLBaseStore): get_prev_content=True ) - self._set_before_and_after(ret, rows, topo_order=False) + self._set_before_and_after(ret, rows, topo_order=from_id is None) if order.lower() == "desc": ret.reverse() |