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

use models::space::space_join_requests::{self, Entity as JoinRequests};
use models::space::{
    AllowJoinLevel, MemberStatus, SpaceJoinRequestState, space_group_permissions,
    space_member_groups, spaces,
};
use models::users::user_notifications::{self, NotificationType};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set};
use serde::{Deserialize, Serialize};
use time::{Duration, OffsetDateTime};
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;

use crate::authority;
use crate::errors::SpaceError;
use crate::services::dto::{UserDto, resolve_space_image_url};
use crate::services::service::SpaceService;
use crate::services::types::JoinRequestSummary;
use crate::share::authz::{MANAGE_MEMBERS, assert_not_banned, require_perm};
use crate::share::lookup;

pub const MAX_JOIN_REQUEST_NOTE_LEN: usize = 512;

pub type JoinRequestPage = export::pagination::Page<JoinRequestDto>;
pub type ExploreSpacePage = export::pagination::Page<ExploreSpaceDto>;

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct JoinRequestDto {
    pub id: Uuid,
    pub space_id: Uuid,
    pub user: Option<UserDto>,
    pub note: Option<String>,
    pub status: SpaceJoinRequestState,
    pub decided_by: Option<UserDto>,
    pub decided_at: Option<OffsetDateTime>,
    pub created_at: OffsetDateTime,
    pub updated_at: OffsetDateTime,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateJoinRequestCmd {
    #[serde(default)]
    pub note: Option<String>,
}

#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct JoinRequestPath {
    pub space_name: String,
    pub request_id: Uuid,
}

#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ListJoinRequestsQuery {
    pub status: Option<SpaceJoinRequestState>,
    pub limit: Option<u64>,
    pub offset: Option<u64>,
}

#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ExploreQuery {
    pub q: Option<String>,
    pub sort: Option<String>,
    pub limit: Option<u64>,
    pub offset: Option<u64>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct ExploreViewerRequestDto {
    pub id: Uuid,
    pub status: SpaceJoinRequestState,
    pub created_at: OffsetDateTime,
    pub decided_at: Option<OffsetDateTime>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cooldown_until: Option<OffsetDateTime>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct ExploreViewerDto {
    pub is_member: bool,
    pub request: Option<ExploreViewerRequestDto>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct ExploreSpaceDto {
    pub id: Uuid,
    pub lower_name: String,
    pub name: String,
    pub description: Option<String>,
    pub avatar_url: Option<String>,
    pub banner_url: Option<String>,
    pub members_count: i64,
    pub allow_join: bool,
    pub allow_join_level: AllowJoinLevel,
    pub created_at: OffsetDateTime,
    pub viewer: Option<ExploreViewerDto>,
}

fn join_request_dto(row: JoinRequestSummary, users: &HashMap<i64, UserDto>) -> JoinRequestDto {
    JoinRequestDto {
        id: row.id,
        space_id: row.organization_id,
        user: users.get(&row.user_id).cloned(),
        note: row.note,
        status: row.status,
        decided_by: row.decided_by.and_then(|uid| users.get(&uid).cloned()),
        decided_at: authority::to_time_opt(row.decided_at),
        created_at: authority::to_time(row.created_at),
        updated_at: authority::to_time(row.updated_at),
    }
}

impl SpaceService {
    async fn assert_join_requirements(
        &self,
        space: &spaces::Model,
        uid: i64,
    ) -> Result<(), SpaceError> {
        match AllowJoinLevel::from_i16(space.allow_join_level) {
            AllowJoinLevel::None => {}
            AllowJoinLevel::NeedMailActiveAndPhoneActive
            | AllowJoinLevel::NeedMailActive
            | AllowJoinLevel::NeedMailActiveAndLastOneMid
            | AllowJoinLevel::NeedMailActiveAndLastOneHours => {
                if !crate::authority::has_verified_email(&self.state, uid).await? {
                    return Err(SpaceError::JoinRequirementNotMet(
                        "an activated email address is required".to_string(),
                    ));
                }
                let now = OffsetDateTime::now_utc();
                let created_at = crate::authority::account_created_at(&self.state, uid).await?;
                match AllowJoinLevel::from_i16(space.allow_join_level) {
                    AllowJoinLevel::NeedMailActiveAndLastOneMid
                        if created_at.is_some_and(|created| created > now - Duration::days(1)) =>
                    {
                        return Err(SpaceError::JoinRequirementNotMet(
                            "the account must be at least one day old".to_string(),
                        ));
                    }
                    AllowJoinLevel::NeedMailActiveAndLastOneHours
                        if created_at.is_some_and(|created| created > now - Duration::hours(1)) =>
                    {
                        return Err(SpaceError::JoinRequirementNotMet(
                            "the account must be at least one hour old".to_string(),
                        ));
                    }
                    _ => {}
                }
            }
        }
        if space.member_must_enable_2fa && !crate::authority::mfa_enabled(&self.state, uid).await? {
            return Err(SpaceError::JoinRequirementNotMet(
                "two-factor authentication is required".to_string(),
            ));
        }
        Ok(())
    }

    async fn approver_uids(&self, space: &spaces::Model) -> Result<Vec<i64>, SpaceError> {
        let mut uids: HashSet<i64> = HashSet::new();
        uids.insert(space.owner_user_id);
        let member_groups = space_member_groups::Entity::find()
            .filter(space_member_groups::Column::SpaceId.eq(space.id))
            .all(self.db().read())
            .await?;
        let group_ids: Vec<Uuid> = {
            let mut ids: Vec<Uuid> = member_groups.iter().map(|mg| mg.group_id).collect();
            ids.sort();
            ids.dedup();
            ids
        };
        if !group_ids.is_empty() {
            let grants = space_group_permissions::Entity::find()
                .filter(space_group_permissions::Column::GroupId.is_in(group_ids))
                .filter(space_group_permissions::Column::Permission.eq(MANAGE_MEMBERS))
                .all(self.db().read())
                .await?;
            let managing_groups: HashSet<Uuid> = grants.into_iter().map(|g| g.group_id).collect();
            let candidates: Vec<i64> = member_groups
                .into_iter()
                .filter(|mg| managing_groups.contains(&mg.group_id))
                .map(|mg| mg.user_id)
                .collect();
            if !candidates.is_empty() {
                let candidate_set: HashSet<i64> = candidates.into_iter().collect();
                let active = crate::authority::list_members(
                    &self.state,
                    space.owner_user_id,
                    space.id,
                    false,
                    false,
                )
                .await?;
                uids.extend(
                    active
                        .into_iter()
                        .filter(|member| candidate_set.contains(&member.user_id))
                        .map(|member| member.user_id),
                );
            }
        }
        Ok(uids.into_iter().collect())
    }

    async fn notify_join_event(
        &self,
        recipient: i64,
        title: String,
        data: serde_json::Value,
    ) -> Result<(), SpaceError> {
        let row = user_notifications::ActiveModel {
            id: Set(Uuid::now_v7()),
            user_id: Set(recipient),
            notification_type: Set(NotificationType::System),
            title: Set(title),
            body: Set(None),
            data_json: Set(Some(data)),
            actor_uid: Set(None),
            read_at: Set(None),
            created_at: Set(OffsetDateTime::now_utc()),
        }
        .insert(self.db().write())
        .await?;
        export::notify::publish_notification(&self.state, recipient, row.to_event_value()).await;
        Ok(())
    }

    pub async fn explore_spaces(
        &self,
        viewer: Option<i64>,
        query: ExploreQuery,
    ) -> Result<ExploreSpacePage, SpaceError> {
        let (limit, offset) = export::pagination::page_bounds(
            query.limit,
            query.offset,
            self.config().space.default_pagination,
            self.config().space.max_pagination,
        );

        let rows = crate::authority::list_public_spaces(
            &self.state,
            query.q.as_deref().unwrap_or_default(),
        )
        .await?;
        let mut ranked = Vec::with_capacity(rows.len());
        for row in rows {
            let count = crate::authority::member_count(&self.state, row.id).await?;
            ranked.push((row, count));
        }
        if query.sort.as_deref() == Some("newest") {
            ranked.sort_by(|a, b| b.0.created_at.cmp(&a.0.created_at));
        } else {
            ranked.sort_by(|a, b| {
                b.1.cmp(&a.1)
                    .then_with(|| b.0.created_at.cmp(&a.0.created_at))
            });
        }
        let total = ranked.len() as i64;
        let rows: Vec<(spaces::Model, i64)> = ranked
            .into_iter()
            .skip(offset as usize)
            .take(limit as usize)
            .collect();

        let mut memberships: HashSet<Uuid> = HashSet::new();
        let mut latest_requests: HashMap<Uuid, space_join_requests::Model> = HashMap::new();
        if let Some(uid) = viewer
            && !rows.is_empty()
        {
            let space_ids: Vec<Uuid> = rows.iter().map(|(space, _)| space.id).collect();
            let joined: HashSet<Uuid> = crate::authority::list_user_spaces(&self.state, uid, uid)
                .await?
                .into_iter()
                .filter(crate::authority::space_is_active)
                .map(|space| space.id)
                .collect();
            memberships = space_ids
                .iter()
                .copied()
                .filter(|id| joined.contains(id))
                .collect();
            let requests = JoinRequests::find()
                .filter(space_join_requests::Column::UserId.eq(uid))
                .filter(space_join_requests::Column::OrganizationId.is_in(space_ids))
                .order_by_desc(space_join_requests::Column::CreatedAt)
                .all(self.db().read())
                .await?;
            for row in requests {
                latest_requests.entry(row.organization_id).or_insert(row);
            }
        }

        let cooldown_hours = self.config().space.join_request_cooldown_hours.max(0);
        let mut items = Vec::with_capacity(rows.len());
        for (space, members_count) in rows {
            let extras = crate::authority::extras(&self.state, space.id).await?;
            let viewer = viewer.map(|_| ExploreViewerDto {
                is_member: memberships.contains(&space.id),
                request: latest_requests
                    .get(&space.id)
                    .map(|row| ExploreViewerRequestDto {
                        id: row.id,
                        status: row.status,
                        created_at: authority::to_time(row.created_at),
                        decided_at: authority::to_time_opt(row.decided_at),
                        cooldown_until: if row.status == SpaceJoinRequestState::Rejected {
                            Some(
                                authority::to_time(row.decided_at.unwrap_or(row.created_at))
                                    + Duration::hours(cooldown_hours),
                            )
                        } else {
                            None
                        },
                    }),
            });
            items.push(ExploreSpaceDto {
                id: space.id,
                lower_name: space.slug.clone(),
                name: space.name,
                description: space.description,
                avatar_url: resolve_space_image_url(
                    &space.slug,
                    "avatar",
                    extras
                        .as_ref()
                        .and_then(|row| row.avatar_url.clone())
                        .or_else(|| space.avatar_url.clone()),
                ),
                banner_url: resolve_space_image_url(
                    &space.slug,
                    "banner",
                    extras.and_then(|row| row.banner_url),
                ),
                members_count,
                allow_join: space.allow_join,
                allow_join_level: AllowJoinLevel::from_i16(space.allow_join_level),
                created_at: authority::to_time(space.created_at),
                viewer,
            });
        }

        Ok(ExploreSpacePage {
            items,
            total,
            limit,
            offset,
        })
    }

    #[tracing::instrument(name = "space.create_join_request", skip_all, fields(caller = %caller, space = %space_name))]
    pub async fn create_join_request(
        &self,
        caller: i64,
        space_name: &str,
        cmd: CreateJoinRequestCmd,
    ) -> Result<JoinRequestDto, SpaceError> {
        if !self
            .state
            .redis
            .quota_allow(
                &format!("quota:space_join_request:{caller}"),
                self.config().quota.space_join_request_per_day,
                86_400,
            )
            .await
        {
            return Err(SpaceError::RateLimited(
                "join request quota exceeded, retry tomorrow".to_string(),
            ));
        }
        let space = lookup::lookup_space_by_name(self, caller, space_name).await?;
        if space.join_policy == models::space::SpaceJoinPolicy::Private || !space.allow_join {
            return Err(SpaceError::JoinNotAllowed(
                "this space does not accept join requests".to_string(),
            ));
        }

        if let Some(member) =
            crate::authority::member(&self.state, caller, space.id, caller).await?
        {
            if member.status() == MemberStatus::Active {
                return Err(SpaceError::Conflict(
                    "user is already a member of this space".to_string(),
                ));
            }
            if member.status() == MemberStatus::Banned {
                return Err(SpaceError::JoinNotAllowed(
                    "you are banned from this space".to_string(),
                ));
            }
        }

        let pending = JoinRequests::find()
            .filter(space_join_requests::Column::OrganizationId.eq(space.id))
            .filter(space_join_requests::Column::UserId.eq(caller))
            .filter(space_join_requests::Column::Status.eq(SpaceJoinRequestState::Pending))
            .one(self.db().read())
            .await?;
        if pending.is_some() {
            return Err(SpaceError::Conflict(
                "a join request is already pending".to_string(),
            ));
        }

        if let Some(last) = JoinRequests::find()
            .filter(space_join_requests::Column::OrganizationId.eq(space.id))
            .filter(space_join_requests::Column::UserId.eq(caller))
            .filter(space_join_requests::Column::Status.eq(SpaceJoinRequestState::Rejected))
            .order_by_desc(space_join_requests::Column::CreatedAt)
            .one(self.db().read())
            .await?
        {
            let cooldown_hours = self.config().space.join_request_cooldown_hours.max(0);
            let decided_at = authority::to_time(last.decided_at.unwrap_or(last.created_at));
            if OffsetDateTime::now_utc() < decided_at + Duration::hours(cooldown_hours) {
                return Err(SpaceError::JoinRequestCooldown(
                    "a declined request must cool down before trying again".to_string(),
                ));
            }
        }

        self.assert_join_requirements(&space, caller).await?;

        let note = cmd
            .note
            .map(|raw| raw.trim().to_string())
            .filter(|value| !value.is_empty());
        if let Some(value) = &note
            && value.chars().count() > MAX_JOIN_REQUEST_NOTE_LEN
        {
            return Err(SpaceError::BadRequest(format!(
                "join request note must be at most {MAX_JOIN_REQUEST_NOTE_LEN} characters"
            )));
        }

        let inserted = self
            .admin_create_join_request(caller, space.id, note, None)
            .await?;

        let data = serde_json::json!({
            "space_id": space.id,
            "space_name": space.slug,
            "request_id": inserted.id,
            "applicant_id": caller,
            "kind": "join_request_submitted",
        });
        for uid in self.approver_uids(&space).await? {
            if uid == caller {
                continue;
            }
            let locale = self.recipient_locale(Some(uid)).await;
            let title =
                i18n::catalog::notification_join_request_submitted_title(locale, &space.name);
            if let Err(error) = self.notify_join_event(uid, title, data.clone()).await {
                tracing::warn!(%error, %uid, space_id = %space.id, "join request notification failed");
            }
        }

        let users: HashMap<i64, UserDto> = self.load_user_briefs(&[caller]).await?;
        Ok(join_request_dto(inserted, &users))
    }

    #[tracing::instrument(name = "space.withdraw_join_request", skip_all, fields(caller = %caller, space = %space_name, request_id = %request_id))]
    pub async fn withdraw_join_request(
        &self,
        caller: i64,
        space_name: &str,
        request_id: Uuid,
    ) -> Result<(), SpaceError> {
        let space = lookup::lookup_space_by_name(self, caller, space_name).await?;
        let request = JoinRequests::find_by_id(request_id)
            .filter(space_join_requests::Column::OrganizationId.eq(space.id))
            .one(self.db().read())
            .await?
            .ok_or(SpaceError::NotFound)?;
        if request.user_id != caller {
            return Err(SpaceError::NotFound);
        }
        if request.status != SpaceJoinRequestState::Pending {
            return Err(SpaceError::Conflict(
                "join request is not pending".to_string(),
            ));
        }
        self.admin_withdraw_join_request(caller, space.id, request_id, None)
            .await?;
        Ok(())
    }

    pub async fn list_join_requests(
        &self,
        caller: i64,
        space_name: &str,
        query: ListJoinRequestsQuery,
    ) -> Result<JoinRequestPage, SpaceError> {
        let space = lookup::lookup_space_by_name(self, caller, space_name).await?;
        require_perm(self, &space, caller, MANAGE_MEMBERS).await?;

        let (limit, offset) = export::pagination::page_bounds(
            query.limit,
            query.offset,
            self.config().space.default_pagination,
            self.config().space.max_pagination,
        );

        let (rows, total) = self
            .admin_list_join_requests(caller, space.id, query.status, limit, offset, None)
            .await?;

        let mut uids: Vec<i64> = rows.iter().filter_map(|row| row.decided_by).collect();
        uids.extend(rows.iter().map(|row| row.user_id));
        let users: HashMap<i64, UserDto> = self.load_user_briefs(&uids).await?;

        Ok(JoinRequestPage {
            items: rows
                .into_iter()
                .map(|row| join_request_dto(row, &users))
                .collect(),
            total: total as i64,
            limit,
            offset,
        })
    }

    async fn decide_join_request(
        &self,
        caller: i64,
        space_name: &str,
        request_id: Uuid,
        approve: bool,
    ) -> Result<JoinRequestDto, SpaceError> {
        let space = lookup::lookup_space_by_name(self, caller, space_name).await?;
        require_perm(self, &space, caller, MANAGE_MEMBERS).await?;

        let request = JoinRequests::find_by_id(request_id)
            .filter(space_join_requests::Column::OrganizationId.eq(space.id))
            .one(self.db().read())
            .await?
            .ok_or(SpaceError::NotFound)?;
        if request.status != SpaceJoinRequestState::Pending {
            return Err(SpaceError::Conflict(
                "join request is not pending".to_string(),
            ));
        }

        if approve {
            assert_not_banned(&self.state, space.id, request.user_id).await?;
            if space.member_must_enable_2fa
                && !crate::authority::mfa_enabled(&self.state, request.user_id).await?
            {
                return Err(SpaceError::Forbidden);
            }
        }

        let updated = if approve {
            self.admin_approve_join_request(caller, space.id, request_id, None)
                .await?
        } else {
            self.admin_reject_join_request(caller, space.id, request_id, None, None)
                .await?
        };

        let applicant_id = request.user_id;
        if approve {
            self.grant_public_room_memberships(space.id, applicant_id)
                .await;
        }

        let locale = self.recipient_locale(Some(applicant_id)).await;
        let title = if approve {
            i18n::catalog::notification_join_request_approved_title(locale, &space.name)
        } else {
            i18n::catalog::notification_join_request_rejected_title(locale, &space.name)
        };
        let data = serde_json::json!({
            "space_id": space.id,
            "space_name": space.slug,
            "request_id": request_id,
            "applicant_id": applicant_id,
            "kind": if approve { "join_request_approved" } else { "join_request_rejected" },
        });
        if let Err(error) = self.notify_join_event(applicant_id, title, data).await {
            tracing::warn!(%error, %applicant_id, %request_id, "join request decision notification failed");
        }

        let users: HashMap<i64, UserDto> = self.load_user_briefs(&[applicant_id, caller]).await?;
        Ok(join_request_dto(updated, &users))
    }

    #[tracing::instrument(name = "space.approve_join_request", skip_all, fields(caller = %caller, space = %space_name, request_id = %request_id))]
    pub async fn approve_join_request(
        &self,
        caller: i64,
        space_name: &str,
        request_id: Uuid,
    ) -> Result<JoinRequestDto, SpaceError> {
        self.decide_join_request(caller, space_name, request_id, true)
            .await
    }

    #[tracing::instrument(name = "space.reject_join_request", skip_all, fields(caller = %caller, space = %space_name, request_id = %request_id))]
    pub async fn reject_join_request(
        &self,
        caller: i64,
        space_name: &str,
        request_id: Uuid,
    ) -> Result<JoinRequestDto, SpaceError> {
        self.decide_join_request(caller, space_name, request_id, false)
            .await
    }
}