diff --git a/.gitignore b/.gitignore index 24a4d865..b0dfa4bf 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,4 @@ AGENTS.md crates/jacquard-lexicon/tests/fixtures/lexicons/atproto crates/jacquard-lexicon/target codegen_plan.md +/lex_js diff --git a/README.md b/README.md index 6e9fdb37..d9b6590d 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ use clap::Parser; use jacquard::CowStr; use jacquard::api::app_bsky::feed::get_timeline::GetTimeline; use jacquard::api::com_atproto::server::create_session::CreateSession; -use jacquard::client::{AuthenticatedClient, Session, XrpcClient}; +use jacquard::client::{BasicClient, Session}; use miette::IntoDiagnostic; #[derive(Parser, Debug)] @@ -49,7 +49,8 @@ async fn main() -> miette::Result<()> { let args = Args::parse(); // Create HTTP client - let mut client = AuthenticatedClient::new(reqwest::Client::new(), args.pds); + let base = url::Url::parse(&args.pds).into_diagnostic()?; + let client = BasicClient::new(base); // Create session let session = Session::from( @@ -65,7 +66,7 @@ async fn main() -> miette::Result<()> { ); println!("logged in as {} ({})", session.handle, session.did); - client.set_session(session); + client.set_session(session).await.into_diagnostic()?; // Fetch timeline println!("\nfetching timeline..."); diff --git a/crates/jacquard-api/Cargo.toml b/crates/jacquard-api/Cargo.toml index dd2f091d..7125f4ab 100644 --- a/crates/jacquard-api/Cargo.toml +++ b/crates/jacquard-api/Cargo.toml @@ -11,6 +11,9 @@ readme.workspace = true exclude.workspace = true license.workspace = true +[package.metadata.docs.rs] +features = [ "com_atproto", "app_bsky", "chat_bsky", "tools_ozone" ] + [features] default = [ "com_atproto"] app_bsky = [] diff --git a/crates/jacquard-common/Cargo.toml b/crates/jacquard-common/Cargo.toml index 58bfd9d2..9f49e435 100644 --- a/crates/jacquard-common/Cargo.toml +++ b/crates/jacquard-common/Cargo.toml @@ -60,3 +60,6 @@ version = "0.13" optional = true default-features = false features = ["arithmetic"] + +[package.metadata.docs.rs] +features = [ "crypto-k256", "crypto-k256", "crypto-p256"] diff --git a/crates/jacquard/Cargo.toml b/crates/jacquard/Cargo.toml index deb8f9d7..495e6347 100644 --- a/crates/jacquard/Cargo.toml +++ b/crates/jacquard/Cargo.toml @@ -12,7 +12,7 @@ exclude.workspace = true license.workspace = true [features] -default = ["api_all"] +default = ["api_all", "dns"] derive = ["dep:jacquard-derive"] api = ["jacquard-api/com_atproto"] api_all = ["api", "jacquard-api/app_bsky", "jacquard-api/chat_bsky", "jacquard-api/tools_ozone"] @@ -42,7 +42,7 @@ serde_html_form.workspace = true serde_ipld_dagcbor.workspace = true serde_json.workspace = true thiserror.workspace = true -tokio = { version = "1", features = ["macros", "rt-multi-thread"] } +tokio = { version = "1", features = ["macros", "rt-multi-thread", "fs"] } hickory-resolver = { version = "0.24", default-features = false, features = ["system-config", "tokio-runtime"], optional = true } url.workspace = true smol_str.workspace = true diff --git a/crates/jacquard/src/client.rs b/crates/jacquard/src/client.rs index 15a85c55..aef99d5e 100644 --- a/crates/jacquard/src/client.rs +++ b/crates/jacquard/src/client.rs @@ -3,19 +3,24 @@ //! This module provides HTTP and XRPC client traits along with an authenticated //! client implementation that manages session tokens. +mod at_client; mod error; mod response; +mod token; +mod xrpc_call; use std::fmt::Display; use std::future::Future; -use bytes::Bytes; +pub use at_client::{AtClient, SendOverrides}; pub use error::{ClientError, Result}; use http::{ HeaderName, HeaderValue, Request, - header::{AUTHORIZATION, CONTENT_TYPE, InvalidHeaderValue}, + header::{AUTHORIZATION, CONTENT_TYPE}, }; pub use response::Response; +pub use token::{FileTokenStore, MemoryTokenStore, TokenStore, TokenStoreError}; +pub use xrpc_call::{CallOptions, XrpcCall, XrpcExt}; use jacquard_common::{ CowStr, IntoStatic, @@ -24,6 +29,7 @@ use jacquard_common::{ xrpc::{XrpcMethod, XrpcRequest}, }, }; +use url::Url; /// Implement HttpClient for reqwest::Client impl HttpClient for reqwest::Client { @@ -61,7 +67,7 @@ impl HttpClient for reqwest::Client { } } -/// HTTP client trait for sending raw HTTP requests +/// HTTP client trait for sending raw HTTP requests. pub trait HttpClient { /// Error type returned by the HTTP client type Error: std::error::Error + Display + Send + Sync + 'static; @@ -71,38 +77,12 @@ pub trait HttpClient { request: Request>, ) -> impl Future>, Self::Error>> + Send; } -/// XRPC client trait for AT Protocol RPC calls -pub trait XrpcClient: HttpClient + Sync { - /// Get the base URI for XRPC requests (e.g., "https://bsky.social") - fn base_uri(&self) -> CowStr<'_>; - /// Get the authorization token for XRPC requests - #[allow(unused_variables)] - fn authorization_token( - &self, - is_refresh: bool, - ) -> impl Future>> + Send { - async { None } - } - /// Get the `atproto-proxy` header. - fn atproto_proxy_header(&self) -> impl Future> + Send { - async { None } - } - /// Get the `atproto-accept-labelers` header. - fn atproto_accept_labelers_header(&self) -> impl Future>> + Send { - async { None } - } - /// Send an XRPC request and get back a response - fn send(&self, request: R) -> impl Future>> + Send - where - Self: Sized + Sync, - { - send_xrpc(self, request) - } -} +// Note: Stateless and stateful XRPC clients are implemented in xrpc_call.rs and at_client.rs pub(crate) const NSID_REFRESH_SESSION: &str = "com.atproto.server.refreshSession"; -/// Authorization token types for XRPC requests +/// Authorization token types for XRPC requests. +#[derive(Debug, Clone)] pub enum AuthorizationToken<'s> { /// Bearer token (access JWT, refresh JWT to refresh the session) Bearer(CowStr<'s>), @@ -110,14 +90,56 @@ pub enum AuthorizationToken<'s> { Dpop(CowStr<'s>), } -impl TryFrom> for HeaderValue { - type Error = InvalidHeaderValue; +/// Basic client wrapper: reqwest transport + in-memory token store. +pub struct BasicClient(AtClient); + +impl BasicClient { + /// Construct a basic client with minimal inputs. + pub fn new(base: Url) -> Self { + Self(AtClient::new( + reqwest::Client::new(), + base, + MemoryTokenStore::default(), + )) + } + + /// Access the inner stateful client. + pub fn inner(&self) -> &AtClient { + &self.0 + } + + /// Send an XRPC request. + pub async fn send(&self, req: R) -> Result> { + self.0.send(req).await + } + + /// Send with per-call overrides. + pub async fn send_with( + &self, + req: R, + overrides: SendOverrides<'_>, + ) -> Result> { + self.0.send_with(req, overrides).await + } - fn try_from(token: AuthorizationToken) -> core::result::Result { - HeaderValue::from_str(&match token { - AuthorizationToken::Bearer(t) => format!("Bearer {t}"), - AuthorizationToken::Dpop(t) => format!("DPoP {t}"), - }) + /// Get current session. + pub async fn session(&self) -> Option { + self.0.session().await + } + + /// Set the session. + pub async fn set_session(&self, session: Session) -> core::result::Result<(), TokenStoreError> { + self.0.set_session(session).await + } + + /// Clear session. + pub async fn clear_session(&self) -> core::result::Result<(), TokenStoreError> { + self.0.clear_session().await + } + + /// Base URL of this client. + pub fn base(&self) -> &Url { + self.0.base() } } @@ -146,86 +168,80 @@ impl From
for HeaderName { } } -/// Generic XRPC send implementation that uses HttpClient -async fn send_xrpc(client: &C, request: R) -> Result> -where - R: XrpcRequest + Send, - C: XrpcClient + ?Sized + Sync, -{ - // Build URI: base_uri + /xrpc/ + NSID - let mut uri = format!("{}/xrpc/{}", client.base_uri(), R::NSID); +/// Build an HTTP request for an XRPC call given base URL and options +pub(crate) fn build_http_request( + base: &Url, + req: &R, + opts: &xrpc_call::CallOptions<'_>, +) -> core::result::Result>, error::TransportError> { + let mut url = base.clone(); + let mut path = url.path().trim_end_matches('/').to_owned(); + path.push_str("/xrpc/"); + path.push_str(R::NSID); + url.set_path(&path); - // Add query parameters for Query methods if let XrpcMethod::Query = R::METHOD { - let qs = serde_html_form::to_string(&request).map_err(error::EncodeError::from)?; + let qs = serde_html_form::to_string(&req) + .map_err(|e| error::TransportError::InvalidRequest(e.to_string()))?; if !qs.is_empty() { - uri.push('?'); - uri.push_str(&qs); + url.set_query(Some(&qs)); + } else { + url.set_query(None); } } - // Build HTTP request let method = match R::METHOD { XrpcMethod::Query => http::Method::GET, XrpcMethod::Procedure(_) => http::Method::POST, }; - let mut builder = Request::builder().method(method).uri(&uri); + let mut builder = Request::builder().method(method).uri(url.as_str()); - // Add Content-Type for procedures if let XrpcMethod::Procedure(encoding) = R::METHOD { builder = builder.header(Header::ContentType, encoding); } - - // Add authorization header - let is_refresh = R::NSID == NSID_REFRESH_SESSION; - if let Some(token) = client.authorization_token(is_refresh).await { - let header_value: HeaderValue = token.try_into().map_err(|e| { + builder = builder.header(http::header::ACCEPT, R::OUTPUT_ENCODING); + + if let Some(token) = &opts.auth { + let hv = match token { + AuthorizationToken::Bearer(t) => { + HeaderValue::from_str(&format!("Bearer {}", t.as_ref())) + } + AuthorizationToken::Dpop(t) => HeaderValue::from_str(&format!("DPoP {}", t.as_ref())), + } + .map_err(|e| { error::TransportError::InvalidRequest(format!("Invalid authorization token: {}", e)) })?; - builder = builder.header(Header::Authorization, header_value); + builder = builder.header(Header::Authorization, hv); } - // Add atproto-proxy header - if let Some(proxy) = client.atproto_proxy_header().await { - builder = builder.header(Header::AtprotoProxy, proxy); + if let Some(proxy) = &opts.atproto_proxy { + builder = builder.header(Header::AtprotoProxy, proxy.as_ref()); } - - // Add atproto-accept-labelers header - if let Some(labelers) = client.atproto_accept_labelers_header().await { - builder = builder.header(Header::AtprotoAcceptLabelers, labelers.join(", ")); + if let Some(labelers) = &opts.atproto_accept_labelers { + if !labelers.is_empty() { + let joined = labelers + .iter() + .map(|s| s.as_ref()) + .collect::>() + .join(", "); + builder = builder.header(Header::AtprotoAcceptLabelers, joined); + } + } + for (name, value) in &opts.extra_headers { + builder = builder.header(name, value); } - // Serialize body for procedures let body = if let XrpcMethod::Procedure(_) = R::METHOD { - request.encode_body()? + req.encode_body() + .map_err(|e| error::TransportError::InvalidRequest(e.to_string()))? } else { vec![] }; - // TODO: make this not panic - let http_request = builder.body(body).expect("Failed to build HTTP request"); - - // Send HTTP request - let http_response = client - .send_http(http_request) - .await - .map_err(|e| error::TransportError::Other(Box::new(e)))?; - - let status = http_response.status(); - let buffer = Bytes::from(http_response.into_body()); - - // XRPC errors come as 400/401 with structured error bodies - // Other error status codes (404, 500, etc.) are generic HTTP errors - if !status.is_success() && !matches!(status.as_u16(), 400 | 401) { - return Err(ClientError::Http(error::HttpError { - status, - body: Some(buffer), - })); - } - - // Response will parse XRPC errors for 400/401, or output for 2xx - Ok(Response::new(buffer, status)) + builder + .body(body) + .map_err(|e| error::TransportError::InvalidRequest(e.to_string())) } /// Session information from `com.atproto.server.createSession` @@ -256,78 +272,17 @@ impl From { - client: C, - base_uri: CowStr<'static>, - session: Option, -} - -impl AuthenticatedClient { - /// Create a new authenticated client with a base URI - /// - /// # Example - /// ```ignore - /// let client = AuthenticatedClient::new( - /// reqwest::Client::new(), - /// CowStr::from("https://bsky.social") - /// ); - /// ``` - pub fn new(client: C, base_uri: CowStr<'static>) -> Self { +impl From> + for Session +{ + fn from( + output: jacquard_api::com_atproto::server::refresh_session::RefreshSessionOutput<'_>, + ) -> Self { Self { - client, - base_uri: base_uri, - session: None, - } - } - - /// Set the session obtained from `createSession` or `refreshSession` - pub fn set_session(&mut self, session: Session) { - self.session = Some(session); - } - - /// Get the current session if one exists - pub fn session(&self) -> Option<&Session> { - self.session.as_ref() - } - - /// Clear the current session locally - /// - /// Note: This only clears the local session state. To properly revoke the session - /// server-side, use `com.atproto.server.deleteSession` before calling this. - pub fn clear_session(&mut self) { - self.session = None; - } -} - -impl HttpClient for AuthenticatedClient { - type Error = C::Error; - - fn send_http( - &self, - request: Request>, - ) -> impl Future>, Self::Error>> { - self.client.send_http(request) - } -} - -impl XrpcClient for AuthenticatedClient { - fn base_uri(&self) -> CowStr<'_> { - self.base_uri.clone() - } - - async fn authorization_token(&self, is_refresh: bool) -> Option> { - if is_refresh { - self.session - .as_ref() - .map(|s| AuthorizationToken::Bearer(s.refresh_jwt.clone())) - } else { - self.session - .as_ref() - .map(|s| AuthorizationToken::Bearer(s.access_jwt.clone())) + access_jwt: output.access_jwt.into_static(), + refresh_jwt: output.refresh_jwt.into_static(), + did: output.did.into_static(), + handle: output.handle.into_static(), } } } diff --git a/crates/jacquard/src/client/at_client.rs b/crates/jacquard/src/client/at_client.rs new file mode 100644 index 00000000..f56f7995 --- /dev/null +++ b/crates/jacquard/src/client/at_client.rs @@ -0,0 +1,232 @@ +use bytes::Bytes; +use url::Url; + +use crate::client::xrpc_call::{CallOptions, XrpcExt}; +use crate::client::{self as super_mod, AuthorizationToken, HttpClient, Response, Session, error}; +use jacquard_common::types::xrpc::XrpcRequest; + +use super::token::TokenStore; + +/// Per-call overrides when sending via `AtClient`. +#[derive(Debug, Default, Clone)] +pub struct SendOverrides<'a> { + /// Optional base URI override for this call. + pub base_uri: Option, + /// Per-request options such as auth, proxy, labelers, extra headers. + pub options: CallOptions<'a>, + /// Whether to auto-refresh on expired/invalid token and retry once. + pub auto_refresh: bool, +} + +impl<'a> SendOverrides<'a> { + /// Construct default overrides (no base override, auto-refresh enabled). + pub fn new() -> Self { + Self { + base_uri: None, + options: CallOptions::default(), + auto_refresh: true, + } + } + /// Override the base URI for this call only. + pub fn base_uri(mut self, base: Url) -> Self { + self.base_uri = Some(base); + self + } + /// Provide a full set of call options (auth/headers/etc.). + pub fn options(mut self, opts: CallOptions<'a>) -> Self { + self.options = opts; + self + } + /// Enable or disable one-shot auto-refresh + retry behavior. + pub fn auto_refresh(mut self, enable: bool) -> Self { + self.auto_refresh = enable; + self + } +} + +/// Stateful client for AT Protocol XRPC with token storage and auto-refresh. +/// +/// Example (file-backed tokens) +/// ```ignore +/// use jacquard::client::{AtClient, FileTokenStore, TokenStore}; +/// use jacquard::api::com_atproto::server::create_session::CreateSession; +/// use jacquard::client::AtClient as _; // method resolution +/// use jacquard::CowStr; +/// +/// #[tokio::main] +/// async fn main() -> miette::Result<()> { +/// let base = url::Url::parse("https://bsky.social")?; +/// let store = FileTokenStore::new("/tmp/jacquard-session.json"); +/// let client = AtClient::new(reqwest::Client::new(), base, store); +/// let session = client +/// .send( +/// CreateSession::new() +/// .identifier(CowStr::from("alice.example")) +/// .password(CowStr::from("app-password")) +/// .build(), +/// ) +/// .await? +/// .into_output()?; +/// client.set_session(session.into()).await?; +/// Ok(()) +/// } +/// ``` +pub struct AtClient { + transport: C, + base: Url, + tokens: S, + refresh_lock: tokio::sync::Mutex<()>, +} + +impl AtClient { + /// Create a new client with a transport, base URL, and token store. + pub fn new(transport: C, base: Url, tokens: S) -> Self { + Self { + transport, + base, + tokens, + refresh_lock: tokio::sync::Mutex::new(()), + } + } + + /// Get the base URL of this client. + pub fn base(&self) -> &Url { + &self.base + } + + /// Access the underlying transport. + pub fn transport(&self) -> &C { + &self.transport + } + + /// Get the current session, if any. + pub async fn session(&self) -> Option { + self.tokens.get().await + } + + /// Set the current session in the token store. + pub async fn set_session(&self, session: Session) -> Result<(), super_mod::TokenStoreError> { + self.tokens.set(session).await + } + + /// Clear the current session from the token store. + pub async fn clear_session(&self) -> Result<(), super_mod::TokenStoreError> { + self.tokens.clear().await + } + + /// Send an XRPC request using the client's base URL and default behavior. + pub async fn send(&self, req: R) -> super_mod::Result> { + self.send_with(req, SendOverrides::new()).await + } + + /// Send an XRPC request with per-call overrides. + pub async fn send_with( + &self, + req: R, + mut overrides: SendOverrides<'_>, + ) -> super_mod::Result> { + let base = overrides + .base_uri + .clone() + .unwrap_or_else(|| self.base.clone()); + let is_refresh = R::NSID == super_mod::NSID_REFRESH_SESSION; + + if overrides.options.auth.is_none() { + if let Some(s) = self.tokens.get().await { + overrides.options.auth = Some(if is_refresh { + AuthorizationToken::Bearer(s.refresh_jwt) + } else { + AuthorizationToken::Bearer(s.access_jwt) + }); + } + } + + let http_request = super_mod::build_http_request(&base, &req, &overrides.options) + .map_err(error::TransportError::from)?; + let http_response = self + .transport + .send_http(http_request) + .await + .map_err(|e| error::TransportError::Other(Box::new(e)))?; + let status = http_response.status(); + let buffer = Bytes::from(http_response.into_body()); + + if !status.is_success() && !matches!(status.as_u16(), 400 | 401) { + return Err(error::HttpError { + status, + body: Some(buffer), + } + .into()); + } + + if overrides.auto_refresh + && !is_refresh + && overrides.options.auth.is_some() + && Self::is_auth_expired(status, &buffer) + { + self.refresh_once().await?; + + let mut retry_opts = overrides.options.clone(); + if let Some(s) = self.tokens.get().await { + retry_opts.auth = Some(AuthorizationToken::Bearer(s.access_jwt)); + } + let http_request = super_mod::build_http_request(&base, &req, &retry_opts) + .map_err(error::TransportError::from)?; + let http_response = self + .transport + .send_http(http_request) + .await + .map_err(|e| error::TransportError::Other(Box::new(e)))?; + let status = http_response.status(); + let buffer = Bytes::from(http_response.into_body()); + + if !status.is_success() && !matches!(status.as_u16(), 400 | 401) { + return Err(error::HttpError { + status, + body: Some(buffer), + } + .into()); + } + return Ok(Response::new(buffer, status)); + } + + Ok(Response::new(buffer, status)) + } + + async fn refresh_once(&self) -> super_mod::Result<()> { + let _guard = self.refresh_lock.lock().await; + let Some(s) = self.tokens.get().await else { + return Err(error::ClientError::Auth(error::AuthError::NotAuthenticated)); + }; + let refresh_token = s.refresh_jwt.clone(); + let refresh_resp = self + .transport + .xrpc(self.base.clone()) + .auth(AuthorizationToken::Bearer(refresh_token)) + .send(jacquard_api::com_atproto::server::refresh_session::RefreshSession) + .await?; + let refreshed = match refresh_resp.into_output() { + Ok(o) => Session::from(o), + Err(_) => return Err(error::ClientError::Auth(error::AuthError::RefreshFailed)), + }; + self.tokens + .set(refreshed) + .await + .map_err(|_| error::ClientError::Auth(error::AuthError::RefreshFailed))?; + Ok(()) + } + + fn is_auth_expired(status: http::StatusCode, buffer: &Bytes) -> bool { + if status.as_u16() == 401 { + return true; + } + if status.as_u16() == 400 { + if let Ok(val) = serde_json::from_slice::(buffer) { + if let Some(code) = val.get("error").and_then(|v| v.as_str()) { + return matches!(code, "ExpiredToken" | "InvalidToken"); + } + } + } + false + } +} diff --git a/crates/jacquard/src/client/token.rs b/crates/jacquard/src/client/token.rs new file mode 100644 index 00000000..05b2fcdb --- /dev/null +++ b/crates/jacquard/src/client/token.rs @@ -0,0 +1,125 @@ +use async_trait::async_trait; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use thiserror::Error; + +use super::Session; +use jacquard_common::IntoStatic; +use jacquard_common::types::string::{Did, Handle}; + +/// Errors emitted by token stores. +#[derive(Debug, Error)] +pub enum TokenStoreError { + /// An underlying I/O or serialization error with context. + #[error("token store error: {0}")] + Other(String), +} + +/// Pluggable session token storage (memory, disk, browser, etc.). +#[async_trait] +pub trait TokenStore: Send + Sync { + /// Get the current session if present. + async fn get(&self) -> Option; + /// Persist the given session. + async fn set(&self, session: Session) -> Result<(), TokenStoreError>; + /// Remove any stored session. + async fn clear(&self) -> Result<(), TokenStoreError>; +} + +/// In-memory token store suitable for short-lived sessions and tests. +#[derive(Default, Clone)] +pub struct MemoryTokenStore(Arc>>); + +#[async_trait] +impl TokenStore for MemoryTokenStore { + async fn get(&self) -> Option { + self.0.read().await.clone() + } + async fn set(&self, session: Session) -> Result<(), TokenStoreError> { + *self.0.write().await = Some(session); + Ok(()) + } + async fn clear(&self) -> Result<(), TokenStoreError> { + *self.0.write().await = None; + Ok(()) + } +} + +/// File-backed token store using a JSON file. +/// +/// Example +/// ```ignore +/// use jacquard::client::{AtClient, FileTokenStore}; +/// let base = url::Url::parse("https://bsky.social").unwrap(); +/// let store = FileTokenStore::new("/tmp/jacquard-session.json"); +/// let client = AtClient::new(reqwest::Client::new(), base, store); +/// ``` +#[derive(Clone, Debug)] +pub struct FileTokenStore { + path: PathBuf, +} + +impl FileTokenStore { + /// Create a new file token store at the given path. + pub fn new(path: impl AsRef) -> Self { + Self { + path: path.as_ref().to_path_buf(), + } + } +} + +#[derive(serde::Serialize, serde::Deserialize)] +struct FileSession { + access_jwt: String, + refresh_jwt: String, + did: String, + handle: String, +} + +#[async_trait] +impl TokenStore for FileTokenStore { + async fn get(&self) -> Option { + let data = tokio::fs::read(&self.path).await.ok()?; + let disk: FileSession = serde_json::from_slice(&data).ok()?; + let did = Did::new_owned(disk.did).ok()?; + let handle = Handle::new_owned(disk.handle).ok()?; + Some(Session { + access_jwt: disk.access_jwt.into(), + refresh_jwt: disk.refresh_jwt.into(), + did: did.into_static(), + handle: handle.into_static(), + }) + } + + async fn set(&self, session: Session) -> Result<(), TokenStoreError> { + let disk = FileSession { + access_jwt: session.access_jwt.to_string(), + refresh_jwt: session.refresh_jwt.to_string(), + did: session.did.to_string(), + handle: session.handle.to_string(), + }; + let buf = + serde_json::to_vec_pretty(&disk).map_err(|e| TokenStoreError::Other(e.to_string()))?; + if let Some(parent) = self.path.parent() { + tokio::fs::create_dir_all(parent) + .await + .map_err(|e| TokenStoreError::Other(e.to_string()))?; + } + let tmp = self.path.with_extension("tmp"); + tokio::fs::write(&tmp, &buf) + .await + .map_err(|e| TokenStoreError::Other(e.to_string()))?; + tokio::fs::rename(&tmp, &self.path) + .await + .map_err(|e| TokenStoreError::Other(e.to_string()))?; + Ok(()) + } + + async fn clear(&self) -> Result<(), TokenStoreError> { + match tokio::fs::remove_file(&self.path).await { + Ok(_) => Ok(()), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(e) => Err(TokenStoreError::Other(e.to_string())), + } + } +} diff --git a/crates/jacquard/src/client/xrpc_call.rs b/crates/jacquard/src/client/xrpc_call.rs new file mode 100644 index 00000000..c83b953d --- /dev/null +++ b/crates/jacquard/src/client/xrpc_call.rs @@ -0,0 +1,154 @@ +use bytes::Bytes; +use http::{HeaderName, HeaderValue}; +use url::Url; + +use crate::CowStr; +use crate::client::{self as super_mod, Response, error}; +use crate::client::{AuthorizationToken, HttpClient}; +use jacquard_common::types::xrpc::XrpcRequest; + +/// Per-request options for XRPC calls. +#[derive(Debug, Default, Clone)] +pub struct CallOptions<'a> { + /// Optional Authorization to apply (`Bearer` or `DPoP`). + pub auth: Option>, + /// `atproto-proxy` header value. + pub atproto_proxy: Option>, + /// `atproto-accept-labelers` header values. + pub atproto_accept_labelers: Option>>, + /// Extra headers to attach to this request. + pub extra_headers: Vec<(HeaderName, HeaderValue)>, +} + +/// Extension for stateless XRPC calls on any `HttpClient`. +/// +/// Example +/// ```ignore +/// use jacquard::client::XrpcExt; +/// use jacquard::api::app_bsky::feed::get_author_feed::GetAuthorFeed; +/// use jacquard::types::ident::AtIdentifier; +/// use miette::IntoDiagnostic; +/// +/// #[tokio::main] +/// async fn main() -> miette::Result<()> { +/// let http = reqwest::Client::new(); +/// let base = url::Url::parse("https://public.api.bsky.app")?; +/// let resp = http +/// .xrpc(base) +/// .send( +/// GetAuthorFeed::new() +/// .actor(AtIdentifier::new_static("pattern.atproto.systems").unwrap()) +/// .limit(5) +/// .build(), +/// ) +/// .await?; +/// let out = resp.into_output()?; +/// println!("author feed:\n{}", serde_json::to_string_pretty(&out).into_diagnostic()?); +/// Ok(()) +/// } +/// ``` +pub trait XrpcExt: HttpClient { + /// Start building an XRPC call for the given base URL. + fn xrpc<'a>(&'a self, base: Url) -> XrpcCall<'a, Self> + where + Self: Sized, + { + XrpcCall { + client: self, + base, + opts: CallOptions::default(), + } + } +} + +impl XrpcExt for T {} + +/// Stateless XRPC call builder. +/// +/// Example (per-request overrides) +/// ```ignore +/// use jacquard::client::{XrpcExt, AuthorizationToken}; +/// use jacquard::api::app_bsky::feed::get_author_feed::GetAuthorFeed; +/// use jacquard::types::ident::AtIdentifier; +/// use jacquard::CowStr; +/// use miette::IntoDiagnostic; +/// +/// #[tokio::main] +/// async fn main() -> miette::Result<()> { +/// let http = reqwest::Client::new(); +/// let base = url::Url::parse("https://public.api.bsky.app")?; +/// let resp = http +/// .xrpc(base) +/// .auth(AuthorizationToken::Bearer(CowStr::from("ACCESS_JWT"))) +/// .accept_labelers(vec![CowStr::from("did:plc:labelerid")]) +/// .header(http::header::USER_AGENT, http::HeaderValue::from_static("jacquard-example")) +/// .send( +/// GetAuthorFeed::new() +/// .actor(AtIdentifier::new_static("pattern.atproto.systems").unwrap()) +/// .limit(5) +/// .build(), +/// ) +/// .await?; +/// let out = resp.into_output()?; +/// println!("{}", serde_json::to_string_pretty(&out).into_diagnostic()?); +/// Ok(()) +/// } +/// ``` +pub struct XrpcCall<'a, C: HttpClient> { + pub(crate) client: &'a C, + pub(crate) base: Url, + pub(crate) opts: CallOptions<'a>, +} + +impl<'a, C: HttpClient> XrpcCall<'a, C> { + /// Apply Authorization to this call. + pub fn auth(mut self, token: AuthorizationToken<'a>) -> Self { + self.opts.auth = Some(token); + self + } + /// Set `atproto-proxy` header for this call. + pub fn proxy(mut self, proxy: CowStr<'a>) -> Self { + self.opts.atproto_proxy = Some(proxy); + self + } + /// Set `atproto-accept-labelers` header(s) for this call. + pub fn accept_labelers(mut self, labelers: Vec>) -> Self { + self.opts.atproto_accept_labelers = Some(labelers); + self + } + /// Add an extra header. + pub fn header(mut self, name: HeaderName, value: HeaderValue) -> Self { + self.opts.extra_headers.push((name, value)); + self + } + /// Replace the builder's options entirely. + pub fn with_options(mut self, opts: CallOptions<'a>) -> Self { + self.opts = opts; + self + } + + /// Send the given typed XRPC request and return a response wrapper. + pub async fn send(self, request: R) -> super_mod::Result> { + let http_request = super_mod::build_http_request(&self.base, &request, &self.opts) + .map_err(error::TransportError::from)?; + + let http_response = self + .client + .send_http(http_request) + .await + .map_err(|e| error::TransportError::Other(Box::new(e)))?; + + let status = http_response.status(); + let buffer = Bytes::from(http_response.into_body()); + + if !status.is_success() && !matches!(status.as_u16(), 400 | 401) { + return Err(error::HttpError { + status, + body: Some(buffer), + } + .into()); + } + + Ok(Response::new(buffer, status)) + } +} diff --git a/crates/jacquard/src/identity/resolver.rs b/crates/jacquard/src/identity/resolver.rs index d6442c2d..2c3b98c3 100644 --- a/crates/jacquard/src/identity/resolver.rs +++ b/crates/jacquard/src/identity/resolver.rs @@ -1,16 +1,16 @@ //! Identity resolution: handle → DID and DID → document, with smart fallbacks. //! //! Fallback order (default): -//! - Handle → DID: DNS TXT (if `dns` feature) → HTTPS well-known → embedded XRPC -//! `resolveHandle` → public API fallback → Slingshot `resolveHandle` (if configured). -//! - DID → Doc: did:web well-known → PLC/slingshot HTTP → embedded XRPC `resolveDid`, +//! - Handle → DID: DNS TXT (if `dns` feature) → HTTPS well-known → PDS XRPC +//! `resolveHandle` (when `pds_fallback` is configured) → public API fallback → Slingshot `resolveHandle` (if configured). +//! - DID → Doc: did:web well-known → PLC/Slingshot HTTP → PDS XRPC `resolveDid` (when configured), //! then Slingshot mini‑doc (partial) if configured. //! //! Parsing returns a `DidDocResponse` so callers can borrow from the response buffer //! and optionally validate the document `id` against the requested DID. -use crate::CowStr; -use crate::client::AuthenticatedClient; +// use crate::CowStr; // not currently needed directly here +use crate::client::XrpcExt; use bon::Builder; use bytes::Bytes; use jacquard_common::IntoStatic; @@ -183,14 +183,14 @@ pub enum DidStep { /// Configurable resolver options. /// /// - `plc_source`: where to fetch did:plc documents (PLC Directory or Slingshot). -/// - `pds_fallback`: optional base URL of a PDS for XRPC fallbacks (auth-aware -/// paths available via helpers that take an `XrpcClient`). +/// - `pds_fallback`: optional base URL of a PDS for XRPC fallbacks (stateless +/// XRPC over reqwest; authentication can be layered as needed). /// - `handle_order`/`did_order`: ordered strategies for resolution. /// - `validate_doc_id`: if true (default), convenience helpers validate doc `id` against the requested DID, /// returning `DocIdMismatch` with the fetched document on mismatch. /// - `public_fallback_for_handle`: if true (default), attempt /// `https://public.api.bsky.app/xrpc/com.atproto.identity.resolveHandle` as an unauth fallback. -/// There is no public fallback for DID documents; when `PdsResolveDid` is chosen and the embedded XRPC +/// There is no public fallback for DID documents; when `PdsResolveDid` is chosen and the PDS XRPC /// client fails, the resolver falls back to Slingshot mini-doc (partial) if `PlcSource::Slingshot` is configured. #[derive(Debug, Clone, Builder)] #[builder(start_fn = new)] @@ -238,7 +238,7 @@ impl Default for ResolverOptions { /// - HTTPS well-known for handles and `did:web` /// - PLC directory or Slingshot for `did:plc` /// - Slingshot `resolveHandle` (unauthenticated) when configured as the PLC source -/// - Auth-aware PDS fallbacks via helpers that accept an `XrpcClient` +/// - PDS fallbacks via helpers that use stateless XRPC on top of reqwest #[async_trait::async_trait] pub trait IdentityResolver { /// Access options for validation decisions in default methods @@ -284,45 +284,18 @@ pub trait IdentityResolver { } /// Default resolver implementation with configurable fallback order. -/// -/// Behavior highlights: -/// - Handle resolution tries DNS TXT (if enabled via `dns` feature), then HTTPS -/// well-known, then Slingshot's unauthenticated `resolveHandle` when -/// `PlcSource::Slingshot` is configured. -/// - DID resolution tries did:web well-known for `did:web`, and the configured -/// PLC base (PLC directory or Slingshot) for `did:plc`. -/// - PDS-authenticated fallbacks (e.g., `resolveHandle`, `resolveDid` on a PDS) -/// are available via helper methods that accept a user-provided `XrpcClient`. -/// -/// Example -/// ```ignore -/// # use jacquard::identity::resolver::{DefaultResolver, ResolverOptions}; -/// # use jacquard::client::{AuthenticatedClient, XrpcClient}; -/// # use jacquard::types::string::Handle; -/// # use jacquard::CowStr; -/// -/// // Build an auth-capable XRPC client (without a session it behaves like public/unauth) -/// let http = reqwest::Client::new(); -/// let xrpc = AuthenticatedClient::new(http.clone(), CowStr::new_static("https://bsky.social")); -/// let resolver = DefaultResolver::new(http, xrpc, ResolverOptions::default()); -/// -/// // Resolve a handle to a DID -/// let did = tokio_test::block_on(async { resolver.resolve_handle(&Handle::new("bad-example.com").unwrap()).await }).unwrap(); -/// ``` -pub struct DefaultResolver { +pub struct DefaultResolver { http: reqwest::Client, - xrpc: C, opts: ResolverOptions, #[cfg(feature = "dns")] dns: Option, } -impl DefaultResolver { +impl DefaultResolver { /// Create a new instance of the default resolver with all options (except DNS) up front - pub fn new(http: reqwest::Client, xrpc: C, opts: ResolverOptions) -> Self { + pub fn new(http: reqwest::Client, opts: ResolverOptions) -> Self { Self { http, - xrpc, opts, #[cfg(feature = "dns")] dns: None, @@ -439,15 +412,20 @@ impl DefaultResolver { } } -impl DefaultResolver { - /// Resolve handle to DID via a PDS XRPC client (auth-aware path) +impl DefaultResolver { + /// Resolve handle to DID via a PDS XRPC call (stateless, unauth by default) pub async fn resolve_handle_via_pds( &self, handle: &Handle<'_>, ) -> Result, IdentityError> { + let pds = match &self.opts.pds_fallback { + Some(u) => u.clone(), + None => return Err(IdentityError::InvalidWellKnown), + }; let req = ResolveHandle::new().handle((*handle).clone()).build(); let resp = self - .xrpc + .http + .xrpc(pds) .send(req) .await .map_err(|e| IdentityError::Xrpc(e.to_string()))?; @@ -464,9 +442,14 @@ impl DefaultResolver { &self, did: &Did<'_>, ) -> Result, IdentityError> { + let pds = match &self.opts.pds_fallback { + Some(u) => u.clone(), + None => return Err(IdentityError::InvalidWellKnown), + }; let req = resolve_did::ResolveDid::new().did(did.clone()).build(); let resp = self - .xrpc + .http + .xrpc(pds) .send(req) .await .map_err(|e| IdentityError::Xrpc(e.to_string()))?; @@ -510,7 +493,7 @@ impl DefaultResolver { } #[async_trait::async_trait] -impl IdentityResolver for DefaultResolver { +impl IdentityResolver for DefaultResolver { fn options(&self) -> &ResolverOptions { &self.opts } @@ -541,7 +524,7 @@ impl IdentityResolver for DefaultRes } } HandleStep::PdsResolveHandle => { - // Prefer embedded XRPC client + // Prefer PDS XRPC via stateless client if let Ok(did) = self.resolve_handle_via_pds(handle).await { return Ok(did); } @@ -630,7 +613,7 @@ impl IdentityResolver for DefaultRes } } DidStep::PdsResolveDid => { - // Try embedded XRPC client for full DID doc + // Try PDS XRPC for full DID doc if let Ok(doc) = self.fetch_did_doc_via_pds_owned(did).await { let buf = serde_json::to_vec(&doc).unwrap_or_default(); return Ok(DidDocResponse { @@ -667,7 +650,7 @@ pub enum IdentityWarning { }, } -impl DefaultResolver { +impl DefaultResolver { /// Resolve a handle to its DID, fetch the DID document, and return doc plus any warnings. /// This applies the default equality check on the document id (error with doc if mismatch). pub async fn resolve_handle_and_doc( @@ -772,11 +755,7 @@ mod tests { #[test] fn did_web_urls() { - let r = DefaultResolver::new( - reqwest::Client::new(), - TestXrpc::new(), - ResolverOptions::default(), - ); + let r = DefaultResolver::new(reqwest::Client::new(), ResolverOptions::default()); assert_eq!( r.test_did_web_url_raw("did:web:example.com"), "https://example.com/.well-known/did.json" @@ -819,11 +798,7 @@ mod tests { #[test] fn slingshot_mini_doc_url_build() { - let r = DefaultResolver::new( - reqwest::Client::new(), - TestXrpc::new(), - ResolverOptions::default(), - ); + let r = DefaultResolver::new(reqwest::Client::new(), ResolverOptions::default()); let base = Url::parse("https://slingshot.microcosm.blue").unwrap(); let url = r.slingshot_mini_doc_url(&base, "bad-example.com").unwrap(); assert_eq!( @@ -873,43 +848,15 @@ mod tests { other => panic!("unexpected: {:?}", other), } } - use crate::client::{HttpClient, XrpcClient}; - use http::Request; - use jacquard_common::CowStr; - - struct TestXrpc { - client: reqwest::Client, - } - impl TestXrpc { - fn new() -> Self { - Self { - client: reqwest::Client::new(), - } - } - } - impl HttpClient for TestXrpc { - type Error = reqwest::Error; - async fn send_http( - &self, - request: Request>, - ) -> Result>, Self::Error> { - self.client.send_http(request).await - } - } - impl XrpcClient for TestXrpc { - fn base_uri(&self) -> CowStr<'_> { - CowStr::from("https://public.api.bsky.app") - } - } } -/// Resolver specialized for unauthenticated/public flows using reqwest + AuthenticatedClient -pub type PublicResolver = DefaultResolver>; +/// Resolver specialized for unauthenticated/public flows using reqwest and stateless XRPC +pub type PublicResolver = DefaultResolver; impl Default for PublicResolver { /// Build a resolver with: /// - reqwest HTTP client - /// - XRPC base https://public.api.bsky.app (unauthenticated) + /// - Public fallbacks enabled for handle resolution /// - default options (DNS enabled if compiled, public fallback for handles enabled) /// /// Example @@ -919,10 +866,8 @@ impl Default for PublicResolver { /// ``` fn default() -> Self { let http = reqwest::Client::new(); - let xrpc = - AuthenticatedClient::new(http.clone(), CowStr::from("https://public.api.bsky.app")); let opts = ResolverOptions::default(); - let resolver = DefaultResolver::new(http, xrpc, opts); + let resolver = DefaultResolver::new(http, opts); #[cfg(feature = "dns")] let resolver = resolver.with_system_dns(); resolver @@ -933,10 +878,9 @@ impl Default for PublicResolver { /// mini-doc fallbacks, unauthenticated by default. pub fn slingshot_resolver_default() -> PublicResolver { let http = reqwest::Client::new(); - let xrpc = AuthenticatedClient::new(http.clone(), CowStr::from("https://public.api.bsky.app")); let mut opts = ResolverOptions::default(); opts.plc_source = PlcSource::slingshot_default(); - let resolver = DefaultResolver::new(http, xrpc, opts); + let resolver = DefaultResolver::new(http, opts); #[cfg(feature = "dns")] let resolver = resolver.with_system_dns(); resolver diff --git a/crates/jacquard/src/lib.rs b/crates/jacquard/src/lib.rs index 68fb71dd..e1747ee6 100644 --- a/crates/jacquard/src/lib.rs +++ b/crates/jacquard/src/lib.rs @@ -24,7 +24,7 @@ //! # use jacquard::CowStr; //! use jacquard::api::app_bsky::feed::get_timeline::GetTimeline; //! use jacquard::api::com_atproto::server::create_session::CreateSession; -//! use jacquard::client::{AuthenticatedClient, Session, XrpcClient}; +//! use jacquard::client::{BasicClient, Session}; //! # use miette::IntoDiagnostic; //! //! # #[derive(Parser, Debug)] @@ -48,7 +48,8 @@ //! let args = Args::parse(); //! //! // Create HTTP client -//! let mut client = AuthenticatedClient::new(reqwest::Client::new(), args.pds); +//! let url = url::Url::parse(&args.pds).unwrap(); +//! let client = BasicClient::new(url); //! //! // Create session //! let session = Session::from( @@ -64,7 +65,7 @@ //! ); //! //! println!("logged in as {} ({})", session.handle, session.did); -//! client.set_session(session); +//! client.set_session(session).await.unwrap(); //! //! // Fetch timeline //! println!("\nfetching timeline..."); @@ -86,6 +87,84 @@ //! } //! ``` //! +//! ## Clients +//! +//! - Stateless XRPC: any `HttpClient` (e.g., `reqwest::Client`) implements `XrpcExt`, +//! which provides `xrpc(base: Url) -> XrpcCall` for per-request calls with +//! optional `CallOptions` (auth, proxy, labelers, headers). Useful when you +//! want to pass auth on each call or build advanced flows. +//! Example +//! ```ignore +//! use jacquard::client::XrpcExt; +//! use jacquard::api::app_bsky::feed::get_author_feed::GetAuthorFeed; +//! use jacquard::types::ident::AtIdentifier; +//! +//! #[tokio::main] +//! async fn main() -> anyhow::Result<()> { +//! let http = reqwest::Client::new(); +//! let base = url::Url::parse("https://public.api.bsky.app")?; +//! let resp = http +//! .xrpc(base) +//! .send( +//! GetAuthorFeed::new() +//! .actor(AtIdentifier::new_static("pattern.atproto.systems").unwrap()) +//! .limit(5) +//! .build(), +//! ) +//! .await?; +//! let out = resp.into_output()?; +//! println!("{}", serde_json::to_string_pretty(&out).into_diagnostic()?); +//! Ok(()) +//! } +//! ``` +//! - Stateful client: `AtClient` holds a base `Url`, a transport, and a +//! `TokenStore` implementation. It automatically sets Authorization and can +//! auto-refresh a session when expired, retrying once. +//! - Convenience wrapper: `BasicClient` is an ergonomic newtype over +//! `AtClient` with a `new(Url)` constructor. +//! +//! Per-request overrides (stateless) +//! ```ignore +//! use jacquard::client::{XrpcExt, AuthorizationToken}; +//! use jacquard::api::app_bsky::feed::get_author_feed::GetAuthorFeed; +//! use jacquard::types::ident::AtIdentifier; +//! use jacquard::CowStr; +//! use miette::IntoDiagnostic; +//! +//! #[tokio::main] +//! async fn main() -> miette::Result<()> { +//! let http = reqwest::Client::new(); +//! let base = url::Url::parse("https://public.api.bsky.app")?; +//! let resp = http +//! .xrpc(base) +//! .auth(AuthorizationToken::Bearer(CowStr::from("ACCESS_JWT"))) +//! .accept_labelers(vec![CowStr::from("did:plc:labelerid")]) +//! .header(http::header::USER_AGENT, http::HeaderValue::from_static("jacquard-example")) +//! .send( +//! GetAuthorFeed::new() +//! .actor(AtIdentifier::new_static("pattern.atproto.systems").unwrap()) +//! .limit(5) +//! .build(), +//! ) +//! .await?; +//! let out = resp.into_output()?; +//! println!("{}", serde_json::to_string_pretty(&out).into_diagnostic()?); +//! Ok(()) +//! } +//! ``` +//! +//! Token storage: +//! - Use `MemoryTokenStore` for ephemeral sessions, tests, and CLIs. +//! - For persistence, `FileTokenStore` stores session tokens as JSON on disk. +//! See `client::token::FileTokenStore` docs for details. +//! Example +//! ```ignore +//! use jacquard::client::{AtClient, FileTokenStore}; +//! let base = url::Url::parse("https://bsky.social").unwrap(); +//! let store = FileTokenStore::new("/tmp/jacquard-session.json"); +//! let client = AtClient::new(reqwest::Client::new(), base, store); +//! ``` +//! #![warn(missing_docs)] diff --git a/crates/jacquard/src/main.rs b/crates/jacquard/src/main.rs index 5e17b11b..2481854a 100644 --- a/crates/jacquard/src/main.rs +++ b/crates/jacquard/src/main.rs @@ -2,7 +2,9 @@ use clap::Parser; use jacquard::CowStr; use jacquard::api::app_bsky::feed::get_timeline::GetTimeline; use jacquard::api::com_atproto::server::create_session::CreateSession; -use jacquard::client::{AuthenticatedClient, Session, XrpcClient}; +use jacquard::client::{BasicClient, Session}; +use jacquard::identity::resolver::{slingshot_resolver_default, IdentityResolver}; +use jacquard::types::string::Handle; use miette::IntoDiagnostic; #[derive(Parser, Debug)] @@ -24,8 +26,11 @@ struct Args { async fn main() -> miette::Result<()> { let args = Args::parse(); - // Create HTTP client - let mut client = AuthenticatedClient::new(reqwest::Client::new(), args.pds); + // Resolve PDS for the handle using the Slingshot-enabled resolver + let resolver = slingshot_resolver_default(); + let handle = Handle::new(args.username.as_ref()).into_diagnostic()?; + let (_did, pds_url) = resolver.pds_for_handle(&handle).await.into_diagnostic()?; + let client = BasicClient::new(pds_url); // Create session let session = Session::from( @@ -41,7 +46,7 @@ async fn main() -> miette::Result<()> { ); println!("logged in as {} ({})", session.handle, session.did); - client.set_session(session); + client.set_session(session).await.into_diagnostic()?; // Fetch timeline println!("\nfetching timeline...");