diff --git a/Cargo.lock b/Cargo.lock index cea4cde..cbbd942 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5965,6 +5965,7 @@ dependencies = [ "form_urlencoded", "idna", "percent-encoding", + "serde", ] [[package]] diff --git a/slingshot/Cargo.toml b/slingshot/Cargo.toml index 1f0dee9..e563b19 100644 --- a/slingshot/Cargo.toml +++ b/slingshot/Cargo.toml @@ -28,4 +28,4 @@ time = { version = "0.3.41", features = ["serde"] } tokio = { version = "1.47.0", features = ["full"] } tokio-util = "0.7.15" tracing-subscriber = { version = "0.3.19", features = ["env-filter"] } -url = "2.5.4" +url = { version = "2.5.4", features = ["serde"] } diff --git a/slingshot/src/error.rs b/slingshot/src/error.rs index 67b3e07..1bdab41 100644 --- a/slingshot/src/error.rs +++ b/slingshot/src/error.rs @@ -91,3 +91,13 @@ pub enum RecordError { #[error("upstream non-atproto bad request")] UpstreamBadBadNotGoodRequest(reqwest::Error), } + +#[derive(Debug, Error)] +pub enum ProxyError { + #[error("failed to parse path: {0}")] + PathParseError(String), + #[error(transparent)] + UrlParseError(#[from] url::ParseError), + #[error(transparent)] + ReqwestError(#[from] reqwest::Error), +} diff --git a/slingshot/src/lib.rs b/slingshot/src/lib.rs index e00e566..71e3c8b 100644 --- a/slingshot/src/lib.rs +++ b/slingshot/src/lib.rs @@ -3,6 +3,7 @@ pub mod error; mod firehose_cache; mod healthcheck; mod identity; +mod proxy; mod record; mod server; @@ -10,5 +11,6 @@ pub use consumer::consume; pub use firehose_cache::firehose_cache; pub use healthcheck::healthcheck; pub use identity::{Identity, IdentityKey}; +pub use proxy::Proxy; pub use record::{CachedRecord, ErrorResponseObject, Repo}; pub use server::serve; diff --git a/slingshot/src/main.rs b/slingshot/src/main.rs index 56d7500..ff47953 100644 --- a/slingshot/src/main.rs +++ b/slingshot/src/main.rs @@ -2,7 +2,7 @@ // use foyer::{Engine, DirectFsDeviceOptions, HybridCacheBuilder}; use metrics_exporter_prometheus::PrometheusBuilder; use slingshot::{ - Identity, Repo, consume, error::MainTaskError, firehose_cache, healthcheck, serve, + Identity, Proxy, Repo, consume, error::MainTaskError, firehose_cache, healthcheck, serve, }; use std::net::SocketAddr; use std::path::PathBuf; @@ -143,16 +143,16 @@ async fn main() -> Result<(), String> { ) .await .map_err(|e| format!("identity setup failed: {e:?}"))?; - - log::info!("identity service ready."); let identity_refresher = identity.clone(); let identity_shutdown = shutdown.clone(); tasks.spawn(async move { identity_refresher.run_refresher(identity_shutdown).await?; Ok(()) }); + log::info!("identity service ready."); let repo = Repo::new(identity.clone()); + let proxy = Proxy::new(repo.clone()); let identity_for_server = identity.clone(); let server_shutdown = shutdown.clone(); @@ -163,6 +163,7 @@ async fn main() -> Result<(), String> { server_cache_handle, identity_for_server, repo, + proxy, args.acme_domain, args.acme_contact, args.acme_cache_path, diff --git a/slingshot/src/proxy.rs b/slingshot/src/proxy.rs new file mode 100644 index 0000000..13edeca --- /dev/null +++ b/slingshot/src/proxy.rs @@ -0,0 +1,487 @@ +use serde::Deserialize; +use url::Url; +use std::{collections::HashMap, time::Duration}; +use crate::{Repo, server::HydrationSource, error::ProxyError}; +use reqwest::Client; +use serde_json::{Map, Value}; + +pub enum ParamValue { + String(Vec), + Int(Vec), + Bool(Vec), +} +pub struct Params(HashMap); + +impl TryFrom> for Params { + type Error = (); // TODO + fn try_from(val: Map) -> Result { + let mut out = HashMap::new(); + for (k, v) in val { + match v { + Value::String(s) => out.insert(k, ParamValue::String(vec![s])), + Value::Bool(b) => out.insert(k, ParamValue::Bool(vec![b])), + Value::Number(n) => { + let Some(i) = n.as_i64() else { + return Err(()); + }; + out.insert(k, ParamValue::Int(vec![i])) + } + Value::Array(a) => { + let Some(first) = a.first() else { + continue; + }; + if first.is_string() { + let mut vals = Vec::with_capacity(a.len()); + for v in a { + let Some(v) = v.as_str() else { + return Err(()); + }; + vals.push(v.to_string()); + } + out.insert(k, ParamValue::String(vals)); + } else if first.is_i64() { + let mut vals = Vec::with_capacity(a.len()); + for v in a { + let Some(v) = v.as_i64() else { + return Err(()); + }; + vals.push(v); + } + out.insert(k, ParamValue::Int(vals)); + } else if first.is_boolean() { + let mut vals = Vec::with_capacity(a.len()); + for v in a { + let Some(v) = v.as_bool() else { + return Err(()); + }; + vals.push(v); + } + out.insert(k, ParamValue::Bool(vals)); + } + todo!(); + } + _ => return Err(()), + }; + } + + Ok(Self(out)) + } +} + +#[derive(Clone)] +pub struct Proxy { + repo: Repo, + client: Client, +} + +impl Proxy { + pub fn new(repo: Repo) -> Self { + let client = Client::builder() + .user_agent(format!( + "microcosm slingshot v{} (contact: @bad-example.com)", + env!("CARGO_PKG_VERSION") + )) + .no_proxy() + .timeout(Duration::from_secs(6)) + .build() + .unwrap(); + Self { repo, client } + } + + pub async fn proxy( + &self, + xrpc: String, + service: String, + params: Option>, + ) -> Result { + + // hackin it to start + + // 1. assume did-web (TODO) and get the did doc + #[derive(Debug, Deserialize)] + struct ServiceDoc { + id: String, + service: Vec, + } + #[derive(Debug, Deserialize)] + struct ServiceItem { + id: String, + #[expect(unused)] + r#type: String, + #[serde(rename = "serviceEndpoint")] + service_endpoint: Url, + } + let dw = service.strip_prefix("did:web:").expect("a did web"); + let (dw, service_id) = dw.split_once("#").expect("whatever"); + let mut dw_url = Url::parse(&format!("https://{dw}"))?; + dw_url.set_path("/.well-known/did.json"); + let doc: ServiceDoc = self.client + .get(dw_url) + .send() + .await? + .error_for_status()? + .json() + .await?; + + assert_eq!(doc.id, format!("did:web:{}", dw)); + + let mut upstream = None; + for ServiceItem { id, service_endpoint, .. } in doc.service { + let Some((_, id)) = id.split_once("#") else { continue; }; + if id != service_id { continue; }; + upstream = Some(service_endpoint); + break; + } + + // 2. proxy the request forward + let mut upstream = upstream.expect("to find it"); + upstream.set_path(&format!("/xrpc/{xrpc}")); // TODO: validate nsid + + if let Some(params) = params { + let mut query = upstream.query_pairs_mut(); + let Params(ps) = params.try_into().expect("valid params"); + for (k, pvs) in ps { + match pvs { + ParamValue::String(s) => { + for s in s { + query.append_pair(&k, &s); + } + } + ParamValue::Int(i) => { + for i in i { + query.append_pair(&k, &i.to_string()); + } + } + ParamValue::Bool(b) => { + for b in b { + query.append_pair(&k, &b.to_string()); + } + } + } + } + } + + // TODO: other headers to proxy + Ok(self.client + .get(upstream) + .send() + .await? + .error_for_status()? + .json() + .await?) + } +} + +#[derive(Debug, PartialEq)] +pub enum PathPart { + Scalar(String), + Vector(String, Option), // key, $type +} + +pub fn parse_record_path(input: &str) -> Result, String> { + let mut out = Vec::new(); + + let mut key_acc = String::new(); + let mut type_acc = String::new(); + let mut in_bracket = false; + let mut chars = input.chars().enumerate(); + while let Some((i, c)) = chars.next() { + match c { + '[' if in_bracket => return Err(format!("nested opening bracket not allowed, at {i}")), + '[' if key_acc.is_empty() => return Err(format!("missing key before opening bracket, at {i}")), + '[' => in_bracket = true, + ']' if in_bracket => { + in_bracket = false; + let key = std::mem::take(&mut key_acc); + let r#type = std::mem::take(&mut type_acc); + let t = if r#type.is_empty() { None } else { Some(r#type) }; + out.push(PathPart::Vector(key, t)); + // peek ahead because we need a dot after array if there's more and i don't want to add more loop state + let Some((i, c)) = chars.next() else { + break; + }; + if c != '.' { + return Err(format!("expected dot after close bracket, found {c:?} at {i}")); + } + } + ']' => return Err(format!("unexpected close bracket at {i}")), + '.' if in_bracket => type_acc.push(c), + '.' if key_acc.is_empty() => return Err(format!("missing key before next segment, at {i}")), + '.' => { + let key = std::mem::take(&mut key_acc); + assert!(type_acc.is_empty()); + out.push(PathPart::Scalar(key)); + } + _ if in_bracket => type_acc.push(c), + _ => key_acc.push(c), + } + } + if in_bracket { + return Err("unclosed bracket".into()); + } + if !key_acc.is_empty() { + out.push(PathPart::Scalar(key_acc)); + } + Ok(out) +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum RefShape { + StrongRef, + AtUri, + AtUriParts, + Did, + Handle, + AtIdentifier, +} + +impl TryFrom<&str> for RefShape { + type Error = String; + fn try_from(s: &str) -> Result { + match s { + "strong-ref" => Ok(Self::StrongRef), + "at-uri" => Ok(Self::AtUri), + "at-uri-parts" => Ok(Self::AtUriParts), + "did" => Ok(Self::Did), + "handle" => Ok(Self::Handle), + "at-identifier" => Ok(Self::AtIdentifier), + _ => Err(format!("unknown shape: {s}")), + } + } +} + +#[derive(Debug, PartialEq)] +pub enum MatchedRef { + AtUri { + uri: String, + cid: Option, + }, + Identifier(String), +} + +pub fn match_shape(shape: RefShape, val: &Value) -> Option { + // TODO: actually validate at-uri format + // TODO: actually validate everything else also + // TODO: should this function normalize identifiers to DIDs probably? + // or just return at-uri parts so the caller can resolve and reassemble + match shape { + RefShape::StrongRef => { + let o = val.as_object()?; + let uri = o.get("uri")?.as_str()?.to_string(); + let cid = o.get("cid")?.as_str()?.to_string(); + Some(MatchedRef::AtUri { uri, cid: Some(cid) }) + } + RefShape::AtUri => { + let uri = val.as_str()?.to_string(); + Some(MatchedRef::AtUri { uri, cid: None }) + } + RefShape::AtUriParts => { + let o = val.as_object()?; + let identifier = o.get("repo").or(o.get("did"))?.as_str()?.to_string(); + let collection = o.get("collection")?.as_str()?.to_string(); + let rkey = o.get("rkey")?.as_str()?.to_string(); + let uri = format!("at://{identifier}/{collection}/{rkey}"); + let cid = o.get("cid").and_then(|v| v.as_str()).map(str::to_string); + Some(MatchedRef::AtUri { uri, cid }) + } + RefShape::Did => { + let id = val.as_str()?; + if !id.starts_with("did:") { + return None; + } + Some(MatchedRef::Identifier(id.to_string())) + } + RefShape::Handle => { + let id = val.as_str()?; + if id.contains(':') { + return None; + } + Some(MatchedRef::Identifier(id.to_string())) + } + RefShape::AtIdentifier => { + Some(MatchedRef::Identifier(val.as_str()?.to_string())) + } + } +} + +// TODO: send back metadata about the matching +pub fn extract_links( + sources: Vec, + skeleton: &Value, +) -> Result, String> { + // collect early to catch errors from the client + // (TODO maybe the handler should do this and pass in the processed stuff probably definitely yeah) + let sources = sources + .into_iter() + .map(|HydrationSource { path, shape }| { + let path_parts = parse_record_path(&path)?; + let shape: RefShape = shape.as_str().try_into()?; + Ok((path_parts, shape)) + }) + .collect::, String>>()?; + + // lazy first impl, just re-walk the skeleton as many times as needed + // not deduplicating for now + let mut out = Vec::new(); + for (path_parts, shape) in sources { + for val in PathWalker::new(&path_parts, skeleton) { + if let Some(matched) = match_shape(shape, val) { + out.push(matched); + } + } + } + + Ok(out) +} + +struct PathWalker<'a> { + todo: Vec<(&'a [PathPart], &'a Value)>, +} +impl<'a> PathWalker<'a> { + fn new(path_parts: &'a [PathPart], skeleton: &'a Value) -> Self { + Self { todo: vec![(path_parts, skeleton)] } + } +} +impl<'a> Iterator for PathWalker<'a> { + type Item = &'a Value; + fn next(&mut self) -> Option { + loop { + let (parts, val) = self.todo.pop()?; + let Some((part, rest)) = parts.split_first() else { + return Some(val); + }; + let Some(o) = val.as_object() else { + continue; + }; + match part { + PathPart::Scalar(k) => { + let Some(v) = o.get(k) else { + continue; + }; + self.todo.push((rest, v)); + } + PathPart::Vector(k, t) => { + let Some(a) = o.get(k).and_then(|v| v.as_array()) else { + continue; + }; + for v in a + .iter() + .rev() + .filter(|c| { + let Some(t) = t else { return true }; + c + .as_object() + .and_then(|o| o.get("$type")) + .and_then(|v| v.as_str()) + .map(|s| s == t) + .unwrap_or(false) + }) + { + self.todo.push((rest, v)) + } + } + } + } + } +} + + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn test_parse_record_path() -> Result<(), Box> { + let cases = [ + ("", vec![]), + ("subject", vec![PathPart::Scalar("subject".into())]), + ("authorDid", vec![PathPart::Scalar("authorDid".into())]), + ("subject.uri", vec![PathPart::Scalar("subject".into()), PathPart::Scalar("uri".into())]), + ("members[]", vec![PathPart::Vector("members".into(), None)]), + ("add[].key", vec![ + PathPart::Vector("add".into(), None), + PathPart::Scalar("key".into()), + ]), + ("a[b]", vec![PathPart::Vector("a".into(), Some("b".into()))]), + ("a[b.c]", vec![PathPart::Vector("a".into(), Some("b.c".into()))]), + ("facets[app.bsky.richtext.facet].features[app.bsky.richtext.facet#mention].did", vec![ + PathPart::Vector("facets".into(), Some("app.bsky.richtext.facet".into())), + PathPart::Vector("features".into(), Some("app.bsky.richtext.facet#mention".into())), + PathPart::Scalar("did".into()), + ]), + ]; + + for (path, expected) in cases { + let parsed = parse_record_path(path)?; + assert_eq!(parsed, expected, "path: {path:?}"); + } + + Ok(()) + } + + #[test] + fn test_match_shape() { + let cases = [ + ("strong-ref", json!(""), None), + ("strong-ref", json!({}), None), + ("strong-ref", json!({ "uri": "abc" }), None), + ("strong-ref", json!({ "cid": "def" }), None), + ( + "strong-ref", + json!({ "uri": "abc", "cid": "def" }), + Some(MatchedRef::AtUri { uri: "abc".to_string(), cid: Some("def".to_string()) }), + ), + ("at-uri", json!({ "uri": "abc" }), None), + ("at-uri", json!({ "uri": "abc", "cid": "def" }), None), + ( + "at-uri", + json!("abc"), + Some(MatchedRef::AtUri { uri: "abc".to_string(), cid: None }), + ), + ("at-uri-parts", json!("abc"), None), + ("at-uri-parts", json!({}), None), + ( + "at-uri-parts", + json!({"repo": "a", "collection": "b", "rkey": "c"}), + Some(MatchedRef::AtUri { uri: "at://a/b/c".to_string(), cid: None }), + ), + ( + "at-uri-parts", + json!({"did": "a", "collection": "b", "rkey": "c"}), + Some(MatchedRef::AtUri { uri: "at://a/b/c".to_string(), cid: None }), + ), + ( + "at-uri-parts", + // 'repo' takes precedence over 'did' + json!({"did": "a", "repo": "z", "collection": "b", "rkey": "c"}), + Some(MatchedRef::AtUri { uri: "at://z/b/c".to_string(), cid: None }), + ), + ( + "at-uri-parts", + json!({"repo": "a", "collection": "b", "rkey": "c", "cid": "def"}), + Some(MatchedRef::AtUri { uri: "at://a/b/c".to_string(), cid: Some("def".to_string()) }), + ), + ( + "at-uri-parts", + json!({"repo": "a", "collection": "b", "rkey": "c", "cid": {}}), + Some(MatchedRef::AtUri { uri: "at://a/b/c".to_string(), cid: None }), + ), + ("did", json!({}), None), + ("did", json!(""), None), + ("did", json!("bad-example.com"), None), + ("did", json!("did:plc:xyz"), Some(MatchedRef::Identifier("did:plc:xyz".to_string()))), + ("handle", json!({}), None), + ("handle", json!("bad-example.com"), Some(MatchedRef::Identifier("bad-example.com".to_string()))), + ("handle", json!("did:plc:xyz"), None), + ("at-identifier", json!({}), None), + ("at-identifier", json!("bad-example.com"), Some(MatchedRef::Identifier("bad-example.com".to_string()))), + ("at-identifier", json!("did:plc:xyz"), Some(MatchedRef::Identifier("did:plc:xyz".to_string()))), + ]; + for (shape, val, expected) in cases { + let s = shape.try_into().unwrap(); + let matched = match_shape(s, &val); + assert_eq!(matched, expected, "shape: {shape:?}, val: {val:?}"); + } + } +} diff --git a/slingshot/src/record.rs b/slingshot/src/record.rs index e01eaa0..39b33e0 100644 --- a/slingshot/src/record.rs +++ b/slingshot/src/record.rs @@ -11,8 +11,8 @@ use url::Url; #[derive(Debug, Serialize, Deserialize)] pub struct RawRecord { - cid: Cid, - record: String, + pub cid: Cid, + pub record: String, } // TODO: should be able to do typed CID diff --git a/slingshot/src/server.rs b/slingshot/src/server.rs index 41f014f..0a84f46 100644 --- a/slingshot/src/server.rs +++ b/slingshot/src/server.rs @@ -1,15 +1,15 @@ use crate::{ - CachedRecord, ErrorResponseObject, Identity, Repo, + CachedRecord, ErrorResponseObject, Identity, Proxy, Repo, error::{RecordError, ServerError}, + proxy::{extract_links, MatchedRef}, + record::RawRecord, }; use atrium_api::types::string::{Cid, Did, Handle, Nsid, RecordKey}; use foyer::HybridCache; use links::at_uri::parse_at_uri as normalize_at_uri; use serde::Serialize; -use std::path::PathBuf; -use std::str::FromStr; -use std::sync::Arc; -use std::time::Instant; +use std::{path::PathBuf, str::FromStr, sync::Arc, time::Instant, collections::HashMap}; +use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use poem::{ @@ -24,6 +24,7 @@ use poem::{ }; use poem_openapi::{ ApiResponse, ContactObject, ExternalDocumentObject, Object, OpenApi, OpenApiService, Tags, + Union, param::Query, payload::Json, types::Example, }; @@ -92,6 +93,13 @@ fn bad_request_handler_resolve_mini(err: poem::Error) -> ResolveMiniIDResponse { })) } +fn bad_request_handler_proxy_query(err: poem::Error) -> ProxyHydrateResponse { + ProxyHydrateResponse::BadRequest(Json(XrpcErrorResponseObject { + error: "InvalidRequest".to_string(), + message: format!("Bad request, here's some info that maybe should not be exposed: {err}"), + })) +} + fn bad_request_handler_resolve_handle(err: poem::Error) -> JustDidResponse { JustDidResponse::BadRequest(Json(XrpcErrorResponseObject { error: "InvalidRequest".to_string(), @@ -190,6 +198,119 @@ enum ResolveMiniIDResponse { BadRequest(XrpcError), } +#[derive(Object)] +struct ProxyHydrationError { + reason: String, +} + +#[derive(Object)] +struct ProxyHydrationPending { + url: String, +} + +#[derive(Object)] +struct ProxyHydrationRecordFound { + record: serde_json::Value, +} + +#[derive(Object)] +struct ProxyHydrationIdentifierFound { + record: MiniDocResponseObject, +} + +// todo: there's gotta be a supertrait that collects these? +use poem_openapi::types::{Type, ToJSON, ParseFromJSON, IsObjectType}; + +#[derive(Union)] +#[oai(discriminator_name = "status", rename_all = "camelCase")] +enum Hydration { + Error(ProxyHydrationError), + Pending(ProxyHydrationPending), + Found(T), +} + +#[derive(Object)] +#[oai(example = true)] +struct ProxyHydrateResponseObject { + /// The original upstream response content + output: serde_json::Value, + /// Any hydrated records + records: HashMap>, + /// Any hydrated identifiers + identifiers: HashMap>, +} +impl Example for ProxyHydrateResponseObject { + fn example() -> Self { + Self { + output: serde_json::json!({}), + records: HashMap::from([ + ("asdf".into(), Hydration::Pending(ProxyHydrationPending { url: "todo".into() })), + ]), + identifiers: HashMap::new(), + } + } +} + +#[derive(ApiResponse)] +#[oai(bad_request_handler = "bad_request_handler_proxy_query")] +enum ProxyHydrateResponse { + #[oai(status = 200)] + Ok(Json), + #[oai(status = 400)] + BadRequest(XrpcError) +} + +#[derive(Object)] +pub struct HydrationSource { + /// Record Path syntax for locating fields + pub path: String, + /// What to expect at the path: 'strong-ref', 'at-uri', 'at-uri-parts', 'did', 'handle', or 'at-identifier'. + /// + /// - `strong-ref`: object in the shape of `com.atproto.repo.strongRef` with `uri` and `cid` keys. + /// - `at-uri`: string, must have all segments present (identifier, collection, rkey) + /// - `at-uri-parts`: object with keys (`repo` or `did`), `collection`, `rkey`, and optional `cid`. Other keys may be present and will be ignored. + /// - `did`: string, `did` format + /// - `handle`: string, `handle` format + /// - `at-identifier`: string, `did` or `handle` format + pub shape: String, +} + +#[derive(Object)] +#[oai(example = true)] +struct ProxyQueryPayload { + /// The NSID of the XRPC you wish to forward + xrpc: String, + /// The destination service the request will be forwarded to + atproto_proxy: String, + /// The `params` for the destination service XRPC endpoint + /// + /// Currently this will be passed along unchecked, but a future version of + /// slingshot may attempt to do lexicon resolution to validate `params` + /// based on the upstream service + params: Option, + /// Paths within the response to look for at-uris that can be hydrated + hydration_sources: Vec, + // todo: deadline thing + +} +impl Example for ProxyQueryPayload { + fn example() -> Self { + Self { + xrpc: "app.bsky.feed.getFeedSkeleton".to_string(), + atproto_proxy: "did:web:blue.mackuba.eu#bsky_fg".to_string(), + params: Some(serde_json::json!({ + "feed": "at://did:plc:oio4hkxaop4ao4wz2pp3f4cr/app.bsky.feed.generator/atproto", + })), + hydration_sources: vec![ + HydrationSource { + path: "feed[].post".to_string(), + shape: "at-uri".to_string(), + } + ], + } + } +} + #[derive(Object)] #[oai(example = true)] struct FoundDidResponseObject { @@ -221,6 +342,7 @@ enum JustDidResponse { struct Xrpc { cache: HybridCache, identity: Identity, + proxy: Arc, repo: Arc, } @@ -550,6 +672,152 @@ impl Xrpc { })) } + /// com.bad-example.proxy.hydrateQueryResponse + /// + /// > [!important] + /// > Unstable! This endpoint is experimental and may change. + /// + /// Fetch + include records referenced from an upstream xrpc query response + #[oai( + path = "/com.bad-example.proxy.hydrateQueryResponse", + method = "post", + tag = "ApiTags::Custom" + )] + async fn proxy_hydrate_query( + &self, + Json(payload): Json, + ) -> ProxyHydrateResponse { + // TODO: the Accept request header, if present, gotta be json + // TODO: find any Authorization header and verify it. TBD about `aud`. + + let params = if let Some(p) = payload.params { + let serde_json::Value::Object(map) = p else { + panic!("params have to be an object"); + }; + Some(map) + } else { None }; + + match self.proxy.proxy( + payload.xrpc, + payload.atproto_proxy, + params, + ).await { + Ok(skeleton) => { + let links = match extract_links(payload.hydration_sources, &skeleton) { + Ok(l) => l, + Err(e) => { + log::warn!("problem extracting: {e:?}"); + return ProxyHydrateResponse::BadRequest(xrpc_error("oop", "sorry, error extracting")) + } + }; + let mut records = HashMap::new(); + let mut identifiers = HashMap::new(); + + enum GetThing { + Record(String, Hydration), + Identifier(String, Hydration), + } + + let (tx, mut rx) = mpsc::channel(1); + + for link in links { + match link { + MatchedRef::AtUri { uri, cid } => { + if records.contains_key(&uri) { + log::warn!("skipping duplicate record without checking cid"); + continue; + } + let mut u = url::Url::parse("https://example.com").unwrap(); + u.query_pairs_mut().append_pair("at_uri", &uri); // BLEH todo + records.insert(uri.clone(), Hydration::Pending(ProxyHydrationPending { + url: format!("/xrpc/blue.microcosm.repo.getRecordByUri?{}", u.query().unwrap()), // TODO better; with cid, etc. + })); + let tx = tx.clone(); + let identity = self.identity.clone(); + let repo = self.repo.clone(); + tokio::task::spawn(async move { + let rest = uri.strip_prefix("at://").unwrap(); + let (identifier, rest) = rest.split_once('/').unwrap(); + let (collection, rkey) = rest.split_once('/').unwrap(); + + let did = if identifier.starts_with("did:") { + Did::new(identifier.to_string()).unwrap() + } else { + let handle = Handle::new(identifier.to_string()).unwrap(); + identity.handle_to_did(handle).await.unwrap().unwrap() + }; + + let res = match repo.get_record( + &did, + &Nsid::new(collection.to_string()).unwrap(), + &RecordKey::new(rkey.to_string()).unwrap(), + &cid.as_ref().map(|s| Cid::from_str(s).unwrap()), + ).await { + Ok(CachedRecord::Deleted) => + Hydration::Error(ProxyHydrationError { + reason: "record deleted".to_string(), + }), + Ok(CachedRecord::Found(RawRecord { cid: found_cid, record })) => { + if let Some(c) = cid && found_cid.as_ref().to_string() != c { + log::warn!("ignoring cid mismatch"); + } + let value = serde_json::from_str(&record).unwrap(); + Hydration::Found(ProxyHydrationRecordFound { + record: value, + }) + } + Err(e) => { + log::warn!("finally oop {e:?}"); + Hydration::Error(ProxyHydrationError { + reason: "failed to fetch record".to_string(), + }) + } + }; + tx.send(GetThing::Record(uri, res)).await + }); + } + MatchedRef::Identifier(id) => { + if identifiers.contains_key(&id) { + continue; + } + let mut u = url::Url::parse("https://example.com").unwrap(); + u.query_pairs_mut().append_pair("identifier", &id); + identifiers.insert(id, Hydration::Pending(ProxyHydrationPending { + url: format!("/xrpc/blue.microcosm.identity.resolveMiniDoc?{}", u.query().unwrap()), // gross + })); + let tx = tx.clone(); + // let doc_fut = self.resolve_mini_doc(); + tokio::task::spawn(async { + + }); + } + } + } + // so the channel can close when all are completed + // (we shoudl be doing a timeout...) + drop(tx); + + while let Some(hydration) = rx.recv().await { + match hydration { + GetThing::Record(uri, h) => { records.insert(uri, h); } + GetThing::Identifier(uri, md) => { identifiers.insert(uri, md); } + }; + } + + ProxyHydrateResponse::Ok(Json(ProxyHydrateResponseObject { + output: skeleton, + records, + identifiers, + })) + } + Err(e) => { + log::warn!("oh no: {e:?}"); + ProxyHydrateResponse::BadRequest(xrpc_error("oop", "sorry")) + } + } + + } + async fn get_record_impl( &self, repo: String, @@ -748,6 +1016,7 @@ pub async fn serve( cache: HybridCache, identity: Identity, repo: Repo, + proxy: Proxy, acme_domain: Option, acme_contact: Option, acme_cache_path: Option, @@ -756,10 +1025,12 @@ pub async fn serve( bind: std::net::SocketAddr, ) -> Result<(), ServerError> { let repo = Arc::new(repo); + let proxy = Arc::new(proxy); let api_service = OpenApiService::new( Xrpc { cache, identity, + proxy, repo, }, "Slingshot", @@ -823,7 +1094,7 @@ where .with( Cors::new() .allow_origin_regex("*") - .allow_methods([Method::GET]) + .allow_methods([Method::GET, Method::POST]) .allow_credentials(false), ) .with(CatchPanic::new())