main
use std::fmt;
use std::time::Duration;
use async_trait::async_trait;
use tokio::sync::{mpsc, oneshot};
use crate::hub::SocketHub;
use crate::session::Input;
use crate::{EmitPacket, SocketPayload};
pub type SocketResult<T> = Result<T, SocketError>;
#[derive(Debug)]
pub enum SocketError {
Rejected(String),
Unavailable(String),
Config(String),
}
impl fmt::Display for SocketError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Rejected(message) => write!(formatter, "rejected: {message}"),
Self::Unavailable(message) => write!(formatter, "unavailable: {message}"),
Self::Config(message) => write!(formatter, "config: {message}"),
}
}
}
impl std::error::Error for SocketError {}
/// 聊天连接生命周期阶段（在线态旁路上报）。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionPhase {
Open,
Heartbeat,
Close,
}
/// 会话出站句柄：业务层（事件投递）只通过它把事件推给具体客户端。
#[derive(Clone)]
pub struct SessionHandle {
input: mpsc::Sender<Input>,
}
impl SessionHandle {
pub(crate) fn new(input: mpsc::Sender<Input>) -> Self {
Self { input }
}
/// 投递一个事件并等待客户端 ACK；返回是否被接受。
pub async fn deliver(&self, packet: EmitPacket, timeout: Duration) -> bool {
let deadline = tokio::time::Instant::now() + timeout;
let (ack, receipt) = oneshot::channel::<SocketPayload>();
if self
.input
.send(Input::Emit {
packet,
ack: Some((ack, deadline)),
accepted: None,
})
.await
.is_err()
{
return false;
}
matches!(tokio::time::timeout_at(deadline, receipt).await, Ok(Ok(_)))
}
pub async fn close(&self) {
let _ = self.input.send(Input::Close).await;
}
pub fn is_closed(&self) -> bool {
self.input.is_closed()
}
pub async fn closed(&self) {
self.input.closed().await
}
}
/// 命名空间连接建立后的会话快照，交给业务层做设备注册/在线态登记。
#[derive(Clone)]
pub struct ConnectedSession {
pub user_id: i64,
pub device_id: String,
pub namespace: String,
pub session_id: String,
pub session_sid: String,
pub access_token: String,
pub first_namespace: bool,
pub hub: SocketHub,
pub handle: SessionHandle,
}
/// 每个 namespace 一个投递 worker 的上下文。
pub struct DeliveryContext {
pub user_id: i64,
pub device_id: String,
pub namespace: String,
pub session_id: String,
pub session_sid: String,
pub access_token: String,
pub hub: SocketHub,
pub handle: SessionHandle,
pub stop: oneshot::Receiver<()>,
}
/// 通道与业务的分界：socket 只负责协议/会话，身份、订阅与投递策略由业务层实现。
#[async_trait]
pub trait SocketBackend: Send + Sync + 'static {
/// 会话凭证（KenApi 会话键）→ user_id。
async fn authenticate(&self, token: &str) -> SocketResult<i64>;
/// 命名空间连接建立后：设备游标注册、组织成员缓存、在线心跳。
async fn on_connect(&self, session: &ConnectedSession) -> SocketResult<()>;
/// 连接心跳（PONG 触发）。
async fn heartbeat(&self, user_id: i64);
/// 在线态旁路上报；失败只记日志。
async fn report_connection(&self, phase: ConnectionPhase, user_id: i64, session: &str);
/// 每个 namespace 的投递 worker（订阅匹配 + 二次鉴权 + 积压消费）。
async fn run_delivery(&self, context: DeliveryContext);
}