use std::time::Duration;

use sea_orm::{ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QuerySelect};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::errors::{SpaceError, SpaceResult};
use crate::services::service::SpaceService;
use crate::services::types::{
    PERMISSION_CACHE_TTL_SECS, PERMISSIONS, PermissionCheck, Resolution, SYSTEM_ROLE_ADMIN,
    SYSTEM_ROLE_OWNER, is_admin_permission, is_known_permission,
};
use crate::services::{member, role, service};
use models::space::{
    PermissionEffect, SpaceMemberRole, space_member_permissions, space_member_roles, space_members,
    space_role_permissions,
};

/// 缓存进 Redis 的判定核心。版本号相同的组织共享同一份核心，
/// 任何写操作推进 `spaces.version` 都会让整份核心失效。
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub(crate) struct Kernel {
    pub version: i32,
    pub is_owner: bool,
    pub is_admin: bool,
    pub role_keys: Vec<String>,
    pub role_allow: Vec<String>,
    pub role_deny: Vec<String>,
    pub direct_allow: Vec<String>,
    pub direct_deny: Vec<String>,
}

fn push_unique(target: &mut Vec<String>, value: &str) {
    if !target.iter().any(|existing| existing == value) {
        target.push(value.to_owned());
    }
}

pub(crate) fn kernel_from<'a>(
    version: i32,
    membership: &space_members::Model,
    role_keys: Vec<String>,
    role_grants: impl IntoIterator<Item = &'a space_role_permissions::Model>,
    direct_grants: impl IntoIterator<Item = &'a space_member_permissions::Model>,
) -> Kernel {
    let is_owner = membership.role == SpaceMemberRole::Owner
        || role_keys.iter().any(|key| key == SYSTEM_ROLE_OWNER);
    let is_admin = is_owner
        || membership.role == SpaceMemberRole::Admin
        || role_keys.iter().any(|key| key == SYSTEM_ROLE_ADMIN);

    let mut kernel = Kernel {
        version,
        is_owner,
        is_admin,
        role_keys,
        ..Default::default()
    };

    for grant in role_grants {
        if !is_known_permission(&grant.permission) {
            continue;
        }
        match grant.effect {
            PermissionEffect::Deny => push_unique(&mut kernel.role_deny, &grant.permission),
            PermissionEffect::Allow => push_unique(&mut kernel.role_allow, &grant.permission),
        }
    }

    for grant in direct_grants {
        if !is_known_permission(&grant.permission) {
            continue;
        }
        match grant.effect {
            PermissionEffect::Deny => push_unique(&mut kernel.direct_deny, &grant.permission),
            PermissionEffect::Allow => push_unique(&mut kernel.direct_allow, &grant.permission),
        }
    }

    kernel
}

pub(crate) async fn load_kernel<C: ConnectionTrait>(
    db: &C,
    version: i32,
    membership: &space_members::Model,
) -> SpaceResult<Kernel> {
    let role_ids = space_member_roles::Entity::find()
        .select_only()
        .column(space_member_roles::Column::RoleId)
        .filter(space_member_roles::Column::MemberId.eq(membership.id))
        .into_tuple::<Uuid>()
        .all(db)
        .await?;
    let role_keys = role::keys_of_ids(db, &role_ids).await?;

    let mut role_grants = Vec::new();
    if !role_ids.is_empty() {
        role_grants = space_role_permissions::Entity::find()
            .filter(space_role_permissions::Column::RoleId.is_in(role_ids))
            .all(db)
            .await?;
    }

    let direct = space_member_permissions::Entity::find()
        .filter(space_member_permissions::Column::MemberId.eq(membership.id))
        .all(db)
        .await?;

    Ok(kernel_from(
        version,
        membership,
        role_keys,
        &role_grants,
        &direct,
    ))
}

fn decide(kernel: &Kernel, permission: &str) -> (&'static str, &'static str) {
    if !is_known_permission(permission) {
        return ("unknown_permission", "default_deny");
    }
    if kernel.is_owner {
        return ("", "owner");
    }
    if kernel.is_admin && is_admin_permission(permission) {
        return ("", "admin");
    }
    // 直挂层优先于角色层，层内 DENY 优先。两层都显式覆盖，
    // 这样"角色给了、成员被单独收回"和"成员单独补授"都能表达。
    if kernel.direct_deny.iter().any(|key| key == permission) {
        return ("direct_deny", "direct");
    }
    if kernel.direct_allow.iter().any(|key| key == permission) {
        return ("", "direct");
    }
    if kernel.role_deny.iter().any(|key| key == permission) {
        return ("role_deny", "role");
    }
    if kernel.role_allow.iter().any(|key| key == permission) {
        return ("", "role");
    }
    ("no_grant", "default_deny")
}

pub(crate) fn allowed_list(kernel: &Kernel) -> Vec<String> {
    PERMISSIONS
        .iter()
        .filter(|permission| decide(kernel, permission).0.is_empty())
        .map(|permission| (*permission).to_owned())
        .collect()
}

async fn cached_kernel(
    db: &db::DbConn,
    cache: &cache::Cache,
    organization_id: Uuid,
    user_id: i64,
    version: i32,
    membership: &space_members::Model,
) -> SpaceResult<Kernel> {
    let key = service::perm_cache_prefix(organization_id, user_id, version);
    if let Ok(Some(bytes)) = cache.get_bytes(&key).await
        && let Ok(kernel) = serde_json::from_slice::<Kernel>(&bytes)
        && kernel.version == version
    {
        return Ok(kernel);
    }
    let kernel = load_kernel(db.read(), version, membership).await?;
    if let Ok(bytes) = serde_json::to_vec(&kernel)
        && let Err(error) = cache
            .set_ex(&key, bytes, Duration::from_secs(PERMISSION_CACHE_TTL_SECS))
            .await
    {
        tracing::warn!(%error, %organization_id, user_id, "failed to cache the permission kernel");
    }
    Ok(kernel)
}

/// 前置闸门：组织状态 + 活跃会籍 + 判定核心。批量校验只走一次。
enum Gate {
    Denied {
        version: i32,
        reason: &'static str,
        matched_by: &'static str,
    },
    Ready {
        version: i32,
        kernel: Kernel,
    },
}

impl Gate {
    fn version(&self) -> i32 {
        match self {
            Self::Denied { version, .. } | Self::Ready { version, .. } => *version,
        }
    }

    fn resolution(&self, permission: &str) -> Resolution {
        match self {
            Self::Denied {
                reason, matched_by, ..
            } => Resolution {
                allowed: false,
                reason,
                matched_by,
                effective: Vec::new(),
            },
            Self::Ready { kernel, .. } => {
                let (reason, matched_by) = decide(kernel, permission);
                Resolution {
                    allowed: reason.is_empty(),
                    reason,
                    matched_by,
                    effective: allowed_list(kernel),
                }
            }
        }
    }
}

async fn gate(
    db: &db::DbConn,
    cache: &cache::Cache,
    organization_id: Uuid,
    user_id: i64,
    _request_id: Option<&str>,
) -> SpaceResult<Gate> {
    let organization = models::space::spaces::Entity::find_by_id(organization_id)
        .one(db.read())
        .await?
        .filter(|row| row.deleted_at.is_none())
        .ok_or(SpaceError::NotFound)?;
    let version = organization.version;
    if organization.status != models::space::SpaceStatus::Active
        || organization.archived_at.is_some()
    {
        return Ok(Gate::Denied {
            version,
            reason: "org_inactive",
            matched_by: "org_status",
        });
    }
    if organization.owner_user_id == user_id {
        return Ok(Gate::Ready {
            version,
            kernel: Kernel {
                version,
                is_owner: true,
                ..Default::default()
            },
        });
    }
    let membership = match member::find_any(db.read(), organization_id, user_id).await? {
        Some(row) => row,
        None => {
            return Ok(Gate::Denied {
                version,
                reason: "not_member",
                matched_by: "membership",
            });
        }
    };
    if membership.banned_at.is_some() {
        return Ok(Gate::Denied {
            version,
            reason: "member_banned",
            matched_by: "membership",
        });
    }
    if membership.left_at.is_some() {
        return Ok(Gate::Denied {
            version,
            reason: "member_left",
            matched_by: "membership",
        });
    }

    let kernel = cached_kernel(db, cache, organization_id, user_id, version, &membership).await?;
    Ok(Gate::Ready { version, kernel })
}

/// 唯一判定入口。顺序固定：组织状态 → 活跃会籍 → Owner → Admin 固有集
/// → 角色 Allow/Deny → 成员直挂 Allow/Deny。
#[tracing::instrument(skip_all, fields(organization_id = %organization_id, user_id = user_id, permission = %permission, request_id = %envelope::context::resolve_request_id(request_id)))]
pub(crate) async fn resolve(
    svc: &SpaceService,
    organization_id: Uuid,
    user_id: i64,
    permission: &str,
    request_id: Option<&str>,
) -> SpaceResult<Resolution> {
    let gate = gate(
        &svc.access.database,
        &svc.access.cache,
        organization_id,
        user_id,
        request_id,
    )
    .await?;
    Ok(gate.resolution(permission))
}

impl SpaceService {
    /// 平台管理员越过会籍判定，但仍受组织状态约束。
    #[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 check_permission(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        user_id: i64,
        permission: &str,
        request_id: Option<&str>,
    ) -> SpaceResult<PermissionCheck> {
        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 gate = gate(
            &self.access.database,
            &self.access.cache,
            organization_id,
            target,
            request_id,
        )
        .await?;
        let resolution = gate.resolution(permission.trim());
        Ok(PermissionCheck {
            allowed: resolution.allowed,
            reason: resolution.reason.to_owned(),
            matched_by: resolution.matched_by.to_owned(),
            organization_version: gate.version(),
        })
    }

    /// 批量校验：组织与会籍只加载一次，权限逐个走同一条判定链。
    #[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 check_permissions(
        &self,
        actor_user_id: i64,
        organization_id: Uuid,
        user_id: i64,
        permissions: &[String],
        request_id: Option<&str>,
    ) -> SpaceResult<Vec<PermissionCheck>> {
        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 gate = gate(
            &self.access.database,
            &self.access.cache,
            organization_id,
            target,
            request_id,
        )
        .await?;
        let version = gate.version();
        Ok(permissions
            .iter()
            .map(|permission| {
                let resolution = gate.resolution(permission.trim());
                PermissionCheck {
                    allowed: resolution.allowed,
                    reason: resolution.reason.to_owned(),
                    matched_by: resolution.matched_by.to_owned(),
                    organization_version: version,
                }
            })
            .collect())
    }
}

pub(crate) async fn effective_for(
    db: &db::DbConn,
    cache: &cache::Cache,
    space_id: Uuid,
    user_id: i64,
) -> SpaceResult<Vec<String>> {
    match gate(db, cache, space_id, user_id, None).await? {
        Gate::Ready { kernel, .. } => Ok(allowed_list(&kernel)),
        Gate::Denied { .. } => Ok(Vec::new()),
    }
}

pub(crate) async fn invalidate(cache: &cache::Cache, organization_id: Uuid) -> SpaceResult<()> {
    for key in [
        format!("auth:org:{organization_id}"),
        format!("auth:org:summary:{organization_id}"),
    ] {
        if let Err(error) = cache.delete(&key).await {
            tracing::warn!(%error, %organization_id, "failed to invalidate organization cache");
        }
    }
    Ok(())
}