538d47b
use chrono::Utc;
use sea_orm::entity::Set;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QueryOrder,
QuerySelect, TransactionTrait,
};
use uuid::Uuid;
use crate::AuthSvc;
use crate::errors::{AuthError, AuthResult};
use crate::events::label;
use crate::models::space::{
OrgJoinMethod, OrgMemberRole, OrganizationStatus, organization_invite_redemptions,
organization_invites, organization_roles,
};
use crate::organization::member;
use crate::organization::service::{bump_version, normalize_reason};
use crate::organization::types::{
DEFAULT_INVITE_TTL_SECS, InviteSummary, SYSTEM_ROLE_MEMBER, SYSTEM_ROLE_OWNER,
};
use crate::share::crypto;
impl AuthSvc {
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, request_id = %request_id.unwrap_or_default()))]
pub async fn create_invite(
&self,
actor_user_id: i64,
organization_id: Uuid,
role_id: Option<Uuid>,
target_user_id: Option<i64>,
max_uses: i32,
ttl_secs: Option<u64>,
request_id: Option<&str>,
) -> AuthResult<(InviteSummary, String)> {
self.create_invite_with_extras(
actor_user_id,
organization_id,
role_id,
target_user_id,
max_uses,
ttl_secs,
None,
&[],
request_id,
)
.await
}
/// 邮箱型邀请与额外角色必须在同一个事务里写入，否则邀请已经可用而扩展字段丢失。
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, request_id = %request_id.unwrap_or_default()))]
pub async fn create_invite_with_extras(
&self,
actor_user_id: i64,
organization_id: Uuid,
role_id: Option<Uuid>,
target_user_id: Option<i64>,
max_uses: i32,
ttl_secs: Option<u64>,
email: Option<String>,
extra_role_ids: &[Uuid],
request_id: Option<&str>,
) -> AuthResult<(InviteSummary, String)> {
self.require_permission(actor_user_id, organization_id, "invite.create", request_id)
.await?;
let organization = self.load_organization(organization_id).await?;
if organization.status != OrganizationStatus::Active {
return Err(AuthError::OrganizationInactive);
}
if max_uses < 0 {
return Err(AuthError::MalformedRequest);
}
if let Some(role_id) = role_id {
let role = organization_roles::Entity::find_by_id(role_id)
.filter(organization_roles::Column::OrganizationId.eq(organization_id))
.one(self.db().read())
.await?
.ok_or(AuthError::OrganizationRoleNotFound)?;
if role.key == SYSTEM_ROLE_OWNER {
return Err(AuthError::Forbidden);
}
}
if !extra_role_ids.is_empty() {
// 额外角色等于提前授予权限，与主动分配角色同等审查。
self.require_permission(actor_user_id, organization_id, "role.assign", request_id)
.await?;
for extra_role_id in extra_role_ids {
let role = organization_roles::Entity::find_by_id(*extra_role_id)
.filter(organization_roles::Column::OrganizationId.eq(organization_id))
.one(self.db().read())
.await?
.ok_or(AuthError::OrganizationRoleNotFound)?;
if role.key == SYSTEM_ROLE_OWNER {
return Err(AuthError::Forbidden);
}
}
}
let email = email
.map(|value| crate::share::validate::validate_email(&value))
.transpose()?;
let method = match target_user_id {
Some(user_id) => {
crate::models::users::users::Entity::find_by_id(user_id)
.filter(crate::models::users::users::Column::DeletedAt.is_null())
.one(self.db().read())
.await?
.ok_or(AuthError::UserNotFound)?;
OrgJoinMethod::InviteTargeted
}
None => OrgJoinMethod::InviteLink,
};
let method = if email.is_some() {
OrgJoinMethod::InviteEmail
} else {
method
};
let extra_role_ids = if extra_role_ids.is_empty() {
None
} else {
Some(
extra_role_ids
.iter()
.map(|id| id.to_string())
.collect::<Vec<_>>()
.join(","),
)
};
// 明文令牌只在这里出现一次，库里只留 sha256。
let token = crypto::random_b64url(32);
let token_hash = crypto::sha256_hex(token.as_bytes());
let ttl = ttl_secs
.unwrap_or(DEFAULT_INVITE_TTL_SECS)
.clamp(60, 90 * 24 * 60 * 60);
let expires_at = Some(Utc::now() + chrono::Duration::seconds(ttl as i64));
let txn = self.db().write().begin().await?;
let row = organization_invites::ActiveModel {
organization_id: Set(organization_id),
method: Set(method),
token_hash: Set(token_hash),
role_id: Set(role_id),
target_user_id: Set(target_user_id),
created_by: Set(actor_user_id),
max_uses: Set(max_uses),
uses: Set(0),
email: Set(email),
extra_role_ids: Set(extra_role_ids),
expires_at: Set(expires_at),
..Default::default()
}
.insert(&txn)
.await?;
bump_version(&txn, organization_id).await?;
self.emit_invite(
&txn,
organization_id,
actor_user_id,
serde_json::json!({
"id": row.id,
"method": match row.method {
OrgJoinMethod::InviteLink => "invite_link",
OrgJoinMethod::InviteTargeted => "invite_targeted",
OrgJoinMethod::InviteEmail => "invite_email",
OrgJoinMethod::Direct => "direct",
},
"target_user_id": target_user_id,
"max_uses": max_uses,
"request_id": request_id,
}),
request_id,
)
.await?;
txn.commit().await?;
self.invalidate_org(organization_id).await?;
let summary = invite_summary(&row);
Ok((summary, token))
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, request_id = %request_id.unwrap_or_default()))]
pub async fn list_invites(
&self,
actor_user_id: i64,
organization_id: Uuid,
include_revoked: bool,
limit: u64,
offset: u64,
request_id: Option<&str>,
) -> AuthResult<Vec<InviteSummary>> {
self.require_permission(actor_user_id, organization_id, "invite.view", request_id)
.await?;
let mut query = organization_invites::Entity::find()
.filter(organization_invites::Column::OrganizationId.eq(organization_id));
if !include_revoked {
query = query.filter(organization_invites::Column::RevokedAt.is_null());
}
let rows = query
.order_by_desc(organization_invites::Column::CreatedAt)
.limit(limit)
.offset(offset)
.all(self.db().read())
.await?;
Ok(rows.iter().map(invite_summary).collect())
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, invite_id = %invite_id, request_id = %request_id.unwrap_or_default()))]
pub async fn revoke_invite(
&self,
actor_user_id: i64,
organization_id: Uuid,
invite_id: Uuid,
reason: Option<&str>,
request_id: Option<&str>,
) -> AuthResult<bool> {
let reason = normalize_reason(reason)?;
let invite = organization_invites::Entity::find_by_id(invite_id)
.filter(organization_invites::Column::OrganizationId.eq(organization_id))
.one(self.db().read())
.await?
.ok_or(AuthError::OrganizationInviteNotFound)?;
if invite.target_user_id != Some(actor_user_id) {
self.require_permission(actor_user_id, organization_id, "invite.revoke", request_id)
.await?;
}
if invite.revoked_at.is_some() {
return Ok(false);
}
let txn = self.db().write().begin().await?;
organization_invites::Entity::update_many()
.col_expr(
organization_invites::Column::RevokedAt,
Expr::value(Utc::now()),
)
.filter(organization_invites::Column::Id.eq(invite_id))
.exec(&txn)
.await?;
let version = bump_version(&txn, organization_id).await?;
self.emit_invite(
&txn,
organization_id,
actor_user_id,
serde_json::json!({
"id": invite_id,
"action": "revoke",
"reason": &reason,
"version": version,
"request_id": request_id,
}),
request_id,
)
.await?;
txn.commit().await?;
self.invalidate_org(organization_id).await?;
Ok(true)
}
/// 兑换邀请。整个过程在一个事务里完成，uses 递增用条件更新防止并发超发：
/// `UPDATE ... WHERE uses < max_uses` 影响 0 行即视为已用尽。
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, request_id = %request_id.unwrap_or_default()))]
pub async fn redeem_invite(
&self,
actor_user_id: i64,
token: &str,
request_id: Option<&str>,
) -> AuthResult<(Uuid, Uuid, bool)> {
let trimmed = token.trim();
if trimmed.is_empty() || trimmed.len() > 128 {
return Err(AuthError::OrganizationInviteNotFound);
}
let token_hash = crypto::sha256_hex(trimmed.as_bytes());
let invite = organization_invites::Entity::find()
.filter(organization_invites::Column::TokenHash.eq(token_hash))
.one(self.db().read())
.await?
.ok_or(AuthError::OrganizationInviteNotFound)?;
if invite.revoked_at.is_some() {
return Err(AuthError::OrganizationInviteExhausted);
}
if let Some(expires_at) = invite.expires_at
&& expires_at <= Utc::now()
{
return Err(AuthError::OrganizationInviteExpired);
}
if let Some(target_user_id) = invite.target_user_id
&& target_user_id != actor_user_id
{
return Err(AuthError::Forbidden);
}
let organization = self.load_organization(invite.organization_id).await?;
if organization.status != OrganizationStatus::Active {
return Err(AuthError::OrganizationInactive);
}
let actor = crate::models::users::users::Entity::find_by_id(actor_user_id)
.filter(crate::models::users::users::Column::DeletedAt.is_null())
.one(self.db().read())
.await?
.ok_or(AuthError::UserNotFound)?;
if let Some(email) = invite
.email
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
&& !actor.email.eq_ignore_ascii_case(email)
{
return Err(AuthError::Forbidden);
}
let txn = self.db().write().begin().await?;
let mut consume = organization_invites::Entity::update_many()
.col_expr(
organization_invites::Column::Uses,
Expr::col(organization_invites::Column::Uses).add(1),
)
.filter(organization_invites::Column::Id.eq(invite.id));
if invite.max_uses > 0 {
consume = consume.filter(organization_invites::Column::Uses.lt(invite.max_uses));
}
let consumed = consume.exec(&txn).await?;
if consumed.rows_affected == 0 {
return Err(AuthError::OrganizationInviteExhausted);
}
let role = match invite.role_id {
Some(role_id) => {
organization_roles::Entity::find_by_id(role_id)
.filter(organization_roles::Column::OrganizationId.eq(invite.organization_id))
.one(&txn)
.await?
}
None => self.default_role_row(&txn, invite.organization_id).await?,
};
let method = match invite.method {
OrgJoinMethod::InviteTargeted => OrgJoinMethod::InviteTargeted,
OrgJoinMethod::InviteEmail => OrgJoinMethod::InviteEmail,
_ => OrgJoinMethod::InviteLink,
};
let already = member::find_any(&txn, invite.organization_id, actor_user_id).await?;
let (row, joined) = member::upsert_join(
&txn,
invite.organization_id,
actor_user_id,
OrgMemberRole::Member,
method,
Some(invite.created_by),
Some(invite.id),
)
.await?;
if let Some(role) = role
&& role.key != SYSTEM_ROLE_OWNER
{
crate::organization::role::bind_role(&txn, row.id, role.id, invite.created_by).await?;
}
if let Some(extra) = invite.extra_role_ids.as_deref() {
for raw in extra.split(',') {
let Ok(role_id) = Uuid::parse_str(raw.trim()) else {
continue;
};
let Some(role) = organization_roles::Entity::find_by_id(role_id)
.filter(organization_roles::Column::OrganizationId.eq(invite.organization_id))
.one(&txn)
.await?
else {
continue;
};
if role.key == SYSTEM_ROLE_OWNER || role.key == SYSTEM_ROLE_MEMBER {
continue;
}
crate::organization::role::bind_role(&txn, row.id, role.id, invite.created_by)
.await?;
}
}
let redeemed = already.is_none();
if redeemed {
let existing = organization_invite_redemptions::Entity::find()
.filter(organization_invite_redemptions::Column::InviteId.eq(invite.id))
.filter(organization_invite_redemptions::Column::UserId.eq(actor_user_id))
.one(&txn)
.await?;
if existing.is_none() {
organization_invite_redemptions::ActiveModel {
invite_id: Set(invite.id),
user_id: Set(actor_user_id),
organization_id: Set(invite.organization_id),
..Default::default()
}
.insert(&txn)
.await?;
}
}
let version = bump_version(&txn, invite.organization_id).await?;
self.emit_invite(
&txn,
invite.organization_id,
actor_user_id,
serde_json::json!({
"id": invite.id,
"action": "redeem",
"user_id": actor_user_id,
"joined": joined,
"version": version,
"request_id": request_id,
}),
request_id,
)
.await?;
txn.commit().await?;
self.invalidate_org(invite.organization_id).await?;
Ok((invite.organization_id, row.id, joined))
}
pub(crate) async fn default_role_row<C: ConnectionTrait>(
&self,
db: &C,
organization_id: Uuid,
) -> AuthResult<Option<organization_roles::Model>> {
Ok(organization_roles::Entity::find()
.filter(organization_roles::Column::OrganizationId.eq(organization_id))
.filter(organization_roles::Column::DefaultRole.eq(true))
.order_by_asc(organization_roles::Column::Rank)
.one(db)
.await?)
}
/// 邀请事件与业务写入同事务入队：提交时一起可见，回滚时一起消失。
pub(crate) async fn emit_invite(
&self,
txn: &sea_orm::DatabaseTransaction,
organization_id: Uuid,
actor_user_id: i64,
payload: serde_json::Value,
_request_id: Option<&str>,
) -> AuthResult<()> {
let labels = crate::events::Labels::new()
.with(label::ORGANIZATION_ID, organization_id)
.with(label::ACTOR_USER_ID, actor_user_id);
self.emit_tx(
txn,
label::ORGANIZATION_INVITE_CREATE,
labels,
payload,
None,
)
.await
}
}
pub(crate) fn invite_summary(row: &organization_invites::Model) -> InviteSummary {
InviteSummary {
id: row.id,
organization_id: row.organization_id,
method: row.method,
role_id: row.role_id,
target_user_id: row.target_user_id,
created_by: row.created_by,
max_uses: row.max_uses,
uses: row.uses,
expires_at: row.expires_at,
revoked_at: row.revoked_at,
created_at: row.created_at,
email: row.email.clone(),
}
}