main
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) {}
}