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

wscall_server/
server_types.rs
