use std::{borrow::ToOwned, pin::pin};

use axum::extract::State;
use conduwuit::{
	Err, Error, Result, debug, debug_info, info, matrix::pdu::PartialPdu, utils, warn,
};
use conduwuit_service::Services;
use futures::StreamExt;
use ruma::{
	OwnedUserId, RoomId, UserId,
	api::{
		error::{ErrorKind, IncompatibleRoomVersionErrorData},
		federation::membership::prepare_join_event,
	},
	assign,
	events::{
		StateEventType,
		room::{
			join_rules::{AllowRule, JoinRule, RoomJoinRulesEventContent},
			member::{MembershipState, RoomMemberEventContent},
		},
	},
};
use serde_json::value::to_raw_value;
use service::rooms::state::RoomMutexGuard;
use tokio::join;

use crate::Ruma;

/// # `GET /_matrix/federation/v1/make_join/{roomId}/{userId}`
///
/// Creates a join template.
#[tracing::instrument(skip_all, fields(room_id = %body.room_id, user_id = %body.user_id, origin = %body.origin()), level = "info")]
pub(crate) async fn create_join_event_template_route(
	State(services): State<crate::State>,
	body: Ruma<prepare_join_event::v1::Request>,
) -> Result<prepare_join_event::v1::Response> {
	if !services.rooms.metadata.exists(&body.room_id).await {
		return Err!(Request(NotFound("Room is unknown to this server.")));
	}
	if !services
		.rooms
		.state_cache
		.server_in_room(services.globals.server_name(), &body.room_id)
		.await
	{
		info!(
			origin = body.origin().as_str(),
			room_id = %body.room_id,
			"Refusing to serve make_join for room we aren't participating in"
		);
		return Err!(Request(NotFound("This server is not participating in that room.")));
	}

	if body.user_id.server_name() != body.origin() {
		return Err!(Request(BadJson("Not allowed to join on behalf of another server/user.")));
	}

	// ACL check origin server
	services
		.rooms
		.event_handler
		.acl_check(body.origin(), &body.room_id)
		.await?;

	if services
		.moderation
		.is_remote_server_forbidden(body.origin())
	{
		warn!(
			"Server {} for remote user {} tried joining room ID {} which has a server name that \
			 is globally forbidden. Rejecting.",
			body.origin(),
			&body.user_id,
			&body.room_id,
		);
		return Err!(Request(Forbidden("Server is banned on this homeserver.")));
	}

	if let Some(server) = body.room_id.server_name() {
		if services.moderation.is_remote_server_forbidden(server) {
			return Err!(Request(Forbidden(warn!(
				"Room ID server name {server} is banned on this homeserver."
			))));
		}
	}

	let room_version = services.rooms.state.get_room_version(&body.room_id).await?;
	let room_version_rules = room_version.rules().unwrap();
	if !body.ver.contains(&room_version) {
		return Err(Error::BadRequest(
			ErrorKind::IncompatibleRoomVersion(IncompatibleRoomVersionErrorData::new(
				room_version,
			)),
			"Room version not supported.",
		));
	}

	let state_lock = services.rooms.state.mutex.lock(body.room_id.as_str()).await;
	let (is_invited, is_joined) = join!(
		services
			.rooms
			.state_cache
			.is_invited(&body.user_id, &body.room_id),
		services
			.rooms
			.state_cache
			.is_joined(&body.user_id, &body.room_id)
	);
	let join_authorized_via_users_server: Option<OwnedUserId> = {
		if is_joined || is_invited {
			// User is already joined or invited and consequently does not need an
			// authorising user
			None
		} else if !room_version_rules.authorization.restricted_join_rule {
			// room version does not support restricted join rules
			None
		} else if user_can_perform_restricted_join(&services, &body.user_id, &body.room_id)
			.await?
		{
			Some(
				select_authorising_user(&services, &body.room_id, &body.user_id, &state_lock)
					.await?,
			)
		} else {
			None
		}
	};
	if services.antispam.check_all_joins() && join_authorized_via_users_server.is_none() {
		if services
			.antispam
			.meowlnir_accept_make_join(body.room_id.clone(), body.user_id.clone())
			.await
			.is_err()
		{
			return Err!(Request(Forbidden("Antispam rejected join request.")));
		}
	}

	let (pdu, _) = services
		.rooms
		.timeline
		.create_event(
			PartialPdu::state(
				body.user_id.to_string(),
				&assign!(RoomMemberEventContent::new(MembershipState::Join), {
					join_authorized_via_users_server,
				}),
			),
			&body.user_id,
			Some(&body.room_id),
			&state_lock,
		)
		.await?;
	drop(state_lock);
	let mut pdu_json = utils::to_canonical_object(&pdu)
		.expect("Barebones PDU should be convertible to canonical JSON");
	pdu_json.remove("event_id");

	Ok(
		assign!(prepare_join_event::v1::Response::new(to_raw_value(&pdu_json).expect("CanonicalJson can be serialized to JSON")), {
			room_version: Some(room_version),
		}),
	)
}

/// Attempts to find a user who is able to issue an invite in the target room.
pub(crate) async fn select_authorising_user<'a>(
	services: &Services,
	room_id: &'a RoomId,
	user_id: &'a UserId,
	state_lock: &'a RoomMutexGuard,
) -> Result<OwnedUserId> {
	let candidates = services.rooms.state_cache.local_users_in_room(room_id);

	let mut candidates = pin!(candidates);

	while let Some(candidate) = candidates.next().await {
		if services
			.rooms
			.state_accessor
			.user_can_invite(room_id, &candidate, user_id, state_lock)
			.await
		{
			return Ok(candidate);
		}
	}

	Err!(Request(UnableToGrantJoin(
		"No user on this server is able to assist in joining."
	)))
}

/// Checks whether the given user can join the given room via a restricted join.
pub(crate) async fn user_can_perform_restricted_join(
	services: &Services,
	user_id: &UserId,
	room_id: &RoomId,
) -> Result<bool> {
	let Ok(join_rules_event_content) = services
		.rooms
		.state_accessor
		.room_state_get_content::<RoomJoinRulesEventContent>(
			room_id,
			&StateEventType::RoomJoinRules,
			"",
		)
		.await
	else {
		// No join rules means there's nothing to authorise (defaults to invite)
		return Ok(false);
	};

	let (JoinRule::Restricted(r) | JoinRule::KnockRestricted(r)) =
		join_rules_event_content.join_rule
	else {
		// This is not a restricted room
		return Ok(false);
	};

	if r.allow.is_empty() {
		// This will never be authorisable, return forbidden.
		return Err!(Request(Forbidden("You are not invited to this room.")));
	}

	let mut could_satisfy = true;
	for allow_rule in &r.allow {
		match allow_rule {
			| AllowRule::RoomMembership(membership) => {
				if !services
					.rooms
					.state_cache
					.server_in_room(services.globals.server_name(), &membership.room_id)
					.await
				{
					// Since we can't check this room, mark could_satisfy as false
					// so that we can return M_UNABLE_TO_AUTHORIZE_JOIN later.
					could_satisfy = false;
					continue;
				}

				if services
					.rooms
					.state_cache
					.is_joined(user_id, &membership.room_id)
					.await
				{
					debug!(
						"User {} is allowed to join room {} via membership in room {}",
						user_id, room_id, membership.room_id
					);
					return Ok(true);
				}
			},
			| other if other.rule_type() == "fi.mau.spam_checker" =>
				return match services
					.antispam
					.meowlnir_accept_make_join(room_id.to_owned(), user_id.to_owned())
					.await
				{
					| Ok(()) => Ok(true),
					| Err(_) => Err!(Request(Forbidden("Antispam rejected join request."))),
				},
			| _ => {
				// We don't recognise this join rule, so we cannot satisfy the request.
				could_satisfy = false;
				debug_info!(
					"Unsupported allow rule in restricted join for room {}: {:?}",
					room_id,
					allow_rule
				);
			},
		}
	}

	if could_satisfy {
		// We were able to check all the restrictions and can be certain that the
		// prospective member is not permitted to join.
		Err!(Request(Forbidden(
			"You do not belong to any of the rooms or spaces required to join this room."
		)))
	} else {
		// We were unable to check all the restrictions. This usually means we aren't in
		// one of the rooms this one is restricted to, ergo can't check its state for
		// the user's membership, and consequently the user *might* be able to join if
		// they ask another server.
		Err!(Request(UnableToAuthorizeJoin(
			"You do not belong to any of the recognised rooms or spaces required to join this \
			 room, but this server is unable to verify every requirement. You may be able to \
			 join via another server."
		)))
	}
}