arminius_lib_db_keys/

user_keys.rs

// SPDX-FileCopyrightText: 2025 BWI GmbH
//
// SPDX-License-Identifier: Apache-2.0

//! Queries related to the `user_keys` table.
//!
//! This table contains both device keys and cross-signing keys.

mod tests;

use std::collections::{BTreeMap, HashMap};

use arminius_lib_db::{Database, DatabaseError, ResultExt as _};
use arminius_lib_model::{types::StreamId, user_keys::DbKeyType};
use ruma::{
	CrossSigningOrDeviceSigningKeyId, DeviceId, OneTimeKeyAlgorithm, OwnedDeviceId, OwnedUserId,
	UInt, UserId,
	encryption::{CrossSigningKey, DeviceKeys},
};

/// Entry in the `user_keys` table.
///
/// Since each of the keys can either be device keys, or cross-signing keys,
/// which have different structures, we use this enum to return records that are
/// either one of the two.
#[derive(Debug)]
pub enum UserKeyRow {
	/// Contains cross-signing keys, see [`CrossSigningKeyRow`].
	CrossSigningKey(CrossSigningKeyRow),
	/// Contains device keys, see [`DeviceKeysRow`].
	DeviceKeys(DeviceKeysRow),
}

/// [`CrossSigningKey`] from the database, along their `NID`.
#[derive(Debug)]
pub struct CrossSigningKeyRow {
	/// The surrogate key of the keys object.
	pub nid: i64,
	/// The actual cross-signing keys.
	pub key: CrossSigningKey,
}

/// [`DeviceKeys`] from the database, along their `NID`.
#[derive(Debug)]
pub struct DeviceKeysRow {
	/// The surrogate key of the keys object.
	pub nid: i64,
	/// The actual device keys.
	pub key: DeviceKeys,
}

/// A user who had their keys changed since the last sync request.
#[derive(Debug)]
pub struct ChangedUserKeys {
	/// The stream ID of the changed user key. Defaults to 0.
	pub stream_id: StreamId,
	/// The Matrix user ID of that user.
	pub user_id: OwnedUserId,
}

/// A lightweight reference to a `user_keys` row, carrying only the fields
/// needed to resolve key type and identity during a cross-signing signature
/// import.
#[derive(Debug, Clone)]
pub struct UserKeyRef {
	/// The surrogate key of the `user_keys` row.
	pub nid: i64,
	/// The key type of the `user_keys` row.
	pub key_type: DbKeyType,
}

/// Extracts the single key id from a cross-signing key's `keys` map, to be
/// stored in the `user_keys.key_id` column.
///
/// A well-formed cross-signing key has exactly one entry in `keys` (per the
/// Matrix spec, the object "must have exactly one property"); anything else is
/// rejected as invalid data rather than silently picking one.
fn single_cross_signing_key_id(key: &CrossSigningKey) -> Result<&str, DatabaseError> {
	let mut ids = key.keys.keys();
	match (ids.next(), ids.next()) {
		(Some(id), None) => Ok(id.as_str()),
		_ => Err(DatabaseError::InvalidData(format!(
			"cross-signing key must have exactly one entry in its 'keys' map, got {}",
			key.keys.len()
		))),
	}
}

/// Queries related to the `user_keys` table.
#[expect(
	trivial_casts,
	reason = "The casts in `sqlx` queries are not regular rust casts, but used to infer custom types
	from SQL queries. See
	<https://docs.rs/sqlx/latest/sqlx/macro.query.html#type-overrides-bind-parameters-postgres-only>"
)]
// Queries are usually used as extension traits so suggest `import as _`.
#[rust_analyzer::prefer_underscore_import]
pub trait UserKeysQueries: Database {
	/// Inserts device keys for a given user and device, and returns its NID.
	///
	/// # Conflict resolution
	///
	/// If the given `device_id` is already present in the table, its row gets
	/// updated with the new `key_object` value **if and only if** the new
	/// one is different.
	#[tracing::instrument(skip(self))]
	async fn upsert_device_keys(
		&mut self,
		user_id: &UserId,
		device_id: &DeviceId,
		key: &DeviceKeys,
	) -> Result<i64, DatabaseError> {
		let key_object =
			serde_json::to_value(key).invalid_data("Failed to serialize device keys")?;
		Ok(sqlx::query_scalar!(
			r#"
			WITH
			the_user AS (
				SELECT nid FROM users WHERE user_id = $1
			),
			the_device AS (
				SELECT
					devices.nid
					FROM devices
					WHERE
						devices.user_nid = (SELECT nid FROM the_user)
						AND devices.device_id = $2
			),
			upserted AS (
				INSERT INTO user_keys (key_type, user_nid, device_nid, key_id, key_object)
					 VALUES (
						'device',
						(SELECT nid FROM the_user),
						(SELECT nid FROM the_device),
						$3,
						$4
					)
					ON CONFLICT(key_type, user_nid, device_nid) DO UPDATE
					SET
						key_object = EXCLUDED.key_object,
						stream_id = EXCLUDED.stream_id
					WHERE
						user_keys.key_object IS DISTINCT FROM EXCLUDED.key_object
					RETURNING nid
			)
			SELECT nid AS "nid!" FROM upserted
			UNION ALL
			SELECT nid AS "nid!"
				FROM user_keys
				WHERE
					NOT EXISTS (SELECT 1 FROM upserted)
					AND key_type = 'device'
					AND user_keys.user_nid = (SELECT nid FROM the_user)
					AND user_keys.device_nid = (SELECT nid FROM the_device)
			"#,
			user_id.as_str(),
			device_id.as_str(),
			// See <https://spec.matrix.org/v1.16/client-server-api/#key-and-signature-security>
			format!("ed25519:{}", device_id),
			key_object,
		)
		.fetch_one(self.connection().await?)
		.await?)
	}

	/// Imports a cross-signing key for a given user idempotently.
	///
	/// Returns true if a new cross-signing key was inserted, false if one
	/// already exists
	#[tracing::instrument(skip(self))]
	async fn import_cross_signing_key(
		&mut self,
		user_id: &UserId,
		key_type: DbKeyType,
		key: &CrossSigningKey,
	) -> Result<bool, DatabaseError> {
		let key_id = single_cross_signing_key_id(key)?;
		let key_object =
			serde_json::to_value(key).invalid_data("Failed to serialize cross-signing key")?;
		let result = sqlx::query!(
			r#"
			INSERT INTO user_keys (key_type, user_nid, device_nid, key_id, key_object)
			SELECT
				$1,
				u.nid,
				NULL,
				$3,
				$4
			FROM
				users u
			WHERE
				u.user_id = $2
			ON CONFLICT (key_type,
				user_nid,
				device_nid)
				DO NOTHING
			"#,
			key_type as DbKeyType,
			user_id.as_str(),
			key_id,
			key_object,
		)
		.execute(self.connection().await?)
		.await?;
		Ok(result.rows_affected() > 0)
	}

	/// Resolves a `user_keys` row by user id and key id, returning a
	/// [`UserKeyRef`] if found, or `None` if no matching row exists.
	///
	/// The `key_id` is matched against both the stored `key_id` column (device
	/// keys, stored as `"ed25519:<device_id>"`) and against keys embedded in
	/// `key_object->'keys'` (cross-signing keys, where the key id is a field
	/// name inside the JSON object). This dual match handles both device keys
	/// and cross-signing keys with a single query.
	///
	/// Returns `None` when:
	/// - The user does not exist in the `users` table (e.g. a remote user whose keys were never
	///   imported).
	/// - No `user_keys` row matches the given `key_id`.
	#[tracing::instrument(skip(self))]
	async fn resolve_user_key(
		&mut self,
		user_id: &UserId,
		key_id: &CrossSigningOrDeviceSigningKeyId,
	) -> Result<Option<UserKeyRef>, DatabaseError> {
		let mut rows = sqlx::query!(
			r#"
			SELECT
				user_keys.nid,
				key_type AS "key_type: DbKeyType"
			FROM
				user_keys
				JOIN users ON user_keys.user_nid = users.nid
			WHERE
				user_id = $1
				AND (key_id = $2
					OR key_object -> 'keys' ? $2)
			"#,
			user_id.as_str(),
			key_id.as_str(),
		)
		.fetch_all(self.connection().await?)
		.await?;
		if rows.len() > 1 {
			return Err(DatabaseError::InvalidData(format!(
				"key id '{key_id}' for user '{user_id}' is ambiguous: it matches {} user_keys rows",
				rows.len()
			)));
		}
		Ok(rows.pop().map(|record| UserKeyRef { nid: record.nid, key_type: record.key_type }))
	}

	/// Resolves a *device* `user_keys` row by user id and **bare device id**
	/// (the `key_type = 'device'` row whose device has this `device_id`).
	///
	/// Cross-signing signatures name their signed target by bare device id,
	/// which does not match the stored `key_id` (`ed25519:<device_id>`) nor a
	/// `key_object->'keys'` field, so [`Self::resolve_user_key`] cannot find
	/// it. This is the device branch of the ADR's signed-key resolution.
	///
	/// Returns `None` if the user, device, or its device-keys row is absent.
	#[tracing::instrument(skip(self))]
	async fn resolve_device_user_key(
		&mut self,
		user_id: &UserId,
		device_id: &DeviceId,
	) -> Result<Option<UserKeyRef>, DatabaseError> {
		let mut rows = sqlx::query!(
			r#"
			SELECT
				user_keys.nid,
				user_keys.key_type AS "key_type: DbKeyType"
			FROM
				user_keys
				JOIN devices ON user_keys.device_nid = devices.nid
				JOIN users ON devices.user_nid = users.nid
			WHERE
				users.user_id = $1
				AND user_keys.key_type = 'device'
				AND devices.device_id = $2
			"#,
			user_id.as_str(),
			device_id.as_str(),
		)
		.fetch_all(self.connection().await?)
		.await?;
		if rows.len() > 1 {
			return Err(DatabaseError::InvalidData(format!(
				"device id '{device_id}' for user '{user_id}' is ambiguous: it matches {} device user_keys rows",
				rows.len()
			)));
		}
		Ok(rows.pop().map(|record| UserKeyRef { nid: record.nid, key_type: record.key_type }))
	}

	/// Resolves a *cross-signing* `user_keys` row (`device_nid IS NULL`) by
	/// user id and the key id a cross-signing signature names its target by.
	///
	/// Synapse names a signed cross-signing-key target by its **bare** public
	/// key (e.g. `"<pubkey>"`), while arminius stores that key id with its
	/// algorithm prefix (`"ed25519:<pubkey>"`), both in the `key_id` column and
	/// as the field name inside `key_object->'keys'`.
	/// [`Self::resolve_user_key`] only matches the exact stored form, so a
	/// bare target never resolves. This method matches the bare id (by
	/// prefixing it) as well as the already prefixed form, scoped to
	/// cross-signing rows so it can never collide with a device key. This is
	/// the cross-signing-key branch of the ADR's signed-key resolution (the
	/// device branch is [`Self::resolve_device_user_key`]).
	///
	/// Returns `None` if the user has no matching cross-signing key.
	#[tracing::instrument(skip(self))]
	async fn resolve_cross_signing_user_key(
		&mut self,
		user_id: &UserId,
		key_id: &str,
	) -> Result<Option<UserKeyRef>, DatabaseError> {
		let mut rows = sqlx::query!(
			r#"
			SELECT
				user_keys.nid,
				key_type AS "key_type: DbKeyType"
			FROM
				user_keys
				JOIN users ON user_keys.user_nid = users.nid
			WHERE
				user_id = $1
				AND device_nid IS NULL
				AND (key_id = $2
					OR key_id = 'ed25519:' || $2
					OR key_object -> 'keys' ? $2
					OR key_object -> 'keys' ? ('ed25519:' || $2))
			"#,
			user_id.as_str(),
			key_id,
		)
		.fetch_all(self.connection().await?)
		.await?;
		if rows.len() > 1 {
			return Err(DatabaseError::InvalidData(format!(
				"cross-signing key id '{key_id}' for user '{user_id}' is ambiguous: it matches {} user_keys rows",
				rows.len()
			)));
		}
		Ok(rows.pop().map(|record| UserKeyRef { nid: record.nid, key_type: record.key_type }))
	}

	/// Imports device keys for a given user and device idempotently.
	///
	/// Inserts into `user_keys` with `key_type='device'` and returns `true` if
	/// a new row was inserted. If a row for the same `(key_type, user_nid,
	/// device_nid)` already exists, the insert is skipped and `false` is
	/// returned, making this safe to replay and resume.
	#[tracing::instrument(skip(self))]
	async fn import_device_keys(
		&mut self,
		user_id: &UserId,
		device_id: &DeviceId,
		key: &DeviceKeys,
	) -> Result<bool, DatabaseError> {
		let key_object =
			serde_json::to_value(key).invalid_data("Failed to serialize device keys")?;
		let key_id = format!("ed25519:{device_id}");
		let result = sqlx::query!(
			r#"
			INSERT INTO user_keys (key_type, user_nid, device_nid, key_id, key_object)
			SELECT
				'device',
				users.nid,
				devices.nid,
				$3,
				$4
			FROM
				devices
				JOIN users ON devices.user_nid = users.nid
			WHERE
				user_id = $1
				AND device_id = $2
			ON CONFLICT (key_type,
				user_nid,
				device_nid)
				DO NOTHING
			"#,
			user_id.as_str(),
			device_id.as_str(),
			key_id,
			key_object,
		)
		.execute(self.connection().await?)
		.await?;
		Ok(result.rows_affected() > 0)
	}

	/// Given a list of users, and an optional list of device IDs for each user,
	/// fetches all of their device keys.
	///
	/// NOTE: If the list of device IDs is empty for a user, fetches all of
	/// their keys (instead of only the ones of that list).
	#[tracing::instrument(skip(self))]
	async fn fetch_device_keys_for_users(
		&mut self,
		device_keys: &BTreeMap<OwnedUserId, Vec<OwnedDeviceId>>,
	) -> Result<BTreeMap<OwnedUserId, BTreeMap<OwnedDeviceId, DeviceKeysRow>>, DatabaseError> {
		// Split the list of users-devices into three lists:
		// - `specific_users` and `specific_devices` contains the pairs to query together
		// - `wildcard_users` contains the users for whom to request all devices
		let (mut specific_users, mut specific_devices, mut wildcard_users) =
			(Vec::new(), Vec::new(), Vec::new());

		for (user, devices) in device_keys {
			if devices.is_empty() {
				wildcard_users.push(user.to_string());
			} else {
				for device in devices {
					specific_users.push(user.to_string());
					specific_devices.push(device.to_string());
				}
			}
		}

		sqlx::query!(
			"
			SELECT
			  users.user_id,
			  devices.device_id,
			  user_keys.nid,
			  user_keys.key_object
			FROM
			  user_keys
			  JOIN devices ON user_keys.device_nid = devices.nid
			  JOIN users ON devices.user_nid = users.nid
			WHERE
			  user_keys.key_type = 'device'
			  AND (
				(
				  users.user_id, devices.device_id
				) IN (
				  SELECT
					user_id,
					device_id
				  FROM
					UNNEST($1 :: TEXT[], $2 :: TEXT[]) AS pairs(user_id, device_id)
				)
				OR users.user_id = ANY($3 :: TEXT[])
			  )
			",
			&specific_users,
			&specific_devices,
			&wildcard_users,
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.try_fold(BTreeMap::new(), |mut result, record| {
			let user_id =
				UserId::parse(record.user_id).invalid_data("Invalid user ID in database")?;
			let key = serde_json::from_value(record.key_object)
				.invalid_data("Invalid device keys object in database")?;

			let user_keys: &mut BTreeMap<OwnedDeviceId, DeviceKeysRow> =
				result.entry(user_id).or_default();
			user_keys.insert(record.device_id.into(), DeviceKeysRow { nid: record.nid, key });

			Ok(result)
		})
	}

	/// Fetches a single cross-signing key from the database.
	#[tracing::instrument(skip(self))]
	async fn fetch_single_cross_signing_key_for_user(
		&mut self,
		user_id: &UserId,
		key_type: DbKeyType,
	) -> Result<Option<CrossSigningKey>, DatabaseError> {
		sqlx::query!(
			"SELECT key_object
					FROM user_keys
					JOIN users ON user_keys.user_nid = users.nid
					WHERE users.user_id = $1 AND user_keys.key_type = $2",
			user_id.to_string(),
			key_type as DbKeyType
		)
		.fetch_optional(self.connection().await?)
		.await?
		.map(|row| {
			serde_json::from_value(row.key_object)
				.invalid_data("Invalid cross-signing key in database")
		})
		.transpose()
	}

	/// Fetches all the cross-signing keys for the given list of `user_ids`.
	///
	/// This query can be performed by both by the client API (when a local user
	/// requests local keys) and by the Federation Input API (when a foreign
	/// homeserver requests local keys). Additionally, the only case this query
	/// should return a User-Signing key is when it is that of a local user
	/// requesting their own keys. This is what the `requester` parameter is
	/// used for, and it should only be set in the local-keys-fetching scenario.
	///
	/// This returns a map from [`OwnedUserId`] to a map from [`DbKeyType`] to
	/// [`CrossSigningKey`].
	#[tracing::instrument(skip(self))]
	async fn fetch_allowed_cross_signing_keys(
		&mut self,
		requester: Option<&UserId>,
		user_ids: &[OwnedUserId],
	) -> Result<HashMap<OwnedUserId, HashMap<DbKeyType, CrossSigningKeyRow>>, DatabaseError> {
		sqlx::query!(
			r#"
			SELECT
				user_id,
				key_type AS "key_type:DbKeyType",
				user_keys.nid,
				key_object
			FROM
				user_keys
				JOIN users ON user_keys.user_nid = users.nid
			WHERE
				user_id = ANY ($1::text[])
				AND (key_type = ANY ('{master,self_signing}')
					-- Users are only allowed to see their own User-Signing key
					OR (key_type = 'user_signing'
						AND user_id = $2))
			"#,
			&user_ids.into_iter().map(ToString::to_string).collect::<Vec<_>>(),
			requester.map(ToString::to_string)
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.try_fold(HashMap::new(), |mut result, record| {
			let user_id =
				UserId::parse(record.user_id).invalid_data("Invalid user ID in database")?;
			let key = serde_json::from_value(record.key_object)
				.invalid_data("Database contains invalid cross-signing key")?;

			let user_keys: &mut HashMap<DbKeyType, CrossSigningKeyRow> =
				result.entry(user_id).or_default();
			user_keys.insert(record.key_type, CrossSigningKeyRow { nid: record.nid, key });

			Ok(result)
		})
	}

	/// Fetches all of a user's cross-signing keys IDs.
	#[tracing::instrument(skip(self))]
	async fn fetch_cross_signing_keys_ids_for_user(
		&mut self,
		user_id: &UserId,
	) -> Result<Vec<String>, DatabaseError> {
		Ok(sqlx::query_scalar!(
			r#"SELECT jsonb_object_keys(key_object->'keys') AS "keys!"
				FROM user_keys
				JOIN users ON user_keys.user_nid = users.nid
				WHERE users.user_id = $1 AND key_type != 'device'"#,
			user_id.to_string()
		)
		.fetch_all(self.connection().await?)
		.await?)
	}

	/// Inserts the given cross-signing keys.
	///
	/// # Conflict resolution
	///
	/// If a key already exists for the given key_type+user_id, its row gets
	/// updated with the new `key_id` and `key_object` if and only if the new
	/// `key_object` is different than the stored one. If they are the same,
	/// nothing happens.
	#[tracing::instrument(skip(self))]
	async fn upsert_cross_signing_key(
		&mut self,
		user_id: &UserId,
		key_type: DbKeyType,
		key: &CrossSigningKey,
	) -> Result<i64, DatabaseError> {
		let key_id = single_cross_signing_key_id(key)?;
		let key_object =
			serde_json::to_value(key).invalid_data("Failed to serialize cross-signing key")?;
		Ok(sqlx::query!(
			"INSERT INTO user_keys (key_type, user_nid, key_id, key_object)
				VALUES (
					$1,
					(SELECT nid FROM users WHERE user_id = $2),
					$3,
					$4
				)
				ON CONFLICT(key_type, user_nid, device_nid)
				DO UPDATE SET
					key_object = EXCLUDED.key_object,
					key_id = EXCLUDED.key_id,
					stream_id = EXCLUDED.stream_id
				-- Only upsert if the keys were different
				WHERE user_keys.key_object IS DISTINCT FROM EXCLUDED.key_object
        RETURNING nid
				",
			// This cast is mandatory to prevent sqlx complaints.
			key_type as DbKeyType,
			user_id.as_str(),
			key_id,
			key_object
		)
		.fetch_one(self.connection().await?)
		.await?
		.nid)
	}

	/// Fetches all the user keys for the given users.
	///
	/// If the list of key types is empty, fetches all the types.
	#[tracing::instrument(skip(self))]
	async fn fetch_user_keys(
		&mut self,
		requested_keys: &[(OwnedUserId, Vec<DbKeyType>)],
	) -> Result<BTreeMap<OwnedUserId, Vec<UserKeyRow>>, DatabaseError> {
		// Split the list of requested user keys into three lists:
		// - `specific_users` and `specific_types` go in pair and they contain the specific key
		//   types to query
		// - `wildcard_users` contains the users for whom to request all the keys
		let (mut specific_users, mut specific_key_types, mut wildcard_users) =
			(Vec::new(), Vec::new(), Vec::new());

		for (user, key_types) in requested_keys {
			if key_types.is_empty() {
				wildcard_users.push(user.to_string());
			} else {
				for key_type in key_types {
					specific_users.push(user.to_string());
					specific_key_types.push(key_type);
				}
			}
		}

		sqlx::query!(
			r#"
			SELECT
				user_id,
				key_type AS "key_type:DbKeyType",
				user_keys.nid,
				key_object
			FROM
				user_keys
				JOIN users ON user_keys.user_nid = users.nid
			WHERE ((users.user_id, user_keys.key_type) IN (
					SELECT
						user_id,
						key_type
					FROM
						UNNEST($1::text[], $2::KEY_TYPE[])))
				OR users.user_id = ANY ($3::text[])
			"#,
			&specific_users,
			&specific_key_types as &[&DbKeyType],
			&wildcard_users
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.try_fold(BTreeMap::new(), |mut result, record| {
			let user_id =
				UserId::parse(record.user_id).invalid_data("Invalid user ID in database")?;

			let user_keys: &mut Vec<UserKeyRow> = result.entry(user_id).or_default();

			let db_key = match record.key_type {
				DbKeyType::Device => {
					let key: DeviceKeys = serde_json::from_value(record.key_object)
						.invalid_data("Database contains invalid device keys object")?;

					UserKeyRow::DeviceKeys(DeviceKeysRow { nid: record.nid, key })
				}
				_ => {
					let key: CrossSigningKey = serde_json::from_value(record.key_object)
						.invalid_data("Database contains invalid cross-signing key object")?;

					UserKeyRow::CrossSigningKey(CrossSigningKeyRow { nid: record.nid, key })
				}
			};
			user_keys.push(db_key);

			Ok(result)
		})
	}

	/// Returns the list of users that either:
	/// - Updated their device identity or cross-signing keys after the given `from_keys_stream_id`
	/// - Joined a room shared with `user_id` after the given `from_event_stream_id`
	///
	/// Or our own user name, if we changed our device keys
	#[tracing::instrument(skip(self))]
	async fn fetch_sync_user_keys_changes(
		&mut self,
		user_id: &UserId,
		from_event_stream_id: StreamId,
		from_keys_stream_id: StreamId,
	) -> Result<Vec<ChangedUserKeys>, DatabaseError> {
		let mut changed = sqlx::query!(
			r#"
			SELECT
				users_sharing_encrypted_room.user_id AS "user_id!",
				MAX(GREATEST (devices.stream_id, user_keys.stream_id, local_signatures.stream_id)) AS max_stream_id
			FROM
				-- Fetch all users sharing a room with the given user
				users_sharing_encrypted_room ($1)
				JOIN devices ON devices.user_nid = users_sharing_encrypted_room.user_nid
				JOIN user_keys ON users_sharing_encrypted_room.user_nid = user_keys.user_nid
				JOIN local_signatures ON local_signatures.signed_key_nid = user_keys.nid
			WHERE
				-- Filter out rooms that users are no longer members of
				users_sharing_encrypted_room.membership_state NOT IN ('leave', 'kick', 'ban')
				-- Either the keys have been updated/uploaded after the given keys stream ID
				AND ((user_keys.stream_id > $2
						AND user_keys.stream_id <= max_safe_seq ('user_keys_streamids'))
					-- Or there were new signatures for that key since the last sync
					OR (local_signatures.stream_id > $2
						AND local_signatures.stream_id <= max_safe_seq ('user_keys_streamids'))
					-- Or the user created a new device since the last sync
					OR (devices.stream_id > $2
						AND devices.stream_id <= max_safe_seq ('user_keys_streamids')))
				-- Or the room has been joined after the given events stream ID
				OR (users_sharing_encrypted_room.membership_state = 'join'
					AND users_sharing_encrypted_room.membership_stream_id > $3
					AND users_sharing_encrypted_room.membership_stream_id <= max_safe_seq ('eventids'))
			GROUP BY
				users_sharing_encrypted_room.user_id
			"#,
			user_id.to_string(),
			from_keys_stream_id,
			from_event_stream_id
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(|row| -> Result<_, DatabaseError> {
			let user_id = UserId::parse(row.user_id).invalid_data("Invalid user ID in database")?;

			Ok(ChangedUserKeys { stream_id: row.max_stream_id.unwrap_or(StreamId::MIN), user_id })
		})
		.collect::<Result<Vec<_>, _>>()?;

		// And also check if our own user keys changed
		if let Some(max_stream_id) = sqlx::query_scalar!(
			r#"
			 SELECT
					 MAX(stream_id) AS max_stream_id
			 FROM (
					SELECT stream_id FROM devices JOIN users ON devices.user_nid = users.nid WHERE users.user_id = $1
					UNION ALL
					SELECT stream_id FROM user_keys JOIN users ON user_keys.user_nid = users.nid WHERE users.user_id = $1
					UNION ALL
					SELECT local_signatures.stream_id FROM user_keys JOIN local_signatures ON local_signatures.signed_key_nid = user_keys.nid JOIN users ON user_keys.user_nid = users.nid WHERE users.user_id = $1
			 ) stream_ids
			 HAVING MAX(stream_id) > $2
      "#,
			user_id.as_str(),
			from_keys_stream_id
		)
		.fetch_optional(self.connection().await?)
		.await?
		{
			changed.push(ChangedUserKeys {
				stream_id: max_stream_id.unwrap_or(StreamId::MIN),
				user_id: user_id.to_owned(),
			});
		}

		Ok(changed)
	}

	/// Returns the list of users that no longer share any encrypted rooms since
	/// the given `from_event_stream_id`.
	#[tracing::instrument(skip(self))]
	async fn fetch_sync_user_keys_left(
		&mut self,
		user_id: &UserId,
		from_event_stream_id: StreamId,
	) -> Result<Vec<OwnedUserId>, DatabaseError> {
		sqlx::query!(
			r#"
			SELECT
				users_sharing_encrypted_room.user_id AS "user_id!"
			FROM
				users_sharing_encrypted_room ($1)
			WHERE
				users_sharing_encrypted_room.membership_stream_id > $2
				AND users_sharing_encrypted_room.membership_stream_id <= max_safe_seq ('eventids')
			GROUP BY
				users_sharing_encrypted_room.user_id
			HAVING
			-- Filter out users who still share encrypted room(s)
			COUNT(*) FILTER (WHERE users_sharing_encrypted_room.membership_state NOT IN ('ban', 'leave', 'knock')) = 0
			"#,
			user_id.to_string(),
			from_event_stream_id
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(|row| UserId::parse(row.user_id).invalid_data("Invalid user ID in database"))
		.collect::<Result<Vec<_>, _>>()
	}

	/// Fetches the count of non-used one-time keys for the given user.
	#[tracing::instrument(skip(self))]
	async fn fetch_otk_counts(
		&mut self,
		user_id: &UserId,
		device_id: &DeviceId,
	) -> Result<BTreeMap<OneTimeKeyAlgorithm, UInt>, DatabaseError> {
		sqlx::query!(
			r#"
			SELECT
				key_algorithm,
				COUNT(*) AS "count!"
			FROM
				one_time_keys
				JOIN devices ON one_time_keys.device_nid = devices.nid
			WHERE
				devices.user_nid = (
					SELECT
						nid
					FROM
						users
					WHERE
						user_id = $1)
				AND devices.device_id = $2
			GROUP BY
				key_algorithm
			"#,
			user_id.to_string(),
			device_id.to_string()
		)
		.fetch_all(self.connection().await?)
		.await?
		.into_iter()
		.map(|row| {
			let count =
				UInt::try_from(row.count as u64).invalid_data("Invalid one-time key count")?;

			Ok((row.key_algorithm.into(), count))
		})
		.collect::<Result<BTreeMap<_, _>, _>>()
	}

	/// Fetches the highest stream ID ever issued for the `user_keys_streamids` stream.
	#[tracing::instrument(skip(self))]
	async fn fetch_current_user_keys_stream_id(&mut self) -> Result<StreamId, DatabaseError> {
		Ok(sqlx::query_scalar!(
			r#"
		SELECT GREATEST(
			(SELECT MAX(devices.stream_id) FROM devices),
			(SELECT MAX(user_keys.stream_id) FROM user_keys),
			(SELECT MAX(local_signatures.stream_id) FROM local_signatures)
		) as max_stream_id
		"#
		)
		.fetch_one(self.connection().await?)
		.await?
		.unwrap_or(0))
	}
}

impl<T> UserKeysQueries for T where T: Database {}