Merge pull request #25 from matrix-org/events_refactor

Event refactor

1 files changed, 76 insertions, 37 deletions

diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index 15adc9dc2c..af8eb5f0f5 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -15,11 +15,12 @@
 
 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 EventContext
 
 import logging
 
@@ -31,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()
@@ -44,6 +43,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(
@@ -57,62 +58,100 @@ class BaseHandler(object):
             )
 
     @defer.inlineCallbacks
-    def _on_new_room_event(self, event, snapshot, extra_destinations=[],
-                           extra_users=[], suppress_auth=False,
-                           do_invite_host=None):
+    def _create_new_client_event(self, builder):
         yield run_on_reactor()
 
-        snapshot.fill_out_prev_events(event)
+        context = EventContext()
+
+        latest_ret = yield self.store.get_latest_events_in_room(
+            builder.room_id,
+        )
+
+        if latest_ret:
+            depth = max([d for _, _, d in latest_ret]) + 1
+        else:
+            depth = 1
+
+        prev_events = [(e, h) for e, h, _ in latest_ret]
+
+        builder.prev_events = prev_events
+        builder.depth = depth
 
-        yield self.state_handler.annotate_event_with_state(event)
+        state_handler = self.state_handler
+        ret = yield state_handler.annotate_context_with_state(
+            builder,
+            context,
+        )
+        prev_state = ret
 
-        yield self.auth.add_auth_events(event)
+        if builder.is_state():
+            builder.prev_state = prev_state
 
-        logger.debug("Signing event...")
+        yield self.auth.add_auth_events(builder, context)
 
         add_hashes_and_signatures(
-            event, self.server_name, self.signing_key
+            builder, self.server_name, self.signing_key
+        )
+
+        event = builder.build()
+
+        logger.debug(
+            "Created event %s with auth_events: %s, current state: %s",
+            event.event_id, context.auth_events, context.current_state,
         )
 
-        logger.debug("Signed event.")
+        defer.returnValue(
+            (event, context,)
+        )
+
+    @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:
-            logger.debug("Authing...")
-            self.auth.check(event, auth_events=event.old_state_events)
-            logger.debug("Authed")
-        else:
-            logger.debug("Suppressed auth.")
+            self.auth.check(event, auth_events=context.auth_events)
 
-        if do_invite_host:
-            federation_handler = self.hs.get_handlers().federation_handler
-            invite_event = yield federation_handler.send_invite(
-                do_invite_host,
-                event
-            )
+        yield self.store.persist_event(event, context=context)
 
-            # FIXME: We need to check if the remote changed anything else
-            event.signatures = invite_event.signatures
+        federation_handler = self.hs.get_handlers().federation_handler
 
-        yield self.store.persist_event(event)
+        if event.type == EventTypes.Member:
+            if event.content["membership"] == Membership.INVITE:
+                invitee = self.hs.parse_userid(event.state_key)
+                if not self.hs.is_mine(invitee):
+                    # TODO: Can we add signature from remote server in a nicer
+                    # way? If we have been invited by a remote server, we need
+                    # to get them to sign the event.
+                    returned_invite = yield federation_handler.send_invite(
+                        invitee.domain,
+                        event,
+                    )
+
+                    # TODO: Make sure the signatures actually are correct.
+                    event.signatures.update(
+                        returned_invite.signatures
+                    )
 
         destinations = set(extra_destinations)
-        # Send a PDU to all hosts who have joined the room.
-
-        for k, s in event.state_events.items():
+        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
                         )
-            except:
+            except SynapseError:
                 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)
+        yield federation_handler.handle_new_event(
+            event,
+            None,
+            destinations=destinations,
+        )