fix: Mobile styling improvements

Closes #1894

.forgejo/actions/rust-toolchain/action.yml

@@ -33,7 +33,7 @@ runs:
         echo "version=$(rustup --version)" >> $GITHUB_OUTPUT
     - name: Cache rustup toolchains
       if: steps.rustup-version.outputs.version == ''
-      uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+      uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
       with:
         path: |
           ~/.rustup

.forgejo/actions/setup-llvm-with-apt/action.yml

@@ -57,7 +57,7 @@ runs:
 
     - name: Check for LLVM cache
       id: cache
-      uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+      uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
       with:
         path: |
           /usr/bin/clang-*

.forgejo/actions/setup-rust/action.yml

@@ -65,7 +65,7 @@ runs:
 
     - name: Cache toolchain binaries
       id: toolchain-cache
-      uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+      uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
       with:
         path: |
           .cargo/bin
@@ -76,7 +76,7 @@ runs:
 
     - name: Cache Cargo registry and git
       id: registry-cache
-      uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+      uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
       with:
         path: |
           .cargo/registry/index

.forgejo/actions/timelord/action.yml

@@ -31,7 +31,7 @@ runs:
 
     - name: Restore binary cache
       id: binary-cache
-      uses: actions/cache/restore@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+      uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
       with:
         path: |
           /usr/share/rust/.cargo/bin
@@ -71,13 +71,13 @@ runs:
 
     - name: Install timelord-cli and git-warp-time
       if: steps.check-binaries.outputs.need-install == 'true'
-      uses: https://github.com/taiki-e/install-action@9e1e5806d4a4822de933115878265be9aaa786d9 # v2
+      uses: https://github.com/taiki-e/install-action@9bcaee1dcae34154180f412e2fa69355a7cda9f6 # v2
       with:
         tool: git-warp-time,timelord-cli@3.0.1
 
     - name: Save binary cache
       if: steps.check-binaries.outputs.need-install == 'true'
-      uses: actions/cache/save@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+      uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
       with:
         path: |
           /usr/share/rust/.cargo/bin
@@ -87,7 +87,7 @@ runs:
 
     - name: Restore timelord cache with fallbacks
       id: timelord-restore
-      uses: actions/cache/restore@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+      uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
       with:
         path: ${{ env.TIMELORD_CACHE_PATH }}
         key: ${{ env.TIMELORD_KEY }}
@@ -114,7 +114,7 @@ runs:
         timelord sync --source-dir ${{ env.TIMELORD_PATH }} --cache-dir ${{ env.TIMELORD_CACHE_PATH }}
 
     - name: Save updated timelord cache immediately
-      uses: actions/cache/save@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+      uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
       with:
         path: ${{ env.TIMELORD_CACHE_PATH }}
         key: ${{ env.TIMELORD_KEY }}

.forgejo/workflows/build-debian.yml

@@ -49,7 +49,7 @@ jobs:
           ref: ${{ github.ref_name }}
 
       - name: Cache Cargo registry
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: |
             ~/.cargo/registry

.forgejo/workflows/build-fedora.yml

@@ -37,7 +37,7 @@ jobs:
 
 
       - name: Cache DNF packages
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: |
             /var/cache/dnf
@@ -47,7 +47,7 @@ jobs:
             dnf-fedora${{ steps.fedora.outputs.version }}-
 
       - name: Cache Cargo registry
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: |
             ~/.cargo/registry
@@ -57,7 +57,7 @@ jobs:
             cargo-fedora${{ steps.fedora.outputs.version }}-
 
       - name: Cache Rust build dependencies
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: |
             ~/rpmbuild/BUILD/*/target/release/deps

.forgejo/workflows/documentation.yml

@@ -37,7 +37,7 @@ jobs:
           node-version: 22
 
       - name: Cache npm dependencies
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: ~/.npm
           key: continuwuity-rspress-${{ steps.runner-env.outputs.slug }}-${{ steps.runner-env.outputs.arch }}-node-${{ steps.runner-env.outputs.node_version }}-${{ hashFiles('package-lock.json') }}

.forgejo/workflows/mirror-images.yml

@@ -55,7 +55,7 @@ jobs:
       #     repositories: continuwuity
 
       - name: Install regsync
-        uses: https://github.com/regclient/actions/regsync-installer@4b4db1dcc7dad75ad67a788a380f75a20cc8a040 # main
+        uses: https://github.com/regclient/actions/regsync-installer@9a2d4216180dbb3e2dccfa60d2dd4afd98e42ec5 # main
 
       - name: Check what images need mirroring
         run: |

.forgejo/workflows/renovate.yml

@@ -55,7 +55,7 @@ jobs:
         run: /usr/local/renovate/node -e 'console.log(`node heap limit = ${require("v8").getHeapStatistics().heap_size_limit / (1024 * 1024)} Mb`)'
 
       - name: Restore renovate repo cache
-        uses: actions/cache/restore@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: |
             /tmp/renovate/cache/renovate/repository
@@ -64,7 +64,7 @@ jobs:
             renovate-repo-cache-
 
       - name: Restore renovate package cache
-        uses: actions/cache/restore@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: |
             /tmp/renovate/cache/renovate/renovate-cache-sqlite
@@ -73,7 +73,7 @@ jobs:
             renovate-package-cache-
 
       - name: Restore renovate OSV cache
-        uses: actions/cache/restore@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: |
             /tmp/osv
@@ -117,7 +117,7 @@ jobs:
 
       - name: Save renovate package cache
         if: always()
-        uses: actions/cache/save@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: |
             /tmp/renovate/cache/renovate/renovate-cache-sqlite
@@ -125,7 +125,7 @@ jobs:
 
       - name: Save renovate OSV cache
         if: always()
-        uses: actions/cache/save@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5
+        uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
         with:
           path: |
             /tmp/osv

Cargo.toml

@@ -45,7 +45,7 @@ version = "1.0.6"
 version = "1.0.0"
 
 [workspace.dependencies.cargo_toml]
-version = "0.22"
+version = "1.0"
 default-features = false
 features = ["features"]
 
@@ -356,6 +356,7 @@ features = [
     "ring-compat",
     "compat-upload-signatures",
     "compat-optional-txn-pdus",
+    "compat-get-3pids",
     "unstable-msc2666",
     "unstable-msc2867",
     "unstable-msc2870",
@@ -402,6 +403,9 @@ default-features = false
 version = "0.11.0"
 default-features = false
 
+[workspace.dependencies.openidconnect]
+version = "4.0.1"
+
 # optional opentelemetry, performance measurements, flamegraphs, etc for performance measurements and monitoring
 [workspace.dependencies.opentelemetry]
 version = "0.32.0"

changelog.d/+12935c96.feature.md

@@ -0,0 +1 @@
+Appservice device management as outlined in MSC4190 (part of Matrix 1.17) is now fully supported. Contributed by @ginger.

changelog.d/1818.feature.md

@@ -0,0 +1,2 @@
+Improved the performance and reliability of fetching missing events, improving network partition recovery. Contributed
+by @nex.

changelog.d/765.feature.md

@@ -0,0 +1 @@
+Added support for linking an external identity provider with OIDC. Contributed by @ginger.

conduwuit-example.toml

@@ -297,7 +297,7 @@
 
 # This item is undocumented. Please contribute documentation for it.
 #
-#max_fetch_prev_events = 192
+#max_fetch_prev_events = 1024
 
 # How many incoming federation transactions the server is willing to be
 # processing at any given time before it becomes overloaded and starts
@@ -643,6 +643,14 @@
 #
 #default_room_acl_deny =
 
+# The number of forward extremities to tolerate in a room before
+# attempting to manually squash them with a "dummy event". Setting this
+# above 20 will hinder its efficacy, and setting it below 5 will cause
+# more dummy events to be sent than necessary (which increases federation
+# traffic).
+#
+#dummy_event_threshold = 10
+
 # Enable OpenTelemetry OTLP tracing export. This replaces the deprecated
 # Jaeger exporter. Traces will be sent via OTLP to a collector (such as
 # Jaeger) that supports the OpenTelemetry Protocol.
@@ -1426,6 +1434,11 @@
 #
 #send_messages_from_ignored_users_to_client = false
 
+# Send "org.matrix.dummy_event" events to the client. This is a debugging
+# option.
+#
+#send_dummy_events_to_clients = false
+
 # Vector list of IPv4 and IPv6 CIDR ranges / subnets *in quotes* that you
 # do not want continuwuity to send outbound requests to. Defaults to
 # RFC1918, unroutable, loopback, multicast, and testnet addresses for
@@ -2015,3 +2028,96 @@
 #   legacy authentication will be unable to log in.
 #
 #compatibility_mode = "hybrid"
+
+#[global.oauth.oidc]
+# Uncommenting this section will enable Continuwuity's support for
+# authenticating users using an OpenID Connect-compatible identity provider.
+# This is referred to as "delegated authentication".
+#
+# IMPORTANT NOTE: When delegated authentication is active, Continuwuity will behave as if
+# the `global.oauth.compatibility_mode` setting is set to `exclusive`.
+# Matrix clients which do not support OAuth login (also referred to as "next-gen auth") will NOT be able
+# to log in while delegated authentication is active.
+
+
+# The OIDC issuer URL. Continuwuity will use OpenID Connect Discovery to
+# automatically fetch the identity provider's metadata from this URL.
+# Generally you should set this to the base domain your identity provider
+# runs on.
+#
+#discovery_url =
+
+# The OAuth client ID for Continuwuity to use when communicating with the
+# identity provider.
+#
+#client_id =
+
+# The OAuth client secret for Continuwuity to use when communicating with
+# the identity provider.
+#
+#client_secret =
+
+# A path to a file which Continuwuity will read the client secret from.
+# If this option is set, it will override `client_secret`.
+#
+# The server will fail to start if the file cannot be read.
+#
+#client_secret_file =
+
+# Additional scopes Continuwuity should request from the IDP. This may be
+# necessary to access certain claims. Continuwuity always requests the
+# `openid` scope.
+#
+#additional_scopes = []
+
+# Whether the user should be prompted to choose a localpart
+# when signing in for the first time. If this is `false`, Continuwuity
+# will attempt to use the value of the `preferred_username_claim`
+# (see below) as the user's localpart. Authentication will
+# fail if this claim is missing or is not a valid localpart.
+#
+#prompt_for_localpart = true
+
+# The claim to use for the user's localpart, if `prompt_for_localpart` is
+# false.
+#
+#preferred_username_claim = "preferred_username"
+
+# The claim which will be used to set the user's email address,
+# either on initial registration or on every login depending on
+# the value of `profile_key_import_mode`. Continuwuity assumes that
+# the IDP has taken care of verifying that the user controls the email
+# address it provides.
+#
+# This option does nothing if SMTP is not configured.
+#
+# If this option is set, and `profile_key_import_mode` is `on_login`,
+# users will not be able to change their email addresses themselves.
+#
+#email_claim = "email"
+
+# Defines how claims returned from the IDP should be mapped to a user's
+# profile data. The profile field named in each key will be set from the
+# claim named in the corresponding value when the user first registers,
+# and possibly on subsequent logins as well, depending on the value of
+# `profile_key_import_mode` (see below).
+#
+# Per-room overrides to the user's display name or avatar will be
+# preserved by the import process.
+#
+# SECURITY NOTE: If the `avatar_url` field is set, Continuwuity will
+# perform a HTTP GET to the URL in the mapped claim and use the returned
+# file as the user's profile picture. Make sure your users are not able
+# to set the value of the mapped claim to an arbitrary URL.
+#
+#profile_key_map = { displayname = "name" }
+
+# When profile keys should be imported from the IDP's claims.
+#
+# - "on_registration": Listed keys will be imported once, when the user
+#   logs in for the first time and their shadow account is created.
+# - "on_login": Listed keys will be imported every time the user logs in.
+#   Additionally, users will not be able to manually edit any listed keys
+#   through their Matrix client.
+#
+#profile_key_import_mode = "on_registration"

docs/reference/admin/debug.md

@@ -10,7 +10,13 @@ ## `!admin debug echo`
 
 ## `!admin debug get-auth-chain`
 
-Get the auth_chain of a PDU
+Loads the auth_chain of a PDU, reporting how long it took
+
+## `!admin debug show-auth-chain`
+
+Walks & displays the auth_chain of a PDU in a mermaid graph format.
+
+This is useless to basically anyone but developers, and is also probably slow and memory hungry.
 
 ## `!admin debug parse-pdu`
 
@@ -44,6 +50,12 @@ ## `!admin debug get-room-state`
 
 Of course the check is still done on the actual client API.
 
+## `!admin debug get-state-at`
+
+Gets all the room state events at the specified event.
+
+State at event might not be available for some PDUs, such as rejected ones.
+
 ## `!admin debug get-signing-keys`
 
 Get and display signing keys from local cache or remote server

docs/reference/admin/index.md

@@ -14,6 +14,7 @@ ## Categories
 - [`!admin appservices`](appservices/): Commands for managing appservices
 - [`!admin users`](users/): Commands for managing local users
 - [`!admin token`](token/): Commands for managing registration tokens
+- [`!admin oidc`](oidc/): Commands for managing OIDC
 - [`!admin rooms`](rooms/): Commands for managing rooms
 - [`!admin federation`](federation/): Commands for managing federation
 - [`!admin server`](server/): Commands for managing the server

docs/reference/admin/oidc.md

@@ -0,0 +1,13 @@
+<!-- This file is generated by `cargo xtask generate-docs`. Do not edit. -->
+# `!admin oidc`
+
+Commands for managing OIDC
+
+
+## `!admin oidc link`
+
+Link a user ID to the given subject claim
+
+## `!admin oidc unlink`
+
+Unlink the given subject claim from its associated user ID

docs/reference/admin/users.md

@@ -12,10 +12,6 @@ ## `!admin users reset-password`
 
 Reset user password
 
-## `!admin users issue-password-reset-link`
-
-Issue a self-service password reset link for a user
-
 ## `!admin users get-email`
 
 Get a user's associated email address
@@ -96,6 +92,14 @@ ## `!admin users list-users`
 
 List local users in the database
 
+## `!admin users list-invited-rooms`
+
+Lists all the rooms (local and remote) that the specified user is invited to
+
+## `!admin users reject-all-invites`
+
+Manually make a user reject all current invites
+
 ## `!admin users list-joined-rooms`
 
 Lists all the rooms (local and remote) that the specified user is joined in

nix/devshell.nix

@@ -9,7 +9,7 @@
     {
       # basic nix shell containing all things necessary to build continuwuity in all flavors manually (on x86_64-linux)
       devShells.default =
-        (inputs.crane.mkLib pkgs).overrideToolchain (pkgs: self'.packages.stable-toolchain).devShell
+        ((inputs.crane.mkLib pkgs).overrideToolchain (pkgs: self'.packages.stable-toolchain)).devShell
           {
             packages = [
               self'.packages.rocksdb

nix/rust.nix

@@ -12,7 +12,7 @@
         target:
         target.fromToolchainName {
           name = (lib.importTOML "${inputs.self}/rust-toolchain.toml").toolchain.channel;
-          sha256 = "sha256-mvUGEOHYJpn3ikC5hckneuGixaC+yGrkMM/liDIDgoU=";
+          sha256 = "sha256-h+t2xTBz5yt2YIO+1VMIIGlCU7gyp2LYOFvaV1nwOXU=";
         };
     in
     {

rust-toolchain.toml

@@ -10,7 +10,7 @@
 
 [toolchain]
 profile = "minimal"
-channel = "1.96.0"
+channel = "1.96.1"
 components = [
     # For rust-analyzer
     "rust-src",

src/admin/admin.rs

@@ -1,5 +1,5 @@
 use clap::Parser;
-use conduwuit::Result;
+use conduwuit::{Err, Result};
 
 use crate::{
 	appservice::{self, AppserviceCommand},
@@ -8,6 +8,7 @@
 	debug::{self, DebugCommand},
 	federation::{self, FederationCommand},
 	media::{self, MediaCommand},
+	oidc::{self, OidcCommand},
 	query::{self, QueryCommand},
 	room::{self, RoomCommand},
 	server::{self, ServerCommand},
@@ -18,44 +19,48 @@
 #[derive(Debug, Parser)]
 #[command(name = conduwuit_core::BRANDING, version = conduwuit_core::version())]
 pub enum AdminCommand {
-	#[command(subcommand)]
 	/// Commands for managing appservices
+	#[command(subcommand)]
 	Appservices(AppserviceCommand),
 
-	#[command(subcommand)]
 	/// Commands for managing local users
+	#[command(subcommand)]
 	Users(UserCommand),
 
-	#[command(subcommand)]
 	/// Commands for managing registration tokens
+	#[command(subcommand)]
 	Token(TokenCommand),
 
+	/// Commands for managing OIDC
 	#[command(subcommand)]
+	Oidc(OidcCommand),
+
 	/// Commands for managing rooms
+	#[command(subcommand)]
 	Rooms(RoomCommand),
 
-	#[command(subcommand)]
 	/// Commands for managing federation
+	#[command(subcommand)]
 	Federation(FederationCommand),
 
-	#[command(subcommand)]
 	/// Commands for managing the server
+	#[command(subcommand)]
 	Server(ServerCommand),
 
-	#[command(subcommand)]
 	/// Commands for managing media
+	#[command(subcommand)]
 	Media(MediaCommand),
 
-	#[command(subcommand)]
 	/// Commands for checking integrity
+	#[command(subcommand)]
 	Check(CheckCommand),
 
-	#[command(subcommand)]
 	/// Commands for debugging things
+	#[command(subcommand)]
 	Debug(DebugCommand),
 
-	#[command(subcommand)]
 	/// Low-level queries for database getters and iterators
+	#[command(subcommand)]
 	Query(QueryCommand),
 }
 
@@ -80,6 +85,16 @@ pub(super) async fn process(command: AdminCommand, context: &Context<'_>) -> Res
 			context.bail_restricted()?;
 			token::process(command, context).await
 		},
+		| Oidc(command) => {
+			// OIDC commands are all restricted
+			context.bail_restricted()?;
+
+			if !context.services.oidc.enabled() {
+				return Err!("OIDC is not configured");
+			}
+
+			oidc::process(command, context).await
+		},
 		| Rooms(command) => room::process(command, context).await,
 		| Federation(command) => federation::process(command, context).await,
 		| Server(command) => server::process(command, context).await,

src/admin/check/commands.rs

@@ -6,7 +6,12 @@
 impl Context<'_> {
 	pub(super) async fn check_all_users(&self) -> Result {
 		let timer = tokio::time::Instant::now();
-		let users = self.services.users.stream().collect::<Vec<_>>().await;
+		let users = self
+			.services
+			.users
+			.stream_local_users()
+			.collect::<Vec<_>>()
+			.await;
 		let query_time = timer.elapsed();
 
 		let total = users.len();

src/admin/debug/commands.rs

@@ -31,6 +31,8 @@
 };
 use tracing_subscriber::EnvFilter;
 
+use crate::PAGE_SIZE;
+
 #[derive(Clone, Copy, Eq, PartialEq)]
 enum NodeStatus {
 	Normal(bool),
@@ -610,7 +612,7 @@ pub(super) async fn force_device_list_updates(&self) -> Result {
 		// Force E2EE device list updates for all users
 		self.services
 			.users
-			.stream()
+			.stream_local_users()
 			.for_each(async |user_id| self.services.users.mark_device_key_update(&user_id).await)
 			.await;
 
@@ -1166,4 +1168,56 @@ pub(super) async fn send_test_email(&self) -> Result {
 
 		Ok(())
 	}
+
+	pub(super) async fn rooms_by_extremity_count(&self, page: Option<usize>) -> Result {
+		let page = page.unwrap_or(1);
+		// My Giant Chain:tm:
+		let mapped: HashMap<OwnedRoomId, u64> = self
+			.services
+			.rooms
+			.state
+			.all_forward_extremities()
+			.ready_fold(HashMap::new(), move |mut map, (room_id, _)| {
+				let count: u64 = map.get(&room_id).copied().unwrap_or(0);
+				map.insert(room_id, count.saturating_add(1));
+				map
+			})
+			.await
+			.into_iter()
+			.filter_map(|(room_id, count)| (count >= 2).then_some((room_id, count)))
+			.collect();
+		if mapped.is_empty() {
+			return Err!("No more rooms.");
+		}
+
+		let mut rooms = mapped.keys().collect::<Vec<_>>();
+		rooms.sort_by_key(|room_id| {
+			mapped
+				.get(*room_id)
+				.copied()
+				.expect("keys must have values")
+		});
+		rooms.reverse();
+
+		let body = rooms
+			.into_iter()
+			.stream()
+			.skip(page.saturating_sub(1).saturating_mul(PAGE_SIZE))
+			.take(PAGE_SIZE)
+			.map(|room_id| {
+				format!(
+					"{room_id}: {}",
+					mapped.get(room_id).copied().expect("keys must have values")
+				)
+			})
+			.collect::<Vec<_>>()
+			.await;
+
+		self.write_str(&format!(
+			"Rooms by extremity count ({}):\n```\n{}\n```",
+			body.len(),
+			body.join("\n")
+		))
+		.await
+	}
 }

src/admin/debug/mod.rs

@@ -245,6 +245,11 @@ pub enum DebugCommand {
 	/// Send a test email to the invoking admin's email address
 	SendTestEmail,
 
+	/// Lists room IDs by forward extremity count in descending order
+	RoomsByExtremityCount {
+		page: Option<usize>,
+	},
+
 	/// Developer test stubs
 	#[command(subcommand)]
 	#[allow(non_snake_case)]

src/admin/mod.rs

@@ -16,6 +16,7 @@
 pub(crate) mod debug;
 pub(crate) mod federation;
 pub(crate) mod media;
+pub(crate) mod oidc;
 pub(crate) mod query;
 pub(crate) mod room;
 pub(crate) mod server;

src/admin/oidc/commands.rs

@@ -0,0 +1,25 @@
+use conduwuit::Result;
+
+use crate::utils::parse_active_local_user_id;
+
+impl crate::Context<'_> {
+	pub(super) async fn oidc_link(&self, user_id: String, subject: String) -> Result {
+		let user_id = parse_active_local_user_id(self.services, &user_id).await?;
+
+		self.services.oidc.link_user(&user_id, &subject);
+
+		self.write_str(&format!("Subject `{subject}` linked to account `{user_id}`."))
+			.await?;
+
+		Ok(())
+	}
+
+	pub(super) async fn oidc_unlink(&self, subject: String) -> Result {
+		self.services.oidc.unlink_user(&subject);
+
+		self.write_str(&format!("Subject `{subject}` unlinked."))
+			.await?;
+
+		Ok(())
+	}
+}

src/admin/oidc/mod.rs

@@ -0,0 +1,22 @@
+mod commands;
+
+use clap::Subcommand;
+use conduwuit::Result;
+use conduwuit_macros::admin_command_dispatch;
+
+#[admin_command_dispatch]
+#[derive(Debug, Subcommand)]
+pub enum OidcCommand {
+	/// Link a user ID to the given subject claim.
+	#[clap(name = "link")]
+	OidcLink {
+		user_id: String,
+		subject: String,
+	},
+
+	/// Unlink the given subject claim from its associated user ID.
+	#[clap(name = "unlink")]
+	OidcUnlink {
+		subject: String,
+	},
+}

src/admin/query/users.rs

@@ -191,8 +191,13 @@ async fn get_latest_backup(&self, user_id: OwnedUserId) -> Result {
 
 	async fn iter_users(&self) -> Result {
 		let timer = tokio::time::Instant::now();
-		let result: Vec<OwnedUserId> =
-			self.services.users.stream().map(Into::into).collect().await;
+		let result: Vec<OwnedUserId> = self
+			.services
+			.users
+			.stream_local_users()
+			.map(Into::into)
+			.collect()
+			.await;
 
 		let query_time = timer.elapsed();
 
@@ -202,7 +207,7 @@ async fn iter_users(&self) -> Result {
 
 	async fn iter_users2(&self) -> Result {
 		let timer = tokio::time::Instant::now();
-		let result: Vec<_> = self.services.users.stream().collect().await;
+		let result: Vec<_> = self.services.users.stream_local_users().collect().await;
 		let result: Vec<_> = result
 			.into_iter()
 			.map(|user_id| String::from_utf8_lossy(user_id.as_bytes()).into_owned())

src/admin/token/commands.rs

@@ -44,7 +44,7 @@ pub(super) async fn issue_token(&self, expires: super::TokenExpires) -> Result {
 			.services
 			.config
 			.oauth
-			.compatibility_mode
+			.compatibility_mode()
 			.oauth_available()
 		{
 			self.write_str(&format!(

src/admin/user/commands.rs

@@ -20,7 +20,7 @@
 		tag::{TagEvent, TagEventContent, TagInfo},
 	},
 };
-use service::users::HashedPassword;
+use service::users::{AccountStatus, HashedPassword};
 
 use crate::{
 	get_room_info,
@@ -54,16 +54,16 @@ pub(super) async fn create_user(&self, username: String, password: Option<String
 			.determine_registration_user_id(Some(username), None, None)
 			.await?;
 
-		let password = HashedPassword::new(
-			&password.unwrap_or_else(|| utils::random_string(AUTO_GEN_PASSWORD_LENGTH)),
-		)?;
+		let password =
+			&password.unwrap_or_else(|| utils::random_string(AUTO_GEN_PASSWORD_LENGTH));
 
 		self.services
 			.users
-			.create_local_account(&user_id, password, None)
-			.await;
+			.create_local_account(&user_id, Some(HashedPassword::new(password)?), None)
+			.await?;
 
-		self.write_str(&format!("Created user {user_id}")).await
+		self.write_str(&format!("Created user {user_id} with password `{password}`"))
+			.await
 	}
 
 	pub(super) async fn deactivate(&self, no_leave_rooms: bool, user_id: String) -> Result {
@@ -103,15 +103,12 @@ pub(super) async fn deactivate(&self, no_leave_rooms: bool, user_id: String) ->
 
 	pub(super) async fn suspend(&self, user_id: String) -> Result {
 		self.bail_restricted()?;
-		let user_id = parse_local_user_id(self.services, &user_id)?;
+		let user_id = parse_active_local_user_id(self.services, &user_id).await?;
 
 		if user_id == self.services.globals.server_user {
 			return Err!("Not allowed to suspend the server service account.",);
 		}
 
-		if !self.services.users.exists(&user_id).await {
-			return Err!("User {user_id} does not exist.");
-		}
 		if self.services.users.is_admin(&user_id).await {
 			return Err!("Admin users cannot be suspended.");
 		}
@@ -127,15 +124,12 @@ pub(super) async fn suspend(&self, user_id: String) -> Result {
 
 	pub(super) async fn unsuspend(&self, user_id: String) -> Result {
 		self.bail_restricted()?;
-		let user_id = parse_local_user_id(self.services, &user_id)?;
+		let user_id = parse_active_local_user_id(self.services, &user_id).await?;
 
 		if user_id == self.services.globals.server_user {
 			return Err!("Not allowed to unsuspend the server service account.",);
 		}
 
-		if !self.services.users.exists(&user_id).await {
-			return Err!("User {user_id} does not exist.");
-		}
 		self.services.users.unsuspend_account(&user_id).await;
 
 		self.write_str(&format!("User {user_id} has been unsuspended."))
@@ -147,6 +141,7 @@ pub(super) async fn reset_password(
 		logout: bool,
 		username: String,
 		password: Option<String>,
+		convert_to_local_account: bool,
 	) -> Result {
 		let user_id = parse_local_user_id(self.services, &username)?;
 
@@ -159,15 +154,37 @@ pub(super) async fn reset_password(
 
 		let new_password =
 			password.unwrap_or_else(|| utils::random_string(AUTO_GEN_PASSWORD_LENGTH));
+		let new_password_hash = HashedPassword::new(&new_password)?;
 
-		self.services
-			.users
-			.set_password(&user_id, Some(HashedPassword::new(&new_password)?));
+		if convert_to_local_account {
+			self.services
+				.users
+				.convert_to_local_account(&user_id, new_password_hash)
+				.await?;
+		} else {
+			match self.services.users.status(&user_id).await {
+				| AccountStatus::Active if !self.services.users.is_shadow(&user_id).await => {
+					self.services
+						.users
+						.set_password(&user_id, new_password_hash)
+						.await?;
+				},
+				| AccountStatus::NotFound => {
+					return Err!("The provided user does not exist.");
+				},
+				| _ => {
+					return Err!(
+						"The provided user is a shadow or deactivated account. To convert it to \
+						 a local account, pass the --convert-to-local-account flag."
+					);
+				},
+			}
 
-		self.write_str(&format!(
-			"Successfully reset the password for user {user_id}: `{new_password}`"
-		))
-		.await?;
+			self.write_str(&format!(
+				"Successfully reset the password for user {user_id}: `{new_password}`"
+			))
+			.await?;
+		}
 
 		if logout {
 			self.services
@@ -919,21 +936,16 @@ pub(super) async fn force_leave_remote_room(
 
 	pub(super) async fn lock(&self, user_id: String) -> Result {
 		self.bail_restricted()?;
-		let user_id = parse_local_user_id(self.services, &user_id)?;
-		assert!(
-			self.services.globals.user_is_local(&user_id),
-			"Parsed user_id must be a local user"
-		);
+		let user_id = parse_active_local_user_id(self.services, &user_id).await?;
+
 		if user_id == self.services.globals.server_user {
 			return Err!("Not allowed to lock the server service account.",);
 		}
 
-		if !self.services.users.exists(&user_id).await {
-			return Err!("User {user_id} does not exist.");
-		}
 		if self.services.users.is_admin(&user_id).await {
 			return Err!("Admin users cannot be locked.");
 		}
+
 		self.services
 			.users
 			.lock_account(&user_id, self.sender_or_service_user())
@@ -945,11 +957,8 @@ pub(super) async fn lock(&self, user_id: String) -> Result {
 
 	pub(super) async fn unlock(&self, user_id: String) -> Result {
 		self.bail_restricted()?;
-		let user_id = parse_local_user_id(self.services, &user_id)?;
-		assert!(
-			self.services.globals.user_is_local(&user_id),
-			"Parsed user_id must be a local user"
-		);
+		let user_id = parse_active_local_user_id(self.services, &user_id).await?;
+
 		self.services.users.unlock_account(&user_id).await;
 
 		self.write_str(&format!("User {user_id} has been unlocked."))
@@ -958,21 +967,16 @@ pub(super) async fn unlock(&self, user_id: String) -> Result {
 
 	pub(super) async fn logout(&self, user_id: String) -> Result {
 		self.bail_restricted()?;
-		let user_id = parse_local_user_id(self.services, &user_id)?;
-		assert!(
-			self.services.globals.user_is_local(&user_id),
-			"Parsed user_id must be a local user"
-		);
+		let user_id = parse_active_local_user_id(self.services, &user_id).await?;
+
 		if user_id == self.services.globals.server_user {
 			return Err!("Not allowed to log out the server service account.",);
 		}
 
-		if !self.services.users.exists(&user_id).await {
-			return Err!("User {user_id} does not exist.");
-		}
 		if self.services.users.is_admin(&user_id).await {
 			return Err!("You cannot forcefully log out admin users.");
 		}
+
 		self.services
 			.users
 			.all_device_ids(&user_id)
@@ -989,18 +993,12 @@ pub(super) async fn logout(&self, user_id: String) -> Result {
 
 	pub(super) async fn disable_login(&self, user_id: String) -> Result {
 		self.bail_restricted()?;
-		let user_id = parse_local_user_id(self.services, &user_id)?;
-		assert!(
-			self.services.globals.user_is_local(&user_id),
-			"Parsed user_id must be a local user"
-		);
+		let user_id = parse_active_local_user_id(self.services, &user_id).await?;
+
 		if user_id == self.services.globals.server_user {
 			return Err!("Not allowed to disable login for the server service account.",);
 		}
 
-		if !self.services.users.exists(&user_id).await {
-			return Err!("User {user_id} does not exist.");
-		}
 		if self.services.users.is_admin(&user_id).await {
 			return Err!("Admin users cannot have their login disallowed.");
 		}
@@ -1014,14 +1012,8 @@ pub(super) async fn disable_login(&self, user_id: String) -> Result {
 
 	pub(super) async fn enable_login(&self, user_id: String) -> Result {
 		self.bail_restricted()?;
-		let user_id = parse_local_user_id(self.services, &user_id)?;
-		assert!(
-			self.services.globals.user_is_local(&user_id),
-			"Parsed user_id must be a local user"
-		);
-		if !self.services.users.exists(&user_id).await {
-			return Err!("User {user_id} does not exist.");
-		}
+		let user_id = parse_active_local_user_id(self.services, &user_id).await?;
+
 		self.services.users.enable_login(&user_id);
 
 		self.write_str(&format!("{user_id} can now log in.")).await
@@ -1129,10 +1121,8 @@ pub(super) async fn change_email(&self, user_id: String, email: Option<String>)
 	}
 
 	pub(super) async fn reset_push_rules(&self, user_id: String) -> Result {
-		let user_id = parse_local_user_id(self.services, &user_id)?;
-		if !self.services.users.is_active(&user_id).await {
-			return Err!("User is not active.");
-		}
+		let user_id = parse_active_local_user_id(self.services, &user_id).await?;
+
 		recreate_push_rules_and_return(self.services, &user_id).await?;
 		self.write_str("Reset user's push rules to the server default.")
 			.await

src/admin/user/mod.rs

@@ -27,6 +27,9 @@ pub enum UserCommand {
 		username: String,
 		/// New password for the user, if unspecified one is generated
 		password: Option<String>,
+
+		#[arg(long)]
+		convert_to_local_account: bool,
 	},
 
 	/// Get a user's associated email address.

src/admin/utils.rs

@@ -54,14 +54,7 @@ pub(crate) async fn parse_active_local_user_id(
 	user_id: &str,
 ) -> Result<OwnedUserId> {
 	let user_id = parse_local_user_id(services, user_id)?;
-
-	if !services.users.exists(&user_id).await {
-		return Err!("User {user_id:?} does not exist on this server.");
-	}
-
-	if services.users.is_deactivated(&user_id).await? {
-		return Err!("User {user_id:?} is deactivated.");
-	}
+	services.users.status(&user_id).await.ensure_active()?;
 
 	Ok(user_id)
 }

src/api/client/account/mod.rs

@@ -128,7 +128,8 @@ pub(crate) async fn change_password_route(
 
 	services
 		.users
-		.set_password(&sender_user, Some(HashedPassword::new(&body.new_password)?));
+		.set_password(&sender_user, HashedPassword::new(&body.new_password)?)
+		.await?;
 
 	if body.logout_devices {
 		// Logout all devices except the current one

src/api/client/account/register.rs

@@ -72,7 +72,7 @@ pub(crate) async fn register_route(
 			.determine_registration_user_id(body.username.clone(), None, Some(appservice_info))
 			.await?;
 
-		services.users.create(&user_id, None)?;
+		services.users.create_shadow_account(&user_id).await?;
 
 		user_id
 	} else {
@@ -97,8 +97,8 @@ pub(crate) async fn register_route(
 
 		services
 			.users
-			.create_local_account(&user_id, password, identity.email)
-			.await;
+			.create_local_account(&user_id, Some(password), identity.email)
+			.await?;
 
 		user_id
 	};
@@ -106,7 +106,7 @@ pub(crate) async fn register_route(
 	let (token, device) = if !body.inhibit_login {
 		// If UIAA is disabled, we can't create a device. In that case only appservices
 		// can reach this point in the first place, so we return an error for them.
-		if !services.config.oauth.compatibility_mode.uiaa_available() {
+		if !services.config.oauth.compatibility_mode().uiaa_available() {
 			return Err!(Request(AppserviceLoginUnsupported(
 				"User-interactive appservice registration is not available on this server."
 			)));

src/api/client/account/threepid.rs

@@ -25,6 +25,10 @@ pub(crate) async fn third_party_route(
 	let sender_user = body.identity.expect_sender_user()?;
 	let mut threepids = vec![];
 
+	if !services.threepid.email_requirement().may_view() {
+		return Ok(get_3pids::v3::Response::new(vec![]));
+	}
+
 	if let Some(email) = services
 		.threepid
 		.get_email_for_localpart(sender_user.localpart())

src/api/client/admin/lock.rs

@@ -0,0 +1,87 @@
+use axum::extract::State;
+use conduwuit::{Err, Result};
+use futures::future::{join, join3};
+use ruma::api::client::admin::{is_user_locked, lock_user};
+
+use crate::Ruma;
+
+/// # `GET /_matrix/client/v1/admin/lock/{userId}`
+///
+/// Check the account lock status of a target user
+pub(crate) async fn get_lock_status(
+	State(services): State<crate::State>,
+	body: Ruma<is_user_locked::v1::Request>,
+) -> Result<is_user_locked::v1::Response> {
+	let (admin, status) = join(
+		services.users.is_admin(body.identity.expect_sender_user()?),
+		services.users.status(&body.user_id),
+	)
+	.await;
+
+	if !admin {
+		return Err!(Request(Forbidden("Only server administrators can use this endpoint")));
+	}
+
+	status.ensure_active()?;
+
+	Ok(is_user_locked::v1::Response::new(
+		services.users.is_locked(&body.user_id).await?,
+	))
+}
+
+/// # `PUT /_matrix/client/v1/admin/lock/{userId}`
+///
+/// Set the account lock status of a target user
+pub(crate) async fn put_lock_status(
+	State(services): State<crate::State>,
+	body: Ruma<lock_user::v1::Request>,
+) -> Result<lock_user::v1::Response> {
+	let sender_user = body.identity.expect_sender_user()?;
+
+	let (sender_admin, status, target_admin) = join3(
+		services.users.is_admin(sender_user),
+		services.users.status(&body.user_id),
+		services.users.is_admin(&body.user_id),
+	)
+	.await;
+
+	if !sender_admin {
+		return Err!(Request(Forbidden("Only server administrators can use this endpoint")));
+	}
+
+	status.ensure_active()?;
+
+	if body.user_id == *sender_user {
+		return Err!(Request(Forbidden("You cannot lock yourself")));
+	}
+
+	if target_admin {
+		return Err!(Request(Forbidden("You cannot lock another server administrator")));
+	}
+
+	if services.users.is_locked(&body.user_id).await? == body.locked {
+		// No change
+		return Ok(lock_user::v1::Response::new(body.locked));
+	}
+
+	let action = if body.locked {
+		services
+			.users
+			.suspend_account(&body.user_id, sender_user)
+			.await;
+		"locked"
+	} else {
+		services.users.unsuspend_account(&body.user_id).await;
+		"unlocked"
+	};
+
+	if services.config.admin_room_notices {
+		// Notify the admin room that an account has been un/suspended
+		services
+			.admin
+			.send_text(&format!("{} has been {} by {}.", body.user_id, action, sender_user))
+			.await;
+	}
+
+	Ok(lock_user::v1::Response::new(body.locked))
+}

src/api/client/admin/mod.rs

@@ -1,3 +1,4 @@
+mod lock;
 mod suspend;
 
-pub(crate) use self::suspend::*;
+pub(crate) use self::{lock::*, suspend::*};

src/api/client/admin/suspend.rs

@@ -1,7 +1,7 @@
 use axum::extract::State;
 use conduwuit::{Err, Result};
 use futures::future::{join, join3};
-use ruminuwuity::admin::{get_suspended, set_suspended};
+use ruma::api::client::admin::{is_user_suspended, suspend_user};
 
 use crate::Ruma;
 
@@ -10,23 +10,21 @@
 /// Check the suspension status of a target user
 pub(crate) async fn get_suspended_status(
 	State(services): State<crate::State>,
-	body: Ruma<get_suspended::v1::Request>,
-) -> Result<get_suspended::v1::Response> {
-	let (admin, active) = join(
+	body: Ruma<is_user_suspended::v1::Request>,
+) -> Result<is_user_suspended::v1::Response> {
+	let (admin, status) = join(
 		services.users.is_admin(body.identity.expect_sender_user()?),
-		services.users.is_active(&body.user_id),
+		services.users.status(&body.user_id),
 	)
 	.await;
+
 	if !admin {
 		return Err!(Request(Forbidden("Only server administrators can use this endpoint")));
 	}
-	if !services.globals.user_is_local(&body.user_id) {
-		return Err!(Request(InvalidParam("Can only check the suspended status of local users")));
-	}
-	if !active {
-		return Err!(Request(NotFound("Unknown user")));
-	}
-	Ok(get_suspended::v1::Response::new(
+
+	status.ensure_active()?;
+
+	Ok(is_user_suspended::v1::Response::new(
 		services.users.is_suspended(&body.user_id).await?,
 	))
 }
@@ -36,13 +34,13 @@ pub(crate) async fn get_suspended_status(
 /// Set the suspension status of a target user
 pub(crate) async fn put_suspended_status(
 	State(services): State<crate::State>,
-	body: Ruma<set_suspended::v1::Request>,
-) -> Result<set_suspended::v1::Response> {
+	body: Ruma<suspend_user::v1::Request>,
+) -> Result<suspend_user::v1::Response> {
 	let sender_user = body.identity.expect_sender_user()?;
 
-	let (sender_admin, active, target_admin) = join3(
+	let (sender_admin, status, target_admin) = join3(
 		services.users.is_admin(sender_user),
-		services.users.is_active(&body.user_id),
+		services.users.status(&body.user_id),
 		services.users.is_admin(&body.user_id),
 	)
 	.await;
@@ -50,21 +48,20 @@ pub(crate) async fn put_suspended_status(
 	if !sender_admin {
 		return Err!(Request(Forbidden("Only server administrators can use this endpoint")));
 	}
-	if !services.globals.user_is_local(&body.user_id) {
-		return Err!(Request(InvalidParam("Can only set the suspended status of local users")));
-	}
-	if !active {
-		return Err!(Request(NotFound("Unknown user")));
-	}
+
+	status.ensure_active()?;
+
 	if body.user_id == *sender_user {
 		return Err!(Request(Forbidden("You cannot suspend yourself")));
 	}
+
 	if target_admin {
 		return Err!(Request(Forbidden("You cannot suspend another server administrator")));
 	}
+
 	if services.users.is_suspended(&body.user_id).await? == body.suspended {
 		// No change
-		return Ok(set_suspended::v1::Response::new(body.suspended));
+		return Ok(suspend_user::v1::Response::new(body.suspended));
 	}
 
 	let action = if body.suspended {
@@ -86,5 +83,5 @@ pub(crate) async fn put_suspended_status(
 			.await;
 	}
 
-	Ok(set_suspended::v1::Response::new(body.suspended))
+	Ok(suspend_user::v1::Response::new(body.suspended))
 }

src/api/client/capabilities.rs

@@ -7,12 +7,12 @@
 	api::client::discovery::get_capabilities::{
 		self,
 		v3::{
-			Capabilities, GetLoginTokenCapability, RoomVersionStability, RoomVersionsCapability,
-			ThirdPartyIdChangesCapability,
+			Capabilities, GetLoginTokenCapability, ProfileFieldsCapability, RoomVersionStability,
+			RoomVersionsCapability, ThirdPartyIdChangesCapability,
 		},
 	},
+	assign,
 };
-use serde_json::json;
 
 use crate::Ruma;
 
@@ -40,22 +40,20 @@ pub(crate) async fn get_capabilities_route(
 	capabilities.get_login_token =
 		GetLoginTokenCapability::new(services.server.config.login_via_existing_session);
 
-	// MSC4133 capability
-	capabilities.set("uk.tcpip.msc4133.profile_fields", json!({"enabled": true}))?;
-
-	capabilities.set(
-		"org.matrix.msc4267.forget_forced_upon_leave",
-		json!({"enabled": services.config.forget_forced_upon_leave}),
-	)?;
+	capabilities.forget_forced_upon_leave.enabled = true;
 
 	if services
 		.users
 		.is_admin(body.identity.expect_sender_user()?)
 		.await
 	{
-		// Advertise suspension API
-		capabilities.set("uk.timedout.msc4323", json!({"suspend": true, "lock": false}))?;
+		capabilities.account_moderation.lock = true;
+		capabilities.account_moderation.suspend = true;
 	}
 
+	capabilities.profile_fields = Some(
+		assign!(ProfileFieldsCapability::new(true), { disallowed: Some(services.oidc.restricted_profile_fields()) }),
+	);
+
 	Ok(get_capabilities::v3::Response::new(capabilities))
 }

src/api/client/device.rs

@@ -121,7 +121,9 @@ pub(crate) async fn delete_device_route(
 	let sender_user = body.identity.expect_sender_user()?;
 
 	// Appservices get to skip UIAA for this endpoint
-	if let Some(sender_device) = body.identity.sender_device() {
+	if !body.identity.is_appservice() {
+		let sender_device = body.identity.expect_sender_device()?;
+
 		// Prompt the user to confirm with their password using UIAA
 		let _ = services
 			.uiaa

src/api/client/keys.rs

@@ -197,6 +197,7 @@ pub(crate) async fn upload_signing_keys_route(
 	if uiaa_needed_to_upload_keys(
 		services,
 		sender_user,
+		body.identity.is_appservice(),
 		body.self_signing_key.as_ref(),
 		body.user_signing_key.as_ref(),
 		body.master_key.as_ref(),
@@ -231,10 +232,16 @@ pub(crate) async fn upload_signing_keys_route(
 async fn uiaa_needed_to_upload_keys(
 	services: crate::State,
 	user_id: &UserId,
+	is_appservice: bool,
 	self_signing_key: Option<&Raw<CrossSigningKey>>,
 	user_signing_key: Option<&Raw<CrossSigningKey>>,
 	master_signing_key: Option<&Raw<CrossSigningKey>>,
 ) -> bool {
+	if is_appservice {
+		// Appservices can skip UIAA for this endpoint
+		return false;
+	}
+
 	let (self_signing_key, user_signing_key, master_signing_key) = (
 		self_signing_key.map(Raw::deserialize).flat_ok(),
 		user_signing_key.map(Raw::deserialize).flat_ok(),

src/api/client/message.rs

@@ -291,7 +291,9 @@ pub(crate) async fn is_ignored_pdu<Pdu>(
 {
 	// exclude Synapse's dummy events from bloating up response bodies. clients
 	// don't need to see this.
-	if event.kind().to_string() == "org.matrix.dummy_event" {
+	if !services.config.send_dummy_events_to_clients
+		&& event.kind().to_string() == "org.matrix.dummy_event"
+	{
 		return Ok(true);
 	}
 

src/api/client/oauth/mod.rs

@@ -37,7 +37,7 @@ pub(crate) fn router(state: crate::State) -> Router<crate::State> {
 		.layer(middleware::from_fn_with_state(
 			state,
 			async |State(state): State<crate::State>, request: Request, next: Next| -> Response {
-				if state.config.oauth.compatibility_mode.oauth_available() {
+				if state.config.oauth.compatibility_mode().oauth_available() {
 					next.run(request).await
 				} else {
 					(StatusCode::NOT_FOUND, "OAuth is unavailable on this server").into_response()

src/api/client/oauth/server_metadata.rs

@@ -21,7 +21,7 @@ pub(crate) async fn get_authorization_server_metadata_route(
 	State(services): State<crate::State>,
 	_body: Ruma<get_authorization_server_metadata::v1::Request>,
 ) -> Result<get_authorization_server_metadata::v1::Response> {
-	if !services.config.oauth.compatibility_mode.oauth_available() {
+	if !services.config.oauth.compatibility_mode().oauth_available() {
 		return Err!(Request(Unrecognized("OAuth is unavailable on this server")));
 	}
 

src/api/client/profile.rs

@@ -1,9 +1,8 @@
 use std::collections::BTreeMap;
 
 use axum::extract::State;
-use conduwuit::{Err, Result, matrix::pdu::PartialPdu, utils::to_canonical_object};
+use conduwuit::{Err, Result};
 use conduwuit_service::Services;
-use futures::StreamExt;
 use ruma::{
 	UserId,
 	api::{
@@ -13,11 +12,10 @@
 		federation,
 	},
 	assign,
-	events::room::member::MembershipState,
-	presence::PresenceState,
 	profile::{ProfileFieldName, ProfileFieldValue},
 };
-use serde_json::{Value, to_value};
+use serde_json::Value;
+use service::users::ProfileFieldChange;
 
 use crate::Ruma;
 
@@ -65,13 +63,24 @@ pub(crate) async fn set_profile_field_route(
 		return Err!(Request(InvalidParam("You may not change a remote user's profile data.")));
 	}
 
-	set_profile_field(
-		&services,
-		&body.user_id,
-		ProfileFieldChange::Set(body.value.clone()),
-		body.propagate_to.clone(),
-	)
-	.await?;
+	if services
+		.oidc
+		.restricted_profile_fields()
+		.contains(&body.value.field_name())
+	{
+		return Err!(Request(Forbidden(
+			"This profile field is controlled by your identity provider."
+		)));
+	}
+
+	services
+		.users
+		.set_profile_field(
+			&body.user_id,
+			ProfileFieldChange::Set(body.value.clone()),
+			body.propagate_to.clone(),
+		)
+		.await?;
 
 	Ok(set_profile_field::v3::Response::new())
 }
@@ -94,13 +103,24 @@ pub(crate) async fn delete_profile_field_route(
 		return Err!(Request(InvalidParam("You may not change a remote user's profile data.")));
 	}
 
-	set_profile_field(
-		&services,
-		&body.user_id,
-		ProfileFieldChange::Delete(body.field.clone()),
-		body.propagate_to.clone(),
-	)
-	.await?;
+	if services
+		.oidc
+		.restricted_profile_fields()
+		.contains(&body.field)
+	{
+		return Err!(Request(Forbidden(
+			"This profile field is controlled by your identity provider."
+		)));
+	}
+
+	services
+		.users
+		.set_profile_field(
+			&body.user_id,
+			ProfileFieldChange::Delete(body.field.clone()),
+			body.propagate_to.clone(),
+		)
+		.await?;
 
 	Ok(delete_profile_field::v3::Response::new())
 }
@@ -110,8 +130,8 @@ async fn fetch_full_profile(
 	user_id: &UserId,
 ) -> Option<BTreeMap<String, Value>> {
 	// If the user exists locally, fetch their local profile
-	if services.users.exists(user_id).await {
-		return Some(get_local_profile(services, user_id).await);
+	if services.users.status(user_id).await.is_found() {
+		return Some(services.users.get_local_profile(user_id).await);
 	}
 
 	// Otherwise ask their homeserver
@@ -135,13 +155,10 @@ async fn fetch_full_profile(
 			continue;
 		};
 
-		let _ = set_profile_field(
-			services,
-			user_id,
-			ProfileFieldChange::Set(value),
-			PropagateTo::None,
-		)
-		.await;
+		let _ = services
+			.users
+			.set_profile_field(user_id, ProfileFieldChange::Set(value), PropagateTo::None)
+			.await;
 	}
 
 	Some(BTreeMap::from_iter(response))
@@ -154,7 +171,7 @@ async fn fetch_profile_field(
 ) -> Result<Option<ProfileFieldValue>> {
 	// If the user exists locally, fetch their local profile field
 	if services.globals.user_is_local(user_id) {
-		return Ok(get_local_profile_field(services, user_id, field).await);
+		return Ok(services.users.get_local_profile_field(user_id, field).await);
 	}
 
 	// Otherwise ask their homeserver
@@ -175,13 +192,14 @@ async fn fetch_profile_field(
 
 	if let Some(value) = response.get(field.as_str()).map(ToOwned::to_owned) {
 		if let Ok(value) = ProfileFieldValue::new(field.as_str(), value) {
-			let _ = set_profile_field(
-				services,
-				user_id,
-				ProfileFieldChange::Set(value.clone()),
-				PropagateTo::None,
-			)
-			.await;
+			let _ = services
+				.users
+				.set_profile_field(
+					user_id,
+					ProfileFieldChange::Set(value.clone()),
+					PropagateTo::None,
+				)
+				.await;
 
 			Ok(Some(value))
 		} else {
@@ -190,260 +208,11 @@ async fn fetch_profile_field(
 			)))
 		}
 	} else {
-		let _ = set_profile_field(
-			services,
-			user_id,
-			ProfileFieldChange::Delete(field),
-			PropagateTo::None,
-		)
-		.await;
+		let _ = services
+			.users
+			.set_profile_field(user_id, ProfileFieldChange::Delete(field), PropagateTo::None)
+			.await;
 
 		Ok(None)
 	}
 }
-
-pub(crate) async fn get_local_profile(
-	services: &Services,
-	user_id: &UserId,
-) -> BTreeMap<String, Value> {
-	let mut profile = BTreeMap::new();
-
-	// Get displayname and avatar_url independently because `all_profile_keys`
-	// doesn't include them
-	for field in [ProfileFieldName::AvatarUrl, ProfileFieldName::DisplayName] {
-		let key = field.as_str().to_owned();
-
-		if let Some(value) = get_local_profile_field(services, user_id, field).await {
-			profile.insert(key, value.value().into_owned());
-		}
-	}
-
-	// Insert all other profile fields
-	let mut all_fields = services.users.all_profile_keys(user_id);
-
-	while let Some((key, value)) = all_fields.next().await {
-		profile.insert(key, value);
-	}
-
-	profile
-}
-
-pub(crate) async fn get_local_profile_field(
-	services: &Services,
-	user_id: &UserId,
-	field: ProfileFieldName,
-) -> Option<ProfileFieldValue> {
-	let value = match field.clone() {
-		| ProfileFieldName::AvatarUrl => services
-			.users
-			.avatar_url(user_id)
-			.await
-			.ok()
-			.map(to_value)
-			.transpose()
-			.expect("converting avatar url to value should succeed"),
-		| ProfileFieldName::DisplayName => services
-			.users
-			.displayname(user_id)
-			.await
-			.ok()
-			.map(to_value)
-			.transpose()
-			.expect("converting displayname to value should succeed"),
-		| other => services
-			.users
-			.profile_key(user_id, other.as_str())
-			.await
-			.ok(),
-	}?;
-
-	Some(
-		ProfileFieldValue::new(field.as_str(), value)
-			.expect("local profile field should be valid"),
-	)
-}
-
-enum ProfileFieldChange {
-	Set(ProfileFieldValue),
-	Delete(ProfileFieldName),
-}
-
-impl ProfileFieldChange {
-	fn field_name(&self) -> ProfileFieldName {
-		match self {
-			| &Self::Delete(ref name) => name.clone(),
-			| &Self::Set(ref value) => value.field_name(),
-		}
-	}
-
-	fn value(&self) -> Option<Value> {
-		if let Self::Set(value) = self {
-			Some(value.value().into_owned())
-		} else {
-			None
-		}
-	}
-}
-
-async fn set_profile_field(
-	services: &Services,
-	user_id: &UserId,
-	change: ProfileFieldChange,
-	propagate_to: PropagateTo,
-) -> Result<()> {
-	const MAX_KEY_LENGTH_BYTES: usize = 255;
-	const MAX_PROFILE_LENGTH_BYTES: usize = 65536;
-
-	let field_name = change.field_name();
-
-	// TODO: The spec mentions special error codes (M_PROFILE_TOO_LARGE,
-	// M_KEY_TOO_LARGE) for profile field size limits, but they're not in its list
-	// of error codes and Ruma doesn't have them. Should we return those, or is
-	// M_TOO_LARGE okay?
-	if field_name.as_str().len() > MAX_KEY_LENGTH_BYTES {
-		return Err!(Request(TooLarge(
-			"Individual profile keys must not exceed {MAX_KEY_LENGTH_BYTES} bytes in length."
-		)));
-	}
-
-	// Serialize the entire profile as canonical JSON, including the new change,
-	// to check if it exceeds 64 KiB
-	{
-		let mut full_profile = get_local_profile(services, user_id).await;
-
-		match &change {
-			| ProfileFieldChange::Set(value) => {
-				full_profile.insert(
-					value.field_name().as_str().to_owned(),
-					value.value().clone().into_owned(),
-				);
-			},
-			| ProfileFieldChange::Delete(key) => {
-				full_profile.remove(key.as_str());
-			},
-		}
-
-		if let Ok(canonical_profile) = to_canonical_object(full_profile) {
-			if serde_json::to_string(&canonical_profile)
-				.expect("should be able to serialize to string")
-				.len() > MAX_PROFILE_LENGTH_BYTES
-			{
-				return Err!(
-					"Profile data must not exceed {MAX_PROFILE_LENGTH_BYTES} bytes in length."
-				);
-			}
-		} else {
-			return Err!(Request(BadJson("Failed to canonicalize profile.")));
-		}
-	}
-
-	// If the user is local and changed their displayname or avatar_url, update it
-	// in all their joined rooms. This is done before updating their profile data
-	// so we can check the old value of the field if `propagate_to` is `unchanged`.
-	if matches!(field_name, ProfileFieldName::AvatarUrl | ProfileFieldName::DisplayName)
-		&& matches!(propagate_to, PropagateTo::All | PropagateTo::Unchanged)
-		&& services.globals.user_is_local(user_id)
-	{
-		let current_displayname = services.users.displayname(user_id).await.ok();
-		let current_avatar_url = services.users.avatar_url(user_id).await.ok();
-
-		let mut all_joined_rooms = services.rooms.state_cache.rooms_joined(user_id);
-
-		while let Some(room_id) = all_joined_rooms.next().await {
-			// TODO: this clobbers any custom fields on the event content
-			let mut current_membership = services
-				.rooms
-				.state_accessor
-				.get_member(&room_id, user_id)
-				.await
-				.expect("should be able to fetch membership event for joined room");
-
-			assert_eq!(
-				current_membership.membership,
-				MembershipState::Join,
-				"user should be joined"
-			);
-
-			// If `propagate_to` is `unchanged`, and the current value of the field we're
-			// updating was changed from its global value in this room, skip it.
-			if matches!(propagate_to, PropagateTo::Unchanged) {
-				let field_changed_from_global = match field_name {
-					| ProfileFieldName::AvatarUrl =>
-						current_membership.avatar_url.as_ref() != current_avatar_url.as_ref(),
-					| ProfileFieldName::DisplayName =>
-						current_membership.displayname.as_ref() != current_displayname.as_ref(),
-					| _ => unreachable!(),
-				};
-
-				if field_changed_from_global {
-					continue;
-				}
-			}
-
-			let state_lock = services.rooms.state.mutex.lock(room_id.as_str()).await;
-
-			// Preserve keys in accordance with the key copying rules
-			current_membership.reason = None;
-			current_membership.join_authorized_via_users_server = None;
-			match &change {
-				| ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(avatar_url)) => {
-					current_membership.avatar_url = Some(avatar_url.clone());
-				},
-				| ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)) => {
-					current_membership.displayname = Some(displayname.clone());
-				},
-				| ProfileFieldChange::Delete(ProfileFieldName::AvatarUrl) => {
-					current_membership.avatar_url = None;
-				},
-				| ProfileFieldChange::Delete(ProfileFieldName::DisplayName) => {
-					current_membership.displayname = None;
-				},
-				| _ => unreachable!(),
-			}
-
-			let _ = services
-				.rooms
-				.timeline
-				.build_and_append_pdu(
-					PartialPdu::state(user_id.to_string(), &current_membership),
-					user_id,
-					Some(&room_id),
-					&state_lock,
-				)
-				.await;
-		}
-
-		if services.config.allow_local_presence {
-			// Send a presence EDU to indicate the profile changed
-			let _ = services
-				.presence
-				.ping_presence(user_id, &PresenceState::Online)
-				.await;
-		}
-	}
-
-	match change {
-		| ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)) => {
-			services
-				.users
-				.set_displayname(user_id, Some(displayname).filter(|dn| !dn.is_empty()));
-		},
-		| ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(avatar_url)) => {
-			services
-				.users
-				.set_avatar_url(user_id, Some(avatar_url).filter(|av| av.is_valid()));
-		},
-		| ProfileFieldChange::Delete(ProfileFieldName::DisplayName) => {
-			services.users.set_displayname(user_id, None);
-		},
-		| ProfileFieldChange::Delete(ProfileFieldName::AvatarUrl) => {
-			services.users.set_avatar_url(user_id, None);
-		},
-		| other =>
-			services
-				.users
-				.set_profile_key(user_id, other.field_name().as_str(), other.value()),
-	}
-
-	Ok(())
-}

src/api/client/report.rs

@@ -149,7 +149,7 @@ pub(crate) async fn report_user_route(
 
 	delay_response().await;
 
-	if !services.users.is_active_local(&body.user_id).await {
+	if !services.users.status(&body.user_id).await.is_found() {
 		// return 200 as to not reveal if the user exists. Recommended by spec.
 		return Ok(report_user::v3::Response::new());
 	}

src/api/client/session.rs

@@ -43,7 +43,7 @@ pub(crate) async fn get_login_types_route(
 	ClientIp(client): ClientIp,
 	_body: Ruma<get_login_types::v3::Request>,
 ) -> Result<get_login_types::v3::Response> {
-	if !services.config.oauth.compatibility_mode.uiaa_available() {
+	if !services.config.oauth.compatibility_mode().uiaa_available() {
 		return Err!(Request(Unrecognized(
 			"User-interactive authentication is not available on this server."
 		)));
@@ -88,14 +88,6 @@ pub async fn handle_login(
 		UserId::parse_with_server_name(user_id_or_localpart, &services.config.server_name)
 			.map_err(|_| err!(Request(InvalidUsername("User ID is malformed"))))?;
 
-	if !services.globals.user_is_local(&user_id) {
-		return Err!(Request(InvalidParam("User ID does not belong to this homeserver")));
-	}
-
-	if services.users.is_deactivated(&user_id).await? {
-		return Err!(Request(UserDeactivated("This account has been deactivated.")));
-	}
-
 	if services.users.is_locked(&user_id).await? {
 		return Err!(Request(UserLocked("This account has been locked.")));
 	}
@@ -128,7 +120,7 @@ pub(crate) async fn login_route(
 	ClientIp(client): ClientIp,
 	body: Ruma<login::v3::Request>,
 ) -> Result<login::v3::Response> {
-	if !services.config.oauth.compatibility_mode.uiaa_available() {
+	if !services.config.oauth.compatibility_mode().uiaa_available() {
 		return match body.login_info {
 			| LoginInfo::ApplicationService(_) => {
 				Err!(Request(AppserviceLoginUnsupported(

src/api/client/user_directory.rs

@@ -32,22 +32,26 @@ pub(crate) async fn search_users_route(
 		.min(LIMIT_MAX);
 
 	let search_term = body.search_term.to_lowercase();
-	let mut users = services.users.stream().broad_filter_map(async |user_id| {
-		let display_name = services.users.displayname(&user_id).await.ok();
 
-		let user_id_matches = user_id.as_str().to_lowercase().contains(&search_term);
+	let mut users = services
+		.users
+		.stream_local_users()
+		.chain(services.users.stream_remote_users())
+		.broad_filter_map(async |user_id| {
+			let display_name = services.users.displayname(&user_id).await.ok();
 
-		let display_name_matches = display_name
-			.as_deref()
-			.map(str::to_lowercase)
-			.is_some_and(|display_name| display_name.contains(&search_term));
+			let user_id_matches = user_id.as_str().to_lowercase().contains(&search_term);
 
-		if !user_id_matches && !display_name_matches {
-			return None;
-		}
+			let display_name_matches = display_name
+				.as_deref()
+				.map(str::to_lowercase)
+				.is_some_and(|display_name| display_name.contains(&search_term));
 
-		let user_in_public_room =
-			services
+			if !user_id_matches && !display_name_matches {
+				return None;
+			}
+
+			let user_in_public_room = services
 				.rooms
 				.state_cache
 				.rooms_joined(&user_id)
@@ -60,22 +64,22 @@ pub(crate) async fn search_users_route(
 						.await
 				});
 
-		let user_sees_user = services
-			.rooms
-			.state_cache
-			.user_sees_user(sender_user, &user_id);
+			let user_sees_user = services
+				.rooms
+				.state_cache
+				.user_sees_user(sender_user, &user_id);
 
-		pin_mut!(user_in_public_room, user_sees_user);
+			pin_mut!(user_in_public_room, user_sees_user);
 
-		if user_in_public_room.or(user_sees_user).await {
-			Some(assign!(search_users::v3::User::new(user_id.clone()), {
-				display_name,
-				avatar_url: services.users.avatar_url(&user_id).await.ok(),
-			}))
-		} else {
-			None
-		}
-	});
+			if user_in_public_room.or(user_sees_user).await {
+				Some(assign!(search_users::v3::User::new(user_id.clone()), {
+					display_name,
+					avatar_url: services.users.avatar_url(&user_id).await.ok(),
+				}))
+			} else {
+				None
+			}
+		});
 
 	let results = users.by_ref().take(limit).collect().await;
 	let limited = users.next().await.is_some();

src/api/router.rs

@@ -181,6 +181,8 @@ pub fn build(router: Router<State>, state: State) -> Router<State> {
 		.ruma_route(&client::get_room_summary)
 		.ruma_route(&client::get_suspended_status)
 		.ruma_route(&client::put_suspended_status)
+		.ruma_route(&client::get_lock_status)
+		.ruma_route(&client::put_lock_status)
 		.ruma_route(&client::well_known_support)
 		.ruma_route(&client::well_known_client)
 		.ruma_route(&client::well_known_policy_server)

src/api/server/get_missing_events.rs

@@ -4,15 +4,11 @@
 use conduwuit::{Err, Event, Result, debug, info, trace, utils::to_canonical_object, warn};
 use ruma::{OwnedEventId, api::federation::event::get_missing_events};
 use serde_json::{json, value::RawValue};
+use service::rooms::event_handler::GET_MISSING_EVENTS_MAX_BATCH_SIZE;
 
 use super::AccessCheck;
 use crate::Ruma;
 
-/// arbitrary number but synapse's is 20 and we can handle lots of these anyways
-const LIMIT_MAX: usize = 50;
-/// spec says default is 10
-const LIMIT_DEFAULT: usize = 10;
-
 /// # `POST /_matrix/federation/v1/get_missing_events/{roomId}`
 ///
 /// Retrieves events that the sender is missing.
@@ -45,8 +41,8 @@ pub(crate) async fn get_missing_events_route(
 	let limit = body
 		.limit
 		.try_into()
-		.unwrap_or(LIMIT_DEFAULT)
-		.min(LIMIT_MAX);
+		.unwrap_or(10)
+		.min(GET_MISSING_EVENTS_MAX_BATCH_SIZE);
 
 	let room_version = services.rooms.state.get_room_version(&body.room_id).await?;
 

src/api/server/query.rs

@@ -7,10 +7,7 @@
 	api::federation::query::{get_profile_information, get_room_information},
 };
 
-use crate::{
-	Ruma,
-	client::{get_local_profile, get_local_profile_field},
-};
+use crate::Ruma;
 
 /// # `GET /_matrix/federation/v1/query/directory`
 ///
@@ -75,15 +72,19 @@ pub(crate) async fn get_profile_information_route(
 	let response = if let Some(field) = &body.field {
 		let mut response = get_profile_information::v1::Response::new();
 
-		if let Some(value) =
-			get_local_profile_field(&services, &body.user_id, field.to_owned()).await
+		if let Some(value) = services
+			.users
+			.get_local_profile_field(&body.user_id, field.to_owned())
+			.await
 		{
 			response.set(value.field_name().as_str().to_owned(), value.value().into_owned());
 		}
 
 		response
 	} else {
-		get_local_profile(&services, &body.user_id)
+		services
+			.users
+			.get_local_profile(&body.user_id)
 			.await
 			.into_iter()
 			.collect()

src/api/server/send.rs

@@ -1,5 +1,5 @@
 use std::{
-	collections::{BTreeMap, HashMap, HashSet},
+	collections::{BTreeMap, HashMap},
 	net::IpAddr,
 	time::{Duration, Instant},
 };
@@ -7,9 +7,8 @@
 use axum::extract::State;
 use axum_client_ip::ClientIp;
 use conduwuit::{
-	Err, Error, Result, debug, debug_warn, err, error,
+	Err, Error, Result, debug, debug_error, debug_warn, err, error,
 	result::LogErr,
-	state_res::lexicographical_topological_sort,
 	trace,
 	utils::{
 		IterStream, ReadyExt, millis_since_unix_epoch,
@@ -25,8 +24,7 @@
 use http::StatusCode;
 use itertools::Itertools;
 use ruma::{
-	CanonicalJsonObject, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId,
-	OwnedUserId, RoomId, ServerName, UInt, UserId,
+	CanonicalJsonObject, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, ServerName, UserId,
 	api::{
 		error::{ErrorKind, LimitExceededErrorData},
 		federation::transactions::{
@@ -39,12 +37,12 @@
 		},
 	},
 	events::receipt::{ReceiptEvent, ReceiptEventContent, ReceiptType},
-	int,
 	serde::Raw,
 	to_device::DeviceIdOrAllDevices,
 };
-use service::transactions::{
-	FederationTxnState, TransactionError, TxnKey, WrappedTransactionResponse,
+use service::{
+	rooms::event_handler::build_local_dag,
+	transactions::{FederationTxnState, TransactionError, TxnKey, WrappedTransactionResponse},
 };
 use tokio::sync::watch::{Receiver, Sender};
 use tracing::instrument;
@@ -133,6 +131,7 @@ async fn wait_for_result(
 }
 
 #[instrument(
+	name="transaction"
 	skip_all,
 	fields(
 		id = ?body.transaction_id.as_str(),
@@ -174,8 +173,14 @@ async fn process_inbound_transaction(
 
 	for (id, result) in &results {
 		if let Err(e) = result {
-			if matches!(e, Error::BadRequest(ErrorKind::NotFound, _)) {
-				debug_warn!("Incoming PDU failed {id}: {e:?}");
+			match e {
+				| Error::BadRequest(
+					ErrorKind::Forbidden | ErrorKind::InvalidParam | ErrorKind::BadJson,
+					..,
+				) => {
+					debug_warn!("Incoming PDU {id} failed: {e:?}");
+				},
+				| _ => debug_error!("Incoming PDU {id} failed: {e:?}"),
 			}
 		}
 	}
@@ -269,74 +274,6 @@ async fn handle(
 	Ok(results)
 }
 
-/// Attempts to build a localised directed acyclic graph out of the given PDUs,
-/// returning them in a topologically sorted order.
-///
-/// This is used to attempt to process PDUs in an order that respects their
-/// dependencies, however it is ultimately the sender's responsibility to send
-/// them in a processable order, so this is just a best effort attempt. It does
-/// not account for power levels or other tie breaks.
-async fn build_local_dag(
-	pdu_map: &HashMap<OwnedEventId, CanonicalJsonObject>,
-) -> Result<Vec<OwnedEventId>> {
-	debug_assert!(pdu_map.len() >= 2, "needless call to build_local_dag with less than 2 PDUs");
-	let mut dag: HashMap<OwnedEventId, HashSet<OwnedEventId>> =
-		HashMap::with_capacity(pdu_map.len());
-	let mut id_origin_ts: HashMap<OwnedEventId, _> = HashMap::with_capacity(pdu_map.len());
-
-	for (event_id, value) in pdu_map {
-		// We already checked that these properties are correct in parse_incoming_pdu,
-		// so it's safe to unwrap here.
-		// We also filter to remove any prev_events that are not in this pdu_map, as we
-		// need to have at least one event with zero out degrees for the lexico-topo
-		// sort below. If there are multiple events with omitted prevs, they will be
-		// ordered by timestamp, then event ID. At that point though, it's unlikely to
-		// matter.
-		let prev_events = value
-			.get("prev_events")
-			.unwrap()
-			.as_array()
-			.unwrap()
-			.iter()
-			.map(|v| EventId::parse(v.as_str().unwrap()).unwrap())
-			.filter(|id| pdu_map.contains_key(id))
-			.collect();
-
-		dag.insert(event_id.clone(), prev_events);
-		let origin_server_ts = value
-			.get("origin_server_ts")
-			.and_then(ruma::CanonicalJsonValue::as_integer)
-			.unwrap_or_default();
-		id_origin_ts.insert(event_id.clone(), origin_server_ts);
-	}
-
-	debug!(count = dag.len(), "Sorting incoming events with partial graph");
-	lexicographical_topological_sort(&dag, &async |node_id| {
-		// Note: we don't bother fetching power levels because that would massively slow
-		// this function down. This is a best-effort attempt to order events correctly
-		// for processing, however ultimately that should be the sender's job.
-		let ts = id_origin_ts
-			.get(&node_id)
-			.copied()
-			.unwrap_or_else(|| int!(0))
-			.to_string()
-			.parse::<u64>()
-			.ok()
-			.and_then(UInt::new)
-			.unwrap_or_default();
-		Ok((int!(0), MilliSecondsSinceUnixEpoch(ts)))
-	})
-	.await
-	.inspect(|sorted| {
-		debug_assert_eq!(
-			sorted.len(),
-			pdu_map.len(),
-			"Sorted graph was not the same size as the input graph"
-		);
-	})
-	.map_err(|e| err!("failed to resolve local graph: {e}"))
-}
-
 async fn handle_room(
 	services: &Services,
 	_client: &IpAddr,
@@ -352,7 +289,7 @@ async fn handle_room(
 		.await;
 
 	let room_id = &room_id;
-	let pdu_map: HashMap<OwnedEventId, CanonicalJsonObject> = pdus
+	let mut pdu_map: HashMap<OwnedEventId, CanonicalJsonObject> = pdus
 		.into_iter()
 		.map(|(_, event_id, value)| (event_id, value))
 		.collect();
@@ -360,7 +297,11 @@ async fn handle_room(
 	// failure (e.g., cycles). This is best-effort; proper ordering is the sender's
 	// responsibility.
 	let sorted_event_ids = if pdu_map.len() >= 2 {
-		build_local_dag(&pdu_map).await.unwrap_or_else(|e| {
+		let refmap = pdu_map
+			.iter()
+			.map(|(event_id, obj)| (event_id.clone(), obj))
+			.collect();
+		build_local_dag(&refmap).await.unwrap_or_else(|e| {
 			debug_warn!("Failed to build local DAG for room {room_id}: {e}");
 			pdu_map.keys().cloned().collect()
 		})
@@ -370,9 +311,8 @@ async fn handle_room(
 	let mut results = Vec::with_capacity(sorted_event_ids.len());
 	for event_id in sorted_event_ids {
 		let value = pdu_map
-			.get(&event_id)
-			.expect("sorted event IDs must be from the original map")
-			.clone();
+			.remove(&event_id)
+			.expect("sorted event IDs must be from the original map");
 		services
 			.server
 			.check_running()
@@ -380,7 +320,8 @@ async fn handle_room(
 		let result = services
 			.rooms
 			.event_handler
-			.handle_incoming_pdu(origin, room_id, &event_id, value, true)
+			.handle_incoming_pdu(origin, room_id, &event_id, value.clone(), true)
+			.boxed()
 			.await
 			.map(|_| ());
 		results.push((event_id, result));
@@ -679,8 +620,9 @@ async fn handle_edu_direct_to_device(
 		.broad_filter_map(|(target_user_id, map)| async move {
 			services
 				.users
-				.is_active_local(&target_user_id)
+				.status(&target_user_id)
 				.await
+				.is_active()
 				.then_some((target_user_id, map))
 		})
 		.for_each_concurrent(automatic_width(), |(target_user_id, map)| {

src/core/Cargo.toml

@@ -9,6 +9,7 @@ repository.workspace = true
 version.workspace = true
 
 [lib]
+doctest = false
 path = "mod.rs"
 crate-type = [
 	"rlib",
@@ -117,6 +118,7 @@ url.workspace = true
 parking_lot.workspace = true
 lock_api.workspace = true
 hyper-util.workspace = true
+openidconnect.workspace = true
 
 [target.'cfg(unix)'.dependencies]
 nix.workspace = true

src/core/config/check.rs

@@ -289,6 +289,15 @@ pub fn check(config: &Config) -> Result {
 		| _ => (),
 	}
 
+	if let Some(oidc) = &config.oauth.oidc
+		&& oidc.client_secret.is_none()
+		&& oidc.client_secret_file.is_none()
+	{
+		return Err!(
+			"Either `client_secret` or `client_secret_file` must be set if OIDC is configured."
+		);
+	}
+
 	Ok(())
 }
 

src/core/config/mod.rs

@@ -4,7 +4,7 @@
 pub mod proxy;
 
 use std::{
-	collections::{BTreeMap, BTreeSet},
+	collections::{BTreeMap, BTreeSet, HashMap},
 	net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
 	path::PathBuf,
 };
@@ -17,10 +17,12 @@
 use figment::providers::{Env, Format, Toml};
 pub use figment::{Figment, value::Value as FigmentValue};
 use lettre::message::Mailbox;
+use openidconnect::{ClientId, ClientSecret, Scope};
 use regex::RegexSet;
 use ruma::{
 	OwnedRoomId, OwnedRoomOrAliasId, OwnedServerName, OwnedUserId, RoomVersionId,
 	api::client::{discovery::discover_support::ContactRole, rtc::RtcTransport},
+	profile::ProfileFieldName,
 	serde::Base64,
 };
 use serde::{Deserialize, Serialize, de::IgnoredAny};
@@ -375,7 +377,7 @@ pub struct Config {
 	#[serde(default = "default_max_request_size")]
 	pub max_request_size: usize,
 
-	/// default: 192
+	/// default: 1024
 	#[serde(default = "default_max_fetch_prev_events")]
 	pub max_fetch_prev_events: u16,
 
@@ -787,6 +789,16 @@ pub struct Config {
 	/// a substitute for moderation bots.
 	pub default_room_acl_deny: Option<Vec<String>>,
 
+	/// The number of forward extremities to tolerate in a room before
+	/// attempting to manually squash them with a "dummy event". Setting this
+	/// above 20 will hinder its efficacy, and setting it below 5 will cause
+	/// more dummy events to be sent than necessary (which increases federation
+	/// traffic).
+	///
+	/// default: 10
+	#[serde(default = "default_extremity_threshold")]
+	pub dummy_event_threshold: u8,
+
 	/// display: nested
 	#[serde(default)]
 	pub well_known: WellKnownConfig,
@@ -1657,6 +1669,11 @@ pub struct Config {
 	#[serde(default)]
 	pub send_messages_from_ignored_users_to_client: bool,
 
+	/// Send "org.matrix.dummy_event" events to the client. This is a debugging
+	/// option.
+	#[serde(default)]
+	pub send_dummy_events_to_clients: bool,
+
 	/// Vector list of IPv4 and IPv6 CIDR ranges / subnets *in quotes* that you
 	/// do not want continuwuity to send outbound requests to. Defaults to
 	/// RFC1918, unroutable, loopback, multicast, and testnet addresses for
@@ -2404,10 +2421,24 @@ pub struct OauthConfig {
 	///   legacy authentication will be unable to log in.
 	///
 	/// default: "hybrid"
-	pub compatibility_mode: OAuthMode,
+	compatibility_mode: OAuthMode,
+
+	/// display: hidden
+	pub oidc: Option<OidcConfig>,
 }
 
-#[derive(Clone, Debug, Default, Deserialize)]
+impl OauthConfig {
+	#[must_use]
+	pub fn compatibility_mode(&self) -> OAuthMode {
+		if self.oidc.is_some() {
+			OAuthMode::Exclusive
+		} else {
+			self.compatibility_mode
+		}
+	}
+}
+
+#[derive(Clone, Copy, Debug, Default, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum OAuthMode {
 	Disabled,
@@ -2424,6 +2455,121 @@ pub fn uiaa_available(&self) -> bool { matches!(self, Self::Disabled | Self::Hyb
 	pub fn oauth_available(&self) -> bool { matches!(self, Self::Hybrid | Self::Exclusive) }
 }
 
+#[derive(Clone, Debug, Deserialize)]
+#[config_example_generator(
+	filename = "conduwuit-example.toml",
+	section = "global.oauth.oidc",
+	optional = "true",
+	subheader = "\
+# Uncommenting this section will enable Continuwuity's support for
+# authenticating users using an OpenID Connect-compatible identity provider.
+# This is referred to as \"delegated authentication\".
+#
+# IMPORTANT NOTE: When delegated authentication is active, Continuwuity will behave as if
+# the `global.oauth.compatibility_mode` setting is set to `exclusive`.
+# Matrix clients which do not support OAuth login (also referred to as \"next-gen auth\") will \
+	             NOT be able
+# to log in while delegated authentication is active."
+)]
+pub struct OidcConfig {
+	/// The OIDC issuer URL. Continuwuity will use OpenID Connect Discovery to
+	/// automatically fetch the identity provider's metadata from this URL.
+	/// Generally you should set this to the base domain your identity provider
+	/// runs on.
+	pub discovery_url: Url,
+
+	/// The OAuth client ID for Continuwuity to use when communicating with the
+	/// identity provider.
+	pub client_id: ClientId,
+
+	/// The OAuth client secret for Continuwuity to use when communicating with
+	/// the identity provider.
+	pub client_secret: Option<ClientSecret>,
+
+	/// A path to a file which Continuwuity will read the client secret from.
+	/// If this option is set, it will override `client_secret`.
+	///
+	/// The server will fail to start if the file cannot be read.
+	pub client_secret_file: Option<PathBuf>,
+
+	/// Additional scopes Continuwuity should request from the IDP. This may be
+	/// necessary to access certain claims. Continuwuity always requests the
+	/// `openid` scope.
+	///
+	/// default: []
+	#[serde(default)]
+	pub additional_scopes: Vec<Scope>,
+
+	/// Whether the user should be prompted to choose a localpart
+	/// when signing in for the first time. If this is `false`, Continuwuity
+	/// will attempt to use the value of the `preferred_username_claim`
+	/// (see below) as the user's localpart. Authentication will
+	/// fail if this claim is missing or is not a valid localpart.
+	///
+	/// default: true
+	#[serde(default = "true_fn")]
+	pub prompt_for_localpart: bool,
+
+	/// The claim to use for the user's localpart, if `prompt_for_localpart` is
+	/// false.
+	///
+	/// default: "preferred_username"
+	#[serde(default = "default_preferred_username_claim")]
+	pub preferred_username_claim: String,
+
+	/// The claim which will be used to set the user's email address,
+	/// either on initial registration or on every login depending on
+	/// the value of `profile_key_import_mode`. Continuwuity assumes that
+	/// the IDP has taken care of verifying that the user controls the email
+	/// address it provides.
+	///
+	/// This option does nothing if SMTP is not configured.
+	///
+	/// If this option is set, and `profile_key_import_mode` is `on_login`,
+	/// users will not be able to change their email addresses themselves.
+	///
+	/// default: "email"
+	pub email_claim: Option<String>,
+
+	/// Defines how claims returned from the IDP should be mapped to a user's
+	/// profile data. The profile field named in each key will be set from the
+	/// claim named in the corresponding value when the user first registers,
+	/// and possibly on subsequent logins as well, depending on the value of
+	/// `profile_key_import_mode` (see below).
+	///
+	/// Per-room overrides to the user's display name or avatar will be
+	/// preserved by the import process.
+	///
+	/// SECURITY NOTE: If the `avatar_url` field is set, Continuwuity will
+	/// perform a HTTP GET to the URL in the mapped claim and use the returned
+	/// file as the user's profile picture. Make sure your users are not able
+	/// to set the value of the mapped claim to an arbitrary URL.
+	///
+	/// default: { displayname = "name" }
+	#[serde(default = "default_profile_key_map")]
+	pub profile_key_map: HashMap<String, String>,
+
+	/// When profile keys should be imported from the IDP's claims.
+	///
+	/// - "on_registration": Listed keys will be imported once, when the user
+	///   logs in for the first time and their shadow account is created.
+	/// - "on_login": Listed keys will be imported every time the user logs in.
+	///   Additionally, users will not be able to manually edit any listed keys
+	///   through their Matrix client.
+	///
+	/// default: "on_registration"
+	#[serde(default)]
+	pub profile_key_import_mode: OidcProfileKeyImportMode,
+}
+
+#[derive(Clone, Debug, Deserialize, Default)]
+#[serde(rename_all = "snake_case")]
+pub enum OidcProfileKeyImportMode {
+	#[default]
+	OnRegistration,
+	OnLogin,
+}
+
 const DEPRECATED_KEYS: &[&str] = &[
 	"cache_capacity",
 	"conduit_cache_capacity_modifier",
@@ -2615,7 +2761,7 @@ fn default_pusher_timeout() -> u64 { 60 }
 
 fn default_pusher_idle_timeout() -> u64 { 15 }
 
-fn default_max_fetch_prev_events() -> u16 { 192_u16 }
+fn default_max_fetch_prev_events() -> u16 { 1024 }
 
 fn default_max_concurrent_inbound_transactions() -> usize { 150 }
 
@@ -2718,6 +2864,8 @@ fn default_rocksdb_stats_level() -> u8 { 1 }
 #[inline]
 pub fn default_default_room_version() -> RoomVersionId { RoomVersionId::V12 }
 
+fn default_extremity_threshold() -> u8 { 10 }
+
 fn default_ip_range_denylist() -> Vec<String> {
 	vec![
 		"127.0.0.0/8".to_owned(),
@@ -2806,3 +2954,9 @@ fn default_client_shutdown_timeout() -> u64 { 15 }
 fn default_sender_shutdown_timeout() -> u64 { 5 }
 
 fn default_terms_language() -> String { "en".to_owned() }
+
+fn default_preferred_username_claim() -> String { "preferred_username".to_owned() }
+
+fn default_profile_key_map() -> HashMap<String, String> {
+	HashMap::from_iter([(ProfileFieldName::DisplayName.to_string(), "name".to_owned())])
+}

src/core/matrix/pdu/partial.rs

@@ -5,7 +5,10 @@
 	events::{MessageLikeEventContent, StateEventContent, TimelineEventType},
 };
 use serde::Deserialize;
-use serde_json::value::{RawValue as RawJsonValue, to_raw_value};
+use serde_json::{
+	json,
+	value::{RawValue as RawJsonValue, to_raw_value},
+};
 
 use super::StateKey;
 
@@ -62,7 +65,7 @@ impl Default for PartialPdu {
 	fn default() -> Self {
 		Self {
 			event_type: "m.room.message".into(),
-			content: Box::<RawJsonValue>::default(),
+			content: to_raw_value(&json!({})).unwrap(),
 			unsigned: None,
 			state_key: None,
 			redacts: None,

src/core/matrix/versions.rs

@@ -23,28 +23,23 @@ pub fn versions() -> Vec<String> {
 		"v1.14".to_owned(),
 		"v1.15".to_owned(),
 		"v1.16".to_owned(),
+		"v1.17".to_owned(),
+		"v1.18".to_owned(),
 	]
 }
 
 #[must_use]
 pub fn unstable_features() -> BTreeMap<String, bool> {
 	BTreeMap::from_iter([
-		("org.matrix.e2e_cross_signing".to_owned(), true),
-		("org.matrix.msc2285.stable".to_owned(), true), /* private read receipts (https://github.com/matrix-org/matrix-spec-proposals/pull/2285) */
-		("uk.half-shot.msc2666.query_mutual_rooms".to_owned(), true), /* query mutual rooms (https://github.com/matrix-org/matrix-spec-proposals/pull/2666) */
-		("org.matrix.msc2836".to_owned(), true), /* threading/threads (https://github.com/matrix-org/matrix-spec-proposals/pull/2836) */
-		("org.matrix.msc2946".to_owned(), true), /* spaces/hierarchy summaries (https://github.com/matrix-org/matrix-spec-proposals/pull/2946) */
-		("org.matrix.msc3026.busy_presence".to_owned(), true), /* busy presence status (https://github.com/matrix-org/matrix-spec-proposals/pull/3026) */
-		("org.matrix.msc3827".to_owned(), true), /* filtering of /publicRooms by room type (https://github.com/matrix-org/matrix-spec-proposals/pull/3827) */
-		("org.matrix.msc3952_intentional_mentions".to_owned(), true), /* intentional mentions (https://github.com/matrix-org/matrix-spec-proposals/pull/3952) */
-		("org.matrix.msc3916.stable".to_owned(), true), /* authenticated media (https://github.com/matrix-org/matrix-spec-proposals/pull/3916) */
-		("org.matrix.msc4180".to_owned(), true), /* stable flag for 3916 (https://github.com/matrix-org/matrix-spec-proposals/pull/4180) */
-		("uk.tcpip.msc4133".to_owned(), true), /* Extending User Profile API with Key:Value Pairs (https://github.com/matrix-org/matrix-spec-proposals/pull/4133) */
-		("us.cloke.msc4175".to_owned(), true), /* Profile field for user time zone (https://github.com/matrix-org/matrix-spec-proposals/pull/4175) */
-		("org.matrix.simplified_msc3575".to_owned(), true), /* Simplified Sliding sync (https://github.com/matrix-org/matrix-spec-proposals/pull/4186) */
-		("uk.timedout.msc4323".to_owned(), true), /* agnostic suspend (https://github.com/matrix-org/matrix-spec-proposals/pull/4323) */
-		("org.matrix.msc4155".to_owned(), true), /* invite filtering (https://github.com/matrix-org/matrix-spec-proposals/pull/4155) */
-		("computer.gingershaped.msc4466".to_owned(), true), /* profile change propagation (https://github.com/matrix-org/matrix-spec-proposals/pull/4466) */
-		("org.matrix.msc4380.stable".to_owned(), true),
+		// query mutual rooms (https://github.com/matrix-org/matrix-spec-proposals/pull/2666)
+		// Expected for spec v1.19
+		("uk.half-shot.msc2666.query_mutual_rooms.stable".to_owned(), true),
+		// Simplified Sliding sync (https://github.com/matrix-org/matrix-spec-proposals/pull/4186)
+		// Expected for spec v1.19
+		("org.matrix.simplified_msc3575".to_owned(), true),
+		// invite filtering (https://github.com/matrix-org/matrix-spec-proposals/pull/4155)
+		("org.matrix.msc4155".to_owned(), true),
+		// profile change propagation (https://github.com/matrix-org/matrix-spec-proposals/pull/4466)
+		("computer.gingershaped.msc4466".to_owned(), true),
 	])
 }

src/core/utils/time.rs

@@ -1,7 +1,10 @@
 pub mod exponential_backoff;
+pub mod jitter;
 
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
+pub use jitter::jitter;
+
 use crate::{Result, err};
 
 #[inline]

src/core/utils/time/jitter.rs

@@ -0,0 +1,16 @@
+use std::{ops::RangeInclusive, time::Duration};
+
+/// Returns a `Duration` that is jittered by a random percentage of the base
+/// duration. The jitter is applied as a random percentage in the range of
+/// `-jitter_range` to `jitter_range`.
+///
+/// # Example
+/// ```
+/// use conduwuit_core::utils::time::jitter;
+/// let sleep_duration = jitter(Duration::from_secs(1), -10..=10);
+/// // Adds a jitter of between -10% and 10% to the duration.
+/// ```
+#[must_use]
+pub fn jitter(base: Duration, jitter_range: RangeInclusive<f64>) -> Duration {
+	base.mul_f64(1.0 + (rand::random_range(jitter_range)) / 100.0)
+}

src/database/maps.rs

@@ -124,6 +124,14 @@ pub(super) fn open_list(db: &Arc<Engine>, maps: &[Descriptor]) -> Result<Maps> {
 		name: "onetimekeyid_onetimekeys",
 		..descriptor::RANDOM_SMALL
 	},
+	Descriptor {
+		name: "openidsubject_localpart",
+		..descriptor::RANDOM_SMALL
+	},
+	Descriptor {
+		name: "openidsubject_currentpictureurl",
+		..descriptor::RANDOM_SMALL
+	},
 	Descriptor {
 		name: "fallbackkeyid_fallbackkey",
 		..descriptor::RANDOM_SMALL
@@ -169,6 +177,10 @@ pub(super) fn open_list(db: &Arc<Engine>, maps: &[Descriptor]) -> Result<Maps> {
 		name: "registrationtoken_info",
 		..descriptor::RANDOM_SMALL
 	},
+	Descriptor {
+		name: "remoteuserid_remoteuser",
+		..descriptor::RANDOM
+	},
 	Descriptor {
 		name: "roomid_invitedcount",
 		..descriptor::RANDOM_SMALL
@@ -195,6 +207,10 @@ pub(super) fn open_list(db: &Arc<Engine>, maps: &[Descriptor]) -> Result<Maps> {
 		val_size_hint: Some(8),
 		..descriptor::RANDOM_SMALL
 	},
+	Descriptor {
+		name: "roomid_mindepth",
+		..descriptor::RANDOM_SMALL
+	},
 	Descriptor {
 		name: "roomserverids",
 		..descriptor::RANDOM_SMALL
@@ -394,6 +410,10 @@ pub(super) fn open_list(db: &Arc<Engine>, maps: &[Descriptor]) -> Result<Maps> {
 		name: "userid_blurhash",
 		..descriptor::DROPPED
 	},
+	Descriptor {
+		name: "userid_deactivated",
+		..descriptor::RANDOM_SMALL
+	},
 	Descriptor {
 		name: "userid_dehydrateddevice",
 		..descriptor::RANDOM_SMALL

src/macros/config.rs

@@ -81,6 +81,12 @@ fn generate_example(input: &ItemStruct, args: &[Meta], write: bool) -> Result<To
 		};
 		file.write_fmt(format_args!("{section_header}"))
 			.expect("written to config file");
+
+		if let Some(subheader) = settings.get("subheader") {
+			file.write_all(subheader.as_bytes())
+				.expect("written to config file");
+			file.write_all(b"\n\n").expect("written to config file");
+		}
 	}
 
 	let mut summary: Vec<TokenStream2> = Vec::new();

src/service/Cargo.toml

@@ -120,6 +120,7 @@ reqwest_recaptcha = { package = "reqwest", version = "0.12.28", default-features
 yansi.workspace = true
 lettre.workspace = true
 serde_urlencoded.workspace = true
+openidconnect.workspace = true
 
 [target.'cfg(all(unix, target_os = "linux"))'.dependencies]
 sd-notify.workspace = true

src/service/admin/create.rs

@@ -48,7 +48,7 @@ pub async fn create_admin_room(services: &Services) -> Result {
 
 	// Create a user for the server
 	let server_user = services.globals.server_user.as_ref();
-	services.users.create(server_user, None)?;
+	services.users.create_shadow_account(server_user).await?;
 
 	let mut create_content = if room_version_rules.authorization.use_room_create_sender {
 		RoomCreateEventContent::new_v1(server_user.into())

src/service/announcements/mod.rs

@@ -18,7 +18,11 @@
 use std::{sync::Arc, time::Duration};
 
 use async_trait::async_trait;
-use conduwuit::{Result, Server, debug, error, utils::response::LimitReadExt, warn};
+use conduwuit::{
+	Result, Server, debug, error,
+	utils::{response::LimitReadExt, time::jitter},
+	warn,
+};
 use database::{Deserialized, Map};
 use ruma::events::{Mentions, room::message::RoomMessageEventContent};
 use serde::Deserialize;
@@ -98,10 +102,7 @@ async fn worker(self: Arc<Self>) -> Result<()> {
 				.ok();
 		}
 
-		let first_check_jitter = {
-			let jitter_percent = rand::random_range(-50.0..=10.0);
-			self.interval.mul_f64(1.0 + jitter_percent / 100.0)
-		};
+		let first_check_jitter = jitter(self.interval, -50.0..=10.0);
 
 		let mut i = interval(self.interval);
 		i.set_missed_tick_behavior(MissedTickBehavior::Delay);

src/service/appservice/mod.rs

@@ -108,17 +108,17 @@ async fn start_appservice(&self, id: String, registration: Registration) -> Resu
 			self.services.globals.server_name(),
 		)?;
 
-		if !self.services.users.exists(&appservice_user_id).await {
-			self.services.users.create(&appservice_user_id, None)?;
-		} else if self
+		if !self
 			.services
 			.users
-			.is_deactivated(&appservice_user_id)
+			.status(&appservice_user_id)
 			.await
-			.unwrap_or(false)
+			.is_found()
 		{
-			// Reactivate the appservice user if it was accidentally deactivated
-			self.services.users.set_password(&appservice_user_id, None);
+			self.services
+				.users
+				.create_shadow_account(&appservice_user_id)
+				.await?;
 		}
 
 		self.registration_info

src/service/emergency/mod.rs

@@ -54,15 +54,22 @@ impl Service {
 	async fn set_emergency_access(&self) -> Result {
 		let server_user = &self.services.globals.server_user;
 
-		self.services.users.set_password(
-			server_user,
-			self.services
-				.config
-				.emergency_password
-				.as_deref()
-				.map(HashedPassword::new)
-				.transpose()?,
-		);
+		match &self.services.config.emergency_password {
+			| Some(emergency_password) => {
+				let emergency_password = HashedPassword::new(emergency_password)?;
+
+				self.services
+					.users
+					.convert_to_local_account(server_user, emergency_password)
+					.await?;
+			},
+			| None => {
+				self.services
+					.users
+					.convert_to_shadow_account(server_user)
+					.await?;
+			},
+		}
 
 		let (ruleset, pwd_set) = match self.services.config.emergency_password {
 			| Some(_) => (Ruleset::server_default(server_user), true),

src/service/media/preview.rs

@@ -211,7 +211,6 @@ pub async fn download_audio(
 		Ok(preview_data)
 	}
 
-	#[cfg(feature = "url_preview")]
 	pub async fn download_media(&self, url: &str) -> Result<(OwnedMxcUri, usize)> {
 		use conduwuit::utils::random_string;
 		use http::header::CONTENT_TYPE;
@@ -268,11 +267,6 @@ pub async fn download_audio(
 		Err!(FeatureDisabled("url_preview"))
 	}
 
-	#[cfg(not(feature = "url_preview"))]
-	pub async fn download_media(&self, _url: &str) -> Result<UrlPreviewData> {
-		Err!(FeatureDisabled("url_preview"))
-	}
-
 	#[cfg(feature = "url_preview")]
 	async fn download_html(&self, url: &str) -> Result<UrlPreviewData> {
 		use webpage::HTML;

src/service/migrations.rs

@@ -41,7 +41,7 @@ pub(crate) async fn migrations(services: &Services) -> Result<()> {
 	// requires recreating the database from scratch.
 	if users_count > 0 {
 		let server_user = &services.globals.server_user;
-		if !services.users.exists(server_user).await {
+		if !services.users.status(server_user).await.is_found() {
 			error!("The {server_user} server user does not exist, and the database is not new.");
 			return Err!(Database(
 				"Cannot reuse an existing database after changing the server name, please \
@@ -228,6 +228,18 @@ async fn migrate(services: &Services) -> Result<()> {
 			.map_err(|e| err!("Failed to run 'fix_local_invite_state' migration': {e}"))?;
 	}
 
+	if services.globals.db.database_version().await < 18 {
+		services.globals.db.bump_database_version(18);
+		info!("Migration: Bumped database version to 18");
+	}
+
+	if db["global"].get(SPLIT_USERID_PASSWORD).await.is_not_found() {
+		info!("Running migration 'split_userid_password'");
+		split_userid_password(services)
+			.await
+			.map_err(|e| err!("Failed to run 'split_userid_password' migration': {e}"))?;
+	}
+
 	assert_eq!(
 		services.globals.db.database_version().await,
 		DATABASE_VERSION,
@@ -242,9 +254,9 @@ async fn migrate(services: &Services) -> Result<()> {
 		if !patterns.is_empty() {
 			services
 				.users
-				.stream()
+				.stream_local_users()
 				.filter_map(async |user_id| {
-					if services.users.is_active_local(&user_id).await {
+					if services.users.status(&user_id).await.is_found() {
 						Some(user_id)
 					} else {
 						None
@@ -774,7 +786,7 @@ async fn fix_local_invite_state(services: &Services) -> Result {
 
 	let db = &services.db;
 	let cork = db.cork_and_sync();
-	let userroomid_invitestate = services.db["userroomid_invitestate"].clone();
+	let userroomid_invitestate = db["userroomid_invitestate"].clone();
 
 	// for each user invited to a room
 	let fixed =  userroomid_invitestate.stream()
@@ -818,3 +830,44 @@ async fn fix_local_invite_state(services: &Services) -> Result {
 	db.db.sort()?;
 	Ok(())
 }
+
+const SPLIT_USERID_PASSWORD: &str = "split_userid_password";
+async fn split_userid_password(services: &Services) -> Result {
+	// Split remote and deactivated users out from the `userid_password` table
+
+	let db = &services.db;
+	let cork = db.cork_and_sync();
+	let userid_password = db["userid_password"].clone();
+	let remoteuserid_remoteuser = db["remoteuserid_remoteuser"].clone();
+	let userid_deactivated = db["userid_deactivated"].clone();
+
+	let remote_users = userid_password
+		.stream::<OwnedUserId, String>()
+		.ignore_err()
+		.fold(0_usize, async |mut remote_users, (user_id, hash)| {
+			if !services.globals.user_is_local(&user_id) {
+				assert!(hash.is_empty(), "non-empty hash {hash} for remote user {user_id}");
+
+				remoteuserid_remoteuser.insert(&user_id, "");
+				userid_password.remove(&user_id);
+				remote_users = remote_users.saturating_add(1);
+			} else if hash.is_empty() {
+				if !(services.appservice.is_exclusive_user_id(&user_id).await
+					|| user_id == services.globals.server_user)
+				{
+					info!("Marking {user_id} as deactivated");
+					userid_deactivated.insert(&user_id, "");
+				}
+			}
+
+			remote_users
+		})
+		.await;
+
+	drop(cork);
+	info!(?remote_users, "Split userid_password.");
+
+	db["global"].insert(FIXED_LOCAL_INVITE_STATE_MARKER, []);
+	db.db.sort()?;
+	Ok(())
+}

src/service/mod.rs

@@ -28,6 +28,7 @@
 pub mod media;
 pub mod moderation;
 pub mod oauth;
+pub mod oidc;
 pub mod presence;
 pub mod pusher;
 pub mod registration_tokens;

src/service/oauth/grant.rs

@@ -160,7 +160,7 @@ pub enum ErrorCode {
 	InvalidClientMetadata,
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, Deserialize)]
 pub struct AuthorizationCodeResponse {
 	pub state: String,
 	pub code: String,

src/service/oidc/mod.rs

@@ -0,0 +1,484 @@
+use std::{collections::HashMap, str::FromStr, sync::Arc, time::Duration};
+
+use async_trait::async_trait;
+use conduwuit::{
+	Result,
+	config::{OidcConfig, OidcProfileKeyImportMode},
+	debug, err, error, info, warn,
+};
+use database::{Deserialized, Map};
+use lettre::Address;
+use openidconnect::{
+	AdditionalClaims, AuthorizationCode, ClientSecret, CsrfToken, EmptyExtraTokenFields,
+	EndpointMaybeSet, EndpointNotSet, EndpointSet, IdTokenClaims, IdTokenFields, IssuerUrl,
+	Nonce, PkceCodeChallenge, PkceCodeVerifier, RedirectUrl, StandardErrorResponse,
+	StandardTokenResponse, TokenResponse,
+	core::{
+		CoreAuthDisplay, CoreAuthPrompt, CoreAuthenticationFlow, CoreErrorResponseType,
+		CoreGenderClaim, CoreJsonWebKey, CoreJweContentEncryptionAlgorithm,
+		CoreJwsSigningAlgorithm, CoreProviderMetadata, CoreRevocableToken,
+		CoreRevocationErrorResponse, CoreTokenIntrospectionResponse, CoreTokenType,
+	},
+	reqwest,
+};
+use ruma::{
+	OwnedUserId, UserId,
+	api::client::profile::PropagateTo,
+	profile::{ProfileFieldName, ProfileFieldValue},
+};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use tokio::{runtime, sync::SetOnce};
+use url::Url;
+
+use crate::{
+	Dep, config, globals, media,
+	oauth::grant::AuthorizationCodeResponse,
+	threepid,
+	users::{self, AccountStatus, ProfileFieldChange},
+};
+
+pub struct Service {
+	services: Services,
+	runtime: runtime::Handle,
+	db: Data,
+	client: Option<OidcClient>,
+}
+
+struct Data {
+	openidsubject_localpart: Arc<Map>,
+	openidsubject_currentpictureurl: Arc<Map>,
+}
+struct Services {
+	config: Dep<config::Service>,
+	globals: Dep<globals::Service>,
+	media: Dep<media::Service>,
+	threepid: Dep<threepid::Service>,
+	users: Dep<users::Service>,
+}
+
+struct OidcClient {
+	config: OidcConfig,
+	client_secret: ClientSecret,
+	machine: SetOnce<OidcClientMachine>,
+	client: reqwest::Client,
+}
+
+type OidcClientMachine = openidconnect::Client<
+	AllClaims,
+	CoreAuthDisplay,
+	CoreGenderClaim,
+	CoreJweContentEncryptionAlgorithm,
+	CoreJsonWebKey,
+	CoreAuthPrompt,
+	StandardErrorResponse<CoreErrorResponseType>,
+	StandardTokenResponse<
+		IdTokenFields<
+			AllClaims,
+			EmptyExtraTokenFields,
+			CoreGenderClaim,
+			CoreJweContentEncryptionAlgorithm,
+			CoreJwsSigningAlgorithm,
+		>,
+		CoreTokenType,
+	>,
+	CoreTokenIntrospectionResponse,
+	CoreRevocableToken,
+	CoreRevocationErrorResponse,
+	EndpointSet,
+	EndpointNotSet,
+	EndpointNotSet,
+	EndpointNotSet,
+	EndpointMaybeSet,
+	EndpointMaybeSet,
+>;
+
+pub type Claims = IdTokenClaims<AllClaims, CoreGenderClaim>;
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct AllClaims {
+	#[serde(flatten)]
+	pub claims: HashMap<String, Value>,
+}
+
+impl AdditionalClaims for AllClaims {}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct PendingSession {
+	pkce_verifier: PkceCodeVerifier,
+	nonce: Nonce,
+	csrf_token: CsrfToken,
+}
+
+pub enum SessionCompletionStatus {
+	NeedsUserId,
+	Complete(OwnedUserId),
+}
+
+#[async_trait]
+impl crate::Service for Service {
+	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
+		Ok(Arc::new(Self {
+			services: Services {
+				config: args.depend::<config::Service>("config"),
+                globals: args.depend::<globals::Service>("globals"),
+				media: args.depend::<media::Service>("media"),
+                threepid: args.depend::<threepid::Service>("threepid"),
+                users: args.depend::<users::Service>("users"),
+			},
+			runtime: args.server.runtime().clone(),
+            db: Data {
+                openidsubject_localpart: args.db["openidsubject_localpart"].clone(),
+				openidsubject_currentpictureurl: args.db["openidsubject_currentpictureurl"].clone(),
+            },
+            client: args.server.config.oauth.oidc.as_ref().map(|config| -> Result<OidcClient> {
+				Ok(OidcClient {
+                    config: config.clone(),
+					client_secret: if let Some(client_secret_file) = &config.client_secret_file {
+						std::fs::read_to_string(client_secret_file)
+							.map(ClientSecret::new)
+							.map_err(|err| err!("Failed to read OIDC client secret file: {err}"))?
+					} else if let Some(client_secret) = &config.client_secret {
+						client_secret.clone()
+					} else {
+						// The config check function should cause an early exit before this happens
+						panic!("neither client secret or client secret file were set");
+					},
+                    machine: SetOnce::new(),
+                    // This isn't in the client service because it has to use the `reqwest` shipped by `openidconnect`
+                    client: reqwest::ClientBuilder::new()
+                        .connect_timeout(Duration::from_secs(args.server.config.request_conn_timeout))
+                        .read_timeout(Duration::from_secs(args.server.config.request_timeout))
+                        .timeout(Duration::from_secs(args.server.config.request_total_timeout))
+                        .pool_idle_timeout(Duration::from_secs(args.server.config.request_idle_timeout))
+                        .pool_max_idle_per_host(args.server.config.request_idle_per_host.into())
+                        .user_agent(conduwuit::user_agent())
+                        .redirect(reqwest::redirect::Policy::none())
+                        .danger_accept_invalid_certs(args.server.config.allow_invalid_tls_certificates_yes_i_know_what_the_fuck_i_am_doing_with_this_and_i_know_this_is_insecure)
+                        .build()
+                        .expect("client should build")
+                })}
+			).transpose()?,
+		}))
+	}
+
+	async fn worker(self: Arc<Self>) -> Result {
+		if let Some(OidcClient { config, client_secret, machine, client }) = &self.client {
+			let redirect_url = self
+				.services
+				.config
+				.get_client_domain()
+				.join(&format!("{}/oidc/complete", conduwuit::ROUTE_PREFIX))
+				.expect("redirect url should be valid");
+
+			let provider_metadata = CoreProviderMetadata::discover_async(
+				IssuerUrl::from_url(config.discovery_url.clone()),
+				client,
+			)
+			.await
+			.map_err(|err| err!("Failed to discover OIDC provider metadata: {err}"))?;
+
+			machine
+				.set(
+					OidcClientMachine::from_provider_metadata(
+						provider_metadata,
+						config.client_id.clone(),
+						Some(client_secret.clone()),
+					)
+					.set_redirect_uri(RedirectUrl::from_url(redirect_url)),
+				)
+				.expect("machine should be empty");
+		}
+
+		Ok(())
+	}
+
+	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
+}
+
+impl Service {
+	const SERVER_MISCONFIGURED: &str =
+		"Identity server is misconfigured. Contact your homeserver's administrator.";
+
+	pub fn enabled(&self) -> bool { self.client.is_some() }
+
+	pub fn restricted_profile_fields(&self) -> Vec<ProfileFieldName> {
+		if let Some(config) = self.client.as_ref().map(|client| &client.config)
+			&& matches!(config.profile_key_import_mode, OidcProfileKeyImportMode::OnLogin)
+		{
+			config
+				.profile_key_map
+				.keys()
+				.map(|key| ProfileFieldName::from(key.as_str()))
+				.collect()
+		} else {
+			vec![]
+		}
+	}
+
+	pub async fn begin_session(&self, prompt: Option<CoreAuthPrompt>) -> (PendingSession, Url) {
+		let OidcClient { machine, config, .. } =
+			self.client.as_ref().expect("oidc should be configured");
+		let machine = machine.wait().await;
+
+		let (pkce_challenge, pkce_verifier) = PkceCodeChallenge::new_random_sha256();
+
+		let mut auth_url = machine
+			.authorize_url(
+				CoreAuthenticationFlow::AuthorizationCode,
+				CsrfToken::new_random,
+				Nonce::new_random,
+			)
+			.add_scopes(config.additional_scopes.iter().cloned())
+			.set_pkce_challenge(pkce_challenge);
+
+		if let Some(prompt) = prompt {
+			auth_url = auth_url.add_prompt(prompt);
+		}
+
+		let (auth_url, csrf_token, nonce) = auth_url.url();
+
+		(PendingSession { pkce_verifier, nonce, csrf_token }, auth_url)
+	}
+
+	pub async fn exchange_code(
+		&self,
+		session: PendingSession,
+		response: AuthorizationCodeResponse,
+	) -> Result<Claims, &'static str> {
+		let Some(OidcClient { machine, client, .. }) = self.client.as_ref() else {
+			return Err("Delegated authentication is not enabled on this server.");
+		};
+
+		let machine = machine.wait().await;
+
+		if session.csrf_token.into_secret() != response.state {
+			return Err("State mismatch.");
+		}
+
+		let token_response = machine
+			.exchange_code(AuthorizationCode::new(response.code))
+			.expect("machine should be configured correctly")
+			.set_pkce_verifier(session.pkce_verifier)
+			.request_async(client)
+			.await
+			.map_err(|err| {
+				error!("Failed to exchange OIDC authorization code: {err}");
+				"Code exchange failed."
+			})?;
+
+		let Some(id_token) = token_response.id_token() else {
+			error!("Identity server did not return an id token");
+			return Err(Self::SERVER_MISCONFIGURED);
+		};
+
+		let claims = id_token
+			.claims(&machine.id_token_verifier(), &session.nonce)
+			.map_err(|err| {
+				error!("Failed to verify id token claims: {err}");
+				Self::SERVER_MISCONFIGURED
+			})?
+			.to_owned();
+
+		info!(subject = claims.subject().as_str(), "Authenticated subject");
+
+		Ok(claims)
+	}
+
+	#[tracing::instrument(skip(self, claims), fields(subject = claims.subject().to_string()))]
+	pub async fn complete_session(
+		&self,
+		claims: &Claims,
+		supplied_user_id: Option<OwnedUserId>,
+	) -> Result<SessionCompletionStatus, &'static str> {
+		let Some(OidcClient { config, .. }) = self.client.as_ref() else {
+			return Err("Delegated authentication is not enabled on this server.");
+		};
+
+		// this is a truly awful hack but we really need all the claims in a map
+		let all_claims = serde_json::to_value(claims)
+			.expect("should be able to serialize claims")
+			.as_object()
+			.expect("claims should be an object")
+			.to_owned();
+
+		debug!(?all_claims);
+
+		let subject = claims.subject().as_str();
+
+		let user_id = if let Ok(localpart) = self
+			.db
+			.openidsubject_localpart
+			.get(subject)
+			.await
+			.deserialized::<String>()
+		{
+			UserId::parse(format!("@{localpart}:{}", self.services.globals.server_name()))
+				.expect("saved localpart should be valid")
+		} else if config.prompt_for_localpart {
+			if let Some(supplied_user_id) = supplied_user_id {
+				supplied_user_id
+			} else {
+				return Ok(SessionCompletionStatus::NeedsUserId);
+			}
+		} else if let Some(preferred_username) = all_claims
+			.get(&config.preferred_username_claim)
+			.and_then(|claim| claim.as_str())
+		{
+			self.services
+				.users
+				.determine_registration_user_id(Some(preferred_username.to_owned()), None, None)
+				.await
+				.map_err(|err| {
+					error!("Preferred username claim is not a valid localpart: {err}");
+					"Your preferred username is not a valid Matrix user ID localpart. Contact \
+					 your homeserver's administrator."
+				})?
+		} else {
+			error!("Preferred username claim was not present or was not a string");
+			return Err(Self::SERVER_MISCONFIGURED);
+		};
+
+		info!(?subject, ?user_id, "User {user_id} successfully authorized with OIDC");
+
+		// Create a shadow account for the user if necessary
+		let new_account_registered = match self.services.users.status(&user_id).await {
+			| AccountStatus::Active => {
+				// Do nothing, an account already exists
+				false
+			},
+			| AccountStatus::NotFound => {
+				// Create a new shadow user
+				self.services
+					.users
+					.create_local_account(&user_id, None, None)
+					.await
+					.map_err(|err| {
+						error!("Failed to create a shadow user for {user_id}: {err}");
+						Self::SERVER_MISCONFIGURED
+					})?;
+
+				info!(?subject, ?user_id, "Shadow user created for {user_id}");
+				true
+			},
+			| AccountStatus::Deactivated => {
+				return Err("Your account has been deactivated.");
+			},
+		};
+
+		self.link_user(&user_id, subject);
+
+		// Import profile fields
+		if matches!(config.profile_key_import_mode, OidcProfileKeyImportMode::OnLogin)
+			|| (matches!(
+				config.profile_key_import_mode,
+				OidcProfileKeyImportMode::OnRegistration
+			) && new_account_registered)
+		{
+			if let Some(email_claim) = &config.email_claim {
+				if let Some(email) = claims.email().map(|email| email.as_str())
+					&& let Ok(address) = Address::from_str(email)
+				{
+					if let Err(err) = self
+						.services
+						.threepid
+						.associate_localpart_email(user_id.localpart(), &address)
+						.await
+					{
+						warn!(?email_claim, ?address, "Failed to associate email address: {err}");
+					}
+				} else {
+					warn!(
+						?email_claim,
+						"Email claim was not present or was not a valid email address"
+					);
+				}
+			}
+
+			let user_id = user_id.clone();
+			let subject = claims.subject().to_string();
+			let profile_key_map = config.profile_key_map.clone();
+			let openidsubject_currentpictureurl = self.db.openidsubject_currentpictureurl.clone();
+			let users = self.services.users.clone();
+			let media = self.services.media.clone();
+
+			let import_task = self.runtime.spawn(async move {
+				for (field, claim) in &profile_key_map {
+					let Some(value) = all_claims.get(claim).cloned() else {
+						warn!(?field, ?claim, "IDP provided no value for this mapped claim");
+						continue;
+					};
+
+					let value = if let Some(picture_url) = value.as_str()
+						&& field == ProfileFieldName::AvatarUrl.as_str()
+						&& openidsubject_currentpictureurl
+							.get(&subject)
+							.await
+							.deserialized::<String>()
+							.ok()
+							.is_none_or(|current_picture| current_picture != picture_url)
+					{
+						match media.download_media(picture_url).await {
+							| Ok((mxc, size)) => {
+								openidsubject_currentpictureurl.insert(&subject, picture_url);
+								info!(?picture_url, ?mxc, ?size, "Downloaded profile picture");
+
+								ProfileFieldValue::AvatarUrl(mxc)
+							},
+							| Err(err) => {
+								warn!(
+									?claim,
+									?picture_url,
+									"Failed to download profile picture: {err}"
+								);
+								continue;
+							},
+						}
+					} else {
+						match ProfileFieldValue::new(field, value.clone()) {
+							| Ok(value) => value,
+							| Err(err) => {
+								warn!(
+									?field,
+									?claim,
+									?value,
+									"Failed to parse claim value for profile field: {err}"
+								);
+								continue;
+							},
+						}
+					};
+
+					if let Err(err) = users
+						.set_profile_field(
+							&user_id,
+							ProfileFieldChange::Set(value),
+							PropagateTo::Unchanged,
+						)
+						.await
+					{
+						warn!(?field, ?claim, "Error while setting profile field: {err}");
+					}
+				}
+
+				info!("Profile import complete");
+			});
+
+			// Only wait for import to complete if this is a new account,
+			// so they see the correct profile information in the account panel
+			if new_account_registered {
+				let _ = import_task.await;
+			}
+		}
+
+		Ok(SessionCompletionStatus::Complete(user_id))
+	}
+
+	pub fn link_user(&self, user_id: &UserId, subject: &str) {
+		self.db
+			.openidsubject_localpart
+			.insert(subject, user_id.localpart());
+	}
+
+	pub fn unlink_user(&self, subject: &str) { self.db.openidsubject_localpart.remove(subject); }
+}

src/service/rooms/event_handler/fetch_and_handle_outliers.rs

@@ -1,235 +1,661 @@
 use std::{
-	collections::{BTreeMap, HashSet, VecDeque, hash_map},
+	collections::{HashMap, HashSet, VecDeque},
 	time::Instant,
 };
 
+use assign::assign;
+#[cfg(debug_assertions)]
+use conduwuit::error;
 use conduwuit::{
-	Event, PduEvent, debug, debug_warn, matrix::event::gen_event_id_canonical_json, trace,
-	utils::continue_exponential_backoff_secs, warn,
+	Err, Event, PduEvent, debug, debug_error, debug_info, debug_warn, err,
+	result::FlatOk,
+	state_res::lexicographical_topological_sort,
+	trace,
+	utils::{IterStream, math::Expected, stream::BroadbandExt},
+	warn,
 };
+use futures::{StreamExt, future::select_ok};
 use ruma::{
-	CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName,
-	api::federation::event::get_event,
+	CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId,
+	OwnedServerName, RoomId, ServerName, UInt,
+	api::federation::event::{get_event, get_missing_events},
+	int,
+	room_version_rules::RoomVersionRules,
 };
 
 use super::get_room_version_rules;
+use crate::rooms::event_handler::parse_incoming_pdu::expect_event_id_array;
+
+pub const GET_MISSING_EVENTS_MAX_BATCH_SIZE: usize = 50;
+
+/// Attempts to build a localised directed acyclic graph out of the given PDUs,
+/// returning them in a topologically sorted order.
+///
+/// This is used to attempt to process PDUs in an order that respects their
+/// dependencies, however it is ultimately the sender's responsibility to send
+/// them in a processable order, so this is just a best effort attempt. It does
+/// not account for power levels or other tie breaks.
+#[allow(clippy::implicit_hasher)]
+pub async fn build_local_dag(
+	pdu_map: &HashMap<OwnedEventId, &CanonicalJsonObject>,
+) -> conduwuit::Result<Vec<OwnedEventId>> {
+	debug_assert!(pdu_map.len() >= 2, "needless call to build_local_dag with less than 2 PDUs");
+	let mut dag: HashMap<OwnedEventId, HashSet<OwnedEventId>> =
+		HashMap::with_capacity(pdu_map.len());
+	let mut id_origin_ts: HashMap<OwnedEventId, _> = HashMap::with_capacity(pdu_map.len());
+
+	for (event_id, value) in pdu_map {
+		// Parse all prev events as event IDs - if they are missing, return an error (we
+		// can't sanely continue in this case), otherwise skip invalid prev events.
+		let prev_events = value
+			.get("prev_events")
+			.and_then(CanonicalJsonValue::as_array)
+			.ok_or_else(|| {
+				err!(Request(BadJson("event JSON for {event_id} is missing prev_events")))
+			})?
+			.iter()
+			.map(|v| v.as_str().and_then(|s| EventId::parse(s).ok()))
+			.filter(|id| id.as_ref().is_some_and(|id| pdu_map.contains_key(id)))
+			.map(Option::unwrap)
+			.collect();
+
+		dag.insert(event_id.clone(), prev_events);
+		let origin_server_ts = value
+			.get("origin_server_ts")
+			.and_then(CanonicalJsonValue::as_integer)
+			.map(i64::from)
+			.map(UInt::try_from)
+			.flat_ok()
+			.unwrap_or_default();
+		id_origin_ts.insert(event_id.clone(), origin_server_ts);
+	}
+
+	debug!(count = dag.len(), "Sorting incoming events with partial graph");
+	lexicographical_topological_sort(&dag, &async |node_id| {
+		// Note: we don't bother fetching power levels because that would massively slow
+		// this function down. This is a best-effort attempt to order events correctly
+		// for processing, however ultimately that should be the sender's job.
+		let ts = id_origin_ts.get(&node_id).copied().unwrap_or_default();
+		Ok((int!(0), MilliSecondsSinceUnixEpoch(ts)))
+	})
+	.await
+	.inspect(|sorted| {
+		debug_assert_eq!(
+			sorted.len(),
+			pdu_map.len(),
+			"Sorted graph was not the same size as the input graph"
+		);
+	})
+	.map_err(|e| err!("failed to resolve local graph: {e}"))
+}
 
 impl super::Service {
-	/// Find the event and auth it. Once the event is validated (steps 1 - 8)
-	/// it is appended to the outliers Tree.
+	/// Uses `POST /_matrix/federation/v1/get_missing_events/{room_id}` to fill
+	/// gaps in the DAG.
 	///
-	/// Returns pdu and if we fetched it over federation the raw json.
+	/// This function walks backwards from `head`, fetching incrementally (by a
+	/// factor of 10) more events until the remote we're fetching from either
+	/// stops returning new events, or the min_depth is reached.
 	///
-	/// a. Look in the main timeline (pduid_pdu tree)
-	/// b. Look at outlier pdu tree
-	/// c. Ask origin server over federation
-	/// d. TODO: Ask other servers over federation?
-	pub(super) async fn fetch_and_handle_outliers<'a, Pdu, Events>(
+	/// This function does not persist the events, but does validate them. The
+	/// caller is responsible for passing them through handle_incoming_pdu or
+	/// related functions.
+	///
+	/// Only the one `via` is asked for missing events, as multiplexing remotes
+	/// may result in the event tree being walked in a gappy or disordered
+	/// manner.
+	///
+	/// ## Parameters
+	///
+	/// - `room_id`: The room's ID.
+	/// - `head`: The event we are potentially missing prev_events for.
+	/// - `tail`: The most recently known events in the graph (typically forward
+	///   extremities).
+	/// - `via`: The server to ask for missing events.
+	/// - `min_depth`: Don't process events with a `depth` lower than this
+	///   value. Not massively useful, but can help short-circuit infinite loops
+	///   and weird edge paths.
+	#[tracing::instrument(name = "get_missing_events_bulk", skip_all)]
+	pub async fn get_missing_events(
 		&self,
-		origin: &'a ServerName,
-		events: Events,
-		create_event: &'a Pdu,
-		room_id: &'a RoomId,
-	) -> Vec<(PduEvent, Option<BTreeMap<String, CanonicalJsonValue>>)>
-	where
-		Pdu: Event + Send + Sync,
-		Events: Iterator<Item = &'a EventId> + Clone + Send,
-	{
-		let back_off = |id| match self
-			.services
-			.globals
-			.bad_event_ratelimiter
-			.write()
-			.entry(id)
+		room_id: &RoomId,
+		head: &PduEvent,
+		tail: Vec<OwnedEventId>,
+		via: &ServerName,
+		min_depth: UInt,
+	) -> conduwuit::Result<HashMap<OwnedEventId, PduEvent>> {
+		let start = Instant::now();
+		#[cfg(debug_assertions)]
 		{
-			| hash_map::Entry::Vacant(e) => {
-				e.insert((Instant::now(), 1));
-			},
-			| hash_map::Entry::Occupied(mut e) => {
-				*e.get_mut() = (Instant::now(), e.get().1.saturating_add(1));
-			},
-		};
-
-		let mut events_with_auth_events = Vec::with_capacity(events.clone().count());
-		trace!("Fetching {} outlier pdus", events.clone().count());
-
-		for id in events {
-			// a. Look in the main timeline (pduid_pdu tree)
-			// b. Look at outlier pdu tree
-			// (get_pdu_json checks both)
-			if let Ok(local_pdu) = self.services.timeline.get_pdu(id).await {
-				trace!("Found {id} in main timeline or outlier tree");
-				events_with_auth_events.push((id.to_owned(), Some(local_pdu), vec![]));
-				continue;
-			}
-
-			// c. Ask origin server over federation
-			// We also handle its auth chain here so we don't get a stack overflow in
-			// handle_outlier_pdu.
-			let mut todo_auth_events: VecDeque<_> = [id.to_owned()].into();
-			let mut events_in_reverse_order = Vec::with_capacity(todo_auth_events.len());
-
-			let mut events_all = HashSet::with_capacity(todo_auth_events.len());
-			while let Some(next_id) = todo_auth_events.pop_front() {
-				if let Some((time, tries)) = self
-					.services
-					.globals
-					.bad_event_ratelimiter
-					.read()
-					.get(&*next_id)
-				{
-					// Exponential backoff
-					const MIN_DURATION: u64 = 60 * 2;
-					const MAX_DURATION: u64 = 60 * 60 * 8;
-					if continue_exponential_backoff_secs(
-						MIN_DURATION,
-						MAX_DURATION,
-						time.elapsed(),
-						*tries,
-					) {
-						debug_warn!(
-							tried = ?*tries,
-							elapsed = ?time.elapsed(),
-							"Backing off from {next_id}",
-						);
-						continue;
+			let missing_count = head
+				.prev_events()
+				.stream()
+				.fold(0_u8, |i, event_id| async move {
+					if self.services.timeline.pdu_exists(event_id).await {
+						i.expected_add(1)
+					} else {
+						i
 					}
-				}
+				})
+				.await;
+			debug_assert_ne!(
+				missing_count, 0,
+				"event passed to get_missing_events is not missing any events (wasteful call)"
+			);
+		};
+		assert!(!tail.is_empty(), "empty tail");
+		assert_ne!(via, self.services.globals.server_name(), "cannot ask ourselves for events");
 
-				if events_all.contains(&next_id) {
+		// The iteration limit is in place to ensure that if the remote server leaves us
+		// in a state of infinite recursion (as old versions of continuwuity and
+		// predecessors would), we give up. However, get_missing_events doesn't return
+		// that many events per-request. Synapse returns 20, and conduwuit+ return 50.
+		// This means with a hard iteration limit, we might give up too early, before
+		// we get a chance to even come close to max_fetch_prev_events. As such, we'll
+		// calculate the limit based on that config option and the aforementioned
+		// averages.
+		let max_fetch = self.services.server.config.max_fetch_prev_events;
+		let iteration_limit = max_fetch.saturating_div(20).max(10);
+
+		let mut discovered = HashMap::with_capacity(head.prev_events.len());
+		let mut latest_events: Vec<OwnedEventId> = vec![head.event_id().to_owned()];
+		debug!(elapsed=?start.elapsed(),
+			%room_id,
+			event_id=%head.event_id(),
+			%iteration_limit,
+			"Fetching any missing events for head event",
+		);
+		for iteration in 0..iteration_limit {
+			let limit = iteration
+				.expected_add(1)
+				.saturating_mul(10)
+				.min(GET_MISSING_EVENTS_MAX_BATCH_SIZE.try_into().expect(
+					"GET_MISSING_EVENTS_MAX_BATCH_SIZE (usize) should fit in u16 (<=65536)",
+				))
+				.max(
+					// This max call ensures we fetch *at least* all the prev events the
+					// head has.
+					u16::try_from(head.prev_events.len())
+						.expect("cannot have more than 20 prev events, which fits in u16"),
+				);
+			debug_info!(elapsed=?start.elapsed(),
+				%limit,
+				%via,
+				%iteration,
+				%iteration_limit,
+				discovered=discovered.len(),
+				%min_depth,
+				"Attempting to gap fill missing events"
+			);
+			let response: get_missing_events::v1::Response = self
+				.services
+				.sending
+				.send_federation_request(
+					via,
+					assign!(
+						get_missing_events::v1::Request::new(
+							room_id.to_owned(),
+							tail.clone(),
+							latest_events.clone()
+						),
+						{limit: limit.into(), min_depth}
+					),
+				)
+				.await?;
+
+			if response.events.is_empty() {
+				debug_info!(
+					elapsed=?start.elapsed(),
+					%via,
+					"Finished gap filling missing events (remote returned no more events)."
+				);
+				break;
+			}
+			debug_info!(
+				elapsed=?start.elapsed(),
+				"Got {} events back from remote",
+				response.events.len()
+			);
+
+			latest_events.clear();
+			for raw_event in response.events {
+				let (_, event_id, pdu_json) = self.parse_incoming_pdu(&raw_event).await?;
+				let pdu = PduEvent::from_id_val(&event_id, pdu_json).map_err(|e| {
+					err!(Request(BadJson("Failed to parse gapfilled event {event_id}: {e}")))
+				})?;
+				if discovered.contains_key(&event_id) {
+					// We already received this event.
+					trace!("Already received {event_id}");
 					continue;
 				}
-
-				if self.services.timeline.pdu_exists(&next_id).await {
-					trace!("Found {next_id} in db");
-					continue;
-				}
-
-				debug!("Fetching {next_id} over federation from {origin}.");
-				match self
+				if self
 					.services
-					.sending
-					.send_federation_request(
-						origin,
-						get_event::v1::Request::new((*next_id).to_owned()),
-					)
+					.timeline
+					.non_outlier_pdu_exists(&event_id)
 					.await
 				{
-					| Ok(res) => {
-						debug!("Got {next_id} over federation from {origin}");
-						let Ok(room_version_rules) = get_room_version_rules(create_event) else {
-							back_off((*next_id).to_owned());
-							continue;
-						};
-
-						let Ok((calculated_event_id, value)) =
-							gen_event_id_canonical_json(&res.pdu, &room_version_rules)
-						else {
-							back_off((*next_id).to_owned());
-							continue;
-						};
-
-						if calculated_event_id != *next_id {
-							warn!(
-								"Server didn't return event id we requested: requested: \
-								 {next_id}, we got {calculated_event_id}. Event: {:?}",
-								&res.pdu
-							);
-						}
-
-						if let Some(auth_events) = value
-							.get("auth_events")
-							.and_then(CanonicalJsonValue::as_array)
-						{
-							for auth_event in auth_events {
-								match serde_json::from_value::<OwnedEventId>(
-									auth_event.clone().into(),
-								) {
-									| Ok(auth_event) => {
-										trace!(
-											"Found auth event id {auth_event} for event \
-											 {next_id}"
-										);
-										todo_auth_events.push_back(auth_event);
-									},
-									| _ => {
-										warn!("Auth event id is not valid");
-									},
-								}
-							}
-						} else {
-							warn!("Auth event list invalid");
-						}
-
-						events_in_reverse_order.push((next_id.clone(), value));
-						events_all.insert(next_id);
-					},
-					| Err(e) => {
-						warn!("Failed to fetch auth event {next_id} from {origin}: {e}");
-						back_off((*next_id).to_owned());
-					},
+					// NOTE: we explicitly check for *non*-outlier events here, as if we end
+					// up discovering outlier events, we will be able to upgrade them
+					// immediately.
+					trace!("Already have {event_id} as a timeline PDU");
+					continue;
 				}
-			}
 
-			events_with_auth_events.push((id.to_owned(), None, events_in_reverse_order));
-		}
+				if pdu.depth < min_depth {
+					debug_warn!(
+						elapsed=?start.elapsed(),
+						"Received PDU with depth {} below min_depth {}",
+						pdu.depth,
+						min_depth
+					);
+					discovered.insert(event_id.clone(), pdu);
+					continue;
+				}
 
-		let mut pdus = Vec::with_capacity(events_with_auth_events.len());
-		for (id, local_pdu, events_in_reverse_order) in events_with_auth_events {
-			// a. Look in the main timeline (pduid_pdu tree)
-			// b. Look at outlier pdu tree
-			// (get_pdu_json checks both)
-			if let Some(local_pdu) = local_pdu {
-				trace!("Found {id} in main timeline or outlier tree");
-				pdus.push((local_pdu.clone(), None));
-			}
-
-			for (next_id, value) in events_in_reverse_order.into_iter().rev() {
-				if let Some((time, tries)) = self
-					.services
-					.globals
-					.bad_event_ratelimiter
-					.read()
-					.get(&*next_id)
-				{
-					// Exponential backoff
-					const MIN_DURATION: u64 = 5 * 60;
-					const MAX_DURATION: u64 = 60 * 60 * 24;
-					if continue_exponential_backoff_secs(
-						MIN_DURATION,
-						MAX_DURATION,
-						time.elapsed(),
-						*tries,
-					) {
-						debug!("Backing off from {next_id}");
+				for prev_event_id in pdu.prev_events() {
+					if discovered.contains_key(prev_event_id) {
+						// We already received this event.
+						trace!("Already received prev event {prev_event_id}");
 						continue;
 					}
+					if self
+						.services
+						.timeline
+						.non_outlier_pdu_exists(prev_event_id)
+						.await
+					{
+						// NOTE: we explicitly check for *non*-outlier events here, as if we end
+						// up discovering outlier events, we will be able to upgrade them
+						// immediately.
+						trace!("Already have prev event {prev_event_id} as a timeline PDU");
+						continue;
+					}
+					if let Ok(outlier) = self.services.timeline.get_pdu(prev_event_id).await {
+						// We already have this PDU as an outlier, don't ask for
+						// it. However, if we are missing any prev events for it, add it to the
+						// latest events anyway.
+						let outlier_missing_prevs = outlier
+							.prev_events()
+							.stream()
+							.fold(0_u8, |i, event_id| async move {
+								if self.services.timeline.pdu_exists(event_id).await {
+									i.expected_add(1)
+								} else {
+									i
+								}
+							})
+							.await;
+						if outlier_missing_prevs > 0 {
+							trace!("Missing {outlier_missing_prevs} PDU(s) for prev event");
+							latest_events.push(prev_event_id.to_owned());
+						}
+						trace!("Had {prev_event_id} as an outlier already, skipping discovery");
+						discovered.insert(prev_event_id.to_owned(), outlier);
+						continue;
+					}
+					trace!("Missing prev {prev_event_id} of {event_id}");
+					latest_events.push(prev_event_id.to_owned());
+				}
+				trace!("Discovered {event_id}");
+				discovered.insert(event_id.clone(), pdu);
+			}
+
+			if latest_events.is_empty() {
+				debug!(elapsed=?start.elapsed(),
+					%limit,
+					%via,
+					%iteration,
+					discovered=discovered.len(),
+					"No more events to fetch."
+				);
+				break;
+			}
+			if discovered.len() >= self.services.server.config.max_fetch_prev_events.into() {
+				// Stupid hack, debug_error!() drops the log to a DEBUG when not in debug mode,
+				// which is bad because this should at least produce a warning. It's an error in
+				// debug mode because this can be important, but typically not much can be done
+				// about it as a user.
+				#[cfg(debug_assertions)]
+				error!(elapsed=?start.elapsed(),
+					discovered=discovered.len(),
+					max_fetch_prev_events=self.services.server.config.max_fetch_prev_events,
+					%iteration,
+					%iteration_limit,
+					%via,
+					event_id=%head.event_id(),
+					%room_id,
+					"Encountered a gap too large to fill, giving up"
+				);
+				#[cfg(not(debug_assertions))]
+				warn!(elapsed=?start.elapsed(),
+					discovered=discovered.len(),
+					max_fetch_prev_events=self.services.server.config.max_fetch_prev_events,
+					%iteration,
+					%iteration_limit,
+					%via,
+					event_id=%head.event_id(),
+					%room_id,
+					"Encountered a gap too large to fill"
+				);
+				break;
+			}
+		}
+
+		trace!(elapsed=?start.elapsed(), "Finished get_missing_events");
+		Ok(discovered)
+	}
+
+	/// Sends a `GET /_matrix/federation/v1/event/{event_id}` request to the
+	/// target `remote`, parses the resulting PDU, and ensures the remote
+	/// returned the correct event.
+	/// Allows `fetch_and_handle_missing_events` to atomically fetch events from
+	/// multiple remotes in parallel.
+	async fn fetch_event_via(
+		&self,
+		remote: OwnedServerName,
+		event_id: OwnedEventId,
+		room_version_rules: &RoomVersionRules,
+	) -> conduwuit::Result<(OwnedEventId, CanonicalJsonObject)> {
+		let res = self
+			.services
+			.sending
+			.send_federation_request(&remote, get_event::v1::Request::new(event_id.clone()))
+			.await?;
+
+		let (calculated_event_id, value) = self
+			.parse_incoming_pdu_with_known_room(&res.pdu, room_version_rules)
+			.await?;
+
+		if calculated_event_id != event_id {
+			Err!(Request(BadJson(warn!(
+				expected=%event_id,
+				received=%calculated_event_id,
+				"Server didn't return event id we requested",
+			))))
+		} else {
+			Ok((event_id, value))
+		}
+	}
+
+	async fn fetch_event_vias(
+		&self,
+		candidates: impl Iterator<Item = &OwnedServerName>,
+		event_id: &EventId,
+		room_version_rules: &RoomVersionRules,
+	) -> conduwuit::Result<(OwnedEventId, CanonicalJsonObject)> {
+		if let Ok(pdu_json) = self.services.timeline.get_pdu_json(event_id).await {
+			return Ok((event_id.to_owned(), pdu_json));
+		}
+		let futures = candidates
+			.map(|remote| {
+				Box::pin(self.fetch_event_via(
+					remote.to_owned(),
+					event_id.to_owned(),
+					room_version_rules,
+				))
+			})
+			.collect::<Vec<_>>();
+		select_ok(futures).await.map(|(res, _)| res)
+	}
+
+	/// Asks remote servers for any individual events that are missing, also
+	/// known as "atomic fetch". Should only be used for fetching missing auth
+	/// events or resolving missing events from state_ids. For all other uses,
+	/// use get_missing_events.
+	///
+	/// This function manually walks auth_events trees in a breadth-first
+	/// search, and persists all fetched events as outliers when all the
+	/// backwards extremities have been resolved.
+	#[tracing::instrument(name = "get_missing_auth_events_atomic", skip_all)]
+	pub(super) async fn fetch_and_handle_auth_events<Pdu>(
+		&self,
+		origin: &ServerName,
+		events: Vec<OwnedEventId>,
+		create_event: &Pdu,
+		room_id: &RoomId,
+	) -> HashMap<OwnedEventId, PduEvent>
+	where
+		Pdu: Event + Send + Sync,
+	{
+		let start = Instant::now();
+		let room_version_rules =
+			&get_room_version_rules(create_event).unwrap_or(RoomVersionRules::V1);
+		let mut candidates = self
+			.services
+			.timeline
+			.candidate_backfill_servers(room_id)
+			.await;
+		candidates.insert(origin.to_owned());
+		assert!(!candidates.is_empty(), "no candidates to fetch missing events from");
+		let mut discovered_events =
+			HashMap::with_capacity(events.len().saturating_add(events.len().saturating_mul(3)));
+		trace!(
+			elapsed=?start.elapsed(),
+			"Fetching {} unknown PDUs on demand from {} candidates",
+			events.len(),
+			candidates.len()
+		);
+
+		let mut seen: HashMap<OwnedEventId, u8> = HashMap::new();
+		for apex_event_id in &events {
+			let mut todo: VecDeque<OwnedEventId> = [apex_event_id.to_owned()].into();
+
+			while let Some(target_id) = todo.pop_front() {
+				if discovered_events.contains_key(&target_id) {
+					continue;
+				}
+				if let Ok(local_pdu) = self.services.timeline.get_pdu(&target_id).await {
+					trace!(elapsed=?start.elapsed(), "Found {target_id} in db");
+					let mut obj = local_pdu.into_canonical_object();
+					obj.remove("event_id");
+					discovered_events.insert(target_id.clone(), obj);
+					continue;
+				}
+				let attempts = seen.get(&*target_id).copied().unwrap_or_default();
+				if attempts >= 5 {
+					debug_error!(
+						elapsed=?start.elapsed(),
+						%attempts,
+						%target_id,
+						"Could not fetch missing event after 5 attempts, giving up"
+					);
+					continue;
 				}
 
-				trace!("Handling outlier {next_id}");
-				match Box::pin(self.handle_outlier_pdu(
-					origin,
-					create_event,
-					&next_id,
-					room_id,
-					value.clone(),
-					true,
-				))
-				.await
+				debug!(elapsed=?start.elapsed(),"Fetching {target_id} over federation");
+				let value = match self
+					.fetch_event_vias(candidates.iter(), &target_id, room_version_rules)
+					.await
 				{
-					| Ok((pdu, json)) =>
-						if next_id == *id {
-							trace!("Handled outlier {next_id} (original request)");
-							pdus.push((pdu, Some(json)));
-						},
+					| Ok((_, x)) => x,
 					| Err(e) => {
-						warn!("Authentication of event {next_id} failed: {e:?}");
-						back_off(next_id);
+						warn!(elapsed=?start.elapsed(),"failed to fetch missing event {target_id} from any candidate: {e}");
+						continue;
 					},
+				};
+				let auth_events =
+					match expect_event_id_array(&value, "auth_events").map_err(|e| {
+						err!(Request(BadJson(warn!(
+							elapsed=?start.elapsed(),
+							event_id=%target_id,
+							"Failed to parse event fetched from remote: {e}"
+						))))
+					}) {
+						| Ok(auth_events) => auth_events,
+						| Err(e) => {
+							warn!(
+								elapsed=?start.elapsed(),
+								?e,
+								"event {target_id} is malformed (bad auth_events), skipping"
+							);
+							continue;
+						},
+					};
+				let mut have_all_auth = true;
+				for auth_event_id in auth_events {
+					if let Ok(local_pdu) = self.services.timeline.get_pdu(&auth_event_id).await {
+						trace!(elapsed=?start.elapsed(),"Found auth event {auth_event_id} in db");
+						let mut obj = local_pdu.into_canonical_object();
+						obj.remove("event_id");
+						discovered_events.insert(auth_event_id.clone(), obj);
+						continue;
+					}
+					if discovered_events.contains_key(&auth_event_id) {
+						trace!(elapsed=?start.elapsed(),%auth_event_id, "Already found auth event");
+						continue;
+					}
+					debug!(elapsed=?start.elapsed(),"Missing auth event {auth_event_id} for event {target_id}");
+					seen.insert(
+						auth_event_id.clone(),
+						seen.get(&auth_event_id)
+							.copied()
+							.unwrap_or_default()
+							.saturating_add(1),
+					);
+					todo.push_back(auth_event_id);
+					have_all_auth = false;
+				}
+				// Insert this PDU back at the end of the queue so that it will be resolved once
+				// all of its auth events have been fetched.
+				if have_all_auth {
+					debug!(elapsed=?start.elapsed(),%target_id, "Have all auth events");
+					discovered_events.insert(target_id, value);
+				} else {
+					debug_warn!(elapsed=?start.elapsed(),
+						"Fetched {target_id} but missing some auth events, will have to re-fetch."
+					);
+					seen.insert(target_id.clone(), attempts.saturating_add(1));
+					todo.push_back(target_id);
 				}
 			}
 		}
-		trace!("Fetched and handled {} outlier pdus", pdus.len());
+
+		let refmap: HashMap<OwnedEventId, &CanonicalJsonObject> = discovered_events
+			.iter()
+			.map(|(id, data)| (id.clone(), data))
+			.collect();
+		let seeded_ordered = build_local_dag(&refmap)
+			.await
+			.expect("failed to build local DAG");
+		let mut pdus = HashMap::with_capacity(seeded_ordered.len());
+		for discovered_event_id in seeded_ordered {
+			let pdu_json = discovered_events.remove(&discovered_event_id).unwrap();
+			debug_info!(
+				elapsed=?start.elapsed(),
+				"Handling missing event {discovered_event_id} as outlier"
+			);
+			assert_eq!(pdu_json.get("event_id"), None, "pdu_json had event_id");
+			match Box::pin(self.handle_outlier_pdu(
+				origin,
+				create_event,
+				&discovered_event_id,
+				room_id,
+				pdu_json,
+			))
+			.await
+			{
+				| Ok((pdu, _)) => {
+					trace!(elapsed=?start.elapsed(), "Persisted {discovered_event_id}");
+					let _ = pdus.insert(discovered_event_id, pdu);
+				},
+				| Err(e) => warn!(
+					elapsed=?start.elapsed(),
+					"Authentication of event {discovered_event_id} failed: {e:?}"
+				),
+			}
+		}
+
+		trace!(
+			elapsed=?start.elapsed(),
+			"Finished fetch_and_handle_missing_events: fetched and handled {} missing PDUs",
+			pdus.len()
+		);
+		pdus.retain(|id, _| events.contains(id)); // Only return state events
+		trace!(elapsed=?start.elapsed(), "Filtered return value down to {} PDUs", pdus.len());
 		pdus
 	}
+
+	/// Similar to `fetch_and_handle_missing_events`, but simply walks the
+	/// prev events tree instead of the auth events tree. Additionally, it does
+	/// not *handle* fetched PDUs in any capacity.
+	#[tracing::instrument(name = "get_missing_prev_events_atomic", skip_all)]
+	pub(super) async fn fetch_prev_events<Pdu>(
+		&self,
+		origin: &ServerName,
+		events: Vec<OwnedEventId>,
+		create_event: &Pdu,
+		room_id: &RoomId,
+	) -> HashMap<OwnedEventId, PduEvent>
+	where
+		Pdu: Event + Send + Sync,
+	{
+		let room_version_rules =
+			&get_room_version_rules(create_event).unwrap_or(RoomVersionRules::V1);
+		let mut candidates = self
+			.services
+			.timeline
+			.candidate_backfill_servers(room_id)
+			.await;
+		candidates.insert(origin.to_owned());
+
+		let mut todo: VecDeque<OwnedEventId> = VecDeque::from(events);
+		let mut discovered_events = HashMap::new();
+		while let Some(next_id) = todo.pop_front() {
+			if discovered_events.len() >= self.services.server.config.max_fetch_prev_events.into()
+			{
+				debug_warn!(
+					"Encountered a gap too large to fill, giving up (fetched {} events)",
+					discovered_events.len()
+				);
+				break;
+			}
+			if discovered_events.contains_key(&next_id) {
+				continue;
+			}
+			let pdu = match self
+				.fetch_event_vias(candidates.iter(), &next_id, room_version_rules)
+				.await
+			{
+				| Ok((_, data)) => data,
+				| Err(e) => {
+					warn!("Failed to fetch prev event {next_id} from any candidate: {e}");
+					continue;
+				},
+			};
+
+			let prev_events = match expect_event_id_array(&pdu, "prev_events").map_err(|e| {
+				err!(Request(BadJson(warn!(
+					event_id=%next_id,
+					"Failed to parse event fetched from remote: {e}"
+				))))
+			}) {
+				| Ok(auth_events) => auth_events,
+				| Err(e) => {
+					warn!(?e, "event {next_id} is malformed (bad prev_events), skipping");
+					continue;
+				},
+			};
+			let missing_prev = prev_events
+				.iter()
+				.stream()
+				.broad_filter_map(|event_id| async {
+					if discovered_events.contains_key(event_id)
+						|| self.services.timeline.pdu_exists(event_id).await
+					{
+						None
+					} else {
+						Some(event_id.to_owned())
+					}
+				})
+				.collect::<Vec<_>>()
+				.await;
+			todo.extend(missing_prev);
+			discovered_events.insert(
+				next_id.clone(),
+				PduEvent::from_id_val(&next_id, pdu).expect("fetched PDU was already validated"),
+			);
+		}
+
+		discovered_events
+	}
 }

src/service/rooms/event_handler/fetch_prev.rs

@@ -1,124 +1,155 @@
-use std::{
-	collections::{BTreeMap, HashMap, HashSet, VecDeque},
-	iter::once,
-};
+use std::{collections::HashMap, time::Instant};
 
 use conduwuit::{
-	Event, PduEvent, Result, debug_warn, err,
-	state_res::{self},
-};
-use futures::{FutureExt, future};
-use ruma::{
-	CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName,
-	int, uint,
+	Event, PduEvent, debug, debug_info, debug_warn, trace,
+	utils::{BoolExt, IterStream, stream::BroadbandExt},
 };
+use futures::StreamExt;
+use ruma::{CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName};
 
-use super::check_room_id;
+use crate::rooms::event_handler::build_local_dag;
 
 impl super::Service {
-	#[allow(clippy::type_complexity)]
-	pub(super) async fn fetch_prev<'a, Pdu, Events>(
+	/// Fetches any missing prev_events for this event and persists them before
+	/// returning.
+	pub(super) async fn fetch_prevs(
 		&self,
-		origin: &ServerName,
-		create_event: &Pdu,
 		room_id: &RoomId,
+		create_event: &PduEvent,
+		incoming_pdu: &PduEvent,
+		origin: &ServerName,
 		first_ts_in_room: MilliSecondsSinceUnixEpoch,
-		initial_set: Events,
-	) -> Result<(
-		Vec<OwnedEventId>,
-		HashMap<OwnedEventId, (PduEvent, BTreeMap<String, CanonicalJsonValue>)>,
-	)>
-	where
-		Pdu: Event + Send + Sync,
-		Events: Iterator<Item = &'a EventId> + Clone + Send,
-	{
-		let num_ids = initial_set.clone().count();
-		let mut eventid_info = HashMap::new();
-		let mut graph: HashMap<OwnedEventId, _> = HashMap::with_capacity(num_ids);
-		let mut todo_outlier_stack: VecDeque<OwnedEventId> =
-			initial_set.map(ToOwned::to_owned).collect();
+	) -> conduwuit::Result<()> {
+		let start = Instant::now();
+		let mut missing = incoming_pdu
+			.prev_events()
+			.stream()
+			.broad_filter_map(|event_id| async move {
+				self.services
+					.timeline
+					.get_non_outlier_pdu_json(event_id)
+					.await
+					.is_ok()
+					.or(|| event_id.to_owned())
+			})
+			.collect::<Vec<_>>()
+			.await;
+		if missing.is_empty() {
+			debug!(elapsed=?start.elapsed(), event_id=%incoming_pdu.event_id(), "No missing prev events.");
+			return Ok(());
+		}
+		debug!(elapsed=?start.elapsed(), %room_id, event_id=%incoming_pdu.event_id(), ?missing, "Fetching previous events");
+		let tail = self
+			.services
+			.state
+			.get_forward_extremities(room_id)
+			.collect::<Vec<_>>()
+			.await;
 
-		let mut amount = 0;
+		let mut gapfilled = self
+			.get_missing_events(
+				room_id,
+				incoming_pdu,
+				tail,
+				origin,
+				self.services
+					.metadata
+					.get_mindepth(room_id)
+					.await
+					.saturating_sub(
+						u8::try_from(incoming_pdu.prev_events.len())
+							.unwrap()
+							.saturating_mul(2)
+							.into(),
+					),
+			)
+			.await?;
+		debug_info!(elapsed=?start.elapsed(), "Fetched {} missing events", gapfilled.len());
+		missing.retain(|eid| !gapfilled.contains_key(eid));
+		if !missing.is_empty() {
+			debug_warn!(elapsed=?start.elapsed(), "Still missing {} events, falling back to atomic fetch.", missing.len());
+			gapfilled.extend(
+				self.fetch_prev_events(origin, missing, create_event, room_id)
+					.await,
+			);
+		}
 
-		while let Some(prev_event_id) = todo_outlier_stack.pop_front() {
-			self.services.server.check_running()?;
+		// Persist all fetched events
+		let mapped = gapfilled
+			.iter()
+			.map(|(eid, evt)| {
+				let mut obj = evt.to_canonical_object();
+				obj.remove("event_id"); // event_id is inserted by backfill_missing_events
+				(eid.clone(), obj)
+			})
+			.collect::<HashMap<_, _>>();
 
+		let to_persist = if mapped.len() <= 1 {
+			mapped.keys().map(ToOwned::to_owned).collect()
+		} else {
+			let refmap: HashMap<OwnedEventId, &CanonicalJsonObject> =
+				mapped.iter().map(|(id, data)| (id.clone(), data)).collect();
+			build_local_dag(&refmap).await?
+		};
+
+		let job_start = Instant::now();
+		trace!("Starting to persist {} prev events", to_persist.len());
+		for (i, event_id) in to_persist.iter().enumerate() {
+			debug!(
+				elapsed=?start.elapsed(),
+				"Persisting fetched prev event: {event_id} ({}/{})",
+				i.saturating_add(1),
+				to_persist.len(),
+			);
+			let obj = mapped.get(event_id).cloned().unwrap();
+			let persist_start = Instant::now();
 			match self
-				.fetch_and_handle_outliers(
-					origin,
-					once(prev_event_id.as_ref()),
-					create_event,
-					room_id,
-				)
-				.boxed()
+				.handle_outlier_pdu(origin, create_event, event_id, room_id, obj)
 				.await
-				.pop()
 			{
-				| Some((pdu, mut json_opt)) => {
-					check_room_id(room_id, &pdu)?;
-
-					let limit = self.services.server.config.max_fetch_prev_events;
-					if amount > limit {
-						debug_warn!("Max prev event limit reached! Limit: {limit}");
-						graph.insert(prev_event_id.clone(), HashSet::new());
-						continue;
-					}
-
-					if json_opt.is_none() {
-						json_opt = self
-							.services
-							.outlier
-							.get_outlier_pdu_json(&prev_event_id)
-							.await
-							.ok();
-					}
-
-					if let Some(json) = json_opt {
-						if pdu.origin_server_ts() > first_ts_in_room {
-							amount = amount.saturating_add(1);
-							for prev_prev in pdu.prev_events() {
-								if !graph.contains_key(prev_prev) {
-									todo_outlier_stack.push_back(prev_prev.to_owned());
-								}
-							}
-
-							graph.insert(
-								prev_event_id.clone(),
-								pdu.prev_events().map(ToOwned::to_owned).collect(),
-							);
-						} else {
-							// Time based check failed
-							graph.insert(prev_event_id.clone(), HashSet::new());
-						}
-
-						eventid_info.insert(prev_event_id.clone(), (pdu, json));
-					} else {
-						// Get json failed, so this was not fetched over federation
-						graph.insert(prev_event_id.clone(), HashSet::new());
-					}
-				},
-				| _ => {
-					// Fetch and handle failed
-					graph.insert(prev_event_id.clone(), HashSet::new());
+				| Ok((pdu, val)) if pdu.origin_server_ts() >= first_ts_in_room => {
+					Box::pin(self.upgrade_outlier_to_timeline_pdu(
+						pdu,
+						val,
+						create_event,
+						origin,
+						room_id,
+					))
+					.await
+					.inspect_err(|e| {
+						debug_warn!(
+							total_elapsed=?start.elapsed(),
+							job_elapsed=?job_start.elapsed(),
+							task_elapsed=?persist_start.elapsed(),
+							"Failed to upgrade prev event {event_id}: {e}",
+						);
+					})
+					.inspect(|_| {
+						debug_info!(
+							total_elapsed=?start.elapsed(),
+							job_elapsed=?job_start.elapsed(),
+							task_elapsed=?persist_start.elapsed(),
+							"Upgraded prev event {event_id}",
+						);
+					})
+					.ok();
 				},
+				| Err(e) => debug_warn!(
+					total_elapsed=?start.elapsed(),
+					job_elapsed=?job_start.elapsed(),
+					task_elapsed=?persist_start.elapsed(),
+					"Failed to persist prev event {event_id}: {e}",
+				),
+				| _ => {},
 			}
 		}
 
-		let event_fetch = |event_id| {
-			let origin_server_ts = eventid_info
-				.get(&event_id)
-				.map_or_else(|| uint!(0), |info| info.0.origin_server_ts().get());
-
-			// This return value is the key used for sorting events,
-			// events are then sorted by power level, time,
-			// and lexically by event_id.
-			future::ok((int!(0), MilliSecondsSinceUnixEpoch(origin_server_ts)))
-		};
-
-		let sorted = state_res::lexicographical_topological_sort(&graph, &event_fetch)
-			.await
-			.map_err(|e| err!(Database(error!("Error sorting prev events: {e}"))))?;
-
-		Ok((sorted, eventid_info))
+		// NOTE because i keep forgetting: the caller persists incoming_pdu.
+		// we only care about its prev events
+		trace!(
+			total_elapsed=?start.elapsed(),
+			persist_elapsed=?job_start.elapsed(),
+		);
+		Ok(())
 	}
 }

src/service/rooms/event_handler/fetch_state.rs

@@ -1,34 +1,43 @@
-use std::collections::{HashMap, hash_map};
-
-use conduwuit::{Err, Event, Result, debug, debug_warn, err};
-use futures::FutureExt;
-use ruma::{
-	EventId, OwnedEventId, RoomId, ServerName, api::federation::event::get_room_state_ids,
-	events::StateEventType,
+use std::{
+	cmp::max,
+	collections::{HashMap, HashSet, hash_map},
+	hash::{BuildHasherDefault, DefaultHasher},
+	time::{Duration, Instant},
 };
 
-use crate::rooms::short::ShortStateKey;
+use conduwuit::{
+	Err, Event, PduEvent, Result, debug, debug_warn, err, info, trace,
+	utils::{BoolExt, IterStream},
+	warn,
+};
+use futures::{StreamExt, TryFutureExt, future::select_ok};
+use ruma::{
+	EventId, OwnedEventId, OwnedRoomId, RoomId, ServerName,
+	api::federation::event::{get_room_state, get_room_state_ids},
+};
+
+use crate::{conduwuit::utils::stream::BroadbandExt, rooms::short::ShortStateKey};
 
 impl super::Service {
-	/// Call /state_ids to find out what the state at this pdu is. We trust the
-	/// server's response to some extent (sic), but we still do a lot of checks
-	/// on the events
-	#[tracing::instrument(
-		level = "debug",
-		skip_all,
-		fields(%origin),
-	)]
-	pub(super) async fn fetch_state<Pdu>(
+	/// Asks a remote server what the state at this event is.
+	/// It first attempts to call `GET /_matrix/federation/v1/state_ids` (fast).
+	/// If any events are missing, they are fetched from the remote, and
+	/// persisted as outliers, before being returned back to this function. If
+	/// we are missing a lot of events locally (>=50), this function falls back
+	/// to requesting the full state in PDU format from the remote (`GET
+	/// /_matrix/federation/v1/state, very slow in large rooms), and persists
+	/// them directly.
+	#[tracing::instrument(skip_all)]
+	pub(super) async fn fetch_state(
 		&self,
 		origin: &ServerName,
-		create_event: &Pdu,
+		create_event: &PduEvent,
 		room_id: &RoomId,
 		event_id: &EventId,
-	) -> Result<Option<HashMap<u64, OwnedEventId>>>
-	where
-		Pdu: Event + Send + Sync,
-	{
-		let res = self
+	) -> Result<HashMap<u64, OwnedEventId>> {
+		let start = Instant::now();
+		trace!(%origin, "Asking remote for state_ids");
+		let res: get_room_state_ids::v1::Response = match self
 			.services
 			.sending
 			.send_federation_request(
@@ -36,21 +45,189 @@ pub(super) async fn fetch_state<Pdu>(
 				get_room_state_ids::v1::Request::new(event_id.to_owned(), room_id.to_owned()),
 			)
 			.await
-			.inspect_err(|e| debug_warn!("Fetching state for event failed: {e}"))?;
+			.inspect_err(
+				|e| debug_warn!(elapsed=?start.elapsed(), "Fetching state for event failed: {e}"),
+			) {
+			| Ok(resp) => Ok(resp),
+			| Err(e) =>
+				if e.is_not_found() {
+					self.fetch_state_ids_from_backfill_servers(
+						event_id.to_owned(),
+						room_id.to_owned(),
+					)
+					.await
+				} else {
+					Err(e)
+				},
+		}?;
 
-		debug!("Fetching state events");
-		let state_ids = res.pdu_ids.iter().map(AsRef::as_ref);
-		let state_vec = self
-			.fetch_and_handle_outliers(origin, state_ids, create_event, room_id)
-			.boxed()
+		debug!(elapsed=?start.elapsed(), events = res.pdu_ids.len(), "Fetching state events");
+		let mut state_events: HashMap<OwnedEventId, PduEvent> =
+			HashMap::with_capacity(res.pdu_ids.len());
+		let to_fetch: Vec<OwnedEventId> = res
+			.pdu_ids
+			.clone()
+			.into_iter()
+			.stream()
+			.broad_filter_map(|event_id| async move {
+				self.services
+					.timeline
+					.pdu_exists(&event_id)
+					.await
+					.or_some(event_id)
+			})
+			.collect()
 			.await;
+		if to_fetch.is_empty() {
+			debug!(elapsed=?start.elapsed(), "All required state events are already known.");
+			state_events = res
+				.pdu_ids
+				.iter()
+				.stream()
+				.broad_filter_map(|event_id| async move {
+					Some((
+						event_id.clone(),
+						self.services
+							.timeline
+							.get_pdu(event_id)
+							.await
+							.expect("Event disappeared between filtering and fetching"),
+					))
+				})
+				.collect()
+				.await;
+			assert_eq!(
+				state_events.len(),
+				res.pdu_ids.len(),
+				"Failed to load all required state events despite allegedly knowing all of them \
+				 already",
+			);
+		} else {
+			let total_count = res.pdu_ids.len();
+			let missing_count = to_fetch.len();
+			let missing_threshold = max(50, total_count >> 2);
+			if missing_count >= missing_threshold {
+				// If there's more than 50 events to fetch, or we're missing 25% or more of the
+				// state, we would need to make a lot of atomic requests, so we'll just try
+				// to fetch the full state from the remote instead.
+				// Since this endpoint might fail in huge rooms, we fall back to atomic fetch
+				// anyway.
+				warn!(
+					elapsed=?start.elapsed(),
+					%missing_count,
+					%total_count,
+					%missing_threshold,
+					"Fetching full state from remote server for event"
+				);
+				let state_response = tokio::time::timeout(
+					Duration::from_secs(30),
+					self.fetch_full_state(origin, create_event, room_id, event_id),
+				)
+				.await;
+				info!(
+					elapsed=?start.elapsed(),
+					%missing_count,
+					%total_count,
+					%missing_threshold,
+					"Fetched full state from remote server for event"
+				);
+				let fetched_state = match state_response {
+					| Ok(Ok(state)) => {
+						// Filter to ensure we only use the PDUs we were expecting, preventing
+						// arbitrary state injection.
+						// Atomic fetch does not have this problem as each PDU is evaluated
+						// individually.
+						let expected: &HashSet<OwnedEventId, BuildHasherDefault<DefaultHasher>> =
+							&HashSet::from_iter(res.pdu_ids.clone());
+						state
+							.into_iter()
+							.stream()
+							.broad_filter_map(|(event_id, pdu)| async move {
+								expected.contains(&event_id).then_some((event_id, pdu))
+							})
+							.collect()
+							.await
+					},
+					| Ok(Err(e)) => {
+						warn!(
+							elapsed=?start.elapsed(),
+							error=?e,
+							%origin,
+							"Failed to fetch full state from remote, falling back to atomic fetch"
+						);
+						self.fetch_and_handle_auth_events(
+							origin,
+							res.pdu_ids.clone(),
+							create_event,
+							room_id,
+						)
+						.await
+					},
+					| Err(e) => {
+						warn!(
+							elapsed=?start.elapsed(),
+							error=?e,
+							%origin,
+							"Remote did not return room state in an acceptable timeframe, falling back to atomic fetch"
+						);
+						self.fetch_and_handle_auth_events(
+							origin,
+							res.pdu_ids.clone(),
+							create_event,
+							room_id,
+						)
+						.await
+					},
+				};
+
+				assert!(
+					!fetched_state.is_empty(),
+					"fetch_full_state or fetch_and_handle_missing_events returned empty state \
+					 map"
+				);
+				state_events.extend(fetched_state);
+			} else {
+				state_events = res
+					.pdu_ids
+					.iter()
+					.stream()
+					.broad_filter_map(|event_id| async move {
+						self.services
+							.timeline
+							.get_pdu(event_id)
+							.await
+							.map(|p| (event_id.to_owned(), p))
+							.ok()
+					})
+					.collect()
+					.await;
+				assert!(
+					!state_events.is_empty(),
+					"Only missing {} events but read-ahead state vec was empty",
+					to_fetch.len()
+				);
+				debug!(
+					elapsed=?start.elapsed(),
+					to_fetch = to_fetch.len(),
+					"Fetching missing events for state from remote"
+				);
+				let fetched_state = self
+					.fetch_and_handle_auth_events(origin, to_fetch, create_event, room_id)
+					.await;
+				state_events.extend(fetched_state);
+			}
+		}
+		if state_events.is_empty() {
+			return Ok(HashMap::new());
+		}
 
 		let mut state: HashMap<ShortStateKey, OwnedEventId> =
-			HashMap::with_capacity(state_vec.len());
-		for (pdu, _) in state_vec {
-			let state_key = pdu
-				.state_key()
-				.ok_or_else(|| err!(Database("Found non-state pdu in state events.")))?;
+			HashMap::with_capacity(state_events.len());
+		debug!(elapsed=?start.elapsed(), events = state_events.len(), "Processing state events");
+		for (event_id, pdu) in state_events {
+			let state_key = pdu.state_key().ok_or_else(|| {
+				err!(Request(BadJson("Found non-state pdu in state events: {event_id}")))
+			})?;
 
 			let shortstatekey = self
 				.services
@@ -62,28 +239,154 @@ pub(super) async fn fetch_state<Pdu>(
 				| hash_map::Entry::Vacant(v) => {
 					v.insert(pdu.event_id().to_owned());
 				},
-				| hash_map::Entry::Occupied(_) => {
-					return Err!(Database(
-						"State event's type and state_key combination exists multiple times: \
-						 {}, {}",
+				| hash_map::Entry::Occupied(existing) => {
+					return Err!(Request(Forbidden(
+						"State event's type and state_key combination exists multiple times \
+						 ({event_id} + {}): ({}, \"{}\")",
+						existing.get(),
 						pdu.kind(),
-						state_key
-					));
+						state_key,
+					)));
 				},
 			}
 		}
+		trace!(elapsed=?start.elapsed(), "fetch_state finished");
+		Ok(state)
+	}
 
-		// The original create event must still be in the state
-		let create_shortstatekey = self
+	async fn fetch_state_ids_from_backfill_servers(
+		&self,
+		event_id: OwnedEventId,
+		room_id: OwnedRoomId,
+	) -> Result<get_room_state_ids::v1::Response> {
+		let candidates = self
 			.services
-			.short
-			.get_shortstatekey(&StateEventType::RoomCreate, "")
-			.await?;
-
-		if state.get(&create_shortstatekey).map(AsRef::as_ref) != Some(create_event.event_id()) {
-			return Err!(Database("Incoming event refers to wrong create event."));
+			.timeline
+			.candidate_backfill_servers(&room_id)
+			.await;
+		if candidates.is_empty() {
+			return Err!(Request(NotFound(
+				"Cannot ask any other servers for the state at this event"
+			)));
 		}
+		debug!(%room_id, ?candidates, "Asking backfill servers for state_ids");
+		let futures = candidates.iter().map(|server_name| {
+			Box::pin(
+				self.services
+					.sending
+					.send_federation_request(
+						server_name,
+						get_room_state_ids::v1::Request::new(event_id.clone(), room_id.clone()),
+					)
+					.inspect_err(|e| {
+						debug_warn!("Fallback fetching state for event failed: {e}");
+					}),
+			)
+		});
+		Ok(select_ok(futures).await?.0)
+	}
 
-		Ok(Some(state))
+	/// Fetches the full state via `GET /_matrix/federation/v1/state` from a
+	/// remote server, and persists all the incoming auth chain events and
+	/// state events as outliers, for use later.
+	///
+	/// Any events that cannot be persisted are dropped with a warning.
+	pub(super) async fn fetch_full_state(
+		&self,
+		origin: &ServerName,
+		create_event: &PduEvent,
+		room_id: &RoomId,
+		event_id: &EventId,
+	) -> Result<HashMap<OwnedEventId, PduEvent>> {
+		let start = Instant::now();
+		trace!("Fetching full state from remote server");
+		let res: get_room_state::v1::Response = self
+			.services
+			.sending
+			.send_federation_request(
+				origin,
+				get_room_state::v1::Request::new(event_id.to_owned(), room_id.to_owned()),
+			)
+			.await
+			.inspect_err(|e| debug_warn!("Fetching state for event failed: {e}"))?;
+		debug!(elapsed=?start.elapsed(), count = res.auth_chain.len(), "Handling incoming auth chain...");
+		res.auth_chain
+			.iter()
+			.stream()
+			.broad_filter_map(|raw_event_json| async {
+				if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok()
+					&& parsed.0 == room_id
+				{
+					Some(parsed)
+				} else {
+					None
+				}
+			})
+			.for_each_concurrent(
+				None,
+				|(incoming_room_id, incoming_event_id, incoming_event_json)| async move {
+					self.handle_outlier_pdu(
+						origin,
+						create_event,
+						&incoming_event_id,
+						&incoming_room_id,
+						incoming_event_json,
+					)
+					.await
+					.inspect_err(|e| {
+						warn!(
+							%incoming_room_id,
+							%incoming_event_id,
+							?e,
+							"Failed to handle auth chain event from state fetch"
+						);
+					})
+					.ok();
+				},
+			)
+			.await;
+		debug!(elapsed=?start.elapsed(), count = res.pdus.len(), "Handling incoming state PDUs...");
+		let r = res
+			.pdus
+			.iter()
+			.stream()
+			.broad_filter_map(|raw_event_json| async {
+				if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok()
+					&& parsed.0 == room_id
+				{
+					Some(parsed)
+				} else {
+					None
+				}
+			})
+			.broad_filter_map(
+				|(incoming_room_id, incoming_event_id, incoming_event_json)| async move {
+					self.handle_outlier_pdu(
+						origin,
+						create_event,
+						&incoming_event_id,
+						&incoming_room_id,
+						incoming_event_json,
+					)
+					.await
+					.inspect_err(|e| {
+						warn!(
+							elapsed=?start.elapsed(),
+							%incoming_room_id,
+							%incoming_event_id,
+							?e,
+							"Failed to handle state event from state fetch"
+						);
+					})
+					.ok()
+				},
+			)
+			.fold(HashMap::new(), |mut acc, (event, _)| async move {
+				acc.insert(event.event_id().to_owned(), event);
+				acc
+			})
+			.await;
+		trace!(elapsed=?start.elapsed(), "fetch_full_state finished");
+		Ok(r)
 	}
 }

src/service/rooms/event_handler/handle_incoming_pdu.rs

@@ -1,14 +1,14 @@
 use std::{
-	collections::{BTreeMap, hash_map},
-	time::Instant,
+	collections::BTreeMap,
+	time::{Duration, Instant},
 };
 
 use conduwuit::{
-	Err, Event, PduEvent, Result, debug::INFO_SPAN_LEVEL, debug_error, debug_info, defer, err,
-	info, trace, utils::stream::IterStream, warn,
+	Err, Event, PduEvent, Result, debug, debug_error, debug_info, debug_warn, defer, err, error,
+	info, matrix::PartialPdu, result::DebugInspect, trace, utils::time::jitter, warn,
 };
 use futures::{
-	FutureExt, TryFutureExt, TryStreamExt,
+	FutureExt, StreamExt,
 	future::{OptionFuture, try_join4},
 };
 use ruma::{
@@ -18,7 +18,7 @@
 		room::member::{MembershipState, RoomMemberEventContent},
 	},
 };
-use tracing::debug;
+use tokio::sync::mpsc;
 
 use crate::rooms::timeline::{RawPduId, pdu_fits};
 
@@ -110,10 +110,9 @@ impl super::Service {
 	/// 14. Check if the event passes auth based on the "current state" of the
 	///     room, if not soft fail it
 	#[tracing::instrument(
-	name = "pdu",
-	level = INFO_SPAN_LEVEL,
-	skip_all,
-	fields(%room_id, %event_id),
+		name = "pdu",
+		skip_all,
+		fields(%room_id, %event_id),
 	)]
 	pub async fn handle_incoming_pdu<'a>(
 		&self,
@@ -153,9 +152,7 @@ pub async fn handle_incoming_pdu<'a>(
 			.and_then(|v| v.as_str())
 			.ok_or_else(|| err!("No sender in object"))
 			.and_then(|v| Ok(UserId::parse(v)?))
-			.map_err(|e| {
-				err!(Request(InvalidParam("PDU does not have a valid sender key: {e}")))
-			})?;
+			.map_err(|e| err!(Request(BadJson("PDU does not have a valid sender key: {e}"))))?;
 
 		let sender_acl_check: OptionFuture<_> = sender
 			.server_name()
@@ -235,7 +232,7 @@ pub async fn handle_incoming_pdu<'a>(
 		}}
 
 		let (incoming_pdu, val) = self
-			.handle_outlier_pdu(origin, create_event, event_id, room_id, value, false)
+			.handle_outlier_pdu(origin, create_event, event_id, room_id, value)
 			.await?;
 
 		// 8. if not timeline event: stop
@@ -243,71 +240,211 @@ pub async fn handle_incoming_pdu<'a>(
 			return Ok(None);
 		}
 
-		// Skip old events
+		// Skip events sent before we joined (they need to be persisted as backfilled
+		// events, not timeline events, which is handled elsewhere).
 		let first_ts_in_room = self
 			.services
 			.timeline
 			.first_pdu_in_room(room_id)
 			.await?
 			.origin_server_ts();
+		if incoming_pdu.origin_server_ts() < first_ts_in_room {
+			return Ok(None);
+		}
 
 		// 9. Fetch any missing prev events doing all checks listed here starting at 1.
 		//    These are timeline events
-		let (sorted_prev_events, mut eventid_info) = self
-			.fetch_prev(
-				origin,
-				create_event,
-				room_id,
-				first_ts_in_room,
-				incoming_pdu.prev_events(),
-			)
-			.await?;
 
-		debug!(
-			events = ?sorted_prev_events,
-			"Handling previous events"
-		);
+		debug!("Fetching and persisting any missing prev events");
+		Box::pin(self.fetch_prevs(
+			room_id,
+			create_event,
+			&incoming_pdu,
+			origin,
+			first_ts_in_room,
+		))
+		.await
+		.debug_inspect_err(|e| {
+			error!("Failed to fetch and persist incoming event's prev_events: {e:?}");
+		})?;
 
-		sorted_prev_events
-			.iter()
-			.try_stream()
-			.map_ok(AsRef::as_ref)
-			.try_for_each(|prev_id| {
-				self.handle_prev_pdu(
-					origin,
-					event_id,
-					room_id,
-					eventid_info.remove(prev_id),
-					create_event,
-					first_ts_in_room,
-					prev_id,
-				)
-				.inspect_err(move |e| {
-					warn!("Prev {prev_id} failed: {e}");
-					match self
-						.services
-						.globals
-						.bad_event_ratelimiter
-						.write()
-						.entry(prev_id.into())
-					{
-						| hash_map::Entry::Vacant(e) => {
-							e.insert((Instant::now(), 1));
-						},
-						| hash_map::Entry::Occupied(mut e) => {
-							let tries = e.get().1.saturating_add(1);
-							*e.get_mut() = (Instant::now(), tries);
-						},
-					}
-				})
-				.map(|_| self.services.server.check_running())
-			})
-			.boxed()
-			.await?;
+		let is_dummy_event = incoming_pdu.event_type().to_string() == "org.matrix.dummy_event";
 
 		// Done with prev events, now handling the incoming event
-		self.upgrade_outlier_to_timeline_pdu(incoming_pdu, val, create_event, origin, room_id)
-			.boxed()
-			.await
+		let pdu_id = Box::pin(self.upgrade_outlier_to_timeline_pdu(
+			incoming_pdu,
+			val,
+			create_event,
+			origin,
+			room_id,
+		))
+		.await?;
+
+		let extremities_count = self
+			.services
+			.state
+			.get_forward_extremities(room_id)
+			.count()
+			.await;
+
+		self.maybe_squash_extremities(room_id, extremities_count, is_dummy_event)
+			.await;
+
+		Ok(pdu_id)
+	}
+
+	/// Conditionally starts an extremity squasher. If there is no waiting
+	/// extremity squasher, a new one is created. Otherwise, the existing one is
+	/// pinged.
+	async fn maybe_squash_extremities(
+		&self,
+		room_id: &RoomId,
+		extremities_count: usize,
+		is_dummy_event: bool,
+	) {
+		let (tx, fut) = {
+			if let Some(tx) = self.extremity_squashers.read().get(room_id)
+				&& !tx.is_closed()
+			{
+				(tx.clone(), None)
+			} else {
+				let mut map = self.extremity_squashers.upgradable_read();
+
+				if let Some(tx) = map.get(room_id)
+					&& !tx.is_closed()
+				{
+					(tx.clone(), None)
+				} else {
+					let (tx, rx) = mpsc::channel(100);
+					map.with_upgraded(|map| map.insert(room_id.to_owned(), tx.clone()));
+
+					(tx, Some(self.spawn_squasher(room_id, rx)))
+				}
+			}
+		};
+
+		if let Some(fut) = fut {
+			fut.await;
+		}
+		let _ = tx.try_send((extremities_count, is_dummy_event));
+	}
+
+	/// Spawns an extremity squasher with the given room and receiver channel.
+	async fn spawn_squasher(&self, room_id: &RoomId, mut rx: mpsc::Receiver<(usize, bool)>) {
+		let Some(service) = self.me.upgrade() else {
+			return;
+		};
+		let room_id = room_id.to_owned();
+
+		self.services.server.runtime().spawn(async move {
+			let mut latest_extremity_count = None;
+			let mut non_dummy_event = false;
+
+			let mut closing = false;
+
+			let waker = tokio::time::sleep(jitter(Duration::from_mins(2), -25.0..=25.0));
+			tokio::pin!(waker);
+
+			loop {
+				tokio::select! {
+					msg = rx.recv() => {
+						if let Some((extremities_count, is_dummy_event)) = msg {
+							latest_extremity_count = Some(extremities_count);
+							non_dummy_event = non_dummy_event || !is_dummy_event;
+							let sleep_duration = if extremities_count >= 20 {
+								// Skip the original sleep duration and send in the next 3-7 seconds as the number of extremities has grown beyond what one squash can reasonably reduce. We still jitter here in case we receive more events in that time that reduce the number anyway, and to account for other servers sending the same squashes.
+								jitter(Duration::from_secs(5), -50.0..=50.0)
+							} else {
+								jitter(Duration::from_mins(1), -50.0..=50.0)
+							};
+							#[allow(clippy::arithmetic_side_effects)]
+							waker.as_mut().reset(tokio::time::Instant::now() + sleep_duration);
+						} else {
+							{let mut map = service.extremity_squashers.write();
+							if let Some(tx) = map.get(&room_id) && tx.is_closed() {
+								map.remove(&room_id);
+							}}
+
+							if let Some(count) = latest_extremity_count {
+								if non_dummy_event && count >= service.services.server.config.dummy_event_threshold.into() {
+									Self::squash_extremities(&service, &room_id, count).await;
+								}
+							}
+							break;
+						}
+					}
+					() = &mut waker, if !closing => {
+						if let Some(count) = latest_extremity_count {
+							if non_dummy_event && count >= service.services.server.config.dummy_event_threshold.into() {
+								Self::squash_extremities(&service, &room_id, count).await;
+							}
+							latest_extremity_count = None;
+							non_dummy_event = false;
+							#[allow(clippy::arithmetic_side_effects)]
+							waker.as_mut().reset(tokio::time::Instant::now() + Duration::from_mins(2));
+						} else {
+							rx.close();
+							closing = true;
+						}
+					}
+					() = service.server_shutdown.notified(), if !closing => {
+						rx.close();
+						closing = true;
+					}
+				}
+			}
+		});
+	}
+
+	/// Squashes extremities in a room by sending dummy events (empty events
+	/// that are hidden from clients) to the room. It will only send ONE dummy
+	/// event to squash. If there are more than 20 extremities, multiple calls
+	/// to `squash_extremities` will be required.
+	/// Sending the dummy event will be attempted by iterating over each local
+	/// user currently joined to the room (including deactivated users) until
+	/// either one of them successfully builds and appends a dummy event PDU, or
+	/// there are no more users to try.
+	async fn squash_extremities(&self, room_id: &RoomId, extremities_count: usize) {
+		debug_warn!(
+			%extremities_count,
+			threshold=%self.services.server.config.dummy_event_threshold,
+			"Attempting to squash extremities after upgrading pdu"
+		);
+		// Try to send a dummy event to squash extremities. See issue #1844
+		let power_levels = self
+			.services
+			.state_accessor
+			.get_room_power_levels(room_id)
+			.await;
+		let mut local_users = self.services.state_cache.local_users_in_room(room_id);
+		while let Some(user_id) = local_users.next().await {
+			if !power_levels.user_can_send_message(&user_id, "org.matrix.dummy_event".into()) {
+				trace!(%user_id, "user does not have power level to send dummy event, skipping");
+				continue;
+			}
+			let state_lock = self.services.state.mutex.lock(room_id).await;
+			if self
+				.services
+				.timeline
+				.build_and_append_pdu(
+					PartialPdu {
+						event_type: "org.matrix.dummy_event".into(),
+						..PartialPdu::default()
+					},
+					&user_id,
+					Some(room_id),
+					&state_lock,
+				)
+				.await
+				.inspect(|_| debug!(sender=%user_id, "Successfully sent a dummy event"))
+				.inspect_err(
+					|e| debug!(sender=%user_id, ?e, "Failed to send a dummy event via user"),
+				)
+				.is_ok()
+			{
+				return;
+			}
+		}
+		debug_warn!("Unable to squash extremities using any local user");
 	}
 }

src/service/rooms/event_handler/handle_outlier_pdu.rs

@@ -1,11 +1,13 @@
 use std::collections::{BTreeMap, HashMap, hash_map};
 
 use conduwuit::{
-	Err, Event, PduEvent, Result, debug, debug_info, debug_warn, err, state_res, trace, warn,
+	Err, Event, PduEvent, Result, debug, debug_info, debug_warn, err, info, state_res, trace,
+	warn,
 };
 use futures::future::ready;
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName,
+	api::federation::authorization::get_event_authorization, canonical_json::redact,
 	events::StateEventType,
 };
 
@@ -16,6 +18,7 @@ impl super::Service {
 	/// Handles a PDU as an outlier, performing basic checks like signatures and
 	/// hashes, proclaimed event auth, and then adding it to the outlier tree.
 	#[allow(clippy::too_many_arguments)]
+	#[tracing::instrument(name="handle_outlier", skip_all, fields(%event_id))]
 	pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 		&self,
 		origin: &'a ServerName,
@@ -23,7 +26,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 		event_id: &'a EventId,
 		room_id: &'a RoomId,
 		mut value: CanonicalJsonObject,
-		auth_events_known: bool,
 	) -> Result<(PduEvent, BTreeMap<String, CanonicalJsonValue>)>
 	where
 		Pdu: Event + Send + Sync,
@@ -47,27 +49,38 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 			.verify_event(&value, &room_version_rules)
 			.await
 		{
-			| Ok(ruma::signatures::Verified::All) => value,
+			| Ok(ruma::signatures::Verified::All) => {
+				if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
+					debug!(
+						"Already have event {event_id} as an outlier or timeline event, not \
+						 re-processing"
+					);
+					value.insert(
+						"event_id".to_owned(),
+						CanonicalJsonValue::String(event_id.as_str().to_owned()),
+					);
+					check_room_id(room_id, &pdu_event)?;
+					return Ok((pdu_event, value));
+				}
+				value
+			},
 			| Ok(ruma::signatures::Verified::Signatures) => {
-				// Redact
-				debug_info!("Calculated hash does not match (redaction): {event_id}");
-				let Ok(obj) =
-					ruma::canonical_json::redact(value, &room_version_rules.redaction, None)
-				else {
-					return Err!(Request(InvalidParam("Redaction failed")));
-				};
-
-				// Skip the PDU if it is redacted and we already have it as an outlier event
-				if self.services.timeline.pdu_exists(event_id).await {
-					return Err!(Request(InvalidParam(
-						"Event was redacted and we already knew about it"
-					)));
+				if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await {
+					debug!(
+						"Received a redacted copy of {event_id}, but we already knew about it. \
+						 Re-using known content instead."
+					);
+					check_room_id(room_id, &pdu_event)?;
+					let obj = pdu_event.to_canonical_object();
+					return Ok((pdu_event, obj));
 				}
 
-				obj
+				debug_info!("Calculated hash does not match (redaction): {event_id}");
+				redact(value, &room_version_rules.redaction, None)
+					.map_err(|e| err!(Request(BadJson("Failed to redact {event_id}: {e}"))))?
 			},
 			| Err(e) => {
-				return Err!(Request(InvalidParam(debug_error!(
+				return Err!(Request(Forbidden(debug_error!(
 					"Signature verification failed for {event_id}: {e}"
 				))));
 			},
@@ -90,67 +103,81 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 		// Fetch all auth events
 		let mut auth_events: HashMap<OwnedEventId, PduEvent> = HashMap::new();
 
-		for aid in pdu_event.auth_events() {
-			if self.services.pdu_metadata.is_event_rejected(aid).await {
+		for auth_event_id in pdu_event.auth_events() {
+			if self
+				.services
+				.pdu_metadata
+				.is_event_rejected(auth_event_id)
+				.await
+			{
 				debug_warn!(
-					"Rejecting incoming event {} which depends on rejected auth event {aid}",
+					"Rejecting incoming event {} which depends on rejected auth event \
+					 {auth_event_id}",
 					event_id,
 				);
 				self.services.pdu_metadata.mark_event_rejected(event_id);
-				return Err!(Request(InvalidParam("Event has rejected auth event: {aid}")));
+				return Err!(Request(Forbidden(
+					"Event has rejected auth event: {auth_event_id}"
+				)));
 			}
 
-			if let Ok(auth_event) = self.services.timeline.get_pdu(aid).await {
+			if let Ok(auth_event) = self.services.timeline.get_pdu(auth_event_id).await {
 				check_room_id(room_id, &auth_event)?;
-				trace!("Found auth event {aid} for outlier event {event_id} locally");
-				auth_events.insert(aid.to_owned(), auth_event);
+				trace!("Found auth event {auth_event_id} for outlier event {event_id} locally");
+				auth_events.insert(auth_event_id.to_owned(), auth_event);
 			} else {
 				debug_warn!(
-					"Could not find auth event {aid} for outlier event {event_id} locally"
+					"Could not find auth event {auth_event_id} for outlier event {event_id} \
+					 locally"
 				);
 			}
 		}
 
 		// Fetch any missing ones & reject invalid ones
-		let missing_auth_events = if auth_events_known {
-			pdu_event
-				.auth_events()
-				.filter(|id| !auth_events.contains_key(*id))
-				.collect::<Vec<_>>()
-		} else {
-			pdu_event.auth_events().collect::<Vec<_>>()
-		};
-		if !missing_auth_events.is_empty() || !auth_events_known {
-			debug_info!(
-				"Fetching {} missing auth events for outlier event {event_id}",
-				missing_auth_events.len()
-			);
-			for (pdu, _) in self
-				.fetch_and_handle_outliers(
+		if auth_events.len() != pdu_event.auth_events().count() {
+			info!("Missing some auth events, asking remote for auth chain");
+			let response: get_event_authorization::v1::Response = self
+				.services
+				.sending
+				.send_federation_request(
 					origin,
-					missing_auth_events.iter().copied(),
-					create_event,
-					room_id,
+					get_event_authorization::v1::Request::new(
+						room_id.to_owned(),
+						event_id.to_owned(),
+					),
 				)
 				.await
-			{
-				auth_events.insert(pdu.event_id().to_owned(), pdu);
+				.map_err(|e| {
+					err!(Request(Forbidden(
+						"Remote server is not divulging incoming event's auth chain: {e}"
+					)))
+				})?;
+			let mut auth_chain_map = HashMap::with_capacity(response.auth_chain.len());
+			for auth_pdu_json in response.auth_chain {
+				let (auth_event_room_id, auth_event_id, auth_pdu_json) =
+					self.parse_incoming_pdu(&auth_pdu_json).await?;
+				if auth_event_room_id != room_id {
+					return Err!(Request(Forbidden(
+						"Auth event {auth_event_id} is in {auth_event_room_id}, not {room_id}."
+					)));
+				}
+				let auth_pdu = PduEvent::from_id_val(&auth_event_id, auth_pdu_json)
+					.map_err(|e| err!(Request(BadJson("Invalid PDU {auth_event_id}: {e}"))))?;
+				auth_chain_map.insert(auth_event_id, auth_pdu);
+			}
+			for auth_event_id in pdu_event.auth_events() {
+				if auth_events.contains_key(auth_event_id) {
+					continue;
+				}
+				if let Some(auth_event) = auth_chain_map.get(auth_event_id) {
+					auth_events.insert(auth_event_id.to_owned(), auth_event.clone());
+				} else {
+					return Err!(Request(Forbidden(
+						"Remote server is not divulging incoming event's auth events (missing: \
+						 {auth_event_id})"
+					)));
+				}
 			}
-		} else {
-			debug!("No missing auth events for outlier event {event_id}");
-		}
-		// reject if we are still missing some
-		let still_missing = pdu_event
-			.auth_events()
-			.filter(|id| !auth_events.contains_key(*id))
-			.collect::<Vec<_>>();
-		if !still_missing.is_empty() {
-			// Don't reject: this could be a temporary condition
-			// TODO: use get_missing_events?
-			return Err!(Request(InvalidParam(
-				"Could not fetch all auth events for outlier event {event_id}, still missing: \
-				 {still_missing:?}"
-			)));
 		}
 
 		// 6. Reject "due to auth events" if the event doesn't pass auth based on the
@@ -181,7 +208,7 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 						.outlier
 						.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
 					self.services.pdu_metadata.mark_event_rejected(event_id);
-					return Err!(Request(InvalidParam(
+					return Err!(Request(Forbidden(
 						"Auth event's type and state_key combination exists multiple times: {}, \
 						 {}",
 						auth_event.kind,
@@ -191,18 +218,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>(
 			}
 		}
 
-		// The original create event must be in the auth events
-		if !matches!(
-			auth_events_by_key.get(&(StateEventType::RoomCreate, String::new().into())),
-			Some(_) | None
-		) {
-			self.services.pdu_metadata.mark_event_rejected(event_id);
-			self.services
-				.outlier
-				.add_pdu_outlier(pdu_event.event_id(), &incoming_pdu);
-			return Err!(Request(InvalidParam("Incoming event refers to wrong create event.")));
-		}
-
 		let state_fetch = |ty: &StateEventType, sk: &str| {
 			let key = (ty.to_owned(), sk.into());
 			ready(auth_events_by_key.get(&key).map(ToOwned::to_owned))

src/service/rooms/event_handler/handle_prev_pdu.rs

@@ -1,95 +0,0 @@
-use std::{collections::BTreeMap, time::Instant};
-
-use conduwuit::{
-	Err, Event, PduEvent, Result, debug::INFO_SPAN_LEVEL, defer,
-	utils::continue_exponential_backoff_secs,
-};
-use ruma::{CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch, RoomId, ServerName};
-use tracing::debug;
-
-impl super::Service {
-	#[allow(clippy::type_complexity)]
-	#[allow(clippy::too_many_arguments)]
-	#[tracing::instrument(
-	name = "prev",
-	level = INFO_SPAN_LEVEL,
-	skip_all,
-	fields(%prev_id),
-	)]
-	pub(super) async fn handle_prev_pdu<'a, Pdu>(
-		&self,
-		origin: &'a ServerName,
-		event_id: &'a EventId,
-		room_id: &'a RoomId,
-		eventid_info: Option<(PduEvent, BTreeMap<String, CanonicalJsonValue>)>,
-		create_event: &'a Pdu,
-		first_ts_in_room: MilliSecondsSinceUnixEpoch,
-		prev_id: &'a EventId,
-	) -> Result
-	where
-		Pdu: Event + Send + Sync,
-	{
-		// Check for disabled again because it might have changed
-		if self.services.metadata.is_disabled(room_id).await {
-			return Err!(Request(Forbidden(debug_warn!(
-				"Federaton of room {room_id} is currently disabled on this server. Request by \
-				 origin {origin} and event ID {event_id}"
-			))));
-		}
-
-		if let Some((time, tries)) = self
-			.services
-			.globals
-			.bad_event_ratelimiter
-			.read()
-			.get(prev_id)
-		{
-			// Exponential backoff
-			const MIN_DURATION: u64 = 5 * 60;
-			const MAX_DURATION: u64 = 60 * 60 * 24;
-			if continue_exponential_backoff_secs(
-				MIN_DURATION,
-				MAX_DURATION,
-				time.elapsed(),
-				*tries,
-			) {
-				debug!(
-					?tries,
-					duration = ?time.elapsed(),
-					"Backing off from prev_event"
-				);
-				return Ok(());
-			}
-		}
-
-		let Some((pdu, json)) = eventid_info else {
-			return Ok(());
-		};
-
-		// Skip old events
-		if pdu.origin_server_ts() < first_ts_in_room {
-			return Ok(());
-		}
-
-		let start_time = Instant::now();
-		self.federation_handletime
-			.write()
-			.insert(room_id.into(), ((*prev_id).to_owned(), start_time));
-
-		defer! {{
-			self.federation_handletime
-				.write()
-				.remove(room_id);
-		}};
-
-		self.upgrade_outlier_to_timeline_pdu(pdu, json, create_event, origin, room_id)
-			.await?;
-
-		debug!(
-			elapsed = ?start_time.elapsed(),
-			"Handled prev_event",
-		);
-
-		Ok(())
-	}
-}

src/service/rooms/event_handler/mod.rs

@@ -4,7 +4,6 @@
 mod fetch_state;
 mod handle_incoming_pdu;
 mod handle_outlier_pdu;
-mod handle_prev_pdu;
 mod parse_incoming_pdu;
 mod policy_server;
 mod resolve_state;
@@ -15,19 +14,21 @@
 
 use async_trait::async_trait;
 use conduwuit::{Err, Event, PduEvent, Result, Server, SyncRwLock, utils::MutexMap};
+pub use fetch_and_handle_outliers::{GET_MISSING_EVENTS_MAX_BATCH_SIZE, build_local_dag};
 use ruma::{
 	OwnedEventId, OwnedRoomId, RoomId, events::room::create::RoomCreateEventContent,
 	room_version_rules::RoomVersionRules,
 };
-use tokio::sync::Notify;
+use tokio::sync::{Notify, mpsc};
 
 use crate::{Dep, globals, rooms, sending, server_keys};
-
 pub struct Service {
 	pub mutex_federation: RoomMutexMap,
 	pub federation_handletime: SyncRwLock<HandleTimeMap>,
+	pub extremity_squashers: SyncRwLock<HashMap<OwnedRoomId, mpsc::Sender<(usize, bool)>>>,
 	services: Services,
 	server_shutdown: Notify,
+	me: std::sync::Weak<Self>,
 }
 
 struct Services {
@@ -53,9 +54,11 @@ struct Services {
 #[async_trait]
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
-		Ok(Arc::new(Self {
+		Ok(Arc::new_cyclic(|s| Self {
+			me: s.clone(),
 			mutex_federation: RoomMutexMap::new(),
 			federation_handletime: HandleTimeMap::new().into(),
+			extremity_squashers: SyncRwLock::new(HashMap::new()),
 			services: Services {
 				globals: args.depend::<globals::Service>("globals"),
 				sending: args.depend::<sending::Service>("sending"),
@@ -91,6 +94,8 @@ async fn memory_usage(&self, out: &mut (dyn Write + Send)) -> Result {
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 
 	fn interrupt(&self) { self.server_shutdown.notify_waiters(); }
+
+	async fn clear_cache(&self) {}
 }
 
 impl Service {

src/service/rooms/event_handler/parse_incoming_pdu.rs

@@ -7,7 +7,7 @@
 use itertools::Itertools;
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, RoomId,
-	RoomVersionId,
+	RoomVersionId, room_version_rules::RoomVersionRules,
 };
 use serde_json::value::RawValue as RawJsonValue;
 
@@ -56,7 +56,10 @@ fn extract_room_id(event_type: &str, pdu: &CanonicalJsonObject) -> Result<OwnedR
 
 /// Parses every entry in an array as an event ID, returning an error if any
 /// step fails.
-fn expect_event_id_array(value: &CanonicalJsonObject, field: &str) -> Result<Vec<OwnedEventId>> {
+pub(super) fn expect_event_id_array(
+	value: &CanonicalJsonObject,
+	field: &str,
+) -> Result<Vec<OwnedEventId>> {
 	value
 		.get(field)
 		.ok_or_else(|| err!(Request(BadJson("missing field `{field}` on PDU"))))?
@@ -102,6 +105,19 @@ pub fn validate_pdu(&self, pdu: &CanonicalJsonObject) -> Result {
 		Ok(())
 	}
 
+	pub async fn parse_incoming_pdu_with_known_room(
+		&self,
+		pdu: &RawJsonValue,
+		room_version_rules: &RoomVersionRules,
+	) -> Result<(OwnedEventId, CanonicalJsonObject)> {
+		let (event_id, value) =
+			gen_event_id_canonical_json(pdu, room_version_rules).map_err(|e| {
+				err!(Request(InvalidParam("Could not convert event to canonical json: {e}")))
+			})?;
+		self.validate_pdu(&value)?;
+		Ok((event_id, value))
+	}
+
 	pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result<Parsed> {
 		let value = serde_json::from_str::<CanonicalJsonObject>(pdu.get()).map_err(|e| {
 			err!(BadServerResponse(debug_warn!("Error parsing incoming event {e:?}")))

src/service/rooms/event_handler/state_at_incoming.rs

@@ -5,7 +5,7 @@
 };
 
 use conduwuit::{
-	Result, debug, err, error,
+	Result, debug, debug_error, err, error,
 	matrix::{Event, StateMap},
 	trace,
 	utils::stream::{BroadbandExt, IterStream, ReadyExt, TryBroadbandExt, TryWidebandExt},
@@ -37,6 +37,7 @@ pub(super) async fn state_at_incoming_degree_one<Pdu>(
 			.pdu_shortstatehash(prev_event)
 			.await
 		else {
+			trace!("No shortstatehash for {prev_event}, cannot calculate one-degree state.");
 			return Ok(None);
 		};
 
@@ -100,6 +101,7 @@ pub(super) async fn state_at_incoming_resolved<Pdu>(
 					.map_ok(move |sstatehash| (sstatehash, prev_event))
 			})
 			.try_collect::<HashMap<_, _>>()
+			.inspect_err(|e| debug_error!("failed to calculate N-degree short state hashes: {e}"))
 			.await
 		else {
 			return Ok(None);

src/service/rooms/event_handler/upgrade_outlier_pdu.rs

@@ -1,8 +1,9 @@
 use std::{borrow::Borrow, sync::Arc, time::Instant};
 
 use conduwuit::{
-	Err, Result, debug, debug_info, err, info, is_equal_to,
+	Err, Result, debug, debug_error, debug_info, err, info, is_equal_to,
 	matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res},
+	result::DebugInspect,
 	trace,
 	utils::{
 		IterStream,
@@ -23,28 +24,17 @@
 };
 
 impl super::Service {
-	pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
+	#[tracing::instrument(name="upgrade_outlier", skip_all, fields(event_id=%incoming_pdu.event_id()))]
+	pub(super) async fn upgrade_outlier_to_timeline_pdu(
 		&self,
 		incoming_pdu: PduEvent,
 		mut val: CanonicalJsonObject,
-		create_event: &Pdu,
+		create_event: &PduEvent,
 		origin: &ServerName,
 		room_id: &RoomId,
-	) -> Result<Option<RawPduId>>
-	where
-		Pdu: Event + Send + Sync,
-	{
-		// Skip the PDU if we already have it as a timeline event
-		if let Ok(pduid) = self
-			.services
-			.timeline
-			.get_pdu_id(incoming_pdu.event_id())
-			.await
-		{
-			return Ok(Some(pduid));
-		}
-
-		let (rejected, soft_failed) = join!(
+	) -> Result<Option<RawPduId>> {
+		let (pduid, rejected, soft_failed) = join!(
+			self.services.timeline.get_pdu_id(incoming_pdu.event_id()),
 			self.services
 				.pdu_metadata
 				.is_event_rejected(incoming_pdu.event_id()),
@@ -52,17 +42,27 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
 				.pdu_metadata
 				.is_event_soft_failed(incoming_pdu.event_id())
 		);
-		if rejected {
-			return Err!(Request(InvalidParam("Event has been rejected")));
+		if let Ok(id) = pduid {
+			trace!(event_id=%incoming_pdu.event_id(), "Skipping upgrade of already upgraded PDU");
+			return Ok(Some(id));
+		} else if rejected {
+			return Err!(Request(Forbidden("Event has been rejected")));
 		} else if soft_failed {
-			return Err!(Request(InvalidParam("Event has been soft-failed")));
+			return Err!(Request(Forbidden("Event has been soft-failed")));
 		}
 
+		assert_eq!(
+			*create_event.kind(),
+			StateEventType::RoomCreate.into(),
+			"tried to upgrade a PDU with a create_event that is not a room create event"
+		);
+
 		debug!(
 			event_id = %incoming_pdu.event_id,
 			"Upgrading PDU from outlier to timeline"
 		);
 		let timer = Instant::now();
+		let min_depth = self.services.metadata.get_mindepth(room_id).await;
 		let room_version_rules = get_room_version_rules(create_event)?;
 
 		// 10. Fetch missing state and auth chain events by calling /state_ids at
@@ -73,21 +73,34 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
 			event_id = %incoming_pdu.event_id,
 			"Resolving state at event"
 		);
-		let mut state_at_incoming_event = if incoming_pdu.prev_events().count() == 1 {
+		let state_at_incoming_event = if incoming_pdu.prev_events().count() == 1 {
 			self.state_at_incoming_degree_one(&incoming_pdu).await?
 		} else {
 			self.state_at_incoming_resolved(&incoming_pdu, room_id, &room_version_rules)
 				.await?
 		};
+		let state_at_incoming_event = match state_at_incoming_event {
+			| Some(s) => s,
+			| None => {
+				trace!("Could not calculate incoming state, asking remote {origin} for it");
+				self.fetch_state(origin, create_event, room_id, incoming_pdu.event_id())
+					.await
+					.debug_inspect_err(|e| {
+						debug_error!("Could not fetch state from {origin}: {e}");
+					})?
+			},
+		};
 
-		if state_at_incoming_event.is_none() {
-			state_at_incoming_event = self
-				.fetch_state(origin, create_event, room_id, incoming_pdu.event_id())
-				.await?;
+		if state_at_incoming_event.is_empty()
+			&& *incoming_pdu.event_type() != StateEventType::RoomCreate.into()
+		{
+			// This can happen if the remote sends an event but cannot be reached to fetch
+			// the state at it, and all other servers in the room (which might just be the
+			// unreachable server) are unable to provide required info.
+			// returning an error here allows the upgrade to be attempted at another time.
+			return Err!(Request(Forbidden("Could not resolve incoming state at event")));
 		}
-
-		let state_at_incoming_event =
-			state_at_incoming_event.expect("we always set this to some above");
+		trace!(state_events = state_at_incoming_event.len(), "Calculated incoming state");
 
 		debug!(
 			event_id = %incoming_pdu.event_id,
@@ -385,6 +398,12 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu<Pdu>(
 
 		// Event has passed all auth/stateres checks
 		drop(state_lock);
+		if incoming_pdu.depth > min_depth && incoming_pdu.state_key().is_some() {
+			self.services
+				.metadata
+				.set_mindepth(room_id, incoming_pdu.depth.into());
+			trace!("Increased room's min depth from {} to {}", min_depth, incoming_pdu.depth);
+		}
 
 		Ok(pdu_id)
 	}

src/service/rooms/membership/mod.rs

@@ -1,7 +1,7 @@
 use std::{collections::HashMap, sync::Arc};
 
 use conduwuit::{
-	Err, Pdu, Result, Server, debug, debug_info, debug_warn, err, error, info, is_true,
+	Err, Event, Pdu, Result, Server, debug, debug_info, debug_warn, err, error, info, is_true,
 	matrix::{
 		StateKey,
 		event::{gen_event_id, gen_event_id_canonical_json},
@@ -34,7 +34,7 @@
 use crate::{
 	Dep, antispam, globals,
 	rooms::{
-		metadata, outlier, pdu_metadata, short,
+		event_handler, metadata, outlier, pdu_metadata, short,
 		state::{self, RoomMutexGuard},
 		state_accessor, state_cache,
 		state_compressor::{self, CompressedState, HashSetCompressStateEvent},
@@ -51,6 +51,7 @@ struct Services {
 	server: Arc<Server>,
 	db: Arc<Database>,
 	antispam: Dep<antispam::Service>,
+	event_handler: Dep<event_handler::Service>,
 	globals: Dep<globals::Service>,
 	metadata: Dep<metadata::Service>,
 	outlier: Dep<outlier::Service>,
@@ -73,6 +74,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				server: args.server.clone(),
 				db: args.db.clone(),
 				antispam: args.depend::<antispam::Service>("antispam"),
+				event_handler: args.depend::<event_handler::Service>("rooms::event_handler"),
 				globals: args.depend::<globals::Service>("globals"),
 				metadata: args.depend::<metadata::Service>("rooms::metadata"),
 				outlier: args.depend::<outlier::Service>("rooms::outlier"),
@@ -380,8 +382,6 @@ pub async fn join_remote_room(
 
 		// It has enough fields to be called a proper event now
 		let mut join_event = join_event_stub;
-
-		info!("Asking {remote_server} for send_join in room {room_id}");
 		let send_join_request = federation::membership::create_join_event::v2::Request::new(
 			room_id.to_owned(),
 			event_id.clone(),
@@ -391,6 +391,18 @@ pub async fn join_remote_room(
 				.await,
 		);
 
+		// NOTE: send_join can take a long time to respond, but from the point of view
+		// of other servers, we may already have finished joining. This means they
+		// sometimes end up sending PDUs to us that we aren't yet ready to accept, and
+		// consequently drop. Holding the mutex over the room while processing mitigates
+		// this.
+		let _room_lock = self
+			.services
+			.event_handler
+			.mutex_federation
+			.lock(room_id.as_str())
+			.await;
+		info!("Asking {remote_server} for send_join in room {room_id}");
 		let send_join_response = match self
 			.services
 			.sending
@@ -576,7 +588,13 @@ pub async fn join_remote_room(
 		if !auth_check {
 			return Err!(Request(Forbidden("Auth check failed")));
 		}
+		let resident_before = self
+			.services
+			.state_cache
+			.server_in_room(self.services.globals.server_name(), room_id)
+			.await;
 
+		let cork = self.services.db.cork_and_flush();
 		info!("Compressing state from send_join");
 		let compressed: CompressedState = self
 			.services
@@ -625,6 +643,10 @@ pub async fn join_remote_room(
 				room_id,
 			)
 			.await?;
+		self.services
+			.metadata
+			.maybe_set_mindepth(room_id, parsed_join_pdu.depth.into())
+			.await;
 
 		info!("Setting final room state for new room");
 		// We set the room state after inserting the pdu, so that we never have a moment
@@ -632,6 +654,23 @@ pub async fn join_remote_room(
 		self.services
 			.state
 			.set_room_state(room_id, statehash_after_join, &state_lock);
+		if !resident_before {
+			// NOTE: We replace local extremities for this room if we were not a resident
+			// before. We might be doing a remote join to satisfy restricted join rules,
+			// so we don't want to do this if we're already a resident. Otherwise, we
+			// want to replace our forward extremities whole-sale in case we were
+			// desynced.
+			info!("Replacing local forward extremities");
+			self.services
+				.state
+				.set_forward_extremities(
+					room_id,
+					std::iter::once(parsed_join_pdu.event_id()),
+					&state_lock,
+				)
+				.await;
+		}
+		drop(cork);
 
 		Ok(())
 	}

src/service/rooms/metadata/mod.rs

@@ -1,9 +1,9 @@
 use std::sync::Arc;
 
 use conduwuit::{Result, utils::stream::TryIgnore};
-use database::Map;
+use database::{Deserialized, Map};
 use futures::{Stream, StreamExt};
-use ruma::{OwnedRoomId, RoomId};
+use ruma::{OwnedRoomId, RoomId, UInt, uint};
 
 use crate::{Dep, rooms};
 
@@ -17,6 +17,7 @@ struct Data {
 	bannedroomids: Arc<Map>,
 	roomid_shortroomid: Arc<Map>,
 	pduid_pdu: Arc<Map>,
+	roomid_mindepth: Arc<Map>,
 }
 
 struct Services {
@@ -31,6 +32,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				bannedroomids: args.db["bannedroomids"].clone(),
 				roomid_shortroomid: args.db["roomid_shortroomid"].clone(),
 				pduid_pdu: args.db["pduid_pdu"].clone(),
+				roomid_mindepth: args.db["roomid_mindepth"].clone(),
 			},
 			services: Services {
 				short: args.depend::<rooms::short::Service>("rooms::short"),
@@ -98,4 +100,25 @@ pub async fn is_banned(&self, room_id: &RoomId) -> bool {
 	pub fn list_banned_rooms(&self) -> impl Stream<Item = OwnedRoomId> + Send + '_ {
 		self.db.bannedroomids.keys().ignore_err()
 	}
+
+	pub async fn get_mindepth(&self, room_id: &RoomId) -> UInt {
+		self.db
+			.roomid_mindepth
+			.get(room_id)
+			.await
+			.deserialized::<UInt>()
+			.unwrap_or_else(|_| uint!(0))
+	}
+
+	pub fn set_mindepth(&self, room_id: &RoomId, min_depth: u64) {
+		self.db
+			.roomid_mindepth
+			.put_raw(room_id.as_bytes(), min_depth.to_be_bytes());
+	}
+
+	pub async fn maybe_set_mindepth(&self, room_id: &RoomId, min_depth: u64) {
+		if min_depth > self.get_mindepth(room_id).await.into() {
+			self.set_mindepth(room_id, min_depth);
+		}
+	}
 }

src/service/rooms/state/mod.rs

@@ -371,6 +371,16 @@ pub async fn get_room_shortstatehash(&self, room_id: &RoomId) -> Result<ShortSta
 			.deserialized()
 	}
 
+	pub fn all_forward_extremities(
+		&self,
+	) -> impl Stream<Item = (OwnedRoomId, OwnedEventId)> + Send {
+		self.db
+			.roomid_pduleaves
+			.keys()
+			.map_ok(|(room_id, event_id): (OwnedRoomId, OwnedEventId)| (room_id, event_id))
+			.ignore_err()
+	}
+
 	pub fn get_forward_extremities<'a>(
 		&'a self,
 		room_id: &'a RoomId,

src/service/rooms/state_cache/mod.rs

@@ -248,7 +248,7 @@ pub fn active_local_users_in_room<'a>(
 	) -> impl Stream<Item = OwnedUserId> + Send + 'a {
 		self.local_users_in_room(room_id)
 			.filter_map(async |user_id| {
-				if self.services.users.is_active(&user_id).await {
+				if self.services.users.status(&user_id).await.is_active() {
 					Some(user_id)
 				} else {
 					None

src/service/rooms/state_cache/update.rs

@@ -30,15 +30,8 @@ pub async fn update_membership(
 	) -> Result {
 		let membership = pdu.get_content::<RoomMemberEventContent>()?;
 
-		// Keep track what remote users exist by adding them as "deactivated" users
-		//
-		// TODO: use futures to update remote profiles without blocking the membership
-		// update
-		#[allow(clippy::collapsible_if)]
-		if !self.services.globals.user_is_local(user_id)
-			&& !self.services.users.exists(user_id).await
-		{
-			self.services.users.create(user_id, None)?;
+		if !self.services.globals.user_is_local(user_id) {
+			self.services.users.record_remote_user(user_id);
 		}
 
 		match &membership.membership {

src/service/rooms/timeline/append.rs

@@ -20,7 +20,7 @@
 };
 use futures::{StreamExt, TryFutureExt};
 use ruma::{
-	CanonicalJsonObject, CanonicalJsonValue, EventId, RoomVersionId, UserId,
+	CanonicalJsonObject, CanonicalJsonValue, EventId, RoomId, RoomVersionId, UserId,
 	events::{
 		GlobalAccountDataEventType, TimelineEventType,
 		push_rules::PushRulesEvent,
@@ -44,7 +44,7 @@ pub async fn append_incoming_pdu<'a, Leaves>(
 		state_ids_compressed: Arc<CompressedState>,
 		soft_fail: bool,
 		state_lock: &'a RoomMutexGuard,
-		room_id: &'a ruma::RoomId,
+		room_id: &'a RoomId,
 	) -> Result<Option<RawPduId>>
 	where
 		Leaves: Iterator<Item = &'a EventId> + Send + 'a,
@@ -147,7 +147,7 @@ pub async fn append_pdu<'a, Leaves>(
 		mut pdu_json: CanonicalJsonObject,
 		leaves: Leaves,
 		state_lock: &'a RoomMutexGuard,
-		room_id: &'a ruma::RoomId,
+		room_id: &'a RoomId,
 	) -> Result<RawPduId>
 	where
 		Leaves: Iterator<Item = &'a EventId> + Send + 'a,
@@ -210,8 +210,8 @@ pub async fn append_pdu<'a, Leaves>(
 		// See if the event matches any known pushers via power level
 		if *pdu.kind() != TimelineEventType::RoomCreate {
 			tokio::join!(
-				self.notify_local_users(pdu, &pdu_id),
-				self.handle_pdu_effects(pdu, &pdu_id, shortroomid)
+				self.notify_local_users(pdu, &pdu_id, room_id),
+				self.handle_pdu_effects(pdu, &pdu_id, room_id, shortroomid)
 					.inspect_err(|e| {
 						error!(
 							"failed to handle PDU effects of incoming PDU {}: {e:?}",
@@ -223,22 +223,23 @@ pub async fn append_pdu<'a, Leaves>(
 			);
 		}
 
-		self.send_to_interested_appservices(pdu, &pdu_id).await;
+		self.send_to_interested_appservices(pdu, &pdu_id, room_id)
+			.await;
 
 		Ok(pdu_id)
 	}
 
 	/// Notifies local users of the incoming event with a power levels context.
-	async fn notify_local_users(&self, pdu: &PduEvent, pdu_id: &RawPduId) {
+	async fn notify_local_users(&self, pdu: &PduEvent, pdu_id: &RawPduId, room_id: &RoomId) {
 		let power_levels = self
 			.services
 			.state_accessor
-			.get_room_power_levels(pdu.room_id().unwrap())
+			.get_room_power_levels(room_id)
 			.await;
 		let mut push_targets: HashSet<_> = self
 			.services
 			.state_cache
-			.active_local_users_in_room(pdu.room_id().unwrap())
+			.active_local_users_in_room(room_id)
 			// Don't notify the sender of their own events, and don't send from ignored users
 			.ready_filter(|user| *user != pdu.sender())
 			.filter_map(|recipient_user| async move {
@@ -254,7 +255,13 @@ async fn notify_local_users(&self, pdu: &PduEvent, pdu_id: &RawPduId) {
 			if let Some(state_key) = pdu.state_key() {
 				match UserId::parse(state_key) {
 					| Ok(target_user_id) => {
-						if self.services.users.is_active_local(&target_user_id).await {
+						if self
+							.services
+							.users
+							.status(&target_user_id)
+							.await
+							.is_active()
+						{
 							push_targets.insert(target_user_id.clone());
 						}
 					},
@@ -285,13 +292,7 @@ async fn notify_local_users(&self, pdu: &PduEvent, pdu_id: &RawPduId) {
 			for action in self
 				.services
 				.pusher
-				.get_actions(
-					user,
-					&rules_for_user,
-					power_levels.clone(),
-					&serialized,
-					pdu.room_id().unwrap(),
-				)
+				.get_actions(user, &rules_for_user, power_levels.clone(), &serialized, room_id)
 				.await
 			{
 				match action {
@@ -331,7 +332,7 @@ async fn notify_local_users(&self, pdu: &PduEvent, pdu_id: &RawPduId) {
 		}
 
 		self.db
-			.increment_notification_counts(pdu.room_id().unwrap(), notifies, highlights);
+			.increment_notification_counts(room_id, notifies, highlights);
 	}
 
 	/// Handles PDU effects based on the type of incoming event.
@@ -341,9 +342,9 @@ async fn handle_pdu_effects(
 		&self,
 		pdu: &PduEvent,
 		pdu_id: &RawPduId,
+		room_id: &RoomId,
 		short_room_id: ShortRoomId,
 	) -> Result {
-		let room_id = pdu.room_id().unwrap();
 		match *pdu.kind() {
 			| TimelineEventType::RoomRedaction => {
 				use RoomVersionId::*;
@@ -456,8 +457,17 @@ async fn is_appservice_interested(
 		&self,
 		appservice: &RegistrationInfo,
 		pdu: &PduEvent,
+		room_id: &RoomId,
 	) -> bool {
-		let room_id = pdu.room_id().unwrap();
+		if self
+			.services
+			.state_cache
+			.appservice_in_room(room_id, appservice)
+			.await
+		{
+			return true;
+		}
+
 		let target = if *pdu.kind() == TimelineEventType::RoomMember {
 			pdu.state_key().and_then(|sk| UserId::parse(sk).ok())
 		} else {
@@ -472,8 +482,13 @@ async fn is_appservice_interested(
 			| _ => (false, false),
 		};
 		let sender_matches_namespace = appservice.users.is_match(pdu.sender().as_str());
+		let room_matches_namespace = appservice.rooms.is_match(room_id.as_str());
 
-		if target_matches_sender || target_matches_namespace || sender_matches_namespace {
+		if target_matches_sender
+			|| target_matches_namespace
+			|| sender_matches_namespace
+			|| room_matches_namespace
+		{
 			return true;
 		}
 
@@ -486,7 +501,12 @@ async fn is_appservice_interested(
 	}
 
 	/// Notifies interested appservices of a new PDU.
-	async fn send_to_interested_appservices(&self, pdu: &PduEvent, pdu_id: &RawPduId) {
+	async fn send_to_interested_appservices(
+		&self,
+		pdu: &PduEvent,
+		pdu_id: &RawPduId,
+		room_id: &RoomId,
+	) {
 		let interested_appservices = self
 			.services
 			.appservice
@@ -498,7 +518,7 @@ async fn send_to_interested_appservices(&self, pdu: &PduEvent, pdu_id: &RawPduId
 		interested_appservices
 			.stream()
 			.broad_filter_map(|appservice| async move {
-				self.is_appservice_interested(&appservice, pdu)
+				self.is_appservice_interested(&appservice, pdu, room_id)
 					.await
 					.then_some(appservice)
 			})

src/service/rooms/timeline/backfill.rs

@@ -224,7 +224,10 @@ pub async fn backfill_pdu(&self, origin: &ServerName, pdu: Box<RawJsonValue>) ->
 
 	/// Determines which servers are trusted enough to provide backfill in a
 	/// room.
-	async fn candidate_backfill_servers(&self, room_id: &RoomId) -> HashSet<OwnedServerName> {
+	pub(crate) async fn candidate_backfill_servers(
+		&self,
+		room_id: &RoomId,
+	) -> HashSet<OwnedServerName> {
 		let mut candidate_backfill_servers = HashSet::new();
 
 		let power_levels = self

src/service/rooms/timeline/mod.rs

@@ -173,6 +173,10 @@ pub async fn get_non_outlier_pdu_json(
 		self.db.get_non_outlier_pdu_json(event_id).await
 	}
 
+	pub async fn non_outlier_pdu_exists(&self, event_id: &EventId) -> bool {
+		self.db.non_outlier_pdu_exists(event_id).await.is_ok()
+	}
+
 	/// Returns the pdu's id.
 	#[inline]
 	pub async fn get_pdu_id(&self, event_id: &EventId) -> Result<RawPduId> {

src/service/services.rs

@@ -11,8 +11,8 @@
 	account_data, admin, announcements, antispam, appservice, client, config, emergency,
 	federation, firstrun, globals, key_backups, mailer,
 	manager::Manager,
-	media, moderation, oauth, presence, pusher, registration_tokens, resolver, rooms, sending,
-	server_keys,
+	media, moderation, oauth, oidc, presence, pusher, registration_tokens, resolver, rooms,
+	sending, server_keys,
 	service::{self, Args, Map, Service},
 	sync, threepid, transactions, uiaa, users,
 };
@@ -28,6 +28,7 @@ pub struct Services {
 	pub key_backups: Arc<key_backups::Service>,
 	pub media: Arc<media::Service>,
 	pub oauth: Arc<oauth::Service>,
+	pub oidc: Arc<oidc::Service>,
 	pub mailer: Arc<mailer::Service>,
 	pub presence: Arc<presence::Service>,
 	pub pusher: Arc<pusher::Service>,
@@ -85,6 +86,7 @@ macro_rules! build {
 			key_backups: build!(key_backups::Service),
 			media: build!(media::Service),
 			oauth: build!(oauth::Service),
+			oidc: build!(oidc::Service),
 			mailer: build!(mailer::Service),
 			presence: build!(presence::Service),
 			pusher: build!(pusher::Service),

src/service/threepid/mod.rs

@@ -1,6 +1,6 @@
 use std::{borrow::Cow, collections::HashMap, sync::Arc};
 
-use conduwuit::{Err, Error, Result, result::FlatOk};
+use conduwuit::{Err, Error, Result, config::OidcProfileKeyImportMode, result::FlatOk};
 use database::{Deserialized, Map};
 use governor::{DefaultKeyedRateLimiter, Quota, RateLimiter};
 use lettre::{Address, message::Mailbox};
@@ -33,7 +33,9 @@ pub enum EmailRequirement {
 	Required,
 	/// Users may change or remove their email.
 	Optional,
-	/// Users may not change their email at all.
+	/// Users may see their email, but may not change it.
+	Locked,
+	/// Email features are disabled.
 	Unavailable,
 }
 
@@ -43,6 +45,9 @@ pub fn may_change(&self) -> bool { matches!(self, Self::Required | Self::Optiona
 
 	#[must_use]
 	pub fn may_remove(&self) -> bool { matches!(self, Self::Optional) }
+
+	#[must_use]
+	pub fn may_view(&self) -> bool { !matches!(self, Self::Unavailable) }
 }
 
 struct Data {
@@ -87,7 +92,14 @@ impl Service {
 	/// Check if users are required to have an email address.
 	pub fn email_requirement(&self) -> EmailRequirement {
 		if let Some(smtp) = &self.services.config.smtp {
-			if smtp.require_email_for_registration || smtp.require_email_for_token_registration {
+			if let Some(oidc) = &self.services.config.oauth.oidc
+				&& matches!(oidc.profile_key_import_mode, OidcProfileKeyImportMode::OnLogin)
+				&& oidc.email_claim.is_some()
+			{
+				EmailRequirement::Locked
+			} else if smtp.require_email_for_registration
+				|| smtp.require_email_for_token_registration
+			{
 				EmailRequirement::Required
 			} else {
 				EmailRequirement::Optional

src/service/uiaa/mod.rs

@@ -314,7 +314,7 @@ async fn create_session(
 				.services
 				.config
 				.oauth
-				.compatibility_mode
+				.compatibility_mode()
 				.uiaa_available()
 		{
 			return Err!(Request(Unrecognized(

src/service/users/account.rs

@@ -6,21 +6,23 @@
 	warn,
 };
 use database::{Deserialized, Json};
-use futures::{FutureExt, Stream, StreamExt, TryFutureExt};
+use futures::{FutureExt, Stream, StreamExt};
 use lettre::Address;
 use ruma::{
 	MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedUserId, UserId,
+	api::client::profile::PropagateTo,
 	events::{
 		GlobalAccountDataEventType, ignored_user_list::IgnoredUserListEvent,
 		push_rules::PushRulesEvent, room::message::RoomMessageEventContent,
 	},
+	profile::ProfileFieldValue,
 	push::Ruleset,
 };
 use ruminuwuity::invite_permission_config::{FilterLevel, InvitePermissionConfigEvent};
 
 use crate::{
 	appservice::RegistrationInfo,
-	users::{HashedPassword, UserSuspension},
+	users::{HashedPassword, UserSuspension, profile::ProfileFieldChange},
 };
 
 /// The status of an access token.
@@ -29,6 +31,30 @@ pub enum AccessTokenStatus {
 	Expired,
 }
 
+/// The status of a user account.
+#[derive(Clone, Copy)]
+pub enum AccountStatus {
+	NotFound,
+	Active,
+	Deactivated,
+}
+
+impl AccountStatus {
+	#[must_use]
+	pub fn is_found(&self) -> bool { !matches!(self, Self::NotFound) }
+
+	#[must_use]
+	pub fn is_active(&self) -> bool { matches!(self, Self::Active) }
+
+	pub fn ensure_active(&self) -> Result<()> {
+		match self {
+			| Self::Active => Ok(()),
+			| Self::Deactivated => Err!(Request(UserDeactivated("This account is deactivated."))),
+			| Self::NotFound => Err!(Request(NotFound("This account does not exist."))),
+		}
+	}
+}
+
 impl super::Service {
 	/// Returns true/false based on whether the recipient/receiving user has
 	/// ignored the sender.
@@ -89,29 +115,33 @@ pub async fn is_admin(&self, user_id: &UserId) -> bool {
 		self.services.admin.user_is_admin(user_id).await
 	}
 
-	/// Create a new user account on this homeserver. Set the password to `None`
-	/// to create a non-local user. Non-local users with a password will return
-	/// an error.
-	#[inline]
-	pub fn create(&self, user_id: &UserId, password: Option<HashedPassword>) -> Result<()> {
-		if !self.services.globals.user_is_local(user_id) && password.is_some() {
-			return Err!("Cannot create a nonlocal user with a set password");
+	/// Create a new shadow user account on this server. Shadow accounts
+	/// have no password and cannot be logged into.
+	pub async fn create_shadow_account(&self, user_id: &UserId) -> Result<()> {
+		assert!(self.services.globals.user_is_local(user_id), "user id must be local");
+
+		if self.status(user_id).await.is_found() {
+			return Err!(Request(UserInUse("An account with this user ID already exists.")));
 		}
 
-		self.set_password(user_id, password);
+		self.db.userid_password.insert(user_id, "");
 
 		Ok(())
 	}
 
-	/// Create a new account for a local human or bot user.
+	/// Create a new account for a local human or bot user. If `password` is
+	/// None, the account will be a shadow account.
 	pub async fn create_local_account(
 		&self,
 		user_id: &UserId,
-		password: HashedPassword,
+		password: Option<HashedPassword>,
 		email: Option<Address>,
-	) {
-		self.create(user_id, Some(password))
-			.expect("should be able to save a new local user. what happened?");
+	) -> Result<()> {
+		self.create_shadow_account(user_id).await?;
+
+		if let Some(password) = password {
+			self.convert_to_local_account(user_id, password).await?;
+		}
 
 		// Set an initial display name
 		{
@@ -123,7 +153,13 @@ pub async fn create_local_account(
 				displayname.push_str(suffix);
 			}
 
-			self.set_displayname(user_id, Some(displayname));
+			self.set_profile_field(
+				user_id,
+				ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)),
+				PropagateTo::None,
+			)
+			.await
+			.expect("should be able to set display name");
 		};
 
 		// Set default push rules
@@ -230,6 +266,8 @@ pub async fn create_local_account(
 		}
 
 		info!("Created new user account for {user_id}");
+
+		Ok(())
 	}
 
 	pub async fn determine_registration_user_id(
@@ -268,28 +306,7 @@ pub async fn determine_registration_user_id(
 				&supplied_username,
 				self.services.globals.server_name(),
 			) {
-				| Ok(user_id) => {
-					if let Err(e) = user_id.validate_strict() {
-						// Unless we are in emergency mode, we should follow synapse's behaviour
-						// on not allowing things like spaces and UTF-8 characters in
-						// usernames
-						if !emergency_mode_enabled {
-							return Err!(Request(InvalidUsername(debug_warn!(
-								"Username {supplied_username} contains disallowed characters or \
-								 spaces: {e}"
-							))));
-						}
-					}
-
-					// Don't allow registration with user IDs that aren't local
-					if !self.services.globals.user_is_local(&user_id) {
-						return Err!(Request(InvalidUsername(
-							"Username {supplied_username} is not local to this server"
-						)));
-					}
-
-					user_id
-				},
+				| Ok(user_id) => user_id,
 				| Err(e) => {
 					return Err!(Request(InvalidUsername(debug_warn!(
 						"Username {supplied_username} is not valid: {e}"
@@ -297,8 +314,27 @@ pub async fn determine_registration_user_id(
 				},
 			};
 
-			if self.exists(&user_id).await {
-				return Err!(Request(UserInUse("User ID is not available.")));
+			if let Err(e) = user_id.validate_strict() {
+				// Unless we are in emergency mode, we should follow synapse's behaviour
+				// on not allowing things like spaces and UTF-8 characters in
+				// usernames
+				if !emergency_mode_enabled {
+					return Err!(Request(InvalidUsername(debug_warn!(
+						"Username {supplied_username} contains disallowed characters or spaces: \
+						 {e}"
+					))));
+				}
+			}
+
+			// Don't allow registration with user IDs that aren't local
+			if !self.services.globals.user_is_local(&user_id) {
+				return Err!(Request(InvalidUsername(
+					"Username {supplied_username} is not local to this server"
+				)));
+			}
+
+			if self.status(&user_id).await.is_found() {
+				return Err!(Request(UserInUse("Username is not available.")));
 			}
 
 			// Check that the user ID is/is not in an appservice's namespace
@@ -329,26 +365,24 @@ pub async fn determine_registration_user_id(
 				)
 				.unwrap();
 
-				if !self.exists(&user_id).await {
+				if !self.status(&user_id).await.is_found() {
 					break Ok(user_id);
 				}
 			}
 		}
 	}
 
-	/// Deactivates an account, removing all of their device IDs and unsetting
-	/// their password.
+	/// Deactivates an account, removing all of their device IDs.
 	pub async fn deactivate_account(&self, user_id: &UserId) -> Result<()> {
+		self.status(user_id).await.ensure_active()?;
+
 		// Remove all associated devices
 		self.all_device_ids(user_id)
 			.for_each(async |device_id| self.remove_device(user_id, &device_id).await)
 			.await;
 
-		// Set the password to "" to indicate a deactivated account. Hashes will never
-		// result in an empty string, so the user will not be able to log in again.
-		// Systems like changing the password without logging in should check if the
-		// account is deactivated.
-		self.set_password(user_id, None);
+		// Mark user as deactivated
+		self.db.userid_deactivated.insert(user_id, "");
 
 		// TODO: Unhook 3PID
 		Ok(())
@@ -393,25 +427,6 @@ pub async fn lock_account(&self, user_id: &UserId, locking_user: &UserId) {
 	/// Unlocks an account, allowing the user to log in and use it again.
 	pub async fn unlock_account(&self, user_id: &UserId) { self.db.userid_lock.remove(user_id); }
 
-	/// Check if the provided user ID belongs to an existing (possibly
-	/// deactivated) account on this homeserver.
-	#[inline]
-	pub async fn exists(&self, user_id: &UserId) -> bool {
-		self.services.globals.user_is_local(user_id)
-			&& self.db.userid_password.get(user_id).await.is_ok()
-	}
-
-	/// Check if account is deactivated (has an empty password). Returns a
-	/// NotFound error if the user does not exist.
-	pub async fn is_deactivated(&self, user_id: &UserId) -> Result<bool> {
-		self.db
-			.userid_password
-			.get(user_id)
-			.map_ok(|val| val.is_empty())
-			.map_err(|_| err!(Request(NotFound("User does not exist."))))
-			.await
-	}
-
 	/// Check if account is suspended. Returns false if the user does not exist.
 	pub async fn is_suspended(&self, user_id: &UserId) -> Result<bool> {
 		match self
@@ -469,18 +484,33 @@ pub async fn is_login_disabled(&self, user_id: &UserId) -> bool {
 			.is_ok()
 	}
 
-	/// Check if account is active (not deactivated)
-	pub async fn is_active(&self, user_id: &UserId) -> bool {
-		!self.is_deactivated(user_id).await.unwrap_or(true)
+	/// Checks the activation status of the provided user ID's account.
+	pub async fn status(&self, user_id: &UserId) -> AccountStatus {
+		if !self.services.globals.user_is_local(user_id) {
+			AccountStatus::NotFound
+		} else if self.db.userid_password.exists(user_id).await.is_ok() {
+			if self.db.userid_deactivated.exists(user_id).await.is_ok() {
+				AccountStatus::Deactivated
+			} else {
+				AccountStatus::Active
+			}
+		} else {
+			AccountStatus::NotFound
+		}
 	}
 
-	/// Check if account is a local user, and is active (not deactivated)
-	pub async fn is_active_local(&self, user_id: &UserId) -> bool {
-		self.services.globals.user_is_local(user_id) && self.is_active(user_id).await
+	/// Checks if a user is a shadow.
+	pub async fn is_shadow(&self, user_id: &UserId) -> bool {
+		self.db
+			.userid_password
+			.get(user_id)
+			.await
+			.deserialized::<String>()
+			.is_ok_and(|hash| hash.is_empty())
 	}
 
 	/// Returns the number of users registered on this server, including
-	/// deactivated users.
+	/// deactivated and shadow users.
 	#[inline]
 	pub async fn count(&self) -> usize { self.db.userid_password.count().await }
 
@@ -522,8 +552,8 @@ pub async fn find_from_token(
 		Some((user_id, device_id, AccessTokenStatus::Valid))
 	}
 
-	/// Returns an iterator over all users on this homeserver.
-	pub fn stream(&self) -> impl Stream<Item = OwnedUserId> + Send {
+	/// Returns an iterator over all local users on this homeserver.
+	pub fn stream_local_users(&self) -> impl Stream<Item = OwnedUserId> + Send {
 		self.db.userid_password.keys().ignore_err()
 	}
 
@@ -540,16 +570,49 @@ pub fn list_local_users(&self) -> impl Stream<Item = OwnedUserId> + Send + '_ {
 	}
 
 	/// Set a user's password.
-	pub fn set_password(&self, user_id: &UserId, password: Option<HashedPassword>) {
-		if let Some(hash) = password {
-			self.db.userid_password.insert(user_id, hash.0);
-		} else {
-			self.db.userid_password.insert(user_id, b"");
+	pub async fn set_password(&self, user_id: &UserId, password: HashedPassword) -> Result {
+		self.status(user_id).await.ensure_active()?;
+
+		self.db.userid_password.insert(user_id, password.0);
+
+		Ok(())
+	}
+
+	/// Convert an existing user to an active local account. This will turn
+	/// shadow accounts into normal accounts and will clear the deactivation
+	/// flag on the user if it is set.
+	pub async fn convert_to_local_account(
+		&self,
+		user_id: &UserId,
+		password: HashedPassword,
+	) -> Result {
+		if !self.status(user_id).await.is_found() {
+			return Err!(Request(NotFound("This account does not exist.")));
 		}
+
+		self.db.userid_deactivated.remove(user_id);
+		self.db.userid_password.insert(user_id, password.0);
+
+		Ok(())
+	}
+
+	/// Convert an existing user to a shadow account. This will clear their
+	/// password and reactivate them if they are deactivated.
+	pub async fn convert_to_shadow_account(&self, user_id: &UserId) -> Result {
+		if !self.status(user_id).await.is_found() {
+			return Err!(Request(NotFound("This account does not exist.")));
+		}
+
+		self.db.userid_deactivated.remove(user_id);
+		self.db.userid_password.insert(user_id, "");
+
+		Ok(())
 	}
 
 	/// Check a user's password.
 	pub async fn check_password(&self, user_id: &UserId, password: &str) -> Result<OwnedUserId> {
+		self.status(user_id).await.ensure_active()?;
+
 		let (hash, user_id): (String, OwnedUserId) =
 			if let Ok(hash) = self.db.userid_password.get(user_id).await.deserialized() {
 				(hash, user_id.to_owned())
@@ -558,21 +621,20 @@ pub async fn check_password(&self, user_id: &UserId, password: &str) -> Result<O
 				// better
 				let lowercase_user_id = UserId::parse(user_id.as_str().to_lowercase()).unwrap();
 
-				if let Ok(hash) = self
+				let hash = self
 					.db
 					.userid_password
 					.get(lowercase_user_id.as_str())
 					.await
 					.deserialized()
-				{
-					(hash, lowercase_user_id)
-				} else {
-					return Err!(Request(Forbidden("This user cannot log in with a password.")));
-				}
+					.expect("user should exist");
+
+				(hash, lowercase_user_id)
 			};
 
 		if hash.is_empty() {
-			return Err!(Request(UserDeactivated("This user is deactivated")));
+			// Cannot log into shadow users
+			return Err!(Request(Forbidden("This user cannot log in with a password.")));
 		}
 
 		utils::hash::verify_password(password, &hash)

src/service/users/dehydrated_device.rs

@@ -21,10 +21,7 @@ pub struct DehydratedDevice {
 impl super::Service {
 	/// Creates or recreates the user's dehydrated device.
 	pub async fn set_dehydrated_device(&self, user_id: &UserId, request: Request) -> Result {
-		assert!(
-			self.exists(user_id).await,
-			"Tried to create dehydrated device for non-existent user"
-		);
+		self.status(user_id).await.ensure_active()?;
 
 		let existing_id = self.get_dehydrated_device_id(user_id).await;
 

src/service/users/device.rs

@@ -4,8 +4,8 @@
 };
 
 use conduwuit::{
-	Err, utils,
-	utils::{ReadyExt, stream::TryIgnore},
+	Err, Result,
+	utils::{self, ReadyExt, stream::TryIgnore},
 };
 use database::{Deserialized, Ignore, Interfix, Json};
 use futures::{Stream, StreamExt};
@@ -18,8 +18,7 @@
 use crate::users::increment;
 
 impl super::Service {
-	/// Adds a new device to a user. The user must exist, otherwise InvalidParam
-	/// is returned.
+	/// Adds a new device to a user.
 	pub async fn create_device(
 		&self,
 		user_id: &UserId,
@@ -28,12 +27,8 @@ pub async fn create_device(
 		token_max_age: Option<Duration>,
 		initial_device_display_name: Option<String>,
 		client_ip: Option<String>,
-	) -> conduwuit::Result<()> {
-		if !self.exists(user_id).await {
-			return Err!(Request(InvalidParam(error!(
-				"Called create_device for non-existent user {user_id}"
-			))));
-		}
+	) -> Result<()> {
+		self.status(user_id).await.ensure_active()?;
 
 		let key = (user_id, device_id);
 		let mut device = Device::new(device_id.into());
@@ -50,7 +45,7 @@ pub async fn create_device(
 	/// Removes a device from a user.
 	pub async fn remove_device(&self, user_id: &UserId, device_id: &DeviceId) {
 		// Remove dehydrated device if this is the dehydrated device
-		let _: conduwuit::Result<_> = self
+		let _ = self
 			.remove_dehydrated_device(user_id, Some(device_id))
 			.await;
 
@@ -97,11 +92,7 @@ pub fn all_device_ids<'a>(
 	}
 
 	/// Gets the access token associated with a device.
-	pub async fn get_token(
-		&self,
-		user_id: &UserId,
-		device_id: &DeviceId,
-	) -> conduwuit::Result<String> {
+	pub async fn get_token(&self, user_id: &UserId, device_id: &DeviceId) -> Result<String> {
 		let key = (user_id, device_id);
 		self.db.userdeviceid_token.qry(&key).await.deserialized()
 	}
@@ -131,7 +122,7 @@ pub async fn set_token(
 		device_id: &DeviceId,
 		token: &str,
 		token_max_age: Option<Duration>,
-	) -> conduwuit::Result<()> {
+	) -> Result<()> {
 		let key = (user_id, device_id);
 		if self.db.userdeviceid_metadata.qry(&key).await.is_err() {
 			return Err!(Database(error!(
@@ -259,7 +250,7 @@ pub async fn update_device_metadata(
 		user_id: &UserId,
 		device_id: &DeviceId,
 		device: &Device,
-	) -> conduwuit::Result<()> {
+	) -> Result<()> {
 		increment(&self.db.userid_devicelistversion, user_id.as_bytes());
 		self.update_device_metadata_no_increment(user_id, device_id, device)
 	}
@@ -273,7 +264,7 @@ fn update_device_metadata_no_increment(
 		user_id: &UserId,
 		device_id: &DeviceId,
 		device: &Device,
-	) -> conduwuit::Result<()> {
+	) -> Result<()> {
 		let key = (user_id, device_id);
 		self.db.userdeviceid_metadata.put(key, Json(device));
 
@@ -312,7 +303,7 @@ pub async fn get_device_metadata(
 		&self,
 		user_id: &UserId,
 		device_id: &DeviceId,
-	) -> conduwuit::Result<Device> {
+	) -> Result<Device> {
 		self.db
 			.userdeviceid_metadata
 			.qry(&(user_id, device_id))
@@ -321,7 +312,7 @@ pub async fn get_device_metadata(
 	}
 
 	/// Gets the most recent device list version for a user.
-	pub async fn get_devicelist_version(&self, user_id: &UserId) -> conduwuit::Result<u64> {
+	pub async fn get_devicelist_version(&self, user_id: &UserId) -> Result<u64> {
 		self.db
 			.userid_devicelistversion
 			.get(user_id)

src/service/users/mod.rs

@@ -4,20 +4,22 @@
 pub(super) mod filters;
 pub(super) mod keys;
 pub(super) mod profile;
+pub(super) mod remote;
 
 use std::{mem, sync::Arc};
 
-pub use account::AccessTokenStatus;
+pub use account::{AccessTokenStatus, AccountStatus};
 use conduwuit::{
 	Err, Error, Result, err,
 	utils::{self},
 };
 use database::Map;
+pub use profile::ProfileFieldChange;
 use ruma::{UserId, api::error::ErrorKind, encryption::CrossSigningKey, serde::Raw};
 use serde::{Deserialize, Serialize};
 
 use crate::{
-	Dep, account_data, admin, appservice, config, firstrun, globals, oauth,
+	Dep, account_data, admin, appservice, config, firstrun, globals, oauth, presence,
 	rooms::{self, alias, membership},
 	threepid,
 };
@@ -61,9 +63,12 @@ struct Services {
 	globals: Dep<globals::Service>,
 	membership: Dep<membership::Service>,
 	oauth: Dep<oauth::Service>,
+	presence: Dep<presence::Service>,
+	state: Dep<rooms::state::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
 	state_cache: Dep<rooms::state_cache::Service>,
 	threepid: Dep<threepid::Service>,
+	timeline: Dep<rooms::timeline::Service>,
 }
 
 struct Data {
@@ -75,11 +80,13 @@ struct Data {
 	logintoken_expiresatuserid: Arc<Map>,
 	todeviceid_events: Arc<Map>,
 	token_userdeviceid: Arc<Map>,
+	remoteuserid_remoteuser: Arc<Map>,
 	userdeviceid_tokenexpires: Arc<Map>,
 	userdeviceid_metadata: Arc<Map>,
 	userdeviceid_token: Arc<Map>,
 	userfilterid_filter: Arc<Map>,
 	userid_avatarurl: Arc<Map>,
+	userid_deactivated: Arc<Map>,
 	userid_dehydrateddevice: Arc<Map>,
 	userid_devicelistversion: Arc<Map>,
 	userid_displayname: Arc<Map>,
@@ -107,10 +114,13 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				globals: args.depend::<globals::Service>("globals"),
 				membership: args.depend::<membership::Service>("membership"),
 				oauth: args.depend::<oauth::Service>("oauth"),
+				presence: args.depend::<presence::Service>("presence"),
+				state: args.depend::<rooms::state::Service>("rooms::state"),
 				state_accessor: args
 					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
 				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
 				threepid: args.depend::<threepid::Service>("threepid"),
+				timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
 			},
 			db: Data {
 				keychangeid_userid: args.db["keychangeid_userid"].clone(),
@@ -121,10 +131,12 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				logintoken_expiresatuserid: args.db["logintoken_expiresatuserid"].clone(),
 				todeviceid_events: args.db["todeviceid_events"].clone(),
 				token_userdeviceid: args.db["token_userdeviceid"].clone(),
+				remoteuserid_remoteuser: args.db["remoteuserid_remoteuser"].clone(),
 				userdeviceid_metadata: args.db["userdeviceid_metadata"].clone(),
 				userdeviceid_token: args.db["userdeviceid_token"].clone(),
 				userfilterid_filter: args.db["userfilterid_filter"].clone(),
 				userid_avatarurl: args.db["userid_avatarurl"].clone(),
+				userid_deactivated: args.db["userid_deactivated"].clone(),
 				userid_dehydrateddevice: args.db["userid_dehydrateddevice"].clone(),
 				userid_devicelistversion: args.db["userid_devicelistversion"].clone(),
 				userid_displayname: args.db["userid_displayname"].clone(),

src/service/users/profile.rs

@@ -1,17 +1,276 @@
-use conduwuit::utils::{ReadyExt, stream::TryIgnore};
+use std::{borrow::Cow, collections::BTreeMap};
+
+use conduwuit::{
+	Err, Result,
+	pdu::PartialPdu,
+	utils::{ReadyExt, stream::TryIgnore, to_canonical_object},
+};
 use database::{Deserialized, Ignore, Interfix, Json};
 use futures::{Stream, StreamExt};
-use ruma::{OwnedMxcUri, UserId};
+use ruma::{
+	OwnedMxcUri, UserId,
+	api::client::profile::PropagateTo,
+	events::room::member::MembershipState,
+	presence::PresenceState,
+	profile::{ProfileFieldName, ProfileFieldValue},
+};
+use serde_json::{Value, to_value};
+
+pub enum ProfileFieldChange {
+	Set(ProfileFieldValue),
+	Delete(ProfileFieldName),
+}
+
+impl ProfileFieldChange {
+	fn field_name(&self) -> ProfileFieldName {
+		match self {
+			| &Self::Delete(ref name) => name.clone(),
+			| &Self::Set(ref value) => value.field_name(),
+		}
+	}
+
+	fn value(&self) -> Option<Cow<'_, Value>> {
+		if let Self::Set(value) = self {
+			Some(value.value())
+		} else {
+			None
+		}
+	}
+}
 
 impl super::Service {
+	pub async fn set_profile_field(
+		&self,
+		user_id: &UserId,
+		change: ProfileFieldChange,
+		propagate_to: PropagateTo,
+	) -> Result<()> {
+		const MAX_KEY_LENGTH_BYTES: usize = 255;
+		const MAX_PROFILE_LENGTH_BYTES: usize = 65536;
+
+		let field_name = change.field_name();
+
+		// TODO: The spec mentions special error codes (M_PROFILE_TOO_LARGE,
+		// M_KEY_TOO_LARGE) for profile field size limits, but they're not in its list
+		// of error codes and Ruma doesn't have them. Should we return those, or is
+		// M_TOO_LARGE okay?
+		if field_name.as_str().len() > MAX_KEY_LENGTH_BYTES {
+			return Err!(Request(TooLarge(
+				"Individual profile keys must not exceed {MAX_KEY_LENGTH_BYTES} bytes in length."
+			)));
+		}
+
+		// Serialize the entire profile as canonical JSON, including the new change,
+		// to check if it exceeds 64 KiB
+		{
+			let mut full_profile = self.get_local_profile(user_id).await;
+
+			match &change {
+				| ProfileFieldChange::Set(value) => {
+					full_profile.insert(
+						value.field_name().as_str().to_owned(),
+						value.value().clone().into_owned(),
+					);
+				},
+				| ProfileFieldChange::Delete(key) => {
+					full_profile.remove(key.as_str());
+				},
+			}
+
+			if let Ok(canonical_profile) = to_canonical_object(full_profile) {
+				if serde_json::to_string(&canonical_profile)
+					.expect("should be able to serialize to string")
+					.len() > MAX_PROFILE_LENGTH_BYTES
+				{
+					return Err!(
+						"Profile data must not exceed {MAX_PROFILE_LENGTH_BYTES} bytes in \
+						 length."
+					);
+				}
+			} else {
+				return Err!(Request(BadJson("Failed to canonicalize profile.")));
+			}
+		}
+
+		// Check if this change would be a no-op
+		if self
+			.get_local_profile_field(user_id, field_name.clone())
+			.await
+			.is_some_and(|value| Some(value.value()) == change.value())
+		{
+			return Ok(());
+		}
+
+		// If the user is local and changed their displayname or avatar_url, update it
+		// in all their joined rooms. This is done before updating their profile data
+		// so we can check the old value of the field if `propagate_to` is `unchanged`.
+		if matches!(field_name, ProfileFieldName::AvatarUrl | ProfileFieldName::DisplayName)
+			&& matches!(propagate_to, PropagateTo::All | PropagateTo::Unchanged)
+			&& self.services.globals.user_is_local(user_id)
+		{
+			let current_displayname = self.displayname(user_id).await.ok();
+			let current_avatar_url = self.avatar_url(user_id).await.ok();
+
+			let mut all_joined_rooms = self.services.state_cache.rooms_joined(user_id);
+
+			while let Some(room_id) = all_joined_rooms.next().await {
+				// TODO: this clobbers any custom fields on the event content
+				let mut current_membership = self
+					.services
+					.state_accessor
+					.get_member(&room_id, user_id)
+					.await
+					.expect("should be able to fetch membership event for joined room");
+
+				assert_eq!(
+					current_membership.membership,
+					MembershipState::Join,
+					"user should be joined"
+				);
+
+				// If `propagate_to` is `unchanged`, and the current value of the field we're
+				// updating was changed from its global value in this room, skip it.
+				if matches!(propagate_to, PropagateTo::Unchanged) {
+					let field_changed_from_global = match field_name {
+						| ProfileFieldName::AvatarUrl =>
+							current_membership.avatar_url.as_ref() != current_avatar_url.as_ref(),
+						| ProfileFieldName::DisplayName =>
+							current_membership.displayname.as_ref()
+								!= current_displayname.as_ref(),
+						| _ => unreachable!(),
+					};
+
+					if field_changed_from_global {
+						continue;
+					}
+				}
+
+				let state_lock = self.services.state.mutex.lock(room_id.as_str()).await;
+
+				// Preserve keys in accordance with the key copying rules
+				current_membership.reason = None;
+				current_membership.join_authorized_via_users_server = None;
+				match &change {
+					| ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(avatar_url)) => {
+						current_membership.avatar_url = Some(avatar_url.clone());
+					},
+					| ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)) => {
+						current_membership.displayname = Some(displayname.clone());
+					},
+					| ProfileFieldChange::Delete(ProfileFieldName::AvatarUrl) => {
+						current_membership.avatar_url = None;
+					},
+					| ProfileFieldChange::Delete(ProfileFieldName::DisplayName) => {
+						current_membership.displayname = None;
+					},
+					| _ => unreachable!(),
+				}
+
+				let _ = self
+					.services
+					.timeline
+					.build_and_append_pdu(
+						PartialPdu::state(user_id.to_string(), &current_membership),
+						user_id,
+						Some(&room_id),
+						&state_lock,
+					)
+					.await;
+			}
+
+			if self.services.config.allow_local_presence {
+				// Send a presence EDU to indicate the profile changed
+				let _ = self
+					.services
+					.presence
+					.ping_presence(user_id, &PresenceState::Online)
+					.await;
+			}
+		}
+
+		match change {
+			| ProfileFieldChange::Set(ProfileFieldValue::DisplayName(displayname)) => {
+				self.set_displayname(user_id, Some(displayname).filter(|dn| !dn.is_empty()));
+			},
+			| ProfileFieldChange::Set(ProfileFieldValue::AvatarUrl(avatar_url)) => {
+				self.set_avatar_url(user_id, Some(avatar_url).filter(|av| av.is_valid()));
+			},
+			| ProfileFieldChange::Delete(ProfileFieldName::DisplayName) => {
+				self.set_displayname(user_id, None);
+			},
+			| ProfileFieldChange::Delete(ProfileFieldName::AvatarUrl) => {
+				self.set_avatar_url(user_id, None);
+			},
+			| other => self.set_profile_key(
+				user_id,
+				other.field_name().as_str(),
+				other.value().map(Cow::into_owned),
+			),
+		}
+
+		Ok(())
+	}
+
+	pub async fn get_local_profile(&self, user_id: &UserId) -> BTreeMap<String, Value> {
+		let mut profile = BTreeMap::new();
+
+		// Get displayname and avatar_url independently because `all_profile_keys`
+		// doesn't include them
+		for field in [ProfileFieldName::AvatarUrl, ProfileFieldName::DisplayName] {
+			let key = field.as_str().to_owned();
+
+			if let Some(value) = self.get_local_profile_field(user_id, field).await {
+				profile.insert(key, value.value().into_owned());
+			}
+		}
+
+		// Insert all other profile fields
+		let mut all_fields = self.all_profile_keys(user_id);
+
+		while let Some((key, value)) = all_fields.next().await {
+			profile.insert(key, value);
+		}
+
+		profile
+	}
+
+	pub async fn get_local_profile_field(
+		&self,
+		user_id: &UserId,
+		field: ProfileFieldName,
+	) -> Option<ProfileFieldValue> {
+		let value = match field.clone() {
+			| ProfileFieldName::AvatarUrl => self
+				.avatar_url(user_id)
+				.await
+				.ok()
+				.map(to_value)
+				.transpose()
+				.expect("converting avatar url to value should succeed"),
+			| ProfileFieldName::DisplayName => self
+				.displayname(user_id)
+				.await
+				.ok()
+				.map(to_value)
+				.transpose()
+				.expect("converting displayname to value should succeed"),
+			| other => self.profile_key(user_id, other.as_str()).await.ok(),
+		}?;
+
+		Some(
+			ProfileFieldValue::new(field.as_str(), value)
+				.expect("local profile field should be valid"),
+		)
+	}
+
 	/// Returns the displayname of a user on this homeserver.
-	pub async fn displayname(&self, user_id: &UserId) -> conduwuit::Result<String> {
+	pub async fn displayname(&self, user_id: &UserId) -> Result<String> {
 		self.db.userid_displayname.get(user_id).await.deserialized()
 	}
 
 	/// Sets a new displayname or removes it if displayname is None. You still
 	/// need to notify all rooms of this change.
-	pub fn set_displayname(&self, user_id: &UserId, displayname: Option<String>) {
+	fn set_displayname(&self, user_id: &UserId, displayname: Option<String>) {
 		if let Some(displayname) = displayname {
 			self.db.userid_displayname.insert(user_id, displayname);
 		} else {
@@ -20,12 +279,12 @@ pub fn set_displayname(&self, user_id: &UserId, displayname: Option<String>) {
 	}
 
 	/// Get the `avatar_url` of a user.
-	pub async fn avatar_url(&self, user_id: &UserId) -> conduwuit::Result<OwnedMxcUri> {
+	pub async fn avatar_url(&self, user_id: &UserId) -> Result<OwnedMxcUri> {
 		self.db.userid_avatarurl.get(user_id).await.deserialized()
 	}
 
 	/// Sets a new avatar_url or removes it if avatar_url is None.
-	pub fn set_avatar_url(&self, user_id: &UserId, avatar_url: Option<OwnedMxcUri>) {
+	fn set_avatar_url(&self, user_id: &UserId, avatar_url: Option<OwnedMxcUri>) {
 		match avatar_url {
 			| Some(avatar_url) => {
 				self.db.userid_avatarurl.insert(user_id, &avatar_url);
@@ -37,11 +296,7 @@ pub fn set_avatar_url(&self, user_id: &UserId, avatar_url: Option<OwnedMxcUri>)
 	}
 
 	/// Gets a specific user profile key
-	pub async fn profile_key(
-		&self,
-		user_id: &UserId,
-		profile_key: &str,
-	) -> conduwuit::Result<serde_json::Value> {
+	pub async fn profile_key(&self, user_id: &UserId, profile_key: &str) -> Result<Value> {
 		let key = (user_id, profile_key);
 		self.db
 			.useridprofilekey_value
@@ -54,7 +309,7 @@ pub async fn profile_key(
 	pub fn all_profile_keys<'a>(
 		&'a self,
 		user_id: &'a UserId,
-	) -> impl Stream<Item = (String, serde_json::Value)> + 'a + Send {
+	) -> impl Stream<Item = (String, Value)> + 'a + Send {
 		type KeyVal<'a> = ((Ignore, String), &'a [u8]);
 
 		let prefix = (user_id, Interfix);
@@ -67,11 +322,11 @@ pub fn all_profile_keys<'a>(
 	}
 
 	/// Sets a new profile key value, removes the key if value is None
-	pub fn set_profile_key(
+	fn set_profile_key(
 		&self,
 		user_id: &UserId,
 		profile_key: &str,
-		profile_key_value: Option<serde_json::Value>,
+		profile_key_value: Option<Value>,
 	) {
 		let key = (user_id, profile_key);
 

src/service/users/remote.rs

@@ -0,0 +1,17 @@
+use conduwuit::utils::stream::TryIgnore;
+use futures::Stream;
+use ruma::{OwnedUserId, UserId};
+
+impl super::Service {
+	/// Record the existence of a remote user.
+	pub fn record_remote_user(&self, user_id: &UserId) {
+		assert!(!self.services.globals.user_is_local(user_id), "user is not remote");
+
+		self.db.remoteuserid_remoteuser.insert(user_id, "");
+	}
+
+	/// Returns a stream over all remote users this server has ever seen.
+	pub fn stream_remote_users(&self) -> impl Stream<Item = OwnedUserId> + Send {
+		self.db.remoteuserid_remoteuser.keys().ignore_err()
+	}
+}