072c704
//! 内部 RPC 的 event 域：EventIngress + Subscription，方法挂在 `EventService` 上。
use internal::Gate;
use internal::server::InternalRouter;
use proto::internal::v1::event_ingress_service_server::{
EventIngressService as EventIngressTrait, EventIngressServiceServer,
};
use proto::internal::v1::subscription_service_server::{
SubscriptionService as SubscriptionTrait, SubscriptionServiceServer,
};
use proto::internal::v1::{
EventFilter, ListSubscriptionsRequest, ListSubscriptionsResponse, PublishRequest,
PublishResponse, RevokeRequest, RevokeResponse, SubscribeRequest, SubscribeResponse,
SubscriptionView, UnsubscribeRequest, UnsubscribeResponse,
};
use tonic::{Request, Response, Status};
use crate::gateway::action::EventProperties;
use crate::gateway::event_service::EventService;
use crate::gateway::subscriptions::SubscriptionRequest;
pub fn register_ingress(
router: InternalRouter,
service: &EventService,
gate: Gate,
) -> InternalRouter {
router.add_service(EventIngressServiceServer::with_interceptor(
IngressRpc {
service: service.clone(),
},
gate,
))
}
pub fn register(router: InternalRouter, service: &EventService, gate: Gate) -> InternalRouter {
router.add_service(SubscriptionServiceServer::with_interceptor(
SubscriptionRpc {
service: service.clone(),
},
gate,
))
}
#[derive(Clone)]
struct IngressRpc {
service: EventService,
}
#[derive(Clone)]
struct SubscriptionRpc {
service: EventService,
}
#[tonic::async_trait]
impl EventIngressTrait for IngressRpc {
async fn publish(
&self,
request: Request<PublishRequest>,
) -> Result<Response<PublishResponse>, Status> {
let meta = internal::interceptor::meta(&request)?;
let request = request.into_inner();
let Some(envelope) = request.identity.as_ref() else {
return Err(Status::unauthenticated("identity envelope is required"));
};
let identity = self.service.verify_identity(envelope).await?;
let body = if request.body.is_empty() {
serde_json::Value::Null
} else {
serde_json::from_slice(&request.body).unwrap_or(serde_json::Value::Null)
};
let properties = filter_to_properties(request.filter);
let delivered = self
.service
.publish(
&identity,
&request.topic,
&request.tag,
properties,
body,
request.personal,
meta.request_id.as_str(),
)
.await?;
Ok(Response::new(PublishResponse {
delivered,
scope: String::new(),
}))
}
async fn revoke(
&self,
request: Request<RevokeRequest>,
) -> Result<Response<RevokeResponse>, Status> {
let request = request.into_inner();
let Some(envelope) = request.identity.as_ref() else {
return Err(Status::unauthenticated("identity envelope is required"));
};
let identity = self.service.verify_identity(envelope).await?;
let closed = self.service.revoke(&identity, &request.reason).await?;
Ok(Response::new(RevokeResponse { closed }))
}
}
#[tonic::async_trait]
impl SubscriptionTrait for SubscriptionRpc {
async fn subscribe(
&self,
request: Request<SubscribeRequest>,
) -> Result<Response<SubscribeResponse>, Status> {
let request = request.into_inner();
let Some(envelope) = request.identity.as_ref() else {
return Err(Status::unauthenticated("identity envelope is required"));
};
let identity = self.service.verify_identity(envelope).await?;
let filter = filter_to_properties(request.filter);
let id = self
.service
.subscribe(
identity.user_id,
&request.device_id,
&request.namespace,
SubscriptionRequest {
topic: request.topic,
tag: request.tag,
id: if request.id.is_empty() {
None
} else {
Some(request.id)
},
filter: filter.clone(),
permission_path: if request.permission_path.is_empty() {
None
} else {
Some(request.permission_path)
},
},
)
.await
.map_err(|error| Status::invalid_argument(error))?;
Ok(Response::new(SubscribeResponse {
id,
scope: scope_label(&filter),
created_at_unix_ms: 0,
}))
}
async fn unsubscribe(
&self,
request: Request<UnsubscribeRequest>,
) -> Result<Response<UnsubscribeResponse>, Status> {
let request = request.into_inner();
let Some(envelope) = request.identity.as_ref() else {
return Err(Status::unauthenticated("identity envelope is required"));
};
let identity = self.service.verify_identity(envelope).await?;
self.service
.unsubscribe(
identity.user_id,
&request.device_id,
&request.namespace,
&request.topic,
&request.tag,
&request.id,
)
.await
.map_err(|error| Status::invalid_argument(error))?;
Ok(Response::new(UnsubscribeResponse { removed: true }))
}
async fn list_subscriptions(
&self,
request: Request<ListSubscriptionsRequest>,
) -> Result<Response<ListSubscriptionsResponse>, Status> {
let request = request.into_inner();
let Some(envelope) = request.identity.as_ref() else {
return Err(Status::unauthenticated("identity envelope is required"));
};
let identity = self.service.verify_identity(envelope).await?;
let subscriptions = self
.service
.list(identity.user_id, &request.device_id, "/")
.await
.map_err(|error| Status::unavailable(error))?;
let views = subscriptions
.into_iter()
.map(|subscription| SubscriptionView {
id: subscription.id,
topic: subscription.topic,
tag: subscription.tag,
filter: properties_to_filter(&subscription.filter),
scope: subscription.scope,
device_id: subscription.device_id,
namespace: subscription.namespace,
})
.collect();
Ok(Response::new(ListSubscriptionsResponse {
subscriptions: views,
}))
}
}
fn scope_label(filter: &EventProperties) -> String {
[
("room", &filter.room_id),
("issue", &filter.issue_id),
("pull", &filter.pull_id),
("repo", &filter.repo_id),
("org", &filter.org_id),
("user", &filter.user_id),
]
.into_iter()
.find_map(|(kind, id)| id.as_ref().map(|id| format!("{kind}:{id}")))
.unwrap_or_default()
}
fn filter_to_properties(filter: Option<EventFilter>) -> EventProperties {
let filter = filter.unwrap_or_default();
EventProperties {
user_id: Some(filter.user_id).filter(|value| !value.is_empty()),
org_id: Some(filter.org_id).filter(|value| !value.is_empty()),
request_id: Some(filter.request_id).filter(|value| !value.is_empty()),
repo_id: Some(filter.repo_id).filter(|value| !value.is_empty()),
repo_name: Some(filter.repo_name).filter(|value| !value.is_empty()),
room_id: Some(filter.room_id).filter(|value| !value.is_empty()),
message_id: Some(filter.message_id).filter(|value| !value.is_empty()),
thread_id: Some(filter.thread_id).filter(|value| !value.is_empty()),
issue_id: Some(filter.issue_id).filter(|value| !value.is_empty()),
pull_id: Some(filter.pull_id).filter(|value| !value.is_empty()),
timestamp: (filter.timestamp != 0).then_some(filter.timestamp),
}
}
fn properties_to_filter(properties: &EventProperties) -> Option<EventFilter> {
Some(EventFilter {
user_id: properties.user_id.clone().unwrap_or_default(),
org_id: properties.org_id.clone().unwrap_or_default(),
request_id: properties.request_id.clone().unwrap_or_default(),
repo_id: properties.repo_id.clone().unwrap_or_default(),
repo_name: properties.repo_name.clone().unwrap_or_default(),
room_id: properties.room_id.clone().unwrap_or_default(),
message_id: properties.message_id.clone().unwrap_or_default(),
thread_id: properties.thread_id.clone().unwrap_or_default(),
issue_id: properties.issue_id.clone().unwrap_or_default(),
pull_id: properties.pull_id.clone().unwrap_or_default(),
timestamp: properties.timestamp.unwrap_or_default(),
})
}