From 3d82b460ccfa304e24a254364e292bdeb8a71af3 Mon Sep 17 00:00:00 2001 From: Orual Date: Thu, 16 Oct 2025 23:33:20 -0400 Subject: [PATCH] moving toward a matched architecture with the http side --- crates/jacquard-common/src/xrpc.rs | 6 +- .../jacquard-common/src/xrpc/subscription.rs | 300 +++++++++++++++++- crates/jacquard-oauth/src/client.rs | 123 ++++--- .../jacquard/src/client/credential_session.rs | 144 ++++++--- 4 files changed, 479 insertions(+), 94 deletions(-) diff --git a/crates/jacquard-common/src/xrpc.rs b/crates/jacquard-common/src/xrpc.rs index 84ddfa49f..88342c0b0 100644 --- a/crates/jacquard-common/src/xrpc.rs +++ b/crates/jacquard-common/src/xrpc.rs @@ -20,7 +20,11 @@ pub use streaming::StreamingResponse; pub mod subscription; #[cfg(feature = "websocket")] -pub use subscription::{MessageEncoding, SubscriptionEndpoint, SubscriptionResp, XrpcSubscription}; +pub use subscription::{ + BasicSubscriptionClient, MessageEncoding, SubscriptionCall, SubscriptionClient, + SubscriptionEndpoint, SubscriptionExt, SubscriptionOptions, SubscriptionResp, + TungsteniteSubscriptionClient, XrpcSubscription, +}; use bytes::Bytes; use http::{ diff --git a/crates/jacquard-common/src/xrpc/subscription.rs b/crates/jacquard-common/src/xrpc/subscription.rs index e60954a74..409bcf35b 100644 --- a/crates/jacquard-common/src/xrpc/subscription.rs +++ b/crates/jacquard-common/src/xrpc/subscription.rs @@ -5,8 +5,11 @@ use serde::{Deserialize, Serialize}; use std::error::Error; +use std::future::Future; +use url::Url; -use crate::IntoStatic; +use crate::websocket::{WebSocketClient, WebSocketConnection}; +use crate::{CowStr, IntoStatic}; /// Encoding format for subscription messages #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -93,3 +96,298 @@ pub trait SubscriptionEndpoint { /// Stream response type type Stream: SubscriptionResp; } + +/// Per-subscription options for WebSocket subscriptions. +#[derive(Debug, Default, Clone)] +pub struct SubscriptionOptions<'a> { + /// Extra headers to attach to this subscription (e.g., Authorization). + pub headers: Vec<(CowStr<'a>, CowStr<'a>)>, +} + +impl IntoStatic for SubscriptionOptions<'_> { + type Output = SubscriptionOptions<'static>; + + fn into_static(self) -> Self::Output { + SubscriptionOptions { + headers: self + .headers + .into_iter() + .map(|(k, v)| (k.into_static(), v.into_static())) + .collect(), + } + } +} + +/// Extension for stateless subscription calls on any `WebSocketClient`. +/// +/// Provides a builder pattern for establishing WebSocket subscriptions with custom options. +pub trait SubscriptionExt: WebSocketClient { + /// Start building a subscription call for the given base URL. + fn subscription<'a>(&'a self, base: Url) -> SubscriptionCall<'a, Self> + where + Self: Sized, + { + SubscriptionCall { + client: self, + base, + opts: SubscriptionOptions::default(), + } + } +} + +impl SubscriptionExt for T {} + +/// Stateless subscription call builder. +/// +/// Provides methods for adding headers and establishing typed subscriptions. +pub struct SubscriptionCall<'a, C: WebSocketClient> { + pub(crate) client: &'a C, + pub(crate) base: Url, + pub(crate) opts: SubscriptionOptions<'a>, +} + +impl<'a, C: WebSocketClient> SubscriptionCall<'a, C> { + /// Add an extra header. + pub fn header(mut self, name: impl Into>, value: impl Into>) -> Self { + self.opts.headers.push((name.into(), value.into())); + self + } + + /// Replace the builder's options entirely. + pub fn with_options(mut self, opts: SubscriptionOptions<'a>) -> Self { + self.opts = opts; + self + } + + /// Subscribe to the given XRPC subscription endpoint. + /// + /// Builds a WebSocket URL from the base, appends the NSID path, + /// encodes query parameters from the subscription type, and connects. + pub async fn subscribe(self, params: &Sub) -> Result + where + Sub: XrpcSubscription, + { + let mut url = self.base.clone(); + let mut path = url.path().trim_end_matches('/').to_owned(); + path.push_str("/xrpc/"); + path.push_str(Sub::NSID); + url.set_path(&path); + + let query_params = params.query_params(); + if !query_params.is_empty() { + let qs = query_params + .iter() + .map(|(k, v)| format!("{}={}", k, v)) + .collect::>() + .join("&"); + url.set_query(Some(&qs)); + } else { + url.set_query(None); + } + + self.client + .connect_with_headers(url, self.opts.headers) + .await + } +} + +/// Stateful subscription client trait. +/// +/// Analogous to `XrpcClient` but for WebSocket subscriptions. +/// Provides a stateful interface for subscribing with configured base URI and options. +#[cfg_attr(not(target_arch = "wasm32"), trait_variant::make(Send))] +pub trait SubscriptionClient: WebSocketClient { + /// Get the base URI for the client. + fn base_uri(&self) -> Url; + + /// Get the subscription options for the client. + fn subscription_opts(&self) -> impl Future> { + async { SubscriptionOptions::default() } + } + + /// Subscribe to an XRPC subscription endpoint using the client's base URI and options. + #[cfg(not(target_arch = "wasm32"))] + fn subscribe( + &self, + params: &Sub, + ) -> impl Future> + where + Sub: XrpcSubscription + Send + Sync, + Self: Sync; + + /// Subscribe to an XRPC subscription endpoint using the client's base URI and options. + #[cfg(target_arch = "wasm32")] + fn subscribe( + &self, + params: &Sub, + ) -> impl Future> + where + Sub: XrpcSubscription + Send + Sync; + + /// Subscribe with custom options. + #[cfg(not(target_arch = "wasm32"))] + fn subscribe_with_opts( + &self, + params: &Sub, + opts: SubscriptionOptions<'_>, + ) -> impl Future> + where + Sub: XrpcSubscription + Send + Sync, + Self: Sync; + + /// Subscribe with custom options. + #[cfg(target_arch = "wasm32")] + fn subscribe_with_opts( + &self, + params: &Sub, + opts: SubscriptionOptions<'_>, + ) -> impl Future> + where + Sub: XrpcSubscription + Send + Sync; +} + +/// Simple stateless subscription client wrapping a WebSocketClient. +/// +/// Analogous to a basic HTTP client but for WebSocket subscriptions. +/// Does not manage sessions or authentication - useful for public subscriptions +/// or when you want to handle auth manually via headers. +pub struct BasicSubscriptionClient { + client: W, + base_uri: Url, + opts: SubscriptionOptions<'static>, +} + +impl BasicSubscriptionClient { + /// Create a new basic subscription client with the given WebSocket client and base URI. + pub fn new(client: W, base_uri: Url) -> Self { + Self { + client, + base_uri, + opts: SubscriptionOptions::default(), + } + } + + /// Create with default options. + pub fn with_options(mut self, opts: SubscriptionOptions<'_>) -> Self { + self.opts = opts.into_static(); + self + } + + /// Get a reference to the inner WebSocket client. + pub fn inner(&self) -> &W { + &self.client + } +} + +impl WebSocketClient for BasicSubscriptionClient { + type Error = W::Error; + + async fn connect(&self, url: Url) -> Result { + self.client.connect(url).await + } + + async fn connect_with_headers( + &self, + url: Url, + headers: Vec<(CowStr<'_>, CowStr<'_>)>, + ) -> Result { + self.client.connect_with_headers(url, headers).await + } +} + +impl SubscriptionClient for BasicSubscriptionClient { + fn base_uri(&self) -> Url { + self.base_uri.clone() + } + + async fn subscription_opts(&self) -> SubscriptionOptions<'_> { + self.opts.clone() + } + + #[cfg(not(target_arch = "wasm32"))] + async fn subscribe( + &self, + params: &Sub, + ) -> Result + where + Sub: XrpcSubscription + Send + Sync, + Self: Sync, + { + let opts = self.subscription_opts().await; + self.subscribe_with_opts(params, opts).await + } + + #[cfg(target_arch = "wasm32")] + async fn subscribe( + &self, + params: &Sub, + ) -> Result + where + Sub: XrpcSubscription + Send + Sync, + { + let opts = self.subscription_opts().await; + self.subscribe_with_opts(params, opts).await + } + + #[cfg(not(target_arch = "wasm32"))] + async fn subscribe_with_opts( + &self, + params: &Sub, + opts: SubscriptionOptions<'_>, + ) -> Result + where + Sub: XrpcSubscription + Send + Sync, + Self: Sync, + { + let base = self.base_uri(); + self.subscription(base) + .with_options(opts) + .subscribe(params) + .await + } + + #[cfg(target_arch = "wasm32")] + async fn subscribe_with_opts( + &self, + params: &Sub, + opts: SubscriptionOptions<'_>, + ) -> Result + where + Sub: XrpcSubscription + Send + Sync, + { + let base = self.base_uri(); + self.subscription(base) + .with_options(opts) + .subscribe(params) + .await + } +} + +/// Type alias for a basic subscription client using the default TungsteniteClient. +/// +/// Provides a simple, stateless WebSocket subscription client without session management. +/// Useful for public subscriptions or when handling authentication manually. +/// +/// # Example +/// +/// ```no_run +/// # use jacquard_common::xrpc::{TungsteniteSubscriptionClient, SubscriptionClient}; +/// # use url::Url; +/// # #[tokio::main] +/// # async fn main() -> Result<(), Box> { +/// let base = Url::parse("wss://bsky.network")?; +/// let client = TungsteniteSubscriptionClient::from_base_uri(base); +/// // let conn = client.subscribe(¶ms).await?; +/// # Ok(()) +/// # } +/// ``` +pub type TungsteniteSubscriptionClient = + BasicSubscriptionClient; + +impl TungsteniteSubscriptionClient { + /// Create a new Tungstenite-backed subscription client with the given base URI. + pub fn from_base_uri(base_uri: Url) -> Self { + let client = crate::websocket::tungstenite_client::TungsteniteClient::new(); + BasicSubscriptionClient::new(client, base_uri) + } +} diff --git a/crates/jacquard-oauth/src/client.rs b/crates/jacquard-oauth/src/client.rs index 439908cde..e6ddef2e4 100644 --- a/crates/jacquard-oauth/src/client.rs +++ b/crates/jacquard-oauth/src/client.rs @@ -436,49 +436,6 @@ where } } -#[cfg(feature = "websocket")] -impl OAuthSession -where - S: ClientAuthStore, - T: OAuthResolver, - W: WebSocketClient, -{ - /// Subscribe to an XRPC WebSocket subscription. - /// - /// Connects to the WebSocket endpoint and threads through DPoP authentication headers. - pub async fn subscribe(&self, params: &Sub) -> Result - where - Sub: XrpcSubscription, - { - let base_uri = self.endpoint().await; - - // Build WebSocket URL - let mut ws_url = base_uri.clone(); - ws_url.set_scheme("wss").ok(); - ws_url.set_path(&format!("/xrpc/{}", Sub::NSID)); - - // Add query params - let query_params = params.query_params(); - if !query_params.is_empty() { - let query_string = serde_html_form::to_string(&query_params).unwrap_or_default(); - ws_url.set_query(Some(&query_string)); - } - - // Thread DPoP auth headers (even though tokio-tungstenite-wasm doesn't support them yet) - let token = self.access_token().await; - let auth_value = match token { - AuthorizationToken::Bearer(t) => format!("Bearer {}", t.as_ref()), - AuthorizationToken::Dpop(t) => format!("DPoP {}", t.as_ref()), - }; - let headers = vec![( - CowStr::from("Authorization"), - CowStr::from(auth_value), - )]; - - self.ws_client.connect_with_headers(ws_url, headers).await - } -} - impl HttpClient for OAuthSession where S: ClientAuthStore + Send + Sync + 'static, @@ -605,3 +562,83 @@ where async { self.client.resolve_did_doc(did).await } } } + +#[cfg(feature = "websocket")] +impl WebSocketClient for OAuthSession +where + S: ClientAuthStore + Send + Sync + 'static, + T: OAuthResolver + Send + Sync + 'static, + W: WebSocketClient + Send + Sync, +{ + type Error = W::Error; + + async fn connect(&self, url: Url) -> std::result::Result { + self.ws_client.connect(url).await + } + + async fn connect_with_headers( + &self, + url: Url, + headers: Vec<(CowStr<'_>, CowStr<'_>)>, + ) -> std::result::Result { + self.ws_client.connect_with_headers(url, headers).await + } +} + +#[cfg(feature = "websocket")] +impl jacquard_common::xrpc::SubscriptionClient for OAuthSession +where + S: ClientAuthStore + Send + Sync + 'static, + T: OAuthResolver + Send + Sync + 'static, + W: WebSocketClient + Send + Sync, +{ + fn base_uri(&self) -> Url { + #[cfg(not(target_arch = "wasm32"))] + if tokio::runtime::Handle::try_current().is_ok() { + return tokio::task::block_in_place(|| self.data.blocking_read().host_url.clone()); + } + + self.data.blocking_read().host_url.clone() + } + + async fn subscription_opts(&self) -> jacquard_common::xrpc::SubscriptionOptions<'_> { + let mut opts = jacquard_common::xrpc::SubscriptionOptions::default(); + let token = self.access_token().await; + let auth_value = match token { + AuthorizationToken::Bearer(t) => format!("Bearer {}", t.as_ref()), + AuthorizationToken::Dpop(t) => format!("DPoP {}", t.as_ref()), + }; + opts.headers.push(( + CowStr::from("Authorization"), + CowStr::from(auth_value), + )); + opts + } + + async fn subscribe( + &self, + params: &Sub, + ) -> std::result::Result + where + Sub: XrpcSubscription + Send + Sync, + { + let opts = self.subscription_opts().await; + self.subscribe_with_opts(params, opts).await + } + + async fn subscribe_with_opts( + &self, + params: &Sub, + opts: jacquard_common::xrpc::SubscriptionOptions<'_>, + ) -> std::result::Result + where + Sub: XrpcSubscription + Send + Sync, + { + use jacquard_common::xrpc::SubscriptionExt; + let base = self.base_uri(); + self.subscription(base) + .with_options(opts) + .subscribe(params) + .await + } +} diff --git a/crates/jacquard/src/client/credential_session.rs b/crates/jacquard/src/client/credential_session.rs index f7bdce9c5..a7abf66f4 100644 --- a/crates/jacquard/src/client/credential_session.rs +++ b/crates/jacquard/src/client/credential_session.rs @@ -411,55 +411,6 @@ where } } -#[cfg(feature = "websocket")] -impl CredentialSession -where - S: SessionStore, - W: WebSocketClient, -{ - /// Subscribe to an XRPC WebSocket subscription. - /// - /// Connects to the WebSocket endpoint and threads through authentication headers. - pub async fn subscribe( - &self, - params: &Sub, - ) -> Result - where - Sub: XrpcSubscription, - { - let base_uri = self.endpoint().await; - - // Build WebSocket URL - let mut ws_url = base_uri.clone(); - ws_url.set_scheme("wss").ok(); - ws_url.set_path(&format!("/xrpc/{}", Sub::NSID)); - - // Add query params - let query_params = params.query_params(); - if !query_params.is_empty() { - let query_string = serde_html_form::to_string(&query_params) - .unwrap_or_default(); - ws_url.set_query(Some(&query_string)); - } - - // Thread auth headers (even though tokio-tungstenite-wasm doesn't support them yet) - let headers = if let Some(token) = self.access_token().await { - let auth_value = match token { - AuthorizationToken::Bearer(t) => format!("Bearer {}", t.as_ref()), - AuthorizationToken::Dpop(t) => format!("DPoP {}", t.as_ref()), - }; - vec![( - CowStr::from("Authorization"), - CowStr::from(auth_value), - )] - } else { - vec![] - }; - - self.ws_client.connect_with_headers(ws_url, headers).await - } -} - impl HttpClient for CredentialSession where S: SessionStore + Send + Sync + 'static, @@ -585,3 +536,98 @@ where async { self.client.resolve_did_doc(did).await } } } + +#[cfg(feature = "websocket")] +impl WebSocketClient for CredentialSession +where + S: SessionStore + Send + Sync + 'static, + T: Send + Sync + 'static, + W: WebSocketClient + Send + Sync, +{ + type Error = W::Error; + + async fn connect(&self, url: Url) -> Result { + self.ws_client.connect(url).await + } + + async fn connect_with_headers( + &self, + url: Url, + headers: Vec<(CowStr<'_>, CowStr<'_>)>, + ) -> Result { + self.ws_client.connect_with_headers(url, headers).await + } +} + +#[cfg(feature = "websocket")] +impl jacquard_common::xrpc::SubscriptionClient for CredentialSession +where + S: SessionStore + Send + Sync + 'static, + T: Send + Sync + 'static, + W: WebSocketClient + Send + Sync, +{ + fn base_uri(&self) -> Url { + #[cfg(not(target_arch = "wasm32"))] + if tokio::runtime::Handle::try_current().is_ok() { + tokio::task::block_in_place(|| { + self.endpoint.blocking_read().clone().unwrap_or( + Url::parse("https://public.bsky.app") + .expect("public appview should be valid url"), + ) + }) + } else { + self.endpoint.blocking_read().clone().unwrap_or( + Url::parse("https://public.bsky.app").expect("public appview should be valid url"), + ) + } + + #[cfg(target_arch = "wasm32")] + { + self.endpoint.blocking_read().clone().unwrap_or( + Url::parse("https://public.bsky.app").expect("public appview should be valid url"), + ) + } + } + + async fn subscription_opts(&self) -> jacquard_common::xrpc::SubscriptionOptions<'_> { + let mut opts = jacquard_common::xrpc::SubscriptionOptions::default(); + if let Some(token) = self.access_token().await { + let auth_value = match token { + AuthorizationToken::Bearer(t) => format!("Bearer {}", t.as_ref()), + AuthorizationToken::Dpop(t) => format!("DPoP {}", t.as_ref()), + }; + opts.headers.push(( + CowStr::from("Authorization"), + CowStr::from(auth_value), + )); + } + opts + } + + async fn subscribe( + &self, + params: &Sub, + ) -> Result + where + Sub: XrpcSubscription + Send + Sync, + { + let opts = self.subscription_opts().await; + self.subscribe_with_opts(params, opts).await + } + + async fn subscribe_with_opts( + &self, + params: &Sub, + opts: jacquard_common::xrpc::SubscriptionOptions<'_>, + ) -> Result + where + Sub: XrpcSubscription + Send + Sync, + { + use jacquard_common::xrpc::SubscriptionExt; + let base = self.base_uri(); + self.subscription(base) + .with_options(opts) + .subscribe(params) + .await + } +} -- 2.51.2