072c704
use std::sync::Arc;
use actor::ActorHost;
use internal::envelope::{EnvelopeVerifier, VerifiedIdentity};
use socket::SocketHub;
use crate::gateway::action::EventProperties;
use crate::gateway::platform::PlatformClient;
use crate::gateway::subscriptions::{Subscription, SubscriptionRequest, Subscriptions};
use crate::gateway::{ConsumeResult, is_personal};
/// 事件域唯一业务入口：socket 路径与内部 RPC 路径共用同一套订阅/授权/投递逻辑。
#[derive(Clone)]
pub struct EventService {
host: ActorHost,
platform: PlatformClient,
subscriptions: Subscriptions,
verifier: Arc<EnvelopeVerifier>,
hub: SocketHub,
}
impl EventService {
pub fn new(
host: ActorHost,
platform: PlatformClient,
subscriptions: Subscriptions,
verifier: Arc<EnvelopeVerifier>,
hub: SocketHub,
) -> Self {
Self {
host,
platform,
subscriptions,
verifier,
hub,
}
}
pub fn host(&self) -> &ActorHost {
&self.host
}
pub fn platform(&self) -> &PlatformClient {
&self.platform
}
pub fn subscriptions(&self) -> &Subscriptions {
&self.subscriptions
}
/// 解密身份信封 + 平台权威复核。
pub async fn verify_identity(
&self,
envelope: &proto::internal::v1::IdentityEnvelope,
) -> Result<VerifiedIdentity, tonic::Status> {
let identity = self.verifier.verify(envelope).await?;
self.platform
.attest(
identity.user_id,
identity.security_version,
&identity.request_id,
)
.await
.map_err(|error| error.into_status())?;
Ok(identity)
}
/// 订阅（浏览器 socket 与内部 RPC 共用）。
pub async fn subscribe(
&self,
user_id: i64,
device_id: &str,
namespace: &str,
request: SubscriptionRequest,
) -> Result<String, String> {
self.subscriptions
.subscribe(&self.platform, user_id, device_id, namespace, request)
.await
}
pub async fn unsubscribe(
&self,
user_id: i64,
device_id: &str,
namespace: &str,
topic: &str,
tag: &str,
id: &str,
) -> Result<(), String> {
self.subscriptions
.unsubscribe(user_id, device_id, namespace, topic, tag, id)
.await
}
pub async fn list(
&self,
user_id: i64,
device_id: &str,
namespace: &str,
) -> Result<Vec<Subscription>, String> {
self.subscriptions.list(user_id, device_id, namespace).await
}
/// 平台定向投递：身份已校验，需再过订阅匹配 + 投递期授权。
pub async fn publish(
&self,
identity: &VerifiedIdentity,
topic: &str,
tag: &str,
properties: EventProperties,
body: serde_json::Value,
personal: bool,
request_id: &str,
) -> Result<u32, tonic::Status> {
if !crate::gateway::action::is_supported(topic, tag) {
return Err(tonic::Status::invalid_argument("unsupported topic or tag"));
}
let personal = personal || is_personal(topic, tag);
let recipient = identity.user_id;
if personal {
if properties
.user_id
.as_deref()
.is_some_and(|value| value != recipient.to_string())
{
return Err(tonic::Status::permission_denied(
"personal events must target the identity user",
));
}
let payload = crate::gateway::action::common_payload(
topic,
tag,
Some(recipient),
recipient,
&properties,
&body,
None,
);
self.host
.publish_event(recipient, payload)
.await
.map_err(|_| tonic::Status::unavailable("delivery store is unavailable"))?;
return Ok(1);
}
let subscribers = self
.subscriptions
.subscribers(topic, tag, &properties)
.await
.map_err(|_| tonic::Status::failed_precondition("subscription store is unavailable"))?;
let mut delivered = 0;
for subscription in subscribers
.into_iter()
.filter(|subscription| subscription.user_id == recipient)
.filter(|subscription| subscription.filter.matches(&properties))
{
let allowed = self
.platform
.authorize(recipient, &subscription.filter, "DELIVER")
.await
.map_err(|error| error.into_status())?;
if !allowed {
continue;
}
if let Some(org) = subscription.filter.org_id.as_deref() {
if !self
.platform
.is_member(recipient, org)
.await
.map_err(|error| error.into_status())?
{
continue;
}
}
let payload = crate::gateway::action::common_payload(
topic,
tag,
properties
.user_id
.as_deref()
.and_then(|value| value.parse::<i64>().ok()),
recipient,
&properties,
&body,
Some(vec![subscription.clone()]),
);
if self.host.publish_event(recipient, payload).await.is_ok() {
delivered += 1;
}
}
let _ = request_id;
Ok(delivered)
}
/// 撤销连接/权限：发控制事件并关闭该用户在线会话。
pub async fn revoke(
&self,
identity: &VerifiedIdentity,
reason: &str,
) -> Result<u32, tonic::Status> {
let payload = serde_json::json!({
"topic": "event",
"tag": "access-revoked",
"control": true,
"event_id": uuid::Uuid::now_v7().to_string(),
"recipient_user_id": identity.user_id,
"properties": {"user_id": identity.user_id},
"body": {"reason": reason},
});
let delivered = self.host.publish_event(identity.user_id, payload).await;
let closed = self.hub.close_user(identity.user_id).await as u32;
delivered.map_err(|_| tonic::Status::unavailable("delivery store is unavailable"))?;
Ok(closed)
}
pub fn consume_result(_: ConsumeResult) {}
}