diff --git a/crates/strata/src/dataset.rs b/crates/strata/src/dataset.rs index 836d301..baa37b1 100644 --- a/crates/strata/src/dataset.rs +++ b/crates/strata/src/dataset.rs @@ -9,6 +9,7 @@ use anyhow::{Result, bail}; use futures::stream::{BoxStream, StreamExt}; use schema::Schema; +use serde::Serialize; use serde_json::Value; use crate::page::Cursor; @@ -17,7 +18,7 @@ use crate::record::Records; /// How a sink should apply a written dataset. Rides as metadata on the existing /// `put` verb (the reserved `disposition` query param) rather than a new verb, so /// every sink shares one write surface. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize)] pub enum Disposition { /// Insert every row (the historical behavior). Re-running adds duplicates. #[default] diff --git a/crates/strata/src/flight/mod.rs b/crates/strata/src/flight/mod.rs index 6062721..ca06cc4 100644 --- a/crates/strata/src/flight/mod.rs +++ b/crates/strata/src/flight/mod.rs @@ -55,12 +55,12 @@ impl StrataFlight { .get(provider) .map_err(|e| Status::not_found(e.to_string()))?; - let schema = provider - .resolve_schema(path) + let endpoint = provider + .resolve(path) .await .map_err(|e| Status::internal(e.to_string()))?; - Ok(strata_schema_to_arrow_schema(&schema)) + Ok(strata_schema_to_arrow_schema(&endpoint.response)) } } @@ -164,10 +164,10 @@ impl FlightService for StrataFlight { .map_err(|e| Status::internal(e.to_string()))?; // Advertise the static schema per endpoint; endpoints whose schema // is dynamic or not record-shaped (e.g. a generic `Row`) are skipped. - for (path, schema) in provider.endpoint_schemas() { - let arrow_schema = strata_schema_to_arrow_schema(&schema); + for endpoint in provider.endpoints() { + let arrow_schema = strata_schema_to_arrow_schema(&endpoint.response); - let descriptor = FlightDescriptor::new_path(vec![name.clone(), path]); + let descriptor = FlightDescriptor::new_path(vec![name.clone(), endpoint.path]); if let Ok(info) = FlightInfo::new().try_with_schema(&arrow_schema) { infos.push(Ok(info.with_descriptor(descriptor))); } diff --git a/crates/strata/src/graphql/mod.rs b/crates/strata/src/graphql/mod.rs index 10b431c..da7a548 100644 --- a/crates/strata/src/graphql/mod.rs +++ b/crates/strata/src/graphql/mod.rs @@ -62,12 +62,13 @@ async fn build_schema(registry: &Arc) -> Result { for table in tables(registry, &mount).await { let path = format!("/tables/{table}"); let provider = registry.get(&mount)?; - if !provider.queryable(&path) { - continue; - } - let Ok(row_schema) = provider.resolve_schema(&path).await else { + let Ok(endpoint) = provider.resolve(&path).await else { continue; }; + if !endpoint.metadata.queryable { + continue; + } + let row_schema = endpoint.response; let type_name = format!("{mount}_{table}"); objects.push(row_object(&type_name, &row_schema)); query = query.field(table_field( diff --git a/crates/strata/src/main.rs b/crates/strata/src/main.rs index d3af86c..6967c5a 100644 --- a/crates/strata/src/main.rs +++ b/crates/strata/src/main.rs @@ -168,8 +168,8 @@ async fn run() -> Result<()> { Command::List { provider: Some(name), } => { - for route in registry.get(&name)?.routes() { - println!("{route}"); + for endpoint in registry.get(&name)?.endpoints() { + println!("{}", endpoint.path); } } Command::Schema { provider: None } => { @@ -248,14 +248,14 @@ async fn run() -> Result<()> { // unreachable at startup shouldn't stop `serve`. match registry .get(&pipe.source.mount)? - .resolve_schema(&pipe.source.path) + .resolve(&pipe.source.path) .await { - Ok(schema) => { + Ok(endpoint) => { for ep in [&pipe.source, &pipe.destination] { let ep = strata_types::Endpoint::new(&ep.mount, strip_query(&ep.path)); - db.upsert_source_schema(&ep, &schema).await?; + db.upsert_source_schema(&ep, &endpoint.response).await?; } } Err(e) => tracing::warn!( diff --git a/crates/strata/src/page.rs b/crates/strata/src/page.rs index edc4416..215adc6 100644 --- a/crates/strata/src/page.rs +++ b/crates/strata/src/page.rs @@ -48,7 +48,7 @@ impl Page { /// `list` route (`.strategy(…)`); the router only stores the signal — the sync /// system alongside `pipe` is what interprets it. The two differ in where a run /// starts, when it stops, and what a re-sync does. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] pub enum ListStrategy { /// Finite backfill by offset/limit. `next` reaches `None` at the tail, so a run /// starts from the beginning and drains to completion. A re-sync re-scans from diff --git a/crates/strata/src/pipe/mod.rs b/crates/strata/src/pipe/mod.rs index f94dfda..667ee9d 100644 --- a/crates/strata/src/pipe/mod.rs +++ b/crates/strata/src/pipe/mod.rs @@ -19,7 +19,8 @@ use sync::Progress; pub async fn run_pass(registry: &Registry, store: &S, pipe: &mut Pipe) -> Result<()> { let src = registry.get(&pipe.source.mount)?; let dst = registry.get(&pipe.destination.mount)?; - let strategy = src.strategy(&pipe.source.path).ok_or_else(|| { + let meta = src.resolve(&pipe.source.path).await?.metadata; + let strategy = meta.strategy.ok_or_else(|| { anyhow!( "source `{}{}` is not a list endpoint (no sync strategy)", pipe.source.mount, @@ -36,7 +37,7 @@ pub async fn run_pass(registry: &Registry, store: &S, pipe: &mut P // The source declares how a sink should apply its rows (merge if it re-emits // updates, else append). The sink still skips existing keys on append. let dst_path = WriteRequest::new(&pipe.destination.path) - .with_disposition(src.disposition(&pipe.source.path)) + .with_disposition(meta.disposition) .path(); let mut chunks = stream.chunks; diff --git a/crates/strata/src/provider.rs b/crates/strata/src/provider.rs index 3652cf2..f209f09 100644 --- a/crates/strata/src/provider.rs +++ b/crates/strata/src/provider.rs @@ -9,13 +9,11 @@ use std::collections::HashMap; use std::sync::Arc; use anyhow::{Result, anyhow, bail}; -use schema::Schema; use serde_json::{Value, json}; use crate::catalog::Catalog; use crate::config::ProviderConfig; -use crate::dataset::{DataStream, Disposition}; -use crate::page::ListStrategy; +use crate::dataset::DataStream; use crate::router::{Body, BoxFuture, EndpointInfo, Method, Response, Router}; /// Implemented by each concrete provider. Knows its state type and how to wire @@ -41,24 +39,11 @@ pub trait ProviderObject: Send + Sync { /// Wire the mount-scoped [`Catalog`] into this provider's router, so its /// handlers can read their own persisted annotations. fn set_catalog(&mut self, catalog: Catalog); - fn routes(&self) -> Vec; - /// Machine-readable description of every endpoint (path, params, response - /// schema), for introspection. + /// Every endpoint, statically described (dynamic resolvers not run). fn endpoints(&self) -> Vec; - /// Static response `DataType` of every endpoint, paired with its pattern. - fn endpoint_schemas(&self) -> Vec<(String, Schema)>; - /// Response `DataType` of the endpoint matching `path`, resolved (running a - /// dynamic schema resolver if the endpoint has one). - fn resolve_schema<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result>; - /// The declared [`ListStrategy`] of the `list` route matching `path`, for the - /// external sync layer to drive its walk. `None` if no list route matches. - fn strategy(&self, path: &str) -> Option; - /// The declared write [`Disposition`] of the `list` route matching `path` — - /// whether a pipe should merge (source re-emits updates) or append. Defaults to - /// `Append`. - fn disposition(&self, path: &str) -> Disposition; - // TODO: Make this items just metadata map on the route - fn queryable(&self, path: &str) -> bool; + /// The read endpoint matching a concrete `path`, with its response schema + /// resolved (running a dynamic resolver if the route has one). + fn resolve<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result>; /// Auto-pick the `List` read by path and return it as a [`DataStream`] — the /// data plane, for callers that consume the Arrow stream (pipe, CLI, Flight /// `do_get`). The router loops the provider's single-page handler internally. @@ -92,39 +77,14 @@ where self.router.set_catalog(catalog); } - fn routes(&self) -> Vec { - self.router.patterns() - } - fn endpoints(&self) -> Vec { self.router.endpoints() } - fn endpoint_schemas(&self) -> Vec<(String, Schema)> { - self.router.schemas() - } - - fn resolve_schema<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result> { + fn resolve<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result> { let state = self.state.clone(); let path = path.to_string(); - Box::pin(async move { - self.router - .resolve_schema(state, &path) - .await - .ok_or_else(|| anyhow!("no endpoint matches `{path}`"))? - }) - } - - fn strategy(&self, path: &str) -> Option { - self.router.strategy(path) - } - - fn disposition(&self, path: &str) -> Disposition { - self.router.disposition(path) - } - - fn queryable(&self, path: &str) -> bool { - self.router.queryable(path) + Box::pin(async move { self.router.resolve(state, &path).await }) } fn read<'a>(&'a self, path: &'a str) -> BoxFuture<'a, Result> { diff --git a/crates/strata/src/router.rs b/crates/strata/src/router.rs index 74ca63f..c69397f 100644 --- a/crates/strata/src/router.rs +++ b/crates/strata/src/router.rs @@ -210,12 +210,6 @@ fn split_path(path: &str) -> impl Iterator { path.split('/').filter(|s| !s.is_empty()) } -/// Split a call path into `(path, query)` at the first `?`. The query (without -/// the `?`) is `""` when absent. -fn split_query(path: &str) -> (&str, &str) { - path.split_once('?').unwrap_or((path, "")) -} - /// The verb an endpoint answers. Reads (`Get`, `List`) take no body. The write /// verbs take a body: `Create` a single typed entity (provider assigns /// identity); `Put` a whole [`Dataset`] (schema + rows) — the write-dual of @@ -305,14 +299,22 @@ struct Entry { schema_resolver: Option>, /// Human-readable description of the endpoint, for introspection. description: Option, - strategy: Option, - /// How a sink should apply this source's rows when piped: `Merge` if the source - /// re-emits updated elements (upsert on key), else `Append`. Independent of the - /// walk `strategy`. - disposition: Option, - /// Whether this list route accepts the `filter`/`fields` read params — the - /// signal a query surface (GraphQL) reads to expose `where` + projection. - queryable: bool, + metadata: RouterMetadata, +} + +impl Entry { + /// Static description of this route (dynamic resolver not run). + fn info(&self) -> EndpointInfo { + EndpointInfo { + method: self.method, + path: self.pattern.as_str().to_string(), + description: self.description.clone(), + params: self.pattern.param_names(), + body: self.body_schema.clone(), + response: self.response_schema.clone(), + metadata: self.metadata.clone(), + } + } } /// Erase a `get` handler `(Arc, Params) -> T`: an entity-plane read. The result @@ -522,9 +524,6 @@ impl Route { self } - /// Declare how a sink should apply this source's rows when piped: `Merge` if the - /// source re-emits updated elements (upsert on key), else the default `Append`. - /// Independent of the walk [`strategy`](Self::strategy). pub fn writes(mut self, disposition: Disposition) -> Self { self.disposition = Some(disposition); self @@ -606,8 +605,6 @@ impl Route { self } - /// Attach a human-readable description of the endpoint, surfaced in the - /// introspection output (`strata schema`) alongside its method and schemas. pub fn description(mut self, description: impl Into) -> Self { self.description = Some(description.into()); self @@ -638,29 +635,58 @@ impl Route { response_schema: self.response_schema.unwrap_or_else(Schema::empty), schema_resolver: self.schema_resolver, description: self.description, - strategy: self.strategy, - disposition: self.disposition, - queryable: self.queryable, + metadata: RouterMetadata { + strategy: self.strategy, + disposition: self.disposition.unwrap_or_default(), + queryable: self.queryable, + }, } } } -/// A machine-readable description of one endpoint, for introspection. -#[derive(Debug, Serialize)] +#[derive(Debug, Clone, Default, Serialize)] +pub struct RouterMetadata { + #[serde(skip_serializing_if = "Option::is_none")] + pub strategy: Option, + pub disposition: Disposition, + pub queryable: bool, +} + +#[derive(Debug, Clone)] pub struct EndpointInfo { - /// The verb this endpoint answers: `get`, `list`, `create`, or `upsert`. pub method: Method, /// The route pattern, e.g. `/repos/:owner/:name`. pub path: String, - /// Human-readable description of the endpoint, if declared. Omitted when unset. - #[serde(skip_serializing_if = "Option::is_none")] pub description: Option, /// Names of the path captures the endpoint takes. pub params: Vec, - /// JSON Schema of the request body, for write verbs. `null` for reads. - pub body: Value, - /// JSON Schema of the value this endpoint returns. - pub response: Value, + /// Request-body schema (empty for reads). + pub body: Schema, + /// Response schema. + pub response: Schema, + pub metadata: RouterMetadata, +} + +impl Serialize for EndpointInfo { + fn serialize(&self, serializer: S) -> Result { + use serde::ser::SerializeMap; + let mut map = serializer.serialize_map(None)?; + map.serialize_entry("method", &self.method)?; + map.serialize_entry("path", &self.path)?; + if let Some(description) = &self.description { + map.serialize_entry("description", description)?; + } + map.serialize_entry("params", &self.params)?; + let body = if self.method.is_write() { + self.body.to_json_schema() + } else { + Value::Null + }; + map.serialize_entry("body", &body)?; + map.serialize_entry("response", &self.response.to_json_schema())?; + map.serialize_entry("metadata", &self.metadata)?; + map.end() + } } /// Maps route patterns to handlers for one provider whose state is `S`. @@ -695,91 +721,46 @@ impl Router { self.routes.push(route.into_entry()); } - /// All registered patterns, in registration order (for `strata list`). - pub fn patterns(&self) -> Vec { - self.routes - .iter() - .map(|r| r.pattern.as_str().to_string()) - .collect() - } - - /// Describe every endpoint (path, params, response schema) for introspection. + /// Describe every endpoint (static schemas; dynamic resolvers not run) for + /// introspection, the CLI listing, and Flight `ListFlights`. pub fn endpoints(&self) -> Vec { - self.routes - .iter() - .map(|r| EndpointInfo { - method: r.method, - path: r.pattern.as_str().to_string(), - description: r.description.clone(), - params: r.pattern.param_names(), - body: if r.method.is_write() { - r.body_schema.to_json_schema() - } else { - Value::Null - }, - response: r.response_schema.to_json_schema(), - }) - .collect() - } - - /// Static response `DataType` of every route, paired with its pattern (for - /// Flight `ListFlights`). Ignores dynamic resolvers. - pub fn schemas(&self) -> Vec<(String, Schema)> { - self.routes - .iter() - .map(|r| (r.pattern.as_str().to_string(), r.response_schema.clone())) - .collect() + self.routes.iter().map(Entry::info).collect() } - /// The `DataType` of the *entity* at `path`. Resolves a **read** route only: - /// the data plane (`List` — exactly what [`dispatch_read`](Self::dispatch_read) - /// streams) if there is one, else the entity plane (`Get`) - pub async fn resolve_schema(&self, state: Arc, path: &str) -> Option> { - let (raw_path, _) = split_query(path); - let read = |method: Method| { - self.routes - .iter() - .find(move |r| r.method == method && r.pattern.match_path(raw_path).is_some()) + pub async fn resolve(&self, state: Arc, path: &str) -> Result { + // Prefer the List (data-plane) schema; fall back to Get for entity-only paths. + let (route, params) = match self.match_route(path, Method::List).await { + Ok(hit) => hit, + Err(_) => self.match_route(path, Method::Get).await?, }; - let route = read(Method::List).or_else(|| read(Method::Get))?; - match &route.schema_resolver { - Some(resolver) => { - let params = route.pattern.match_path(raw_path).unwrap_or_default(); - Some(resolver(state, params).await) - } - None => Some(Ok(route.response_schema.clone())), + let mut info = route.info(); + if let Some(resolver) = &route.schema_resolver { + info.response = resolver(state, params).await?; } + Ok(info) } - /// The declared [`ListStrategy`] of the `list` route matching `path`, or `None` - /// if no list route matches. The router only reports the signal; the external - /// sync layer is what interprets it. - pub fn strategy(&self, path: &str) -> Option { - let (raw_path, _) = split_query(path); - self.routes - .iter() - .find(|r| r.method == Method::List && r.pattern.match_path(raw_path).is_some()) - .and_then(|r| r.strategy) - } - - /// The declared write [`Disposition`] of the `list` route matching `path` — - /// whether a sink should merge (source re-emits updates) or append. Defaults to - /// `Append` when the route declares nothing or no list route matches. - pub fn disposition(&self, path: &str) -> Disposition { - let (raw_path, _) = split_query(path); - self.routes - .iter() - .find(|r| r.method == Method::List && r.pattern.match_path(raw_path).is_some()) - .and_then(|r| r.disposition) - .unwrap_or_default() - } - - pub fn queryable(&self, path: &str) -> bool { - let (raw_path, _) = split_query(path); - self.routes - .iter() - .find(|r| r.method == Method::List && r.pattern.match_path(raw_path).is_some()) - .is_some_and(|r| r.queryable) + async fn match_route(&self, path: &str, method: Method) -> Result<(&Entry, Params)> { + let (raw_path, query) = path.split_once('?').unwrap_or((path, "")); + for route in &self.routes { + if route.method == method + && let Some(mut params) = route.pattern.match_path(raw_path) + { + params.set_query(query); + if let Some(source) = &self.schema_source { + params.schema = source.schema(raw_path.to_string()).await?; + } + return Ok((route, params)); + } + } + bail!( + "no {method} route matches `{raw_path}`. known routes:\n {}", + self.routes + .iter() + .map(|r| r.pattern.as_str()) + .collect::>() + .join("\n ") + ) } /// Dispatch an explicit verb: find the route matching both `path` and @@ -792,8 +773,8 @@ impl Router { path: &str, body: Option, ) -> Result { - self.run(state, path, body, move |m| m == method, &method.to_string()) - .await + let (route, params) = self.match_route(path, method).await?; + (route.handler)(state, params, body).await } /// Auto-pick the `List` route matching `path` and return it as a real @@ -804,53 +785,22 @@ impl Router { /// each [`Chunk`] carrying its page's cursor as a checkpoint. Entity reads /// (`get`) are named explicitly via [`Router::dispatch`]; they aren't streams. pub async fn dispatch_read(&self, state: Arc, path: &str) -> Result { - let (raw_path, query) = split_query(path); - let route = self - .routes - .iter() - .find(|r| r.method == Method::List && r.pattern.match_path(raw_path).is_some()) - .ok_or_else(|| { - anyhow!( - "no list route matches `{raw_path}`. known routes:\n {}", - self.patterns().join("\n ") - ) - })?; - - // Path captures + the persisted annotation schema, resolved once (both are - // per-path, identical across pages). - let base = route.pattern.match_path(raw_path).unwrap_or_default(); - let annotations = match &self.schema_source { - Some(source) => source.schema(raw_path.to_string()).await?, - None => None, - }; + let (route, base) = self.match_route(path, Method::List).await?; - // The declared `DataType` of the stream (run the per-request resolver once). let schema = match &route.schema_resolver { - Some(resolver) => { - let mut params = base.clone(); - params.set_query(query); - params.schema = annotations.clone(); - resolver(state.clone(), params).await? - } + Some(resolver) => resolver(state.clone(), base.clone()).await?, None => route.response_schema.clone(), }; let handler = route.handler.clone(); - let query = query.to_string(); - // Unfold the single-page handler by cursor. State: `Some(None)` = first - // page, `Some(Some(token))` = resume, `None` = done. let chunks = futures::stream::unfold(Some(None::), move |token_state| { let handler = handler.clone(); let state = state.clone(); let base = base.clone(); - let annotations = annotations.clone(); - let query = query.clone(); async move { let token = token_state?; let mut params = base.clone(); - params.set_query(&query); - params.schema = annotations.clone(); if let Some(token) = &token { params.set_cursor(token); } @@ -875,37 +825,9 @@ impl Router { Ok(DataStream { schema, chunks }) } - /// Shared matcher: find the first route whose pattern matches `path` and - /// whose method satisfies `want`, then invoke it. - async fn run( - &self, - state: Arc, - path: &str, - body: Option, - want: impl Fn(Method) -> bool, - verb: &str, - ) -> Result { - let (raw_path, query) = split_query(path); - for route in &self.routes { - if want(route.method) - && let Some(mut params) = route.pattern.match_path(raw_path) - { - params.set_query(query); - if let Some(source) = &self.schema_source { - params.schema = source.schema(raw_path.to_string()).await?; - } - return (route.handler)(state, params, body).await; - } - } - bail!( - "no {verb} route matches `{raw_path}`. known routes:\n {}", - self.patterns().join("\n ") - ) - } - pub fn validate(&self) -> anyhow::Result<()> { for route in &self.routes { - if matches!(route.method, Method::List) && route.strategy.is_none() { + if matches!(route.method, Method::List) && route.metadata.strategy.is_none() { anyhow::bail!( "list method {:?} does not implement strategy", route.pattern.as_str() @@ -968,7 +890,7 @@ mod tests { } #[tokio::test] - async fn resolve_schema_picks_the_read_route_never_the_write() -> anyhow::Result<()> { + async fn resolve_picks_the_read_route_never_the_write() -> anyhow::Result<()> { let mut router: Router<()> = Router::new(); // `put` is registered *first* on `/rows` — resolving must still answer with // the `list` row schema, not the write's `WriteMeta`. @@ -977,26 +899,20 @@ mod tests { // A `get`-only path falls back to the entity plane. router.add(Route::new().path("/one").get(get_row)); - let rows = router - .resolve_schema(Arc::new(()), "/rows") - .await - .expect("works")?; - assert_eq!(rows, Row::schema()); + let rows = router.resolve(Arc::new(()), "/rows").await?; + assert_eq!(rows.response, Row::schema()); + assert_eq!(rows.method, Method::List); - let one = router - .resolve_schema(Arc::new(()), "/one") - .await - .expect("works")?; - assert_eq!(one, Row::schema()); + let one = router.resolve(Arc::new(()), "/one").await?; + assert_eq!(one.response, Row::schema()); Ok(()) } #[tokio::test] - async fn resolve_schema_ignores_write_only_paths() { + async fn resolve_ignores_write_only_paths() { let mut router: Router<()> = Router::new(); router.add(Route::new().path("/sink").put(put_rows)); - // No read route: there's no entity to describe. - assert!(router.resolve_schema(Arc::new(()), "/sink").await.is_none()); + assert!(router.resolve(Arc::new(()), "/sink").await.is_err()); } }