Merge branch 'correct-sendtxn' into pushers

Make clippy happy (needless-return, etc.)

See merge request famedly/conduit!23

.gitignore

@@ -2,3 +2,4 @@
 **/*.rs.bk
 
 Rocket.toml
+conduit.toml

.gitlab/issue_templates/Issue Template.md

@@ -0,0 +1,15 @@
+# Headline
+
+### Description
+
+
+
+
+
+
+
+
+
+
+
+/label ~conduit

Cargo.toml

@@ -4,7 +4,7 @@ description = "A Matrix homeserver written in Rust"
 license = "Apache-2.0"
 authors = ["timokoesters <timo@koesters.xyz>"]
 homepage = "https://conduit.rs"
-repository = "https://git.koesters.xyz/timo/conduit"
+repository = "https://gitlab.com/famedly/conduit"
 readme = "README.md"
 version = "0.1.0"
 edition = "2018"
@@ -14,51 +14,64 @@ edition = "2018"
 [dependencies]
 # Used to handle requests
 # TODO: This can become optional as soon as proper configs are supported
-rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "1f1f44f336e5a172361fc1860461bb03667b1ed2", features = ["tls"] } # Used to handle requests
+rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "93e62c86eddf7cc9a7fc40b044182f83f0d7d92a", features = ["tls"] } # Used to handle requests
 #rocket = { git = "https://github.com/timokoesters/Rocket.git", branch = "empty_parameters", default-features = false, features = ["tls"] }
 
 # Used for matrix spec type definitions and helpers
-ruma = { git = "https://github.com/ruma/ruma", features = ["rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks", "unstable-exhaustive-types"], rev = "e8882fe8142d7b55ed4c8ccc6150946945f9e237" }
-# ruma = { git = "https://github.com/DevinR528/ruma", features = ["rand", "client-api", "federation-api", "unstable-exhaustive-types", "unstable-pre-spec", "unstable-synapse-quirks"], branch = "unstable-join" }
-# ruma = { path = "../ruma/ruma", features = ["unstable-exhaustive-types", "rand", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks"] }
+#ruma = { git = "https://github.com/ruma/ruma", features = ["rand", "appservice-api", "client-api", "federation-api", "unstable-pre-spec", "unstable-synapse-quirks", "unstable-exhaustive-types"], rev = "0a10afe6dacc2b7a50a8002c953d10b7fb4e37bc" }
+# ruma = { git = "https://github.com/DevinR528/ruma", features = ["rand", "client-api", "federation-api", "unstable-exhaustive-types", "unstable-pre-spec", "unstable-synapse-quirks"], branch = "verified-export" }
+ruma = { path = "../ruma/ruma", features = ["unstable-exhaustive-types", "rand", "client-api", "federation-api", "push-gateway-api", "unstable-pre-spec", "unstable-synapse-quirks"] }
 
 # Used when doing state resolution
-# state-res = { git = "https://github.com/timokoesters/state-res", branch = "spec-comp", features = ["unstable-pre-spec"] }
-state-res = { git = "https://github.com/ruma/state-res", branch = "timo-spec-comp", features = ["unstable-pre-spec", "gen-eventid"] }
-# state-res = { path = "../../state-res", features = ["unstable-pre-spec", "gen-eventid"] }
+# state-res = { git = "https://github.com/timokoesters/state-res", branch = "timo-spec-comp", features = ["unstable-pre-spec"] }
+# TODO: remove the gen-eventid feature
+#state-res = { git = "https://github.com/ruma/state-res", branch = "main", features = ["unstable-pre-spec", "gen-eventid"] }
+# state-res = { git = "https://github.com/ruma/state-res", rev = "791c66d73cf064d09db0cdf767d5fef43a343425", features = ["unstable-pre-spec", "gen-eventid"] }
+state-res = { path = "../state-res", features = ["unstable-pre-spec", "gen-eventid"] }
 
 # Used for long polling and federation sender, should be the same as rocket::tokio
-tokio = { version = "0.2.23" }
+tokio = "1.2.0"
 # Used for storing data permanently
 sled = { version = "0.34.6", default-features = false }
 # Used for emitting log entries
-log = "0.4.11"
+log = "0.4.14"
 # Used for rocket<->ruma conversions
-http = "0.2.1"
+http = "0.2.3"
 # Used to find data directory for default db path
 directories = "3.0.1"
-# Used for number types for ruma
-js_int = "0.1.9"
 # Used for ruma wrapper
-serde_json = { version = "1.0.60", features = ["raw_value"] }
+serde_json = { version = "1.0.64", features = ["raw_value"] }
+# Used for appservice registration files
+serde_yaml = "0.8.17"
 # Used for pdu definition
-serde = "1.0.117"
+serde = "1.0.123"
 # Used for secure identifiers
-rand = "0.7.3"
+rand = "0.8.3"
 # Used to hash passwords
 rust-argon2 = "0.8.3"
 # Used to send requests
-reqwest = "0.10.9"
+reqwest = { version = "0.11.1" }
 # Used for conduit::Error type
-thiserror = "1.0.22"
+thiserror = "1.0.24"
 # Used to generate thumbnails for images
-image = { version = "0.23.12", default-features = false, features = ["jpeg", "png", "gif"] }
+image = { version = "0.23.14", default-features = false, features = ["jpeg", "png", "gif"] }
 # Used to encode server public key
 base64 = "0.13.0"
 # Used when hashing the state
-ring = "0.16.19"
+ring = "0.16.20"
 # Used when querying the SRV record of other servers
-trust-dns-resolver = "0.19.6"
+trust-dns-resolver = "0.20.0"
+# Used to find matching events for appservices
+regex = "1.4.3"
+# jwt jsonwebtokens
+jsonwebtoken = "7.2.0"
+# Performance measurements
+tracing = "0.1.25"
+opentelemetry = "0.12.0"
+tracing-subscriber = "0.2.16"
+tracing-opentelemetry = "0.11.0"
+opentelemetry-jaeger = "0.11.0"
+pretty_env_logger = "0.4.0"
 
 [features]
 default = ["conduit_bin"]
@@ -73,3 +86,26 @@ required-features = ["conduit_bin"]
 [lib]
 name = "conduit"
 path = "src/lib.rs"
+
+[package.metadata.deb]
+name = "matrix-conduit"
+maintainer = "Paul van Tilburg <paul@luon.net>"
+copyright = "2020, Timo Kösters <timo@koesters.xyz>"
+license-file = ["LICENSE", "3"]
+depends = "$auto, ca-certificates"
+extended-description = """\
+A fast Matrix homeserver that is optimized for smaller, personal servers, \
+instead of a server that has high scalability."""
+section = "net"
+priority = "optional"
+assets = [
+  ["debian/env.local", "etc/matrix-conduit/local", "644"],
+  ["debian/README.Debian", "usr/share/doc/matrix-conduit/", "644"],
+  ["README.md", "usr/share/doc/matrix-conduit/", "644"],
+  ["target/release/conduit", "usr/sbin/matrix-conduit", "755"],
+]
+conf-files = [
+  "/etc/matrix-conduit/local"
+]
+maintainer-scripts = "debian/"
+systemd-units = { unit-name = "matrix-conduit" }

DEPLOY.md

@@ -0,0 +1,164 @@
+# Deploying Conduit
+
+## Getting help
+
+If you run into any problems while setting up Conduit, write an email to `support@conduit.rs`, ask us in `#conduit:matrix.org` or [open an issue on GitLab](https://gitlab.com/famedly/conduit/-/issues/new).
+
+## Installing Conduit
+
+You have to download the binary that fits your machine. Run `uname -m` to see
+what you need. Now copy the right url:
+- x84_64: `https://conduit.rs/master/x86_64/conduit-bin`
+- armv7: `https://conduit.rs/master/armv7/conduit-bin`
+- armv8: `https://conduit.rs/master/armv8/conduit-bin`
+- arm: `https://conduit.rs/master/arm/conduit-bin`
+
+```bash
+$ sudo wget -O /usr/local/bin/matrix-conduit <url>
+$ sudo chmod +x /usr/local/bin/matrix-conduit
+```
+
+
+## Setting up a systemd service
+
+Now we'll set up a systemd service for Conduit, so it's easy to start/stop
+Conduit and set it to autostart when your server reboots. Simply paste the
+default systemd service you can find below into
+`/etc/systemd/system/conduit.service`.
+
+```systemd
+[Unit]
+Description=Conduit Matrix Server
+After=network.target
+
+[Service]
+Environment="CONDUIT_CONFIG=/etc/matrix-conduit/conduit.toml"
+User=root
+Group=root
+Restart=always
+ExecStart=/usr/local/bin/matrix-conduit
+
+[Install]
+WantedBy=multi-user.target
+```
+
+Finally, run
+```bash
+$ sudo systemctl daemon-reload
+```
+
+
+## Creating the Conduit configuration file
+
+Now we need to create the Conduit's config file in `/etc/matrix-conduit/conduit.toml`. Paste this in **and take a moment to read it. You need to change at least the server name.**
+```toml
+[global]
+# The server_name is the name of this server. It is used as a suffix for user
+# and room ids. Examples: matrix.org, conduit.rs
+# The Conduit server needs to be reachable at https://your.server.name/ on port
+# 443 (client-server) and 8448 (federation) OR you can create /.well-known
+# files to redirect requests. See
+# https://matrix.org/docs/spec/client_server/latest#get-well-known-matrix-client
+# and https://matrix.org/docs/spec/server_server/r0.1.4#get-well-known-matrix-server
+# for more information
+
+# YOU NEED TO EDIT THIS
+#server_name = "your.server.name"
+
+# This is the only directory where Conduit will save its data
+database_path = "/var/lib/matrix-conduit/conduit_db"
+
+# The port Conduit will be running on. You need to set up a reverse proxy in
+# your web server (e.g. apache or nginx), so all requests to /_matrix on port
+# 443 and 8448 will be forwarded to the Conduit instance running on this port
+port = 6167
+
+# Max size for uploads
+max_request_size = 20_000_000 # in bytes
+
+# Disabling registration means no new users will be able to register on this server
+allow_registration = false
+
+# Disable encryption, so no new encrypted rooms can be created
+# Note: existing rooms will continue to work
+allow_encryption = true
+allow_federation = true
+
+#cache_capacity = 1073741824 # in bytes, 1024 * 1024 * 1024
+#max_concurrent_requests = 4 # How many requests Conduit sends to other servers at the same time
+#workers = 4 # default: cpu core count * 2
+
+address = "127.0.0.1" # This makes sure Conduit can only be reached using the reverse proxy
+```
+
+
+## Setting up the Reverse Proxy
+
+This depends on whether you use Apache, Nginx or another web server.
+
+### Apache
+
+Create `/etc/apache2/sites-enabled/050-conduit.conf` and copy-and-paste this:
+```
+Listen 8448
+
+<VirtualHost *:443 *:8448>
+
+ServerName your.server.name # EDIT THIS
+
+AllowEncodedSlashes NoDecode
+ProxyPass /_matrix/ http://localhost:6167/
+ProxyPassReverse /_matrix/ http://localhost:6167/
+
+Include /etc/letsencrypt/options-ssl-apache.conf
+SSLCertificateFile /etc/letsencrypt/live/your.server.name/fullchain.pem # EDIT THIS
+SSLCertificateKeyFile /etc/letsencrypt/live/your.server.name/privkey.pem # EDIT THIS
+</VirtualHost>
+```
+
+**You need to make some edits again.** When you are done, run
+```bash
+$ sudo systemctl reload apache2
+```
+
+
+### Nginx
+
+If you use Nginx and not Apache, add the following server section inside the
+http section of `/etc/nginx/nginx.conf`
+```
+server {
+    listen 443;
+    listen 8448;
+    server_name your.server.name; # EDIT THIS
+
+    location /_matrix/ {
+        proxy_pass http://localhost:6167/_matrix/;
+    }
+}
+```
+**You need to make some edits again.** When you are done, run
+```bash
+$ sudo systemctl reload nginx
+```
+
+
+## SSL Certificate
+
+The easiest way to get an SSL certificate, if you don't have one already, is to install `certbot` and run this:
+```bash
+$ sudo certbot -d your.server.name
+```
+
+
+## You're done!
+
+Now you can start Conduit with:
+```bash
+$ sudo systemctl start conduit
+```
+
+Set it to start automatically when your system boots with:
+```bash
+$ sudo systemctl enable conduit
+```

DEPLOY_FROM_SOURCE.md

@@ -1,103 +0,0 @@
-# Deploy from source
-
-## Prerequisites
-
-Make sure you have `libssl-dev` and `pkg-config` installed and the [rust toolchain](https://rustup.rs) is available on at least on user.
-
-
-## Install Conduit
-
-```bash
-$ sudo useradd -m conduit
-$ sudo -u conduit cargo install --git "https://git.koesters.xyz/timo/conduit.git"
-```
-
-
-## Setup systemd service
-
-In this guide, we set up a systemd service for Conduit, so it's easy to start, stop Conduit and set it to autostart when your server reboots. Paste the default systemd service below and configure it to fit your setup (in /etc/systemd/system/conduit.service).
-
-```systemd
-[Unit]
-Description=Conduit
-After=network.target
-
-[Service]
-Environment="ROCKET_SERVER_NAME=YOURSERVERNAME.HERE" # EDIT THIS
-
-Environment="ROCKET_PORT=14004" # Reverse proxy port
-
-#Environment="ROCKET_MAX_REQUEST_SIZE=20000000" # in bytes
-#Environment="ROCKET_REGISTRATION_DISABLED=true"
-#Environment="ROCKET_ENCRYPTION_DISABLED=true"
-#Environment="ROCKET_FEDERATION_ENABLED=true"
-#Environment="ROCKET_LOG=normal" # Detailed logging
-
-Environment="ROCKET_ENV=production"
-User=conduit
-Group=conduit
-Type=simple
-Restart=always
-ExecStart=/home/conduit/.cargo/bin/conduit
-
-[Install]
-WantedBy=multi-user.target
-```
-
-Finally, run
-```bash
-$ sudo systemctl daemon-reload
-```
-
-
-## Setup Reverse Proxy
-
-This depends on whether you use Apache, Nginx or something else. For Apache it looks like this (in /etc/apache2/sites-enabled/050-conduit.conf):
-```
-<VirtualHost *:443>
-
-ServerName conduit.koesters.xyz # EDIT THIS
-
-AllowEncodedSlashes NoDecode
-
-ServerAlias conduit.koesters.xyz # EDIT THIS
-
-ProxyPreserveHost On
-ProxyRequests off
-AllowEncodedSlashes NoDecode
-ProxyPass / http://localhost:14004/ nocanon
-ProxyPassReverse / http://localhost:14004/ nocanon
-
-Include /etc/letsencrypt/options-ssl-apache.conf
-
-# EDIT THESE:
-SSLCertificateFile /etc/letsencrypt/live/conduit.koesters.xyz/fullchain.pem
-SSLCertificateKeyFile /etc/letsencrypt/live/conduit.koesters.xyz/privkey.pem
-</VirtualHost>
-```
-
-Then run
-```bash
-$ sudo systemctl reload apache2
-```
-
-
-## SSL Certificate
-
-The easiest way to get an SSL certificate for the domain is to install `certbot` and run this:
-```bash
-$ sudo certbot -d conduit.koesters.xyz
-```
-
-
-## You're done!
-
-Now you can start Conduit with
-```bash
-$ sudo systemctl start conduit
-```
-
-and set it to start automatically when your system boots with
-```bash
-$ sudo systemctl enable conduit
-```

README.md

@@ -17,13 +17,12 @@ example) and register on the `https://conduit.koesters.xyz` homeserver.
 
 #### How can I deploy my own?
 
-##### From source
+##### Deploy
 
-Clone the repo, build it with `cargo build --release` and call the binary
-(target/release/conduit) from somewhere like a systemd script. [Read
-more](DEPLOY_FROM_SOURCE.md)
+Download or compile a conduit binary and call it from somewhere like a systemd script. [Read
+more](DEPLOY.md)
 
-##### Using Docker
+##### Deploy using Docker
 
 Pull and run the docker image with
 

Rocket-example.toml

@@ -1,31 +0,0 @@
-[global]
-# The name of this server
-# Note: If server name != hostname, you need a .well-known file for federation
-# to work
-server_name = "your.server.name"
-
-port = 14004
-
-# Max size for uploads
-#max_request_size = 20_000_000 # in bytes, ~20 MB
-
-# Disable registration. No new users will be able to register on this server
-#registration_disabled = true
-
-# Disable encryption, so no new encrypted rooms can be created
-# Note: existing rooms will continue to work
-#encryption_disabled = true
-
-#federation_enabled = true
-
-# Default path is in this user's data
-#database_path = "/home/timo/MyConduitServer"
-
-# You should probably leave this at 0.0.0.0
-address = "0.0.0.0"
-
-# TLS support
-# Note: Not necessary when using a reverse proxy:
-#[global.tls]
-#certs = "/etc/letsencrypt/live/your.server.name/fullchain.pem"
-#key = "/etc/letsencrypt/live/your.server.name/privkey.pem"

conduit-example.toml

@@ -0,0 +1,40 @@
+[global]
+# The server_name is the name of this server. It is used as a suffix for user
+# and room ids. Examples: matrix.org, conduit.rs
+# The Conduit server needs to be reachable at https://your.server.name/ on port
+# 443 (client-server) and 8448 (federation) OR you can create /.well-known
+# files to redirect requests. See
+# https://matrix.org/docs/spec/client_server/latest#get-well-known-matrix-client
+# and https://matrix.org/docs/spec/server_server/r0.1.4#get-well-known-matrix-server
+# for more information
+
+# YOU NEED TO EDIT THIS
+#server_name = "your.server.name"
+
+# This is the only directly where Conduit will save its data
+database_path = "/var/lib/conduit/conduit.db"
+
+# The port Conduit will be running on. You need to set up a reverse proxy in
+# your web server (e.g. apache or nginx), so all requests to /_matrix on port
+# 443 and 8448 will be forwarded to the Conduit instance running on this port
+port = 6167
+
+# Max size for uploads
+max_request_size = 20_000_000 # in bytes
+
+# Disable registration. No new users will be able to register on this server
+#allow_registration = false
+
+# Disable encryption, so no new encrypted rooms can be created
+# Note: existing rooms will continue to work
+#allow_encryption = false
+#allow_federation = false
+
+# Enable jaeger to support monitoring and troubleshooting through jaeger
+#allow_jaeger = false
+
+#cache_capacity = 1073741824 # in bytes, 1024 * 1024 * 1024
+#max_concurrent_requests = 4 # How many requests Conduit sends to other servers at the same time
+#workers = 4 # default: cpu core count * 2
+
+address = "127.0.0.1" # This makes sure Conduit can only be reached using the reverse proxy

debian/README.Debian

@@ -0,0 +1,29 @@
+Conduit for Debian
+==================
+
+Configuration
+-------------
+
+When installed, Debconf handles the configuration of the homeserver (host)name,
+the address and port it listens on. These configuration variables end up in
+/etc/matrix-conduit/debian.
+
+You can tweak more detailed settings by uncommenting and setting the variables
+in /etc/matrix-conduit/local. This involves settings such as the maximum file
+size for download/upload, enabling federation, etc.
+
+Running
+-------
+
+The package uses the matrix-conduit.service systemd unit file to start and
+stop Conduit. It loads the configuration files mentioned above to set up the
+environment before running the server.
+
+This package assumes by default that Conduit is placed behind a reverse proxy
+such as Apache or nginx. This default deployment entails just listening on
+127.0.0.1 and the free port 14004 and is reachable via a client using the URL
+http://localhost:14004.
+
+At a later stage this packaging may support also setting up TLS and running
+stand-alone.  In this case, however, you need to set up some certificates and
+renewal, for it to work properly.

debian/config

@@ -0,0 +1,17 @@
+#!/bin/sh
+set -e
+
+# Source debconf library.
+. /usr/share/debconf/confmodule
+
+# Ask for the Matrix homeserver name, address and port.
+db_input high matrix-conduit/hostname || true
+db_go
+
+db_input low matrix-conduit/address || true
+db_go
+
+db_input medium matrix-conduit/port || true
+db_go
+
+exit 0

debian/env.local

@@ -0,0 +1,33 @@
+# Conduit homeserver local configuration
+#
+# Conduit is an application based on the Rocket web framework.
+# Configuration of Conduit happens via Debconf (see the resulting config in
+# `/etc/matrix-conduit/debian`) and optionally by uncommenting and tweaking the
+# variables in this file below.
+
+# The maximum size of a Matrix HTTP requests in bytes.
+#
+# This mostly affects the size of files that can be downloaded/uploaded.
+# It defaults to 20971520 (20MB).
+#ROCKET_MAX_REQUEST_SIZE=20971520
+
+# Whether user registration is allowed.
+#
+# User registration is not disabled by default.
+#ROCKET_REGISTRATION_DISABLED=false
+
+# Whether encryption is enabled.
+#
+# (End-to-end) encryption is not disabled by default.
+#ROCKET_ENCRYPTION_DISABLED=false
+
+# Whether federation with other Matrix servers is enabled.
+#
+# Federation is not enabled by default; it is still experimental.
+#ROCKET_FEDERATION_ENABLED=false
+
+# The log level of the homeserver.
+#
+# The log level is "critical" by default.
+# Allowed values are: "off", "normal", "debug", "critical"
+#ROCKET_LOG="critical"

debian/matrix-conduit.service

@@ -0,0 +1,49 @@
+[Unit]
+Description=Conduit Matrix homeserver
+After=network.target
+
+[Service]
+User=_matrix-conduit
+Group=_matrix-conduit
+Type=simple
+
+AmbientCapabilities=
+CapabilityBoundingSet=
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+NoNewPrivileges=yes
+ProtectClock=yes
+ProtectControlGroups=yes
+ProtectHome=yes
+ProtectHostname=yes
+ProtectKernelLogs=yes
+ProtectKernelModules=yes
+ProtectKernelTunables=yes
+ProtectSystem=strict
+PrivateDevices=yes
+PrivateMounts=yes
+PrivateTmp=yes
+PrivateUsers=yes
+RemoveIPC=yes
+RestrictAddressFamilies=AF_INET AF_INET6
+RestrictNamespaces=yes
+RestrictRealtime=yes
+RestrictSUIDSGID=yes
+SystemCallArchitectures=native
+SystemCallFilter=@system-service
+SystemCallErrorNumber=EPERM
+StateDirectory=matrix-conduit
+
+Environment="ROCKET_ENV=production"
+Environment="ROCKET_DATABASE_PATH=/var/lib/matrix-conduit"
+EnvironmentFile=/etc/matrix-conduit/debian
+EnvironmentFile=/etc/matrix-conduit/local
+
+ExecStart=/usr/sbin/matrix-conduit
+Restart=on-failure
+RestartSec=10
+StartLimitInterval=1m
+StartLimitBurst=5
+
+[Install]
+WantedBy=multi-user.target

debian/postinst

@@ -0,0 +1,73 @@
+#!/bin/sh
+set -e
+
+. /usr/share/debconf/confmodule
+
+CONDUIT_CONFIG_PATH=/etc/matrix-conduit
+CONDUIT_CONFIG_FILE="$CONDUIT_CONFIG_PATH/debian"
+CONDUIT_DATABASE_PATH=/var/lib/matrix-conduit
+
+case "$1" in
+  configure)
+    # Create the `_matrix-conduit` user if it does not exist yet.
+    if ! getent passwd _matrix-conduit > /dev/null ; then
+      echo 'Adding system user for the Conduit Matrix homeserver' 1>&2
+      adduser --system --group --quiet \
+        --home $CONDUIT_DATABASE_PATH \
+        --disabled-login \
+        --force-badname \
+        _matrix-conduit
+    fi
+
+    # Create the database path if it does not exist yet.
+    if [ ! -d "$CONDUIT_DATABASE_PATH" ]; then
+      mkdir -p "$CONDUIT_DATABASE_PATH"
+      chown _matrix-conduit "$CONDUIT_DATABASE_PATH"
+    fi
+
+    # Write the debconf values in the config.
+    db_get matrix-conduit/hostname
+    ROCKET_SERVER_NAME="$RET"
+    db_get matrix-conduit/address
+    ROCKET_ADDRESS="$RET"
+    db_get matrix-conduit/port
+    ROCKET_PORT="$RET"
+    cat >"$CONDUIT_CONFIG_FILE" << EOF
+# Conduit homeserver Debian configuration
+#
+# Conduit is an application based on the Rocket web framework.
+# Configuration of Conduit happens via Debconf (of which the resulting config
+# is in this file) and optionally by uncommenting and tweaking the variables in
+# /etc/matrix-conduit/local.
+
+# THIS FILE IS GENERATED BY DEBCONF AND WILL BE OVERRIDDEN!
+#
+# Please make changes by running:
+#
+# \$ dpkg-reconfigure matrix-conduit
+#
+# or by providing overriding changes in /etc/matrix-conduit/local.
+
+# The server (host)name of the Matrix homeserver.
+#
+# This is the hostname the homeserver will be reachable at via a client.
+ROCKET_SERVER_NAME="$ROCKET_SERVER_NAME"
+
+# The address the Matrix homeserver listens on.
+#
+# By default the server listens on address 0.0.0.0. Change this to 127.0.0.1 to
+# only listen on the localhost when using a reverse proxy.
+ROCKET_ADDRESS="$ROCKET_ADDRESS"
+
+# The port of the Matrix homeserver.
+#
+# This port is could be any available port if accessed by a reverse proxy.
+# By default the server listens on port 8000.
+ROCKET_PORT="$ROCKET_PORT"
+
+# THIS FILE IS GENERATED BY DEBCONF AND WILL BE OVERRIDDEN!
+EOF
+    ;;
+esac
+
+#DEBHELPER#

debian/postrm

@@ -0,0 +1,22 @@
+#!/bin/sh
+set -e
+
+CONDUIT_CONFIG_PATH=/etc/matrix-conduit
+CONDUIT_DATABASE_PATH=/var/lib/matrix-conduit
+
+case $1 in
+  purge)
+    # Per https://www.debian.org/doc/debian-policy/ch-files.html#behavior
+    # "configuration files must be preserved when the package is removed, and
+    #  only deleted when the package is purged."
+    if [ -d "$CONDUIT_CONFIG_PATH" ]; then
+      rm -r "$CONDUIT_CONFIG_PATH"
+    fi
+
+    if [ -d "$CONDUIT_DATABASE_PATH" ]; then
+      rm -r "$CONDUIT_DATABASE_PATH"
+    fi
+    ;;
+esac
+
+#DEBHELPER#

debian/templates

@@ -0,0 +1,21 @@
+Template: matrix-conduit/hostname
+Type: string
+Default: localhost
+Description: The server (host)name of the Matrix homeserver
+ This is the hostname the homeserver will be reachable at via a client.
+ .
+ If set to "localhost", you can connect with a client locally and clients
+ from other hosts and also other homeservers will not be able to reach you!
+
+Template: matrix-conduit/address
+Type: string
+Default: 127.0.0.1
+Description: The listen address of the Matrix homeserver
+ This is the address the homeserver will listen on. Leave it set to 127.0.0.1
+ when using a reverse proxy.
+
+Template: matrix-conduit/port
+Type: string
+Default: 14004
+Description: The port of the Matrix homeserver
+ This port is most often just accessed by a reverse proxy.

rustfmt.toml

@@ -1 +1 @@
-merge_imports = true
+imports_granularity="Crate"

src/appservice_server.rs

@@ -0,0 +1,104 @@
+use crate::{utils, Error, Result};
+use http::header::{HeaderValue, CONTENT_TYPE};
+use log::warn;
+use ruma::api::OutgoingRequest;
+use std::{
+    convert::{TryFrom, TryInto},
+    fmt::Debug,
+    time::Duration,
+};
+
+pub async fn send_request<T: OutgoingRequest>(
+    globals: &crate::database::globals::Globals,
+    registration: serde_yaml::Value,
+    request: T,
+) -> Result<T::IncomingResponse>
+where
+    T: Debug,
+{
+    let destination = registration.get("url").unwrap().as_str().unwrap();
+    let hs_token = registration.get("hs_token").unwrap().as_str().unwrap();
+
+    let mut http_request = request
+        .try_into_http_request(&destination, Some(""))
+        .unwrap();
+
+    let mut parts = http_request.uri().clone().into_parts();
+    let old_path_and_query = parts.path_and_query.unwrap().as_str().to_owned();
+    let symbol = if old_path_and_query.contains('?') {
+        "&"
+    } else {
+        "?"
+    };
+
+    parts.path_and_query = Some(
+        (old_path_and_query + symbol + "access_token=" + hs_token)
+            .parse()
+            .unwrap(),
+    );
+    *http_request.uri_mut() = parts.try_into().expect("our manipulation is always valid");
+
+    http_request.headers_mut().insert(
+        CONTENT_TYPE,
+        HeaderValue::from_str("application/json").unwrap(),
+    );
+
+    let mut reqwest_request = reqwest::Request::try_from(http_request)
+        .expect("all http requests are valid reqwest requests");
+
+    *reqwest_request.timeout_mut() = Some(Duration::from_secs(30));
+
+    let url = reqwest_request.url().clone();
+    let reqwest_response = globals.reqwest_client().execute(reqwest_request).await;
+
+    // Because reqwest::Response -> http::Response is complicated:
+    match reqwest_response {
+        Ok(mut reqwest_response) => {
+            let status = reqwest_response.status();
+            let mut http_response = http::Response::builder().status(status);
+            let headers = http_response.headers_mut().unwrap();
+
+            for (k, v) in reqwest_response.headers_mut().drain() {
+                if let Some(key) = k {
+                    headers.insert(key, v);
+                }
+            }
+
+            let status = reqwest_response.status();
+
+            let body = reqwest_response
+                .bytes()
+                .await
+                .unwrap_or_else(|e| {
+                    warn!("server error: {}", e);
+                    Vec::new().into()
+                }) // TODO: handle timeout
+                .into_iter()
+                .collect::<Vec<_>>();
+
+            if status != 200 {
+                warn!(
+                    "Appservice returned bad response {} {}\n{}\n{:?}",
+                    destination,
+                    status,
+                    url,
+                    utils::string_from_bytes(&body)
+                );
+            }
+
+            let response = T::IncomingResponse::try_from(
+                http_response
+                    .body(body)
+                    .expect("reqwest body is valid http body"),
+            );
+            response.map_err(|_| {
+                warn!(
+                    "Appservice returned invalid response bytes {}\n{}",
+                    destination, url
+                );
+                Error::BadServerResponse("Server returned bad response.")
+            })
+        }
+        Err(e) => Err(e.into()),
+    }
+}

src/client_server/account.rs

@@ -40,6 +40,7 @@ const GUEST_NAME_LENGTH: usize = 10;
     feature = "conduit_bin",
     get("/_matrix/client/r0/register/available", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_register_available_route(
     db: State<'_, Database>,
     body: Ruma<get_username_availability::Request<'_>>,
@@ -82,11 +83,12 @@ pub async fn get_register_available_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/register", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn register_route(
     db: State<'_, Database>,
     body: Ruma<register::Request<'_>>,
 ) -> ConduitResult<register::Response> {
-    if db.globals.registration_disabled() {
+    if !db.globals.allow_registration() {
         return Err(Error::BadRequest(
             ErrorKind::Forbidden,
             "Registration has been disabled.",
@@ -139,18 +141,20 @@ pub async fn register_route(
         auth_error: None,
     };
 
-    if let Some(auth) = &body.auth {
-        let (worked, uiaainfo) =
-            db.uiaa
-                .try_auth(&user_id, "".into(), auth, &uiaainfo, &db.users, &db.globals)?;
-        if !worked {
+    if !body.from_appservice {
+        if let Some(auth) = &body.auth {
+            let (worked, uiaainfo) =
+                db.uiaa
+                    .try_auth(&user_id, "".into(), auth, &uiaainfo, &db.users, &db.globals)?;
+            if !worked {
+                return Err(Error::Uiaa(uiaainfo));
+            }
+        // Success!
+        } else {
+            uiaainfo.session = Some(utils::random_string(SESSION_ID_LENGTH));
+            db.uiaa.create(&user_id, "".into(), &uiaainfo)?;
             return Err(Error::Uiaa(uiaainfo));
         }
-    // Success!
-    } else {
-        uiaainfo.session = Some(utils::random_string(SESSION_ID_LENGTH));
-        db.uiaa.create(&user_id, "".into(), &uiaainfo)?;
-        return Err(Error::Uiaa(uiaainfo));
     }
 
     if missing_username {
@@ -237,10 +241,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // 2. Make conduit bot join
@@ -261,10 +262,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // 3. Power levels
@@ -298,10 +296,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // 4.1 Join Rules
@@ -318,10 +313,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // 4.2 History Visibility
@@ -340,10 +332,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // 4.3 Guest Access
@@ -360,10 +349,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // 6. Events implied by name and topic
@@ -382,10 +368,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         db.rooms.build_and_append_pdu(
@@ -401,10 +384,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // Room alias
@@ -426,10 +406,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         db.rooms.set_alias(&alias, Some(&room_id), &db.globals)?;
@@ -452,10 +429,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
         db.rooms.build_and_append_pdu(
             PduBuilder {
@@ -474,26 +448,16 @@ pub async fn register_route(
             },
             &user_id,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // Send welcome message
         db.rooms.build_and_append_pdu(
             PduBuilder {
                 event_type: EventType::RoomMessage,
-                content: serde_json::to_value(message::MessageEventContent::Text(
-                    message::TextMessageEventContent {
-                        body: "Thanks for trying out Conduit! This software is still in development, so expect many bugs and missing features. If you have federation enabled, you can join the Conduit chat room by typing `/join #conduit:matrix.org`. **Important: Please don't join any other Matrix rooms over federation without permission from the room's admins.** Some actions might trigger bugs in other server implementations, breaking the chat for everyone else.".to_owned(),
-                        formatted: Some(message::FormattedBody {
-                            format: message::MessageFormat::Html,
-                            body: "Thanks for trying out Conduit! This software is still in development, so expect many bugs and missing features. If you have federation enabled, you can join the Conduit chat room by typing <code>/join #conduit:matrix.org</code>. <strong>Important: Please don't join any other Matrix rooms over federation without permission from the room's admins.</strong> Some actions might trigger bugs in other server implementations, breaking the chat for everyone else.".to_owned(),
-                        }),
-                        relates_to: None,
-                        new_content: None,
-                    },
+                content: serde_json::to_value(message::MessageEventContent::text_html(
+                        "Thanks for trying out Conduit! This software is still in development, so expect many bugs and missing features. If you have federation enabled, you can join the Conduit chat room by typing `/join #conduit:matrix.org`. **Important: Please don't join any other Matrix rooms over federation without permission from the room's admins.** Some actions might trigger bugs in other server implementations, breaking the chat for everyone else.".to_owned(),
+                        "Thanks for trying out Conduit! This software is still in development, so expect many bugs and missing features. If you have federation enabled, you can join the Conduit chat room by typing <code>/join #conduit:matrix.org</code>. <strong>Important: Please don't join any other Matrix rooms over federation without permission from the room's admins.</strong> Some actions might trigger bugs in other server implementations, breaking the chat for everyone else.".to_owned(),
                 ))
                 .expect("event is valid, we just created it"),
                 unsigned: None,
@@ -502,10 +466,7 @@ pub async fn register_route(
             },
             &conduit_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
     }
 
@@ -532,6 +493,7 @@ pub async fn register_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/account/password", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn change_password_route(
     db: State<'_, Database>,
     body: Ruma<change_password::Request<'_>>,
@@ -570,16 +532,16 @@ pub async fn change_password_route(
 
     db.users.set_password(&sender_user, &body.new_password)?;
 
-    // TODO: Read logout_devices field when it's available and respect that, currently not supported in Ruma
-    // See: https://github.com/ruma/ruma/issues/107
-    // Logout all devices except the current one
-    for id in db
-        .users
-        .all_device_ids(&sender_user)
-        .filter_map(|id| id.ok())
-        .filter(|id| id != sender_device)
-    {
-        db.users.remove_device(&sender_user, &id)?;
+    if body.logout_devices {
+        // Logout all devices except the current one
+        for id in db
+            .users
+            .all_device_ids(&sender_user)
+            .filter_map(|id| id.ok())
+            .filter(|id| id != sender_device)
+        {
+            db.users.remove_device(&sender_user, &id)?;
+        }
     }
 
     db.flush().await?;
@@ -596,6 +558,7 @@ pub async fn change_password_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/account/whoami", data = "<body>")
 )]
+#[tracing::instrument(skip(body))]
 pub async fn whoami_route(body: Ruma<whoami::Request>) -> ConduitResult<whoami::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
     Ok(whoami::Response {
@@ -616,6 +579,7 @@ pub async fn whoami_route(body: Ruma<whoami::Request>) -> ConduitResult<whoami::
     feature = "conduit_bin",
     post("/_matrix/client/r0/account/deactivate", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn deactivate_route(
     db: State<'_, Database>,
     body: Ruma<deactivate::Request<'_>>,
@@ -677,10 +641,7 @@ pub async fn deactivate_route(
             },
             &sender_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
     }
 
@@ -696,3 +657,17 @@ pub async fn deactivate_route(
     }
     .into())
 }
+
+/*/
+#[cfg_attr(
+    feature = "conduit_bin",
+    get("/_matrix/client/r0/account/3pid", data = "<body>")
+)]
+pub async fn third_party_route(
+    body: Ruma<account::add_3pid::Request<'_>>,
+) -> ConduitResult<account::add_3pid::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    Ok(account::add_3pid::Response::default().into())
+}
+*/

src/client_server/alias.rs

@@ -1,7 +1,9 @@
 use super::State;
-use crate::{server_server, ConduitResult, Database, Error, Ruma};
+use crate::{ConduitResult, Database, Error, Ruma};
+use regex::Regex;
 use ruma::{
     api::{
+        appservice,
         client::{
             error::ErrorKind,
             r0::alias::{create_alias, delete_alias, get_alias},
@@ -18,6 +20,7 @@ use rocket::{delete, get, put};
     feature = "conduit_bin",
     put("/_matrix/client/r0/directory/room/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn create_alias_route(
     db: State<'_, Database>,
     body: Ruma<create_alias::Request<'_>>,
@@ -38,6 +41,7 @@ pub async fn create_alias_route(
     feature = "conduit_bin",
     delete("/_matrix/client/r0/directory/room/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn delete_alias_route(
     db: State<'_, Database>,
     body: Ruma<delete_alias::Request<'_>>,
@@ -53,6 +57,7 @@ pub async fn delete_alias_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/directory/room/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_alias_route(
     db: State<'_, Database>,
     body: Ruma<get_alias::Request<'_>>,
@@ -65,23 +70,59 @@ pub async fn get_alias_helper(
     room_alias: &RoomAliasId,
 ) -> ConduitResult<get_alias::Response> {
     if room_alias.server_name() != db.globals.server_name() {
-        let response = server_server::send_request(
-            &db.globals,
-            room_alias.server_name().to_owned(),
-            federation::query::get_room_information::v1::Request { room_alias },
-        )
-        .await?;
+        let response = db
+            .sending
+            .send_federation_request(
+                &db.globals,
+                room_alias.server_name(),
+                federation::query::get_room_information::v1::Request { room_alias },
+            )
+            .await?;
 
         return Ok(get_alias::Response::new(response.room_id, response.servers).into());
     }
 
-    let room_id = db
-        .rooms
-        .id_from_alias(&room_alias)?
-        .ok_or(Error::BadRequest(
-            ErrorKind::NotFound,
-            "Room with alias not found.",
-        ))?;
+    let mut room_id = None;
+    match db.rooms.id_from_alias(&room_alias)? {
+        Some(r) => room_id = Some(r),
+        None => {
+            for (_id, registration) in db.appservice.iter_all().filter_map(|r| r.ok()) {
+                let aliases = registration
+                    .get("namespaces")
+                    .and_then(|ns| ns.get("aliases"))
+                    .and_then(|users| users.get("regex"))
+                    .and_then(|regex| regex.as_str())
+                    .and_then(|regex| Regex::new(regex).ok());
+
+                if aliases.map_or(false, |aliases| aliases.is_match(room_alias.as_str()))
+                    && db
+                        .sending
+                        .send_appservice_request(
+                            &db.globals,
+                            registration,
+                            appservice::query::query_room_alias::v1::Request { room_alias },
+                        )
+                        .await
+                        .is_ok()
+                {
+                    room_id = Some(db.rooms.id_from_alias(&room_alias)?.ok_or_else(|| {
+                        Error::bad_config("Appservice lied to us. Room does not exist.")
+                    })?);
+                    break;
+                }
+            }
+        }
+    };
+
+    let room_id = match room_id {
+        Some(room_id) => room_id,
+        None => {
+            return Err(Error::BadRequest(
+                ErrorKind::NotFound,
+                "Room with alias not found.",
+            ))
+        }
+    };
 
     Ok(get_alias::Response::new(room_id, vec![db.globals.server_name().to_owned()]).into())
 }

src/client_server/backup.rs

@@ -17,6 +17,7 @@ use rocket::{delete, get, post, put};
     feature = "conduit_bin",
     post("/_matrix/client/unstable/room_keys/version", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn create_backup_route(
     db: State<'_, Database>,
     body: Ruma<create_backup::Request>,
@@ -35,6 +36,7 @@ pub async fn create_backup_route(
     feature = "conduit_bin",
     put("/_matrix/client/unstable/room_keys/version/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn update_backup_route(
     db: State<'_, Database>,
     body: Ruma<update_backup::Request<'_>>,
@@ -52,6 +54,7 @@ pub async fn update_backup_route(
     feature = "conduit_bin",
     get("/_matrix/client/unstable/room_keys/version", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_latest_backup_route(
     db: State<'_, Database>,
     body: Ruma<get_latest_backup::Request>,
@@ -79,6 +82,7 @@ pub async fn get_latest_backup_route(
     feature = "conduit_bin",
     get("/_matrix/client/unstable/room_keys/version/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_backup_route(
     db: State<'_, Database>,
     body: Ruma<get_backup::Request<'_>>,
@@ -105,6 +109,7 @@ pub async fn get_backup_route(
     feature = "conduit_bin",
     delete("/_matrix/client/unstable/room_keys/version/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn delete_backup_route(
     db: State<'_, Database>,
     body: Ruma<delete_backup::Request<'_>>,
@@ -123,6 +128,7 @@ pub async fn delete_backup_route(
     feature = "conduit_bin",
     put("/_matrix/client/unstable/room_keys/keys", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn add_backup_keys_route(
     db: State<'_, Database>,
     body: Ruma<add_backup_keys::Request<'_>>,
@@ -156,6 +162,7 @@ pub async fn add_backup_keys_route(
     feature = "conduit_bin",
     put("/_matrix/client/unstable/room_keys/keys/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn add_backup_key_sessions_route(
     db: State<'_, Database>,
     body: Ruma<add_backup_key_sessions::Request<'_>>,
@@ -187,6 +194,7 @@ pub async fn add_backup_key_sessions_route(
     feature = "conduit_bin",
     put("/_matrix/client/unstable/room_keys/keys/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn add_backup_key_session_route(
     db: State<'_, Database>,
     body: Ruma<add_backup_key_session::Request<'_>>,
@@ -215,6 +223,7 @@ pub async fn add_backup_key_session_route(
     feature = "conduit_bin",
     get("/_matrix/client/unstable/room_keys/keys", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_backup_keys_route(
     db: State<'_, Database>,
     body: Ruma<get_backup_keys::Request<'_>>,
@@ -230,6 +239,7 @@ pub async fn get_backup_keys_route(
     feature = "conduit_bin",
     get("/_matrix/client/unstable/room_keys/keys/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_backup_key_sessions_route(
     db: State<'_, Database>,
     body: Ruma<get_backup_key_sessions::Request<'_>>,
@@ -247,6 +257,7 @@ pub async fn get_backup_key_sessions_route(
     feature = "conduit_bin",
     get("/_matrix/client/unstable/room_keys/keys/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_backup_key_session_route(
     db: State<'_, Database>,
     body: Ruma<get_backup_key_session::Request<'_>>,
@@ -270,6 +281,7 @@ pub async fn get_backup_key_session_route(
     feature = "conduit_bin",
     delete("/_matrix/client/unstable/room_keys/keys", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn delete_backup_keys_route(
     db: State<'_, Database>,
     body: Ruma<delete_backup_keys::Request<'_>>,
@@ -292,6 +304,7 @@ pub async fn delete_backup_keys_route(
     feature = "conduit_bin",
     delete("/_matrix/client/unstable/room_keys/keys/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn delete_backup_key_sessions_route(
     db: State<'_, Database>,
     body: Ruma<delete_backup_key_sessions::Request<'_>>,
@@ -314,6 +327,7 @@ pub async fn delete_backup_key_sessions_route(
     feature = "conduit_bin",
     delete("/_matrix/client/unstable/room_keys/keys/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn delete_backup_key_session_route(
     db: State<'_, Database>,
     body: Ruma<delete_backup_key_session::Request<'_>>,

src/client_server/capabilities.rs

@@ -1,5 +1,10 @@
 use crate::ConduitResult;
-use ruma::{api::client::r0::capabilities::get_capabilities, RoomVersionId};
+use ruma::{
+    api::client::r0::capabilities::{
+        get_capabilities, Capabilities, RoomVersionStability, RoomVersionsCapability,
+    },
+    RoomVersionId,
+};
 use std::collections::BTreeMap;
 
 #[cfg(feature = "conduit_bin")]
@@ -9,26 +14,17 @@ use rocket::get;
 ///
 /// Get information on this server's supported feature set and other relevent capabilities.
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/client/r0/capabilities"))]
+#[tracing::instrument]
 pub async fn get_capabilities_route() -> ConduitResult<get_capabilities::Response> {
     let mut available = BTreeMap::new();
-    available.insert(
-        RoomVersionId::Version5,
-        get_capabilities::RoomVersionStability::Stable,
-    );
-    available.insert(
-        RoomVersionId::Version6,
-        get_capabilities::RoomVersionStability::Stable,
-    );
+    available.insert(RoomVersionId::Version5, RoomVersionStability::Stable);
+    available.insert(RoomVersionId::Version6, RoomVersionStability::Stable);
 
-    Ok(get_capabilities::Response {
-        capabilities: get_capabilities::Capabilities {
-            change_password: get_capabilities::ChangePasswordCapability::default(), // enabled by default
-            room_versions: get_capabilities::RoomVersionsCapability {
-                default: RoomVersionId::Version6,
-                available,
-            },
-            custom_capabilities: BTreeMap::new(),
-        },
-    }
-    .into())
+    let mut capabilities = Capabilities::new();
+    capabilities.room_versions = RoomVersionsCapability {
+        default: RoomVersionId::Version6,
+        available,
+    };
+
+    Ok(get_capabilities::Response { capabilities }.into())
 }

src/client_server/config.rs

@@ -16,13 +16,14 @@ use rocket::{get, put};
     feature = "conduit_bin",
     put("/_matrix/client/r0/user/<_>/account_data/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn set_global_account_data_route(
     db: State<'_, Database>,
     body: Ruma<set_global_account_data::Request<'_>>,
 ) -> ConduitResult<set_global_account_data::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
-    let content = serde_json::from_str::<serde_json::Value>(body.data.get())
+    let data = serde_json::from_str(body.data.get())
         .map_err(|_| Error::BadRequest(ErrorKind::BadJson, "Data is invalid."))?;
 
     let event_type = body.event_type.to_string();
@@ -32,10 +33,7 @@ pub async fn set_global_account_data_route(
         sender_user,
         event_type.clone().into(),
         &BasicEvent {
-            content: CustomEventContent {
-                event_type,
-                json: content,
-            },
+            content: CustomEventContent { event_type, data },
         },
         &db.globals,
     )?;
@@ -49,6 +47,7 @@ pub async fn set_global_account_data_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/user/<_>/account_data/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_global_account_data_route(
     db: State<'_, Database>,
     body: Ruma<get_global_account_data::Request<'_>>,

src/client_server/context.rs

@@ -10,6 +10,7 @@ use rocket::get;
     feature = "conduit_bin",
     get("/_matrix/client/r0/rooms/<_>/context/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_context_route(
     db: State<'_, Database>,
     body: Ruma<get_context::Request<'_>>,

src/client_server/device.rs

@@ -16,6 +16,7 @@ use rocket::{delete, get, post, put};
     feature = "conduit_bin",
     get("/_matrix/client/r0/devices", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_devices_route(
     db: State<'_, Database>,
     body: Ruma<get_devices::Request>,
@@ -35,6 +36,7 @@ pub async fn get_devices_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/devices/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_device_route(
     db: State<'_, Database>,
     body: Ruma<get_device::Request<'_>>,
@@ -53,6 +55,7 @@ pub async fn get_device_route(
     feature = "conduit_bin",
     put("/_matrix/client/r0/devices/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn update_device_route(
     db: State<'_, Database>,
     body: Ruma<update_device::Request<'_>>,
@@ -78,6 +81,7 @@ pub async fn update_device_route(
     feature = "conduit_bin",
     delete("/_matrix/client/r0/devices/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn delete_device_route(
     db: State<'_, Database>,
     body: Ruma<delete_device::Request<'_>>,
@@ -126,6 +130,7 @@ pub async fn delete_device_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/delete_devices", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn delete_devices_route(
     db: State<'_, Database>,
     body: Ruma<delete_devices::Request<'_>>,

src/client_server/directory.rs

@@ -1,5 +1,5 @@
 use super::State;
-use crate::{server_server, ConduitResult, Database, Error, Result, Ruma};
+use crate::{ConduitResult, Database, Error, Result, Ruma};
 use log::info;
 use ruma::{
     api::{
@@ -21,7 +21,7 @@ use ruma::{
         EventType,
     },
     serde::Raw,
-    ServerName,
+    ServerName, UInt,
 };
 
 #[cfg(feature = "conduit_bin")]
@@ -31,6 +31,7 @@ use rocket::{get, post, put};
     feature = "conduit_bin",
     post("/_matrix/client/r0/publicRooms", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_public_rooms_filtered_route(
     db: State<'_, Database>,
     body: Ruma<get_public_rooms_filtered::Request<'_>>,
@@ -50,6 +51,7 @@ pub async fn get_public_rooms_filtered_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/publicRooms", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_public_rooms_route(
     db: State<'_, Database>,
     body: Ruma<get_public_rooms::Request<'_>>,
@@ -78,6 +80,7 @@ pub async fn get_public_rooms_route(
     feature = "conduit_bin",
     put("/_matrix/client/r0/directory/list/room/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn set_room_visibility_route(
     db: State<'_, Database>,
     body: Ruma<set_room_visibility::Request<'_>>,
@@ -107,6 +110,7 @@ pub async fn set_room_visibility_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/directory/list/room/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_room_visibility_route(
     db: State<'_, Database>,
     body: Ruma<get_room_visibility::Request<'_>>,
@@ -124,7 +128,7 @@ pub async fn get_room_visibility_route(
 pub async fn get_public_rooms_filtered_helper(
     db: &Database,
     server: Option<&ServerName>,
-    limit: Option<js_int::UInt>,
+    limit: Option<UInt>,
     since: Option<&str>,
     filter: &IncomingFilter,
     _network: &IncomingRoomNetwork,
@@ -133,19 +137,21 @@ pub async fn get_public_rooms_filtered_helper(
         .clone()
         .filter(|server| *server != db.globals.server_name().as_str())
     {
-        let response = server_server::send_request(
-            &db.globals,
-            other_server.to_owned(),
-            federation::directory::get_public_rooms_filtered::v1::Request {
-                limit,
-                since: since.as_deref(),
-                filter: Filter {
-                    generic_search_term: filter.generic_search_term.as_deref(),
+        let response = db
+            .sending
+            .send_federation_request(
+                &db.globals,
+                other_server,
+                federation::directory::get_public_rooms_filtered::v1::Request {
+                    limit,
+                    since: since.as_deref(),
+                    filter: Filter {
+                        generic_search_term: filter.generic_search_term.as_deref(),
+                    },
+                    room_network: RoomNetwork::Matrix,
                 },
-                room_network: RoomNetwork::Matrix,
-            },
-        )
-        .await?;
+            )
+            .await?;
 
         return Ok(get_public_rooms_filtered::Response {
             chunk: response

src/client_server/filter.rs

@@ -5,6 +5,7 @@ use ruma::api::client::r0::filter::{self, create_filter, get_filter};
 use rocket::{get, post};
 
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/client/r0/user/<_>/filter/<_>"))]
+#[tracing::instrument]
 pub async fn get_filter_route() -> ConduitResult<get_filter::Response> {
     // TODO
     Ok(get_filter::Response::new(filter::IncomingFilterDefinition {
@@ -18,6 +19,7 @@ pub async fn get_filter_route() -> ConduitResult<get_filter::Response> {
 }
 
 #[cfg_attr(feature = "conduit_bin", post("/_matrix/client/r0/user/<_>/filter"))]
+#[tracing::instrument]
 pub async fn create_filter_route() -> ConduitResult<create_filter::Response> {
     // TODO
     Ok(create_filter::Response::new(utils::random_string(10)).into())

src/client_server/keys.rs

@@ -22,6 +22,7 @@ use rocket::{get, post};
     feature = "conduit_bin",
     post("/_matrix/client/r0/keys/upload", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn upload_keys_route(
     db: State<'_, Database>,
     body: Ruma<upload_keys::Request>,
@@ -70,6 +71,7 @@ pub async fn upload_keys_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/keys/query", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_keys_route(
     db: State<'_, Database>,
     body: Ruma<get_keys::Request<'_>>,
@@ -150,6 +152,7 @@ pub async fn get_keys_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/keys/claim", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn claim_keys_route(
     db: State<'_, Database>,
     body: Ruma<claim_keys::Request>,
@@ -183,6 +186,7 @@ pub async fn claim_keys_route(
     feature = "conduit_bin",
     post("/_matrix/client/unstable/keys/device_signing/upload", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn upload_signing_keys_route(
     db: State<'_, Database>,
     body: Ruma<upload_signing_keys::Request<'_>>,
@@ -240,6 +244,7 @@ pub async fn upload_signing_keys_route(
     feature = "conduit_bin",
     post("/_matrix/client/unstable/keys/signatures/upload", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn upload_signatures_route(
     db: State<'_, Database>,
     body: Ruma<upload_signatures::Request>,
@@ -300,6 +305,7 @@ pub async fn upload_signatures_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/keys/changes", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_key_changes_route(
     db: State<'_, Database>,
     body: Ruma<get_key_changes::Request<'_>>,

src/client_server/media.rs

@@ -1,7 +1,5 @@
 use super::State;
-use crate::{
-    database::media::FileMeta, server_server, utils, ConduitResult, Database, Error, Ruma,
-};
+use crate::{database::media::FileMeta, utils, ConduitResult, Database, Error, Ruma};
 use ruma::api::client::{
     error::ErrorKind,
     r0::media::{create_content, get_content, get_content_thumbnail, get_media_config},
@@ -14,6 +12,7 @@ use std::convert::TryInto;
 const MXC_LENGTH: usize = 32;
 
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/media/r0/config"))]
+#[tracing::instrument(skip(db))]
 pub async fn get_media_config_route(
     db: State<'_, Database>,
 ) -> ConduitResult<get_media_config::Response> {
@@ -27,6 +26,7 @@ pub async fn get_media_config_route(
     feature = "conduit_bin",
     post("/_matrix/media/r0/upload", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn create_content_route(
     db: State<'_, Database>,
     body: Ruma<create_content::Request<'_>>,
@@ -45,13 +45,18 @@ pub async fn create_content_route(
 
     db.flush().await?;
 
-    Ok(create_content::Response { content_uri: mxc }.into())
+    Ok(create_content::Response {
+        content_uri: mxc,
+        blurhash: None,
+    }
+    .into())
 }
 
 #[cfg_attr(
     feature = "conduit_bin",
     get("/_matrix/media/r0/download/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_content_route(
     db: State<'_, Database>,
     body: Ruma<get_content::Request<'_>>,
@@ -71,16 +76,18 @@ pub async fn get_content_route(
         }
         .into())
     } else if &*body.server_name != db.globals.server_name() && body.allow_remote {
-        let get_content_response = server_server::send_request(
-            &db.globals,
-            body.server_name.clone(),
-            get_content::Request {
-                allow_remote: false,
-                server_name: &body.server_name,
-                media_id: &body.media_id,
-            },
-        )
-        .await?;
+        let get_content_response = db
+            .sending
+            .send_federation_request(
+                &db.globals,
+                &body.server_name,
+                get_content::Request {
+                    allow_remote: false,
+                    server_name: &body.server_name,
+                    media_id: &body.media_id,
+                },
+            )
+            .await?;
 
         db.media.create(
             mxc,
@@ -99,6 +106,7 @@ pub async fn get_content_route(
     feature = "conduit_bin",
     get("/_matrix/media/r0/thumbnail/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_content_thumbnail_route(
     db: State<'_, Database>,
     body: Ruma<get_content_thumbnail::Request<'_>>,
@@ -118,19 +126,21 @@ pub async fn get_content_thumbnail_route(
     )? {
         Ok(get_content_thumbnail::Response { file, content_type }.into())
     } else if &*body.server_name != db.globals.server_name() && body.allow_remote {
-        let get_thumbnail_response = server_server::send_request(
-            &db.globals,
-            body.server_name.clone(),
-            get_content_thumbnail::Request {
-                allow_remote: false,
-                height: body.height,
-                width: body.width,
-                method: body.method,
-                server_name: &body.server_name,
-                media_id: &body.media_id,
-            },
-        )
-        .await?;
+        let get_thumbnail_response = db
+            .sending
+            .send_federation_request(
+                &db.globals,
+                &body.server_name,
+                get_content_thumbnail::Request {
+                    allow_remote: false,
+                    height: body.height,
+                    width: body.width,
+                    method: body.method.clone(),
+                    server_name: &body.server_name,
+                    media_id: &body.media_id,
+                },
+            )
+            .await?;
 
         db.media.upload_thumbnail(
             mxc,

src/client_server/membership.rs

@@ -2,9 +2,9 @@ use super::State;
 use crate::{
     client_server,
     pdu::{PduBuilder, PduEvent},
-    server_server, utils, ConduitResult, Database, Error, Result, Ruma,
+    utils, ConduitResult, Database, Error, Result, Ruma,
 };
-use log::warn;
+use log::{info, warn};
 use ruma::{
     api::{
         client::{
@@ -21,11 +21,9 @@ use ruma::{
     serde::{to_canonical_value, CanonicalJsonObject, Raw},
     EventId, RoomId, RoomVersionId, ServerName, UserId,
 };
-use state_res::StateEvent;
 use std::{
-    collections::{BTreeMap, HashMap, HashSet},
+    collections::{BTreeMap, HashMap},
     convert::TryFrom,
-    iter,
     sync::Arc,
 };
 
@@ -36,6 +34,7 @@ use rocket::{get, post};
     feature = "conduit_bin",
     post("/_matrix/client/r0/rooms/<_>/join", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn join_room_by_id_route(
     db: State<'_, Database>,
     body: Ruma<join_room_by_id::Request<'_>>,
@@ -54,6 +53,7 @@ pub async fn join_room_by_id_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/join/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn join_room_by_id_or_alias_route(
     db: State<'_, Database>,
     body: Ruma<join_room_by_id_or_alias::Request<'_>>,
@@ -88,6 +88,7 @@ pub async fn join_room_by_id_or_alias_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/rooms/<_>/leave", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn leave_room_route(
     db: State<'_, Database>,
     body: Ruma<leave_room::Request<'_>>,
@@ -124,10 +125,7 @@ pub async fn leave_room_route(
         },
         &sender_user,
         &body.room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     db.flush().await?;
@@ -139,6 +137,7 @@ pub async fn leave_room_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/rooms/<_>/invite", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn invite_user_route(
     db: State<'_, Database>,
     body: Ruma<invite_user::Request<'_>>,
@@ -163,10 +162,7 @@ pub async fn invite_user_route(
             },
             &sender_user,
             &body.room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         db.flush().await?;
@@ -181,6 +177,7 @@ pub async fn invite_user_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/rooms/<_>/kick", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn kick_user_route(
     db: State<'_, Database>,
     body: Ruma<kick_user::Request<'_>>,
@@ -218,10 +215,7 @@ pub async fn kick_user_route(
         },
         &sender_user,
         &body.room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     db.flush().await?;
@@ -233,6 +227,7 @@ pub async fn kick_user_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/rooms/<_>/ban", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn ban_user_route(
     db: State<'_, Database>,
     body: Ruma<ban_user::Request<'_>>,
@@ -277,10 +272,7 @@ pub async fn ban_user_route(
         },
         &sender_user,
         &body.room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     db.flush().await?;
@@ -292,6 +284,7 @@ pub async fn ban_user_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/rooms/<_>/unban", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn unban_user_route(
     db: State<'_, Database>,
     body: Ruma<unban_user::Request<'_>>,
@@ -328,10 +321,7 @@ pub async fn unban_user_route(
         },
         &sender_user,
         &body.room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     db.flush().await?;
@@ -343,6 +333,7 @@ pub async fn unban_user_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/rooms/<_>/forget", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn forget_room_route(
     db: State<'_, Database>,
     body: Ruma<forget_room::Request<'_>>,
@@ -360,6 +351,7 @@ pub async fn forget_room_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/joined_rooms", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn joined_rooms_route(
     db: State<'_, Database>,
     body: Ruma<joined_rooms::Request>,
@@ -380,6 +372,7 @@ pub async fn joined_rooms_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/rooms/<_>/members", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_member_events_route(
     db: State<'_, Database>,
     body: Ruma<get_member_events::Request<'_>>,
@@ -396,9 +389,10 @@ pub async fn get_member_events_route(
     Ok(get_member_events::Response {
         chunk: db
             .rooms
-            .room_state_type(&body.room_id, &EventType::RoomMember)?
-            .values()
-            .map(|pdu| pdu.to_member_event())
+            .room_state_full(&body.room_id)?
+            .iter()
+            .filter(|(key, _)| key.0 == EventType::RoomMember)
+            .map(|(_, pdu)| pdu.to_member_event())
             .collect(),
     }
     .into())
@@ -408,6 +402,7 @@ pub async fn get_member_events_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/rooms/<_>/joined_members", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn joined_members_route(
     db: State<'_, Database>,
     body: Ruma<joined_members::Request<'_>>,
@@ -442,6 +437,7 @@ pub async fn joined_members_route(
     Ok(joined_members::Response { joined }.into())
 }
 
+#[tracing::instrument(skip(db))]
 async fn join_room_by_id_helper(
     db: &Database,
     sender_user: Option<&UserId>,
@@ -458,16 +454,18 @@ async fn join_room_by_id_helper(
         ));
 
         for remote_server in servers {
-            let make_join_response = server_server::send_request(
-                &db.globals,
-                remote_server.clone(),
-                federation::membership::create_join_event_template::v1::Request {
-                    room_id,
-                    user_id: sender_user,
-                    ver: &[RoomVersionId::Version5, RoomVersionId::Version6],
-                },
-            )
-            .await;
+            let make_join_response = db
+                .sending
+                .send_federation_request(
+                    &db.globals,
+                    remote_server,
+                    federation::membership::create_join_event_template::v1::Request {
+                        room_id,
+                        user_id: sender_user,
+                        ver: &[RoomVersionId::Version5, RoomVersionId::Version6],
+                    },
+                )
+                .await;
 
             make_join_response_and_server = make_join_response.map(|r| (r, remote_server));
 
@@ -535,16 +533,18 @@ async fn join_room_by_id_helper(
         // It has enough fields to be called a proper event now
         let join_event = join_event_stub;
 
-        let send_join_response = server_server::send_request(
-            &db.globals,
-            remote_server.clone(),
-            federation::membership::create_join_event::v2::Request {
-                room_id,
-                event_id: &event_id,
-                pdu: PduEvent::convert_to_outgoing_federation_event(join_event.clone()),
-            },
-        )
-        .await?;
+        let send_join_response = db
+            .sending
+            .send_federation_request(
+                &db.globals,
+                remote_server,
+                federation::membership::create_join_event::v2::Request {
+                    room_id,
+                    event_id: &event_id,
+                    pdu: PduEvent::convert_to_outgoing_federation_event(join_event.clone()),
+                },
+            )
+            .await?;
 
         let add_event_id = |pdu: &Raw<Pdu>| -> Result<(EventId, CanonicalJsonObject)> {
             let mut value = serde_json::from_str(pdu.json().get())
@@ -565,131 +565,103 @@ async fn join_room_by_id_helper(
             Ok((event_id, value))
         };
 
-        let room_state = send_join_response.room_state.state.iter().map(add_event_id);
+        let count = db.globals.next_count()?;
 
-        let state_events = room_state
-            .clone()
-            .map(|pdu: Result<(EventId, CanonicalJsonObject)>| Ok(pdu?.0))
-            .chain(iter::once(Ok(event_id.clone()))) // Add join event we just created
-            .collect::<Result<HashSet<EventId>>>()?;
+        let mut pdu_id = room_id.as_bytes().to_vec();
+        pdu_id.push(0xff);
+        pdu_id.extend_from_slice(&count.to_be_bytes());
 
-        let auth_chain = send_join_response
+        let pdu = PduEvent::from_id_val(&event_id, join_event.clone())
+            .map_err(|_| Error::BadServerResponse("Invalid PDU in send_join response."))?;
+
+        let mut state = HashMap::new();
+
+        for pdu in send_join_response
             .room_state
-            .auth_chain
+            .state
             .iter()
-            .map(add_event_id);
-
-        let mut event_map = room_state
-            .chain(auth_chain)
-            .chain(iter::once(Ok((event_id, join_event)))) // Add join event we just created
+            .map(add_event_id)
             .map(|r| {
                 let (event_id, value) = r?;
-                state_res::StateEvent::from_id_canon_obj(event_id.clone(), value.clone())
+                PduEvent::from_id_val(&event_id, value.clone())
                     .map(|ev| (event_id, Arc::new(ev)))
                     .map_err(|e| {
                         warn!("{:?}: {}", value, e);
                         Error::BadServerResponse("Invalid PDU in send_join response.")
                     })
             })
-            .collect::<Result<BTreeMap<EventId, Arc<StateEvent>>>>()?;
-
-        let control_events = event_map
-            .values()
-            .filter(|pdu| pdu.is_power_event())
-            .map(|pdu| pdu.event_id())
-            .collect::<Vec<_>>();
-
-        // These events are not guaranteed to be sorted but they are resolved according to spec
-        // we auth them anyways to weed out faulty/malicious server. The following is basically the
-        // full state resolution algorithm.
-        let event_ids = event_map.keys().cloned().collect::<Vec<_>>();
-
-        let sorted_control_events = state_res::StateResolution::reverse_topological_power_sort(
-            &room_id,
-            &control_events,
-            &mut event_map,
-            &db.rooms,
-            &event_ids,
-        );
-
-        // Auth check each event against the "partial" state created by the preceding events
-        let resolved_control_events = state_res::StateResolution::iterative_auth_check(
-            room_id,
-            &RoomVersionId::Version6,
-            &sorted_control_events,
-            &BTreeMap::new(), // We have no "clean/resolved" events to add (these extend the `resolved_control_events`)
-            &mut event_map,
-            &db.rooms,
-        )
-        .expect("iterative auth check failed on resolved events");
-
-        // This removes the control events that failed auth, leaving the resolved
-        // to be mainline sorted. In the actual `state_res::StateResolution::resolve`
-        // function both are removed since these are all events we don't know of
-        // we must keep track of everything to add to our DB.
-        let events_to_sort = event_map
-            .keys()
-            .filter(|id| {
-                !sorted_control_events.contains(id)
-                    || resolved_control_events.values().any(|rid| *id == rid)
-            })
-            .cloned()
-            .collect::<Vec<_>>();
-
-        let power_level = resolved_control_events.get(&(EventType::RoomPowerLevels, "".into()));
-        // Sort the remaining non control events
-        let sorted_event_ids = state_res::StateResolution::mainline_sort(
-            room_id,
-            &events_to_sort,
-            power_level,
-            &mut event_map,
-            &db.rooms,
-        );
-
-        let resolved_events = state_res::StateResolution::iterative_auth_check(
-            room_id,
-            &RoomVersionId::Version6,
-            &sorted_event_ids,
-            &resolved_control_events,
-            &mut event_map,
-            &db.rooms,
-        )
-        .expect("iterative auth check failed on resolved events");
-
-        let mut state = HashMap::new();
-
-        // filter the events that failed the auth check keeping the remaining events
-        // sorted correctly
-        for ev_id in sorted_event_ids
-            .iter()
-            .filter(|id| resolved_events.values().any(|rid| rid == *id))
         {
-            // this is a `state_res::StateEvent` that holds a `ruma::Pdu`
-            let pdu = event_map
-                .get(ev_id)
-                .expect("Found event_id in sorted events that is not in resolved state");
+            let (id, pdu) = pdu?;
+            info!("adding {} to outliers: {:#?}", id, pdu);
+            db.rooms.add_pdu_outlier(&pdu)?;
+            if let Some(state_key) = &pdu.state_key {
+                if pdu.kind == EventType::RoomMember {
+                    let target_user_id = UserId::try_from(state_key.clone()).map_err(|_| {
+                        Error::BadServerResponse("Invalid user id in send_join response.")
+                    })?;
 
-            // We do not rebuild the PDU in this case only insert to DB
-            let count = db.globals.next_count()?;
-            let mut pdu_id = room_id.as_bytes().to_vec();
-            pdu_id.push(0xff);
-            pdu_id.extend_from_slice(&count.to_be_bytes());
-            db.rooms.append_pdu(
-                &PduEvent::from(&**pdu),
-                &utils::to_canonical_object(&**pdu).expect("Pdu is valid canonical object"),
-                count,
-                pdu_id.clone().into(),
-                &db.globals,
-                &db.account_data,
-                &db.admin,
-            )?;
-
-            if state_events.contains(ev_id) {
-                state.insert((pdu.kind(), pdu.state_key()), pdu_id);
+                    // Update our membership info, we do this here incase a user is invited
+                    // and immediately leaves we need the DB to record the invite event for auth
+                    db.rooms.update_membership(
+                        &pdu.room_id,
+                        &target_user_id,
+                        serde_json::from_value::<member::MemberEventContent>(pdu.content.clone())
+                            .map_err(|_| {
+                            Error::BadRequest(
+                                ErrorKind::InvalidParam,
+                                "Invalid member event content.",
+                            )
+                        })?,
+                        &pdu.sender,
+                        &db.account_data,
+                        &db.globals,
+                    )?;
+                }
+                let mut long_id = room_id.as_bytes().to_vec();
+                long_id.push(0xff);
+                long_id.extend_from_slice(id.as_bytes());
+                state.insert((pdu.kind.clone(), state_key.clone()), long_id);
             }
         }
 
-        db.rooms.force_state(room_id, state)?;
+        state.insert(
+            (
+                pdu.kind.clone(),
+                pdu.state_key.clone().expect("join event has state key"),
+            ),
+            pdu_id.clone(),
+        );
+
+        db.rooms.force_state(room_id, state, &db.globals)?;
+
+        for pdu in send_join_response
+            .room_state
+            .auth_chain
+            .iter()
+            .map(add_event_id)
+            .map(|r| {
+                let (event_id, value) = r?;
+                PduEvent::from_id_val(&event_id, value.clone())
+                    .map(|ev| (event_id, Arc::new(ev)))
+                    .map_err(|e| {
+                        warn!("{:?}: {}", value, e);
+                        Error::BadServerResponse("Invalid PDU in send_join response.")
+                    })
+            })
+        {
+            let (id, pdu) = pdu?;
+            info!("adding {} to outliers: {:#?}", id, pdu);
+            db.rooms.add_pdu_outlier(&pdu)?;
+        }
+
+        db.rooms.append_pdu(
+            &pdu,
+            utils::to_canonical_object(&pdu).expect("Pdu is valid canonical object"),
+            db.globals.next_count()?,
+            pdu_id.into(),
+            &[pdu.event_id.clone()],
+            db,
+        )?;
     } else {
         let event = member::MemberEventContent {
             membership: member::MembershipState::Join,
@@ -709,10 +681,7 @@ async fn join_room_by_id_helper(
             },
             &sender_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
     }
 

src/client_server/message.rs

@@ -8,7 +8,10 @@ use ruma::{
     events::EventContent,
     EventId,
 };
-use std::convert::{TryFrom, TryInto};
+use std::{
+    collections::BTreeMap,
+    convert::{TryFrom, TryInto},
+};
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, put};
@@ -17,12 +20,13 @@ use rocket::{get, put};
     feature = "conduit_bin",
     put("/_matrix/client/r0/rooms/<_>/send/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn send_message_event_route(
     db: State<'_, Database>,
     body: Ruma<send_message_event::Request<'_>>,
 ) -> ConduitResult<send_message_event::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
-    let sender_device = body.sender_device.as_ref().expect("user is authenticated");
+    let sender_device = body.sender_device.as_deref();
 
     // Check if this is a new transaction id
     if let Some(response) =
@@ -46,7 +50,7 @@ pub async fn send_message_event_route(
         return Ok(send_message_event::Response { event_id }.into());
     }
 
-    let mut unsigned = serde_json::Map::new();
+    let mut unsigned = BTreeMap::new();
     unsigned.insert("transaction_id".to_owned(), body.txn_id.clone().into());
 
     let event_id = db.rooms.build_and_append_pdu(
@@ -65,10 +69,7 @@ pub async fn send_message_event_route(
         },
         &sender_user,
         &body.room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     db.transaction_ids.add_txnid(
@@ -87,6 +88,7 @@ pub async fn send_message_event_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/rooms/<_>/messages", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_message_events_route(
     db: State<'_, Database>,
     body: Ruma<get_message_events::Request<'_>>,

src/client_server/mod.rs

@@ -75,6 +75,7 @@ const SESSION_ID_LENGTH: usize = 256;
 
 #[cfg(feature = "conduit_bin")]
 #[options("/<_..>")]
+#[tracing::instrument]
 pub async fn options_route() -> ConduitResult<send_event_to_device::Response> {
     Ok(send_event_to_device::Response.into())
 }

src/client_server/presence.rs

@@ -10,6 +10,7 @@ use rocket::put;
     feature = "conduit_bin",
     put("/_matrix/client/r0/presence/<_>/status", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn set_presence_route(
     db: State<'_, Database>,
     body: Ruma<set_presence::Request<'_>>,

src/client_server/profile.rs

@@ -19,6 +19,7 @@ use std::convert::TryInto;
     feature = "conduit_bin",
     put("/_matrix/client/r0/profile/<_>/displayname", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn set_displayname_route(
     db: State<'_, Database>,
     body: Ruma<set_display_name::Request<'_>>,
@@ -63,10 +64,7 @@ pub async fn set_displayname_route(
             },
             &sender_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // Presence update
@@ -101,6 +99,7 @@ pub async fn set_displayname_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/profile/<_>/displayname", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_displayname_route(
     db: State<'_, Database>,
     body: Ruma<get_display_name::Request<'_>>,
@@ -115,6 +114,7 @@ pub async fn get_displayname_route(
     feature = "conduit_bin",
     put("/_matrix/client/r0/profile/<_>/avatar_url", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn set_avatar_url_route(
     db: State<'_, Database>,
     body: Ruma<set_avatar_url::Request<'_>>,
@@ -159,10 +159,7 @@ pub async fn set_avatar_url_route(
             },
             &sender_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
 
         // Presence update
@@ -197,6 +194,7 @@ pub async fn set_avatar_url_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/profile/<_>/avatar_url", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_avatar_url_route(
     db: State<'_, Database>,
     body: Ruma<get_avatar_url::Request<'_>>,
@@ -211,6 +209,7 @@ pub async fn get_avatar_url_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/profile/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_profile_route(
     db: State<'_, Database>,
     body: Ruma<get_profile::Request<'_>>,

src/client_server/push.rs

@@ -1,21 +1,29 @@
 use super::State;
 use crate::{ConduitResult, Database, Error, Ruma};
-use log::warn;
 use ruma::{
     api::client::{
         error::ErrorKind,
-        r0::push::{get_pushers, get_pushrules_all, set_pushrule, set_pushrule_enabled},
+        r0::push::{
+            delete_pushrule, get_pushers, get_pushrule, get_pushrule_actions, get_pushrule_enabled,
+            get_pushrules_all, set_pusher, set_pushrule, set_pushrule_actions,
+            set_pushrule_enabled, RuleKind,
+        },
+    },
+    events::{push_rules, EventType},
+    push::{
+        ConditionalPushRuleInit, ContentPushRule, OverridePushRule, PatternedPushRuleInit,
+        RoomPushRule, SenderPushRule, SimplePushRuleInit, UnderridePushRule,
     },
-    events::EventType,
 };
 
 #[cfg(feature = "conduit_bin")]
-use rocket::{get, post, put};
+use rocket::{delete, get, post, put};
 
 #[cfg_attr(
     feature = "conduit_bin",
     get("/_matrix/client/r0/pushrules", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_pushrules_all_route(
     db: State<'_, Database>,
     body: Ruma<get_pushrules_all::Request>,
@@ -24,7 +32,7 @@ pub async fn get_pushrules_all_route(
 
     let event = db
         .account_data
-        .get::<ruma::events::push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
         .ok_or(Error::BadRequest(
             ErrorKind::NotFound,
             "PushRules event not found.",
@@ -36,16 +44,206 @@ pub async fn get_pushrules_all_route(
     .into())
 }
 
-#[cfg_attr(feature = "conduit_bin", put(
-    "/_matrix/client/r0/pushrules/<_>/<_>/<_>",
-    //data = "<body>"
-))]
+#[cfg_attr(
+    feature = "conduit_bin",
+    get("/_matrix/client/r0/pushrules/<_>/<_>/<_>", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn get_pushrule_route(
+    db: State<'_, Database>,
+    body: Ruma<get_pushrule::Request<'_>>,
+) -> ConduitResult<get_pushrule::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    let event = db
+        .account_data
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .ok_or(Error::BadRequest(
+            ErrorKind::NotFound,
+            "PushRules event not found.",
+        ))?;
+
+    let global = event.content.global;
+    let rule = match body.kind {
+        RuleKind::Override => global
+            .override_
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.clone().into()),
+        RuleKind::Underride => global
+            .underride
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.clone().into()),
+        RuleKind::Sender => global
+            .sender
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.clone().into()),
+        RuleKind::Room => global
+            .room
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.clone().into()),
+        RuleKind::Content => global
+            .content
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.clone().into()),
+        RuleKind::_Custom(_) => None,
+    };
+
+    if let Some(rule) = rule {
+        Ok(get_pushrule::Response { rule }.into())
+    } else {
+        Err(Error::BadRequest(
+            ErrorKind::NotFound,
+            "Push rule not found.",
+        ))
+    }
+}
+
+#[cfg_attr(
+    feature = "conduit_bin",
+    put("/_matrix/client/r0/pushrules/<_>/<_>/<_>", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
 pub async fn set_pushrule_route(
     db: State<'_, Database>,
-    //body: Ruma<set_pushrule::Request>,
+    body: Ruma<set_pushrule::Request<'_>>,
 ) -> ConduitResult<set_pushrule::Response> {
-    // TODO
-    warn!("TODO: set_pushrule_route");
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    if body.scope != "global" {
+        return Err(Error::BadRequest(
+            ErrorKind::InvalidParam,
+            "Scopes other than 'global' are not supported.",
+        ));
+    }
+
+    let mut event = db
+        .account_data
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .ok_or(Error::BadRequest(
+            ErrorKind::NotFound,
+            "PushRules event not found.",
+        ))?;
+
+    let global = &mut event.content.global;
+    match body.kind {
+        RuleKind::Override => {
+            if let Some(rule) = global
+                .override_
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.override_.remove(&rule);
+            }
+
+            global.override_.insert(OverridePushRule(
+                ConditionalPushRuleInit {
+                    actions: body.actions.clone(),
+                    default: false,
+                    enabled: true,
+                    rule_id: body.rule_id.clone(),
+                    conditions: body.conditions.clone(),
+                }
+                .into(),
+            ));
+        }
+        RuleKind::Underride => {
+            if let Some(rule) = global
+                .underride
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.underride.remove(&rule);
+            }
+
+            global.underride.insert(UnderridePushRule(
+                ConditionalPushRuleInit {
+                    actions: body.actions.clone(),
+                    default: false,
+                    enabled: true,
+                    rule_id: body.rule_id.clone(),
+                    conditions: body.conditions.clone(),
+                }
+                .into(),
+            ));
+        }
+        RuleKind::Sender => {
+            if let Some(rule) = global
+                .sender
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.sender.remove(&rule);
+            }
+
+            global.sender.insert(SenderPushRule(
+                SimplePushRuleInit {
+                    actions: body.actions.clone(),
+                    default: false,
+                    enabled: true,
+                    rule_id: body.rule_id.clone(),
+                }
+                .into(),
+            ));
+        }
+        RuleKind::Room => {
+            if let Some(rule) = global
+                .room
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.room.remove(&rule);
+            }
+
+            global.room.insert(RoomPushRule(
+                SimplePushRuleInit {
+                    actions: body.actions.clone(),
+                    default: false,
+                    enabled: true,
+                    rule_id: body.rule_id.clone(),
+                }
+                .into(),
+            ));
+        }
+        RuleKind::Content => {
+            if let Some(rule) = global
+                .content
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.content.remove(&rule);
+            }
+
+            global.content.insert(ContentPushRule(
+                PatternedPushRuleInit {
+                    actions: body.actions.clone(),
+                    default: false,
+                    enabled: true,
+                    rule_id: body.rule_id.clone(),
+                    pattern: body.pattern.clone().unwrap_or_default(),
+                }
+                .into(),
+            ));
+        }
+        RuleKind::_Custom(_) => {}
+    }
+
+    db.account_data.update(
+        None,
+        &sender_user,
+        EventType::PushRules,
+        &event,
+        &db.globals,
+    )?;
 
     db.flush().await?;
 
@@ -54,33 +252,463 @@ pub async fn set_pushrule_route(
 
 #[cfg_attr(
     feature = "conduit_bin",
-    put("/_matrix/client/r0/pushrules/<_>/<_>/<_>/enabled")
+    get("/_matrix/client/r0/pushrules/<_>/<_>/<_>/actions", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
+pub async fn get_pushrule_actions_route(
+    db: State<'_, Database>,
+    body: Ruma<get_pushrule_actions::Request<'_>>,
+) -> ConduitResult<get_pushrule_actions::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    if body.scope != "global" {
+        return Err(Error::BadRequest(
+            ErrorKind::InvalidParam,
+            "Scopes other than 'global' are not supported.",
+        ));
+    }
+
+    let mut event = db
+        .account_data
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .ok_or(Error::BadRequest(
+            ErrorKind::NotFound,
+            "PushRules event not found.",
+        ))?;
+
+    let global = &mut event.content.global;
+    let actions = match body.kind {
+        RuleKind::Override => global
+            .override_
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.actions.clone()),
+        RuleKind::Underride => global
+            .underride
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.actions.clone()),
+        RuleKind::Sender => global
+            .sender
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.actions.clone()),
+        RuleKind::Room => global
+            .room
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.actions.clone()),
+        RuleKind::Content => global
+            .content
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map(|rule| rule.0.actions.clone()),
+        RuleKind::_Custom(_) => None,
+    };
+
+    db.flush().await?;
+
+    Ok(get_pushrule_actions::Response {
+        actions: actions.unwrap_or_default(),
+    }
+    .into())
+}
+
+#[cfg_attr(
+    feature = "conduit_bin",
+    put("/_matrix/client/r0/pushrules/<_>/<_>/<_>/actions", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn set_pushrule_actions_route(
+    db: State<'_, Database>,
+    body: Ruma<set_pushrule_actions::Request<'_>>,
+) -> ConduitResult<set_pushrule_actions::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    if body.scope != "global" {
+        return Err(Error::BadRequest(
+            ErrorKind::InvalidParam,
+            "Scopes other than 'global' are not supported.",
+        ));
+    }
+
+    let mut event = db
+        .account_data
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .ok_or(Error::BadRequest(
+            ErrorKind::NotFound,
+            "PushRules event not found.",
+        ))?;
+
+    let global = &mut event.content.global;
+    match body.kind {
+        RuleKind::Override => {
+            if let Some(mut rule) = global
+                .override_
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.override_.remove(&rule);
+                rule.0.actions = body.actions.clone();
+                global.override_.insert(rule);
+            }
+        }
+        RuleKind::Underride => {
+            if let Some(mut rule) = global
+                .underride
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.underride.remove(&rule);
+                rule.0.actions = body.actions.clone();
+                global.underride.insert(rule);
+            }
+        }
+        RuleKind::Sender => {
+            if let Some(mut rule) = global
+                .sender
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.sender.remove(&rule);
+                rule.0.actions = body.actions.clone();
+                global.sender.insert(rule);
+            }
+        }
+        RuleKind::Room => {
+            if let Some(mut rule) = global
+                .room
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.room.remove(&rule);
+                rule.0.actions = body.actions.clone();
+                global.room.insert(rule);
+            }
+        }
+        RuleKind::Content => {
+            if let Some(mut rule) = global
+                .content
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.content.remove(&rule);
+                rule.0.actions = body.actions.clone();
+                global.content.insert(rule);
+            }
+        }
+        RuleKind::_Custom(_) => {}
+    };
+
+    db.account_data.update(
+        None,
+        &sender_user,
+        EventType::PushRules,
+        &event,
+        &db.globals,
+    )?;
+
+    db.flush().await?;
+
+    Ok(set_pushrule_actions::Response.into())
+}
+
+#[cfg_attr(
+    feature = "conduit_bin",
+    get("/_matrix/client/r0/pushrules/<_>/<_>/<_>/enabled", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn get_pushrule_enabled_route(
+    db: State<'_, Database>,
+    body: Ruma<get_pushrule_enabled::Request<'_>>,
+) -> ConduitResult<get_pushrule_enabled::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    if body.scope != "global" {
+        return Err(Error::BadRequest(
+            ErrorKind::InvalidParam,
+            "Scopes other than 'global' are not supported.",
+        ));
+    }
+
+    let mut event = db
+        .account_data
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .ok_or(Error::BadRequest(
+            ErrorKind::NotFound,
+            "PushRules event not found.",
+        ))?;
+
+    let global = &mut event.content.global;
+    let enabled = match body.kind {
+        RuleKind::Override => global
+            .override_
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map_or(false, |rule| rule.0.enabled),
+        RuleKind::Underride => global
+            .underride
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map_or(false, |rule| rule.0.enabled),
+        RuleKind::Sender => global
+            .sender
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map_or(false, |rule| rule.0.enabled),
+        RuleKind::Room => global
+            .room
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map_or(false, |rule| rule.0.enabled),
+        RuleKind::Content => global
+            .content
+            .iter()
+            .find(|rule| rule.0.rule_id == body.rule_id)
+            .map_or(false, |rule| rule.0.enabled),
+        RuleKind::_Custom(_) => false,
+    };
+
+    db.flush().await?;
+
+    Ok(get_pushrule_enabled::Response { enabled }.into())
+}
+
+#[cfg_attr(
+    feature = "conduit_bin",
+    put("/_matrix/client/r0/pushrules/<_>/<_>/<_>/enabled", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
 pub async fn set_pushrule_enabled_route(
     db: State<'_, Database>,
+    body: Ruma<set_pushrule_enabled::Request<'_>>,
 ) -> ConduitResult<set_pushrule_enabled::Response> {
-    // TODO
-    warn!("TODO: set_pushrule_enabled_route");
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    if body.scope != "global" {
+        return Err(Error::BadRequest(
+            ErrorKind::InvalidParam,
+            "Scopes other than 'global' are not supported.",
+        ));
+    }
+
+    let mut event = db
+        .account_data
+        .get::<ruma::events::push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .ok_or(Error::BadRequest(
+            ErrorKind::NotFound,
+            "PushRules event not found.",
+        ))?;
+
+    let global = &mut event.content.global;
+    match body.kind {
+        RuleKind::Override => {
+            if let Some(mut rule) = global
+                .override_
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.override_.remove(&rule);
+                rule.0.enabled = body.enabled;
+                global.override_.insert(rule);
+            }
+        }
+        RuleKind::Underride => {
+            if let Some(mut rule) = global
+                .underride
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.underride.remove(&rule);
+                rule.0.enabled = body.enabled;
+                global.underride.insert(rule);
+            }
+        }
+        RuleKind::Sender => {
+            if let Some(mut rule) = global
+                .sender
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.sender.remove(&rule);
+                rule.0.enabled = body.enabled;
+                global.sender.insert(rule);
+            }
+        }
+        RuleKind::Room => {
+            if let Some(mut rule) = global
+                .room
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.room.remove(&rule);
+                rule.0.enabled = body.enabled;
+                global.room.insert(rule);
+            }
+        }
+        RuleKind::Content => {
+            if let Some(mut rule) = global
+                .content
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.content.remove(&rule);
+                rule.0.enabled = body.enabled;
+                global.content.insert(rule);
+            }
+        }
+        RuleKind::_Custom(_) => {}
+    }
+
+    db.account_data.update(
+        None,
+        &sender_user,
+        EventType::PushRules,
+        &event,
+        &db.globals,
+    )?;
 
     db.flush().await?;
 
     Ok(set_pushrule_enabled::Response.into())
 }
 
-#[cfg_attr(feature = "conduit_bin", get("/_matrix/client/r0/pushers"))]
-pub async fn get_pushers_route() -> ConduitResult<get_pushers::Response> {
-    Ok(get_pushers::Response {
-        pushers: Vec::new(),
+#[cfg_attr(
+    feature = "conduit_bin",
+    delete("/_matrix/client/r0/pushrules/<_>/<_>/<_>", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn delete_pushrule_route(
+    db: State<'_, Database>,
+    body: Ruma<delete_pushrule::Request<'_>>,
+) -> ConduitResult<delete_pushrule::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    if body.scope != "global" {
+        return Err(Error::BadRequest(
+            ErrorKind::InvalidParam,
+            "Scopes other than 'global' are not supported.",
+        ));
     }
-    .into())
-}
+
+    let mut event = db
+        .account_data
+        .get::<push_rules::PushRulesEvent>(None, &sender_user, EventType::PushRules)?
+        .ok_or(Error::BadRequest(
+            ErrorKind::NotFound,
+            "PushRules event not found.",
+        ))?;
+
+    let global = &mut event.content.global;
+    match body.kind {
+        RuleKind::Override => {
+            if let Some(rule) = global
+                .override_
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.override_.remove(&rule);
+            }
+        }
+        RuleKind::Underride => {
+            if let Some(rule) = global
+                .underride
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.underride.remove(&rule);
+            }
+        }
+        RuleKind::Sender => {
+            if let Some(rule) = global
+                .sender
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.sender.remove(&rule);
+            }
+        }
+        RuleKind::Room => {
+            if let Some(rule) = global
+                .room
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.room.remove(&rule);
+            }
+        }
+        RuleKind::Content => {
+            if let Some(rule) = global
+                .content
+                .iter()
+                .find(|rule| rule.0.rule_id == body.rule_id)
+                .cloned()
+            {
+                global.content.remove(&rule);
+            }
+        }
+        RuleKind::_Custom(_) => {}
+    }
+
+    db.account_data.update(
+        None,
+        &sender_user,
+        EventType::PushRules,
+        &event,
+        &db.globals,
+    )?;
 
-#[cfg_attr(feature = "conduit_bin", post("/_matrix/client/r0/pushers/set"))]
-pub async fn set_pushers_route(db: State<'_, Database>) -> ConduitResult<get_pushers::Response> {
     db.flush().await?;
 
+    Ok(delete_pushrule::Response.into())
+}
+
+#[cfg_attr(
+    feature = "conduit_bin",
+    get("/_matrix/client/r0/pushers", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn get_pushers_route(
+    db: State<'_, Database>,
+    body: Ruma<get_pushers::Request>,
+) -> ConduitResult<get_pushers::Response> {
+    let sender = body.sender_user.as_ref().expect("authenticated endpoint");
+
     Ok(get_pushers::Response {
-        pushers: Vec::new(),
+        pushers: db.pusher.get_pusher(sender)?,
     }
     .into())
 }
+
+#[cfg_attr(
+    feature = "conduit_bin",
+    post("/_matrix/client/r0/pushers/set", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn set_pushers_route(
+    db: State<'_, Database>,
+    body: Ruma<set_pusher::Request>,
+) -> ConduitResult<set_pusher::Response> {
+    let sender = body.sender_user.as_ref().expect("authenticated endpoint");
+    let pusher = body.pusher.clone();
+
+    db.pusher.set_pusher(sender, pusher)?;
+
+    db.flush().await?;
+
+    Ok(set_pusher::Response::default().into())
+}

src/client_server/read_marker.rs

@@ -1,7 +1,12 @@
 use super::State;
 use crate::{ConduitResult, Database, Error, Ruma};
 use ruma::{
-    api::client::{error::ErrorKind, r0::read_marker::set_read_marker},
+    api::client::{
+        error::ErrorKind,
+        r0::{
+            capabilities::get_capabilities, read_marker::set_read_marker, receipt::create_receipt,
+        },
+    },
     events::{AnyEphemeralRoomEvent, AnyEvent, EventType},
 };
 
@@ -13,6 +18,7 @@ use std::{collections::BTreeMap, time::SystemTime};
     feature = "conduit_bin",
     post("/_matrix/client/r0/rooms/<_>/read_markers", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn set_read_marker_route(
     db: State<'_, Database>,
     body: Ruma<set_read_marker::Request<'_>>,
@@ -76,3 +82,58 @@ pub async fn set_read_marker_route(
 
     Ok(set_read_marker::Response.into())
 }
+
+#[cfg_attr(
+    feature = "conduit_bin",
+    post("/_matrix/client/r0/rooms/<_>/receipt/<_>/<_>", data = "<body>")
+)]
+#[tracing::instrument(skip(db, body))]
+pub async fn create_receipt_route(
+    db: State<'_, Database>,
+    body: Ruma<create_receipt::Request<'_>>,
+) -> ConduitResult<create_receipt::Response> {
+    let sender_user = body.sender_user.as_ref().expect("user is authenticated");
+
+    db.rooms.edus.private_read_set(
+        &body.room_id,
+        &sender_user,
+        db.rooms
+            .get_pdu_count(&body.event_id)?
+            .ok_or(Error::BadRequest(
+                ErrorKind::InvalidParam,
+                "Event does not exist.",
+            ))?,
+        &db.globals,
+    )?;
+
+    let mut user_receipts = BTreeMap::new();
+    user_receipts.insert(
+        sender_user.clone(),
+        ruma::events::receipt::Receipt {
+            ts: Some(SystemTime::now()),
+        },
+    );
+    let mut receipt_content = BTreeMap::new();
+    receipt_content.insert(
+        body.event_id.to_owned(),
+        ruma::events::receipt::Receipts {
+            read: Some(user_receipts),
+        },
+    );
+
+    db.rooms.edus.readreceipt_update(
+        &sender_user,
+        &body.room_id,
+        AnyEvent::Ephemeral(AnyEphemeralRoomEvent::Receipt(
+            ruma::events::receipt::ReceiptEvent {
+                content: ruma::events::receipt::ReceiptEventContent(receipt_content),
+                room_id: body.room_id.clone(),
+            },
+        )),
+        &db.globals,
+    )?;
+
+    db.flush().await?;
+
+    Ok(create_receipt::Response.into())
+}

src/client_server/redact.rs

@@ -12,6 +12,7 @@ use rocket::put;
     feature = "conduit_bin",
     put("/_matrix/client/r0/rooms/<_>/redact/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn redact_event_route(
     db: State<'_, Database>,
     body: Ruma<redact_event::Request<'_>>,
@@ -31,10 +32,7 @@ pub async fn redact_event_route(
         },
         &sender_user,
         &body.room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     db.flush().await?;

src/client_server/room.rs

@@ -22,6 +22,7 @@ use rocket::{get, post};
     feature = "conduit_bin",
     post("/_matrix/client/r0/createRoom", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn create_room_route(
     db: State<'_, Database>,
     body: Ruma<create_room::Request<'_>>,
@@ -65,10 +66,7 @@ pub async fn create_room_route(
         },
         &sender_user,
         &room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     // 2. Let the room creator join
@@ -89,10 +87,7 @@ pub async fn create_room_route(
         },
         &sender_user,
         &room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     // 3. Power levels
@@ -133,10 +128,7 @@ pub async fn create_room_route(
         },
         &sender_user,
         &room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     // 4. Events set by preset
@@ -172,10 +164,7 @@ pub async fn create_room_route(
         },
         &sender_user,
         &room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     // 4.2 History Visibility
@@ -192,10 +181,7 @@ pub async fn create_room_route(
         },
         &sender_user,
         &room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     // 4.3 Guest Access
@@ -220,10 +206,7 @@ pub async fn create_room_route(
         },
         &sender_user,
         &room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     // 5. Events listed in initial_state
@@ -234,19 +217,12 @@ pub async fn create_room_route(
         .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Invalid initial state event."))?;
 
         // Silently skip encryption events if they are not allowed
-        if pdu_builder.event_type == EventType::RoomEncryption && db.globals.encryption_disabled() {
+        if pdu_builder.event_type == EventType::RoomEncryption && !db.globals.allow_encryption() {
             continue;
         }
 
-        db.rooms.build_and_append_pdu(
-            pdu_builder,
-            &sender_user,
-            &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
-        )?;
+        db.rooms
+            .build_and_append_pdu(pdu_builder, &sender_user, &room_id, &db)?;
     }
 
     // 6. Events implied by name and topic
@@ -266,10 +242,7 @@ pub async fn create_room_route(
             },
             &sender_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
     }
 
@@ -287,10 +260,7 @@ pub async fn create_room_route(
             },
             &sender_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
     }
 
@@ -313,10 +283,7 @@ pub async fn create_room_route(
             },
             &sender_user,
             &room_id,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
     }
 
@@ -340,6 +307,7 @@ pub async fn create_room_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/rooms/<_>/event/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_room_event_route(
     db: State<'_, Database>,
     body: Ruma<get_room_event::Request<'_>>,
@@ -367,6 +335,7 @@ pub async fn get_room_event_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/rooms/<_room_id>/upgrade", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn upgrade_room_route(
     db: State<'_, Database>,
     body: Ruma<upgrade_room::Request<'_>>,
@@ -403,10 +372,7 @@ pub async fn upgrade_room_route(
         },
         sender_user,
         &body.room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     // Get the old room federations status
@@ -446,10 +412,7 @@ pub async fn upgrade_room_route(
         },
         sender_user,
         &replacement_room,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     // Join the new room
@@ -470,10 +433,7 @@ pub async fn upgrade_room_route(
         },
         sender_user,
         &replacement_room,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     // Recommended transferable state events list from the specs
@@ -506,10 +466,7 @@ pub async fn upgrade_room_route(
             },
             sender_user,
             &replacement_room,
-            &db.globals,
-            &db.sending,
-            &db.admin,
-            &db.account_data,
+            &db,
         )?;
     }
 
@@ -552,10 +509,7 @@ pub async fn upgrade_room_route(
         },
         sender_user,
         &body.room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     db.flush().await?;

src/client_server/search.rs

@@ -11,6 +11,7 @@ use std::collections::BTreeMap;
     feature = "conduit_bin",
     post("/_matrix/client/r0/search", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn search_events_route(
     db: State<'_, Database>,
     body: Ruma<search_events::Request<'_>>,
@@ -77,8 +78,8 @@ pub async fn search_events_route(
 
     Ok(search_events::Response::new(ResultCategories {
         room_events: ResultRoomEvents {
-            count: None,             // TODO? maybe not
-            groups: BTreeMap::new(), // TODO
+            count: Some((results.len() as u32).into()), // TODO: set this to none. Element shouldn't depend on it
+            groups: BTreeMap::new(),                    // TODO
             next_batch,
             results,
             state: BTreeMap::new(), // TODO

src/client_server/session.rs

@@ -8,6 +8,13 @@ use ruma::{
     },
     UserId,
 };
+use serde::Deserialize;
+
+#[derive(Debug, Deserialize)]
+struct Claims {
+    sub: String,
+    exp: usize,
+}
 
 #[cfg(feature = "conduit_bin")]
 use rocket::{get, post};
@@ -17,6 +24,7 @@ use rocket::{get, post};
 /// Get the homeserver's supported login types. One of these should be used as the `type` field
 /// when logging in.
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/client/r0/login"))]
+#[tracing::instrument]
 pub async fn get_login_types_route() -> ConduitResult<get_login_types::Response> {
     Ok(get_login_types::Response::new(vec![get_login_types::LoginType::Password]).into())
 }
@@ -35,49 +43,71 @@ pub async fn get_login_types_route() -> ConduitResult<get_login_types::Response>
     feature = "conduit_bin",
     post("/_matrix/client/r0/login", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn login_route(
     db: State<'_, Database>,
     body: Ruma<login::Request<'_>>,
 ) -> ConduitResult<login::Response> {
     // Validate login method
-    let user_id =
-        // TODO: Other login methods
-        if let (login::IncomingUserInfo::MatrixId(username), login::IncomingLoginInfo::Password { password }) =
-            (&body.user, &body.login_info)
-        {
-            let user_id = UserId::parse_with_server_name(username.to_string(), db.globals.server_name())
-                .map_err(|_| Error::BadRequest(
-                    ErrorKind::InvalidUsername,
-                    "Username is invalid."
-                ))?;
-            let hash = db.users.password_hash(&user_id)?
-                .ok_or(Error::BadRequest(
-                    ErrorKind::Forbidden,
-                    "Wrong username or password."
-                ))?;
+    // TODO: Other login methods
+    let user_id = match &body.login_info {
+        login::IncomingLoginInfo::Password { password } => {
+            let username = if let login::IncomingUserInfo::MatrixId(matrix_id) = &body.user {
+                matrix_id
+            } else {
+                return Err(Error::BadRequest(ErrorKind::Forbidden, "Bad login type."));
+            };
+            let user_id =
+                UserId::parse_with_server_name(username.to_owned(), db.globals.server_name())
+                    .map_err(|_| {
+                        Error::BadRequest(ErrorKind::InvalidUsername, "Username is invalid.")
+                    })?;
+            let hash = db.users.password_hash(&user_id)?.ok_or(Error::BadRequest(
+                ErrorKind::Forbidden,
+                "Wrong username or password.",
+            ))?;
 
             if hash.is_empty() {
                 return Err(Error::BadRequest(
                     ErrorKind::UserDeactivated,
-                    "The user has been deactivated"
+                    "The user has been deactivated",
                 ));
             }
 
-            let hash_matches =
-                argon2::verify_encoded(&hash, password.as_bytes()).unwrap_or(false);
+            let hash_matches = argon2::verify_encoded(&hash, password.as_bytes()).unwrap_or(false);
 
             if !hash_matches {
-                return Err(Error::BadRequest(ErrorKind::Forbidden, "Wrong username or password."));
+                return Err(Error::BadRequest(
+                    ErrorKind::Forbidden,
+                    "Wrong username or password.",
+                ));
             }
 
             user_id
-        } else {
-            return Err(Error::BadRequest(ErrorKind::Forbidden, "Bad login type."));
-        };
+        }
+        login::IncomingLoginInfo::Token { token } => {
+            if let Some(jwt_decoding_key) = db.globals.jwt_decoding_key() {
+                let token = jsonwebtoken::decode::<Claims>(
+                    &token,
+                    &jwt_decoding_key,
+                    &jsonwebtoken::Validation::default(),
+                )
+                .map_err(|_| Error::BadRequest(ErrorKind::InvalidUsername, "Token is invalid."))?;
+                let username = token.claims.sub;
+                UserId::parse_with_server_name(username, db.globals.server_name()).map_err(
+                    |_| Error::BadRequest(ErrorKind::InvalidUsername, "Username is invalid."),
+                )?
+            } else {
+                return Err(Error::BadRequest(
+                    ErrorKind::Unknown,
+                    "Token login is not supported (server has no jwt decoding key).",
+                ));
+            }
+        }
+    };
 
     // Generate new device id if the user didn't specify one
     let device_id = body
-        .body
         .device_id
         .clone()
         .unwrap_or_else(|| utils::random_string(DEVICE_ID_LENGTH).into());
@@ -85,14 +115,23 @@ pub async fn login_route(
     // Generate a new token for the device
     let token = utils::random_string(TOKEN_LENGTH);
 
-    // TODO: Don't always create a new device
-    // Add device
-    db.users.create_device(
-        &user_id,
-        &device_id,
-        &token,
-        body.initial_device_display_name.clone(),
-    )?;
+    // Determine if device_id was provided and exists in the db for this user
+    let device_exists = body.device_id.as_ref().map_or(false, |device_id| {
+        db.users
+            .all_device_ids(&user_id)
+            .any(|x| x.as_ref().map_or(false, |v| v == device_id))
+    });
+
+    if device_exists {
+        db.users.set_token(&user_id, &device_id, &token)?;
+    } else {
+        db.users.create_device(
+            &user_id,
+            &device_id,
+            &token,
+            body.initial_device_display_name.clone(),
+        )?;
+    }
 
     info!("{} logged in", user_id);
 
@@ -118,6 +157,7 @@ pub async fn login_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/logout", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn logout_route(
     db: State<'_, Database>,
     body: Ruma<logout::Request>,
@@ -145,6 +185,7 @@ pub async fn logout_route(
     feature = "conduit_bin",
     post("/_matrix/client/r0/logout/all", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn logout_all_route(
     db: State<'_, Database>,
     body: Ruma<logout_all::Request>,

src/client_server/state.rs

@@ -22,6 +22,7 @@ use rocket::{get, put};
     feature = "conduit_bin",
     put("/_matrix/client/r0/rooms/<_>/state/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn send_state_event_for_key_route(
     db: State<'_, Database>,
     body: Ruma<send_state_event_for_key::Request<'_>>,
@@ -55,6 +56,7 @@ pub async fn send_state_event_for_key_route(
     feature = "conduit_bin",
     put("/_matrix/client/r0/rooms/<_>/state/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn send_state_event_for_empty_key_route(
     db: State<'_, Database>,
     body: Ruma<send_state_event_for_empty_key::Request<'_>>,
@@ -63,8 +65,8 @@ pub async fn send_state_event_for_empty_key_route(
     let Ruma {
         body,
         sender_user,
-        sender_device: _,
         json_body,
+        ..
     } = body;
 
     let json = serde_json::from_str::<serde_json::Value>(
@@ -96,6 +98,7 @@ pub async fn send_state_event_for_empty_key_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/rooms/<_>/state", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_state_events_route(
     db: State<'_, Database>,
     body: Ruma<get_state_events::Request<'_>>,
@@ -142,6 +145,7 @@ pub async fn get_state_events_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/rooms/<_>/state/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_state_events_for_key_route(
     db: State<'_, Database>,
     body: Ruma<get_state_events_for_key::Request<'_>>,
@@ -193,6 +197,7 @@ pub async fn get_state_events_for_key_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/rooms/<_>/state/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_state_events_for_empty_key_route(
     db: State<'_, Database>,
     body: Ruma<get_state_events_for_empty_key::Request<'_>>,
@@ -234,7 +239,7 @@ pub async fn get_state_events_for_empty_key_route(
         .1;
 
     Ok(get_state_events_for_empty_key::Response {
-        content: serde_json::value::to_raw_value(&event)
+        content: serde_json::value::to_raw_value(&event.content)
             .map_err(|_| Error::bad_database("Invalid event content in database"))?,
     }
     .into())
@@ -284,10 +289,7 @@ pub async fn send_state_event_for_key_helper(
         },
         &sender_user,
         &room_id,
-        &db.globals,
-        &db.sending,
-        &db.admin,
-        &db.account_data,
+        &db,
     )?;
 
     Ok(event_id)

src/client_server/sync.rs

@@ -30,6 +30,7 @@ use std::{
     feature = "conduit_bin",
     get("/_matrix/client/r0/sync", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn sync_events_route(
     db: State<'_, Database>,
     body: Ruma<sync_events::Request<'_>>,
@@ -91,147 +92,267 @@ pub async fn sync_events_route(
 
         // They /sync response doesn't always return all messages, so we say the output is
         // limited unless there are events in non_timeline_pdus
-        let mut limited = false;
-
-        let mut state_pdus = Vec::new();
-        for (_, pdu) in non_timeline_pdus {
-            if pdu.state_key.is_some() {
-                state_pdus.push(pdu);
-            }
-            limited = true;
-        }
+        let limited = non_timeline_pdus.next().is_some();
 
         // Database queries:
-        let encrypted_room = db
-            .rooms
-            .room_state_get(&room_id, &EventType::RoomEncryption, "")?
-            .is_some();
+
+        let current_state_hash = db.rooms.current_state_hash(&room_id)?;
 
         // These type is Option<Option<_>>. The outer Option is None when there is no event between
         // since and the current room state, meaning there should be no updates.
         // The inner Option is None when there is an event, but there is no state hash associated
         // with it. This can happen for the RoomCreate event, so all updates should arrive.
-        let first_pdu_after_since = db.rooms.pdus_after(sender_user, &room_id, since).next();
+        let first_pdu_before_since = db.rooms.pdus_until(sender_user, &room_id, since).next();
+        let pdus_after_since = db
+            .rooms
+            .pdus_after(sender_user, &room_id, since)
+            .next()
+            .is_some();
 
-        let since_state_hash = first_pdu_after_since
+        let since_state_hash = first_pdu_before_since
             .as_ref()
             .map(|pdu| db.rooms.pdu_state_hash(&pdu.as_ref().ok()?.0).ok()?);
 
-        let since_members = since_state_hash.as_ref().map(|state_hash| {
-            state_hash.as_ref().and_then(|state_hash| {
-                db.rooms
-                    .state_type(&state_hash, &EventType::RoomMember)
-                    .ok()
-            })
-        });
+        let (
+            heroes,
+            joined_member_count,
+            invited_member_count,
+            joined_since_last_sync,
+            state_events,
+        ) = if pdus_after_since && Some(&current_state_hash) != since_state_hash.as_ref() {
+            let current_state = db.rooms.room_state_full(&room_id)?;
+            let current_members = current_state
+                .iter()
+                .filter(|(key, _)| key.0 == EventType::RoomMember)
+                .map(|(key, value)| (&key.1, value)) // Only keep state key
+                .collect::<Vec<_>>();
+            let encrypted_room = current_state
+                .get(&(EventType::RoomEncryption, "".to_owned()))
+                .is_some();
+            let since_state = since_state_hash.as_ref().map(|state_hash| {
+                state_hash
+                    .as_ref()
+                    .and_then(|state_hash| db.rooms.state_full(&room_id, &state_hash).ok())
+            });
 
-        let since_encryption = since_state_hash.as_ref().map(|state_hash| {
-            state_hash.as_ref().and_then(|state_hash| {
-                db.rooms
-                    .state_get(&state_hash, &EventType::RoomEncryption, "")
-                    .ok()
-            })
-        });
+            let since_encryption = since_state.as_ref().map(|state| {
+                state
+                    .as_ref()
+                    .map(|state| state.get(&(EventType::RoomEncryption, "".to_owned())))
+            });
 
-        let current_members = db.rooms.room_state_type(&room_id, &EventType::RoomMember)?;
+            // Calculations:
+            let new_encrypted_room =
+                encrypted_room && since_encryption.map_or(true, |encryption| encryption.is_none());
 
-        // Calculations:
-        let new_encrypted_room =
-            encrypted_room && since_encryption.map_or(false, |encryption| encryption.is_none());
+            let send_member_count = since_state.as_ref().map_or(true, |since_state| {
+                since_state.as_ref().map_or(true, |since_state| {
+                    current_members.len()
+                        != since_state
+                            .iter()
+                            .filter(|(key, _)| key.0 == EventType::RoomMember)
+                            .count()
+                })
+            });
 
-        let send_member_count = since_members.as_ref().map_or(false, |since_members| {
-            since_members.as_ref().map_or(true, |since_members| {
-                current_members.len() != since_members.len()
-            })
-        });
+            let since_sender_member = since_state.as_ref().map(|since_state| {
+                since_state.as_ref().and_then(|state| {
+                    state
+                        .get(&(EventType::RoomMember, sender_user.as_str().to_owned()))
+                        .and_then(|pdu| {
+                            serde_json::from_value::<
+                                Raw<ruma::events::room::member::MemberEventContent>,
+                            >(pdu.content.clone())
+                            .expect("Raw::from_value always works")
+                            .deserialize()
+                            .map_err(|_| Error::bad_database("Invalid PDU in database."))
+                            .ok()
+                        })
+                })
+            });
 
-        let since_sender_member = since_members.as_ref().map(|since_members| {
-            since_members.as_ref().and_then(|members| {
-                members.get(sender_user.as_str()).and_then(|pdu| {
-                    serde_json::from_value::<Raw<ruma::events::room::member::MemberEventContent>>(
-                        pdu.content.clone(),
-                    )
+            if encrypted_room {
+                for (user_id, current_member) in current_members {
+                    let current_membership = serde_json::from_value::<
+                        Raw<ruma::events::room::member::MemberEventContent>,
+                    >(current_member.content.clone())
                     .expect("Raw::from_value always works")
                     .deserialize()
-                    .map_err(|_| Error::bad_database("Invalid PDU in database."))
-                    .ok()
-                })
-            })
-        });
+                    .map_err(|_| Error::bad_database("Invalid PDU in database."))?
+                    .membership;
 
-        if encrypted_room {
-            for (user_id, current_member) in current_members {
-                let current_membership = serde_json::from_value::<
-                    Raw<ruma::events::room::member::MemberEventContent>,
-                >(current_member.content.clone())
-                .expect("Raw::from_value always works")
-                .deserialize()
-                .map_err(|_| Error::bad_database("Invalid PDU in database."))?
-                .membership;
-
-                let since_membership =
-                    since_members
-                        .as_ref()
-                        .map_or(MembershipState::Join, |members| {
-                            members
-                                .as_ref()
-                                .and_then(|members| {
-                                    members.get(&user_id).and_then(|since_member| {
-                                        serde_json::from_value::<
-                                            Raw<ruma::events::room::member::MemberEventContent>,
-                                        >(
-                                            since_member.content.clone()
-                                        )
-                                        .expect("Raw::from_value always works")
-                                        .deserialize()
-                                        .map_err(|_| {
-                                            Error::bad_database("Invalid PDU in database.")
-                                        })
-                                        .ok()
+                    let since_membership =
+                        since_state
+                            .as_ref()
+                            .map_or(MembershipState::Leave, |since_state| {
+                                since_state
+                                    .as_ref()
+                                    .and_then(|since_state| {
+                                        since_state
+                                            .get(&(EventType::RoomMember, user_id.clone()))
+                                            .and_then(|since_member| {
+                                                serde_json::from_value::<
+                                                Raw<ruma::events::room::member::MemberEventContent>,
+                                            >(
+                                                since_member.content.clone()
+                                            )
+                                            .expect("Raw::from_value always works")
+                                            .deserialize()
+                                            .map_err(|_| {
+                                                Error::bad_database("Invalid PDU in database.")
+                                            })
+                                            .ok()
+                                            })
                                     })
-                                })
-                                .map_or(MembershipState::Leave, |member| member.membership)
-                        });
+                                    .map_or(MembershipState::Leave, |member| member.membership)
+                            });
 
-                let user_id = UserId::try_from(user_id)
-                    .map_err(|_| Error::bad_database("Invalid UserId in member PDU."))?;
+                    let user_id = UserId::try_from(user_id.clone())
+                        .map_err(|_| Error::bad_database("Invalid UserId in member PDU."))?;
 
-                match (since_membership, current_membership) {
-                    (MembershipState::Leave, MembershipState::Join) => {
-                        // A new user joined an encrypted room
-                        if !share_encrypted_room(&db, &sender_user, &user_id, &room_id) {
-                            device_list_updates.insert(user_id);
+                    match (since_membership, current_membership) {
+                        (MembershipState::Leave, MembershipState::Join) => {
+                            // A new user joined an encrypted room
+                            if !share_encrypted_room(&db, &sender_user, &user_id, &room_id) {
+                                device_list_updates.insert(user_id);
+                            }
                         }
+                        (MembershipState::Join, MembershipState::Leave) => {
+                            // Write down users that have left encrypted rooms we are in
+                            left_encrypted_users.insert(user_id);
+                        }
+                        _ => {}
                     }
-                    (MembershipState::Join, MembershipState::Leave) => {
-                        // Write down users that have left encrypted rooms we are in
-                        left_encrypted_users.insert(user_id);
-                    }
-                    _ => {}
                 }
             }
-        }
 
-        let joined_since_last_sync = since_sender_member.map_or(false, |member| {
-            member.map_or(true, |member| member.membership != MembershipState::Join)
-        });
+            let joined_since_last_sync = since_sender_member.map_or(true, |member| {
+                member.map_or(true, |member| member.membership != MembershipState::Join)
+            });
 
-        if joined_since_last_sync && encrypted_room || new_encrypted_room {
-            // If the user is in a new encrypted room, give them all joined users
-            device_list_updates.extend(
-                db.rooms
-                    .room_members(&room_id)
-                    .filter_map(|user_id| Some(user_id.ok()?))
-                    .filter(|user_id| {
-                        // Don't send key updates from the sender to the sender
-                        sender_user != user_id
-                    })
-                    .filter(|user_id| {
-                        // Only send keys if the sender doesn't share an encrypted room with the target already
-                        !share_encrypted_room(&db, sender_user, user_id, &room_id)
-                    }),
-            );
-        }
+            if joined_since_last_sync && encrypted_room || new_encrypted_room {
+                // If the user is in a new encrypted room, give them all joined users
+                device_list_updates.extend(
+                    db.rooms
+                        .room_members(&room_id)
+                        .filter_map(|user_id| Some(user_id.ok()?))
+                        .filter(|user_id| {
+                            // Don't send key updates from the sender to the sender
+                            sender_user != user_id
+                        })
+                        .filter(|user_id| {
+                            // Only send keys if the sender doesn't share an encrypted room with the target already
+                            !share_encrypted_room(&db, sender_user, user_id, &room_id)
+                        }),
+                );
+            }
+
+            let (joined_member_count, invited_member_count, heroes) = if send_member_count {
+                let joined_member_count = db.rooms.room_members(&room_id).count();
+                let invited_member_count = db.rooms.room_members_invited(&room_id).count();
+
+                // Recalculate heroes (first 5 members)
+                let mut heroes = Vec::new();
+
+                if joined_member_count + invited_member_count <= 5 {
+                    // Go through all PDUs and for each member event, check if the user is still joined or
+                    // invited until we have 5 or we reach the end
+
+                    for hero in db
+                        .rooms
+                        .all_pdus(&sender_user, &room_id)?
+                        .filter_map(|pdu| pdu.ok()) // Ignore all broken pdus
+                        .filter(|(_, pdu)| pdu.kind == EventType::RoomMember)
+                        .map(|(_, pdu)| {
+                            let content = serde_json::from_value::<
+                                Raw<ruma::events::room::member::MemberEventContent>,
+                            >(pdu.content.clone())
+                            .expect("Raw::from_value always works")
+                            .deserialize()
+                            .map_err(|_| {
+                                Error::bad_database("Invalid member event in database.")
+                            })?;
+
+                            if let Some(state_key) = &pdu.state_key {
+                                let user_id =
+                                    UserId::try_from(state_key.clone()).map_err(|_| {
+                                        Error::bad_database("Invalid UserId in member PDU.")
+                                    })?;
+
+                                // The membership was and still is invite or join
+                                if matches!(
+                                    content.membership,
+                                    MembershipState::Join | MembershipState::Invite
+                                ) && (db.rooms.is_joined(&user_id, &room_id)?
+                                    || db.rooms.is_invited(&user_id, &room_id)?)
+                                {
+                                    Ok::<_, Error>(Some(state_key.clone()))
+                                } else {
+                                    Ok(None)
+                                }
+                            } else {
+                                Ok(None)
+                            }
+                        })
+                        .filter_map(|u| u.ok()) // Filter out buggy users
+                        // Filter for possible heroes
+                        .filter_map(|u| u)
+                    {
+                        if heroes.contains(&hero) || hero == sender_user.as_str() {
+                            continue;
+                        }
+
+                        heroes.push(hero);
+                    }
+                }
+
+                (
+                    Some(joined_member_count),
+                    Some(invited_member_count),
+                    heroes,
+                )
+            } else {
+                (None, None, Vec::new())
+            };
+
+            let state_events = if dbg!(joined_since_last_sync) {
+                current_state
+                    .into_iter()
+                    .map(|(_, pdu)| pdu.to_sync_state_event())
+                    .collect()
+            } else {
+                match since_state {
+                    None => Vec::new(),
+                    Some(Some(since_state)) => current_state
+                        .iter()
+                        .filter(|(key, value)| {
+                            since_state.get(key).map(|e| &e.event_id) != Some(&value.event_id)
+                        })
+                        .filter(|(_, value)| {
+                            !timeline_pdus.iter().any(|(_, timeline_pdu)| {
+                                timeline_pdu.kind == value.kind
+                                    && timeline_pdu.state_key == value.state_key
+                            })
+                        })
+                        .map(|(_, pdu)| pdu.to_sync_state_event())
+                        .collect(),
+                    Some(None) => current_state
+                        .iter()
+                        .map(|(_, pdu)| pdu.to_sync_state_event())
+                        .collect(),
+                }
+            };
+
+            (
+                heroes,
+                joined_member_count,
+                invited_member_count,
+                joined_since_last_sync,
+                state_events,
+            )
+        } else {
+            (Vec::new(), None, None, false, Vec::new())
+        };
 
         // Look for device list updates in this room
         device_list_updates.extend(
@@ -240,71 +361,6 @@ pub async fn sync_events_route(
                 .filter_map(|r| r.ok()),
         );
 
-        let (joined_member_count, invited_member_count, heroes) = if send_member_count {
-            let joined_member_count = db.rooms.room_members(&room_id).count();
-            let invited_member_count = db.rooms.room_members_invited(&room_id).count();
-
-            // Recalculate heroes (first 5 members)
-            let mut heroes = Vec::new();
-
-            if joined_member_count + invited_member_count <= 5 {
-                // Go through all PDUs and for each member event, check if the user is still joined or
-                // invited until we have 5 or we reach the end
-
-                for hero in db
-                    .rooms
-                    .all_pdus(&sender_user, &room_id)?
-                    .filter_map(|pdu| pdu.ok()) // Ignore all broken pdus
-                    .filter(|(_, pdu)| pdu.kind == EventType::RoomMember)
-                    .map(|(_, pdu)| {
-                        let content = serde_json::from_value::<
-                            Raw<ruma::events::room::member::MemberEventContent>,
-                        >(pdu.content.clone())
-                        .expect("Raw::from_value always works")
-                        .deserialize()
-                        .map_err(|_| Error::bad_database("Invalid member event in database."))?;
-
-                        if let Some(state_key) = &pdu.state_key {
-                            let user_id = UserId::try_from(state_key.clone()).map_err(|_| {
-                                Error::bad_database("Invalid UserId in member PDU.")
-                            })?;
-
-                            // The membership was and still is invite or join
-                            if matches!(
-                                content.membership,
-                                MembershipState::Join | MembershipState::Invite
-                            ) && (db.rooms.is_joined(&user_id, &room_id)?
-                                || db.rooms.is_invited(&user_id, &room_id)?)
-                            {
-                                Ok::<_, Error>(Some(state_key.clone()))
-                            } else {
-                                Ok(None)
-                            }
-                        } else {
-                            Ok(None)
-                        }
-                    })
-                    .filter_map(|u| u.ok()) // Filter out buggy users
-                    // Filter for possible heroes
-                    .filter_map(|u| u)
-                {
-                    if heroes.contains(&hero) || hero == sender_user.as_str() {
-                        continue;
-                    }
-
-                    heroes.push(hero);
-                }
-            }
-
-            (
-                Some(joined_member_count),
-                Some(invited_member_count),
-                heroes,
-            )
-        } else {
-            (None, None, Vec::new())
-        };
-
         let notification_count = if send_notification_counts {
             if let Some(last_read) = db.rooms.edus.private_read_get(&room_id, &sender_user)? {
                 Some(
@@ -334,7 +390,7 @@ pub async fn sync_events_route(
             })?;
 
         let room_events = timeline_pdus
-            .into_iter()
+            .iter()
             .map(|(_, pdu)| pdu.to_sync_room_event())
             .collect::<Vec<_>>();
 
@@ -384,17 +440,8 @@ pub async fn sync_events_route(
                 prev_batch,
                 events: room_events,
             },
-            // TODO: state before timeline
             state: sync_events::State {
-                events: if joined_since_last_sync {
-                    db.rooms
-                        .room_state_full(&room_id)?
-                        .into_iter()
-                        .map(|(_, pdu)| pdu.to_sync_state_event())
-                        .collect()
-                } else {
-                    Vec::new()
-                },
+                events: state_events,
             },
             ephemeral: sync_events::Ephemeral { events: edus },
         };
@@ -456,7 +503,12 @@ pub async fn sync_events_route(
             })
             .and_then(|state_hash| {
                 db.rooms
-                    .state_get(&state_hash, &EventType::RoomMember, sender_user.as_str())
+                    .state_get(
+                        &room_id,
+                        &state_hash,
+                        &EventType::RoomMember,
+                        sender_user.as_str(),
+                    )
                     .ok()?
                     .ok_or_else(|| Error::bad_database("State hash in db doesn't have a state."))
                     .ok()
@@ -651,9 +703,10 @@ pub async fn sync_events_route(
         if duration.as_secs() > 30 {
             duration = Duration::from_secs(30);
         }
-        let mut delay = tokio::time::delay_for(duration);
+        let delay = tokio::time::sleep(duration);
+        tokio::pin!(delay);
         tokio::select! {
-            _ = &mut delay => {}
+            _ = &mut delay, if delay.is_elapsed() => {}
             _ = watcher => {}
         }
     }
@@ -661,6 +714,7 @@ pub async fn sync_events_route(
     Ok(response.into())
 }
 
+#[tracing::instrument(skip(db))]
 fn share_encrypted_room(
     db: &Database,
     sender_user: &UserId,

src/client_server/tag.rs

@@ -13,6 +13,7 @@ use rocket::{delete, get, put};
     feature = "conduit_bin",
     put("/_matrix/client/r0/user/<_>/rooms/<_>/tags/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn update_tag_route(
     db: State<'_, Database>,
     body: Ruma<create_tag::Request<'_>>,
@@ -49,6 +50,7 @@ pub async fn update_tag_route(
     feature = "conduit_bin",
     delete("/_matrix/client/r0/user/<_>/rooms/<_>/tags/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn delete_tag_route(
     db: State<'_, Database>,
     body: Ruma<delete_tag::Request<'_>>,
@@ -82,6 +84,7 @@ pub async fn delete_tag_route(
     feature = "conduit_bin",
     get("/_matrix/client/r0/user/<_>/rooms/<_>/tags", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn get_tags_route(
     db: State<'_, Database>,
     body: Ruma<get_tags::Request<'_>>,

src/client_server/thirdparty.rs

@@ -10,6 +10,7 @@ use std::collections::BTreeMap;
     feature = "conduit_bin",
     get("/_matrix/client/r0/thirdparty/protocols")
 )]
+#[tracing::instrument]
 pub async fn get_protocols_route() -> ConduitResult<get_protocols::Response> {
     warn!("TODO: get_protocols_route");
     Ok(get_protocols::Response {

src/client_server/to_device.rs

@@ -12,12 +12,13 @@ use rocket::put;
     feature = "conduit_bin",
     put("/_matrix/client/r0/sendToDevice/<_>/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn send_event_to_device_route(
     db: State<'_, Database>,
     body: Ruma<send_event_to_device::Request<'_>>,
 ) -> ConduitResult<send_event_to_device::Response> {
     let sender_user = body.sender_user.as_ref().expect("user is authenticated");
-    let sender_device = body.sender_device.as_ref().expect("user is authenticated");
+    let sender_device = body.sender_device.as_deref();
 
     // Check if this is a new transaction id
     if db

src/client_server/typing.rs

@@ -10,6 +10,7 @@ use rocket::put;
     feature = "conduit_bin",
     put("/_matrix/client/r0/rooms/<_>/typing/<_>", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub fn create_typing_event_route(
     db: State<'_, Database>,
     body: Ruma<create_typing_event::Request<'_>>,

src/client_server/unversioned.rs

@@ -15,6 +15,7 @@ use rocket::get;
 /// Note: Unstable features are used while developing new features. Clients should avoid using
 /// unstable features in their stable releases
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/client/versions"))]
+#[tracing::instrument]
 pub async fn get_supported_versions_route() -> ConduitResult<get_supported_versions::Response> {
     let mut resp =
         get_supported_versions::Response::new(vec!["r0.5.0".to_owned(), "r0.6.0".to_owned()]);

src/client_server/user_directory.rs

@@ -9,6 +9,7 @@ use rocket::post;
     feature = "conduit_bin",
     post("/_matrix/client/r0/user_directory/search", data = "<body>")
 )]
+#[tracing::instrument(skip(db, body))]
 pub async fn search_users_route(
     db: State<'_, Database>,
     body: Ruma<search_users::Request<'_>>,

src/client_server/voip.rs

@@ -1,13 +1,18 @@
-use crate::{ConduitResult, Error};
-use ruma::api::client::{error::ErrorKind, r0::message::send_message_event};
+use crate::ConduitResult;
+use ruma::api::client::r0::voip::get_turn_server_info;
+use std::time::Duration;
 
 #[cfg(feature = "conduit_bin")]
 use rocket::get;
 
 #[cfg_attr(feature = "conduit_bin", get("/_matrix/client/r0/voip/turnServer"))]
-pub async fn turn_server_route() -> ConduitResult<send_message_event::Response> {
-    Err(Error::BadRequest(
-        ErrorKind::NotFound,
-        "There is no turn server yet.",
-    ))
+#[tracing::instrument]
+pub async fn turn_server_route() -> ConduitResult<get_turn_server_info::Response> {
+    Ok(get_turn_server_info::Response {
+        username: "".to_owned(),
+        password: "".to_owned(),
+        uris: Vec::new(),
+        ttl: Duration::from_secs(60 * 60 * 24),
+    }
+    .into())
 }

src/database.rs

@@ -1,8 +1,10 @@
 pub mod account_data;
 pub mod admin;
+pub mod appservice;
 pub mod globals;
 pub mod key_backups;
 pub mod media;
+pub mod pusher;
 pub mod rooms;
 pub mod sending;
 pub mod transaction_ids;
@@ -16,30 +18,45 @@ use log::info;
 use rocket::futures::{self, channel::mpsc};
 use ruma::{DeviceId, ServerName, UserId};
 use serde::Deserialize;
-use std::{convert::TryInto, fs::remove_dir_all};
+use std::{
+    collections::HashMap,
+    fs::remove_dir_all,
+    sync::{Arc, RwLock},
+};
+use tokio::sync::Semaphore;
 
-#[derive(Clone, Deserialize)]
+#[derive(Clone, Debug, Deserialize)]
 pub struct Config {
-    #[serde(default = "default_server_name")]
     server_name: Box<ServerName>,
-    database_path: Option<String>,
+    database_path: String,
     #[serde(default = "default_cache_capacity")]
-    cache_capacity: u64,
+    cache_capacity: u32,
     #[serde(default = "default_max_request_size")]
     max_request_size: u32,
-    #[serde(default)]
-    registration_disabled: bool,
-    #[serde(default)]
-    encryption_disabled: bool,
-    #[serde(default)]
-    federation_enabled: bool,
+    #[serde(default = "default_max_concurrent_requests")]
+    max_concurrent_requests: u16,
+    #[serde(default = "true_fn")]
+    allow_registration: bool,
+    #[serde(default = "true_fn")]
+    allow_encryption: bool,
+    #[serde(default = "false_fn")]
+    allow_federation: bool,
+    #[serde(default = "false_fn")]
+    pub allow_jaeger: bool,
+    jwt_secret: Option<String>,
+    #[serde(default = "Vec::new")]
+    trusted_servers: Vec<Box<ServerName>>,
 }
 
-fn default_server_name() -> Box<ServerName> {
-    "localhost".try_into().expect("")
+fn false_fn() -> bool {
+    false
 }
 
-fn default_cache_capacity() -> u64 {
+fn true_fn() -> bool {
+    true
+}
+
+fn default_cache_capacity() -> u32 {
     1024 * 1024 * 1024
 }
 
@@ -47,6 +64,10 @@ fn default_max_request_size() -> u32 {
     20 * 1024 * 1024 // Default to 20 MB
 }
 
+fn default_max_concurrent_requests() -> u16 {
+    4
+}
+
 #[derive(Clone)]
 pub struct Database {
     pub globals: globals::Globals,
@@ -59,6 +80,8 @@ pub struct Database {
     pub transaction_ids: transaction_ids::TransactionIds,
     pub sending: sending::Sending,
     pub admin: admin::Admin,
+    pub appservice: appservice::Appservice,
+    pub pusher: pusher::PushData,
     pub _db: sled::Db,
 }
 
@@ -76,37 +99,22 @@ impl Database {
     }
 
     /// Load an existing database or create a new one.
-    pub fn load_or_create(config: Config) -> Result<Self> {
-        let path = config
-            .database_path
-            .clone()
-            .map(Ok::<_, Error>)
-            .unwrap_or_else(|| {
-                let path = ProjectDirs::from("xyz", "koesters", "conduit")
-                    .ok_or_else(|| {
-                        Error::bad_config("The OS didn't return a valid home directory path.")
-                    })?
-                    .data_dir()
-                    .join(config.server_name.as_str());
-
-                Ok(path
-                    .to_str()
-                    .ok_or_else(|| Error::bad_config("Database path contains invalid unicode."))?
-                    .to_owned())
-            })?;
-
+    pub async fn load_or_create(config: Config) -> Result<Self> {
         let db = sled::Config::default()
-            .path(&path)
-            .cache_capacity(config.cache_capacity)
-            .print_profile_on_drop(true)
+            .path(&config.database_path)
+            .cache_capacity(config.cache_capacity as u64)
             .open()?;
 
-        info!("Opened sled database at {}", path);
+        info!("Opened sled database at {}", config.database_path);
 
         let (admin_sender, admin_receiver) = mpsc::unbounded();
 
         let db = Self {
-            globals: globals::Globals::load(db.open_tree("global")?, config)?,
+            globals: globals::Globals::load(
+                db.open_tree("global")?,
+                db.open_tree("servertimeout_signingkey")?,
+                config,
+            )?,
             users: users::Users {
                 userid_password: db.open_tree("userid_password")?,
                 userid_displayname: db.open_tree("userid_displayname")?,
@@ -155,9 +163,12 @@ impl Database {
                 roomuserid_invited: db.open_tree("roomuserid_invited")?,
                 userroomid_left: db.open_tree("userroomid_left")?,
 
+                statekey_short: db.open_tree("statekey_short")?,
                 stateid_pduid: db.open_tree("stateid_pduid")?,
                 pduid_statehash: db.open_tree("pduid_statehash")?,
                 roomid_statehash: db.open_tree("roomid_statehash")?,
+                eventid_outlierpdu: db.open_tree("roomeventid_outlierpdu")?,
+                prevevent_parent: db.open_tree("prevevent_parent")?,
             },
             account_data: account_data::AccountData {
                 roomuserdataid_accountdata: db.open_tree("roomuserdataid_accountdata")?,
@@ -176,10 +187,16 @@ impl Database {
             sending: sending::Sending {
                 servernamepduids: db.open_tree("servernamepduids")?,
                 servercurrentpdus: db.open_tree("servercurrentpdus")?,
+                maximum_requests: Arc::new(Semaphore::new(10)),
             },
             admin: admin::Admin {
                 sender: admin_sender,
             },
+            appservice: appservice::Appservice {
+                cached_registrations: Arc::new(RwLock::new(HashMap::new())),
+                id_appserviceregistrations: db.open_tree("id_appserviceregistrations")?,
+            },
+            pusher: pusher::PushData::new(&db)?,
             _db: db,
         };
 

src/database/account_data.rs

@@ -74,6 +74,7 @@ impl AccountData {
     }
 
     /// Returns all changes to the account data that happened after `since`.
+    #[tracing::instrument(skip(self))]
     pub fn changes_since(
         &self,
         room_id: Option<&RoomId>,

src/database/admin.rs

@@ -7,10 +7,11 @@ use ruma::{
     events::{room::message, EventType},
     UserId,
 };
-use tokio::select;
 
 pub enum AdminCommand {
-    SendTextMessage(message::TextMessageEventContent),
+    RegisterAppservice(serde_yaml::Value),
+    ListAppservices,
+    SendMessage(message::MessageEventContent),
 }
 
 #[derive(Clone)]
@@ -41,31 +42,48 @@ impl Admin {
                 .unwrap();
 
             if conduit_room.is_none() {
-                warn!("Conduit instance does not have an #admins room. Logging to that room will not work.");
+                warn!("Conduit instance does not have an #admins room. Logging to that room will not work. Restart Conduit after creating a user to fix this.");
             }
 
+            let send_message = |message: message::MessageEventContent| {
+                if let Some(conduit_room) = &conduit_room {
+                    db.rooms
+                        .build_and_append_pdu(
+                            PduBuilder {
+                                event_type: EventType::RoomMessage,
+                                content: serde_json::to_value(message)
+                                    .expect("event is valid, we just created it"),
+                                unsigned: None,
+                                state_key: None,
+                                redacts: None,
+                            },
+                            &conduit_user,
+                            &conduit_room,
+                            &db,
+                        )
+                        .unwrap();
+                }
+            };
+
             loop {
-                select! {
+                tokio::select! {
                     Some(event) = receiver.next() => {
                         match event {
-                            AdminCommand::SendTextMessage(message) => {
-                                if let Some(conduit_room) = &conduit_room {
-                                    db.rooms.build_and_append_pdu(
-                                        PduBuilder {
-                                            event_type: EventType::RoomMessage,
-                                            content: serde_json::to_value(message).expect("event is valid, we just created it"),
-                                            unsigned: None,
-                                            state_key: None,
-                                            redacts: None,
-                                        },
-                                        &conduit_user,
-                                        &conduit_room,
-                                        &db.globals,
-                                        &db.sending,
-                                        &db.admin,
-                                        &db.account_data,
-                                    ).unwrap();
-                                }
+                            AdminCommand::RegisterAppservice(yaml) => {
+                                db.appservice.register_appservice(yaml).unwrap(); // TODO handle error
+                            }
+                            AdminCommand::ListAppservices => {
+                                let appservices = db.appservice.iter_ids().collect::<Vec<_>>();
+                                let count = appservices.len();
+                                let output = format!(
+                                    "Appservices ({}): {}",
+                                    count,
+                                    appservices.into_iter().filter_map(|r| r.ok()).collect::<Vec<_>>().join(", ")
+                                );
+                                send_message(message::MessageEventContent::text_plain(output));
+                            }
+                            AdminCommand::SendMessage(message) => {
+                                send_message(message);
                             }
                         }
                     }

src/database/appservice.rs

@@ -0,0 +1,69 @@
+use crate::{utils, Error, Result};
+use std::{
+    collections::HashMap,
+    sync::{Arc, RwLock},
+};
+
+#[derive(Clone)]
+pub struct Appservice {
+    pub(super) cached_registrations: Arc<RwLock<HashMap<String, serde_yaml::Value>>>,
+    pub(super) id_appserviceregistrations: sled::Tree,
+}
+
+impl Appservice {
+    pub fn register_appservice(&self, yaml: serde_yaml::Value) -> Result<()> {
+        // TODO: Rumaify
+        let id = yaml.get("id").unwrap().as_str().unwrap();
+        self.id_appserviceregistrations
+            .insert(id, serde_yaml::to_string(&yaml).unwrap().as_bytes())?;
+        self.cached_registrations
+            .write()
+            .unwrap()
+            .insert(id.to_owned(), yaml);
+
+        Ok(())
+    }
+
+    pub fn get_registration(&self, id: &str) -> Result<Option<serde_yaml::Value>> {
+        self.cached_registrations
+            .read()
+            .unwrap()
+            .get(id)
+            .map_or_else(
+                || {
+                    Ok(self
+                        .id_appserviceregistrations
+                        .get(id)?
+                        .map(|bytes| {
+                            Ok::<_, Error>(serde_yaml::from_slice(&bytes).map_err(|_| {
+                                Error::bad_database(
+                                    "Invalid registration bytes in id_appserviceregistrations.",
+                                )
+                            })?)
+                        })
+                        .transpose()?)
+                },
+                |r| Ok(Some(r.clone())),
+            )
+    }
+
+    pub fn iter_ids(&self) -> impl Iterator<Item = Result<String>> {
+        self.id_appserviceregistrations.iter().keys().map(|id| {
+            Ok(utils::string_from_bytes(&id?).map_err(|_| {
+                Error::bad_database("Invalid id bytes in id_appserviceregistrations.")
+            })?)
+        })
+    }
+
+    pub fn iter_all<'a>(
+        &'a self,
+    ) -> impl Iterator<Item = Result<(String, serde_yaml::Value)>> + 'a {
+        self.iter_ids().filter_map(|id| id.ok()).map(move |id| {
+            Ok((
+                id.clone(),
+                self.get_registration(&id)?
+                    .expect("iter_ids only returns appservices that exist"),
+            ))
+        })
+    }
+}

src/database/globals.rs

@@ -1,20 +1,39 @@
 use crate::{database::Config, utils, Error, Result};
 use log::error;
-use ruma::ServerName;
-use std::sync::Arc;
+use ruma::{
+    api::federation::discovery::{ServerSigningKeys, VerifyKey},
+    ServerName, ServerSigningKeyId,
+};
+use std::{
+    collections::{BTreeMap, HashMap},
+    sync::{Arc, RwLock},
+    time::Duration,
+};
+use trust_dns_resolver::TokioAsyncResolver;
 
 pub const COUNTER: &str = "c";
 
+pub type DestinationCache = Arc<RwLock<HashMap<Box<ServerName>, (String, Option<String>)>>>;
+type WellKnownMap = HashMap<Box<ServerName>, (String, Option<String>)>;
+
 #[derive(Clone)]
 pub struct Globals {
+    pub actual_destination_cache: Arc<RwLock<WellKnownMap>>, // actual_destination, host
     pub(super) globals: sled::Tree,
+    config: Config,
     keypair: Arc<ruma::signatures::Ed25519KeyPair>,
     reqwest_client: reqwest::Client,
-    config: Config,
+    dns_resolver: TokioAsyncResolver,
+    jwt_decoding_key: Option<jsonwebtoken::DecodingKey<'static>>,
+    pub(super) servertimeout_signingkey: sled::Tree, // ServerName + Timeout Timestamp -> algorithm:key + pubkey
 }
 
 impl Globals {
-    pub fn load(globals: sled::Tree, config: Config) -> Result<Self> {
+    pub fn load(
+        globals: sled::Tree,
+        servertimeout_signingkey: sled::Tree,
+        config: Config,
+    ) -> Result<Self> {
         let bytes = &*globals
             .update_and_fetch("keypair", utils::generate_keypair)?
             .expect("utils::generate_keypair always returns Some");
@@ -49,11 +68,29 @@ impl Globals {
             }
         };
 
+        let reqwest_client = reqwest::Client::builder()
+            .connect_timeout(Duration::from_secs(30))
+            .timeout(Duration::from_secs(60 * 3))
+            .pool_max_idle_per_host(1)
+            .build()
+            .unwrap();
+
+        let jwt_decoding_key = config
+            .jwt_secret
+            .as_ref()
+            .map(|secret| jsonwebtoken::DecodingKey::from_secret(secret.as_bytes()).into_static());
+
         Ok(Self {
             globals,
-            keypair: Arc::new(keypair),
-            reqwest_client: reqwest::Client::new(),
             config,
+            keypair: Arc::new(keypair),
+            reqwest_client,
+            dns_resolver: TokioAsyncResolver::tokio_from_system_conf().map_err(|_| {
+                Error::bad_config("Failed to set up trust dns resolver with system config.")
+            })?,
+            actual_destination_cache: Arc::new(RwLock::new(HashMap::new())),
+            servertimeout_signingkey,
+            jwt_decoding_key,
         })
     }
 
@@ -92,15 +129,87 @@ impl Globals {
         self.config.max_request_size
     }
 
-    pub fn registration_disabled(&self) -> bool {
-        self.config.registration_disabled
+    pub fn allow_registration(&self) -> bool {
+        self.config.allow_registration
     }
 
-    pub fn encryption_disabled(&self) -> bool {
-        self.config.encryption_disabled
+    pub fn allow_encryption(&self) -> bool {
+        self.config.allow_encryption
     }
 
-    pub fn federation_enabled(&self) -> bool {
-        self.config.federation_enabled
+    pub fn allow_federation(&self) -> bool {
+        self.config.allow_federation
+    }
+
+    pub fn trusted_servers(&self) -> &[Box<ServerName>] {
+        &self.config.trusted_servers
+    }
+
+    pub fn dns_resolver(&self) -> &TokioAsyncResolver {
+        &self.dns_resolver
+    }
+
+    pub fn jwt_decoding_key(&self) -> Option<&jsonwebtoken::DecodingKey<'_>> {
+        self.jwt_decoding_key.as_ref()
+    }
+
+    /// TODO: the key valid until timestamp is only honored in room version > 4
+    /// Remove the outdated keys and insert the new ones.
+    ///
+    /// This doesn't actually check that the keys provided are newer than the old set.
+    pub fn add_signing_key(&self, origin: &ServerName, keys: &ServerSigningKeys) -> Result<()> {
+        let mut key1 = origin.as_bytes().to_vec();
+        key1.push(0xff);
+
+        let mut key2 = key1.clone();
+
+        let ts = keys
+            .valid_until_ts
+            .duration_since(std::time::UNIX_EPOCH)
+            .expect("time is valid")
+            .as_millis() as u64;
+
+        key1.extend_from_slice(&ts.to_be_bytes());
+        key2.extend_from_slice(&(ts + 1).to_be_bytes());
+
+        self.servertimeout_signingkey.insert(
+            key1,
+            serde_json::to_vec(&keys.verify_keys).expect("ServerSigningKeys are a valid string"),
+        )?;
+
+        self.servertimeout_signingkey.insert(
+            key2,
+            serde_json::to_vec(&keys.old_verify_keys)
+                .expect("ServerSigningKeys are a valid string"),
+        )?;
+
+        Ok(())
+    }
+
+    /// This returns an empty `Ok(BTreeMap<..>)` when there are no keys found for the server.
+    pub fn signing_keys_for(
+        &self,
+        origin: &ServerName,
+    ) -> Result<BTreeMap<ServerSigningKeyId, VerifyKey>> {
+        let mut response = BTreeMap::new();
+
+        let now = crate::utils::millis_since_unix_epoch();
+
+        for item in self.servertimeout_signingkey.scan_prefix(origin.as_bytes()) {
+            let (k, bytes) = item?;
+            let valid_until = k
+                .splitn(2, |&b| b == 0xff)
+                .nth(1)
+                .map(crate::utils::u64_from_bytes)
+                .ok_or_else(|| Error::bad_database("Invalid signing keys."))?
+                .map_err(|_| Error::bad_database("Invalid signing key valid until bytes"))?;
+            // If these keys are still valid use em!
+            if valid_until > now {
+                let btree: BTreeMap<_, _> = serde_json::from_slice(&bytes)
+                    .map_err(|_| Error::bad_database("Invalid BTreeMap<> of signing keys"))?;
+                response.extend(btree);
+            }
+        }
+        Ok(response)
     }
 }

src/database/key_backups.rs

@@ -2,7 +2,7 @@ use crate::{utils, Error, Result};
 use ruma::{
     api::client::{
         error::ErrorKind,
-        r0::backup::{BackupAlgorithm, KeyData, Sessions},
+        r0::backup::{BackupAlgorithm, KeyBackupData, RoomKeyBackup},
     },
     RoomId, UserId,
 };
@@ -129,7 +129,7 @@ impl KeyBackups {
         version: &str,
         room_id: &RoomId,
         session_id: &str,
-        key_data: &KeyData,
+        key_data: &KeyBackupData,
         globals: &super::globals::Globals,
     ) -> Result<()> {
         let mut key = user_id.to_string().as_bytes().to_vec();
@@ -153,7 +153,7 @@ impl KeyBackups {
 
         self.backupkeyid_backup.insert(
             &key,
-            &*serde_json::to_string(&key_data).expect("KeyData::to_string always works"),
+            &*serde_json::to_string(&key_data).expect("KeyBackupData::to_string always works"),
         )?;
 
         Ok(())
@@ -182,13 +182,17 @@ impl KeyBackups {
         .to_string())
     }
 
-    pub fn get_all(&self, user_id: &UserId, version: &str) -> Result<BTreeMap<RoomId, Sessions>> {
+    pub fn get_all(
+        &self,
+        user_id: &UserId,
+        version: &str,
+    ) -> Result<BTreeMap<RoomId, RoomKeyBackup>> {
         let mut prefix = user_id.to_string().as_bytes().to_vec();
         prefix.push(0xff);
         prefix.extend_from_slice(version.as_bytes());
         prefix.push(0xff);
 
-        let mut rooms = BTreeMap::<RoomId, Sessions>::new();
+        let mut rooms = BTreeMap::<RoomId, RoomKeyBackup>::new();
 
         for result in self.backupkeyid_backup.scan_prefix(&prefix).map(|r| {
             let (key, value) = r?;
@@ -211,15 +215,16 @@ impl KeyBackups {
             )
             .map_err(|_| Error::bad_database("backupkeyid_backup room_id is invalid room id."))?;
 
-            let key_data = serde_json::from_slice(&value)
-                .map_err(|_| Error::bad_database("KeyData in backupkeyid_backup is invalid."))?;
+            let key_data = serde_json::from_slice(&value).map_err(|_| {
+                Error::bad_database("KeyBackupData in backupkeyid_backup is invalid.")
+            })?;
 
             Ok::<_, Error>((room_id, session_id, key_data))
         }) {
             let (room_id, session_id, key_data) = result?;
             rooms
                 .entry(room_id)
-                .or_insert_with(|| Sessions {
+                .or_insert_with(|| RoomKeyBackup {
                     sessions: BTreeMap::new(),
                 })
                 .sessions
@@ -234,7 +239,7 @@ impl KeyBackups {
         user_id: &UserId,
         version: &str,
         room_id: &RoomId,
-    ) -> BTreeMap<String, KeyData> {
+    ) -> BTreeMap<String, KeyBackupData> {
         let mut prefix = user_id.to_string().as_bytes().to_vec();
         prefix.push(0xff);
         prefix.extend_from_slice(version.as_bytes());
@@ -257,7 +262,7 @@ impl KeyBackups {
                     })?;
 
                 let key_data = serde_json::from_slice(&value).map_err(|_| {
-                    Error::bad_database("KeyData in backupkeyid_backup is invalid.")
+                    Error::bad_database("KeyBackupData in backupkeyid_backup is invalid.")
                 })?;
 
                 Ok::<_, Error>((session_id, key_data))
@@ -272,7 +277,7 @@ impl KeyBackups {
         version: &str,
         room_id: &RoomId,
         session_id: &str,
-    ) -> Result<Option<KeyData>> {
+    ) -> Result<Option<KeyBackupData>> {
         let mut key = user_id.to_string().as_bytes().to_vec();
         key.push(0xff);
         key.extend_from_slice(version.as_bytes());
@@ -284,8 +289,9 @@ impl KeyBackups {
         self.backupkeyid_backup
             .get(&key)?
             .map(|value| {
-                serde_json::from_slice(&value)
-                    .map_err(|_| Error::bad_database("KeyData in backupkeyid_backup is invalid."))
+                serde_json::from_slice(&value).map_err(|_| {
+                    Error::bad_database("KeyBackupData in backupkeyid_backup is invalid.")
+                })
             })
             .transpose()
     }

src/database/media.rs

@@ -290,7 +290,12 @@ impl Media {
                     file: thumbnail_bytes.to_vec(),
                 }))
             } else {
-                Ok(None)
+                // Couldn't parse file to generate thumbnail, send original
+                Ok(Some(FileMeta {
+                    filename,
+                    content_type,
+                    file: file.to_vec(),
+                }))
             }
         } else {
             Ok(None)

src/database/pusher.rs

@@ -0,0 +1,546 @@
+use crate::{Database, Error, PduEvent, Result};
+use log::{error, info, warn};
+use ruma::{
+    api::{
+        client::r0::push::{Pusher, PusherKind},
+        push_gateway::send_event_notification::{
+            self,
+            v1::{Device, Notification, NotificationCounts, NotificationPriority},
+        },
+        OutgoingRequest,
+    },
+    events::room::{
+        member::{MemberEventContent, MembershipState},
+        message::{MessageEventContent, MessageType, TextMessageEventContent},
+        power_levels::PowerLevelsEventContent,
+    },
+    events::EventType,
+    push::{Action, PushCondition, PushFormat, Ruleset, Tweak},
+    uint, UInt, UserId,
+};
+
+use std::{convert::TryFrom, fmt::Debug, time::Duration};
+
+#[derive(Debug, Clone)]
+pub struct PushData {
+    /// UserId + pushkey -> Pusher
+    pub(super) senderkey_pusher: sled::Tree,
+}
+
+impl PushData {
+    pub fn new(db: &sled::Db) -> Result<Self> {
+        Ok(Self {
+            senderkey_pusher: db.open_tree("senderkey_pusher")?,
+        })
+    }
+
+    pub fn set_pusher(&self, sender: &UserId, pusher: Pusher) -> Result<()> {
+        println!("CCCCCCCCCCCCCCCCCCCCCc");
+        dbg!(&pusher);
+        let mut key = sender.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(pusher.pushkey.as_bytes());
+
+        // There are 2 kinds of pushers but the spec says: null deletes the pusher.
+        if pusher.kind.is_none() {
+            return self
+                .senderkey_pusher
+                .remove(key)
+                .map(|_| ())
+                .map_err(Into::into);
+        }
+
+        self.senderkey_pusher.insert(
+            dbg!(key),
+            &*serde_json::to_string(&pusher).expect("Pusher is valid JSON string"),
+        )?;
+
+        Ok(())
+    }
+
+    pub fn get_pusher(&self, sender: &UserId) -> Result<Vec<Pusher>> {
+        let mut prefix = sender.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        self.senderkey_pusher
+            .scan_prefix(dbg!(prefix))
+            .values()
+            .map(|push| {
+                println!("DDDDDDDDDDDDDDDDDDDDDDDDDD");
+                let push =
+                    dbg!(push).map_err(|_| Error::bad_database("Invalid push bytes in db."))?;
+                Ok(serde_json::from_slice(&*push)
+                    .map_err(|_| Error::bad_database("Invalid Pusher in db."))?)
+            })
+            .collect()
+    }
+}
+
+pub async fn send_request<T: OutgoingRequest>(
+    globals: &crate::database::globals::Globals,
+    destination: &str,
+    request: T,
+) -> Result<T::IncomingResponse>
+where
+    T: Debug,
+{
+    let destination = destination.replace("/_matrix/push/v1/notify", "");
+
+    let http_request = request
+        .try_into_http_request(&destination, Some(""))
+        .map_err(|e| {
+            warn!("Failed to find destination {}: {}", destination, e);
+            Error::BadServerResponse("Invalid destination")
+        })?;
+
+    let reqwest_request = reqwest::Request::try_from(http_request)
+        .expect("all http requests are valid reqwest requests");
+
+    // TODO: we could keep this very short and let expo backoff do it's thing...
+    //*reqwest_request.timeout_mut() = Some(Duration::from_secs(5));
+
+    let url = reqwest_request.url().clone();
+    let reqwest_response = globals
+        .reqwest_client()
+        .execute(dbg!(reqwest_request))
+        .await;
+
+    // Because reqwest::Response -> http::Response is complicated:
+    match reqwest_response {
+        Ok(mut reqwest_response) => {
+            let status = reqwest_response.status();
+            let mut http_response = http::Response::builder().status(status);
+            let headers = http_response.headers_mut().unwrap();
+
+            for (k, v) in reqwest_response.headers_mut().drain() {
+                if let Some(key) = k {
+                    headers.insert(key, v);
+                }
+            }
+
+            let status = reqwest_response.status();
+
+            let body = reqwest_response
+                .bytes()
+                .await
+                .unwrap_or_else(|e| {
+                    warn!("server error {}", e);
+                    Vec::new().into()
+                }) // TODO: handle timeout
+                .into_iter()
+                .collect::<Vec<_>>();
+
+            if status != 200 {
+                info!(
+                    "Push gateway returned bad response {} {}\n{}\n{:?}",
+                    destination,
+                    status,
+                    url,
+                    crate::utils::string_from_bytes(&body)
+                );
+            }
+
+            let response = T::IncomingResponse::try_from(
+                http_response
+                    .body(body)
+                    .expect("reqwest body is valid http body"),
+            );
+            response.map_err(|_| {
+                info!(
+                    "Push gateway returned invalid response bytes {}\n{}",
+                    destination, url
+                );
+                Error::BadServerResponse("Push gateway returned bad response.")
+            })
+        }
+        Err(e) => Err(e.into()),
+    }
+}
+
+pub async fn send_push_notice(
+    user: &UserId,
+    unread: UInt,
+    pushers: &[Pusher],
+    ruleset: Ruleset,
+    pdu: &PduEvent,
+    db: &Database,
+) -> Result<()> {
+    if let Some(msgtype) = pdu.content.get("msgtype").and_then(|b| b.as_str()) {
+        if msgtype == "m.notice" {
+            return Ok(());
+        }
+    }
+
+    for rule in ruleset.into_iter() {
+        // TODO: can actions contain contradictory Actions
+        if rule
+            .actions
+            .iter()
+            .any(|act| matches!(act, ruma::push::Action::DontNotify))
+            || !rule.enabled
+        {
+            continue;
+        }
+
+        match dbg!(rule.rule_id.as_str()) {
+            ".m.rule.master" => {}
+            ".m.rule.suppress_notices" => {
+                if pdu.kind == EventType::RoomMessage
+                    && pdu
+                        .content
+                        .get("msgtype")
+                        .map_or(false, |ty| ty == "m.notice")
+                {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.invite_for_me" => {
+                if let EventType::RoomMember = &pdu.kind {
+                    if pdu.state_key.as_deref() == Some(user.as_str())
+                        && serde_json::from_value::<MemberEventContent>(pdu.content.clone())
+                            .map_err(|_| Error::bad_database("PDU contained bad message content"))?
+                            .membership
+                            == MembershipState::Invite
+                    {
+                        let tweaks = rule
+                            .actions
+                            .iter()
+                            .filter_map(|a| match a {
+                                Action::SetTweak(tweak) => Some(tweak.clone()),
+                                _ => None,
+                            })
+                            .collect::<Vec<_>>();
+                        send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                            .await?;
+                        break;
+                    }
+                }
+            }
+            ".m.rule.member_event" => {
+                if let EventType::RoomMember = &pdu.kind {
+                    // TODO use this?
+                    let _member = serde_json::from_value::<MemberEventContent>(pdu.content.clone())
+                        .map_err(|_| Error::bad_database("PDU contained bad message content"))?;
+                    if let Some(conditions) = rule.conditions {
+                        if conditions.iter().any(|cond| match cond {
+                            PushCondition::EventMatch { key, pattern } => {
+                                let mut json =
+                                    serde_json::to_value(pdu).expect("PDU is valid JSON");
+                                for key in key.split('.') {
+                                    json = json[key].clone();
+                                }
+                                // TODO: this is baddddd
+                                json.to_string().contains(pattern)
+                            }
+                            _ => false,
+                        }) {
+                            let tweaks = rule
+                                .actions
+                                .iter()
+                                .filter_map(|a| match a {
+                                    Action::SetTweak(tweak) => Some(tweak.clone()),
+                                    _ => None,
+                                })
+                                .collect::<Vec<_>>();
+                            send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                                .await?;
+                            break;
+                        }
+                    }
+                }
+            }
+            ".m.rule.contains_display_name" => {
+                if let EventType::RoomMessage = &pdu.kind {
+                    let msg_content =
+                        serde_json::from_value::<MessageEventContent>(pdu.content.clone())
+                            .map_err(|_| {
+                                Error::bad_database("PDU contained bad message content")
+                            })?;
+                    if let MessageType::Text(TextMessageEventContent { body, .. }) =
+                        &msg_content.msgtype
+                    {
+                        if body.contains(user.localpart()) {
+                            let tweaks = rule
+                                .actions
+                                .iter()
+                                .filter_map(|a| match a {
+                                    Action::SetTweak(tweak) => Some(tweak.clone()),
+                                    _ => None,
+                                })
+                                .collect::<Vec<_>>();
+                            send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                                .await?;
+                            break;
+                        }
+                    }
+                }
+            }
+            ".m.rule.tombstone" => {
+                if pdu.kind == EventType::RoomTombstone && pdu.state_key.as_deref() == Some("") {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.roomnotif" => {
+                if let EventType::RoomMessage = &pdu.kind {
+                    let msg_content =
+                        serde_json::from_value::<MessageEventContent>(pdu.content.clone())
+                            .map_err(|_| {
+                                Error::bad_database("PDU contained bad message content")
+                            })?;
+                    if let MessageType::Text(TextMessageEventContent { body, .. }) =
+                        &msg_content.msgtype
+                    {
+                        let power_level_cmp = |pl: PowerLevelsEventContent| {
+                            &pl.notifications.room
+                                <= pl.users.get(&pdu.sender).unwrap_or(&ruma::int!(0))
+                        };
+                        let deserialize = |pl: PduEvent| {
+                            serde_json::from_value::<PowerLevelsEventContent>(pl.content).ok()
+                        };
+                        if body.contains("@room")
+                            && db
+                                .rooms
+                                .room_state_get(&pdu.room_id, &EventType::RoomPowerLevels, "")?
+                                .map(|(_, pl)| pl)
+                                .map(deserialize)
+                                .flatten()
+                                .map_or(false, power_level_cmp)
+                        {
+                            let tweaks = rule
+                                .actions
+                                .iter()
+                                .filter_map(|a| match a {
+                                    Action::SetTweak(tweak) => Some(tweak.clone()),
+                                    _ => None,
+                                })
+                                .collect::<Vec<_>>();
+                            send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                                .await?;
+                            break;
+                        }
+                    }
+                }
+            }
+            ".m.rule.contains_user_name" => {
+                if let EventType::RoomMessage = &pdu.kind {
+                    let msg_content =
+                        serde_json::from_value::<MessageEventContent>(pdu.content.clone())
+                            .map_err(|_| {
+                                Error::bad_database("PDU contained bad message content")
+                            })?;
+                    if let MessageType::Text(TextMessageEventContent { body, .. }) =
+                        &msg_content.msgtype
+                    {
+                        if body.contains(user.localpart()) {
+                            let tweaks = rule
+                                .actions
+                                .iter()
+                                .filter_map(|a| match a {
+                                    Action::SetTweak(tweak) => Some(tweak.clone()),
+                                    _ => None,
+                                })
+                                .collect::<Vec<_>>();
+                            send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str())
+                                .await?;
+                            break;
+                        }
+                    }
+                }
+            }
+            ".m.rule.call" => {
+                if pdu.kind == EventType::CallInvite {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.encrypted_room_one_to_one" => {
+                if db.rooms.room_members(&pdu.room_id).count() == 2
+                    && pdu.kind == EventType::RoomEncrypted
+                {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.room_one_to_one" => {
+                if db.rooms.room_members(&pdu.room_id).count() == 2
+                    && pdu.kind == EventType::RoomMessage
+                {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.message" => {
+                if pdu.kind == EventType::RoomMessage {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            ".m.rule.encrypted" => {
+                if pdu.kind == EventType::RoomEncrypted {
+                    let tweaks = rule
+                        .actions
+                        .iter()
+                        .filter_map(|a| match a {
+                            Action::SetTweak(tweak) => Some(tweak.clone()),
+                            _ => None,
+                        })
+                        .collect::<Vec<_>>();
+                    send_notice(unread, pushers, tweaks, pdu, db, rule.rule_id.as_str()).await?;
+                    break;
+                }
+            }
+            _ => {}
+        }
+    }
+
+    Ok(())
+}
+
+async fn send_notice(
+    unread: UInt,
+    pushers: &[Pusher],
+    tweaks: Vec<Tweak>,
+    event: &PduEvent,
+    db: &Database,
+    name: &str,
+) -> Result<()> {
+    println!("BBBBBBBBBBBBBBBr");
+    let (http, _emails): (Vec<&Pusher>, _) = dbg!(pushers)
+        .iter()
+        .partition(|pusher| pusher.kind == Some(PusherKind::Http));
+
+    // TODO:
+    // Two problems with this
+    // 1. if "event_id_only" is the only format kind it seems we should never add more info
+    // 2. can pusher/devices have conflicting formats
+    for pusher in dbg!(http) {
+        let event_id_only = pusher.data.format == Some(PushFormat::EventIdOnly);
+        let url = if let Some(url) = pusher.data.url.as_ref() {
+            url
+        } else {
+            error!("Http Pusher must have URL specified.");
+            continue;
+        };
+
+        let mut device = Device::new(pusher.app_id.clone(), pusher.pushkey.clone());
+        let mut data_minus_url = pusher.data.clone();
+        // The url must be stripped off according to spec
+        data_minus_url.url = None;
+        device.data = Some(data_minus_url);
+
+        // Tweaks are only added if the format is NOT event_id_only
+        if !event_id_only {
+            device.tweaks = tweaks.clone();
+        }
+
+        let d = &[device];
+        let mut notifi = Notification::new(d);
+
+        notifi.prio = NotificationPriority::Low;
+        notifi.event_id = Some(&event.event_id);
+        notifi.room_id = Some(&event.room_id);
+        // TODO: missed calls
+        notifi.counts = NotificationCounts::new(unread, uint!(0));
+
+        if event.kind == EventType::RoomEncrypted
+            || tweaks
+                .iter()
+                .any(|t| matches!(t, Tweak::Highlight(true) | Tweak::Sound(_)))
+        {
+            notifi.prio = NotificationPriority::High
+        }
+
+        if event_id_only {
+            error!("SEND PUSH NOTICE `{}`", name);
+            send_request(
+                &db.globals,
+                &url,
+                send_event_notification::v1::Request::new(notifi),
+            )
+            .await?;
+        } else {
+            notifi.sender = Some(&event.sender);
+            notifi.event_type = Some(&event.kind);
+            notifi.content = serde_json::value::to_raw_value(&event.content).ok();
+
+            if event.kind == EventType::RoomMember {
+                notifi.user_is_target = event.state_key.as_deref() == Some(event.sender.as_str());
+            }
+
+            let user_name = db.users.displayname(&event.sender)?;
+            notifi.sender_display_name = user_name.as_deref();
+            let room_name = db
+                .rooms
+                .room_state_get(&event.room_id, &EventType::RoomName, "")?
+                .map(|(_, pdu)| match pdu.content.get("name") {
+                    Some(serde_json::Value::String(s)) => Some(s.to_string()),
+                    _ => None,
+                })
+                .flatten();
+            notifi.room_name = room_name.as_deref();
+
+            error!("SEND PUSH NOTICE Full `{}`", name);
+            send_request(
+                &db.globals,
+                &url,
+                send_event_notification::v1::Request::new(notifi),
+            )
+            .await?;
+        }
+    }
+
+    // TODO: email
+    // for email in emails {}
+
+    Ok(())
+}

src/database/rooms/edus.rs

@@ -1,5 +1,4 @@
 use crate::{utils, Error, Result};
-use js_int::UInt;
 use ruma::{
     events::{
         presence::{PresenceEvent, PresenceEventContent},
@@ -7,7 +6,7 @@ use ruma::{
     },
     presence::PresenceState,
     serde::Raw,
-    RoomId, UserId,
+    RoomId, UInt, UserId,
 };
 use std::{
     collections::HashMap,
@@ -71,6 +70,7 @@ impl RoomEdus {
     }
 
     /// Returns an iterator over the most recent read_receipts in a room that happened after the event with id `since`.
+    #[tracing::instrument(skip(self))]
     pub fn readreceipts_since(
         &self,
         room_id: &RoomId,
@@ -116,6 +116,7 @@ impl RoomEdus {
     }
 
     /// Returns the private read marker.
+    #[tracing::instrument(skip(self))]
     pub fn private_read_get(&self, room_id: &RoomId, user_id: &UserId) -> Result<Option<u64>> {
         let mut key = room_id.to_string().as_bytes().to_vec();
         key.push(0xff);
@@ -257,6 +258,7 @@ impl RoomEdus {
     }
 
     /// Returns the count of the last typing update in this room.
+    #[tracing::instrument(skip(self, globals))]
     pub fn last_typing_update(
         &self,
         room_id: &RoomId,
@@ -340,6 +342,7 @@ impl RoomEdus {
     }
 
     /// Resets the presence timeout, so the user will stay in their current presence state.
+    #[tracing::instrument(skip(self))]
     pub fn ping_presence(&self, user_id: &UserId) -> Result<()> {
         self.userid_lastpresenceupdate.insert(
             &user_id.to_string().as_bytes(),
@@ -386,8 +389,6 @@ impl RoomEdus {
             .take_while(|(_, timestamp)| current_timestamp - timestamp > 5 * 60_000)
         // 5 Minutes
         {
-            self.userid_lastpresenceupdate.remove(&user_id_bytes)?;
-
             // Send new presence events to set the user offline
             let count = globals.next_count()?.to_be_bytes();
             let user_id = utils::string_from_bytes(&user_id_bytes)
@@ -421,12 +422,18 @@ impl RoomEdus {
                     .expect("PresenceEvent can be serialized"),
                 )?;
             }
+
+            self.userid_lastpresenceupdate.insert(
+                &user_id.to_string().as_bytes(),
+                &utils::millis_since_unix_epoch().to_be_bytes(),
+            )?;
         }
 
         Ok(())
     }
 
     /// Returns an iterator over the most recent presence updates that happened after the event with id `since`.
+    #[tracing::instrument(skip(self, globals, rooms))]
     pub fn presence_since(
         &self,
         room_id: &RoomId,

src/database/sending.rs

@@ -1,82 +1,132 @@
-use std::{collections::HashMap, convert::TryFrom, time::SystemTime};
+use std::{
+    collections::HashMap,
+    convert::TryFrom,
+    fmt::Debug,
+    sync::Arc,
+    time::{Duration, Instant, SystemTime},
+};
 
-use crate::{server_server, utils, Error, PduEvent, Result};
+use crate::{
+    appservice_server, database::pusher, server_server, utils, Database, Error, PduEvent, Result,
+};
 use federation::transactions::send_transaction_message;
-use log::{debug, warn};
+use log::{info, warn};
+use ring::digest;
 use rocket::futures::stream::{FuturesUnordered, StreamExt};
-use ruma::{api::federation, ServerName};
+use ruma::{
+    api::{appservice, federation, OutgoingRequest},
+    events::{push_rules, EventType},
+    uint, ServerName, UInt,
+};
 use sled::IVec;
-use tokio::select;
+use tokio::{select, sync::Semaphore};
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum OutgoingKind {
+    Appservice(Box<ServerName>),
+    Push(Vec<u8>),
+    Normal(Box<ServerName>),
+}
 
 #[derive(Clone)]
 pub struct Sending {
     /// The state for a given state hash.
-    pub(super) servernamepduids: sled::Tree, // ServernamePduId = ServerName + PduId
-    pub(super) servercurrentpdus: sled::Tree, // ServerCurrentPdus = ServerName + PduId (pduid can be empty for reservation)
+    pub(super) servernamepduids: sled::Tree, // ServernamePduId = (+ / $)ServerName / UserId + PduId
+    pub(super) servercurrentpdus: sled::Tree, // ServerCurrentPdus = (+ / $)ServerName / UserId + PduId (pduid can be empty for reservation)
+    pub(super) maximum_requests: Arc<Semaphore>,
 }
 
 impl Sending {
-    pub fn start_handler(&self, globals: &super::globals::Globals, rooms: &super::rooms::Rooms) {
+    pub fn start_handler(&self, db: &Database) {
         let servernamepduids = self.servernamepduids.clone();
         let servercurrentpdus = self.servercurrentpdus.clone();
-        let rooms = rooms.clone();
-        let globals = globals.clone();
+
+        let db = db.clone();
 
         tokio::spawn(async move {
             let mut futures = FuturesUnordered::new();
 
             // Retry requests we could not finish yet
-            let mut current_transactions = HashMap::new();
+            let mut current_transactions = HashMap::<OutgoingKind, Vec<IVec>>::new();
 
-            for (server, pdu) in servercurrentpdus
+            for (key, outgoing_kind, pdu) in servercurrentpdus
                 .iter()
                 .filter_map(|r| r.ok())
-                .map(|(key, _)| {
-                    let mut parts = key.splitn(2, |&b| b == 0xff);
-                    let server = parts.next().expect("splitn always returns one element");
-                    let pdu = parts.next().ok_or_else(|| {
-                        Error::bad_database("Invalid bytes in servercurrentpdus.")
-                    })?;
-
-                    Ok::<_, Error>((
-                        Box::<ServerName>::try_from(utils::string_from_bytes(&server).map_err(
-                            |_| {
-                                Error::bad_database(
-                                    "Invalid server bytes in server_currenttransaction",
-                                )
-                            },
-                        )?)
-                        .map_err(|_| {
-                            Error::bad_database(
-                                "Invalid server string in server_currenttransaction",
-                            )
-                        })?,
-                        IVec::from(pdu),
-                    ))
+                .filter_map(|(key, _)| {
+                    Self::parse_servercurrentpdus(&key)
+                        .ok()
+                        .map(|(k, p)| (key, k, p))
                 })
-                .filter_map(|r| r.ok())
-                .filter(|(_, pdu)| !pdu.is_empty()) // Skip reservation key
-                .take(50)
-            // This should not contain more than 50 anyway
             {
-                current_transactions
-                    .entry(server)
-                    .or_insert_with(Vec::new)
-                    .push(pdu);
+                if pdu.is_empty() {
+                    // Remove old reservation key
+                    servercurrentpdus.remove(key).unwrap();
+                    continue;
+                }
+
+                let entry = current_transactions
+                    .entry(outgoing_kind)
+                    .or_insert_with(Vec::new);
+
+                if entry.len() > 30 {
+                    warn!("Dropping some current pdus because too many were queued. This should not happen.");
+                    servercurrentpdus.remove(key).unwrap();
+                    continue;
+                }
+
+                entry.push(pdu);
             }
 
-            for (server, pdus) in current_transactions {
-                futures.push(Self::handle_event(server, pdus, &globals, &rooms));
+            for (outgoing_kind, pdus) in current_transactions {
+                // Create new reservation
+                let mut prefix = match &outgoing_kind {
+                    OutgoingKind::Appservice(server) => {
+                        let mut p = b"+".to_vec();
+                        p.extend_from_slice(server.as_bytes());
+                        p
+                    }
+                    OutgoingKind::Push(id) => {
+                        let mut p = b"$".to_vec();
+                        p.extend_from_slice(&id);
+                        p
+                    }
+                    OutgoingKind::Normal(server) => {
+                        let mut p = Vec::new();
+                        p.extend_from_slice(server.as_bytes());
+                        p
+                    }
+                };
+                prefix.push(0xff);
+                servercurrentpdus.insert(prefix, &[]).unwrap();
+
+                futures.push(Self::handle_event(outgoing_kind.clone(), pdus, &db));
             }
 
+            let mut last_failed_try: HashMap<OutgoingKind, (u32, Instant)> = HashMap::new();
+
             let mut subscriber = servernamepduids.watch_prefix(b"");
             loop {
                 select! {
-                    Some(server) = futures.next() => {
-                        debug!("sending response: {:?}", &server);
-                        match server {
-                            Ok((server, _response)) => {
-                                let mut prefix = server.as_bytes().to_vec();
+                    Some(response) = futures.next() => {
+                        match response {
+                            Ok(outgoing_kind) => {
+                                let mut prefix = match &outgoing_kind {
+                                    OutgoingKind::Appservice(server) => {
+                                        let mut p = b"+".to_vec();
+                                        p.extend_from_slice(server.as_bytes());
+                                        p
+                                    }
+                                    OutgoingKind::Push(id) => {
+                                        let mut p = b"$".to_vec();
+                                        p.extend_from_slice(&id);
+                                        p
+                                    },
+                                    OutgoingKind::Normal(server) => {
+                                        let mut p = vec![];
+                                        p.extend_from_slice(server.as_bytes());
+                                        p
+                                    },
+                                };
                                 prefix.push(0xff);
 
                                 for key in servercurrentpdus
@@ -98,7 +148,7 @@ impl Sending {
                                     .map(|k| {
                                         k.subslice(prefix.len(), k.len() - prefix.len())
                                     })
-                                    .take(50)
+                                    .take(30)
                                     .collect::<Vec<_>>();
 
                                 if !new_pdus.is_empty() {
@@ -109,15 +159,49 @@ impl Sending {
                                         servernamepduids.remove(&current_key).unwrap();
                                     }
 
-                                    futures.push(Self::handle_event(server, new_pdus, &globals, &rooms));
+                                    futures.push(
+                                        Self::handle_event(
+                                            outgoing_kind.clone(),
+                                            new_pdus,
+                                            &db,
+                                        )
+                                    );
                                 } else {
                                     servercurrentpdus.remove(&prefix).unwrap();
                                     // servercurrentpdus with the prefix should be empty now
                                 }
                             }
-                            Err((server, e)) => {
-                                warn!("Couldn't send transaction to {}: {}", server, e)
-                                // TODO: exponential backoff
+                            Err((outgoing_kind, e)) => {
+                                info!("Couldn't send transaction to {:?}\n{}", outgoing_kind, e);
+                                let mut prefix = match &outgoing_kind {
+                                    OutgoingKind::Appservice(serv) => {
+                                        let mut p = b"+".to_vec();
+                                        p.extend_from_slice(serv.as_bytes());
+                                        p
+                                    },
+                                    OutgoingKind::Push(id) => {
+                                        let mut p = b"$".to_vec();
+                                        p.extend_from_slice(&id);
+                                        p
+                                    },
+                                    OutgoingKind::Normal(serv) => {
+                                        let mut p = vec![];
+                                        p.extend_from_slice(serv.as_bytes());
+                                        p
+                                    },
+                                };
+
+                                prefix.push(0xff);
+
+                                last_failed_try.insert(outgoing_kind.clone(), match last_failed_try.get(&outgoing_kind) {
+                                    Some(last_failed) => {
+                                        (last_failed.0+1, Instant::now())
+                                    },
+                                    None => {
+                                        (1, Instant::now())
+                                    }
+                                });
+                                servercurrentpdus.remove(&prefix).unwrap();
                             }
                         };
                     },
@@ -126,24 +210,65 @@ impl Sending {
                             let servernamepduid = key.clone();
                             let mut parts = servernamepduid.splitn(2, |&b| b == 0xff);
 
-                            if let Some((server, pdu_id)) = utils::string_from_bytes(
+                            let exponential_backoff = |(tries, instant): &(u32, Instant)| {
+                                // Fail if a request has failed recently (exponential backoff)
+                                let mut min_elapsed_duration = Duration::from_secs(60) * (*tries) * (*tries);
+                                if min_elapsed_duration > Duration::from_secs(60*60*24) {
+                                    min_elapsed_duration = Duration::from_secs(60*60*24);
+                                }
+
+                                instant.elapsed() < min_elapsed_duration
+                            };
+                            if let Some((outgoing_kind, pdu_id)) = utils::string_from_bytes(
                                     parts
                                         .next()
                                         .expect("splitn will always return 1 or more elements"),
                                 )
-                                .map_err(|_| Error::bad_database("ServerName in servernamepduid bytes are invalid."))
-                                .and_then(|server_str| Box::<ServerName>::try_from(server_str)
-                                    .map_err(|_| Error::bad_database("ServerName in servernamepduid is invalid.")))
-                                .ok()
-                                .and_then(|server| parts
+                                .map_err(|_| Error::bad_database("[Utf8] ServerName in servernamepduid bytes are invalid."))
+                                .and_then(|ident_str| {
+                                    // Appservices start with a plus
+                                    Ok(if ident_str.starts_with('+') {
+                                        OutgoingKind::Appservice(
+                                            Box::<ServerName>::try_from(&ident_str[1..])
+                                                .map_err(|_| Error::bad_database("ServerName in servernamepduid is invalid."))?
+                                        )
+                                    } else if ident_str.starts_with('$') {
+                                        OutgoingKind::Push(ident_str[1..].as_bytes().to_vec())
+                                    } else {
+                                        OutgoingKind::Normal(
+                                            Box::<ServerName>::try_from(ident_str)
+                                                .map_err(|_| Error::bad_database("ServerName in servernamepduid is invalid."))?
+                                        )
+                                    })
+                                })
+                                .and_then(|outgoing_kind| parts
                                     .next()
                                     .ok_or_else(|| Error::bad_database("Invalid servernamepduid in db."))
-                                    .ok()
-                                    .map(|pdu_id| (server, pdu_id))
+                                    .map(|pdu_id| (outgoing_kind, pdu_id))
                                 )
-                                // TODO: exponential backoff
-                                .filter(|(server, _)| {
-                                    let mut prefix = server.to_string().as_bytes().to_vec();
+                                .ok()
+                                .filter(|(outgoing_kind, _)| {
+                                    if last_failed_try.get(outgoing_kind).map_or(false, exponential_backoff) {
+                                        return false;
+                                    }
+
+                                    let mut prefix = match outgoing_kind {
+                                        OutgoingKind::Appservice(serv) => {
+                                            let mut p = b"+".to_vec();
+                                            p.extend_from_slice(serv.as_bytes());
+                                            p
+                                    },
+                                        OutgoingKind::Push(id) => {
+                                            let mut p = b"$".to_vec();
+                                            p.extend_from_slice(&id);
+                                            p
+                                        },
+                                        OutgoingKind::Normal(serv) => {
+                                            let mut p = vec![];
+                                            p.extend_from_slice(serv.as_bytes());
+                                            p
+                                        },
+                                    };
                                     prefix.push(0xff);
 
                                     servercurrentpdus
@@ -154,7 +279,13 @@ impl Sending {
                                 servercurrentpdus.insert(&key, &[]).unwrap();
                                 servernamepduids.remove(&key).unwrap();
 
-                                futures.push(Self::handle_event(server, vec![pdu_id.into()], &globals, &rooms));
+                                futures.push(
+                                    Self::handle_event(
+                                        outgoing_kind,
+                                        vec![pdu_id.into()],
+                                        &db,
+                                    )
+                                );
                             }
                         }
                     }
@@ -163,6 +294,26 @@ impl Sending {
         });
     }
 
+    #[tracing::instrument(skip(self))]
+    pub fn send_push_pdu(&self, pdu_id: &[u8]) -> Result<()> {
+        // Make sure we don't cause utf8 errors when parsing to a String...
+        let pduid = String::from_utf8_lossy(pdu_id).as_bytes().to_vec();
+
+        // these are valid ServerName chars
+        // (byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'.')
+        let mut key = b"$".to_vec();
+        // keep each pdu push unique
+        key.extend_from_slice(pduid.as_slice());
+        key.push(0xff);
+        key.extend_from_slice(pdu_id);
+        self.servernamepduids.insert(key, b"")?;
+
+        println!("AAAA");
+
+        Ok(())
+    }
+
+    #[tracing::instrument(skip(self))]
     pub fn send_pdu(&self, server: &ServerName, pdu_id: &[u8]) -> Result<()> {
         let mut key = server.as_bytes().to_vec();
         key.push(0xff);
@@ -172,56 +323,287 @@ impl Sending {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
+    pub fn send_pdu_appservice(&self, appservice_id: &str, pdu_id: &[u8]) -> Result<()> {
+        let mut key = b"+".to_vec();
+        key.extend_from_slice(appservice_id.as_bytes());
+        key.push(0xff);
+        key.extend_from_slice(pdu_id);
+        self.servernamepduids.insert(key, b"")?;
+
+        Ok(())
+    }
+
+    #[tracing::instrument]
+    fn calculate_hash(keys: &[IVec]) -> Vec<u8> {
+        // We only hash the pdu's event ids, not the whole pdu
+        let bytes = keys.join(&0xff);
+        let hash = digest::digest(&digest::SHA256, &bytes);
+        hash.as_ref().to_owned()
+    }
+
+    #[tracing::instrument(skip(db))]
     async fn handle_event(
-        server: Box<ServerName>,
+        kind: OutgoingKind,
         pdu_ids: Vec<IVec>,
-        globals: &super::globals::Globals,
-        rooms: &super::rooms::Rooms,
-    ) -> std::result::Result<
-        (Box<ServerName>, send_transaction_message::v1::Response),
-        (Box<ServerName>, Error),
-    > {
-        let pdu_jsons = pdu_ids
-            .iter()
-            .map(|pdu_id| {
-                Ok::<_, (Box<ServerName>, Error)>(
-                    // TODO: check room version and remove event_id if needed
-                    serde_json::from_str(
-                        PduEvent::convert_to_outgoing_federation_event(
-                            rooms
-                                .get_pdu_json_from_id(pdu_id)
+        db: &Database,
+    ) -> std::result::Result<OutgoingKind, (OutgoingKind, Error)> {
+        match dbg!(&kind) {
+            OutgoingKind::Appservice(server) => {
+                let pdu_jsons = pdu_ids
+                    .iter()
+                    .map(|pdu_id| {
+                        Ok::<_, (Box<ServerName>, Error)>(
+                            db.rooms
+                                .get_pdu_from_id(pdu_id)
                                 .map_err(|e| (server.clone(), e))?
                                 .ok_or_else(|| {
                                     (
                                         server.clone(),
                                         Error::bad_database(
-                                            "Event in servernamepduids not found in db.",
+                                            "[Appservice] Event in servernamepduids not found in ",
+                                        ),
+                                    )
+                                })?
+                                .to_any_event(),
+                        )
+                    })
+                    .filter_map(|r| r.ok())
+                    .collect::<Vec<_>>();
+                let permit = db.sending.maximum_requests.acquire().await;
+                let response = appservice_server::send_request(
+                    &db.globals,
+                    db.appservice
+                        .get_registration(server.as_str())
+                        .unwrap()
+                        .unwrap(), // TODO: handle error
+                    appservice::event::push_events::v1::Request {
+                        events: &pdu_jsons,
+                        txn_id: &base64::encode_config(
+                            Self::calculate_hash(&pdu_ids),
+                            base64::URL_SAFE_NO_PAD,
+                        ),
+                    },
+                )
+                .await
+                .map(|_response| kind.clone())
+                .map_err(|e| (kind, e));
+
+                drop(permit);
+
+                response
+            }
+            OutgoingKind::Push(id) => {
+                let pdus = pdu_ids
+                    .iter()
+                    .map(|pdu_id| {
+                        Ok::<_, (Vec<u8>, Error)>(
+                            db.rooms
+                                .get_pdu_from_id(pdu_id)
+                                .map_err(|e| (id.clone(), e))?
+                                .ok_or_else(|| {
+                                    (
+                                        id.clone(),
+                                        Error::bad_database(
+                                            "[Push] Event in servernamepduids not found in db.",
                                         ),
                                     )
                                 })?,
                         )
-                        .json()
-                        .get(),
-                    )
-                    .expect("Raw<..> is always valid"),
-                )
-            })
-            .filter_map(|r| r.ok())
-            .collect::<Vec<_>>();
+                    })
+                    .filter_map(|r| r.ok())
+                    .collect::<Vec<_>>();
 
-        server_server::send_request(
-            &globals,
-            server.clone(),
-            send_transaction_message::v1::Request {
-                origin: globals.server_name(),
-                pdus: &pdu_jsons,
-                edus: &[],
-                origin_server_ts: SystemTime::now(),
-                transaction_id: &utils::random_string(16),
-            },
-        )
-        .await
-        .map(|response| (server.clone(), response))
-        .map_err(|e| (server, e))
+                for pdu in dbg!(&pdus) {
+                    // Redacted events are not notification targets (we don't send push for them)
+                    if pdu.unsigned.get("redacted_because").is_some() {
+                        continue;
+                    }
+
+                    for user in db.rooms.room_members(&pdu.room_id) {
+                        let user = user.map_err(|e| (OutgoingKind::Push(id.clone()), e))?;
+
+                        // Don't notify the user of their own events
+                        if user == pdu.sender {
+                            continue;
+                        }
+
+                        let pushers = dbg!(db.pusher.get_pusher(&user))
+                            .map_err(|e| (OutgoingKind::Push(id.clone()), e))?;
+
+                        let rules_for_user = db
+                            .account_data
+                            .get::<push_rules::PushRulesEvent>(None, &user, EventType::PushRules)
+                            .map_err(|e| (OutgoingKind::Push(id.clone()), e))?
+                            .map(|ev| ev.content.global)
+                            .unwrap_or_else(|| crate::push_rules::default_pushrules(&user));
+
+                        let unread: UInt = if let Some(last_read) = db
+                            .rooms
+                            .edus
+                            .private_read_get(&pdu.room_id, &user)
+                            .map_err(|e| (OutgoingKind::Push(id.clone()), e))?
+                        {
+                            (db.rooms
+                                .pdus_since(&user, &pdu.room_id, last_read)
+                                .map_err(|e| (OutgoingKind::Push(id.clone()), e))?
+                                .filter_map(|pdu| pdu.ok()) // Filter out buggy events
+                                .filter(|(_, pdu)| {
+                                    matches!(
+                                        pdu.kind.clone(),
+                                        EventType::RoomMessage | EventType::RoomEncrypted
+                                    )
+                                })
+                                .count() as u32)
+                                .into()
+                        } else {
+                            // Just return zero unread messages
+                            uint!(0)
+                        };
+
+                        let permit = db.sending.maximum_requests.acquire().await;
+                        let _response = pusher::send_push_notice(
+                            &user,
+                            unread,
+                            &pushers,
+                            rules_for_user,
+                            pdu,
+                            db,
+                        )
+                        .await
+                        .map(|_response| kind.clone())
+                        .map_err(|e| (kind.clone(), e));
+
+                        drop(permit);
+                    }
+                }
+                Ok(OutgoingKind::Push(id.clone()))
+            }
+            OutgoingKind::Normal(server) => {
+                let pdu_jsons = pdu_ids
+                    .iter()
+                    .map(|pdu_id| {
+                        Ok::<_, (OutgoingKind, Error)>(
+                            // TODO: check room version and remove event_id if needed
+                            serde_json::from_str(
+                                PduEvent::convert_to_outgoing_federation_event(
+                                    db.rooms
+                                        .get_pdu_json_from_id(pdu_id)
+                                        .map_err(|e| (OutgoingKind::Normal(server.clone()), e))?
+                                        .ok_or_else(|| {
+                                            (
+                                                OutgoingKind::Normal(server.clone()),
+                                                Error::bad_database(
+                                                    "[Normal] Event in servernamepduids not found in db.",
+                                                ),
+                                            )
+                                        })?,
+                                )
+                                .json()
+                                .get(),
+                            )
+                            .expect("Raw<..> is always valid"),
+                        )
+                    })
+                    .filter_map(|r| r.ok())
+                    .collect::<Vec<_>>();
+
+                let permit = db.sending.maximum_requests.acquire().await;
+
+                info!("sending pdus to {}: {:#?}", server, pdu_jsons);
+                let response = server_server::send_request(
+                    &db.globals,
+                    &*server,
+                    send_transaction_message::v1::Request {
+                        origin: db.globals.server_name(),
+                        pdus: &pdu_jsons,
+                        edus: &[],
+                        origin_server_ts: SystemTime::now(),
+                        transaction_id: &base64::encode_config(
+                            Self::calculate_hash(&pdu_ids),
+                            base64::URL_SAFE_NO_PAD,
+                        ),
+                    },
+                )
+                .await
+                .map(|response| {
+                    info!("server response: {:?}", response);
+                    kind.clone()
+                })
+                .map_err(|e| (kind, e));
+
+                drop(permit);
+
+                response
+            }
+        }
+    }
+
+    fn parse_servercurrentpdus(key: &IVec) -> Result<(OutgoingKind, IVec)> {
+        let mut parts = key.splitn(2, |&b| b == 0xff);
+        let server = parts.next().expect("splitn always returns one element");
+        let pdu = parts
+            .next()
+            .ok_or_else(|| Error::bad_database("Invalid bytes in servercurrentpdus."))?;
+
+        let server = utils::string_from_bytes(&server).map_err(|_| {
+            Error::bad_database("Invalid server bytes in server_currenttransaction")
+        })?;
+
+        // Appservices start with a plus
+        Ok::<_, Error>(if server.starts_with('+') {
+            (
+                OutgoingKind::Appservice(Box::<ServerName>::try_from(server).map_err(|_| {
+                    Error::bad_database("Invalid server string in server_currenttransaction")
+                })?),
+                IVec::from(pdu),
+            )
+        } else if server.starts_with('$') {
+            (
+                OutgoingKind::Push(server.as_bytes().to_vec()),
+                IVec::from(pdu),
+            )
+        } else {
+            (
+                OutgoingKind::Normal(Box::<ServerName>::try_from(server).map_err(|_| {
+                    Error::bad_database("Invalid server string in server_currenttransaction")
+                })?),
+                IVec::from(pdu),
+            )
+        })
+    }
+
+    #[tracing::instrument(skip(self, globals))]
+    pub async fn send_federation_request<T: OutgoingRequest>(
+        &self,
+        globals: &crate::database::globals::Globals,
+        destination: &ServerName,
+        request: T,
+    ) -> Result<T::IncomingResponse>
+    where
+        T: Debug,
+    {
+        let permit = self.maximum_requests.acquire().await;
+        let response = server_server::send_request(globals, destination, request).await;
+        drop(permit);
+
+        response
+    }
+
+    #[tracing::instrument(skip(self, globals))]
+    pub async fn send_appservice_request<T: OutgoingRequest>(
+        &self,
+        globals: &crate::database::globals::Globals,
+        registration: serde_yaml::Value,
+        request: T,
+    ) -> Result<T::IncomingResponse>
+    where
+        T: Debug,
+    {
+        let permit = self.maximum_requests.acquire().await;
+        let response = appservice_server::send_request(globals, registration, request).await;
+        drop(permit);
+
+        response
     }
 }

src/database/transaction_ids.rs

@@ -11,13 +11,13 @@ impl TransactionIds {
     pub fn add_txnid(
         &self,
         user_id: &UserId,
-        device_id: &DeviceId,
+        device_id: Option<&DeviceId>,
         txn_id: &str,
         data: &[u8],
     ) -> Result<()> {
         let mut key = user_id.as_bytes().to_vec();
         key.push(0xff);
-        key.extend_from_slice(device_id.as_bytes());
+        key.extend_from_slice(device_id.map(|d| d.as_bytes()).unwrap_or_default());
         key.push(0xff);
         key.extend_from_slice(txn_id.as_bytes());
 
@@ -29,12 +29,12 @@ impl TransactionIds {
     pub fn existing_txnid(
         &self,
         user_id: &UserId,
-        device_id: &DeviceId,
+        device_id: Option<&DeviceId>,
         txn_id: &str,
     ) -> Result<Option<IVec>> {
         let mut key = user_id.as_bytes().to_vec();
         key.push(0xff);
-        key.extend_from_slice(device_id.as_bytes());
+        key.extend_from_slice(device_id.map(|d| d.as_bytes()).unwrap_or_default());
         key.push(0xff);
         key.extend_from_slice(txn_id.as_bytes());
 

src/database/users.rs

@@ -1,5 +1,4 @@
 use crate::{utils, Error, Result};
-use js_int::UInt;
 use ruma::{
     api::client::{
         error::ErrorKind,
@@ -11,7 +10,7 @@ use ruma::{
     encryption::DeviceKeys,
     events::{AnyToDeviceEvent, EventType},
     serde::Raw,
-    DeviceId, DeviceKeyAlgorithm, DeviceKeyId, UserId,
+    DeviceId, DeviceKeyAlgorithm, DeviceKeyId, UInt, UserId,
 };
 use std::{collections::BTreeMap, convert::TryFrom, mem, time::SystemTime};
 
@@ -252,7 +251,7 @@ impl Users {
     }
 
     /// Replaces the access token of one device.
-    fn set_token(&self, user_id: &UserId, device_id: &DeviceId, token: &str) -> Result<()> {
+    pub fn set_token(&self, user_id: &UserId, device_id: &DeviceId, token: &str) -> Result<()> {
         let mut userdeviceid = user_id.to_string().as_bytes().to_vec();
         userdeviceid.push(0xff);
         userdeviceid.extend_from_slice(device_id.as_bytes());
@@ -312,6 +311,7 @@ impl Users {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn last_one_time_keys_update(&self, user_id: &UserId) -> Result<u64> {
         self.userid_lastonetimekeyupdate
             .get(&user_id.to_string().as_bytes())?
@@ -365,6 +365,7 @@ impl Users {
             .transpose()
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn count_one_time_keys(
         &self,
         user_id: &UserId,
@@ -564,6 +565,7 @@ impl Users {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn keys_changed(
         &self,
         user_or_room_id: &str,
@@ -739,6 +741,7 @@ impl Users {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn get_to_device_events(
         &self,
         user_id: &UserId,
@@ -761,6 +764,7 @@ impl Users {
         Ok(events)
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn remove_to_device_events(
         &self,
         user_id: &UserId,

src/error.rs

@@ -1,4 +1,8 @@
-use std::{collections::HashMap, sync::RwLock, time::Duration, time::Instant};
+use std::{
+    collections::HashMap,
+    sync::RwLock,
+    time::{Duration, Instant},
+};
 
 use log::error;
 use ruma::{
@@ -34,7 +38,7 @@ pub enum Error {
         #[from]
         source: image::error::ImageError,
     },
-    #[error("Could not connect to server.")]
+    #[error("Could not connect to server: {source}")]
     ReqwestError {
         #[from]
         source: reqwest::Error,
@@ -121,29 +125,46 @@ impl log::Log for ConduitLogger {
     fn log(&self, record: &log::Record<'_>) {
         let output = format!("{} - {}", record.level(), record.args());
 
-        println!("{}", output);
+        let match_mod_path =
+            |path: &str| path.starts_with("conduit::") || path.starts_with("state");
 
         if self.enabled(record.metadata())
-            && record
-                .module_path()
-                .map_or(false, |path| path.starts_with("conduit::"))
+            && (record.module_path().map_or(false, match_mod_path)
+                || record
+                    .module_path()
+                    .map_or(true, |path| !path.starts_with("rocket::")) // Rockets logs are annoying
+                    && record.metadata().level() <= log::Level::Warn)
         {
+            let first_line = output
+                .lines()
+                .next()
+                .expect("lines always returns one item");
+
+            eprintln!("{}", output);
+
+            let mute_duration = match record.metadata().level() {
+                log::Level::Error => Duration::from_secs(60 * 5), // 5 minutes
+                log::Level::Warn => Duration::from_secs(60 * 60 * 24), // A day
+                _ => Duration::from_secs(60 * 60 * 24 * 7),       // A week
+            };
+
             if self
                 .last_logs
                 .read()
                 .unwrap()
-                .get(&output)
-                .map_or(false, |i| i.elapsed() < Duration::from_secs(60 * 30))
+                .get(first_line)
+                .map_or(false, |i| i.elapsed() < mute_duration)
+            // Don't post this log again for some time
             {
                 return;
             }
 
             if let Ok(mut_last_logs) = &mut self.last_logs.try_write() {
-                mut_last_logs.insert(output.clone(), Instant::now());
+                mut_last_logs.insert(first_line.to_owned(), Instant::now());
             }
 
-            self.db.admin.send(AdminCommand::SendTextMessage(
-                message::TextMessageEventContent::plain(output),
+            self.db.admin.send(AdminCommand::SendMessage(
+                message::MessageEventContent::notice_plain(output),
             ));
         }
     }

src/lib.rs

@@ -1,3 +1,4 @@
+pub mod appservice_server;
 pub mod client_server;
 mod database;
 mod error;

src/main.rs

@@ -1,5 +1,6 @@
 #![warn(rust_2018_idioms)]
 
+pub mod appservice_server;
 pub mod client_server;
 pub mod server_server;
 
@@ -10,20 +11,47 @@ mod push_rules;
 mod ruma_wrapper;
 mod utils;
 
+use database::Config;
 pub use database::Database;
-pub use error::{ConduitLogger, Error, Result};
-use log::LevelFilter;
+pub use error::{Error, Result};
 pub use pdu::PduEvent;
 pub use rocket::State;
+use ruma::api::client::error::ErrorKind;
 pub use ruma_wrapper::{ConduitResult, Ruma, RumaResponse};
 
-use rocket::{fairing::AdHoc, routes};
+use log::LevelFilter;
+use rocket::{
+    catch, catchers,
+    fairing::AdHoc,
+    figment::{
+        providers::{Env, Format, Toml},
+        Figment,
+    },
+    routes, Request,
+};
+use tracing::span;
+use tracing_subscriber::{prelude::*, Registry};
 
-fn setup_rocket() -> rocket::Rocket {
+fn setup_rocket() -> (rocket::Rocket, Config) {
     // Force log level off, so we can use our own logger
-    std::env::set_var("ROCKET_LOG_LEVEL", "off");
+    std::env::set_var("CONDUIT_LOG_LEVEL", "off");
 
-    rocket::ignite()
+    let config =
+        Figment::from(rocket::Config::release_default())
+            .merge(
+                Toml::file(Env::var("CONDUIT_CONFIG").expect(
+                    "The CONDUIT_CONFIG env var needs to be set. Example: /etc/conduit.toml",
+                ))
+                .nested(),
+            )
+            .merge(Env::prefixed("CONDUIT_").global());
+
+    let parsed_config = config
+        .extract::<Config>()
+        .expect("It looks like your config is invalid. Please take a look at the error");
+    let parsed_config2 = parsed_config.clone();
+
+    let rocket = rocket::custom(config)
         .mount(
             "/",
             routes![
@@ -40,7 +68,12 @@ fn setup_rocket() -> rocket::Rocket {
                 client_server::get_capabilities_route,
                 client_server::get_pushrules_all_route,
                 client_server::set_pushrule_route,
+                client_server::get_pushrule_route,
                 client_server::set_pushrule_enabled_route,
+                client_server::get_pushrule_enabled_route,
+                client_server::get_pushrule_actions_route,
+                client_server::set_pushrule_actions_route,
+                client_server::delete_pushrule_route,
                 client_server::get_room_event_route,
                 client_server::get_filter_route,
                 client_server::create_filter_route,
@@ -69,6 +102,7 @@ fn setup_rocket() -> rocket::Rocket {
                 client_server::get_backup_key_sessions_route,
                 client_server::get_backup_keys_route,
                 client_server::set_read_marker_route,
+                client_server::create_receipt_route,
                 client_server::create_typing_event_route,
                 client_server::create_room_route,
                 client_server::redact_event_route,
@@ -122,6 +156,7 @@ fn setup_rocket() -> rocket::Rocket {
                 client_server::get_key_changes_route,
                 client_server::get_pushers_route,
                 client_server::set_pushers_route,
+                // client_server::third_party_route,
                 client_server::upgrade_room_route,
                 server_server::get_server_version_route,
                 server_server::get_server_keys_route,
@@ -133,24 +168,76 @@ fn setup_rocket() -> rocket::Rocket {
                 server_server::get_profile_information_route,
             ],
         )
+        .register(catchers![
+            not_found_catcher,
+            forbidden_catcher,
+            unknown_token_catcher,
+            missing_token_catcher,
+            bad_json_catcher
+        ])
         .attach(AdHoc::on_attach("Config", |rocket| async {
-            let data =
-                Database::load_or_create(rocket.figment().extract().expect("config is valid"))
-                    .expect("config is valid");
+            let data = Database::load_or_create(parsed_config2)
+                .await
+                .expect("config is valid");
 
-            data.sending.start_handler(&data.globals, &data.rooms);
-            log::set_boxed_logger(Box::new(ConduitLogger {
-                db: data.clone(),
-                last_logs: Default::default(),
-            }))
-            .unwrap();
-            log::set_max_level(LevelFilter::Info);
+            data.sending.start_handler(&data);
 
             Ok(rocket.manage(data))
-        }))
+        }));
+
+    (rocket, parsed_config)
 }
 
 #[rocket::main]
 async fn main() {
-    setup_rocket().launch().await.unwrap();
+    let (rocket, config) = setup_rocket();
+
+    if config.allow_jaeger {
+        let (tracer, _uninstall) = opentelemetry_jaeger::new_pipeline()
+            .with_service_name("conduit")
+            .install()
+            .unwrap();
+        let telemetry = tracing_opentelemetry::layer().with_tracer(tracer);
+        Registry::default().with(telemetry).try_init().unwrap();
+
+        let root = span!(tracing::Level::INFO, "app_start", work_units = 2);
+        let _enter = root.enter();
+
+        rocket.launch().await.unwrap();
+    } else {
+        pretty_env_logger::init();
+
+        let root = span!(tracing::Level::INFO, "app_start", work_units = 2);
+        let _enter = root.enter();
+
+        rocket.launch().await.unwrap();
+    }
+}
+
+#[catch(404)]
+fn not_found_catcher(_: &Request<'_>) -> String {
+    "404 Not Found".to_owned()
+}
+
+#[catch(580)]
+fn forbidden_catcher() -> Result<()> {
+    Err(Error::BadRequest(ErrorKind::Forbidden, "Forbidden."))
+}
+
+#[catch(581)]
+fn unknown_token_catcher() -> Result<()> {
+    Err(Error::BadRequest(
+        ErrorKind::UnknownToken { soft_logout: false },
+        "Unknown token.",
+    ))
+}
+
+#[catch(582)]
+fn missing_token_catcher() -> Result<()> {
+    Err(Error::BadRequest(ErrorKind::MissingToken, "Missing token."))
+}
+
+#[catch(583)]
+fn bad_json_catcher() -> Result<()> {
+    Err(Error::BadRequest(ErrorKind::BadJson, "Bad json."))
 }

src/pdu.rs

@@ -1,23 +1,17 @@
 use crate::Error;
-use js_int::UInt;
 use ruma::{
     events::{
         pdu::EventHash, room::member::MemberEventContent, AnyEvent, AnyRoomEvent, AnyStateEvent,
         AnyStrippedStateEvent, AnySyncRoomEvent, AnySyncStateEvent, EventType, StateEvent,
     },
     serde::{to_canonical_value, CanonicalJsonObject, CanonicalJsonValue, Raw},
-    EventId, RoomId, RoomVersionId, ServerName, ServerSigningKeyId, UserId,
+    EventId, RoomId, RoomVersionId, ServerName, ServerSigningKeyId, UInt, UserId,
 };
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use std::{
-    collections::BTreeMap,
-    convert::{TryFrom, TryInto},
-    sync::Arc,
-    time::UNIX_EPOCH,
-};
+use std::{cmp::Ordering, collections::BTreeMap, convert::TryFrom, time::UNIX_EPOCH};
 
-#[derive(Deserialize, Serialize, Debug)]
+#[derive(Clone, Deserialize, Serialize, Debug)]
 pub struct PduEvent {
     pub event_id: EventId,
     pub room_id: RoomId,
@@ -33,13 +27,14 @@ pub struct PduEvent {
     pub auth_events: Vec<EventId>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub redacts: Option<EventId>,
-    #[serde(default, skip_serializing_if = "serde_json::Map::is_empty")]
-    pub unsigned: serde_json::Map<String, serde_json::Value>,
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub unsigned: BTreeMap<String, serde_json::Value>,
     pub hashes: EventHash,
     pub signatures: BTreeMap<Box<ServerName>, BTreeMap<ServerSigningKeyId, String>>,
 }
 
 impl PduEvent {
+    #[tracing::instrument(skip(self))]
     pub fn redact(&mut self, reason: &PduEvent) -> crate::Result<()> {
         self.unsigned.clear();
 
@@ -86,6 +81,7 @@ impl PduEvent {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn to_sync_room_event(&self) -> Raw<AnySyncRoomEvent> {
         let mut json = json!({
             "content": self.content,
@@ -107,6 +103,7 @@ impl PduEvent {
     }
 
     /// This only works for events that are also AnyRoomEvents.
+    #[tracing::instrument(skip(self))]
     pub fn to_any_event(&self) -> Raw<AnyEvent> {
         let mut json = json!({
             "content": self.content,
@@ -128,6 +125,7 @@ impl PduEvent {
         serde_json::from_value(json).expect("Raw::from_value always works")
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn to_room_event(&self) -> Raw<AnyRoomEvent> {
         let mut json = json!({
             "content": self.content,
@@ -149,6 +147,7 @@ impl PduEvent {
         serde_json::from_value(json).expect("Raw::from_value always works")
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn to_state_event(&self) -> Raw<AnyStateEvent> {
         let json = json!({
             "content": self.content,
@@ -164,20 +163,27 @@ impl PduEvent {
         serde_json::from_value(json).expect("Raw::from_value always works")
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn to_sync_state_event(&self) -> Raw<AnySyncStateEvent> {
-        let json = json!({
-            "content": self.content,
-            "type": self.kind,
-            "event_id": self.event_id,
-            "sender": self.sender,
-            "origin_server_ts": self.origin_server_ts,
-            "unsigned": self.unsigned,
-            "state_key": self.state_key,
-        });
+        let json = format!(
+            r#"{{"content":{},"type":"{}","event_id":"{}","sender":"{}","origin_server_ts":{},"unsigned":{},"state_key":"{}"}}"#,
+            self.content,
+            self.kind,
+            self.event_id,
+            self.sender,
+            self.origin_server_ts,
+            serde_json::to_string(&self.unsigned).expect("Map::to_string always works"),
+            self.state_key
+                .as_ref()
+                .expect("state events have state keys")
+        );
 
-        serde_json::from_value(json).expect("Raw::from_value always works")
+        Raw::from_json(
+            serde_json::value::RawValue::from_string(json).expect("our string is valid json"),
+        )
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn to_stripped_state_event(&self) -> Raw<AnyStrippedStateEvent> {
         let json = json!({
             "content": self.content,
@@ -189,6 +195,7 @@ impl PduEvent {
         serde_json::from_value(json).expect("Raw::from_value always works")
     }
 
+    #[tracing::instrument(skip(self))]
     pub fn to_member_event(&self) -> Raw<StateEvent<MemberEventContent>> {
         let json = json!({
             "content": self.content,
@@ -206,6 +213,7 @@ impl PduEvent {
     }
 
     /// This does not return a full `Pdu` it is only to satisfy ruma's types.
+    #[tracing::instrument]
     pub fn convert_to_outgoing_federation_event(
         mut pdu_json: CanonicalJsonObject,
     ) -> Raw<ruma::events::pdu::Pdu> {
@@ -227,72 +235,95 @@ impl PduEvent {
         )
         .expect("Raw::from_value always works")
     }
-}
 
-impl From<&state_res::StateEvent> for PduEvent {
-    fn from(pdu: &state_res::StateEvent) -> Self {
-        Self {
-            event_id: pdu.event_id(),
-            room_id: pdu.room_id().clone(),
-            sender: pdu.sender().clone(),
-            origin_server_ts: (pdu
-                .origin_server_ts()
-                .duration_since(UNIX_EPOCH)
-                .expect("time is valid")
-                .as_millis() as u64)
-                .try_into()
-                .expect("time is valid"),
-            kind: pdu.kind(),
-            content: pdu.content().clone(),
-            state_key: Some(pdu.state_key()),
-            prev_events: pdu.prev_event_ids(),
-            depth: *pdu.depth(),
-            auth_events: pdu.auth_events(),
-            redacts: pdu.redacts().cloned(),
-            unsigned: pdu.unsigned().clone().into_iter().collect(),
-            hashes: pdu.hashes().clone(),
-            signatures: pdu.signatures(),
-        }
+    pub fn from_id_val(
+        event_id: &EventId,
+        mut json: CanonicalJsonObject,
+    ) -> Result<Self, serde_json::Error> {
+        json.insert(
+            "event_id".to_string(),
+            to_canonical_value(event_id).expect("event_id is a valid Value"),
+        );
+
+        serde_json::from_value(serde_json::to_value(json).expect("valid JSON"))
     }
 }
 
-impl PduEvent {
-    pub fn convert_for_state_res(&self) -> Arc<state_res::StateEvent> {
-        Arc::new(
-            // For consistency of eventId (just in case) we use the one
-            // generated by conduit for everything.
-            state_res::StateEvent::from_id_value(
-                self.event_id.clone(),
-                json!({
-                    "event_id": self.event_id,
-                    "room_id": self.room_id,
-                    "sender": self.sender,
-                    "origin_server_ts": self.origin_server_ts,
-                    "type": self.kind,
-                    "content": self.content,
-                    "state_key": self.state_key,
-                    "prev_events": self.prev_events,
-                    "depth": self.depth,
-                    "auth_events": self.auth_events,
-                    "redacts": self.redacts,
-                    "unsigned": self.unsigned,
-                    "hashes": self.hashes,
-                    "signatures": self.signatures,
-                }),
-            )
-            .expect("all conduit PDUs are state events"),
-        )
+impl state_res::Event for PduEvent {
+    fn event_id(&self) -> &EventId {
+        &self.event_id
+    }
+
+    fn room_id(&self) -> &RoomId {
+        &self.room_id
+    }
+
+    fn sender(&self) -> &UserId {
+        &self.sender
+    }
+    fn kind(&self) -> EventType {
+        self.kind.clone()
+    }
+
+    fn content(&self) -> serde_json::Value {
+        self.content.clone()
+    }
+    fn origin_server_ts(&self) -> std::time::SystemTime {
+        UNIX_EPOCH + std::time::Duration::from_millis(self.origin_server_ts.into())
+    }
+
+    fn state_key(&self) -> Option<String> {
+        self.state_key.clone()
+    }
+    fn prev_events(&self) -> Vec<EventId> {
+        self.prev_events.to_vec()
+    }
+    fn depth(&self) -> &UInt {
+        &self.depth
+    }
+    fn auth_events(&self) -> Vec<EventId> {
+        self.auth_events.to_vec()
+    }
+    fn redacts(&self) -> Option<&EventId> {
+        self.redacts.as_ref()
+    }
+    fn hashes(&self) -> &EventHash {
+        &self.hashes
+    }
+    fn signatures(&self) -> BTreeMap<Box<ServerName>, BTreeMap<ruma::ServerSigningKeyId, String>> {
+        self.signatures.clone()
+    }
+    fn unsigned(&self) -> &BTreeMap<String, serde_json::Value> {
+        &self.unsigned
+    }
+}
+
+// These impl's allow us to dedup state snapshots when resolving state
+// for incoming events (federation/send/{txn}).
+impl Eq for PduEvent {}
+impl PartialEq for PduEvent {
+    fn eq(&self, other: &Self) -> bool {
+        self.event_id == other.event_id
+    }
+}
+impl PartialOrd for PduEvent {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        self.event_id.partial_cmp(&other.event_id)
+    }
+}
+impl Ord for PduEvent {
+    fn cmp(&self, other: &Self) -> Ordering {
+        self.event_id.cmp(&other.event_id)
     }
 }
 
 /// Generates a correct eventId for the incoming pdu.
 ///
-/// Returns a tuple of the new `EventId` and the PDU with the eventId inserted as a `serde_json::Value`.
-pub(crate) fn process_incoming_pdu(
+/// Returns a tuple of the new `EventId` and the PDU as a `BTreeMap<String, CanonicalJsonValue>`.
+pub(crate) fn gen_event_id_canonical_json(
     pdu: &Raw<ruma::events::pdu::Pdu>,
 ) -> (EventId, CanonicalJsonObject) {
-    let mut value =
-        serde_json::from_str(pdu.json().get()).expect("A Raw<...> is always valid JSON");
+    let value = serde_json::from_str(pdu.json().get()).expect("A Raw<...> is always valid JSON");
 
     let event_id = EventId::try_from(&*format!(
         "${}",
@@ -301,11 +332,6 @@ pub(crate) fn process_incoming_pdu(
     ))
     .expect("ruma's reference hashes are valid event ids");
 
-    value.insert(
-        "event_id".to_owned(),
-        to_canonical_value(&event_id).expect("EventId is a valid CanonicalJsonValue"),
-    );
-
     (event_id, value)
 }
 
@@ -315,7 +341,7 @@ pub struct PduBuilder {
     #[serde(rename = "type")]
     pub event_type: EventType,
     pub content: serde_json::Value,
-    pub unsigned: Option<serde_json::Map<String, serde_json::Value>>,
+    pub unsigned: Option<BTreeMap<String, serde_json::Value>>,
     pub state_key: Option<String>,
     pub redacts: Option<EventId>,
 }

src/ruma_wrapper.rs

@@ -1,6 +1,6 @@
 use crate::Error;
 use ruma::{
-    api::{AuthScheme, OutgoingRequest},
+    api::{AuthScheme, IncomingRequest, OutgoingRequest},
     identifiers::{DeviceId, UserId},
     Outgoing,
 };
@@ -12,7 +12,7 @@ use std::{
 #[cfg(feature = "conduit_bin")]
 use {
     crate::utils,
-    log::warn,
+    log::{debug, warn},
     rocket::{
         data::{
             ByteUnit, Data, FromDataFuture, FromTransformedData, Transform, TransformFuture,
@@ -29,22 +29,20 @@ use {
 
 /// This struct converts rocket requests into ruma structs by converting them into http requests
 /// first.
-pub struct Ruma<T: Outgoing> {
+pub struct Ruma<T: Outgoing + OutgoingRequest> {
     pub body: T::Incoming,
     pub sender_user: Option<UserId>,
     pub sender_device: Option<Box<DeviceId>>,
     pub json_body: Option<Box<serde_json::value::RawValue>>, // This is None when body is not a valid string
+    pub from_appservice: bool,
 }
 
 #[cfg(feature = "conduit_bin")]
 impl<'a, T: Outgoing + OutgoingRequest> FromTransformedData<'a> for Ruma<T>
 where
-    <T as Outgoing>::Incoming: TryFrom<http::request::Request<std::vec::Vec<u8>>> + std::fmt::Debug,
-    <<T as Outgoing>::Incoming as std::convert::TryFrom<
-        http::request::Request<std::vec::Vec<u8>>,
-    >>::Error: std::fmt::Debug,
+    T::Incoming: IncomingRequest,
 {
-    type Error = (); // TODO: Better error handling
+    type Error = ();
     type Owned = Data;
     type Borrowed = Self::Owned;
 
@@ -66,28 +64,73 @@ where
                 .await
                 .expect("database was loaded");
 
-            let (sender_user, sender_device) = match T::METADATA.authentication {
-                AuthScheme::AccessToken | AuthScheme::QueryOnlyAccessToken => {
-                    // Get token from header or query value
-                    let token = match request
-                        .headers()
-                        .get_one("Authorization")
-                        .map(|s| s[7..].to_owned()) // Split off "Bearer "
-                        .or_else(|| request.get_query_value("access_token").and_then(|r| r.ok()))
-                    {
-                        // TODO: M_MISSING_TOKEN
-                        None => return Failure((Status::Unauthorized, ())),
-                        Some(token) => token,
-                    };
+            // Get token from header or query value
+            let token = request
+                .headers()
+                .get_one("Authorization")
+                .map(|s| s[7..].to_owned()) // Split off "Bearer "
+                .or_else(|| request.get_query_value("access_token").and_then(|r| r.ok()));
 
-                    // Check if token is valid
-                    match db.users.find_from_token(&token).unwrap() {
-                        // TODO: M_UNKNOWN_TOKEN
-                        None => return Failure((Status::Unauthorized, ())),
-                        Some((user_id, device_id)) => (Some(user_id), Some(device_id.into())),
+            let (sender_user, sender_device, from_appservice) = if let Some((_id, registration)) =
+                db.appservice
+                    .iter_all()
+                    .filter_map(|r| r.ok())
+                    .find(|(_id, registration)| {
+                        registration
+                            .get("as_token")
+                            .and_then(|as_token| as_token.as_str())
+                            .map_or(false, |as_token| token.as_deref() == Some(as_token))
+                    }) {
+                match T::METADATA.authentication {
+                    AuthScheme::AccessToken | AuthScheme::QueryOnlyAccessToken => {
+                        let user_id = request.get_query_value::<String>("user_id").map_or_else(
+                            || {
+                                UserId::parse_with_server_name(
+                                    registration
+                                        .get("sender_localpart")
+                                        .unwrap()
+                                        .as_str()
+                                        .unwrap(),
+                                    db.globals.server_name(),
+                                )
+                                .unwrap()
+                            },
+                            |string| {
+                                UserId::try_from(string.expect("parsing to string always works"))
+                                    .unwrap()
+                            },
+                        );
+
+                        if !db.users.exists(&user_id).unwrap() {
+                            // Forbidden
+                            return Failure((Status::raw(580), ()));
+                        }
+
+                        // TODO: Check if appservice is allowed to be that user
+                        (Some(user_id), None, true)
                     }
+                    AuthScheme::ServerSignatures => (None, None, true),
+                    AuthScheme::None => (None, None, true),
+                }
+            } else {
+                match T::METADATA.authentication {
+                    AuthScheme::AccessToken | AuthScheme::QueryOnlyAccessToken => {
+                        if let Some(token) = token {
+                            match db.users.find_from_token(&token).unwrap() {
+                                // Unknown Token
+                                None => return Failure((Status::raw(581), ())),
+                                Some((user_id, device_id)) => {
+                                    (Some(user_id), Some(device_id.into()), false)
+                                }
+                            }
+                        } else {
+                            // Missing Token
+                            return Failure((Status::raw(582), ()));
+                        }
+                    }
+                    AuthScheme::ServerSignatures => (None, None, false),
+                    AuthScheme::None => (None, None, false),
                 }
-                _ => (None, None),
             };
 
             let mut http_request = http::Request::builder()
@@ -103,9 +146,8 @@ where
             handle.read_to_end(&mut body).await.unwrap();
 
             let http_request = http_request.body(body.clone()).unwrap();
-            log::debug!("{:?}", http_request);
-
-            match <T as Outgoing>::Incoming::try_from(http_request) {
+            debug!("{:?}", http_request);
+            match <T::Incoming as IncomingRequest>::try_from_http_request(http_request) {
                 Ok(t) => Success(Ruma {
                     body: t,
                     sender_user,
@@ -114,17 +156,18 @@ where
                     json_body: utils::string_from_bytes(&body)
                         .ok()
                         .and_then(|s| serde_json::value::RawValue::from_string(s).ok()),
+                    from_appservice,
                 }),
                 Err(e) => {
                     warn!("{:?}", e);
-                    Failure((Status::BadRequest, ()))
+                    Failure((Status::raw(583), ()))
                 }
             }
         })
     }
 }
 
-impl<T: Outgoing> Deref for Ruma<T> {
+impl<T: Outgoing + OutgoingRequest> Deref for Ruma<T> {
     type Target = T::Incoming;
 
     fn deref(&self) -> &Self::Target {

src/utils.rs

@@ -55,6 +55,7 @@ pub fn random_string(length: usize) -> String {
     thread_rng()
         .sample_iter(&rand::distributions::Alphanumeric)
         .take(length)
+        .map(char::from)
         .collect()
 }
 

tests/Complement.Dockerfile

@@ -9,7 +9,7 @@ ARG SCCACHE_ENDPOINT
 ARG SCCACHE_S3_USE_SSL
 
 COPY . .
-RUN cargo build
+RUN test -e cached_target/release/conduit || cargo build --release
 
 FROM valkum/docker-rust-ci:latest
 WORKDIR /workdir

tests/sytest/are-we-synapse-yet.list

@@ -91,14 +91,15 @@ snd PUT /rooms/:room_id/send/:event_type/:txn_id deduplicates the same txn id
 get GET /rooms/:room_id/messages returns a message
 get GET /rooms/:room_id/messages lazy loads members correctly 
 typ PUT /rooms/:room_id/typing/:user_id sets typing notification 
+typ Typing notifications don't leak (3 subtests)
 rst GET /rooms/:room_id/state/m.room.power_levels can fetch levels
 rst PUT /rooms/:room_id/state/m.room.power_levels can set levels
 rst PUT power_levels should not explode if the old power levels were empty
 rst Both GET and PUT work
 rct POST /rooms/:room_id/receipt can create receipts 
 red POST /rooms/:room_id/read_markers can create read marker
-med POST /media/v1/upload can create an upload
-med GET /media/v1/download can fetch the value again
+med POST /media/r0/upload can create an upload 
+med GET /media/r0/download can fetch the value again 
 cap GET /capabilities is present and well formed for registered user 
 cap GET /r0/capabilities is not public
 reg Register with a recaptcha 
@@ -296,7 +297,7 @@ fgt Forgotten room messages cannot be paginated
 fgt Forgetting room does not show up in v2 /sync 
 fgt Can forget room you've been kicked from 
 fgt Can't forget room you're still in
-mem Can re-join room if re-invited
+fgt Can re-join room if re-invited
 ath Only original members of the room can see messages from erased users 
 mem /joined_rooms returns only joined rooms
 mem /joined_members return joined members 
@@ -455,6 +456,19 @@ rmv User can invite remote user to room with version 5
 rmv Remote user can backfill in a room with version 5 
 rmv Can reject invites over federation for rooms with version 5 
 rmv Can receive redactions from regular users over federation in room version 5 
+rmv User can create and send/receive messages in a room with version 6 
+rmv User can create and send/receive messages in a room with version 6 (2 subtests)
+rmv local user can join room with version 6 
+rmv User can invite local user to room with version 6 
+rmv remote user can join room with version 6 
+rmv User can invite remote user to room with version 6 
+rmv Remote user can backfill in a room with version 6 
+rmv Can reject invites over federation for rooms with version 6 
+rmv Can receive redactions from regular users over federation in room version 6 
+rmv Inbound federation rejects invites which include invalid JSON for room version 6
+rmv Outbound federation rejects invite response which include invalid JSON for room version 6
+rmv Inbound federation rejects invite rejections which include invalid JSON for room version 6
+rmv Server rejects invalid JSON in a version 6 room
 pre Presence changes are reported to local room members 
 f,pre Presence changes are also reported to remote room members 
 pre Presence changes to UNAVAILABLE are reported to local room members 
@@ -482,24 +496,24 @@ f,dvk If remote user leaves room, changes device and rejoins we see update in /k
 dvk Get left notifs in sync and /keys/changes when other user leaves 
 dvk Get left notifs for other users in sync and /keys/changes when user leaves 
 f,dvk If user leaves room, remote user changes device and rejoins we see update in /sync and /keys/changes 
-dvk Can create backup version
-dvk Can update backup version
-dvk Responds correctly when backup is empty
-dvk Can backup keys
-dvk Can update keys with better versions
-dvk Will not update keys with worse versions
-dvk Will not back up to an old backup version
-dvk Can delete backup
-dvk Deleted & recreated backups are empty
-dvk Can create more than 10 backup versions
-dvk Can upload self-signing keys
-dvk Fails to upload self-signing keys with no auth
-dvk Fails to upload self-signing key without master key
-dvk Changing master key notifies local users
-dvk Changing user-signing key notifies local users
-f,dvk can fetch self-signing keys over federation
-f,dvk uploading self-signing key notifies over federation
-f,dvk uploading signed devices gets propagated over federation
+dkb Can create backup version 
+dkb Can update backup version 
+dkb Responds correctly when backup is empty 
+dkb Can backup keys 
+dkb Can update keys with better versions 
+dkb Will not update keys with worse versions 
+dkb Will not back up to an old backup version 
+dkb Can delete backup 
+dkb Deleted & recreated backups are empty 
+dkb Can create more than 10 backup versions 
+xsk Can upload self-signing keys 
+xsk Fails to upload self-signing keys with no auth 
+xsk Fails to upload self-signing key without master key 
+xsk Changing master key notifies local users 
+xsk Changing user-signing key notifies local users 
+f,xsk can fetch self-signing keys over federation 
+f,xsk uploading self-signing key notifies over federation 
+f,xsk uploading signed devices gets propagated over federation 
 tag Can add tag
 tag Can remove tag
 tag Can list tags for a room
@@ -536,11 +550,11 @@ std Can recv device messages until they are acknowledged
 std Device messages with the same txn_id are deduplicated 
 std Device messages wake up /sync 
 std Can recv device messages over federation 
-std Device messages over federation wake up /sync
+fsd Device messages over federation wake up /sync 
 std Can send messages with a wildcard device id 
 std Can send messages with a wildcard device id to two devices 
 std Wildcard device messages wake up /sync 
-std Wildcard device messages over federation wake up /sync
+fsd Wildcard device messages over federation wake up /sync 
 adm /whois 
 nsp /purge_history 
 nsp /purge_history by ts 
@@ -578,6 +592,7 @@ frv A pair of servers can establish a join in a v2 room
 fsj Outbound federation rejects send_join responses with no m.room.create event 
 frv Outbound federation rejects m.room.create events with an unknown room version 
 fsj Event with an invalid signature in the send_join response should not cause room join to fail 
+fsj Inbound: send_join rejects invalid JSON for room version 6
 fed Outbound federation can send events
 fed Inbound federation can receive events 
 fed Inbound federation can receive redacted events 
@@ -636,6 +651,7 @@ fst Name/topic keys are correct
 fau Remote servers cannot set power levels in rooms without existing powerlevels 
 fau Remote servers should reject attempts by non-creators to set the power levels 
 fau Inbound federation rejects typing notifications from wrong remote 
+fau Users cannot set notifications powerlevel higher than their own
 fed Forward extremities remain so even after the next events are populated as outliers 
 fau Banned servers cannot send events 
 fau Banned servers cannot /make_join 
@@ -815,6 +831,7 @@ psh Trying to get push rules with unknown scope fails with 400
 psh Trying to get push rules with unknown template fails with 400 
 psh Trying to get push rules with unknown attribute fails with 400 
 psh Trying to get push rules with unknown rule_id fails with 404
+psh Rooms with names are correctly named in pushes
 v1s GET /initialSync with non-numeric 'limit' 
 v1s GET /events with non-numeric 'limit' 
 v1s GET /events with negative 'limit' 
@@ -823,7 +840,7 @@ ath Event size limits
 syn Check creating invalid filters returns 4xx 
 f,pre New federated private chats get full presence information (SYN-115) 
 pre Left room members do not cause problems for presence 
-crm Rooms can be created with an initial invite list (SYN-205)
+crm Rooms can be created with an initial invite list (SYN-205) (1 subtests)
 typ Typing notifications don't leak 
 ban Non-present room members cannot ban others 
 psh Getting push rules doesn't corrupt the cache SYN-390 
@@ -834,3 +851,16 @@ gst Real user can call /events on another world_readable room (SYN-606)
 gst Events come down the correct room
 pub Asking for a remote rooms list, but supplying the local server's name, returns the local rooms list
 std Can send a to-device message to two users which both receive it using /sync
+fme Outbound federation will ignore a missing event with bad JSON for room version 6
+fbk Outbound federation rejects backfill containing invalid JSON for events in room version 6
+jso Invalid JSON integers
+jso Invalid JSON floats
+jso Invalid JSON special values
+inv Can invite users to invite-only rooms (2 subtests)
+plv setting 'm.room.name' respects room powerlevel (2 subtests)
+psh Messages that notify from another user increment notification_count
+psh Messages that org.matrix.msc2625.mark_unread from another user increment org.matrix.msc2625.unread_count
+dvk Can claim one time key using POST (2 subtests)
+fdk Can query remote device keys using POST (1 subtests)
+fdk Can claim remote one time key using POST (2 subtests)
+fmj Inbound /make_join rejects attempts to join rooms where all users have left

tests/sytest/are-we-synapse-yet.py

@@ -33,6 +33,8 @@ import sys
 
 test_mappings = {
     "nsp": "Non-Spec API",
+    "unk": "Unknown API (no group specified)",
+    "app": "Application Services API",
     "f": "Federation", # flag to mark test involves federation
 
     "federation_apis": {
@@ -50,6 +52,7 @@ test_mappings = {
         "fpb": "Public Room API",
         "fdk": "Device Key APIs",
         "fed": "Federation API",
+		"fsd": "Send-to-Device APIs",
     },
 
     "client_apis": {
@@ -61,6 +64,8 @@ test_mappings = {
         "pro": "Profile",
         "dev": "Devices",
         "dvk": "Device Keys",
+        "dkb": "Device Key Backup",
+        "xsk": "Cross-signing Keys",
         "pre": "Presence",
         "crm": "Create Room",
         "syn": "Sync API",
@@ -98,7 +103,7 @@ test_mappings = {
         "adm": "Server Admin API",
         "ign": "Ignore Users",
         "udr": "User Directory APIs",
-        "app": "Application Services API",
+		"jso": "Enforced canonical JSON",
     },
 }
 
@@ -156,6 +161,8 @@ def print_stats(header_name, gid_to_tests, gid_to_name, verbose):
     total_tests = 0
     for gid, tests in gid_to_tests.items():
         group_total = len(tests)
+        if group_total == 0:
+            continue
         group_passing = 0
         test_names_and_marks = []
         for name, passing in tests.items():
@@ -186,7 +193,6 @@ def main(results_tap_path, verbose):
     test_name_to_group_id = {}
     fed_tests = set()
     client_tests = set()
-    groupless_tests = set()
     with open("./are-we-synapse-yet.list", "r") as f:
         for line in f.readlines():
             test_name = " ".join(line.split(" ")[1:]).strip()
@@ -212,8 +218,12 @@ def main(results_tap_path, verbose):
             #   test_name: OK
             # }
         },
+        "appservice": {
+            "app": {},
+        },
         "nonspec": {
-            "nsp": {}
+            "nsp": {},
+            "unk": {}
         },
     }
     with open(results_tap_path, "r") as f:
@@ -224,10 +234,11 @@ def main(results_tap_path, verbose):
             name = test_result["name"]
             group_id = test_name_to_group_id.get(name)
             if not group_id:
-                groupless_tests.add(name)
-                # raise Exception("The test '%s' doesn't have a group" % (name,))
+                summary["nonspec"]["unk"][name] = test_result["ok"]
             if group_id == "nsp":
                 summary["nonspec"]["nsp"][name] = test_result["ok"]
+            elif group_id == "app":
+                summary["appservice"]["app"][name] = test_result["ok"]
             elif group_id in test_mappings["federation_apis"]:
                 group = summary["federation"].get(group_id, {})
                 group[name] = test_result["ok"]
@@ -243,12 +254,7 @@ def main(results_tap_path, verbose):
     print_stats("Non-Spec APIs", summary["nonspec"], test_mappings, verbose)
     print_stats("Client-Server APIs", summary["client"], test_mappings["client_apis"], verbose)
     print_stats("Federation APIs", summary["federation"], test_mappings["federation_apis"], verbose)
-    if verbose:
-        print("The following tests don't have a group:")
-        for name in groupless_tests:
-            print("  %s" % (name,))
-    else:
-        print("%d tests don't have a group" % len(groupless_tests))
+    print_stats("Application Services APIs", summary["appservice"], test_mappings, verbose)