main
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<JoinRequestDto>;
pub type ExploreSpacePage = api::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, 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<String>,
pub sort: Option<ExploreSort>,
pub order: Option<SortOrder>,
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 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.access.database.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.access.database.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, 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<i64>,
query: ExploreQuery,
) -> Result<ExploreSpacePage, SpaceError> {
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<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, 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<JoinRequestDto, SpaceError> {
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<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, 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<JoinRequestPage, SpaceError> {
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<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,
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, 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<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
}
}