fraiseql_core/runtime/subscription/
protocol.rs

1//! GraphQL over `WebSocket` subscription protocol types.
2//!
3//! Implements the `graphql-ws` protocol (v5+) message framing for
4//! client-to-server and server-to-client subscription communication.
5
6use std::collections::HashMap;
7
8use serde::{Deserialize, Serialize};
9
10/// Client-to-server message types.
11#[derive(Debug, Clone, PartialEq, Eq)]
12#[non_exhaustive]
13pub enum ClientMessageType {
14    /// Connection initialization.
15    ConnectionInit,
16    /// Ping (keepalive).
17    Ping,
18    /// Pong response.
19    Pong,
20    /// Subscribe to operation.
21    Subscribe,
22    /// Complete/unsubscribe from operation.
23    Complete,
24    /// Legacy `graphql-ws` connection termination — mapped to a graceful
25    /// close so the connection and its subscriptions do not outlive the
26    /// client's stated intent (#786).
27    ConnectionTerminate,
28}
29
30impl ClientMessageType {
31    /// Parse message type from string.
32    #[must_use]
33    #[allow(clippy::should_implement_trait)] // Reason: returns Option<Self> (unknown types yield None), not a FromStr-compatible Result
34    pub fn from_str(s: &str) -> Option<Self> {
35        match s {
36            "connection_init" => Some(Self::ConnectionInit),
37            "ping" => Some(Self::Ping),
38            "pong" => Some(Self::Pong),
39            "subscribe" => Some(Self::Subscribe),
40            "complete" => Some(Self::Complete),
41            "connection_terminate" => Some(Self::ConnectionTerminate),
42            _ => None,
43        }
44    }
45
46    /// Get string representation.
47    #[must_use]
48    pub const fn as_str(&self) -> &'static str {
49        match self {
50            Self::ConnectionInit => "connection_init",
51            Self::Ping => "ping",
52            Self::Pong => "pong",
53            Self::Subscribe => "subscribe",
54            Self::Complete => "complete",
55            Self::ConnectionTerminate => "connection_terminate",
56        }
57    }
58}
59
60/// Server-to-client message types.
61#[derive(Debug, Clone, PartialEq, Eq)]
62#[non_exhaustive]
63pub enum ServerMessageType {
64    /// Connection acknowledged.
65    ConnectionAck,
66    /// Ping (keepalive).
67    Ping,
68    /// Pong response.
69    Pong,
70    /// Subscription data.
71    Next,
72    /// Operation error.
73    Error,
74    /// Operation complete.
75    Complete,
76}
77
78impl ServerMessageType {
79    /// Get string representation.
80    #[must_use]
81    pub const fn as_str(&self) -> &'static str {
82        match self {
83            Self::ConnectionAck => "connection_ack",
84            Self::Ping => "ping",
85            Self::Pong => "pong",
86            Self::Next => "next",
87            Self::Error => "error",
88            Self::Complete => "complete",
89        }
90    }
91}
92
93/// Client message (from `WebSocket` client).
94#[derive(Debug, Clone, Deserialize)]
95pub struct ClientMessage {
96    /// Message type.
97    #[serde(rename = "type")]
98    pub message_type: String,
99
100    /// Operation ID (for subscribe/complete).
101    #[serde(default)]
102    pub id: Option<String>,
103
104    /// Payload (connection params or subscription query).
105    #[serde(default)]
106    pub payload: Option<serde_json::Value>,
107}
108
109impl ClientMessage {
110    /// Parse the message type.
111    #[must_use]
112    pub fn parsed_type(&self) -> Option<ClientMessageType> {
113        ClientMessageType::from_str(&self.message_type)
114    }
115
116    /// Extract connection parameters from `connection_init` payload.
117    #[must_use]
118    pub const fn connection_params(&self) -> Option<&serde_json::Value> {
119        self.payload.as_ref()
120    }
121
122    /// Extract subscription query from subscribe payload.
123    #[must_use]
124    pub fn subscription_payload(&self) -> Option<SubscribePayload> {
125        self.payload.as_ref().and_then(|p| serde_json::from_value(p.clone()).ok())
126    }
127}
128
129/// Subscribe message payload.
130#[derive(Debug, Clone, Deserialize, Serialize)]
131pub struct SubscribePayload {
132    /// GraphQL query string.
133    pub query: String,
134
135    /// Optional operation name.
136    #[serde(rename = "operationName")]
137    #[serde(default)]
138    pub operation_name: Option<String>,
139
140    /// Query variables.
141    #[serde(default)]
142    pub variables: HashMap<String, serde_json::Value>,
143
144    /// Extensions (e.g., persisted query hash).
145    #[serde(default)]
146    pub extensions: HashMap<String, serde_json::Value>,
147}
148
149/// Server message (to `WebSocket` client).
150#[derive(Debug, Clone, Serialize)]
151pub struct ServerMessage {
152    /// Message type.
153    #[serde(rename = "type")]
154    pub message_type: String,
155
156    /// Operation ID (for next/error/complete).
157    #[serde(skip_serializing_if = "Option::is_none")]
158    pub id: Option<String>,
159
160    /// Payload (data, errors, or ack payload).
161    #[serde(skip_serializing_if = "Option::is_none")]
162    pub payload: Option<serde_json::Value>,
163}
164
165impl ServerMessage {
166    /// Create `connection_ack` message.
167    #[must_use]
168    pub fn connection_ack(payload: Option<serde_json::Value>) -> Self {
169        Self {
170            message_type: ServerMessageType::ConnectionAck.as_str().to_string(),
171            id: None,
172            payload,
173        }
174    }
175
176    /// Create ping message.
177    #[must_use]
178    pub fn ping(payload: Option<serde_json::Value>) -> Self {
179        Self {
180            message_type: ServerMessageType::Ping.as_str().to_string(),
181            id: None,
182            payload,
183        }
184    }
185
186    /// Create pong message.
187    #[must_use]
188    pub fn pong(payload: Option<serde_json::Value>) -> Self {
189        Self {
190            message_type: ServerMessageType::Pong.as_str().to_string(),
191            id: None,
192            payload,
193        }
194    }
195
196    /// Create next (data) message.
197    #[must_use]
198    #[allow(clippy::needless_pass_by_value)] // Reason: data is moved into serde_json::json! macro to construct the payload object
199    pub fn next(id: impl Into<String>, data: serde_json::Value) -> Self {
200        Self {
201            message_type: ServerMessageType::Next.as_str().to_string(),
202            id:           Some(id.into()),
203            payload:      Some(serde_json::json!({ "data": data })),
204        }
205    }
206
207    /// Create a next (data) message that also carries GraphQL `extensions`.
208    ///
209    /// Mirrors [`Self::next`] but populates the `extensions` field of the
210    /// graphql-transport-ws `ExecutionResult` payload — the spec-blessed,
211    /// safely-ignorable slot used to deliver the Change-Spine envelope to
212    /// subscription clients (#425).
213    #[must_use]
214    #[allow(clippy::needless_pass_by_value)] // Reason: data/extensions are moved into the json! macro
215    pub fn next_with_extensions(
216        id: impl Into<String>,
217        data: serde_json::Value,
218        extensions: serde_json::Value,
219    ) -> Self {
220        Self {
221            message_type: ServerMessageType::Next.as_str().to_string(),
222            id:           Some(id.into()),
223            payload:      Some(serde_json::json!({ "data": data, "extensions": extensions })),
224        }
225    }
226
227    /// Create error message.
228    #[must_use]
229    #[allow(clippy::needless_pass_by_value)] // Reason: errors is consumed by serde_json::to_value, which requires an owned value
230    pub fn error(id: impl Into<String>, errors: Vec<GraphQLError>) -> Self {
231        Self {
232            message_type: ServerMessageType::Error.as_str().to_string(),
233            id:           Some(id.into()),
234            payload:      Some(serde_json::to_value(errors).unwrap_or_default()),
235        }
236    }
237
238    /// Create complete message.
239    #[must_use]
240    pub fn complete(id: impl Into<String>) -> Self {
241        Self {
242            message_type: ServerMessageType::Complete.as_str().to_string(),
243            id:           Some(id.into()),
244            payload:      None,
245        }
246    }
247
248    /// Serialize to JSON string.
249    ///
250    /// # Errors
251    ///
252    /// Returns error if serialization fails.
253    pub fn to_json(&self) -> Result<String, serde_json::Error> {
254        serde_json::to_string(self)
255    }
256}
257
258pub use fraiseql_error::{GraphQLError, GraphQLErrorLocation as ErrorLocation};
259
260/// Close codes for `WebSocket` connection.
261#[derive(Debug, Clone, Copy, PartialEq, Eq)]
262#[non_exhaustive]
263pub enum CloseCode {
264    /// Normal closure.
265    Normal               = 1000,
266    /// Server going away (drain/shutdown, RFC 6455 §7.4.1).
267    GoingAway            = 1001,
268    /// Client violated protocol.
269    ProtocolError        = 1002,
270    /// Invalid or undecodable message (graphql-transport-ws spec: Bad Request).
271    BadRequest           = 4400,
272    /// Internal server error.
273    InternalError        = 1011,
274    /// Connection initialization timeout.
275    ConnectionInitTimeout = 4408,
276    /// Too many initialization requests.
277    TooManyInitRequests  = 4429,
278    /// Subscriber already exists (duplicate ID).
279    SubscriberAlreadyExists = 4409,
280    /// Unauthorized.
281    Unauthorized         = 4401,
282    /// Subscription not found (invalid ID on complete).
283    SubscriptionNotFound = 4404,
284}
285
286impl CloseCode {
287    /// Get the close code value.
288    #[must_use]
289    pub const fn code(self) -> u16 {
290        self as u16
291    }
292
293    /// Get the close reason message.
294    #[must_use]
295    pub const fn reason(self) -> &'static str {
296        match self {
297            Self::Normal => "Normal closure",
298            Self::GoingAway => "Server shutting down",
299            Self::ProtocolError => "Protocol error",
300            Self::BadRequest => "Bad Request",
301            Self::InternalError => "Internal server error",
302            Self::ConnectionInitTimeout => "Connection initialization timeout",
303            Self::TooManyInitRequests => "Too many initialization requests",
304            Self::SubscriberAlreadyExists => "Subscriber already exists",
305            Self::Unauthorized => "Unauthorized",
306            Self::SubscriptionNotFound => "Subscription not found",
307        }
308    }
309}
fraiseql_core/runtime/subscription/protocol.rs

fraiseql_core/runtime/subscription/
protocol.rs
