From 7da961f4d6efa434b41e77b608d2dce2e76a1395 Mon Sep 17 00:00:00 2001 From: phil Date: Tue, 3 Jun 2025 15:03:35 -0400 Subject: [PATCH] deserialize multiple collections from query blehhhhhh --- ufos/src/lib.rs | 1 - ufos/src/qs_query.rs | 73 --------------------------- ufos/src/server/collections_query.rs | 72 ++++++++++++++++++++++++++ ufos/src/server/cors.rs | 23 +++++++++ ufos/src/{server.rs => server/mod.rs} | 72 ++++++++++---------------- 5 files changed, 121 insertions(+), 120 deletions(-) delete mode 100644 ufos/src/qs_query.rs create mode 100644 ufos/src/server/collections_query.rs create mode 100644 ufos/src/server/cors.rs rename ufos/src/{server.rs => server/mod.rs} (89%) diff --git a/ufos/src/lib.rs b/ufos/src/lib.rs index 8ad6162..aee41d1 100644 --- a/ufos/src/lib.rs +++ b/ufos/src/lib.rs @@ -3,7 +3,6 @@ pub mod db_types; pub mod error; pub mod file_consumer; pub mod index_html; -pub mod qs_query; pub mod server; pub mod storage; pub mod storage_fjall; diff --git a/ufos/src/qs_query.rs b/ufos/src/qs_query.rs deleted file mode 100644 index 7cbab8e..0000000 --- a/ufos/src/qs_query.rs +++ /dev/null @@ -1,73 +0,0 @@ -use async_trait::async_trait; -use dropshot::{ - ApiEndpointBodyContentType, ExclusiveExtractor, ExtractorMetadata, HttpError, RequestContext, - RequestInfo, ServerContext, SharedExtractor, -}; -/// copied from https://github.com/oxidecomputer/dropshot/blob/695e1d8872c988c43066eb0848c87c127eeda361/dropshot/src/extractor/query.rs -/// Apache 2.0: https://github.com/oxidecomputer/dropshot/blob/695e1d8872c988c43066eb0848c87c127eeda361/LICENSE -use schemars::JsonSchema; -use serde::de::DeserializeOwned; - -/// `VecsAllowedQuery` is an extractor used to deserialize an -/// instance of `QueryType` from an HTTP request's query string. `QueryType` -/// is any structure of yours that implements [serde::Deserialize] and -/// [schemars::JsonSchema]. See the crate documentation for more information. -#[derive(Debug)] -pub struct VecsAllowedQuery { - inner: QueryType, -} -impl VecsAllowedQuery { - // TODO drop this in favor of Deref? + Display and Debug for convenience? - pub fn into_inner(self) -> QueryType { - self.inner - } -} - -/// Given an HTTP request, pull out the query string and attempt to deserialize -/// it as an instance of `QueryType`. -fn http_request_load_query( - request: &RequestInfo, -) -> Result, HttpError> -where - QueryType: DeserializeOwned + JsonSchema + Send + Sync, -{ - let raw_query_string = request.uri().query().unwrap_or(""); - // TODO-correctness: are query strings defined to be urlencoded in this way? - match serde_qs::from_str(raw_query_string) { - Ok(q) => Ok(VecsAllowedQuery { inner: q }), - Err(e) => Err(HttpError::for_bad_request( - None, - format!("unable to parse query string: {}", e), - )), - } -} - -// The `SharedExtractor` implementation for Query describes how to -// construct an instance of `Query` from an HTTP request: namely, by -// parsing the query string to an instance of `QueryType`. -// TODO-cleanup We shouldn't have to use the "'static" bound on `QueryType` -// here. It seems like we ought to be able to use 'async_trait, but that -// doesn't seem to be defined. -#[async_trait] -impl SharedExtractor for VecsAllowedQuery -where - QueryType: JsonSchema + DeserializeOwned + Send + Sync + 'static, -{ - async fn from_request( - rqctx: &RequestContext, - ) -> Result, HttpError> { - http_request_load_query(&rqctx.request) - } - - fn metadata(body_content_type: ApiEndpointBodyContentType) -> ExtractorMetadata { - // HACK: would love to use Query here but it "helpfully" panics when it sees a Vec. - // we can't really get at enough of Query's logic to use it directly, sadly, so the - // resulting openapi docs suck (query params are listed as body payload, example - // codes make no sense, etc.) - // - // trying to hack the resulting ExtractorMetadata to look like Query's is a pain: - // things almost work out but then something in dropshot won't be `pub` and it falls - // apart. maybe it's possible, i didn't get it in the time i had. - dropshot::TypedBody::::metadata(body_content_type) - } -} diff --git a/ufos/src/server/collections_query.rs b/ufos/src/server/collections_query.rs new file mode 100644 index 0000000..d5daecb --- /dev/null +++ b/ufos/src/server/collections_query.rs @@ -0,0 +1,72 @@ +use crate::Nsid; +use async_trait::async_trait; +use dropshot::{ + ApiEndpointBodyContentType, ExtractorMetadata, HttpError, Query, RequestContext, ServerContext, + SharedExtractor, +}; +use schemars::JsonSchema; +use serde::Deserialize; +use std::collections::HashSet; + +/// The real type that gets deserialized +#[derive(Debug, Deserialize, JsonSchema)] +pub struct MultiCollectionQuery { + pub collection: Vec, +} + +/// The fake corresponding type for docs that dropshot won't freak out about a +/// vec for +#[derive(Deserialize, JsonSchema)] +#[allow(dead_code)] +struct MultiCollectionQueryForDocs { + /// One or more collection [NSID](https://atproto.com/specs/nsid)s + /// + /// Pass this parameter multiple times to specify multiple collections, like + /// `collection=app.bsky.feed.like&collection=app.bsky.feed.post` + collection: String, +} + +impl TryFrom for HashSet { + type Error = HttpError; + fn try_from(mcq: MultiCollectionQuery) -> Result { + let mut out = HashSet::with_capacity(mcq.collection.len()); + for c in mcq.collection { + let nsid = Nsid::new(c).map_err(|e| { + HttpError::for_bad_request( + None, + format!("failed to convert collection to an NSID: {e:?}"), + ) + })?; + out.insert(nsid); + } + Ok(out) + } +} + +// The `SharedExtractor` implementation for Query describes how to +// construct an instance of `Query` from an HTTP request: namely, by +// parsing the query string to an instance of `QueryType`. +#[async_trait] +impl SharedExtractor for MultiCollectionQuery { + async fn from_request( + ctx: &RequestContext, + ) -> Result { + let raw_query = ctx.request.uri().query().unwrap_or(""); + let q = serde_qs::from_str(raw_query).map_err(|e| { + HttpError::for_bad_request(None, format!("unable to parse query string: {}", e)) + })?; + Ok(q) + } + + fn metadata(body_content_type: ApiEndpointBodyContentType) -> ExtractorMetadata { + // HACK: query type switcheroo: passing MultiCollectionQuery to + // `metadata` would "helpfully" panic because dropshot believes we can + // only have scalar types in a query. + // + // so instead we have a fake second type whose only job is to look the + // same as MultiCollectionQuery exept that it has `String` instead of + // `Vec`, which dropshot will accept, and generate ~close-enough + // docs for. + as SharedExtractor>::metadata(body_content_type) + } +} diff --git a/ufos/src/server/cors.rs b/ufos/src/server/cors.rs new file mode 100644 index 0000000..61a8815 --- /dev/null +++ b/ufos/src/server/cors.rs @@ -0,0 +1,23 @@ +use dropshot::{HttpError, HttpResponseHeaders, HttpResponseOk}; +use schemars::JsonSchema; +use serde::Serialize; + +pub type OkCorsResponse = Result>, HttpError>; + +/// Helper for constructing Ok responses: return OkCors(T).into() +/// (not happy with this yet) +pub struct OkCors(pub T); + +impl From> for OkCorsResponse +where + T: Serialize + JsonSchema + Send + Sync, +{ + fn from(ok: OkCors) -> OkCorsResponse { + let mut res = HttpResponseHeaders::new_unnamed(HttpResponseOk(ok.0)); + res.headers_mut() + .insert("access-control-allow-origin", "*".parse().unwrap()); + Ok(res) + } +} + +// TODO: cors for HttpError diff --git a/ufos/src/server.rs b/ufos/src/server/mod.rs similarity index 89% rename from ufos/src/server.rs rename to ufos/src/server/mod.rs index 782219c..c515de0 100644 --- a/ufos/src/server.rs +++ b/ufos/src/server/mod.rs @@ -1,10 +1,14 @@ +mod collections_query; +mod cors; + use crate::index_html::INDEX_HTML; -use crate::qs_query::VecsAllowedQuery; use crate::storage::StoreReader; use crate::store_types::{HourTruncatedCursor, WeekTruncatedCursor}; use crate::{ConsumerInfo, Cursor, JustCount, Nsid, NsidCount, OrderCollectionsBy, UFOsRecord}; use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; use chrono::{DateTime, Utc}; +use collections_query::MultiCollectionQuery; +use cors::{OkCors, OkCorsResponse}; use dropshot::endpoint; use dropshot::ApiDescription; use dropshot::Body; @@ -12,8 +16,6 @@ use dropshot::ConfigDropshot; use dropshot::ConfigLogging; use dropshot::ConfigLoggingLevel; use dropshot::HttpError; -use dropshot::HttpResponseHeaders; -use dropshot::HttpResponseOk; use dropshot::Query; use dropshot::RequestContext; use dropshot::ServerBuilder; @@ -76,7 +78,7 @@ async fn index(_ctx: RequestContext) -> Result, HttpErro }] async fn get_openapi(ctx: RequestContext) -> OkCorsResponse { let spec = (*ctx.context().spec).clone(); - ok_cors(spec) + OkCors(spec).into() } #[derive(Debug, Serialize, JsonSchema)] @@ -105,11 +107,12 @@ async fn get_meta_info(ctx: RequestContext) -> OkCorsResponse .await .map_err(failed_to_get("consumer info"))?; - ok_cors(MetaInfo { + OkCors(MetaInfo { storage_name: storage.name(), storage: storage_info, consumer, }) + .into() } // TODO: replace with normal (🙃) multi-qs value somehow @@ -194,12 +197,11 @@ async fn get_records_by_collections( .map(|r| r.into()) .collect(); - ok_cors(records) + OkCors(records).into() } #[derive(Debug, Deserialize, JsonSchema)] -struct TotalSeenCollectionsQuery { - collection: Vec, // JsonSchema not implemented for Nsid :( +struct CollectionsStatsQuery { /// Limit stats to those seen after this UTC datetime /// /// default: 1 week ago @@ -216,38 +218,22 @@ struct TotalCounts { } /// Collection stats /// -/// Get stats for a collection over a specific time period -/// -/// API docs note: the **Body** fields here are actually query parameters!! -/// -/// Due to limitations with dropshot's query parsing (no support for sequences), -/// this is kind of the best i could do for now. sadly. +/// Get record statistics for collections during a specific time period #[endpoint { method = GET, path = "/collections/stats" }] -async fn get_records_total_seen( +async fn get_collection_stats( ctx: RequestContext, - query: VecsAllowedQuery, + collections_query: MultiCollectionQuery, + query: Query, ) -> OkCorsResponse> { let Context { storage, .. } = ctx.context(); let q = query.into_inner(); + let collections: HashSet = collections_query.try_into()?; - log::warn!("collection: {:?}", q.collection); - - let mut collections = Vec::with_capacity(q.collection.len()); - for c in q.collection { - let Ok(nsid) = Nsid::new(c.clone()) else { - return Err(HttpError::for_bad_request( - None, - format!("could not parse collection to nsid: {c}"), - )); - }; - collections.push(nsid); - } - - let since = q.since.map(dt_to_cursor).transpose()?; - let until = q.until.map(dt_to_cursor).transpose()?; + let _since = q.since.map(dt_to_cursor).transpose()?; + let _until = q.until.map(dt_to_cursor).transpose()?; let mut seen_by_collection = HashMap::with_capacity(collections.len()); @@ -266,7 +252,7 @@ async fn get_records_total_seen( ); } - ok_cors(seen_by_collection) + OkCors(seen_by_collection).into() } #[derive(Debug, Serialize, JsonSchema)] @@ -315,7 +301,9 @@ struct CollectionsQuery { order: Option, } -/// List collections (with stats) +/// List collections +/// +/// With statistics. /// /// ## To fetch a full list: /// @@ -385,10 +373,11 @@ async fn get_collections( let next_cursor = next_cursor.map(|c| URL_SAFE_NO_PAD.encode(c)); - ok_cors(CollectionsResponse { + OkCors(CollectionsResponse { collections, cursor: next_cursor, }) + .into() } #[derive(Debug, Deserialize, JsonSchema)] @@ -439,7 +428,7 @@ async fn get_timeseries( let step = if let Some(secs) = q.step { if secs < 3600 { let msg = format!("step is too small: {}", secs); - return Err(HttpError::for_bad_request(None, msg)); + Err(HttpError::for_bad_request(None, msg))?; } (secs / 3600) * 3600 // trucate to hour } else { @@ -465,7 +454,7 @@ async fn get_timeseries( .map(|(k, v)| (k.to_string(), v.iter().map(Into::into).collect())) .collect(); - ok_cors(CollectionTimeseriesResponse { range, series }) + OkCors(CollectionTimeseriesResponse { range, series }).into() } pub async fn serve(storage: impl StoreReader + 'static) -> Result<(), String> { @@ -481,7 +470,7 @@ pub async fn serve(storage: impl StoreReader + 'static) -> Result<(), String> { api.register(get_openapi).unwrap(); api.register(get_meta_info).unwrap(); api.register(get_records_by_collections).unwrap(); - api.register(get_records_total_seen).unwrap(); + api.register(get_collection_stats).unwrap(); api.register(get_collections).unwrap(); api.register(get_timeseries).unwrap(); @@ -514,12 +503,3 @@ pub async fn serve(storage: impl StoreReader + 'static) -> Result<(), String> { .map_err(|error| format!("failed to start server: {}", error))? .await } - -/// awkward helpers -type OkCorsResponse = Result>, HttpError>; -fn ok_cors(t: T) -> OkCorsResponse { - let mut res = HttpResponseHeaders::new_unnamed(HttpResponseOk(t)); - res.headers_mut() - .insert("access-control-allow-origin", "*".parse().unwrap()); - Ok(res) -} -- 2.51.2