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::{Map, 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    /// Moves the params payload out and binds it without cloning.
211    ///
212    /// `serde_json::from_value` consumes its input, so taking ownership of the
213    /// stored `Value` first avoids the deep copy that [`Self::bind`] performs.
214    /// The stored params are replaced with `Value::Null`; this is intended for
215    /// the `typed_route` / `validated_route` wrappers whose handlers receive the
216    /// typed value directly and no longer need the raw params.
217    pub(crate) fn bind_take<T>(&mut self) -> Result<T, ApiError>
218    where
219        T: DeserializeOwned,
220    {
221        let params = std::mem::take(&mut self.params);
222        serde_json::from_value(params)
223            .map_err(|source| ApiError::bad_request(format!("invalid params: {source}")))
224    }
225
226    /// Binds params and runs `ValidateParams` on the result.
227    pub fn bind_and_validate<T>(&self) -> Result<T, ApiError>
228    where
229        T: DeserializeOwned + ValidateParams,
230    {
231        let params: T = self.bind()?;
232        params.validate()?;
233        Ok(params)
234    }
235
236    /// Binds params and runs `validator::Validate` on the result.
237    pub fn bind_validated<T>(&self) -> Result<T, ApiError>
238    where
239        T: DeserializeOwned + Validate,
240    {
241        let params: T = self.bind()?;
242        params.validate().map_err(|source| {
243            ApiError::bad_request("params validation failed").with_details(json!({
244                "validation_errors": validation::errors_to_details(&source),
245            }))
246        })?;
247        Ok(params)
248    }
249
250    /// Returns all attachments that accompanied the request.
251    pub fn attachments(&self) -> &[FileAttachment] {
252        &self.attachments
253    }
254
255    /// Returns the raw metadata payload.
256    pub fn metadata(&self) -> &Value {
257        &self.metadata
258    }
259
260    /// Returns a server handle for outbound event operations.
261    pub fn server(&self) -> &ServerHandle {
262        &self.server
263    }
264
265    /// Returns a simplified JSON view of incoming attachments.
266    pub fn attachment_summaries(&self) -> Vec<Value> {
267        self.attachments
268            .iter()
269            .map(|attachment| {
270                json!({
271                    "id": attachment.id,
272                    "name": attachment.name,
273                    "content_type": attachment.content_type,
274                    "size": attachment.size(),
275                })
276            })
277            .collect()
278    }
279}
280
281/// Event context passed to event handlers.
282#[derive(Clone)]
283pub struct EventContext {
284    pub(crate) connection_id: String,
285    pub(crate) peer_addr: Option<SocketAddr>,
286    pub(crate) event_id: u64,
287    pub(crate) name: String,
288    pub(crate) data: Map<String, Value>,
289    pub(crate) attachments: Vec<FileAttachment>,
290    pub(crate) metadata: Value,
291    pub(crate) server: ServerHandle,
292}
293
294impl EventContext {
295    /// Returns the logical connection id for the current client.
296    pub fn connection_id(&self) -> &str {
297        &self.connection_id
298    }
299
300    /// Returns the peer socket address when available.
301    pub fn peer_addr(&self) -> Option<SocketAddr> {
302        self.peer_addr
303    }
304
305    /// Returns the peer IP as a string when available.
306    pub fn peer_ip(&self) -> Option<String> {
307        self.peer_addr.map(|addr| addr.ip().to_string())
308    }
309
310    /// Returns the event correlation id.
311    pub fn event_id(&self) -> u64 {
312        self.event_id
313    }
314
315    /// Returns the event name.
316    pub fn name(&self) -> &str {
317        &self.name
318    }
319
320    /// Returns the event data object.
321    pub fn data(&self) -> &Map<String, Value> {
322        &self.data
323    }
324
325    /// Returns attachments that accompanied the event.
326    pub fn attachments(&self) -> &[FileAttachment] {
327        &self.attachments
328    }
329
330    /// Returns the raw metadata payload.
331    pub fn metadata(&self) -> &Value {
332        &self.metadata
333    }
334
335    /// Returns a server handle for outbound event operations.
336    pub fn server(&self) -> &ServerHandle {
337        &self.server
338    }
339}
340
341/// Context passed to the global exception handler.
342#[derive(Clone)]
343pub struct ExceptionContext {
344    pub connection_id: String,
345    pub request_id: Option<u64>,
346    pub target: String,
347    pub message_kind: &'static str,
348    pub error: ApiError,
349}
350
351/// Application-level error returned from handlers.
352#[derive(Debug, Clone, Error)]
353#[error("{code}: {message}")]
354pub struct ApiError {
355    pub code: String,
356    pub message: String,
357    pub status: u16,
358    pub details: Option<Value>,
359}
360
361impl ApiError {
362    /// Constructs a 400 bad request error.
363    pub fn bad_request(message: impl Into<String>) -> Self {
364        Self::new("bad_request", message, 400)
365    }
366
367    /// Constructs a 404 not found error.
368    pub fn not_found(message: impl Into<String>) -> Self {
369        Self::new("not_found", message, 404)
370    }
371
372    /// Constructs a 500 internal error.
373    pub fn internal(message: impl Into<String>) -> Self {
374        Self::new("internal_error", message, 500)
375    }
376
377    /// Constructs an error with an explicit code and status.
378    pub fn new(code: impl Into<String>, message: impl Into<String>, status: u16) -> Self {
379        Self {
380            code: code.into(),
381            message: message.into(),
382            status,
383            details: None,
384        }
385    }
386
387    /// Attaches structured details to the error.
388    pub fn with_details(mut self, details: Value) -> Self {
389        self.details = Some(details);
390        self
391    }
392
393    /// Converts the error into a transport payload.
394    pub fn into_payload(self) -> ErrorPayload {
395        ErrorPayload {
396            code: self.code,
397            message: self.message,
398            status: self.status,
399            details: self.details,
400        }
401    }
402}
403
404/// Errors produced by the reusable server runtime.
405#[derive(Debug, Error)]
406pub enum ServerError {
407    #[error("io error: {0}")]
408    Io(#[from] std::io::Error),
409    #[error("websocket error: {0}")]
410    WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
411    #[error("protocol error: {0}")]
412    Protocol(#[from] ProtocolError),
413    #[error("api error: {0:?}")]
414    Api(#[from] ApiError),
415    #[error("connection idle timeout: {0}")]
416    IdleTimeout(String),
417    #[error("outbound queue is full for connection {0}")]
418    OutboundQueueFull(String),
419}
wscall_server/server_types.rs

wscall_server/
server_types.rs
