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

fraiseql_core/runtime/subscription/
protocol.rs
