use sea_orm::entity::Set;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{
    ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter,
    QueryOrder, QuerySelect, TransactionTrait,
};
use uuid::Uuid;

use crate::errors::{SpaceError, SpaceResult};
use crate::services::service::{
    SpaceService, bump_version, normalize_role_key, normalize_role_name,
};
use crate::services::types::{
    CreateRoleBody, MemberSummary, PermissionGrant, PermissionGrant as Grant, RoleSummary,
    SYSTEM_ROLE_ADMIN, SYSTEM_ROLE_MEMBER, SYSTEM_ROLE_OWNER, SetRolePermissionsBody,
    is_known_permission,
};
use crate::services::{error, member};
use events::label;
use models::space::{PermissionEffect, space_member_roles, space_role_permissions, space_roles};

/// 内置角色的固有授权。OWNER 不需要显式授权（解析链直接短路），
/// ADMIN 的固有集写在 permissions.rs 的白名单里，这里只落 MEMBER 的默认值。
const MEMBER_DEFAULTS: &[&str] = &[
    "org.view",
    "member.view",
    "role.view",
    "space.view_space",
    "space.read",
    "space.message",
];

pub(crate) async fn seed_system_roles<C: ConnectionTrait>(
    db: &C,
    organization_id: Uuid,
) -> SpaceResult<Uuid> {
    let owner = space_roles::ActiveModel {
        organization_id: Set(organization_id),
        key: Set(SYSTEM_ROLE_OWNER.to_owned()),
        name: Set("Owner".to_owned()),
        description: Set(Some("full control of the organization".to_owned())),
        rank: Set(100),
        system: Set(true),
        default_role: Set(false),
        ..Default::default()
    }
    .insert(db)
    .await?;
    let admin = space_roles::ActiveModel {
        organization_id: Set(organization_id),
        key: Set(SYSTEM_ROLE_ADMIN.to_owned()),
        name: Set("Admin".to_owned()),
        description: Set(Some("manages members, roles and invites".to_owned())),
        rank: Set(50),
        system: Set(true),
        default_role: Set(false),
        ..Default::default()
    }
    .insert(db)
    .await?;
    let member = space_roles::ActiveModel {
        organization_id: Set(organization_id),
        key: Set(SYSTEM_ROLE_MEMBER.to_owned()),
        name: Set("Member".to_owned()),
        description: Set(Some("default role for every new member".to_owned())),
        rank: Set(0),
        system: Set(true),
        default_role: Set(true),
        ..Default::default()
    }
    .insert(db)
    .await?;

    for permission in MEMBER_DEFAULTS {
        space_role_permissions::ActiveModel {
            role_id: Set(member.id),
            permission: Set((*permission).to_owned()),
            effect: Set(PermissionEffect::Allow),
            ..Default::default()
        }
        .insert(db)
        .await?;
    }
    let _ = admin;
    Ok(owner.id)
}

pub(crate) async fn find_by_key<C: ConnectionTrait>(
    organization_id: Uuid,
    key: &str,
    db: &C,
) -> SpaceResult<Option<space_roles::Model>> {
    Ok(space_roles::Entity::find()
        .filter(space_roles::Column::OrganizationId.eq(organization_id))
        .filter(space_roles::Column::Key.eq(key))
        .one(db)
        .await?)
}

pub(crate) async fn keys_of_ids<C: ConnectionTrait>(
    db: &C,
    role_ids: &[Uuid],
) -> SpaceResult<Vec<String>> {
    if role_ids.is_empty() {
        return Ok(Vec::new());
    }
    Ok(space_roles::Entity::find()
        .select_only()
        .column(space_roles::Column::Key)
        .filter(space_roles::Column::Id.is_in(role_ids.to_vec()))
        .order_by_asc(space_roles::Column::Rank)
        .into_tuple::<String>()
        .all(db)
        .await?)
}

pub async fn role_keys_of<C: ConnectionTrait>(db: &C, member_id: Uuid) -> SpaceResult<Vec<String>> {
    let ids = space_member_roles::Entity::find()
        .select_only()
        .column(space_member_roles::Column::RoleId)
        .filter(space_member_roles::Column::MemberId.eq(member_id))
        .into_tuple::<Uuid>()
        .all(db)
        .await?;
    keys_of_ids(db, &ids).await
}

pub(crate) async fn grants_of<C: ConnectionTrait>(
    db: &C,
    role_id: Uuid,
) -> SpaceResult<Vec<PermissionGrant>> {
    let rows = space_role_permissions::Entity::find()
        .filter(space_role_permissions::Column::RoleId.eq(role_id))
        .order_by_asc(space_role_permissions::Column::Permission)
        .all(db)
        .await?;
    Ok(rows
        .into_iter()
        .map(|row| Grant {
            permission: row.permission,
            effect: row.effect,
        })
        .collect())
}

/// 绑定幂等：唯一键冲突直接吞掉，重复授予不该报错。
pub(crate) async fn bind_role<C: ConnectionTrait>(
    db: &C,
    member_id: Uuid,
    role_id: Uuid,
    granted_by: i64,
) -> SpaceResult<()> {
    let existing = space_member_roles::Entity::find()
        .filter(space_member_roles::Column::MemberId.eq(member_id))
        .filter(space_member_roles::Column::RoleId.eq(role_id))
        .one(db)
        .await?;
    if existing.is_some() {
        return Ok(());
    }
    space_member_roles::ActiveModel {
        member_id: Set(member_id),
        role_id: Set(role_id),
        granted_by: Set(Some(granted_by)),
        ..Default::default()
    }
    .insert(db)
    .await?;
    Ok(())
}

pub(crate) async fn unbind_role<C: ConnectionTrait>(
    db: &C,
    member_id: Uuid,
    role_id: Uuid,
) -> SpaceResult<u64> {
    let result = space_member_roles::Entity::delete_many()
        .filter(space_member_roles::Column::MemberId.eq(member_id))
        .filter(space_member_roles::Column::RoleId.eq(role_id))
        .exec(db)
        .await?;
    Ok(result.rows_affected)
}

pub(crate) async fn clear_bindings<C: ConnectionTrait>(db: &C, member_id: Uuid) -> SpaceResult<()> {
    space_member_roles::Entity::delete_many()
        .filter(space_member_roles::Column::MemberId.eq(member_id))
        .exec(db)
        .await?;
    Ok(())
}

pub(crate) async fn load_keys<C: ConnectionTrait>(
    organization_id: Uuid,
    keys: &[String],
    db: &C,
) -> SpaceResult<Vec<Uuid>> {
    if keys.is_empty() {
        return Ok(Vec::new());
    }
    let mut out = Vec::with_capacity(keys.len());
    for key in keys {
        let row = find_by_key(organization_id, key.trim(), db)
            .await?
            .ok_or(SpaceError::SpaceRoleNotFound)?;
        if row.key == SYSTEM_ROLE_OWNER {
            return Err(SpaceError::Forbidden);
        }
        out.push(row.id);
    }
    Ok(out)
}

async fn replace_grants<C: ConnectionTrait>(
    db: &C,
    role_id: Uuid,
    org_version_holder: &Uuid,
    grants: &[PermissionGrant],
) -> SpaceResult<()> {
    space_role_permissions::Entity::delete_many()
        .filter(space_role_permissions::Column::RoleId.eq(role_id))
        .exec(db)
        .await?;
    let _ = org_version_holder;
    for grant in grants {
        space_role_permissions::ActiveModel {
            role_id: Set(role_id),
            permission: Set(grant.permission.clone()),
            effect: Set(grant.effect),
            ..Default::default()
        }
        .insert(db)
        .await?;
    }
    Ok(())
}

impl SpaceService {
    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn list_roles(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        request_id: Option<&str>,
    ) -> SpaceResult<Vec<RoleSummary>> {
        self.require_permission(actor_user_id, organization_id, "role.view", request_id)
            .await?;
        let rows = space_roles::Entity::find()
            .filter(space_roles::Column::OrganizationId.eq(organization_id))
            .order_by_desc(space_roles::Column::Rank)
            .order_by_asc(space_roles::Column::Key)
            .all(self.access.database.read())
            .await?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            out.push(self.role_summary(&row).await?);
        }
        Ok(out)
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn create_role(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        body: CreateRoleBody,
        request_id: Option<&str>,
    ) -> SpaceResult<RoleSummary> {
        self.require_permission(actor_user_id, organization_id, "role.create", request_id)
            .await?;
        let key = normalize_role_key(&body.key)?;
        let name = normalize_role_name(&body.name)?;
        let grants = Self::validate_grants(&body.permissions)?;
        self.require_grantable(actor_user_id, organization_id, &grants, request_id)
            .await?;
        if find_by_key(organization_id, &key, self.access.database.read())
            .await?
            .is_some()
        {
            return Err(SpaceError::SpaceRoleKeyTaken);
        }

        let txn = self.access.database.write().begin().await?;
        let role = space_roles::ActiveModel {
            organization_id: Set(organization_id),
            key: Set(key.clone()),
            name: Set(name.clone()),
            description: Set(body.description.clone()),
            rank: Set(body.rank),
            system: Set(false),
            default_role: Set(body.default_role),
            self_assignable: Set(body.self_assignable),
            ..Default::default()
        }
        .insert(&txn)
        .await
        .map_err(error::map_insert_conflict)?;
        replace_grants(&txn, role.id, &organization_id, &grants).await?;
        let version = bump_version(&txn, organization_id).await?;
        let refreshed = load_role(&txn, organization_id, role.id).await?;
        self.emit_role(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": role.id,
                "key": &key,
                "action": "create",
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        let mut summary = self.role_summary(&refreshed).await?;
        summary.version = version;
        Ok(summary)
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, role_id = %role_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn set_role_permissions(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        role_id: Uuid,
        body: SetRolePermissionsBody,
        request_id: Option<&str>,
    ) -> SpaceResult<RoleSummary> {
        self.require_permission(
            actor_user_id,
            organization_id,
            "role.permission.update",
            request_id,
        )
        .await?;
        let role = load_role(self.access.database.read(), organization_id, role_id).await?;
        if role.key == SYSTEM_ROLE_OWNER {
            return Err(SpaceError::SpaceRoleSystem);
        }
        let grants = Self::validate_grants(&body.permissions)?;
        self.require_grantable(actor_user_id, organization_id, &grants, request_id)
            .await?;

        let txn = self.access.database.write().begin().await?;
        replace_grants(&txn, role_id, &organization_id, &grants).await?;
        space_roles::Entity::update_many()
            .col_expr(
                space_roles::Column::Version,
                Expr::col(space_roles::Column::Version).add(1),
            )
            .col_expr(
                space_roles::Column::UpdatedAt,
                Expr::value(chrono::Utc::now()),
            )
            .filter(space_roles::Column::Id.eq(role_id))
            .exec(&txn)
            .await?;
        let version = bump_version(&txn, organization_id).await?;
        let refreshed = load_role(&txn, organization_id, role_id).await?;
        self.emit_role(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": role_id,
                "key": &refreshed.key,
                "action": "set_permissions",
                "permissions": body.permissions.iter().map(|g| &g.permission).collect::<Vec<_>>(),
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        let mut summary = self.role_summary(&refreshed).await?;
        summary.version = version;
        Ok(summary)
    }

    #[allow(clippy::too_many_arguments)]
    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, role_id = %role_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn update_role(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        role_id: Uuid,
        name: Option<String>,
        description: Option<String>,
        rank: Option<i32>,
        default_role: Option<bool>,
        self_assignable: Option<bool>,
        expected_version: i32,
        request_id: Option<&str>,
    ) -> SpaceResult<RoleSummary> {
        self.require_permission(actor_user_id, organization_id, "role.update", request_id)
            .await?;
        let role = load_role(self.access.database.read(), organization_id, role_id).await?;
        let name = name.as_deref().map(normalize_role_name).transpose()?;
        if expected_version > 0 && role.version != expected_version {
            return Err(SpaceError::VersionMismatch {
                current: role.version,
            });
        }
        if role.key == SYSTEM_ROLE_OWNER
            && (rank.is_some() || default_role.is_some() || self_assignable.is_some())
        {
            return Err(SpaceError::SpaceRoleSystem);
        }
        // 系统角色的排名、默认身份、自助加入都是结构属性，任何改法都会改变权限基线。
        if role.system && (rank.is_some() || default_role.is_some() || self_assignable.is_some()) {
            return Err(SpaceError::SpaceRoleSystem);
        }

        let txn = self.access.database.write().begin().await?;
        if let Some(true) = default_role {
            // 默认角色唯一：换默认之前先清掉旧的，否则新成员会被绑两个默认角色。
            space_roles::Entity::update_many()
                .col_expr(space_roles::Column::DefaultRole, Expr::value(false))
                .filter(space_roles::Column::OrganizationId.eq(organization_id))
                .exec(&txn)
                .await?;
        }
        let mut active: space_roles::ActiveModel = role.clone().into();
        if let Some(value) = name {
            active.name = Set(value);
        }
        if let Some(value) = description {
            active.description = Set(if value.trim().is_empty() {
                None
            } else {
                Some(value)
            });
        }
        if let Some(value) = rank {
            active.rank = Set(value);
        }
        if let Some(value) = default_role {
            active.default_role = Set(value);
        }
        if let Some(value) = self_assignable {
            if role.system {
                return Err(SpaceError::SpaceRoleSystem);
            }
            active.self_assignable = Set(value);
        }
        active.update(&txn).await?;
        let version = bump_version(&txn, organization_id).await?;
        let refreshed = load_role(&txn, organization_id, role_id).await?;
        self.emit_role(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": role_id,
                "key": &role.key,
                "action": "update",
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        let mut summary = self.role_summary(&refreshed).await?;
        summary.version = version;
        Ok(summary)
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, role_id = %role_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn delete_role(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        role_id: Uuid,
        request_id: Option<&str>,
    ) -> SpaceResult<bool> {
        self.require_permission(actor_user_id, organization_id, "role.delete", request_id)
            .await?;
        let role = load_role(self.access.database.read(), organization_id, role_id).await?;
        if role.system {
            return Err(SpaceError::SpaceRoleSystem);
        }
        let bound = space_member_roles::Entity::find()
            .filter(space_member_roles::Column::RoleId.eq(role_id))
            .count(self.access.database.read())
            .await?;
        if bound > 0 {
            return Err(SpaceError::SpaceRoleInUse);
        }

        let txn = self.access.database.write().begin().await?;
        space_role_permissions::Entity::delete_many()
            .filter(space_role_permissions::Column::RoleId.eq(role_id))
            .exec(&txn)
            .await?;
        space_roles::Entity::delete_many()
            .filter(space_roles::Column::Id.eq(role_id))
            .exec(&txn)
            .await?;
        let version = bump_version(&txn, organization_id).await?;
        self.emit_role(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": role_id,
                "key": &role.key,
                "action": "delete",
                "version": version,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        Ok(true)
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, member_id = %member_id, role_key = %role_key, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn assign_role(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Uuid,
        role_key: &str,
        request_id: Option<&str>,
    ) -> SpaceResult<Vec<String>> {
        let member = crate::services::member::load_by_id(
            self.access.database.read(),
            organization_id,
            member_id,
        )
        .await?;
        if member.banned_at.is_some() || member.left_at.is_some() {
            return Err(SpaceError::SpaceMemberNotFound);
        }
        let role = find_by_key(
            organization_id,
            role_key.trim(),
            self.access.database.read(),
        )
        .await?
        .ok_or(SpaceError::SpaceRoleNotFound)?;
        if role.key == SYSTEM_ROLE_OWNER {
            return Err(SpaceError::Forbidden);
        }
        if actor_user_id == member.user_id {
            if role.system || !role.self_assignable {
                return Err(SpaceError::Forbidden);
            }
        } else {
            self.require_permission(actor_user_id, organization_id, "role.assign", request_id)
                .await?;
            // 系统角色等同于整套权限包：代绑时必须具备其中的每一条权限。
            if role.system {
                let grants = grants_of(self.access.database.read(), role.id).await?;
                self.require_grantable(actor_user_id, organization_id, &grants, request_id)
                    .await?;
            }
        }

        let txn = self.access.database.write().begin().await?;
        bind_role(&txn, member_id, role.id, actor_user_id).await?;
        let version = bump_version(&txn, organization_id).await?;
        let keys = role_keys_of(&txn, member_id).await?;
        self.emit_role(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "member_id": member_id,
                "role_key": &role.key,
                "action": "assign",
                "version": version,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        Ok(keys)
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, member_id = %member_id, role_key = %role_key, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn unassign_role(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Uuid,
        role_key: &str,
        request_id: Option<&str>,
    ) -> SpaceResult<Vec<String>> {
        let role = find_by_key(
            organization_id,
            role_key.trim(),
            self.access.database.read(),
        )
        .await?
        .ok_or(SpaceError::SpaceRoleNotFound)?;
        if role.key == SYSTEM_ROLE_OWNER {
            return Err(SpaceError::SpaceRoleSystem);
        }
        let member = crate::services::member::load_by_id(
            self.access.database.read(),
            organization_id,
            member_id,
        )
        .await?;
        if actor_user_id == member.user_id {
            if !role.self_assignable {
                return Err(SpaceError::Forbidden);
            }
        } else {
            // 解绑系统角色必须与绑定同权：否则 Admin 绑定后没人能收回。
            self.require_permission(actor_user_id, organization_id, "role.assign", request_id)
                .await?;
            if role.system {
                let grants = grants_of(self.access.database.read(), role.id).await?;
                self.require_grantable(actor_user_id, organization_id, &grants, request_id)
                    .await?;
            }
        }

        let txn = self.access.database.write().begin().await?;
        unbind_role(&txn, member_id, role.id).await?;
        let version = bump_version(&txn, organization_id).await?;
        let keys = role_keys_of(&txn, member_id).await?;
        self.emit_role(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "member_id": member.id,
                "role_key": &role.key,
                "action": "unassign",
                "version": version,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        Ok(keys)
    }

    /// 成员直挂权限：整表替换语义。`permission` 必须在白名单内，否则直接拒绝。
    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, member_id = %member_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn set_member_permissions(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Uuid,
        grants: Vec<PermissionGrant>,
        request_id: Option<&str>,
    ) -> SpaceResult<Vec<String>> {
        self.require_permission(
            actor_user_id,
            organization_id,
            "role.permission.update",
            request_id,
        )
        .await?;
        let member = crate::services::member::load_by_id(
            self.access.database.read(),
            organization_id,
            member_id,
        )
        .await?;
        if member.role == models::space::SpaceMemberRole::Owner {
            return Err(SpaceError::SpaceOwnerRequired);
        }
        let grants = Self::validate_grants(&grants)?;
        self.require_grantable(actor_user_id, organization_id, &grants, request_id)
            .await?;

        let txn = self.access.database.write().begin().await?;
        models::space::space_member_permissions::Entity::delete_many()
            .filter(models::space::space_member_permissions::Column::MemberId.eq(member_id))
            .exec(&txn)
            .await?;
        for grant in &grants {
            models::space::space_member_permissions::ActiveModel {
                member_id: Set(member_id),
                permission: Set(grant.permission.clone()),
                effect: Set(grant.effect),
                granted_by: Set(Some(actor_user_id)),
                ..Default::default()
            }
            .insert(&txn)
            .await?;
        }
        let version = bump_version(&txn, organization_id).await?;
        self.emit_role(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "member_id": member_id,
                "granted": grants.len(),
                "version": version,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        self.member_permissions(member_id).await
    }

    pub(crate) async fn member_permissions(&self, member_id: Uuid) -> SpaceResult<Vec<String>> {
        let rows = models::space::space_member_permissions::Entity::find()
            .filter(models::space::space_member_permissions::Column::MemberId.eq(member_id))
            .order_by_asc(models::space::space_member_permissions::Column::Permission)
            .all(self.access.database.read())
            .await?;
        Ok(rows
            .into_iter()
            .filter(|row| is_known_permission(&row.permission))
            .map(|row| row.permission)
            .collect())
    }

    /// 角色事件与业务写入同事务入队：提交时一起可见，回滚时一起消失。
    pub(crate) async fn emit_role(
        &self,
        txn: &sea_orm::DatabaseTransaction,
        organization_id: Uuid,
        actor_user_id: i64,
        payload: serde_json::Value,
        request_id: Option<&str>,
    ) -> SpaceResult<()> {
        let labels = events::Labels::new()
            .with(label::ORGANIZATION_ID, organization_id)
            .with(label::ACTOR_USER_ID, actor_user_id);
        self.emit_tx(
            txn,
            label::ORGANIZATION_ROLE_UPDATE,
            labels,
            payload,
            request_id,
        )
        .await
    }

    /// 单个角色详情。
    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, role_id = %role_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn get_role(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        role_id: Uuid,
        request_id: Option<&str>,
    ) -> SpaceResult<RoleSummary> {
        self.require_permission(actor_user_id, organization_id, "role.view", request_id)
            .await?;
        let row = load_role(self.access.database.read(), organization_id, role_id).await?;
        self.role_summary(&row).await
    }

    /// 自助加入：只接受 `self_assignable = true` 的非内置角色。
    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, role_id = %role_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn join_role(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        role_id: Uuid,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        let role = load_role(self.access.database.read(), organization_id, role_id).await?;
        if role.system || !role.self_assignable {
            return Err(SpaceError::Forbidden);
        }
        let row = member::find_active(self.access.database.read(), organization_id, actor_user_id)
            .await?
            .ok_or(SpaceError::Forbidden)?;

        let txn = self.access.database.write().begin().await?;
        bind_role(&txn, row.id, role_id, actor_user_id).await?;
        let version = bump_version(&txn, organization_id).await?;
        let refreshed = member::load_by_id(&txn, organization_id, row.id).await?;
        self.emit_member(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": row.id,
                "user_id": actor_user_id,
                "action": "join_role",
                "role_id": role_id,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        let mut summary = self.member_summary(&refreshed).await?;
        summary.version = version;
        Ok(summary)
    }

    /// 自助退出：同样只接受自助角色；重复退出视为成功。
    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, role_id = %role_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn leave_role(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        role_id: Uuid,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        let role = load_role(self.access.database.read(), organization_id, role_id).await?;
        if role.system || !role.self_assignable {
            return Err(SpaceError::Forbidden);
        }
        let row = member::find_active(self.access.database.read(), organization_id, actor_user_id)
            .await?
            .ok_or(SpaceError::Forbidden)?;

        let txn = self.access.database.write().begin().await?;
        unbind_role(&txn, row.id, role_id).await?;
        let version = bump_version(&txn, organization_id).await?;
        let refreshed = member::load_by_id(&txn, organization_id, row.id).await?;
        self.emit_member(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": row.id,
                "user_id": actor_user_id,
                "action": "leave_role",
                "role_id": role_id,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        let mut summary = self.member_summary(&refreshed).await?;
        summary.version = version;
        Ok(summary)
    }
}

async fn load_role<C: ConnectionTrait>(
    db: &C,
    organization_id: Uuid,
    role_id: Uuid,
) -> SpaceResult<space_roles::Model> {
    space_roles::Entity::find_by_id(role_id)
        .filter(space_roles::Column::OrganizationId.eq(organization_id))
        .one(db)
        .await?
        .ok_or(SpaceError::SpaceRoleNotFound)
}