2 files changed, 54 insertions, 4 deletions
diff --git a/synapse/storage/event_actions.py b/synapse/storage/event_actions.py
index 593b1714c7..40ac8e2d27 100644
--- a/synapse/storage/event_actions.py
+++ b/synapse/storage/event_actions.py
@@ -24,18 +24,67 @@ logger = logging.getLogger(__name__)
 
 class EventActionsStore(SQLBaseStore):
     @defer.inlineCallbacks
-    def set_actions_for_event(self, event_id, user_id, profile_tag, actions):
+    def set_actions_for_event(self, event, user_id, profile_tag, actions):
         actionsJson = json.dumps(actions)
 
         ret = yield self.runInteraction(
             "_set_actions_for_event",
             self._simple_upsert_txn,
             EventActionsTable.table_name,
-            {'event_id': event_id, 'user_id': user_id, 'profile_tag': profile_tag},
+            {
+                'room_id': event['room_id'],
+                'event_id': event['event_id'],
+                'user_id': user_id,
+                'profile_tag': profile_tag
+            },
             {'actions': actionsJson}
         )
         defer.returnValue(ret)
 
+    @defer.inlineCallbacks
+    def get_unread_event_actions_by_room(self, room_id, last_read_event_id):
+        #events = yield self._get_events(
+        #    [last_read_event_id],
+        #    check_redacted=False
+        #)
+
+        def _get_unread_event_actions_by_room(txn):
+            sql = (
+                "SELECT stream_ordering, topological_ordering"
+                " FROM events"
+                " WHERE room_id = ? AND event_id = ?"
+            )
+            txn.execute(
+                sql, (room_id, last_read_event_id)
+            )
+            results = txn.fetchall()
+            if len(results) == 0:
+                return []
+
+            stream_ordering = results[0][0]
+            topological_ordering = results[0][1]
+
+            sql = (
+                "SELECT ea.actions"
+                " FROM event_actions ea, events e"
+                " WHERE ea.room_id = e.room_id"
+                " AND ea.event_id = e.event_id"
+                " AND ea.room_id = ?"
+                " AND ("
+                "       e.topological_ordering > ?"
+                "       OR (e.topological_ordering == ? AND e.stream_ordering > ?)"
+                ")"
+            )
+            txn.execute(sql,
+                (room_id, topological_ordering, topological_ordering, stream_ordering)
+            )
+            return txn.fetchall()
+
+        ret = yield self.runInteraction(
+            "get_unread_event_actions_by_room",
+            _get_unread_event_actions_by_room
+        )
+        defer.returnValue(ret)
 
 class EventActionsTable(object):
     table_name = "event_actions"
diff --git a/synapse/storage/schema/delta/27/event_actions.sql b/synapse/storage/schema/delta/27/event_actions.sql
index 1246823a00..bbdaee990e 100644
--- a/synapse/storage/schema/delta/27/event_actions.sql
+++ b/synapse/storage/schema/delta/27/event_actions.sql
@@ -14,12 +14,13 @@
  */
 
 CREATE TABLE IF NOT EXISTS event_actions(
+    room_id TEXT NOT NULL,
     event_id TEXT NOT NULL,
     user_id TEXT NOT NULL,
     profile_tag VARCHAR(32),
     actions TEXT NOT NULL,
-    CONSTRAINT event_id_user_id_profile_tag_uniqueness UNIQUE (event_id, user_id, profile_tag)
+    CONSTRAINT event_id_user_id_profile_tag_uniqueness UNIQUE (room_id, event_id, user_id, profile_tag)
 );
 
 
-CREATE INDEX event_actions_event_id_user_id_profile_tag on event_actions(event_id, user_id, profile_tag);
+CREATE INDEX event_actions_room_id_event_id_user_id_profile_tag on event_actions(room_id, event_id, user_id, profile_tag);