02d8172
use crate::errors::{SpaceError, SpaceResult};
use crate::services::{error, member, permission, role};
use cache::Cache;
use chrono::Utc;
use config::AppConfig;
use config::auth::AuthConfig;
use db::DbConn;
use events::{EventHandler, Labels, label};
use export::{AppState, Database};
use models::users::users;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{
ActiveModelTrait, ColumnTrait, Condition, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
QuerySelect, Set, TransactionTrait,
};
use serde::Serialize;
use std::sync::Arc;
use storage::BlobStore;
use time::OffsetDateTime;
use uuid::Uuid;
async fn actor_details(
conn: &impl sea_orm::ConnectionTrait,
labels: &Labels,
) -> Option<serde_json::Value> {
let uid = labels.get(label::USER_ID)?.parse::<i64>().ok()?;
let user = models::users::users::Entity::find_by_id(uid)
.one(conn)
.await
.ok()??;
Some(serde_json::json!({
"id": user.id,
"username": user.username,
"display_name": user.displayname,
"email": user.email,
"is_admin": user.is_admin,
"bot": user.bot,
}))
}
#[derive(Clone)]
pub struct SpaceService {
pub state: AppState,
db_conn: DbConn,
}
impl SpaceService {
pub fn new(state: AppState) -> Self {
let db_conn = DbConn::new(
state.database.primary().clone(),
state.database.replicas().to_vec(),
);
Self { state, db_conn }
}
pub fn db_conn(&self) -> &DbConn {
&self.db_conn
}
pub fn cache(&self) -> &Cache {
&self.state.cache
}
pub fn events(&self) -> Option<&EventHandler> {
self.state.events.as_ref()
}
pub async fn emit_tx<T>(
&self,
conn: &impl sea_orm::ConnectionTrait,
kind: &str,
labels: Labels,
payload: T,
request_id: Option<&str>,
) -> Result<(), SpaceError>
where
T: Serialize,
{
let trace_id = request_id
.map(str::to_owned)
.or_else(envelope::context::request_id)
.unwrap_or_default();
let mut payload = serde_json::to_value(&payload).unwrap_or(serde_json::Value::Null);
if let Some(actor) = actor_details(conn, &labels).await {
payload["actor"] = actor;
}
events::outbox::enqueue_at(
conn,
kind,
labels,
&payload,
&trace_id,
chrono::Utc::now(),
request_id,
)
.await?;
Ok(())
}
pub fn config(&self) -> &AppConfig {
&self.state.config
}
pub fn auth(&self) -> &AuthConfig {
&self.state.config.auth
}
pub async fn require_session(
&self,
session: &session::Session,
_request_id: Option<&str>,
) -> Result<i64, SpaceError> {
let user_id = session.user_id()?.ok_or(SpaceError::Unauthorized)?;
let version = session
.security_version()?
.ok_or(SpaceError::Unauthorized)?;
let id = session
.get::<String>("auth_session_id")?
.ok_or(SpaceError::Unauthorized)?;
let kenauth = self
.state
.kenauth
.as_deref()
.ok_or_else(|| SpaceError::Internal("kenoauth is not configured".to_owned()))?;
let _ = kenauth.get_session(user_id, version, &id).await?;
Ok(user_id)
}
pub fn db(&self) -> &Arc<dyn Database> {
&self.state.database
}
pub fn storage(&self) -> &dyn BlobStore {
self.state.storage.as_ref()
}
/// Language for one recipient's space mail.
pub(crate) async fn recipient_locale(&self, uid: Option<i64>) -> i18n::UiLocale {
let fallback = i18n::config_fallback(self.config().email.default_locale());
match uid {
Some(uid) => i18n::resolve_user_locale(self.db().read(), uid, fallback).await,
None => fallback,
}
}
}
impl SpaceService {
pub(crate) async fn grant_public_room_memberships(&self, space_id: Uuid, user_id: i64) {
let public_rooms = models::room::rooms::Entity::find()
.filter(models::room::rooms::Column::SpaceId.eq(space_id))
.filter(models::room::rooms::Column::IsPrivate.eq(false))
.filter(models::room::rooms::Column::IsActive.eq(true))
.all(self.db().read())
.await;
let Ok(public_rooms) = public_rooms else {
tracing::warn!(%space_id, %user_id, "public room lookup failed");
return;
};
for room in &public_rooms {
let now = OffsetDateTime::now_utc();
let rm = models::room::room_members::ActiveModel {
role: Set(models::room::room_members::RoomMemberRole::Member),
room_id: Set(room.id),
user_id: Set(user_id),
nick: Set(None),
joined_at: Set(now),
left_at: Set(None),
created_at: Set(now),
updated_at: Set(Some(now)),
};
if let Err(e) = rm.insert(self.db().write()).await {
tracing::warn!(error = %e, room_id = %room.id, %user_id, "failed to create room_members record");
}
}
}
pub async fn load_user_briefs(
&self,
uids: &[i64],
) -> Result<
std::collections::HashMap<i64, crate::services::dto::UserDto>,
crate::errors::SpaceError,
> {
Ok(export::users::load_user_briefs(&self.state, uids).await?)
}
}
use crate::services::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 models::space::{
PermissionEffect, SpaceMemberRole, SpaceStatus, space_invite_redemptions, space_invites,
space_member_permissions, space_member_roles, space_members, space_role_permissions,
space_roles, spaces,
};
pub(crate) fn perm_cache_prefix(organization_id: Uuid, user_id: i64) -> String {
format!("auth:org:perm:{organization_id}:{user_id}")
}
impl SpaceService {
/// 组织写操作的统一收尾：推进 version，然后清掉本组织全部派生缓存。
/// version 同时是权限缓存的版本键与乐观锁基准，必须与写操作同事务推进。
pub(crate) async fn invalidate_org(&self, organization_id: Uuid) -> SpaceResult<()> {
let members = space_members::Entity::find()
.select_only()
.column(space_members::Column::UserId)
.filter(space_members::Column::OrganizationId.eq(organization_id))
.into_tuple::<i64>()
.all(self.db_conn().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>,
) -> SpaceResult<()> {
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) -> SpaceResult<Option<i32>> {
Ok(spaces::Entity::find_by_id(organization_id)
.select_only()
.column(spaces::Column::Version)
.into_tuple::<i32>()
.one(self.db_conn().read())
.await?)
}
/// `expected_version <= 0` 表示调用方不做乐观锁校验。
pub(crate) async fn check_version(
&self,
organization_id: Uuid,
expected: i32,
) -> SpaceResult<()> {
if expected <= 0 {
return Ok(());
}
let current = self
.peek_org_version(organization_id)
.await?
.ok_or(SpaceError::SpaceNotFound)?;
if current != expected {
return Err(SpaceError::VersionMismatch { current });
}
Ok(())
}
pub(crate) async fn is_platform_admin(&self, user_id: i64) -> SpaceResult<bool> {
Ok(users::Entity::find_by_id(user_id)
.select_only()
.column(users::Column::IsAdmin)
.into_tuple::<bool>()
.one(self.db_conn().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>,
) -> SpaceResult<()> {
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>,
) -> SpaceResult<()> {
for grant in grants {
if grant.effect == PermissionEffect::Allow {
self.require_permission(user_id, organization_id, &grant.permission, request_id)
.await?;
}
}
Ok(())
}
/// 组织事件在业务事务**内部**入队：业务写入与事件行同生共死。
///
/// 写不进 outbox 就让整个事务回滚（`SpaceError` 向上传播），这是刻意选择：
/// 宁可这次写操作失败重试，也不要出现「组织改成了、消费方永远不知道」。
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>,
) -> SpaceResult<()> {
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_space(
&self,
actor_user_id: i64,
body: CreateOrganizationBody,
request_id: Option<&str>,
) -> SpaceResult<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_conn().read())
.await?
.ok_or(SpaceError::UserNotFound)?;
if spaces::Entity::find()
.filter(spaces::Column::Slug.eq(slug.clone()))
.one(self.db_conn().read())
.await?
.is_some()
{
return Err(SpaceError::SpaceSlugTaken);
}
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_conn().write().begin().await?;
let organization = spaces::ActiveModel {
slug: Set(slug),
name: Set(name),
description: Set(description),
avatar_url: Set(avatar_url),
status: Set(SpaceStatus::Active),
owner_user_id: Set(actor_user_id),
join_policy: Set(body
.join_policy
.unwrap_or(models::space::SpaceJoinPolicy::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_space(
&self,
actor_user_id: i64,
organization_id: Option<Uuid>,
slug: Option<&str>,
request_id: Option<&str>,
) -> SpaceResult<(OrganizationSummary, Option<MemberSummary>)> {
let organization = match (
organization_id,
slug.map(str::trim).filter(|value| !value.is_empty()),
) {
(Some(id), _) => {
spaces::Entity::find_by_id(id)
.filter(spaces::Column::DeletedAt.is_null())
.one(self.db_conn().read())
.await?
}
(None, Some(slug)) => {
spaces::Entity::find()
.filter(spaces::Column::Slug.eq(slug.to_owned()))
.filter(spaces::Column::DeletedAt.is_null())
.one(self.db_conn().read())
.await?
}
(None, None) => None,
}
.ok_or(SpaceError::SpaceNotFound)?;
let summary = self
.organization_summary(&organization, Some(actor_user_id))
.await?;
// 非成员只看公开字段：my_role 为空即可，不额外补一层 DTO。
let detail =
match member::find_any(self.db_conn().read(), organization.id, actor_user_id).await? {
Some(row) => Some(self.member_summary(&row).await?),
None => None,
};
Ok((summary, detail))
}
pub async fn space_id_by_name(
&self,
actor_user_id: i64,
space_name: &str,
) -> SpaceResult<Uuid> {
let slug = space_name.trim().to_lowercase();
let (summary, _) = self
.get_space(actor_user_id, None, Some(&slug), None)
.await?;
Ok(summary.id)
}
pub(crate) async fn organization_summary(
&self,
organization: &spaces::Model,
actor_user_id: Option<i64>,
) -> SpaceResult<OrganizationSummary> {
let member_count = space_members::Entity::find()
.filter(space_members::Column::OrganizationId.eq(organization.id))
.filter(space_members::Column::LeftAt.is_null())
.filter(space_members::Column::BannedAt.is_null())
.count(self.db_conn().read())
.await? as i64;
let my_role = match actor_user_id {
Some(user_id) => member::find_active(self.db_conn().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_spaces(
&self,
actor_user_id: i64,
user_id: i64,
include_archived: bool,
limit: u64,
offset: u64,
request_id: Option<&str>,
) -> SpaceResult<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(SpaceError::Forbidden);
}
let ids = space_members::Entity::find()
.select_only()
.column(space_members::Column::OrganizationId)
.filter(space_members::Column::UserId.eq(target))
.filter(space_members::Column::LeftAt.is_null())
.filter(space_members::Column::BannedAt.is_null())
.into_tuple::<Uuid>()
.all(self.db_conn().read())
.await?;
if ids.is_empty() {
return Ok(Vec::new());
}
let mut query = spaces::Entity::find()
.filter(spaces::Column::DeletedAt.is_null())
.filter(spaces::Column::Id.is_in(ids));
if !include_archived {
query = query.filter(spaces::Column::Status.ne(SpaceStatus::Archived));
}
let rows = query
.order_by_desc(spaces::Column::UpdatedAt)
.limit(limit)
.offset(offset)
.all(self.db_conn().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_spaces(
&self,
query: &str,
limit: u64,
offset: u64,
) -> SpaceResult<Vec<OrganizationSummary>> {
let mut select = spaces::Entity::find()
.filter(spaces::Column::DeletedAt.is_null())
.filter(spaces::Column::Status.eq(SpaceStatus::Active))
.filter(spaces::Column::JoinPolicy.ne(models::space::SpaceJoinPolicy::Private));
let needle = query.trim();
if !needle.is_empty() {
select = select.filter(
Condition::any()
.add(spaces::Column::Slug.contains(needle))
.add(spaces::Column::Name.contains(needle)),
);
}
let rows = select
.order_by_desc(spaces::Column::CreatedAt)
.limit(limit)
.offset(offset)
.all(self.db_conn().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_space(
&self,
actor_user_id: i64,
organization_id: Uuid,
body: UpdateOrganizationBody,
request_id: Option<&str>,
) -> SpaceResult<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_conn().write().begin().await?;
let current = spaces::Entity::find_by_id(organization_id)
.filter(spaces::Column::DeletedAt.is_null())
.one(&txn)
.await?
.ok_or(SpaceError::SpaceNotFound)?;
if body.expected_version > 0 && current.version != body.expected_version {
return Err(SpaceError::VersionMismatch {
current: current.version,
});
}
let mut active: spaces::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_space(
&self,
actor_user_id: i64,
organization_id: Uuid,
reason: Option<&str>,
request_id: Option<&str>,
) -> SpaceResult<OrganizationSummary> {
self.require_permission(actor_user_id, organization_id, "org.archive", request_id)
.await?;
let reason = normalize_reason(reason)?;
let txn = self.db_conn().write().begin().await?;
spaces::Entity::update_many()
.col_expr(spaces::Column::Status, Expr::value(SpaceStatus::Archived))
.col_expr(spaces::Column::ArchivedAt, Expr::value(Utc::now()))
.col_expr(spaces::Column::SuspendedReason, Expr::value(reason))
.filter(spaces::Column::Id.eq(organization_id))
.filter(spaces::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_space(
&self,
actor_user_id: i64,
organization_id: Uuid,
request_id: Option<&str>,
) -> SpaceResult<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(SpaceError::Forbidden);
}
let txn = self.db_conn().write().begin().await?;
spaces::Entity::update_many()
.col_expr(spaces::Column::Status, Expr::value(SpaceStatus::Active))
.col_expr(
spaces::Column::ArchivedAt,
Expr::value(None::<chrono::DateTime<Utc>>),
)
.col_expr(spaces::Column::SuspendedReason, Expr::value(None::<String>))
.filter(spaces::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_space(
&self,
actor_user_id: i64,
organization_id: Uuid,
force: bool,
reason: Option<&str>,
request_id: Option<&str>,
) -> SpaceResult<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(SpaceError::Forbidden);
}
let reason = normalize_reason(reason)?;
if !force {
let active = space_members::Entity::find()
.filter(space_members::Column::OrganizationId.eq(organization_id))
.filter(space_members::Column::LeftAt.is_null())
.filter(space_members::Column::BannedAt.is_null())
.count(self.db_conn().read())
.await?;
if active > 0 {
return Err(SpaceError::SpaceHasMembers);
}
}
let txn = self.db_conn().write().begin().await?;
let member_ids = space_members::Entity::find()
.select_only()
.column(space_members::Column::Id)
.filter(space_members::Column::OrganizationId.eq(organization_id))
.into_tuple::<Uuid>()
.all(&txn)
.await?;
// 权限缓存键依赖 user_id，而成员行在本事务里会被物理删除，所以先把 user_id 取出来。
let member_user_ids = space_members::Entity::find()
.select_only()
.column(space_members::Column::UserId)
.filter(space_members::Column::OrganizationId.eq(organization_id))
.into_tuple::<i64>()
.all(&txn)
.await?;
if !member_ids.is_empty() {
space_member_permissions::Entity::delete_many()
.filter(space_member_permissions::Column::MemberId.is_in(member_ids.clone()))
.exec(&txn)
.await?;
space_member_roles::Entity::delete_many()
.filter(space_member_roles::Column::MemberId.is_in(member_ids))
.exec(&txn)
.await?;
space_members::Entity::delete_many()
.filter(space_members::Column::OrganizationId.eq(organization_id))
.exec(&txn)
.await?;
}
space_invite_redemptions::Entity::delete_many()
.filter(space_invite_redemptions::Column::OrganizationId.eq(organization_id))
.exec(&txn)
.await?;
space_invites::Entity::delete_many()
.filter(space_invites::Column::OrganizationId.eq(organization_id))
.exec(&txn)
.await?;
let role_ids = space_roles::Entity::find()
.select_only()
.column(space_roles::Column::Id)
.filter(space_roles::Column::OrganizationId.eq(organization_id))
.into_tuple::<Uuid>()
.all(&txn)
.await?;
if !role_ids.is_empty() {
space_role_permissions::Entity::delete_many()
.filter(space_role_permissions::Column::RoleId.is_in(role_ids))
.exec(&txn)
.await?;
}
space_roles::Entity::delete_many()
.filter(space_roles::Column::OrganizationId.eq(organization_id))
.exec(&txn)
.await?;
spaces::Entity::update_many()
.col_expr(spaces::Column::DeletedAt, Expr::value(Utc::now()))
.col_expr(spaces::Column::SuspendedReason, Expr::value(reason.clone()))
.filter(spaces::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>,
) -> SpaceResult<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(SpaceError::Forbidden);
}
if organization.status != SpaceStatus::Active {
return Err(SpaceError::SpaceInactive);
}
if target_user_id == organization.owner_user_id {
return Err(SpaceError::SpaceMemberExists);
}
let target = member::find_active(self.db_conn().read(), organization_id, target_user_id)
.await?
.ok_or(SpaceError::SpaceMemberNotFound)?;
let owner_role =
role::find_by_key(organization_id, SYSTEM_ROLE_OWNER, self.db_conn().read())
.await?
.ok_or_else(|| SpaceError::internal("owner role is missing"))?;
let admin_role =
role::find_by_key(organization_id, SYSTEM_ROLE_ADMIN, self.db_conn().read())
.await?
.ok_or_else(|| SpaceError::internal("admin role is missing"))?;
let txn = self.db_conn().write().begin().await?;
// 唯一 OWNER 的不变式落在两处：spaces.owner_user_id 与成员行的粗粒度角色。
// 两个改写放在同一事务里，读到中间态的可能性为零。
if let Some(previous) =
member::find_any(&txn, organization_id, organization.owner_user_id).await?
{
let mut active: space_members::ActiveModel = previous.clone().into();
active.role = Set(SpaceMemberRole::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: space_members::ActiveModel = target.clone().into();
active.role = Set(SpaceMemberRole::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?;
spaces::Entity::update_many()
.col_expr(spaces::Column::OwnerUserId, Expr::value(target_user_id))
.filter(spaces::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 async fn load_organization(&self, organization_id: Uuid) -> SpaceResult<spaces::Model> {
spaces::Entity::find_by_id(organization_id)
.filter(spaces::Column::DeletedAt.is_null())
.one(self.db_conn().read())
.await?
.ok_or(SpaceError::SpaceNotFound)
}
pub async fn member_summary(&self, row: &space_members::Model) -> SpaceResult<MemberSummary> {
let user = users::Entity::find_by_id(row.user_id)
.one(self.db_conn().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_conn().read(), row.id).await?,
permissions: permission::effective(self.db_conn().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: &space_roles::Model) -> SpaceResult<RoleSummary> {
let permissions = role::grants_of(self.db_conn().read(), row.id).await?;
let member_count = space_member_roles::Entity::find()
.filter(space_member_roles::Column::RoleId.eq(row.id))
.count(self.db_conn().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) -> SpaceResult<Option<Uuid>> {
Ok(space_roles::Entity::find()
.filter(space_roles::Column::OrganizationId.eq(organization_id))
.filter(space_roles::Column::DefaultRole.eq(true))
.order_by_asc(space_roles::Column::Rank)
.one(self.db_conn().read())
.await?
.map(|row| row.id))
}
pub(crate) fn validate_grants(grants: &[PermissionGrant]) -> SpaceResult<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(SpaceError::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 async fn bump_version<C>(db: &C, organization_id: Uuid) -> SpaceResult<i32>
where
C: sea_orm::ConnectionTrait,
{
let bumped = spaces::Entity::update_many()
.col_expr(
spaces::Column::Version,
Expr::col(spaces::Column::Version).add(1),
)
.col_expr(spaces::Column::UpdatedAt, Expr::value(Utc::now()))
.filter(spaces::Column::Id.eq(organization_id))
.exec(db)
.await?;
if bumped.rows_affected == 0 {
return Err(SpaceError::SpaceNotFound);
}
spaces::Entity::find_by_id(organization_id)
.select_only()
.column(spaces::Column::Version)
.into_tuple::<i32>()
.one(db)
.await?
.ok_or(SpaceError::SpaceNotFound)
}
pub(crate) fn normalize_slug(raw: &str) -> SpaceResult<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(SpaceError::MalformedRequest);
}
Ok(slug)
}
pub(crate) fn normalize_name(raw: &str) -> SpaceResult<String> {
let name = raw.trim();
if name.is_empty() || name.len() > MAX_ORG_NAME_LEN {
return Err(SpaceError::MalformedRequest);
}
Ok(name.to_owned())
}
pub(crate) fn normalize_description(raw: Option<&str>) -> SpaceResult<Option<String>> {
match raw.map(str::trim) {
None | Some("") => Ok(None),
Some(value) if value.len() > MAX_ORG_DESCRIPTION_LEN => Err(SpaceError::MalformedRequest),
Some(value) => Ok(Some(value.to_owned())),
}
}
pub(crate) fn normalize_avatar(raw: Option<&str>) -> SpaceResult<Option<String>> {
match raw.map(str::trim) {
None | Some("") => Ok(None),
Some(value) if value.len() > 512 || !value.starts_with("https://") => {
Err(SpaceError::MalformedRequest)
}
Some(value) => Ok(Some(value.to_owned())),
}
}
pub(crate) fn normalize_reason(raw: Option<&str>) -> SpaceResult<Option<String>> {
match raw.map(str::trim) {
None | Some("") => Ok(None),
Some(value) if value.len() > 512 => Err(SpaceError::MalformedRequest),
Some(value) => Ok(Some(value.to_owned())),
}
}
pub(crate) fn normalize_role_key(raw: &str) -> SpaceResult<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(SpaceError::MalformedRequest);
}
Ok(key)
}
pub(crate) fn normalize_role_name(raw: &str) -> SpaceResult<String> {
let name = raw.trim();
if name.is_empty() || name.len() > MAX_ROLE_NAME_LEN {
return Err(SpaceError::MalformedRequest);
}
Ok(name.to_owned())
}
pub(crate) fn valid_join_level(value: i16) -> SpaceResult<i16> {
(0..=4)
.contains(&value)
.then_some(value)
.ok_or(SpaceError::MalformedRequest)
}