Very first cut of calculating actions for events as they come in. Doesn't store them yet. Not very efficient.

4 files changed, 81 insertions, 1 deletions

diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index 5fd20285d2..a8e8c4f5af 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -19,9 +19,12 @@ from synapse.api.errors import LimitExceededError, SynapseError, AuthError
 from synapse.crypto.event_signing import add_hashes_and_signatures
 from synapse.api.constants import Membership, EventTypes
 from synapse.types import UserID, RoomAlias
+from synapse.push.action_generator import ActionGenerator
 
 from synapse.util.logcontext import PreserveLoggingContext
 
+from synapse.events.utils import serialize_event
+
 import logging
 
 
@@ -264,6 +267,11 @@ class BaseHandler(object):
             event, context=context
         )
 
+        action_generator = ActionGenerator(self.store)
+        yield action_generator.handle_event(serialize_event(
+            event, self.clock.time_msec()
+        ))
+
         destinations = set(extra_destinations)
         for k, s in context.current_state.items():
             try:
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 2855f2d7c3..18289eb529 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -32,10 +32,12 @@ from synapse.crypto.event_signing import (
 )
 from synapse.types import UserID
 
-from synapse.events.utils import prune_event
+from synapse.events.utils import prune_event, serialize_event
 
 from synapse.util.retryutils import NotRetryingDestination
 
+from synapse.push.action_generator import ActionGenerator
+
 from twisted.internet import defer
 
 import itertools
@@ -1113,6 +1115,11 @@ class FederationHandler(BaseHandler):
             current_state=current_state,
         )
 
+        action_generator = ActionGenerator(self.store)
+        yield action_generator.handle_event(serialize_event(
+            event, self.clock.time_msec())
+        )
+
         defer.returnValue((context, event_stream_id, max_stream_id))
 
     @defer.inlineCallbacks
@@ -1139,6 +1146,12 @@ class FederationHandler(BaseHandler):
             is_new_state=(not outliers and not backfilled),
         )
 
+        for ev_info in event_infos:
+            action_generator = ActionGenerator(self.store)
+            yield action_generator.handle_event(serialize_event(
+                ev_info["event"], self.clock.time_msec())
+            )
+
     @defer.inlineCallbacks
     def _persist_auth_tree(self, auth_events, state, event):
         """Checks the auth chain is valid (and passes auth checks) for the
diff --git a/synapse/push/action_generator.py b/synapse/push/action_generator.py
new file mode 100644
index 0000000000..508eeaed95
--- /dev/null
+++ b/synapse/push/action_generator.py
@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015 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 twisted.internet import defer
+
+import push_rule_evaluator
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class ActionGenerator:
+    def __init__(self, store):
+        self.store = store
+        # really we want to get all user ids and all profile tags too,
+        # since we want the actions for each profile tag for every user and
+        # also actions for a client with no profile tag for each user.
+        # Currently the event stream doesn't support profile tags on an
+        # event stream, so we just run the rules for a client with no profile
+        # tag (ie. we just need all the users).
+
+    @defer.inlineCallbacks
+    def handle_event(self, event):
+        users = yield self.store.get_users_in_room(event['room_id'])
+        logger.error("users in room: %r", users)
+
+        for uid in users:
+            evaluator = yield push_rule_evaluator.\
+                evaluator_for_user_name_and_profile_tag(
+                    uid, None, event['room_id'], self.store
+            )
+            actions = yield evaluator.actions_for_event(event)
+            logger.info("actions for user %s: %s", uid, actions)
diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py
index 2e5eddd259..f230faa25e 100644
--- a/synapse/storage/registration.py
+++ b/synapse/storage/registration.py
@@ -292,6 +292,18 @@ class RegistrationStore(SQLBaseStore):
         defer.returnValue(None)
 
     @defer.inlineCallbacks
+    def get_all_user_ids(self):
+        """Returns all user ids registered on this homeserver"""
+        return self.runInteraction(
+            "get_all_user_ids",
+            self._get_all_user_ids_txn
+        )
+
+    def _get_all_user_ids_txn(self, txn):
+        txn.execute("SELECT name from users")
+        return [r[0] for r in txn.fetchall()]
+
+    @defer.inlineCallbacks
     def count_all_users(self):
         """Counts all users registered on the homeserver."""
         def _count_users(txn):