use std::fmt;
use std::time::Duration;

use config::event::platform::PlatformClientConfig;
use internal::Connection;
use proto::internal::v1::identity_service_client::IdentityServiceClient;
use proto::internal::v1::{
    AttestIdentityRequest, CredentialKind, PublishRequest, ResolveIdentityRequest,
};

use crate::gateway::action::EventProperties;

#[derive(Debug, Clone)]
pub enum PlatformError {
    Config(String),
    Rejected(String),
    Unavailable(String),
}

impl fmt::Display for PlatformError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Config(message) => write!(f, "config: {message}"),
            Self::Rejected(message) => write!(f, "rejected: {message}"),
            Self::Unavailable(message) => write!(f, "unavailable: {message}"),
        }
    }
}

impl std::error::Error for PlatformError {}

impl PlatformError {
    pub fn into_status(self) -> tonic::Status {
        match self {
            Self::Config(message) => tonic::Status::failed_precondition(message),
            Self::Rejected(message) => tonic::Status::permission_denied(message),
            Self::Unavailable(_) => tonic::Status::unavailable("platform service is unavailable"),
        }
    }
}

impl From<tonic::Status> for PlatformError {
    fn from(status: tonic::Status) -> Self {
        match status.code() {
            tonic::Code::Unauthenticated
            | tonic::Code::PermissionDenied
            | tonic::Code::InvalidArgument => Self::Rejected(status.message().to_string()),
            _ => Self::Unavailable(status.message().to_string()),
        }
    }
}

/// 事件进程对基础核心平台的唯一出口。
#[derive(Clone)]
pub struct PlatformClient {
    rpc: Connection,
    http: reqwest::Client,
    resource_api_endpoint: String,
    resource_service_key: String,
    timeout: Duration,
}

impl PlatformClient {
    pub fn new(config: &PlatformClientConfig) -> Result<Self, PlatformError> {
        config.validate().map_err(PlatformError::Config)?;
        let timeout = Duration::from_secs(config.timeout_secs.max(1));
        let rpc = Connection::connect(&config.rpc_endpoint, &config.rpc_key, timeout)
            .map_err(|error| PlatformError::Config(error.to_string()))?;
        let http = reqwest::Client::builder()
            .timeout(timeout)
            .build()
            .map_err(|error| PlatformError::Config(error.to_string()))?;
        Ok(Self {
            rpc,
            http,
            resource_api_endpoint: config
                .resource_api_endpoint
                .trim_end_matches('/')
                .to_string(),
            resource_service_key: config.resource_api_service_key.trim().to_string(),
            timeout,
        })
    }

    pub fn rpc(&self) -> &Connection {
        &self.rpc
    }

    /// KenApi 会话 cookie 的值 → user_id。
    pub async fn authenticate(&self, credential: &str) -> Result<i64, PlatformError> {
        if credential.is_empty() {
            return Err(PlatformError::Rejected(
                "session credential is empty".to_string(),
            ));
        }
        let mut client =
            IdentityServiceClient::with_interceptor(self.rpc.channel(), self.rpc.interceptor());
        let request = tonic::Request::new(ResolveIdentityRequest {
            kind: CredentialKind::Session as i32,
            credential: credential.to_string(),
            required_scopes: vec!["event".to_string()],
            request_id: String::new(),
        });
        let response = tokio::time::timeout(self.timeout, client.resolve(request))
            .await
            .map_err(|_| PlatformError::Unavailable("identity resolve timed out".to_string()))??;
        response
            .into_inner()
            .principal
            .map(|principal| principal.user_id)
            .ok_or_else(|| PlatformError::Rejected("identity has no principal".to_string()))
    }

    /// 信封身份的权威复核（用户仍有效、安全版本一致）。
    pub async fn attest(
        &self,
        user_id: i64,
        security_version: i64,
        request_id: &str,
    ) -> Result<(), PlatformError> {
        let mut client =
            IdentityServiceClient::with_interceptor(self.rpc.channel(), self.rpc.interceptor());
        let request = tonic::Request::new(AttestIdentityRequest {
            user_id,
            security_version,
            request_id: request_id.to_string(),
        });
        let response = tokio::time::timeout(self.timeout, client.attest(request))
            .await
            .map_err(|_| PlatformError::Unavailable("identity attest timed out".to_string()))??;
        let response = response.into_inner();
        if response.active {
            Ok(())
        } else {
            Err(PlatformError::Rejected(
                "identity is no longer active".to_string(),
            ))
        }
    }

    /// 资源订阅/投递授权：按 filter 命中哪个域，就调哪个域的授权 RPC。
    pub async fn authorize(
        &self,
        user_id: i64,
        filter: &EventProperties,
        action: &str,
    ) -> Result<bool, PlatformError> {
        if let Some(room_id) = filter.room_id.as_deref() {
            let mut client = proto::internal::v1::chat_authorization_service_client::ChatAuthorizationServiceClient::with_interceptor(
                self.rpc.channel(),
                self.rpc.interceptor(),
            );
            let request = tonic::Request::new(proto::internal::v1::AuthorizeRoomRequest {
                user_id,
                room_id: room_id.to_string(),
                action: action.to_string(),
                request_id: String::new(),
            });
            return Ok(
                tokio::time::timeout(self.timeout, client.authorize_room(request))
                    .await
                    .map_err(|_| {
                        PlatformError::Unavailable("room authorization timed out".to_string())
                    })??
                    .into_inner()
                    .allowed,
            );
        }
        if let Some(issue_id) = filter.issue_id.as_deref() {
            let mut client = proto::internal::v1::issue_authorization_service_client::IssueAuthorizationServiceClient::with_interceptor(
                self.rpc.channel(),
                self.rpc.interceptor(),
            );
            let request = tonic::Request::new(proto::internal::v1::AuthorizeIssueRequest {
                user_id,
                issue_id: issue_id.to_string(),
                action: action.to_string(),
                request_id: String::new(),
            });
            return Ok(
                tokio::time::timeout(self.timeout, client.authorize_issue(request))
                    .await
                    .map_err(|_| {
                        PlatformError::Unavailable("issue authorization timed out".to_string())
                    })??
                    .into_inner()
                    .allowed,
            );
        }
        if let Some(pull_id) = filter.pull_id.as_deref() {
            let mut client = proto::internal::v1::pull_authorization_service_client::PullAuthorizationServiceClient::with_interceptor(
                self.rpc.channel(),
                self.rpc.interceptor(),
            );
            let request = tonic::Request::new(proto::internal::v1::AuthorizePullRequest {
                user_id,
                pull_id: pull_id.to_string(),
                action: action.to_string(),
                request_id: String::new(),
            });
            return Ok(
                tokio::time::timeout(self.timeout, client.authorize_pull(request))
                    .await
                    .map_err(|_| {
                        PlatformError::Unavailable("pull authorization timed out".to_string())
                    })??
                    .into_inner()
                    .allowed,
            );
        }
        if let Some(repo_id) = filter.repo_id.as_deref() {
            let mut client = proto::internal::v1::space_authorization_service_client::SpaceAuthorizationServiceClient::with_interceptor(
                self.rpc.channel(),
                self.rpc.interceptor(),
            );
            let request = tonic::Request::new(proto::internal::v1::AuthorizeResourceRequest {
                user_id,
                repo_id: repo_id.to_string(),
                space_id: String::new(),
                action: action.to_string(),
                request_id: String::new(),
            });
            return Ok(
                tokio::time::timeout(self.timeout, client.authorize_repo(request))
                    .await
                    .map_err(|_| {
                        PlatformError::Unavailable("repo authorization timed out".to_string())
                    })??
                    .into_inner()
                    .allowed,
            );
        }
        // 没有资源维度的订阅（personal / cross-user）由调用方各自校验。
        Ok(true)
    }

    /// 组织成员校验：组织数据在 KenApi 的 space 域（space id == org id）。
    pub async fn is_member(
        &self,
        user_id: i64,
        organization_id: &str,
    ) -> Result<bool, PlatformError> {
        let mut client = proto::internal::v1::space_authorization_service_client::SpaceAuthorizationServiceClient::with_interceptor(
            self.rpc.channel(),
            self.rpc.interceptor(),
        );
        let request = tonic::Request::new(proto::internal::v1::AuthorizeResourceRequest {
            user_id,
            repo_id: String::new(),
            space_id: organization_id.to_string(),
            action: "DELIVER".to_string(),
            request_id: String::new(),
        });
        Ok(
            tokio::time::timeout(self.timeout, client.authorize_space(request))
                .await
                .map_err(|_| {
                    PlatformError::Unavailable("organization authorization timed out".to_string())
                })??
                .into_inner()
                .allowed,
        )
    }

    /// 组织成员 id（事件扇出用）：组织数据在 KenApi space 域。
    pub async fn member_ids(&self, organization_id: &str) -> Result<Vec<i64>, PlatformError> {
        let mut client = proto::internal::v1::space_authorization_service_client::SpaceAuthorizationServiceClient::with_interceptor(
            self.rpc.channel(),
            self.rpc.interceptor(),
        );
        let request = tonic::Request::new(proto::internal::v1::SpaceMembersRequest {
            space_id: organization_id.to_string(),
            request_id: String::new(),
        });
        Ok(
            tokio::time::timeout(self.timeout, client.list_members(request))
                .await
                .map_err(|_| {
                    PlatformError::Unavailable("organization members timed out".to_string())
                })??
                .into_inner()
                .user_ids,
        )
    }

    /// 上报聊天连接生命周期（平台据此维护在线态、抑制 Web Push）。
    pub async fn report_connection(
        &self,
        phase: &str,
        user_id: i64,
        session: &str,
    ) -> Result<(), PlatformError> {
        if self.resource_service_key.is_empty() {
            return Ok(());
        }
        let path = match phase {
            "open" => "/internal/chat/connections",
            "heartbeat" => "/internal/chat/connections/heartbeat",
            "close" => "/internal/chat/connections/close",
            _ => return Ok(()),
        };
        let response = self
            .http
            .post(format!("{}{path}", self.resource_api_endpoint))
            .header("content-type", "application/json")
            .header("x-gitdata-service-key", &self.resource_service_key)
            .json(&serde_json::json!({"user_id": user_id, "session_id": session}))
            .send()
            .await
            .map_err(|error| PlatformError::Unavailable(error.to_string()))?;
        if response.status().is_success() {
            Ok(())
        } else {
            Err(PlatformError::Unavailable(format!(
                "chat connection report returned {}",
                response.status()
            )))
        }
    }

    pub fn publish_client(
        &self,
    ) -> proto::internal::v1::event_ingress_service_client::EventIngressServiceClient<
        tonic::service::interceptor::InterceptedService<
            tonic::transport::Channel,
            internal::RpcInterceptor,
        >,
    > {
        proto::internal::v1::event_ingress_service_client::EventIngressServiceClient::with_interceptor(
            self.rpc.channel(),
            self.rpc.interceptor(),
        )
    }

    pub fn _unused_publish(_: PublishRequest) {}
}