uqa_client/
http_engine.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7use std::ffi::OsStr;
8use std::fmt;
9use std::net::IpAddr;
10use std::time::Duration;
11
12use reqwest::header::CONTENT_TYPE;
13use reqwest::{Client, Response, Url};
14use secrecy::{ExposeSecret, SecretString};
15use serde::de::DeserializeOwned;
16use serde::Serialize;
17use uqa_sql::{AsyncSQLEngine, SQLParam, SQLResult};
18
19use crate::cli_connection;
20use crate::server_error_envelope::ServerErrorEnvelope;
21use crate::sql_batch_execution::SQLBatchWireResponse;
22use crate::sql_execution::SQLWireResponse;
23use crate::{HttpEngineError, SQLBatchExecution, SQLExecution, SQLStatement, SQLStream};
24
25const JSON_CONTENT_TYPE: &str = "application/json";
26const NDJSON_CONTENT_TYPE: &str = "application/x-ndjson";
27const REQUEST_ID_HEADER: &str = "x-request-id";
28const MAX_JSON_RESPONSE_BYTES: usize = 65 * 1024 * 1024;
29const MAX_ERROR_RESPONSE_BYTES: usize = 64 * 1024;
30
31/// Authenticated client for the SQL API shared by local and Cloud UQA nodes.
32#[derive(Clone)]
33pub struct HttpEngine {
34    http: Client,
35    base_url: Url,
36    credential: SecretString,
37}
38
39#[derive(Serialize)]
40struct SQLBatchRequest<'a> {
41    statements: &'a [SQLStatement],
42}
43
44impl HttpEngine {
45    /// Connect to one UQA data-plane origin.
46    ///
47    /// Plain HTTP is accepted only for loopback local nodes. Cloud endpoints must use HTTPS.
48    pub fn new(base_url: &str, credential: SecretString) -> Result<Self, HttpEngineError> {
49        if credential.expose_secret().is_empty() {
50            return Err(HttpEngineError::InvalidCredential);
51        }
52        let base_url = parse_base_url(base_url)?;
53        let http = Client::builder()
54            .no_proxy()
55            .connect_timeout(Duration::from_secs(10))
56            .redirect(reqwest::redirect::Policy::none())
57            .user_agent(concat!("uqa-client/", env!("CARGO_PKG_VERSION")))
58            .build()
59            .map_err(HttpEngineError::build_client)?;
60        Ok(Self {
61            http,
62            base_url,
63            credential,
64        })
65    }
66
67    /// Read `UQA_URL` and `UQA_TOKEN`, as emitted by `uqa ... connection --format env`.
68    pub fn from_env() -> Result<Self, HttpEngineError> {
69        let base_url = std::env::var("UQA_URL")
70            .map_err(|_| HttpEngineError::MissingEnvironmentVariable("UQA_URL"))?;
71        let credential = std::env::var("UQA_TOKEN")
72            .map_err(|_| HttpEngineError::MissingEnvironmentVariable("UQA_TOKEN"))?;
73        Self::new(&base_url, SecretString::from(credential))
74    }
75
76    /// Resolve a local project through the installed `uqa` CLI, then connect to its data plane.
77    ///
78    /// The CLI is invoked only during construction. Subsequent SQL calls use HTTP directly.
79    pub async fn local(project: &str) -> Result<Self, HttpEngineError> {
80        Self::local_with_cli(project, "uqa").await
81    }
82
83    /// Resolve a local project through a specific `uqa` CLI executable.
84    pub async fn local_with_cli(
85        project: &str,
86        cli_path: impl AsRef<OsStr>,
87    ) -> Result<Self, HttpEngineError> {
88        let connection = cli_connection::resolve_local(cli_path.as_ref(), project).await?;
89        Self::new(&connection.url, connection.token)
90    }
91
92    /// Resolve a Cloud project and optional organization through the installed `uqa` CLI.
93    ///
94    /// Passing `None` uses the CLI's current default organization. The CLI is invoked only during
95    /// construction; subsequent SQL calls use HTTP directly.
96    pub async fn cloud(project: &str, organization: Option<&str>) -> Result<Self, HttpEngineError> {
97        Self::cloud_with_cli(project, organization, "uqa").await
98    }
99
100    /// Resolve a Cloud project through a specific `uqa` CLI executable.
101    pub async fn cloud_with_cli(
102        project: &str,
103        organization: Option<&str>,
104        cli_path: impl AsRef<OsStr>,
105    ) -> Result<Self, HttpEngineError> {
106        let connection =
107            cli_connection::resolve_cloud(cli_path.as_ref(), project, organization).await?;
108        Self::new(&connection.url, connection.token)
109    }
110
111    /// Execute one materialized SQL statement through `POST /v1/sql`.
112    pub async fn sql(
113        &self,
114        query: &str,
115        params: &[SQLParam],
116    ) -> Result<SQLResult, HttpEngineError> {
117        Ok(self.sql_with_metadata(query, params).await?.into_result())
118    }
119
120    /// Execute one materialized SQL statement and preserve its request ID.
121    pub async fn sql_with_metadata(
122        &self,
123        query: &str,
124        params: &[SQLParam],
125    ) -> Result<SQLExecution, HttpEngineError> {
126        let statement = SQLStatement::new(query, params)?;
127        let response = self
128            .authorized(self.http.post(self.endpoint("v1/sql")?))
129            .json(&statement)
130            .send()
131            .await
132            .map_err(HttpEngineError::transport)?;
133        let request_id = response_request_id(&response)?;
134        let result = decode_json_response::<SQLWireResponse>(response).await?;
135        validate_request_id(&request_id, &result.request_id)?;
136        Ok(SQLExecution::from_wire(result))
137    }
138
139    /// Execute every statement atomically through `POST /v1/sql/batch`.
140    pub async fn sql_batch(
141        &self,
142        statements: &[(&str, &[SQLParam])],
143    ) -> Result<Vec<SQLResult>, HttpEngineError> {
144        Ok(self
145            .sql_batch_with_metadata(statements)
146            .await?
147            .into_results())
148    }
149
150    /// Execute an atomic SQL batch and preserve its request ID.
151    pub async fn sql_batch_with_metadata(
152        &self,
153        statements: &[(&str, &[SQLParam])],
154    ) -> Result<SQLBatchExecution, HttpEngineError> {
155        let statements = statements
156            .iter()
157            .map(|(query, params)| SQLStatement::new(*query, params))
158            .collect::<Result<Vec<_>, _>>()?;
159        let response = self
160            .authorized(self.http.post(self.endpoint("v1/sql/batch")?))
161            .json(&SQLBatchRequest {
162                statements: &statements,
163            })
164            .send()
165            .await
166            .map_err(HttpEngineError::transport)?;
167        let request_id = response_request_id(&response)?;
168        let result = decode_json_response::<SQLBatchWireResponse>(response).await?;
169        validate_request_id(&request_id, &result.request_id)?;
170        Ok(SQLBatchExecution::from_wire(result))
171    }
172
173    /// Start an incremental SQL request through `POST /v1/sql/stream`.
174    pub async fn sql_stream(
175        &self,
176        query: &str,
177        params: &[SQLParam],
178    ) -> Result<SQLStream, HttpEngineError> {
179        let statement = SQLStatement::new(query, params)?;
180        let response = self
181            .authorized(self.http.post(self.endpoint("v1/sql/stream")?))
182            .header(reqwest::header::ACCEPT, NDJSON_CONTENT_TYPE)
183            .json(&statement)
184            .send()
185            .await
186            .map_err(HttpEngineError::transport)?;
187        if !response.status().is_success() {
188            return Err(error_from_response(response).await);
189        }
190        validate_content_type(&response, NDJSON_CONTENT_TYPE)?;
191        let request_id = response_request_id(&response)?;
192        Ok(SQLStream::new(response, request_id))
193    }
194
195    fn authorized(&self, request: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
196        request.bearer_auth(self.credential.expose_secret())
197    }
198
199    fn endpoint(&self, path: &str) -> Result<Url, HttpEngineError> {
200        self.base_url
201            .join(path)
202            .map_err(|_| HttpEngineError::InvalidBaseURL)
203    }
204}
205
206impl AsyncSQLEngine for HttpEngine {
207    type Error = HttpEngineError;
208
209    async fn sql<'a>(
210        &'a self,
211        query: &'a str,
212        params: &'a [SQLParam],
213    ) -> Result<SQLResult, Self::Error> {
214        HttpEngine::sql(self, query, params).await
215    }
216
217    async fn sql_batch<'a>(
218        &'a self,
219        statements: &'a [(&'a str, &'a [SQLParam])],
220    ) -> Result<Vec<SQLResult>, Self::Error> {
221        HttpEngine::sql_batch(self, statements).await
222    }
223}
224
225impl fmt::Debug for HttpEngine {
226    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
227        formatter
228            .debug_struct("HttpEngine")
229            .field("base_url", &"[REDACTED]")
230            .field("credential", &"[REDACTED]")
231            .finish()
232    }
233}
234
235fn parse_base_url(source: &str) -> Result<Url, HttpEngineError> {
236    let url = Url::parse(source).map_err(|_| HttpEngineError::InvalidBaseURL)?;
237    let valid_origin = url.username().is_empty()
238        && url.password().is_none()
239        && url.query().is_none()
240        && url.fragment().is_none()
241        && url.path() == "/"
242        && url.host_str().is_some();
243    if !valid_origin || !matches!(url.scheme(), "http" | "https") {
244        return Err(HttpEngineError::InvalidBaseURL);
245    }
246    if url.scheme() == "http" && !url.host_str().is_some_and(is_loopback_host) {
247        return Err(HttpEngineError::InsecureRemoteURL);
248    }
249    Ok(url)
250}
251
252fn is_loopback_host(host: &str) -> bool {
253    let host = host
254        .strip_prefix('[')
255        .and_then(|host| host.strip_suffix(']'))
256        .unwrap_or(host);
257    host.eq_ignore_ascii_case("localhost")
258        || host
259            .parse::<IpAddr>()
260            .is_ok_and(|address| address.is_loopback())
261}
262
263async fn decode_json_response<T: DeserializeOwned>(
264    response: Response,
265) -> Result<T, HttpEngineError> {
266    if !response.status().is_success() {
267        return Err(error_from_response(response).await);
268    }
269    validate_content_type(&response, JSON_CONTENT_TYPE)?;
270    let body = read_bounded(response, MAX_JSON_RESPONSE_BYTES).await?;
271    serde_json::from_slice(&body).map_err(HttpEngineError::InvalidResponse)
272}
273
274async fn error_from_response(response: Response) -> HttpEngineError {
275    let status = response.status();
276    if let Err(error) = validate_content_type(&response, JSON_CONTENT_TYPE) {
277        return error;
278    }
279    let header_request_id = match response_request_id(&response) {
280        Ok(request_id) => request_id,
281        Err(error) => return error,
282    };
283    let body = match read_bounded(response, MAX_ERROR_RESPONSE_BYTES).await {
284        Ok(body) => body,
285        Err(error) => return error,
286    };
287    let Ok(envelope) = serde_json::from_slice::<ServerErrorEnvelope>(&body) else {
288        return HttpEngineError::Server {
289            status,
290            code: "HTTP_ERROR".to_owned(),
291            message: "UQA returned a non-success response".to_owned(),
292            request_id: Some(header_request_id),
293        };
294    };
295    if header_request_id != envelope.request_id {
296        return HttpEngineError::ResponseRequestIdMismatch;
297    }
298    HttpEngineError::Server {
299        status,
300        code: envelope.error.code,
301        message: envelope.error.message,
302        request_id: Some(envelope.request_id),
303    }
304}
305
306async fn read_bounded(
307    mut response: Response,
308    maximum_bytes: usize,
309) -> Result<Vec<u8>, HttpEngineError> {
310    if response
311        .content_length()
312        .is_some_and(|length| length > maximum_bytes as u64)
313    {
314        return Err(HttpEngineError::ResponseTooLarge);
315    }
316    let mut body = Vec::new();
317    while let Some(chunk) = response.chunk().await.map_err(HttpEngineError::transport)? {
318        if body.len().saturating_add(chunk.len()) > maximum_bytes {
319            return Err(HttpEngineError::ResponseTooLarge);
320        }
321        body.extend_from_slice(&chunk);
322    }
323    Ok(body)
324}
325
326fn validate_content_type(response: &Response, expected: &str) -> Result<(), HttpEngineError> {
327    let valid = response
328        .headers()
329        .get(CONTENT_TYPE)
330        .and_then(|value| value.to_str().ok())
331        .and_then(|value| value.split(';').next())
332        .is_some_and(|value| value.trim().eq_ignore_ascii_case(expected));
333    if valid {
334        Ok(())
335    } else {
336        Err(HttpEngineError::UnexpectedContentType)
337    }
338}
339
340fn response_request_id(response: &Response) -> Result<String, HttpEngineError> {
341    response
342        .headers()
343        .get(REQUEST_ID_HEADER)
344        .and_then(|value| value.to_str().ok())
345        .filter(|value| !value.is_empty())
346        .map(str::to_owned)
347        .ok_or(HttpEngineError::MissingRequestId)
348}
349
350fn validate_request_id(header: &str, body: &str) -> Result<(), HttpEngineError> {
351    if header == body {
352        Ok(())
353    } else {
354        Err(HttpEngineError::ResponseRequestIdMismatch)
355    }
356}
357
358#[cfg(test)]
359mod tests {
360    use super::*;
361
362    #[test]
363    fn base_url_rejects_credentials_paths_and_remote_plain_http() {
364        for source in [
365            "http://user:secret@127.0.0.1:8432/",
366            "http://127.0.0.1:8432/v1",
367            "http://example.com/",
368            "ftp://127.0.0.1/",
369        ] {
370            assert!(parse_base_url(source).is_err(), "accepted {source}");
371        }
372        assert!(parse_base_url("http://127.0.0.1:8432/").is_ok());
373        assert!(parse_base_url("http://[::1]:8432/").is_ok());
374        assert!(parse_base_url("https://example.com/").is_ok());
375    }
376
377    #[test]
378    fn debug_output_redacts_endpoint_and_credential() {
379        let credential = "uqa_db_customer-secret";
380        let client =
381            HttpEngine::new("http://127.0.0.1:8432/", SecretString::from(credential)).unwrap();
382        let debug = format!("{client:?}");
383        assert!(!debug.contains("127.0.0.1"));
384        assert!(!debug.contains(credential));
385    }
386}
uqa_client/http_engine.rs

uqa_client/
http_engine.rs
