phoenix_channel_client/client/
endpoint.rs

1use std::rc::Rc;
2
3use base64::{Engine as _, engine::general_purpose::STANDARD_NO_PAD};
4use futures::future::LocalBoxFuture;
5use phoenix_channel_runtime::V2_PROTOCOL_VERSION;
6use thiserror::Error;
7use url::Url;
8
9use super::ConnectContext;
10
11const AUTH_TOKEN_PREFIX: &str = "base64url.bearer.phx.";
12
13/// Query parameters and Phoenix 1.8 authentication for a connection attempt.
14#[derive(Clone, Default, Eq, PartialEq)]
15pub struct ConnectionConfig {
16    /// Query parameters appended to the endpoint URL.
17    pub params: Vec<(String, String)>,
18    /// Token sent through Phoenix's authenticated WebSocket subprotocol.
19    pub auth_token: Option<String>,
20}
21
22impl std::fmt::Debug for ConnectionConfig {
23    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24        formatter
25            .debug_struct("ConnectionConfig")
26            .field("params", &self.params)
27            .field("auth_token", &self.auth_token.as_ref().map(|_| "redacted"))
28            .finish()
29    }
30}
31
32impl ConnectionConfig {
33    /// Appends one query parameter.
34    pub fn param(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
35        self.params.push((name.into(), value.into()));
36        self
37    }
38
39    /// Sets the Phoenix 1.8 authentication token.
40    pub fn auth_token(mut self, token: impl Into<String>) -> Self {
41        self.auth_token = Some(token.into());
42        self
43    }
44}
45
46/// Async callback that refreshes connection parameters and authentication.
47pub type ConnectionConfigLoader =
48    Rc<dyn Fn(ConnectContext) -> LocalBoxFuture<'static, Result<ConnectionConfig, String>>>;
49
50/// Creates a loader that clones the same configuration for every attempt.
51pub fn static_connection_config(config: ConnectionConfig) -> ConnectionConfigLoader {
52    Rc::new(move |_| {
53        let config = config.clone();
54        Box::pin(async move { Ok(config) })
55    })
56}
57
58/// A Phoenix socket base URL and per-attempt connection configuration.
59#[derive(Clone)]
60pub struct Endpoint {
61    url: Url,
62    config_loader: ConnectionConfigLoader,
63}
64
65impl Endpoint {
66    /// Parses a `ws` or `wss` Phoenix socket URL.
67    ///
68    /// `/websocket` is appended when the supplied path does not already end
69    /// with it.
70    pub fn new(url: impl AsRef<str>) -> Result<Self, EndpointError> {
71        let mut url = Url::parse(url.as_ref())?;
72        match url.scheme() {
73            "ws" | "wss" => {}
74            scheme => return Err(EndpointError::UnsupportedScheme(scheme.to_owned())),
75        }
76        if !url.path().trim_end_matches('/').ends_with("/websocket") {
77            let path = format!("{}/websocket", url.path().trim_end_matches('/'));
78            url.set_path(&path);
79        }
80        Ok(Self {
81            url,
82            config_loader: static_connection_config(ConnectionConfig::default()),
83        })
84    }
85
86    /// Uses a static connection configuration for every attempt.
87    pub fn connection_config(mut self, config: ConnectionConfig) -> Self {
88        self.config_loader = static_connection_config(config);
89        self
90    }
91
92    /// Installs a loader that is evaluated for every connection attempt.
93    pub fn connection_config_loader(mut self, loader: ConnectionConfigLoader) -> Self {
94        self.config_loader = loader;
95        self
96    }
97
98    /// Resolves query parameters, protocol version, and authentication protocols.
99    pub async fn resolve(
100        &self,
101        context: ConnectContext,
102    ) -> Result<ResolvedEndpoint, EndpointError> {
103        let config = (self.config_loader)(context)
104            .await
105            .map_err(EndpointError::Config)?;
106        let mut url = self.url.clone();
107        let existing = url
108            .query_pairs()
109            .filter(|(name, _)| name != "vsn")
110            .map(|(name, value)| (name.into_owned(), value.into_owned()))
111            .collect::<Vec<_>>();
112        url.set_query(None);
113        {
114            let mut query = url.query_pairs_mut();
115            for (name, value) in existing.into_iter().chain(config.params) {
116                query.append_pair(&name, &value);
117            }
118            query.append_pair("vsn", V2_PROTOCOL_VERSION);
119        }
120
121        let protocols = config.auth_token.map_or_else(Vec::new, |token| {
122            vec![
123                "phoenix".to_owned(),
124                format!("{AUTH_TOKEN_PREFIX}{}", STANDARD_NO_PAD.encode(token)),
125            ]
126        });
127        Ok(ResolvedEndpoint {
128            url: url.into(),
129            protocols,
130        })
131    }
132}
133
134/// Fully resolved URL and WebSocket subprotocols for one connection attempt.
135#[derive(Clone, Eq, PartialEq)]
136pub struct ResolvedEndpoint {
137    /// WebSocket URL including query parameters and `vsn=2.0.0`.
138    pub url: String,
139    /// Requested WebSocket subprotocols, including encoded authentication.
140    pub protocols: Vec<String>,
141}
142
143impl std::fmt::Debug for ResolvedEndpoint {
144    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
145        let protocols = self
146            .protocols
147            .iter()
148            .map(|protocol| {
149                if protocol.starts_with(AUTH_TOKEN_PREFIX) {
150                    "redacted"
151                } else {
152                    protocol.as_str()
153                }
154            })
155            .collect::<Vec<_>>();
156        formatter
157            .debug_struct("ResolvedEndpoint")
158            .field("url", &self.url)
159            .field("protocols", &protocols)
160            .finish()
161    }
162}
163
164/// Invalid URL or failure while refreshing endpoint configuration.
165#[derive(Clone, Debug, Error, Eq, PartialEq)]
166pub enum EndpointError {
167    /// The endpoint string was not a valid URL.
168    #[error("invalid endpoint URL: {0}")]
169    InvalidUrl(String),
170    /// The endpoint scheme was neither `ws` nor `wss`.
171    #[error("unsupported endpoint URL scheme: {0}")]
172    UnsupportedScheme(String),
173    /// The connection configuration loader failed.
174    #[error("connection configuration failed: {0}")]
175    Config(String),
176}
177
178impl From<url::ParseError> for EndpointError {
179    fn from(error: url::ParseError) -> Self {
180        Self::InvalidUrl(error.to_string())
181    }
182}
183
184#[cfg(test)]
185mod tests {
186    use super::*;
187
188    #[test]
189    fn resolves_the_websocket_path_version_params_and_auth_protocol() {
190        futures::executor::block_on(async {
191            let endpoint = Endpoint::new("wss://example.test/socket?existing=yes")
192                .unwrap()
193                .connection_config(
194                    ConnectionConfig::default()
195                        .param("user_id", "7")
196                        .auth_token("1234"),
197                );
198
199            let resolved = endpoint
200                .resolve(ConnectContext { attempt: 0 })
201                .await
202                .unwrap();
203
204            assert_eq!(
205                resolved.url,
206                "wss://example.test/socket/websocket?existing=yes&user_id=7&vsn=2.0.0"
207            );
208            assert_eq!(
209                resolved.protocols,
210                ["phoenix", "base64url.bearer.phx.MTIzNA"]
211            );
212        });
213    }
214
215    #[test]
216    fn debug_output_redacts_authentication_secrets() {
217        let config = ConnectionConfig::default().auth_token("secret-token");
218        let debug = format!("{config:?}");
219        assert!(!debug.contains("secret-token"));
220
221        let endpoint = futures::executor::block_on(async {
222            Endpoint::new("wss://example.test/socket")
223                .unwrap()
224                .connection_config(config)
225                .resolve(ConnectContext { attempt: 0 })
226                .await
227                .unwrap()
228        });
229        let debug = format!("{endpoint:?}");
230        assert!(!debug.contains("c2VjcmV0LXRva2Vu"));
231    }
232
233    #[test]
234    fn reloads_connection_configuration_for_every_attempt() {
235        futures::executor::block_on(async {
236            let endpoint = Endpoint::new("ws://example.test/socket/websocket?vsn=1.0.0")
237                .unwrap()
238                .connection_config_loader(Rc::new(|context| {
239                    Box::pin(async move {
240                        Ok(ConnectionConfig::default()
241                            .param("attempt", context.attempt.to_string()))
242                    })
243                }));
244
245            assert_eq!(
246                endpoint
247                    .resolve(ConnectContext { attempt: 3 })
248                    .await
249                    .unwrap()
250                    .url,
251                "ws://example.test/socket/websocket?attempt=3&vsn=2.0.0"
252            );
253        });
254    }
255}
phoenix_channel_client/client/endpoint.rs

phoenix_channel_client/client/
endpoint.rs
