13aafa3
use chrono::Utc;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{
ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect,
TransactionTrait,
};
use tonic::{Request, Response, Status};
use uuid::Uuid;
use crate::AuthService;
use crate::api::{context, convert, error, store};
use crate::errors::{AuthError, AuthResult};
use crate::models::space::{
MessageNotifications as DomainNotifications, OrgJoinMethod as DomainJoin,
OrgMemberRole as DomainRole, OrganizationJoinPolicy as DomainJoinPolicy,
OrganizationJoinRequestStatus as DomainRequestStatus, OrganizationStatus as DomainStatus,
PermissionEffect as DomainEffect, organization_member_permissions, organization_member_roles,
organization_members, organization_roles, organizations,
};
use crate::protos::auth::internal::v1::{
GetOrganizationRequest as AdminGetOrganizationRequest,
GetOrganizationResponse as AdminGetOrganizationResponse,
ListOrganizationsRequest as AdminListOrganizationsRequest,
ListOrganizationsResponse as AdminListOrganizationsResponse, ListUserMembershipsRequest,
ListUserMembershipsResponse, OrganizationAdminMembership as AdminMembershipProto,
OrganizationAdminRecord as AdminRecordProto, OrganizationAdminStatus as AdminStatusProto,
PurgeUserMembershipsRequest, PurgeUserMembershipsResponse, ResumeOrganizationRequest,
ResumeOrganizationResponse, SuspendOrganizationRequest, SuspendOrganizationResponse,
};
use crate::protos::auth::v1::organization_service_server::{
OrganizationService, OrganizationServiceServer,
};
use crate::protos::auth::v1::{
OrganizationJoinMethod as OrgJoinMethodProto, OrganizationMemberRole as OrgMemberRoleProto,
OrganizationStatus as OrganizationStatusProto, PermissionEffect as PermissionEffectProto, *,
};
use crate::services::organization::types::{
AddMemberBody, CreateOrganizationBody, CreateRoleBody, JoinRequestSummary, JoinRuleSummary,
MAX_BATCH_PERMISSIONS, PermissionGrant, SetMemberRoleBody, SetRolePermissionsBody,
UpdateBanBody, UpdateMemberBody, UpdateOrganizationBody,
};
use crate::services::organization::{member, role, service};
fn uuid_of(value: &str, request_id: Option<&str>) -> Result<Uuid, Status> {
Uuid::parse_str(value.trim())
.map_err(|_| error::invalid_argument("identifier is not a uuid", request_id))
}
fn status_code(value: DomainStatus) -> i32 {
OrganizationStatusProto::from(value) as i32
}
fn role_code(value: DomainRole) -> i32 {
OrgMemberRoleProto::from(value) as i32
}
fn effect_code(value: DomainEffect) -> i32 {
PermissionEffectProto::from(value) as i32
}
fn notifications_code(value: DomainNotifications) -> &'static str {
match value {
DomainNotifications::All => "all",
DomainNotifications::Mentions => "mentions",
DomainNotifications::Nothing => "nothing",
}
}
fn notifications_from(value: &str) -> Option<DomainNotifications> {
match value.trim() {
"all" => Some(DomainNotifications::All),
"mentions" => Some(DomainNotifications::Mentions),
"nothing" => Some(DomainNotifications::Nothing),
_ => None,
}
}
fn join_code(value: DomainJoin) -> i32 {
OrgJoinMethodProto::from(value) as i32
}
fn join_policy_model(value: i32) -> Option<DomainJoinPolicy> {
match OrganizationJoinPolicy::try_from(value).ok()? {
OrganizationJoinPolicy::Open => Some(DomainJoinPolicy::Open),
OrganizationJoinPolicy::Approval => Some(DomainJoinPolicy::Approval),
OrganizationJoinPolicy::InviteOnly => Some(DomainJoinPolicy::InviteOnly),
OrganizationJoinPolicy::Private => Some(DomainJoinPolicy::Private),
OrganizationJoinPolicy::Unspecified => None,
}
}
fn join_policy_code(value: DomainJoinPolicy) -> i32 {
match value {
DomainJoinPolicy::Open => OrganizationJoinPolicy::Open as i32,
DomainJoinPolicy::Approval => OrganizationJoinPolicy::Approval as i32,
DomainJoinPolicy::InviteOnly => OrganizationJoinPolicy::InviteOnly as i32,
DomainJoinPolicy::Private => OrganizationJoinPolicy::Private as i32,
}
}
fn org_info(
summary: crate::services::organization::types::OrganizationSummary,
) -> OrganizationInfo {
OrganizationInfo {
id: summary.id.to_string(),
slug: summary.slug,
name: summary.name,
description: summary.description.unwrap_or_default(),
avatar_url: summary.avatar_url.unwrap_or_default(),
status: status_code(summary.status),
owner_user_id: summary.owner_user_id,
member_count: summary.member_count as i32,
version: summary.version,
created_at: Some(convert::timestamp_from_datetime(summary.created_at)),
updated_at: Some(convert::timestamp_from_datetime(summary.updated_at)),
archived_at: convert::datetime_opt(summary.archived_at),
my_role: summary.my_role.map(role_code).unwrap_or(0),
join_policy: join_policy_code(summary.join_policy),
allow_join: summary.allow_join,
require_email_verified: summary.require_email_verified,
require_mfa: summary.require_mfa,
welcome_message: summary.welcome_message.unwrap_or_default(),
member_must_enable_2fa: summary.member_must_enable_2fa,
allow_invite: summary.allow_invite,
allow_join_level: i32::from(summary.allow_join_level),
}
}
fn member_info(
summary: crate::services::organization::types::MemberSummary,
) -> OrganizationMemberInfo {
OrganizationMemberInfo {
id: summary.id.to_string(),
organization_id: summary.organization_id.to_string(),
user_id: summary.user_id,
username: summary.username,
displayname: summary.displayname.unwrap_or_default(),
role: role_code(summary.role),
role_keys: summary.role_keys,
permissions: summary.permissions,
joined_at: Some(convert::timestamp_from_datetime(summary.joined_at)),
left_at: convert::datetime_opt(summary.left_at),
banned_at: convert::datetime_opt(summary.banned_at),
invited_by: summary.invited_by.unwrap_or_default(),
join_method: join_code(summary.join_method),
alias_name: summary.alias_name.unwrap_or_default(),
alias_avatar_url: summary.alias_avatar_url.unwrap_or_default(),
message_notifications: notifications_code(summary.message_notifications).to_owned(),
muted: summary.muted,
mute_until: convert::datetime_opt(summary.mute_until),
mobile_push: summary.mobile_push,
suppress_everyone: summary.suppress_everyone,
suppress_roles: summary.suppress_roles,
hide_muted_channels: summary.hide_muted_channels,
ban_reason: summary.ban_reason.unwrap_or_default(),
ban_expires_at: convert::datetime_opt(summary.ban_expires_at),
ban_allow_agent_join: summary.ban_allow_agent_join,
ban_allow_speak: summary.ban_allow_speak,
ban_allow_push_repo: summary.ban_allow_push_repo,
ban_allow_comment_any: summary.ban_allow_comment_any,
version: summary.version,
last_active_at: convert::datetime_opt(summary.last_active_at),
}
}
fn role_info(summary: crate::services::organization::types::RoleSummary) -> OrganizationRoleInfo {
OrganizationRoleInfo {
id: summary.id.to_string(),
organization_id: summary.organization_id.to_string(),
key: summary.key,
name: summary.name,
description: summary.description.unwrap_or_default(),
rank: summary.rank,
system: summary.system,
default_role: summary.default_role,
permissions: summary
.permissions
.into_iter()
.map(|grant| OrganizationPermissionGrant {
permission: grant.permission,
effect: effect_code(grant.effect),
})
.collect(),
member_count: summary.member_count as i32,
created_at: Some(convert::timestamp_from_datetime(summary.created_at)),
updated_at: Some(convert::timestamp_from_datetime(summary.updated_at)),
self_assignable: summary.self_assignable,
version: summary.version,
color: summary.color.unwrap_or_default(),
emoji: summary.emoji.unwrap_or_default(),
icon_hash: summary.icon_hash.unwrap_or_default(),
position: summary.position,
}
}
fn invite_info(
summary: crate::services::organization::types::InviteSummary,
) -> OrganizationInviteInfo {
OrganizationInviteInfo {
id: summary.id.to_string(),
organization_id: summary.organization_id.to_string(),
method: join_code(summary.method),
role_id: summary.role_id.map(|id| id.to_string()).unwrap_or_default(),
target_user_id: summary.target_user_id.unwrap_or_default(),
created_by: summary.created_by,
max_uses: summary.max_uses,
uses: summary.uses,
expires_at: convert::datetime_opt(summary.expires_at),
revoked_at: convert::datetime_opt(summary.revoked_at),
created_at: Some(convert::timestamp_from_datetime(summary.created_at)),
email: summary.email.unwrap_or_default(),
}
}
fn grants(rows: Vec<OrganizationPermissionGrant>) -> Vec<PermissionGrant> {
rows.into_iter()
.filter_map(|row| match PermissionEffectProto::try_from(row.effect) {
Ok(PermissionEffectProto::Allow) => Some(PermissionGrant {
permission: row.permission,
effect: DomainEffect::Allow,
}),
Ok(PermissionEffectProto::Deny) => Some(PermissionGrant {
permission: row.permission,
effect: DomainEffect::Deny,
}),
_ => None,
})
.collect()
}
fn model_role(value: i32) -> Option<DomainRole> {
match OrgMemberRoleProto::try_from(value) {
Ok(OrgMemberRoleProto::Member) => Some(DomainRole::Member),
Ok(OrgMemberRoleProto::Admin) => Some(DomainRole::Admin),
Ok(OrgMemberRoleProto::Owner) => Some(DomainRole::Owner),
_ => None,
}
}
#[derive(Clone)]
pub struct OrganizationApi {
svc: AuthService,
}
impl OrganizationApi {
pub fn new(svc: AuthService) -> Self {
Self { svc }
}
pub fn server(
self,
) -> tonic::service::interceptor::InterceptedService<
OrganizationServiceServer<Self>,
context::Interceptor,
> {
let gate = context::Interceptor::new(self.svc.auth().rpc_key.clone());
OrganizationServiceServer::with_interceptor(self, gate)
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn org_info(
&self,
actor: i64,
id: Uuid,
request_id: Option<&str>,
) -> AuthResult<OrganizationInfo> {
let (summary, _) = self
.svc
.get_organization(actor, Some(id), None, request_id)
.await?;
Ok(org_info(summary))
}
}
#[tonic::async_trait]
impl OrganizationService for OrganizationApi {
async fn create_organization(
&self,
request: Request<CreateOrganizationRequest>,
) -> Result<Response<CreateOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.create_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("CreateOrganization", &result);
result
}
async fn get_organization(
&self,
request: Request<GetOrganizationRequest>,
) -> Result<Response<GetOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.get_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetOrganization", &result);
result
}
async fn list_organizations(
&self,
request: Request<ListOrganizationsRequest>,
) -> Result<Response<ListOrganizationsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.list_organizations_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListOrganizations", &result);
result
}
async fn list_public_organizations(
&self,
request: Request<ListPublicOrganizationsRequest>,
) -> Result<Response<ListPublicOrganizationsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.list_public_organizations_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListPublicOrganizations", &result);
result
}
async fn update_organization(
&self,
request: Request<UpdateOrganizationRequest>,
) -> Result<Response<UpdateOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.update_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("UpdateOrganization", &result);
result
}
async fn archive_organization(
&self,
request: Request<ArchiveOrganizationRequest>,
) -> Result<Response<ArchiveOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.archive_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ArchiveOrganization", &result);
result
}
async fn restore_organization(
&self,
request: Request<RestoreOrganizationRequest>,
) -> Result<Response<RestoreOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.restore_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RestoreOrganization", &result);
result
}
async fn delete_organization(
&self,
request: Request<DeleteOrganizationRequest>,
) -> Result<Response<DeleteOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.delete_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("DeleteOrganization", &result);
result
}
async fn transfer_ownership(
&self,
request: Request<TransferOwnershipRequest>,
) -> Result<Response<TransferOwnershipResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.transfer_ownership_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("TransferOwnership", &result);
result
}
async fn add_member(
&self,
request: Request<AddMemberRequest>,
) -> Result<Response<AddMemberResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.add_member_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("AddMember", &result);
result
}
async fn remove_member(
&self,
request: Request<RemoveMemberRequest>,
) -> Result<Response<RemoveMemberResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.remove_member_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RemoveMember", &result);
result
}
async fn leave_organization(
&self,
request: Request<LeaveOrganizationRequest>,
) -> Result<Response<LeaveOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.leave_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("LeaveOrganization", &result);
result
}
async fn list_members(
&self,
request: Request<ListMembersRequest>,
) -> Result<Response<ListMembersResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.list_members_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListMembers", &result);
result
}
async fn set_member_role(
&self,
request: Request<SetMemberRoleRequest>,
) -> Result<Response<SetMemberRoleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.set_member_role_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("SetMemberRole", &result);
result
}
async fn ban_member(
&self,
request: Request<BanMemberRequest>,
) -> Result<Response<BanMemberResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.ban_member_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("BanMember", &result);
result
}
async fn unban_member(
&self,
request: Request<UnbanMemberRequest>,
) -> Result<Response<UnbanMemberResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.unban_member_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("UnbanMember", &result);
result
}
async fn create_role(
&self,
request: Request<CreateRoleRequest>,
) -> Result<Response<CreateRoleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.create_role_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("CreateRole", &result);
result
}
async fn update_role(
&self,
request: Request<UpdateRoleRequest>,
) -> Result<Response<UpdateRoleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.update_role_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("UpdateRole", &result);
result
}
async fn delete_role(
&self,
request: Request<DeleteRoleRequest>,
) -> Result<Response<DeleteRoleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.delete_role_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("DeleteRole", &result);
result
}
async fn list_roles(
&self,
request: Request<ListRolesRequest>,
) -> Result<Response<ListRolesResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.list_roles_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListRoles", &result);
result
}
async fn set_role_permissions(
&self,
request: Request<SetRolePermissionsRequest>,
) -> Result<Response<SetRolePermissionsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.set_role_permissions_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("SetRolePermissions", &result);
result
}
async fn assign_role(
&self,
request: Request<AssignRoleRequest>,
) -> Result<Response<AssignRoleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.assign_role_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("AssignRole", &result);
result
}
async fn unassign_role(
&self,
request: Request<UnassignRoleRequest>,
) -> Result<Response<UnassignRoleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.unassign_role_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("UnassignRole", &result);
result
}
async fn set_member_permissions(
&self,
request: Request<SetMemberPermissionsRequest>,
) -> Result<Response<SetMemberPermissionsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.set_member_permissions_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("SetMemberPermissions", &result);
result
}
async fn check_permission(
&self,
request: Request<CheckPermissionRequest>,
) -> Result<Response<CheckPermissionResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.check_permission_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("CheckPermission", &result);
result
}
async fn check_permissions(
&self,
request: Request<CheckPermissionsRequest>,
) -> Result<Response<CheckPermissionsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.check_permissions_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("CheckPermissions", &result);
result
}
async fn create_invite(
&self,
request: Request<CreateInviteRequest>,
) -> Result<Response<CreateInviteResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.create_invite_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("CreateInvite", &result);
result
}
async fn revoke_invite(
&self,
request: Request<RevokeInviteRequest>,
) -> Result<Response<RevokeInviteResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.revoke_invite_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RevokeInvite", &result);
result
}
async fn list_invites(
&self,
request: Request<ListInvitesRequest>,
) -> Result<Response<ListInvitesResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.list_invites_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListInvites", &result);
result
}
async fn redeem_invite(
&self,
request: Request<RedeemInviteRequest>,
) -> Result<Response<RedeemInviteResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.redeem_invite_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RedeemInvite", &result);
result
}
async fn get_member(
&self,
request: Request<GetMemberRequest>,
) -> Result<Response<GetMemberResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.get_member_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetMember", &result);
result
}
async fn update_member(
&self,
request: Request<UpdateMemberRequest>,
) -> Result<Response<UpdateMemberResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.update_member_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("UpdateMember", &result);
result
}
async fn update_self_member(
&self,
request: Request<UpdateMemberRequest>,
) -> Result<Response<UpdateMemberResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.update_self_member_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("UpdateSelfMember", &result);
result
}
async fn list_bans(
&self,
request: Request<ListBansRequest>,
) -> Result<Response<ListBansResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.list_bans_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListBans", &result);
result
}
async fn get_ban(
&self,
request: Request<GetBanRequest>,
) -> Result<Response<GetBanResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.get_ban_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetBan", &result);
result
}
async fn update_ban(
&self,
request: Request<UpdateBanRequest>,
) -> Result<Response<UpdateBanResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.update_ban_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("UpdateBan", &result);
result
}
async fn get_role(
&self,
request: Request<GetRoleRequest>,
) -> Result<Response<GetRoleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.get_role_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetRole", &result);
result
}
async fn join_role(
&self,
request: Request<JoinRoleRequest>,
) -> Result<Response<JoinRoleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.join_role_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("JoinRole", &result);
result
}
async fn leave_role(
&self,
request: Request<LeaveRoleRequest>,
) -> Result<Response<LeaveRoleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.leave_role_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("LeaveRole", &result);
result
}
async fn create_join_rule(
&self,
request: Request<CreateJoinRuleRequest>,
) -> Result<Response<CreateJoinRuleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.create_join_rule_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("CreateJoinRule", &result);
result
}
async fn update_join_rule(
&self,
request: Request<UpdateJoinRuleRequest>,
) -> Result<Response<UpdateJoinRuleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.update_join_rule_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("UpdateJoinRule", &result);
result
}
async fn delete_join_rule(
&self,
request: Request<DeleteJoinRuleRequest>,
) -> Result<Response<DeleteJoinRuleResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.delete_join_rule_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("DeleteJoinRule", &result);
result
}
async fn list_join_rules(
&self,
request: Request<ListJoinRulesRequest>,
) -> Result<Response<ListJoinRulesResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.list_join_rules_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListJoinRules", &result);
result
}
async fn create_join_request(
&self,
request: Request<CreateJoinRequestRequest>,
) -> Result<Response<CreateJoinRequestResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.create_join_request_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("CreateJoinRequest", &result);
result
}
async fn list_join_requests(
&self,
request: Request<ListJoinRequestsRequest>,
) -> Result<Response<ListJoinRequestsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.list_join_requests_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListJoinRequests", &result);
result
}
async fn withdraw_join_request(
&self,
request: Request<WithdrawJoinRequestRequest>,
) -> Result<Response<WithdrawJoinRequestResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.withdraw_join_request_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("WithdrawJoinRequest", &result);
result
}
async fn approve_join_request(
&self,
request: Request<ApproveJoinRequestRequest>,
) -> Result<Response<ApproveJoinRequestResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.approve_join_request_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ApproveJoinRequest", &result);
result
}
async fn reject_join_request(
&self,
request: Request<RejectJoinRequestRequest>,
) -> Result<Response<RejectJoinRequestResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.reject_join_request_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("RejectJoinRequest", &result);
result
}
async fn ensure_bot_member(
&self,
request: Request<EnsureBotMemberRequest>,
) -> Result<Response<EnsureBotMemberResponse>, Status> {
let meta = context::meta(&request).await?;
let timer = crate::oltp::metrics::ActiveRpcGuard::new("auth.v1.OrganizationService");
let result = self
.ensure_bot_member_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("EnsureBotMember", &result);
result
}
}
impl OrganizationApi {
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn create_organization_inner(
&self,
request_id: Option<&str>,
request: CreateOrganizationRequest,
) -> Result<Response<CreateOrganizationResponse>, Status> {
let summary = self
.svc
.create_organization(
request.actor_user_id,
CreateOrganizationBody {
slug: request.slug,
name: request.name,
description: Some(request.description).filter(|value| !value.trim().is_empty()),
avatar_url: Some(request.avatar_url).filter(|value| !value.trim().is_empty()),
join_policy: request.join_policy.and_then(join_policy_model),
allow_join: request.allow_join,
require_email_verified: request.require_email_verified,
require_mfa: request.require_mfa,
welcome_message: request.welcome_message,
member_must_enable_2fa: request.member_must_enable_2fa,
allow_invite: request.allow_invite,
allow_join_level: request.allow_join_level.map(|value| value as i16),
},
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let (organization, membership) = self
.svc
.get_organization(request.actor_user_id, Some(summary.id), None, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(CreateOrganizationResponse {
organization: Some(org_info(organization)),
member: membership.map(member_info),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn get_organization_inner(
&self,
request_id: Option<&str>,
request: GetOrganizationRequest,
) -> Result<Response<GetOrganizationResponse>, Status> {
let id = if request.organization_id.trim().is_empty() {
None
} else {
Some(uuid_of(&request.organization_id, request_id)?)
};
let slug = Some(request.slug).filter(|value| !value.trim().is_empty());
let (organization, membership) = self
.svc
.get_organization(request.actor_user_id, id, slug.as_deref(), request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(GetOrganizationResponse {
organization: Some(org_info(organization)),
member: membership.map(member_info),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_organizations_inner(
&self,
request_id: Option<&str>,
request: ListOrganizationsRequest,
) -> Result<Response<ListOrganizationsResponse>, Status> {
let page = request.page.unwrap_or_default();
let offset = store::parse_page_token(&page.page_token)
.ok_or_else(|| error::invalid_argument("page_token is malformed", request_id))?;
let page_size = convert::page_size(page.page_size);
let rows = self
.svc
.list_organizations(
request.actor_user_id,
request.user_id,
request.include_archived,
page_size as u64,
offset,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let next_page_token = if convert::has_more(offset, page_size, rows.len()) {
store::page_token(offset + rows.len() as u64)
} else {
String::new()
};
Ok(Response::new(ListOrganizationsResponse {
organizations: rows.into_iter().map(org_info).collect(),
page: Some(crate::protos::auth::common::v1::PageResponse { next_page_token }),
}))
}
async fn list_public_organizations_inner(
&self,
request_id: Option<&str>,
request: ListPublicOrganizationsRequest,
) -> Result<Response<ListPublicOrganizationsResponse>, Status> {
let page = request.page.unwrap_or_default();
let offset = store::parse_page_token(&page.page_token)
.ok_or_else(|| error::invalid_argument("page_token is malformed", request_id))?;
let page_size = convert::page_size(page.page_size);
let rows = self
.svc
.list_public_organizations(&request.query, page_size as u64, offset)
.await
.map_err(|err| error::status(&err, request_id))?;
let next_page_token = if convert::has_more(offset, page_size, rows.len()) {
store::page_token(offset + rows.len() as u64)
} else {
String::new()
};
Ok(Response::new(ListPublicOrganizationsResponse {
organizations: rows.into_iter().map(org_info).collect(),
page: Some(crate::protos::auth::common::v1::PageResponse { next_page_token }),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn update_organization_inner(
&self,
request_id: Option<&str>,
request: UpdateOrganizationRequest,
) -> Result<Response<UpdateOrganizationResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let summary = self
.svc
.update_organization(
request.actor_user_id,
id,
UpdateOrganizationBody {
name: request.name,
description: request.description,
avatar_url: request.avatar_url,
expected_version: request.expected_version,
join_policy: request.join_policy.and_then(join_policy_model),
allow_join: request.allow_join,
require_email_verified: request.require_email_verified,
require_mfa: request.require_mfa,
welcome_message: request.welcome_message,
member_must_enable_2fa: request.member_must_enable_2fa,
allow_invite: request.allow_invite,
allow_join_level: request.allow_join_level.map(|value| value as i16),
},
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(UpdateOrganizationResponse {
organization: Some(org_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn archive_organization_inner(
&self,
request_id: Option<&str>,
request: ArchiveOrganizationRequest,
) -> Result<Response<ArchiveOrganizationResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let summary = self
.svc
.archive_organization(
request.actor_user_id,
id,
Some(request.reason.as_str()),
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(ArchiveOrganizationResponse {
organization: Some(org_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn restore_organization_inner(
&self,
request_id: Option<&str>,
request: RestoreOrganizationRequest,
) -> Result<Response<RestoreOrganizationResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let summary = self
.svc
.restore_organization(request.actor_user_id, id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(RestoreOrganizationResponse {
organization: Some(org_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn delete_organization_inner(
&self,
request_id: Option<&str>,
request: DeleteOrganizationRequest,
) -> Result<Response<DeleteOrganizationResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let deleted = self
.svc
.delete_organization(
request.actor_user_id,
id,
request.force,
Some(request.reason.as_str()),
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(DeleteOrganizationResponse { deleted }))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn transfer_ownership_inner(
&self,
request_id: Option<&str>,
request: TransferOwnershipRequest,
) -> Result<Response<TransferOwnershipResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let organization = self
.svc
.load_organization(id)
.await
.map_err(|err| error::status(&err, request_id))?;
let previous_owner_id = organization.owner_user_id;
let previous_owner = member::find_any(self.svc.db().read(), id, previous_owner_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let summary = self
.svc
.transfer_ownership(
request.actor_user_id,
id,
request.target_user_id,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let previous = match previous_owner {
Some(row) => Some(
self.svc
.member_summary(&row)
.await
.map_err(|err| error::status(&err, request_id))?,
),
None => None,
};
let new_owner = member::find_active(self.svc.db().read(), id, request.target_user_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let new_owner = match new_owner {
Some(row) => Some(
self.svc
.member_summary(&row)
.await
.map_err(|err| error::status(&err, request_id))?,
),
None => None,
};
Ok(Response::new(TransferOwnershipResponse {
organization: Some(org_info(summary)),
previous_owner: previous.map(member_info),
new_owner: new_owner.map(member_info),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn add_member_inner(
&self,
request_id: Option<&str>,
request: AddMemberRequest,
) -> Result<Response<AddMemberResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let (summary, created) = self
.svc
.add_member(
request.actor_user_id,
id,
AddMemberBody {
user_id: request.user_id,
role: model_role(request.role),
role_keys: request.role_keys,
},
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(AddMemberResponse {
member: Some(member_info(summary)),
created,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn remove_member_inner(
&self,
request_id: Option<&str>,
request: RemoveMemberRequest,
) -> Result<Response<RemoveMemberResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let member_id = if request.member_id.trim().is_empty() {
None
} else {
Some(uuid_of(&request.member_id, request_id)?)
};
let removed = self
.svc
.remove_member(
request.actor_user_id,
id,
member_id,
request.user_id,
Some(request.reason.as_str()),
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(RemoveMemberResponse { removed }))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn leave_organization_inner(
&self,
request_id: Option<&str>,
request: LeaveOrganizationRequest,
) -> Result<Response<LeaveOrganizationResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let left = self
.svc
.leave_organization(request.actor_user_id, id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(LeaveOrganizationResponse { left }))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_members_inner(
&self,
request_id: Option<&str>,
request: ListMembersRequest,
) -> Result<Response<ListMembersResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let page = request.page.unwrap_or_default();
let offset = store::parse_page_token(&page.page_token)
.ok_or_else(|| error::invalid_argument("page_token is malformed", request_id))?;
let page_size = convert::page_size(page.page_size);
let rows = self
.svc
.list_members(
request.actor_user_id,
id,
request.include_inactive,
request.include_banned,
page_size as u64,
offset,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let next_page_token = if convert::has_more(offset, page_size, rows.len()) {
store::page_token(offset + rows.len() as u64)
} else {
String::new()
};
Ok(Response::new(ListMembersResponse {
members: rows.into_iter().map(member_info).collect(),
page: Some(crate::protos::auth::common::v1::PageResponse { next_page_token }),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn set_member_role_inner(
&self,
request_id: Option<&str>,
request: SetMemberRoleRequest,
) -> Result<Response<SetMemberRoleResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let member_id = uuid_of(&request.member_id, request_id)?;
let role = model_role(request.role)
.ok_or_else(|| error::invalid_argument("role is required", request_id))?;
let summary = self
.svc
.set_member_role(
request.actor_user_id,
id,
member_id,
SetMemberRoleBody {
role,
expected_version: request.expected_version,
},
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(SetMemberRoleResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn ban_member_inner(
&self,
request_id: Option<&str>,
request: BanMemberRequest,
) -> Result<Response<BanMemberResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let member_id = member_id_of(
&self.svc,
id,
&request.member_id,
request.user_id,
request_id,
)
.await?;
let summary = self
.svc
.ban_member(
request.actor_user_id,
id,
member_id,
Some(request.reason.as_str()),
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(BanMemberResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn unban_member_inner(
&self,
request_id: Option<&str>,
request: UnbanMemberRequest,
) -> Result<Response<UnbanMemberResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let member_id = member_id_of(
&self.svc,
id,
&request.member_id,
request.user_id,
request_id,
)
.await?;
let summary = self
.svc
.unban_member(request.actor_user_id, id, member_id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(UnbanMemberResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn create_role_inner(
&self,
request_id: Option<&str>,
request: CreateRoleRequest,
) -> Result<Response<CreateRoleResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let summary = self
.svc
.create_role(
request.actor_user_id,
id,
CreateRoleBody {
key: request.key,
name: request.name,
description: Some(request.description).filter(|value| !value.trim().is_empty()),
rank: request.rank,
default_role: request.default_role,
self_assignable: request.self_assignable,
permissions: grants(request.permissions),
},
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(CreateRoleResponse {
role: Some(role_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn update_role_inner(
&self,
request_id: Option<&str>,
request: UpdateRoleRequest,
) -> Result<Response<UpdateRoleResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let role_id = uuid_of(&request.role_id, request_id)?;
let summary = self
.svc
.update_role(
request.actor_user_id,
id,
role_id,
request.name,
request.description,
request.rank,
request.default_role,
request.self_assignable,
request.expected_version,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(UpdateRoleResponse {
role: Some(role_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn delete_role_inner(
&self,
request_id: Option<&str>,
request: DeleteRoleRequest,
) -> Result<Response<DeleteRoleResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let role_id = uuid_of(&request.role_id, request_id)?;
let deleted = self
.svc
.delete_role(request.actor_user_id, id, role_id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(DeleteRoleResponse { deleted }))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_roles_inner(
&self,
request_id: Option<&str>,
request: ListRolesRequest,
) -> Result<Response<ListRolesResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let rows = self
.svc
.list_roles(request.actor_user_id, id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(ListRolesResponse {
roles: rows.into_iter().map(role_info).collect(),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn set_role_permissions_inner(
&self,
request_id: Option<&str>,
request: SetRolePermissionsRequest,
) -> Result<Response<SetRolePermissionsResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let role_id = uuid_of(&request.role_id, request_id)?;
let summary = self
.svc
.set_role_permissions(
request.actor_user_id,
id,
role_id,
SetRolePermissionsBody {
permissions: grants(request.permissions),
},
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(SetRolePermissionsResponse {
role: Some(role_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn assign_role_inner(
&self,
request_id: Option<&str>,
request: AssignRoleRequest,
) -> Result<Response<AssignRoleResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let member_id = member_id_of(
&self.svc,
id,
&request.member_id,
request.user_id,
request_id,
)
.await?;
let role_id = uuid_of(&request.role_id, request_id)?;
let key = self.role_key(id, role_id, request_id).await?;
self.svc
.assign_role(request.actor_user_id, id, member_id, &key, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
self.member_response(
id,
member_id,
request_id,
AssignRoleResponse { member: None },
)
.await
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn unassign_role_inner(
&self,
request_id: Option<&str>,
request: UnassignRoleRequest,
) -> Result<Response<UnassignRoleResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let member_id = member_id_of(
&self.svc,
id,
&request.member_id,
request.user_id,
request_id,
)
.await?;
let role_id = uuid_of(&request.role_id, request_id)?;
let key = self.role_key(id, role_id, request_id).await?;
self.svc
.unassign_role(request.actor_user_id, id, member_id, &key, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let row = member::load_by_id(self.svc.db().read(), id, member_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let summary = self
.svc
.member_summary(&row)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(UnassignRoleResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn set_member_permissions_inner(
&self,
request_id: Option<&str>,
request: SetMemberPermissionsRequest,
) -> Result<Response<SetMemberPermissionsResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let member_id = uuid_of(&request.member_id, request_id)?;
self.svc
.set_member_permissions(
request.actor_user_id,
id,
member_id,
grants(request.permissions),
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let row = member::load_by_id(self.svc.db().read(), id, member_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let summary = self
.svc
.member_summary(&row)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(SetMemberPermissionsResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn check_permission_inner(
&self,
request_id: Option<&str>,
request: CheckPermissionRequest,
) -> Result<Response<CheckPermissionResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let check = self
.svc
.check_permission(
request.actor_user_id,
id,
request.user_id,
&request.permission,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(CheckPermissionResponse {
allowed: check.allowed,
reason: check.reason,
matched_by: check.matched_by,
organization_version: check.organization_version,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn check_permissions_inner(
&self,
request_id: Option<&str>,
request: CheckPermissionsRequest,
) -> Result<Response<CheckPermissionsResponse>, Status> {
if request.permissions.len() > MAX_BATCH_PERMISSIONS {
return Err(error::invalid_argument(
"too many permissions in one request",
request_id,
));
}
let id = uuid_of(&request.organization_id, request_id)?;
let checks = self
.svc
.check_permissions(
request.actor_user_id,
id,
request.user_id,
&request.permissions,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let version = checks
.first()
.map(|check| check.organization_version)
.unwrap_or_default();
let mut allowed = Vec::with_capacity(checks.len());
let mut reasons = Vec::with_capacity(checks.len());
for check in checks {
allowed.push(check.allowed);
reasons.push(check.reason);
}
Ok(Response::new(CheckPermissionsResponse {
allowed,
reasons,
organization_version: version,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn create_invite_inner(
&self,
request_id: Option<&str>,
request: CreateInviteRequest,
) -> Result<Response<CreateInviteResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let role_id = if request.role_id.trim().is_empty() {
None
} else {
Some(uuid_of(&request.role_id, request_id)?)
};
let target_user_id = Some(request.target_user_id).filter(|value| *value > 0);
// 额外角色先解成 UUID：格式错直接拒绝，不在落库时变成脏字符串。
let mut extra_role_ids = Vec::with_capacity(request.role_ids.len());
for raw in &request.role_ids {
let trimmed = raw.trim();
if trimmed.is_empty() {
continue;
}
extra_role_ids.push(uuid_of(trimmed, request_id)?);
}
let ttl_secs = convert::datetime(&request.expires_at.unwrap_or_default()).map(|expires| {
let seconds = (expires - Utc::now()).num_seconds();
seconds.clamp(60, 90 * 24 * 60 * 60) as u64
});
let (mut summary, token) = self
.svc
.create_invite_with_extras(
request.actor_user_id,
id,
role_id,
target_user_id,
request.max_uses,
ttl_secs,
Some(request.email.trim().to_lowercase()).filter(|value| !value.is_empty()),
&extra_role_ids,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let email = request.email.trim().to_lowercase();
if !email.is_empty() {
summary.email = Some(email);
summary.method = DomainJoin::InviteEmail;
}
Ok(Response::new(CreateInviteResponse {
invite: Some(invite_info(summary)),
token,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn revoke_invite_inner(
&self,
request_id: Option<&str>,
request: RevokeInviteRequest,
) -> Result<Response<RevokeInviteResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let invite_id = uuid_of(&request.invite_id, request_id)?;
let revoked = self
.svc
.revoke_invite(request.actor_user_id, id, invite_id, None, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(RevokeInviteResponse { revoked }))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_invites_inner(
&self,
request_id: Option<&str>,
request: ListInvitesRequest,
) -> Result<Response<ListInvitesResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let page = request.page.unwrap_or_default();
let offset = store::parse_page_token(&page.page_token)
.ok_or_else(|| error::invalid_argument("page_token is malformed", request_id))?;
let page_size = convert::page_size(page.page_size);
let rows = self
.svc
.list_invites(
request.actor_user_id,
id,
request.include_revoked,
page_size as u64,
offset,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let next_page_token = if convert::has_more(offset, page_size, rows.len()) {
store::page_token(offset + rows.len() as u64)
} else {
String::new()
};
Ok(Response::new(ListInvitesResponse {
invites: rows.into_iter().map(invite_info).collect(),
page: Some(crate::protos::auth::common::v1::PageResponse { next_page_token }),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn redeem_invite_inner(
&self,
request_id: Option<&str>,
request: RedeemInviteRequest,
) -> Result<Response<RedeemInviteResponse>, Status> {
let (organization_id, member_id, joined) = self
.svc
.redeem_invite(request.actor_user_id, &request.token, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let organization = self
.org_info(request.actor_user_id, organization_id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let row = member::load_by_id(self.svc.db().read(), organization_id, member_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let summary = self
.svc
.member_summary(&row)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(RedeemInviteResponse {
organization: Some(organization),
member: Some(member_info(summary)),
joined,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn get_member_inner(
&self,
request_id: Option<&str>,
request: GetMemberRequest,
) -> Result<Response<GetMemberResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let member_id = if request.member_id.trim().is_empty() {
None
} else {
Some(uuid_of(&request.member_id, request_id)?)
};
let summary = self
.svc
.get_member(
request.actor_user_id,
id,
member_id,
request.user_id,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(GetMemberResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn update_member_inner(
&self,
request_id: Option<&str>,
request: UpdateMemberRequest,
) -> Result<Response<UpdateMemberResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let member_id = member_id_of(
&self.svc,
id,
&request.member_id,
request.user_id,
request_id,
)
.await?;
let notifications = match request.message_notifications.as_deref() {
None | Some("") => None,
Some(value) => Some(notifications_from(value).ok_or_else(|| {
error::invalid_argument("message_notifications is invalid", request_id)
})?),
};
let mute_until = if request.clear_mute_until.unwrap_or(false) {
Some(None)
} else {
match request.mute_until {
Some(value) => Some(Some(convert::datetime(&value).ok_or_else(|| {
error::invalid_argument("mute_until is invalid", request_id)
})?)),
None => None,
}
};
let body = UpdateMemberBody {
alias_name: request.alias_name.map(Some),
alias_avatar_url: request.alias_avatar_url.map(Some),
message_notifications: notifications,
muted: request.muted,
mute_until,
mobile_push: request.mobile_push,
suppress_everyone: request.suppress_everyone,
suppress_roles: request.suppress_roles,
hide_muted_channels: request.hide_muted_channels,
expected_version: request.expected_version,
};
let summary = if request.user_id != 0 && request.user_id == request.actor_user_id {
self.svc
.update_self_member(request.actor_user_id, id, body, request_id)
.await
} else {
self.svc
.update_member(request.actor_user_id, id, member_id, body, request_id)
.await
}
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(UpdateMemberResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn update_self_member_inner(
&self,
request_id: Option<&str>,
request: UpdateMemberRequest,
) -> Result<Response<UpdateMemberResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let notifications = match request.message_notifications.as_deref() {
None | Some("") => None,
Some(value) => Some(notifications_from(value).ok_or_else(|| {
error::invalid_argument("message_notifications is invalid", request_id)
})?),
};
let mute_until = if request.clear_mute_until.unwrap_or(false) {
Some(None)
} else {
match request.mute_until {
Some(value) => Some(Some(convert::datetime(&value).ok_or_else(|| {
error::invalid_argument("mute_until is invalid", request_id)
})?)),
None => None,
}
};
let body = UpdateMemberBody {
alias_name: request.alias_name.map(Some),
alias_avatar_url: request.alias_avatar_url.map(Some),
message_notifications: notifications,
muted: request.muted,
mute_until,
mobile_push: request.mobile_push,
suppress_everyone: request.suppress_everyone,
suppress_roles: request.suppress_roles,
hide_muted_channels: request.hide_muted_channels,
expected_version: request.expected_version,
};
let summary = self
.svc
.update_self_member(request.actor_user_id, id, body, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(UpdateMemberResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_bans_inner(
&self,
request_id: Option<&str>,
request: ListBansRequest,
) -> Result<Response<ListBansResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let page = request.page.unwrap_or_default();
let offset = store::parse_page_token(&page.page_token)
.ok_or_else(|| error::invalid_argument("page_token is malformed", request_id))?;
let page_size = convert::page_size(page.page_size);
let rows = self
.svc
.list_bans(
request.actor_user_id,
id,
page_size as u64,
offset,
request_id,
)
.await
.map_err(|err| error::status(&err, request_id))?;
let next_page_token = if convert::has_more(offset, page_size, rows.len()) {
store::page_token(offset + rows.len() as u64)
} else {
String::new()
};
Ok(Response::new(ListBansResponse {
bans: rows.into_iter().map(member_info).collect(),
page: Some(crate::protos::auth::common::v1::PageResponse { next_page_token }),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn get_ban_inner(
&self,
request_id: Option<&str>,
request: GetBanRequest,
) -> Result<Response<GetBanResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let summary = self
.svc
.get_ban(request.actor_user_id, id, request.user_id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(GetBanResponse {
ban: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn update_ban_inner(
&self,
request_id: Option<&str>,
request: UpdateBanRequest,
) -> Result<Response<UpdateBanResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let expires_at = if request.clear_expires_at.unwrap_or(false) {
Some(None)
} else {
match request.expires_at {
Some(value) => Some(Some(convert::datetime(&value).ok_or_else(|| {
error::invalid_argument("expires_at is invalid", request_id)
})?)),
None => None,
}
};
let body = UpdateBanBody {
reason: request.reason.map(Some),
expires_at,
allow_agent_join: request.allow_agent_join,
allow_speak: request.allow_speak,
allow_push_repo: request.allow_push_repo,
allow_comment_any: request.allow_comment_any,
};
let summary = self
.svc
.update_ban(request.actor_user_id, id, request.user_id, body, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(UpdateBanResponse {
ban: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn get_role_inner(
&self,
request_id: Option<&str>,
request: GetRoleRequest,
) -> Result<Response<GetRoleResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let role_id = uuid_of(&request.role_id, request_id)?;
let summary = self
.svc
.get_role(request.actor_user_id, id, role_id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(GetRoleResponse {
role: Some(role_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn join_role_inner(
&self,
request_id: Option<&str>,
request: JoinRoleRequest,
) -> Result<Response<JoinRoleResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let role_id = uuid_of(&request.role_id, request_id)?;
let summary = self
.svc
.join_role(request.actor_user_id, id, role_id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(JoinRoleResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn leave_role_inner(
&self,
request_id: Option<&str>,
request: LeaveRoleRequest,
) -> Result<Response<LeaveRoleResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let role_id = uuid_of(&request.role_id, request_id)?;
let summary = self
.svc
.leave_role(request.actor_user_id, id, role_id, request_id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(Response::new(LeaveRoleResponse {
member: Some(member_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn create_join_rule_inner(
&self,
request_id: Option<&str>,
request: CreateJoinRuleRequest,
) -> Result<Response<CreateJoinRuleResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let role_id = uuid_of(&request.role_id, request_id)?;
let rule = self
.svc
.create_join_rule(
request.actor_user_id,
organization_id,
role_id,
request.trigger,
request.matcher,
request.enabled,
request_id,
)
.await
.map_err(|value| error::status(&value, request_id))?;
Ok(Response::new(CreateJoinRuleResponse {
rule: Some(join_rule_info(rule)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn update_join_rule_inner(
&self,
request_id: Option<&str>,
request: UpdateJoinRuleRequest,
) -> Result<Response<UpdateJoinRuleResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let rule_id = uuid_of(&request.rule_id, request_id)?;
let role_id = request
.role_id
.filter(|value| !value.trim().is_empty())
.map(|value| uuid_of(&value, request_id))
.transpose()?;
let rule = self
.svc
.update_join_rule(
request.actor_user_id,
organization_id,
rule_id,
role_id,
request.trigger,
request.matcher,
request.enabled,
request_id,
)
.await
.map_err(|value| error::status(&value, request_id))?;
Ok(Response::new(UpdateJoinRuleResponse {
rule: Some(join_rule_info(rule)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn delete_join_rule_inner(
&self,
request_id: Option<&str>,
request: DeleteJoinRuleRequest,
) -> Result<Response<DeleteJoinRuleResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let rule_id = uuid_of(&request.rule_id, request_id)?;
let deleted = self
.svc
.delete_join_rule(request.actor_user_id, organization_id, rule_id, request_id)
.await
.map_err(|value| error::status(&value, request_id))?;
Ok(Response::new(DeleteJoinRuleResponse { deleted }))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_join_rules_inner(
&self,
request_id: Option<&str>,
request: ListJoinRulesRequest,
) -> Result<Response<ListJoinRulesResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let rules = self
.svc
.list_join_rules(request.actor_user_id, organization_id, request_id)
.await
.map_err(|value| error::status(&value, request_id))?;
Ok(Response::new(ListJoinRulesResponse {
rules: rules.into_iter().map(join_rule_info).collect(),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn create_join_request_inner(
&self,
request_id: Option<&str>,
request: CreateJoinRequestRequest,
) -> Result<Response<CreateJoinRequestResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let note = (!request.note.trim().is_empty()).then_some(request.note);
let summary = self
.svc
.create_join_request(request.actor_user_id, organization_id, note, request_id)
.await
.map_err(|value| error::status(&value, request_id))?;
Ok(Response::new(CreateJoinRequestResponse {
request: Some(join_request_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn list_join_requests_inner(
&self,
request_id: Option<&str>,
request: ListJoinRequestsRequest,
) -> Result<Response<ListJoinRequestsResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let status = request_status_from(request.status);
let page = request.page.unwrap_or_default();
let offset = store::parse_page_token(&page.page_token)
.ok_or_else(|| error::invalid_argument("page_token is malformed", request_id))?;
let page_size = convert::page_size(page.page_size);
// PageResponse 只有 next_page_token，总数不对外暴露。
let (rows, _total) = self
.svc
.list_join_requests(
request.actor_user_id,
organization_id,
status,
page_size as u64,
offset,
request_id,
)
.await
.map_err(|value| error::status(&value, request_id))?;
let row_count = rows.len();
let next_page_token = if convert::has_more(offset, page_size, row_count) {
store::page_token(offset + row_count as u64)
} else {
String::new()
};
Ok(Response::new(ListJoinRequestsResponse {
requests: rows.into_iter().map(join_request_info).collect(),
page: Some(crate::protos::auth::common::v1::PageResponse { next_page_token }),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn withdraw_join_request_inner(
&self,
request_id: Option<&str>,
request: WithdrawJoinRequestRequest,
) -> Result<Response<WithdrawJoinRequestResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let join_request_id = uuid_of(&request.request_id, request_id)?;
let summary = self
.svc
.withdraw_join_request(
request.actor_user_id,
organization_id,
join_request_id,
request_id,
)
.await
.map_err(|value| error::status(&value, request_id))?;
Ok(Response::new(WithdrawJoinRequestResponse {
request: Some(join_request_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn approve_join_request_inner(
&self,
request_id: Option<&str>,
request: ApproveJoinRequestRequest,
) -> Result<Response<ApproveJoinRequestResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let join_request_id = uuid_of(&request.request_id, request_id)?;
let summary = self
.svc
.approve_join_request(
request.actor_user_id,
organization_id,
join_request_id,
request_id,
)
.await
.map_err(|value| error::status(&value, request_id))?;
let member = self
.svc
.member_summary(
&crate::models::space::organization_members::Entity::find()
.filter(
crate::models::space::organization_members::Column::OrganizationId
.eq(organization_id),
)
.filter(
crate::models::space::organization_members::Column::UserId
.eq(summary.user_id),
)
.one(self.svc.db().read())
.await
.map_err(|value| error::status(&AuthError::from(value), request_id))?
.ok_or_else(|| {
error::status(&AuthError::OrganizationMemberNotFound, request_id)
})?,
)
.await
.map_err(|value| error::status(&value, request_id))?;
Ok(Response::new(ApproveJoinRequestResponse {
request: Some(join_request_info(summary)),
member: Some(member_info(member)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn reject_join_request_inner(
&self,
request_id: Option<&str>,
request: RejectJoinRequestRequest,
) -> Result<Response<RejectJoinRequestResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let join_request_id = uuid_of(&request.request_id, request_id)?;
let note = (!request.note.trim().is_empty()).then_some(request.note);
let summary = self
.svc
.reject_join_request(
request.actor_user_id,
organization_id,
join_request_id,
note,
request_id,
)
.await
.map_err(|value| error::status(&value, request_id))?;
Ok(Response::new(RejectJoinRequestResponse {
request: Some(join_request_info(summary)),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn ensure_bot_member_inner(
&self,
request_id: Option<&str>,
request: EnsureBotMemberRequest,
) -> Result<Response<EnsureBotMemberResponse>, Status> {
let organization_id = uuid_of(&request.organization_id, request_id)?;
let (summary, created) = self
.svc
.ensure_bot_member(
organization_id,
request.user_id,
request.role_keys,
request.role_ids,
request_id,
)
.await
.map_err(|value| error::status(&value, request_id))?;
Ok(Response::new(EnsureBotMemberResponse {
member: Some(member_info(summary)),
created,
}))
}
}
impl OrganizationApi {
async fn role_key(
&self,
organization_id: Uuid,
role_id: Uuid,
request_id: Option<&str>,
) -> Result<String, Status> {
organization_roles::Entity::find_by_id(role_id)
.filter(organization_roles::Column::OrganizationId.eq(organization_id))
.one(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?
.map(|row| row.key)
.ok_or_else(|| error::status(&AuthError::OrganizationRoleNotFound, request_id))
}
async fn member_response(
&self,
organization_id: Uuid,
member_id: Uuid,
request_id: Option<&str>,
mut response: AssignRoleResponse,
) -> Result<Response<AssignRoleResponse>, Status> {
let row = member::load_by_id(self.svc.db().read(), organization_id, member_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let summary = self
.svc
.member_summary(&row)
.await
.map_err(|err| error::status(&err, request_id))?;
response.member = Some(member_info(summary));
Ok(Response::new(response))
}
}
/// 平台侧管理面。与成员视角的服务分开，因为这里的判定与 actor 的组织身份无关。
#[derive(Clone)]
pub struct OrganizationAdminApi {
svc: AuthService,
}
impl OrganizationAdminApi {
pub fn new(svc: AuthService) -> Self {
Self { svc }
}
pub fn server(
self,
) -> tonic::service::interceptor::InterceptedService<
crate::protos::auth::internal::v1::organization_admin_service_server::OrganizationAdminServiceServer<Self>,
context::Interceptor,
>{
use crate::protos::auth::internal::v1::organization_admin_service_server::OrganizationAdminServiceServer;
let gate = context::Interceptor::new(self.svc.auth().rpc_key.clone());
OrganizationAdminServiceServer::with_interceptor(self, gate)
}
async fn admin_record(
&self,
row: &organizations::Model,
request_id: Option<&str>,
) -> Result<AdminRecordProto, Status> {
let member_count = organization_members::Entity::find()
.filter(organization_members::Column::OrganizationId.eq(row.id))
.filter(organization_members::Column::LeftAt.is_null())
.filter(organization_members::Column::BannedAt.is_null())
.count(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let role_count = organization_roles::Entity::find()
.filter(organization_roles::Column::OrganizationId.eq(row.id))
.count(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
Ok(AdminRecordProto {
id: row.id.to_string(),
slug: row.slug.clone(),
name: row.name.clone(),
status: admin_status_code(row.status),
owner_user_id: row.owner_user_id,
member_count: member_count as i32,
role_count: role_count as i32,
version: row.version,
created_at: Some(convert::timestamp_from_datetime(row.created_at)),
updated_at: Some(convert::timestamp_from_datetime(row.updated_at)),
archived_at: convert::datetime_opt(row.archived_at),
suspended_reason: row.suspended_reason.clone().unwrap_or_default(),
})
}
async fn membership(
&self,
row: &organization_members::Model,
organization: &organizations::Model,
request_id: Option<&str>,
) -> Result<AdminMembershipProto, Status> {
let role_keys = role::role_keys_of(self.svc.db().read(), row.id)
.await
.map_err(|err| error::status(&err, request_id))?;
Ok(AdminMembershipProto {
organization_id: row.organization_id.to_string(),
slug: organization.slug.clone(),
name: organization.name.clone(),
user_id: row.user_id,
role: match row.role {
DomainRole::Owner => "owner",
DomainRole::Admin => "admin",
DomainRole::Member => "member",
}
.to_owned(),
role_keys,
joined_at: Some(convert::timestamp_from_datetime(row.joined_at)),
banned_at: convert::datetime_opt(row.banned_at),
})
}
async fn set_status(
&self,
organization_id: Uuid,
status: DomainStatus,
reason: Option<String>,
request_id: Option<&str>,
) -> Result<AdminRecordProto, Status> {
let txn = self
.svc
.db()
.write()
.begin()
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let updated = organizations::Entity::update_many()
.col_expr(organizations::Column::Status, Expr::value(status))
.col_expr(organizations::Column::SuspendedReason, Expr::value(reason))
.col_expr(
organizations::Column::ArchivedAt,
Expr::value(if status == DomainStatus::Archived {
Some(Utc::now())
} else {
None::<chrono::DateTime<Utc>>
}),
)
.filter(organizations::Column::Id.eq(organization_id))
.filter(organizations::Column::DeletedAt.is_null())
.exec(&txn)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
if updated.rows_affected == 0 {
return Err(error::status(&AuthError::OrganizationNotFound, request_id));
}
service::bump_version(&txn, organization_id)
.await
.map_err(|err| error::status(&err, request_id))?;
txn.commit()
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
self.svc
.invalidate_org(organization_id)
.await
.map_err(|err| error::status(&err, request_id))?;
let row = self
.svc
.load_organization(organization_id)
.await
.map_err(|err| error::status(&err, request_id))?;
self.admin_record(&row, request_id).await
}
}
#[tonic::async_trait]
impl crate::protos::auth::internal::v1::organization_admin_service_server::OrganizationAdminService
for OrganizationAdminApi
{
async fn list_organizations(
&self,
request: Request<AdminListOrganizationsRequest>,
) -> Result<Response<AdminListOrganizationsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer =
crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.OrganizationAdminService");
let result = self
.admin_list_organizations_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListOrganizations", &result);
result
}
async fn get_organization(
&self,
request: Request<AdminGetOrganizationRequest>,
) -> Result<Response<AdminGetOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer =
crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.OrganizationAdminService");
let result = self
.admin_get_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("GetOrganization", &result);
result
}
async fn suspend_organization(
&self,
request: Request<SuspendOrganizationRequest>,
) -> Result<Response<SuspendOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer =
crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.OrganizationAdminService");
let result = self
.admin_suspend_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("SuspendOrganization", &result);
result
}
async fn resume_organization(
&self,
request: Request<ResumeOrganizationRequest>,
) -> Result<Response<ResumeOrganizationResponse>, Status> {
let meta = context::meta(&request).await?;
let timer =
crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.OrganizationAdminService");
let result = self
.admin_resume_organization_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ResumeOrganization", &result);
result
}
async fn purge_user_memberships(
&self,
request: Request<PurgeUserMembershipsRequest>,
) -> Result<Response<PurgeUserMembershipsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer =
crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.OrganizationAdminService");
let result = self
.admin_purge_user_memberships_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("PurgeUserMemberships", &result);
result
}
async fn list_user_memberships(
&self,
request: Request<ListUserMembershipsRequest>,
) -> Result<Response<ListUserMembershipsResponse>, Status> {
let meta = context::meta(&request).await?;
let timer =
crate::oltp::metrics::ActiveRpcGuard::new("auth.internal.v1.OrganizationAdminService");
let result = self
.admin_list_user_memberships_inner(meta.request_id(), request.into_inner())
.await;
timer.finish("ListUserMemberships", &result);
result
}
}
impl OrganizationAdminApi {
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn admin_list_organizations_inner(
&self,
request_id: Option<&str>,
request: AdminListOrganizationsRequest,
) -> Result<Response<AdminListOrganizationsResponse>, Status> {
let limit = if request.limit == 0 || request.limit > convert::MAX_PAGE_SIZE as u64 {
convert::DEFAULT_PAGE_SIZE as u64
} else {
request.limit
};
let mut query =
organizations::Entity::find().filter(organizations::Column::DeletedAt.is_null());
if let Some(status) = admin_status_model(request.status) {
query = query.filter(organizations::Column::Status.eq(status));
}
let needle = request.query.trim().to_owned();
if !needle.is_empty() {
query = query.filter(
sea_orm::Condition::any()
.add(organizations::Column::Slug.starts_with(&needle))
.add(organizations::Column::Name.starts_with(&needle)),
);
}
let rows = query
.order_by_desc(organizations::Column::CreatedAt)
.limit(limit)
.offset(request.offset)
.all(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let mut records = Vec::with_capacity(rows.len());
for row in &rows {
records.push(self.admin_record(row, request_id).await?);
}
let next_offset = if rows.len() as u64 == limit {
request.offset + rows.len() as u64
} else {
0
};
Ok(Response::new(AdminListOrganizationsResponse {
organizations: records,
next_offset,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn admin_get_organization_inner(
&self,
request_id: Option<&str>,
request: AdminGetOrganizationRequest,
) -> Result<Response<AdminGetOrganizationResponse>, Status> {
let row = if request.organization_id.trim().is_empty() {
let slug = request.slug.trim();
organizations::Entity::find()
.filter(organizations::Column::Slug.eq(slug))
.filter(organizations::Column::DeletedAt.is_null())
.one(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?
.ok_or_else(|| error::status(&AuthError::OrganizationNotFound, request_id))?
} else {
let id = uuid_of(&request.organization_id, request_id)?;
self.svc
.load_organization(id)
.await
.map_err(|err| error::status(&err, request_id))?
};
let members = organization_members::Entity::find()
.filter(organization_members::Column::OrganizationId.eq(row.id))
.order_by_desc(organization_members::Column::JoinedAt)
.limit(20)
.all(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let mut recent = Vec::with_capacity(members.len());
for membership_row in &members {
recent.push(self.membership(membership_row, &row, request_id).await?);
}
Ok(Response::new(AdminGetOrganizationResponse {
organization: Some(self.admin_record(&row, request_id).await?),
recent_members: recent,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn admin_suspend_organization_inner(
&self,
request_id: Option<&str>,
request: SuspendOrganizationRequest,
) -> Result<Response<SuspendOrganizationResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let record = self
.set_status(
id,
DomainStatus::Suspended,
Some(request.reason).filter(|value| !value.trim().is_empty()),
request_id,
)
.await?;
Ok(Response::new(SuspendOrganizationResponse {
organization: Some(record),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn admin_resume_organization_inner(
&self,
request_id: Option<&str>,
request: ResumeOrganizationRequest,
) -> Result<Response<ResumeOrganizationResponse>, Status> {
let id = uuid_of(&request.organization_id, request_id)?;
let record = self
.set_status(id, DomainStatus::Active, None, request_id)
.await?;
Ok(Response::new(ResumeOrganizationResponse {
organization: Some(record),
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn admin_purge_user_memberships_inner(
&self,
request_id: Option<&str>,
request: PurgeUserMembershipsRequest,
) -> Result<Response<PurgeUserMembershipsResponse>, Status> {
let rows = organization_members::Entity::find()
.filter(organization_members::Column::UserId.eq(request.user_id))
.filter(organization_members::Column::LeftAt.is_null())
.filter(organization_members::Column::BannedAt.is_null())
.all(self.svc.db().write())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let mut removed = 0;
let mut blocked = Vec::new();
let mut touched = Vec::new();
// 全部成员关系放在一个事务里：逐行提交时中途失败会留下半套数据。
let txn = self
.svc
.db()
.write()
.begin()
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
for row in rows {
// 条件更新：并发改写成 Owner 时这条语句不会命中，判定与写入不可分割。
let updated = organization_members::Entity::update_many()
.col_expr(
organization_members::Column::LeftAt,
Expr::value(Utc::now()),
)
.col_expr(
organization_members::Column::Version,
Expr::col(organization_members::Column::Version).add(1),
)
.filter(organization_members::Column::Id.eq(row.id))
.filter(organization_members::Column::Role.ne(DomainRole::Owner))
.filter(organization_members::Column::LeftAt.is_null())
.exec(&txn)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
if updated.rows_affected == 0 {
blocked.push(row.organization_id.to_string());
continue;
}
organization_member_roles::Entity::delete_many()
.filter(organization_member_roles::Column::MemberId.eq(row.id))
.exec(&txn)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
organization_member_permissions::Entity::delete_many()
.filter(organization_member_permissions::Column::MemberId.eq(row.id))
.exec(&txn)
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
service::bump_version(&txn, row.organization_id)
.await
.map_err(|err| error::status(&err, request_id))?;
removed += 1;
if !touched.contains(&row.organization_id) {
touched.push(row.organization_id);
}
}
txn.commit()
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
for organization_id in touched {
let _ = self.svc.invalidate_org(organization_id).await;
}
Ok(Response::new(PurgeUserMembershipsResponse {
removed,
blocked_organization_ids: blocked,
}))
}
#[tracing::instrument(skip_all, fields(request_id = %request_id.unwrap_or_default()))]
async fn admin_list_user_memberships_inner(
&self,
request_id: Option<&str>,
request: ListUserMembershipsRequest,
) -> Result<Response<ListUserMembershipsResponse>, Status> {
let mut query = organization_members::Entity::find()
.filter(organization_members::Column::UserId.eq(request.user_id));
if !request.include_inactive {
query = query
.filter(organization_members::Column::LeftAt.is_null())
.filter(organization_members::Column::BannedAt.is_null());
}
let rows = query
.order_by_desc(organization_members::Column::JoinedAt)
.all(self.svc.db().read())
.await
.map_err(|err| error::status(&AuthError::from(err), request_id))?;
let mut memberships = Vec::with_capacity(rows.len());
for row in &rows {
let organization = match self.svc.load_organization(row.organization_id).await {
Ok(organization) => organization,
Err(AuthError::OrganizationNotFound) => continue,
Err(err) => return Err(error::status(&err, request_id)),
};
memberships.push(self.membership(row, &organization, request_id).await?);
}
Ok(Response::new(ListUserMembershipsResponse { memberships }))
}
}
async fn member_id_of(
svc: &AuthService,
organization_id: Uuid,
member_id: &str,
user_id: i64,
request_id: Option<&str>,
) -> Result<Uuid, Status> {
if !member_id.trim().is_empty() {
return uuid_of(member_id, request_id);
}
let row = member::find_any(svc.db().read(), organization_id, user_id)
.await
.map_err(|value| error::status(&value, request_id))?;
row.map(|row| row.id)
.ok_or_else(|| error::status(&AuthError::OrganizationMemberNotFound, request_id))
}
fn admin_status_code(value: DomainStatus) -> i32 {
let proto = match value {
DomainStatus::Active => AdminStatusProto::Active,
DomainStatus::Suspended => AdminStatusProto::Suspended,
DomainStatus::Archived => AdminStatusProto::Archived,
};
proto as i32
}
fn admin_status_model(value: i32) -> Option<DomainStatus> {
match AdminStatusProto::try_from(value) {
Ok(AdminStatusProto::Active) => Some(DomainStatus::Active),
Ok(AdminStatusProto::Suspended) => Some(DomainStatus::Suspended),
Ok(AdminStatusProto::Archived) => Some(DomainStatus::Archived),
_ => None,
}
}
fn join_rule_info(summary: JoinRuleSummary) -> JoinRuleInfo {
JoinRuleInfo {
id: summary.id.to_string(),
organization_id: summary.organization_id.to_string(),
role_id: summary.role_id.to_string(),
trigger: summary.trigger,
matcher: summary.matcher,
enabled: summary.enabled,
created_at: Some(convert::timestamp_from_datetime(summary.created_at)),
updated_at: Some(convert::timestamp_from_datetime(summary.updated_at)),
}
}
fn request_status_code(value: DomainRequestStatus) -> i32 {
match value {
DomainRequestStatus::Pending => OrganizationJoinRequestStatus::Pending as i32,
DomainRequestStatus::Approved => OrganizationJoinRequestStatus::Approved as i32,
DomainRequestStatus::Rejected => OrganizationJoinRequestStatus::Rejected as i32,
DomainRequestStatus::Withdrawn => OrganizationJoinRequestStatus::Withdrawn as i32,
}
}
fn request_status_from(value: i32) -> Option<DomainRequestStatus> {
match OrganizationJoinRequestStatus::try_from(value).ok()? {
OrganizationJoinRequestStatus::Pending => Some(DomainRequestStatus::Pending),
OrganizationJoinRequestStatus::Approved => Some(DomainRequestStatus::Approved),
OrganizationJoinRequestStatus::Rejected => Some(DomainRequestStatus::Rejected),
OrganizationJoinRequestStatus::Withdrawn => Some(DomainRequestStatus::Withdrawn),
OrganizationJoinRequestStatus::Unspecified => None,
}
}
fn join_request_info(summary: JoinRequestSummary) -> JoinRequestInfo {
JoinRequestInfo {
id: summary.id.to_string(),
organization_id: summary.organization_id.to_string(),
user_id: summary.user_id,
username: summary.username,
displayname: summary.displayname.unwrap_or_default(),
status: request_status_code(summary.status),
note: summary.note.unwrap_or_default(),
decided_by: summary.decided_by.unwrap_or_default(),
decided_at: convert::datetime_opt(summary.decided_at),
created_at: Some(convert::timestamp_from_datetime(summary.created_at)),
updated_at: Some(convert::timestamp_from_datetime(summary.updated_at)),
}
}