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::NotificationType; use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder}; use serde::{Deserialize, Serialize}; use time::{Duration, OffsetDateTime}; use utoipa::{IntoParams, ToSchema}; use uuid::Uuid; use api::pagination::SortOrder; 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 = api::pagination::Page; pub type ExploreSpacePage = api::pagination::Page; #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct JoinRequestDto { pub id: Uuid, pub space_id: Uuid, pub user: Option, pub note: Option, pub status: SpaceJoinRequestState, pub decided_by: Option, pub decided_at: Option, pub created_at: OffsetDateTime, pub updated_at: OffsetDateTime, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct CreateJoinRequestCmd { #[serde(default)] pub note: Option, } #[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, pub limit: Option, pub offset: Option, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, ToSchema)] #[serde(rename_all = "snake_case")] pub enum ExploreSort { Members, Newest, } impl ExploreSort { fn default_order(self) -> SortOrder { SortOrder::Desc } } #[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)] #[serde(deny_unknown_fields)] pub struct ExploreQuery { pub q: Option, pub sort: Option, pub order: Option, pub limit: Option, pub offset: Option, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct ExploreViewerRequestDto { pub id: Uuid, pub status: SpaceJoinRequestState, pub created_at: OffsetDateTime, pub decided_at: Option, #[serde(skip_serializing_if = "Option::is_none")] pub cooldown_until: Option, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct ExploreViewerDto { pub is_member: bool, pub request: Option, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct ExploreSpaceDto { pub id: Uuid, pub lower_name: String, pub name: String, pub description: Option, pub avatar_url: Option, pub banner_url: Option, pub members_count: i64, pub allow_join: bool, pub allow_join_level: AllowJoinLevel, pub created_at: OffsetDateTime, pub viewer: Option, } fn join_request_dto(row: JoinRequestSummary, users: &HashMap) -> 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 approver_uids(&self, space: &spaces::Model) -> Result, SpaceError> { let mut uids: HashSet = 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.access.database.read()) .await?; let group_ids: Vec = { let mut ids: Vec = 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.access.database.read()) .await?; let managing_groups: HashSet = grants.into_iter().map(|g| g.group_id).collect(); let candidates: Vec = 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 = candidates.into_iter().collect(); let active = crate::authority::list_members(self, 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> { self.notifications .create( recipient, NotificationType::System, title, None, Some(data), None, ) .await?; Ok(()) } pub async fn explore_spaces( &self, viewer: Option, query: ExploreQuery, ) -> Result { let (limit, offset) = api::pagination::page_bounds( query.limit, query.offset, self.policy.settings.space.default_pagination, self.policy.settings.space.max_pagination, )?; let rows = crate::authority::list_public_spaces(self, 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, row.id).await?; ranked.push((row, count)); } let sort = query.sort.unwrap_or(ExploreSort::Members); let descending = query.order.unwrap_or(sort.default_order()) == SortOrder::Desc; match sort { ExploreSort::Newest => ranked.sort_by(|a, b| { if descending { b.0.created_at.cmp(&a.0.created_at) } else { a.0.created_at.cmp(&b.0.created_at) } }), ExploreSort::Members => ranked.sort_by(|a, b| { if descending { b.1.cmp(&a.1) .then_with(|| b.0.created_at.cmp(&a.0.created_at)) } else { a.1.cmp(&b.1) .then_with(|| b.0.created_at.cmp(&a.0.created_at)) } }), } let total = ranked.len() as u64; let rows: Vec<(spaces::Model, i64)> = ranked .into_iter() .skip(offset as usize) .take(limit as usize) .collect(); let mut memberships: HashSet = HashSet::new(); let mut latest_requests: HashMap = HashMap::new(); if let Some(uid) = viewer && !rows.is_empty() { let space_ids: Vec = rows.iter().map(|(space, _)| space.id).collect(); let joined: HashSet = crate::authority::list_user_spaces(self, 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.access.database.read()) .await?; for row in requests { latest_requests.entry(row.organization_id).or_insert(row); } } let cooldown_hours = self .policy .settings .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.access.database.read(), 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 { self.policy.allow_join_request(caller).await?; let space = lookup::lookup_space_by_name(self, 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, 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.access.database.write()) .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.access.database.read()) .await? { let cooldown_hours = self .policy .settings .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.policy.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) = ¬e && 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.policy.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 = 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, space_name).await?; let request = JoinRequests::find_by_id(request_id) .filter(space_join_requests::Column::OrganizationId.eq(space.id)) .one(self.access.database.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 { let space = lookup::lookup_space_by_name(self, space_name).await?; require_perm(self, &space, caller, MANAGE_MEMBERS).await?; let (limit, offset) = api::pagination::page_bounds( query.limit, query.offset, self.policy.settings.space.default_pagination, self.policy.settings.space.max_pagination, )?; let (rows, total) = self .admin_list_join_requests(caller, space.id, query.status, limit, offset, None) .await?; let mut uids: Vec = rows.iter().filter_map(|row| row.decided_by).collect(); uids.extend(rows.iter().map(|row| row.user_id)); let users: HashMap = self.load_user_briefs(&uids).await?; Ok(JoinRequestPage { items: rows .into_iter() .map(|row| join_request_dto(row, &users)) .collect(), total: total, limit, offset, }) } async fn decide_join_request( &self, caller: i64, space_name: &str, request_id: Uuid, approve: bool, ) -> Result { let space = lookup::lookup_space_by_name(self, 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.access.database.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, space.id, request.user_id).await?; if space.member_must_enable_2fa && !crate::authority::mfa_enabled(&self.policy.accounts, 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; let locale = self.policy.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 = 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 { 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 { self.decide_join_request(caller, space_name, request_id, false) .await } }