wscall_server/
server_types.rs

1use std::net::SocketAddr;
2use std::sync::Arc;
3use std::sync::atomic::AtomicU64;
4
5use bytes::Bytes;
6use dashmap::DashMap;
7use serde::de::DeserializeOwned;
8use serde_json::{Value, json};
9use thiserror::Error;
10use tokio::sync::mpsc;
11use validator::Validate;
12use wscall_protocol::{
13    EncryptionKind, ErrorPayload, FileAttachment, FrameCodec, PacketEnvelope, ProtocolError,
14};
15
16use crate::validation;
17
18pub(crate) enum ServerOutbound {
19    /// Pre-encoded frame bytes. All push/response paths encode once up front
20    /// (often enabling parallel encoding across concurrent handlers) and the
21    /// writer task only ships bytes, which keeps the writer cheap and shared
22    /// across recipients for broadcasts.
23    PreEncoded(Bytes),
24    /// An unencoded packet that the writer must encode using the per-connection
25    /// codec (used in ECDH mode where each connection has a unique session key).
26    Packet(PacketEnvelope),
27    Pong(Vec<u8>),
28    Close,
29}
30
31/// Per-connection entry stored in the live client table.
32///
33/// In PSK mode all connections share the server-wide codec. In ECDH mode
34/// each connection has a unique session key; the per-connection codec is
35/// passed directly to the writer/reader tasks at spawn time.
36pub(crate) struct ClientEntry {
37    pub sender: mpsc::Sender<ServerOutbound>,
38}
39
40/// Shared, lock-free table of live client outbound channels.
41pub(crate) struct ServerState {
42    pub clients: DashMap<String, ClientEntry>,
43    /// Per-server event id counter for server-originated events.
44    pub next_event_id: AtomicU64,
45}
46
47impl ServerState {
48    pub fn new() -> Self {
49        Self {
50            clients: DashMap::new(),
51            next_event_id: AtomicU64::new(0),
52        }
53    }
54}
55
56impl Default for ServerState {
57    fn default() -> Self {
58        Self::new()
59    }
60}
61
62/// Handle that can be cloned into request and event handlers for server push operations.
63///
64/// Carries a clone of the [`FrameCodec`] so that push operations can pre-encode
65/// frames once and avoid per-recipient work in the writer task.
66#[derive(Clone)]
67pub struct ServerHandle {
68    pub(crate) state: Arc<ServerState>,
69    pub(crate) codec: FrameCodec,
70    pub(crate) default_encryption: EncryptionKind,
71    /// Whether the server is in ECDH mode (per-connection session keys).
72    pub(crate) is_ecdh: bool,
73}
74
75/// Context passed to server connection lifecycle handlers.
76#[derive(Clone)]
77pub struct ServerConnectionContext {
78    pub(crate) connection_id: String,
79    pub(crate) peer_addr: Option<SocketAddr>,
80    pub(crate) server: ServerHandle,
81}
82
83impl ServerConnectionContext {
84    /// Returns the logical connection id for the current client.
85    pub fn connection_id(&self) -> &str {
86        &self.connection_id
87    }
88
89    /// Returns the peer socket address when available.
90    pub fn peer_addr(&self) -> Option<SocketAddr> {
91        self.peer_addr
92    }
93
94    /// Returns the peer IP as a string when available.
95    pub fn peer_ip(&self) -> Option<String> {
96        self.peer_addr.map(|addr| addr.ip().to_string())
97    }
98
99    /// Returns a server handle for outbound event operations.
100    pub fn server(&self) -> &ServerHandle {
101        &self.server
102    }
103}
104
105/// Context passed to server disconnect lifecycle handlers.
106#[derive(Clone)]
107pub struct ServerDisconnectContext {
108    pub(crate) connection_id: String,
109    pub(crate) peer_addr: Option<SocketAddr>,
110    pub(crate) reason: String,
111    pub(crate) server: ServerHandle,
112}
113
114impl ServerDisconnectContext {
115    /// Returns the logical connection id for the disconnected client.
116    pub fn connection_id(&self) -> &str {
117        &self.connection_id
118    }
119
120    /// Returns the peer socket address when available.
121    pub fn peer_addr(&self) -> Option<SocketAddr> {
122        self.peer_addr
123    }
124
125    /// Returns the peer IP as a string when available.
126    pub fn peer_ip(&self) -> Option<String> {
127        self.peer_addr.map(|addr| addr.ip().to_string())
128    }
129
130    /// Returns the human-readable disconnect reason.
131    pub fn reason(&self) -> &str {
132        &self.reason
133    }
134
135    /// Returns a server handle for outbound event operations.
136    pub fn server(&self) -> &ServerHandle {
137        &self.server
138    }
139}
140
141/// Request context passed to API route handlers.
142#[derive(Clone)]
143pub struct ApiContext {
144    pub(crate) connection_id: String,
145    pub(crate) peer_addr: Option<SocketAddr>,
146    pub(crate) request_id: u64,
147    pub(crate) route: String,
148    pub(crate) params: Value,
149    pub(crate) attachments: Vec<FileAttachment>,
150    pub(crate) metadata: Value,
151    pub(crate) server: ServerHandle,
152}
153
154/// Trait for custom parameter validation after JSON binding.
155pub trait ValidateParams {
156    fn validate(&self) -> Result<(), ApiError>;
157}
158
159impl ApiContext {
160    /// Returns the logical connection id for the current client.
161    pub fn connection_id(&self) -> &str {
162        &self.connection_id
163    }
164
165    /// Returns the peer socket address when available.
166    pub fn peer_addr(&self) -> Option<SocketAddr> {
167        self.peer_addr
168    }
169
170    /// Returns the peer IP as a string when available.
171    pub fn peer_ip(&self) -> Option<String> {
172        self.peer_addr.map(|addr| addr.ip().to_string())
173    }
174
175    /// Returns the request correlation id.
176    pub fn request_id(&self) -> u64 {
177        self.request_id
178    }
179
180    /// Returns the matched route name.
181    pub fn route(&self) -> &str {
182        &self.route
183    }
184
185    /// Returns the raw JSON params payload.
186    pub fn params(&self) -> &Value {
187        &self.params
188    }
189
190    /// Looks up a single parameter by key.
191    pub fn param(&self, key: &str) -> Option<&Value> {
192        self.params.as_object()?.get(key)
193    }
194
195    /// Looks up a required parameter or returns a bad request error.
196    pub fn require_param(&self, key: &str) -> Result<&Value, ApiError> {
197        self.param(key)
198            .ok_or_else(|| ApiError::bad_request(format!("missing required param: {key}")))
199    }
200
201    /// Binds the raw params payload into a strongly typed value.
202    pub fn bind<T>(&self) -> Result<T, ApiError>
203    where
204        T: DeserializeOwned,
205    {
206        serde_json::from_value(self.params.clone())
207            .map_err(|source| ApiError::bad_request(format!("invalid params: {source}")))
208    }
209
210    /// Binds params and runs `ValidateParams` on the result.
211    pub fn bind_and_validate<T>(&self) -> Result<T, ApiError>
212    where
213        T: DeserializeOwned + ValidateParams,
214    {
215        let params: T = self.bind()?;
216        params.validate()?;
217        Ok(params)
218    }
219
220    /// Binds params and runs `validator::Validate` on the result.
221    pub fn bind_validated<T>(&self) -> Result<T, ApiError>
222    where
223        T: DeserializeOwned + Validate,
224    {
225        let params: T = self.bind()?;
226        params.validate().map_err(|source| {
227            ApiError::bad_request("params validation failed").with_details(json!({
228                "validation_errors": validation::errors_to_details(&source),
229            }))
230        })?;
231        Ok(params)
232    }
233
234    /// Returns all attachments that accompanied the request.
235    pub fn attachments(&self) -> &[FileAttachment] {
236        &self.attachments
237    }
238
239    /// Returns the raw metadata payload.
240    pub fn metadata(&self) -> &Value {
241        &self.metadata
242    }
243
244    /// Returns a server handle for outbound event operations.
245    pub fn server(&self) -> &ServerHandle {
246        &self.server
247    }
248
249    /// Returns a simplified JSON view of incoming attachments.
250    pub fn attachment_summaries(&self) -> Vec<Value> {
251        self.attachments
252            .iter()
253            .map(|attachment| {
254                json!({
255                    "id": attachment.id,
256                    "name": attachment.name,
257                    "content_type": attachment.content_type,
258                    "size": attachment.size,
259                })
260            })
261            .collect()
262    }
263}
264
265/// Event context passed to event handlers.
266#[derive(Clone)]
267pub struct EventContext {
268    pub(crate) connection_id: String,
269    pub(crate) peer_addr: Option<SocketAddr>,
270    pub(crate) event_id: u64,
271    pub(crate) name: String,
272    pub(crate) data: Value,
273    pub(crate) attachments: Vec<FileAttachment>,
274    pub(crate) metadata: Value,
275    pub(crate) storage_id: Option<u64>,
276    pub(crate) server: ServerHandle,
277}
278
279impl EventContext {
280    /// Returns the logical connection id for the current client.
281    pub fn connection_id(&self) -> &str {
282        &self.connection_id
283    }
284
285    /// Returns the peer socket address when available.
286    pub fn peer_addr(&self) -> Option<SocketAddr> {
287        self.peer_addr
288    }
289
290    /// Returns the peer IP as a string when available.
291    pub fn peer_ip(&self) -> Option<String> {
292        self.peer_addr.map(|addr| addr.ip().to_string())
293    }
294
295    /// Returns the event correlation id.
296    pub fn event_id(&self) -> u64 {
297        self.event_id
298    }
299
300    /// Returns the event name.
301    pub fn name(&self) -> &str {
302        &self.name
303    }
304
305    /// Returns the raw JSON event data.
306    pub fn data(&self) -> &Value {
307        &self.data
308    }
309
310    /// Returns attachments that accompanied the event.
311    pub fn attachments(&self) -> &[FileAttachment] {
312        &self.attachments
313    }
314
315    /// Returns the raw metadata payload.
316    pub fn metadata(&self) -> &Value {
317        &self.metadata
318    }
319
320    /// Returns the storage id when the event carries one.
321    pub fn storage_id(&self) -> Option<u64> {
322        self.storage_id
323    }
324
325    /// Returns a server handle for outbound event operations.
326    pub fn server(&self) -> &ServerHandle {
327        &self.server
328    }
329}
330
331/// Context passed to the global exception handler.
332#[derive(Clone)]
333pub struct ExceptionContext {
334    pub connection_id: String,
335    pub request_id: Option<u64>,
336    pub target: String,
337    pub message_kind: &'static str,
338    pub error: ApiError,
339}
340
341/// Application-level error returned from handlers.
342#[derive(Debug, Clone, Error)]
343#[error("{code}: {message}")]
344pub struct ApiError {
345    pub code: String,
346    pub message: String,
347    pub status: u16,
348    pub details: Option<Value>,
349}
350
351impl ApiError {
352    /// Constructs a 400 bad request error.
353    pub fn bad_request(message: impl Into<String>) -> Self {
354        Self::new("bad_request", message, 400)
355    }
356
357    /// Constructs a 404 not found error.
358    pub fn not_found(message: impl Into<String>) -> Self {
359        Self::new("not_found", message, 404)
360    }
361
362    /// Constructs a 500 internal error.
363    pub fn internal(message: impl Into<String>) -> Self {
364        Self::new("internal_error", message, 500)
365    }
366
367    /// Constructs an error with an explicit code and status.
368    pub fn new(code: impl Into<String>, message: impl Into<String>, status: u16) -> Self {
369        Self {
370            code: code.into(),
371            message: message.into(),
372            status,
373            details: None,
374        }
375    }
376
377    /// Attaches structured details to the error.
378    pub fn with_details(mut self, details: Value) -> Self {
379        self.details = Some(details);
380        self
381    }
382
383    /// Converts the error into a transport payload.
384    pub fn into_payload(self) -> ErrorPayload {
385        ErrorPayload {
386            code: self.code,
387            message: self.message,
388            status: self.status,
389            details: self.details,
390        }
391    }
392}
393
394/// Errors produced by the reusable server runtime.
395#[derive(Debug, Error)]
396pub enum ServerError {
397    #[error("io error: {0}")]
398    Io(#[from] std::io::Error),
399    #[error("websocket error: {0}")]
400    WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
401    #[error("protocol error: {0}")]
402    Protocol(#[from] ProtocolError),
403    #[error("api error: {0:?}")]
404    Api(#[from] ApiError),
405    #[error("connection idle timeout: {0}")]
406    IdleTimeout(String),
407    #[error("outbound queue is full for connection {0}")]
408    OutboundQueueFull(String),
409}
wscall_server/server_types.rs

wscall_server/
server_types.rs
