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_emails; 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, pub email: String, pub groups: Vec, pub created_at: OffsetDateTime, pub expires_at: Option, pub accepted_at: Option, pub revoked_at: Option, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct CreateInviteCmd { #[serde(default)] pub user_id: Option, #[serde(default)] pub email: Option, #[serde(default)] pub group_ids: Vec, #[serde( default, deserialize_with = "export::serde_datetime::deserialize_opt_datetime" )] pub expires_at: Option, } #[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; 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 { let space = lookup_space_by_name(self, 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 { user::Entity::find_by_id(uid) .one(self.db().read()) .await? .ok_or(SpaceError::NotFound)?; if let Some(ref em) = email_norm { let match_email = user_emails::Entity::find() .filter(user_emails::Column::UserId.eq(uid)) .filter(user_emails::Column::LowerEmail.eq(em)) .filter(user_emails::Column::IsActivated.eq(true)) .filter(user_emails::Column::DeletedAt.is_null()) .one(self.db().read()) .await?; if match_email.is_none() { return Err(SpaceError::BadRequest( "email does not match user's activated email".to_string(), )); } } } else if let Some(ref em) = email_norm { let email_row = user_emails::Entity::find() .filter(user_emails::Column::LowerEmail.eq(em)) .filter(user_emails::Column::IsActivated.eq(true)) .filter(user_emails::Column::DeletedAt.is_null()) .one(self.db().read()) .await?; if let Some(er) = email_row { target_uid = Some(er.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 = 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 = email_norm; if recipient_email.is_none() && let Some(uid) = target_uid { let primary = user_emails::Entity::find() .filter(user_emails::Column::UserId.eq(uid)) .filter(user_emails::Column::IsPrimary.eq(true)) .filter(user_emails::Column::IsActivated.eq(true)) .filter(user_emails::Column::DeletedAt.is_null()) .one(self.db().read()) .await?; if let Some(p) = primary { recipient_email = Some(p.lower_email); } } 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#"

{}

{}

{}

"#, 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 = 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 { let space = lookup_space_by_name(self, 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 = 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() { user_emails::Entity::find() .filter(user_emails::Column::UserId.eq(caller)) .filter(user_emails::Column::LowerEmail.eq(&invite.email)) .filter(user_emails::Column::IsActivated.eq(true)) .filter(user_emails::Column::DeletedAt.is_null()) .one(self.db().read()) .await? .is_some() } 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 = 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 = spaces::Entity::find_by_id(invite.space_id) .one(self.db().read()) .await? .ok_or(SpaceError::NotFound)?; if !space.is_active { return Err(SpaceError::NotFound); } assert_not_banned(self.db().read(), space.id, caller).await?; if space.member_must_enable_2fa { let caller_user = user::Entity::find_by_id(caller) .one(self.db().read()) .await? .ok_or(SpaceError::NotFound)?; if !caller_user.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 redeemed = client.redeem_invite(caller, &cmd.code).await?; let existing_member = space_members::Entity::find_by_id((invite.space_id, caller)) .one(self.db().read()) .await?; if let Some(m) = &existing_member && m.status == MemberStatus::Active { return Err(SpaceError::Conflict( "user is already a member of this space".to_string(), )); } let now = OffsetDateTime::now_utc(); let member_am: Option = existing_member.map(|m| m.into()); let member_row = space_members::ActiveModel { space_id: Set(invite.space_id), user_id: Set(caller), alias_name: Set(None), alias_avatar_url: Set(None), status: Set(MemberStatus::Active), last_active_at: Set(None), left_at: Set(None), invited_at: Set(Some(invite.created_at)), invited_by: Set(None), joined_at: Set(now), created_at: Set(now), updated_at: Set(Some(now)), message_notifications: Set(MessageNotifications::All), muted: Set(false), mute_until: Set(None), mobile_push: Set(false), suppress_everyone: Set(false), suppress_roles: Set(false), hide_muted_channels: Set(false), ..Default::default() }; let invite_space_id = invite.space_id; 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 { if let Some(am) = member_am { let mut am = am; am.status = Set(MemberStatus::Active); am.left_at = Set(None); am.updated_at = Set(Some(now)); am.update(txn).await?; } else { member_row.insert(txn).await?; } 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?; } spaces::Entity::update_many() .filter(spaces::Column::Id.eq(space.id)) .col_expr(spaces::Column::NumsMembers, Expr::cust("nums_members + 1")) .exec(txn) .await?; invite_am.update(txn).await?; Ok(()) }) }) .await?; if let Some(member) = redeemed.member.as_ref() { crate::sync::upsert_member(self.db().write(), invite_space_id, member).await?; } Ok(()) } pub async fn list_invites( &self, caller: i64, space_name: &str, q: PageQuery, ) -> Result { let space = lookup_space_by_name(self, 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 = 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> = HashMap::new(); for ig in all_invite_groups { groups_map .entry(ig.invite_id) .or_default() .push(ig.group_id); } let user_ids: Vec = rows.iter().filter_map(|i| i.user_id).collect(); let user_map: HashMap = 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, }) } }