use std::{
	cmp,
	collections::{BTreeMap, HashMap},
	future::ready,
	pin::Pin,
	sync::Arc,
	time::Instant,
};

use conduwuit::{
	Err, Error, Event, Pdu, PduCount, Result, debug, debug_info, debug_warn, err, error, info,
	result::NotFound,
	trace,
	utils::{
		IterStream, ReadyExt,
		stream::{TryExpect, TryIgnore, WidebandExt, automatic_width},
		to_canonical_object,
	},
	warn,
};
use database::{Deserialized, Ignore, Json};
use futures::{FutureExt, StreamExt, TryStreamExt};
use itertools::Itertools;
use ruma::{
	OwnedDeviceId, OwnedEventId, OwnedRoomId, OwnedUserId, UserId,
	api::OAuthClientScope,
	events::{
		AnyStrippedStateEvent, GlobalAccountDataEventType, StateEventType,
		push_rules::PushRulesEvent,
		room::member::{MembershipState, RoomMemberEventContent},
	},
	push::Ruleset,
	serde::{Raw, from_raw_json_value},
};
use serde_json::{Value, value::to_raw_value};

use crate::{Services, media, rooms, rooms::short::ShortStateHash};

type MigrationFn =
	for<'a> fn(&'a Services) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;

enum DatabaseMigrationStep {
	Migration(&'static str, MigrationFn),
	VersionBump(u64),
}

static MIGRATIONS: &[DatabaseMigrationStep] = const {
	use DatabaseMigrationStep::*;

	macro_rules! migration {
		($name:expr => $function:path) => {
			Migration($name, |services| {
				Box::pin(async {
					let db = &services.db;

					if db["global"].get($name).await.is_not_found() {
						info!("Starting migration {}", $name);
						let cork = db.cork_and_sync();

						$function(services)
							.await
							.map_err(|err| err!("Failed to run migration {}: {err}", $name))?;

						drop(cork);
						db["global"].insert($name, []);
						db.db.sort()?;
					}

					Ok(())
				})
			})
		};

		($name:ident) => {
// cargo fmt really wants this to be left-aligned, I guess?
migration!(stringify!($name) => $name)
		};
	}

	// When you're writing a migration that makes backwards-incompatible
	// changes, you need to use `VersionBump` to keep old versions from
	// accepting the new database. The `VersionBump` command should come
	// _before_ your migration, which ensures that old versions will reject
	// the new database even if the server crashes or restarts while your
	// migration is running.

	// If your migration is backwards-compatible, meaning that its changes can
	// be understood by a version of continuwuity that has not run it, it
	// doesn't need a version bump. Use your best discretion when deciding what
	// "be understood" means. Also, make sure to add a comment that explains
	// why the bump was necessary, for the sake of future developers.

	// Database versions should increase monotonically (you'll get a const-eval
	// panic if they don't) and linearly. I don't know why conduit skipped
	// from 13 to 17, don't bother asking.

	&[
		// Begin pre-fork logic. We don't have insight into why this works
		// the way it does, so there are no comments here.
		VersionBump(12),
		migration!(fix_push_rule_names),
		VersionBump(13),
		migration!(fix_server_default_push_rules),
		VersionBump(17),
		migration!("feat_sha256_media" => media::migrations::migrate_sha256_media),
		migration!(fix_bad_double_separator_in_state_cache),
		migration!(retroactively_fix_bad_data_from_roomuserid_joined),
		migration!(fix_referencedevents_missing_sep),
		migration!(fix_readreceiptid_readreceipt_duplicates),
		// End pre-fork logic.

		// v18: `fix_corrupt_msc4133_fields` will change the
		// `us.cloke.msc4175.tz` field in a way that old versions don't
		// understand.
		VersionBump(18),
		migration!(fix_corrupt_msc4133_fields),
		// v19: `populate_userroomid_leftstate_table` will put data in the
		// values of `userroomid_leftstate` that old versions will fail to
		// deserialize because they expect the values to be empty.
		VersionBump(19),
		migration!(populate_userroomid_leftstate_table),
		// This is backwards compatible because it fixes some data that clients
		// care about but we don't.
		migration!(fix_local_invite_state),
		// v20: Old versions expect remote users to be tracked in
		// `userid_password`. This migration moves them to a different keyspace.
		VersionBump(20),
		migration!(split_userid_password),
		// This is backwards compatible because old versions will recreate the
		// keyspace if it's missing.
		migration!("drop_roomsynctoken_shortstatehash" => obliterate_roomsynctoken_shortstatehash_with_extreme_prejudice),
		// v21: `unembed_unsigned_info` will remove `unsigned` data from PDUs
		// that old versions expect to be embedded.
		VersionBump(21),
		// This migration can take several hours on old databases, so we run it
		// in the background to keep the service manager from killing us for
		// taking too long to start. Future maintainers: !!! BE VERY CAREFUL !!!
		// if you need to write more migrations like this! This one is fine
		// because it deletes data that we don't check anymore, but migrations
		// which replace existing data that we do check should NOT run in
		// the background. If they do, the server will finish starting up and
		// might access data that the migration hasn't gotten to yet.
		Migration("unembed_unsigned_info", |services| {
			Box::pin(async {
				if services.db["global"]
					.get(b"unembed_unsigned_info")
					.await
					.is_not_found()
				{
					let db = services.db.clone();
					let metadata = services.rooms.metadata.clone();
					let timeline = services.rooms.timeline.clone();

					services.server.runtime().spawn(async move {
						unembed_unsigned_info(metadata, timeline).await;
						db["global"].insert(b"unembed_unsigned_info", []);
					});
				}

				Ok(())
			})
		}),
		// This is backwards compatible because old versions will recreate the
		// keyspace if it's missing.
		migration!(drop_server_signingkeys),
		// v22: Old versions will fail to deserialize the fixed OAuth scope
		// names, preventing users from signing in.
		VersionBump(22),
		migration!(fix_oauth_scope_names),
		// Removes any extremities in roomid_pduleaves that we do not know about
		// (and consequently should not reference).
		migration!(remove_unknown_extremities),
		// Re-calculates extremities for any rooms that actually lack any
		// extremities at all.
		migration!(consistency_check_extremity_tracker),
		// This is backwards compatible because old versions never read the
		// account data entries it removes.
		migration!(remove_replaced_account_data),
	]
};

pub(crate) const DATABASE_VERSION: u64 = const {
	let mut latest_version = 0_u64;
	// TODO: refactor once slice iter is const stable
	let mut i = 0;

	while i < MIGRATIONS.len() {
		if let DatabaseMigrationStep::VersionBump(version) = &MIGRATIONS[i] {
			if *version > latest_version {
				latest_version = *version;
			} else {
				panic!("Database version must monotonically increase");
			}
		}
		i += 1;
	}

	latest_version
};

// This should match the first `VersionBump` in `MIGRATIONS`.
const OLDEST_SUPPORTED_VERSION: u64 = 11;

pub(crate) async fn migrations(services: &Services) -> Result<()> {
	let users_count = services.users.count().await;

	// Matrix resource ownership is based on the server name; changing it
	// requires recreating the database from scratch.
	if users_count > 0 {
		let server_user = &services.globals.server_user;
		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 \
				 delete the old one first.",
			));
		}
	}

	if users_count > 0 {
		migrate(services).await
	} else {
		fresh(services).await
	}
}

async fn fresh(services: &Services) -> Result<()> {
	info!("Creating new fresh database");
	let db = &services.db;

	services.globals.db.bump_database_version(DATABASE_VERSION);

	for step in MIGRATIONS {
		if let DatabaseMigrationStep::Migration(key, _) = step {
			db["global"].insert(key, []);
		}
	}

	// Create the admin room and server user on first run
	info!("Creating admin room and server user");
	crate::admin::create_admin_room(services)
		.boxed()
		.await
		.inspect_err(|e| error!("Failed to create admin room during db init: {e}"))?;

	info!("Created new database with version {DATABASE_VERSION}");

	Ok(())
}

/// Apply any migrations
async fn migrate(services: &Services) -> Result<()> {
	if services.globals.db.database_version().await < OLDEST_SUPPORTED_VERSION {
		return Err!(Database(
			"Database schema version {} is too old to migrate. Please use an old version of \
			 conduwuit to migrate to version 11 and then run Continuwuity again.",
			services.globals.db.database_version().await
		));
	}

	for step in MIGRATIONS {
		match step {
			| DatabaseMigrationStep::Migration(_, migration) => {
				migration(services).await?;
			},
			| DatabaseMigrationStep::VersionBump(version) => {
				if services.globals.db.database_version().await < *version {
					services.globals.db.bump_database_version(*version);
					info!("Bumped database version to {version}.");
				}
			},
		}
	}

	if services.config.media_startup_check {
		info!("Starting media startup integrity check.");
		let now = Instant::now();
		media::migrations::checkup_sha256_media(services)
			.await
			.map_err(|e| err!("Failed to verify media integrity: {e}"))?;
		info!(
			"Finished media startup integrity check in {} seconds.",
			now.elapsed().as_secs_f32()
		);
	}

	assert_eq!(
		services.globals.db.database_version().await,
		DATABASE_VERSION,
		"Failed asserting local database version {} is equal to known latest continuwuity \
		 database version {}",
		services.globals.db.database_version().await,
		DATABASE_VERSION,
	);

	{
		let patterns = services.globals.forbidden_usernames();
		if !patterns.is_empty() {
			services
				.users
				.stream_local_users()
				.filter_map(async |user_id| {
					if services.users.status(&user_id).await.is_found() {
						Some(user_id)
					} else {
						None
					}
				})
				.ready_for_each(|user_id| {
					let matches = patterns.matches(user_id.localpart());
					if matches.matched_any() {
						warn!(
							"User {} matches the following forbidden username patterns: {}",
							user_id.to_string(),
							matches
								.into_iter()
								.map(|x| &patterns.patterns()[x])
								.join(", ")
						);
					}
				})
				.await;
		}
	}

	{
		let patterns = services.globals.forbidden_alias_names();
		if !patterns.is_empty() {
			for room_id in services.rooms.metadata.iter_ids().collect::<Vec<_>>().await {
				services
					.rooms
					.alias
					.local_aliases_for_room(&room_id)
					.ready_for_each(|room_alias| {
						let matches = patterns.matches(room_alias.alias());
						if matches.matched_any() {
							warn!(
								"Room with alias {} ({}) matches the following forbidden room \
								 name patterns: {}",
								room_alias,
								&room_id,
								matches
									.into_iter()
									.map(|x| &patterns.patterns()[x])
									.join(", ")
							);
						}
					})
					.await;
			}
		}
	}

	info!("Loaded RocksDB database with schema version {DATABASE_VERSION}");

	Ok(())
}

async fn fix_push_rule_names(services: &Services) -> Result<()> {
	for username in &services
		.users
		.stream_local_users()
		.ready_filter(|user| services.globals.user_is_local(user))
		.collect::<Vec<OwnedUserId>>()
		.await
	{
		let user = match UserId::parse_with_server_name(username.as_str(), &services.server.name)
		{
			| Ok(u) => u,
			| Err(e) => {
				warn!("Invalid username {username}: {e}");
				continue;
			},
		};

		let Some(mut account_data) = services
			.account_data
			.get_global::<PushRulesEvent>(&user, GlobalAccountDataEventType::PushRules)
			.await
			.ok()
		else {
			// Skip users with no push rules
			continue;
		};

		let rules_list = &mut account_data.content.global;

		//content rule
		{
			let content_rule_transformation =
				[".m.rules.contains_user_name", ".m.rule.contains_user_name"];

			let rule = rules_list.content.get(content_rule_transformation[0]);

			if let Some(rule) = rule {
				let mut rule = rule.clone();
				content_rule_transformation[1].clone_into(&mut rule.rule_id);
				rules_list
					.content
					.shift_remove(content_rule_transformation[0]);

				rules_list.content.insert(rule);
			}
		}

		//underride rules
		{
			let underride_rule_transformation = [
				[".m.rules.call", ".m.rule.call"],
				[".m.rules.room_one_to_one", ".m.rule.room_one_to_one"],
				[".m.rules.encrypted_room_one_to_one", ".m.rule.encrypted_room_one_to_one"],
				[".m.rules.message", ".m.rule.message"],
				[".m.rules.encrypted", ".m.rule.encrypted"],
			];

			for transformation in underride_rule_transformation {
				let rule = rules_list.underride.get(transformation[0]);
				if let Some(rule) = rule {
					let mut rule = rule.clone();
					transformation[1].clone_into(&mut rule.rule_id);
					rules_list.underride.shift_remove(transformation[0]);
					rules_list.underride.insert(rule);
				}
			}
		}

		services
			.account_data
			.update(
				None,
				&user,
				GlobalAccountDataEventType::PushRules.to_string().into(),
				&serde_json::to_value(account_data).expect("to json value always works"),
			)
			.await?;
	}

	Ok(())
}

async fn fix_server_default_push_rules(services: &Services) -> Result<()> {
	for username in &services
		.users
		.stream_local_users()
		.ready_filter(|user| services.globals.user_is_local(user))
		.collect::<Vec<OwnedUserId>>()
		.await
	{
		let user = match UserId::parse_with_server_name(username.as_str(), &services.server.name)
		{
			| Ok(u) => u,
			| Err(e) => {
				warn!("Invalid username {username}: {e}");
				continue;
			},
		};

		let Some(mut account_data) = services
			.account_data
			.get_global::<PushRulesEvent>(&user, GlobalAccountDataEventType::PushRules)
			.await
			.ok()
		else {
			// Skip users with no push rules
			continue;
		};

		let user_default_rules = Ruleset::server_default(&user);
		account_data
			.content
			.global
			.update_with_server_default(user_default_rules);

		services
			.account_data
			.update(
				None,
				&user,
				GlobalAccountDataEventType::PushRules.to_string().into(),
				&serde_json::to_value(account_data).expect("to json value always works"),
			)
			.await?;
	}

	Ok(())
}

async fn fix_bad_double_separator_in_state_cache(services: &Services) -> Result<()> {
	info!("Fixing bad double separator in state_cache roomuserid_joined");

	let db = &services.db;
	let roomuserid_joined = &db["roomuserid_joined"];

	let mut iter_count: usize = 0;
	roomuserid_joined
		.raw_stream()
		.ignore_err()
		.ready_for_each(|(key, value)| {
			let mut key = key.to_vec();
			iter_count = iter_count.saturating_add(1);
			debug_info!(%iter_count);
			let first_sep_index = key
				.iter()
				.position(|&i| i == 0xFF)
				.expect("found 0xFF delim");

			if key
				.iter()
				.get(first_sep_index..=first_sep_index.saturating_add(1))
				.copied()
				.collect_vec()
				== vec![0xFF, 0xFF]
			{
				debug_warn!("Found bad key: {key:?}");
				roomuserid_joined.remove(&key);

				key.remove(first_sep_index);
				debug_warn!("Fixed key: {key:?}");
				roomuserid_joined.insert(&key, value);
			}
		})
		.await;

	info!("Finished fixing");
	Ok(())
}

async fn retroactively_fix_bad_data_from_roomuserid_joined(services: &Services) -> Result<()> {
	info!("Retroactively fixing bad data from broken roomuserid_joined");

	let room_ids = services.rooms.metadata.iter_ids().collect::<Vec<_>>().await;

	for room_id in &room_ids {
		debug_info!("Fixing room {room_id}");

		let users_in_room: Vec<OwnedUserId> = services
			.rooms
			.state_cache
			.room_members(room_id)
			.collect()
			.await;

		let joined_members = users_in_room
			.iter()
			.stream()
			.filter(|user_id| {
				services
					.rooms
					.state_accessor
					.get_member(room_id, user_id)
					.map(|member| {
						member.is_ok_and(|member| member.membership == MembershipState::Join)
					})
			})
			.collect::<Vec<_>>()
			.await;

		let non_joined_members = users_in_room
			.iter()
			.stream()
			.filter(|user_id| {
				services
					.rooms
					.state_accessor
					.get_member(room_id, user_id)
					.map(|member| {
						member.is_ok_and(|member| member.membership == MembershipState::Join)
					})
			})
			.collect::<Vec<_>>()
			.await;

		for user_id in &joined_members {
			debug_info!("User is joined, marking as joined");
			services.rooms.state_cache.mark_as_joined(user_id, room_id);
		}

		for user_id in &non_joined_members {
			debug_info!("User is left or banned, marking as left");
			services
				.rooms
				.state_cache
				.mark_as_left(user_id, room_id, None)
				.await;
		}
	}

	for room_id in &room_ids {
		debug_info!(
			"Updating joined count for room {room_id} to fix servers in room after correcting \
			 membership states"
		);

		services
			.rooms
			.state_cache
			.update_joined_count(room_id)
			.await;
	}

	info!("Finished fixing");
	Ok(())
}

async fn fix_referencedevents_missing_sep(services: &Services) -> Result {
	info!("Fixing missing record separator between room_id and event_id in referencedevents");

	let referencedevents = services.db["referencedevents"].clone();

	let totals: (usize, usize) = (0, 0);
	let (total, fixed) = referencedevents
		.raw_stream()
		.expect_ok()
		.enumerate()
		.ready_fold(totals, |mut a, (i, (key, val))| {
			debug_assert!(val.is_empty(), "expected no value");

			let has_sep = key.contains(&database::SEP);

			if !has_sep {
				let key_str = std::str::from_utf8(key).expect("key not utf-8");
				let room_id_len = key_str.find('$').expect("missing '$' in key");
				let (room_id, event_id) = key_str.split_at(room_id_len);
				debug!(?a, "fixing {room_id}, {event_id}");

				let new_key = (room_id, event_id);
				referencedevents.put_raw(new_key, val);
				referencedevents.remove(key);
			}

			a.0 = cmp::max(i, a.0);
			a.1 = a.1.saturating_add((!has_sep).into());
			a
		})
		.await;

	info!(?total, ?fixed, "Fixed missing record separators in 'referencedevents'.");
	Ok(())
}

async fn fix_readreceiptid_readreceipt_duplicates(services: &Services) -> Result {
	type ArrayId = String;
	type Key = (OwnedRoomId, u64, OwnedUserId);

	info!("Fixing undeleted entries in readreceiptid_readreceipt...");

	let db = &services.db;
	let readreceiptid_readreceipt = db["readreceiptid_readreceipt"].clone();

	let mut cur_room: Option<ArrayId> = None;
	let mut cur_user: Option<ArrayId> = None;
	let (mut total, mut fixed): (usize, usize) = (0, 0);
	readreceiptid_readreceipt
		.keys()
		.expect_ok()
		.ready_for_each(|key: Key| {
			let (ref room_id, _, ref user_id) = key;
			let last_room = cur_room.replace(room_id.as_str().into());

			let last_user = cur_user.replace(user_id.as_str().into());

			let is_dup = cur_room == last_room && cur_user == last_user;
			if is_dup {
				readreceiptid_readreceipt.del(key);
			}

			fixed = fixed.saturating_add(is_dup.into());
			total = total.saturating_add(1);
		})
		.await;

	info!(?total, ?fixed, "Fixed undeleted entries in readreceiptid_readreceipt.");
	Ok(())
}

async fn fix_corrupt_msc4133_fields(services: &Services) -> Result {
	// Due to an old bug, some conduwuit databases have `us.cloke.msc4175.tz`
	// user profile fields with raw strings instead of quoted JSON ones.
	// This migration fixes that.

	use serde_json::{Value, from_slice};
	type KeyVal<'a> = ((OwnedUserId, String), &'a [u8]);

	info!("Fixing corrupted `us.cloke.msc4175.tz` fields...");

	let db = &services.db;
	let useridprofilekey_value = db["useridprofilekey_value"].clone();

	let (total, fixed) = useridprofilekey_value
		.stream()
		.try_fold(
			(0_usize, 0_usize),
			async |(mut total, mut fixed),
			       ((user, key), value): KeyVal<'_>|
			       -> Result<(usize, usize)> {
				match from_slice::<Value>(value) {
					// corrupted timezone field
					| Err(_) if key == "us.cloke.msc4175.tz" => {
						let new_value = Value::String(String::from_utf8(value.to_vec())?);
						useridprofilekey_value.put((user, key), Json(new_value));
						fixed = fixed.saturating_add(1);
					},
					// corrupted value for some other key
					| Err(error) => {
						warn!(
							"deleting MSC4133 key {} for user {} due to deserialization \
							 failure: {}",
							key, user, error
						);
						useridprofilekey_value.del((user, key));
					},
					// other key with no issues
					| Ok(_) => {
						// do nothing
					},
				}

				total = total.saturating_add(1);

				Ok((total, fixed))
			},
		)
		.await?;

	info!(?total, ?fixed, "Fixed corrupted `us.cloke.msc4175.tz` fields.");
	Ok(())
}

async fn populate_userroomid_leftstate_table(services: &Services) -> Result {
	type KeyVal = (Key, Raw<Option<Pdu>>);
	type Key = (OwnedUserId, OwnedRoomId);

	let db = &services.db;
	let userroomid_leftstate = db["userroomid_leftstate"].clone();

	let (total, fixed, _) = userroomid_leftstate
		.stream()
		.try_fold(
			(0_usize, 0_usize, HashMap::<OwnedRoomId, ShortStateHash>::new()),
			async |(mut total, mut fixed, mut shortstatehash_cache): (
				usize,
				usize,
				HashMap<_, _>,
			),
			       ((user_id, room_id), state): KeyVal|
			       -> Result<(usize, usize, HashMap<_, _>)> {
				if state.deserialize().is_err() {
					let latest_shortstatehash =
						if let Some(shortstatehash) = shortstatehash_cache.get(&room_id) {
							*shortstatehash
						} else if let Ok(shortstatehash) =
							services.rooms.state.get_room_shortstatehash(&room_id).await
						{
							shortstatehash_cache.insert(room_id.clone(), shortstatehash);
							shortstatehash
						} else {
							warn!(%room_id, %user_id, "room has no shortstatehash");
							return Ok((total, fixed, shortstatehash_cache));
						};

					let leave_state_event = services
						.rooms
						.state_accessor
						.state_get(
							latest_shortstatehash,
							&StateEventType::RoomMember,
							user_id.as_str(),
						)
						.await;

					match leave_state_event {
						| Ok(leave_state_event) => {
							userroomid_leftstate.put((user_id, room_id), Json(leave_state_event));
							fixed = fixed.saturating_add(1);
						},
						| Err(_) => {
							warn!(
								%room_id,
								%user_id,
								"room cached as left has no leave event for user, removing \
								 cache entry"
							);
							userroomid_leftstate.del((user_id, room_id));
						},
					}
				}

				total = total.saturating_add(1);
				Ok((total, fixed, shortstatehash_cache))
			},
		)
		.await?;

	info!(?total, ?fixed, "Fixed entries in `userroomid_leftstate`.");
	Ok(())
}

async fn fix_local_invite_state(services: &Services) -> Result {
	// Clean up the effects of !1249 by caching stripped state for invites

	type KeyVal = (Key, Raw<Vec<AnyStrippedStateEvent>>);
	type Key = (OwnedUserId, OwnedRoomId);

	let db = &services.db;
	let userroomid_invitestate = db["userroomid_invitestate"].clone();

	// for each user invited to a room
	let fixed =  userroomid_invitestate.stream()
		// if they're a local user on this homeserver
		.try_filter(|((user_id, _), _): &KeyVal| ready(services.globals.user_is_local(user_id)))
		.and_then(async |((user_id, room_id), stripped_state): KeyVal| Ok::<_,
			Error>((user_id.clone(), room_id.clone(), stripped_state.deserialize
		().unwrap_or_else(|e| {
			trace!("Failed to deserialize: {:?}", stripped_state.json());
			warn!(
				%user_id,
				%room_id,
				"Failed to deserialize stripped state for invite, removing from db: {e}"
			);
			userroomid_invitestate.del((&user_id, &room_id));
			vec![]
		}))))
		.try_fold(0_usize, async |mut fixed, (user_id, room_id, stripped_state)| {
			// and their invite state is None
			if stripped_state.is_empty()
				// and they are actually invited to the room
				&& let Ok(membership_event) = services.rooms.state_accessor.room_state_get(&room_id, &StateEventType::RoomMember, user_id.as_str()).await
				&& membership_event.get_content::<RoomMemberEventContent>().is_ok_and(|content| content.membership == MembershipState::Invite)
				// and the invite was sent by a local user
				&& services.globals.user_is_local(&membership_event.sender) {

				// build and save stripped state for their invite in the database
				let stripped_state = services.rooms.state.summary_stripped(&membership_event, &room_id, &user_id, false).await;
				userroomid_invitestate.put((&user_id, &room_id), Json(stripped_state));
				fixed = fixed.saturating_add(1);
			}

			Ok(fixed)
		})
		.await?;

	info!(?fixed, "Fixed local invite state cache entries.");
	Ok(())
}

async fn split_userid_password(services: &Services) -> Result {
	// Split remote and deactivated users out from the `userid_password` table

	let db = &services.db;
	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_user_id(&user_id)
					|| user_id == services.globals.server_user)
				{
					info!("Marking {user_id} as deactivated");
					userid_deactivated.insert(&user_id, "");
				}
			}

			remote_users
		})
		.await;

	info!(?remote_users, "Split userid_password.");

	Ok(())
}

async fn obliterate_roomsynctoken_shortstatehash_with_extreme_prejudice(
	services: &Services,
) -> Result {
	services.db.db.drop_column("roomsynctoken_shortstatehash")?;

	info!("Cleared roomsynctoken_shortstatehash.");

	Ok(())
}

#[tracing::instrument(name = "unembed_unsigned_info", skip_all)]
async fn unembed_unsigned_info(
	metadata: Arc<rooms::metadata::Service>,
	timeline: Arc<rooms::timeline::Service>,
) {
	type Map = BTreeMap<String, Box<serde_json::value::RawValue>>;
	info!(
		"Starting background migration. This may take a long time, and may cause your server to \
		 lag. Please avoid restarting until the migration is complete."
	);

	let start = Instant::now();
	let mut total_migrated = 0_usize;
	let mut all_rooms = metadata.iter_ids();
	while let Some(room_id) = all_rooms.next().await {
		info!("Migrating PDUs in {room_id}");
		let mut pdus = std::pin::pin!(timeline.all_pdus(&room_id));
		let mut migrated = 0_usize;
		while let Some((_, pdu)) = pdus.next().await {
			let mut pdu = pdu.clone();
			let mut unsigned: Map = pdu
				.unsigned()
				.and_then(|u| from_raw_json_value::<Map, Error>(u).ok())
				.unwrap_or_default();
			unsigned.remove("age");
			unsigned.remove("membership");
			unsigned.remove("prev_content");
			unsigned.remove("prev_sender");

			if let Some(redacted_because) = unsigned
				.remove("redacted_because")
				.and_then(|r| from_raw_json_value::<Map, Error>(&r).ok())
			{
				match redacted_because
					.get("event_id")
					.and_then(|e| from_raw_json_value::<OwnedEventId, Error>(e).ok())
				{
					| None => {},
					| Some(event_id) => {
						unsigned.insert(
							"org.continuwuity.redacted_by".to_owned(),
							to_raw_value(event_id.as_str()).expect("str must be raw value"),
						);
					},
				}
			}
			pdu.unsigned = Some(to_raw_value(&unsigned).expect("unsigned must be valid JSON"));
			let pdu_id = timeline
				.get_pdu_id(pdu.event_id())
				.await
				.expect("PDU must have an id");
			timeline
				.replace_pdu(&pdu_id, &to_canonical_object(pdu).expect("PDU must be valid JSON"))
				.await
				.expect("must be able to replace PDU after migrating unsigned data");
			migrated = migrated.saturating_add(1);
		}
		total_migrated = total_migrated.saturating_add(migrated);
		info!(elapsed=?start.elapsed(), "Migrated {migrated} PDUs in {room_id}");
	}

	info!(elapsed=?start.elapsed(), total_migrated_pdus=total_migrated, "Finished migration.");
}

async fn drop_server_signingkeys(services: &Services) -> Result<()> {
	services.db.db.drop_column("server_signingkeys")?;

	info!("Cleared server_signingkeys.");

	Ok(())
}

async fn fix_oauth_scope_names(services: &Services) -> Result {
	let db = &services.db;
	let userdeviceid_oauthsessioninfo = db["userdeviceid_oauthsessioninfo"].clone();

	userdeviceid_oauthsessioninfo
		.stream::<(OwnedUserId, OwnedDeviceId), Value>()
		.ignore_err()
		.for_each(async |((user_id, device_id), mut session_info)| {
			let map = session_info.as_object_mut().unwrap();
			let scopes = map.get_mut("scopes").unwrap().as_array_mut().unwrap();

			if let Some(old_scope_index) = scopes
				.iter()
				.position(|value| *value == Value::String("ClientApi".to_owned()))
			{
				scopes[old_scope_index] =
					Value::String(OAuthClientScope::ApiFullAccess.to_string());
			}

			userdeviceid_oauthsessioninfo.put((user_id, device_id), Json(session_info));
		})
		.await;

	info!("Fixed OAuth session scope names");

	Ok(())
}

#[tracing::instrument(skip_all)]
async fn remove_unknown_extremities(services: &Services) -> Result {
	services.db["roomid_pduleaves"]
		.keys()
		.ignore_err()
		.for_each_concurrent(
			automatic_width(),
			|(room_id, event_id): (OwnedRoomId, OwnedEventId)| async move {
				if !services.rooms.timeline.pdu_exists(&event_id).await {
					info!("Removing unknown extremity in {room_id}: {event_id}");
					services
						.rooms
						.state
						.remove_forward_extremity(&room_id, &event_id);
				} else if !services
					.rooms
					.pdu_metadata
					.is_event_accepted(&event_id)
					.await
				{
					info!("Removing failed extremity in {room_id}: {event_id}");
					services
						.rooms
						.state
						.remove_forward_extremity(&room_id, &event_id);
				}
			},
		)
		.await;
	Ok(())
}

async fn remove_replaced_account_data(services: &Services) -> Result {
	type Key = (Option<OwnedRoomId>, OwnedUserId, u64, String);
	type Latest = (Ignore, Ignore, u64, Ignore);

	let entries = services.db["roomuserdataid_accountdata"].clone();
	let latest_of_type = services.db["roomusertype_roomuserdataid"].clone();

	let keys: Vec<Key> = entries.keys().ignore_err().collect().await;
	let mut removed: usize = 0;
	for (room_id, user_id, count, event_type) in &keys {
		let latest = latest_of_type
			.qry(&(room_id, user_id, event_type))
			.await
			.deserialized::<Latest>();

		if latest.is_ok_and(|(_, _, latest, _)| latest != *count) {
			entries.del((room_id, user_id, count, event_type));
			removed = removed.saturating_add(1);
		}
	}

	info!(
		total = keys.len(),
		?removed,
		"Removed replaced entries in 'roomuserdataid_accountdata'."
	);
	Ok(())
}

#[tracing::instrument(skip_all)]
async fn consistency_check_extremity_tracker(services: &Services) -> Result {
	let mut rooms = services.rooms.metadata.iter_ids();

	// For every room the server is a member of,
	while let Some(room_id) = rooms.next().await {
		// If this room has no extremities, re-calculate
		let extremity_count = services
			.rooms
			.state
			.get_forward_extremities(&room_id)
			.count()
			.await;

		if extremity_count == 0 {
			warn!(
				"Room {room_id} is missing extremity information, recalculating. This may take \
				 a long time if this room is old."
			);
			let epoch_ts = services
				.rooms
				.timeline
				.first_pdu_in_room(&room_id)
				.await
				.map_or_default(|e| e.origin_server_ts().0);
			services
				.rooms
				.timeline
				.pdus(&room_id, Some(PduCount::Normal(0)))
				.ignore_err()
				.for_each(|(_, pdu)| {
					let room_id = &room_id;
					async move {
						if pdu.origin_server_ts < epoch_ts {
							return;
						}
						if !services
							.rooms
							.pdu_metadata
							.is_event_accepted(pdu.event_id())
							.await
						{
							return;
						}
						services
							.rooms
							.timeline
							.progress_extremities(&pdu)
							.await
							.inspect_err(|e| {
								error!("Failed to progress extremities in {room_id}: {e:?}");
							})
							.ok();
					}
				})
				.await;
		} else if extremity_count > 20 {
			warn!(
				%extremity_count,
				"Room {room_id} has an abnormal number of forward extremities, retaining latest 20"
			);
			let mut thinned = services
				.rooms
				.state
				.get_forward_extremities(&room_id)
				.wide_filter_map(|event_id| {
					let room_id = &room_id;
					async move {
						let Ok(pdu) =
							services.rooms.timeline.get_non_outlier_pdu(&event_id).await
						else {
							services
								.rooms
								.state
								.remove_forward_extremity(room_id, &event_id);
							return None;
						};
						Some(pdu)
					}
				})
				.collect::<Vec<_>>()
				.await;
			thinned.sort_by_key(|pdu| {
				(
					i64::from(pdu.depth).saturating_mul(-1),
					i64::from(pdu.origin_server_ts).saturating_mul(-1),
				)
			});
			thinned
				.into_iter()
				.stream()
				.skip(20)
				.ready_for_each(move |pdu| {
					services
						.rooms
						.state
						.remove_forward_extremity(&room_id, pdu.event_id());
				})
				.await;
		}
	}

	Ok(())
}