use std::{collections::HashSet, iter::once};

use conduwuit::trace;
use conduwuit_core::{
	Err, Result,
	matrix::{event::Event, pdu::PartialPdu},
	utils::{IterStream, ReadyExt},
};
use futures::{FutureExt, StreamExt};
use ruma::{
	OwnedEventId, OwnedServerName, RoomId, UserId,
	events::{
		TimelineEventType,
		room::{
			member::{MembershipState, RoomMemberEventContent},
			redaction::RoomRedactionEventContent,
		},
	},
};

use super::{ExtractBody, RoomMutexGuard};

impl super::Service {
	/// Creates a new persisted data unit and adds it to a room. This function
	/// takes a roomid_mutex_state, meaning that only this function is able to
	/// mutate the room state.
	#[tracing::instrument(skip(self, state_lock, partial_pdu), level = "trace")]
	pub async fn build_and_append_pdu(
		&self,
		partial_pdu: PartialPdu,
		sender: &UserId,
		room_id: Option<&RoomId>,
		state_lock: &RoomMutexGuard,
	) -> Result<OwnedEventId> {
		let (pdu, pdu_json) = self
			.create_hash_and_sign_event(partial_pdu, sender, room_id, state_lock)
			.await?;

		let room_id = pdu.room_id_or_hash();
		if self.services.admin.is_admin_room(&room_id).await {
			self.check_pdu_for_admin_room(&pdu, sender).boxed().await?;
		}

		// If redaction event is not authorized, do not append it to the timeline
		if *pdu.kind() == TimelineEventType::RoomRedaction {
			use ruma::RoomVersionId::*;
			trace!("Running redaction checks for room {room_id}");
			match self.services.state.get_room_version(&room_id).await? {
				| V1 | V2 | V3 | V4 | V5 | V6 | V7 | V8 | V9 | V10 => {
					if let Some(redact_id) = pdu.redacts() {
						if !self
							.services
							.state_accessor
							.user_can_redact(redact_id, pdu.sender(), &room_id, false)
							.await?
						{
							return Err!(Request(Forbidden("User cannot redact this event.")));
						}
					}
				},
				| _ => {
					let content: RoomRedactionEventContent = pdu.get_content()?;
					if let Some(redact_id) = &content.redacts {
						if !self
							.services
							.state_accessor
							.user_can_redact(redact_id, pdu.sender(), &room_id, false)
							.await?
						{
							return Err!(Request(Forbidden("User cannot redact this event.")));
						}
					}
				},
			}
		}

		if *pdu.kind() == TimelineEventType::RoomMember {
			trace!("Running room member checks for room {room_id}");
			let content: RoomMemberEventContent = pdu.get_content()?;

			if content.join_authorized_via_users_server.is_some()
				&& content.membership != MembershipState::Join
			{
				return Err!(Request(BadJson(
					"join_authorised_via_users_server is only for member joins"
				)));
			}

			if content
				.join_authorized_via_users_server
				.as_ref()
				.is_some_and(|authorising_user| {
					!self.services.globals.user_is_local(authorising_user)
				}) {
				return Err!(Request(InvalidParam(
					"Authorising user does not belong to this homeserver"
				)));
			}
		}
		if *pdu.kind() == TimelineEventType::RoomCreate {
			trace!("Creating shortroomid for {room_id}");
			self.services
				.short
				.get_or_create_shortroomid(&room_id)
				.await;
		}

		// We append to state before appending the pdu, so we don't have a moment in
		// time with the pdu without it's state. This is okay because append_pdu can't
		// fail.
		trace!("Appending {} state for room {room_id}", pdu.event_id());
		let statehashid = self.services.state.append_to_state(&pdu, &room_id).await?;
		trace!("State hash ID for {room_id}: {statehashid:?}");

		trace!("Generating raw ID for PDU {}", pdu.event_id());
		let pdu_id = self
			.append_pdu(
				&pdu,
				pdu_json,
				// Since this PDU references all pdu_leaves we can update the leaves
				// of the room
				once(pdu.event_id()),
				state_lock,
				&room_id,
			)
			.boxed()
			.await?;

		// Process admin commands for locally sent events
		if *pdu.kind() == TimelineEventType::RoomMessage {
			let content: ExtractBody = pdu.get_content()?;
			if let Some(body) = content.body {
				if let Some(source) = self
					.services
					.admin
					.is_admin_command(&pdu, &body, true)
					.await
				{
					self.services.admin.command_with_sender(
						body,
						Some(pdu.event_id().into()),
						source,
						pdu.sender.clone(),
					)?;
				}
			}
		}

		// We set the room state after inserting the pdu, so that we never have a moment
		// in time where events in the current room state do not exist
		trace!("Setting room state for room {room_id}");
		self.services
			.state
			.set_room_state(&room_id, statehashid, state_lock);

		self.services.sync.wake_all_joined(&room_id).await;

		if *pdu.kind() == TimelineEventType::RoomMember {
			if let Some(target_user) = pdu
				.state_key
				.as_ref()
				.and_then(|state_key| UserId::parse(state_key.as_str()).ok())
			{
				if self.services.globals.user_is_local(&target_user) {
					self.services.sync.wake(&target_user).await;
				}
			}
		}

		let mut servers: HashSet<OwnedServerName> = self
			.services
			.state_cache
			.room_servers(&room_id)
			.collect()
			.await;

		// In case we are kicking or banning a user, we need to inform their server of
		// the change
		if *pdu.kind() == TimelineEventType::RoomMember {
			if let Some(state_key_uid) = &pdu
				.state_key
				.as_ref()
				.and_then(|state_key| UserId::parse(state_key.as_str()).ok())
			{
				servers.insert(state_key_uid.server_name().to_owned());
			}
		}

		// Remove our server from the server list since it will be added to it by
		// room_servers() and/or the if statement above
		servers.remove(self.services.globals.server_name());

		trace!("Sending PDU {} to {} servers", pdu.event_id(), servers.len());
		self.services
			.sending
			.send_pdu_servers(servers.stream(), &pdu_id)
			.await?;

		trace!("Event {} in room {:?} has been appended", pdu.event_id(), room_id);
		Ok(pdu.event_id().to_owned())
	}

	/// Assert invariants about the admin room, to prevent (for example) all
	/// admins from leaving or being banned from the room
	#[tracing::instrument(skip_all, level = "debug")]
	async fn check_pdu_for_admin_room<Pdu>(&self, pdu: &Pdu, sender: &UserId) -> Result
	where
		Pdu: Event + Send + Sync,
	{
		match pdu.kind() {
			| TimelineEventType::RoomEncryption => {
				return Err!(Request(Forbidden(error!(
					"Encryption not supported in admins room."
				))));
			},
			| TimelineEventType::RoomMember => {
				let target = pdu
					.state_key()
					.filter(|v| v.starts_with('@'))
					.unwrap_or(sender.as_str());

				let server_user = &self.services.globals.server_user.to_string();

				let content: RoomMemberEventContent = pdu.get_content()?;
				match content.membership {
					| MembershipState::Leave => {
						if target == server_user {
							return Err!(Request(Forbidden(error!(
								"Server user cannot leave the admins room."
							))));
						}

						let count = self
							.services
							.state_cache
							.room_members(&pdu.room_id_or_hash())
							.ready_filter(|user| self.services.globals.user_is_local(user))
							.ready_filter(|user| *user != target)
							.boxed()
							.count()
							.await;

						if count < 2 {
							return Err!(Request(Forbidden(error!(
								"Last admin cannot leave the admins room."
							))));
						}
					},

					| MembershipState::Ban if pdu.state_key().is_some() => {
						if target == server_user {
							return Err!(Request(Forbidden(error!(
								"Server cannot be banned from admins room."
							))));
						}

						let count = self
							.services
							.state_cache
							.room_members(&pdu.room_id_or_hash())
							.ready_filter(|user| self.services.globals.user_is_local(user))
							.ready_filter(|user| *user != target)
							.boxed()
							.count()
							.await;

						if count < 2 {
							return Err!(Request(Forbidden(error!(
								"Last admin cannot be banned from admins room."
							))));
						}
					},
					| _ => {},
				}
			},
			| _ => {},
		}

		Ok(())
	}
}
