b453d5d
use std::collections::{HashMap, HashSet};
use models::space::space_group_permissions;
use models::space::space_join_request::{self, SpaceJoinRequestStatus};
use models::space::space_member_groups;
use models::space::space_members::{self, MemberStatus};
use models::space::spaces::{self, AllowJoinLevel};
use models::users::user;
use models::users::user_emails;
use models::users::user_notifications::{self, NotificationType};
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, Condition, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
QuerySelect, Set, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use time::{Duration, OffsetDateTime};
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::errors::{SpaceError, is_unique_violation};
use crate::services::members::{MemberActivationDefaults, activate_member_row};
use crate::services::service::SpaceService;
use crate::services::spaces::{UserDto, resolve_space_image_url};
use crate::share::authz::{MANAGE_MEMBERS, assert_not_banned, require_perm};
use crate::share::lookup::lookup_space_by_name;
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: SpaceJoinRequestStatus,
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<SpaceJoinRequestStatus>,
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: SpaceJoinRequestStatus,
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: space_join_request::Model,
users: &HashMap<i64, UserDto>,
) -> JoinRequestDto {
JoinRequestDto {
id: row.id,
space_id: row.space_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: row.decided_at,
created_at: row.created_at,
updated_at: row.updated_at,
}
}
impl SpaceService {
async fn has_activated_email(&self, uid: i64) -> Result<bool, SpaceError> {
let count = user_emails::Entity::find()
.filter(user_emails::Column::UserId.eq(uid))
.filter(user_emails::Column::IsActivated.eq(true))
.filter(user_emails::Column::DeletedAt.is_null())
.count(self.db().read())
.await?;
Ok(count > 0)
}
async fn assert_join_requirements(
&self,
space: &spaces::Model,
uid: i64,
applicant: &user::Model,
) -> Result<(), SpaceError> {
match space.allow_join_level {
AllowJoinLevel::None => {}
AllowJoinLevel::NeedMailActiveAndPhoneActive
| AllowJoinLevel::NeedMailActive
| AllowJoinLevel::NeedMailActiveAndLastOneMid
| AllowJoinLevel::NeedMailActiveAndLastOneHours => {
if !self.has_activated_email(uid).await? {
return Err(SpaceError::JoinRequirementNotMet(
"an activated email address is required".to_string(),
));
}
let now = OffsetDateTime::now_utc();
match space.allow_join_level {
AllowJoinLevel::NeedMailActiveAndLastOneMid
if applicant.created_at > now - Duration::days(1) =>
{
return Err(SpaceError::JoinRequirementNotMet(
"the account must be at least one day old".to_string(),
));
}
AllowJoinLevel::NeedMailActiveAndLastOneHours
if applicant.created_at > 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 && !applicant.totp_enabled {
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_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 active: Vec<i64> = space_members::Entity::find()
.filter(space_members::Column::SpaceId.eq(space.id))
.filter(space_members::Column::UserId.is_in(candidates.clone()))
.filter(space_members::Column::Status.eq(MemberStatus::Active))
.all(self.db().read())
.await?
.into_iter()
.map(|m| m.user_id)
.collect();
uids.extend(active);
}
}
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 max_pagination = self.config().space.max_pagination;
let default_pagination = self.config().space.default_pagination;
let limit = query
.limit
.unwrap_or(default_pagination)
.clamp(1, max_pagination);
let offset = query
.offset
.unwrap_or(0)
.clamp(0, export::pagination::MAX_OFFSET);
let mut select = spaces::Entity::find()
.filter(spaces::Column::IsActive.eq(true))
.filter(spaces::Column::IsPrivate.eq(false));
if let Some(raw) = query
.q
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
let needle = raw.to_ascii_lowercase();
select = select.filter(
Condition::any()
.add(spaces::Column::LowerName.contains(needle))
.add(spaces::Column::Name.contains(raw)),
);
}
select = if query.sort.as_deref() == Some("newest") {
select.order_by_desc(spaces::Column::CreatedAt)
} else {
select
.order_by_desc(spaces::Column::NumsMembers)
.order_by_desc(spaces::Column::CreatedAt)
};
let total = select.clone().count(self.db().read()).await? as i64;
let rows = select
.limit(limit)
.offset(offset)
.all(self.db().read())
.await?;
let mut memberships: HashSet<Uuid> = HashSet::new();
let mut latest_requests: HashMap<Uuid, space_join_request::Model> = HashMap::new();
if let Some(uid) = viewer
&& !rows.is_empty()
{
let space_ids: Vec<Uuid> = rows.iter().map(|s| s.id).collect();
memberships = space_members::Entity::find()
.filter(space_members::Column::UserId.eq(uid))
.filter(space_members::Column::SpaceId.is_in(space_ids.clone()))
.filter(space_members::Column::Status.eq(MemberStatus::Active))
.all(self.db().read())
.await?
.into_iter()
.map(|m| m.space_id)
.collect();
let requests = space_join_request::Entity::find()
.filter(space_join_request::Column::UserId.eq(uid))
.filter(space_join_request::Column::SpaceId.is_in(space_ids))
.order_by_desc(space_join_request::Column::CreatedAt)
.all(self.db().read())
.await?;
for row in requests {
latest_requests.entry(row.space_id).or_insert(row);
}
}
let cooldown_hours = self.config().space.join_request_cooldown_hours.max(0);
let items = rows
.into_iter()
.map(|s| {
let viewer = viewer.map(|_| ExploreViewerDto {
is_member: memberships.contains(&s.id),
request: latest_requests
.get(&s.id)
.map(|row| ExploreViewerRequestDto {
id: row.id,
status: row.status,
created_at: row.created_at,
decided_at: row.decided_at,
cooldown_until: if row.status == SpaceJoinRequestStatus::Rejected {
Some(
row.decided_at.unwrap_or(row.created_at)
+ Duration::hours(cooldown_hours),
)
} else {
None
},
}),
});
ExploreSpaceDto {
id: s.id,
lower_name: s.lower_name.clone(),
name: s.name,
description: s.description,
avatar_url: resolve_space_image_url(&s.lower_name, "avatar", s.avatar_url),
banner_url: resolve_space_image_url(&s.lower_name, "banner", s.banner_url),
members_count: s.nums_members,
allow_join: s.allow_join,
allow_join_level: s.allow_join_level,
created_at: s.created_at,
viewer,
}
})
.collect();
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_space_by_name(self, space_name).await?;
if space.is_private || !space.allow_join {
return Err(SpaceError::JoinNotAllowed(
"this space does not accept join requests".to_string(),
));
}
let applicant = user::Entity::find_by_id(caller)
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
let existing = space_members::Entity::find_by_id((space.id, caller))
.one(self.db().read())
.await?;
if existing.is_some_and(|member| member.status == MemberStatus::Active) {
return Err(SpaceError::Conflict(
"user is already a member of this space".to_string(),
));
}
if let Some(ban) = models::space::space_bans::Entity::find_by_id((space.id, caller))
.one(self.db().read())
.await?
&& (ban.expires_at.is_none() || ban.expires_at > Some(OffsetDateTime::now_utc()))
{
return Err(SpaceError::JoinNotAllowed(
"you are banned from this space".to_string(),
));
}
let pending = space_join_request::Entity::find()
.filter(space_join_request::Column::SpaceId.eq(space.id))
.filter(space_join_request::Column::UserId.eq(caller))
.filter(space_join_request::Column::Status.eq(SpaceJoinRequestStatus::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) = space_join_request::Entity::find()
.filter(space_join_request::Column::SpaceId.eq(space.id))
.filter(space_join_request::Column::UserId.eq(caller))
.filter(space_join_request::Column::Status.eq(SpaceJoinRequestStatus::Rejected))
.order_by_desc(space_join_request::Column::CreatedAt)
.one(self.db().read())
.await?
{
let cooldown_hours = self.config().space.join_request_cooldown_hours.max(0);
let decided_at = 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, &applicant)
.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 client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
let created = client
.create_join_request(
caller,
&space.id.to_string(),
note.as_deref().unwrap_or_default(),
)
.await?;
let request_id = Uuid::parse_str(&created.id).map_err(|_| {
SpaceError::Internal("kenoauth returned a non-uuid request".to_string())
})?;
let now = OffsetDateTime::now_utc();
let request_row = space_join_request::ActiveModel {
id: Set(request_id),
space_id: Set(space.id),
user_id: Set(caller),
status: Set(SpaceJoinRequestStatus::Pending),
note: Set(note),
decided_by: Set(None),
decided_at: Set(None),
created_at: Set(now),
updated_at: Set(now),
};
let inserted = match request_row.insert(self.db().write()).await {
Ok(row) => row,
Err(error) if is_unique_violation(&error) => {
return Err(SpaceError::Conflict(
"a join request is already pending".to_string(),
));
}
Err(error) => return Err(error.into()),
};
let data = serde_json::json!({
"space_id": space.id,
"space_name": space.lower_name,
"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_space_by_name(self, space_name).await?;
let request = space_join_request::Entity::find_by_id(request_id)
.filter(space_join_request::Column::SpaceId.eq(space.id))
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
if request.user_id != caller {
return Err(SpaceError::NotFound);
}
if request.status != SpaceJoinRequestStatus::Pending {
return Err(SpaceError::Conflict(
"join request is not pending".to_string(),
));
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
client
.withdraw_join_request(caller, &space.id.to_string(), &request_id.to_string())
.await?;
let now = OffsetDateTime::now_utc();
let updated = space_join_request::Entity::update_many()
.col_expr(
space_join_request::Column::Status,
Expr::value(SpaceJoinRequestStatus::Withdrawn),
)
.col_expr(
space_join_request::Column::DecidedAt,
Expr::value(Some(now)),
)
.col_expr(space_join_request::Column::UpdatedAt, Expr::value(now))
.filter(space_join_request::Column::Id.eq(request_id))
.filter(space_join_request::Column::Status.eq(SpaceJoinRequestStatus::Pending))
.exec(self.db().write())
.await?;
if updated.rows_affected == 0 {
return Err(SpaceError::Conflict(
"join request is not pending".to_string(),
));
}
Ok(())
}
pub async fn list_join_requests(
&self,
caller: i64,
space_name: &str,
query: ListJoinRequestsQuery,
) -> Result<JoinRequestPage, SpaceError> {
let space = lookup_space_by_name(self, space_name).await?;
require_perm(self, &space, caller, MANAGE_MEMBERS).await?;
let max_pagination = self.config().space.max_pagination;
let default_pagination = self.config().space.default_pagination;
let limit = query
.limit
.unwrap_or(default_pagination)
.clamp(1, max_pagination);
let offset = query
.offset
.unwrap_or(0)
.clamp(0, export::pagination::MAX_OFFSET);
let mut select = space_join_request::Entity::find()
.filter(space_join_request::Column::SpaceId.eq(space.id));
if let Some(status) = query.status {
select = select.filter(space_join_request::Column::Status.eq(status));
}
select = select.order_by_desc(space_join_request::Column::CreatedAt);
let total = select.clone().count(self.db().read()).await? as i64;
let rows = select
.limit(limit)
.offset(offset)
.all(self.db().read())
.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,
limit,
offset,
})
}
async fn decide_join_request(
&self,
caller: i64,
space_name: &str,
request_id: Uuid,
approve: bool,
) -> Result<JoinRequestDto, SpaceError> {
let space = lookup_space_by_name(self, space_name).await?;
require_perm(self, &space, caller, MANAGE_MEMBERS).await?;
let request = space_join_request::Entity::find_by_id(request_id)
.filter(space_join_request::Column::SpaceId.eq(space.id))
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
if request.status != SpaceJoinRequestStatus::Pending {
return Err(SpaceError::Conflict(
"join request is not pending".to_string(),
));
}
if approve {
assert_not_banned(self.db().read(), space.id, request.user_id).await?;
let applicant = user::Entity::find_by_id(request.user_id)
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
if space.member_must_enable_2fa && !applicant.totp_enabled {
return Err(SpaceError::Forbidden);
}
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
let decided_member = if approve {
let (_, member) = client
.approve_join_request(caller, &space.id.to_string(), &request_id.to_string())
.await?;
member
} else {
client
.reject_join_request(caller, &space.id.to_string(), &request_id.to_string(), "")
.await?;
None
};
let now = OffsetDateTime::now_utc();
let status = if approve {
SpaceJoinRequestStatus::Approved
} else {
SpaceJoinRequestStatus::Rejected
};
let space_id = space.id;
let applicant_id = request.user_id;
self.db()
.primary()
.transaction::<_, (), SpaceError>(|txn| {
Box::pin(async move {
let updated = space_join_request::Entity::update_many()
.col_expr(space_join_request::Column::Status, Expr::value(status))
.col_expr(
space_join_request::Column::DecidedBy,
Expr::value(Some(caller)),
)
.col_expr(
space_join_request::Column::DecidedAt,
Expr::value(Some(now)),
)
.col_expr(space_join_request::Column::UpdatedAt, Expr::value(now))
.filter(space_join_request::Column::Id.eq(request_id))
.filter(
space_join_request::Column::Status.eq(SpaceJoinRequestStatus::Pending),
)
.exec(txn)
.await?;
if updated.rows_affected == 0 {
return Err(SpaceError::Conflict(
"join request is not pending".to_string(),
));
}
if approve {
activate_member_row(
txn,
space_id,
applicant_id,
now,
MemberActivationDefaults::default(),
)
.await?;
}
Ok(())
})
})
.await?;
if approve {
if let Some(member) = decided_member.as_ref() {
crate::sync::upsert_member(self.db().write(), space_id, member).await?;
}
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.lower_name,
"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 updated = space_join_request::Entity::find_by_id(request_id)
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
let users: HashMap<i64, UserDto> = self.load_user_briefs(&[updated.user_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
}
}