538d47b
use chrono::Utc;
use sea_orm::entity::Set;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{
ActiveModelTrait, ColumnTrait, Condition, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
QuerySelect, TransactionTrait,
};
use uuid::Uuid;
use crate::AuthSvc;
use crate::errors::{AuthError, AuthResult};
use crate::events::{Labels, label};
use crate::models::space::{
OrgMemberRole, OrganizationStatus, PermissionEffect, organization_invite_redemptions,
organization_invites, organization_member_permissions, organization_member_roles,
organization_members, organization_role_permissions, organization_roles, organizations,
};
use crate::models::users::users;
use crate::organization::types::{
CreateOrganizationBody, MAX_ORG_DESCRIPTION_LEN, MAX_ORG_NAME_LEN, MAX_PERMISSION_LEN,
MAX_ROLE_KEY_LEN, MAX_ROLE_NAME_LEN, MAX_SLUG_LEN, MemberSummary, OrganizationSummary,
PermissionGrant, RoleSummary, SYSTEM_ROLE_ADMIN, SYSTEM_ROLE_OWNER, UpdateOrganizationBody,
is_known_permission,
};
use crate::organization::{error, member, permission, role};
pub(crate) fn perm_cache_prefix(organization_id: Uuid, user_id: i64) -> String {
format!("auth:org:perm:{organization_id}:{user_id}")
}
impl AuthSvc {
/// 组织写操作的统一收尾：推进 version，然后清掉本组织全部派生缓存。
/// version 同时是权限缓存的版本键与乐观锁基准，必须与写操作同事务推进。
pub(crate) async fn invalidate_org(&self, organization_id: Uuid) -> AuthResult<()> {
let members = organization_members::Entity::find()
.select_only()
.column(organization_members::Column::UserId)
.filter(organization_members::Column::OrganizationId.eq(organization_id))
.into_tuple::<i64>()
.all(self.db().read())
.await?;
self.drop_org_cache(organization_id, members).await
}
/// 成员行已随事务删除时用这一版：user_id 由调用方在事务内收集后传入。
pub(crate) async fn drop_org_cache(
&self,
organization_id: Uuid,
members: Vec<i64>,
) -> AuthResult<()> {
let cache = self.cache();
for key in [
format!("auth:org:{organization_id}"),
format!("auth:org:summary:{organization_id}"),
] {
if let Err(error) = cache.delete(&key).await {
tracing::warn!(%error, %organization_id, "failed to drop the organization cache");
}
}
for user_id in members {
if let Err(error) = cache
.delete(&perm_cache_prefix(organization_id, user_id))
.await
{
tracing::warn!(%error, %organization_id, user_id, "failed to drop the permission cache");
}
}
Ok(())
}
pub(crate) async fn peek_org_version(&self, organization_id: Uuid) -> AuthResult<Option<i32>> {
Ok(organizations::Entity::find_by_id(organization_id)
.select_only()
.column(organizations::Column::Version)
.into_tuple::<i32>()
.one(self.db().read())
.await?)
}
/// `expected_version <= 0` 表示调用方不做乐观锁校验。
pub(crate) async fn check_version(
&self,
organization_id: Uuid,
expected: i32,
) -> AuthResult<()> {
if expected <= 0 {
return Ok(());
}
let current = self
.peek_org_version(organization_id)
.await?
.ok_or(AuthError::OrganizationNotFound)?;
if current != expected {
return Err(AuthError::VersionMismatch { current });
}
Ok(())
}
pub(crate) async fn is_platform_admin(&self, user_id: i64) -> AuthResult<bool> {
Ok(users::Entity::find_by_id(user_id)
.select_only()
.column(users::Column::IsAdmin)
.into_tuple::<bool>()
.one(self.db().read())
.await?
.unwrap_or(false))
}
/// 唯一权限闸门。所有组织内写操作先过这里，语义与 CheckPermission 完全一致。
pub(crate) async fn require_permission(
&self,
user_id: i64,
organization_id: Uuid,
permission: &str,
request_id: Option<&str>,
) -> AuthResult<()> {
let outcome =
permission::resolve(self, organization_id, user_id, permission, request_id).await?;
if outcome.allowed {
return Ok(());
}
Err(error::reason_error(outcome.reason, permission))
}
/// 授权上限：只能授出自己当前确实持有的权限，不能借新建/改角色提权。
pub(crate) async fn require_grantable(
&self,
user_id: i64,
organization_id: Uuid,
grants: &[PermissionGrant],
request_id: Option<&str>,
) -> AuthResult<()> {
for grant in grants {
if grant.effect == PermissionEffect::Allow {
self.require_permission(user_id, organization_id, &grant.permission, request_id)
.await?;
}
}
Ok(())
}
/// 组织事件在业务事务**内部**入队：业务写入与事件行同生共死。
///
/// 写不进 outbox 就让整个事务回滚（`AuthError` 向上传播），这是刻意选择：
/// 宁可这次写操作失败重试，也不要出现「组织改成了、消费方永远不知道」。
async fn emit_org<T: serde::Serialize>(
&self,
txn: &sea_orm::DatabaseTransaction,
kind: &str,
organization_id: Uuid,
actor_user_id: i64,
payload: T,
request_id: Option<&str>,
) -> AuthResult<()> {
let labels = Labels::new()
.with(label::ORGANIZATION_ID, organization_id)
.with(label::ACTOR_USER_ID, actor_user_id);
self.emit_tx(txn, kind, labels, payload, request_id).await
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, request_id = %request_id.unwrap_or_default()))]
pub async fn create_organization(
&self,
actor_user_id: i64,
body: CreateOrganizationBody,
request_id: Option<&str>,
) -> AuthResult<OrganizationSummary> {
let slug = normalize_slug(&body.slug)?;
let name = normalize_name(&body.name)?;
let description = normalize_description(body.description.as_deref())?;
let avatar_url = normalize_avatar(body.avatar_url.as_deref())?;
users::Entity::find_by_id(actor_user_id)
.filter(users::Column::DeletedAt.is_null())
.one(self.db().read())
.await?
.ok_or(AuthError::UserNotFound)?;
if organizations::Entity::find()
.filter(organizations::Column::Slug.eq(slug.clone()))
.one(self.db().read())
.await?
.is_some()
{
return Err(AuthError::OrganizationSlugTaken);
}
let welcome_message = match body.welcome_message.as_deref() {
Some(value) if value.trim().is_empty() => None,
Some(value) => normalize_description(Some(value))?,
None => None,
};
let allow_join_level = valid_join_level(body.allow_join_level.unwrap_or(0))?;
let txn = self.db().write().begin().await?;
let organization = organizations::ActiveModel {
slug: Set(slug),
name: Set(name),
description: Set(description),
avatar_url: Set(avatar_url),
status: Set(OrganizationStatus::Active),
owner_user_id: Set(actor_user_id),
join_policy: Set(body
.join_policy
.unwrap_or(crate::models::space::OrganizationJoinPolicy::InviteOnly)),
allow_join: Set(body.allow_join.unwrap_or(false)),
require_email_verified: Set(body.require_email_verified.unwrap_or(true)),
require_mfa: Set(body.require_mfa.unwrap_or(false)),
welcome_message: Set(welcome_message),
member_must_enable_2fa: Set(body.member_must_enable_2fa.unwrap_or(false)),
allow_invite: Set(body.allow_invite.unwrap_or(true)),
allow_join_level: Set(allow_join_level),
..Default::default()
}
.insert(&txn)
.await
.map_err(error::map_insert_conflict)?;
let owner_role = role::seed_system_roles(&txn, organization.id).await?;
let membership = member::insert_owner(&txn, organization.id, actor_user_id).await?;
role::bind_role(&txn, membership.id, owner_role, actor_user_id).await?;
self.emit_org(
&txn,
label::ORGANIZATION_LIFE_CYCLE_CREATE,
organization.id,
actor_user_id,
serde_json::json!({
"id": organization.id,
"slug": &organization.slug,
"name": &organization.name,
"request_id": request_id,
}),
request_id,
)
.await?;
txn.commit().await?;
let summary = self
.organization_summary(&organization, Some(actor_user_id))
.await?;
Ok(summary)
}
#[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 get_organization(
&self,
actor_user_id: i64,
organization_id: Option<Uuid>,
slug: Option<&str>,
request_id: Option<&str>,
) -> AuthResult<(OrganizationSummary, Option<MemberSummary>)> {
let organization = match (
organization_id,
slug.map(str::trim).filter(|value| !value.is_empty()),
) {
(Some(id), _) => {
organizations::Entity::find_by_id(id)
.filter(organizations::Column::DeletedAt.is_null())
.one(self.db().read())
.await?
}
(None, Some(slug)) => {
organizations::Entity::find()
.filter(organizations::Column::Slug.eq(slug.to_owned()))
.filter(organizations::Column::DeletedAt.is_null())
.one(self.db().read())
.await?
}
(None, None) => None,
}
.ok_or(AuthError::OrganizationNotFound)?;
let summary = self
.organization_summary(&organization, Some(actor_user_id))
.await?;
// 非成员只看公开字段：my_role 为空即可，不额外补一层 DTO。
let detail =
match member::find_any(self.db().read(), organization.id, actor_user_id).await? {
Some(row) => Some(self.member_summary(&row).await?),
None => None,
};
Ok((summary, detail))
}
pub(crate) async fn organization_summary(
&self,
organization: &organizations::Model,
actor_user_id: Option<i64>,
) -> AuthResult<OrganizationSummary> {
let member_count = organization_members::Entity::find()
.filter(organization_members::Column::OrganizationId.eq(organization.id))
.filter(organization_members::Column::LeftAt.is_null())
.filter(organization_members::Column::BannedAt.is_null())
.count(self.db().read())
.await? as i64;
let my_role = match actor_user_id {
Some(user_id) => member::find_active(self.db().read(), organization.id, user_id)
.await?
.map(|row| row.role),
None => None,
};
Ok(OrganizationSummary {
id: organization.id,
slug: organization.slug.clone(),
name: organization.name.clone(),
description: organization.description.clone(),
avatar_url: organization.avatar_url.clone(),
status: organization.status,
owner_user_id: organization.owner_user_id,
member_count,
version: organization.version,
created_at: organization.created_at,
updated_at: organization.updated_at,
archived_at: organization.archived_at,
my_role,
join_policy: organization.join_policy,
allow_join: organization.allow_join,
require_email_verified: organization.require_email_verified,
require_mfa: organization.require_mfa,
welcome_message: organization.welcome_message.clone(),
member_must_enable_2fa: organization.member_must_enable_2fa,
allow_invite: organization.allow_invite,
allow_join_level: organization.allow_join_level,
})
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, user_id = user_id, request_id = %request_id.unwrap_or_default()))]
pub async fn list_organizations(
&self,
actor_user_id: i64,
user_id: i64,
include_archived: bool,
limit: u64,
offset: u64,
request_id: Option<&str>,
) -> AuthResult<Vec<OrganizationSummary>> {
let target = if user_id == 0 { actor_user_id } else { user_id };
if target != actor_user_id && !self.is_platform_admin(actor_user_id).await? {
return Err(AuthError::Forbidden);
}
let ids = organization_members::Entity::find()
.select_only()
.column(organization_members::Column::OrganizationId)
.filter(organization_members::Column::UserId.eq(target))
.filter(organization_members::Column::LeftAt.is_null())
.filter(organization_members::Column::BannedAt.is_null())
.into_tuple::<Uuid>()
.all(self.db().read())
.await?;
if ids.is_empty() {
return Ok(Vec::new());
}
let mut query = organizations::Entity::find()
.filter(organizations::Column::DeletedAt.is_null())
.filter(organizations::Column::Id.is_in(ids));
if !include_archived {
query = query.filter(organizations::Column::Status.ne(OrganizationStatus::Archived));
}
let rows = query
.order_by_desc(organizations::Column::UpdatedAt)
.limit(limit)
.offset(offset)
.all(self.db().read())
.await?;
let mut out = Vec::with_capacity(rows.len());
for row in rows {
out.push(self.organization_summary(&row, Some(target)).await?);
}
Ok(out)
}
/// 公开目录：ACTIVE 且非 PRIVATE 的组织。不要求成员身份。
pub async fn list_public_organizations(
&self,
query: &str,
limit: u64,
offset: u64,
) -> AuthResult<Vec<OrganizationSummary>> {
let mut select = organizations::Entity::find()
.filter(organizations::Column::DeletedAt.is_null())
.filter(organizations::Column::Status.eq(OrganizationStatus::Active))
.filter(
organizations::Column::JoinPolicy
.ne(crate::models::space::OrganizationJoinPolicy::Private),
);
let needle = query.trim();
if !needle.is_empty() {
select = select.filter(
Condition::any()
.add(organizations::Column::Slug.contains(needle))
.add(organizations::Column::Name.contains(needle)),
);
}
let rows = select
.order_by_desc(organizations::Column::CreatedAt)
.limit(limit)
.offset(offset)
.all(self.db().read())
.await?;
let mut out = Vec::with_capacity(rows.len());
for row in rows {
out.push(self.organization_summary(&row, None).await?);
}
Ok(out)
}
#[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 update_organization(
&self,
actor_user_id: i64,
organization_id: Uuid,
body: UpdateOrganizationBody,
request_id: Option<&str>,
) -> AuthResult<OrganizationSummary> {
self.require_permission(actor_user_id, organization_id, "org.update", request_id)
.await?;
self.check_version(organization_id, body.expected_version)
.await?;
let name = body.name.as_deref().map(normalize_name).transpose()?;
let description = match body.description.as_deref() {
Some(value) => Some(normalize_description(Some(value))?),
None => None,
};
let avatar_url = match body.avatar_url.as_deref() {
Some(value) => Some(normalize_avatar(Some(value))?),
None => None,
};
let txn = self.db().write().begin().await?;
let current = organizations::Entity::find_by_id(organization_id)
.filter(organizations::Column::DeletedAt.is_null())
.one(&txn)
.await?
.ok_or(AuthError::OrganizationNotFound)?;
if body.expected_version > 0 && current.version != body.expected_version {
return Err(AuthError::VersionMismatch {
current: current.version,
});
}
let mut active: organizations::ActiveModel = current.into();
if let Some(value) = name {
active.name = Set(value);
}
if let Some(value) = description {
active.description = Set(value);
}
if let Some(value) = avatar_url {
active.avatar_url = Set(value);
}
if let Some(value) = body.join_policy {
active.join_policy = Set(value);
}
if let Some(value) = body.allow_join {
active.allow_join = Set(value);
}
if let Some(value) = body.require_email_verified {
active.require_email_verified = Set(value);
}
if let Some(value) = body.require_mfa {
active.require_mfa = Set(value);
}
if let Some(value) = body.welcome_message.clone() {
let trimmed = value.trim();
active.welcome_message = Set(if trimmed.is_empty() {
None
} else {
Some(trimmed.to_string())
});
}
if let Some(value) = body.member_must_enable_2fa {
active.member_must_enable_2fa = Set(value);
}
if let Some(value) = body.allow_invite {
active.allow_invite = Set(value);
}
if let Some(value) = body.allow_join_level {
active.allow_join_level = Set(valid_join_level(value)?);
}
let updated = active.update(&txn).await?;
let version = bump_version(&txn, organization_id).await?;
self.emit_org(
&txn,
label::ORGANIZATION_LIFE_CYCLE_UPDATE,
organization_id,
actor_user_id,
serde_json::json!({
"id": organization_id,
"name": &updated.name,
"request_id": request_id,
}),
request_id,
)
.await?;
txn.commit().await?;
self.invalidate_org(organization_id).await?;
let refreshed = self.load_organization(organization_id).await?;
let mut summary = self
.organization_summary(&refreshed, Some(actor_user_id))
.await?;
summary.version = version;
Ok(summary)
}
#[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 archive_organization(
&self,
actor_user_id: i64,
organization_id: Uuid,
reason: Option<&str>,
request_id: Option<&str>,
) -> AuthResult<OrganizationSummary> {
self.require_permission(actor_user_id, organization_id, "org.archive", request_id)
.await?;
let reason = normalize_reason(reason)?;
let txn = self.db().write().begin().await?;
organizations::Entity::update_many()
.col_expr(
organizations::Column::Status,
Expr::value(OrganizationStatus::Archived),
)
.col_expr(organizations::Column::ArchivedAt, Expr::value(Utc::now()))
.col_expr(organizations::Column::SuspendedReason, Expr::value(reason))
.filter(organizations::Column::Id.eq(organization_id))
.filter(organizations::Column::DeletedAt.is_null())
.exec(&txn)
.await?;
let version = bump_version(&txn, organization_id).await?;
self.emit_org(
&txn,
label::ORGANIZATION_LIFE_CYCLE_UPDATE,
organization_id,
actor_user_id,
serde_json::json!({
"id": organization_id,
"action": "archive",
"request_id": request_id,
}),
request_id,
)
.await?;
txn.commit().await?;
self.invalidate_org(organization_id).await?;
let refreshed = self.load_organization(organization_id).await?;
let mut summary = self
.organization_summary(&refreshed, Some(actor_user_id))
.await?;
summary.version = version;
Ok(summary)
}
#[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 restore_organization(
&self,
actor_user_id: i64,
organization_id: Uuid,
request_id: Option<&str>,
) -> AuthResult<OrganizationSummary> {
// 归档组织的权限判定一律不通过，所以这里不能用 require_permission：
// 恢复动作由 OWNER 或平台管理员执行。
let organization = self.load_organization(organization_id).await?;
if organization.owner_user_id != actor_user_id
&& !self.is_platform_admin(actor_user_id).await?
{
return Err(AuthError::Forbidden);
}
let txn = self.db().write().begin().await?;
organizations::Entity::update_many()
.col_expr(
organizations::Column::Status,
Expr::value(OrganizationStatus::Active),
)
.col_expr(
organizations::Column::ArchivedAt,
Expr::value(None::<chrono::DateTime<Utc>>),
)
.col_expr(
organizations::Column::SuspendedReason,
Expr::value(None::<String>),
)
.filter(organizations::Column::Id.eq(organization_id))
.exec(&txn)
.await?;
let version = bump_version(&txn, organization_id).await?;
self.emit_org(
&txn,
label::ORGANIZATION_LIFE_CYCLE_UPDATE,
organization_id,
actor_user_id,
serde_json::json!({
"id": organization_id,
"action": "restore",
"request_id": request_id,
}),
request_id,
)
.await?;
txn.commit().await?;
self.invalidate_org(organization_id).await?;
let refreshed = self.load_organization(organization_id).await?;
let mut summary = self
.organization_summary(&refreshed, Some(actor_user_id))
.await?;
summary.version = version;
Ok(summary)
}
/// 删除分两种语义：`force = false` 时仍有活跃成员会被拦下，
/// `force = true` 时才连成员与角色一并清理。软删标记写在 `deleted_at` 上，
/// 但权限缓存与派生数据必须真的清掉，否则残留的授予会在重建同名组织时复活。
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, force = force, request_id = %request_id.unwrap_or_default()))]
pub async fn delete_organization(
&self,
actor_user_id: i64,
organization_id: Uuid,
force: bool,
reason: Option<&str>,
request_id: Option<&str>,
) -> AuthResult<bool> {
let organization = self.load_organization(organization_id).await?;
if organization.owner_user_id != actor_user_id
&& !self.is_platform_admin(actor_user_id).await?
{
return Err(AuthError::Forbidden);
}
let reason = normalize_reason(reason)?;
if !force {
let active = organization_members::Entity::find()
.filter(organization_members::Column::OrganizationId.eq(organization_id))
.filter(organization_members::Column::LeftAt.is_null())
.filter(organization_members::Column::BannedAt.is_null())
.count(self.db().read())
.await?;
if active > 0 {
return Err(AuthError::OrganizationHasMembers);
}
}
let txn = self.db().write().begin().await?;
let member_ids = organization_members::Entity::find()
.select_only()
.column(organization_members::Column::Id)
.filter(organization_members::Column::OrganizationId.eq(organization_id))
.into_tuple::<Uuid>()
.all(&txn)
.await?;
// 权限缓存键依赖 user_id，而成员行在本事务里会被物理删除，所以先把 user_id 取出来。
let member_user_ids = organization_members::Entity::find()
.select_only()
.column(organization_members::Column::UserId)
.filter(organization_members::Column::OrganizationId.eq(organization_id))
.into_tuple::<i64>()
.all(&txn)
.await?;
if !member_ids.is_empty() {
organization_member_permissions::Entity::delete_many()
.filter(organization_member_permissions::Column::MemberId.is_in(member_ids.clone()))
.exec(&txn)
.await?;
organization_member_roles::Entity::delete_many()
.filter(organization_member_roles::Column::MemberId.is_in(member_ids))
.exec(&txn)
.await?;
organization_members::Entity::delete_many()
.filter(organization_members::Column::OrganizationId.eq(organization_id))
.exec(&txn)
.await?;
}
organization_invite_redemptions::Entity::delete_many()
.filter(organization_invite_redemptions::Column::OrganizationId.eq(organization_id))
.exec(&txn)
.await?;
organization_invites::Entity::delete_many()
.filter(organization_invites::Column::OrganizationId.eq(organization_id))
.exec(&txn)
.await?;
let role_ids = organization_roles::Entity::find()
.select_only()
.column(organization_roles::Column::Id)
.filter(organization_roles::Column::OrganizationId.eq(organization_id))
.into_tuple::<Uuid>()
.all(&txn)
.await?;
if !role_ids.is_empty() {
organization_role_permissions::Entity::delete_many()
.filter(organization_role_permissions::Column::RoleId.is_in(role_ids))
.exec(&txn)
.await?;
}
organization_roles::Entity::delete_many()
.filter(organization_roles::Column::OrganizationId.eq(organization_id))
.exec(&txn)
.await?;
organizations::Entity::update_many()
.col_expr(organizations::Column::DeletedAt, Expr::value(Utc::now()))
.col_expr(
organizations::Column::SuspendedReason,
Expr::value(reason.clone()),
)
.filter(organizations::Column::Id.eq(organization_id))
.exec(&txn)
.await?;
bump_version(&txn, organization_id).await?;
self.emit_org(
&txn,
label::ORGANIZATION_LIFE_CYCLE_DELETE,
organization_id,
actor_user_id,
serde_json::json!({
"id": organization_id,
"force": force,
"reason": &reason,
"request_id": request_id,
}),
request_id,
)
.await?;
txn.commit().await?;
self.drop_org_cache(organization_id, member_user_ids)
.await?;
Ok(true)
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, target_user_id = target_user_id, request_id = %request_id.unwrap_or_default()))]
pub async fn transfer_ownership(
&self,
actor_user_id: i64,
organization_id: Uuid,
target_user_id: i64,
request_id: Option<&str>,
) -> AuthResult<OrganizationSummary> {
let organization = self.load_organization(organization_id).await?;
if organization.owner_user_id != actor_user_id
&& !self.is_platform_admin(actor_user_id).await?
{
return Err(AuthError::Forbidden);
}
if organization.status != OrganizationStatus::Active {
return Err(AuthError::OrganizationInactive);
}
if target_user_id == organization.owner_user_id {
return Err(AuthError::OrganizationMemberExists);
}
let target = member::find_active(self.db().read(), organization_id, target_user_id)
.await?
.ok_or(AuthError::OrganizationMemberNotFound)?;
let owner_role = role::find_by_key(organization_id, SYSTEM_ROLE_OWNER, self.db().read())
.await?
.ok_or_else(|| AuthError::internal("owner role is missing"))?;
let admin_role = role::find_by_key(organization_id, SYSTEM_ROLE_ADMIN, self.db().read())
.await?
.ok_or_else(|| AuthError::internal("admin role is missing"))?;
let txn = self.db().write().begin().await?;
// 唯一 OWNER 的不变式落在两处：organizations.owner_user_id 与成员行的粗粒度角色。
// 两个改写放在同一事务里，读到中间态的可能性为零。
if let Some(previous) =
member::find_any(&txn, organization_id, organization.owner_user_id).await?
{
let mut active: organization_members::ActiveModel = previous.clone().into();
active.role = Set(OrgMemberRole::Admin);
active.update(&txn).await?;
role::unbind_role(&txn, previous.id, owner_role.id).await?;
role::bind_role(&txn, previous.id, admin_role.id, actor_user_id).await?;
}
let mut active: organization_members::ActiveModel = target.clone().into();
active.role = Set(OrgMemberRole::Owner);
active.update(&txn).await?;
role::unbind_role(&txn, target.id, admin_role.id).await?;
role::bind_role(&txn, target.id, owner_role.id, actor_user_id).await?;
organizations::Entity::update_many()
.col_expr(
organizations::Column::OwnerUserId,
Expr::value(target_user_id),
)
.filter(organizations::Column::Id.eq(organization_id))
.exec(&txn)
.await?;
let version = bump_version(&txn, organization_id).await?;
self.emit_org(
&txn,
label::ORGANIZATION_MEMBER_UPDATE,
organization_id,
actor_user_id,
serde_json::json!({
"id": organization_id,
"action": "transfer_ownership",
"previous_owner": organization.owner_user_id,
"new_owner": target_user_id,
"request_id": request_id,
}),
request_id,
)
.await?;
txn.commit().await?;
self.invalidate_org(organization_id).await?;
let refreshed = self.load_organization(organization_id).await?;
let mut summary = self
.organization_summary(&refreshed, Some(actor_user_id))
.await?;
summary.version = version;
Ok(summary)
}
pub(crate) async fn load_organization(
&self,
organization_id: Uuid,
) -> AuthResult<organizations::Model> {
organizations::Entity::find_by_id(organization_id)
.filter(organizations::Column::DeletedAt.is_null())
.one(self.db().read())
.await?
.ok_or(AuthError::OrganizationNotFound)
}
pub(crate) async fn member_summary(
&self,
row: &organization_members::Model,
) -> AuthResult<MemberSummary> {
let user = users::Entity::find_by_id(row.user_id)
.one(self.db().read())
.await?;
Ok(MemberSummary {
id: row.id,
organization_id: row.organization_id,
user_id: row.user_id,
username: user
.as_ref()
.map(|u| u.username.clone())
.unwrap_or_default(),
displayname: user.and_then(|u| u.displayname),
role: row.role,
role_keys: role::role_keys_of(self.db().read(), row.id).await?,
permissions: permission::effective(self.db().read(), row).await?,
joined_at: row.joined_at,
left_at: row.left_at,
banned_at: row.banned_at,
invited_by: row.invited_by,
join_method: row.join_method,
alias_name: row.alias_name.clone(),
alias_avatar_url: row.alias_avatar_url.clone(),
message_notifications: row.message_notifications,
muted: row.muted,
mute_until: row.mute_until,
mobile_push: row.mobile_push,
suppress_everyone: row.suppress_everyone,
suppress_roles: row.suppress_roles,
hide_muted_channels: row.hide_muted_channels,
ban_reason: row.ban_reason.clone(),
ban_expires_at: row.ban_expires_at,
ban_allow_agent_join: row.ban_allow_agent_join,
ban_allow_speak: row.ban_allow_speak,
ban_allow_push_repo: row.ban_allow_push_repo,
ban_allow_comment_any: row.ban_allow_comment_any,
version: row.version,
last_active_at: row.last_active_at,
})
}
pub(crate) async fn role_summary(
&self,
row: &organization_roles::Model,
) -> AuthResult<RoleSummary> {
let permissions = role::grants_of(self.db().read(), row.id).await?;
let member_count = organization_member_roles::Entity::find()
.filter(organization_member_roles::Column::RoleId.eq(row.id))
.count(self.db().read())
.await? as i64;
Ok(RoleSummary {
id: row.id,
organization_id: row.organization_id,
key: row.key.clone(),
name: row.name.clone(),
description: row.description.clone(),
rank: row.rank,
system: row.system,
default_role: row.default_role,
self_assignable: row.self_assignable,
permissions,
member_count,
created_at: row.created_at,
updated_at: row.updated_at,
version: row.version,
color: row.color.clone(),
emoji: row.emoji.clone(),
icon_hash: row.icon_hash.clone(),
position: row.position,
})
}
pub(crate) async fn default_role_id(&self, organization_id: Uuid) -> AuthResult<Option<Uuid>> {
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(self.db().read())
.await?
.map(|row| row.id))
}
pub(crate) fn validate_grants(grants: &[PermissionGrant]) -> AuthResult<Vec<PermissionGrant>> {
let mut out = Vec::with_capacity(grants.len());
for grant in grants {
let permission = grant.permission.trim();
if permission.is_empty()
|| permission.len() > MAX_PERMISSION_LEN
|| !is_known_permission(permission)
{
return Err(AuthError::InvalidPermission {
permission: grant.permission.clone(),
});
}
out.push(PermissionGrant {
permission: permission.to_owned(),
effect: match grant.effect {
PermissionEffect::Deny => PermissionEffect::Deny,
_ => PermissionEffect::Allow,
},
});
}
Ok(out)
}
}
/// 在写事务内推进 version 并回读新值。
pub(crate) async fn bump_version<C>(db: &C, organization_id: Uuid) -> AuthResult<i32>
where
C: sea_orm::ConnectionTrait,
{
let bumped = organizations::Entity::update_many()
.col_expr(
organizations::Column::Version,
Expr::col(organizations::Column::Version).add(1),
)
.col_expr(organizations::Column::UpdatedAt, Expr::value(Utc::now()))
.filter(organizations::Column::Id.eq(organization_id))
.exec(db)
.await?;
if bumped.rows_affected == 0 {
return Err(AuthError::OrganizationNotFound);
}
organizations::Entity::find_by_id(organization_id)
.select_only()
.column(organizations::Column::Version)
.into_tuple::<i32>()
.one(db)
.await?
.ok_or(AuthError::OrganizationNotFound)
}
pub(crate) fn normalize_slug(raw: &str) -> AuthResult<String> {
let slug = raw.trim().to_lowercase();
if slug.is_empty()
|| slug.len() > MAX_SLUG_LEN
|| !slug
.chars()
.all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '-')
|| slug.starts_with('-')
|| slug.ends_with('-')
{
return Err(AuthError::MalformedRequest);
}
Ok(slug)
}
pub(crate) fn normalize_name(raw: &str) -> AuthResult<String> {
let name = raw.trim();
if name.is_empty() || name.len() > MAX_ORG_NAME_LEN {
return Err(AuthError::MalformedRequest);
}
Ok(name.to_owned())
}
pub(crate) fn normalize_description(raw: Option<&str>) -> AuthResult<Option<String>> {
match raw.map(str::trim) {
None | Some("") => Ok(None),
Some(value) if value.len() > MAX_ORG_DESCRIPTION_LEN => Err(AuthError::MalformedRequest),
Some(value) => Ok(Some(value.to_owned())),
}
}
pub(crate) fn normalize_avatar(raw: Option<&str>) -> AuthResult<Option<String>> {
match raw.map(str::trim) {
None | Some("") => Ok(None),
Some(value) if value.len() > 512 || !value.starts_with("https://") => {
Err(AuthError::MalformedRequest)
}
Some(value) => Ok(Some(value.to_owned())),
}
}
pub(crate) fn normalize_reason(raw: Option<&str>) -> AuthResult<Option<String>> {
match raw.map(str::trim) {
None | Some("") => Ok(None),
Some(value) if value.len() > 512 => Err(AuthError::MalformedRequest),
Some(value) => Ok(Some(value.to_owned())),
}
}
pub(crate) fn normalize_role_key(raw: &str) -> AuthResult<String> {
let key = raw.trim().to_lowercase();
if key.is_empty()
|| key.len() > MAX_ROLE_KEY_LEN
|| !key
.chars()
.all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '_' || ch == '-')
{
return Err(AuthError::MalformedRequest);
}
Ok(key)
}
pub(crate) fn normalize_role_name(raw: &str) -> AuthResult<String> {
let name = raw.trim();
if name.is_empty() || name.len() > MAX_ROLE_NAME_LEN {
return Err(AuthError::MalformedRequest);
}
Ok(name.to_owned())
}
pub(crate) fn valid_join_level(value: i16) -> AuthResult<i16> {
(0..=4)
.contains(&value)
.then_some(value)
.ok_or(AuthError::MalformedRequest)
}