072c704
use std::collections::HashMap;
use std::time::Duration;
use actor::{ActorHost, PushStore};
use async_trait::async_trait;
use socket::{
ConnectedSession, ConnectionPhase, DeliveryContext, SocketBackend, SocketError, SocketPayload,
};
use crate::gateway::platform::{PlatformClient, PlatformError};
use crate::gateway::subscriptions::Subscriptions;
const IDLE_FALLBACK: Duration = Duration::from_secs(30);
const STANDBY_WAIT: Duration = Duration::from_secs(2);
/// 单条订阅的投递授权结论缓存时长；与身份复核的 60 秒节奏一致。
const AUTHORIZATION_TTL: Duration = Duration::from_secs(60);
/// socket 通道与事件业务之间的适配层：身份、订阅、投递策略都落在这里。
#[derive(Clone)]
pub struct RealtimeBackend {
host: ActorHost,
platform: PlatformClient,
subscriptions: Subscriptions,
push_store: PushStore,
}
impl RealtimeBackend {
pub fn new(
host: ActorHost,
platform: PlatformClient,
subscriptions: Subscriptions,
push_store: PushStore,
) -> Self {
Self {
host,
platform,
subscriptions,
push_store,
}
}
}
#[async_trait]
impl SocketBackend for RealtimeBackend {
async fn authenticate(&self, token: &str) -> Result<i64, SocketError> {
self.platform
.authenticate(token)
.await
.map_err(|error| match error {
PlatformError::Rejected(message) => SocketError::Rejected(message),
PlatformError::Unavailable(message) => SocketError::Unavailable(message),
PlatformError::Config(message) => SocketError::Config(message),
})
}
async fn on_connect(&self, session: &ConnectedSession) -> Result<(), SocketError> {
self.push_store
.register_device(session.user_id, &session.device_id, &session.namespace)
.await
.map_err(SocketError::Unavailable)?;
if session.first_namespace {
self.host
.socket_heartbeat(session.user_id)
.await
.map_err(|error| SocketError::Unavailable(error.to_string()))?;
}
Ok(())
}
async fn heartbeat(&self, user_id: i64) {
let _ = self.host.socket_heartbeat(user_id).await;
}
async fn report_connection(&self, phase: ConnectionPhase, user_id: i64, session: &str) {
let phase = match phase {
ConnectionPhase::Open => "open",
ConnectionPhase::Heartbeat => "heartbeat",
ConnectionPhase::Close => "close",
};
if let Err(error) = self
.platform
.report_connection(phase, user_id, session)
.await
{
tracing::warn!(%error, "chat connection report failed");
}
}
async fn run_delivery(&self, context: DeliveryContext) {
let store = self.push_store.clone();
let subscriptions = self.subscriptions.clone();
let platform = self.platform.clone();
let hub = context.hub.clone();
let input = context.handle.clone();
let user = context.user_id;
let device = context.device_id.clone();
let namespace = context.namespace.clone();
let token = context.access_token.clone();
let owner = format!("{}:{}", context.session_sid, context.session_id);
let mut stopped = context.stop;
let mut shutdown = hub.shutdown_receiver();
let notification = hub.notification(user).await;
let work = async {
let mut verified: Option<tokio::time::Instant> = None;
let mut authorized: HashMap<String, tokio::time::Instant> = HashMap::new();
loop {
let changed = notification.notified();
tokio::pin!(changed);
changed.as_mut().enable();
if verified.is_none_or(|verified: tokio::time::Instant| {
verified.elapsed() >= Duration::from_secs(60)
}) {
match platform.authenticate(&token).await {
Ok(current) if current == user => {
verified = Some(tokio::time::Instant::now())
}
Err(PlatformError::Unavailable(_)) => {
tokio::time::sleep(Duration::from_secs(1)).await;
continue;
}
_ => {
input.close().await;
break;
}
}
}
if !matches!(
store.claim(user, &device, &namespace, &owner).await,
Ok(true)
) {
tokio::time::sleep(STANDBY_WAIT).await;
continue;
}
let (cursor, next) = match store.next(user, &device, &namespace).await {
Ok(next) => next,
Err(_) => {
tokio::time::sleep(Duration::from_secs(1)).await;
continue;
}
};
let Some((delivery, mut payload)) = next else {
tokio::select! {
_ = &mut changed => {}
_ = tokio::time::sleep(IDLE_FALLBACK) => {}
}
continue;
};
let authorization = async {
if let Some(destinations) = payload.get("destinations") {
let destinations: Vec<crate::gateway::Subscription> =
serde_json::from_value(destinations.clone())
.map_err(|error| PlatformError::Unavailable(error.to_string()))?;
for subscription in destinations {
if subscription.device_id != device
|| subscription.namespace != namespace
{
continue;
}
if authorized
.get(&subscription.id)
.is_some_and(|at| at.elapsed() < AUTHORIZATION_TTL)
{
return Ok(true);
}
if !subscriptions
.contains(&subscription)
.await
.map_err(PlatformError::Unavailable)?
{
continue;
}
if !platform
.authorize(user, &subscription.filter, "DELIVER")
.await?
{
continue;
}
if let Some(org) = subscription.filter.org_id.as_deref() {
if !platform.is_member(user, org).await? {
continue;
}
}
authorized.insert(subscription.id.clone(), tokio::time::Instant::now());
return Ok(true);
}
return Ok(false);
}
if payload.get("source").and_then(serde_json::Value::as_str) == Some("socket") {
return Ok(payload.get("namespace").and_then(serde_json::Value::as_str)
== Some(namespace.as_str()));
}
if payload.get("control").and_then(serde_json::Value::as_bool) != Some(true) {
if let Some(org) = payload
.pointer("/properties/org_id")
.and_then(serde_json::Value::as_str)
{
return platform.is_member(user, org).await;
}
}
Ok(true)
}
.await;
let allowed = match authorization {
Ok(allowed) => allowed,
Err(PlatformError::Rejected(_)) => {
input.close().await;
break;
}
Err(_) => {
tokio::time::sleep(Duration::from_secs(1)).await;
continue;
}
};
if allowed {
let previous_sequence = cursor
.split_once('-')
.and_then(|(number, _)| number.parse::<u64>().ok());
let sequence = delivery
.split_once('-')
.and_then(|(number, _)| number.parse::<u64>().ok());
if previous_sequence.and_then(|previous| previous.checked_add(1)) != sequence {
payload["resync_required"] = serde_json::Value::Bool(true);
}
let resync = payload
.get("resync_required")
.and_then(serde_json::Value::as_bool)
== Some(true);
let binaries: Vec<Vec<u8>> = payload
.get("binaries")
.cloned()
.and_then(|value| serde_json::from_value(value).ok())
.unwrap_or_default();
if let Some(object) = payload.as_object_mut() {
object.remove("binaries");
object.remove("destinations");
}
let packet = crate::gateway::EmitPacket {
namespace: namespace.clone(),
event: if resync { "resync" } else { "event" }.to_string(),
args: vec![payload],
binaries,
};
if !input.deliver(packet, Duration::from_secs(15)).await {
tokio::time::sleep(Duration::from_secs(1)).await;
continue;
}
}
if !matches!(
store
.acknowledge(user, &device, &namespace, &owner, &cursor, &delivery)
.await,
Ok(true)
) {
tokio::time::sleep(Duration::from_secs(1)).await;
}
}
};
if !*shutdown.borrow() {
tokio::select! {
_ = &mut stopped => {}
_ = shutdown.changed() => {}
_ = input.closed() => {}
_ = work => {}
}
}
store.release(user, &device, &namespace, &owner).await;
let _ = SocketPayload::default();
}
}