diff --git a/synapse/__init__.py b/synapse/__init__.py
index 1e7b2ab272..47fc1b2ea4 100644
--- a/synapse/__init__.py
+++ b/synapse/__init__.py
@@ -15,3 +15,5 @@
""" This is a reference implementation of a synapse home server.
"""
+
+__version__ = "0.0.1"
diff --git a/synapse/api/auth.py b/synapse/api/auth.py
index 91ec0995f9..646f6dc06c 100644
--- a/synapse/api/auth.py
+++ b/synapse/api/auth.py
@@ -19,8 +19,7 @@ from twisted.internet import defer
from synapse.api.constants import Membership
from synapse.api.errors import AuthError, StoreError, Codes
-from synapse.api.events.room import (RoomTopicEvent, RoomMemberEvent,
- MessageEvent, FeedbackEvent)
+from synapse.api.events.room import RoomMemberEvent
import logging
@@ -44,19 +43,19 @@ class Auth(object):
be raised only if raises=True.
"""
try:
- if event.type in [RoomTopicEvent.TYPE, MessageEvent.TYPE,
- FeedbackEvent.TYPE]:
- self._check_joined_room(
- member=snapshot.membership_state,
- user_id=snapshot.user_id,
- room_id=snapshot.room_id,
- )
- defer.returnValue(True)
- elif event.type == RoomMemberEvent.TYPE:
- allowed = yield self.is_membership_change_allowed(event)
- defer.returnValue(allowed)
+ if hasattr(event, "room_id"):
+ if event.type == RoomMemberEvent.TYPE:
+ allowed = yield self.is_membership_change_allowed(event)
+ defer.returnValue(allowed)
+ else:
+ self._check_joined_room(
+ member=snapshot.membership_state,
+ user_id=snapshot.user_id,
+ room_id=snapshot.room_id,
+ )
+ defer.returnValue(True)
else:
- raise AuthError(500, "Unknown event type %s" % event.type)
+ raise AuthError(500, "Unknown event: %s" % event)
except AuthError as e:
logger.info("Event auth check failed on event %s with msg: %s",
event, e.msg)
@@ -83,6 +82,8 @@ class Auth(object):
@defer.inlineCallbacks
def is_membership_change_allowed(self, event):
+ target_user_id = event.state_key
+
# does this room even exist
room = yield self.store.get_room(event.room_id)
if not room:
@@ -100,7 +101,7 @@ class Auth(object):
# get info about the target
try:
target = yield self.store.get_room_member(
- user_id=event.target_user_id,
+ user_id=target_user_id,
room_id=event.room_id)
except:
target = None
@@ -114,12 +115,12 @@ class Auth(object):
raise AuthError(403, "You are not in room %s." % event.room_id)
elif target_in_room: # the target is already in the room.
raise AuthError(403, "%s is already in the room." %
- event.target_user_id)
+ target_user_id)
elif Membership.JOIN == membership:
# Joins are valid iff caller == target and they were:
# invited: They are accepting the invitation
# joined: It's a NOOP
- if event.user_id != event.target_user_id:
+ if event.user_id != target_user_id:
raise AuthError(403, "Cannot force another user to join.")
elif room.is_public:
pass # anyone can join public rooms.
@@ -129,10 +130,10 @@ class Auth(object):
elif Membership.LEAVE == membership:
if not caller_in_room: # trying to leave a room you aren't joined
raise AuthError(403, "You are not in room %s." % event.room_id)
- elif event.target_user_id != event.user_id:
+ elif target_user_id != event.user_id:
# trying to force another user to leave
raise AuthError(403, "Cannot force %s to leave." %
- event.target_user_id)
+ target_user_id)
else:
raise AuthError(500, "Unknown membership %s" % membership)
diff --git a/synapse/api/constants.py b/synapse/api/constants.py
index 1ff1af76ec..2af5424029 100644
--- a/synapse/api/constants.py
+++ b/synapse/api/constants.py
@@ -23,6 +23,7 @@ class Membership(object):
JOIN = u"join"
KNOCK = u"knock"
LEAVE = u"leave"
+ LIST = (INVITE, JOIN, KNOCK, LEAVE)
class Feedback(object):
diff --git a/synapse/api/events/__init__.py b/synapse/api/events/__init__.py
index 921fd08832..f9653e0b2a 100644
--- a/synapse/api/events/__init__.py
+++ b/synapse/api/events/__init__.py
@@ -41,16 +41,17 @@ class SynapseEvent(JsonEncodedObject):
"room_id",
"user_id", # sender/initiator
"content", # HTTP body, JSON
+ "state_key",
]
internal_keys = [
"is_state",
- "state_key",
"prev_events",
"prev_state",
"depth",
"destinations",
"origin",
+ "outlier",
]
required_keys = [
diff --git a/synapse/api/events/factory.py b/synapse/api/events/factory.py
index b61dac7acd..c2cdcddf41 100644
--- a/synapse/api/events/factory.py
+++ b/synapse/api/events/factory.py
@@ -33,16 +33,21 @@ class EventFactory(object):
RoomConfigEvent
]
- def __init__(self):
+ def __init__(self, hs):
self._event_list = {} # dict of TYPE to event class
for event_class in EventFactory._event_classes:
self._event_list[event_class.TYPE] = event_class
+ self.clock = hs.get_clock()
+
def create_event(self, etype=None, **kwargs):
kwargs["type"] = etype
if "event_id" not in kwargs:
kwargs["event_id"] = random_string(10)
+ if "ts" not in kwargs:
+ kwargs["ts"] = int(self.clock.time_msec())
+
if etype in self._event_list:
handler = self._event_list[etype]
else:
diff --git a/synapse/api/events/room.py b/synapse/api/events/room.py
index 42459f3f21..2a7b5e8aba 100644
--- a/synapse/api/events/room.py
+++ b/synapse/api/events/room.py
@@ -13,6 +13,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from synapse.api.constants import Membership
+from synapse.api.errors import SynapseError
from . import SynapseEvent
@@ -59,15 +61,15 @@ class RoomMemberEvent(SynapseEvent):
TYPE = "m.room.member"
valid_keys = SynapseEvent.valid_keys + [
- "target_user_id", # target
+ # target is the state_key
"membership", # action
]
def __init__(self, **kwargs):
- if "target_user_id" in kwargs:
- kwargs["state_key"] = kwargs["target_user_id"]
if "membership" not in kwargs:
kwargs["membership"] = kwargs.get("content", {}).get("membership")
+ if not kwargs["membership"] in Membership.LIST:
+ raise SynapseError(400, "Bad membership value.")
super(RoomMemberEvent, self).__init__(**kwargs)
def get_content_template(self):
@@ -108,7 +110,7 @@ class InviteJoinEvent(SynapseEvent):
TYPE = "m.room.invite_join"
valid_keys = SynapseEvent.valid_keys + [
- "target_user_id",
+ # target_user_id is the state_key
"target_host",
]
diff --git a/synapse/api/notifier.py b/synapse/api/notifier.py
index 9f622df6bb..ec9c4e513d 100644
--- a/synapse/api/notifier.py
+++ b/synapse/api/notifier.py
@@ -56,11 +56,11 @@ class Notifier(object):
# invites MUST prod the person being invited, who won't be in the room.
if (event.type == RoomMemberEvent.TYPE and
event.content["membership"] == Membership.INVITE):
- member_list.append(event.target_user_id)
+ member_list.append(event.state_key)
# similarly, LEAVEs must be sent to the person leaving
if (event.type == RoomMemberEvent.TYPE and
event.content["membership"] == Membership.LEAVE):
- member_list.append(event.target_user_id)
+ member_list.append(event.state_key)
for user_id in member_list:
if user_id in self.stored_event_listeners:
diff --git a/synapse/api/streams/event.py b/synapse/api/streams/event.py
index a5c8b2b31f..fe44a488bc 100644
--- a/synapse/api/streams/event.py
+++ b/synapse/api/streams/event.py
@@ -19,9 +19,6 @@ from twisted.internet import defer
from synapse.api.errors import EventStreamError
from synapse.api.events import SynapseEvent
-from synapse.api.events.room import (
- RoomMemberEvent, MessageEvent, FeedbackEvent, RoomTopicEvent
-)
from synapse.api.streams import PaginationStream, StreamData
import logging
diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py
index ca102236cf..f210d26629 100755
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -31,12 +31,14 @@ from synapse.api.urls import (
)
from daemonize import Daemonize
+import twisted.manhole.telnet
import argparse
import logging
import logging.config
import sqlite3
import os
+import re
logger = logging.getLogger(__name__)
@@ -56,7 +58,7 @@ class SynapseHomeServer(HomeServer):
return File("webclient") # TODO configurable?
def build_resource_for_content_repo(self):
- return ContentRepoResource("uploads", self.auth)
+ return ContentRepoResource(self, self.upload_dir, self.auth)
def build_db_pool(self):
""" Set up all the dbs. Since all the *.sql have IF NOT EXISTS, so we
@@ -235,8 +237,10 @@ def setup():
parser.add_argument('--pid-file', dest="pid", help="When running as a "
"daemon, the file to store the pid in",
default="hs.pid")
- parser.add_argument("-w", "--webclient", dest="webclient",
- action="store_true", help="Host the web client.")
+ parser.add_argument("-W", "--webclient", dest="webclient", default=True,
+ action="store_false", help="Don't host a web client.")
+ parser.add_argument("--manhole", dest="manhole", type=int, default=None,
+ help="Turn on the twisted telnet manhole service.")
args = parser.parse_args()
verbosity = int(args.verbose) if args.verbose else None
@@ -255,9 +259,16 @@ def setup():
logger.info("Server hostname: %s", args.host)
+ if re.search(":[0-9]+$", args.host):
+ domain_with_port = args.host
+ else:
+ domain_with_port = "%s:%s" % (args.host, args.port)
+
hs = SynapseHomeServer(
args.host,
- db_name=db_name
+ domain_with_port=domain_with_port,
+ upload_dir=os.path.abspath("uploads"),
+ db_name=db_name,
)
# This object doesn't need to be saved because it's set as the handler for
@@ -273,6 +284,13 @@ def setup():
hs.build_db_pool()
+ if args.manhole:
+ f = twisted.manhole.telnet.ShellFactory()
+ f.username = "matrix"
+ f.password = "rabbithole"
+ f.namespace['hs'] = hs
+ reactor.listenTCP(args.manhole, f, interface='127.0.0.1')
+
if args.daemonize:
daemon = Daemonize(
app="synapse-homeserver",
diff --git a/synapse/federation/replication.py b/synapse/federation/replication.py
index 8030d0963f..cf634a64b2 100644
--- a/synapse/federation/replication.py
+++ b/synapse/federation/replication.py
@@ -509,10 +509,10 @@ class _TransactionQueue(object):
# a transaction in progress. If we do, stick it in the pending_pdus
# table and we'll get back to it later.
- destinations = [
+ destinations = set([
d for d in pdu.destinations
if d != self.server_name
- ]
+ ])
logger.debug("Sending to: %s", str(destinations))
diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index c2f4685c92..3f07b5aa4a 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -24,4 +24,5 @@ class BaseHandler(object):
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.hs = hs
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index aa3bf273f7..2680678de7 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -32,6 +32,15 @@ logger = logging.getLogger(__name__)
class FederationHandler(BaseHandler):
"""Handles events that originated from federation."""
+ def __init__(self, hs):
+ super(FederationHandler, self).__init__(hs)
+
+ self.distributor.observe(
+ "user_joined_room",
+ self._on_user_joined
+ )
+
+ self.waiting_for_join_list = {}
@log_function
@defer.inlineCallbacks
@@ -56,7 +65,7 @@ class FederationHandler(BaseHandler):
content.update({"membership": Membership.JOIN})
new_event = self.event_factory.create_event(
etype=RoomMemberEvent.TYPE,
- target_user_id=event.user_id,
+ state_key=event.user_id,
room_id=event.room_id,
user_id=event.user_id,
membership=Membership.JOIN,
@@ -103,6 +112,13 @@ class FederationHandler(BaseHandler):
if not backfilled:
yield self.notifier.on_new_room_event(event, store_id)
+ if event.type == RoomMemberEvent.TYPE:
+ if event.membership == Membership.JOIN:
+ user = self.hs.parse_userid(event.target_user_id)
+ self.distributor.fire(
+ "user_joined_room", user=user, room_id=event.room_id
+ )
+
@log_function
@defer.inlineCallbacks
@@ -152,12 +168,14 @@ class FederationHandler(BaseHandler):
yield federation.handle_new_event(new_event)
- store_id = yield self.store.persist_event(new_event)
- self.notifier.on_new_room_event(new_event, store_id)
+ # TODO (erikj): Time out here.
+ d = defer.Deferred()
+ self.waiting_for_join_list.setdefault((joinee, room_id), []).append(d)
+ yield d
try:
yield self.store.store_room(
- event.room_id,
+ room_id,
"",
is_public=False
)
@@ -166,3 +184,10 @@ class FederationHandler(BaseHandler):
defer.returnValue(True)
+
+
+ @log_function
+ def _on_user_joined(self, user, room_id):
+ waiters = self.waiting_for_join_list.get((user.to_string(), room_id), [])
+ while waiters:
+ waiters.pop().callback(None)
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 540e114b82..be10162db5 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -142,6 +142,10 @@ class PresenceHandler(BaseHandler):
@defer.inlineCallbacks
def is_presence_visible(self, observer_user, observed_user):
+ defer.returnValue(True)
+ return
+ # FIXME (erikj): This code path absolutely kills the database.
+
assert(observed_user.is_mine)
if observer_user == observed_user:
@@ -155,12 +159,11 @@ class PresenceHandler(BaseHandler):
if allowed_by_subscription:
defer.returnValue(True)
- rm_handler = self.homeserver.get_handlers().room_member_handler
- for room_id in (yield rm_handler.get_rooms_for_user(observer_user)):
- if observed_user in (yield rm_handler.get_room_members(room_id)):
- defer.returnValue(True)
+ share_room = yield self.store.do_users_share_a_room(
+ [observer_user, observed_user]
+ )
- defer.returnValue(False)
+ defer.returnValue(share_room)
@defer.inlineCallbacks
def get_state(self, target_user, auth_user):
@@ -187,6 +190,10 @@ class PresenceHandler(BaseHandler):
@defer.inlineCallbacks
def set_state(self, target_user, auth_user, state):
+ return
+ # TODO (erikj): Turn this back on. Why did we end up sending EDUs
+ # everywhere?
+
if not target_user.is_mine:
raise SynapseError(400, "User is not hosted on this Home Server")
@@ -667,7 +674,7 @@ class PresenceHandler(BaseHandler):
def push_update_to_clients(self, observer_user, observed_user,
statuscache):
- state = statuscache.make_event(user=observed_user, clock=self.clock)
+ statuscache.make_event(user=observed_user, clock=self.clock)
self.notifier.on_new_user_event(
observer_user.to_string(),
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 53ccaa0991..7e34b4a6fc 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -20,15 +20,14 @@ from synapse.types import UserID, RoomAlias, RoomID
from synapse.api.constants import Membership
from synapse.api.errors import RoomError, StoreError, SynapseError
from synapse.api.events.room import (
- RoomTopicEvent, MessageEvent, InviteJoinEvent, RoomMemberEvent,
- RoomConfigEvent
+ RoomTopicEvent, RoomMemberEvent, RoomConfigEvent
)
from synapse.api.streams.event import EventStream, EventsStreamData
+from synapse.handlers.presence import PresenceStreamData
from synapse.util import stringutils
from ._base import BaseHandler
import logging
-import json
logger = logging.getLogger(__name__)
@@ -260,21 +259,38 @@ class MessageHandler(BaseHandler):
membership_list=[Membership.INVITE, Membership.JOIN]
)
- ret = []
+ rooms_ret = []
+
+ now_rooms_token = yield self.store.get_room_events_max_id()
+
+ # FIXME (erikj): Fix this.
+ presence_stream = PresenceStreamData(self.hs)
+ now_presence_token = yield presence_stream.max_token()
+ presence = yield presence_stream.get_rows(
+ user_id, 0, now_presence_token, None, None
+ )
+
+ # FIXME (erikj): We need to not generate this token,
+ now_token = "%s_%s" % (now_rooms_token, now_presence_token)
for event in room_list:
d = {
"room_id": event.room_id,
"membership": event.membership,
}
- ret.append(d)
+
+ if event.membership == Membership.INVITE:
+ d["inviter"] = event.user_id
+
+ rooms_ret.append(d)
if event.membership != Membership.JOIN:
continue
try:
messages, token = yield self.store.get_recent_events_for_room(
event.room_id,
- limit=50,
+ limit=10,
+ end_token=now_rooms_token,
)
d["messages"] = {
@@ -282,10 +298,15 @@ class MessageHandler(BaseHandler):
"start": token[0],
"end": token[1],
}
+
+ current_state = yield self.store.get_current_state(event.room_id)
+ d["state"] = [c.get_dict() for c in current_state]
except:
logger.exception("Failed to get snapshot")
- logger.debug("snapshot_all_rooms returning: %s", ret)
+ ret = {"rooms": rooms_ret, "presence": presence[0], "end": now_token}
+
+ # logger.debug("snapshot_all_rooms returning: %s", ret)
defer.returnValue(ret)
@@ -377,7 +398,7 @@ class RoomCreationHandler(BaseHandler):
content = {"membership": Membership.JOIN}
join_event = self.event_factory.create_event(
etype=RoomMemberEvent.TYPE,
- target_user_id=user_id,
+ state_key=user_id,
room_id=room_id,
user_id=user_id,
membership=Membership.JOIN,
@@ -505,6 +526,7 @@ class RoomMemberHandler(BaseHandler):
Raises:
SynapseError if there was a problem changing the membership.
"""
+ target_user_id = event.state_key
snapshot = yield self.store.snapshot_room(
event.room_id, event.user_id,
@@ -512,7 +534,7 @@ class RoomMemberHandler(BaseHandler):
)
## TODO(markjh): get prev state from snapshot.
prev_state = yield self.store.get_room_member(
- event.target_user_id, event.room_id
+ target_user_id, event.room_id
)
if prev_state:
@@ -569,7 +591,7 @@ class RoomMemberHandler(BaseHandler):
content.update({"membership": Membership.JOIN})
new_event = self.event_factory.create_event(
etype=RoomMemberEvent.TYPE,
- target_user_id=joinee.to_string(),
+ state_key=joinee.to_string(),
room_id=room_id,
user_id=joinee.to_string(),
membership=Membership.JOIN,
@@ -586,7 +608,7 @@ class RoomMemberHandler(BaseHandler):
@defer.inlineCallbacks
def _do_join(self, event, snapshot, room_host=None, do_auth=True):
- joinee = self.hs.parse_userid(event.target_user_id)
+ joinee = self.hs.parse_userid(event.state_key)
# room_id = RoomID.from_string(event.room_id, self.hs)
room_id = event.room_id
@@ -697,16 +719,17 @@ class RoomMemberHandler(BaseHandler):
# If we're inviting someone, then we should also send it to that
# HS.
+ target_user_id = event.state_key
if membership == Membership.INVITE:
host = UserID.from_string(
- event.target_user_id, self.hs
+ target_user_id, self.hs
).domain
destinations.append(host)
# If we are joining a remote HS, include that.
if membership == Membership.JOIN:
host = UserID.from_string(
- event.target_user_id, self.hs
+ target_user_id, self.hs
).domain
destinations.append(host)
diff --git a/synapse/http/server.py b/synapse/http/server.py
index c28d9a33f9..66f966fcaa 100644
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -212,8 +212,9 @@ class ContentRepoResource(resource.Resource):
"""
isLeaf = True
- def __init__(self, directory, auth):
+ def __init__(self, hs, directory, auth):
resource.Resource.__init__(self)
+ self.hs = hs
self.directory = directory
self.auth = auth
@@ -250,7 +251,8 @@ class ContentRepoResource(resource.Resource):
file_ext = re.sub("[^a-z]", "", file_ext)
suffix += "." + file_ext
- file_path = os.path.join(self.directory, prefix + main_part + suffix)
+ file_name = prefix + main_part + suffix
+ file_path = os.path.join(self.directory, file_name)
logger.info("User %s is uploading a file to path %s",
auth_user.to_string(),
file_path)
@@ -259,8 +261,8 @@ class ContentRepoResource(resource.Resource):
attempts = 0
while os.path.exists(file_path):
main_part = random_string(24)
- file_path = os.path.join(self.directory,
- prefix + main_part + suffix)
+ file_name = prefix + main_part + suffix
+ file_path = os.path.join(self.directory, file_name)
attempts += 1
if attempts > 25: # really? Really?
raise SynapseError(500, "Unable to create file.")
@@ -272,11 +274,14 @@ class ContentRepoResource(resource.Resource):
# servers.
# TODO: A little crude here, we could do this better.
- filename = request.path.split(self.directory + "/")[1]
+ filename = request.path.split('/')[-1]
# be paranoid
filename = re.sub("[^0-9A-z.-_]", "", filename)
file_path = self.directory + "/" + filename
+
+ logger.debug("Searching for %s", file_path)
+
if os.path.isfile(file_path):
# filename has the content type
base64_contentype = filename.split(".")[1]
@@ -304,6 +309,10 @@ class ContentRepoResource(resource.Resource):
self._async_render(request)
return server.NOT_DONE_YET
+ def render_OPTIONS(self, request):
+ respond_with_json_bytes(request, 200, {}, send_cors=True)
+ return server.NOT_DONE_YET
+
@defer.inlineCallbacks
def _async_render(self, request):
try:
@@ -313,8 +322,15 @@ class ContentRepoResource(resource.Resource):
with open(fname, "wb") as f:
f.write(request.content.read())
+
+ # FIXME (erikj): These should use constants.
+ file_name = os.path.basename(fname)
+ url = "http://%s/matrix/content/%s" % (
+ self.hs.domain_with_port, file_name
+ )
+
respond_with_json_bytes(request, 200,
- json.dumps({"content_token": fname}),
+ json.dumps({"content_token": url}),
send_cors=True)
except CodeMessageException as e:
diff --git a/synapse/rest/base.py b/synapse/rest/base.py
index 6a88cbe866..e855d293e5 100644
--- a/synapse/rest/base.py
+++ b/synapse/rest/base.py
@@ -15,6 +15,7 @@
""" This module contains base REST classes for constructing REST servlets. """
from synapse.api.urls import CLIENT_PREFIX
+from synapse.rest.transactions import HttpTransactionStore
import re
@@ -59,6 +60,7 @@ class RestServlet(object):
self.handlers = hs.get_handlers()
self.event_factory = hs.get_event_factory()
self.auth = hs.get_auth()
+ self.txns = HttpTransactionStore()
def register(self, http_server):
""" Register this servlet with the given HTTP server. """
diff --git a/synapse/rest/register.py b/synapse/rest/register.py
index eb457562b9..f17ec11cf4 100644
--- a/synapse/rest/register.py
+++ b/synapse/rest/register.py
@@ -33,10 +33,10 @@ class RegisterRestServlet(RestServlet):
try:
register_json = json.loads(request.content.read())
if "password" in register_json:
- password = register_json["password"]
+ password = register_json["password"].encode("utf-8")
if type(register_json["user_id"]) == unicode:
- desired_user_id = register_json["user_id"]
+ desired_user_id = register_json["user_id"].encode("utf-8")
if urllib.quote(desired_user_id) != desired_user_id:
raise SynapseError(
400,
diff --git a/synapse/rest/room.py b/synapse/rest/room.py
index f5b547b963..6771da8fcd 100644
--- a/synapse/rest/room.py
+++ b/synapse/rest/room.py
@@ -18,9 +18,10 @@ from twisted.internet import defer
from base import RestServlet, client_path_pattern
from synapse.api.errors import SynapseError, Codes
-from synapse.api.events.room import (RoomTopicEvent, MessageEvent,
- RoomMemberEvent, FeedbackEvent)
-from synapse.api.constants import Feedback, Membership
+from synapse.api.events.room import (
+ MessageEvent, RoomMemberEvent, FeedbackEvent
+)
+from synapse.api.constants import Feedback
from synapse.api.streams import PaginationConfig
import json
@@ -95,46 +96,76 @@ class RoomCreateRestServlet(RestServlet):
return (200, {})
-class RoomTopicRestServlet(RestServlet):
- PATTERN = client_path_pattern("/rooms/(?P<room_id>[^/]*)/topic$")
+class RoomStateEventRestServlet(RestServlet):
+ def register(self, http_server):
+ # /room/$roomid/state/$eventtype
+ no_state_key = "/rooms/(?P<room_id>[^/]*)/state/(?P<event_type>[^/]*)$"
- def get_event_type(self):
- return RoomTopicEvent.TYPE
+ # /room/$roomid/state/$eventtype/$statekey
+ state_key = ("/rooms/(?P<room_id>[^/]*)/state/" +
+ "(?P<event_type>[^/]*)/(?P<state_key>[^/]*)$")
+
+ http_server.register_path("GET",
+ client_path_pattern(state_key),
+ self.on_GET)
+ http_server.register_path("PUT",
+ client_path_pattern(state_key),
+ self.on_PUT)
+ http_server.register_path("GET",
+ client_path_pattern(no_state_key),
+ self.on_GET_no_state_key)
+ http_server.register_path("PUT",
+ client_path_pattern(no_state_key),
+ self.on_PUT_no_state_key)
+
+ def on_GET_no_state_key(self, request, room_id, event_type):
+ return self.on_GET(request, room_id, event_type, "")
+
+ def on_PUT_no_state_key(self, request, room_id, event_type):
+ return self.on_PUT(request, room_id, event_type, "")
@defer.inlineCallbacks
- def on_GET(self, request, room_id):
+ def on_GET(self, request, room_id, event_type, state_key):
user = yield self.auth.get_user_by_req(request)
msg_handler = self.handlers.message_handler
data = yield msg_handler.get_room_data(
user_id=user.to_string(),
room_id=urllib.unquote(room_id),
- event_type=RoomTopicEvent.TYPE,
- state_key="",
+ event_type=urllib.unquote(event_type),
+ state_key=urllib.unquote(state_key),
)
if not data:
- raise SynapseError(404, "Topic not found.", errcode=Codes.NOT_FOUND)
- defer.returnValue((200, data.content))
+ raise SynapseError(404, "Event not found.", errcode=Codes.NOT_FOUND)
+ defer.returnValue((200, data[0].get_dict()["content"]))
@defer.inlineCallbacks
- def on_PUT(self, request, room_id):
+ def on_PUT(self, request, room_id, event_type, state_key):
user = yield self.auth.get_user_by_req(request)
+ event_type = urllib.unquote(event_type)
content = _parse_json(request)
event = self.event_factory.create_event(
- etype=self.get_event_type(),
+ etype=event_type,
content=content,
room_id=urllib.unquote(room_id),
user_id=user.to_string(),
+ state_key=urllib.unquote(state_key)
)
-
- msg_handler = self.handlers.message_handler
- yield msg_handler.store_room_data(
- event=event
- )
- defer.returnValue((200, ""))
+ if event_type == RoomMemberEvent.TYPE:
+ # membership events are special
+ handler = self.handlers.room_member_handler
+ yield handler.change_membership(event)
+ defer.returnValue((200, ""))
+ else:
+ # store random bits of state
+ msg_handler = self.handlers.message_handler
+ yield msg_handler.store_room_data(
+ event=event
+ )
+ defer.returnValue((200, ""))
class JoinRoomAliasServlet(RestServlet):
@@ -157,73 +188,6 @@ class JoinRoomAliasServlet(RestServlet):
defer.returnValue((200, ret_dict))
-class RoomMemberRestServlet(RestServlet):
- PATTERN = client_path_pattern("/rooms/(?P<room_id>[^/]*)/members/"
- + "(?P<target_user_id>[^/]*)/state$")
-
- def get_event_type(self):
- return RoomMemberEvent.TYPE
-
- @defer.inlineCallbacks
- def on_GET(self, request, room_id, target_user_id):
- room_id = urllib.unquote(room_id)
- user = yield self.auth.get_user_by_req(request)
-
- handler = self.handlers.room_member_handler
- member = yield handler.get_room_member(
- room_id,
- urllib.unquote(target_user_id),
- user.to_string())
- if not member:
- raise SynapseError(404, "Member not found.",
- errcode=Codes.NOT_FOUND)
- defer.returnValue((200, member.content))
-
- @defer.inlineCallbacks
- def on_DELETE(self, request, roomid, target_user_id):
- user = yield self.auth.get_user_by_req(request)
-
- event = self.event_factory.create_event(
- etype=self.get_event_type(),
- target_user_id=urllib.unquote(target_user_id),
- room_id=urllib.unquote(roomid),
- user_id=user.to_string(),
- membership=Membership.LEAVE,
- content={"membership": Membership.LEAVE}
- )
-
- handler = self.handlers.room_member_handler
- yield handler.change_membership(event)
- defer.returnValue((200, ""))
-
- @defer.inlineCallbacks
- def on_PUT(self, request, roomid, target_user_id):
- user = yield self.auth.get_user_by_req(request)
-
- content = _parse_json(request)
- if "membership" not in content:
- raise SynapseError(400, "No membership key.",
- errcode=Codes.BAD_JSON)
-
- valid_membership_values = [Membership.JOIN, Membership.INVITE]
- if (content["membership"] not in valid_membership_values):
- raise SynapseError(400, "Membership value must be %s." % (
- valid_membership_values,), errcode=Codes.BAD_JSON)
-
- event = self.event_factory.create_event(
- etype=self.get_event_type(),
- target_user_id=urllib.unquote(target_user_id),
- room_id=urllib.unquote(roomid),
- user_id=user.to_string(),
- membership=content["membership"],
- content=content
- )
-
- handler = self.handlers.room_member_handler
- yield handler.change_membership(event)
- defer.returnValue((200, ""))
-
-
class MessageRestServlet(RestServlet):
PATTERN = client_path_pattern("/rooms/(?P<room_id>[^/]*)/messages/"
+ "(?P<sender_id>[^/]*)/(?P<msg_id>[^/]*)$")
@@ -285,7 +249,7 @@ class FeedbackRestServlet(RestServlet):
@defer.inlineCallbacks
def on_GET(self, request, room_id, msg_sender_id, msg_id, fb_sender_id,
feedback_type):
- user = yield (self.auth.get_user_by_req(request))
+ yield (self.auth.get_user_by_req(request))
# TODO (erikj): Implement this?
raise NotImplementedError("Getting feedback is not supported")
@@ -354,7 +318,8 @@ class RoomMemberListRestServlet(RestServlet):
user_id=user.to_string())
for event in members["chunk"]:
- target_user = self.hs.parse_userid(event["target_user_id"])
+ # FIXME: should probably be state_key here, not user_id
+ target_user = self.hs.parse_userid(event["user_id"])
# Presence is an optional cache; don't fail if we can't fetch it
try:
presence_state = yield self.handlers.presence_handler.get_state(
@@ -400,6 +365,52 @@ class RoomTriggerBackfill(RestServlet):
res = [event.get_dict() for event in events]
defer.returnValue((200, res))
+
+class RoomMembershipRestServlet(RestServlet):
+
+ def register(self, http_server):
+ # /rooms/$roomid/[invite|join|leave]
+ PATTERN = ("/rooms/(?P<room_id>[^/]*)/" +
+ "(?P<membership_action>join|invite|leave)")
+ register_txn_path(self, PATTERN, http_server)
+
+ @defer.inlineCallbacks
+ def on_POST(self, request, room_id, membership_action):
+ user = yield self.auth.get_user_by_req(request)
+
+ content = _parse_json(request)
+
+ # target user is you unless it is an invite
+ state_key = user.to_string()
+ if membership_action == "invite":
+ if "user_id" not in content:
+ raise SynapseError(400, "Missing user_id key.")
+ state_key = content["user_id"]
+
+ event = self.event_factory.create_event(
+ etype=RoomMemberEvent.TYPE,
+ content={"membership": unicode(membership_action)},
+ room_id=urllib.unquote(room_id),
+ user_id=user.to_string(),
+ state_key=state_key
+ )
+ handler = self.handlers.room_member_handler
+ yield handler.change_membership(event)
+ defer.returnValue((200, ""))
+
+ @defer.inlineCallbacks
+ def on_PUT(self, request, room_id, membership_action, txn_id):
+ try:
+ defer.returnValue(self.txns.get_client_transaction(request, txn_id))
+ except:
+ pass
+
+ response = yield self.on_POST(request, room_id, membership_action)
+
+ self.txns.store_client_transaction(request, txn_id, response)
+ defer.returnValue(response)
+
+
def _parse_json(request):
try:
content = json.loads(request.content.read())
@@ -411,9 +422,32 @@ def _parse_json(request):
raise SynapseError(400, "Content not JSON.", errcode=Codes.NOT_JSON)
+def register_txn_path(servlet, regex_string, http_server):
+ """Registers a transaction-based path.
+
+ This registers two paths:
+ PUT regex_string/$txnid
+ POST regex_string
+
+ Args:
+ regex_string (str): The regex string to register. Must NOT have a
+ trailing $ as this string will be appended to.
+ http_server : The http_server to register paths with.
+ """
+ http_server.register_path(
+ "POST",
+ client_path_pattern(regex_string + "$"),
+ servlet.on_POST
+ )
+ http_server.register_path(
+ "PUT",
+ client_path_pattern(regex_string + "/(?P<txn_id>[^/]*)$"),
+ servlet.on_PUT
+ )
+
+
def register_servlets(hs, http_server):
- RoomTopicRestServlet(hs).register(http_server)
- RoomMemberRestServlet(hs).register(http_server)
+ RoomStateEventRestServlet(hs).register(http_server)
MessageRestServlet(hs).register(http_server)
FeedbackRestServlet(hs).register(http_server)
RoomCreateRestServlet(hs).register(http_server)
@@ -421,3 +455,4 @@ def register_servlets(hs, http_server):
RoomMessageListRestServlet(hs).register(http_server)
JoinRoomAliasServlet(hs).register(http_server)
RoomTriggerBackfill(hs).register(http_server)
+ RoomMembershipRestServlet(hs).register(http_server)
diff --git a/synapse/rest/transactions.py b/synapse/rest/transactions.py
new file mode 100644
index 0000000000..b8aa1ef11c
--- /dev/null
+++ b/synapse/rest/transactions.py
@@ -0,0 +1,96 @@
+# -*- 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.
+
+"""This module contains logic for storing HTTP PUT transactions. This is used
+to ensure idempotency when performing PUTs using the REST API."""
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class HttpTransactionStore(object):
+
+ def __init__(self):
+ # { key : (txn_id, response) }
+ self.transactions = {}
+
+ def get_response(self, key, txn_id):
+ """Retrieve a response for this request.
+
+ Args:
+ key (str): A transaction-independent key for this request. Typically
+ this is a combination of the path (without the transaction id) and
+ the user's access token.
+ txn_id (str): The transaction ID for this request
+ Returns:
+ A tuple of (HTTP response code, response content) or None.
+ """
+ try:
+ logger.debug("get_response Key: %s TxnId: %s", key, txn_id)
+ (last_txn_id, response) = self.transactions[key]
+ if txn_id == last_txn_id:
+ logger.info("get_response: Returning a response for %s", key)
+ return response
+ except KeyError:
+ pass
+ return None
+
+ def store_response(self, key, txn_id, response):
+ """Stores an HTTP response tuple.
+
+ Args:
+ key (str): A transaction-independent key for this request. Typically
+ this is a combination of the path (without the transaction id) and
+ the user's access token.
+ txn_id (str): The transaction ID for this request.
+ response (tuple): A tuple of (HTTP response code, response content)
+ """
+ logger.debug("store_response Key: %s TxnId: %s", key, txn_id)
+ self.transactions[key] = (txn_id, response)
+
+ def store_client_transaction(self, request, txn_id, response):
+ """Stores the request/response pair of an HTTP transaction.
+
+ Args:
+ request (twisted.web.http.Request): The twisted HTTP request. This
+ request must have the transaction ID as the last path segment.
+ response (tuple): A tuple of (response code, response dict)
+ txn_id (str): The transaction ID for this request.
+ """
+ self.store_response(self._get_key(request), txn_id, response)
+
+ def get_client_transaction(self, request, txn_id):
+ """Retrieves a stored response if there was one.
+
+ Args:
+ request (twisted.web.http.Request): The twisted HTTP request. This
+ request must have the transaction ID as the last path segment.
+ txn_id (str): The transaction ID for this request.
+ Returns:
+ The response tuple.
+ Raises:
+ KeyError if the transaction was not found.
+ """
+ response = self.get_response(self._get_key(request), txn_id)
+ if response is None:
+ raise KeyError("Transaction not found.")
+ return response
+
+ def _get_key(self, request):
+ token = request.args["access_token"][0]
+ path_without_txn_id = request.path.rsplit("/", 1)[0]
+ return path_without_txn_id + "/" + token
+
+
diff --git a/synapse/server.py b/synapse/server.py
index d4c2481483..c5b0a32757 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -159,7 +159,7 @@ class HomeServer(BaseHomeServer):
return DataStore(self)
def build_event_factory(self):
- return EventFactory()
+ return EventFactory(self)
def build_handlers(self):
return Handlers(self)
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 63afca9ad7..5e52e9fecf 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -16,8 +16,9 @@
from twisted.internet import defer
from synapse.api.events.room import (
- RoomMemberEvent, MessageEvent, RoomTopicEvent, FeedbackEvent,
- RoomConfigEvent, RoomNameEvent,
+ RoomMemberEvent, RoomTopicEvent, FeedbackEvent,
+# RoomConfigEvent,
+ RoomNameEvent,
)
from synapse.util.logutils import log_function
@@ -116,6 +117,11 @@ class DataStore(RoomMemberStore, RoomStore,
if stream_ordering is not None:
vals["stream_ordering"] = stream_ordering
+ if hasattr(event, "outlier"):
+ vals["outlier"] = event.outlier
+ else:
+ vals["outlier"] = False
+
unrec = {
k: v
for k, v in event.get_full_dict().items()
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index cfbe85d798..33d56f47ce 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -296,6 +296,11 @@ class SQLBaseStore(object):
def _parse_event_from_row(self, row_dict):
d = copy.deepcopy({k: v for k, v in row_dict.items() if v})
+
+ d.pop("stream_ordering", None)
+ d.pop("topological_ordering", None)
+ d.pop("processed", None)
+
d.update(json.loads(row_dict["unrecognized_keys"]))
d["content"] = json.loads(d["content"])
del d["unrecognized_keys"]
diff --git a/synapse/storage/feedback.py b/synapse/storage/feedback.py
index d2d1e1d1e8..336192ad01 100644
--- a/synapse/storage/feedback.py
+++ b/synapse/storage/feedback.py
@@ -15,11 +15,7 @@
from twisted.internet import defer
-from ._base import SQLBaseStore, Table
-from synapse.api.events.room import FeedbackEvent
-
-import collections
-import json
+from ._base import SQLBaseStore
class FeedbackStore(SQLBaseStore):
diff --git a/synapse/storage/room.py b/synapse/storage/room.py
index 1ae3220197..d1f1a232f8 100644
--- a/synapse/storage/room.py
+++ b/synapse/storage/room.py
@@ -18,12 +18,10 @@ from twisted.internet import defer
from sqlite3 import IntegrityError
from synapse.api.errors import StoreError
-from synapse.api.events.room import RoomTopicEvent
from ._base import SQLBaseStore, Table
import collections
-import json
import logging
logger = logging.getLogger(__name__)
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index 1df043cd36..5038aeea03 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -15,15 +15,10 @@
from twisted.internet import defer
-from synapse.types import UserID
-from synapse.api.constants import Membership
-from synapse.api.events.room import RoomMemberEvent
-
-from ._base import SQLBaseStore, Table
+from ._base import SQLBaseStore
+from synapse.api.constants import Membership
-import collections
-import json
import logging
logger = logging.getLogger(__name__)
@@ -34,14 +29,15 @@ class RoomMemberStore(SQLBaseStore):
def _store_room_member_txn(self, txn, event):
"""Store a room member in the database.
"""
- domain = self.hs.parse_userid(event.target_user_id).domain
+ target_user_id = event.state_key
+ domain = self.hs.parse_userid(target_user_id).domain
self._simple_insert_txn(
txn,
"room_memberships",
{
"event_id": event.event_id,
- "user_id": event.target_user_id,
+ "user_id": target_user_id,
"sender": event.user_id,
"room_id": event.room_id,
"membership": event.membership,
@@ -145,7 +141,28 @@ class RoomMemberStore(SQLBaseStore):
rows = yield self._execute_and_decode(sql, *where_values)
- logger.debug("_get_members_query Got rows %s", rows)
+ # logger.debug("_get_members_query Got rows %s", rows)
results = [self._parse_event_from_row(r) for r in rows]
defer.returnValue(results)
+
+ @defer.inlineCallbacks
+ def do_users_share_a_room(self, user_list):
+ """ Checks whether a list of users share a room.
+ """
+ user_list_clause = " OR ".join(["m.user_id = ?"] * len(user_list))
+ sql = (
+ "SELECT m.room_id FROM room_memberships as m "
+ "INNER JOIN current_state_events as c "
+ "ON m.event_id = c.event_id "
+ "WHERE m.membership = 'join' "
+ "AND (%(clause)s) "
+ "GROUP BY m.room_id HAVING COUNT(m.room_id) = ?"
+ ) % {"clause": user_list_clause}
+
+ args = user_list
+ args.append(len(user_list))
+
+ rows = yield self._execute(None, sql, *args)
+
+ defer.returnValue(len(rows) > 0)
diff --git a/synapse/storage/schema/im.sql b/synapse/storage/schema/im.sql
index ea04261ff0..e92f21ef3b 100644
--- a/synapse/storage/schema/im.sql
+++ b/synapse/storage/schema/im.sql
@@ -22,9 +22,15 @@ CREATE TABLE IF NOT EXISTS events(
content TEXT NOT NULL,
unrecognized_keys TEXT,
processed BOOL NOT NULL,
+ outlier BOOL NOT NULL,
CONSTRAINT ev_uniq UNIQUE (event_id)
);
+CREATE INDEX IF NOT EXISTS events_event_id ON events (event_id);
+CREATE INDEX IF NOT EXISTS events_stream_ordering ON events (stream_ordering);
+CREATE INDEX IF NOT EXISTS events_topological_ordering ON events (topological_ordering);
+CREATE INDEX IF NOT EXISTS events_room_id ON events (room_id);
+
CREATE TABLE IF NOT EXISTS state_events(
event_id TEXT NOT NULL,
room_id TEXT NOT NULL,
@@ -33,6 +39,12 @@ CREATE TABLE IF NOT EXISTS state_events(
prev_state TEXT
);
+CREATE UNIQUE INDEX IF NOT EXISTS state_events_event_id ON state_events (event_id);
+CREATE INDEX IF NOT EXISTS state_events_room_id ON state_events (room_id);
+CREATE INDEX IF NOT EXISTS state_events_type ON state_events (type);
+CREATE INDEX IF NOT EXISTS state_events_state_key ON state_events (state_key);
+
+
CREATE TABLE IF NOT EXISTS current_state_events(
event_id TEXT NOT NULL,
room_id TEXT NOT NULL,
@@ -41,6 +53,11 @@ CREATE TABLE IF NOT EXISTS current_state_events(
CONSTRAINT curr_uniq UNIQUE (room_id, type, state_key) ON CONFLICT REPLACE
);
+CREATE INDEX IF NOT EXISTS curr_events_event_id ON current_state_events (event_id);
+CREATE INDEX IF NOT EXISTS current_state_events_room_id ON current_state_events (room_id);
+CREATE INDEX IF NOT EXISTS current_state_events_type ON current_state_events (type);
+CREATE INDEX IF NOT EXISTS current_state_events_state_key ON current_state_events (state_key);
+
CREATE TABLE IF NOT EXISTS room_memberships(
event_id TEXT NOT NULL,
user_id TEXT NOT NULL,
@@ -49,6 +66,10 @@ CREATE TABLE IF NOT EXISTS room_memberships(
membership TEXT NOT NULL
);
+CREATE INDEX IF NOT EXISTS room_memberships_event_id ON room_memberships (event_id);
+CREATE INDEX IF NOT EXISTS room_memberships_room_id ON room_memberships (room_id);
+CREATE INDEX IF NOT EXISTS room_memberships_user_id ON room_memberships (user_id);
+
CREATE TABLE IF NOT EXISTS feedback(
event_id TEXT NOT NULL,
feedback_type TEXT,
@@ -77,5 +98,6 @@ CREATE TABLE IF NOT EXISTS rooms(
CREATE TABLE IF NOT EXISTS room_hosts(
room_id TEXT NOT NULL,
- host TEXT NOT NULL
+ host TEXT NOT NULL,
+ CONSTRAINT room_hosts_uniq UNIQUE (room_id, host) ON CONFLICT IGNORE
);
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index 7460bf28d7..ac887e2957 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -37,10 +37,8 @@ from twisted.internet import defer
from ._base import SQLBaseStore
from synapse.api.errors import SynapseError
-from synapse.api.constants import Membership
from synapse.util.logutils import log_function
-import json
import logging
@@ -177,6 +175,7 @@ class StreamStore(SQLBaseStore):
"((room_id IN (%(current)s)) OR "
"(event_id IN (%(invites)s))) "
"AND e.stream_ordering > ? AND e.stream_ordering < ? "
+ "AND e.outlier = 0 "
"ORDER BY stream_ordering ASC LIMIT %(limit)d "
) % {
"current": current_room_membership_sql,
@@ -224,7 +223,7 @@ class StreamStore(SQLBaseStore):
sql = (
"SELECT * FROM events "
- "WHERE room_id = ? AND %(bounds)s "
+ "WHERE outlier = 0 AND room_id = ? AND %(bounds)s "
"ORDER BY topological_ordering %(order)s, stream_ordering %(order)s %(limit)s "
) % {"bounds": bounds, "order": order, "limit": limit_str}
@@ -249,15 +248,14 @@ class StreamStore(SQLBaseStore):
)
@defer.inlineCallbacks
- def get_recent_events_for_room(self, room_id, limit, with_feedback=False):
+ def get_recent_events_for_room(self, room_id, limit, end_token,
+ with_feedback=False):
# TODO (erikj): Handle compressed feedback
- end_token = yield self.get_room_events_max_id()
-
sql = (
"SELECT * FROM events "
"WHERE room_id = ? AND stream_ordering <= ? "
- "ORDER BY topological_ordering, stream_ordering DESC LIMIT ? "
+ "ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ? "
)
rows = yield self._execute_and_decode(
|
