Reference Matrix Home Server

1 files changed, 697 insertions, 0 deletions

diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
new file mode 100644
index 0000000000..38db4b1d67
--- /dev/null
+++ b/synapse/handlers/presence.py
@@ -0,0 +1,697 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014 matrix.org
+#
+# 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
+
+from synapse.api.errors import SynapseError, AuthError
+from synapse.api.constants import PresenceState
+from synapse.api.streams import StreamData
+
+from ._base import BaseHandler
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+# TODO(paul): Maybe there's one of these I can steal from somewhere
+def partition(l, func):
+    """Partition the list by the result of func applied to each element."""
+    ret = {}
+
+    for x in l:
+        key = func(x)
+        if key not in ret:
+            ret[key] = []
+        ret[key].append(x)
+
+    return ret
+
+
+def partitionbool(l, func):
+    def boolfunc(x):
+        return bool(func(x))
+
+    ret = partition(l, boolfunc)
+    return ret.get(True, []), ret.get(False, [])
+
+
+class PresenceHandler(BaseHandler):
+
+    def __init__(self, hs):
+        super(PresenceHandler, self).__init__(hs)
+
+        self.homeserver = hs
+
+        distributor = hs.get_distributor()
+        distributor.observe("registered_user", self.registered_user)
+
+        distributor.observe(
+            "started_user_eventstream", self.started_user_eventstream
+        )
+        distributor.observe(
+            "stopped_user_eventstream", self.stopped_user_eventstream
+        )
+
+        distributor.observe("user_joined_room",
+            self.user_joined_room
+        )
+
+        distributor.declare("collect_presencelike_data")
+
+        distributor.declare("changed_presencelike_data")
+        distributor.observe(
+            "changed_presencelike_data", self.changed_presencelike_data
+        )
+
+        self.distributor = distributor
+
+        self.federation = hs.get_replication_layer()
+
+        self.federation.register_edu_handler(
+            "m.presence", self.incoming_presence
+        )
+        self.federation.register_edu_handler(
+            "m.presence_invite",
+            lambda origin, content: self.invite_presence(
+                observed_user=hs.parse_userid(content["observed_user"]),
+                observer_user=hs.parse_userid(content["observer_user"]),
+            )
+        )
+        self.federation.register_edu_handler(
+            "m.presence_accept",
+            lambda origin, content: self.accept_presence(
+                observed_user=hs.parse_userid(content["observed_user"]),
+                observer_user=hs.parse_userid(content["observer_user"]),
+            )
+        )
+        self.federation.register_edu_handler(
+            "m.presence_deny",
+            lambda origin, content: self.deny_presence(
+                observed_user=hs.parse_userid(content["observed_user"]),
+                observer_user=hs.parse_userid(content["observer_user"]),
+            )
+        )
+
+        # IN-MEMORY store, mapping local userparts to sets of local users to
+        # be informed of state changes.
+        self._local_pushmap = {}
+        # map local users to sets of remote /domain names/ who are interested
+        # in them
+        self._remote_sendmap = {}
+        # map remote users to sets of local users who're interested in them
+        self._remote_recvmap = {}
+
+        # map any user to a UserPresenceCache
+        self._user_cachemap = {}
+        self._user_cachemap_latest_serial = 0
+
+    def _get_or_make_usercache(self, user):
+        """If the cache entry doesn't exist, initialise a new one."""
+        if user not in self._user_cachemap:
+            self._user_cachemap[user] = UserPresenceCache()
+        return self._user_cachemap[user]
+
+    def _get_or_offline_usercache(self, user):
+        """If the cache entry doesn't exist, return an OFFLINE one but do not
+        store it into the cache."""
+        if user in self._user_cachemap:
+            return self._user_cachemap[user]
+        else:
+            statuscache = UserPresenceCache()
+            statuscache.update({"state": PresenceState.OFFLINE}, user)
+            return statuscache
+
+    def registered_user(self, user):
+        self.store.create_presence(user.localpart)
+
+    @defer.inlineCallbacks
+    def is_presence_visible(self, observer_user, observed_user):
+        assert(observed_user.is_mine)
+
+        if observer_user == observed_user:
+            defer.returnValue(True)
+
+        allowed_by_subscription = yield self.store.is_presence_visible(
+            observed_localpart=observed_user.localpart,
+            observer_userid=observer_user.to_string(),
+        )
+
+        if allowed_by_subscription:
+            defer.returnValue(True)
+
+        # TODO(paul): Check same channel
+
+        defer.returnValue(False)
+
+    @defer.inlineCallbacks
+    def get_state(self, target_user, auth_user):
+        if target_user.is_mine:
+            visible = yield self.is_presence_visible(observer_user=auth_user,
+                observed_user=target_user
+            )
+
+            if visible:
+                state = yield self.store.get_presence_state(
+                    target_user.localpart
+                )
+                defer.returnValue(state)
+            else:
+                raise SynapseError(404, "Presence information not visible")
+        else:
+            # TODO(paul): Have remote server send us permissions set
+            defer.returnValue(
+                    self._get_or_offline_usercache(target_user).get_state()
+            )
+
+    @defer.inlineCallbacks
+    def set_state(self, target_user, auth_user, state):
+        if not target_user.is_mine:
+            raise SynapseError(400, "User is not hosted on this Home Server")
+
+        if target_user != auth_user:
+            raise AuthError(400, "Cannot set another user's displayname")
+
+        # TODO(paul): Sanity-check 'state'
+        if "status_msg" not in state:
+            state["status_msg"] = None
+
+        for k in state.keys():
+            if k not in ("state", "status_msg"):
+                raise SynapseError(
+                    400, "Unexpected presence state key '%s'" % (k,)
+                )
+
+        logger.debug("Updating presence state of %s to %s",
+                     target_user.localpart, state["state"])
+
+        state_to_store = dict(state)
+
+        yield defer.DeferredList([
+            self.store.set_presence_state(
+                target_user.localpart, state_to_store
+            ),
+            self.distributor.fire(
+                "collect_presencelike_data", target_user, state
+            ),
+        ])
+
+        now_online = state["state"] != PresenceState.OFFLINE
+        was_polling = target_user in self._user_cachemap
+
+        if now_online and not was_polling:
+            self.start_polling_presence(target_user, state=state)
+        elif not now_online and was_polling:
+            self.stop_polling_presence(target_user)
+
+        # TODO(paul): perform a presence push as part of start/stop poll so
+        #   we don't have to do this all the time
+        self.changed_presencelike_data(target_user, state)
+
+        if not now_online:
+            del self._user_cachemap[target_user]
+
+    def changed_presencelike_data(self, user, state):
+        statuscache = self._get_or_make_usercache(user)
+
+        self._user_cachemap_latest_serial += 1
+        statuscache.update(state, serial=self._user_cachemap_latest_serial)
+
+        self.push_presence(user, statuscache=statuscache)
+
+    def started_user_eventstream(self, user):
+        # TODO(paul): Use "last online" state
+        self.set_state(user, user, {"state": PresenceState.ONLINE})
+
+    def stopped_user_eventstream(self, user):
+        # TODO(paul): Save current state as "last online" state
+        self.set_state(user, user, {"state": PresenceState.OFFLINE})
+
+    @defer.inlineCallbacks
+    def user_joined_room(self, user, room_id):
+        localusers = set()
+        remotedomains = set()
+
+        rm_handler = self.homeserver.get_handlers().room_member_handler
+        yield rm_handler.fetch_room_distributions_into(room_id,
+                localusers=localusers, remotedomains=remotedomains,
+                ignore_user=user)
+
+        if user.is_mine:
+            yield self._send_presence_to_distribution(srcuser=user,
+                localusers=localusers, remotedomains=remotedomains,
+                statuscache=self._get_or_offline_usercache(user),
+            )
+
+        for srcuser in localusers:
+            yield self._send_presence(srcuser=srcuser, destuser=user,
+                statuscache=self._get_or_offline_usercache(srcuser),
+            )
+
+    @defer.inlineCallbacks
+    def send_invite(self, observer_user, observed_user):
+        if not observer_user.is_mine:
+            raise SynapseError(400, "User is not hosted on this Home Server")
+
+        yield self.store.add_presence_list_pending(
+            observer_user.localpart, observed_user.to_string()
+        )
+
+        if observed_user.is_mine:
+            yield self.invite_presence(observed_user, observer_user)
+        else:
+            yield self.federation.send_edu(
+                destination=observed_user.domain,
+                edu_type="m.presence_invite",
+                content={
+                    "observed_user": observed_user.to_string(),
+                    "observer_user": observer_user.to_string(),
+                }
+            )
+
+    @defer.inlineCallbacks
+    def _should_accept_invite(self, observed_user, observer_user):
+        if not observed_user.is_mine:
+            defer.returnValue(False)
+
+        row = yield self.store.has_presence_state(observed_user.localpart)
+        if not row:
+            defer.returnValue(False)
+
+        # TODO(paul): Eventually we'll ask the user's permission for this
+        # before accepting. For now just accept any invite request
+        defer.returnValue(True)
+
+    @defer.inlineCallbacks
+    def invite_presence(self, observed_user, observer_user):
+        accept = yield self._should_accept_invite(observed_user, observer_user)
+
+        if accept:
+            yield self.store.allow_presence_visible(
+                observed_user.localpart, observer_user.to_string()
+            )
+
+        if observer_user.is_mine:
+            if accept:
+                yield self.accept_presence(observed_user, observer_user)
+            else:
+                yield self.deny_presence(observed_user, observer_user)
+        else:
+            edu_type = "m.presence_accept" if accept else "m.presence_deny"
+
+            yield self.federation.send_edu(
+                destination=observer_user.domain,
+                edu_type=edu_type,
+                content={
+                    "observed_user": observed_user.to_string(),
+                    "observer_user": observer_user.to_string(),
+                }
+            )
+
+    @defer.inlineCallbacks
+    def accept_presence(self, observed_user, observer_user):
+        yield self.store.set_presence_list_accepted(
+            observer_user.localpart, observed_user.to_string()
+        )
+
+        self.start_polling_presence(observer_user, target_user=observed_user)
+
+    @defer.inlineCallbacks
+    def deny_presence(self, observed_user, observer_user):
+        yield self.store.del_presence_list(
+            observer_user.localpart, observed_user.to_string()
+        )
+
+        # TODO(paul): Inform the user somehow?
+
+    @defer.inlineCallbacks
+    def drop(self, observed_user, observer_user):
+        if not observer_user.is_mine:
+            raise SynapseError(400, "User is not hosted on this Home Server")
+
+        yield self.store.del_presence_list(
+            observer_user.localpart, observed_user.to_string()
+        )
+
+        self.stop_polling_presence(observer_user, target_user=observed_user)
+
+    @defer.inlineCallbacks
+    def get_presence_list(self, observer_user, accepted=None):
+        if not observer_user.is_mine:
+            raise SynapseError(400, "User is not hosted on this Home Server")
+
+        presence = yield self.store.get_presence_list(
+            observer_user.localpart, accepted=accepted
+        )
+
+        for p in presence:
+            observed_user = self.hs.parse_userid(p.pop("observed_user_id"))
+            p["observed_user"] = observed_user
+            p.update(self._get_or_offline_usercache(observed_user).get_state())
+
+        defer.returnValue(presence)
+
+    @defer.inlineCallbacks
+    def start_polling_presence(self, user, target_user=None, state=None):
+        logger.debug("Start polling for presence from %s", user)
+
+        if target_user:
+            target_users = [target_user]
+        else:
+            presence = yield self.store.get_presence_list(
+                user.localpart, accepted=True
+            )
+            target_users = [
+                self.hs.parse_userid(x["observed_user_id"]) for x in presence
+            ]
+
+        if state is None:
+            state = yield self.store.get_presence_state(user.localpart)
+
+        localusers, remoteusers = partitionbool(
+            target_users,
+            lambda u: u.is_mine
+        )
+
+        for target_user in localusers:
+            self._start_polling_local(user, target_user)
+
+        deferreds = []
+        remoteusers_by_domain = partition(remoteusers, lambda u: u.domain)
+        for domain in remoteusers_by_domain:
+            remoteusers = remoteusers_by_domain[domain]
+
+            deferreds.append(self._start_polling_remote(
+                user, domain, remoteusers
+            ))
+
+        yield defer.DeferredList(deferreds)
+
+    def _start_polling_local(self, user, target_user):
+        target_localpart = target_user.localpart
+
+        if not self.is_presence_visible(observer_user=user,
+            observed_user=target_user):
+            return
+
+        if target_localpart not in self._local_pushmap:
+            self._local_pushmap[target_localpart] = set()
+
+        self._local_pushmap[target_localpart].add(user)
+
+        self.push_update_to_clients(
+            observer_user=user,
+            observed_user=target_user,
+            statuscache=self._get_or_offline_usercache(target_user),
+        )
+
+    def _start_polling_remote(self, user, domain, remoteusers):
+        for u in remoteusers:
+            if u not in self._remote_recvmap:
+                self._remote_recvmap[u] = set()
+
+            self._remote_recvmap[u].add(user)
+
+        return self.federation.send_edu(
+            destination=domain,
+            edu_type="m.presence",
+            content={"poll": [u.to_string() for u in remoteusers]}
+        )
+
+    def stop_polling_presence(self, user, target_user=None):
+        logger.debug("Stop polling for presence from %s", user)
+
+        if not target_user or target_user.is_mine:
+            self._stop_polling_local(user, target_user=target_user)
+
+        deferreds = []
+
+        if target_user:
+            raise NotImplementedError("TODO: remove one user")
+
+        remoteusers = [u for u in self._remote_recvmap
+                       if user in self._remote_recvmap[u]]
+        remoteusers_by_domain = partition(remoteusers, lambda u: u.domain)
+
+        for domain in remoteusers_by_domain:
+            remoteusers = remoteusers_by_domain[domain]
+
+            deferreds.append(
+                self._stop_polling_remote(user, domain, remoteusers)
+            )
+
+        return defer.DeferredList(deferreds)
+
+    def _stop_polling_local(self, user, target_user):
+        for localpart in self._local_pushmap.keys():
+            if target_user and localpart != target_user.localpart:
+                continue
+
+            if user in self._local_pushmap[localpart]:
+                self._local_pushmap[localpart].remove(user)
+
+            if not self._local_pushmap[localpart]:
+                del self._local_pushmap[localpart]
+
+    def _stop_polling_remote(self, user, domain, remoteusers):
+        for u in remoteusers:
+            self._remote_recvmap[u].remove(user)
+
+            if not self._remote_recvmap[u]:
+                del self._remote_recvmap[u]
+
+        return self.federation.send_edu(
+            destination=domain,
+            edu_type="m.presence",
+            content={"unpoll": [u.to_string() for u in remoteusers]}
+        )
+
+    @defer.inlineCallbacks
+    def push_presence(self, user, statuscache):
+        assert(user.is_mine)
+
+        logger.debug("Pushing presence update from %s", user)
+
+        localusers = set(self._local_pushmap.get(user.localpart, set()))
+        remotedomains = set(self._remote_sendmap.get(user.localpart, set()))
+
+        # Reflect users' status changes back to themselves, so UIs look nice
+        # and also user is informed of server-forced pushes
+        localusers.add(user)
+
+        rm_handler = self.homeserver.get_handlers().room_member_handler
+        room_ids = yield rm_handler.get_rooms_for_user(user)
+
+        for room_id in room_ids:
+            yield rm_handler.fetch_room_distributions_into(
+                room_id, localusers=localusers, remotedomains=remotedomains,
+                ignore_user=user,
+            )
+
+        if not localusers and not remotedomains:
+            defer.returnValue(None)
+
+        yield self._send_presence_to_distribution(user,
+            localusers=localusers, remotedomains=remotedomains,
+            statuscache=statuscache
+        )
+
+    def _send_presence(self, srcuser, destuser, statuscache):
+        if destuser.is_mine:
+            self.push_update_to_clients(
+                observer_user=destuser,
+                observed_user=srcuser,
+                statuscache=statuscache)
+            return defer.succeed(None)
+        else:
+            return self._push_presence_remote(srcuser, destuser.domain,
+                state=statuscache.get_state()
+            )
+
+    @defer.inlineCallbacks
+    def _send_presence_to_distribution(self, srcuser, localusers=set(),
+            remotedomains=set(), statuscache=None):
+
+        for u in localusers:
+            logger.debug(" | push to local user %s", u)
+            self.push_update_to_clients(
+                observer_user=u,
+                observed_user=srcuser,
+                statuscache=statuscache,
+            )
+
+        deferreds = []
+        for domain in remotedomains:
+            logger.debug(" | push to remote domain %s", domain)
+            deferreds.append(self._push_presence_remote(srcuser, domain,
+                state=statuscache.get_state())
+            )
+
+        yield defer.DeferredList(deferreds)
+
+    @defer.inlineCallbacks
+    def _push_presence_remote(self, user, destination, state=None):
+        if state is None:
+            state = yield self.store.get_presence_state(user.localpart)
+            yield self.distributor.fire(
+                "collect_presencelike_data", user, state
+            )
+
+        yield self.federation.send_edu(
+            destination=destination,
+            edu_type="m.presence",
+            content={
+                "push": [
+                    dict(user_id=user.to_string(), **state),
+                ],
+            }
+        )
+
+    @defer.inlineCallbacks
+    def incoming_presence(self, origin, content):
+        deferreds = []
+
+        for push in content.get("push", []):
+            user = self.hs.parse_userid(push["user_id"])
+
+            logger.debug("Incoming presence update from %s", user)
+
+            observers = set(self._remote_recvmap.get(user, set()))
+
+            rm_handler = self.homeserver.get_handlers().room_member_handler
+            room_ids = yield rm_handler.get_rooms_for_user(user)
+
+            for room_id in room_ids:
+                yield rm_handler.fetch_room_distributions_into(
+                    room_id, localusers=observers, ignore_user=user
+                )
+
+            if not observers:
+                break
+
+            state = dict(push)
+            del state["user_id"]
+
+            statuscache = self._get_or_make_usercache(user)
+
+            self._user_cachemap_latest_serial += 1
+            statuscache.update(state, serial=self._user_cachemap_latest_serial)
+
+            for observer_user in observers:
+                self.push_update_to_clients(
+                    observer_user=observer_user,
+                    observed_user=user,
+                    statuscache=statuscache,
+                )
+
+            if state["state"] == PresenceState.OFFLINE:
+                del self._user_cachemap[user]
+
+        for poll in content.get("poll", []):
+            user = self.hs.parse_userid(poll)
+
+            if not user.is_mine:
+                continue
+
+            # TODO(paul) permissions checks
+
+            if not user in self._remote_sendmap:
+                self._remote_sendmap[user] = set()
+
+            self._remote_sendmap[user].add(origin)
+
+            deferreds.append(self._push_presence_remote(user, origin))
+
+        for unpoll in content.get("unpoll", []):
+            user = self.hs.parse_userid(unpoll)
+
+            if not user.is_mine:
+                continue
+
+            if user in self._remote_sendmap:
+                self._remote_sendmap[user].remove(origin)
+
+                if not self._remote_sendmap[user]:
+                    del self._remote_sendmap[user]
+
+        yield defer.DeferredList(deferreds)
+
+    def push_update_to_clients(self, observer_user, observed_user,
+                               statuscache):
+        self.notifier.on_new_user_event(
+            observer_user.to_string(),
+            event_data=statuscache.make_event(user=observed_user),
+            stream_type=PresenceStreamData,
+            store_id=statuscache.serial
+        )
+
+
+class PresenceStreamData(StreamData):
+    def __init__(self, hs):
+        super(PresenceStreamData, self).__init__(hs)
+        self.presence = hs.get_handlers().presence_handler
+
+    def get_rows(self, user_id, from_key, to_key, limit):
+        cachemap = self.presence._user_cachemap
+
+        # TODO(paul): limit, and filter by visibility
+        updates = [(k, cachemap[k]) for k in cachemap
+                   if from_key < cachemap[k].serial <= to_key]
+
+        if updates:
+            latest_serial = max([x[1].serial for x in updates])
+            data = [x[1].make_event(user=x[0]) for x in updates]
+            return ((data, latest_serial))
+        else:
+            return (([], self.presence._user_cachemap_latest_serial))
+
+    def max_token(self):
+        return self.presence._user_cachemap_latest_serial
+
+PresenceStreamData.EVENT_TYPE = PresenceStreamData
+
+
+class UserPresenceCache(object):
+    """Store an observed user's state and status message.
+
+    Includes the update timestamp.
+    """
+    def __init__(self):
+        self.state = {}
+        self.serial = None
+
+    def update(self, state, serial):
+        self.state.update(state)
+        # Delete keys that are now 'None'
+        for k in self.state.keys():
+            if self.state[k] is None:
+                del self.state[k]
+
+        self.serial = serial
+
+        if "status_msg" in state:
+            self.status_msg = state["status_msg"]
+        else:
+            self.status_msg = None
+
+    def get_state(self):
+        # clone it so caller can't break our cache
+        return dict(self.state)
+
+    def make_event(self, user):
+        content = self.get_state()
+        content["user_id"] = user.to_string()
+
+        return {"type": "m.presence", "content": content}