implement MSC4166, returning M_NOT_FOUND if no TURN URIs are set

Signed-off-by: strawberry <strawberry@puppygock.gay>

Cargo.toml

@@ -19,15 +19,12 @@ license = "Apache-2.0"
 # See also `rust-toolchain.toml`
 readme = "README.md"
 repository = "https://github.com/girlbossceo/conduwuit"
-rust-version = "1.80.1"
+rust-version = "1.80.0"
 version = "0.4.6"
 
 [workspace.metadata.crane]
 name = "conduit"
 
-[workspace.dependencies.arrayvec]
-version = "0.7.4"
-
 [workspace.dependencies.const-str]
 version = "0.5.7"
 
@@ -69,7 +66,7 @@ version = "0.8.5"
 
 # Used for the http request / response body type for Ruma endpoints used with reqwest
 [workspace.dependencies.bytes]
-version = "1.7.1"
+version = "1.6.1"
 
 [workspace.dependencies.http-body-util]
 version = "0.1.1"
@@ -78,7 +75,7 @@ version = "0.1.1"
 version = "1.1.0"
 
 [workspace.dependencies.regex]
-version = "1.10.6"
+version = "1.10.4"
 
 [workspace.dependencies.axum]
 version = "0.7.5"
@@ -99,7 +96,7 @@ default-features = false
 features = ["typed-header", "tracing"]
 
 [workspace.dependencies.axum-server]
-version = "0.7.1"
+version = "0.6.0"
 default-features = false
 features = ["tls-rustls"]
 
@@ -107,7 +104,7 @@ features = ["tls-rustls"]
 version = "0.6.0"
 
 [workspace.dependencies.tower]
-version = "0.5.0"
+version = "0.4.13"
 default-features = false
 features = ["util"]
 
@@ -140,7 +137,7 @@ default-features = false
 features = ["rc"]
 
 [workspace.dependencies.serde_json]
-version = "1.0.124"
+version = "1.0.121"
 default-features = false
 features = ["raw_value"]
 
@@ -193,7 +190,7 @@ default-features = false
 
 # used for conduit's CLI and admin room command parsing
 [workspace.dependencies.clap]
-version = "4.5.15"
+version = "4.5.9"
 default-features = false
 features = [
 	"std",
@@ -307,7 +304,7 @@ version = "0.1.2"
 [workspace.dependencies.ruma]
 git = "https://github.com/girlbossceo/ruwuma"
 #branch = "conduwuit-changes"
-rev = "11cbd52af1039f3d4a0b96333acd1d1687ba01aa"
+rev = "c76e2873c1593a3308d4ba3e0e4a1db65acf8536"
 features = [
     "compat",
     "rand",
@@ -431,11 +428,11 @@ features = [
 ]
 
 [workspace.dependencies.rustyline-async]
-version = "0.4.3"
+version = "0.4.2"
 default-features = false
 
 [workspace.dependencies.termimad]
-version = "0.30.0"
+version = "0.29.4"
 default-features = false
 
 [workspace.dependencies.checked_ops]
@@ -473,11 +470,17 @@ rev = "4d78a14a5e03f539b8c6b475aefa08bb14e4de91"
 git = "https://github.com/girlbossceo/tracing"
 rev = "4d78a14a5e03f539b8c6b475aefa08bb14e4de91"
 
+# fixes hyper graceful shutdowns [https://github.com/programatik29/axum-server/issues/114]
+# https://github.com/girlbossceo/axum-server/commit/8e3368d899079818934e61cc9c839abcbbcada8a
+[patch.crates-io.axum-server]
+git = "https://github.com/girlbossceo/axum-server"
+rev = "8e3368d899079818934e61cc9c839abcbbcada8a"
+
 # adds a tab completion callback: https://github.com/girlbossceo/rustyline-async/commit/de26100b0db03e419a3d8e1dd26895d170d1fe50
 # adds event for CTRL+\: https://github.com/girlbossceo/rustyline-async/commit/67d8c49aeac03a5ef4e818f663eaa94dd7bf339b
 [patch.crates-io.rustyline-async]
 git = "https://github.com/girlbossceo/rustyline-async"
-rev = "9654cc84e19241f6e19021eb8e677892656f5071"
+rev = "de26100b0db03e419a3d8e1dd26895d170d1fe50"
 
 #
 # Our crates
@@ -566,17 +569,7 @@ lto = "fat"
 [profile.release-max-perf.build-override]
 inherits = "release-max-perf"
 opt-level = 0
-codegen-units = 32
 #rustflags = [
-#	'-Crelocation-model=pic',
-#	'-Ctarget-feature=-crt-static',
-#	'-Clink-arg=-Wl,--no-gc-sections',
-#]
-
-[profile.release-max-perf.package.conduit_macros]
-inherits = "release-max-perf.build-override"
-#rustflags = [
-#	'-Crelocation-model=pic',
 #	'-Ctarget-feature=-crt-static',
 #]
 

conduwuit-example.toml

@@ -516,31 +516,6 @@ allow_profile_lookup_federation_requests = true
 # Defaults to false as this uses more CPU when compressing.
 #rocksdb_bottommost_compression = false
 
-# Level of statistics collection. Some admin commands to display database statistics may require
-# this option to be set. Database performance may be impacted by higher settings.
-#
-# Option is a number ranging from 0 to 6:
-# 0 = No statistics.
-# 1 = No statistics in release mode (default).
-# 2 to 3 = Statistics with no performance impact.
-# 3 to 5 = Statistics with possible performance impact.
-# 6 = All statistics.
-#
-# Defaults to 1 (No statistics, except in debug-mode)
-#rocksdb_stats_level = 1
-
-# Database repair mode (for RocksDB SST corruption)
-#
-# Use this option when the server reports corruption while running or panics. If the server refuses
-# to start use the recovery mode options first. Corruption errors containing the acronym 'SST' which
-# occur after startup will likely require this option.
-#
-# - Backing up your database directory is recommended prior to running the repair.
-# - Disabling repair mode and restarting the server is recommended after running the repair.
-#
-# Defaults to false
-#rocksdb_repair = false
-
 # Database recovery mode (for RocksDB WAL corruption)
 #
 # Use this option when the server reports corruption and refuses to start. Set mode 2 (PointInTime)

docs/deploying/docker.md

@@ -17,9 +17,9 @@ OCI images for conduwuit are available in the registries listed below.
 | GitLab Registry | [registry.gitlab.com/conduwuit/conduwuit:main][gl]   | ![Image Size][shield-main]    | Stable main branch.   |
 | Docker Hub      | [docker.io/girlbossceo/conduwuit:main][dh]               | ![Image Size][shield-main]    | Stable main branch.   |
 
-[dh]: https://hub.docker.com/r/girlbossceo/conduwuit
+[dh]: https://hub.docker.com/repository/docker/girlbossceo/conduwuit
 [gh]: https://github.com/girlbossceo/conduwuit/pkgs/container/conduwuit
-[gl]: https://gitlab.com/conduwuit/conduwuit/container_registry/6369729
+[gl]: https://gitlab.com/conduwuit/conduwuit/container_registry/6351657
 [shield-latest]: https://img.shields.io/docker/image-size/girlbossceo/conduwuit/latest
 [shield-main]: https://img.shields.io/docker/image-size/girlbossceo/conduwuit/main
 

flake.lock

@@ -9,11 +9,11 @@
         "nixpkgs-stable": "nixpkgs-stable"
       },
       "locked": {
-        "lastModified": 1723558887,
+        "lastModified": 1720542474,
-        "narHash": "sha256-FBaRmikNnCX+HQJXAfNJyiOl3sM1NrlxvcFwQDqwCyk=",
+        "narHash": "sha256-aKjJ/4l2I9+wNGTaOGRsuS3M1+IoTibqgEMPDikXm04=",
         "owner": "zhaofengli",
         "repo": "attic",
-        "rev": "26b9417bde03edc6280d1f7ce709cd619cdb72d4",
+        "rev": "6139576a3ce6bb992e0f6c3022528ec233e45f00",
         "type": "github"
       },
       "original": {
@@ -27,15 +27,15 @@
       "inputs": {
         "devenv": "devenv",
         "flake-compat": "flake-compat_3",
-        "git-hooks": "git-hooks",
+        "nixpkgs": "nixpkgs_3",
-        "nixpkgs": "nixpkgs_3"
+        "pre-commit-hooks": "pre-commit-hooks"
       },
       "locked": {
-        "lastModified": 1723252655,
+        "lastModified": 1719923519,
-        "narHash": "sha256-uOxcrNEMTOHvGHYyqDWnUpkhkY29l01KhjkgVQU2gXg=",
+        "narHash": "sha256-7Rhljj2fsklFRsu+eq7N683Z9qukmreMEj5C1GqCrSA=",
         "owner": "cachix",
         "repo": "cachix",
-        "rev": "bd558fd32f55d6df1c3d496bade90335a1a6044d",
+        "rev": "4e9e71f78b9500fa6210cf1eaa4d75bdbab777c3",
         "type": "github"
       },
       "original": {
@@ -81,11 +81,11 @@
     "complement": {
       "flake": false,
       "locked": {
-        "lastModified": 1722323564,
+        "lastModified": 1720637557,
-        "narHash": "sha256-6w6/N8walz4Ayc9zu7iySqJRmGFukhkaICLn4dweAcA=",
+        "narHash": "sha256-oZz6nCmFmdJZpC+K1iOG2KkzTI6rlAmndxANPDVU7X0=",
         "owner": "matrix-org",
         "repo": "complement",
-        "rev": "6e4426a9e63233f9821a4d2382bfed145244183f",
+        "rev": "0d14432e010482ea9e13a6f7c47c1533c0c9d62f",
         "type": "github"
       },
       "original": {
@@ -123,11 +123,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1722960479,
+        "lastModified": 1720546058,
-        "narHash": "sha256-NhCkJJQhD5GUib8zN9JrmYGMwt4lCRp6ZVNzIiYCl0Y=",
+        "narHash": "sha256-iU2yVaPIZm5vMGdlT0+57vdB/aPq/V5oZFBRwYw+HBM=",
         "owner": "ipetkov",
         "repo": "crane",
-        "rev": "4c6c77920b8d44cd6660c1621dea6b3fc4b4c4f4",
+        "rev": "2d83156f23c43598cf44e152c33a59d3892f8b29",
         "type": "github"
       },
       "original": {
@@ -151,15 +151,15 @@
         ],
         "pre-commit-hooks": [
           "cachix",
-          "git-hooks"
+          "pre-commit-hooks"
         ]
       },
       "locked": {
-        "lastModified": 1723156315,
+        "lastModified": 1719759336,
-        "narHash": "sha256-0JrfahRMJ37Rf1i0iOOn+8Z4CLvbcGNwa2ChOAVrp/8=",
+        "narHash": "sha256-3a34VL/QnHprl5gMy9xlx6d8J+iNp+W88Ex8smkgH9M=",
         "owner": "cachix",
         "repo": "devenv",
-        "rev": "ff5eb4f2accbcda963af67f1a1159e3f6c7f5f91",
+        "rev": "bb32aa986f2f695385e54428d0eaf7d05b31466e",
         "type": "github"
       },
       "original": {
@@ -209,11 +209,11 @@
         "rust-analyzer-src": "rust-analyzer-src"
       },
       "locked": {
-        "lastModified": 1723530607,
+        "lastModified": 1720852044,
-        "narHash": "sha256-FaXZZLLDW1D+pj7UgrIslDS8XjMMG3Pus5gAvUYWQS0=",
+        "narHash": "sha256-3NBYz8VuXuKU+8ONd9NFafCNjPEGHIZQ2Mdoam1a4mY=",
         "owner": "nix-community",
         "repo": "fenix",
-        "rev": "296d44c440302980824c5f3b67e477cf0522e0c1",
+        "rev": "5087b12a595ee73131a944d922f24d81dae05725",
         "type": "github"
       },
       "original": {
@@ -272,6 +272,22 @@
       }
     },
     "flake-compat_4": {
+      "flake": false,
+      "locked": {
+        "lastModified": 1696426674,
+        "narHash": "sha256-kvjfFW7WAETZlt09AgDn1MrtKzP7t90Vf7vypd3OL1U=",
+        "owner": "edolstra",
+        "repo": "flake-compat",
+        "rev": "0f9255e01c2351cc7d116c072cb317785dd33b33",
+        "type": "github"
+      },
+      "original": {
+        "owner": "edolstra",
+        "repo": "flake-compat",
+        "type": "github"
+      }
+    },
+    "flake-compat_5": {
       "flake": false,
       "locked": {
         "lastModified": 1696426674,
@@ -340,38 +356,11 @@
         "type": "github"
       }
     },
-    "git-hooks": {
-      "inputs": {
-        "flake-compat": [
-          "cachix",
-          "flake-compat"
-        ],
-        "gitignore": "gitignore",
-        "nixpkgs": [
-          "cachix",
-          "nixpkgs"
-        ],
-        "nixpkgs-stable": "nixpkgs-stable_2"
-      },
-      "locked": {
-        "lastModified": 1723202784,
-        "narHash": "sha256-qbhjc/NEGaDbyy0ucycubq4N3//gDFFH3DOmp1D3u1Q=",
-        "owner": "cachix",
-        "repo": "git-hooks.nix",
-        "rev": "c7012d0c18567c889b948781bc74a501e92275d1",
-        "type": "github"
-      },
-      "original": {
-        "owner": "cachix",
-        "repo": "git-hooks.nix",
-        "type": "github"
-      }
-    },
     "gitignore": {
       "inputs": {
         "nixpkgs": [
           "cachix",
-          "git-hooks",
+          "pre-commit-hooks",
           "nixpkgs"
         ]
       },
@@ -392,11 +381,11 @@
     "liburing": {
       "flake": false,
       "locked": {
-        "lastModified": 1723553052,
+        "lastModified": 1720798442,
-        "narHash": "sha256-sjqmYP/tvQKmsTD6FOd8m7aEja2hhEyyB3aTxVXEbuE=",
+        "narHash": "sha256-gtPppAoksMLW4GuruQ36nf4EAqIA1Bs6V9Xcx8dBxrQ=",
         "owner": "axboe",
         "repo": "liburing",
-        "rev": "401b3e4bde2316d35e93b548269c72bbb64adc02",
+        "rev": "1d674f83b7d0f07553ac44d99a401b05853d9dbe",
         "type": "github"
       },
       "original": {
@@ -569,16 +558,16 @@
     },
     "nixpkgs-stable_2": {
       "locked": {
-        "lastModified": 1720386169,
+        "lastModified": 1718811006,
-        "narHash": "sha256-NGKVY4PjzwAa4upkGtAMz1npHGoRzWotlSnVlqI40mo=",
+        "narHash": "sha256-0Y8IrGhRmBmT7HHXlxxepg2t8j1X90++qRN3lukGaIk=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "194846768975b7ad2c4988bdb82572c00222c0d7",
+        "rev": "03d771e513ce90147b65fe922d87d3a0356fc125",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "nixos-24.05",
+        "ref": "nixos-23.11",
         "repo": "nixpkgs",
         "type": "github"
       }
@@ -601,11 +590,11 @@
     },
     "nixpkgs_3": {
       "locked": {
-        "lastModified": 1722813957,
+        "lastModified": 1719848872,
-        "narHash": "sha256-IAoYyYnED7P8zrBFMnmp7ydaJfwTnwcnqxUElC1I26Y=",
+        "narHash": "sha256-H3+EC5cYuq+gQW8y0lSrrDZfH71LB4DAf+TDFyvwCNA=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "cb9a96f23c491c081b38eab96d22fa958043c9fa",
+        "rev": "00d80d13810dbfea8ab4ed1009b09100cca86ba8",
         "type": "github"
       },
       "original": {
@@ -617,11 +606,11 @@
     },
     "nixpkgs_4": {
       "locked": {
-        "lastModified": 1723362943,
+        "lastModified": 1720768451,
-        "narHash": "sha256-dFZRVSgmJkyM0bkPpaYRtG/kRMRTorUIDj8BxoOt1T4=",
+        "narHash": "sha256-EYekUHJE2gxeo2pM/zM9Wlqw1Uw2XTJXOSAO79ksc4Y=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "a58bc8ad779655e790115244571758e8de055e3d",
+        "rev": "7e7c39ea35c5cdd002cd4588b03a3fb9ece6fad9",
         "type": "github"
       },
       "original": {
@@ -657,6 +646,30 @@
         "type": "github"
       }
     },
+    "pre-commit-hooks": {
+      "inputs": {
+        "flake-compat": "flake-compat_4",
+        "gitignore": "gitignore",
+        "nixpkgs": [
+          "cachix",
+          "nixpkgs"
+        ],
+        "nixpkgs-stable": "nixpkgs-stable_2"
+      },
+      "locked": {
+        "lastModified": 1719259945,
+        "narHash": "sha256-F1h+XIsGKT9TkGO3omxDLEb/9jOOsI6NnzsXFsZhry4=",
+        "owner": "cachix",
+        "repo": "pre-commit-hooks.nix",
+        "rev": "0ff4381bbb8f7a52ca4a851660fc7a437a4c6e07",
+        "type": "github"
+      },
+      "original": {
+        "owner": "cachix",
+        "repo": "pre-commit-hooks.nix",
+        "type": "github"
+      }
+    },
     "rocksdb": {
       "flake": false,
       "locked": {
@@ -681,7 +694,7 @@
         "complement": "complement",
         "crane": "crane_2",
         "fenix": "fenix",
-        "flake-compat": "flake-compat_4",
+        "flake-compat": "flake-compat_5",
         "flake-utils": "flake-utils_3",
         "liburing": "liburing",
         "nix-filter": "nix-filter",
@@ -692,11 +705,11 @@
     "rust-analyzer-src": {
       "flake": false,
       "locked": {
-        "lastModified": 1723473250,
+        "lastModified": 1720717809,
-        "narHash": "sha256-Ls0e6R4FmGUFXZlUcm6ZQaVNJ4Yj/nua4SSctXIopao=",
+        "narHash": "sha256-6I+fm+nTLF/iaj7ffiFGlSY7POmubwUaPA/Wq0Bm53M=",
         "owner": "rust-lang",
         "repo": "rust-analyzer",
-        "rev": "32a86cb1dad2b208e8f36f1bb50c2e4806b0371f",
+        "rev": "ffbc5ad993d5cd2f3b8bcf9a511165470944ab91",
         "type": "github"
       },
       "original": {

flake.nix

@@ -24,7 +24,7 @@
         file = ./rust-toolchain.toml;
 
         # See also `rust-toolchain.toml`
-        sha256 = "sha256-3jVIIf5XPnUU1CRaTyAiO0XHVbJl12MSx3eucTXCjtE=";
+        sha256 = "sha256-6eN/GKzjVSjEhGO9FhWObkRFaE1Jf+uqMSdQnb8lcB4=";
       };
 
       mkScope = pkgs: pkgs.lib.makeScope pkgs.newScope (self: {

rust-toolchain.toml

@@ -11,7 +11,7 @@
 # If you're having trouble making the relevant changes, bug a maintainer.
 
 [toolchain]
-channel = "1.80.1"
+channel = "1.80.0"
 components = [
     # For rust-analyzer
     "rust-src",

src/admin/command.rs

@@ -1,9 +1,6 @@
-use std::time::SystemTime;
+use service::Services;
-
-use conduit_service::Services;
 
 pub(crate) struct Command<'a> {
 	pub(crate) services: &'a Services,
 	pub(crate) body: &'a [&'a str],
-	pub(crate) timer: SystemTime,
 }

src/admin/debug/commands.rs

@@ -1,12 +1,16 @@
 use std::{
 	collections::{BTreeMap, HashMap},
 	fmt::Write,
-	sync::Arc,
+	sync::{Arc, Mutex},
 	time::{Instant, SystemTime},
 };
 
 use api::client::validate_and_add_event_id;
-use conduit::{debug, debug_error, err, info, trace, utils, warn, Error, PduEvent, Result};
+use conduit::{
+	debug, info, log,
+	log::{capture, Capture},
+	utils, warn, Error, PduEvent, Result,
+};
 use ruma::{
 	api::{client::error::ErrorKind, federation::event::get_room_state},
 	events::room::message::RoomMessageEventContent,
@@ -209,7 +213,7 @@ pub(super) async fn get_remote_pdu(
 				Error::BadRequest(ErrorKind::Unknown, "Received response from server but failed to parse PDU")
 			})?;
 
-			trace!("Attempting to parse PDU: {:?}", &response.pdu);
+			debug!("Attempting to parse PDU: {:?}", &response.pdu);
 			let parsed_pdu = {
 				let parsed_result = self
 					.services
@@ -234,7 +238,8 @@ pub(super) async fn get_remote_pdu(
 
 			debug!("Attempting to fetch homeserver signing keys for {server}");
 			self.services
-				.server_keys
+				.rooms
+				.event_handler
 				.fetch_required_signing_keys(parsed_pdu.iter().map(|(_event_id, event, _room_id)| event), &pub_key_map)
 				.await
 				.unwrap_or_else(|e| {
@@ -447,7 +452,8 @@ pub(super) async fn verify_json(&self) -> Result<RoomMessageEventContent> {
 			let pub_key_map = RwLock::new(BTreeMap::new());
 
 			self.services
-				.server_keys
+				.rooms
+				.event_handler
 				.fetch_required_signing_keys([&value], &pub_key_map)
 				.await?;
 
@@ -565,7 +571,8 @@ pub(super) async fn force_set_room_state_from_server(
 
 	info!("Fetching required signing keys for all the state events we got");
 	self.services
-		.server_keys
+		.rooms
+		.event_handler
 		.fetch_required_signing_keys(events.iter().map(|(_event_id, event, _room_id)| event), &pub_key_map)
 		.await?;
 
@@ -580,8 +587,8 @@ pub(super) async fn force_set_room_state_from_server(
 		};
 
 		let pdu = PduEvent::from_id_val(&event_id, value.clone()).map_err(|e| {
-			debug_error!("Invalid PDU in fetching remote room state PDUs response: {value:#?}");
+			warn!("Invalid PDU in fetching remote room state PDUs response: {} {:?}", e, value);
-			err!(BadServerResponse(debug_error!("Invalid PDU in send_join response: {e:?}")))
+			Error::BadServerResponse("Invalid PDU in send_join response.")
 		})?;
 
 		self.services
@@ -684,7 +691,8 @@ pub(super) async fn get_verify_keys(
 	let signature_ids: Vec<String> = Vec::new();
 	let keys = self
 		.services
-		.server_keys
+		.rooms
+		.event_handler
 		.fetch_signing_keys_for_server(&server_name, signature_ids)
 		.await?;
 
@@ -713,14 +721,30 @@ pub(super) async fn resolve_true_destination(
 		));
 	}
 
+	let filter: &capture::Filter = &|data| {
+		data.level() <= log::Level::DEBUG
+			&& data.mod_name().starts_with("conduit")
+			&& matches!(data.span_name(), "actual" | "well-known" | "srv")
+	};
+
+	let state = &self.services.server.log.capture;
+	let logs = Arc::new(Mutex::new(String::new()));
+	let capture = Capture::new(state, Some(filter), capture::fmt_markdown(logs.clone()));
+
+	let capture_scope = capture.start();
 	let actual = self
 		.services
 		.resolver
 		.resolve_actual_dest(&server_name, !no_cache)
 		.await?;
+	drop(capture_scope);
 
-	let msg = format!("Destination: {}\nHostname URI: {}", actual.dest, actual.host,);
+	let msg = format!(
-
+		"{}\nDestination: {}\nHostname URI: {}",
+		logs.lock().expect("locked"),
+		actual.dest,
+		actual.host,
+	);
 	Ok(RoomMessageEventContent::text_markdown(msg))
 }
 
@@ -808,24 +832,3 @@ pub(super) async fn list_dependencies(&self, names: bool) -> Result<RoomMessageE
 
 	Ok(RoomMessageEventContent::notice_markdown(out))
 }
-
-#[admin_command]
-pub(super) async fn database_stats(
-	&self, property: Option<String>, map: Option<String>,
-) -> Result<RoomMessageEventContent> {
-	let property = property.unwrap_or_else(|| "rocksdb.stats".to_owned());
-	let map_name = map.as_ref().map_or(utils::string::EMPTY, String::as_str);
-
-	let mut out = String::new();
-	for (name, map) in self.services.db.iter_maps() {
-		if !map_name.is_empty() && *map_name != *name {
-			continue;
-		}
-
-		let res = map.property(&property)?;
-		let res = res.trim();
-		writeln!(out, "##### {name}:\n```\n{res}\n```")?;
-	}
-
-	Ok(RoomMessageEventContent::notice_markdown(out))
-}

src/admin/debug/mod.rs

@@ -184,14 +184,6 @@ pub(super) enum DebugCommand {
 		names: bool,
 	},
 
-	/// - Get database statistics
-	DatabaseStats {
-		property: Option<String>,
-
-		#[arg(short, long, alias("column"))]
-		map: Option<String>,
-	},
-
 	/// - Developer test stubs
 	#[command(subcommand)]
 	#[allow(non_snake_case)]

src/admin/handler.rs

@@ -1,21 +1,7 @@
-use std::{
+use std::{panic::AssertUnwindSafe, sync::Arc, time::Instant};
-	panic::AssertUnwindSafe,
-	sync::{Arc, Mutex},
-	time::SystemTime,
-};
 
 use clap::{CommandFactory, Parser};
-use conduit::{
+use conduit::{checked, error, trace, utils::string::common_prefix, Error, Result};
-	debug, error,
-	log::{
-		capture,
-		capture::Capture,
-		fmt::{markdown_table, markdown_table_head},
-	},
-	trace,
-	utils::string::{collect_stream, common_prefix},
-	Error, Result,
-};
 use futures_util::future::FutureExt;
 use ruma::{
 	events::{
@@ -25,10 +11,9 @@ use ruma::{
 	OwnedEventId,
 };
 use service::{
-	admin::{CommandInput, CommandOutput, HandlerFuture, HandlerResult},
+	admin::{CommandInput, CommandOutput, CommandResult, HandlerResult},
 	Services,
 };
-use tracing::Level;
 
 use crate::{admin, admin::AdminCommand, Command};
 
@@ -36,12 +21,12 @@ use crate::{admin, admin::AdminCommand, Command};
 pub(super) fn complete(line: &str) -> String { complete_command(AdminCommand::command(), line) }
 
 #[must_use]
-pub(super) fn handle(services: Arc<Services>, command: CommandInput) -> HandlerFuture {
+pub(super) fn handle(services: Arc<Services>, command: CommandInput) -> HandlerResult {
 	Box::pin(handle_command(services, command))
 }
 
 #[tracing::instrument(skip_all, name = "admin")]
-async fn handle_command(services: Arc<Services>, command: CommandInput) -> HandlerResult {
+async fn handle_command(services: Arc<Services>, command: CommandInput) -> CommandResult {
 	AssertUnwindSafe(Box::pin(process_command(services, &command)))
 		.catch_unwind()
 		.await
@@ -49,24 +34,13 @@ async fn handle_command(services: Arc<Services>, command: CommandInput) -> Handl
 		.or_else(|error| handle_panic(&error, command))
 }
 
-async fn process_command(services: Arc<Services>, input: &CommandInput) -> CommandOutput {
+async fn process_command(services: Arc<Services>, command: &CommandInput) -> CommandOutput {
-	let (command, args, body) = match parse(&services, input) {
+	process(services, &command.command)
-		Err(error) => return error,
-		Ok(parsed) => parsed,
-	};
-
-	let context = Command {
-		services: &services,
-		body: &body,
-		timer: SystemTime::now(),
-	};
-
-	process(&context, command, &args)
 		.await
-		.and_then(|content| reply(content, input.reply_id.clone()))
+		.and_then(|content| reply(content, command.reply_id.clone()))
 }
 
-fn handle_panic(error: &Error, command: CommandInput) -> HandlerResult {
+fn handle_panic(error: &Error, command: CommandInput) -> CommandResult {
 	let link = "Please submit a [bug report](https://github.com/girlbossceo/conduwuit/issues/new). 🥺";
 	let msg = format!("Panic occurred while processing command:\n```\n{error:#?}\n```\n{link}");
 	let content = RoomMessageEventContent::notice_markdown(msg);
@@ -85,61 +59,69 @@ fn reply(mut content: RoomMessageEventContent, reply_id: Option<OwnedEventId>) -
 }
 
 // Parse and process a message from the admin room
-async fn process(context: &Command<'_>, command: AdminCommand, args: &[String]) -> CommandOutput {
+async fn process(services: Arc<Services>, msg: &str) -> CommandOutput {
-	let filter: &capture::Filter =
+	let lines = msg.lines().filter(|l| !l.trim().is_empty());
-		&|data| data.level() <= Level::DEBUG && data.our_modules() && data.scope.contains(&"admin");
+	let command = lines
-	let logs = Arc::new(Mutex::new(
+		.clone()
-		collect_stream(|s| markdown_table_head(s)).expect("markdown table header"),
+		.next()
-	));
+		.expect("each string has at least one line");
-
+	let (parsed, body) = match parse_command(command) {
-	let capture = Capture::new(
+		Ok(parsed) => parsed,
-		&context.services.server.log.capture,
+		Err(error) => {
-		Some(filter),
+			let server_name = services.globals.server_name();
-		capture::fmt(markdown_table, logs.clone()),
+			let message = error.replace("server.name", server_name.as_str());
-	);
+			return Some(RoomMessageEventContent::notice_markdown(message));
-
+		},
-	let capture_scope = capture.start();
-	let result = Box::pin(admin::process(command, context)).await;
-	drop(capture_scope);
-
-	debug!(
-		ok = result.is_ok(),
-		elapsed = ?context.timer.elapsed(),
-		command = ?args,
-		"command processed"
-	);
-
-	let logs = logs.lock().expect("locked");
-	let output = match result {
-		Err(error) => format!("{logs}\nEncountered an error while handling the command:\n```\n{error:#?}\n```"),
-		Ok(reply) => format!("{logs}\n{}", reply.body()), //TODO: content is recreated to add logs
 	};
 
-	Some(RoomMessageEventContent::notice_markdown(output))
+	let body = parse_body(AdminCommand::command(), &body, lines.skip(1).collect()).expect("trailing body parsed");
-}
+	let context = Command {
-
+		services: &services,
-// Parse chat messages from the admin room into an AdminCommand object
+		body: &body,
-fn parse<'a>(
+	};
-	services: &Arc<Services>, input: &'a CommandInput,
+	let timer = Instant::now();
-) -> Result<(AdminCommand, Vec<String>, Vec<&'a str>), CommandOutput> {
+	let result = Box::pin(admin::process(parsed, &context)).await;
-	let lines = input.command.lines().filter(|line| !line.trim().is_empty());
+	let elapsed = timer.elapsed();
-	let command_line = lines.clone().next().expect("command missing first line");
+	conduit::debug!(?command, ok = result.is_ok(), "command processed in {elapsed:?}");
-	let body = lines.skip(1).collect();
+	match result {
-	match parse_command(command_line) {
+		Ok(reply) => Some(reply),
-		Ok((command, args)) => Ok((command, args, body)),
+		Err(error) => Some(RoomMessageEventContent::notice_markdown(format!(
-		Err(error) => {
+			"Encountered an error while handling the command:\n```\n{error:#?}\n```"
-			let message = error
+		))),
-				.to_string()
-				.replace("server.name", services.globals.server_name().as_str());
-			Err(Some(RoomMessageEventContent::notice_markdown(message)))
-		},
 	}
 }
 
-fn parse_command(line: &str) -> Result<(AdminCommand, Vec<String>)> {
+// Parse chat messages from the admin room into an AdminCommand object
-	let argv = parse_line(line);
+fn parse_command(command_line: &str) -> Result<(AdminCommand, Vec<String>), String> {
-	let command = AdminCommand::try_parse_from(&argv)?;
+	let argv = parse_line(command_line);
-	Ok((command, argv))
+	let com = AdminCommand::try_parse_from(&argv).map_err(|error| error.to_string())?;
+	Ok((com, argv))
+}
+
+fn parse_body<'a>(mut cmd: clap::Command, body: &'a [String], lines: Vec<&'a str>) -> Result<Vec<&'a str>> {
+	let mut start = 1;
+	'token: for token in body.iter().skip(1) {
+		let cmd_ = cmd.clone();
+		for sub in cmd_.get_subcommands() {
+			if sub.get_name() == *token {
+				start = checked!(start + 1)?;
+				cmd = sub.clone();
+				continue 'token;
+			}
+		}
+
+		// positional arguments have to be skipped too
+		let num_posargs = cmd_.get_positionals().count();
+		start = checked!(start + num_posargs)?;
+		break;
+	}
+
+	Ok(body
+		.iter()
+		.skip(start)
+		.map(String::as_str)
+		.chain(lines)
+		.collect::<Vec<&'a str>>())
 }
 
 fn complete_command(mut cmd: clap::Command, line: &str) -> String {

src/admin/room/moderation.rs

@@ -547,7 +547,7 @@ async fn list_banned_rooms(&self) -> Result<RoomMessageEventContent> {
 			rooms.reverse();
 
 			let output_plain = format!(
-				"Rooms Banned ({}):\n```\n{}\n```",
+				"Rooms Banned ({}):\n```\n{}```",
 				rooms.len(),
 				rooms
 					.iter()

src/api/Cargo.toml

@@ -47,6 +47,7 @@ hmac.workspace = true
 http.workspace = true
 http-body-util.workspace = true
 hyper.workspace = true
+image.workspace = true
 ipaddress.workspace = true
 jsonwebtoken.workspace = true
 log.workspace = true
@@ -59,6 +60,7 @@ serde.workspace = true
 sha-1.workspace = true
 tokio.workspace = true
 tracing.workspace = true
+webpage.workspace = true
 
 [lints]
 workspace = true

src/api/client/account.rs

@@ -240,7 +240,7 @@ pub(crate) async fn register_route(
 
 	// If `new_user_displayname_suffix` is set, registration will push whatever
 	// content is set to the user's display name with a space before it
-	if !services.globals.new_user_displayname_suffix().is_empty() && body.appservice_info.is_none() {
+	if !services.globals.new_user_displayname_suffix().is_empty() {
 		write!(displayname, " {}", services.globals.config.new_user_displayname_suffix)
 			.expect("should be able to write to string buffer");
 	}

src/api/client/keys.rs

@@ -4,7 +4,7 @@ use std::{
 };
 
 use axum::extract::State;
-use conduit::{utils, utils::math::continue_exponential_backoff_secs, Err, Error, Result};
+use conduit::{debug, utils, utils::math::continue_exponential_backoff_secs, Error, Result};
 use futures_util::{stream::FuturesUnordered, StreamExt};
 use ruma::{
 	api::{
@@ -362,7 +362,8 @@ pub(crate) async fn get_keys_helper<F: Fn(&UserId) -> bool + Send>(
 				const MIN: u64 = 5 * 60;
 				const MAX: u64 = 60 * 60 * 24;
 				if continue_exponential_backoff_secs(MIN, MAX, time.elapsed(), *tries) {
-					return (server, Err!(BadServerResponse("bad query from {server:?}, still backing off")));
+					debug!("Backing off query from {server:?}");
+					return (server, Err(Error::BadServerResponse("bad query, still backing off")));
 				}
 			}
 

src/api/client/media.rs

@@ -1,11 +1,11 @@
 #![allow(deprecated)]
 
-use std::time::Duration;
+use std::{io::Cursor, time::Duration};
 
 use axum::extract::State;
 use axum_client_ip::InsecureClientIp;
 use conduit::{
-	debug_info, debug_warn, err, info,
+	debug, debug_info, debug_warn, err, info,
 	utils::{
 		self,
 		content_disposition::{content_disposition_type, make_content_disposition, sanitise_filename},
@@ -13,16 +13,27 @@ use conduit::{
 	},
 	warn, Err, Error, Result,
 };
-use ruma::api::client::media::{
+use image::io::Reader as ImgReader;
-	create_content, get_content, get_content_as_filename, get_content_thumbnail, get_media_config, get_media_preview,
+use ipaddress::IPAddress;
+use reqwest::Url;
+use ruma::api::client::{
+	error::ErrorKind,
+	media::{
+		create_content, get_content, get_content_as_filename, get_content_thumbnail, get_media_config,
+		get_media_preview,
+	},
 };
 use service::{
-	media::{FileMeta, MXC_LENGTH},
+	media::{FileMeta, UrlPreviewData},
 	Services,
 };
+use webpage::HTML;
 
 use crate::{Ruma, RumaResponse};
 
+/// generated MXC ID (`media-id`) length
+const MXC_LENGTH: usize = 32;
+
 /// Cache control for immutable objects
 const CACHE_CONTROL_IMMUTABLE: &str = "public,max-age=31536000,immutable";
 
@@ -65,12 +76,12 @@ pub(crate) async fn get_media_preview_route(
 	let sender_user = body.sender_user.as_ref().expect("user is authenticated");
 
 	let url = &body.url;
-	if !services.media.url_preview_allowed(url) {
+	if !url_preview_allowed(&services, url) {
-		debug_info!(%sender_user, %url, "URL is not allowed to be previewed");
+		debug_info!(%sender_user, "URL is not allowed to be previewed: {url}");
-		return Err!(Request(Forbidden("URL is not allowed to be previewed")));
+		return Err(Error::BadRequest(ErrorKind::forbidden(), "URL is not allowed to be previewed"));
 	}
 
-	match services.media.get_url_preview(url).await {
+	match get_url_preview(&services, url).await {
 		Ok(preview) => {
 			let res = serde_json::value::to_raw_value(&preview).map_err(|e| {
 				warn!(%sender_user, "Failed to convert UrlPreviewData into a serde json value: {e}");
@@ -205,7 +216,10 @@ pub(crate) async fn get_content_route(
 			body.timeout_ms,
 		)
 		.await
-		.map_err(|e| err!(Request(NotFound(debug_warn!("Fetching media `{mxc}` failed: {e:?}")))))?;
+		.map_err(|e| {
+			debug_warn!("Fetching media `{}` failed: {:?}", mxc, e);
+			Error::BadRequest(ErrorKind::NotFound, "Remote media error.")
+		})?;
 
 		let content_disposition = Some(make_content_disposition(
 			&response.content_type,
@@ -221,7 +235,7 @@ pub(crate) async fn get_content_route(
 			cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_owned()),
 		})
 	} else {
-		Err!(Request(NotFound("Media not found.")))
+		Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
 	}
 }
 
@@ -308,10 +322,13 @@ pub(crate) async fn get_content_as_filename_route(
 					cache_control: Some(CACHE_CONTROL_IMMUTABLE.into()),
 				})
 			},
-			Err(e) => Err!(Request(NotFound(debug_warn!("Fetching media `{mxc}` failed: {e:?}")))),
+			Err(e) => {
+				debug_warn!("Fetching media `{}` failed: {:?}", mxc, e);
+				Err(Error::BadRequest(ErrorKind::NotFound, "Remote media error."))
+			},
 		}
 	} else {
-		Err!(Request(NotFound("Media not found.")))
+		Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
 	}
 }
 
@@ -362,10 +379,10 @@ pub(crate) async fn get_content_thumbnail_route(
 			&mxc,
 			body.width
 				.try_into()
-				.map_err(|e| err!(Request(InvalidParam("Width is invalid: {e:?}"))))?,
+				.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Width is invalid."))?,
 			body.height
 				.try_into()
-				.map_err(|e| err!(Request(InvalidParam("Height is invalid: {e:?}"))))?,
+				.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Height is invalid."))?,
 		)
 		.await?
 	{
@@ -388,7 +405,7 @@ pub(crate) async fn get_content_thumbnail_route(
 			// we'll lie to the client and say the blocked server's media was not found and
 			// log. the client has no way of telling anyways so this is a security bonus.
 			debug_warn!("Received request for media `{}` on blocklisted server", mxc);
-			return Err!(Request(NotFound("Media not found.")));
+			return Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."));
 		}
 
 		match services
@@ -437,10 +454,13 @@ pub(crate) async fn get_content_thumbnail_route(
 					content_disposition,
 				})
 			},
-			Err(e) => Err!(Request(NotFound(debug_warn!("Fetching media `{mxc}` failed: {e:?}")))),
+			Err(e) => {
+				debug_warn!("Fetching media `{}` failed: {:?}", mxc, e);
+				Err(Error::BadRequest(ErrorKind::NotFound, "Remote media error."))
+			},
 		}
 	} else {
-		Err!(Request(NotFound("Media not found.")))
+		Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."))
 	}
 }
 
@@ -478,7 +498,7 @@ async fn get_remote_content(
 		// we'll lie to the client and say the blocked server's media was not found and
 		// log. the client has no way of telling anyways so this is a security bonus.
 		debug_warn!("Received request for media `{mxc}` on blocklisted server");
-		return Err!(Request(NotFound("Media not found.")));
+		return Err(Error::BadRequest(ErrorKind::NotFound, "Media not found."));
 	}
 
 	let content_response = services
@@ -520,3 +540,222 @@ async fn get_remote_content(
 		cache_control: Some(CACHE_CONTROL_IMMUTABLE.to_owned()),
 	})
 }
+
+async fn download_image(services: &Services, client: &reqwest::Client, url: &str) -> Result<UrlPreviewData> {
+	let image = client.get(url).send().await?.bytes().await?;
+	let mxc = format!("mxc://{}/{}", services.globals.server_name(), utils::random_string(MXC_LENGTH));
+
+	services
+		.media
+		.create(None, &mxc, None, None, &image)
+		.await?;
+
+	let (width, height) = match ImgReader::new(Cursor::new(&image)).with_guessed_format() {
+		Err(_) => (None, None),
+		Ok(reader) => match reader.into_dimensions() {
+			Err(_) => (None, None),
+			Ok((width, height)) => (Some(width), Some(height)),
+		},
+	};
+
+	Ok(UrlPreviewData {
+		image: Some(mxc),
+		image_size: Some(image.len()),
+		image_width: width,
+		image_height: height,
+		..Default::default()
+	})
+}
+
+async fn download_html(services: &Services, client: &reqwest::Client, url: &str) -> Result<UrlPreviewData> {
+	let mut response = client.get(url).send().await?;
+
+	let mut bytes: Vec<u8> = Vec::new();
+	while let Some(chunk) = response.chunk().await? {
+		bytes.extend_from_slice(&chunk);
+		if bytes.len() > services.globals.url_preview_max_spider_size() {
+			debug!(
+				"Response body from URL {} exceeds url_preview_max_spider_size ({}), not processing the rest of the \
+				 response body and assuming our necessary data is in this range.",
+				url,
+				services.globals.url_preview_max_spider_size()
+			);
+			break;
+		}
+	}
+	let body = String::from_utf8_lossy(&bytes);
+	let Ok(html) = HTML::from_string(body.to_string(), Some(url.to_owned())) else {
+		return Err(Error::BadRequest(ErrorKind::Unknown, "Failed to parse HTML"));
+	};
+
+	let mut data = match html.opengraph.images.first() {
+		None => UrlPreviewData::default(),
+		Some(obj) => download_image(services, client, &obj.url).await?,
+	};
+
+	let props = html.opengraph.properties;
+
+	/* use OpenGraph title/description, but fall back to HTML if not available */
+	data.title = props.get("title").cloned().or(html.title);
+	data.description = props.get("description").cloned().or(html.description);
+
+	Ok(data)
+}
+
+async fn request_url_preview(services: &Services, url: &str) -> Result<UrlPreviewData> {
+	if let Ok(ip) = IPAddress::parse(url) {
+		if !services.globals.valid_cidr_range(&ip) {
+			return Err(Error::BadServerResponse("Requesting from this address is forbidden"));
+		}
+	}
+
+	let client = &services.client.url_preview;
+	let response = client.head(url).send().await?;
+
+	if let Some(remote_addr) = response.remote_addr() {
+		if let Ok(ip) = IPAddress::parse(remote_addr.ip().to_string()) {
+			if !services.globals.valid_cidr_range(&ip) {
+				return Err(Error::BadServerResponse("Requesting from this address is forbidden"));
+			}
+		}
+	}
+
+	let Some(content_type) = response
+		.headers()
+		.get(reqwest::header::CONTENT_TYPE)
+		.and_then(|x| x.to_str().ok())
+	else {
+		return Err(Error::BadRequest(ErrorKind::Unknown, "Unknown Content-Type"));
+	};
+	let data = match content_type {
+		html if html.starts_with("text/html") => download_html(services, client, url).await?,
+		img if img.starts_with("image/") => download_image(services, client, url).await?,
+		_ => return Err(Error::BadRequest(ErrorKind::Unknown, "Unsupported Content-Type")),
+	};
+
+	services.media.set_url_preview(url, &data).await?;
+
+	Ok(data)
+}
+
+async fn get_url_preview(services: &Services, url: &str) -> Result<UrlPreviewData> {
+	if let Some(preview) = services.media.get_url_preview(url).await {
+		return Ok(preview);
+	}
+
+	// ensure that only one request is made per URL
+	let _request_lock = services.media.url_preview_mutex.lock(url).await;
+
+	match services.media.get_url_preview(url).await {
+		Some(preview) => Ok(preview),
+		None => request_url_preview(services, url).await,
+	}
+}
+
+fn url_preview_allowed(services: &Services, url_str: &str) -> bool {
+	let url: Url = match Url::parse(url_str) {
+		Ok(u) => u,
+		Err(e) => {
+			warn!("Failed to parse URL from a str: {}", e);
+			return false;
+		},
+	};
+
+	if ["http", "https"]
+		.iter()
+		.all(|&scheme| scheme != url.scheme().to_lowercase())
+	{
+		debug!("Ignoring non-HTTP/HTTPS URL to preview: {}", url);
+		return false;
+	}
+
+	let host = match url.host_str() {
+		None => {
+			debug!("Ignoring URL preview for a URL that does not have a host (?): {}", url);
+			return false;
+		},
+		Some(h) => h.to_owned(),
+	};
+
+	let allowlist_domain_contains = services.globals.url_preview_domain_contains_allowlist();
+	let allowlist_domain_explicit = services.globals.url_preview_domain_explicit_allowlist();
+	let denylist_domain_explicit = services.globals.url_preview_domain_explicit_denylist();
+	let allowlist_url_contains = services.globals.url_preview_url_contains_allowlist();
+
+	if allowlist_domain_contains.contains(&"*".to_owned())
+		|| allowlist_domain_explicit.contains(&"*".to_owned())
+		|| allowlist_url_contains.contains(&"*".to_owned())
+	{
+		debug!("Config key contains * which is allowing all URL previews. Allowing URL {}", url);
+		return true;
+	}
+
+	if !host.is_empty() {
+		if denylist_domain_explicit.contains(&host) {
+			debug!(
+				"Host {} is not allowed by url_preview_domain_explicit_denylist (check 1/4)",
+				&host
+			);
+			return false;
+		}
+
+		if allowlist_domain_explicit.contains(&host) {
+			debug!("Host {} is allowed by url_preview_domain_explicit_allowlist (check 2/4)", &host);
+			return true;
+		}
+
+		if allowlist_domain_contains
+			.iter()
+			.any(|domain_s| domain_s.contains(&host.clone()))
+		{
+			debug!("Host {} is allowed by url_preview_domain_contains_allowlist (check 3/4)", &host);
+			return true;
+		}
+
+		if allowlist_url_contains
+			.iter()
+			.any(|url_s| url.to_string().contains(&url_s.to_string()))
+		{
+			debug!("URL {} is allowed by url_preview_url_contains_allowlist (check 4/4)", &host);
+			return true;
+		}
+
+		// check root domain if available and if user has root domain checks
+		if services.globals.url_preview_check_root_domain() {
+			debug!("Checking root domain");
+			match host.split_once('.') {
+				None => return false,
+				Some((_, root_domain)) => {
+					if denylist_domain_explicit.contains(&root_domain.to_owned()) {
+						debug!(
+							"Root domain {} is not allowed by url_preview_domain_explicit_denylist (check 1/3)",
+							&root_domain
+						);
+						return true;
+					}
+
+					if allowlist_domain_explicit.contains(&root_domain.to_owned()) {
+						debug!(
+							"Root domain {} is allowed by url_preview_domain_explicit_allowlist (check 2/3)",
+							&root_domain
+						);
+						return true;
+					}
+
+					if allowlist_domain_contains
+						.iter()
+						.any(|domain_s| domain_s.contains(&root_domain.to_owned()))
+					{
+						debug!(
+							"Root domain {} is allowed by url_preview_domain_contains_allowlist (check 3/3)",
+							&root_domain
+						);
+						return true;
+					}
+				},
+			}
+		}
+	}
+
+	false
+}

src/api/client/membership.rs

@@ -8,11 +8,11 @@ use std::{
 use axum::extract::State;
 use axum_client_ip::InsecureClientIp;
 use conduit::{
-	debug, debug_error, debug_warn, err, error, info,
+	debug, debug_warn, error, info,
 	pdu::{gen_event_id_canonical_json, PduBuilder},
 	trace, utils,
 	utils::math::continue_exponential_backoff_secs,
-	warn, Err, Error, PduEvent, Result,
+	warn, Error, PduEvent, Result,
 };
 use ruma::{
 	api::{
@@ -403,7 +403,6 @@ pub(crate) async fn kick_user_route(
 				unsigned: None,
 				state_key: Some(body.user_id.to_string()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&body.room_id,
@@ -466,7 +465,6 @@ pub(crate) async fn ban_user_route(
 				unsigned: None,
 				state_key: Some(body.user_id.to_string()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&body.room_id,
@@ -514,7 +512,6 @@ pub(crate) async fn unban_user_route(
 				unsigned: None,
 				state_key: Some(body.user_id.to_string()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&body.room_id,
@@ -665,12 +662,6 @@ pub async fn join_room_by_id_helper(
 ) -> Result<join_room_by_id::v3::Response> {
 	let state_lock = services.rooms.state.mutex.lock(room_id).await;
 
-	let user_is_guest = services.users.is_deactivated(sender_user).unwrap_or(false);
-
-	if matches!(services.rooms.state_accessor.guest_can_join(room_id), Ok(false)) && user_is_guest {
-		return Err!(Request(Forbidden("Guests are not allowed to join this room")));
-	}
-
 	if matches!(services.rooms.state_cache.is_joined(sender_user, room_id), Ok(true)) {
 		debug_warn!("{sender_user} is already joined in {room_id}");
 		return Ok(join_room_by_id::v3::Response {
@@ -714,11 +705,11 @@ async fn join_room_by_id_helper_remote(
 		{
 			room_version
 		},
-		_ => return Err!(BadServerResponse("Room version is not supported")),
+		_ => return Err(Error::BadServerResponse("Room version is not supported")),
 	};
 
 	let mut join_event_stub: CanonicalJsonObject = serde_json::from_str(make_join_response.event.get())
-		.map_err(|e| err!(BadServerResponse("Invalid make_join event json received from server: {e:?}")))?;
+		.map_err(|_| Error::BadServerResponse("Invalid make_join event json received from server."))?;
 
 	let join_authorized_via_users_server = join_event_stub
 		.get("content")
@@ -885,14 +876,15 @@ async fn join_room_by_id_helper_remote(
 
 	info!("Parsing join event");
 	let parsed_join_pdu = PduEvent::from_id_val(event_id, join_event.clone())
-		.map_err(|e| err!(BadServerResponse("Invalid join event PDU: {e:?}")))?;
+		.map_err(|_| Error::BadServerResponse("Invalid join event PDU."))?;
 
 	let mut state = HashMap::new();
 	let pub_key_map = RwLock::new(BTreeMap::new());
 
 	info!("Fetching join signing keys");
 	services
-		.server_keys
+		.rooms
+		.event_handler
 		.fetch_join_signing_keys(&send_join_response, &room_version_id, &pub_key_map)
 		.await?;
 
@@ -908,8 +900,8 @@ async fn join_room_by_id_helper_remote(
 		};
 
 		let pdu = PduEvent::from_id_val(&event_id, value.clone()).map_err(|e| {
-			debug_warn!("Invalid PDU in send_join response: {value:#?}");
+			warn!("Invalid PDU in send_join response: {} {:?}", e, value);
-			err!(BadServerResponse("Invalid PDU in send_join response: {e:?}"))
+			Error::BadServerResponse("Invalid PDU in send_join response.")
 		})?;
 
 		services.rooms.outlier.add_pdu_outlier(&event_id, &value)?;
@@ -1103,7 +1095,6 @@ async fn join_room_by_id_helper_local(
 				unsigned: None,
 				state_key: Some(sender_user.to_string()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			room_id,
@@ -1132,10 +1123,10 @@ async fn join_room_by_id_helper_local(
 			{
 				room_version_id
 			},
-			_ => return Err!(BadServerResponse("Room version is not supported")),
+			_ => return Err(Error::BadServerResponse("Room version is not supported")),
 		};
 		let mut join_event_stub: CanonicalJsonObject = serde_json::from_str(make_join_response.event.get())
-			.map_err(|e| err!(BadServerResponse("Invalid make_join event json received from server: {e:?}")))?;
+			.map_err(|_| Error::BadServerResponse("Invalid make_join event json received from server."))?;
 		let join_authorized_via_users_server = join_event_stub
 			.get("content")
 			.map(|s| {
@@ -1239,7 +1230,8 @@ async fn join_room_by_id_helper_local(
 			drop(state_lock);
 			let pub_key_map = RwLock::new(BTreeMap::new());
 			services
-				.server_keys
+				.rooms
+				.event_handler
 				.fetch_required_signing_keys([&signed_value], &pub_key_map)
 				.await?;
 			services
@@ -1260,7 +1252,7 @@ async fn join_room_by_id_helper_local(
 async fn make_join_request(
 	services: &Services, sender_user: &UserId, room_id: &RoomId, servers: &[OwnedServerName],
 ) -> Result<(federation::membership::prepare_join_event::v1::Response, OwnedServerName)> {
-	let mut make_join_response_and_server = Err!(BadServerResponse("No server available to assist in joining."));
+	let mut make_join_response_and_server = Err(Error::BadServerResponse("No server available to assist in joining."));
 
 	let mut make_join_counter: u16 = 0;
 	let mut incompatible_room_version_count: u8 = 0;
@@ -1301,7 +1293,8 @@ async fn make_join_request(
 					"15 servers have responded with M_INCOMPATIBLE_ROOM_VERSION or M_UNSUPPORTED_ROOM_VERSION, \
 					 assuming that Conduwuit does not support the room {room_id}: {e}"
 				);
-				make_join_response_and_server = Err!(BadServerResponse("Room version is not supported by Conduwuit"));
+				make_join_response_and_server =
+					Err(Error::BadServerResponse("Room version is not supported by Conduwuit"));
 				return make_join_response_and_server;
 			}
 
@@ -1309,7 +1302,8 @@ async fn make_join_request(
 				warn!(
 					"50 servers failed to provide valid make_join response, assuming no server can assist in joining."
 				);
-				make_join_response_and_server = Err!(BadServerResponse("No server available to assist in joining."));
+				make_join_response_and_server =
+					Err(Error::BadServerResponse("No server available to assist in joining."));
 				return make_join_response_and_server;
 			}
 		}
@@ -1329,8 +1323,8 @@ pub async fn validate_and_add_event_id(
 	pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
 ) -> Result<(OwnedEventId, CanonicalJsonObject)> {
 	let mut value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
-		debug_error!("Invalid PDU in server response: {pdu:#?}");
+		error!("Invalid PDU in server response: {:?}: {:?}", pdu, e);
-		err!(BadServerResponse("Invalid PDU in server response: {e:?}"))
+		Error::BadServerResponse("Invalid PDU in server response")
 	})?;
 	let event_id = EventId::parse(format!(
 		"${}",
@@ -1366,15 +1360,15 @@ pub async fn validate_and_add_event_id(
 		const MIN: u64 = 60 * 5;
 		const MAX: u64 = 60 * 60 * 24;
 		if continue_exponential_backoff_secs(MIN, MAX, time.elapsed(), *tries) {
-			return Err!(BadServerResponse("bad event {event_id:?}, still backing off"));
+			debug!("Backing off from {event_id}");
+			return Err(Error::BadServerResponse("bad event, still backing off"));
 		}
 	}
 
 	if let Err(e) = ruma::signatures::verify_event(&*pub_key_map.read().await, &value, room_version) {
-		debug_error!("Event {event_id} failed verification {pdu:#?}");
+		warn!("Event {} failed verification {:?} {}", event_id, pdu, e);
-		let e = Err!(BadServerResponse(debug_error!("Event {event_id} failed verification: {e:?}")));
 		back_off(event_id).await;
-		return e;
+		return Err(Error::BadServerResponse("Event failed verification."));
 	}
 
 	value.insert("event_id".to_owned(), CanonicalJsonValue::String(event_id.as_str().to_owned()));
@@ -1416,7 +1410,6 @@ pub(crate) async fn invite_helper(
 					unsigned: None,
 					state_key: Some(user_id.to_string()),
 					redacts: None,
-					timestamp: None,
 				},
 				sender_user,
 				room_id,
@@ -1481,7 +1474,8 @@ pub(crate) async fn invite_helper(
 		.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Origin field is invalid."))?;
 
 		services
-			.server_keys
+			.rooms
+			.event_handler
 			.fetch_required_signing_keys([&value], &pub_key_map)
 			.await?;
 
@@ -1528,7 +1522,6 @@ pub(crate) async fn invite_helper(
 				unsigned: None,
 				state_key: Some(user_id.to_string()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			room_id,
@@ -1646,7 +1639,6 @@ pub async fn leave_room(services: &Services, user_id: &UserId, room_id: &RoomId,
 					unsigned: None,
 					state_key: Some(user_id.to_string()),
 					redacts: None,
-					timestamp: None,
 				},
 				user_id,
 				room_id,
@@ -1659,7 +1651,7 @@ pub async fn leave_room(services: &Services, user_id: &UserId, room_id: &RoomId,
 }
 
 async fn remote_leave_room(services: &Services, user_id: &UserId, room_id: &RoomId) -> Result<()> {
-	let mut make_leave_response_and_server = Err!(BadServerResponse("No server available to assist in leaving."));
+	let mut make_leave_response_and_server = Err(Error::BadServerResponse("No server available to assist in leaving."));
 
 	let invite_state = services
 		.rooms
@@ -1716,11 +1708,11 @@ async fn remote_leave_room(services: &Services, user_id: &UserId, room_id: &Room
 		{
 			version
 		},
-		_ => return Err!(BadServerResponse("Room version is not supported")),
+		_ => return Err(Error::BadServerResponse("Room version is not supported")),
 	};
 
 	let mut leave_event_stub = serde_json::from_str::<CanonicalJsonObject>(make_leave_response.event.get())
-		.map_err(|e| err!(BadServerResponse("Invalid make_leave event json received from server: {e:?}")))?;
+		.map_err(|_| Error::BadServerResponse("Invalid make_leave event json received from server."))?;
 
 	// TODO: Is origin needed?
 	leave_event_stub.insert(

src/api/client/message.rs

@@ -84,11 +84,6 @@ pub(crate) async fn send_message_event_route(
 				unsigned: Some(unsigned),
 				state_key: None,
 				redacts: None,
-				timestamp: if body.appservice_info.is_some() {
-					body.timestamp
-				} else {
-					None
-				},
 			},
 			sender_user,
 			&body.room_id,

src/api/client/profile.rs

@@ -257,12 +257,6 @@ pub(crate) async fn get_profile_route(
 pub async fn update_displayname(
 	services: &Services, user_id: OwnedUserId, displayname: Option<String>, all_joined_rooms: Vec<OwnedRoomId>,
 ) -> Result<()> {
-	let current_display_name = services.users.displayname(&user_id).unwrap_or_default();
-
-	if displayname == current_display_name {
-		return Ok(());
-	}
-
 	services
 		.users
 		.set_displayname(&user_id, displayname.clone())
@@ -295,7 +289,6 @@ pub async fn update_displayname(
 					unsigned: None,
 					state_key: Some(user_id.to_string()),
 					redacts: None,
-					timestamp: None,
 				},
 				room_id,
 			))
@@ -312,13 +305,6 @@ pub async fn update_avatar_url(
 	services: &Services, user_id: OwnedUserId, avatar_url: Option<OwnedMxcUri>, blurhash: Option<String>,
 	all_joined_rooms: Vec<OwnedRoomId>,
 ) -> Result<()> {
-	let current_avatar_url = services.users.avatar_url(&user_id).unwrap_or_default();
-	let current_blurhash = services.users.blurhash(&user_id).unwrap_or_default();
-
-	if current_avatar_url == avatar_url && current_blurhash == blurhash {
-		return Ok(());
-	}
-
 	services
 		.users
 		.set_avatar_url(&user_id, avatar_url.clone())
@@ -356,7 +342,6 @@ pub async fn update_avatar_url(
 					unsigned: None,
 					state_key: Some(user_id.to_string()),
 					redacts: None,
-					timestamp: None,
 				},
 				room_id,
 			))

src/api/client/redact.rs

@@ -34,7 +34,6 @@ pub(crate) async fn redact_event_route(
 				unsigned: None,
 				state_key: None,
 				redacts: Some(body.event_id.into()),
-				timestamp: None,
 			},
 			sender_user,
 			&body.room_id,

src/api/client/room.rs

@@ -20,6 +20,7 @@ use ruma::{
 			tombstone::RoomTombstoneEventContent,
 			topic::RoomTopicEventContent,
 		},
+		space::child::SpaceChildEventContent,
 		StateEventType, TimelineEventType,
 	},
 	int,
@@ -197,7 +198,6 @@ pub(crate) async fn create_room_route(
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&room_id,
@@ -226,7 +226,6 @@ pub(crate) async fn create_room_route(
 				unsigned: None,
 				state_key: Some(sender_user.to_string()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&room_id,
@@ -264,7 +263,6 @@ pub(crate) async fn create_room_route(
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&room_id,
@@ -288,7 +286,6 @@ pub(crate) async fn create_room_route(
 					unsigned: None,
 					state_key: Some(String::new()),
 					redacts: None,
-					timestamp: None,
 				},
 				sender_user,
 				&room_id,
@@ -315,7 +312,6 @@ pub(crate) async fn create_room_route(
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&room_id,
@@ -335,7 +331,6 @@ pub(crate) async fn create_room_route(
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&room_id,
@@ -358,7 +353,6 @@ pub(crate) async fn create_room_route(
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&room_id,
@@ -412,7 +406,6 @@ pub(crate) async fn create_room_route(
 					unsigned: None,
 					state_key: Some(String::new()),
 					redacts: None,
-					timestamp: None,
 				},
 				sender_user,
 				&room_id,
@@ -435,7 +428,6 @@ pub(crate) async fn create_room_route(
 					unsigned: None,
 					state_key: Some(String::new()),
 					redacts: None,
-					timestamp: None,
 				},
 				sender_user,
 				&room_id,
@@ -590,7 +582,6 @@ pub(crate) async fn upgrade_room_route(
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&body.room_id,
@@ -628,22 +619,14 @@ pub(crate) async fn upgrade_room_route(
 			V1 | V2 | V3 | V4 | V5 | V6 | V7 | V8 | V9 | V10 => {
 				create_event_content.insert(
 					"creator".into(),
-					json!(&sender_user).try_into().map_err(|e| {
+					json!(&sender_user)
-						info!("Error forming creation event: {e}");
+						.try_into()
-						Error::BadRequest(ErrorKind::BadJson, "Error forming creation event")
+						.map_err(|_| err!(Request(BadJson("Error forming creation event"))))?,
-					})?,
 				);
 			},
-			V11 => {
-				// "creator" key no longer exists in V11 rooms
-				create_event_content.remove("creator");
-			},
 			_ => {
-				warn!("Unexpected or unsupported room version {}", body.new_version);
+				// "creator" key no longer exists in V11+ rooms
-				return Err(Error::BadRequest(
+				create_event_content.remove("creator");
-					ErrorKind::BadJson,
-					"Unexpected or unsupported room version found",
-				));
 			},
 		}
 	}
@@ -652,26 +635,100 @@ pub(crate) async fn upgrade_room_route(
 		"room_version".into(),
 		json!(&body.new_version)
 			.try_into()
-			.map_err(|_| Error::BadRequest(ErrorKind::BadJson, "Error forming creation event"))?,
+			.map_err(|_| err!(Request(BadJson("Error forming creation event"))))?,
 	);
 	create_event_content.insert(
 		"predecessor".into(),
 		json!(predecessor)
 			.try_into()
-			.map_err(|_| Error::BadRequest(ErrorKind::BadJson, "Error forming creation event"))?,
+			.map_err(|_| err!(Request(BadJson("Error forming creation event"))))?,
 	);
 
-	// Validate creation event content
+	// if the room was a space:
-	if serde_json::from_str::<CanonicalJsonObject>(
+	// - migrate m.space.child and/or m.space.parent
-		to_raw_value(&create_event_content)
+	// - add space type to replacement room m.room.create
-			.expect("Error forming creation event")
+	//
-			.get(),
+	// as apart of MSC4168
-	)
+	if services
-	.is_err()
+		.rooms
+		.state_accessor
+		.get_room_type(&body.room_id)
+		.unwrap_or(None)
+		.is_some_and(|room_type| room_type == ruma::room::RoomType::Space)
 	{
-		return Err(Error::BadRequest(ErrorKind::BadJson, "Error forming creation event"));
+		create_event_content.insert(
+			"type".into(),
+			json!(ruma::room::RoomType::Space)
+				.try_into()
+				.map_err(|_| err!(Request(BadJson("Error forming creation event"))))?,
+		);
+
+		if let Some(event_content) = services
+			.rooms
+			.state_accessor
+			.room_state_get(&body.room_id, &StateEventType::SpaceChild, body.room_id.as_str())
+			.unwrap_or(None)
+		{
+			// space contents that are empty are not apart of the space
+			if !event_content.content.get().is_empty() {
+				// remove all but us from via
+				let mut new_content: SpaceChildEventContent = serde_json::from_str(event_content.content.get())
+					.map_err(|_| err!(Database(error!("Invalid m.space.child content in database"))))?;
+				new_content.via = vec![services.globals.config.server_name.clone()];
+
+				services
+					.rooms
+					.timeline
+					.build_and_append_pdu(
+						PduBuilder {
+							event_type: StateEventType::SpaceChild.to_string().into(),
+							content: to_raw_value(&new_content).expect("we just created it"),
+							unsigned: None,
+							state_key: Some(replacement_room.to_string()),
+							redacts: None,
+						},
+						sender_user,
+						&replacement_room,
+						&state_lock,
+					)
+					.await?;
+			}
+		}
+
+		if let Some(event_content) = services
+			.rooms
+			.state_accessor
+			.room_state_get(&body.room_id, &StateEventType::SpaceParent, body.room_id.as_str())
+			.unwrap_or(None)
+		{
+			// space contents that are empty are not apart of the space
+			if !event_content.content.get().is_empty() {
+				// remove all but us from via
+				let mut new_content: SpaceChildEventContent = serde_json::from_str(event_content.content.get())
+					.map_err(|_| err!(Database(error!("Invalid m.space.child content in database"))))?;
+				new_content.via = vec![services.globals.config.server_name.clone()];
+
+				services
+					.rooms
+					.timeline
+					.build_and_append_pdu(
+						PduBuilder {
+							event_type: StateEventType::SpaceParent.to_string().into(),
+							content: to_raw_value(&new_content).expect("we just created it"),
+							unsigned: None,
+							state_key: Some(replacement_room.to_string()),
+							redacts: None,
+						},
+						sender_user,
+						&replacement_room,
+						&state_lock,
+					)
+					.await?;
+			}
+		}
 	}
 
+	// m.room.create
 	services
 		.rooms
 		.timeline
@@ -682,7 +739,6 @@ pub(crate) async fn upgrade_room_route(
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&replacement_room,
@@ -711,7 +767,6 @@ pub(crate) async fn upgrade_room_route(
 				unsigned: None,
 				state_key: Some(sender_user.to_string()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&replacement_room,
@@ -740,7 +795,6 @@ pub(crate) async fn upgrade_room_route(
 					unsigned: None,
 					state_key: Some(String::new()),
 					redacts: None,
-					timestamp: None,
 				},
 				sender_user,
 				&replacement_room,
@@ -768,11 +822,11 @@ pub(crate) async fn upgrade_room_route(
 			.rooms
 			.state_accessor
 			.room_state_get(&body.room_id, &StateEventType::RoomPowerLevels, "")?
-			.ok_or_else(|| Error::bad_database("Found room without m.room.create event."))?
+			.ok_or_else(|| err!(Database(error!("Found room without m.room.create event."))))?
 			.content
 			.get(),
 	)
-	.map_err(|_| Error::bad_database("Invalid room event in database."))?;
+	.map_err(|_| err!(Database(error!("Invalid room event in database."))))?;
 
 	// Setting events_default and invite to the greater of 50 and users_default + 1
 	let new_level = max(
@@ -780,9 +834,7 @@ pub(crate) async fn upgrade_room_route(
 		power_levels_event_content
 			.users_default
 			.checked_add(int!(1))
-			.ok_or_else(|| {
+			.ok_or_else(|| err!(Request(BadJson("users_default power levels event content is not valid"))))?,
-				Error::BadRequest(ErrorKind::BadJson, "users_default power levels event content is not valid")
-			})?,
 	);
 	power_levels_event_content.events_default = new_level;
 	power_levels_event_content.invite = new_level;
@@ -799,7 +851,6 @@ pub(crate) async fn upgrade_room_route(
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			sender_user,
 			&body.room_id,
@@ -847,7 +898,7 @@ fn default_power_levels_content(
 
 	if let Some(power_level_content_override) = power_level_content_override {
 		let json: JsonObject = serde_json::from_str(power_level_content_override.json().get())
-			.map_err(|_| Error::BadRequest(ErrorKind::BadJson, "Invalid power_level_content_override."))?;
+			.map_err(|_| err!(Request(BadJson("Invalid power_level_content_override"))))?;
 
 		for (key, value) in json {
 			power_levels_content[key] = value;

src/api/client/session.rs

@@ -34,9 +34,8 @@ struct Claims {
 ///
 /// Get the supported login types of this server. One of these should be used as
 /// the `type` field when logging in.
-#[tracing::instrument(skip_all, fields(%client), name = "login")]
 pub(crate) async fn get_login_types_route(
-	InsecureClientIp(client): InsecureClientIp, _body: Ruma<get_login_types::v3::Request>,
+	_body: Ruma<get_login_types::v3::Request>,
 ) -> Result<get_login_types::v3::Response> {
 	Ok(get_login_types::v3::Response::new(vec![
 		get_login_types::v3::LoginType::Password(PasswordLoginType::default()),

src/api/client/state.rs

@@ -43,11 +43,6 @@ pub(crate) async fn send_state_event_for_key_route(
 			&body.event_type,
 			&body.body.body,
 			body.state_key.clone(),
-			if body.appservice_info.is_some() {
-				body.timestamp
-			} else {
-				None
-			},
 		)
 		.await?
 		.into(),
@@ -177,7 +172,7 @@ pub(crate) async fn get_state_events_for_empty_key_route(
 
 async fn send_state_event_for_key_helper(
 	services: &Services, sender: &UserId, room_id: &RoomId, event_type: &StateEventType,
-	json: &Raw<AnyStateEventContent>, state_key: String, timestamp: Option<ruma::MilliSecondsSinceUnixEpoch>,
+	json: &Raw<AnyStateEventContent>, state_key: String,
 ) -> Result<Arc<EventId>> {
 	allowed_to_send_state_event(services, room_id, event_type, json).await?;
 	let state_lock = services.rooms.state.mutex.lock(room_id).await;
@@ -191,7 +186,6 @@ async fn send_state_event_for_key_helper(
 				unsigned: None,
 				state_key: Some(state_key),
 				redacts: None,
-				timestamp,
 			},
 			sender,
 			room_id,

src/api/mod.rs

@@ -10,7 +10,8 @@ extern crate conduit_service as service;
 pub(crate) use conduit::{debug_info, pdu::PduEvent, utils, Error, Result};
 pub(crate) use service::services;
 
-pub(crate) use self::router::{Ruma, RumaResponse, State};
+pub use crate::router::State;
+pub(crate) use crate::router::{Ruma, RumaResponse};
 
 conduit::mod_ctor! {}
 conduit::mod_dtor! {}

src/api/router.rs

@@ -3,7 +3,8 @@ mod auth;
 mod handler;
 mod request;
 mod response;
-pub mod state;
+
+use std::sync::Arc;
 
 use axum::{
 	response::IntoResponse,
@@ -14,9 +15,11 @@ use conduit::{err, Server};
 use http::Uri;
 
 use self::handler::RouterExt;
-pub(super) use self::{args::Args as Ruma, response::RumaResponse, state::State};
+pub(super) use self::{args::Args as Ruma, response::RumaResponse};
 use crate::{client, server};
 
+pub type State = Arc<service::Services>;
+
 pub fn build(router: Router<State>, server: &Server) -> Router<State> {
 	let config = &server.config;
 	let router = router

src/api/router/auth.rs

@@ -261,7 +261,8 @@ async fn auth_server(
 	};
 
 	let keys_result = services
-		.server_keys
+		.rooms
+		.event_handler
 		.fetch_signing_keys_for_server(origin, vec![x_matrix.key.to_string()])
 		.await;
 

src/api/router/state.rs

@@ -1,79 +0,0 @@
-use std::{ops::Deref, sync::Arc};
-
-use conduit_service::Services;
-
-#[derive(Clone, Copy)]
-pub struct State {
-	services: *const Services,
-}
-
-pub struct Guard {
-	services: Arc<Services>,
-}
-
-pub fn create(services: Arc<Services>) -> (State, Guard) {
-	let state = State {
-		services: Arc::into_raw(services.clone()),
-	};
-
-	let guard = Guard {
-		services,
-	};
-
-	(state, guard)
-}
-
-impl Drop for Guard {
-	fn drop(&mut self) {
-		let ptr = Arc::as_ptr(&self.services);
-		// SAFETY: Parity with Arc::into_raw() called in create(). This revivifies the
-		// Arc lost to State so it can be dropped, otherwise Services will leak.
-		let arc = unsafe { Arc::from_raw(ptr) };
-		debug_assert!(
-			Arc::strong_count(&arc) > 1,
-			"Services usually has more than one reference and is not dropped here"
-		);
-	}
-}
-
-impl Deref for State {
-	type Target = Services;
-
-	fn deref(&self) -> &Self::Target {
-		deref(&self.services).expect("dereferenced Services pointer in State must not be null")
-	}
-}
-
-/// SAFETY: State is a thin wrapper containing a raw const pointer to Services
-/// in lieu of an Arc. Services is internally threadsafe. If State contains
-/// additional fields this notice should be reevaluated.
-unsafe impl Send for State {}
-
-/// SAFETY: State is a thin wrapper containing a raw const pointer to Services
-/// in lieu of an Arc. Services is internally threadsafe. If State contains
-/// additional fields this notice should be reevaluated.
-unsafe impl Sync for State {}
-
-fn deref(services: &*const Services) -> Option<&Services> {
-	// SAFETY: We replaced Arc<Services> with *const Services in State. This is
-	// worth about 10 clones (20 reference count updates) for each request handled.
-	// Though this is not an incredibly large quantity, it's woefully unnecessary
-	// given the context as explained below; though it is not currently known to be
-	// a performance bottleneck, the front-line position justifies preempting it.
-	//
-	// Services is created prior to the axum/tower stack and Router, and prior
-	// to serving any requests through the handlers in this crate. It is then
-	// dropped only after all requests have completed, the listening sockets
-	// have been closed, axum/tower has been dropped. Thus Services is
-	// expected to live at least as long as any instance of State, making the
-	// constant updates to the prior Arc unnecessary to keep Services alive.
-	//
-	// Nevertheless if it is possible to accomplish this by annotating State
-	// with a lifetime to hold a reference (and be aware I have made a
-	// significant effort trying to make this work) this unsafety may not be
-	// necessary. It is either very difficult or impossible to get a
-	// lifetime'ed reference through Router / RumaHandler; though it is
-	// possible to pass a reference through axum's `with_state()` in trivial
-	// configurations as the only requirement of a State is Clone.
-	unsafe { services.as_ref() }
-}

src/api/server/make_join.rs

@@ -139,7 +139,6 @@ pub(crate) async fn create_join_event_template_route(
 			unsigned: None,
 			state_key: Some(body.user_id.to_string()),
 			redacts: None,
-			timestamp: None,
 		},
 		&body.user_id,
 		&body.room_id,

src/api/server/make_leave.rs

@@ -57,7 +57,6 @@ pub(crate) async fn create_leave_event_template_route(
 			unsigned: None,
 			state_key: Some(body.user_id.to_string()),
 			redacts: None,
-			timestamp: None,
 		},
 		&body.user_id,
 		&body.room_id,

src/api/server/send.rs

@@ -105,7 +105,8 @@ async fn handle_pdus(
 	let pub_key_map = RwLock::new(BTreeMap::new());
 	if !parsed_pdus.is_empty() {
 		services
-			.server_keys
+			.rooms
+			.event_handler
 			.fetch_required_signing_keys(parsed_pdus.iter().map(|(_event_id, event, _room_id)| event), &pub_key_map)
 			.await
 			.unwrap_or_else(|e| warn!("Could not fetch all signatures for PDUs from {origin}: {e:?}"));

src/api/server/send_join.rs

@@ -138,7 +138,8 @@ async fn create_join_event(
 	}
 
 	services
-		.server_keys
+		.rooms
+		.event_handler
 		.fetch_required_signing_keys([&value], &pub_key_map)
 		.await?;
 

src/api/server/send_leave.rs

@@ -150,7 +150,8 @@ async fn create_leave_event(
 	.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "origin is not a server name."))?;
 
 	services
-		.server_keys
+		.rooms
+		.event_handler
 		.fetch_required_signing_keys([&value], &pub_key_map)
 		.await?;
 

src/core/Cargo.toml

@@ -51,13 +51,11 @@ sha256_media = []
 
 [dependencies]
 argon2.workspace = true
-arrayvec.workspace = true
 axum.workspace = true
 bytes.workspace = true
 cargo_toml.workspace = true
 checked_ops.workspace = true
 chrono.workspace = true
-clap.workspace = true
 conduit-macros.workspace = true
 const-str.workspace = true
 ctor.workspace = true

src/core/config/mod.rs

@@ -236,8 +236,6 @@ pub struct Config {
 	pub rocksdb_compaction_ioprio_idle: bool,
 	#[serde(default = "true_fn")]
 	pub rocksdb_compaction: bool,
-	#[serde(default = "default_rocksdb_stats_level")]
-	pub rocksdb_stats_level: u8,
 
 	pub emergency_password: Option<String>,
 
@@ -289,7 +287,7 @@ pub struct Config {
 
 	#[serde(default = "true_fn")]
 	pub media_startup_check: bool,
-	#[serde(default)]
+	#[serde(default = "true_fn")]
 	pub media_compat_file_link: bool,
 	#[serde(default = "Vec::new")]
 	pub prevent_media_downloads_from: Vec<OwnedServerName>,
@@ -720,7 +718,6 @@ impl fmt::Display for Config {
 			&self.rocksdb_compaction_ioprio_idle.to_string(),
 		);
 		line("RocksDB Compaction enabled", &self.rocksdb_compaction.to_string());
-		line("RocksDB Statistics level", &self.rocksdb_stats_level.to_string());
 		line("Media integrity checks on startup", &self.media_startup_check.to_string());
 		line("Media compatibility filesystem links", &self.media_compat_file_link.to_string());
 		line("Prevent Media Downloads From", {
@@ -1005,8 +1002,6 @@ fn default_rocksdb_compression_level() -> i32 { 32767 }
 #[allow(clippy::doc_markdown)]
 fn default_rocksdb_bottommost_compression_level() -> i32 { 32767 }
 
-fn default_rocksdb_stats_level() -> u8 { 1 }
-
 // I know, it's a great name
 #[must_use]
 pub fn default_default_room_version() -> RoomVersionId { RoomVersionId::V10 }

src/core/debug.rs

@@ -14,7 +14,7 @@ pub use crate::utils::debug::*;
 #[macro_export]
 macro_rules! debug_event {
 	( $level:expr, $($x:tt)+ ) => {
-		if $crate::debug::logging() {
+		if cfg!(debug_assertions) && cfg!(not(feature = "dev_release_log_level")) {
 			::tracing::event!( $level, $($x)+ )
 		} else {
 			::tracing::debug!( $($x)+ )
@@ -88,7 +88,3 @@ pub fn panic_str(p: &Box<dyn Any + Send>) -> &'static str { p.downcast_ref::<&st
 #[inline(always)]
 #[must_use]
 pub fn type_name<T>(_: &T) -> &'static str { std::any::type_name::<T>() }
-
-#[must_use]
-#[inline]
-pub const fn logging() -> bool { cfg!(debug_assertions) && cfg!(not(feature = "dev_release_log_level")) }

src/core/error/err.rs

@@ -41,149 +41,60 @@ macro_rules! Err {
 
 #[macro_export]
 macro_rules! err {
-	(Request(Forbidden($level:ident!($($args:tt)+)))) => {{
+	(Config($item:literal, $($args:expr),*)) => {{
-		let mut buf = String::new();
+		$crate::error!(config = %$item, $($args),*);
+		$crate::error::Error::Config($item, $crate::format_maybe!($($args),*))
+	}};
+
+	(Request(Forbidden($level:ident!($($args:expr),*)))) => {{
+		$crate::$level!($($args),*);
 		$crate::error::Error::Request(
 			::ruma::api::client::error::ErrorKind::forbidden(),
-			$crate::err_log!(buf, $level, $($args)+),
+			$crate::format_maybe!($($args),*),
 			::http::StatusCode::BAD_REQUEST
 		)
 	}};
 
-	(Request(Forbidden($($args:tt)+))) => {
+	(Request(Forbidden($($args:expr),*))) => {
 		$crate::error::Error::Request(
 			::ruma::api::client::error::ErrorKind::forbidden(),
-			$crate::format_maybe!($($args)+),
+			$crate::format_maybe!($($args),*),
 			::http::StatusCode::BAD_REQUEST
 		)
 	};
 
-	(Request($variant:ident($level:ident!($($args:tt)+)))) => {{
+	(Request($variant:ident($level:ident!($($args:expr),*)))) => {{
-		let mut buf = String::new();
+		$crate::$level!($($args),*);
 		$crate::error::Error::Request(
 			::ruma::api::client::error::ErrorKind::$variant,
-			$crate::err_log!(buf, $level, $($args)+),
+			$crate::format_maybe!($($args),*),
 			::http::StatusCode::BAD_REQUEST
 		)
 	}};
 
-	(Request($variant:ident($($args:tt)+))) => {
+	(Request($variant:ident($($args:expr),*))) => {
 		$crate::error::Error::Request(
 			::ruma::api::client::error::ErrorKind::$variant,
-			$crate::format_maybe!($($args)+),
+			$crate::format_maybe!($($args),*),
 			::http::StatusCode::BAD_REQUEST
 		)
 	};
 
-	(Config($item:literal, $($args:tt)+)) => {{
+	($variant:ident($level:ident!($($args:expr),*))) => {{
-		let mut buf = String::new();
+		$crate::$level!($($args),*);
-		$crate::error::Error::Config($item, $crate::err_log!(buf, error, config = %$item, $($args)+))
+		$crate::error::Error::$variant($crate::format_maybe!($($args),*))
 	}};
 
-	($variant:ident($level:ident!($($args:tt)+))) => {{
+	($variant:ident($($args:expr),*)) => {
-		let mut buf = String::new();
+		$crate::error::Error::$variant($crate::format_maybe!($($args),*))
-		$crate::error::Error::$variant($crate::err_log!(buf, $level, $($args)+))
+	};
+
+	($level:ident!($($args:expr),*)) => {{
+		$crate::$level!($($args),*);
+		$crate::error::Error::Err($crate::format_maybe!($($args),*))
 	}};
 
-	($variant:ident($($args:tt)+)) => {
+	($($args:expr),*) => {
-		$crate::error::Error::$variant($crate::format_maybe!($($args)+))
+		$crate::error::Error::Err($crate::format_maybe!($($args),*))
-	};
-
-	($level:ident!($($args:tt)+)) => {{
-		let mut buf = String::new();
-		$crate::error::Error::Err($crate::err_log!(buf, $level, $($args)+))
-	}};
-
-	($($args:tt)+) => {
-		$crate::error::Error::Err($crate::format_maybe!($($args)+))
-	};
-}
-
-/// A trinity of integration between tracing, logging, and Error. This is a
-/// customization of tracing::event! with the primary purpose of sharing the
-/// error string, fieldset parsing and formatting. An added benefit is that we
-/// can share the same callsite metadata for the source of our Error and the
-/// associated logging and tracing event dispatches.
-#[macro_export]
-macro_rules! err_log {
-	($out:ident, $level:ident, $($fields:tt)+) => {{
-		use std::{fmt, fmt::Write};
-
-		use ::tracing::{
-			callsite, callsite2, level_enabled, metadata, valueset, Callsite, Event, __macro_support,
-			__tracing_log,
-			field::{Field, ValueSet, Visit},
-			Level,
-		};
-
-		const LEVEL: Level = $crate::err_lev!($level);
-		static __CALLSITE: callsite::DefaultCallsite = callsite2! {
-			name: std::concat! {
-				"event ",
-				std::file!(),
-				":",
-				std::line!(),
-			},
-			kind: metadata::Kind::EVENT,
-			target: std::module_path!(),
-			level: LEVEL,
-			fields: $($fields)+,
-		};
-
-		let visit = &mut |vs: ValueSet<'_>| {
-			struct Visitor<'a>(&'a mut String);
-			impl Visit for Visitor<'_> {
-				fn record_debug(&mut self, field: &Field, val: &dyn fmt::Debug) {
-					if field.name() == "message" {
-						write!(self.0, "{:?}", val).expect("stream error");
-					} else {
-						write!(self.0, " {}={:?}", field.name(), val).expect("stream error");
-					}
-				}
-			}
-
-			let meta = __CALLSITE.metadata();
-			let enabled = level_enabled!(LEVEL) && {
-				let interest = __CALLSITE.interest();
-				!interest.is_never() && __macro_support::__is_enabled(meta, interest)
-			};
-
-			if enabled {
-				Event::dispatch(meta, &vs);
-			}
-
-			__tracing_log!(LEVEL, __CALLSITE, &vs);
-			vs.record(&mut Visitor(&mut $out));
-		};
-
-		(visit)(valueset!(__CALLSITE.metadata().fields(), $($fields)+));
-		($out).into()
-	}}
-}
-
-#[macro_export]
-macro_rules! err_lev {
-	(debug_warn) => {
-		if $crate::debug::logging() {
-			::tracing::Level::WARN
-		} else {
-			::tracing::Level::DEBUG
-		}
-	};
-
-	(debug_error) => {
-		if $crate::debug::logging() {
-			::tracing::Level::ERROR
-		} else {
-			::tracing::Level::DEBUG
-		}
-	};
-
-	(warn) => {
-		::tracing::Level::WARN
-	};
-
-	(error) => {
-		::tracing::Level::ERROR
 	};
 }

src/core/error/mod.rs

@@ -5,7 +5,8 @@ mod response;
 
 use std::{any::Any, borrow::Cow, convert::Infallible, fmt};
 
-pub use self::log::*;
+pub use log::*;
+
 use crate::error;
 
 #[derive(thiserror::Error)]
@@ -62,8 +63,6 @@ pub enum Error {
 	TomlSerError(#[from] toml::ser::Error),
 	#[error("{0}")]
 	TomlDeError(#[from] toml::de::Error),
-	#[error("{0}")]
-	Clap(#[from] clap::error::Error),
 
 	// ruma
 	#[error("{0}")]
@@ -88,13 +87,13 @@ pub enum Error {
 	// conduwuit
 	#[error("Arithmetic operation failed: {0}")]
 	Arithmetic(Cow<'static, str>),
-	#[error("{0}")]
-	BadServerResponse(Cow<'static, str>),
 	#[error("There was a problem with the '{0}' directive in your configuration: {1}")]
 	Config(&'static str, Cow<'static, str>),
 	#[error("{0}")]
 	Database(Cow<'static, str>),
 	#[error("{0}")]
+	BadServerResponse(&'static str),
+	#[error("{0}")]
 	Conflict(&'static str), // This is only needed for when a room alias already exists
 
 	// unique / untyped

src/core/info/mod.rs

@@ -6,6 +6,3 @@ pub mod rustc;
 pub mod version;
 
 pub use conduit_macros::rustc_flags_capture;
-
-pub const MODULE_ROOT: &str = const_str::split!(std::module_path!(), "::")[0];
-pub const CRATE_PREFIX: &str = const_str::split!(MODULE_ROOT, '_')[0];

src/core/log/capture/data.rs

@@ -2,20 +2,15 @@ use tracing::Level;
 use tracing_core::{span::Current, Event};
 
 use super::{layer::Value, Layer};
-use crate::{info, utils::string::EMPTY};
 
 pub struct Data<'a> {
 	pub layer: &'a Layer,
 	pub event: &'a Event<'a>,
 	pub current: &'a Current,
-	pub values: &'a [Value],
+	pub values: Option<&'a mut [Value]>,
-	pub scope: &'a [&'static str],
 }
 
 impl Data<'_> {
-	#[must_use]
-	pub fn our_modules(&self) -> bool { self.mod_name().starts_with(info::CRATE_PREFIX) }
-
 	#[must_use]
 	pub fn level(&self) -> Level { *self.event.metadata().level() }
 
@@ -23,13 +18,15 @@ impl Data<'_> {
 	pub fn mod_name(&self) -> &str { self.event.metadata().module_path().unwrap_or_default() }
 
 	#[must_use]
-	pub fn span_name(&self) -> &str { self.current.metadata().map_or(EMPTY, |s| s.name()) }
+	pub fn span_name(&self) -> &str { self.current.metadata().map_or("", |s| s.name()) }
 
 	#[must_use]
 	pub fn message(&self) -> &str {
 		self.values
+			.as_ref()
+			.expect("values are not composed for a filter")
 			.iter()
 			.find(|(k, _)| *k == "message")
-			.map_or(EMPTY, |(_, v)| v.as_str())
+			.map_or("", |(_, v)| v.as_str())
 	}
 }

src/core/log/capture/layer.rs

@@ -1,25 +1,21 @@
 use std::{fmt, sync::Arc};
 
-use arrayvec::ArrayVec;
 use tracing::field::{Field, Visit};
 use tracing_core::{Event, Subscriber};
 use tracing_subscriber::{layer::Context, registry::LookupSpan};
 
 use super::{Capture, Data, State};
 
+pub type Value = (&'static str, String);
+
 pub struct Layer {
 	state: Arc<State>,
 }
 
 struct Visitor {
-	values: Values,
+	values: Vec<Value>,
 }
 
-type Values = ArrayVec<Value, 32>;
-pub type Value = (&'static str, String);
-
-type ScopeNames = ArrayVec<&'static str, 32>;
-
 impl Layer {
 	#[inline]
 	pub fn new(state: &Arc<State>) -> Self {
@@ -55,9 +51,8 @@ fn handle<S>(layer: &Layer, capture: &Capture, event: &Event<'_>, ctx: &Context<
 where
 	S: Subscriber + for<'a> LookupSpan<'a>,
 {
-	let names = ScopeNames::new();
 	let mut visitor = Visitor {
-		values: Values::new(),
+		values: Vec::new(),
 	};
 	event.record(&mut visitor);
 
@@ -66,8 +61,7 @@ where
 		layer,
 		event,
 		current: &ctx.current_span(),
-		values: &visitor.values,
+		values: Some(&mut visitor.values),
-		scope: &names,
 	});
 }
 
@@ -75,21 +69,12 @@ fn filter<S>(layer: &Layer, capture: &Capture, event: &Event<'_>, ctx: &Context<
 where
 	S: Subscriber + for<'a> LookupSpan<'a>,
 {
-	let values = Values::new();
-	let mut names = ScopeNames::new();
-	if let Some(scope) = ctx.event_scope(event) {
-		for span in scope {
-			names.push(span.name());
-		}
-	}
-
 	capture.filter.as_ref().map_or(true, |filter| {
 		filter(Data {
 			layer,
 			event,
 			current: &ctx.current_span(),
-			values: &values,
+			values: None,
-			scope: &names,
 		})
 	})
 }

src/core/log/fmt.rs

@@ -5,7 +5,7 @@ use crate::Result;
 
 pub fn html<S>(out: &mut S, level: &Level, span: &str, msg: &str) -> Result<()>
 where
-	S: Write + ?Sized,
+	S: Write,
 {
 	let color = color::code_tag(level);
 	let level = level.as_str().to_uppercase();
@@ -19,7 +19,7 @@ where
 
 pub fn markdown<S>(out: &mut S, level: &Level, span: &str, msg: &str) -> Result<()>
 where
-	S: Write + ?Sized,
+	S: Write,
 {
 	let level = level.as_str().to_uppercase();
 	writeln!(out, "`{level:>5}` `{span:^12}` `{msg}`")?;
@@ -29,19 +29,19 @@ where
 
 pub fn markdown_table<S>(out: &mut S, level: &Level, span: &str, msg: &str) -> Result<()>
 where
-	S: Write + ?Sized,
+	S: Write,
 {
 	let level = level.as_str().to_uppercase();
-	writeln!(out, "| {level:>5} | {span:^12} | {msg} |")?;
+	writeln!(out, "| `{level:>5}` | `{span:^12}` | `{msg} |")?;
 
 	Ok(())
 }
 
 pub fn markdown_table_head<S>(out: &mut S) -> Result<()>
 where
-	S: Write + ?Sized,
+	S: Write,
 {
-	write!(out, "| level | span | message |\n| ------: | :-----: | :------- |\n")?;
+	write!(out, "| level | span | message |\n|------:|:----:|:--------|\n")?;
 
 	Ok(())
 }

src/core/mod.rs

@@ -16,7 +16,7 @@ pub use error::Error;
 pub use info::{rustc_flags_capture, version, version::version};
 pub use pdu::{PduBuilder, PduCount, PduEvent};
 pub use server::Server;
-pub use utils::{ctor, dtor, implement};
+pub use utils::{ctor, dtor};
 
 pub use crate as conduit_core;
 

src/core/pdu/builder.rs

@@ -1,6 +1,6 @@
 use std::{collections::BTreeMap, sync::Arc};
 
-use ruma::{events::TimelineEventType, EventId, MilliSecondsSinceUnixEpoch};
+use ruma::{events::TimelineEventType, EventId};
 use serde::Deserialize;
 use serde_json::value::RawValue as RawJsonValue;
 
@@ -13,8 +13,4 @@ pub struct PduBuilder {
 	pub unsigned: Option<BTreeMap<String, serde_json::Value>>,
 	pub state_key: Option<String>,
 	pub redacts: Option<Arc<EventId>>,
-	/// For timestamped messaging, should only be used for appservices
-	///
-	/// Will be set to current time if None
-	pub timestamp: Option<MilliSecondsSinceUnixEpoch>,
 }

src/core/pdu/mod.rs

@@ -23,7 +23,7 @@ use serde_json::{
 	value::{to_raw_value, RawValue as RawJsonValue},
 };
 
-use crate::{err, warn, Error};
+use crate::{warn, Error};
 
 #[derive(Deserialize)]
 struct ExtractRedactedBecause {
@@ -396,8 +396,10 @@ impl Ord for PduEvent {
 pub fn gen_event_id_canonical_json(
 	pdu: &RawJsonValue, room_version_id: &RoomVersionId,
 ) -> crate::Result<(OwnedEventId, CanonicalJsonObject)> {
-	let value: CanonicalJsonObject = serde_json::from_str(pdu.get())
+	let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
-		.map_err(|e| err!(BadServerResponse(warn!("Error parsing incoming event: {e:?}"))))?;
+		warn!("Error parsing incoming event {:?}: {:?}", pdu, e);
+		Error::BadServerResponse("Invalid PDU in server response")
+	})?;
 
 	let event_id = format!(
 		"${}",

src/core/utils/mod.rs

@@ -17,7 +17,6 @@ use std::cmp::{self, Ordering};
 
 pub use ::ctor::{ctor, dtor};
 pub use bytes::{increment, u64_from_bytes, u64_from_u8, u64_from_u8x8};
-pub use conduit_macros::implement;
 pub use debug::slice_truncated as debug_slice_truncated;
 pub use hash::calculate_hash;
 pub use html::Escape as HtmlEscape;

src/core/utils/string.rs

@@ -8,12 +8,12 @@ pub const EMPTY: &str = "";
 /// arguments are provided the first is assumed to be a format string.
 #[macro_export]
 macro_rules! format_maybe {
-	($s:literal $(,)?) => {
+	($s:literal) => {
 		if $crate::is_format!($s) { std::format!($s).into() } else { $s.into() }
 	};
 
-	($s:literal, $($args:tt)+) => {
+	($($args:expr),*) => {
-		std::format!($s, $($args)+).into()
+		std::format!($($args),*).into()
 	};
 }
 
@@ -24,20 +24,6 @@ macro_rules! is_format {
 	($s:literal) => {
 		::const_str::contains!($s, "{") && ::const_str::contains!($s, "}")
 	};
-
-	($($s:tt)+) => {
-		false
-	};
-}
-
-#[inline]
-pub fn collect_stream<F>(func: F) -> Result<String>
-where
-	F: FnOnce(&mut dyn std::fmt::Write) -> Result<()>,
-{
-	let mut out = String::new();
-	func(&mut out)?;
-	Ok(out)
 }
 
 #[inline]

src/database/database.rs

@@ -2,12 +2,7 @@ use std::{ops::Index, sync::Arc};
 
 use conduit::{Result, Server};
 
-use crate::{
+use crate::{cork::Cork, maps, maps::Maps, Engine, Map};
-	cork::Cork,
-	maps,
-	maps::{Maps, MapsKey, MapsVal},
-	Engine, Map,
-};
 
 pub struct Database {
 	pub db: Arc<Engine>,
@@ -35,9 +30,6 @@ impl Database {
 	#[inline]
 	#[must_use]
 	pub fn cork_and_sync(&self) -> Cork { Cork::new(&self.db, true, true) }
-
-	#[inline]
-	pub fn iter_maps(&self) -> impl Iterator<Item = (&MapsKey, &MapsVal)> + '_ { self.map.iter() }
 }
 
 impl Index<&str> for Database {

src/database/engine.rs

@@ -1,6 +1,5 @@
 use std::{
 	collections::{BTreeSet, HashMap},
-	ffi::CStr,
 	fmt::Write,
 	path::PathBuf,
 	sync::{atomic::AtomicU32, Arc, Mutex, RwLock},
@@ -10,8 +9,7 @@ use conduit::{debug, error, info, utils::time::rfc2822_from_seconds, warn, Err,
 use rocksdb::{
 	backup::{BackupEngine, BackupEngineOptions},
 	perf::get_memory_usage_stats,
-	AsColumnFamilyRef, BoundColumnFamily, Cache, ColumnFamilyDescriptor, DBCommon, DBWithThreadMode, Env,
+	BoundColumnFamily, Cache, ColumnFamilyDescriptor, DBCommon, DBWithThreadMode, Env, MultiThreaded, Options,
-	MultiThreaded, Options,
 };
 
 use crate::{
@@ -242,20 +240,6 @@ impl Engine {
 			},
 		}
 	}
-
-	/// Query for database property by null-terminated name which is expected to
-	/// have a result with an integer representation. This is intended for
-	/// low-overhead programmatic use.
-	pub(crate) fn property_integer(&self, cf: &impl AsColumnFamilyRef, name: &CStr) -> Result<u64> {
-		result(self.db.property_int_value_cf(cf, name))
-			.and_then(|val| val.map_or_else(|| Err!("Property {name:?} not found."), Ok))
-	}
-
-	/// Query for database property by name receiving the result in a string.
-	pub(crate) fn property(&self, cf: &impl AsColumnFamilyRef, name: &str) -> Result<String> {
-		result(self.db.property_value_cf(cf, name))
-			.and_then(|val| val.map_or_else(|| Err!("Property {name:?} not found."), Ok))
-	}
 }
 
 pub(crate) fn repair(db_opts: &Options, path: &PathBuf) -> Result<()> {

src/database/map.rs

@@ -1,4 +1,4 @@
-use std::{ffi::CStr, future::Future, mem::size_of, pin::Pin, sync::Arc};
+use std::{future::Future, mem::size_of, pin::Pin, sync::Arc};
 
 use conduit::{utils, Result};
 use rocksdb::{
@@ -189,10 +189,6 @@ impl Map {
 		self.watchers.watch(prefix)
 	}
 
-	pub fn property_integer(&self, name: &CStr) -> Result<u64> { self.db.property_integer(&self.cf(), name) }
-
-	pub fn property(&self, name: &str) -> Result<String> { self.db.property(&self.cf(), name) }
-
 	#[inline]
 	pub fn name(&self) -> &str { &self.name }
 

src/database/maps.rs

@@ -4,9 +4,7 @@ use conduit::Result;
 
 use crate::{Engine, Map};
 
-pub type Maps = BTreeMap<MapsKey, MapsVal>;
+pub type Maps = BTreeMap<String, Arc<Map>>;
-pub(crate) type MapsVal = Arc<Map>;
-pub(crate) type MapsKey = String;
 
 pub(crate) fn open(db: &Arc<Engine>) -> Result<Maps> { open_list(db, MAPS) }
 

src/database/opts.rs

@@ -2,8 +2,8 @@ use std::{cmp, collections::HashMap};
 
 use conduit::{utils, Config};
 use rocksdb::{
-	statistics::StatsLevel, BlockBasedOptions, Cache, DBCompactionStyle, DBCompressionType, DBRecoveryMode, Env,
+	BlockBasedOptions, Cache, DBCompactionStyle, DBCompressionType, DBRecoveryMode, Env, LogLevel, Options,
-	LogLevel, Options, UniversalCompactOptions, UniversalCompactionStopStyle,
+	UniversalCompactOptions, UniversalCompactionStopStyle,
 };
 
 /// Create database-wide options suitable for opening the database. This also
@@ -13,11 +13,6 @@ use rocksdb::{
 /// through cf_options().
 pub(crate) fn db_options(config: &Config, env: &mut Env, row_cache: &Cache, col_cache: &Cache) -> Options {
 	const MIN_PARALLELISM: usize = 2;
-	const DEFAULT_STATS_LEVEL: StatsLevel = if cfg!(debug_assertions) {
-		StatsLevel::ExceptDetailedTimers
-	} else {
-		StatsLevel::DisableAll
-	};
 
 	let mut opts = Options::default();
 
@@ -73,18 +68,8 @@ pub(crate) fn db_options(config: &Config, env: &mut Env, row_cache: &Cache, col_
 	set_compression_defaults(&mut opts, config);
 
 	// Misc
-	opts.create_if_missing(true);
 	opts.set_disable_auto_compactions(!config.rocksdb_compaction);
-
+	opts.create_if_missing(true);
-	opts.set_statistics_level(match config.rocksdb_stats_level {
-		0 => StatsLevel::DisableAll,
-		1 => DEFAULT_STATS_LEVEL,
-		2 => StatsLevel::ExceptHistogramOrTimers,
-		3 => StatsLevel::ExceptTimers,
-		4 => StatsLevel::ExceptDetailedTimers,
-		5 => StatsLevel::ExceptTimeForMutex,
-		6_u8..=u8::MAX => StatsLevel::All,
-	});
 
 	// Default: https://github.com/facebook/rocksdb/wiki/WAL-Recovery-Modes#ktoleratecorruptedtailrecords
 	//

src/router/layers.rs

@@ -6,7 +6,6 @@ use axum::{
 };
 use axum_client_ip::SecureClientIpSource;
 use conduit::{error, Result, Server};
-use conduit_api::router::state::Guard;
 use conduit_service::Services;
 use http::{
 	header::{self, HeaderName},
@@ -36,7 +35,7 @@ const CONDUWUIT_CSP: &[&str] = &[
 
 const CONDUWUIT_PERMISSIONS_POLICY: &[&str] = &["interest-cohort=()", "browsing-topics=()"];
 
-pub(crate) fn build(services: &Arc<Services>) -> Result<(Router, Guard)> {
+pub(crate) fn build(services: &Arc<Services>) -> Result<Router> {
 	let server = &services.server;
 	let layers = ServiceBuilder::new();
 
@@ -48,7 +47,7 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<(Router, Guard)> {
 
 	let layers = layers
 		.sensitive_headers([header::AUTHORIZATION])
-		.layer(axum::middleware::from_fn_with_state(Arc::clone(services), request::spawn))
+		.layer(axum::middleware::from_fn_with_state(Arc::clone(server), request::spawn))
 		.layer(
 			TraceLayer::new_for_http()
 				.make_span_with(tracing_span::<_>)
@@ -56,7 +55,7 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<(Router, Guard)> {
 				.on_request(DefaultOnRequest::new().level(Level::TRACE))
 				.on_response(DefaultOnResponse::new().level(Level::DEBUG)),
 		)
-		.layer(axum::middleware::from_fn_with_state(Arc::clone(services), request::handle))
+		.layer(axum::middleware::from_fn_with_state(Arc::clone(server), request::handle))
 		.layer(SecureClientIpSource::ConnectInfo.into_extension())
 		.layer(SetResponseHeaderLayer::if_not_present(
 			HeaderName::from_static("origin-agent-cluster"), // https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Origin-Agent-Cluster
@@ -86,8 +85,7 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<(Router, Guard)> {
 		.layer(body_limit_layer(server))
 		.layer(CatchPanicLayer::custom(catch_panic));
 
-	let (router, guard) = router::build(services);
+	Ok(router::build(services).layer(layers))
-	Ok((router.layer(layers), guard))
 }
 
 #[cfg(any(feature = "zstd_compression", feature = "gzip_compression", feature = "brotli_compression"))]

src/router/request.rs

@@ -4,15 +4,13 @@ use axum::{
 	extract::State,
 	response::{IntoResponse, Response},
 };
-use conduit::{debug, debug_error, debug_warn, defer, err, error, trace, Result};
+use conduit::{debug, debug_error, debug_warn, defer, err, error, trace, Result, Server};
-use conduit_service::Services;
 use http::{Method, StatusCode, Uri};
 
 #[tracing::instrument(skip_all, level = "debug")]
 pub(crate) async fn spawn(
-	State(services): State<Arc<Services>>, req: http::Request<axum::body::Body>, next: axum::middleware::Next,
+	State(server): State<Arc<Server>>, req: http::Request<axum::body::Body>, next: axum::middleware::Next,
 ) -> Result<Response, StatusCode> {
-	let server = &services.server;
 	if !server.running() {
 		debug_warn!("unavailable pending shutdown");
 		return Err(StatusCode::SERVICE_UNAVAILABLE);
@@ -36,9 +34,8 @@ pub(crate) async fn spawn(
 
 #[tracing::instrument(skip_all, name = "handle")]
 pub(crate) async fn handle(
-	State(services): State<Arc<Services>>, req: http::Request<axum::body::Body>, next: axum::middleware::Next,
+	State(server): State<Arc<Server>>, req: http::Request<axum::body::Body>, next: axum::middleware::Next,
 ) -> Result<Response, StatusCode> {
-	let server = &services.server;
 	if !server.running() {
 		debug_warn!(
 			method = %req.method(),

src/router/router.rs

@@ -2,20 +2,19 @@ use std::sync::Arc;
 
 use axum::{response::IntoResponse, routing::get, Router};
 use conduit::Error;
-use conduit_api::router::{state, state::Guard};
+use conduit_api::State;
 use conduit_service::Services;
 use http::{StatusCode, Uri};
 use ruma::api::client::error::ErrorKind;
 
-pub(crate) fn build(services: &Arc<Services>) -> (Router, Guard) {
+pub(crate) fn build(services: &Arc<Services>) -> Router {
-	let router = Router::<state::State>::new();
+	let router = Router::<State>::new();
-	let (state, guard) = state::create(services.clone());
+	let state = services.clone();
-	let router = conduit_api::router::build(router, &services.server)
+
+	conduit_api::router::build(router, &services.server)
 		.route("/", get(it_works))
 		.fallback(not_found)
-		.with_state(state);
+		.with_state(state)
-
-	(router, guard)
 }
 
 async fn not_found(_uri: Uri) -> impl IntoResponse {

src/router/serve/mod.rs

@@ -18,7 +18,7 @@ pub(super) async fn serve(
 	let server = &services.server;
 	let config = &server.config;
 	let addrs = config.get_bind_addrs();
-	let (app, _guard) = layers::build(&services)?;
+	let app = layers::build(&services)?;
 
 	if cfg!(unix) && config.unix_socket_path.is_some() {
 		unix::serve(server, app, shutdown).await

src/service/Cargo.toml

@@ -69,7 +69,6 @@ termimad.optional = true
 tokio.workspace = true
 tracing.workspace = true
 url.workspace = true
-webpage.workspace = true
 
 [lints]
 workspace = true

src/service/admin/create.rs

@@ -71,7 +71,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,
@@ -100,7 +99,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(server_user.to_string()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,
@@ -126,7 +124,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,
@@ -146,7 +143,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,
@@ -166,7 +162,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,
@@ -186,7 +181,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,
@@ -207,7 +201,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,
@@ -228,7 +221,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,
@@ -253,7 +245,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,
@@ -280,7 +271,6 @@ pub async fn create_admin_room(services: &Services) -> Result<()> {
 				unsigned: None,
 				state_key: Some(String::new()),
 				redacts: None,
-				timestamp: None,
 			},
 			server_user,
 			&room_id,

src/service/admin/grant.rs

@@ -50,7 +50,6 @@ impl super::Service {
 					unsigned: None,
 					state_key: Some(user_id.to_string()),
 					redacts: None,
-					timestamp: None,
 				},
 				server_user,
 				&room_id,
@@ -76,7 +75,6 @@ impl super::Service {
 					unsigned: None,
 					state_key: Some(user_id.to_string()),
 					redacts: None,
-					timestamp: None,
 				},
 				user_id,
 				&room_id,
@@ -102,7 +100,6 @@ impl super::Service {
 					unsigned: None,
 					state_key: Some(String::new()),
 					redacts: None,
-					timestamp: None,
 				},
 				server_user,
 				&room_id,
@@ -122,7 +119,6 @@ impl super::Service {
                 unsigned: None,
                 state_key: None,
                 redacts: None,
-                timestamp: None,
             },
             server_user,
             &room_id,

src/service/admin/mod.rs

@@ -51,9 +51,9 @@ pub struct CommandInput {
 }
 
 pub type Completer = fn(&str) -> String;
-pub type Handler = fn(Arc<crate::Services>, CommandInput) -> HandlerFuture;
+pub type Handler = fn(Arc<crate::Services>, CommandInput) -> HandlerResult;
-pub type HandlerFuture = Pin<Box<dyn Future<Output = HandlerResult> + Send>>;
+pub type HandlerResult = Pin<Box<dyn Future<Output = CommandResult> + Send>>;
-pub type HandlerResult = Result<CommandOutput>;
+pub type CommandResult = Result<CommandOutput, Error>;
 pub type CommandOutput = Option<RoomMessageEventContent>;
 
 const COMMAND_QUEUE_LIMIT: usize = 512;
@@ -173,7 +173,7 @@ impl Service {
 		}
 	}
 
-	async fn process_command(&self, command: CommandInput) -> HandlerResult {
+	async fn process_command(&self, command: CommandInput) -> CommandResult {
 		let Some(services) = self
 			.services
 			.services
@@ -260,7 +260,6 @@ impl Service {
 			unsigned: None,
 			state_key: None,
 			redacts: None,
-			timestamp: None,
 		};
 
 		if let Err(e) = self
@@ -290,7 +289,6 @@ impl Service {
 			unsigned: None,
 			state_key: None,
 			redacts: None,
-			timestamp: None,
 		};
 
 		self.services

src/service/globals/migrations.rs

@@ -743,7 +743,6 @@ async fn db_lt_13(services: &Services) -> Result<()> {
 /// again.
 async fn migrate_sha256_media(services: &Services) -> Result<()> {
 	let db = &services.db;
-	let config = &services.server.config;
 
 	warn!("Migrating legacy base64 file names to sha256 file names");
 	let mediaid_file = &db["mediaid_file"];
@@ -760,9 +759,7 @@ async fn migrate_sha256_media(services: &Services) -> Result<()> {
 	for (old_path, path) in changes {
 		if old_path.exists() {
 			tokio::fs::rename(&old_path, &path).await?;
-			if config.media_compat_file_link {
+			tokio::fs::symlink(&path, &old_path).await?;
-				tokio::fs::symlink(&path, &old_path).await?;
-			}
 		}
 	}
 
@@ -825,14 +822,8 @@ async fn handle_media_check(
 
 	let (mediaid_file, mediaid_user) = dbs;
 
-	let new_exists = files.contains(new_path);
 	let old_exists = files.contains(old_path);
-	let old_is_symlink = || async {
+	let new_exists = files.contains(new_path);
-		tokio::fs::symlink_metadata(old_path)
-			.await
-			.map_or(false, |md| md.is_symlink())
-	};
-
 	if !old_exists && !new_exists {
 		error!(
 			media_id = ?encode_key(key), ?new_path, ?old_path,
@@ -858,29 +849,10 @@ async fn handle_media_check(
 			"Legacy media found without sha256 migration. Fixing..."
 		);
 
-		debug_assert!(
-			old_is_symlink().await,
-			"Legacy media not expected to be a symlink without an existing sha256 migration."
-		);
-
 		tokio::fs::rename(&old_path, &new_path).await?;
 		tokio::fs::symlink(&new_path, &old_path).await?;
 	}
 
-	if !config.media_compat_file_link && old_exists && old_is_symlink().await {
-		debug_warn!(
-			media_id = ?encode_key(key), ?new_path, ?old_path,
-			"Legacy link found but compat disabled. Cleansing symlink..."
-		);
-
-		debug_assert!(
-			new_exists,
-			"sha256 migration into new file expected prior to cleaning legacy symlink here."
-		);
-
-		tokio::fs::remove_file(&old_path).await?;
-	}
-
 	Ok(())
 }
 

src/service/media/data.rs

@@ -4,7 +4,7 @@ use conduit::{debug, debug_info, utils::string_from_bytes, Error, Result};
 use database::{Database, Map};
 use ruma::api::client::error::ErrorKind;
 
-use super::preview::UrlPreviewData;
+use crate::media::UrlPreviewData;
 
 pub(crate) struct Data {
 	mediaid_file: Arc<Map>,

src/service/media/mod.rs

@@ -1,5 +1,4 @@
 mod data;
-mod preview;
 mod tests;
 mod thumbnail;
 
@@ -7,15 +6,16 @@ use std::{path::PathBuf, sync::Arc, time::SystemTime};
 
 use async_trait::async_trait;
 use base64::{engine::general_purpose, Engine as _};
-use conduit::{debug, debug_error, err, error, trace, utils, utils::MutexMap, Err, Result, Server};
+use conduit::{debug, debug_error, err, error, utils, utils::MutexMap, Err, Result, Server};
 use data::{Data, Metadata};
 use ruma::{OwnedMxcUri, OwnedUserId};
+use serde::Serialize;
 use tokio::{
 	fs,
 	io::{AsyncReadExt, AsyncWriteExt, BufReader},
 };
 
-use crate::{client, globals, Dep};
+use crate::{globals, Dep};
 
 #[derive(Debug)]
 pub struct FileMeta {
@@ -24,32 +24,43 @@ pub struct FileMeta {
 	pub content_disposition: Option<String>,
 }
 
+#[derive(Serialize, Default)]
+pub struct UrlPreviewData {
+	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:title"))]
+	pub title: Option<String>,
+	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:description"))]
+	pub description: Option<String>,
+	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image"))]
+	pub image: Option<String>,
+	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "matrix:image:size"))]
+	pub image_size: Option<usize>,
+	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image:width"))]
+	pub image_width: Option<u32>,
+	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image:height"))]
+	pub image_height: Option<u32>,
+}
+
 pub struct Service {
-	url_preview_mutex: MutexMap<String, ()>,
-	pub(crate) db: Data,
 	services: Services,
+	pub(crate) db: Data,
+	pub url_preview_mutex: MutexMap<String, ()>,
 }
 
 struct Services {
 	server: Arc<Server>,
-	client: Dep<client::Service>,
 	globals: Dep<globals::Service>,
 }
 
-/// generated MXC ID (`media-id`) length
-pub const MXC_LENGTH: usize = 32;
-
 #[async_trait]
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		Ok(Arc::new(Self {
-			url_preview_mutex: MutexMap::new(),
-			db: Data::new(args.db),
 			services: Services {
 				server: args.server.clone(),
-				client: args.depend::<client::Service>("client"),
 				globals: args.depend::<globals::Service>("globals"),
 			},
+			db: Data::new(args.db),
+			url_preview_mutex: MutexMap::new(),
 		}))
 	}
 
@@ -88,15 +99,10 @@ impl Service {
 	pub async fn delete(&self, mxc: &str) -> Result<()> {
 		if let Ok(keys) = self.db.search_mxc_metadata_prefix(mxc) {
 			for key in keys {
-				trace!(?mxc, ?key, "Deleting from filesystem");
+				self.remove_media_file(&key).await?;
-				if let Err(e) = self.remove_media_file(&key).await {
-					error!(?mxc, ?key, "Failed to remove media file: {e}");
-				}
 
-				trace!(?mxc, ?key, "Deleting from database");
+				debug!("Deleting MXC {mxc} from database");
-				if let Err(e) = self.db.delete_file_mxc(mxc) {
+				self.db.delete_file_mxc(mxc)?;
-					error!(?mxc, ?key, "Failed to remove media from database: {e}");
-				}
 			}
 
 			Ok(())
@@ -218,6 +224,22 @@ impl Service {
 		Ok(deletion_count)
 	}
 
+	pub async fn get_url_preview(&self, url: &str) -> Option<UrlPreviewData> { self.db.get_url_preview(url) }
+
+	/// TODO: use this?
+	#[allow(dead_code)]
+	pub async fn remove_url_preview(&self, url: &str) -> Result<()> {
+		// TODO: also remove the downloaded image
+		self.db.remove_url_preview(url)
+	}
+
+	pub async fn set_url_preview(&self, url: &str, data: &UrlPreviewData) -> Result<()> {
+		let now = SystemTime::now()
+			.duration_since(SystemTime::UNIX_EPOCH)
+			.expect("valid system time");
+		self.db.set_url_preview(url, data, now)
+	}
+
 	pub async fn create_media_dir(&self) -> Result<()> {
 		let dir = self.get_media_dir();
 		Ok(fs::create_dir_all(dir).await?)

src/service/media/preview.rs

@@ -1,274 +0,0 @@
-use std::{io::Cursor, time::SystemTime};
-
-use conduit::{debug, utils, warn, Err, Result};
-use conduit_core::implement;
-use image::ImageReader as ImgReader;
-use ipaddress::IPAddress;
-use serde::Serialize;
-use url::Url;
-use webpage::HTML;
-
-use super::{Service, MXC_LENGTH};
-
-#[derive(Serialize, Default)]
-pub struct UrlPreviewData {
-	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:title"))]
-	pub title: Option<String>,
-	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:description"))]
-	pub description: Option<String>,
-	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image"))]
-	pub image: Option<String>,
-	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "matrix:image:size"))]
-	pub image_size: Option<usize>,
-	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image:width"))]
-	pub image_width: Option<u32>,
-	#[serde(skip_serializing_if = "Option::is_none", rename(serialize = "og:image:height"))]
-	pub image_height: Option<u32>,
-}
-
-#[implement(Service)]
-pub async fn remove_url_preview(&self, url: &str) -> Result<()> {
-	// TODO: also remove the downloaded image
-	self.db.remove_url_preview(url)
-}
-
-#[implement(Service)]
-pub async fn set_url_preview(&self, url: &str, data: &UrlPreviewData) -> Result<()> {
-	let now = SystemTime::now()
-		.duration_since(SystemTime::UNIX_EPOCH)
-		.expect("valid system time");
-	self.db.set_url_preview(url, data, now)
-}
-
-#[implement(Service)]
-pub async fn download_image(&self, url: &str) -> Result<UrlPreviewData> {
-	let client = &self.services.client.url_preview;
-	let image = client.get(url).send().await?.bytes().await?;
-	let mxc = format!(
-		"mxc://{}/{}",
-		self.services.globals.server_name(),
-		utils::random_string(MXC_LENGTH)
-	);
-
-	self.create(None, &mxc, None, None, &image).await?;
-
-	let (width, height) = match ImgReader::new(Cursor::new(&image)).with_guessed_format() {
-		Err(_) => (None, None),
-		Ok(reader) => match reader.into_dimensions() {
-			Err(_) => (None, None),
-			Ok((width, height)) => (Some(width), Some(height)),
-		},
-	};
-
-	Ok(UrlPreviewData {
-		image: Some(mxc),
-		image_size: Some(image.len()),
-		image_width: width,
-		image_height: height,
-		..Default::default()
-	})
-}
-
-#[implement(Service)]
-pub async fn get_url_preview(&self, url: &str) -> Result<UrlPreviewData> {
-	if let Some(preview) = self.db.get_url_preview(url) {
-		return Ok(preview);
-	}
-
-	// ensure that only one request is made per URL
-	let _request_lock = self.url_preview_mutex.lock(url).await;
-
-	match self.db.get_url_preview(url) {
-		Some(preview) => Ok(preview),
-		None => self.request_url_preview(url).await,
-	}
-}
-
-#[implement(Service)]
-async fn request_url_preview(&self, url: &str) -> Result<UrlPreviewData> {
-	if let Ok(ip) = IPAddress::parse(url) {
-		if !self.services.globals.valid_cidr_range(&ip) {
-			return Err!(BadServerResponse("Requesting from this address is forbidden"));
-		}
-	}
-
-	let client = &self.services.client.url_preview;
-	let response = client.head(url).send().await?;
-
-	if let Some(remote_addr) = response.remote_addr() {
-		if let Ok(ip) = IPAddress::parse(remote_addr.ip().to_string()) {
-			if !self.services.globals.valid_cidr_range(&ip) {
-				return Err!(BadServerResponse("Requesting from this address is forbidden"));
-			}
-		}
-	}
-
-	let Some(content_type) = response
-		.headers()
-		.get(reqwest::header::CONTENT_TYPE)
-		.and_then(|x| x.to_str().ok())
-	else {
-		return Err!(Request(Unknown("Unknown Content-Type")));
-	};
-	let data = match content_type {
-		html if html.starts_with("text/html") => self.download_html(url).await?,
-		img if img.starts_with("image/") => self.download_image(url).await?,
-		_ => return Err!(Request(Unknown("Unsupported Content-Type"))),
-	};
-
-	self.set_url_preview(url, &data).await?;
-
-	Ok(data)
-}
-
-#[implement(Service)]
-async fn download_html(&self, url: &str) -> Result<UrlPreviewData> {
-	let client = &self.services.client.url_preview;
-	let mut response = client.get(url).send().await?;
-
-	let mut bytes: Vec<u8> = Vec::new();
-	while let Some(chunk) = response.chunk().await? {
-		bytes.extend_from_slice(&chunk);
-		if bytes.len() > self.services.globals.url_preview_max_spider_size() {
-			debug!(
-				"Response body from URL {} exceeds url_preview_max_spider_size ({}), not processing the rest of the \
-				 response body and assuming our necessary data is in this range.",
-				url,
-				self.services.globals.url_preview_max_spider_size()
-			);
-			break;
-		}
-	}
-	let body = String::from_utf8_lossy(&bytes);
-	let Ok(html) = HTML::from_string(body.to_string(), Some(url.to_owned())) else {
-		return Err!(Request(Unknown("Failed to parse HTML")));
-	};
-
-	let mut data = match html.opengraph.images.first() {
-		None => UrlPreviewData::default(),
-		Some(obj) => self.download_image(&obj.url).await?,
-	};
-
-	let props = html.opengraph.properties;
-
-	/* use OpenGraph title/description, but fall back to HTML if not available */
-	data.title = props.get("title").cloned().or(html.title);
-	data.description = props.get("description").cloned().or(html.description);
-
-	Ok(data)
-}
-
-#[implement(Service)]
-pub fn url_preview_allowed(&self, url_str: &str) -> bool {
-	let url: Url = match Url::parse(url_str) {
-		Ok(u) => u,
-		Err(e) => {
-			warn!("Failed to parse URL from a str: {}", e);
-			return false;
-		},
-	};
-
-	if ["http", "https"]
-		.iter()
-		.all(|&scheme| scheme != url.scheme().to_lowercase())
-	{
-		debug!("Ignoring non-HTTP/HTTPS URL to preview: {}", url);
-		return false;
-	}
-
-	let host = match url.host_str() {
-		None => {
-			debug!("Ignoring URL preview for a URL that does not have a host (?): {}", url);
-			return false;
-		},
-		Some(h) => h.to_owned(),
-	};
-
-	let allowlist_domain_contains = self
-		.services
-		.globals
-		.url_preview_domain_contains_allowlist();
-	let allowlist_domain_explicit = self
-		.services
-		.globals
-		.url_preview_domain_explicit_allowlist();
-	let denylist_domain_explicit = self.services.globals.url_preview_domain_explicit_denylist();
-	let allowlist_url_contains = self.services.globals.url_preview_url_contains_allowlist();
-
-	if allowlist_domain_contains.contains(&"*".to_owned())
-		|| allowlist_domain_explicit.contains(&"*".to_owned())
-		|| allowlist_url_contains.contains(&"*".to_owned())
-	{
-		debug!("Config key contains * which is allowing all URL previews. Allowing URL {}", url);
-		return true;
-	}
-
-	if !host.is_empty() {
-		if denylist_domain_explicit.contains(&host) {
-			debug!(
-				"Host {} is not allowed by url_preview_domain_explicit_denylist (check 1/4)",
-				&host
-			);
-			return false;
-		}
-
-		if allowlist_domain_explicit.contains(&host) {
-			debug!("Host {} is allowed by url_preview_domain_explicit_allowlist (check 2/4)", &host);
-			return true;
-		}
-
-		if allowlist_domain_contains
-			.iter()
-			.any(|domain_s| domain_s.contains(&host.clone()))
-		{
-			debug!("Host {} is allowed by url_preview_domain_contains_allowlist (check 3/4)", &host);
-			return true;
-		}
-
-		if allowlist_url_contains
-			.iter()
-			.any(|url_s| url.to_string().contains(&url_s.to_string()))
-		{
-			debug!("URL {} is allowed by url_preview_url_contains_allowlist (check 4/4)", &host);
-			return true;
-		}
-
-		// check root domain if available and if user has root domain checks
-		if self.services.globals.url_preview_check_root_domain() {
-			debug!("Checking root domain");
-			match host.split_once('.') {
-				None => return false,
-				Some((_, root_domain)) => {
-					if denylist_domain_explicit.contains(&root_domain.to_owned()) {
-						debug!(
-							"Root domain {} is not allowed by url_preview_domain_explicit_denylist (check 1/3)",
-							&root_domain
-						);
-						return true;
-					}
-
-					if allowlist_domain_explicit.contains(&root_domain.to_owned()) {
-						debug!(
-							"Root domain {} is allowed by url_preview_domain_explicit_allowlist (check 2/3)",
-							&root_domain
-						);
-						return true;
-					}
-
-					if allowlist_domain_contains
-						.iter()
-						.any(|domain_s| domain_s.contains(&root_domain.to_owned()))
-					{
-						debug!(
-							"Root domain {} is allowed by url_preview_domain_contains_allowlist (check 3/3)",
-							&root_domain
-						);
-						return true;
-					}
-				},
-			}
-		}
-	}
-
-	false
-}

src/service/mod.rs

@@ -18,7 +18,6 @@ pub mod pusher;
 pub mod resolver;
 pub mod rooms;
 pub mod sending;
-pub mod server_keys;
 pub mod transaction_ids;
 pub mod uiaa;
 pub mod updates;

src/service/pusher/mod.rs

@@ -3,7 +3,7 @@ mod data;
 use std::{fmt::Debug, mem, sync::Arc};
 
 use bytes::BytesMut;
-use conduit::{debug_error, err, trace, utils::string_from_bytes, warn, Err, PduEvent, Result};
+use conduit::{debug_info, info, trace, utils::string_from_bytes, warn, Error, PduEvent, Result};
 use ipaddress::IPAddress;
 use ruma::{
 	api::{
@@ -84,9 +84,8 @@ impl Service {
 		let http_request = request
 			.try_into_http_request::<BytesMut>(&dest, SendAccessToken::IfRequired(""), &VERSIONS)
 			.map_err(|e| {
-				err!(BadServerResponse(warn!(
+				warn!("Failed to find destination {dest} for push gateway: {e}");
-					"Failed to find destination {dest} for push gateway: {e}"
+				Error::BadServerResponse("Invalid push gateway destination")
-				)))
 			})?
 			.map(BytesMut::freeze);
 
@@ -96,7 +95,7 @@ impl Service {
 			trace!("Checking request URL for IP");
 			if let Ok(ip) = IPAddress::parse(url_host) {
 				if !self.services.globals.valid_cidr_range(&ip) {
-					return Err!(BadServerResponse("Not allowed to send requests to this IP"));
+					return Err(Error::BadServerResponse("Not allowed to send requests to this IP"));
 				}
 			}
 		}
@@ -111,7 +110,7 @@ impl Service {
 				if let Some(remote_addr) = response.remote_addr() {
 					if let Ok(ip) = IPAddress::parse(remote_addr.ip().to_string()) {
 						if !self.services.globals.valid_cidr_range(&ip) {
-							return Err!(BadServerResponse("Not allowed to send requests to this IP"));
+							return Err(Error::BadServerResponse("Not allowed to send requests to this IP"));
 						}
 					}
 				}
@@ -130,10 +129,10 @@ impl Service {
 				let body = response.bytes().await?; // TODO: handle timeout
 
 				if !status.is_success() {
-					debug_error!("Push gateway response body: {:?}", string_from_bytes(&body));
+					info!("Push gateway {dest} returned unsuccessful HTTP response ({status})");
-					return Err!(BadServerResponse(error!(
+					debug_info!("Push gateway response body: {:?}", string_from_bytes(&body));
-						"Push gateway {dest} returned unsuccessful HTTP response: {status}"
+
-					)));
+					return Err(Error::BadServerResponse("Push gateway returned unsuccessful response"));
 				}
 
 				let response = T::IncomingResponse::try_from_http_response(
@@ -141,11 +140,13 @@ impl Service {
 						.body(body)
 						.expect("reqwest body is valid http body"),
 				);
-				response
+				response.map_err(|e| {
-					.map_err(|e| err!(BadServerResponse(error!("Push gateway {dest} returned invalid response: {e}"))))
+					warn!("Push gateway {dest} returned invalid response bytes: {e}");
+					Error::BadServerResponse("Push gateway returned bad/invalid response")
+				})
 			},
 			Err(e) => {
-				debug_error!("Could not send request to pusher {dest}: {e}");
+				warn!("Could not send request to pusher {dest}: {e}");
 				Err(e.into())
 			},
 		}
@@ -164,7 +165,7 @@ impl Service {
 			.room_state_get(&pdu.room_id, &StateEventType::RoomPowerLevels, "")?
 			.map(|ev| {
 				serde_json::from_str(ev.content.get())
-					.map_err(|e| err!(Database("invalid m.room.power_levels event: {e:?}")))
+					.map_err(|_| Error::bad_database("invalid m.room.power_levels event"))
 			})
 			.transpose()?
 			.unwrap_or_default();
@@ -180,8 +181,8 @@ impl Service {
 			};
 
 			if notify.is_some() {
-				return Err!(Database(
+				return Err(Error::bad_database(
-					r#"Malformed pushrule contains more than one of these actions: ["dont_notify", "notify", "coalesce"]"#
+					r#"Malformed pushrule contains more than one of these actions: ["dont_notify", "notify", "coalesce"]"#,
 				));
 			}
 

src/service/resolver/actual.rs

@@ -4,7 +4,7 @@ use std::{
 	sync::Arc,
 };
 
-use conduit::{debug, debug_error, debug_info, debug_warn, err, trace, Err, Result};
+use conduit::{debug, debug_error, debug_info, debug_warn, trace, Err, Error, Result};
 use hickory_resolver::{error::ResolveError, lookup::SrvLookup};
 use ipaddress::IPAddress;
 use ruma::ServerName;
@@ -329,8 +329,10 @@ impl super::Service {
 			dest.is_ip_literal() || !IPAddress::is_valid(dest.host()),
 			"Destination is not an IP literal."
 		);
-		let ip = IPAddress::parse(dest.host())
+		let ip = IPAddress::parse(dest.host()).map_err(|e| {
-			.map_err(|e| err!(BadServerResponse(debug_error!("Failed to parse IP literal from string: {e}"))))?;
+			debug_error!("Failed to parse IP literal from string: {}", e);
+			Error::BadServerResponse("Invalid IP address")
+		})?;
 
 		self.validate_ip(&ip)?;
 
@@ -339,7 +341,7 @@ impl super::Service {
 
 	pub(crate) fn validate_ip(&self, ip: &IPAddress) -> Result<()> {
 		if !self.services.globals.valid_cidr_range(ip) {
-			return Err!(BadServerResponse("Not allowed to send requests to this IP"));
+			return Err(Error::BadServerResponse("Not allowed to send requests to this IP"));
 		}
 
 		Ok(())

src/service/rooms/event_handler/mod.rs

@@ -1,4 +1,5 @@
 mod parse_incoming_pdu;
+mod signing_keys;
 
 use std::{
 	collections::{hash_map, BTreeMap, HashMap, HashSet},
@@ -34,7 +35,7 @@ use ruma::{
 use tokio::sync::RwLock;
 
 use super::state_compressor::CompressedStateEvent;
-use crate::{globals, rooms, sending, server_keys, Dep};
+use crate::{globals, rooms, sending, Dep};
 
 pub struct Service {
 	services: Services,
@@ -49,7 +50,6 @@ struct Services {
 	metadata: Dep<rooms::metadata::Service>,
 	outlier: Dep<rooms::outlier::Service>,
 	pdu_metadata: Dep<rooms::pdu_metadata::Service>,
-	server_keys: Dep<server_keys::Service>,
 	short: Dep<rooms::short::Service>,
 	state: Dep<rooms::state::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
@@ -77,7 +77,6 @@ impl crate::Service for Service {
 				auth_chain: args.depend::<rooms::auth_chain::Service>("rooms::auth_chain"),
 				metadata: args.depend::<rooms::metadata::Service>("rooms::metadata"),
 				outlier: args.depend::<rooms::outlier::Service>("rooms::outlier"),
-				server_keys: args.depend::<server_keys::Service>("server_keys"),
 				pdu_metadata: args.depend::<rooms::pdu_metadata::Service>("rooms::pdu_metadata"),
 				short: args.depend::<rooms::short::Service>("rooms::short"),
 				state: args.depend::<rooms::state::Service>("rooms::state"),
@@ -1188,19 +1187,17 @@ impl Service {
 
 			// We go through all the signatures we see on the PDUs and their unresolved
 			// dependencies and fetch the corresponding signing keys
-			self.services
+			self.fetch_required_signing_keys(
-				.server_keys
+				events_with_auth_events
-				.fetch_required_signing_keys(
+					.iter()
-					events_with_auth_events
+					.flat_map(|(_id, _local_pdu, events)| events)
-						.iter()
+					.map(|(_event_id, event)| event),
-						.flat_map(|(_id, _local_pdu, events)| events)
+				pub_key_map,
-						.map(|(_event_id, event)| event),
+			)
-					pub_key_map,
+			.await
-				)
+			.unwrap_or_else(|e| {
-				.await
+				warn!("Could not fetch all signatures for PDUs from {}: {:?}", origin, e);
-				.unwrap_or_else(|e| {
+			});
-					warn!("Could not fetch all signatures for PDUs from {}: {:?}", origin, e);
-				});
 
 			let mut pdus = Vec::with_capacity(events_with_auth_events.len());
 			for (id, local_pdu, events_in_reverse_order) in events_with_auth_events {

src/service/rooms/event_handler/parse_incoming_pdu.rs

@@ -1,18 +1,18 @@
-use conduit::{debug_warn, err, pdu::gen_event_id_canonical_json, Err, Result};
+use conduit::{pdu::gen_event_id_canonical_json, warn, Err, Error, Result};
-use ruma::{CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId};
+use ruma::{api::client::error::ErrorKind, CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId};
 use serde_json::value::RawValue as RawJsonValue;
 
 impl super::Service {
 	pub fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result<(OwnedEventId, CanonicalJsonObject, OwnedRoomId)> {
 		let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
-			debug_warn!("Error parsing incoming event {pdu:#?}");
+			warn!("Error parsing incoming event {pdu:?}: {e:?}");
-			err!(BadServerResponse("Error parsing incoming event {e:?}"))
+			Error::BadServerResponse("Invalid PDU in server response")
 		})?;
 
 		let room_id: OwnedRoomId = value
 			.get("room_id")
 			.and_then(|id| RoomId::parse(id.as_str()?).ok())
-			.ok_or(err!(Request(InvalidParam("Invalid room id in pdu"))))?;
+			.ok_or(Error::BadRequest(ErrorKind::InvalidParam, "Invalid room id in pdu"))?;
 
 		let Ok(room_version_id) = self.services.state.get_room_version(&room_id) else {
 			return Err!("Server is not in room {room_id}");

src/service/rooms/event_handler/signing_keys.rs

@@ -1,10 +1,9 @@
 use std::{
 	collections::{BTreeMap, HashMap, HashSet},
-	sync::Arc,
 	time::{Duration, SystemTime},
 };
 
-use conduit::{debug, debug_error, debug_warn, err, error, info, trace, warn, Err, Result};
+use conduit::{debug, error, info, trace, warn, Error, Result};
 use futures_util::{stream::FuturesUnordered, StreamExt};
 use ruma::{
 	api::federation::{
@@ -22,31 +21,7 @@ use ruma::{
 use serde_json::value::RawValue as RawJsonValue;
 use tokio::sync::{RwLock, RwLockWriteGuard};
 
-use crate::{globals, sending, Dep};
+impl super::Service {
-
-pub struct Service {
-	services: Services,
-}
-
-struct Services {
-	globals: Dep<globals::Service>,
-	sending: Dep<sending::Service>,
-}
-
-impl crate::Service for Service {
-	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
-		Ok(Arc::new(Self {
-			services: Services {
-				globals: args.depend::<globals::Service>("globals"),
-				sending: args.depend::<sending::Service>("sending"),
-			},
-		}))
-	}
-
-	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
-}
-
-impl Service {
 	pub async fn fetch_required_signing_keys<'a, E>(
 		&'a self, events: E, pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
 	) -> Result<()>
@@ -57,13 +32,13 @@ impl Service {
 		for event in events {
 			for (signature_server, signature) in event
 				.get("signatures")
-				.ok_or(err!(BadServerResponse("No signatures in server response pdu.")))?
+				.ok_or(Error::BadServerResponse("No signatures in server response pdu."))?
 				.as_object()
-				.ok_or(err!(BadServerResponse("Invalid signatures object in server response pdu.")))?
+				.ok_or(Error::BadServerResponse("Invalid signatures object in server response pdu."))?
 			{
-				let signature_object = signature.as_object().ok_or(err!(BadServerResponse(
+				let signature_object = signature.as_object().ok_or(Error::BadServerResponse(
 					"Invalid signatures content object in server response pdu.",
-				)))?;
+				))?;
 
 				for signature_id in signature_object.keys() {
 					server_key_ids
@@ -94,12 +69,10 @@ impl Service {
 			.map(|(signature_server, signature_ids)| async {
 				let fetch_res = self
 					.fetch_signing_keys_for_server(
-						signature_server.as_str().try_into().map_err(|e| {
+						signature_server.as_str().try_into().map_err(|_| {
 							(
 								signature_server.clone(),
-								err!(BadServerResponse(
+								Error::BadServerResponse("Invalid servername in signatures of server response pdu."),
-									"Invalid servername in signatures of server response pdu: {e:?}"
-								)),
 							)
 						})?,
 						signature_ids.into_iter().collect(), // HashSet to Vec
@@ -109,9 +82,7 @@ impl Service {
 				match fetch_res {
 					Ok(keys) => Ok((signature_server, keys)),
 					Err(e) => {
-						debug_error!(
+						warn!("Signature verification failed: Could not fetch signing key for {signature_server}: {e}",);
-							"Signature verification failed: Could not fetch signing key for {signature_server}: {e}",
-						);
 						Err((signature_server, e))
 					},
 				}
@@ -127,7 +98,7 @@ impl Service {
 						.insert(signature_server.clone(), keys);
 				},
 				Err((signature_server, e)) => {
-					debug_warn!("Failed to fetch keys for {signature_server}: {e:?}");
+					warn!("Failed to fetch keys for {}: {:?}", signature_server, e);
 				},
 			}
 		}
@@ -145,37 +116,35 @@ impl Service {
 		pub_key_map: &mut RwLockWriteGuard<'_, BTreeMap<String, BTreeMap<String, Base64>>>,
 	) -> Result<()> {
 		let value: CanonicalJsonObject = serde_json::from_str(pdu.get()).map_err(|e| {
-			debug_error!("Invalid PDU in server response: {pdu:#?}");
+			error!("Invalid PDU in server response: {:?}: {:?}", pdu, e);
-			err!(BadServerResponse(error!("Invalid PDU in server response: {e:?}")))
+			Error::BadServerResponse("Invalid PDU in server response")
 		})?;
 
 		let signatures = value
 			.get("signatures")
-			.ok_or(err!(BadServerResponse("No signatures in server response pdu.")))?
+			.ok_or(Error::BadServerResponse("No signatures in server response pdu."))?
 			.as_object()
-			.ok_or(err!(BadServerResponse("Invalid signatures object in server response pdu.")))?;
+			.ok_or(Error::BadServerResponse("Invalid signatures object in server response pdu."))?;
 
 		for (signature_server, signature) in signatures {
-			let signature_object = signature.as_object().ok_or(err!(BadServerResponse(
+			let signature_object = signature.as_object().ok_or(Error::BadServerResponse(
 				"Invalid signatures content object in server response pdu.",
-			)))?;
+			))?;
 
 			let signature_ids = signature_object.keys().cloned().collect::<Vec<_>>();
 
 			let contains_all_ids =
 				|keys: &BTreeMap<String, Base64>| signature_ids.iter().all(|id| keys.contains_key(id));
 
-			let origin = <&ServerName>::try_from(signature_server.as_str()).map_err(|e| {
+			let origin = <&ServerName>::try_from(signature_server.as_str())
-				err!(BadServerResponse(
+				.map_err(|_| Error::BadServerResponse("Invalid servername in signatures of server response pdu."))?;
-					"Invalid servername in signatures of server response pdu: {e:?}"
-				))
-			})?;
 
 			if servers.contains_key(origin) || pub_key_map.contains_key(origin.as_str()) {
 				continue;
 			}
 
-			debug!("Loading signing keys for {origin}");
+			debug!("Loading signing keys for {}", origin);
+
 			let result: BTreeMap<_, _> = self
 				.services
 				.globals
@@ -185,7 +154,7 @@ impl Service {
 				.collect();
 
 			if !contains_all_ids(&result) {
-				debug_warn!("Signing key not loaded for {origin}");
+				debug!("Signing key not loaded for {}", origin);
 				servers.insert(origin.to_owned(), BTreeMap::new());
 			}
 
@@ -202,7 +171,7 @@ impl Service {
 		pub_key_map: &RwLock<BTreeMap<String, BTreeMap<String, Base64>>>,
 	) -> Result<()> {
 		for server in self.services.globals.trusted_servers() {
-			debug!("Asking batch signing keys from trusted server {server}");
+			debug!("Asking batch signing keys from trusted server {}", server);
 			match self
 				.services
 				.sending
@@ -215,16 +184,14 @@ impl Service {
 				.await
 			{
 				Ok(keys) => {
-					debug!("Got signing keys: {keys:?}");
+					debug!("Got signing keys: {:?}", keys);
 					let mut pkm = pub_key_map.write().await;
 					for k in keys.server_keys {
 						let k = match k.deserialize() {
 							Ok(key) => key,
 							Err(e) => {
-								warn!(
+								warn!("Received error {e} while fetching keys from trusted server {server}");
-									"Received error {e} while fetching keys from trusted server {server}: {:#?}",
+								warn!("{}", k.into_json());
-									k.into_json()
-								);
 								continue;
 							},
 						};
@@ -244,10 +211,13 @@ impl Service {
 						pkm.insert(k.server_name.to_string(), result);
 					}
 				},
-				Err(e) => error!(
+				Err(e) => {
-					"Failed sending batched key request to trusted key server {server} for the remote servers \
+					warn!(
-					 {servers:?}: {e}"
+						"Failed sending batched key request to trusted key server {server} for the remote servers \
-				),
+						 {:?}: {e}",
+						servers
+					);
+				},
 			}
 		}
 
@@ -483,6 +453,7 @@ impl Service {
 			}
 		} else {
 			info!("query_trusted_key_servers_first is set to false, querying {origin} first");
+
 			debug!("Asking {origin} for their signing keys over federation");
 			if let Some(server_key) = self
 				.services
@@ -540,7 +511,7 @@ impl Service {
 							.filter_map(|e| e.deserialize().ok())
 							.collect::<Vec<_>>()
 					}) {
-					debug!("Got signing keys: {server_keys:?}");
+					debug!("Got signing keys: {:?}", server_keys);
 					for k in server_keys {
 						self.services
 							.globals
@@ -565,6 +536,7 @@ impl Service {
 			}
 		}
 
-		Err!(BadServerResponse(warn!("Failed to find public key for server {origin:?}")))
+		warn!("Failed to find public key for server: {origin}");
+		Err(Error::BadServerResponse("Failed to find public key for server"))
 	}
 }

src/service/rooms/state_accessor/mod.rs

@@ -350,7 +350,6 @@ impl Service {
 			unsigned: None,
 			state_key: Some(target_user.into()),
 			redacts: None,
-			timestamp: None,
 		};
 
 		Ok(self

src/service/rooms/timeline/mod.rs

@@ -41,7 +41,7 @@ use tokio::sync::RwLock;
 use self::data::Data;
 use crate::{
 	account_data, admin, appservice, appservice::NamespaceRegex, globals, pusher, rooms,
-	rooms::state_compressor::CompressedStateEvent, sending, server_keys, Dep,
+	rooms::state_compressor::CompressedStateEvent, sending, Dep,
 };
 
 // Update Relationships
@@ -86,7 +86,6 @@ struct Services {
 	pdu_metadata: Dep<rooms::pdu_metadata::Service>,
 	read_receipt: Dep<rooms::read_receipt::Service>,
 	sending: Dep<sending::Service>,
-	server_keys: Dep<server_keys::Service>,
 	user: Dep<rooms::user::Service>,
 	pusher: Dep<pusher::Service>,
 	threads: Dep<rooms::threads::Service>,
@@ -115,7 +114,6 @@ impl crate::Service for Service {
 				pdu_metadata: args.depend::<rooms::pdu_metadata::Service>("rooms::pdu_metadata"),
 				read_receipt: args.depend::<rooms::read_receipt::Service>("rooms::read_receipt"),
 				sending: args.depend::<sending::Service>("sending"),
-				server_keys: args.depend::<server_keys::Service>("server_keys"),
 				user: args.depend::<rooms::user::Service>("rooms::user"),
 				pusher: args.depend::<pusher::Service>("pusher"),
 				threads: args.depend::<rooms::threads::Service>("rooms::threads"),
@@ -638,7 +636,6 @@ impl Service {
 			unsigned,
 			state_key,
 			redacts,
-			timestamp,
 		} = pdu_builder;
 
 		let prev_events: Vec<_> = self
@@ -706,14 +703,9 @@ impl Service {
 			room_id: room_id.to_owned(),
 			sender: sender.to_owned(),
 			origin: None,
-			origin_server_ts: timestamp.map_or_else(
+			origin_server_ts: utils::millis_since_unix_epoch()
-				|| {
+				.try_into()
-					utils::millis_since_unix_epoch()
+				.expect("time is valid"),
-						.try_into()
-						.expect("u64 fits into UInt")
-				},
-				|ts| ts.get(),
-			),
 			kind: event_type,
 			content,
 			state_key,
@@ -1189,7 +1181,7 @@ impl Service {
 		}
 
 		self.services
-			.server_keys
+			.event_handler
 			.fetch_required_signing_keys([&value], pub_key_map)
 			.await?;
 

src/service/sending/appservice.rs

@@ -1,7 +1,7 @@
 use std::{fmt::Debug, mem};
 
 use bytes::BytesMut;
-use conduit::{debug_error, err, trace, utils, warn, Err, Result};
+use conduit::{debug_error, trace, utils, warn, Error, Result};
 use reqwest::Client;
 use ruma::api::{appservice::Registration, IncomingResponse, MatrixVersion, OutgoingRequest, SendAccessToken};
 
@@ -26,7 +26,10 @@ where
 	let hs_token = registration.hs_token.as_str();
 	let mut http_request = request
 		.try_into_http_request::<BytesMut>(&dest, SendAccessToken::IfRequired(hs_token), &VERSIONS)
-		.map_err(|e| err!(BadServerResponse(warn!("Failed to find destination {dest}: {e}"))))?
+		.map_err(|e| {
+			warn!("Failed to find destination {dest}: {e}");
+			Error::BadServerResponse("Invalid appservice destination")
+		})?
 		.map(BytesMut::freeze);
 
 	let mut parts = http_request.uri().clone().into_parts();
@@ -66,11 +69,13 @@ where
 	let body = response.bytes().await?; // TODO: handle timeout
 
 	if !status.is_success() {
-		debug_error!("Appservice response bytes: {:?}", utils::string_from_bytes(&body));
+		warn!(
-		return Err!(BadServerResponse(error!(
 			"Appservice \"{}\" returned unsuccessful HTTP response {status} at {dest}",
 			registration.id
-		)));
+		);
+		debug_error!("Appservice response bytes: {:?}", utils::string_from_bytes(&body));
+
+		return Err(Error::BadServerResponse("Appservice returned unsuccessful HTTP response"));
 	}
 
 	let response = T::IncomingResponse::try_from_http_response(
@@ -80,9 +85,7 @@ where
 	);
 
 	response.map(Some).map_err(|e| {
-		err!(BadServerResponse(error!(
+		warn!("Appservice \"{}\" returned invalid response bytes {dest}: {e}", registration.id);
-			"Appservice \"{}\" returned invalid response bytes {dest}: {e}",
+		Error::BadServerResponse("Appservice returned bad/invalid response")
-			registration.id
-		)))
 	})
 }

src/service/sending/resolve.rs

@@ -0,0 +1,343 @@
+use std::{
+	fmt::Debug,
+	net::{IpAddr, SocketAddr},
+};
+
+use conduit::{debug, debug_error, debug_info, debug_warn, trace, Err, Error, Result};
+use hickory_resolver::{error::ResolveError, lookup::SrvLookup};
+use ipaddress::IPAddress;
+use ruma::ServerName;
+
+use crate::{
+	resolver::{add_port_to_hostname, get_ip_with_port, CachedDest, CachedOverride, FedDest},
+	services,
+};
+
+#[derive(Clone, Debug)]
+pub(crate) struct ActualDest {
+	pub(crate) dest: FedDest,
+	pub(crate) host: String,
+	pub(crate) string: String,
+	pub(crate) cached: bool,
+}
+
+#[tracing::instrument(skip_all, name = "resolve")]
+pub(crate) async fn get_actual_dest(server_name: &ServerName) -> Result<ActualDest> {
+	let cached;
+	let cached_result = services().resolver.get_cached_destination(server_name);
+
+	let CachedDest {
+		dest,
+		host,
+		..
+	} = if let Some(result) = cached_result {
+		cached = true;
+		result
+	} else {
+		cached = false;
+		validate_dest(server_name)?;
+		resolve_actual_dest(server_name, true).await?
+	};
+
+	let string = dest.clone().into_https_string();
+	Ok(ActualDest {
+		dest,
+		host,
+		string,
+		cached,
+	})
+}
+
+/// Returns: `actual_destination`, host header
+/// Implemented according to the specification at <https://matrix.org/docs/spec/server_server/r0.1.4#resolving-server-names>
+/// Numbers in comments below refer to bullet points in linked section of
+/// specification
+#[tracing::instrument(skip_all, name = "actual")]
+pub async fn resolve_actual_dest(dest: &ServerName, cache: bool) -> Result<CachedDest> {
+	trace!("Finding actual destination for {dest}");
+	let mut host = dest.as_str().to_owned();
+	let actual_dest = match get_ip_with_port(dest.as_str()) {
+		Some(host_port) => actual_dest_1(host_port)?,
+		None => {
+			if let Some(pos) = dest.as_str().find(':') {
+				actual_dest_2(dest, cache, pos).await?
+			} else if let Some(delegated) = request_well_known(dest.as_str()).await? {
+				actual_dest_3(&mut host, cache, delegated).await?
+			} else if let Some(overrider) = query_srv_record(dest.as_str()).await? {
+				actual_dest_4(&host, cache, overrider).await?
+			} else {
+				actual_dest_5(dest, cache).await?
+			}
+		},
+	};
+
+	// Can't use get_ip_with_port here because we don't want to add a port
+	// to an IP address if it wasn't specified
+	let host = if let Ok(addr) = host.parse::<SocketAddr>() {
+		FedDest::Literal(addr)
+	} else if let Ok(addr) = host.parse::<IpAddr>() {
+		FedDest::Named(addr.to_string(), ":8448".to_owned())
+	} else if let Some(pos) = host.find(':') {
+		let (host, port) = host.split_at(pos);
+		FedDest::Named(host.to_owned(), port.to_owned())
+	} else {
+		FedDest::Named(host, ":8448".to_owned())
+	};
+
+	debug!("Actual destination: {actual_dest:?} hostname: {host:?}");
+	Ok(CachedDest {
+		dest: actual_dest,
+		host: host.into_uri_string(),
+		expire: CachedDest::default_expire(),
+	})
+}
+
+fn actual_dest_1(host_port: FedDest) -> Result<FedDest> {
+	debug!("1: IP literal with provided or default port");
+	Ok(host_port)
+}
+
+async fn actual_dest_2(dest: &ServerName, cache: bool, pos: usize) -> Result<FedDest> {
+	debug!("2: Hostname with included port");
+	let (host, port) = dest.as_str().split_at(pos);
+	conditional_query_and_cache_override(host, host, port.parse::<u16>().unwrap_or(8448), cache).await?;
+	Ok(FedDest::Named(host.to_owned(), port.to_owned()))
+}
+
+async fn actual_dest_3(host: &mut String, cache: bool, delegated: String) -> Result<FedDest> {
+	debug!("3: A .well-known file is available");
+	*host = add_port_to_hostname(&delegated).into_uri_string();
+	match get_ip_with_port(&delegated) {
+		Some(host_and_port) => actual_dest_3_1(host_and_port),
+		None => {
+			if let Some(pos) = delegated.find(':') {
+				actual_dest_3_2(cache, delegated, pos).await
+			} else {
+				trace!("Delegated hostname has no port in this branch");
+				if let Some(overrider) = query_srv_record(&delegated).await? {
+					actual_dest_3_3(cache, delegated, overrider).await
+				} else {
+					actual_dest_3_4(cache, delegated).await
+				}
+			}
+		},
+	}
+}
+
+fn actual_dest_3_1(host_and_port: FedDest) -> Result<FedDest> {
+	debug!("3.1: IP literal in .well-known file");
+	Ok(host_and_port)
+}
+
+async fn actual_dest_3_2(cache: bool, delegated: String, pos: usize) -> Result<FedDest> {
+	debug!("3.2: Hostname with port in .well-known file");
+	let (host, port) = delegated.split_at(pos);
+	conditional_query_and_cache_override(host, host, port.parse::<u16>().unwrap_or(8448), cache).await?;
+	Ok(FedDest::Named(host.to_owned(), port.to_owned()))
+}
+
+async fn actual_dest_3_3(cache: bool, delegated: String, overrider: FedDest) -> Result<FedDest> {
+	debug!("3.3: SRV lookup successful");
+	let force_port = overrider.port();
+	conditional_query_and_cache_override(&delegated, &overrider.hostname(), force_port.unwrap_or(8448), cache).await?;
+	if let Some(port) = force_port {
+		Ok(FedDest::Named(delegated, format!(":{port}")))
+	} else {
+		Ok(add_port_to_hostname(&delegated))
+	}
+}
+
+async fn actual_dest_3_4(cache: bool, delegated: String) -> Result<FedDest> {
+	debug!("3.4: No SRV records, just use the hostname from .well-known");
+	conditional_query_and_cache_override(&delegated, &delegated, 8448, cache).await?;
+	Ok(add_port_to_hostname(&delegated))
+}
+
+async fn actual_dest_4(host: &str, cache: bool, overrider: FedDest) -> Result<FedDest> {
+	debug!("4: No .well-known; SRV record found");
+	let force_port = overrider.port();
+	conditional_query_and_cache_override(host, &overrider.hostname(), force_port.unwrap_or(8448), cache).await?;
+	if let Some(port) = force_port {
+		Ok(FedDest::Named(host.to_owned(), format!(":{port}")))
+	} else {
+		Ok(add_port_to_hostname(host))
+	}
+}
+
+async fn actual_dest_5(dest: &ServerName, cache: bool) -> Result<FedDest> {
+	debug!("5: No SRV record found");
+	conditional_query_and_cache_override(dest.as_str(), dest.as_str(), 8448, cache).await?;
+	Ok(add_port_to_hostname(dest.as_str()))
+}
+
+#[tracing::instrument(skip_all, name = "well-known")]
+async fn request_well_known(dest: &str) -> Result<Option<String>> {
+	trace!("Requesting well known for {dest}");
+	if !services().resolver.has_cached_override(dest) {
+		query_and_cache_override(dest, dest, 8448).await?;
+	}
+
+	let response = services()
+		.client
+		.well_known
+		.get(&format!("https://{dest}/.well-known/matrix/server"))
+		.send()
+		.await;
+
+	trace!("response: {:?}", response);
+	if let Err(e) = &response {
+		debug!("error: {e:?}");
+		return Ok(None);
+	}
+
+	let response = response?;
+	if !response.status().is_success() {
+		debug!("response not 2XX");
+		return Ok(None);
+	}
+
+	let text = response.text().await?;
+	trace!("response text: {:?}", text);
+	if text.len() >= 12288 {
+		debug_warn!("response contains junk");
+		return Ok(None);
+	}
+
+	let body: serde_json::Value = serde_json::from_str(&text).unwrap_or_default();
+
+	let m_server = body
+		.get("m.server")
+		.unwrap_or(&serde_json::Value::Null)
+		.as_str()
+		.unwrap_or_default();
+
+	if ruma::identifiers_validation::server_name::validate(m_server).is_err() {
+		debug_error!("response content missing or invalid");
+		return Ok(None);
+	}
+
+	debug_info!("{:?} found at {:?}", dest, m_server);
+	Ok(Some(m_server.to_owned()))
+}
+
+#[inline]
+async fn conditional_query_and_cache_override(overname: &str, hostname: &str, port: u16, cache: bool) -> Result<()> {
+	if cache {
+		query_and_cache_override(overname, hostname, port).await
+	} else {
+		Ok(())
+	}
+}
+
+#[tracing::instrument(skip_all, name = "ip")]
+async fn query_and_cache_override(overname: &'_ str, hostname: &'_ str, port: u16) -> Result<()> {
+	match services()
+		.resolver
+		.resolver
+		.lookup_ip(hostname.to_owned())
+		.await
+	{
+		Err(e) => handle_resolve_error(&e),
+		Ok(override_ip) => {
+			if hostname != overname {
+				debug_info!("{overname:?} overriden by {hostname:?}");
+			}
+
+			services().resolver.set_cached_override(
+				overname.to_owned(),
+				CachedOverride {
+					ips: override_ip.iter().collect(),
+					port,
+					expire: CachedOverride::default_expire(),
+				},
+			);
+
+			Ok(())
+		},
+	}
+}
+
+#[tracing::instrument(skip_all, name = "srv")]
+async fn query_srv_record(hostname: &'_ str) -> Result<Option<FedDest>> {
+	fn handle_successful_srv(srv: &SrvLookup) -> Option<FedDest> {
+		srv.iter().next().map(|result| {
+			FedDest::Named(
+				result.target().to_string().trim_end_matches('.').to_owned(),
+				format!(":{}", result.port()),
+			)
+		})
+	}
+
+	async fn lookup_srv(hostname: &str) -> Result<SrvLookup, ResolveError> {
+		debug!("querying SRV for {:?}", hostname);
+		let hostname = hostname.trim_end_matches('.');
+		services()
+			.resolver
+			.resolver
+			.srv_lookup(hostname.to_owned())
+			.await
+	}
+
+	let hostnames = [format!("_matrix-fed._tcp.{hostname}."), format!("_matrix._tcp.{hostname}.")];
+
+	for hostname in hostnames {
+		match lookup_srv(&hostname).await {
+			Ok(result) => return Ok(handle_successful_srv(&result)),
+			Err(e) => handle_resolve_error(&e)?,
+		}
+	}
+
+	Ok(None)
+}
+
+#[allow(clippy::single_match_else)]
+fn handle_resolve_error(e: &ResolveError) -> Result<()> {
+	use hickory_resolver::error::ResolveErrorKind;
+
+	match *e.kind() {
+		ResolveErrorKind::NoRecordsFound {
+			..
+		} => {
+			// Raise to debug_warn if we can find out the result wasn't from cache
+			debug!("{e}");
+			Ok(())
+		},
+		_ => Err!(error!("DNS {e}")),
+	}
+}
+
+fn validate_dest(dest: &ServerName) -> Result<()> {
+	if dest == services().globals.server_name() {
+		return Err!("Won't send federation request to ourselves");
+	}
+
+	if dest.is_ip_literal() || IPAddress::is_valid(dest.host()) {
+		validate_dest_ip_literal(dest)?;
+	}
+
+	Ok(())
+}
+
+fn validate_dest_ip_literal(dest: &ServerName) -> Result<()> {
+	trace!("Destination is an IP literal, checking against IP range denylist.",);
+	debug_assert!(
+		dest.is_ip_literal() || !IPAddress::is_valid(dest.host()),
+		"Destination is not an IP literal."
+	);
+	let ip = IPAddress::parse(dest.host()).map_err(|e| {
+		debug_error!("Failed to parse IP literal from string: {}", e);
+		Error::BadServerResponse("Invalid IP address")
+	})?;
+
+	validate_ip(&ip)?;
+
+	Ok(())
+}
+
+pub(crate) fn validate_ip(ip: &IPAddress) -> Result<()> {
+	if !services().globals.valid_cidr_range(ip) {
+		return Err(Error::BadServerResponse("Not allowed to send requests to this IP"));
+	}
+
+	Ok(())
+}

src/service/sending/send.rs

@@ -1,7 +1,7 @@
 use std::{fmt::Debug, mem};
 
 use conduit::{
-	debug, debug_error, debug_info, debug_warn, err, error::inspect_debug_log, trace, utils::string::EMPTY, Err, Error,
+	debug, debug_error, debug_info, debug_warn, error::inspect_debug_log, trace, utils::string::EMPTY, Err, Error,
 	Result,
 };
 use http::{header::AUTHORIZATION, HeaderValue};
@@ -73,7 +73,7 @@ impl super::Service {
 		trace!("Preparing request");
 		let mut http_request = req
 			.try_into_http_request::<Vec<u8>>(&actual.string, SATIR, &VERSIONS)
-			.map_err(|e| err!(BadServerResponse("Invalid destination: {e:?}")))?;
+			.map_err(|_| Error::BadServerResponse("Invalid destination"))?;
 
 		sign_request::<T>(&self.services.globals, dest, &mut http_request);
 
@@ -150,7 +150,10 @@ where
 		);
 	}
 
-	response.map_err(|e| err!(BadServerResponse("Server returned bad 200 response: {e:?}")))
+	match response {
+		Err(_) => Err(Error::BadServerResponse("Server returned bad 200 response.")),
+		Ok(response) => Ok(response),
+	}
 }
 
 fn handle_error<T>(

src/service/services.rs

@@ -12,7 +12,7 @@ use tokio::sync::Mutex;
 use crate::{
 	account_data, admin, appservice, client, emergency, globals, key_backups,
 	manager::Manager,
-	media, presence, pusher, resolver, rooms, sending, server_keys, service,
+	media, presence, pusher, resolver, rooms, sending, service,
 	service::{Args, Map, Service},
 	transaction_ids, uiaa, updates, users,
 };
@@ -31,7 +31,6 @@ pub struct Services {
 	pub resolver: Arc<resolver::Service>,
 	pub rooms: rooms::Service,
 	pub sending: Arc<sending::Service>,
-	pub server_keys: Arc<server_keys::Service>,
 	pub transaction_ids: Arc<transaction_ids::Service>,
 	pub uiaa: Arc<uiaa::Service>,
 	pub updates: Arc<updates::Service>,
@@ -95,7 +94,6 @@ impl Services {
 				user: build!(rooms::user::Service),
 			},
 			sending: build!(sending::Service),
-			server_keys: build!(server_keys::Service),
 			transaction_ids: build!(transaction_ids::Service),
 			uiaa: build!(uiaa::Service),
 			updates: build!(updates::Service),