From 2d1dfb3b34583a4de7e1e53f685c2564a7fc731f Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 28 Oct 2014 16:42:35 +0000
Subject: Begin implementing all the PDU storage stuff in Events land

---
 synapse/storage/schema/event_edges.sql | 51 ++++++++++++++++++++++++++++++++++
 1 file changed, 51 insertions(+)
 create mode 100644 synapse/storage/schema/event_edges.sql

(limited to 'synapse/storage/schema/event_edges.sql')

diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql
new file mode 100644
index 0000000000..6a28314ece
--- /dev/null
+++ b/synapse/storage/schema/event_edges.sql
@@ -0,0 +1,51 @@
+
+CREATE TABLE IF NOT EXISTS event_forward_extremities(
+    event_id TEXT,
+    room_id TEXT,
+    CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE
+);
+
+CREATE INDEX IF NOT EXISTS ev_extrem_room ON event_forward_extremities(room_id);
+CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id);
+--
+
+CREATE TABLE IF NOT EXISTS event_backward_extremities(
+    event_id TEXT,
+    room_id TEXT,
+    CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE
+);
+
+CREATE INDEX IF NOT EXISTS ev_b_extrem_room ON event_backward_extremities(room_id);
+CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id);
+--
+
+CREATE TABLE IF NOT EXISTS event_edges(
+    event_id TEXT,
+    prev_event_id TEXT,
+    room_id TEXT,
+    CONSTRAINT uniqueness UNIQUE (event_id, prev_event_id, room_id)
+);
+
+CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id);
+CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id);
+--
+
+
+CREATE TABLE IF NOT EXISTS room_depth(
+    room_id TEXT,
+    min_depth INTEGER,
+    CONSTRAINT uniqueness UNIQUE (room_id)
+);
+
+CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id);
+--
+
+create TABLE IF NOT EXISTS event_destinations(
+    event_id TEXT,
+    destination TEXT,
+    delivered_ts INTEGER DEFAULT 0, -- or 0 if not delivered
+    CONSTRAINT uniqueness UNIQUE (event_id, destination) ON CONFLICT REPLACE
+);
+
+CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id);
+--
\ No newline at end of file
-- 
cgit 1.5.1


From e7858b6d7ef37849a3d2d5004743cdd21ec330a8 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 29 Oct 2014 16:59:24 +0000
Subject: Start filling out and using new events tables

---
 synapse/federation/pdu_codec.py        | 12 +++--
 synapse/handlers/_base.py              |  4 ++
 synapse/handlers/federation.py         | 90 +++++++++++++++++++---------------
 synapse/state.py                       | 11 +++--
 synapse/storage/__init__.py            | 45 ++++++++++-------
 synapse/storage/_base.py               | 33 ++++++++++---
 synapse/storage/event_federation.py    | 49 ++++++++++++------
 synapse/storage/schema/event_edges.sql |  8 ++-
 8 files changed, 159 insertions(+), 93 deletions(-)

(limited to 'synapse/storage/schema/event_edges.sql')

diff --git a/synapse/federation/pdu_codec.py b/synapse/federation/pdu_codec.py
index 2cd591410b..dccbccb85b 100644
--- a/synapse/federation/pdu_codec.py
+++ b/synapse/federation/pdu_codec.py
@@ -48,8 +48,8 @@ class PduCodec(object):
         kwargs["room_id"] = pdu.context
         kwargs["etype"] = pdu.pdu_type
         kwargs["prev_events"] = [
-            encode_event_id(i, o)
-            for i, o in pdu.prev_pdus
+            (encode_event_id(i, o), s)
+            for i, o, s in pdu.prev_pdus
         ]
 
         if hasattr(pdu, "prev_state_id") and hasattr(pdu, "prev_state_origin"):
@@ -82,7 +82,13 @@ class PduCodec(object):
         d["pdu_type"] = event.type
 
         if hasattr(event, "prev_events"):
-            d["prev_pdus"] = [decode_event_id(e) for e in event.prev_events]
+            def f(e, s):
+                i, o = decode_event_id(e, self.server_name)
+                return i, o, s
+            d["prev_pdus"] = [
+                f(e, s)
+                for e, s in event.prev_events
+            ]
 
         if hasattr(event, "prev_state"):
             d["prev_state_id"], d["prev_state_origin"] = (
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index cd6c35f194..787a01efc5 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -16,6 +16,8 @@
 from twisted.internet import defer
 from synapse.api.errors import LimitExceededError
 
+from synapse.util.async import run_on_reactor
+
 class BaseHandler(object):
 
     def __init__(self, hs):
@@ -45,6 +47,8 @@ class BaseHandler(object):
     @defer.inlineCallbacks
     def _on_new_room_event(self, event, snapshot, extra_destinations=[],
                            extra_users=[], suppress_auth=False):
+        yield run_on_reactor()
+
         snapshot.fill_out_prev_events(event)
 
         yield self.state_handler.annotate_state_groups(event)
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index b575986fc3..5f86ed03fa 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -22,6 +22,7 @@ from synapse.api.constants import Membership
 from synapse.util.logutils import log_function
 from synapse.federation.pdu_codec import PduCodec, encode_event_id
 from synapse.api.errors import SynapseError
+from synapse.util.async import run_on_reactor
 
 from twisted.internet import defer, reactor
 
@@ -81,6 +82,8 @@ class FederationHandler(BaseHandler):
             processing.
         """
 
+        yield run_on_reactor()
+
         pdu = self.pdu_codec.pdu_from_event(event)
 
         if not hasattr(pdu, "destinations") or not pdu.destinations:
@@ -102,6 +105,8 @@ class FederationHandler(BaseHandler):
             self.room_queues[event.room_id].append(pdu)
             return
 
+        logger.debug("Processing event: %s", event.event_id)
+
         if state:
             state = [self.pdu_codec.event_from_pdu(p) for p in state]
 
@@ -216,58 +221,65 @@ class FederationHandler(BaseHandler):
         assert(event.state_key == joinee)
         assert(event.room_id == room_id)
 
-        self.room_queues[room_id] = []
-
-        event.event_id = self.event_factory.create_event_id()
-        event.content = content
+        event.outlier = False
 
-        state = yield self.replication_layer.send_join(
-            target_host,
-            self.pdu_codec.pdu_from_event(event)
-        )
+        self.room_queues[room_id] = []
 
-        state = [self.pdu_codec.event_from_pdu(p) for p in state]
+        try:
+            event.event_id = self.event_factory.create_event_id()
+            event.content = content
 
-        logger.debug("do_invite_join state: %s", state)
+            state = yield self.replication_layer.send_join(
+                target_host,
+                self.pdu_codec.pdu_from_event(event)
+            )
 
-        is_new_state = yield self.state_handler.annotate_state_groups(
-            event,
-            state=state
-        )
+            state = [self.pdu_codec.event_from_pdu(p) for p in state]
 
-        try:
-            yield self.store.store_room(
-                room_id=room_id,
-                room_creator_user_id="",
-                is_public=False
-            )
-        except:
-            # FIXME
-            pass
+            logger.debug("do_invite_join state: %s", state)
 
-        for e in state:
-            # FIXME: Auth these.
             is_new_state = yield self.state_handler.annotate_state_groups(
-                e,
+                event,
+                state=state
             )
 
+            logger.debug("do_invite_join event: %s", event)
+
+            try:
+                yield self.store.store_room(
+                    room_id=room_id,
+                    room_creator_user_id="",
+                    is_public=False
+                )
+            except:
+                # FIXME
+                pass
+
+            for e in state:
+                # FIXME: Auth these.
+                e.outlier = True
+
+                yield self.state_handler.annotate_state_groups(
+                    e,
+                )
+
+                yield self.store.persist_event(
+                    e,
+                    backfilled=False,
+                    is_new_state=False
+                )
+
             yield self.store.persist_event(
-                e,
+                event,
                 backfilled=False,
-                is_new_state=False
+                is_new_state=is_new_state
             )
+        finally:
+            room_queue = self.room_queues[room_id]
+            del self.room_queues[room_id]
 
-        yield self.store.persist_event(
-            event,
-            backfilled=False,
-            is_new_state=is_new_state
-        )
-
-        room_queue = self.room_queues[room_id]
-        del self.room_queues[room_id]
-
-        for p in room_queue:
-            yield self.on_receive_pdu(p, backfilled=False)
+            for p in room_queue:
+                yield self.on_receive_pdu(p, backfilled=False)
 
         defer.returnValue(True)
 
diff --git a/synapse/state.py b/synapse/state.py
index cc6a7db96b..993c4f18d3 100644
--- a/synapse/state.py
+++ b/synapse/state.py
@@ -143,7 +143,9 @@ class StateHandler(object):
             defer.returnValue(False)
             return
 
-        new_state = yield self.resolve_state_groups(event.prev_events)
+        new_state = yield self.resolve_state_groups(
+            [e for e, _ in event.prev_events]
+        )
 
         event.old_state_events = copy.deepcopy(new_state)
 
@@ -157,12 +159,11 @@ class StateHandler(object):
 
     @defer.inlineCallbacks
     def get_current_state(self, room_id, event_type=None, state_key=""):
-        # FIXME: HACK!
-        pdus = yield self.store.get_latest_pdus_in_context(room_id)
+        events = yield self.store.get_latest_events_in_room(room_id)
 
         event_ids = [
-            encode_event_id(pdu_id, origin)
-            for pdu_id, origin, _ in pdus
+            e_id
+            for e_id, _ in events
         ]
 
         res = yield self.resolve_state_groups(event_ids)
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index f89e518690..d75c366834 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -71,6 +71,7 @@ SCHEMAS = [
     "state",
     "signatures",
     "event_edges",
+    "event_signatures",
 ]
 
 
@@ -134,7 +135,8 @@ class DataStore(RoomMemberStore, RoomStore,
                 "type",
                 "room_id",
                 "content",
-                "unrecognized_keys"
+                "unrecognized_keys",
+                "depth",
             ],
             allow_none=allow_none,
         )
@@ -263,7 +265,12 @@ class DataStore(RoomMemberStore, RoomStore,
         vals["unrecognized_keys"] = json.dumps(unrec)
 
         try:
-            self._simple_insert_txn(txn, "events", vals)
+            self._simple_insert_txn(
+                txn,
+                "events",
+                vals,
+                or_replace=(not outlier),
+            )
         except:
             logger.warn(
                 "Failed to persist, probably duplicate: %s",
@@ -307,13 +314,14 @@ class DataStore(RoomMemberStore, RoomStore,
                 }
             )
 
-        signatures = event.signatures.get(event.origin, {})
+        if hasattr(event, "signatures"):
+            signatures = event.signatures.get(event.origin, {})
 
-        for key_id, signature_base64 in signatures.items():
-            signature_bytes = decode_base64(signature_base64)
-            self._store_event_origin_signature_txn(
-                txn, event.event_id, key_id, signature_bytes,
-            )
+            for key_id, signature_base64 in signatures.items():
+                signature_bytes = decode_base64(signature_base64)
+                self._store_event_origin_signature_txn(
+                    txn, event.event_id, event.origin, key_id, signature_bytes,
+                )
 
         for prev_event_id, prev_hashes in event.prev_events:
             for alg, hash_base64 in prev_hashes.items():
@@ -323,10 +331,10 @@ class DataStore(RoomMemberStore, RoomStore,
                 )
 
         # TODO
-        (ref_alg, ref_hash_bytes) = compute_pdu_event_reference_hash(pdu)
-        self._store_event_reference_hash_txn(
-            txn, event.event_id, ref_alg, ref_hash_bytes
-        )
+        # (ref_alg, ref_hash_bytes) = compute_pdu_event_reference_hash(pdu)
+        # self._store_event_reference_hash_txn(
+        #    txn, event.event_id, ref_alg, ref_hash_bytes
+        # )
 
         self._update_min_depth_for_room_txn(txn, event.room_id, event.depth)
 
@@ -412,9 +420,7 @@ class DataStore(RoomMemberStore, RoomStore,
         """
         def _snapshot(txn):
             membership_state = self._get_room_member(txn, user_id, room_id)
-            prev_events = self._get_latest_events_in_room(
-                txn, room_id
-            )
+            prev_events = self._get_latest_events_in_room(txn, room_id)
 
             if state_type is not None and state_key is not None:
                 prev_state_pdu = self._get_current_state_pdu(
@@ -469,12 +475,12 @@ class Snapshot(object):
             return
 
         event.prev_events = [
-            (p_id, origin, hashes)
-            for p_id, origin, hashes, _ in self.prev_events
+            (event_id, hashes)
+            for event_id, hashes, _ in self.prev_events
         ]
 
         if self.prev_events:
-            event.depth = max([int(v) for _, _, _, v in self.prev_events]) + 1
+            event.depth = max([int(v) for _, _, v in self.prev_events]) + 1
         else:
             event.depth = 0
 
@@ -533,9 +539,10 @@ def prepare_database(db_conn):
             db_conn.commit()
 
     else:
-        sql_script = "BEGIN TRANSACTION;"
+        sql_script = "BEGIN TRANSACTION;\n"
         for sql_loc in SCHEMAS:
             sql_script += read_schema(sql_loc)
+            sql_script += "\n"
         sql_script += "COMMIT TRANSACTION;"
         c.executescript(sql_script)
         db_conn.commit()
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 30732caa83..464b12f032 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -19,10 +19,12 @@ from twisted.internet import defer
 from synapse.api.errors import StoreError
 from synapse.api.events.utils import prune_event
 from synapse.util.logutils import log_function
+from syutil.base64util import encode_base64
 
 import collections
 import copy
 import json
+import sys
 import time
 
 
@@ -67,6 +69,9 @@ class LoggingTransaction(object):
             return self.txn.execute(
                 sql, *args, **kwargs
             )
+        except:
+                logger.exception("[SQL FAIL] {%s}", self.name)
+                raise
         finally:
             end = time.clock() * 1000
             sql_logger.debug("[SQL time] {%s} %f", self.name, end - start)
@@ -85,14 +90,20 @@ class SQLBaseStore(object):
         """Wraps the .runInteraction() method on the underlying db_pool."""
         def inner_func(txn, *args, **kwargs):
             start = time.clock() * 1000
-            txn_id = str(SQLBaseStore._TXN_ID)
-            SQLBaseStore._TXN_ID += 1
+            txn_id = SQLBaseStore._TXN_ID
+
+            # We don't really need these to be unique, so lets stop it from
+            # growing really large.
+            self._TXN_ID = (self._TXN_ID + 1) % (sys.maxint - 1)
 
-            name = "%s-%s" % (desc, txn_id, )
+            name = "%s-%x" % (desc, txn_id, )
 
             transaction_logger.debug("[TXN START] {%s}", name)
             try:
                 return func(LoggingTransaction(txn, name), *args, **kwargs)
+            except:
+                logger.exception("[TXN FAIL] {%s}", name)
+                raise
             finally:
                 end = time.clock() * 1000
                 transaction_logger.debug(
@@ -189,7 +200,6 @@ class SQLBaseStore(object):
               statement returns no rows
         """
         return self._simple_selectupdate_one(
-            "_simple_select_one",
             table, keyvalues, retcols=retcols, allow_none=allow_none
         )
 
@@ -215,11 +225,11 @@ class SQLBaseStore(object):
             txn,
             table=table,
             keyvalues=keyvalues,
-            retcols=retcol,
+            retcol=retcol,
         )
 
         if ret:
-            return ret[retcol]
+            return ret[0]
         else:
             if allow_none:
                 return None
@@ -434,6 +444,17 @@ class SQLBaseStore(object):
         sql = "SELECT * FROM events WHERE event_id = ?"
 
         for ev in events:
+            signatures = self._get_event_origin_signatures_txn(
+                txn, ev.event_id,
+            )
+
+            ev.signatures = {
+                k: encode_base64(v) for k, v in signatures.items()
+            }
+
+            prev_events = self._get_latest_events_in_room(txn, ev.room_id)
+            ev.prev_events = [(e_id, s,) for e_id, s, _ in prev_events]
+
             if hasattr(ev, "prev_state"):
                 # Load previous state_content.
                 # TODO: Should we be pulling this out above?
diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py
index 7688fc550f..5f94c31818 100644
--- a/synapse/storage/event_federation.py
+++ b/synapse/storage/event_federation.py
@@ -24,6 +24,13 @@ logger = logging.getLogger(__name__)
 
 class EventFederationStore(SQLBaseStore):
 
+    def get_latest_events_in_room(self, room_id):
+        return self.runInteraction(
+            "get_latest_events_in_room",
+            self._get_latest_events_in_room,
+            room_id,
+        )
+
     def _get_latest_events_in_room(self, txn, room_id):
         self._simple_select_onecol_txn(
             txn,
@@ -34,12 +41,25 @@ class EventFederationStore(SQLBaseStore):
             retcol="event_id",
         )
 
+        sql = (
+            "SELECT e.event_id, e.depth FROM events as e "
+            "INNER JOIN event_forward_extremities as f "
+            "ON e.event_id = f.event_id "
+            "WHERE f.room_id = ?"
+        )
+
+        txn.execute(sql, (room_id, ))
+
         results = []
-        for pdu_id, origin, depth in txn.fetchall():
-            hashes = self._get_prev_event_hashes_txn(txn, pdu_id, origin)
-            sha256_bytes = hashes["sha256"]
-            prev_hashes = {"sha256": encode_base64(sha256_bytes)}
-            results.append((pdu_id, origin, prev_hashes, depth))
+        for event_id, depth in txn.fetchall():
+            hashes = self._get_prev_event_hashes_txn(txn, event_id)
+            prev_hashes = {
+                k: encode_base64(v) for k, v in hashes.items()
+                if k == "sha256"
+            }
+            results.append((event_id, prev_hashes, depth))
+
+        return results
 
     def _get_min_depth_interaction(self, txn, room_id):
         min_depth = self._simple_select_one_onecol_txn(
@@ -70,21 +90,21 @@ class EventFederationStore(SQLBaseStore):
 
     def _handle_prev_events(self, txn, outlier, event_id, prev_events,
                             room_id):
-        for e_id in prev_events:
+        for e_id, _ in prev_events:
             # TODO (erikj): This could be done as a bulk insert
             self._simple_insert_txn(
                 txn,
                 table="event_edges",
                 values={
                     "event_id": event_id,
-                    "prev_event": e_id,
+                    "prev_event_id": e_id,
                     "room_id": room_id,
                 }
             )
 
         # Update the extremities table if this is not an outlier.
         if not outlier:
-            for e_id in prev_events:
+            for e_id, _ in prev_events:
                 # TODO (erikj): This could be done as a bulk insert
                 self._simple_delete_txn(
                     txn,
@@ -116,7 +136,7 @@ class EventFederationStore(SQLBaseStore):
 
             # Insert all the prev_pdus as a backwards thing, they'll get
             # deleted in a second if they're incorrect anyway.
-            for e_id in prev_events:
+            for e_id, _ in prev_events:
                 # TODO (erikj): This could be done as a bulk insert
                 self._simple_insert_txn(
                     txn,
@@ -130,14 +150,11 @@ class EventFederationStore(SQLBaseStore):
             # Also delete from the backwards extremities table all ones that
             # reference pdus that we have already seen
             query = (
-                "DELETE FROM %(event_back)s as b WHERE EXISTS ("
-                "SELECT 1 FROM %(events)s AS events "
+                "DELETE FROM event_backward_extremities WHERE EXISTS ("
+                "SELECT 1 FROM events "
                 "WHERE "
-                "b.event_id = events.event_id "
+                "event_backward_extremities.event_id = events.event_id "
                 "AND not events.outlier "
                 ")"
-            ) % {
-                "event_back": "event_backward_extremities",
-                "events": "events",
-            }
+            )
             txn.execute(query)
\ No newline at end of file
diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql
index 6a28314ece..e5f768c705 100644
--- a/synapse/storage/schema/event_edges.sql
+++ b/synapse/storage/schema/event_edges.sql
@@ -7,7 +7,7 @@ CREATE TABLE IF NOT EXISTS event_forward_extremities(
 
 CREATE INDEX IF NOT EXISTS ev_extrem_room ON event_forward_extremities(room_id);
 CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id);
---
+
 
 CREATE TABLE IF NOT EXISTS event_backward_extremities(
     event_id TEXT,
@@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS event_backward_extremities(
 
 CREATE INDEX IF NOT EXISTS ev_b_extrem_room ON event_backward_extremities(room_id);
 CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id);
---
+
 
 CREATE TABLE IF NOT EXISTS event_edges(
     event_id TEXT,
@@ -28,7 +28,6 @@ CREATE TABLE IF NOT EXISTS event_edges(
 
 CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id);
 CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id);
---
 
 
 CREATE TABLE IF NOT EXISTS room_depth(
@@ -38,7 +37,7 @@ CREATE TABLE IF NOT EXISTS room_depth(
 );
 
 CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id);
---
+
 
 create TABLE IF NOT EXISTS event_destinations(
     event_id TEXT,
@@ -48,4 +47,3 @@ create TABLE IF NOT EXISTS event_destinations(
 );
 
 CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id);
---
\ No newline at end of file
-- 
cgit 1.5.1


From 4317c8e5835f0c15bf882f737d3e3c2a5b85f73f Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 6 Nov 2014 15:10:55 +0000
Subject: Implement new replace_state and changed prev_state

`prev_state` is now a list of previous state ids, similiar to
prev_events. `replace_state` now points to what we think was replaced.
---
 synapse/api/events/__init__.py         |  1 +
 synapse/handlers/directory.py          |  5 +-
 synapse/handlers/federation.py         |  4 +-
 synapse/handlers/message.py            | 11 ++--
 synapse/handlers/profile.py            |  6 +--
 synapse/handlers/room.py               | 16 ++----
 synapse/rest/room.py                   |  2 +-
 synapse/state.py                       | 39 ++------------
 synapse/storage/__init__.py            | 92 +++++++++++++++++++++++++---------
 synapse/storage/_base.py               | 66 +++++++++++++++++-------
 synapse/storage/event_federation.py    | 64 ++++++++++++++++++++---
 synapse/storage/schema/event_edges.sql | 40 ++++++++++-----
 synapse/util/jsonobject.py             |  2 +-
 13 files changed, 220 insertions(+), 128 deletions(-)

(limited to 'synapse/storage/schema/event_edges.sql')

diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py
index 168b812311..fc3f350570 100644
--- a/synapse/api/events/__init__.py
+++ b/synapse/api/events/__init__.py
@@ -60,6 +60,7 @@ class SynapseEvent(JsonEncodedObject):
         "age_ts",
         "prev_content",
         "prev_state",
+        "replaces_state",
         "redacted_because",
         "origin_server_ts",
     ]
diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py
index 6e897e915d..164363cdc5 100644
--- a/synapse/handlers/directory.py
+++ b/synapse/handlers/directory.py
@@ -147,10 +147,7 @@ class DirectoryHandler(BaseHandler):
             content={"aliases": aliases},
         )
 
-        snapshot = yield self.store.snapshot_room(
-            room_id=room_id,
-            user_id=user_id,
-        )
+        snapshot = yield self.store.snapshot_room(event)
 
         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 1464a60937..513ec9a5e3 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -313,9 +313,7 @@ class FederationHandler(BaseHandler):
             state_key=user_id,
         )
 
-        snapshot = yield self.store.snapshot_room(
-            event.room_id, event.user_id,
-        )
+        snapshot = yield self.store.snapshot_room(event)
         snapshot.fill_out_prev_events(event)
 
         yield self.state_handler.annotate_state_groups(event)
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index c6f6ab14d1..8394013df3 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -81,7 +81,7 @@ class MessageHandler(BaseHandler):
         user = self.hs.parse_userid(event.user_id)
         assert user.is_mine, "User must be our own: %s" % (user,)
 
-        snapshot = yield self.store.snapshot_room(event.room_id, event.user_id)
+        snapshot = yield self.store.snapshot_room(event)
 
         yield self._on_new_room_event(
             event, snapshot, suppress_auth=suppress_auth
@@ -141,12 +141,7 @@ class MessageHandler(BaseHandler):
             SynapseError if something went wrong.
         """
 
-        snapshot = yield self.store.snapshot_room(
-            event.room_id,
-            event.user_id,
-            state_type=event.type,
-            state_key=event.state_key,
-        )
+        snapshot = yield self.store.snapshot_room(event)
 
         yield self._on_new_room_event(event, snapshot)
 
@@ -214,7 +209,7 @@ class MessageHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def send_feedback(self, event):
-        snapshot = yield self.store.snapshot_room(event.room_id, event.user_id)
+        snapshot = yield self.store.snapshot_room(event)
 
         # store message in db
         yield self._on_new_room_event(event, snapshot)
diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py
index 4cd0a06093..e47814483a 100644
--- a/synapse/handlers/profile.py
+++ b/synapse/handlers/profile.py
@@ -17,7 +17,6 @@ from twisted.internet import defer
 
 from synapse.api.errors import SynapseError, AuthError, CodeMessageException
 from synapse.api.constants import Membership
-from synapse.api.events.room import RoomMemberEvent
 
 from ._base import BaseHandler
 
@@ -196,10 +195,7 @@ class ProfileHandler(BaseHandler):
         )
 
         for j in joins:
-            snapshot = yield self.store.snapshot_room(
-                j.room_id, j.state_key, RoomMemberEvent.TYPE,
-                j.state_key
-            )
+            snapshot = yield self.store.snapshot_room(j)
 
             content = {
                 "membership": j.content["membership"],
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index f176ad39bf..55c893eb58 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -122,10 +122,7 @@ class RoomCreationHandler(BaseHandler):
 
         @defer.inlineCallbacks
         def handle_event(event):
-            snapshot = yield self.store.snapshot_room(
-                room_id=room_id,
-                user_id=user_id,
-            )
+            snapshot = yield self.store.snapshot_room(event)
 
             logger.debug("Event: %s", event)
 
@@ -364,10 +361,8 @@ class RoomMemberHandler(BaseHandler):
         """
         target_user_id = event.state_key
 
-        snapshot = yield self.store.snapshot_room(
-            event.room_id, event.user_id,
-            RoomMemberEvent.TYPE, target_user_id
-        )
+        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
@@ -442,10 +437,7 @@ class RoomMemberHandler(BaseHandler):
             content=content,
         )
 
-        snapshot = yield self.store.snapshot_room(
-            room_id, joinee.to_string(), RoomMemberEvent.TYPE,
-            joinee.to_string()
-        )
+        snapshot = yield self.store.snapshot_room(new_event)
 
         yield self._do_join(new_event, snapshot, room_host=host, do_auth=True)
 
diff --git a/synapse/rest/room.py b/synapse/rest/room.py
index ec0ce78fda..997895dab0 100644
--- a/synapse/rest/room.py
+++ b/synapse/rest/room.py
@@ -138,7 +138,7 @@ class RoomStateEventRestServlet(RestServlet):
             raise SynapseError(
                 404, "Event not found.", errcode=Codes.NOT_FOUND
             )
-        defer.returnValue((200, data[0].get_dict()["content"]))
+        defer.returnValue((200, data.get_dict()["content"]))
 
     @defer.inlineCallbacks
     def on_PUT(self, request, room_id, event_type, state_key):
diff --git a/synapse/state.py b/synapse/state.py
index 32744e047c..97a8160a33 100644
--- a/synapse/state.py
+++ b/synapse/state.py
@@ -45,40 +45,6 @@ class StateHandler(object):
         self.server_name = hs.hostname
         self.hs = hs
 
-    @defer.inlineCallbacks
-    @log_function
-    def handle_new_event(self, event, snapshot):
-        """ Given an event this works out if a) we have sufficient power level
-        to update the state and b) works out what the prev_state should be.
-
-        Returns:
-            Deferred: Resolved with a boolean indicating if we successfully
-            updated the state.
-
-        Raised:
-            AuthError
-        """
-        # This needs to be done in a transaction.
-
-        if not hasattr(event, "state_key"):
-            return
-
-        # Now I need to fill out the prev state and work out if it has auth
-        # (w.r.t. to power levels)
-
-        snapshot.fill_out_prev_events(event)
-        yield self.annotate_state_groups(event)
-
-        if event.old_state_events:
-            current_state = event.old_state_events.get(
-                (event.type, event.state_key)
-            )
-
-            if current_state:
-                event.prev_state = current_state.event_id
-
-        defer.returnValue(True)
-
     @defer.inlineCallbacks
     @log_function
     def annotate_state_groups(self, event, old_state=None):
@@ -111,7 +77,10 @@ class StateHandler(object):
         event.old_state_events = copy.deepcopy(new_state)
 
         if hasattr(event, "state_key"):
-            new_state[(event.type, event.state_key)] = event
+            key = (event.type, event.state_key)
+            if key in new_state:
+                event.replaces_state = new_state[key].event_id
+            new_state[key] = event
 
         event.state_group = None
         event.state_events = new_state
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 6b8fed4502..2d62fc2ed0 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -242,8 +242,8 @@ class DataStore(RoomMemberStore, RoomStore,
                 "state_key": event.state_key,
             }
 
-            if hasattr(event, "prev_state"):
-                vals["prev_state"] = event.prev_state
+            if hasattr(event, "replaces_state"):
+                vals["prev_state"] = event.replaces_state
 
             self._simple_insert_txn(txn, "state_events", vals)
 
@@ -258,6 +258,40 @@ class DataStore(RoomMemberStore, RoomStore,
                 }
             )
 
+            for e_id, h in event.prev_state:
+                self._simple_insert_txn(
+                    txn,
+                    table="event_edges",
+                    values={
+                        "event_id": event.event_id,
+                        "prev_event_id": e_id,
+                        "room_id": event.room_id,
+                        "is_state": 1,
+                    },
+                    or_ignore=True,
+                )
+
+            if not backfilled:
+                self._simple_insert_txn(
+                    txn,
+                    table="state_forward_extremities",
+                    values={
+                        "event_id": event.event_id,
+                        "room_id": event.room_id,
+                        "type": event.type,
+                        "state_key": event.state_key,
+                    }
+                )
+
+                for prev_state_id, _ in event.prev_state:
+                    self._simple_delete_txn(
+                        txn,
+                        table="state_forward_extremities",
+                        keyvalues={
+                            "event_id": prev_state_id,
+                        }
+                    )
+
         for hash_alg, hash_base64 in event.hashes.items():
             hash_bytes = decode_base64(hash_base64)
             self._store_event_content_hash_txn(
@@ -357,7 +391,7 @@ class DataStore(RoomMemberStore, RoomStore,
             ],
         )
 
-    def snapshot_room(self, room_id, user_id, state_type=None, state_key=None):
+    def snapshot_room(self, event):
         """Snapshot the room for an update by a user
         Args:
             room_id (synapse.types.RoomId): The room to snapshot.
@@ -368,16 +402,29 @@ class DataStore(RoomMemberStore, RoomStore,
             synapse.storage.Snapshot: A snapshot of the state of the room.
         """
         def _snapshot(txn):
-            membership_state = self._get_room_member(txn, user_id, room_id)
-            prev_events = self._get_latest_events_in_room(txn, room_id)
+            prev_events = self._get_latest_events_in_room(
+                txn,
+                event.room_id
+            )
+
+            prev_state = None
+            state_key = None
+            if hasattr(event, "state_key"):
+                state_key = event.state_key
+                prev_state = self._get_latest_state_in_room(
+                    txn,
+                    event.room_id,
+                    type=event.type,
+                    state_key=state_key,
+                )
 
             return Snapshot(
                 store=self,
-                room_id=room_id,
-                user_id=user_id,
+                room_id=event.room_id,
+                user_id=event.user_id,
                 prev_events=prev_events,
-                membership_state=membership_state,
-                state_type=state_type,
+                prev_state=prev_state,
+                state_type=event.type,
                 state_key=state_key,
             )
 
@@ -400,30 +447,29 @@ class Snapshot(object):
     """
 
     def __init__(self, store, room_id, user_id, prev_events,
-                 membership_state, state_type=None, state_key=None,
-                 prev_state_pdu=None):
+                 prev_state, state_type=None, state_key=None):
         self.store = store
         self.room_id = room_id
         self.user_id = user_id
         self.prev_events = prev_events
-        self.membership_state = membership_state
+        self.prev_state = prev_state
         self.state_type = state_type
         self.state_key = state_key
-        self.prev_state_pdu = prev_state_pdu
 
     def fill_out_prev_events(self, event):
-        if hasattr(event, "prev_events"):
-            return
+        if not hasattr(event, "prev_events"):
+            event.prev_events = [
+                (event_id, hashes)
+                for event_id, hashes, _ in self.prev_events
+            ]
 
-        event.prev_events = [
-            (event_id, hashes)
-            for event_id, hashes, _ in self.prev_events
-        ]
+            if self.prev_events:
+                event.depth = max([int(v) for _, _, v in self.prev_events]) + 1
+            else:
+                event.depth = 0
 
-        if self.prev_events:
-            event.depth = max([int(v) for _, _, v in self.prev_events]) + 1
-        else:
-            event.depth = 0
+        if not hasattr(event, "prev_state") and self.prev_state is not None:
+            event.prev_state = self.prev_state
 
 
 def schema_path(schema):
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 7d445b4633..7821fc4726 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -245,7 +245,6 @@ class SQLBaseStore(object):
 
         return [r[0] for r in txn.fetchall()]
 
-
     def _simple_select_onecol(self, table, keyvalues, retcol):
         """Executes a SELECT query on the named table, which returns a list
         comprising of the values of the named column from the selected rows.
@@ -273,17 +272,30 @@ class SQLBaseStore(object):
             keyvalues : dict of column names and values to select the rows with
             retcols : list of strings giving the names of the columns to return
         """
+        return self.runInteraction(
+            "_simple_select_list",
+            self._simple_select_list_txn,
+            table, keyvalues, retcols
+        )
+
+    def _simple_select_list_txn(self, txn, table, keyvalues, retcols):
+        """Executes a SELECT query on the named table, which may return zero or
+        more rows, returning the result as a list of dicts.
+
+        Args:
+            txn : Transaction object
+            table : string giving the table name
+            keyvalues : dict of column names and values to select the rows with
+            retcols : list of strings giving the names of the columns to return
+        """
         sql = "SELECT %s FROM %s WHERE %s" % (
             ", ".join(retcols),
             table,
-            " AND ".join("%s = ?" % (k) for k in keyvalues)
+            " AND ".join("%s = ?" % (k, ) for k in keyvalues)
         )
 
-        def func(txn):
-            txn.execute(sql, keyvalues.values())
-            return self.cursor_to_dict(txn)
-
-        return self.runInteraction("_simple_select_list", func)
+        txn.execute(sql, keyvalues.values())
+        return self.cursor_to_dict(txn)
 
     def _simple_update_one(self, table, keyvalues, updatevalues,
                            retcols=None):
@@ -417,6 +429,10 @@ class SQLBaseStore(object):
         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"])
@@ -450,16 +466,32 @@ class SQLBaseStore(object):
                 k: encode_base64(v) for k, v in signatures.items()
             }
 
-            ev.prev_events = self._get_prev_events(txn, ev.event_id)
-
-            if hasattr(ev, "prev_state"):
-                # Load previous state_content.
-                # TODO: Should we be pulling this out above?
-                cursor = txn.execute(select_event_sql, (ev.prev_state,))
-                prevs = self.cursor_to_dict(cursor)
-                if prevs:
-                    prev = self._parse_event_from_row(prevs[0])
-                    ev.prev_content = prev.content
+            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
+            ]
+
+            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
 
             if not hasattr(ev, "redacted"):
                 logger.debug("Doesn't have redacted key: %s", ev)
diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py
index f427aba879..180a764134 100644
--- a/synapse/storage/event_federation.py
+++ b/synapse/storage/event_federation.py
@@ -69,19 +69,21 @@ class EventFederationStore(SQLBaseStore):
 
         return results
 
-    def _get_prev_events(self, txn, event_id):
-        prev_ids = self._simple_select_onecol_txn(
+    def _get_latest_state_in_room(self, txn, room_id, type, state_key):
+        event_ids = self._simple_select_onecol_txn(
             txn,
-            table="event_edges",
+            table="state_forward_extremities",
             keyvalues={
-                "event_id": event_id,
+                "room_id": room_id,
+                "type": type,
+                "state_key": state_key,
             },
-            retcol="prev_event_id",
+            retcol="event_id",
         )
 
         results = []
-        for prev_event_id in prev_ids:
-            hashes = self._get_event_reference_hashes_txn(txn, prev_event_id)
+        for event_id in event_ids:
+            hashes = self._get_event_reference_hashes_txn(txn, event_id)
             prev_hashes = {
                 k: encode_base64(v) for k, v in hashes.items()
                 if k == "sha256"
@@ -90,6 +92,53 @@ class EventFederationStore(SQLBaseStore):
 
         return results
 
+    def _get_prev_events(self, txn, event_id):
+        results = self._get_prev_events_and_state(
+            txn,
+            event_id,
+            is_state=0,
+        )
+
+        return [(e_id, h, ) for e_id, h, _ in results]
+
+    def _get_prev_state(self, txn, event_id):
+        results = self._get_prev_events_and_state(
+            txn,
+            event_id,
+            is_state=1,
+        )
+
+        return [(e_id, h, ) for e_id, h, _ in results]
+
+    def _get_prev_events_and_state(self, txn, event_id, is_state=None):
+        keyvalues = {
+            "event_id": event_id,
+        }
+
+        if is_state is not None:
+            keyvalues["is_state"] = is_state
+
+        res = self._simple_select_list_txn(
+            txn,
+            table="event_edges",
+            keyvalues=keyvalues,
+            retcols=["prev_event_id", "is_state"],
+        )
+
+        results = []
+        for d in res:
+            hashes = self._get_event_reference_hashes_txn(
+                txn,
+                d["prev_event_id"]
+            )
+            prev_hashes = {
+                k: encode_base64(v) for k, v in hashes.items()
+                if k == "sha256"
+            }
+            results.append((d["prev_event_id"], prev_hashes, d["is_state"]))
+
+        return results
+
     def get_min_depth(self, room_id):
         return self.runInteraction(
             "get_min_depth",
@@ -135,6 +184,7 @@ class EventFederationStore(SQLBaseStore):
                     "event_id": event_id,
                     "prev_event_id": e_id,
                     "room_id": room_id,
+                    "is_state": 0,
                 },
                 or_ignore=True,
             )
diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql
index e5f768c705..51695826a8 100644
--- a/synapse/storage/schema/event_edges.sql
+++ b/synapse/storage/schema/event_edges.sql
@@ -1,7 +1,7 @@
 
 CREATE TABLE IF NOT EXISTS event_forward_extremities(
-    event_id TEXT,
-    room_id TEXT,
+    event_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
     CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE
 );
 
@@ -10,8 +10,8 @@ CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id);
 
 
 CREATE TABLE IF NOT EXISTS event_backward_extremities(
-    event_id TEXT,
-    room_id TEXT,
+    event_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
     CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE
 );
 
@@ -20,10 +20,11 @@ CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id
 
 
 CREATE TABLE IF NOT EXISTS event_edges(
-    event_id TEXT,
-    prev_event_id TEXT,
-    room_id TEXT,
-    CONSTRAINT uniqueness UNIQUE (event_id, prev_event_id, room_id)
+    event_id TEXT NOT NULL,
+    prev_event_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    is_state INTEGER NOT NULL,
+    CONSTRAINT uniqueness UNIQUE (event_id, prev_event_id, room_id, is_state)
 );
 
 CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id);
@@ -31,8 +32,8 @@ CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id);
 
 
 CREATE TABLE IF NOT EXISTS room_depth(
-    room_id TEXT,
-    min_depth INTEGER,
+    room_id TEXT NOT NULL,
+    min_depth INTEGER NOT NULL,
     CONSTRAINT uniqueness UNIQUE (room_id)
 );
 
@@ -40,10 +41,25 @@ CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id);
 
 
 create TABLE IF NOT EXISTS event_destinations(
-    event_id TEXT,
-    destination TEXT,
+    event_id TEXT NOT NULL,
+    destination TEXT NOT NULL,
     delivered_ts INTEGER DEFAULT 0, -- or 0 if not delivered
     CONSTRAINT uniqueness UNIQUE (event_id, destination) ON CONFLICT REPLACE
 );
 
 CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id);
+
+
+CREATE TABLE IF NOT EXISTS state_forward_extremities(
+    event_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    type TEXT NOT NULL,
+    state_key TEXT NOT NULL,
+    CONSTRAINT uniqueness UNIQUE (event_id, room_id) ON CONFLICT REPLACE
+);
+
+CREATE INDEX IF NOT EXISTS st_extrem_keys ON state_forward_extremities(
+    room_id, type, state_key
+);
+CREATE INDEX IF NOT EXISTS st_extrem_id ON state_forward_extremities(event_id);
+
diff --git a/synapse/util/jsonobject.py b/synapse/util/jsonobject.py
index c91eb897a8..e79b68f661 100644
--- a/synapse/util/jsonobject.py
+++ b/synapse/util/jsonobject.py
@@ -80,7 +80,7 @@ class JsonEncodedObject(object):
 
     def get_full_dict(self):
         d = {
-            k: v for (k, v) in self.__dict__.items()
+            k: _encode(v) for (k, v) in self.__dict__.items()
             if k in self.valid_keys or k in self.internal_keys
         }
         d.update(self.unrecognized_keys)
-- 
cgit 1.5.1


From bf6b72eb558cca94e209a541188079750bfefea0 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 6 Nov 2014 18:42:18 +0000
Subject: Start implementing auth chains

---
 synapse/api/auth.py                    |  3 +-
 synapse/api/events/__init__.py         |  2 +-
 synapse/handlers/_base.py              | 59 ++++++++++++++++++++++++++++++++--
 synapse/storage/__init__.py            | 12 ++++++-
 synapse/storage/_base.py               |  2 ++
 synapse/storage/event_federation.py    | 21 ++++++++++++
 synapse/storage/schema/event_edges.sql | 10 ++++++
 synapse/storage/signatures.py          | 12 +++++++
 8 files changed, 115 insertions(+), 6 deletions(-)

(limited to 'synapse/storage/schema/event_edges.sql')

diff --git a/synapse/api/auth.py b/synapse/api/auth.py
index bb25c4ec55..e1302553d7 100644
--- a/synapse/api/auth.py
+++ b/synapse/api/auth.py
@@ -21,8 +21,7 @@ from synapse.api.constants import Membership, JoinRules
 from synapse.api.errors import AuthError, StoreError, Codes, SynapseError
 from synapse.api.events.room import (
     RoomMemberEvent, RoomPowerLevelsEvent, RoomRedactionEvent,
-    RoomJoinRulesEvent, InviteJoinEvent,
-    RoomCreateEvent,
+    RoomJoinRulesEvent, RoomCreateEvent,
 )
 from synapse.util.logutils import log_function
 
diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py
index 84d3a98365..513a48f568 100644
--- a/synapse/api/events/__init__.py
+++ b/synapse/api/events/__init__.py
@@ -61,7 +61,7 @@ class SynapseEvent(JsonEncodedObject):
         "replaces_state",
         "redacted_because",
         "origin_server_ts",
-        "auth_chains",
+        "auth_events",
     ]
 
     internal_keys = [
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index 509f7b550c..2613fa7fce 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -14,11 +14,15 @@
 # limitations under the License.
 
 from twisted.internet import defer
-from synapse.api.errors import LimitExceededError
 
+from synapse.api.errors import LimitExceededError
 from synapse.util.async import run_on_reactor
-
 from synapse.crypto.event_signing import add_hashes_and_signatures
+from synapse.api.events.room import (
+    RoomCreateEvent, RoomMemberEvent, RoomPowerLevelsEvent, RoomJoinRulesEvent,
+)
+from synapse.api.constants import Membership, JoinRules
+from syutil.base64util import encode_base64
 
 import logging
 
@@ -55,6 +59,53 @@ class BaseHandler(object):
                 retry_after_ms=int(1000*(time_allowed - time_now)),
             )
 
+    @defer.inlineCallbacks
+    def _add_auth(self, event):
+        if event.type == RoomCreateEvent.TYPE:
+            event.auth_events = []
+            return
+
+        auth_events = []
+
+        key = (RoomPowerLevelsEvent.TYPE, "", )
+        power_level_event = event.old_state_events.get(key)
+
+        if power_level_event:
+            auth_events.append(power_level_event.event_id)
+
+        key = (RoomJoinRulesEvent.TYPE, "", )
+        join_rule_event = event.old_state_events.get(key)
+
+        key = (RoomMemberEvent.TYPE, event.user_id, )
+        member_event = event.old_state_events.get(key)
+
+        if join_rule_event:
+            join_rule = join_rule_event.content.get("join_rule")
+            is_public = join_rule == JoinRules.PUBLIC if join_rule else False
+
+            if event.type == RoomMemberEvent.TYPE:
+                if event.content["membership"] == Membership.JOIN:
+                    if is_public:
+                        auth_events.append(join_rule_event.event_id)
+                elif member_event:
+                    auth_events.append(member_event.event_id)
+
+        if member_event:
+            if member_event.content["membership"] == Membership.JOIN:
+                auth_events.append(member_event.event_id)
+
+        hashes = yield self.store.get_event_reference_hashes(
+            auth_events
+        )
+        hashes = [
+            {
+                k: encode_base64(v) for k, v in h.items()
+                if k == "sha256"
+            }
+            for h in hashes
+        ]
+        event.auth_events = zip(auth_events, hashes)
+
     @defer.inlineCallbacks
     def _on_new_room_event(self, event, snapshot, extra_destinations=[],
                            extra_users=[], suppress_auth=False):
@@ -64,6 +115,8 @@ class BaseHandler(object):
 
         yield self.state_handler.annotate_state_groups(event)
 
+        yield self._add_auth(event)
+
         logger.debug("Signing event...")
 
         add_hashes_and_signatures(
@@ -76,6 +129,8 @@ class BaseHandler(object):
             logger.debug("Authing...")
             self.auth.check(event, raises=True)
             logger.debug("Authed")
+        else:
+            logger.debug("Suppressed auth.")
 
         yield self.store.persist_event(event)
 
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 2a1970914f..48ad4d864f 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -19,7 +19,6 @@ from synapse.api.events.room import (
     RoomMemberEvent, RoomTopicEvent, FeedbackEvent,
     RoomNameEvent,
     RoomJoinRulesEvent,
-    RoomPowerLevelsEvent,
     RoomRedactionEvent,
 )
 
@@ -302,6 +301,17 @@ class DataStore(RoomMemberStore, RoomStore,
                     txn, event.event_id, prev_event_id, alg, hash_bytes
                 )
 
+        for auth_id, _ in event.auth_events:
+            self._simple_insert_txn(
+                txn,
+                table="event_auth",
+                values={
+                    "event_id": event.event_id,
+                    "room_id": event.room_id,
+                    "auth_id": auth_id,
+                },
+            )
+
         (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event)
         self._store_event_reference_hash_txn(
             txn, event.event_id, ref_alg, ref_hash_bytes
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 7821fc4726..9aa404695d 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -474,6 +474,8 @@ class SQLBaseStore(object):
                 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)
diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py
index 180a764134..86c68ebf87 100644
--- a/synapse/storage/event_federation.py
+++ b/synapse/storage/event_federation.py
@@ -139,6 +139,27 @@ class EventFederationStore(SQLBaseStore):
 
         return results
 
+    def _get_auth_events(self, txn, event_id):
+        auth_ids = self._simple_select_onecol_txn(
+            txn,
+            table="event_auth",
+            keyvalues={
+                "event_id": event_id,
+            },
+            retcol="auth_id",
+        )
+
+        results = []
+        for auth_id in auth_ids:
+            hashes = self._get_event_reference_hashes_txn(txn, auth_id)
+            prev_hashes = {
+                k: encode_base64(v) for k, v in hashes.items()
+                if k == "sha256"
+            }
+            results.append((auth_id, prev_hashes))
+
+        return results
+
     def get_min_depth(self, room_id):
         return self.runInteraction(
             "get_min_depth",
diff --git a/synapse/storage/schema/event_edges.sql b/synapse/storage/schema/event_edges.sql
index 51695826a8..be1c72a775 100644
--- a/synapse/storage/schema/event_edges.sql
+++ b/synapse/storage/schema/event_edges.sql
@@ -63,3 +63,13 @@ CREATE INDEX IF NOT EXISTS st_extrem_keys ON state_forward_extremities(
 );
 CREATE INDEX IF NOT EXISTS st_extrem_id ON state_forward_extremities(event_id);
 
+
+CREATE TABLE IF NOT EXISTS event_auth(
+    event_id TEXT NOT NULL,
+    auth_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    CONSTRAINT uniqueness UNIQUE (event_id, auth_id, room_id)
+);
+
+CREATE INDEX IF NOT EXISTS evauth_edges_id ON event_auth(event_id);
+CREATE INDEX IF NOT EXISTS evauth_edges_auth_id ON event_auth(auth_id);
\ No newline at end of file
diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py
index b4b3d5d7ea..84a49088a2 100644
--- a/synapse/storage/signatures.py
+++ b/synapse/storage/signatures.py
@@ -55,6 +55,18 @@ class SignatureStore(SQLBaseStore):
             or_ignore=True,
         )
 
+    def get_event_reference_hashes(self, event_ids):
+        def f(txn):
+            return [
+                self._get_event_reference_hashes_txn(txn, ev)
+                for ev in event_ids
+            ]
+
+        return self.runInteraction(
+            "get_event_reference_hashes",
+            f
+        )
+
     def _get_event_reference_hashes_txn(self, txn, event_id):
         """Get all the hashes for a given PDU.
         Args:
-- 
cgit 1.5.1

