use chrono::{DateTime, Utc};
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::role;
use crate::services::service::{SpaceService, bump_version, normalize_reason};
use crate::services::types::{
    AddMemberBody, MAX_ALIAS_AVATAR_URL_LEN, MAX_ALIAS_NAME_LEN, MAX_REASON_LEN, MemberSummary,
    SetMemberRoleBody, UpdateBanBody, UpdateMemberBody, normalize_alias,
};
use events::label;
use models::space::{SpaceJoinMethod, SpaceMemberRole, space_member_permissions, space_members};
use models::users::users;

pub async fn find_any<C: ConnectionTrait>(
    db: &C,
    organization_id: Uuid,
    user_id: i64,
) -> SpaceResult<Option<space_members::Model>> {
    Ok(space_members::Entity::find()
        .filter(space_members::Column::OrganizationId.eq(organization_id))
        .filter(space_members::Column::UserId.eq(user_id))
        .one(db)
        .await?)
}

pub async fn find_active<C: ConnectionTrait>(
    db: &C,
    organization_id: Uuid,
    user_id: i64,
) -> SpaceResult<Option<space_members::Model>> {
    Ok(space_members::Entity::find()
        .filter(space_members::Column::OrganizationId.eq(organization_id))
        .filter(space_members::Column::UserId.eq(user_id))
        .filter(space_members::Column::LeftAt.is_null())
        .filter(space_members::Column::BannedAt.is_null())
        .one(db)
        .await?)
}

pub async fn load_by_id<C: ConnectionTrait>(
    db: &C,
    organization_id: Uuid,
    member_id: Uuid,
) -> SpaceResult<space_members::Model> {
    space_members::Entity::find_by_id(member_id)
        .filter(space_members::Column::OrganizationId.eq(organization_id))
        .one(db)
        .await?
        .ok_or(SpaceError::SpaceMemberNotFound)
}

pub(crate) async fn insert_owner<C: ConnectionTrait>(
    db: &C,
    organization_id: Uuid,
    user_id: i64,
) -> SpaceResult<space_members::Model> {
    Ok(space_members::ActiveModel {
        organization_id: Set(organization_id),
        user_id: Set(user_id),
        role: Set(SpaceMemberRole::Owner),
        joined_at: Set(Utc::now()),
        join_method: Set(SpaceJoinMethod::Direct),
        ..Default::default()
    }
    .insert(db)
    .await?)
}

/// 会籍写入的统一入口：已有行就复活，没有就新建。
/// `left_at` / `banned_at` 同时清空才算复活，两个字段各自记录一次"被赶出去"的原因，
/// 直接覆盖会丢掉历史，所以只清标志位，历史留在事件与兑换记录里。
pub(crate) async fn upsert_join<C: ConnectionTrait>(
    db: &C,
    organization_id: Uuid,
    user_id: i64,
    role: SpaceMemberRole,
    method: SpaceJoinMethod,
    invited_by: Option<i64>,
    invite_id: Option<Uuid>,
) -> SpaceResult<(space_members::Model, bool)> {
    match find_any(db, organization_id, user_id).await? {
        Some(existing) => {
            if existing.banned_at.is_some() {
                return Err(SpaceError::SpaceMemberBanned);
            }
            if existing.left_at.is_none() {
                return Ok((existing, false));
            }
            let mut active: space_members::ActiveModel = existing.into();
            active.role = Set(role);
            active.joined_at = Set(Utc::now());
            active.left_at = Set(None);
            active.ban_reason = Set(None);
            active.invited_by = Set(invited_by);
            active.join_method = Set(method);
            active.invite_id = Set(invite_id);
            active.version = Set(active.version.unwrap() + 1);
            Ok((active.update(db).await?, true))
        }
        None => {
            let row = space_members::ActiveModel {
                organization_id: Set(organization_id),
                user_id: Set(user_id),
                role: Set(role),
                joined_at: Set(Utc::now()),
                invited_by: Set(invited_by),
                join_method: Set(method),
                invite_id: Set(invite_id),
                ..Default::default()
            }
            .insert(db)
            .await?;
            Ok((row, true))
        }
    }
}

impl SpaceService {
    // 分页与过滤开关直接对应查询参数，拆结构体没有收益。
    #[allow(clippy::too_many_arguments)]
    #[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 admin_list_members(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        include_inactive: bool,
        include_banned: bool,
        limit: u64,
        offset: u64,
        request_id: Option<&str>,
    ) -> SpaceResult<api::pagination::Page<MemberSummary>> {
        self.require_permission(actor_user_id, organization_id, "member.view", request_id)
            .await?;
        let mut query = space_members::Entity::find()
            .filter(space_members::Column::OrganizationId.eq(organization_id));
        if !include_inactive {
            query = query.filter(space_members::Column::LeftAt.is_null());
        }
        if !include_banned {
            query = query.filter(space_members::Column::BannedAt.is_null());
        }
        let total = query.clone().count(self.access.database.read()).await?;
        let rows = query
            .order_by_asc(space_members::Column::JoinedAt)
            .limit(limit)
            .offset(offset)
            .all(self.access.database.read())
            .await?;
        let items = self.member_summaries(&rows).await?;
        Ok(api::pagination::Page::new(items, total, limit, offset))
    }

    #[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 admin_add_member(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        body: AddMemberBody,
        request_id: Option<&str>,
    ) -> SpaceResult<(MemberSummary, bool)> {
        self.require_permission(actor_user_id, organization_id, "member.add", request_id)
            .await?;
        let role = body.role.unwrap_or(SpaceMemberRole::Member);
        if role == SpaceMemberRole::Owner {
            return Err(SpaceError::Forbidden);
        }
        // 角色绑定属于 `role.assign` 范畴：只有 `member.add` 的人不能把新成员直接绑进高权角色。
        if !body.role_keys.is_empty() {
            self.require_permission(actor_user_id, organization_id, "role.assign", request_id)
                .await?;
        }
        let role_keys = role::load_keys(
            organization_id,
            &body.role_keys,
            self.access.database.read(),
        )
        .await?;
        users::Entity::find_by_id(body.user_id)
            .filter(users::Column::DeletedAt.is_null())
            .one(self.access.database.read())
            .await?
            .ok_or(SpaceError::UserNotFound)?;

        let txn = self.access.database.write().begin().await?;
        let (row, created) = upsert_join(
            &txn,
            organization_id,
            body.user_id,
            role,
            SpaceJoinMethod::Direct,
            Some(actor_user_id),
            None,
        )
        .await?;
        for role_id in &role_keys {
            role::bind_role(&txn, row.id, *role_id, actor_user_id).await?;
        }
        if created && let Some(default_role) = self.default_role_id(organization_id).await? {
            role::bind_role(&txn, row.id, default_role, actor_user_id).await?;
        }
        bump_version(&txn, organization_id).await?;
        let refreshed = load_by_id(&txn, organization_id, row.id).await?;
        self.emit_member(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": refreshed.id,
                "user_id": refreshed.user_id,
                "action": "add",
                "created": created,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        // 返回的 version 必须是成员行自己的版本：组织版本对齐的是另一条乐观锁链。
        let summary = self.member_summary(&refreshed).await?;
        Ok((summary, created))
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, user_id = user_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn admin_remove_member(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Option<Uuid>,
        user_id: i64,
        reason: Option<&str>,
        request_id: Option<&str>,
    ) -> SpaceResult<bool> {
        self.require_permission(actor_user_id, organization_id, "member.remove", request_id)
            .await?;
        let reason = normalize_reason(reason)?;
        let row = match member_id {
            Some(id) => load_by_id(self.access.database.read(), organization_id, id).await?,
            None => find_any(self.access.database.read(), organization_id, user_id)
                .await?
                .ok_or(SpaceError::SpaceMemberNotFound)?,
        };
        if row.role == SpaceMemberRole::Owner {
            return Err(SpaceError::SpaceOwnerRequired);
        }
        if row.left_at.is_some() {
            return Ok(false);
        }

        let txn = self.access.database.write().begin().await?;
        space_members::Entity::update_many()
            .col_expr(space_members::Column::LeftAt, Expr::value(Utc::now()))
            .col_expr(
                space_members::Column::BanReason,
                Expr::value(reason.clone()),
            )
            .col_expr(
                space_members::Column::Version,
                Expr::col(space_members::Column::Version).add(1),
            )
            .filter(space_members::Column::Id.eq(row.id))
            .exec(&txn)
            .await?;
        // 离组即失去全部授予：不清掉绑定，复活时会带着旧角色回来，等于悄悄恢复权限。
        role::clear_bindings(&txn, row.id).await?;
        space_member_permissions::Entity::delete_many()
            .filter(space_member_permissions::Column::MemberId.eq(row.id))
            .exec(&txn)
            .await?;
        let version = bump_version(&txn, organization_id).await?;
        self.emit_member(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": row.id,
                "user_id": row.user_id,
                "action": "remove",
                "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, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn leave_organization(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        request_id: Option<&str>,
    ) -> SpaceResult<bool> {
        let row = find_any(self.access.database.read(), organization_id, actor_user_id)
            .await?
            .ok_or(SpaceError::SpaceMemberNotFound)?;
        if row.role == SpaceMemberRole::Owner {
            return Err(SpaceError::SpaceOwnerRequired);
        }
        if row.left_at.is_some() {
            return Ok(false);
        }

        let txn = self.access.database.write().begin().await?;
        space_members::Entity::update_many()
            .col_expr(space_members::Column::LeftAt, Expr::value(Utc::now()))
            .col_expr(
                space_members::Column::Version,
                Expr::col(space_members::Column::Version).add(1),
            )
            .filter(space_members::Column::Id.eq(row.id))
            .exec(&txn)
            .await?;
        role::clear_bindings(&txn, row.id).await?;
        space_member_permissions::Entity::delete_many()
            .filter(space_member_permissions::Column::MemberId.eq(row.id))
            .exec(&txn)
            .await?;
        let version = bump_version(&txn, organization_id).await?;
        self.emit_member(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": row.id,
                "user_id": row.user_id,
                "action": "leave",
                "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, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn set_member_role(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Uuid,
        body: SetMemberRoleBody,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        self.require_permission(
            actor_user_id,
            organization_id,
            "member.role.update",
            request_id,
        )
        .await?;
        if body.role == SpaceMemberRole::Owner {
            return Err(SpaceError::Forbidden);
        }
        let row = load_by_id(self.access.database.read(), organization_id, member_id).await?;
        if row.role == SpaceMemberRole::Owner {
            return Err(SpaceError::SpaceOwnerRequired);
        }
        if body.expected_version > 0 && row.version != body.expected_version {
            return Err(SpaceError::VersionMismatch {
                current: row.version,
            });
        }

        let owner_role =
            role::find_by_key(organization_id, "owner", self.access.database.read()).await?;
        let admin_role =
            role::find_by_key(organization_id, "admin", self.access.database.read()).await?;
        let member_role =
            role::find_by_key(organization_id, "member", self.access.database.read()).await?;

        let txn = self.access.database.write().begin().await?;
        space_members::Entity::update_many()
            .col_expr(space_members::Column::Role, Expr::value(body.role))
            .col_expr(
                space_members::Column::Version,
                Expr::col(space_members::Column::Version).add(1),
            )
            .filter(space_members::Column::Id.eq(member_id))
            .exec(&txn)
            .await?;
        // 粗粒度角色与系统角色绑定必须同步，否则解析链上会出现"粗粒度是 ADMIN、
        // 绑定里还留着 member"这种自相矛盾的组合。
        let target = match body.role {
            SpaceMemberRole::Admin => admin_role.clone(),
            _ => member_role.clone(),
        };
        for system in [owner_role, admin_role, member_role].into_iter().flatten() {
            role::unbind_role(&txn, member_id, system.id).await?;
        }
        if let Some(system) = target {
            role::bind_role(&txn, member_id, system.id, actor_user_id).await?;
        }
        let version = bump_version(&txn, organization_id).await?;
        let refreshed = load_by_id(&txn, organization_id, member_id).await?;
        self.emit_member(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": member_id,
                "user_id": row.user_id,
                "action": "set_role",
                "role": refreshed.role,
                "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, member_id = %member_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn ban_member(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Uuid,
        reason: Option<&str>,
        expires_at: Option<DateTime<Utc>>,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        self.require_permission(actor_user_id, organization_id, "member.ban", request_id)
            .await?;
        let reason = match reason.map(str::trim) {
            None | Some("") => None,
            Some(value) if value.len() > MAX_REASON_LEN => {
                return Err(SpaceError::MalformedRequest);
            }
            Some(value) => Some(value.to_owned()),
        };
        let row = load_by_id(self.access.database.read(), organization_id, member_id).await?;
        if row.role == SpaceMemberRole::Owner {
            return Err(SpaceError::SpaceOwnerRequired);
        }
        if row.banned_at.is_some() {
            return self.member_summary(&row).await;
        }

        let txn = self.access.database.write().begin().await?;
        space_members::Entity::update_many()
            .col_expr(space_members::Column::BannedAt, Expr::value(Utc::now()))
            .col_expr(
                space_members::Column::BanReason,
                Expr::value(reason.clone()),
            )
            .col_expr(
                space_members::Column::BanExpiresAt,
                Expr::value(expires_at),
            )
            .col_expr(
                space_members::Column::Version,
                Expr::col(space_members::Column::Version).add(1),
            )
            .filter(space_members::Column::Id.eq(member_id))
            .exec(&txn)
            .await?;
        // 封禁优先于一切授权，清空授予让"解封后权限自己回来"不可能发生。
        role::clear_bindings(&txn, member_id).await?;
        space_member_permissions::Entity::delete_many()
            .filter(space_member_permissions::Column::MemberId.eq(member_id))
            .exec(&txn)
            .await?;
        let version = bump_version(&txn, organization_id).await?;
        let refreshed = load_by_id(&txn, organization_id, member_id).await?;
        self.emit_member(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": member_id,
                "user_id": row.user_id,
                "action": "ban",
                "reason": &reason,
                "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, member_id = %member_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn unban_member(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Uuid,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        self.require_permission(actor_user_id, organization_id, "member.ban", request_id)
            .await?;
        let row = load_by_id(self.access.database.read(), organization_id, member_id).await?;
        if row.banned_at.is_none() {
            return self.member_summary(&row).await;
        }

        let txn = self.access.database.write().begin().await?;
        // 解封必须把封禁的全部附带状态一起清掉，否则残留的到期时间与四个 allow_* 位会继续生效。
        let unlocked = space_members::Entity::update_many()
            .col_expr(
                space_members::Column::BannedAt,
                Expr::value(None::<chrono::DateTime<Utc>>),
            )
            .col_expr(
                space_members::Column::BanReason,
                Expr::value(None::<String>),
            )
            .col_expr(
                space_members::Column::BanExpiresAt,
                Expr::value(None::<chrono::DateTime<Utc>>),
            )
            .col_expr(space_members::Column::BanAllowAgentJoin, Expr::value(false))
            .col_expr(space_members::Column::BanAllowSpeak, Expr::value(false))
            .col_expr(space_members::Column::BanAllowPushRepo, Expr::value(false))
            .col_expr(
                space_members::Column::BanAllowCommentAny,
                Expr::value(false),
            )
            .col_expr(
                space_members::Column::Version,
                Expr::col(space_members::Column::Version).add(1),
            )
            .filter(space_members::Column::Id.eq(member_id))
            .filter(space_members::Column::BannedAt.is_not_null())
            .exec(&txn)
            .await?;
        if unlocked.rows_affected != 1 {
            txn.rollback().await?;
            return Err(SpaceError::VersionMismatch {
                current: row.version,
            });
        }
        bump_version(&txn, organization_id).await?;
        let refreshed = load_by_id(&txn, organization_id, member_id).await?;
        self.emit_member(
            &txn,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": member_id,
                "user_id": row.user_id,
                "action": "unban",
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

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

    /// 按会籍 id 或 user_id 读会籍；两者都提供时以 member_id 为准。
    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, user_id = user_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn admin_get_member(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Option<Uuid>,
        user_id: i64,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        self.require_permission(actor_user_id, organization_id, "member.view", request_id)
            .await?;
        let row = match member_id {
            Some(id) => load_by_id(self.access.database.read(), organization_id, id).await?,
            None => find_any(self.access.database.read(), organization_id, user_id)
                .await?
                .ok_or(SpaceError::SpaceMemberNotFound)?,
        };
        self.member_summary(&row).await
    }

    /// 管理端更新会籍：别名、头像、通知偏好、免打扰。封禁中的会籍只走 `update_ban`。
    #[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 admin_update_member(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Uuid,
        body: UpdateMemberBody,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        self.require_permission(
            actor_user_id,
            organization_id,
            "member.role.update",
            request_id,
        )
        .await?;
        self.apply_member_update(
            actor_user_id,
            organization_id,
            member_id,
            body,
            "update",
            request_id,
        )
        .await
    }

    /// 成员自助更新：只允许改自己的别名、头像与通知偏好。
    #[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 update_self_member(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        body: UpdateMemberBody,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        let row = find_active(self.access.database.read(), organization_id, actor_user_id)
            .await?
            .ok_or(SpaceError::Forbidden)?;
        self.apply_member_update(
            actor_user_id,
            organization_id,
            row.id,
            body,
            "self_update",
            request_id,
        )
        .await
    }

    async fn apply_member_update(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        member_id: Uuid,
        body: UpdateMemberBody,
        action: &'static str,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        // 读取、版本校验与写回必须在同一事务内，否则并发封禁/离组/改角色会被旧快照覆盖。
        let txn = self.access.database.write().begin().await?;
        let row = load_by_id(&txn, organization_id, member_id).await?;
        if body.expected_version > 0 && row.version != body.expected_version {
            txn.rollback().await?;
            return Err(SpaceError::VersionMismatch {
                current: row.version,
            });
        }
        if action == "update" {
            if row.role == SpaceMemberRole::Owner {
                txn.rollback().await?;
                return Err(SpaceError::SpaceOwnerRequired);
            }
            if row.banned_at.is_some() {
                txn.rollback().await?;
                return Err(SpaceError::Forbidden);
            }
        } else if row.banned_at.is_some() || row.left_at.is_some() {
            txn.rollback().await?;
            return Err(SpaceError::Forbidden);
        }
        let alias_name = match body.alias_name {
            None => row.alias_name.clone(),
            Some(None) => None,
            Some(Some(value)) => normalize_alias_field(value, MAX_ALIAS_NAME_LEN)?,
        };
        let alias_avatar_url = match body.alias_avatar_url {
            None => row.alias_avatar_url.clone(),
            Some(None) => None,
            Some(Some(value)) => normalize_alias_field(value, MAX_ALIAS_AVATAR_URL_LEN)?,
        };
        let message_notifications = body
            .message_notifications
            .unwrap_or(row.message_notifications);
        let muted = body.muted.unwrap_or(row.muted);
        let mute_until = match body.mute_until {
            None => row.mute_until,
            Some(value) => value,
        };
        let mobile_push = body.mobile_push.unwrap_or(row.mobile_push);
        let suppress_everyone = body.suppress_everyone.unwrap_or(row.suppress_everyone);
        let suppress_roles = body.suppress_roles.unwrap_or(row.suppress_roles);
        let hide_muted_channels = body.hide_muted_channels.unwrap_or(row.hide_muted_channels);
        // 条件写回：只有版本未变时才生效，防止旧快照覆盖并发已提交的封禁/离组/角色变更。
        let written = space_members::Entity::update_many()
            .col_expr(
                space_members::Column::AliasName,
                Expr::value(alias_name.clone()),
            )
            .col_expr(
                space_members::Column::AliasAvatarUrl,
                Expr::value(alias_avatar_url.clone()),
            )
            .col_expr(
                space_members::Column::MessageNotifications,
                Expr::value(message_notifications),
            )
            .col_expr(space_members::Column::Muted, Expr::value(muted))
            .col_expr(space_members::Column::MuteUntil, Expr::value(mute_until))
            .col_expr(space_members::Column::MobilePush, Expr::value(mobile_push))
            .col_expr(
                space_members::Column::SuppressEveryone,
                Expr::value(suppress_everyone),
            )
            .col_expr(
                space_members::Column::SuppressRoles,
                Expr::value(suppress_roles),
            )
            .col_expr(
                space_members::Column::HideMutedChannels,
                Expr::value(hide_muted_channels),
            )
            .col_expr(space_members::Column::Version, Expr::value(row.version + 1))
            .filter(space_members::Column::Id.eq(row.id))
            .filter(space_members::Column::Version.eq(row.version))
            .exec(&txn)
            .await?;
        if written.rows_affected != 1 {
            txn.rollback().await?;
            return Err(SpaceError::VersionMismatch {
                current: row.version,
            });
        }
        let version = bump_version(&txn, organization_id).await?;
        let refreshed = 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": row.user_id,
                "action": action,
                "alias_name": alias_name,
                "alias_avatar_url": alias_avatar_url,
                "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)
    }

    /// 封禁列表：只看 `banned_at` 非空的会籍。
    #[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 admin_list_bans(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        limit: u64,
        offset: u64,
        request_id: Option<&str>,
    ) -> SpaceResult<api::pagination::Page<MemberSummary>> {
        self.require_permission(actor_user_id, organization_id, "member.ban", request_id)
            .await?;
        let query = space_members::Entity::find()
            .filter(space_members::Column::OrganizationId.eq(organization_id))
            .filter(space_members::Column::BannedAt.is_not_null());
        let total = query.clone().count(self.access.database.read()).await?;
        let rows = query
            .order_by_desc(space_members::Column::BannedAt)
            .limit(limit)
            .offset(offset)
            .all(self.access.database.read())
            .await?;
        let items = self.member_summaries(&rows).await?;
        Ok(api::pagination::Page::new(items, total, limit, offset))
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, user_id = user_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn admin_get_ban(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        user_id: i64,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        self.require_permission(actor_user_id, organization_id, "member.ban", request_id)
            .await?;
        let row = find_any(self.access.database.read(), organization_id, user_id)
            .await?
            .ok_or(SpaceError::SpaceMemberNotFound)?;
        if row.banned_at.is_none() {
            return Err(SpaceError::SpaceMemberNotFound);
        }
        self.member_summary(&row).await
    }

    /// 改封禁原因、到期时间与四个放行位；会籍必须处于封禁中。
    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, user_id = user_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn admin_update_ban(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        user_id: i64,
        body: UpdateBanBody,
        request_id: Option<&str>,
    ) -> SpaceResult<MemberSummary> {
        self.require_permission(actor_user_id, organization_id, "member.ban", request_id)
            .await?;
        let row = find_any(self.access.database.read(), organization_id, user_id)
            .await?
            .ok_or(SpaceError::SpaceMemberNotFound)?;
        if row.banned_at.is_none() {
            return Err(SpaceError::SpaceMemberNotFound);
        }
        let reason = match body.reason {
            None => row.ban_reason.clone(),
            Some(value) => match value {
                None => None,
                Some(value) => {
                    let trimmed = value.trim();
                    if trimmed.len() > MAX_REASON_LEN {
                        return Err(SpaceError::MalformedRequest);
                    }
                    Some(trimmed.to_owned()).filter(|value| !value.is_empty())
                }
            },
        };
        let expires_at = match body.expires_at {
            None => row.ban_expires_at,
            Some(value) => value,
        };

        let txn = self.access.database.write().begin().await?;
        let mut active: space_members::ActiveModel = row.clone().into();
        active.ban_reason = Set(reason.clone());
        active.ban_expires_at = Set(expires_at);
        if let Some(value) = body.allow_agent_join {
            active.ban_allow_agent_join = Set(value);
        }
        if let Some(value) = body.allow_speak {
            active.ban_allow_speak = Set(value);
        }
        if let Some(value) = body.allow_push_repo {
            active.ban_allow_push_repo = Set(value);
        }
        if let Some(value) = body.allow_comment_any {
            active.ban_allow_comment_any = Set(value);
        }
        active.version = Set(row.version + 1);
        active.update(&txn).await?;
        let version = bump_version(&txn, organization_id).await?;
        let refreshed = 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": row.user_id,
                "action": "update_ban",
                "expires_at": expires_at,
                "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)
    }

    /// 成员事件与业务写入同事务入队：提交时一起可见，回滚时一起消失。
    pub(crate) async fn emit_member(
        &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_MEMBER_UPDATE,
            labels,
            payload,
            request_id,
        )
        .await
    }
}

/// 别名/别名头像的规范化：空白视为清空，超长按 400 处理。
fn normalize_alias_field(value: String, max: usize) -> SpaceResult<Option<String>> {
    normalize_alias(&value, "alias", max).map_err(|_| SpaceError::MalformedRequest)
}