diff --git a/CHANGELOG.md b/CHANGELOG.md index fb91c83..3131621 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ User-visible changes are recorded here. ## Unreleased +### Added + +- Embed the data-only HTTP API in an existing host with `http::DataService`, + using a mountable Axum router or a direct request handler. Reuse SQLRest's + parsing, deadlines and errors without binding standalone listeners. + ## 0.0.1 - 2026-09-15 ### Added diff --git a/README.md b/README.md index 2898188..6effaad 100644 --- a/README.md +++ b/README.md @@ -78,6 +78,10 @@ Read [interfaces](docs/interfaces.md), [execution](docs/execution.md), [registry](docs/registry.md), [migrations](docs/migrations.md), and [HTTP](docs/http.md) for details. Embedded users call `Registry` from Tokio; standalone callers use the same core through HTTP. +Hosts with an existing HTTP server can mount `http::DataService::router()` with +Axum's `nest_service` or +call `DataService::handle()` without binding SQLRest listeners; see +[HTTP embedding](docs/http.md#embedding-and-shutdown). Important boundaries: diff --git a/docs/http.md b/docs/http.md index 9abf338..6a8d159 100644 --- a/docs/http.md +++ b/docs/http.md @@ -82,6 +82,51 @@ shutdown cancels incomplete uploads. Other management routes ignore bodies. ## Embedding and shutdown +To mount data HTTP access inside an existing Axum host without opening either +SQLRest listener, use `DataService`: + +```rust +use axum::Router; +use sqlrest::{http::DataService, registry::Registry}; + +fn mount(registry: Registry) -> (Router, DataService) { + let data = DataService::new(registry); + let app = Router::new().nest_service("/api/sql", data.router()); + (app, data) +} +``` + +Requests use `/api/sql/db/{name}/...`; Axum strips `/api/sql` before dispatch. +The host must apply its login and per-database authorization before forwarding +requests. No management endpoints are installed. Continue using Registry methods +for publication and status in the host's trusted code. + +Use `nest_service`, not `nest`: SQLRest's router uses a fallback handler, and +Axum's `route_layer` does not cover nested fallbacks. With `nest_service`, apply +the host's authorization `route_layer` **after** mounting the service so it +protects every method under the mount, including unknown paths. The mount alone +does not authenticate requests. If forwarding through a custom fallback instead, +use middleware that also covers fallbacks (such as `layer`), not `route_layer`. + +For a custom adapter, `data.handle(request).await` accepts an +`axum::extract::Request` and returns an `axum::response::Response`. Supply the +URI as `/db/{name}/...`, preserving its encoded path and query; strip only the +outer mount prefix, without decoding/re-encoding parameters. Both entrypoints +reuse the standalone server's request parsing, deadlines, error serialization, +HEAD handling and execution. Host middleware may intentionally impose additional +limits; use a fallback or all-method forwarding route to preserve SQLRest's +explicit HEAD/OPTIONS and JSON 404/405 behavior. + +Keep a `DataService` clone for lifecycle management. During host shutdown, +call `data.shutdown().await` while request tasks and Tokio are still running, +alongside draining the host's own HTTP server. It closes the entire shared +Registry, cancels incomplete uploads on this service and its clones, and waits +for core cleanup. It does not stop host listeners. Dropping a clone/router does +not shut down the Registry. Independently constructed services have independent +upload-cancellation tokens; prefer clones when mounting the same service more +than once. OpenAPI remains available through the Registry; use the external +server URL `/api/sql/db/{name}` when generating it. + Open with `Registry::open(workspace).await` in a live Tokio runtime, or pass it to `http::Server::bind` / `from_listeners`. Await `serve(shutdown_future)` to serve both listeners. `Registry::shutdown().await` closes global admission and drains diff --git a/src/http.rs b/src/http.rs index 31cf3d5..3fd6400 100644 --- a/src/http.rs +++ b/src/http.rs @@ -1,4 +1,4 @@ -//! Independent data and management listeners over the shared Registry. +//! Embeddable data HTTP service and independent data/management listeners. use crate::{ SqlrestError, params::Input, @@ -16,6 +16,68 @@ use std::{future::Future, net::SocketAddr}; use tokio::{net::TcpListener, task::JoinSet}; use tokio_util::sync::CancellationToken; +/// Data-only HTTP access to a shared Registry, without binding any listeners. +/// +/// Requests use `/db/{name}/...`. The host owns authentication and authorization +/// for each database and strips any outer mount prefix before calling `handle`. +/// Clones share shutdown state; dropping a clone does not shut down the service. +#[derive(Clone)] +pub struct DataService { + state: ServiceState, +} + +impl DataService { + pub fn new(registry: Registry) -> Self { + Self { + state: ServiceState { + registry, + stop: CancellationToken::new(), + }, + } + } + + /// Handle an HTTP request using the standalone server's complete data contract. + pub async fn handle(&self, request: Request) -> Response { + data_handler(State(self.state.clone()), request).await + } + + /// Mountable router with no management routes or implicit HEAD/OPTIONS. + /// + /// Mount with `Router::nest_service`, which strips the mount prefix and lets + /// the host apply `route_layer` authorization to the mounted service. + /// Do not use `Router::nest` with host `route_layer`: this router uses a + /// fallback handler, and `route_layer` does not protect nested fallbacks. + /// + /// ``` + /// use axum::Router; + /// use sqlrest::{http::DataService, registry::Registry}; + /// + /// fn mount(registry: Registry) -> (Router, DataService) { + /// let data = DataService::new(registry); + /// let app = Router::new().nest_service("/api/sql", data.router()); + /// // The host must add its authentication/authorization middleware. + /// // Requests now use /api/sql/db/{name}/... + /// (app, data) + /// } + /// ``` + pub fn router(&self) -> Router { + Router::new() + .fallback(data_handler) + .with_state(self.state.clone()) + } + + /// Close admission, cancel incomplete uploads, and await core cleanup. + /// + /// This shuts down the entire shared Registry, including other services + /// using it. Keep Tokio alive until this future completes. The host remains + /// responsible for stopping its own listeners. + pub async fn shutdown(&self) -> Result<(), SqlrestError> { + self.state.registry.start_shutdown()?; + self.state.stop.cancel(); + self.state.registry.shutdown().await + } +} + pub struct Server { registry: Registry, data: TcpListener, @@ -77,9 +139,10 @@ impl Server { }; // A fallback handler keeps explicit HEAD/OPTIONS and our JSON errors; // it does not install implicit GET-to-HEAD or CORS behavior. - let data = Router::new() - .fallback(data_handler) - .with_state(state.clone()); + let data = DataService { + state: state.clone(), + } + .router(); let management = Router::new().fallback(management_handler).with_state(state); let mut servers = JoinSet::new(); let data_stop = stop.clone(); diff --git a/tests/embedded_http_contract.rs b/tests/embedded_http_contract.rs new file mode 100644 index 0000000..e31a00f --- /dev/null +++ b/tests/embedded_http_contract.rs @@ -0,0 +1,253 @@ +use axum::{ + Router, + body::{Body, Bytes, to_bytes}, + http::{Request, StatusCode}, +}; +use serde_json::{Value, json}; +use sqlrest::{ + http::DataService, + registry::{Outcome, PublishRequest, Registry}, +}; +use std::{fs, time::Duration}; +use tower::ServiceExt; + +async fn service(timeout_ms: u64) -> (tempfile::TempDir, Registry, DataService) { + let directory = tempfile::tempdir().unwrap(); + let root = directory.path().join("databases/test"); + fs::create_dir_all(root.join("interfaces/echo/[id]")).unwrap(); + fs::create_dir_all(root.join("interfaces/broken")).unwrap(); + fs::create_dir_all(root.join("migrations")).unwrap(); + fs::write( + root.join("interfaces/echo/[id]/post.sql"), + "SELECT ${path.id:string} AS id, ${body.value:string} AS value", + ) + .unwrap(); + fs::write( + root.join("interfaces/echo/[id]/post.response.yaml"), + r#"{"type":"object","properties":{"id":{"type":"string"},"value":{"type":"string"}},"required":["id","value"],"additionalProperties":false}"#, + ) + .unwrap(); + fs::write( + root.join("interfaces/broken/get.sql"), + "SELECT private_missing_function() AS value", + ) + .unwrap(); + fs::write( + root.join("interfaces/broken/get.response.yaml"), + r#"{"type":"object","properties":{"value":{"type":"string"}},"required":["value"],"additionalProperties":false}"#, + ) + .unwrap(); + let registry = Registry::open(directory.path()).await.unwrap(); + let config: PublishRequest = serde_json::from_value(json!({ + "database": {"kind": "turso"}, + "limits": {"request_timeout_ms": timeout_ms} + })) + .unwrap(); + let operation = registry.publish("test", config).unwrap(); + let result = registry.wait_operation("test", operation).await.unwrap(); + assert_eq!(result.outcome, Outcome::Succeeded); + let service = DataService::new(registry.clone()); + (directory, registry, service) +} + +fn post(uri: &str, body: Body) -> Request { + Request::post(uri) + .header("content-type", "application/json") + .body(body) + .unwrap() +} + +async fn json_body(response: axum::response::Response) -> Value { + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()).unwrap() +} + +#[tokio::test] +async fn data_service_mounts_without_listeners_and_preserves_http_contract() { + let (_directory, _registry, service) = service(5000).await; + drop(service.clone()); + let app = Router::new().nest_service("/corva", service.router()); + let response = app + .clone() + .oneshot(post( + "/corva/db/test/echo/a%252Fb", + Body::from(r#"{"value":"hello"}"#), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + json_body(response).await, + json!({"records":[{"id":"a%2Fb","value":"hello"}]}) + ); + + for (uri, body, expected) in [ + ( + "/db/test/echo/a%2Fb", + r#"{"value":"hello"}"#, + "invalid_path", + ), + ( + "/db/test/echo/a", + r#"{"value":1}"#, + "parameter_type_mismatch", + ), + ] { + let response = service.handle(post(uri, Body::from(body))).await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(json_body(response).await["error"]["code"], expected); + } + + let response = app + .clone() + .oneshot( + Request::head("/corva/db/test/broken") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED); + assert_eq!(response.headers()["allow"], "GET"); + assert!( + to_bytes(response.into_body(), usize::MAX) + .await + .unwrap() + .is_empty() + ); + + let response = service + .handle(Request::get("/db/test/broken").body(Body::empty()).unwrap()) + .await; + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert!(!String::from_utf8_lossy(&bytes).contains("private_missing_function")); + + let response = app + .oneshot( + Request::get("/corva/databases") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + service.shutdown().await.unwrap(); +} + +#[tokio::test] +async fn mounted_data_service_respects_host_route_authorization() { + async fn authorize( + request: axum::extract::Request, + next: axum::middleware::Next, + ) -> axum::response::Response { + use axum::response::IntoResponse; + + if request + .headers() + .get("authorization") + .and_then(|v| v.to_str().ok()) + != Some("Bearer test-token") + { + return StatusCode::UNAUTHORIZED.into_response(); + } + next.run(request).await + } + + let (_directory, _registry, service) = service(5000).await; + let app = Router::new() + .route("/health", axum::routing::get(|| async { "ok" })) + .nest_service("/corva", service.router()) + .route_layer(axum::middleware::from_fn(authorize)); + + // Include an ordinary host route so replacing nest_service with nest would + // silently bypass auth on SQLRest, rather than merely panic in route_layer. + for (method, uri) in [ + ("GET", "/health"), + ("POST", "/corva/db/test/echo/a"), + ("HEAD", "/corva/db/test/echo/a"), + ("OPTIONS", "/corva/db/test/echo/a"), + ("GET", "/corva/unknown"), + ("GET", "/corva"), + ("GET", "/corva/"), + ] { + let response = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(uri) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::UNAUTHORIZED, + "{method} {uri}" + ); + } + + let mut request = post( + "/corva/db/test/echo/a%252Fb", + Body::from(r#"{"value":"authorized"}"#), + ); + request + .headers_mut() + .insert("authorization", "Bearer test-token".parse().unwrap()); + let response = app.oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + json_body(response).await, + json!({"records":[{"id":"a%2Fb","value":"authorized"}]}) + ); + service.shutdown().await.unwrap(); +} + +fn pending_body() -> Body { + Body::from_stream(futures_util::stream::pending::>()) +} + +#[tokio::test] +async fn embedded_upload_uses_request_deadline() { + let (_directory, _registry, service) = service(30).await; + let response = tokio::time::timeout( + Duration::from_secs(2), + service.handle(post("/db/test/echo/a", pending_body())), + ) + .await + .unwrap(); + assert_eq!( + json_body(response).await["error"]["code"], + "execution_timeout" + ); + service.shutdown().await.unwrap(); +} + +#[tokio::test] +async fn embedded_shutdown_cancels_upload_and_closes_shared_registry() { + let (_directory, registry, service) = service(60000).await; + let request_service = service.clone(); + let task = tokio::spawn(async move { + request_service + .handle(post("/db/test/echo/a", pending_body())) + .await + }); + tokio::time::timeout(Duration::from_secs(2), async { + while registry.status("test").unwrap().active_requests == 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(2), service.shutdown()) + .await + .unwrap() + .unwrap(); + let response = task.await.unwrap(); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + let response = service + .handle(post("/db/test/echo/a", Body::from(r#"{"value":"x"}"#))) + .await; + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); +}