538d47b
use chrono::Utc;
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
QuerySelect, Set, TransactionTrait,
};
use uuid::Uuid;
use crate::models::space::{
OrgJoinMethod, OrgMemberRole, OrganizationJoinRequestStatus, organization_join_requests,
organization_join_rules, organization_members, organization_roles,
};
use crate::models::users::users;
use crate::organization::types::{
AddMemberBody, JoinRequestSummary, JoinRuleSummary, SYSTEM_ROLE_MEMBER, SYSTEM_ROLE_OWNER,
};
use crate::organization::{member, service};
use crate::{AuthError, AuthResult, AuthSvc};
fn rule_summary(row: &organization_join_rules::Model) -> JoinRuleSummary {
JoinRuleSummary {
id: row.id,
organization_id: row.organization_id,
role_id: row.role_id,
trigger: row.trigger.clone(),
matcher: row.matcher.clone(),
enabled: row.enabled,
created_at: row.created_at,
updated_at: row.updated_at,
}
}
impl AuthSvc {
async fn join_request_summary(
&self,
row: &organization_join_requests::Model,
) -> AuthResult<JoinRequestSummary> {
let user = users::Entity::find_by_id(row.user_id)
.one(self.db().read())
.await?;
Ok(JoinRequestSummary {
id: row.id,
organization_id: row.organization_id,
user_id: row.user_id,
username: user
.as_ref()
.map(|value| value.username.clone())
.unwrap_or_default(),
displayname: user.and_then(|value| value.displayname),
status: row.status,
note: row.note.clone(),
decided_by: row.decided_by,
decided_at: row.decided_at,
created_at: row.created_at,
updated_at: row.updated_at,
})
}
// 加入规则的可选字段本身就多，参数表与 HTTP 请求体一一对应，拆结构体只会增加噪音。
#[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_join_rule(
&self,
actor_user_id: i64,
organization_id: Uuid,
role_id: Uuid,
trigger: String,
matcher: String,
enabled: bool,
request_id: Option<&str>,
) -> AuthResult<JoinRuleSummary> {
self.require_permission(actor_user_id, organization_id, "role.update", request_id)
.await?;
let trigger = trigger.trim().to_string();
let matcher = matcher.trim().to_string();
if trigger.is_empty() || matcher.is_empty() {
return Err(AuthError::MalformedRequest);
}
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)?;
let row = organization_join_rules::ActiveModel {
organization_id: Set(organization_id),
role_id: Set(role.id),
trigger: Set(trigger),
matcher: Set(matcher),
enabled: Set(enabled),
..Default::default()
}
.insert(self.db().write())
.await?;
service::bump_version(self.db().write(), organization_id).await?;
Ok(rule_summary(&row))
}
// 同上：可选字段多，保持与请求体一致的参数顺序。
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, rule_id = %rule_id, request_id = %request_id.unwrap_or_default()))]
pub async fn update_join_rule(
&self,
actor_user_id: i64,
organization_id: Uuid,
rule_id: Uuid,
role_id: Option<Uuid>,
trigger: Option<String>,
matcher: Option<String>,
enabled: Option<bool>,
request_id: Option<&str>,
) -> AuthResult<JoinRuleSummary> {
self.require_permission(actor_user_id, organization_id, "role.update", request_id)
.await?;
let row = organization_join_rules::Entity::find_by_id(rule_id)
.filter(organization_join_rules::Column::OrganizationId.eq(organization_id))
.one(self.db().read())
.await?
.ok_or(AuthError::OrganizationNotFound)?;
let mut active: organization_join_rules::ActiveModel = row.into();
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)?;
active.role_id = Set(role.id);
}
if let Some(trigger) = trigger {
let trigger = trigger.trim().to_string();
if trigger.is_empty() {
return Err(AuthError::MalformedRequest);
}
active.trigger = Set(trigger);
}
if let Some(matcher) = matcher {
let matcher = matcher.trim().to_string();
if matcher.is_empty() {
return Err(AuthError::MalformedRequest);
}
active.matcher = Set(matcher);
}
if let Some(enabled) = enabled {
active.enabled = Set(enabled);
}
let updated = active.update(self.db().write()).await?;
service::bump_version(self.db().write(), organization_id).await?;
Ok(rule_summary(&updated))
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, rule_id = %rule_id, request_id = %request_id.unwrap_or_default()))]
pub async fn delete_join_rule(
&self,
actor_user_id: i64,
organization_id: Uuid,
rule_id: Uuid,
request_id: Option<&str>,
) -> AuthResult<bool> {
self.require_permission(actor_user_id, organization_id, "role.update", request_id)
.await?;
let deleted = organization_join_rules::Entity::delete_many()
.filter(organization_join_rules::Column::Id.eq(rule_id))
.filter(organization_join_rules::Column::OrganizationId.eq(organization_id))
.exec(self.db().write())
.await?;
if deleted.rows_affected > 0 {
service::bump_version(self.db().write(), organization_id).await?;
}
Ok(deleted.rows_affected > 0)
}
#[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_join_rules(
&self,
actor_user_id: i64,
organization_id: Uuid,
request_id: Option<&str>,
) -> AuthResult<Vec<JoinRuleSummary>> {
self.require_permission(actor_user_id, organization_id, "role.view", request_id)
.await?;
let rows = organization_join_rules::Entity::find()
.filter(organization_join_rules::Column::OrganizationId.eq(organization_id))
.order_by_asc(organization_join_rules::Column::CreatedAt)
.all(self.db().read())
.await?;
Ok(rows.iter().map(rule_summary).collect())
}
#[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_join_request(
&self,
actor_user_id: i64,
organization_id: Uuid,
note: Option<String>,
request_id: Option<&str>,
) -> AuthResult<JoinRequestSummary> {
let organization = self.load_organization(organization_id).await?;
if organization.status != crate::models::space::OrganizationStatus::Active {
return Err(AuthError::OrganizationInactive);
}
if let Some(existing) =
member::find_any(self.db().read(), organization_id, actor_user_id).await?
{
if existing.banned_at.is_some() {
return Err(AuthError::OrganizationMemberBanned);
}
if existing.left_at.is_none() {
return Err(AuthError::OrganizationMemberExists);
}
}
if let Some(pending) = organization_join_requests::Entity::find()
.filter(organization_join_requests::Column::OrganizationId.eq(organization_id))
.filter(organization_join_requests::Column::UserId.eq(actor_user_id))
.filter(
organization_join_requests::Column::Status
.eq(OrganizationJoinRequestStatus::Pending),
)
.one(self.db().read())
.await?
{
let mut active: organization_join_requests::ActiveModel = pending.into();
active.note = Set(note);
let updated = active.update(self.db().write()).await?;
return self.join_request_summary(&updated).await;
}
let row = organization_join_requests::ActiveModel {
organization_id: Set(organization_id),
user_id: Set(actor_user_id),
status: Set(OrganizationJoinRequestStatus::Pending),
note: Set(note),
..Default::default()
}
.insert(self.db().write())
.await?;
self.join_request_summary(&row).await
}
#[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_join_requests(
&self,
actor_user_id: i64,
organization_id: Uuid,
status: Option<OrganizationJoinRequestStatus>,
limit: u64,
offset: u64,
request_id: Option<&str>,
) -> AuthResult<(Vec<JoinRequestSummary>, u64)> {
self.require_permission(actor_user_id, organization_id, "member.view", request_id)
.await?;
let mut select = organization_join_requests::Entity::find()
.filter(organization_join_requests::Column::OrganizationId.eq(organization_id));
if let Some(status) = status {
select = select.filter(organization_join_requests::Column::Status.eq(status));
}
let total = select.clone().count(self.db().read()).await?;
let rows = select
.order_by_desc(organization_join_requests::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.join_request_summary(&row).await?);
}
Ok((out, total))
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, join_request_id = %join_request_id, request_id = %request_id.unwrap_or_default()))]
pub async fn withdraw_join_request(
&self,
actor_user_id: i64,
organization_id: Uuid,
join_request_id: Uuid,
request_id: Option<&str>,
) -> AuthResult<JoinRequestSummary> {
let row = organization_join_requests::Entity::find_by_id(join_request_id)
.filter(organization_join_requests::Column::OrganizationId.eq(organization_id))
.one(self.db().read())
.await?
.ok_or(AuthError::OrganizationMemberNotFound)?;
if row.user_id != actor_user_id {
return Err(AuthError::Forbidden);
}
if row.status != OrganizationJoinRequestStatus::Pending {
return Err(AuthError::Forbidden);
}
let mut active: organization_join_requests::ActiveModel = row.into();
active.status = Set(OrganizationJoinRequestStatus::Withdrawn);
let updated = active.update(self.db().write()).await?;
self.join_request_summary(&updated).await
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, join_request_id = %join_request_id, request_id = %request_id.unwrap_or_default()))]
async fn finish_join_request(
&self,
actor_user_id: i64,
organization_id: Uuid,
join_request_id: Uuid,
status: OrganizationJoinRequestStatus,
note: Option<String>,
request_id: Option<&str>,
) -> AuthResult<JoinRequestSummary> {
self.require_permission(actor_user_id, organization_id, "member.add", request_id)
.await?;
let row = organization_join_requests::Entity::find_by_id(join_request_id)
.filter(organization_join_requests::Column::OrganizationId.eq(organization_id))
.one(self.db().read())
.await?
.ok_or(AuthError::OrganizationMemberNotFound)?;
if row.status != OrganizationJoinRequestStatus::Pending {
return Err(AuthError::Forbidden);
}
if status == OrganizationJoinRequestStatus::Approved {
let default_role = organization_roles::Entity::find()
.filter(organization_roles::Column::OrganizationId.eq(organization_id))
.filter(organization_roles::Column::Key.eq(SYSTEM_ROLE_MEMBER))
.one(self.db().read())
.await?;
self.add_member(
actor_user_id,
organization_id,
AddMemberBody {
user_id: row.user_id,
role: Some(OrgMemberRole::Member),
role_keys: default_role
.filter(|role| role.default_role)
.map(|role| vec![role.key])
.unwrap_or_default(),
},
request_id,
)
.await?;
}
let mut active: organization_join_requests::ActiveModel = row.into();
active.status = Set(status);
active.decided_by = Set(Some(actor_user_id));
active.decided_at = Set(Some(Utc::now()));
if let Some(note) = note {
active.note = Set(Some(note));
}
let updated = active.update(self.db().write()).await?;
self.join_request_summary(&updated).await
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, join_request_id = %join_request_id, request_id = %request_id.unwrap_or_default()))]
pub async fn approve_join_request(
&self,
actor_user_id: i64,
organization_id: Uuid,
join_request_id: Uuid,
request_id: Option<&str>,
) -> AuthResult<JoinRequestSummary> {
self.finish_join_request(
actor_user_id,
organization_id,
join_request_id,
OrganizationJoinRequestStatus::Approved,
None,
request_id,
)
.await
}
#[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, join_request_id = %join_request_id, request_id = %request_id.unwrap_or_default()))]
pub async fn reject_join_request(
&self,
actor_user_id: i64,
organization_id: Uuid,
join_request_id: Uuid,
note: Option<String>,
request_id: Option<&str>,
) -> AuthResult<JoinRequestSummary> {
self.finish_join_request(
actor_user_id,
organization_id,
join_request_id,
OrganizationJoinRequestStatus::Rejected,
note,
request_id,
)
.await
}
#[tracing::instrument(skip_all, fields(organization_id = %organization_id, user_id = user_id, request_id = %request_id.unwrap_or_default()))]
pub async fn ensure_bot_member(
&self,
organization_id: Uuid,
user_id: i64,
role_keys: Vec<String>,
role_ids: Vec<String>,
request_id: Option<&str>,
) -> AuthResult<(crate::organization::types::MemberSummary, bool)> {
let txn = self.db().write().begin().await?;
// 机器人成员的角色上限：OWNER 只能由人类走成员面转移，不能通过内部接口直接塞入。
let user = crate::models::users::users::Entity::find_by_id(user_id)
.filter(crate::models::users::users::Column::DeletedAt.is_null())
.one(&txn)
.await?
.ok_or(AuthError::UserNotFound)?;
if !user.bot {
return Err(AuthError::Forbidden);
}
let (row, created) = member::upsert_join(
&txn,
organization_id,
user_id,
OrgMemberRole::Member,
OrgJoinMethod::Direct,
None,
None,
)
.await?;
let mut roles = Vec::new();
for key in role_keys {
if let Some(role) = organization_roles::Entity::find()
.filter(organization_roles::Column::OrganizationId.eq(organization_id))
.filter(organization_roles::Column::Key.eq(key))
.one(&txn)
.await?
{
if role.key == SYSTEM_ROLE_OWNER {
return Err(AuthError::Forbidden);
}
roles.push(role);
}
}
for id in role_ids {
let Ok(id) = Uuid::parse_str(id.trim()) else {
continue;
};
if let Some(role) = organization_roles::Entity::find_by_id(id)
.filter(organization_roles::Column::OrganizationId.eq(organization_id))
.one(&txn)
.await?
{
if role.key == SYSTEM_ROLE_OWNER {
return Err(AuthError::Forbidden);
}
roles.push(role);
}
}
for role in roles {
let existing = crate::models::space::organization_member_roles::Entity::find()
.filter(
crate::models::space::organization_member_roles::Column::MemberId.eq(row.id),
)
.filter(crate::models::space::organization_member_roles::Column::RoleId.eq(role.id))
.one(&txn)
.await?;
if existing.is_none() {
crate::models::space::organization_member_roles::ActiveModel {
member_id: Set(row.id),
role_id: Set(role.id),
..Default::default()
}
.insert(&txn)
.await?;
}
}
service::bump_version(&txn, organization_id).await?;
txn.commit().await?;
let summary = self.member_summary(&row).await?;
Ok((summary, created))
}
#[tracing::instrument(skip_all, fields(organization_id = %organization_id, user_id = user_id, request_id = %request_id.unwrap_or_default()))]
pub async fn touch_member_activity(
&self,
organization_id: Uuid,
user_id: i64,
request_id: Option<&str>,
) {
let _ = organization_members::Entity::update_many()
.col_expr(
organization_members::Column::LastActiveAt,
Expr::value(Utc::now()),
)
.filter(organization_members::Column::OrganizationId.eq(organization_id))
.filter(organization_members::Column::UserId.eq(user_id))
.exec(self.db().write())
.await;
}
}