From ec2b5d8c284451179c0f7b72c46dda0dd81f6f5f Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 1 Dec 2014 16:21:17 +0000 Subject: Store full JSON of events in db --- synapse/storage/__init__.py | 19 +++++++++++ synapse/storage/_base.py | 77 ++++++++++++------------------------------- synapse/storage/schema/im.sql | 10 ++++++ 3 files changed, 50 insertions(+), 56 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index f15e3dfe62..205d125642 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -155,6 +155,25 @@ class DataStore(RoomMemberStore, RoomStore, if hasattr(event, "outlier"): outlier = event.outlier + event_dict = { + k: v + for k, v in event.get_full_dict().items() + if k not in [ + "redacted", + "redacted_because", + ] + } + + self._simple_insert_txn( + txn, + table="event_json", + values={ + "event_id": event.event_id, + "json": json.dumps(event_dict, separators=(',', ':')), + }, + or_replace=True, + ) + vals = { "topological_ordering": event.depth, "event_id": event.event_id, diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 4881f03368..bb61c20150 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -479,66 +479,30 @@ class SQLBaseStore(object): ) def _parse_events_txn(self, txn, rows): - events = [self._parse_event_from_row(r) for r in rows] - - select_event_sql = ( - "SELECT * FROM events WHERE event_id = ? ORDER BY rowid asc" - ) - - for i, ev in enumerate(events): - signatures = self._get_event_signatures_txn( - txn, ev.event_id, + event_ids = [r["event_id"] for r in rows] + + events = [] + for event_id in event_ids: + js = self._simple_select_one_onecol_txn( + txn, + table="event_json", + keyvalues={"event_id": event_id}, + retcol="json", + allow_none=True, ) - ev.signatures = { - n: { - k: encode_base64(v) for k, v in s.items() - } - for n, s in signatures.items() - } - - hashes = self._get_event_content_hashes_txn( - txn, ev.event_id, - ) + if not js: + # FIXME (erikj): What should we actually do here? + continue - ev.hashes = { - k: encode_base64(v) for k, v in hashes.items() - } - - prevs = self._get_prev_events_and_state(txn, ev.event_id) - - ev.prev_events = [ - (e_id, h) - for e_id, h, is_state in prevs - if is_state == 0 - ] - - ev.auth_events = self._get_auth_events(txn, ev.event_id) - - if hasattr(ev, "state_key"): - ev.prev_state = [ - (e_id, h) - for e_id, h, is_state in prevs - if is_state == 1 - ] - - if hasattr(ev, "replaces_state"): - # Load previous state_content. - # FIXME (erikj): Handle multiple prev_states. - cursor = txn.execute( - select_event_sql, - (ev.replaces_state,) - ) - prevs = self.cursor_to_dict(cursor) - if prevs: - prev = self._parse_event_from_row(prevs[0]) - ev.prev_content = prev.content + d = json.loads(js) - if not hasattr(ev, "redacted"): - logger.debug("Doesn't have redacted key: %s", ev) - ev.redacted = self._has_been_redacted_txn(txn, ev) + ev = self.event_factory.create_event( + etype=d["type"], + **d + ) - if ev.redacted: + if hasattr(ev, "redacted") and ev.redacted: # Get the redaction event. select_event_sql = "SELECT * FROM events WHERE event_id = ?" txn.execute(select_event_sql, (ev.redacted,)) @@ -549,9 +513,10 @@ class SQLBaseStore(object): if del_evs: ev = prune_event(ev) - events[i] = ev ev.redacted_because = del_evs[0] + events.append(ev) + return events def _has_been_redacted_txn(self, txn, event): diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql index 8ba732a23b..cb0c494ddf 100644 --- a/synapse/storage/schema/im.sql +++ b/synapse/storage/schema/im.sql @@ -32,6 +32,16 @@ CREATE INDEX IF NOT EXISTS events_stream_ordering ON events (stream_ordering); CREATE INDEX IF NOT EXISTS events_topological_ordering ON events (topological_ordering); CREATE INDEX IF NOT EXISTS events_room_id ON events (room_id); + +CREATE TABLE IF NOT EXISTS event_json( + event_id TEXT NOT NULL, + json BLOB NOT NULL, + CONSTRAINT ev_j_uniq UNIQUE (event_id) +); + +CREATE INDEX IF NOT EXISTS event_json_id ON event_json(event_id); + + CREATE TABLE IF NOT EXISTS state_events( event_id TEXT NOT NULL, room_id TEXT NOT NULL, -- cgit 1.4.1 From 5d7c9ab7898f2721aa3f60ab76c53dc44322be77 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 4 Dec 2014 11:27:59 +0000 Subject: Begin converting things to use the new Event structure --- synapse/api/auth.py | 11 ++----- synapse/events/__init__.py | 3 ++ synapse/federation/replication.py | 12 +++----- synapse/handlers/_base.py | 65 +++++++++++++++++++++++++++++++-------- synapse/handlers/federation.py | 9 ++---- synapse/handlers/room.py | 1 + synapse/state.py | 18 ++++++++--- synapse/storage/signatures.py | 16 ++++++++++ 8 files changed, 96 insertions(+), 39 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 50d4be113f..5261c3e3bf 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -393,18 +393,11 @@ class Auth(object): if member_event.content["membership"] == Membership.JOIN: auth_events.append(member_event.event_id) - hashes = yield self.store.get_event_reference_hashes( + auth_events = yield self.store.add_event_hashes( auth_events ) - hashes = [ - { - k: encode_base64(v) for k, v in h.items() - if k == "sha256" - } - for h in hashes - ] - defer.returnValue(zip(auth_events, hashes)) + defer.returnValue(auth_events) @log_function def _can_send_event(self, event, auth_events): diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index 6748198917..6a05ba2d16 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -81,6 +81,9 @@ class EventBase(object): type = _event_dict_property("type") user_id = _event_dict_property("sender") + def is_state(self): + return hasattr(self, "state_key") + def get_dict(self): d = dict(self._original) d.update({ diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 01f87fe423..bd56a4c108 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -112,7 +112,7 @@ class ReplicationLayer(object): self.query_handlers[query_type] = handler @log_function - def send_pdu(self, pdu): + def send_pdu(self, pdu, destinations): """Informs the replication layer about a new PDU generated within the home server that should be transmitted to others. @@ -131,7 +131,7 @@ class ReplicationLayer(object): logger.debug("[%s] transaction_layer.enqueue_pdu... ", pdu.event_id) # TODO, add errback, etc. - self._transaction_queue.enqueue_pdu(pdu, order) + self._transaction_queue.enqueue_pdu(pdu, destinations, order) logger.debug( "[%s] transaction_layer.enqueue_pdu... done", @@ -705,15 +705,13 @@ class _TransactionQueue(object): @defer.inlineCallbacks @log_function - def enqueue_pdu(self, pdu, order): + def enqueue_pdu(self, pdu, destinations, order): # We loop through all destinations to see whether we already have # a transaction in progress. If we do, stick it in the pending_pdus # table and we'll get back to it later. - destinations = set([ - d for d in pdu.destinations - if d != self.server_name - ]) + destinations = set(destinations) + destinations.remove(self.server_name) logger.debug("Sending to: %s", str(destinations)) diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index a45715bf60..890b51be30 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -15,11 +15,11 @@ from twisted.internet import defer -from synapse.api.errors import LimitExceededError +from synapse.api.errors import LimitExceededError, SynapseError from synapse.util.async import run_on_reactor from synapse.crypto.event_signing import add_hashes_and_signatures from synapse.api.events.room import RoomMemberEvent -from synapse.api.constants import Membership +from synapse.api.constants import Membership, EventTypes from synapse.events.snapshot import EventSnapshot, EventContext @@ -59,7 +59,7 @@ class BaseHandler(object): ) @defer.inlineCallbacks - def _handle_new_client_event(self, builder): + def _create_new_client_event(self, builder): latest_ret = yield self.store.get_latest_events_in_room( builder.room_id, ) @@ -67,16 +67,27 @@ class BaseHandler(object): depth = max([d for _, _, d in latest_ret]) prev_events = [(e, h) for e, h, _ in latest_ret] - group, curr_state = yield self.state_handler.resolve_state_groups( - [e for e, _ in prev_events] - ) + state_handler = self.state_handler + if builder.is_state(): + ret = yield state_handler.resolve_state_groups( + [e for e, _ in prev_events], + event_type=builder.event_type, + state_key=builder.state_key, + ) - snapshot = EventSnapshot( - prev_events=prev_events, - depth=depth, - current_state=curr_state, - current_state_group=group, - ) + group, curr_state, prev_state = ret + + prev_state = yield self.store.add_event_hashes( + prev_state + ) + + builder.prev_state = prev_state + else: + group, curr_state, _ = yield state_handler.resolve_state_groups( + [e for e, _ in prev_events], + ) + + builder.internal_metadata.state_group = group builder.prev_events = prev_events builder.depth = depth @@ -103,9 +114,39 @@ class BaseHandler(object): auth_events=curr_auth_events, ) + defer.returnValue( + (event, context,) + ) + + @defer.inlineCallbacks + def _handle_new_client_event(self, event, context): + # We now need to go and hit out to wherever we need to hit out to. + self.auth.check(event, auth_events=context.auth_events) + yield self.store.persist_event(event) + destinations = set() + for k, s in context.current_state.items(): + try: + if k[0] == EventTypes.Member: + if s.content["membership"] == Membership.JOIN: + destinations.add( + self.hs.parse_userid(s.state_key).domain + ) + except SynapseError: + logger.warn( + "Failed to get destination from event %s", s.event_id + ) + + yield self.notifier.on_new_room_event(event) + + federation_handler = self.hs.get_handlers().federation_handler + yield federation_handler.handle_new_event( + event, + None, + destinations=destinations, + ) @defer.inlineCallbacks def _on_new_room_event(self, event, snapshot, extra_destinations=[], diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 925eb5376e..7bd36e415e 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -76,7 +76,7 @@ class FederationHandler(BaseHandler): @log_function @defer.inlineCallbacks - def handle_new_event(self, event, snapshot): + def handle_new_event(self, event, snapshot, destinations): """ Takes in an event from the client to server side, that has already been authed and handled by the state module, and sends it to any remote home servers that may be interested. @@ -92,12 +92,7 @@ class FederationHandler(BaseHandler): yield run_on_reactor() - pdu = event - - if not hasattr(pdu, "destinations") or not pdu.destinations: - pdu.destinations = [] - - yield self.replication_layer.send_pdu(pdu) + yield self.replication_layer.send_pdu(event, destinations) @log_function @defer.inlineCallbacks diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 6e1c37df03..52a9788823 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -378,6 +378,7 @@ class RoomMemberHandler(BaseHandler): else: # This is not a JOIN, so we can handle it normally. + # FIXME: This isn't idempotency. if prev_state and prev_state.membership == event.membership: # double same action, treat this event as a NOOP. defer.returnValue({}) diff --git a/synapse/state.py b/synapse/state.py index 430665f7ba..8a556a27f6 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -89,7 +89,7 @@ class StateHandler(object): ids = [e for e, _ in event.prev_events] ret = yield self.resolve_state_groups(ids) - state_group, new_state = ret + state_group, new_state, _ = ret event.old_state_events = copy.deepcopy(new_state) @@ -137,7 +137,7 @@ class StateHandler(object): @defer.inlineCallbacks @log_function - def resolve_state_groups(self, event_ids): + def resolve_state_groups(self, event_ids, event_type=None, state_key=""): """ Given a list of event_ids this method fetches the state at each event, resolves conflicts between them and returns them. @@ -156,7 +156,10 @@ class StateHandler(object): (e.type, e.state_key): e for e in state_list } - defer.returnValue((name, state)) + prev_state = state.get((event_type, state_key), None) + if prev_state: + prev_state = prev_state.event_id + defer.returnValue((name, state, [prev_state])) state = {} for group, g_state in state_groups.items(): @@ -177,6 +180,13 @@ class StateHandler(object): if len(v.values()) > 1 } + if event_type: + prev_states = conflicted_state.get( + (event_type, state_key), {} + ).keys() + else: + prev_states = [] + try: new_state = {} new_state.update(unconflicted_state) @@ -186,7 +196,7 @@ class StateHandler(object): logger.exception("Failed to resolve state") raise - defer.returnValue((None, new_state)) + defer.returnValue((None, new_state, prev_states)) def _get_power_level_from_event_state(self, event, user_id): if hasattr(event, "old_state_events") and event.old_state_events: diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index eea4f21065..e2f11c7ffc 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -15,6 +15,8 @@ from _base import SQLBaseStore +from syutil.base64util import encode_base64 + class SignatureStore(SQLBaseStore): """Persistence for event signatures and hashes""" @@ -67,6 +69,20 @@ class SignatureStore(SQLBaseStore): f ) + def add_event_hashes(self, event_ids): + hashes = yield self.store.get_event_reference_hashes( + event_ids + ) + hashes = [ + { + k: encode_base64(v) for k, v in h.items() + if k == "sha256" + } + for h in hashes + ] + + defer.returnValue(zip(event_ids, hashes)) + def _get_event_reference_hashes_txn(self, txn, event_id): """Get all the hashes for a given PDU. Args: -- cgit 1.4.1 From c31dba86ec40853f27c70ae13409ca3332052cc1 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 4 Dec 2014 15:50:01 +0000 Subject: Convert rest and handlers to use new event structure --- synapse/crypto/event_signing.py | 2 +- synapse/events/__init__.py | 8 +-- synapse/events/builder.py | 5 +- synapse/federation/replication.py | 18 ++---- synapse/handlers/_base.py | 16 +++-- synapse/handlers/directory.py | 20 +++--- synapse/handlers/federation.py | 19 +++--- synapse/handlers/message.py | 23 ++++++- synapse/handlers/profile.py | 19 +++--- synapse/handlers/room.py | 130 ++++++++++++++++---------------------- synapse/rest/base.py | 2 +- synapse/rest/room.py | 112 ++++++++++++++------------------ synapse/server.py | 8 +++ synapse/storage/signatures.py | 5 +- 14 files changed, 188 insertions(+), 199 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/crypto/event_signing.py b/synapse/crypto/event_signing.py index 209f9d73fe..b189f0bb2b 100644 --- a/synapse/crypto/event_signing.py +++ b/synapse/crypto/event_signing.py @@ -15,7 +15,7 @@ # limitations under the License. -from synapse.api.events.utils import prune_event +from synapse.events.utils import prune_event from syutil.jsonutil import encode_canonical_json from syutil.base64util import encode_base64, decode_base64 from syutil.crypto.jsonsign import sign_json diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index 6a05ba2d16..58edf2bc8f 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -85,10 +85,10 @@ class EventBase(object): return hasattr(self, "state_key") def get_dict(self): - d = dict(self._original) + d = dict(self._event_dict) d.update({ - "signatures": self._signatures, - "unsigned": self._unsigned, + "signatures": self.signatures, + "unsigned": self.unsigned, }) return d @@ -128,7 +128,7 @@ class FrozenEvent(EventBase): @staticmethod def from_event(event): e = FrozenEvent( - event.event_dict() + event.get_pdu_json() ) e.internal_metadata = event.internal_metadata diff --git a/synapse/events/builder.py b/synapse/events/builder.py index 39b4d2a2ab..0b8caf9318 100644 --- a/synapse/events/builder.py +++ b/synapse/events/builder.py @@ -22,10 +22,13 @@ from synapse.util.stringutils import random_string class EventBuilder(EventBase): def __init__(self, key_values={}): - super(FrozenEvent, self).__init__( + super(EventBuilder, self).__init__( key_values, ) + def update_event_key(self, key, value): + self._event_dict[key] = value + def update_event_keys(self, other_dict): self._event_dict.update(other_dict) diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index bd56a4c108..b11df9e5c6 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -74,6 +74,7 @@ class ReplicationLayer(object): self._clock = hs.get_clock() self.event_factory = hs.get_event_factory() + self.event_builder_factory = hs.get_event_builder_factory() def set_handler(self, handler): """Sets the handler that the replication layer will use to communicate @@ -658,19 +659,14 @@ class ReplicationLayer(object): return "" % self.server_name def event_from_pdu_json(self, pdu_json, outlier=False): - #TODO: Check we have all the PDU keys here - pdu_json.setdefault("hashes", {}) - pdu_json.setdefault("signatures", {}) - sender = pdu_json.pop("sender", None) - if sender is not None: - pdu_json["user_id"] = sender - state_hash = pdu_json.get("unsigned", {}).pop("state_hash", None) - if state_hash is not None: - pdu_json["state_hash"] = state_hash - return self.event_factory.create_event( - pdu_json["type"], outlier=outlier, **pdu_json + builder = self.event_builder_factory.new( + pdu_json ) + builder.internal_metadata = outlier + + return builder.build() + class _TransactionQueue(object): """This class makes sure we only have one transaction in flight at diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 890b51be30..4052d0e1e7 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -46,6 +46,8 @@ class BaseHandler(object): self.signing_key = hs.config.signing_key[0] self.server_name = hs.hostname + self.event_builder_factory = hs.get_event_builder_factory() + def ratelimit(self, user_id): time_now = self.clock.time() allowed, time_allowed = self.ratelimiter.send_message( @@ -92,7 +94,7 @@ class BaseHandler(object): builder.prev_events = prev_events builder.depth = depth - auth_events = yield self.auth.get_event_auth(builder, curr_state) + auth_events = yield self.auth.get_auth_events(builder, curr_state) builder.update_event_key("auth_events", auth_events) @@ -105,7 +107,7 @@ class BaseHandler(object): auth_ids = zip(*auth_events)[0] curr_auth_events = { k: v - for k, v in curr_state + for k, v in curr_state.items() if v.event_id in auth_ids } @@ -119,14 +121,16 @@ class BaseHandler(object): ) @defer.inlineCallbacks - def _handle_new_client_event(self, event, context): + def handle_new_client_event(self, event, context, extra_destinations=[], + extra_users=[], suppress_auth=False): # We now need to go and hit out to wherever we need to hit out to. - self.auth.check(event, auth_events=context.auth_events) + if not suppress_auth: + self.auth.check(event, auth_events=context.auth_events) yield self.store.persist_event(event) - destinations = set() + destinations = set(extra_destinations) for k, s in context.current_state.items(): try: if k[0] == EventTypes.Member: @@ -139,7 +143,7 @@ class BaseHandler(object): "Failed to get destination from event %s", s.event_id ) - yield self.notifier.on_new_room_event(event) + yield self.notifier.on_new_room_event(event, extra_users=extra_users) federation_handler = self.hs.get_handlers().federation_handler yield federation_handler.handle_new_event( diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py index b95c4b8bf7..76fb897f20 100644 --- a/synapse/handlers/directory.py +++ b/synapse/handlers/directory.py @@ -148,16 +148,12 @@ class DirectoryHandler(BaseHandler): def send_room_alias_update_event(self, user_id, room_id): aliases = yield self.store.get_aliases_for_room(room_id) - event = self.event_factory.create_event( - etype=RoomAliasesEvent.TYPE, - state_key=self.hs.hostname, - room_id=room_id, - user_id=user_id, - content={"aliases": aliases}, - ) - - snapshot = yield self.store.snapshot_room(event) + msg_handler = self.hs.get_handlers().message_handler + yield msg_handler.handle_event({ + "type": RoomAliasesEvent.TYPE, + "state_key": self.hs.hostname, + "room_id": room_id, + "sender": user_id, + "content": {"aliases": aliases}, + }) - yield self._on_new_room_event( - event, snapshot, extra_users=[user_id], suppress_auth=True - ) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 7bd36e415e..b4a28ea3cb 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -421,16 +421,17 @@ class FederationHandler(BaseHandler): join event for the room and return that. We don *not* persist or process it until the other server has signed it and sent it back. """ - event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, - content={"membership": Membership.JOIN}, - room_id=context, - user_id=user_id, - state_key=user_id, - ) + builder = self.event_builder_factory.new({ + "type": RoomMemberEvent.TYPE, + "content": {"membership": Membership.JOIN}, + "room_id": context, + "sender": user_id, + "state_key": user_id, + }) - snapshot = yield self.store.snapshot_room(event) - snapshot.fill_out_prev_events(event) + event, context = yield self._create_new_client_event( + builder=builder, + ) yield self.state_handler.annotate_event_with_state(event) yield self.auth.add_auth_events(event) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 269d6622e1..485d8e8179 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -15,7 +15,7 @@ from twisted.internet import defer -from synapse.api.constants import Membership +from synapse.api.constants import EventTypes, Membership from synapse.api.errors import RoomError from synapse.streams.config import PaginationConfig from synapse.util.logcontext import PreserveLoggingContext @@ -133,6 +133,27 @@ class MessageHandler(BaseHandler): defer.returnValue(chunk) + @defer.inlineCallbacks + def handle_event(self, event_dict): + builder = self.event_builder_factory.new(event_dict) + + event, context = yield self._create_new_client_event( + builder=builder, + ) + + # TODO: self.validator.validate(event) + + if event.type == EventTypes.Member: + member_handler = self.hs.get_handlers().room_member_handler + yield member_handler.change_membership(event, context) + else: + yield self.handle_new_client_event( + event=event, + context=context, + ) + + defer.returnValue(event) + @defer.inlineCallbacks def store_room_data(self, event=None): """ Stores data for a room. diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index 0116ba5358..f2abbc5df9 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -210,14 +210,11 @@ class ProfileHandler(BaseHandler): "collect_presencelike_data", user, content ) - new_event = self.event_factory.create_event( - etype=j.type, - room_id=j.room_id, - state_key=j.state_key, - content=content, - user_id=j.state_key, - ) - - yield self._on_new_room_event( - new_event, snapshot, suppress_auth=True - ) + msg_handler = self.hs.get_handlers().message_handler + yield msg_handler.handle_event({ + "type": j.type, + "room_id": j.room_id, + "state_key": j.state_key, + "content": content, + "sender": j.state_key, + }) diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 52a9788823..f0ffd62b7f 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -123,59 +123,37 @@ class RoomCreationHandler(BaseHandler): user, room_id, is_public=is_public ) - room_member_handler = self.hs.get_handlers().room_member_handler - - @defer.inlineCallbacks - def handle_event(event): - snapshot = yield self.store.snapshot_room(event) - - logger.debug("Event: %s", event) - - if event.type == RoomMemberEvent.TYPE: - yield room_member_handler.change_membership( - event, - do_auth=True - ) - else: - yield self._on_new_room_event( - event, snapshot, extra_users=[user], suppress_auth=True - ) + msg_handler = self.hs.get_handlers().message_handler for event in creation_events: - yield handle_event(event) + yield msg_handler.handle_event(event) if "name" in config: name = config["name"] - name_event = self.event_factory.create_event( - etype=RoomNameEvent.TYPE, - room_id=room_id, - user_id=user_id, - content={"name": name}, - ) - - yield handle_event(name_event) + yield msg_handler.handle_event({ + "type": RoomNameEvent.TYPE, + "room_id": room_id, + "sender": user_id, + "content": {"name": name}, + }) if "topic" in config: topic = config["topic"] - topic_event = self.event_factory.create_event( - etype=RoomTopicEvent.TYPE, - room_id=room_id, - user_id=user_id, - content={"topic": topic}, - ) + yield msg_handler.handle_event({ + "type": RoomTopicEvent.TYPE, + "room_id": room_id, + "sender": user_id, + "content": {"topic": topic}, + }) - yield handle_event(topic_event) - - content = {"membership": Membership.INVITE} for invitee in invite_list: - invite_event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, - state_key=invitee, - room_id=room_id, - user_id=user_id, - content=content - ) - yield handle_event(invite_event) + yield msg_handler.handle_event({ + "type": RoomMemberEvent.TYPE, + "state_key": invitee, + "room_id": room_id, + "user_id": user_id, + "content": {"membership": Membership.INVITE}, + }) result = {"room_id": room_id} @@ -192,22 +170,25 @@ class RoomCreationHandler(BaseHandler): event_keys = { "room_id": room_id, - "user_id": creator_id, + "sender": creator_id, } - def create(etype, **content): - return self.event_factory.create_event( - etype=etype, - content=content, - **event_keys - ) + def create(etype, content): + e = { + "type": etype, + "content": content, + } + + e.update(event_keys) + + return e creation_event = create( etype=RoomCreateEvent.TYPE, - creator=creator.to_string(), + content={"creator": creator.to_string()}, ) - join_event = self.event_factory.create_event( + join_event = create( etype=RoomMemberEvent.TYPE, state_key=creator_id, content={ @@ -216,7 +197,7 @@ class RoomCreationHandler(BaseHandler): **event_keys ) - power_levels_event = self.event_factory.create_event( + power_levels_event = create( etype=RoomPowerLevelsEvent.TYPE, content={ "users": { @@ -233,13 +214,12 @@ class RoomCreationHandler(BaseHandler): "kick": 50, "redact": 50 }, - **event_keys ) join_rule = JoinRules.PUBLIC if is_public else JoinRules.INVITE join_rules_event = create( etype=RoomJoinRulesEvent.TYPE, - join_rule=join_rule, + content={"join_rule": join_rule}, ) return [ @@ -351,7 +331,7 @@ class RoomMemberHandler(BaseHandler): defer.returnValue(member) @defer.inlineCallbacks - def change_membership(self, event=None, do_auth=True): + def change_membership(self, event, context, do_auth=True): """ Change the membership status of a user in a room. Args: @@ -361,8 +341,6 @@ class RoomMemberHandler(BaseHandler): """ target_user_id = event.state_key - snapshot = yield self.store.snapshot_room(event) - ## TODO(markjh): get prev state from snapshot. prev_state = yield self.store.get_room_member( target_user_id, event.room_id @@ -374,7 +352,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, snapshot, do_auth=do_auth) + yield self._do_join(event, context, do_auth=do_auth) else: # This is not a JOIN, so we can handle it normally. @@ -387,7 +365,7 @@ class RoomMemberHandler(BaseHandler): yield self._do_local_membership_update( event, membership=event.content["membership"], - snapshot=snapshot, + context=context, do_auth=do_auth, ) @@ -409,23 +387,21 @@ class RoomMemberHandler(BaseHandler): host = hosts[0] content.update({"membership": Membership.JOIN}) - new_event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, - state_key=joinee.to_string(), - room_id=room_id, - user_id=joinee.to_string(), - membership=Membership.JOIN, - content=content, - ) - - snapshot = yield self.store.snapshot_room(new_event) + event, context = yield self.create_new_client_event({ + "type": RoomMemberEvent.TYPE, + "state_key": joinee.to_string(), + "room_id": room_id, + "sender": joinee.to_string(), + "membership": Membership.JOIN, + "content": content, + }) - yield self._do_join(new_event, snapshot, room_host=host, do_auth=True) + yield self._do_join(event, context, room_host=host, do_auth=True) defer.returnValue({"room_id": room_id}) @defer.inlineCallbacks - def _do_join(self, event, snapshot, room_host=None, do_auth=True): + def _do_join(self, event, context, room_host=None, do_auth=True): joinee = self.hs.parse_userid(event.state_key) # room_id = RoomID.from_string(event.room_id, self.hs) room_id = event.room_id @@ -470,7 +446,7 @@ class RoomMemberHandler(BaseHandler): 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, snapshot + room_host, room_id, event.user_id, event.content, context ) # We want to do the _do_update inside the room lock. @@ -480,7 +456,7 @@ class RoomMemberHandler(BaseHandler): yield self._do_local_membership_update( event, membership=event.content["membership"], - snapshot=snapshot, + context=context, do_auth=do_auth, ) @@ -530,7 +506,7 @@ class RoomMemberHandler(BaseHandler): defer.returnValue(room_ids) @defer.inlineCallbacks - def _do_local_membership_update(self, event, membership, snapshot, + def _do_local_membership_update(self, event, membership, context, do_auth): yield run_on_reactor() @@ -543,9 +519,9 @@ class RoomMemberHandler(BaseHandler): else: do_invite_host = None - yield self._on_new_room_event( + yield self.handle_new_client_event( event, - snapshot, + context, extra_users=[target_user], suppress_auth=(not do_auth), do_invite_host=do_invite_host, diff --git a/synapse/rest/base.py b/synapse/rest/base.py index 79fc4dfb84..72bb66ddda 100644 --- a/synapse/rest/base.py +++ b/synapse/rest/base.py @@ -63,7 +63,7 @@ class RestServlet(object): self.hs = hs self.handlers = hs.get_handlers() - self.event_factory = hs.get_event_factory() + self.builder_factory = hs.get_event_builder_factory() self.auth = hs.get_auth() self.txns = HttpTransactionStore() diff --git a/synapse/rest/room.py b/synapse/rest/room.py index 3147d7a60b..3d78b4ff5c 100644 --- a/synapse/rest/room.py +++ b/synapse/rest/room.py @@ -117,10 +117,10 @@ class RoomStateEventRestServlet(RestServlet): self.on_PUT_no_state_key) def on_GET_no_state_key(self, request, room_id, event_type): - return self.on_GET(request, room_id, event_type, "") + return self.on_GET(request, room_id, event_type, None) def on_PUT_no_state_key(self, request, room_id, event_type): - return self.on_PUT(request, room_id, event_type, "") + return self.on_PUT(request, room_id, event_type, None) @defer.inlineCallbacks def on_GET(self, request, room_id, event_type, state_key): @@ -147,28 +147,18 @@ class RoomStateEventRestServlet(RestServlet): content = _parse_json(request) - event = self.event_factory.create_event( - etype=event_type, # already urldecoded - content=content, - room_id=urllib.unquote(room_id), - user_id=user.to_string(), - state_key=urllib.unquote(state_key) - ) - - self.validator.validate(event) + msg_handler = self.handlers.message_handler + yield msg_handler.handle_event( + { + "type": event_type, + "content": content, + "room_id": room_id, + "sender": user.to_string(), + "state_key": urllib.unquote(state_key), + } + ) - if event_type == RoomMemberEvent.TYPE: - # membership events are special - handler = self.handlers.room_member_handler - yield handler.change_membership(event) - defer.returnValue((200, {})) - else: - # store random bits of state - msg_handler = self.handlers.message_handler - yield msg_handler.store_room_data( - event=event - ) - defer.returnValue((200, {})) + defer.returnValue((200, {})) # TODO: Needs unit testing for generic events + feedback @@ -184,17 +174,15 @@ class RoomSendEventRestServlet(RestServlet): user = yield self.auth.get_user_by_req(request) content = _parse_json(request) - event = self.event_factory.create_event( - etype=urllib.unquote(event_type), - room_id=urllib.unquote(room_id), - user_id=user.to_string(), - content=content - ) - - self.validator.validate(event) - msg_handler = self.handlers.message_handler - yield msg_handler.send_message(event) + event = yield msg_handler.handle_event( + { + "type": urllib.unquote(event_type), + "content": content, + "room_id": urllib.unquote(room_id), + "sender": user.to_string(), + } + ) defer.returnValue((200, {"event_id": event.event_id})) @@ -251,18 +239,17 @@ class JoinRoomAliasServlet(RestServlet): ret_dict = yield handler.join_room_alias(user, identifier) defer.returnValue((200, ret_dict)) else: # room id - event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, - content={"membership": Membership.JOIN}, - room_id=urllib.unquote(identifier.to_string()), - user_id=user.to_string(), - state_key=user.to_string() + msg_handler = self.handlers.message_handler + yield msg_handler.handle_event( + { + "type": RoomMemberEvent.TYPE, + "content": {"membership": Membership.JOIN}, + "room_id": urllib.unquote(identifier.to_string()), + "sender": user.to_string(), + "state_key": user.to_string(), + } ) - self.validator.validate(event) - - handler = self.handlers.room_member_handler - yield handler.change_membership(event) defer.returnValue((200, {})) @defer.inlineCallbacks @@ -414,18 +401,17 @@ class RoomMembershipRestServlet(RestServlet): if membership_action == "kick": membership_action = "leave" - event = self.event_factory.create_event( - etype=RoomMemberEvent.TYPE, - content={"membership": unicode(membership_action)}, - room_id=urllib.unquote(room_id), - user_id=user.to_string(), - state_key=state_key + msg_handler = self.handlers.message_handler + yield msg_handler.handle_event( + { + "type": RoomMemberEvent.TYPE, + "content": {"membership": unicode(membership_action)}, + "room_id": urllib.unquote(room_id), + "sender": user.to_string(), + "state_key": state_key, + } ) - self.validator.validate(event) - - handler = self.handlers.room_member_handler - yield handler.change_membership(event) defer.returnValue((200, {})) @defer.inlineCallbacks @@ -453,18 +439,16 @@ class RoomRedactEventRestServlet(RestServlet): user = yield self.auth.get_user_by_req(request) content = _parse_json(request) - event = self.event_factory.create_event( - etype=RoomRedactionEvent.TYPE, - room_id=urllib.unquote(room_id), - user_id=user.to_string(), - content=content, - redacts=urllib.unquote(event_id), - ) - - self.validator.validate(event) - msg_handler = self.handlers.message_handler - yield msg_handler.send_message(event) + event = yield msg_handler.handle_event( + { + "type": RoomRedactionEvent.TYPE, + "content": content, + "room_id": urllib.unquote(room_id), + "sender": user.to_string(), + "redacts": urllib.unquote(event_id), + } + ) defer.returnValue((200, {"event_id": event.event_id})) diff --git a/synapse/server.py b/synapse/server.py index c3b54221d6..8bc27bbc3c 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -36,6 +36,7 @@ from synapse.util.lockutils import LockManager from synapse.streams.events import EventSources from synapse.api.ratelimiting import Ratelimiter from synapse.crypto.keyring import Keyring +from synapse.events.builder import EventBuilderFactory class BaseHomeServer(object): @@ -82,6 +83,7 @@ class BaseHomeServer(object): 'ratelimiter', 'keyring', 'event_validator', + 'event_builder_factory', ] def __init__(self, hostname, **kwargs): @@ -231,6 +233,12 @@ class HomeServer(BaseHomeServer): def build_event_validator(self): return EventValidator(self) + def build_event_builder_factory(self): + return EventBuilderFactory( + clock=self.get_clock(), + hostname=self.hostname, + ) + def register_servlets(self): """ Register all servlets associated with this HomeServer. """ diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index e2f11c7ffc..3a705119fd 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.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 from syutil.base64util import encode_base64 @@ -69,8 +71,9 @@ class SignatureStore(SQLBaseStore): f ) + @defer.inlineCallbacks def add_event_hashes(self, event_ids): - hashes = yield self.store.get_event_reference_hashes( + hashes = yield self.get_event_reference_hashes( event_ids ) hashes = [ -- cgit 1.4.1 From 6630e1b5795667fd947cc5b0d5d2b00da97325e3 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 5 Dec 2014 16:20:48 +0000 Subject: Start making more things use EventContext rather than event.* --- synapse/api/auth.py | 33 +++++---- synapse/events/__init__.py | 35 ++++++++- synapse/events/utils.py | 16 ++++ synapse/handlers/_base.py | 164 ++++++++++++++++++----------------------- synapse/handlers/federation.py | 19 +++-- synapse/server.py | 2 +- synapse/state.py | 33 +++++++++ synapse/storage/__init__.py | 23 +++--- synapse/storage/_base.py | 8 +- synapse/storage/state.py | 13 +++- 10 files changed, 212 insertions(+), 134 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 5261c3e3bf..3f2e58a5ef 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -351,27 +351,27 @@ class Auth(object): return self.store.is_server_admin(user) @defer.inlineCallbacks - def get_auth_events(self, event, current_state): - if event.type == RoomCreateEvent.TYPE: - event.auth_events = [] + def add_auth_events(self, builder, context): + if builder.type == RoomCreateEvent.TYPE: + builder.auth_events = [] return auth_events = [] key = (RoomPowerLevelsEvent.TYPE, "", ) - power_level_event = current_state.get(key) + power_level_event = context.current_state.get(key) if power_level_event: auth_events.append(power_level_event.event_id) key = (RoomJoinRulesEvent.TYPE, "", ) - join_rule_event = current_state.get(key) + join_rule_event = context.current_state.get(key) - key = (RoomMemberEvent.TYPE, event.user_id, ) - member_event = current_state.get(key) + key = (RoomMemberEvent.TYPE, builder.user_id, ) + member_event = context.current_state.get(key) key = (RoomCreateEvent.TYPE, "", ) - create_event = current_state.get(key) + create_event = context.current_state.get(key) if create_event: auth_events.append(create_event.event_id) @@ -381,8 +381,8 @@ class Auth(object): else: is_public = False - if event.type == RoomMemberEvent.TYPE: - e_type = event.content["membership"] + if builder.type == RoomMemberEvent.TYPE: + e_type = builder.content["membership"] if e_type in [Membership.JOIN, Membership.INVITE]: if join_rule_event: auth_events.append(join_rule_event.event_id) @@ -393,11 +393,18 @@ class Auth(object): if member_event.content["membership"] == Membership.JOIN: auth_events.append(member_event.event_id) - auth_events = yield self.store.add_event_hashes( - auth_events + auth_ids = [(a.event_id, h) for a, h in auth_events] + auth_events_entries = yield self.store.add_event_hashes( + auth_ids ) - defer.returnValue(auth_events) + builder.auth_events = auth_events_entries + + context.auth_events = { + k: v + for k, v in context.current_state.items() + if v.event_id in auth_ids + } @log_function def _can_send_event(self, event, auth_events): diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index 58edf2bc8f..e81b995d39 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -17,8 +17,8 @@ from frozendict import frozendict def _freeze(o): - if isinstance(o, dict): - return frozendict({k: _freeze(v) for k,v in o.items()}) + if isinstance(o, dict) or isinstance(o, frozendict): + return frozendict({k: _freeze(v) for k, v in o.items()}) if isinstance(o, basestring): return o @@ -31,6 +31,21 @@ def _freeze(o): return o +def _unfreeze(o): + if isinstance(o, frozendict) or isinstance(o, dict): + return dict({k: _unfreeze(v) for k, v in o.items()}) + + if isinstance(o, basestring): + return o + + try: + return [_unfreeze(i) for i in o] + except TypeError: + pass + + return o + + class _EventInternalMetadata(object): def __init__(self, internal_metadata_dict): self.__dict__ = internal_metadata_dict @@ -69,6 +84,7 @@ class EventBase(object): ) auth_events = _event_dict_property("auth_events") + depth = _event_dict_property("depth") content = _event_dict_property("content") event_id = _event_dict_property("event_id") hashes = _event_dict_property("hashes") @@ -81,6 +97,10 @@ class EventBase(object): type = _event_dict_property("type") user_id = _event_dict_property("sender") + @property + def membership(self): + return self.content["membership"] + def is_state(self): return hasattr(self, "state_key") @@ -134,3 +154,14 @@ class FrozenEvent(EventBase): e.internal_metadata = event.internal_metadata return e + + def get_dict(self): + # We need to unfreeze what we return + + d = _unfreeze(self._event_dict) + d.update({ + "signatures": self.signatures, + "unsigned": self.unsigned, + }) + + return d diff --git a/synapse/events/utils.py b/synapse/events/utils.py index 412f690f08..1b05ee0a95 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -14,6 +14,7 @@ # limitations under the License. from synapse.api.constants import EventTypes +from . import EventBase def prune_event(event): @@ -80,3 +81,18 @@ def prune_event(event): allowed_fields["content"] = new_content return type(event)(allowed_fields) + + +def serialize_event(hs, e): + # FIXME(erikj): To handle the case of presence events and the like + if not isinstance(e, EventBase): + return e + + # Should this strip out None's? + d = {k: v for k, v in e.get_dict().items()} + if "age_ts" in d["unsigned"]: + now = int(hs.get_clock().time_msec()) + d["unsigned"]["age"] = now - d["unsigned"]["age_ts"] + del d["unsigned"]["age_ts"] + + return d \ No newline at end of file diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 4052d0e1e7..810ce138ff 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -62,6 +62,8 @@ class BaseHandler(object): @defer.inlineCallbacks def _create_new_client_event(self, builder): + context = EventContext() + latest_ret = yield self.store.get_latest_events_in_room( builder.room_id, ) @@ -69,34 +71,26 @@ class BaseHandler(object): depth = max([d for _, _, d in latest_ret]) prev_events = [(e, h) for e, h, _ in latest_ret] - state_handler = self.state_handler - if builder.is_state(): - ret = yield state_handler.resolve_state_groups( - [e for e, _ in prev_events], - event_type=builder.event_type, - state_key=builder.state_key, - ) + builder.prev_events = prev_events + builder.depth = depth - group, curr_state, prev_state = ret + state_handler = self.state_handler + ret = yield state_handler.annotate_context_with_state( + builder, + context, + ) + group, prev_state = ret + if builder.is_state(): prev_state = yield self.store.add_event_hashes( prev_state ) builder.prev_state = prev_state - else: - group, curr_state, _ = yield state_handler.resolve_state_groups( - [e for e, _ in prev_events], - ) builder.internal_metadata.state_group = group - builder.prev_events = prev_events - builder.depth = depth - - auth_events = yield self.auth.get_auth_events(builder, curr_state) - - builder.update_event_key("auth_events", auth_events) + yield self.auth.add_auth_events(builder, context) add_hashes_and_signatures( builder, self.server_name, self.signing_key @@ -104,18 +98,6 @@ class BaseHandler(object): event = builder.build() - auth_ids = zip(*auth_events)[0] - curr_auth_events = { - k: v - for k, v in curr_state.items() - if v.event_id in auth_ids - } - - context = EventContext( - current_state=curr_state, - auth_events=curr_auth_events, - ) - defer.returnValue( (event, context,) ) @@ -128,7 +110,7 @@ class BaseHandler(object): if not suppress_auth: self.auth.check(event, auth_events=context.auth_events) - yield self.store.persist_event(event) + yield self.store.persist_event(event, context=context) destinations = set(extra_destinations) for k, s in context.current_state.items(): @@ -152,63 +134,63 @@ class BaseHandler(object): destinations=destinations, ) - @defer.inlineCallbacks - def _on_new_room_event(self, event, snapshot, extra_destinations=[], - extra_users=[], suppress_auth=False, - do_invite_host=None): - yield run_on_reactor() - - snapshot.fill_out_prev_events(event) - - yield self.state_handler.annotate_event_with_state(event) - - yield self.auth.add_auth_events(event) - - logger.debug("Signing event...") - - add_hashes_and_signatures( - event, self.server_name, self.signing_key - ) - - logger.debug("Signed event.") - - if not suppress_auth: - logger.debug("Authing...") - self.auth.check(event, auth_events=event.old_state_events) - logger.debug("Authed") - else: - logger.debug("Suppressed auth.") - - if do_invite_host: - federation_handler = self.hs.get_handlers().federation_handler - invite_event = yield federation_handler.send_invite( - do_invite_host, - event - ) - - # FIXME: We need to check if the remote changed anything else - event.signatures = invite_event.signatures - - yield self.store.persist_event(event) - - destinations = set(extra_destinations) - # Send a PDU to all hosts who have joined the room. - - for k, s in event.state_events.items(): - try: - if k[0] == RoomMemberEvent.TYPE: - if s.content["membership"] == Membership.JOIN: - destinations.add( - self.hs.parse_userid(s.state_key).domain - ) - except: - logger.warn( - "Failed to get destination from event %s", s.event_id - ) - - event.destinations = list(destinations) - - yield self.notifier.on_new_room_event(event, extra_users=extra_users) - - federation_handler = self.hs.get_handlers().federation_handler - yield federation_handler.handle_new_event(event, snapshot) + # @defer.inlineCallbacks + # def _on_new_room_event(self, event, snapshot, extra_destinations=[], + # extra_users=[], suppress_auth=False, + # do_invite_host=None): + # yield run_on_reactor() + # + # snapshot.fill_out_prev_events(event) + # + # yield self.state_handler.annotate_event_with_state(event) + # + # yield self.auth.add_auth_events(event) + # + # logger.debug("Signing event...") + # + # add_hashes_and_signatures( + # event, self.server_name, self.signing_key + # ) + # + # logger.debug("Signed event.") + # + # if not suppress_auth: + # logger.debug("Authing...") + # self.auth.check(event, auth_events=event.old_state_events) + # logger.debug("Authed") + # else: + # logger.debug("Suppressed auth.") + # + # if do_invite_host: + # federation_handler = self.hs.get_handlers().federation_handler + # invite_event = yield federation_handler.send_invite( + # do_invite_host, + # event + # ) + # + # # FIXME: We need to check if the remote changed anything else + # event.signatures = invite_event.signatures + # + # yield self.store.persist_event(event) + # + # destinations = set(extra_destinations) + # # Send a PDU to all hosts who have joined the room. + # + # for k, s in event.state_events.items(): + # try: + # if k[0] == RoomMemberEvent.TYPE: + # if s.content["membership"] == Membership.JOIN: + # destinations.add( + # self.hs.parse_userid(s.state_key).domain + # ) + # except: + # logger.warn( + # "Failed to get destination from event %s", s.event_id + # ) + # + # event.destinations = list(destinations) + # + # yield self.notifier.on_new_room_event(event, extra_users=extra_users) + # + # federation_handler = self.hs.get_handlers().federation_handler + # yield federation_handler.handle_new_event(event, snapshot) diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index b4a28ea3cb..5264e3eafc 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -17,7 +17,8 @@ from ._base import BaseHandler -from synapse.api.events.utils import prune_event +from synapse.events.snapshot import EventContext +from synapse.events.utils import prune_event from synapse.api.errors import ( AuthError, FederationError, SynapseError, StoreError, ) @@ -416,7 +417,7 @@ class FederationHandler(BaseHandler): @defer.inlineCallbacks @log_function - def on_make_join_request(self, context, user_id): + def on_make_join_request(self, room_id, user_id): """ We've received a /make_join/ request, so we create a partial join event for the room and return that. We don *not* persist or process it until the other server has signed it and sent it back. @@ -424,7 +425,7 @@ class FederationHandler(BaseHandler): builder = self.event_builder_factory.new({ "type": RoomMemberEvent.TYPE, "content": {"membership": Membership.JOIN}, - "room_id": context, + "room_id": room_id, "sender": user_id, "state_key": user_id, }) @@ -433,9 +434,7 @@ class FederationHandler(BaseHandler): builder=builder, ) - yield self.state_handler.annotate_event_with_state(event) - yield self.auth.add_auth_events(event) - self.auth.check(event, auth_events=event.old_state_events) + self.auth.check(event, auth_events=context.auth_events) pdu = event @@ -505,7 +504,9 @@ class FederationHandler(BaseHandler): """ event = pdu - event.outlier = True + context = EventContext() + + event.internal_metadata.outlier = True event.signatures.update( compute_event_signature( @@ -515,10 +516,11 @@ class FederationHandler(BaseHandler): ) ) - yield self.state_handler.annotate_event_with_state(event) + yield self.state_handler.annotate_context_with_state(event, context) yield self.store.persist_event( event, + context=context, backfilled=False, ) @@ -640,6 +642,7 @@ class FederationHandler(BaseHandler): @defer.inlineCallbacks def _handle_new_event(self, event, state=None, backfilled=False, current_state=None, fetch_missing=True): + context = EventContext() is_new_state = yield self.state_handler.annotate_event_with_state( event, old_state=state diff --git a/synapse/server.py b/synapse/server.py index 8bc27bbc3c..0d0f3af3f4 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -20,7 +20,7 @@ # Imports required for the default HomeServer() implementation from synapse.federation import initialize_http_replication -from synapse.api.events import serialize_event +from synapse.events.utils import serialize_event from synapse.api.events.factory import EventFactory from synapse.api.events.validator import EventValidator from synapse.notifier import Notifier diff --git a/synapse/state.py b/synapse/state.py index 8a556a27f6..cbb4243fad 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -135,6 +135,39 @@ class StateHandler(object): defer.returnValue(res[1].values()) + @defer.inlineCallbacks + def annotate_context_with_state(self, event, context): + if event.is_state(): + ret = yield self.resolve_state_groups( + [e for e, _ in event.prev_events], + event_type=event.event_type, + state_key=event.state_key, + ) + else: + ret = yield self.resolve_state_groups( + [e for e, _ in event.prev_events], + ) + + group, curr_state, prev_state = ret + + context.current_state = curr_state + + prev_state = yield self.store.add_event_hashes( + prev_state + ) + + if hasattr(event, "auth_events") and event.auth_events: + auth_ids = zip(*event.auth_events)[0] + context.auth_events = { + k: v + for k, v in context.current_state.items() + if v.event_id in auth_ids + } + + defer.returnValue( + (group, prev_state) + ) + @defer.inlineCallbacks @log_function def resolve_state_groups(self, event_ids, event_type=None, state_key=""): diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 205d125642..f172c2690a 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -21,6 +21,7 @@ from synapse.api.events.room import ( ) from synapse.util.logutils import log_function +from synapse.util.frozenutils import FrozenEncoder from .directory import DirectoryStore from .feedback import FeedbackStore @@ -93,8 +94,8 @@ class DataStore(RoomMemberStore, RoomStore, @defer.inlineCallbacks @log_function - def persist_event(self, event, backfilled=False, is_new_state=True, - current_state=None): + def persist_event(self, event, context, backfilled=False, + is_new_state=True, current_state=None): stream_ordering = None if backfilled: if not self.min_token_deferred.called: @@ -107,6 +108,7 @@ class DataStore(RoomMemberStore, RoomStore, "persist_event", self._persist_event_txn, event=event, + context=context, backfilled=backfilled, stream_ordering=stream_ordering, is_new_state=is_new_state, @@ -138,8 +140,9 @@ class DataStore(RoomMemberStore, RoomStore, defer.returnValue(event[0]) @log_function - def _persist_event_txn(self, txn, event, backfilled, stream_ordering=None, - is_new_state=True, current_state=None): + def _persist_event_txn(self, txn, event, context, backfilled, + stream_ordering=None, is_new_state=True, + current_state=None): if event.type == RoomMemberEvent.TYPE: self._store_room_member_txn(txn, event) elif event.type == FeedbackEvent.TYPE: @@ -152,12 +155,12 @@ class DataStore(RoomMemberStore, RoomStore, self._store_redaction(txn, event) outlier = False - if hasattr(event, "outlier"): - outlier = event.outlier + if hasattr(event.internal_metadata, "outlier"): + outlier = event.internal_metadata.outlier event_dict = { k: v - for k, v in event.get_full_dict().items() + for k, v in event.get_dict().items() if k not in [ "redacted", "redacted_because", @@ -179,7 +182,7 @@ class DataStore(RoomMemberStore, RoomStore, "event_id": event.event_id, "type": event.type, "room_id": event.room_id, - "content": json.dumps(event.content), + "content": json.dumps(event.content, cls=FrozenEncoder), "processed": True, "outlier": outlier, "depth": event.depth, @@ -190,7 +193,7 @@ class DataStore(RoomMemberStore, RoomStore, unrec = { k: v - for k, v in event.get_full_dict().items() + for k, v in event.get_dict().items() if k not in vals.keys() and k not in [ "redacted", "redacted_because", @@ -225,7 +228,7 @@ class DataStore(RoomMemberStore, RoomStore, room_id=event.room_id, ) - self._store_state_groups_txn(txn, event) + self._store_state_groups_txn(txn, event, context) if current_state: txn.execute( diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index bb61c20150..c56c3a0b0f 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -15,7 +15,8 @@ import logging from synapse.api.errors import StoreError -from synapse.api.events.utils import prune_event +from synapse.events import FrozenEvent +from synapse.events.utils import prune_event from synapse.util.logutils import log_function from synapse.util.logcontext import PreserveLoggingContext, LoggingContext from syutil.base64util import encode_base64 @@ -497,10 +498,7 @@ class SQLBaseStore(object): d = json.loads(js) - ev = self.event_factory.create_event( - etype=d["type"], - **d - ) + ev = FrozenEvent(d) if hasattr(ev, "redacted") and ev.redacted: # Get the redaction event. diff --git a/synapse/storage/state.py b/synapse/storage/state.py index e0f44b3e59..b8e721ad72 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -86,11 +86,16 @@ class StateStore(SQLBaseStore): self._store_state_groups_txn, event ) - def _store_state_groups_txn(self, txn, event): - if event.state_events is None: + def _store_state_groups_txn(self, txn, event, context): + if context.current_state_events is None: return - state_group = event.state_group + state_events = context.current_state_events + + if event.is_state(): + state_events[(event.type, event.state_key)] = event + + state_group = context.state_group if not state_group: state_group = self._simple_insert_txn( txn, @@ -102,7 +107,7 @@ class StateStore(SQLBaseStore): or_ignore=True, ) - for state in event.state_events.values(): + for state in context.state_events.values(): self._simple_insert_txn( txn, table="state_groups_state", -- cgit 1.4.1 From d044121168672c657e595525af9b588c8769e9bb Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 8 Dec 2014 09:08:26 +0000 Subject: Various typos and bug fixes. --- synapse/api/auth.py | 16 ++++--- synapse/events/snapshot.py | 3 +- synapse/events/utils.py | 2 + synapse/federation/replication.py | 12 +++-- synapse/handlers/_base.py | 8 ++-- synapse/handlers/federation.py | 97 +++++++++++++++------------------------ synapse/state.py | 16 +++++-- synapse/storage/state.py | 6 +-- 8 files changed, 80 insertions(+), 80 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 3f2e58a5ef..821e3ba5e2 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -24,6 +24,7 @@ from synapse.api.events.room import ( RoomJoinRulesEvent, RoomCreateEvent, RoomAliasesEvent, ) from synapse.util.logutils import log_function +from synapse.util.async import run_on_reactor from syutil.base64util import encode_base64 import logging @@ -352,17 +353,19 @@ class Auth(object): @defer.inlineCallbacks def add_auth_events(self, builder, context): + yield run_on_reactor() + if builder.type == RoomCreateEvent.TYPE: builder.auth_events = [] return - auth_events = [] + auth_ids = [] key = (RoomPowerLevelsEvent.TYPE, "", ) power_level_event = context.current_state.get(key) if power_level_event: - auth_events.append(power_level_event.event_id) + auth_ids.append(power_level_event.event_id) key = (RoomJoinRulesEvent.TYPE, "", ) join_rule_event = context.current_state.get(key) @@ -373,7 +376,7 @@ class Auth(object): key = (RoomCreateEvent.TYPE, "", ) create_event = context.current_state.get(key) if create_event: - auth_events.append(create_event.event_id) + auth_ids.append(create_event.event_id) if join_rule_event: join_rule = join_rule_event.content.get("join_rule") @@ -385,15 +388,14 @@ class Auth(object): e_type = builder.content["membership"] if e_type in [Membership.JOIN, Membership.INVITE]: if join_rule_event: - auth_events.append(join_rule_event.event_id) + auth_ids.append(join_rule_event.event_id) if member_event and not is_public: - auth_events.append(member_event.event_id) + auth_ids.append(member_event.event_id) elif member_event: if member_event.content["membership"] == Membership.JOIN: - auth_events.append(member_event.event_id) + auth_ids.append(member_event.event_id) - auth_ids = [(a.event_id, h) for a, h in auth_events] auth_events_entries = yield self.store.add_event_hashes( auth_ids ) diff --git a/synapse/events/snapshot.py b/synapse/events/snapshot.py index ca15ec09ae..e0cbacc19c 100644 --- a/synapse/events/snapshot.py +++ b/synapse/events/snapshot.py @@ -58,6 +58,7 @@ class EventCache(object): class EventContext(object): - def __init__(self, current_state, auth_events): + def __init__(self, current_state=None, auth_events=None): self.current_state = current_state self.auth_events = auth_events + self.state_group = None diff --git a/synapse/events/utils.py b/synapse/events/utils.py index 1b05ee0a95..485f075406 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -95,4 +95,6 @@ def serialize_event(hs, e): d["unsigned"]["age"] = now - d["unsigned"]["age_ts"] del d["unsigned"]["age_ts"] + d["user_id"] = d.pop("sender", None) + return d \ No newline at end of file diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index b11df9e5c6..3af24ee46d 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -558,7 +558,13 @@ class ReplicationLayer(object): origin, pdu.event_id, do_auth=False ) - if existing and (not existing.outlier or pdu.outlier): + already_seen = ( + existing and ( + not existing.internal_metadata.outlier + or pdu.internal_metadata.outlier + ) + ) + if already_seen: logger.debug("Already seen pdu %s", pdu.event_id) defer.returnValue({}) return @@ -596,7 +602,7 @@ class ReplicationLayer(object): # ) # Get missing pdus if necessary. - if not pdu.outlier: + if not pdu.internal_metadata.outlier: # We only backfill backwards to the min depth. min_depth = yield self.handler.get_min_depth_for_context( pdu.room_id @@ -663,7 +669,7 @@ class ReplicationLayer(object): pdu_json ) - builder.internal_metadata = outlier + builder.internal_metadata.outlier = outlier return builder.build() diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 810ce138ff..0bff644192 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -62,6 +62,8 @@ class BaseHandler(object): @defer.inlineCallbacks def _create_new_client_event(self, builder): + yield run_on_reactor() + context = EventContext() latest_ret = yield self.store.get_latest_events_in_room( @@ -79,7 +81,7 @@ class BaseHandler(object): builder, context, ) - group, prev_state = ret + prev_state = ret if builder.is_state(): prev_state = yield self.store.add_event_hashes( @@ -88,8 +90,6 @@ class BaseHandler(object): builder.prev_state = prev_state - builder.internal_metadata.state_group = group - yield self.auth.add_auth_events(builder, context) add_hashes_and_signatures( @@ -105,6 +105,8 @@ class BaseHandler(object): @defer.inlineCallbacks def handle_new_client_event(self, event, context, extra_destinations=[], extra_users=[], suppress_auth=False): + yield run_on_reactor() + # We now need to go and hit out to wherever we need to hit out to. if not suppress_auth: diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 5264e3eafc..38ee32d26e 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -149,7 +149,7 @@ class FederationHandler(BaseHandler): event.room_id, self.server_name ) - if not is_in_room and not event.outlier: + if not is_in_room and not event.internal_metadata.outlier: logger.debug("Got event for room we're not in.") replication_layer = self.replication_layer @@ -160,7 +160,7 @@ class FederationHandler(BaseHandler): ) for e in auth_chain: - e.outlier = True + e.internal_metadata.outlier = True try: yield self._handle_new_event(e, fetch_missing=False) except: @@ -180,7 +180,7 @@ class FederationHandler(BaseHandler): if state: for e in state: - e.outlier = True + e.internal_metadata.outlier = True try: yield self._handle_new_event(e) except: @@ -254,11 +254,18 @@ class FederationHandler(BaseHandler): event = pdu # FIXME (erikj): Not sure this actually works :/ - yield self.state_handler.annotate_event_with_state(event) + context = EventContext() + yield self.state_handler.annotate_context_with_state(event, context) - events.append(event) + events.append( + (event, context) + ) - yield self.store.persist_event(event, backfilled=True) + yield self.store.persist_event( + event, + context=context, + backfilled=True + ) defer.returnValue(events) @@ -326,7 +333,7 @@ class FederationHandler(BaseHandler): assert(event.state_key == joinee) assert(event.room_id == room_id) - event.outlier = False + event.internal_metadata.outlier = False self.room_queues[room_id] = [] @@ -369,7 +376,7 @@ class FederationHandler(BaseHandler): pass for e in auth_chain: - e.outlier = True + e.internal_metadata.outlier = True try: yield self._handle_new_event(e, fetch_missing=False) except: @@ -380,7 +387,7 @@ class FederationHandler(BaseHandler): for e in state: # FIXME: Auth these. - e.outlier = True + e.internal_metadata.outlier = True try: yield self._handle_new_event( e, @@ -448,7 +455,7 @@ class FederationHandler(BaseHandler): """ event = pdu - event.outlier = False + event.internal_metadata.outlier = False yield self._handle_new_event(event) @@ -643,70 +650,42 @@ class FederationHandler(BaseHandler): def _handle_new_event(self, event, state=None, backfilled=False, current_state=None, fetch_missing=True): context = EventContext() - is_new_state = yield self.state_handler.annotate_event_with_state( + yield self.state_handler.annotate_context_with_state( event, old_state=state ) - if event.old_state_events: - known_ids = set( - [s.event_id for s in event.old_state_events.values()] - ) - for e_id, _ in event.auth_events: - if e_id not in known_ids: - e = yield self.store.get_event( - e_id, - allow_none=True, - ) - - if not e: - # TODO: Do some conflict res to make sure that we're - # not the ones who are wrong. - logger.info( - "Rejecting %s as %s not in %s", - event.event_id, e_id, known_ids, - ) - raise AuthError(403, "Auth events are stale") + is_new_state = not event.internal_metadata.outlier - auth_events = event.old_state_events - else: - # We need to get the auth events from somewhere. - - # TODO: Don't just hit the DBs? - - auth_events = {} - for e_id, _ in event.auth_events: + known_ids = set( + [s.event_id for s in context.auth_events.values()] + ) + for e_id, _ in event.auth_events: + if e_id not in known_ids: e = yield self.store.get_event( e_id, + context, allow_none=True, ) if not e: - e = yield self.replication_layer.get_pdu( - event.origin, e_id, outlier=True + # TODO: Do some conflict res to make sure that we're + # not the ones who are wrong. + logger.info( + "Rejecting %s as %s not in %s", + event.event_id, e_id, known_ids, ) + raise AuthError(403, "Auth events are stale") - if e and fetch_missing: - try: - yield self.on_receive_pdu(event.origin, e, False) - except: - logger.exception( - "Failed to parse auth event %s", - e_id, - ) + context.auth_events[(e.type, e.state_key)] = e - if not e: - logger.warn("Can't find auth event %s.", e_id) + if event.type == RoomMemberEvent.TYPE and not event.auth_events: + if len(event.prev_events) == 1: + c = yield self.store.get_event(event.prev_events[0][0]) + if c.type == RoomCreateEvent.TYPE: + context.auth_events[(c.type, c.state_key)] = c - auth_events[(e.type, e.state_key)] = e - - if event.type == RoomMemberEvent.TYPE and not event.auth_events: - if len(event.prev_events) == 1: - c = yield self.store.get_event(event.prev_events[0][0]) - if c.type == RoomCreateEvent.TYPE: - auth_events[(c.type, c.state_key)] = c - - self.auth.check(event, auth_events=auth_events) + self.auth.check(event, auth_events=context.auth_events) yield self.store.persist_event( event, diff --git a/synapse/state.py b/synapse/state.py index cbb4243fad..464cbae564 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -136,7 +136,16 @@ class StateHandler(object): defer.returnValue(res[1].values()) @defer.inlineCallbacks - def annotate_context_with_state(self, event, context): + def annotate_context_with_state(self, event, context, old_state=None): + yield run_on_reactor() + + if old_state: + context.current_state = { + (s.type, s.state_key): s for s in old_state + } + context.state_group = None + defer.returnValue([]) + if event.is_state(): ret = yield self.resolve_state_groups( [e for e, _ in event.prev_events], @@ -151,6 +160,7 @@ class StateHandler(object): group, curr_state, prev_state = ret context.current_state = curr_state + context.state_group = group prev_state = yield self.store.add_event_hashes( prev_state @@ -164,9 +174,7 @@ class StateHandler(object): if v.event_id in auth_ids } - defer.returnValue( - (group, prev_state) - ) + defer.returnValue(prev_state) @defer.inlineCallbacks @log_function diff --git a/synapse/storage/state.py b/synapse/storage/state.py index b8e721ad72..afe3e5edea 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -87,10 +87,10 @@ class StateStore(SQLBaseStore): ) def _store_state_groups_txn(self, txn, event, context): - if context.current_state_events is None: + if context.current_state is None: return - state_events = context.current_state_events + state_events = context.current_state if event.is_state(): state_events[(event.type, event.state_key)] = event @@ -107,7 +107,7 @@ class StateStore(SQLBaseStore): or_ignore=True, ) - for state in context.state_events.values(): + for state in state_events.values(): self._simple_insert_txn( txn, table="state_groups_state", -- cgit 1.4.1 From ba3d1e2fc04c2b2de6a309c5aa1a1e8024d192e1 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 8 Dec 2014 12:01:25 +0000 Subject: Remove unused import --- synapse/api/auth.py | 1 - synapse/storage/_base.py | 1 - 2 files changed, 2 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/api/auth.py b/synapse/api/auth.py index 821e3ba5e2..cd0deeb0e6 100644 --- a/synapse/api/auth.py +++ b/synapse/api/auth.py @@ -25,7 +25,6 @@ from synapse.api.events.room import ( ) from synapse.util.logutils import log_function from synapse.util.async import run_on_reactor -from syutil.base64util import encode_base64 import logging diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index c56c3a0b0f..b0f454b90b 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -19,7 +19,6 @@ from synapse.events import FrozenEvent from synapse.events.utils import prune_event from synapse.util.logutils import log_function from synapse.util.logcontext import PreserveLoggingContext, LoggingContext -from syutil.base64util import encode_base64 from twisted.internet import defer -- cgit 1.4.1 From aa3f66cf7ff15b292afc6f251c1a77e154b8b675 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 9 Dec 2014 13:35:26 +0000 Subject: Change the way we implement get_events to be less sucky --- synapse/storage/__init__.py | 24 +++++++----------------- synapse/storage/_base.py | 34 +++++++++++++++------------------- 2 files changed, 22 insertions(+), 36 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index f172c2690a..7a09c33613 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -119,25 +119,15 @@ class DataStore(RoomMemberStore, RoomStore, @defer.inlineCallbacks def get_event(self, event_id, allow_none=False): - events_dict = yield self._simple_select_one( - "events", - {"event_id": event_id}, - [ - "event_id", - "type", - "room_id", - "content", - "unrecognized_keys", - "depth", - ], - allow_none=allow_none, - ) + events = yield self._get_events([event_id]) - if not events_dict: - defer.returnValue(None) + if not events: + if allow_none: + defer.returnValue(None) + else: + raise RuntimeError("Could not find event %s" % (event_id,)) - event = yield self._parse_events([events_dict]) - defer.returnValue(event[0]) + defer.returnValue(events[0]) @log_function def _persist_event_txn(self, txn, event, context, backfilled, diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index b0f454b90b..72f88cb2aa 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -461,32 +461,18 @@ class SQLBaseStore(object): **d ) - def _get_events_txn(self, txn, event_ids): - # FIXME (erikj): This should be batched? - - sql = "SELECT * FROM events WHERE event_id = ? ORDER BY rowid asc" - - event_rows = [] - for e_id in event_ids: - c = txn.execute(sql, (e_id,)) - event_rows.extend(self.cursor_to_dict(c)) - - return self._parse_events_txn(txn, event_rows) - - def _parse_events(self, rows): + def _get_events(self, event_ids): return self.runInteraction( - "_parse_events", self._parse_events_txn, rows + "_get_events", self._get_events_txn, event_ids ) - def _parse_events_txn(self, txn, rows): - event_ids = [r["event_id"] for r in rows] - + def _get_events_txn(self, txn, event_ids): events = [] - for event_id in event_ids: + for e_id in event_ids: js = self._simple_select_one_onecol_txn( txn, table="event_json", - keyvalues={"event_id": event_id}, + keyvalues={"event_id": e_id}, retcol="json", allow_none=True, ) @@ -516,6 +502,16 @@ class SQLBaseStore(object): return events + def _parse_events(self, rows): + return self.runInteraction( + "_parse_events", self._parse_events_txn, rows + ) + + def _parse_events_txn(self, txn, rows): + event_ids = [r["event_id"] for r in rows] + + return self._get_events_txn(txn, event_ids) + def _has_been_redacted_txn(self, txn, event): sql = "SELECT event_id FROM redactions WHERE redacts = ?" txn.execute(sql, (event.event_id,)) -- cgit 1.4.1 From 95aa903ffa77effcbca2a510744c3c3fa9b46ed3 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 10 Dec 2014 10:06:12 +0000 Subject: Try and figure out how and why signatures are being changed. --- synapse/crypto/event_signing.py | 3 ++- synapse/events/__init__.py | 18 +++++++------ synapse/events/builder.py | 7 ++--- synapse/federation/replication.py | 9 ++++--- synapse/handlers/federation.py | 54 ++++++++++++++++++++++++++++++++++++--- synapse/handlers/message.py | 1 - synapse/handlers/room.py | 7 ++--- synapse/state.py | 11 ++++++++ synapse/storage/__init__.py | 10 -------- 9 files changed, 86 insertions(+), 34 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/crypto/event_signing.py b/synapse/crypto/event_signing.py index 15de0f5ae3..21c19c971d 100644 --- a/synapse/crypto/event_signing.py +++ b/synapse/crypto/event_signing.py @@ -82,8 +82,9 @@ def compute_event_signature(event, signature_name, signing_key): redact_json = tmp_event.get_pdu_json() redact_json.pop("age_ts", None) redact_json.pop("unsigned", None) - logger.debug("Signing event: %s", redact_json) + logger.debug("Signing event: %s", encode_canonical_json(redact_json)) redact_json = sign_json(redact_json, signature_name, signing_key) + logger.debug("Signed event: %s", encode_canonical_json(redact_json)) return redact_json["signatures"] diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index 230daf30d6..ed02138706 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -15,6 +15,8 @@ from frozendict import frozendict +import copy + def _freeze(o): if isinstance(o, dict) or isinstance(o, frozendict): @@ -48,7 +50,7 @@ def _unfreeze(o): class _EventInternalMetadata(object): def __init__(self, internal_metadata_dict): - self.__dict__ = internal_metadata_dict + self.__dict__ = copy.deepcopy(internal_metadata_dict) def get_dict(self): return dict(self.__dict__) @@ -74,10 +76,10 @@ def _event_dict_property(key): class EventBase(object): def __init__(self, event_dict, signatures={}, unsigned={}, internal_metadata_dict={}): - self.signatures = signatures - self.unsigned = unsigned + self.signatures = copy.deepcopy(signatures) + self.unsigned = copy.deepcopy(unsigned) - self._event_dict = event_dict + self._event_dict = copy.deepcopy(event_dict) self.internal_metadata = _EventInternalMetadata( internal_metadata_dict @@ -131,11 +133,11 @@ class EventBase(object): class FrozenEvent(EventBase): - def __init__(self, event_dict, signatures={}, unsigned={}): - event_dict = dict(event_dict) + def __init__(self, event_dict): + event_dict = copy.deepcopy(event_dict) - signatures.update(event_dict.pop("signatures", {})) - unsigned.update(event_dict.pop("unsigned", {})) + signatures = copy.deepcopy(event_dict.pop("signatures", {})) + unsigned = copy.deepcopy(event_dict.pop("unsigned", {})) frozen_dict = _freeze(event_dict) diff --git a/synapse/events/builder.py b/synapse/events/builder.py index 127b8fa904..642264e9f3 100644 --- a/synapse/events/builder.py +++ b/synapse/events/builder.py @@ -54,10 +54,9 @@ class EventBuilderFactory(object): return e_id.to_string() def new(self, key_values={}): - if "event_id" not in key_values: - key_values["event_id"] = self.create_event_id() + key_values["event_id"] = self.create_event_id() - time_now = self.clock.time_msec() + time_now = int(self.clock.time_msec()) key_values.setdefault("origin", self.hostname) key_values.setdefault("origin_server_ts", time_now) @@ -66,4 +65,6 @@ class EventBuilderFactory(object): age = key_values["unsigned"].pop("age", 0) key_values["unsigned"].setdefault("age_ts", time_now - age) + key_values["signatures"] = {} + return EventBuilder(key_values=key_values,) diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index d4cd79b7ac..a4600b0b40 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -25,6 +25,7 @@ from .persistence import TransactionActions from synapse.util.logutils import log_function from synapse.util.logcontext import PreserveLoggingContext +from synapse.events import FrozenEvent import logging @@ -439,7 +440,9 @@ class ReplicationLayer(object): @defer.inlineCallbacks def on_send_join_request(self, origin, content): + logger.debug("on_send_join_request: content: %s", content) pdu = self.event_from_pdu_json(content) + logger.debug("on_send_join_request: pdu sigs: %s", pdu.signatures) res_pdus = yield self.handler.on_send_join_request(origin, pdu) time_now = self._clock.time_msec() defer.returnValue((200, { @@ -665,13 +668,13 @@ class ReplicationLayer(object): return "" % self.server_name def event_from_pdu_json(self, pdu_json, outlier=False): - builder = self.event_builder_factory.new( + event = FrozenEvent( pdu_json ) - builder.internal_metadata.outlier = outlier + event.internal_metadata.outlier = outlier - return builder.build() + return event class _TransactionQueue(object): diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 827c86c9da..9ae3e5eca4 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -459,10 +459,22 @@ class FederationHandler(BaseHandler): """ event = pdu + logger.debug( + "on_send_join_request: Got event: %s, signatures: %s", + event.event_id, + event.signatures, + ) + event.internal_metadata.outlier = False context = yield self._handle_new_event(event) + logger.debug( + "on_send_join_request: After _handle_new_event: %s, sigs: %s", + event.event_id, + event.signatures, + ) + extra_users = [] if event.type == RoomMemberEvent.TYPE: target_user_id = event.state_key @@ -496,6 +508,12 @@ class FederationHandler(BaseHandler): "Failed to get destination from event %s", s.event_id ) + logger.debug( + "on_send_join_request: Sending event: %s, signatures: %s", + event.event_id, + event.signatures, + ) + yield self.replication_layer.send_pdu(new_pdu, destinations) auth_chain = yield self.store.get_auth_chain(event.event_id) @@ -652,12 +670,23 @@ class FederationHandler(BaseHandler): def _handle_new_event(self, event, state=None, backfilled=False, current_state=None, fetch_missing=True): context = EventContext() + + logger.debug( + "_handle_new_event: Before annotate: %s, sigs: %s", + event.event_id, event.signatures, + ) + yield self.state_handler.annotate_context_with_state( event, context, old_state=state ) + logger.debug( + "_handle_new_event: Before auth fetch: %s, sigs: %s", + event.event_id, event.signatures, + ) + is_new_state = not event.internal_metadata.outlier known_ids = set( @@ -666,29 +695,43 @@ class FederationHandler(BaseHandler): for e_id, _ in event.auth_events: if e_id not in known_ids: e = yield self.store.get_event( - e_id, - allow_none=True, + e_id, allow_none=True, ) if not e: # TODO: Do some conflict res to make sure that we're # not the ones who are wrong. logger.info( - "Rejecting %s as %s not in %s", + "Rejecting %s as %s not in db or %s", event.event_id, e_id, known_ids, ) raise AuthError(403, "Auth events are stale") context.auth_events[(e.type, e.state_key)] = e + logger.debug( + "_handle_new_event: Before hack: %s, sigs: %s", + event.event_id, event.signatures, + ) + if event.type == RoomMemberEvent.TYPE and not event.auth_events: if len(event.prev_events) == 1: c = yield self.store.get_event(event.prev_events[0][0]) if c.type == RoomCreateEvent.TYPE: context.auth_events[(c.type, c.state_key)] = c + logger.debug( + "_handle_new_event: Before auth check: %s, sigs: %s", + event.event_id, event.signatures, + ) + self.auth.check(event, auth_events=context.auth_events) + logger.debug( + "_handle_new_event: Before persist_event: %s, sigs: %s", + event.event_id, event.signatures, + ) + yield self.store.persist_event( event, context=context, @@ -697,4 +740,9 @@ class FederationHandler(BaseHandler): current_state=current_state, ) + logger.debug( + "_handle_new_event: After persist_event: %s, sigs: %s", + event.event_id, event.signatures, + ) + defer.returnValue(context) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 8ee560d79a..13fa0be7b4 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -137,7 +137,6 @@ class MessageHandler(BaseHandler): def handle_event(self, event_dict): builder = self.event_builder_factory.new(event_dict) - if builder.type == EventTypes.Member: membership = builder.content.get("membership", None) if membership == Membership.JOIN: diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 509763ebc7..93732a9c87 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -436,19 +436,16 @@ 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( + yield handler.do_invite_join( room_host, room_id, event.user_id, event.get_dict()["content"], # FIXME To get a non-frozen dict context ) - - # We want to do the _do_update inside the room lock. - if not have_joined: + else: logger.debug("Doing normal join") yield self._do_local_membership_update( diff --git a/synapse/state.py b/synapse/state.py index ebec0ad9dc..7fdf596006 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -144,6 +144,17 @@ class StateHandler(object): (s.type, s.state_key): s for s in old_state } context.state_group = None + + if hasattr(event, "auth_events") and event.auth_events: + auth_ids = zip(*event.auth_events)[0] + context.auth_events = { + k: v + for k, v in context.current_state.items() + if v.event_id in auth_ids + } + else: + context.auth_events = {} + defer.returnValue([]) if event.is_state(): diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 7a09c33613..07b1665bf6 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -312,16 +312,6 @@ class DataStore(RoomMemberStore, RoomStore, txn, event.event_id, hash_alg, hash_bytes, ) - if hasattr(event, "signatures"): - logger.debug("sigs: %s", event.signatures) - for name, sigs in event.signatures.items(): - for key_id, signature_base64 in sigs.items(): - signature_bytes = decode_base64(signature_base64) - self._store_event_signature_txn( - txn, event.event_id, name, key_id, - signature_bytes, - ) - for prev_event_id, prev_hashes in event.prev_events: for alg, hash_base64 in prev_hashes.items(): hash_bytes = decode_base64(hash_base64) -- cgit 1.4.1 From 02db7eb209af879a0168a371b4cb1c2ad0fcab49 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 10 Dec 2014 14:02:48 +0000 Subject: Fix bug when uploading state with empty state_key --- synapse/events/__init__.py | 10 +++++-- synapse/handlers/_base.py | 66 ++++--------------------------------------- synapse/handlers/message.py | 15 ---------- synapse/rest/room.py | 20 +++++++------ synapse/storage/__init__.py | 4 ++- synapse/storage/schema/im.sql | 2 ++ 6 files changed, 29 insertions(+), 88 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index 5f41933174..bd3ffb59cc 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -118,6 +118,9 @@ class EventBase(object): return d + def get(self, key, default): + return self._event_dict.get(key, default) + def get_internal_metadata_dict(self): return self.internal_metadata.get_dict() @@ -165,6 +168,9 @@ class FrozenEvent(EventBase): return _unfreeze(super(FrozenEvent, self).get_dict()) def __str__(self): + return self.__repr__() + + def __repr__(self): return "" % ( - self.event_id, self.type, self.state_key, - ) + self.event_id, self.type, self.get("state_key", None), + ) \ No newline at end of file diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index 6b5f6e7cd8..fdd3151877 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -97,6 +97,11 @@ class BaseHandler(object): event = builder.build() + logger.debug( + "Created event %s with auth_events: %s, current state: %s", + event.event_id, context.auth_events, context.current_state, + ) + defer.returnValue( (event, context,) ) @@ -147,64 +152,3 @@ class BaseHandler(object): None, destinations=destinations, ) - - # @defer.inlineCallbacks - # def _on_new_room_event(self, event, snapshot, extra_destinations=[], - # extra_users=[], suppress_auth=False, - # do_invite_host=None): - # yield run_on_reactor() - # - # snapshot.fill_out_prev_events(event) - # - # yield self.state_handler.annotate_event_with_state(event) - # - # yield self.auth.add_auth_events(event) - # - # logger.debug("Signing event...") - # - # add_hashes_and_signatures( - # event, self.server_name, self.signing_key - # ) - # - # logger.debug("Signed event.") - # - # if not suppress_auth: - # logger.debug("Authing...") - # self.auth.check(event, auth_events=event.old_state_events) - # logger.debug("Authed") - # else: - # logger.debug("Suppressed auth.") - # - # if do_invite_host: - # federation_handler = self.hs.get_handlers().federation_handler - # invite_event = yield federation_handler.send_invite( - # do_invite_host, - # event - # ) - # - # # FIXME: We need to check if the remote changed anything else - # event.signatures = invite_event.signatures - # - # yield self.store.persist_event(event) - # - # destinations = set(extra_destinations) - # # Send a PDU to all hosts who have joined the room. - # - # for k, s in event.state_events.items(): - # try: - # if k[0] == RoomMemberEvent.TYPE: - # if s.content["membership"] == Membership.JOIN: - # destinations.add( - # self.hs.parse_userid(s.state_key).domain - # ) - # except: - # logger.warn( - # "Failed to get destination from event %s", s.event_id - # ) - # - # event.destinations = list(destinations) - # - # yield self.notifier.on_new_room_event(event, extra_users=extra_users) - # - # federation_handler = self.hs.get_handlers().federation_handler - # yield federation_handler.handle_new_event(event, snapshot) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 13fa0be7b4..9043b945e4 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -165,21 +165,6 @@ class MessageHandler(BaseHandler): defer.returnValue(event) - @defer.inlineCallbacks - def store_room_data(self, event=None): - """ Stores data for a room. - - Args: - event : The room path event - stamp_event (bool) : True to stamp event content with server keys. - Raises: - SynapseError if something went wrong. - """ - - snapshot = yield self.store.snapshot_room(event) - - yield self._on_new_room_event(event, snapshot) - @defer.inlineCallbacks def get_room_data(self, user_id=None, room_id=None, event_type=None, state_key=""): diff --git a/synapse/rest/room.py b/synapse/rest/room.py index 3d78b4ff5c..22e1244e57 100644 --- a/synapse/rest/room.py +++ b/synapse/rest/room.py @@ -147,16 +147,18 @@ class RoomStateEventRestServlet(RestServlet): content = _parse_json(request) + event_dict = { + "type": event_type, + "content": content, + "room_id": urllib.unquote(room_id), + "sender": user.to_string(), + } + + if state_key is not None: + event_dict["state_key"] = urllib.unquote(state_key) + msg_handler = self.handlers.message_handler - yield msg_handler.handle_event( - { - "type": event_type, - "content": content, - "room_id": room_id, - "sender": user.to_string(), - "state_key": urllib.unquote(state_key), - } - ) + yield msg_handler.handle_event(event_dict) defer.returnValue((200, {})) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 7068424441..f8d895082d 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -39,6 +39,7 @@ from .state import StateStore from .signatures import SignatureStore from syutil.base64util import decode_base64 +from syutil.jsonutil import encode_canonical_json from synapse.crypto.event_signing import compute_event_reference_hash @@ -162,7 +163,8 @@ class DataStore(RoomMemberStore, RoomStore, table="event_json", values={ "event_id": event.event_id, - "json": json.dumps(event_dict, separators=(',', ':')), + "room_id": event.room_id, + "json": encode_canonical_json(event_dict), }, or_replace=True, ) diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql index cb0c494ddf..0300bb29e1 100644 --- a/synapse/storage/schema/im.sql +++ b/synapse/storage/schema/im.sql @@ -35,11 +35,13 @@ CREATE INDEX IF NOT EXISTS events_room_id ON events (room_id); CREATE TABLE IF NOT EXISTS event_json( event_id TEXT NOT NULL, + room_id TEXT NOT NULL, json BLOB NOT NULL, CONSTRAINT ev_j_uniq UNIQUE (event_id) ); CREATE INDEX IF NOT EXISTS event_json_id ON event_json(event_id); +CREATE INDEX IF NOT EXISTS event_json_room_id ON event_json(room_id); CREATE TABLE IF NOT EXISTS state_events( -- cgit 1.4.1 From 1d2a0040cff8d04cdc7d7d09d8f04a5d628fa9dd Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 10 Dec 2014 15:55:03 +0000 Subject: Fix bug where we clobbered old state group values --- synapse/handlers/federation.py | 9 +++++++++ synapse/storage/__init__.py | 3 ++- synapse/storage/schema/state.sql | 3 ++- 3 files changed, 13 insertions(+), 2 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index e5deb8a9ef..2201cd977e 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -346,6 +346,8 @@ class FederationHandler(BaseHandler): event.get_pdu_json() ) + handled_events = set() + try: builder.event_id = self.event_factory.create_event_id() builder.origin = self.hs.hostname @@ -371,6 +373,10 @@ class FederationHandler(BaseHandler): auth_chain = ret["auth_chain"] auth_chain.sort(key=lambda e: e.depth) + handled_events.update([s.event_id for s in state]) + handled_events.update([a.event_id for a in auth_chain]) + handled_events.add(new_event.event_id) + logger.debug("do_invite_join auth_chain: %s", auth_chain) logger.debug("do_invite_join state: %s", state) @@ -426,6 +432,9 @@ class FederationHandler(BaseHandler): del self.room_queues[room_id] for p, origin in room_queue: + if p.event_id in handled_events: + continue + try: self.on_receive_pdu(origin, p, backfilled=False) except: diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index f8d895082d..2db2e9720f 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -220,7 +220,8 @@ class DataStore(RoomMemberStore, RoomStore, room_id=event.room_id, ) - self._store_state_groups_txn(txn, event, context) + if not outlier: + self._store_state_groups_txn(txn, event, context) if current_state: txn.execute( diff --git a/synapse/storage/schema/state.sql b/synapse/storage/schema/state.sql index 44f7aafb27..2c48d6daca 100644 --- a/synapse/storage/schema/state.sql +++ b/synapse/storage/schema/state.sql @@ -29,7 +29,8 @@ CREATE TABLE IF NOT EXISTS state_groups_state( CREATE TABLE IF NOT EXISTS event_to_state_groups( event_id TEXT NOT NULL, - state_group INTEGER NOT NULL + state_group INTEGER NOT NULL, + CONSTRAINT event_to_state_groups_uniq UNIQUE (event_id) ); CREATE INDEX IF NOT EXISTS state_groups_id ON state_groups(id); -- cgit 1.4.1 From 02e4c181714b94e86cf621307f2dfc765a62ad83 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 10 Dec 2014 18:00:36 +0000 Subject: Remove dead code --- synapse/handlers/_base.py | 2 -- synapse/storage/_base.py | 25 ------------------------- 2 files changed, 27 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index fdd3151877..14f75ecbc2 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -32,10 +32,8 @@ class BaseHandler(object): def __init__(self, hs): self.store = hs.get_datastore() - self.event_factory = hs.get_event_factory() self.auth = hs.get_auth() 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.ratelimiter = hs.get_ratelimiter() diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 935a167f6f..12239fa074 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -436,31 +436,6 @@ class SQLBaseStore(object): return self.runInteraction("_simple_max_id", func) - def _parse_event_from_row(self, row_dict): - d = copy.deepcopy({k: v for k, v in row_dict.items()}) - - d.pop("stream_ordering", None) - d.pop("topological_ordering", None) - d.pop("processed", None) - d["origin_server_ts"] = d.pop("ts", 0) - replaces_state = d.pop("prev_state", None) - - if replaces_state: - d["replaces_state"] = replaces_state - - d.update(json.loads(row_dict["unrecognized_keys"])) - d["content"] = json.loads(d["content"]) - del d["unrecognized_keys"] - - if "age_ts" not in d: - # For compatibility - d["age_ts"] = d.get("origin_server_ts", 0) - - return self.event_factory.create_event( - etype=d["type"], - **d - ) - def _get_events(self, event_ids): return self.runInteraction( "_get_events", self._get_events_txn, event_ids -- cgit 1.4.1 From 8cdebce470869613658543cb79ed5dd97a5f0548 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 11 Dec 2014 13:25:19 +0000 Subject: Fix redactions. Fix 'age' key --- synapse/events/__init__.py | 1 + synapse/events/builder.py | 13 ++++++--- synapse/events/utils.py | 21 +++++++++++++++ synapse/state.py | 12 +++++++++ synapse/storage/_base.py | 60 ++++++++++++++++++++++++----------------- tests/storage/test_redaction.py | 6 ++--- 6 files changed, 83 insertions(+), 30 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index 7103b937af..98d7f0e324 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -97,6 +97,7 @@ class EventBase(object): origin_server_ts = _event_dict_property("origin_server_ts") prev_events = _event_dict_property("prev_events") prev_state = _event_dict_property("prev_state") + redacts = _event_dict_property("redacts") room_id = _event_dict_property("room_id") sender = _event_dict_property("sender") state_key = _event_dict_property("state_key") diff --git a/synapse/events/builder.py b/synapse/events/builder.py index 642264e9f3..9579b1fe8b 100644 --- a/synapse/events/builder.py +++ b/synapse/events/builder.py @@ -19,11 +19,18 @@ from synapse.types import EventID from synapse.util.stringutils import random_string +import copy + class EventBuilder(EventBase): def __init__(self, key_values={}): + signatures = copy.deepcopy(key_values.pop("signatures", {})) + unsigned = copy.deepcopy(key_values.pop("unsigned", {})) + super(EventBuilder, self).__init__( key_values, + signatures=signatures, + unsigned=unsigned ) def update_event_key(self, key, value): @@ -61,9 +68,9 @@ class EventBuilderFactory(object): key_values.setdefault("origin", self.hostname) key_values.setdefault("origin_server_ts", time_now) - if "unsigned" in key_values: - age = key_values["unsigned"].pop("age", 0) - key_values["unsigned"].setdefault("age_ts", time_now - age) + key_values.setdefault("unsigned", {}) + age = key_values["unsigned"].pop("age", 0) + key_values["unsigned"].setdefault("age_ts", time_now - age) key_values["signatures"] = {} diff --git a/synapse/events/utils.py b/synapse/events/utils.py index f5e135e3d0..6d9c9352e2 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -80,6 +80,11 @@ def prune_event(event): allowed_fields["content"] = new_content + allowed_fields["unsigned"] = {} + + if "age_ts" in event.unsigned: + allowed_fields["unsigned"]["age_ts"] = event.unsigned["age_ts"] + return type(event)(allowed_fields) @@ -97,4 +102,20 @@ def serialize_event(hs, e): d["user_id"] = d.pop("sender", None) + if "redacted_because" in e.unsigned: + d["redacted_because"] = serialize_event( + hs, e.unsigned["redacted_because"] + ) + + del d["unsigned"]["redacted_because"] + + if "redacted_by" in e.unsigned: + d["redacted_by"] = e.unsigned["redacted_by"] + del d["unsigned"]["redacted_by"] + + del d["auth_events"] + del d["prev_events"] + del d["hashes"] + del d["signatures"] + return d diff --git a/synapse/state.py b/synapse/state.py index 7fdf596006..5bfa73fb46 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -155,6 +155,12 @@ class StateHandler(object): else: context.auth_events = {} + if event.is_state(): + key = (event.type, event.state_key) + if key in context.current_state: + replaces = context.current_state[key] + event.unsigned["replaces_state"] = replaces.event_id + defer.returnValue([]) if event.is_state(): @@ -177,6 +183,12 @@ class StateHandler(object): prev_state ) + if event.is_state(): + key = (event.type, event.state_key) + if key in context.current_state: + replaces = context.current_state[key] + event.unsigned["replaces_state"] = replaces.event_id + if hasattr(event, "auth_events") and event.auth_events: auth_ids = zip(*event.auth_events)[0] context.auth_events = { diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 12239fa074..ffc26d4a61 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -444,38 +444,50 @@ class SQLBaseStore(object): def _get_events_txn(self, txn, event_ids): events = [] for e_id in event_ids: - js = self._simple_select_one_onecol_txn( - txn, - table="event_json", - keyvalues={"event_id": e_id}, - retcol="json", - allow_none=True, - ) + ev = self._get_event_txn(txn, e_id) - if not js: - # FIXME (erikj): What should we actually do here? - continue + if ev: + events.append(ev) - d = json.loads(js) + return events - ev = FrozenEvent(d) + def _get_event_txn(self, txn, event_id, check_redacted=True): + sql = ( + "SELECT json, r.event_id FROM event_json as e " + "LEFT JOIN redactions as r ON e.event_id = r.redacts " + "WHERE e.event_id = ? " + "LIMIT 1 " + ) - if hasattr(ev, "redacted") and ev.redacted: - # Get the redaction event. - select_event_sql = "SELECT * FROM events WHERE event_id = ?" - txn.execute(select_event_sql, (ev.redacted,)) + txn.execute(sql, (event_id,)) - del_evs = self._parse_events_txn( - txn, self.cursor_to_dict(txn) - ) + res = txn.fetchone() - if del_evs: - ev = prune_event(ev) - ev.redacted_because = del_evs[0] + if not res: + return None - events.append(ev) + js, redacted = res - return events + d = json.loads(js) + + ev = FrozenEvent(d) + + if check_redacted and redacted: + ev = prune_event(ev) + + ev.unsigned["redacted_by"] = redacted + # Get the redaction event. + + because = self._get_event_txn( + txn, + redacted, + check_redacted=False + ) + + if because: + ev.unsigned["redacted_because"] = because + + return ev def _parse_events(self, rows): return self.runInteraction( diff --git a/tests/storage/test_redaction.py b/tests/storage/test_redaction.py index e8671ae3a2..d81f7add1c 100644 --- a/tests/storage/test_redaction.py +++ b/tests/storage/test_redaction.py @@ -149,7 +149,7 @@ class RedactionTestCase(unittest.TestCase): event, ) - self.assertFalse(hasattr(event, "redacted_because")) + self.assertFalse("redacted_because" in event.unsigned) # Redact event reason = "Because I said so" @@ -179,7 +179,7 @@ class RedactionTestCase(unittest.TestCase): event, ) - self.assertTrue(hasattr(event, "redacted_because")) + self.assertTrue("redacted_because" in event.unsigned) self.assertObjectHasAttributes( { @@ -187,7 +187,7 @@ class RedactionTestCase(unittest.TestCase): "user_id": self.u_alice.to_string(), "content": {"reason": reason}, }, - event.redacted_because, + event.unsigned["redacted_because"], ) @defer.inlineCallbacks -- cgit 1.4.1 From 9191292b0f657f6210e88f16ffd9a182bfab8170 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 11 Dec 2014 15:16:55 +0000 Subject: Fix prev_content --- synapse/events/utils.py | 8 ++++++++ synapse/storage/_base.py | 6 ++++++ tests/storage/test_stream.py | 5 ++--- 3 files changed, 16 insertions(+), 3 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/events/utils.py b/synapse/events/utils.py index 6d9c9352e2..4ab770dd5f 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -113,6 +113,14 @@ def serialize_event(hs, e): d["redacted_by"] = e.unsigned["redacted_by"] del d["unsigned"]["redacted_by"] + if "replaces_state" in e.unsigned: + d["replaces_state"] = e.unsigned["replaces_state"] + del d["unsigned"]["replaces_state"] + + if "prev_content" in e.unsigned: + d["prev_content"] = e.unsigned["prev_content"] + del d["unsigned"]["prev_content"] + del d["auth_events"] del d["prev_events"] del d["hashes"] diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index ffc26d4a61..e9cf73a8e2 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -487,6 +487,12 @@ class SQLBaseStore(object): if because: ev.unsigned["redacted_because"] = because + if "replaces_state" in ev.unsigned: + ev.unsigned["prev_content"] = self._get_event_txn( + txn, + ev.unsigned["replaces_state"], + ).get_dict()["content"] + return ev def _parse_events(self, rows): diff --git a/tests/storage/test_stream.py b/tests/storage/test_stream.py index cba65bb9f0..4865a5c142 100644 --- a/tests/storage/test_stream.py +++ b/tests/storage/test_stream.py @@ -58,7 +58,7 @@ class StreamStoreTestCase(unittest.TestCase): self.depth = 1 @defer.inlineCallbacks - def inject_room_member(self, room, user, membership, replaces_state=None): + def inject_room_member(self, room, user, membership): self.depth += 1 builder = self.event_builder_factory.new({ @@ -215,7 +215,6 @@ class StreamStoreTestCase(unittest.TestCase): event2 = yield self.inject_room_member( self.room1, self.u_alice, Membership.JOIN, - replaces_state=event1.event_id, ) end = yield self.store.get_room_events_max_id() @@ -233,6 +232,6 @@ class StreamStoreTestCase(unittest.TestCase): event = results[0] self.assertTrue( - hasattr(event, "prev_content"), + "prev_content" in event.unsigned, msg="No prev_content key" ) -- cgit 1.4.1 From 0b0436923819a0252a7a2a6f70a1f929b45b9114 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 11 Dec 2014 15:56:01 +0000 Subject: Fix public room joining by making sure replaces_state never points to itself. --- synapse/handlers/federation.py | 14 +++++++------- synapse/state.py | 3 ++- synapse/storage/_base.py | 6 ++++-- 3 files changed, 13 insertions(+), 10 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index 2201cd977e..17779475b8 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -349,7 +349,7 @@ class FederationHandler(BaseHandler): handled_events = set() try: - builder.event_id = self.event_factory.create_event_id() + builder.event_id = self.event_builder_factory.create_event_id() builder.origin = self.hs.hostname builder.content = content @@ -593,13 +593,13 @@ class FederationHandler(BaseHandler): } event = yield self.store.get_event(event_id) - if hasattr(event, "state_key"): + if event and event.is_state(): # Get previous state - if hasattr(event, "replaces_state") and event.replaces_state: - prev_event = yield self.store.get_event( - event.replaces_state - ) - results[(event.type, event.state_key)] = prev_event + if "replaces_state" in event.unsigned: + prev_id = event.unsigned["replaces_state"] + if prev_id != event.event_id: + prev_event = yield self.store.get_event(prev_id) + results[(event.type, event.state_key)] = prev_event else: del results[(event.type, event.state_key)] diff --git a/synapse/state.py b/synapse/state.py index 5bfa73fb46..f9ab5faf9e 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -159,7 +159,8 @@ class StateHandler(object): key = (event.type, event.state_key) if key in context.current_state: replaces = context.current_state[key] - event.unsigned["replaces_state"] = replaces.event_id + if replaces.event_id != event.event_id: # Paranoia check + event.unsigned["replaces_state"] = replaces.event_id defer.returnValue([]) diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index e9cf73a8e2..b6f8817b62 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -451,7 +451,8 @@ class SQLBaseStore(object): return events - def _get_event_txn(self, txn, event_id, check_redacted=True): + def _get_event_txn(self, txn, event_id, check_redacted=True, + get_prev_content=True): sql = ( "SELECT json, r.event_id FROM event_json as e " "LEFT JOIN redactions as r ON e.event_id = r.redacts " @@ -487,10 +488,11 @@ class SQLBaseStore(object): if because: ev.unsigned["redacted_because"] = because - if "replaces_state" in ev.unsigned: + if get_prev_content and "replaces_state" in ev.unsigned: ev.unsigned["prev_content"] = self._get_event_txn( txn, ev.unsigned["replaces_state"], + get_prev_content=False, ).get_dict()["content"] return ev -- cgit 1.4.1 From 75085bb4d16d73df019a671ba483f28b9071d0ff Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 12 Dec 2014 14:34:34 +0000 Subject: Pyflakes --- synapse/storage/_base.py | 1 - 1 file changed, 1 deletion(-) (limited to 'synapse/storage') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index b6f8817b62..1967290ad2 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -23,7 +23,6 @@ from synapse.util.logcontext import PreserveLoggingContext, LoggingContext from twisted.internet import defer import collections -import copy import json import sys import time -- cgit 1.4.1 From c39beb55593cd08abe67eddd902627d6171cb245 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 12 Dec 2014 14:53:37 +0000 Subject: Store json as UTF-8 and not bytes --- synapse/storage/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'synapse/storage') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 3bc3622169..079ddac605 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -168,7 +168,7 @@ class DataStore(RoomMemberStore, RoomStore, values={ "event_id": event.event_id, "room_id": event.room_id, - "json": encode_canonical_json(event_dict), + "json": encode_canonical_json(event_dict).decode("UTF-8"), }, or_replace=True, ) -- cgit 1.4.1 From c8dd3314d673fce90a53520475cdb19d5358dd34 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 15 Dec 2014 13:55:22 +0000 Subject: Fix bug where we ignored event_edge_hashes table --- scripts/check_event_hash.py | 3 +++ synapse/events/utils.py | 1 + synapse/storage/__init__.py | 1 - synapse/storage/_base.py | 1 - synapse/storage/event_federation.py | 11 ++++++----- 5 files changed, 10 insertions(+), 7 deletions(-) (limited to 'synapse/storage') diff --git a/scripts/check_event_hash.py b/scripts/check_event_hash.py index 7c32f8102a..679afbd268 100644 --- a/scripts/check_event_hash.py +++ b/scripts/check_event_hash.py @@ -18,6 +18,9 @@ class dictobj(dict): def get_full_dict(self): return dict(self) + def get_pdu_json(self): + return dict(self) + def main(): parser = argparse.ArgumentParser() diff --git a/synapse/events/utils.py b/synapse/events/utils.py index 4ab770dd5f..94f3f15f52 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -42,6 +42,7 @@ def prune_event(event): "auth_events", "origin", "origin_server_ts", + "membership", ] new_content = {} diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 079ddac605..d0ea304b3d 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -91,7 +91,6 @@ class DataStore(RoomMemberStore, RoomStore, def __init__(self, hs): super(DataStore, self).__init__(hs) - self.event_factory = hs.get_event_factory() self.hs = hs self.min_token_deferred = self._get_min_token() diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 1967290ad2..31d5163c19 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -83,7 +83,6 @@ 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() @defer.inlineCallbacks diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 6c559f8f63..ced066f407 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -177,14 +177,15 @@ class EventFederationStore(SQLBaseStore): retcols=["prev_event_id", "is_state"], ) + hashes = self._get_prev_event_hashes_txn(txn, event_id) + results = [] for d in res: - hashes = self._get_event_reference_hashes_txn( - txn, - d["prev_event_id"] - ) + edge_hash = self._get_event_reference_hashes_txn(txn, d["prev_event_id"]) + edge_hash.update(hashes.get(d["prev_event_id"], {})) prev_hashes = { - k: encode_base64(v) for k, v in hashes.items() + k: encode_base64(v) + for k, v in edge_hash.items() if k == "sha256" } results.append((d["prev_event_id"], prev_hashes, d["is_state"])) -- cgit 1.4.1 From f280929a12314cfdb680881e2458bb047b7346ce Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 15 Dec 2014 17:31:36 +0000 Subject: Use frozenutils --- synapse/events/__init__.py | 36 ++++-------------------------------- synapse/storage/__init__.py | 3 +-- 2 files changed, 5 insertions(+), 34 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index d9dfe5e3f3..f8fbb18e3c 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -15,37 +15,9 @@ from frozendict import frozendict -import copy - - -def _freeze(o): - if isinstance(o, dict) or isinstance(o, frozendict): - return frozendict({k: _freeze(v) for k, v in o.items()}) - - if isinstance(o, basestring): - return o - - try: - return tuple([_freeze(i) for i in o]) - except TypeError: - pass - - return o +from synapse.util.frozenutils import freeze, unfreeze - -def _unfreeze(o): - if isinstance(o, frozendict) or isinstance(o, dict): - return dict({k: _unfreeze(v) for k, v in o.items()}) - - if isinstance(o, basestring): - return o - - try: - return [_unfreeze(i) for i in o] - except TypeError: - pass - - return o +import copy class _EventInternalMetadata(object): @@ -147,7 +119,7 @@ class FrozenEvent(EventBase): signatures = copy.deepcopy(event_dict.pop("signatures", {})) unsigned = copy.deepcopy(event_dict.pop("unsigned", {})) - frozen_dict = _freeze(event_dict) + frozen_dict = freeze(event_dict) super(FrozenEvent, self).__init__( frozen_dict, @@ -167,7 +139,7 @@ class FrozenEvent(EventBase): def get_dict(self): # We need to unfreeze what we return - return _unfreeze(super(FrozenEvent, self).get_dict()) + return unfreeze(super(FrozenEvent, self).get_dict()) def __str__(self): return self.__repr__() diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index d0ea304b3d..e75eaa92d5 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -21,7 +21,6 @@ from synapse.api.events.room import ( ) from synapse.util.logutils import log_function -from synapse.util.frozenutils import FrozenEncoder from .directory import DirectoryStore from .feedback import FeedbackStore @@ -177,7 +176,7 @@ class DataStore(RoomMemberStore, RoomStore, "event_id": event.event_id, "type": event.type, "room_id": event.room_id, - "content": json.dumps(event.content, cls=FrozenEncoder), + "content": json.dumps(event.get_dict()["content"]), "processed": True, "outlier": outlier, "depth": event.depth, -- cgit 1.4.1 From 3c77d13aa5375274e267a0ea898ce6267fb67cdc Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 16 Dec 2014 11:29:05 +0000 Subject: Kill off synapse.api.events.* --- synapse/api/constants.py | 1 + synapse/api/events/__init__.py | 148 -------------------------- synapse/api/events/factory.py | 90 ---------------- synapse/api/events/room.py | 170 ----------------------------- synapse/api/events/utils.py | 85 --------------- synapse/api/events/validator.py | 87 --------------- synapse/federation/replication.py | 1 - synapse/handlers/directory.py | 4 +- synapse/handlers/federation.py | 22 ++-- synapse/handlers/message.py | 1 - synapse/handlers/room.py | 28 +++-- synapse/rest/__init__.py | 2 +- synapse/rest/base.py | 2 - synapse/rest/room.py | 9 +- synapse/server.py | 10 -- synapse/state.py | 4 +- synapse/storage/__init__.py | 16 ++- tests/events/__init__.py | 15 --- tests/events/test_events.py | 217 -------------------------------------- tests/handlers/test_federation.py | 7 +- tests/handlers/test_room.py | 25 ++--- tests/storage/test_redaction.py | 23 ++-- tests/storage/test_room.py | 8 +- tests/storage/test_roommember.py | 5 +- tests/storage/test_stream.py | 11 +- tests/utils.py | 8 +- 26 files changed, 74 insertions(+), 925 deletions(-) delete mode 100644 synapse/api/events/__init__.py delete mode 100644 synapse/api/events/factory.py delete mode 100644 synapse/api/events/room.py delete mode 100644 synapse/api/events/utils.py delete mode 100644 synapse/api/events/validator.py delete mode 100644 tests/events/__init__.py delete mode 100644 tests/events/test_events.py (limited to 'synapse/storage') diff --git a/synapse/api/constants.py b/synapse/api/constants.py index b668da4a26..4fc8b79a40 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -68,6 +68,7 @@ class EventTypes(object): PowerLevels = "m.room.power_levels" Aliases = "m.room.aliases" Redaction = "m.room.redaction" + Feedback = "m.room.message.feedback" # These are used for validation Message = "m.room.message" diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py deleted file mode 100644 index 22939d011a..0000000000 --- a/synapse/api/events/__init__.py +++ /dev/null @@ -1,148 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 OpenMarket Ltd -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from synapse.util.jsonobject import JsonEncodedObject - - -def serialize_event(hs, e): - # FIXME(erikj): To handle the case of presence events and the like - if not isinstance(e, SynapseEvent): - return e - - # Should this strip out None's? - d = {k: v for k, v in e.get_dict().items()} - if "age_ts" in d: - d["age"] = int(hs.get_clock().time_msec()) - d["age_ts"] - del d["age_ts"] - - return d - - -class SynapseEvent(JsonEncodedObject): - - """Base class for Synapse events. These are JSON objects which must abide - by a certain well-defined structure. - """ - - # Attributes that are currently assumed by the federation side: - # Mandatory: - # - event_id - # - room_id - # - type - # - is_state - # - # Optional: - # - state_key (mandatory when is_state is True) - # - prev_events (these can be filled out by the federation layer itself.) - # - prev_state - - valid_keys = [ - "event_id", - "type", - "room_id", - "user_id", # sender/initiator - "content", # HTTP body, JSON - "state_key", - "age_ts", - "prev_content", - "replaces_state", - "redacted_because", - "origin_server_ts", - ] - - internal_keys = [ - "is_state", - "depth", - "destinations", - "origin", - "outlier", - "redacted", - "prev_events", - "hashes", - "signatures", - "prev_state", - "auth_events", - "state_hash", - ] - - required_keys = [ - "event_id", - "room_id", - "content", - ] - - outlier = False - - def __init__(self, raises=True, **kwargs): - super(SynapseEvent, self).__init__(**kwargs) - # if "content" in kwargs: - # self.check_json(self.content, raises=raises) - - def get_content_template(self): - """ Retrieve the JSON template for this event as a dict. - - The template must be a dict representing the JSON to match. Only - required keys should be present. The values of the keys in the template - are checked via type() to the values of the same keys in the actual - event JSON. - - NB: If loading content via json.loads, you MUST define strings as - unicode. - - For example: - Content: - { - "name": u"bob", - "age": 18, - "friends": [u"mike", u"jill"] - } - Template: - { - "name": u"string", - "age": 0, - "friends": [u"string"] - } - The values "string" and 0 could be anything, so long as the types - are the same as the content. - """ - raise NotImplementedError("get_content_template not implemented.") - - def get_pdu_json(self, time_now=None): - pdu_json = self.get_full_dict() - pdu_json.pop("destinations", None) - pdu_json.pop("outlier", None) - pdu_json.pop("replaces_state", None) - pdu_json.pop("redacted", None) - pdu_json.pop("prev_content", None) - state_hash = pdu_json.pop("state_hash", None) - if state_hash is not None: - pdu_json.setdefault("unsigned", {})["state_hash"] = state_hash - content = pdu_json.get("content", {}) - content.pop("prev", None) - if time_now is not None and "age_ts" in pdu_json: - age = time_now - pdu_json["age_ts"] - pdu_json.setdefault("unsigned", {})["age"] = int(age) - del pdu_json["age_ts"] - user_id = pdu_json.pop("user_id") - pdu_json["sender"] = user_id - return pdu_json - - -class SynapseStateEvent(SynapseEvent): - - def __init__(self, **kwargs): - if "state_key" not in kwargs: - kwargs["state_key"] = "" - super(SynapseStateEvent, self).__init__(**kwargs) diff --git a/synapse/api/events/factory.py b/synapse/api/events/factory.py deleted file mode 100644 index 1b84e2b445..0000000000 --- a/synapse/api/events/factory.py +++ /dev/null @@ -1,90 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 OpenMarket Ltd -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from synapse.api.events.room import ( - RoomTopicEvent, MessageEvent, RoomMemberEvent, FeedbackEvent, - InviteJoinEvent, RoomConfigEvent, RoomNameEvent, GenericEvent, - RoomPowerLevelsEvent, RoomJoinRulesEvent, - RoomCreateEvent, - RoomRedactionEvent, -) - -from synapse.types import EventID - -from synapse.util.stringutils import random_string - - -class EventFactory(object): - - _event_classes = [ - RoomTopicEvent, - RoomNameEvent, - MessageEvent, - RoomMemberEvent, - FeedbackEvent, - InviteJoinEvent, - RoomConfigEvent, - RoomPowerLevelsEvent, - RoomJoinRulesEvent, - RoomCreateEvent, - RoomRedactionEvent, - ] - - 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() - self.hs = hs - - self.event_id_count = 0 - - def create_event_id(self): - i = str(self.event_id_count) - self.event_id_count += 1 - - local_part = str(int(self.clock.time())) + i + random_string(5) - - e_id = EventID.create(local_part, self.hs.hostname) - - return e_id.to_string() - - def create_event(self, etype=None, **kwargs): - kwargs["type"] = etype - if "event_id" not in kwargs: - kwargs["event_id"] = self.create_event_id() - kwargs["origin"] = self.hs.hostname - else: - ev_id = self.hs.parse_eventid(kwargs["event_id"]) - kwargs["origin"] = ev_id.domain - - if "origin_server_ts" not in kwargs: - kwargs["origin_server_ts"] = int(self.clock.time_msec()) - - # The "age" key is a delta timestamp that should be converted into an - # absolute timestamp the minute we see it. - if "age" in kwargs: - kwargs["age_ts"] = int(self.clock.time_msec()) - int(kwargs["age"]) - del kwargs["age"] - elif "age_ts" not in kwargs: - kwargs["age_ts"] = int(self.clock.time_msec()) - - if etype in self._event_list: - handler = self._event_list[etype] - else: - handler = GenericEvent - - return handler(**kwargs) diff --git a/synapse/api/events/room.py b/synapse/api/events/room.py deleted file mode 100644 index 8c4ac45d02..0000000000 --- a/synapse/api/events/room.py +++ /dev/null @@ -1,170 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 OpenMarket Ltd -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from synapse.api.constants import Feedback, Membership -from synapse.api.errors import SynapseError -from . import SynapseEvent, SynapseStateEvent - - -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" - - valid_keys = SynapseEvent.valid_keys + [ - # target is the state_key - "membership", # action - ] - - def __init__(self, **kwargs): - if "membership" not in kwargs: - kwargs["membership"] = kwargs.get("content", {}).get("membership") - if not kwargs["membership"] in Membership.LIST: - raise SynapseError(400, "Bad membership value.") - super(RoomMemberEvent, self).__init__(**kwargs) - - def get_content_template(self): - return {"membership": u"string"} - - -class MessageEvent(SynapseEvent): - TYPE = "m.room.message" - - valid_keys = SynapseEvent.valid_keys + [ - "msg_id", # unique per room + user combo - ] - - def __init__(self, **kwargs): - super(MessageEvent, self).__init__(**kwargs) - - def get_content_template(self): - return {"msgtype": u"string"} - - -class FeedbackEvent(SynapseEvent): - TYPE = "m.room.message.feedback" - - valid_keys = SynapseEvent.valid_keys - - def __init__(self, **kwargs): - super(FeedbackEvent, self).__init__(**kwargs) - if not kwargs["content"]["type"] in Feedback.LIST: - raise SynapseError(400, "Bad feedback value.") - - def get_content_template(self): - return { - "type": u"string", - "target_event_id": u"string" - } - - -class InviteJoinEvent(SynapseEvent): - TYPE = "m.room.invite_join" - - valid_keys = SynapseEvent.valid_keys + [ - # target_user_id is the state_key - "target_host", - ] - - def __init__(self, **kwargs): - super(InviteJoinEvent, self).__init__(**kwargs) - - def get_content_template(self): - return {} - - -class RoomConfigEvent(SynapseEvent): - TYPE = "m.room.config" - - def __init__(self, **kwargs): - kwargs["state_key"] = "" - super(RoomConfigEvent, self).__init__(**kwargs) - - def get_content_template(self): - return {} - - -class RoomCreateEvent(SynapseStateEvent): - TYPE = "m.room.create" - - def get_content_template(self): - return {} - - -class RoomJoinRulesEvent(SynapseStateEvent): - TYPE = "m.room.join_rules" - - def get_content_template(self): - return {} - - -class RoomPowerLevelsEvent(SynapseStateEvent): - TYPE = "m.room.power_levels" - - def get_content_template(self): - return {} - - -class RoomAliasesEvent(SynapseStateEvent): - TYPE = "m.room.aliases" - - def get_content_template(self): - return {} - - -class RoomRedactionEvent(SynapseEvent): - TYPE = "m.room.redaction" - - valid_keys = SynapseEvent.valid_keys + ["redacts"] - - def get_content_template(self): - return {} diff --git a/synapse/api/events/utils.py b/synapse/api/events/utils.py deleted file mode 100644 index d6019d56eb..0000000000 --- a/synapse/api/events/utils.py +++ /dev/null @@ -1,85 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 OpenMarket Ltd -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from .room import ( - RoomMemberEvent, RoomJoinRulesEvent, RoomPowerLevelsEvent, - RoomAliasesEvent, RoomCreateEvent, -) - - -def prune_event(event): - """ Returns a pruned version of the given event, which removes all keys we - don't know about or think could potentially be dodgy. - - This is used when we "redact" an event. We want to remove all fields that - the user has specified, but we do want to keep necessary information like - type, state_key etc. - """ - event_type = event.type - - allowed_keys = [ - "event_id", - "user_id", - "room_id", - "hashes", - "signatures", - "content", - "type", - "state_key", - "depth", - "prev_events", - "prev_state", - "auth_events", - "origin", - "origin_server_ts", - ] - - new_content = {} - - def add_fields(*fields): - for field in fields: - if field in event.content: - new_content[field] = event.content[field] - - if event_type == RoomMemberEvent.TYPE: - add_fields("membership") - elif event_type == RoomCreateEvent.TYPE: - add_fields("creator") - elif event_type == RoomJoinRulesEvent.TYPE: - add_fields("join_rule") - elif event_type == RoomPowerLevelsEvent.TYPE: - add_fields( - "users", - "users_default", - "events", - "events_default", - "events_default", - "state_default", - "ban", - "kick", - "redact", - ) - elif event_type == RoomAliasesEvent.TYPE: - add_fields("aliases") - - allowed_fields = { - k: v - for k, v in event.get_full_dict().items() - if k in allowed_keys - } - - allowed_fields["content"] = new_content - - return type(event)(**allowed_fields) diff --git a/synapse/api/events/validator.py b/synapse/api/events/validator.py deleted file mode 100644 index 067215f6ef..0000000000 --- a/synapse/api/events/validator.py +++ /dev/null @@ -1,87 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 OpenMarket Ltd -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from synapse.api.errors import SynapseError, Codes - - -class EventValidator(object): - def __init__(self, hs): - pass - - def validate(self, event): - """Checks the given JSON content abides by the rules of the template. - - Args: - content : A JSON object to check. - raises: True to raise a SynapseError if the check fails. - Returns: - True if the content passes the template. Returns False if the check - fails and raises=False. - Raises: - SynapseError if the check fails and raises=True. - """ - # recursively call to inspect each layer - err_msg = self._check_json_template( - event.content, - event.get_content_template() - ) - if err_msg: - raise SynapseError(400, err_msg, Codes.BAD_JSON) - else: - return True - - def _check_json_template(self, content, template): - """Check content and template matches. - - If the template is a dict, each key in the dict will be validated with - the content, else it will just compare the types of content and - template. This basic type check is required because this function will - be recursively called and could be called with just strs or ints. - - Args: - content: The content to validate. - template: The validation template. - Returns: - str: An error message if the validation fails, else None. - """ - if type(content) != type(template): - return "Mismatched types: %s" % template - - if type(template) == dict: - for key in template: - if key not in content: - return "Missing %s key" % key - - if type(content[key]) != type(template[key]): - return "Key %s is of the wrong type (got %s, want %s)" % ( - key, type(content[key]), type(template[key])) - - if type(content[key]) == dict: - # we must go deeper - msg = self._check_json_template( - content[key], - template[key] - ) - if msg: - return msg - elif type(content[key]) == list: - # make sure each item type in content matches the template - for entry in content[key]: - msg = self._check_json_template( - entry, - template[key][0] - ) - if msg: - return msg diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py index 6388bb98e2..9f8aadccca 100644 --- a/synapse/federation/replication.py +++ b/synapse/federation/replication.py @@ -74,7 +74,6 @@ class ReplicationLayer(object): self._clock = hs.get_clock() - self.event_factory = hs.get_event_factory() self.event_builder_factory = hs.get_event_builder_factory() def set_handler(self, handler): diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py index 4b0869cd9f..404baea796 100644 --- a/synapse/handlers/directory.py +++ b/synapse/handlers/directory.py @@ -18,7 +18,7 @@ from twisted.internet import defer from ._base import BaseHandler from synapse.api.errors import SynapseError, Codes, CodeMessageException -from synapse.api.events.room import RoomAliasesEvent +from synapse.api.constants import EventTypes import logging @@ -150,7 +150,7 @@ class DirectoryHandler(BaseHandler): msg_handler = self.hs.get_handlers().message_handler yield msg_handler.create_and_send_event({ - "type": RoomAliasesEvent.TYPE, + "type": EventTypes.Aliases, "state_key": self.hs.hostname, "room_id": room_id, "sender": user_id, diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index c00f5a7031..16a104c0e2 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -22,8 +22,7 @@ from synapse.events.utils import prune_event from synapse.api.errors import ( AuthError, FederationError, SynapseError, StoreError, ) -from synapse.api.events.room import RoomMemberEvent, RoomCreateEvent -from synapse.api.constants import Membership +from synapse.api.constants import EventTypes, Membership from synapse.util.logutils import log_function from synapse.util.async import run_on_reactor from synapse.crypto.event_signing import ( @@ -225,7 +224,7 @@ class FederationHandler(BaseHandler): if not backfilled: extra_users = [] - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: target_user_id = event.state_key target_user = self.hs.parse_userid(target_user_id) extra_users.append(target_user) @@ -234,7 +233,7 @@ class FederationHandler(BaseHandler): event, extra_users=extra_users ) - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: if event.membership == Membership.JOIN: user = self.hs.parse_userid(event.state_key) yield self.distributor.fire( @@ -333,7 +332,8 @@ class FederationHandler(BaseHandler): event = pdu # We should assert some things. - assert(event.type == RoomMemberEvent.TYPE) + # FIXME: Do this in a nicer way + assert(event.type == EventTypes.Member) assert(event.user_id == joinee) assert(event.state_key == joinee) assert(event.room_id == room_id) @@ -450,7 +450,7 @@ class FederationHandler(BaseHandler): process it until the other server has signed it and sent it back. """ builder = self.event_builder_factory.new({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "content": {"membership": Membership.JOIN}, "room_id": room_id, "sender": user_id, @@ -492,7 +492,7 @@ class FederationHandler(BaseHandler): ) extra_users = [] - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: target_user_id = event.state_key target_user = self.hs.parse_userid(target_user_id) extra_users.append(target_user) @@ -501,7 +501,7 @@ class FederationHandler(BaseHandler): event, extra_users=extra_users ) - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: if event.content["membership"] == Membership.JOIN: user = self.hs.parse_userid(event.state_key) yield self.distributor.fire( @@ -514,7 +514,7 @@ class FederationHandler(BaseHandler): for k, s in context.current_state.items(): try: - if k[0] == RoomMemberEvent.TYPE: + if k[0] == EventTypes.Member: if s.content["membership"] == Membership.JOIN: destinations.add( self.hs.parse_userid(s.state_key).domain @@ -731,10 +731,10 @@ class FederationHandler(BaseHandler): event.event_id, event.signatures, ) - if event.type == RoomMemberEvent.TYPE and not event.auth_events: + if event.type == EventTypes.Member and not event.auth_events: if len(event.prev_events) == 1: c = yield self.store.get_event(event.prev_events[0][0]) - if c.type == RoomCreateEvent.TYPE: + if c.type == EventTypes.Create: context.auth_events[(c.type, c.state_key)] = c logger.debug( diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 1eed38c6d1..baf372fdad 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -35,7 +35,6 @@ class MessageHandler(BaseHandler): super(MessageHandler, self).__init__(hs) self.hs = hs self.clock = hs.get_clock() - self.event_factory = hs.get_event_factory() self.validator = EventValidator() @defer.inlineCallbacks diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index f7cc869225..8567d7409d 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -17,12 +17,8 @@ from twisted.internet import defer from synapse.types import UserID, RoomAlias, RoomID -from synapse.api.constants import Membership, JoinRules +from synapse.api.constants import EventTypes, Membership, JoinRules from synapse.api.errors import StoreError, SynapseError -from synapse.api.events.room import ( - RoomMemberEvent, RoomCreateEvent, RoomPowerLevelsEvent, - RoomTopicEvent, RoomNameEvent, RoomJoinRulesEvent, -) from synapse.util import stringutils from synapse.util.async import run_on_reactor from ._base import BaseHandler @@ -131,7 +127,7 @@ class RoomCreationHandler(BaseHandler): if "name" in config: name = config["name"] yield msg_handler.create_and_send_event({ - "type": RoomNameEvent.TYPE, + "type": EventTypes.Name, "room_id": room_id, "sender": user_id, "content": {"name": name}, @@ -140,7 +136,7 @@ class RoomCreationHandler(BaseHandler): if "topic" in config: topic = config["topic"] yield msg_handler.create_and_send_event({ - "type": RoomTopicEvent.TYPE, + "type": EventTypes.Topic, "room_id": room_id, "sender": user_id, "content": {"topic": topic}, @@ -148,7 +144,7 @@ class RoomCreationHandler(BaseHandler): for invitee in invite_list: yield msg_handler.create_and_send_event({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "state_key": invitee, "room_id": room_id, "user_id": user_id, @@ -186,12 +182,12 @@ class RoomCreationHandler(BaseHandler): return e creation_event = create( - etype=RoomCreateEvent.TYPE, + etype=EventTypes.Create, content={"creator": creator.to_string()}, ) join_event = create( - etype=RoomMemberEvent.TYPE, + etype=EventTypes.Member, state_key=creator_id, content={ "membership": Membership.JOIN, @@ -199,15 +195,15 @@ class RoomCreationHandler(BaseHandler): ) power_levels_event = create( - etype=RoomPowerLevelsEvent.TYPE, + etype=EventTypes.PowerLevels, content={ "users": { creator.to_string(): 100, }, "users_default": 0, "events": { - RoomNameEvent.TYPE: 100, - RoomPowerLevelsEvent.TYPE: 100, + EventTypes.Name: 100, + EventTypes.PowerLevels: 100, }, "events_default": 0, "state_default": 50, @@ -219,7 +215,7 @@ class RoomCreationHandler(BaseHandler): join_rule = JoinRules.PUBLIC if is_public else JoinRules.INVITE join_rules_event = create( - etype=RoomJoinRulesEvent.TYPE, + etype=EventTypes.JoinRules, content={"join_rule": join_rule}, ) @@ -344,7 +340,7 @@ class RoomMemberHandler(BaseHandler): target_user_id = event.state_key prev_state = context.current_state.get( - (RoomMemberEvent.TYPE, target_user_id), + (EventTypes.Member, target_user_id), None ) @@ -396,7 +392,7 @@ class RoomMemberHandler(BaseHandler): content.update({"membership": Membership.JOIN}) builder = self.event_builder_factory.new({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "state_key": joinee.to_string(), "room_id": room_id, "sender": joinee.to_string(), diff --git a/synapse/rest/__init__.py b/synapse/rest/__init__.py index e391e5678d..a59630ec96 100644 --- a/synapse/rest/__init__.py +++ b/synapse/rest/__init__.py @@ -28,7 +28,7 @@ class RestServletFactory(object): speaking, they serve as wrappers around events and the handlers that process them. - See synapse.api.events for information on synapse events. + See synapse.events for information on synapse events. """ def __init__(self, hs): diff --git a/synapse/rest/base.py b/synapse/rest/base.py index 72bb66ddda..06eda2587c 100644 --- a/synapse/rest/base.py +++ b/synapse/rest/base.py @@ -67,8 +67,6 @@ class RestServlet(object): self.auth = hs.get_auth() self.txns = HttpTransactionStore() - self.validator = hs.get_event_validator() - def register(self, http_server): """ Register this servlet with the given HTTP server. """ if hasattr(self, "PATTERN"): diff --git a/synapse/rest/room.py b/synapse/rest/room.py index 1a527d27c1..0e2d5fbaae 100644 --- a/synapse/rest/room.py +++ b/synapse/rest/room.py @@ -19,8 +19,7 @@ from twisted.internet import defer from base import RestServlet, client_path_pattern from synapse.api.errors import SynapseError, Codes from synapse.streams.config import PaginationConfig -from synapse.api.events.room import RoomMemberEvent, RoomRedactionEvent -from synapse.api.constants import Membership +from synapse.api.constants import EventTypes, Membership import json import logging @@ -239,7 +238,7 @@ class JoinRoomAliasServlet(RestServlet): msg_handler = self.handlers.message_handler yield msg_handler.create_and_send_event( { - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "content": {"membership": Membership.JOIN}, "room_id": identifier.to_string(), "sender": user.to_string(), @@ -403,7 +402,7 @@ class RoomMembershipRestServlet(RestServlet): msg_handler = self.handlers.message_handler yield msg_handler.create_and_send_event( { - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "content": {"membership": unicode(membership_action)}, "room_id": room_id, "sender": user.to_string(), @@ -441,7 +440,7 @@ class RoomRedactEventRestServlet(RestServlet): msg_handler = self.handlers.message_handler event = yield msg_handler.create_and_send_event( { - "type": RoomRedactionEvent.TYPE, + "type": EventTypes.Redaction, "content": content, "room_id": room_id, "sender": user.to_string(), diff --git a/synapse/server.py b/synapse/server.py index 1c0703c51f..e4021481e8 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -21,8 +21,6 @@ # Imports required for the default HomeServer() implementation from synapse.federation import initialize_http_replication from synapse.events.utils import serialize_event -from synapse.api.events.factory import EventFactory -from synapse.api.events.validator import EventValidator from synapse.notifier import Notifier from synapse.api.auth import Auth from synapse.handlers import Handlers @@ -66,7 +64,6 @@ class BaseHomeServer(object): 'persistence_service', 'replication_layer', 'datastore', - 'event_factory', 'handlers', 'auth', 'rest_servlet_factory', @@ -83,7 +80,6 @@ class BaseHomeServer(object): 'event_sources', 'ratelimiter', 'keyring', - 'event_validator', 'event_builder_factory', ] @@ -198,9 +194,6 @@ class HomeServer(BaseHomeServer): def build_datastore(self): return DataStore(self) - def build_event_factory(self): - return EventFactory(self) - def build_handlers(self): return Handlers(self) @@ -231,9 +224,6 @@ class HomeServer(BaseHomeServer): def build_keyring(self): return Keyring(self) - def build_event_validator(self): - return EventValidator(self) - def build_event_builder_factory(self): return EventBuilderFactory( clock=self.get_clock(), diff --git a/synapse/state.py b/synapse/state.py index f9ab5faf9e..d2763cdd9a 100644 --- a/synapse/state.py +++ b/synapse/state.py @@ -18,7 +18,7 @@ from twisted.internet import defer from synapse.util.logutils import log_function from synapse.util.async import run_on_reactor -from synapse.api.events.room import RoomPowerLevelsEvent +from synapse.api.constants import EventTypes from collections import namedtuple @@ -271,7 +271,7 @@ class StateHandler(object): def _get_power_level_from_event_state(self, event, user_id): if hasattr(event, "old_state_events") and event.old_state_events: - key = (RoomPowerLevelsEvent.TYPE, "", ) + key = (EventTypes.PowerLevels, "", ) power_level_event = event.old_state_events.get(key) level = None if power_level_event: diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index e75eaa92d5..5c079da5ba 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -15,12 +15,8 @@ from twisted.internet import defer -from synapse.api.events.room import ( - RoomMemberEvent, RoomTopicEvent, FeedbackEvent, RoomNameEvent, - RoomRedactionEvent, -) - from synapse.util.logutils import log_function +from synapse.api.constants import EventTypes from .directory import DirectoryStore from .feedback import FeedbackStore @@ -136,15 +132,15 @@ class DataStore(RoomMemberStore, RoomStore, def _persist_event_txn(self, txn, event, context, backfilled, stream_ordering=None, is_new_state=True, current_state=None): - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: self._store_room_member_txn(txn, event) - elif event.type == FeedbackEvent.TYPE: + elif event.type == EventTypes.Feedback: self._store_feedback_txn(txn, event) - elif event.type == RoomNameEvent.TYPE: + elif event.type == EventTypes.Name: self._store_room_name_txn(txn, event) - elif event.type == RoomTopicEvent.TYPE: + elif event.type == EventTypes.Topic: self._store_room_topic_txn(txn, event) - elif event.type == RoomRedactionEvent.TYPE: + elif event.type == EventTypes.Redaction: self._store_redaction(txn, event) outlier = False diff --git a/tests/events/__init__.py b/tests/events/__init__.py deleted file mode 100644 index 9bff9ec169..0000000000 --- a/tests/events/__init__.py +++ /dev/null @@ -1,15 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 OpenMarket Ltd -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - diff --git a/tests/events/test_events.py b/tests/events/test_events.py deleted file mode 100644 index 91d1d44fee..0000000000 --- a/tests/events/test_events.py +++ /dev/null @@ -1,217 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014 OpenMarket Ltd -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from synapse.api.events import SynapseEvent -from synapse.api.events.validator import EventValidator -from synapse.api.errors import SynapseError - -from tests import unittest - - -class SynapseTemplateCheckTestCase(unittest.TestCase): - - def setUp(self): - self.validator = EventValidator(None) - - def tearDown(self): - pass - - def test_top_level_keys(self): - template = { - "person": {}, - "friends": ["string"] - } - - content = { - "person": {"name": "bob"}, - "friends": ["jill", "mike"] - } - - event = MockSynapseEvent(template) - event.content = content - self.assertTrue(self.validator.validate(event)) - - content = { - "person": {"name": "bob"}, - "friends": ["jill"], - "enemies": ["mike"] - } - event.content = content - self.assertTrue(self.validator.validate(event)) - - content = { - "person": {"name": "bob"}, - # missing friends - "enemies": ["mike", "jill"] - } - event.content = content - self.assertRaises( - SynapseError, - self.validator.validate, - event - ) - - def test_lists(self): - template = { - "person": {}, - "friends": [{"name":"string"}] - } - - content = { - "person": {"name": "bob"}, - "friends": ["jill", "mike"] # should be in objects - } - - event = MockSynapseEvent(template) - event.content = content - self.assertRaises( - SynapseError, - self.validator.validate, - event - ) - - content = { - "person": {"name": "bob"}, - "friends": [{"name": "jill"}, {"name": "mike"}] - } - event.content = content - self.assertTrue(self.validator.validate(event)) - - def test_nested_lists(self): - template = { - "results": { - "families": [ - { - "name": "string", - "members": [ - {} - ] - } - ] - } - } - - content = { - "results": { - "families": [ - { - "name": "Smith", - "members": [ - "Alice", "Bob" # wrong types - ] - } - ] - } - } - - event = MockSynapseEvent(template) - event.content = content - self.assertRaises( - SynapseError, - self.validator.validate, - event - ) - - content = { - "results": { - "families": [ - { - "name": "Smith", - "members": [ - {"name": "Alice"}, {"name": "Bob"} - ] - } - ] - } - } - event.content = content - self.assertTrue(self.validator.validate(event)) - - def test_nested_keys(self): - template = { - "person": { - "attributes": { - "hair": "string", - "eye": "string" - }, - "age": 0, - "fav_books": ["string"] - } - } - event = MockSynapseEvent(template) - - content = { - "person": { - "attributes": { - "hair": "brown", - "eye": "green", - "skin": "purple" - }, - "age": 33, - "fav_books": ["lotr", "hobbit"], - "fav_music": ["abba", "beatles"] - } - } - - event.content = content - self.assertTrue(self.validator.validate(event)) - - content = { - "person": { - "attributes": { - "hair": "brown" - # missing eye - }, - "age": 33, - "fav_books": ["lotr", "hobbit"], - "fav_music": ["abba", "beatles"] - } - } - - event.content = content - self.assertRaises( - SynapseError, - self.validator.validate, - event - ) - - content = { - "person": { - "attributes": { - "hair": "brown", - "eye": "green", - "skin": "purple" - }, - "age": 33, - "fav_books": "nothing", # should be a list - } - } - - event.content = content - self.assertRaises( - SynapseError, - self.validator.validate, - event - ) - - -class MockSynapseEvent(SynapseEvent): - - def __init__(self, template): - self.template = template - - def get_content_template(self): - return self.template - diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index 91f7351087..ed351367cc 100644 --- a/tests/handlers/test_federation.py +++ b/tests/handlers/test_federation.py @@ -16,10 +16,7 @@ from twisted.internet import defer from tests import unittest -from synapse.api.events.room import ( - MessageEvent, -) - +from synapse.api.constants import EventTypes from synapse.events import FrozenEvent from synapse.handlers.federation import FederationHandler from synapse.server import HomeServer @@ -79,7 +76,7 @@ class FederationTestCase(unittest.TestCase): @defer.inlineCallbacks def test_msg(self): pdu = FrozenEvent({ - "type": MessageEvent.TYPE, + "type": EventTypes.Message, "room_id": "foo", "content": {"msgtype": u"fooo"}, "origin_server_ts": 0, diff --git a/tests/handlers/test_room.py b/tests/handlers/test_room.py index 9c63f2a882..83493cae20 100644 --- a/tests/handlers/test_room.py +++ b/tests/handlers/test_room.py @@ -17,9 +17,6 @@ from twisted.internet import defer from tests import unittest -from synapse.api.events.room import ( - RoomMemberEvent, -) from synapse.api.constants import EventTypes, Membership from synapse.handlers.room import RoomMemberHandler, RoomCreationHandler from synapse.handlers.profile import ProfileHandler @@ -102,7 +99,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): content = {"membership": Membership.INVITE} builder = self.hs.get_event_builder_factory().new({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "sender": user_id, "state_key": target_user_id, "room_id": room_id, @@ -115,11 +112,11 @@ class RoomMemberHandlerTestCase(unittest.TestCase): def annotate(_, ctx): ctx.current_state = { - (RoomMemberEvent.TYPE, "@alice:green"): self._create_member( + (EventTypes.Member, "@alice:green"): self._create_member( user_id="@alice:green", room_id=room_id, ), - (RoomMemberEvent.TYPE, "@bob:red"): self._create_member( + (EventTypes.Member, "@bob:red"): self._create_member( user_id="@bob:red", room_id=room_id, ), @@ -131,7 +128,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): def add_auth(_, ctx): ctx.auth_events = ctx.current_state[ - (RoomMemberEvent.TYPE, "@bob:red") + (EventTypes.Member, "@bob:red") ] return defer.succeed(True) @@ -181,7 +178,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): self.distributor.observe("user_joined_room", join_signal_observer) builder = self.hs.get_event_builder_factory().new({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "sender": user_id, "state_key": user_id, "room_id": room_id, @@ -194,7 +191,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): def annotate(_, ctx): ctx.current_state = { - (RoomMemberEvent.TYPE, "@bob:red"): self._create_member( + (EventTypes.Member, "@bob:red"): self._create_member( user_id="@bob:red", room_id=room_id, membership=Membership.INVITE @@ -207,7 +204,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): def add_auth(_, ctx): ctx.auth_events = ctx.current_state[ - (RoomMemberEvent.TYPE, "@bob:red") + (EventTypes.Member, "@bob:red") ] return defer.succeed(True) @@ -238,7 +235,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): def _create_member(self, user_id, room_id, membership=Membership.JOIN): builder = self.hs.get_event_builder_factory().new({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "sender": user_id, "state_key": user_id, "room_id": room_id, @@ -254,7 +251,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): user = self.hs.parse_userid(user_id) builder = self.hs.get_event_builder_factory().new({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "sender": user_id, "state_key": user_id, "room_id": room_id, @@ -267,7 +264,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): def annotate(_, ctx): ctx.current_state = { - (RoomMemberEvent.TYPE, "@bob:red"): self._create_member( + (EventTypes.Member, "@bob:red"): self._create_member( user_id="@bob:red", room_id=room_id, membership=Membership.JOIN @@ -280,7 +277,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase): def add_auth(_, ctx): ctx.auth_events = ctx.current_state[ - (RoomMemberEvent.TYPE, "@bob:red") + (EventTypes.Member, "@bob:red") ] return defer.succeed(True) diff --git a/tests/storage/test_redaction.py b/tests/storage/test_redaction.py index f670e154c3..9806fbc69b 100644 --- a/tests/storage/test_redaction.py +++ b/tests/storage/test_redaction.py @@ -18,10 +18,7 @@ from tests import unittest from twisted.internet import defer from synapse.server import HomeServer -from synapse.api.constants import Membership -from synapse.api.events.room import ( - RoomMemberEvent, MessageEvent, RoomRedactionEvent, -) +from synapse.api.constants import EventTypes, Membership from tests.utils import SQLiteMemoryDbPool, MockKey @@ -64,7 +61,7 @@ class RedactionTestCase(unittest.TestCase): content = {"membership": membership} content.update(extra_content) builder = self.event_builder_factory.new({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "sender": user.to_string(), "state_key": user.to_string(), "room_id": room.to_string(), @@ -84,7 +81,7 @@ class RedactionTestCase(unittest.TestCase): self.depth += 1 builder = self.event_builder_factory.new({ - "type": MessageEvent.TYPE, + "type": EventTypes.Message, "sender": user.to_string(), "state_key": user.to_string(), "room_id": room.to_string(), @@ -102,7 +99,7 @@ class RedactionTestCase(unittest.TestCase): @defer.inlineCallbacks def inject_redaction(self, room, event_id, user, reason): builder = self.event_builder_factory.new({ - "type": RoomRedactionEvent.TYPE, + "type": EventTypes.Redaction, "sender": user.to_string(), "state_key": user.to_string(), "room_id": room.to_string(), @@ -142,7 +139,7 @@ class RedactionTestCase(unittest.TestCase): self.assertObjectHasAttributes( { - "type": MessageEvent.TYPE, + "type": EventTypes.Message, "user_id": self.u_alice.to_string(), "content": {"body": "t", "msgtype": "message"}, }, @@ -176,7 +173,7 @@ class RedactionTestCase(unittest.TestCase): self.assertObjectHasAttributes( { - "type": MessageEvent.TYPE, + "type": EventTypes.Message, "user_id": self.u_alice.to_string(), "content": {}, }, @@ -185,7 +182,7 @@ class RedactionTestCase(unittest.TestCase): self.assertObjectHasAttributes( { - "type": RoomRedactionEvent.TYPE, + "type": EventTypes.Redaction, "user_id": self.u_alice.to_string(), "content": {"reason": reason}, }, @@ -221,7 +218,7 @@ class RedactionTestCase(unittest.TestCase): self.assertObjectHasAttributes( { - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "user_id": self.u_bob.to_string(), "content": {"membership": Membership.JOIN, "blue": "red"}, }, @@ -253,7 +250,7 @@ class RedactionTestCase(unittest.TestCase): self.assertObjectHasAttributes( { - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "user_id": self.u_bob.to_string(), "content": {"membership": Membership.JOIN}, }, @@ -262,7 +259,7 @@ class RedactionTestCase(unittest.TestCase): self.assertObjectHasAttributes( { - "type": RoomRedactionEvent.TYPE, + "type": EventTypes.Redaction, "user_id": self.u_alice.to_string(), "content": {"reason": reason}, }, diff --git a/tests/storage/test_room.py b/tests/storage/test_room.py index 4ff02c306b..11761fe29a 100644 --- a/tests/storage/test_room.py +++ b/tests/storage/test_room.py @@ -18,9 +18,7 @@ from tests import unittest from twisted.internet import defer from synapse.server import HomeServer -from synapse.api.events.room import ( - RoomNameEvent, RoomTopicEvent -) +from synapse.api.constants import EventTypes from tests.utils import SQLiteMemoryDbPool @@ -131,7 +129,7 @@ class RoomEventsStoreTestCase(unittest.TestCase): name = u"A-Room-Name" yield self.inject_room_event( - etype=RoomNameEvent.TYPE, + etype=EventTypes.Name, name=name, content={"name": name}, depth=1, @@ -154,7 +152,7 @@ class RoomEventsStoreTestCase(unittest.TestCase): topic = u"A place for things" yield self.inject_room_event( - etype=RoomTopicEvent.TYPE, + etype=EventTypes.Topic, topic=topic, content={"topic": topic}, depth=1, diff --git a/tests/storage/test_roommember.py b/tests/storage/test_roommember.py index 6df09952d1..a23a8189df 100644 --- a/tests/storage/test_roommember.py +++ b/tests/storage/test_roommember.py @@ -18,8 +18,7 @@ from tests import unittest from twisted.internet import defer from synapse.server import HomeServer -from synapse.api.constants import Membership -from synapse.api.events.room import RoomMemberEvent +from synapse.api.constants import EventTypes, Membership from tests.utils import SQLiteMemoryDbPool, MockKey @@ -61,7 +60,7 @@ class RoomMemberStoreTestCase(unittest.TestCase): @defer.inlineCallbacks def inject_room_member(self, room, user, membership, replaces_state=None): builder = self.event_builder_factory.new({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "sender": user.to_string(), "state_key": user.to_string(), "room_id": room.to_string(), diff --git a/tests/storage/test_stream.py b/tests/storage/test_stream.py index 4865a5c142..9247fc579e 100644 --- a/tests/storage/test_stream.py +++ b/tests/storage/test_stream.py @@ -18,8 +18,7 @@ from tests import unittest from twisted.internet import defer from synapse.server import HomeServer -from synapse.api.constants import Membership -from synapse.api.events.room import RoomMemberEvent, MessageEvent +from synapse.api.constants import EventTypes, Membership from tests.utils import SQLiteMemoryDbPool, MockKey @@ -62,7 +61,7 @@ class StreamStoreTestCase(unittest.TestCase): self.depth += 1 builder = self.event_builder_factory.new({ - "type": RoomMemberEvent.TYPE, + "type": EventTypes.Member, "sender": user.to_string(), "state_key": user.to_string(), "room_id": room.to_string(), @@ -82,7 +81,7 @@ class StreamStoreTestCase(unittest.TestCase): self.depth += 1 builder = self.event_builder_factory.new({ - "type": MessageEvent.TYPE, + "type": EventTypes.Message, "sender": user.to_string(), "state_key": user.to_string(), "room_id": room.to_string(), @@ -125,7 +124,7 @@ class StreamStoreTestCase(unittest.TestCase): self.assertObjectHasAttributes( { - "type": MessageEvent.TYPE, + "type": EventTypes.Message, "user_id": self.u_alice.to_string(), "content": {"body": "test", "msgtype": "message"}, }, @@ -162,7 +161,7 @@ class StreamStoreTestCase(unittest.TestCase): self.assertObjectHasAttributes( { - "type": MessageEvent.TYPE, + "type": EventTypes.Message, "user_id": self.u_alice.to_string(), "content": {"body": "test", "msgtype": "message"}, }, diff --git a/tests/utils.py b/tests/utils.py index 70a221550c..731e03f517 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -15,15 +15,11 @@ from synapse.http.server import HttpServer from synapse.api.errors import cs_error, CodeMessageException, StoreError -from synapse.api.constants import Membership +from synapse.api.constants import EventTypes from synapse.storage import prepare_database from synapse.util.logcontext import LoggingContext -from synapse.api.events.room import ( - RoomMemberEvent, MessageEvent -) - from twisted.internet import defer, reactor from twisted.enterprise.adbapi import ConnectionPool @@ -276,7 +272,7 @@ class MemoryDataStore(object): return defer.succeed([]) def persist_event(self, event): - if event.type == RoomMemberEvent.TYPE: + if event.type == EventTypes.Member: room_id = event.room_id user = event.state_key membership = event.membership -- cgit 1.4.1 From 882dc8dcab434d24b0e902be3b4545ceabe9bdbe Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 16 Dec 2014 13:17:09 +0000 Subject: Persist internal_metadata --- synapse/events/__init__.py | 5 +++-- synapse/storage/__init__.py | 5 +++++ synapse/storage/_base.py | 7 ++++--- synapse/storage/schema/im.sql | 1 + 4 files changed, 13 insertions(+), 5 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py index 1ec79a6788..984f14fce4 100644 --- a/synapse/events/__init__.py +++ b/synapse/events/__init__.py @@ -111,7 +111,7 @@ class EventBase(object): class FrozenEvent(EventBase): - def __init__(self, event_dict): + def __init__(self, event_dict, internal_metadata_dict={}): event_dict = copy.deepcopy(event_dict) signatures = copy.deepcopy(event_dict.pop("signatures", {})) @@ -122,7 +122,8 @@ class FrozenEvent(EventBase): super(FrozenEvent, self).__init__( frozen_dict, signatures=signatures, - unsigned=unsigned + unsigned=unsigned, + internal_metadata_dict=internal_metadata_dict, ) @staticmethod diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 5c079da5ba..26f205ae8f 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -156,12 +156,17 @@ class DataStore(RoomMemberStore, RoomStore, ] } + metadata_json = encode_canonical_json( + event.internal_metadata.get_dict() + ) + self._simple_insert_txn( txn, table="event_json", values={ "event_id": event.event_id, "room_id": event.room_id, + "internal_metadata": metadata_json.decode("UTF-8"), "json": encode_canonical_json(event_dict).decode("UTF-8"), }, or_replace=True, diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 31d5163c19..6dc857c4aa 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -452,7 +452,7 @@ class SQLBaseStore(object): def _get_event_txn(self, txn, event_id, check_redacted=True, get_prev_content=True): sql = ( - "SELECT json, r.event_id FROM event_json as e " + "SELECT internal_metadata, json, r.event_id FROM event_json as e " "LEFT JOIN redactions as r ON e.event_id = r.redacts " "WHERE e.event_id = ? " "LIMIT 1 " @@ -465,11 +465,12 @@ class SQLBaseStore(object): if not res: return None - js, redacted = res + internal_metadata, js, redacted = res d = json.loads(js) + internal_metadata = json.loads(internal_metadata) - ev = FrozenEvent(d) + ev = FrozenEvent(d, internal_metadata_dict=internal_metadata) if check_redacted and redacted: ev = prune_event(ev) diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql index 0300bb29e1..253f9f779b 100644 --- a/synapse/storage/schema/im.sql +++ b/synapse/storage/schema/im.sql @@ -36,6 +36,7 @@ CREATE INDEX IF NOT EXISTS events_room_id ON events (room_id); CREATE TABLE IF NOT EXISTS event_json( event_id TEXT NOT NULL, room_id TEXT NOT NULL, + internal_metadata NOT NULL, json BLOB NOT NULL, CONSTRAINT ev_j_uniq UNIQUE (event_id) ); -- cgit 1.4.1 From 35f4f6b07019a9815fb06529412c35a65c3cd285 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 16 Dec 2014 13:27:53 +0000 Subject: Update upgrade script --- scripts/upgrade_db_to_v0.5.5.py | 13 +++++++++++-- synapse/storage/__init__.py | 2 +- 2 files changed, 12 insertions(+), 3 deletions(-) (limited to 'synapse/storage') diff --git a/scripts/upgrade_db_to_v0.5.5.py b/scripts/upgrade_db_to_v0.5.5.py index be9d07b2df..aa0d7667ac 100644 --- a/scripts/upgrade_db_to_v0.5.5.py +++ b/scripts/upgrade_db_to_v0.5.5.py @@ -31,6 +31,7 @@ delta_sql = """ CREATE TABLE IF NOT EXISTS event_json( event_id TEXT NOT NULL, room_id TEXT NOT NULL, + internal_metadata NOT NULL, json BLOB NOT NULL, CONSTRAINT ev_j_uniq UNIQUE (event_id) ); @@ -79,7 +80,7 @@ class Store(object): d.setdefault("unsigned", {})["age_ts"] = d.pop("age_ts") - d.pop("outlier", None) + outlier = d.pop("outlier", False) # d.pop("membership", None) @@ -87,7 +88,10 @@ class Store(object): d.pop("replaces_state", None) - events.append(EventBuilder(d)) + b = EventBuilder(d) + b.internal_metadata.outlier = outlier + + events.append(b) for i, ev in enumerate(events): signatures = self._get_event_signatures_txn( @@ -251,12 +255,17 @@ def reinsert_events(cursor, server_name, signing_key): event.get_dict() ).decode("UTF-8") + metadata_json = encode_canonical_json( + event.internal_metadata.get_dict() + ).decode("UTF-8") + store._simple_insert_txn( cursor, table="event_json", values={ "event_id": event.event_id, "room_id": event.room_id, + "internal_metadata": metadata_json, "json": event_json, }, or_replace=True, diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 26f205ae8f..cc1dcc2e74 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -156,7 +156,7 @@ class DataStore(RoomMemberStore, RoomStore, ] } - metadata_json = encode_canonical_json( + metadata_json = encode_canonical_json( event.internal_metadata.get_dict() ) -- cgit 1.4.1