acbf7ea
use std::collections::HashMap;
use models::space::space_group_permissions;
use models::space::space_groups;
use models::space::space_invite;
use models::space::space_invite_groups;
use models::space::space_member_groups;
use models::space::space_members::{self, MemberStatus, MessageNotifications};
use models::space::spaces;
use models::users::user;
use models::users::user_notifications;
use models::users::user_notifications::NotificationType;
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
QuerySelect, Set, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::SpaceError;
use crate::services::service::SpaceService;
use crate::services::spaces::{PageQuery, UserDto};
use crate::share::authz::{MANAGE_INVITES, assert_not_banned, effective_permissions, require_perm};
use crate::share::lookup::lookup_space_by_name;
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct InviteDto {
pub id: Uuid,
pub space_id: Uuid,
pub user: Option<UserDto>,
pub email: String,
pub groups: Vec<Uuid>,
pub created_at: OffsetDateTime,
pub expires_at: Option<OffsetDateTime>,
pub accepted_at: Option<OffsetDateTime>,
pub revoked_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateInviteCmd {
#[serde(default)]
pub user_id: Option<i64>,
#[serde(default)]
pub email: Option<String>,
#[serde(default)]
pub group_ids: Vec<Uuid>,
#[serde(
default,
deserialize_with = "export::serde_datetime::deserialize_opt_datetime"
)]
pub expires_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct AcceptInviteCmd {
pub code: String,
}
fn proto_timestamp(value: OffsetDateTime) -> prost_types::Timestamp {
prost_types::Timestamp {
seconds: value.unix_timestamp(),
nanos: value.nanosecond() as i32,
}
}
pub type InvitePage = export::pagination::Page<InviteDto>;
fn validate_email(email: &str, max_len: usize) -> bool {
if email.is_empty() || email.chars().count() > max_len {
return false;
}
if email.chars().any(|c| c.is_whitespace()) {
return false;
}
let parts: Vec<&str> = email.split('@').collect();
if parts.len() != 2 || parts[0].is_empty() || parts[1].is_empty() {
return false;
}
true
}
impl SpaceService {
#[tracing::instrument(name = "space.create_invite", skip_all, fields(caller = %caller, space = %space_name))]
pub async fn create_invite(
&self,
caller: i64,
space_name: &str,
cmd: CreateInviteCmd,
) -> Result<InviteDto, SpaceError> {
let space = lookup_space_by_name(self, caller, space_name).await?;
require_perm(self, &space, caller, MANAGE_INVITES).await?;
let email_norm = cmd
.email
.map(|e| e.trim().to_lowercase())
.filter(|e| !e.is_empty());
if let Some(ref em) = email_norm
&& !validate_email(em, self.config().space.max_invite_email_len)
{
return Err(SpaceError::BadRequest("invalid invite email".to_string()));
}
if cmd.user_id.is_none() && email_norm.is_none() {
return Err(SpaceError::BadRequest(
"either user_id or email must be provided".to_string(),
));
}
let mut group_ids = cmd.group_ids;
group_ids.sort();
group_ids.dedup();
if group_ids.is_empty() {
return Err(SpaceError::BadRequest(
"at least one group is required".to_string(),
));
}
let mut target_uid = cmd.user_id;
if let Some(uid) = target_uid {
let public = export::users::get_public_user(&self.state, uid)
.await?
.ok_or(SpaceError::NotFound)?;
if let Some(ref em) = email_norm
&& (!public.email_verified || !public.email.eq_ignore_ascii_case(em))
{
return Err(SpaceError::BadRequest(
"email does not match user's activated email".to_string(),
));
}
} else if let Some(ref em) = email_norm
&& let Some(public) = export::users::find_by_email(&self.state, em).await?
&& public.email_verified
{
target_uid = Some(public.user_id);
}
let group_count = space_groups::Entity::find()
.filter(space_groups::Column::SpaceId.eq(space.id))
.filter(space_groups::Column::Id.is_in(group_ids.clone()))
.count(self.db().read())
.await?;
if group_count != group_ids.len() as u64 {
return Err(SpaceError::BadRequest(
"some groups do not belong to this space".to_string(),
));
}
if caller != space.owner_id {
let caller_perms = effective_permissions(&self.state, &space, caller)
.await?
.into_vec();
let group_perms = space_group_permissions::Entity::find()
.filter(space_group_permissions::Column::GroupId.is_in(group_ids.clone()))
.all(self.db().read())
.await?;
if group_perms
.iter()
.any(|gp| !caller_perms.contains(&gp.permission))
{
return Err(SpaceError::Forbidden);
}
}
if let Some(uid) = target_uid {
let active_user_inv = space_invite::Entity::find()
.filter(space_invite::Column::SpaceId.eq(space.id))
.filter(space_invite::Column::UserId.eq(uid))
.filter(space_invite::Column::AcceptedAt.is_null())
.filter(space_invite::Column::RevokedAt.is_null())
.one(self.db().read())
.await?;
if active_user_inv.is_some() {
return Err(SpaceError::Conflict(
"active invite for this user already exists".to_string(),
));
}
}
if let Some(ref em) = email_norm {
let active_email_inv = space_invite::Entity::find()
.filter(space_invite::Column::SpaceId.eq(space.id))
.filter(space_invite::Column::Email.eq(em))
.filter(space_invite::Column::AcceptedAt.is_null())
.filter(space_invite::Column::RevokedAt.is_null())
.one(self.db().read())
.await?;
if active_email_inv.is_some() {
return Err(SpaceError::Conflict(
"active invite for this email already exists".to_string(),
));
}
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
let method = if target_uid.is_some() { 3 } else { 4 };
let role_ids: Vec<String> = group_ids.iter().map(|value| value.to_string()).collect();
let (invite_info, accept_code) = client
.create_invite(
caller,
&space.id.to_string(),
method,
1,
cmd.expires_at.map(proto_timestamp),
role_ids,
target_uid.unwrap_or(0),
email_norm.as_deref().unwrap_or_default(),
)
.await?;
let invite_id = Uuid::parse_str(&invite_info.id)
.map_err(|_| SpaceError::Internal("kenoauth returned a non-uuid invite".to_string()))?;
let now = OffsetDateTime::now_utc();
let final_email = email_norm.clone().unwrap_or_default();
let invite_row = space_invite::ActiveModel {
id: Set(invite_id),
space_id: Set(space.id),
user_id: Set(target_uid),
email: Set(final_email.clone()),
accept_code: Set(accept_code.clone()),
created_at: Set(now),
expires_at: Set(cmd.expires_at),
accepted_at: Set(None),
revoked_at: Set(None),
};
let group_ids_clone = group_ids.clone();
let inserted = self
.db()
.primary()
.transaction::<_, space_invite::Model, SpaceError>(|txn| {
Box::pin(async move {
let inv = invite_row.insert(txn).await?;
for gid in group_ids_clone {
let grp_row = space_invite_groups::ActiveModel {
invite_id: Set(invite_id),
group_id: Set(gid),
};
grp_row.insert(txn).await?;
}
Ok(inv)
})
})
.await?;
if let Some(invitee) = inserted.user_id {
let data = serde_json::json!({
"space_id": space.id,
// Routes address a space by its `lower_name`, so the inbox can
// link straight to it after the invitee accepts.
"space_name": space.lower_name,
"invite_id": invite_id,
// Same credential the invite email carries: without it the
// accept button in the inbox has nothing to send.
"accept_code": accept_code,
});
let row = user_notifications::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(invitee),
notification_type: Set(NotificationType::Invite),
title: Set(i18n::catalog::notification_invite_title(
self.recipient_locale(Some(invitee)).await,
&space.name,
)),
body: Set(None),
data_json: Set(Some(data)),
actor_uid: Set(Some(caller)),
read_at: Set(None),
created_at: Set(OffsetDateTime::now_utc()),
}
.insert(self.db().write())
.await?;
export::notify::publish_notification(&self.state, invitee, row.to_event_value()).await;
}
let mut recipient_email: Option<String> = email_norm;
if recipient_email.is_none()
&& let Some(uid) = target_uid
&& let Some(public) = export::users::get_public_user(&self.state, uid).await?
&& public.email_verified
&& !public.email.trim().is_empty()
{
recipient_email = Some(public.email.to_lowercase());
}
if let Some(recipient) = recipient_email {
// `recipient_email` may be a plain address with no account yet; the
// invitee uid is the account signal when it exists.
let locale = self.recipient_locale(target_uid).await;
let subject = i18n::catalog::space_invite_subject(locale, &space.name);
// 邮件链接落在 SPA 的接受页：它校验邀请码，并带着 `space` 提示把
// 新成员直接送进刚加入的空间。
let accept_url = format!(
"{}/spaces/invites/accept?code={accept_code}&space={}",
self.config().web.web_base_url,
space.lower_name
);
let text = i18n::catalog::space_invite_text(locale, &space.name, &accept_url);
let html = format!(
r#"<p>{}</p><p><a href="{}" style="display:inline-block;padding:8px 16px;background:#0066cc;color:#fff;text-decoration:none;border-radius:4px;">{}</a></p><p>{}</p>"#,
i18n::catalog::space_invite_body(locale, &space.name),
accept_url,
i18n::catalog::space_invite_accept(locale),
i18n::catalog::space_invite_or_copy(locale, &accept_url)
);
if let Err(err) = self.send_email(&recipient, &subject, &text, &html).await {
tracing::warn!(error = %err, "failed to send space invite email");
}
}
let user_map: HashMap<i64, UserDto> = match inserted.user_id {
Some(uid) => self.load_user_briefs(&[uid]).await?,
None => HashMap::new(),
};
let user_brief = inserted.user_id.and_then(|uid| user_map.get(&uid).cloned());
Ok(InviteDto {
id: inserted.id,
space_id: inserted.space_id,
user: user_brief,
email: inserted.email,
groups: group_ids,
created_at: inserted.created_at,
expires_at: inserted.expires_at,
accepted_at: inserted.accepted_at,
revoked_at: inserted.revoked_at,
})
}
#[tracing::instrument(name = "space.revoke_invite", skip_all, fields(caller = %caller, space = %space_name, invite_id = %invite_id))]
pub async fn revoke_invite(
&self,
caller: i64,
space_name: &str,
invite_id: Uuid,
) -> Result<InviteDto, SpaceError> {
let space = lookup_space_by_name(self, caller, space_name).await?;
require_perm(self, &space, caller, MANAGE_INVITES).await?;
let invite = space_invite::Entity::find_by_id(invite_id)
.filter(space_invite::Column::SpaceId.eq(space.id))
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
client
.revoke_invite(caller, &space.id.to_string(), &invite_id.to_string())
.await?;
let now = OffsetDateTime::now_utc();
let mut am: space_invite::ActiveModel = invite.into();
am.revoked_at = Set(Some(now));
let updated = am.update(self.db().write()).await?;
let groups = space_invite_groups::Entity::find()
.filter(space_invite_groups::Column::InviteId.eq(updated.id))
.all(self.db().read())
.await?;
let group_ids = groups.into_iter().map(|g| g.group_id).collect();
let user_map: HashMap<i64, UserDto> = match updated.user_id {
Some(uid) => self.load_user_briefs(&[uid]).await?,
None => HashMap::new(),
};
let user_brief = updated.user_id.and_then(|uid| user_map.get(&uid).cloned());
Ok(InviteDto {
id: updated.id,
space_id: updated.space_id,
user: user_brief,
email: updated.email,
groups: group_ids,
created_at: updated.created_at,
expires_at: updated.expires_at,
accepted_at: updated.accepted_at,
revoked_at: updated.revoked_at,
})
}
/// The invitee declines: revoke the row so the inviter's invite list stops
/// reporting it as pending. Only the addressed invitee may do this.
#[tracing::instrument(name = "space.reject_invite", skip_all, fields(caller = %caller))]
pub async fn reject_invite(&self, caller: i64, accept_code: &str) -> Result<(), SpaceError> {
let invite = space_invite::Entity::find()
.filter(space_invite::Column::AcceptCode.eq(accept_code))
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
if invite.user_id != Some(caller) {
return Err(SpaceError::Forbidden);
}
if invite.accepted_at.is_some() {
return Err(SpaceError::Conflict("invite already accepted".to_string()));
}
if invite.revoked_at.is_some() {
return Ok(());
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
client
.revoke_invite(caller, &invite.space_id.to_string(), &invite.id.to_string())
.await?;
space_invite::Entity::update_many()
.col_expr(
space_invite::Column::RevokedAt,
Expr::value(Some(OffsetDateTime::now_utc())),
)
.filter(space_invite::Column::Id.eq(invite.id))
.filter(space_invite::Column::AcceptedAt.is_null())
.filter(space_invite::Column::RevokedAt.is_null())
.exec(self.db().write())
.await?;
Ok(())
}
#[tracing::instrument(name = "space.accept_invite", skip_all, fields(caller = %caller))]
pub async fn accept_invite(&self, caller: i64, cmd: AcceptInviteCmd) -> Result<(), SpaceError> {
if !self
.state
.redis
.quota_allow(&format!("space:invite_accept:{caller}"), 10, 60)
.await
{
return Err(SpaceError::RateLimited(
"too many invite acceptance attempts, retry in 60s".to_string(),
));
}
let invite = space_invite::Entity::find()
.filter(space_invite::Column::AcceptCode.eq(&cmd.code))
.one(self.db().read())
.await?
.ok_or(SpaceError::NotFound)?;
let is_assigned_user = invite.user_id == Some(caller);
let is_assigned_email = if !invite.email.is_empty() {
export::users::get_public_user(&self.state, caller)
.await?
.is_some_and(|user| {
user.email_verified && user.email.eq_ignore_ascii_case(&invite.email)
})
} else {
false
};
if !is_assigned_user && !is_assigned_email {
return Err(SpaceError::Forbidden);
}
if invite.revoked_at.is_some() {
return Err(SpaceError::Conflict("invite has been revoked".to_string()));
}
if invite.accepted_at.is_some() {
return Err(SpaceError::Conflict("invite already accepted".to_string()));
}
if let Some(exp) = invite.expires_at
&& exp < OffsetDateTime::now_utc()
{
return Err(SpaceError::Conflict("invite has expired".to_string()));
}
let inv_groups = space_invite_groups::Entity::find()
.filter(space_invite_groups::Column::InviteId.eq(invite.id))
.all(self.db().read())
.await?;
let group_ids: Vec<Uuid> = inv_groups.into_iter().map(|g| g.group_id).collect();
if group_ids.is_empty() {
return Err(SpaceError::Conflict(
"invite has no valid groups".to_string(),
));
}
let valid_count = space_groups::Entity::find()
.filter(space_groups::Column::SpaceId.eq(invite.space_id))
.filter(space_groups::Column::Id.is_in(group_ids.clone()))
.count(self.db().read())
.await?;
if valid_count != group_ids.len() as u64 {
return Err(SpaceError::Conflict(
"invite has invalid groups".to_string(),
));
}
let space = crate::authority::space_by_id(&self.state, caller, invite.space_id).await?;
if !space.is_active {
return Err(SpaceError::NotFound);
}
assert_not_banned(&self.state, space.id, caller).await?;
if space.member_must_enable_2fa
&& !crate::authority::mfa_enabled(&self.state, caller).await?
{
return Err(SpaceError::Forbidden);
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_string()))?;
let redeemed = client.redeem_invite(caller, &cmd.code).await?;
if let Some(member) =
crate::authority::member(&self.state, caller, invite.space_id, caller).await?
&& member.status == MemberStatus::Active
{
return Err(SpaceError::Conflict(
"user is already a member of this space".to_string(),
));
}
let now = OffsetDateTime::now_utc();
let invite_space_id = invite.space_id;
let invite_created_at = invite.created_at;
let mut invite_am: space_invite::ActiveModel = invite.into();
invite_am.accepted_at = Set(Some(now));
self.db()
.primary()
.transaction::<_, (), SpaceError>(|txn| {
Box::pin(async move {
for gid in group_ids {
let mg = space_member_groups::ActiveModel {
space_id: Set(space.id),
user_id: Set(caller),
group_id: Set(gid),
created_at: Set(now),
};
mg.insert(txn).await?;
}
invite_am.update(txn).await?;
Ok(())
})
})
.await?;
let _ = redeemed;
crate::authority::upsert_member_extras(
&self.state,
invite_space_id,
caller,
Some(invite_created_at),
)
.await?;
Ok(())
}
pub async fn list_invites(
&self,
caller: i64,
space_name: &str,
q: PageQuery,
) -> Result<InvitePage, SpaceError> {
let space = lookup_space_by_name(self, caller, space_name).await?;
require_perm(self, &space, caller, MANAGE_INVITES).await?;
let max_pagination = self.config().space.max_pagination;
let default_pagination = self.config().space.default_pagination;
let limit = q
.limit
.unwrap_or(default_pagination)
.clamp(1, max_pagination);
let offset = q
.offset
.unwrap_or(0)
.clamp(0, export::pagination::MAX_OFFSET);
let select = space_invite::Entity::find()
.filter(space_invite::Column::SpaceId.eq(space.id))
.order_by_desc(space_invite::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 invite_ids: Vec<Uuid> = rows.iter().map(|i| i.id).collect();
let all_invite_groups = space_invite_groups::Entity::find()
.filter(space_invite_groups::Column::InviteId.is_in(invite_ids))
.all(self.db().read())
.await?;
let mut groups_map: HashMap<Uuid, Vec<Uuid>> = HashMap::new();
for ig in all_invite_groups {
groups_map
.entry(ig.invite_id)
.or_default()
.push(ig.group_id);
}
let user_ids: Vec<i64> = rows.iter().filter_map(|i| i.user_id).collect();
let user_map: HashMap<i64, UserDto> = self.load_user_briefs(&user_ids).await?;
let items = rows
.into_iter()
.map(|i| {
let user_brief = i.user_id.and_then(|uid| user_map.get(&uid).cloned());
let groups = groups_map.remove(&i.id).unwrap_or_default();
InviteDto {
id: i.id,
space_id: i.space_id,
user: user_brief,
email: i.email,
groups,
created_at: i.created_at,
expires_at: i.expires_at,
accepted_at: i.accepted_at,
revoked_at: i.revoked_at,
}
})
.collect();
Ok(InvitePage {
items,
total,
limit,
offset,
})
}
}