use std::collections::HashMap;

use crate::errors::{SpaceError, SpaceResult};
use crate::services::{error, member, permission, role};
use chrono::Utc;
use config::AppConfig;
use events::{Labels, label};
use models::users::users;
use sea_orm::sea_query::{Expr, ExprTrait};
use sea_orm::{
    ActiveModelTrait, ColumnTrait, Condition, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
    QuerySelect, Set, TransactionTrait,
};
use serde::Serialize;
use uuid::Uuid;

#[derive(Clone)]
pub struct SpaceService {
    pub(crate) access: crate::Access,
    pub(crate) images: crate::services::images::SpaceImages,
    pub(crate) policy: crate::services::policy::SpacePolicy,
    pub(crate) notifications: push::notify::Notifications,
}

impl SpaceService {
    pub fn new(
        access: crate::Access,
        images: crate::services::images::SpaceImages,
        policy: crate::services::policy::SpacePolicy,
        notifications: push::notify::Notifications,
    ) -> Self {
        Self {
            access,
            images,
            policy,
            notifications,
        }
    }

    pub async fn emit_tx<T>(
        &self,
        conn: &impl sea_orm::ConnectionTrait,
        kind: events::EventKind,
        labels: Labels,
        payload: T,
        request_id: Option<&str>,
    ) -> Result<(), SpaceError>
    where
        T: Serialize,
    {
        events::emit(
            conn,
            kind,
            labels.with_opt(label::REQUEST_ID, request_id),
            payload,
        )
        .await?;
        Ok(())
    }
}

impl SpaceService {
    pub async fn load_user_briefs(
        &self,
        uids: &[i64],
    ) -> Result<
        std::collections::HashMap<i64, crate::services::dto::UserDto>,
        crate::errors::SpaceError,
    > {
        Ok(auth::users::load_user_briefs(&self.policy.accounts, uids).await?)
    }
}

use crate::services::types::{
    CreateOrganizationBody, MAX_ORG_DESCRIPTION_LEN, MAX_ORG_NAME_LEN, MAX_PERMISSION_LEN,
    MAX_ROLE_KEY_LEN, MAX_ROLE_NAME_LEN, MAX_SLUG_LEN, MemberSummary, OrganizationSummary,
    PermissionGrant, RoleSummary, SYSTEM_ROLE_ADMIN, SYSTEM_ROLE_OWNER, UpdateOrganizationBody,
    is_known_permission,
};
use models::space::{
    PermissionEffect, SpaceMemberRole, SpaceStatus, space_invite_redemptions, space_invites,
    space_member_permissions, space_member_roles, space_members, space_role_permissions,
    space_roles, spaces,
};

pub(crate) fn perm_cache_prefix(organization_id: Uuid, user_id: i64, version: i32) -> String {
    format!("auth:org:perm:{organization_id}:v{version}:{user_id}")
}

impl SpaceService {
    /// 组织写操作提交后的统一收尾：清掉本组织非版本化的派生缓存。
    /// 权限缓存 key 已编入 version，写操作只需在事务内推进 version，旧 key 自然失活。
    pub(crate) async fn invalidate_org(&self, organization_id: Uuid) -> SpaceResult<()> {
        crate::services::permission::invalidate(&self.access.cache, organization_id).await
    }

    pub(crate) async fn peek_org_version(&self, organization_id: Uuid) -> SpaceResult<Option<i32>> {
        Ok(spaces::Entity::find_by_id(organization_id)
            .select_only()
            .column(spaces::Column::Version)
            .into_tuple::<i32>()
            .one(self.access.database.read())
            .await?)
    }

    /// `expected_version <= 0` 表示调用方不做乐观锁校验。
    pub(crate) async fn check_version(
        &self,
        organization_id: Uuid,
        expected: i32,
    ) -> SpaceResult<()> {
        if expected <= 0 {
            return Ok(());
        }
        let current = self
            .peek_org_version(organization_id)
            .await?
            .ok_or(SpaceError::SpaceNotFound)?;
        if current != expected {
            return Err(SpaceError::VersionMismatch { current });
        }
        Ok(())
    }

    pub(crate) async fn is_platform_admin(&self, user_id: i64) -> SpaceResult<bool> {
        Ok(users::Entity::find_by_id(user_id)
            .select_only()
            .column(users::Column::IsAdmin)
            .into_tuple::<bool>()
            .one(self.access.database.read())
            .await?
            .unwrap_or(false))
    }

    /// 唯一权限闸门。所有组织内写操作先过这里，语义与 CheckPermission 完全一致。
    pub(crate) async fn require_permission(
        &self,
        user_id: i64,
        organization_id: Uuid,
        permission: &str,
        request_id: Option<&str>,
    ) -> SpaceResult<()> {
        let outcome =
            permission::resolve(self, organization_id, user_id, permission, request_id).await?;
        if outcome.allowed {
            return Ok(());
        }
        Err(error::reason_error(outcome.reason, permission))
    }

    /// 授权上限：只能授出自己当前确实持有的权限，不能借新建/改角色提权。
    pub(crate) async fn require_grantable(
        &self,
        user_id: i64,
        organization_id: Uuid,
        grants: &[PermissionGrant],
        request_id: Option<&str>,
    ) -> SpaceResult<()> {
        for grant in grants {
            if grant.effect == PermissionEffect::Allow {
                self.require_permission(user_id, organization_id, &grant.permission, request_id)
                    .await?;
            }
        }
        Ok(())
    }

    /// 组织事件在业务事务**内部**入队：业务写入与事件行同生共死。
    ///
    /// 写不进 outbox 就让整个事务回滚（`SpaceError` 向上传播），这是刻意选择：
    /// 宁可这次写操作失败重试，也不要出现「组织改成了、消费方永远不知道」。
    async fn emit_org<T: serde::Serialize>(
        &self,
        txn: &sea_orm::DatabaseTransaction,
        kind: events::EventKind,
        organization_id: Uuid,
        actor_user_id: i64,
        payload: T,
        request_id: Option<&str>,
    ) -> SpaceResult<()> {
        let labels = Labels::new()
            .with(label::ORGANIZATION_ID, organization_id)
            .with(label::ACTOR_USER_ID, actor_user_id);
        self.emit_tx(txn, kind, labels, payload, request_id).await
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn create_space(
        &self,
        actor_user_id: i64,
        body: CreateOrganizationBody,
        request_id: Option<&str>,
    ) -> SpaceResult<OrganizationSummary> {
        let slug = normalize_slug(&body.slug)?;
        let name = normalize_name(&body.name)?;
        let description = normalize_description(body.description.as_deref())?;
        let avatar_url = normalize_avatar(body.avatar_url.as_deref())?;

        users::Entity::find_by_id(actor_user_id)
            .filter(users::Column::DeletedAt.is_null())
            .one(self.access.database.read())
            .await?
            .ok_or(SpaceError::UserNotFound)?;
        if spaces::Entity::find()
            .filter(spaces::Column::Slug.eq(slug.clone()))
            .one(self.access.database.read())
            .await?
            .is_some()
        {
            return Err(SpaceError::SpaceSlugTaken);
        }

        let welcome_message = match body.welcome_message.as_deref() {
            Some(value) if value.trim().is_empty() => None,
            Some(value) => normalize_description(Some(value))?,
            None => None,
        };
        let allow_join_level = valid_join_level(body.allow_join_level.unwrap_or(0))?;

        let txn = self.access.database.write().begin().await?;
        let organization = spaces::ActiveModel {
            slug: Set(slug),
            name: Set(name),
            description: Set(description),
            avatar_url: Set(avatar_url),
            status: Set(SpaceStatus::Active),
            owner_user_id: Set(actor_user_id),
            join_policy: Set(body
                .join_policy
                .unwrap_or(models::space::SpaceJoinPolicy::InviteOnly)),
            allow_join: Set(body.allow_join.unwrap_or(false)),
            require_email_verified: Set(body.require_email_verified.unwrap_or(true)),
            require_mfa: Set(body.require_mfa.unwrap_or(false)),
            welcome_message: Set(welcome_message),
            member_must_enable_2fa: Set(body.member_must_enable_2fa.unwrap_or(false)),
            allow_invite: Set(body.allow_invite.unwrap_or(true)),
            allow_join_level: Set(allow_join_level),
            ..Default::default()
        }
        .insert(&txn)
        .await
        .map_err(error::map_insert_conflict)?;

        let owner_role = role::seed_system_roles(&txn, organization.id).await?;
        let membership = member::insert_owner(&txn, organization.id, actor_user_id).await?;
        role::bind_role(&txn, membership.id, owner_role, actor_user_id).await?;
        self.emit_org(
            &txn,
            label::ORGANIZATION_LIFE_CYCLE_CREATE,
            organization.id,
            actor_user_id,
            serde_json::json!({
                "id": organization.id,
                "slug": &organization.slug,
                "name": &organization.name,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        let summary = self
            .organization_summary(&organization, Some(actor_user_id))
            .await?;
        Ok(summary)
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = ?organization_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn get_space(
        &self,
        actor_user_id: i64,
        organization_id: Option<Uuid>,
        slug: Option<&str>,
        request_id: Option<&str>,
    ) -> SpaceResult<(OrganizationSummary, Option<MemberSummary>)> {
        let organization = match (
            organization_id,
            slug.map(str::trim).filter(|value| !value.is_empty()),
        ) {
            (Some(id), _) => {
                spaces::Entity::find_by_id(id)
                    .filter(spaces::Column::DeletedAt.is_null())
                    .one(self.access.database.read())
                    .await?
            }
            (None, Some(slug)) => {
                spaces::Entity::find()
                    .filter(spaces::Column::Slug.eq(slug.to_owned()))
                    .filter(spaces::Column::DeletedAt.is_null())
                    .one(self.access.database.read())
                    .await?
            }
            (None, None) => None,
        }
        .ok_or(SpaceError::SpaceNotFound)?;

        let summary = self
            .organization_summary(&organization, Some(actor_user_id))
            .await?;
        // 非成员只看公开字段：my_role 为空即可，不额外补一层 DTO。
        let detail =
            match member::find_any(self.access.database.read(), organization.id, actor_user_id)
                .await?
            {
                Some(row) => Some(self.member_summary(&row).await?),
                None => None,
            };
        Ok((summary, detail))
    }

    pub async fn space_id_by_name(
        &self,
        actor_user_id: i64,
        space_name: &str,
    ) -> SpaceResult<Uuid> {
        let slug = space_name.trim().to_lowercase();
        let (summary, _) = self
            .get_space(actor_user_id, None, Some(&slug), None)
            .await?;
        Ok(summary.id)
    }

    pub(crate) async fn organization_summary(
        &self,
        organization: &spaces::Model,
        actor_user_id: Option<i64>,
    ) -> SpaceResult<OrganizationSummary> {
        let member_count = space_members::Entity::find()
            .filter(space_members::Column::OrganizationId.eq(organization.id))
            .filter(space_members::Column::LeftAt.is_null())
            .filter(space_members::Column::BannedAt.is_null())
            .count(self.access.database.read())
            .await? as i64;
        let my_role = match actor_user_id {
            Some(user_id) => {
                member::find_active(self.access.database.read(), organization.id, user_id)
                    .await?
                    .map(|row| row.role)
            }
            None => None,
        };
        Ok(OrganizationSummary {
            id: organization.id,
            slug: organization.slug.clone(),
            name: organization.name.clone(),
            description: organization.description.clone(),
            avatar_url: organization.avatar_url.clone(),
            status: organization.status,
            owner_user_id: organization.owner_user_id,
            member_count,
            version: organization.version,
            created_at: organization.created_at,
            updated_at: organization.updated_at,
            archived_at: organization.archived_at,
            my_role,
            join_policy: organization.join_policy,
            allow_join: organization.allow_join,
            require_email_verified: organization.require_email_verified,
            require_mfa: organization.require_mfa,
            welcome_message: organization.welcome_message.clone(),
            member_must_enable_2fa: organization.member_must_enable_2fa,
            allow_invite: organization.allow_invite,
            allow_join_level: organization.allow_join_level,
        })
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, user_id = user_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn list_spaces(
        &self,
        actor_user_id: i64,
        user_id: i64,
        include_archived: bool,
        limit: u64,
        offset: u64,
        request_id: Option<&str>,
    ) -> SpaceResult<api::pagination::Page<OrganizationSummary>> {
        let target = if user_id == 0 { actor_user_id } else { user_id };
        if target != actor_user_id && !self.is_platform_admin(actor_user_id).await? {
            return Err(SpaceError::Forbidden);
        }

        let ids = space_members::Entity::find()
            .select_only()
            .column(space_members::Column::OrganizationId)
            .filter(space_members::Column::UserId.eq(target))
            .filter(space_members::Column::LeftAt.is_null())
            .filter(space_members::Column::BannedAt.is_null())
            .into_tuple::<Uuid>()
            .all(self.access.database.read())
            .await?;
        if ids.is_empty() {
            return Ok(api::pagination::Page::new(Vec::new(), 0, limit, offset));
        }

        let mut query = spaces::Entity::find()
            .filter(spaces::Column::DeletedAt.is_null())
            .filter(spaces::Column::Id.is_in(ids));
        if !include_archived {
            query = query.filter(spaces::Column::Status.ne(SpaceStatus::Archived));
        }
        let total = query.clone().count(self.access.database.read()).await?;
        let rows = query
            .order_by_desc(spaces::Column::UpdatedAt)
            .limit(limit)
            .offset(offset)
            .all(self.access.database.read())
            .await?;

        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            out.push(self.organization_summary(&row, Some(target)).await?);
        }
        Ok(api::pagination::Page::new(out, total, limit, offset))
    }

    /// 公开目录：ACTIVE 且非 PRIVATE 的组织。不要求成员身份。
    pub async fn list_public_spaces(
        &self,
        query: &str,
        limit: u64,
        offset: u64,
    ) -> SpaceResult<Vec<OrganizationSummary>> {
        let mut select = spaces::Entity::find()
            .filter(spaces::Column::DeletedAt.is_null())
            .filter(spaces::Column::Status.eq(SpaceStatus::Active))
            .filter(spaces::Column::JoinPolicy.ne(models::space::SpaceJoinPolicy::Private));
        let needle = query.trim();
        if !needle.is_empty() {
            select = select.filter(
                Condition::any()
                    .add(spaces::Column::Slug.contains(needle))
                    .add(spaces::Column::Name.contains(needle)),
            );
        }
        let rows = select
            .order_by_desc(spaces::Column::CreatedAt)
            .limit(limit)
            .offset(offset)
            .all(self.access.database.read())
            .await?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            out.push(self.organization_summary(&row, None).await?);
        }
        Ok(out)
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn update_space(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        body: UpdateOrganizationBody,
        request_id: Option<&str>,
    ) -> SpaceResult<OrganizationSummary> {
        self.require_permission(actor_user_id, organization_id, "org.update", request_id)
            .await?;
        self.check_version(organization_id, body.expected_version)
            .await?;

        let name = body.name.as_deref().map(normalize_name).transpose()?;
        let description = match body.description.as_deref() {
            Some(value) => Some(normalize_description(Some(value))?),
            None => None,
        };
        let avatar_url = match body.avatar_url.as_deref() {
            Some(value) => Some(normalize_avatar(Some(value))?),
            None => None,
        };

        let txn = self.access.database.write().begin().await?;
        let current = spaces::Entity::find_by_id(organization_id)
            .filter(spaces::Column::DeletedAt.is_null())
            .one(&txn)
            .await?
            .ok_or(SpaceError::SpaceNotFound)?;
        if body.expected_version > 0 && current.version != body.expected_version {
            return Err(SpaceError::VersionMismatch {
                current: current.version,
            });
        }
        let mut active: spaces::ActiveModel = current.into();
        if let Some(value) = name {
            active.name = Set(value);
        }
        if let Some(value) = description {
            active.description = Set(value);
        }
        if let Some(value) = avatar_url {
            active.avatar_url = Set(value);
        }
        if let Some(value) = body.join_policy {
            active.join_policy = Set(value);
        }
        if let Some(value) = body.allow_join {
            active.allow_join = Set(value);
        }
        if let Some(value) = body.require_email_verified {
            active.require_email_verified = Set(value);
        }
        if let Some(value) = body.require_mfa {
            active.require_mfa = Set(value);
        }
        if let Some(value) = body.welcome_message.clone() {
            let trimmed = value.trim();
            active.welcome_message = Set(if trimmed.is_empty() {
                None
            } else {
                Some(trimmed.to_string())
            });
        }
        if let Some(value) = body.member_must_enable_2fa {
            active.member_must_enable_2fa = Set(value);
        }
        if let Some(value) = body.allow_invite {
            active.allow_invite = Set(value);
        }
        if let Some(value) = body.allow_join_level {
            active.allow_join_level = Set(valid_join_level(value)?);
        }
        let updated = active.update(&txn).await?;
        let version = bump_version(&txn, organization_id).await?;
        self.emit_org(
            &txn,
            label::ORGANIZATION_LIFE_CYCLE_UPDATE,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": organization_id,
                "name": &updated.name,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        let refreshed = self.load_organization(organization_id).await?;
        let mut summary = self
            .organization_summary(&refreshed, Some(actor_user_id))
            .await?;
        summary.version = version;
        Ok(summary)
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn archive_space(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        reason: Option<&str>,
        request_id: Option<&str>,
    ) -> SpaceResult<OrganizationSummary> {
        self.require_permission(actor_user_id, organization_id, "org.archive", request_id)
            .await?;
        let reason = normalize_reason(reason)?;

        let txn = self.access.database.write().begin().await?;
        spaces::Entity::update_many()
            .col_expr(spaces::Column::Status, Expr::value(SpaceStatus::Archived))
            .col_expr(spaces::Column::ArchivedAt, Expr::value(Utc::now()))
            .col_expr(spaces::Column::SuspendedReason, Expr::value(reason))
            .filter(spaces::Column::Id.eq(organization_id))
            .filter(spaces::Column::DeletedAt.is_null())
            .exec(&txn)
            .await?;
        let version = bump_version(&txn, organization_id).await?;
        self.emit_org(
            &txn,
            label::ORGANIZATION_LIFE_CYCLE_UPDATE,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": organization_id,
                "action": "archive",
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        let refreshed = self.load_organization(organization_id).await?;
        let mut summary = self
            .organization_summary(&refreshed, Some(actor_user_id))
            .await?;
        summary.version = version;
        Ok(summary)
    }

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn restore_space(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        request_id: Option<&str>,
    ) -> SpaceResult<OrganizationSummary> {
        // 归档组织的权限判定一律不通过，所以这里不能用 require_permission：
        // 恢复动作由 OWNER 或平台管理员执行。
        let organization = self.load_organization(organization_id).await?;
        if organization.owner_user_id != actor_user_id
            && !self.is_platform_admin(actor_user_id).await?
        {
            return Err(SpaceError::Forbidden);
        }

        let txn = self.access.database.write().begin().await?;
        spaces::Entity::update_many()
            .col_expr(spaces::Column::Status, Expr::value(SpaceStatus::Active))
            .col_expr(
                spaces::Column::ArchivedAt,
                Expr::value(None::<chrono::DateTime<Utc>>),
            )
            .col_expr(spaces::Column::SuspendedReason, Expr::value(None::<String>))
            .filter(spaces::Column::Id.eq(organization_id))
            .exec(&txn)
            .await?;
        let version = bump_version(&txn, organization_id).await?;
        self.emit_org(
            &txn,
            label::ORGANIZATION_LIFE_CYCLE_UPDATE,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": organization_id,
                "action": "restore",
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        let refreshed = self.load_organization(organization_id).await?;
        let mut summary = self
            .organization_summary(&refreshed, Some(actor_user_id))
            .await?;
        summary.version = version;
        Ok(summary)
    }

    /// 删除分两种语义：`force = false` 时仍有活跃成员会被拦下，
    /// `force = true` 时才连成员与角色一并清理。软删标记写在 `deleted_at` 上，
    /// 但权限缓存与派生数据必须真的清掉，否则残留的授予会在重建同名组织时复活。
    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, force = force, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn delete_space(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        force: bool,
        reason: Option<&str>,
        request_id: Option<&str>,
    ) -> SpaceResult<bool> {
        let organization = self.load_organization(organization_id).await?;
        if organization.owner_user_id != actor_user_id
            && !self.is_platform_admin(actor_user_id).await?
        {
            return Err(SpaceError::Forbidden);
        }
        let reason = normalize_reason(reason)?;

        if !force {
            let active = space_members::Entity::find()
                .filter(space_members::Column::OrganizationId.eq(organization_id))
                .filter(space_members::Column::LeftAt.is_null())
                .filter(space_members::Column::BannedAt.is_null())
                .count(self.access.database.read())
                .await?;
            if active > 0 {
                return Err(SpaceError::SpaceHasMembers);
            }
        }

        let txn = self.access.database.write().begin().await?;
        let member_ids = space_members::Entity::find()
            .select_only()
            .column(space_members::Column::Id)
            .filter(space_members::Column::OrganizationId.eq(organization_id))
            .into_tuple::<Uuid>()
            .all(&txn)
            .await?;
        if !member_ids.is_empty() {
            space_member_permissions::Entity::delete_many()
                .filter(space_member_permissions::Column::MemberId.is_in(member_ids.clone()))
                .exec(&txn)
                .await?;
            space_member_roles::Entity::delete_many()
                .filter(space_member_roles::Column::MemberId.is_in(member_ids))
                .exec(&txn)
                .await?;
            space_members::Entity::delete_many()
                .filter(space_members::Column::OrganizationId.eq(organization_id))
                .exec(&txn)
                .await?;
        }
        space_invite_redemptions::Entity::delete_many()
            .filter(space_invite_redemptions::Column::OrganizationId.eq(organization_id))
            .exec(&txn)
            .await?;
        space_invites::Entity::delete_many()
            .filter(space_invites::Column::OrganizationId.eq(organization_id))
            .exec(&txn)
            .await?;
        let role_ids = space_roles::Entity::find()
            .select_only()
            .column(space_roles::Column::Id)
            .filter(space_roles::Column::OrganizationId.eq(organization_id))
            .into_tuple::<Uuid>()
            .all(&txn)
            .await?;
        if !role_ids.is_empty() {
            space_role_permissions::Entity::delete_many()
                .filter(space_role_permissions::Column::RoleId.is_in(role_ids))
                .exec(&txn)
                .await?;
        }
        space_roles::Entity::delete_many()
            .filter(space_roles::Column::OrganizationId.eq(organization_id))
            .exec(&txn)
            .await?;
        spaces::Entity::update_many()
            .col_expr(spaces::Column::DeletedAt, Expr::value(Utc::now()))
            .col_expr(spaces::Column::SuspendedReason, Expr::value(reason.clone()))
            .filter(spaces::Column::Id.eq(organization_id))
            .exec(&txn)
            .await?;
        bump_version(&txn, organization_id).await?;
        self.emit_org(
            &txn,
            label::ORGANIZATION_LIFE_CYCLE_DELETE,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": organization_id,
                "force": force,
                "reason": &reason,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

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

    #[tracing::instrument(skip_all, fields(actor_user_id = actor_user_id, organization_id = %organization_id, target_user_id = target_user_id, request_id = %envelope::context::resolve_request_id(request_id)))]
    pub async fn transfer_ownership(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        target_user_id: i64,
        request_id: Option<&str>,
    ) -> SpaceResult<OrganizationSummary> {
        let organization = self.load_organization(organization_id).await?;
        if organization.owner_user_id != actor_user_id
            && !self.is_platform_admin(actor_user_id).await?
        {
            return Err(SpaceError::Forbidden);
        }
        if organization.status != SpaceStatus::Active {
            return Err(SpaceError::SpaceInactive);
        }
        if target_user_id == organization.owner_user_id {
            return Err(SpaceError::SpaceMemberExists);
        }
        let target =
            member::find_active(self.access.database.read(), organization_id, target_user_id)
                .await?
                .ok_or(SpaceError::SpaceMemberNotFound)?;
        let owner_role = role::find_by_key(
            organization_id,
            SYSTEM_ROLE_OWNER,
            self.access.database.read(),
        )
        .await?
        .ok_or_else(|| SpaceError::internal("owner role is missing"))?;
        let admin_role = role::find_by_key(
            organization_id,
            SYSTEM_ROLE_ADMIN,
            self.access.database.read(),
        )
        .await?
        .ok_or_else(|| SpaceError::internal("admin role is missing"))?;

        let txn = self.access.database.write().begin().await?;
        // 唯一 OWNER 的不变式落在两处：spaces.owner_user_id 与成员行的粗粒度角色。
        // 两个改写放在同一事务里，读到中间态的可能性为零。
        if let Some(previous) =
            member::find_any(&txn, organization_id, organization.owner_user_id).await?
        {
            let mut active: space_members::ActiveModel = previous.clone().into();
            active.role = Set(SpaceMemberRole::Admin);
            active.update(&txn).await?;
            role::unbind_role(&txn, previous.id, owner_role.id).await?;
            role::bind_role(&txn, previous.id, admin_role.id, actor_user_id).await?;
        }
        let mut active: space_members::ActiveModel = target.clone().into();
        active.role = Set(SpaceMemberRole::Owner);
        active.update(&txn).await?;
        role::unbind_role(&txn, target.id, admin_role.id).await?;
        role::bind_role(&txn, target.id, owner_role.id, actor_user_id).await?;
        spaces::Entity::update_many()
            .col_expr(spaces::Column::OwnerUserId, Expr::value(target_user_id))
            .filter(spaces::Column::Id.eq(organization_id))
            .exec(&txn)
            .await?;
        let version = bump_version(&txn, organization_id).await?;
        self.emit_org(
            &txn,
            label::ORGANIZATION_MEMBER_UPDATE,
            organization_id,
            actor_user_id,
            serde_json::json!({
                "id": organization_id,
                "action": "transfer_ownership",
                "previous_owner": organization.owner_user_id,
                "new_owner": target_user_id,
                "request_id": request_id,
            }),
            request_id,
        )
        .await?;
        txn.commit().await?;

        self.invalidate_org(organization_id).await?;
        let refreshed = self.load_organization(organization_id).await?;
        let mut summary = self
            .organization_summary(&refreshed, Some(actor_user_id))
            .await?;
        summary.version = version;
        Ok(summary)
    }

    pub async fn load_organization(&self, organization_id: Uuid) -> SpaceResult<spaces::Model> {
        spaces::Entity::find_by_id(organization_id)
            .filter(spaces::Column::DeletedAt.is_null())
            .one(self.access.database.read())
            .await?
            .ok_or(SpaceError::SpaceNotFound)
    }

    /// 一次取齐一页成员的用户名、角色与权限，避免逐行回表。
    pub async fn member_summaries(
        &self,
        rows: &[space_members::Model],
    ) -> SpaceResult<Vec<MemberSummary>> {
        if rows.is_empty() {
            return Ok(Vec::new());
        }
        let member_ids: Vec<Uuid> = rows.iter().map(|row| row.id).collect();
        let user_ids: Vec<i64> = rows.iter().map(|row| row.user_id).collect();

        let users: HashMap<i64, (String, Option<String>)> = users::Entity::find()
            .select_only()
            .column(users::Column::Id)
            .column(users::Column::Username)
            .column(users::Column::Displayname)
            .filter(users::Column::Id.is_in(user_ids))
            .into_tuple::<(i64, String, Option<String>)>()
            .all(self.access.database.read())
            .await?
            .into_iter()
            .map(|(id, username, displayname)| (id, (username, displayname)))
            .collect();

        let bindings = space_member_roles::Entity::find()
            .filter(space_member_roles::Column::MemberId.is_in(member_ids.clone()))
            .all(self.access.database.read())
            .await?;
        let mut role_ids_by_member: HashMap<Uuid, Vec<Uuid>> = HashMap::new();
        let mut role_ids: Vec<Uuid> = Vec::new();
        for binding in bindings {
            role_ids_by_member
                .entry(binding.member_id)
                .or_default()
                .push(binding.role_id);
            if !role_ids.contains(&binding.role_id) {
                role_ids.push(binding.role_id);
            }
        }

        let roles: HashMap<Uuid, space_roles::Model> = if role_ids.is_empty() {
            HashMap::new()
        } else {
            space_roles::Entity::find()
                .filter(space_roles::Column::Id.is_in(role_ids.clone()))
                .all(self.access.database.read())
                .await?
                .into_iter()
                .map(|role| (role.id, role))
                .collect()
        };

        let mut role_grants: HashMap<Uuid, Vec<space_role_permissions::Model>> = HashMap::new();
        if !role_ids.is_empty() {
            let grants = space_role_permissions::Entity::find()
                .filter(space_role_permissions::Column::RoleId.is_in(role_ids))
                .all(self.access.database.read())
                .await?;
            for grant in grants {
                role_grants.entry(grant.role_id).or_default().push(grant);
            }
        }

        let mut direct_grants: HashMap<Uuid, Vec<space_member_permissions::Model>> = HashMap::new();
        let direct = space_member_permissions::Entity::find()
            .filter(space_member_permissions::Column::MemberId.is_in(member_ids))
            .all(self.access.database.read())
            .await?;
        for grant in direct {
            direct_grants
                .entry(grant.member_id)
                .or_default()
                .push(grant);
        }

        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let (username, displayname) = users.get(&row.user_id).cloned().unwrap_or_default();
            let member_role_ids = role_ids_by_member
                .get(&row.id)
                .map(Vec::as_slice)
                .unwrap_or_default();
            let mut role_keys: Vec<(i32, String)> = member_role_ids
                .iter()
                .filter_map(|role_id| roles.get(role_id))
                .map(|role| (role.rank, role.key.clone()))
                .collect();
            role_keys.sort_by_key(|(rank, _)| *rank);
            let kernel = permission::kernel_from(
                0,
                row,
                role_keys.into_iter().map(|(_, key)| key).collect(),
                member_role_ids
                    .iter()
                    .filter_map(|role_id| role_grants.get(role_id))
                    .flatten(),
                direct_grants.get(&row.id).into_iter().flatten(),
            );
            out.push(MemberSummary {
                id: row.id,
                organization_id: row.organization_id,
                user_id: row.user_id,
                username,
                displayname,
                role: row.role,
                role_keys: kernel.role_keys.clone(),
                permissions: permission::allowed_list(&kernel),
                joined_at: row.joined_at,
                left_at: row.left_at,
                banned_at: row.banned_at,
                invited_by: row.invited_by,
                join_method: row.join_method,
                alias_name: row.alias_name.clone(),
                alias_avatar_url: row.alias_avatar_url.clone(),
                message_notifications: row.message_notifications,
                muted: row.muted,
                mute_until: row.mute_until,
                mobile_push: row.mobile_push,
                suppress_everyone: row.suppress_everyone,
                suppress_roles: row.suppress_roles,
                hide_muted_channels: row.hide_muted_channels,
                ban_reason: row.ban_reason.clone(),
                ban_expires_at: row.ban_expires_at,
                ban_allow_agent_join: row.ban_allow_agent_join,
                ban_allow_speak: row.ban_allow_speak,
                ban_allow_push_repo: row.ban_allow_push_repo,
                ban_allow_comment_any: row.ban_allow_comment_any,
                version: row.version,
                last_active_at: row.last_active_at,
            });
        }
        Ok(out)
    }

    pub async fn member_summary(&self, row: &space_members::Model) -> SpaceResult<MemberSummary> {
        self.member_summaries(std::slice::from_ref(row))
            .await?
            .into_iter()
            .next()
            .ok_or(SpaceError::SpaceMemberNotFound)
    }

    pub(crate) async fn role_summary(&self, row: &space_roles::Model) -> SpaceResult<RoleSummary> {
        let permissions = role::grants_of(self.access.database.read(), row.id).await?;
        let member_count = space_member_roles::Entity::find()
            .filter(space_member_roles::Column::RoleId.eq(row.id))
            .count(self.access.database.read())
            .await? as i64;
        Ok(RoleSummary {
            id: row.id,
            organization_id: row.organization_id,
            key: row.key.clone(),
            name: row.name.clone(),
            description: row.description.clone(),
            rank: row.rank,
            system: row.system,
            default_role: row.default_role,
            self_assignable: row.self_assignable,
            permissions,
            member_count,
            created_at: row.created_at,
            updated_at: row.updated_at,
            version: row.version,
            color: row.color.clone(),
            emoji: row.emoji.clone(),
            icon_hash: row.icon_hash.clone(),
            position: row.position,
        })
    }

    pub(crate) async fn default_role_id(&self, organization_id: Uuid) -> SpaceResult<Option<Uuid>> {
        Ok(space_roles::Entity::find()
            .filter(space_roles::Column::OrganizationId.eq(organization_id))
            .filter(space_roles::Column::DefaultRole.eq(true))
            .order_by_asc(space_roles::Column::Rank)
            .one(self.access.database.read())
            .await?
            .map(|row| row.id))
    }

    pub(crate) fn validate_grants(grants: &[PermissionGrant]) -> SpaceResult<Vec<PermissionGrant>> {
        let mut out = Vec::with_capacity(grants.len());
        for grant in grants {
            let permission = grant.permission.trim();
            if permission.is_empty()
                || permission.len() > MAX_PERMISSION_LEN
                || !is_known_permission(permission)
            {
                return Err(SpaceError::InvalidPermission {
                    permission: grant.permission.clone(),
                });
            }
            out.push(PermissionGrant {
                permission: permission.to_owned(),
                effect: match grant.effect {
                    PermissionEffect::Deny => PermissionEffect::Deny,
                    _ => PermissionEffect::Allow,
                },
            });
        }
        Ok(out)
    }
}

/// 在写事务内推进 version 并回读新值。
pub async fn bump_version<C>(db: &C, organization_id: Uuid) -> SpaceResult<i32>
where
    C: sea_orm::ConnectionTrait,
{
    let bumped = spaces::Entity::update_many()
        .col_expr(
            spaces::Column::Version,
            Expr::col(spaces::Column::Version).add(1),
        )
        .col_expr(spaces::Column::UpdatedAt, Expr::value(Utc::now()))
        .filter(spaces::Column::Id.eq(organization_id))
        .exec(db)
        .await?;
    if bumped.rows_affected == 0 {
        return Err(SpaceError::SpaceNotFound);
    }
    spaces::Entity::find_by_id(organization_id)
        .select_only()
        .column(spaces::Column::Version)
        .into_tuple::<i32>()
        .one(db)
        .await?
        .ok_or(SpaceError::SpaceNotFound)
}

pub(crate) fn normalize_slug(raw: &str) -> SpaceResult<String> {
    let slug = raw.trim().to_lowercase();
    if slug.is_empty()
        || slug.len() > MAX_SLUG_LEN
        || !slug
            .chars()
            .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '-')
        || slug.starts_with('-')
        || slug.ends_with('-')
    {
        return Err(SpaceError::MalformedRequest);
    }
    Ok(slug)
}

pub(crate) fn normalize_name(raw: &str) -> SpaceResult<String> {
    let name = raw.trim();
    if name.is_empty() || name.len() > MAX_ORG_NAME_LEN {
        return Err(SpaceError::MalformedRequest);
    }
    Ok(name.to_owned())
}

pub(crate) fn normalize_description(raw: Option<&str>) -> SpaceResult<Option<String>> {
    match raw.map(str::trim) {
        None | Some("") => Ok(None),
        Some(value) if value.len() > MAX_ORG_DESCRIPTION_LEN => Err(SpaceError::MalformedRequest),
        Some(value) => Ok(Some(value.to_owned())),
    }
}

pub(crate) fn normalize_avatar(raw: Option<&str>) -> SpaceResult<Option<String>> {
    match raw.map(str::trim) {
        None | Some("") => Ok(None),
        Some(value) if value.len() > 512 || !value.starts_with("https://") => {
            Err(SpaceError::MalformedRequest)
        }
        Some(value) => Ok(Some(value.to_owned())),
    }
}

pub(crate) fn normalize_reason(raw: Option<&str>) -> SpaceResult<Option<String>> {
    match raw.map(str::trim) {
        None | Some("") => Ok(None),
        Some(value) if value.len() > 512 => Err(SpaceError::MalformedRequest),
        Some(value) => Ok(Some(value.to_owned())),
    }
}

pub(crate) fn normalize_role_key(raw: &str) -> SpaceResult<String> {
    let key = raw.trim().to_lowercase();
    if key.is_empty()
        || key.len() > MAX_ROLE_KEY_LEN
        || !key
            .chars()
            .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '_' || ch == '-')
    {
        return Err(SpaceError::MalformedRequest);
    }
    Ok(key)
}

pub(crate) fn normalize_role_name(raw: &str) -> SpaceResult<String> {
    let name = raw.trim();
    if name.is_empty() || name.len() > MAX_ROLE_NAME_LEN {
        return Err(SpaceError::MalformedRequest);
    }
    Ok(name.to_owned())
}

pub(crate) fn valid_join_level(value: i16) -> SpaceResult<i16> {
    (0..=4)
        .contains(&value)
        .then_some(value)
        .ok_or(SpaceError::MalformedRequest)
}

#[derive(Clone)]
pub struct SpaceSettings {
    pub space: config::space::SpaceConfig,
    pub quota: config::quota::QuotaConfig,
    pub email: config::email::EmailConfig,
}
impl SpaceSettings {
    pub fn from_config(config: &AppConfig) -> Self {
        Self {
            space: config.space.clone(),
            quota: config.quota.clone(),
            email: config.email.clone(),
        }
    }
}

impl SpaceService {
    pub async fn allow_room_creation(&self, caller: i64) -> Result<(), crate::errors::SpaceError> {
        self.policy.allow_room_creation(caller).await
    }
}