From f7b5ee5e4e567810517775709f6e791365cef5af Mon Sep 17 00:00:00 2001 From: Rafa Moreno Date: Sat, 17 Oct 2026 20:23:12 +0200 Subject: [PATCH 1/3] Scope cluster endpoints to organizations and send deployment headers --- src/client.rs | 101 +++++------- src/client/routing.rs | 266 +++++++++++++++++++++++++++++++ src/client/routing_tests.rs | 277 +++++++++++++++++++++++++++++++++ src/commands/auth/selection.rs | 2 +- src/commands/cluster.rs | 77 ++++----- src/commands/insert.rs | 1 - tests/api_contracts.rs | 6 +- tests/browser_login.rs | 6 +- tests/format.rs | 2 +- tests/sql_input.rs | 2 +- 10 files changed, 617 insertions(+), 123 deletions(-) create mode 100644 src/client/routing.rs create mode 100644 src/client/routing_tests.rs diff --git a/src/client.rs b/src/client.rs index 5da9ccc..f141a2a 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,3 +1,7 @@ +mod routing; +#[cfg(test)] +mod routing_tests; + use anyhow::{Context, Result}; use flate2::write::GzEncoder; use flate2::Compression; @@ -18,6 +22,7 @@ pub struct ApiClient { pub base_url: String, pub token: Option, client: Client, + routing: routing::RoutingCache, } impl ApiClient { @@ -26,6 +31,18 @@ impl ApiClient { base_url, token, client: Client::new(), + routing: routing::RoutingCache::default(), + } + } + + fn unrouted_request(&self, method: reqwest::Method, path: &str) -> RequestBuilder { + let request = with_client_header( + self.client + .request(method, format!("{}{}", self.base_url, path)), + ); + match &self.token { + Some(token) => request.bearer_auth(token), + None => request, } } @@ -39,13 +56,9 @@ impl ApiClient { body: &Value, timeout: Duration, ) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)) - .json(body) - .timeout(timeout); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self + .routed_request_with_timeout(reqwest::Method::POST, path, timeout)? + .json(body); let resp = req.send().context("failed to connect to server")?; handle_response(resp) } @@ -60,11 +73,7 @@ impl ApiClient { &self, path: &str, ) -> Result<(T, Option)> { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::POST, path)?; let resp = req.send().context("failed to connect to server")?; let query_id = resp .headers() @@ -80,99 +89,69 @@ impl ApiClient { path: &str, idempotency_key: Option<&str>, ) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)); + let mut req = self.routed_request(reqwest::Method::POST, path)?; if let Some(key) = idempotency_key { req = req.header("idempotency-key", key); } - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } let resp = req.send().context("failed to connect to server")?; handle_response(resp) } /// POST without a body to an endpoint that replies without content. pub fn post_no_content(&self, path: &str) -> Result<()> { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::POST, path)?; let resp = req.send().context("failed to connect to server")?; check_status(resp) } /// DELETE to an endpoint that replies without content. pub fn delete_no_content(&self, path: &str) -> Result<()> { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.delete(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::DELETE, path)?; let resp = req.send().context("failed to connect to server")?; check_status(resp) } pub fn patch(&self, path: &str, body: &Value) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.patch(&url)).json(body); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self + .routed_request(reqwest::Method::PATCH, path)? + .json(body); let resp = req.send().context("failed to connect to server")?; handle_response(resp) } pub fn delete(&self, path: &str) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.delete(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::DELETE, path)?; let resp = req.send().context("failed to connect to server")?; handle_response(resp) } pub fn get(&self, path: &str) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.get(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::GET, path)?; let resp = req.send().context("failed to connect to server")?; - handle_response(resp) + let value: Value = handle_response(resp)?; + self.remember_routing_metadata(path, &value); + serde_json::from_value(value).context("failed to parse server response") } /// POST with a pre-serialized JSON string body, gzip-compressed. pub fn post_compressed(&self, path: &str, body: &str) -> Result { - let url = format!("{}{}", self.base_url, path); let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); encoder .write_all(body.as_bytes()) .context("gzip compression failed")?; let compressed = encoder.finish().context("gzip finalization failed")?; - let mut req = self - .client - .post(&url) + let req = self + .routed_request(reqwest::Method::POST, path)? .header("content-type", "application/json") .header("content-encoding", "gzip") .body(compressed); - req = with_client_header(req); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } let resp = req.send().context("failed to connect to server")?; handle_response(resp) } /// POST that returns raw text (for queries that return ClickHouse results directly). pub fn post_raw(&self, path: &str, body: &Value) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.post(&url)).json(body); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::POST, path)?.json(body); let resp = req.send().context("failed to connect to server")?; let status = resp.status(); let text = resp.text().context("failed to read response body")?; @@ -184,11 +163,7 @@ impl ApiClient { /// GET that returns raw text. pub fn get_raw(&self, path: &str) -> Result { - let url = format!("{}{}", self.base_url, path); - let mut req = with_client_header(self.client.get(&url)); - if let Some(ref token) = self.token { - req = req.bearer_auth(token); - } + let req = self.routed_request(reqwest::Method::GET, path)?; let resp = req.send().context("failed to connect to server")?; let status = resp.status(); let text = resp.text().context("failed to read response body")?; @@ -205,7 +180,7 @@ fn with_client_header(req: RequestBuilder) -> RequestBuilder { .header(RAWTREE_CLIENT_VERSION_HEADER, RAWTREE_CLIENT_VERSION_VALUE) } -fn handle_response(resp: reqwest::blocking::Response) -> Result { +fn handle_response(resp: Response) -> Result { let status = resp.status(); let text = resp.text().context("failed to read response body")?; if !status.is_success() { @@ -214,7 +189,7 @@ fn handle_response(resp: reqwest::blocking::Response) -> Re serde_json::from_str(&text).context("failed to parse server response") } -fn check_status(resp: reqwest::blocking::Response) -> Result<()> { +fn check_status(resp: Response) -> Result<()> { let status = resp.status(); if status.is_success() { return Ok(()); diff --git a/src/client/routing.rs b/src/client/routing.rs new file mode 100644 index 0000000..89e3da4 --- /dev/null +++ b/src/client/routing.rs @@ -0,0 +1,266 @@ +use super::{handle_response, ApiClient}; +use anyhow::{bail, Context, Result}; +use reqwest::{blocking::RequestBuilder, Method, Url}; +use serde::Deserialize; +use std::{ + cell::RefCell, + collections::HashMap, + time::{Duration, Instant}, +}; + +#[derive(Clone, Deserialize)] +struct Organization { + organization_id: String, + name: String, +} +#[derive(Deserialize)] +struct Organizations { + organizations: Vec, +} +#[derive(Clone, Deserialize)] +struct Cluster { + id: String, + name: String, + #[serde(default)] + deployment_id: Option, + #[serde(default)] + cloud_region: Option, +} +#[derive(Deserialize)] +struct Clusters { + clusters: Vec, +} + +#[derive(Default)] +pub(super) struct RoutingCache { + organizations: RefCell>>, + clusters: RefCell)>>, +} + +impl ApiClient { + pub(super) fn remember_routing_metadata(&self, path: &str, value: &serde_json::Value) { + if path == "/v1/organizations" { + if let Ok(response) = serde_json::from_value::(value.clone()) { + *self.routing.organizations.borrow_mut() = Some(response.organizations); + } + } else if path.starts_with("/v1/organizations/") && path.ends_with("/clusters") { + if let Ok(response) = serde_json::from_value::(value.clone()) { + self.routing + .clusters + .borrow_mut() + .insert(path.to_owned(), (Instant::now(), response.clusters)); + } + } + } + + pub fn clusters_path(&self, organization: Option<&str>) -> Result { + self.clusters_path_before(organization, None) + } + + fn clusters_path_before( + &self, + organization: Option<&str>, + deadline: Option, + ) -> Result { + if organization.is_none() + && !self + .token + .as_deref() + .is_some_and(|token| token.starts_with("rt_")) + { + bail!("No organization selected. Pass --org or select an organization first."); + } + // IDs need no discovery. Names remain supported by CLI flags/configuration. + if let Some(id) = organization.filter(|id| is_uuid(id)) { + return Ok(format!("/v1/organizations/{id}/clusters")); + } + if self.routing.organizations.borrow().is_none() { + let response: Organizations = handle_response( + before_deadline( + self.unrouted_request(Method::GET, "/v1/organizations"), + deadline, + )? + .send()?, + )?; + *self.routing.organizations.borrow_mut() = Some(response.organizations); + } + let organizations = self.routing.organizations.borrow(); + let matched = organizations.as_ref().and_then(|items| { + items + .iter() + .find(|item| organization.is_none_or(|name| item.name == name)) + }); + let Some(matched) = matched else { + bail!("Organization not found. Pass --org with an organization you can access.") + }; + if !is_uuid(&matched.organization_id) { + bail!("Invalid organization ID returned by the API."); + } + Ok(format!( + "/v1/organizations/{}/clusters", + matched.organization_id + )) + } + + pub(super) fn routed_request(&self, method: Method, path: &str) -> Result { + self.request_before(method, path, None) + } + + pub(super) fn routed_request_with_timeout( + &self, + method: Method, + path: &str, + timeout: Duration, + ) -> Result { + let deadline = Instant::now() + .checked_add(timeout) + .context("invalid request timeout")?; + self.request_before(method, path, Some(deadline)) + } + + fn request_before( + &self, + method: Method, + path: &str, + deadline: Option, + ) -> Result { + let pair = self.deployment_for(path, deadline)?; + if method != Method::GET && path.starts_with("/v1/organizations") { + self.routing.clusters.borrow_mut().clear(); + *self.routing.organizations.borrow_mut() = None; + } + let mut request = self.unrouted_request(method, path); + if let Some((deployment, region)) = pair { + request = request + .header("X-RawTree-Deployment-Id", deployment) + .header("X-RawTree-Region", region); + } + before_deadline(request, deadline) + } + + fn deployment_for( + &self, + path: &str, + deadline: Option, + ) -> Result> { + if self + .token + .as_deref() + .is_none_or(|token| token.starts_with("rt_")) + { + return Ok(None); + } + let url = Url::parse(&format!("{}{}", self.base_url, path)).context("invalid API URL")?; + let Some((organization, cluster)) = scope(&url) else { + return Ok(None); + }; + let collection = self.clusters_path_before(Some(&organization), deadline)?; + let cached = self + .routing + .clusters + .borrow() + .get(&collection) + .filter(|(time, _)| time.elapsed() < Duration::from_secs(30)) + .map(|(_, clusters)| clusters.clone()); + let clusters = match cached { + Some(clusters) => clusters, + None => { + let response: Clusters = handle_response( + before_deadline(self.unrouted_request(Method::GET, &collection), deadline)? + .send()?, + )?; + self.routing + .clusters + .borrow_mut() + .insert(collection, (Instant::now(), response.clusters.clone())); + response.clusters + } + }; + let target = clusters.iter().find(|item| match &cluster { + ClusterSelector::Id(id) => item.id == *id, + ClusterSelector::Name(name) => item.name == *name, + }); + let pair = target.and_then(|item| { + item.deployment_id + .as_deref() + .zip(item.cloud_region.as_deref()) + }); + Ok(pair + .filter(|(deployment, region)| valid_routing(deployment, region)) + .map(|(deployment, region)| (deployment.to_owned(), region.to_owned()))) + } +} + +fn before_deadline(request: RequestBuilder, deadline: Option) -> Result { + match deadline { + Some(deadline) => { + let remaining = deadline + .checked_duration_since(Instant::now()) + .filter(|duration| !duration.is_zero()) + .context("request timed out during routing discovery")?; + Ok(request.timeout(remaining)) + } + None => Ok(request), + } +} + +enum ClusterSelector { + Id(String), + Name(String), +} + +fn scope(url: &Url) -> Option<(String, ClusterSelector)> { + let query: HashMap<_, _> = url.query_pairs().into_owned().collect(); + let parts: Vec<_> = url.path_segments()?.collect(); + let nested = parts.len() >= 4 + && parts[0] == "v1" + && parts[1] == "organizations" + && parts[3] == "clusters"; + let organization = if nested { + urlencoding::decode(parts[2]).ok()?.into_owned() + } else { + query + .get("organization_id") + .or_else(|| query.get("organization"))? + .clone() + }; + let cluster = + if nested && parts.len() >= 5 && !matches!(parts[4], "metrics" | "verify-s3-access") { + ClusterSelector::Id(urlencoding::decode(parts[4]).ok()?.into_owned()) + } else if let Some(id) = query.get("cluster_id") { + ClusterSelector::Id(id.clone()) + } else { + ClusterSelector::Name(query.get("cluster")?.clone()) + }; + Some((organization, cluster)) +} + +fn is_uuid(value: &str) -> bool { + value.len() == 36 + && value.bytes().enumerate().all(|(index, byte)| { + if matches!(index, 8 | 13 | 18 | 23) { + byte == b'-' + } else { + byte.is_ascii_hexdigit() + } + }) +} + +// Match the backend/worker syntax; do not maintain a catalog of AWS regions. +fn valid_routing(deployment: &str, region: &str) -> bool { + let mut parts = region.rsplit('-'); + let number = parts.next().unwrap_or_default(); + let names: Vec<_> = parts.collect(); + deployment.len() == 8 + && deployment + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + && region.len() <= 63 + && !number.starts_with('0') + && !number.is_empty() + && number.bytes().all(|byte| byte.is_ascii_digit()) + && names.len() >= 2 + && names + .iter() + .all(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_lowercase())) +} diff --git a/src/client/routing_tests.rs b/src/client/routing_tests.rs new file mode 100644 index 0000000..3ff6d03 --- /dev/null +++ b/src/client/routing_tests.rs @@ -0,0 +1,277 @@ +use super::ApiClient; +use serde_json::{json, Value}; +use std::collections::HashMap; +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::TcpListener; +use std::thread; +use std::time::{Duration, Instant}; + +const ORG: &str = "00000000-0000-4000-8000-000000000001"; +struct Request { + path: String, + headers: HashMap, + body: Vec, +} + +fn server(responses: Vec<(&'static str, Value)>) -> (String, thread::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let handle = thread::spawn(move || { + let mut requests = Vec::new(); + for (status, payload) in responses { + let deadline = Instant::now() + Duration::from_secs(5); + let mut socket = loop { + match listener.accept() { + Ok((socket, _)) => break socket, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + assert!(Instant::now() < deadline, "missing HTTP request"); + thread::sleep(Duration::from_millis(5)); + } + Err(error) => panic!("accept: {error}"), + } + }; + socket.set_nonblocking(false).unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut reader = BufReader::new(socket.try_clone().unwrap()); + let mut first = String::new(); + reader.read_line(&mut first).unwrap(); + let mut headers = HashMap::new(); + loop { + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + if line == "\r\n" { + break; + } + let (key, value) = line.split_once(':').unwrap(); + headers.insert(key.to_ascii_lowercase(), value.trim().to_owned()); + } + let length = headers + .get("content-length") + .map(|value| value.parse().unwrap()) + .unwrap_or(0); + let mut body = vec![0; length]; + reader.read_exact(&mut body).unwrap(); + let response = payload.to_string(); + write!(socket, "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response}", response.len()).unwrap(); + requests.push(Request { + path: first.trim().to_owned(), + headers, + body, + }); + } + requests + }); + (base, handle) +} + +fn metadata(pair: Value) -> Vec<(&'static str, Value)> { + vec![ + ( + "200 OK", + json!({"organizations": [{"organization_id": ORG, "name": "acme"}]}), + ), + ( + "200 OK", + json!({"clusters": [{"id": "cluster-id", "name": "production", "deployment_id": pair["deployment_id"], "cloud_region": pair["cloud_region"]}]}), + ), + ] +} + +#[test] +fn session_requests_resolve_scope_and_reuse_metadata_for_queries_and_gzip() { + let mut responses = metadata(json!({"deployment_id": "12abcdef", "cloud_region": "eu-west-1"})); + responses.extend([ + ("200 OK", json!({"ok":true})), + ("200 OK", json!({"ok":true})), + ]); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let path = "/v1/query?organization=acme&cluster=production"; + let _: Value = client.post(path, &json!({"query": "SELECT 1"})).unwrap(); + let _: Value = client + .post_compressed( + "/v1/tables/events?organization=acme&cluster=production", + "{\"value\":1}", + ) + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests[0].path, "GET /v1/organizations HTTP/1.1"); + assert_eq!( + requests[1].path, + format!("GET /v1/organizations/{ORG}/clusters HTTP/1.1") + ); + assert!(!requests[0].headers.contains_key("x-rawtree-region")); + for request in &requests[2..] { + assert_eq!(request.headers["x-rawtree-deployment-id"], "12abcdef"); + assert_eq!(request.headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(request.headers["authorization"], "Bearer session.fixture"); + assert_eq!(request.headers["x-rawtree-client"], "cli"); + } + assert_eq!(requests[3].headers["content-encoding"], "gzip"); + let mut decoded = String::new(); + flate2::read::GzDecoder::new(requests[3].body.as_slice()) + .read_to_string(&mut decoded) + .unwrap(); + assert_eq!(decoded, "{\"value\":1}"); +} + +#[test] +fn absent_partial_and_invalid_metadata_keep_default_routing() { + for pair in [ + json!({}), + json!({"deployment_id": "12abcdef"}), + json!({"deployment_id": "12abcdef99", "cloud_region": "eu-west-1"}), + json!({"deployment_id": "12abcdef", "cloud_region": "bad.invalid"}), + ] { + let mut responses = metadata(pair); + responses.push(("200 OK", json!({}))); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("oauth.fixture".to_string())); + let _: Value = client + .get("/v1/databases?organization=acme&cluster=production") + .unwrap(); + let requests = server.join().unwrap(); + assert!(!requests[2].headers.contains_key("x-rawtree-region")); + assert!(!requests[2].headers.contains_key("x-rawtree-deployment-id")); + } +} + +#[test] +fn api_keys_skip_discovery_and_routing_headers() { + let (base, server) = server(vec![("200 OK", json!({}))]); + let client = ApiClient::new(base, Some("rt_fixture".to_string())); + client + .post_raw( + "/v1/query?organization=acme&cluster=production", + &json!({"query":"SELECT 1"}), + ) + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests.len(), 1); + assert!(!requests[0].headers.contains_key("x-rawtree-region")); + assert_eq!(requests[0].headers["authorization"], "Bearer rt_fixture"); +} + +#[test] +fn cluster_id_actions_use_organization_id_and_serving_headers() { + let mut responses = metadata(json!({"deployment_id": "12abcdef", "cloud_region": "us-east-1"})); + responses.remove(0); // Organization ID is already known: no name resolution. + responses.push(("200 OK", json!({}))); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let collection = client.clusters_path(Some(ORG)).unwrap(); + let _: Value = client + .post_empty(&format!("{collection}/cluster-id/stop")) + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!( + requests[1].path, + format!("POST {collection}/cluster-id/stop HTTP/1.1") + ); + assert_eq!(requests[1].headers["x-rawtree-deployment-id"], "12abcdef"); +} + +#[test] +fn discovery_failure_does_not_execute_the_operation() { + let mut responses = metadata(json!({})); + responses[1] = ( + "503 Service Unavailable", + json!({"message":"discovery unavailable"}), + ); + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + assert!(client + .get::("/v1/databases?organization=acme&cluster=production") + .is_err()); + assert_eq!(server.join().unwrap().len(), 2); +} + +#[test] +fn cluster_creation_invalidates_previously_discovered_empty_collection() { + let responses = vec![ + ("200 OK", json!({"clusters": []})), + ("200 OK", json!({"id": "new-id"})), + ( + "200 OK", + json!({"clusters": [{"id":"new-id", "name":"new", "deployment_id":"12abcdef", "cloud_region":"eu-west-1"}]}), + ), + ("200 OK", json!({"data":[]})), + ]; + let (base, server) = server(responses); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let collection = client.clusters_path(Some(ORG)).unwrap(); + let _: Value = client.get(&collection).unwrap(); + let _: Value = client.post(&collection, &json!({"name":"new"})).unwrap(); + let _: Value = client + .get(&format!( + "/v1/databases?organization_id={ORG}&cluster_id=new-id" + )) + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests[2].path, format!("GET {collection} HTTP/1.1")); + assert_eq!(requests[3].headers["x-rawtree-region"], "eu-west-1"); +} + +#[test] +fn explicit_timeout_includes_routing_discovery() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let server = thread::spawn(move || { + let (socket, _) = listener.accept().unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut reader = BufReader::new(socket); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert_eq!(line.trim(), "GET /v1/organizations HTTP/1.1"); + thread::sleep(Duration::from_millis(400)); + }); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let started = Instant::now(); + let result = client.post_with_timeout::( + "/v1/query?organization=acme&cluster=production", + &json!({"query":"SELECT 1"}), + Duration::from_millis(50), + ); + assert!(result.is_err()); + assert!(started.elapsed() < Duration::from_millis(300)); + server.join().unwrap(); +} + +#[test] +fn explicit_id_and_name_selectors_cannot_collide() { + let (base, server) = server(vec![ + ( + "200 OK", + json!({"clusters":[ + {"id":"other-id", "name":"target-id", "deployment_id":"abcdef12", "cloud_region":"ap-south-1"}, + {"id":"target-id", "name":"production", "deployment_id":"12abcdef", "cloud_region":"eu-west-1"} + ]}), + ), + ("200 OK", json!({})), + ("200 OK", json!({})), + ("200 OK", json!({})), + ]); + let client = ApiClient::new(base, Some("session.fixture".to_string())); + let _: Value = client + .get(&format!("/v1/organizations/{ORG}/clusters/target-id")) + .unwrap(); + let _: Value = client + .get(&format!( + "/v1/metrics?organization_id={ORG}&cluster_id=target-id" + )) + .unwrap(); + let _: Value = client + .get(&format!( + "/v1/databases?organization_id={ORG}&cluster=target-id" + )) + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests[1].headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(requests[2].headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(requests[3].headers["x-rawtree-region"], "ap-south-1"); +} diff --git a/src/commands/auth/selection.rs b/src/commands/auth/selection.rs index d025342..96724ba 100644 --- a/src/commands/auth/selection.rs +++ b/src/commands/auth/selection.rs @@ -265,7 +265,7 @@ fn list_clusters_for_organization( client: &ApiClient, organization_name: &str, ) -> Result> { - let path = org::scoped_path("/v1/clusters", Some(organization_name), None); + let path = client.clusters_path(Some(organization_name))?; let resp: ListClustersResponse = client.get(&path)?; Ok(resp.clusters.into_iter().map(|item| item.name).collect()) } diff --git a/src/commands/cluster.rs b/src/commands/cluster.rs index 32798ab..61dffbe 100644 --- a/src/commands/cluster.rs +++ b/src/commands/cluster.rs @@ -135,7 +135,7 @@ pub fn create( options.idle_timeout_minutes, ); - let value: Value = client.post(&clusters_collection_path(options.organization), &body)?; + let value: Value = client.post(&client.clusters_path(options.organization)?, &body)?; let created: ClusterItem = serde_json::from_value(value.clone()).context("invalid cluster response from server")?; @@ -190,7 +190,7 @@ pub fn update( } let value: Value = client.patch( - &cluster_path(&cluster.id, None, organization), + &cluster_path(client, &cluster.id, None, organization)?, &Value::Object(body), )?; let updated: ClusterItem = @@ -363,7 +363,7 @@ pub fn delete( ) -> Result<()> { let (_, resp) = load_dedicated_clusters(client, organization)?; let cluster = resolve_cluster(&resp.clusters, name_or_id)?.clone(); - let path = cluster_path(&cluster.id, None, organization); + let path = cluster_path(client, &cluster.id, None, organization)?; let result: DeleteClusterResponse = client.delete(&path)?; let value = delete_output(&cluster, result.deleted); if result.deleted { @@ -411,7 +411,7 @@ fn load_dedicated_clusters( client: &ApiClient, organization: Option<&str>, ) -> Result<(Value, ListClustersResponse)> { - let path = clusters_collection_path(organization); + let path = client.clusters_path(organization)?; let value: Value = client.get(&path)?; let resp = serde_json::from_value(value.clone()).context("invalid clusters response from server")?; @@ -420,7 +420,7 @@ fn load_dedicated_clusters( fn load_cluster_sizes(client: &ApiClient) -> Result<(Value, ClusterSizesResponse)> { let value: Value = client - .get("/v1/clusters/sizes") + .get("/v1/cluster-sizes") .context("invalid cluster sizes response from server")?; let response: ClusterSizesResponse = serde_json::from_value(value.clone()) .context("invalid cluster sizes response from server")?; @@ -509,7 +509,12 @@ fn request_lifecycle( ) -> Result<()> { let (_, resp) = load_dedicated_clusters(client, organization)?; let cluster = resolve_cluster(&resp.clusters, name_or_id)?; - let path = cluster_path(&cluster.id, Some(action.path_segment()), organization); + let path = cluster_path( + client, + &cluster.id, + Some(action.path_segment()), + organization, + )?; let value: Value = client.post_empty(&path)?; let updated: ClusterItem = serde_json::from_value(value.clone()).context("invalid cluster response from server")?; @@ -529,24 +534,22 @@ fn request_lifecycle( Ok(()) } -fn clusters_collection_path(organization: Option<&str>) -> String { - match organization { - Some(name) => format!("/v1/clusters?organization={}", urlencoding::encode(name)), - None => "/v1/clusters".to_string(), - } -} - -fn cluster_path(cluster_id: &str, action: Option<&str>, organization: Option<&str>) -> String { - let mut path = format!("/v1/clusters/{}", urlencoding::encode(cluster_id)); +fn cluster_path( + client: &ApiClient, + cluster_id: &str, + action: Option<&str>, + organization: Option<&str>, +) -> Result { + let mut path = format!( + "{}/{}", + client.clusters_path(organization)?, + urlencoding::encode(cluster_id) + ); if let Some(action) = action { path.push('/'); path.push_str(action); } - if let Some(name) = organization { - path.push_str("?organization="); - path.push_str(&urlencoding::encode(name)); - } - path + Ok(path) } fn format_phase(phase: &str) -> String { @@ -603,39 +606,13 @@ mod tests { use serde_json::json; use super::{ - cluster_path, cluster_size_json, clusters_collection_path, create_request_body, - default_cluster_after_delete, delete_output, empty_list_message, format_created_at, - format_idle_timeout, format_phase, format_size_per_replica, renamed_default_cluster, - resolve_cluster, resolve_cluster_size, ClusterItem, ClusterResources, ClusterSizeItem, - ClusterStatus, ListClustersResponse, + cluster_size_json, create_request_body, default_cluster_after_delete, delete_output, + empty_list_message, format_created_at, format_idle_timeout, format_phase, + format_size_per_replica, renamed_default_cluster, resolve_cluster, resolve_cluster_size, + ClusterItem, ClusterResources, ClusterSizeItem, ClusterStatus, ListClustersResponse, }; use crate::cli::ClusterSizeArg; - #[test] - fn collection_path_encodes_organization() { - assert_eq!( - clusters_collection_path(Some("team alpha")), - "/v1/clusters?organization=team%20alpha" - ); - assert_eq!(clusters_collection_path(None), "/v1/clusters"); - } - - #[test] - fn lifecycle_paths_encode_cluster_and_organization() { - assert_eq!( - cluster_path("cluster/id", Some("stop"), Some("team alpha")), - "/v1/clusters/cluster%2Fid/stop?organization=team%20alpha" - ); - assert_eq!( - cluster_path("cluster-id", Some("resume"), None), - "/v1/clusters/cluster-id/resume" - ); - assert_eq!( - cluster_path("cluster-id", None, Some("team alpha")), - "/v1/clusters/cluster-id?organization=team%20alpha" - ); - } - #[test] fn cluster_size_catalog_resolves_cpu_and_memory_pairs() { let sizes = vec![ diff --git a/src/commands/insert.rs b/src/commands/insert.rs index 2f9bcde..6ff7b08 100644 --- a/src/commands/insert.rs +++ b/src/commands/insert.rs @@ -5,7 +5,6 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::mpsc; use std::sync::Arc; use std::thread; -use std::time::Duration; use anyhow::{bail, Context, Result}; use indicatif::{ProgressBar, ProgressStyle}; diff --git a/tests/api_contracts.rs b/tests/api_contracts.rs index 6d49495..9ffc1f3 100644 --- a/tests/api_contracts.rs +++ b/tests/api_contracts.rs @@ -36,11 +36,11 @@ fn database_create_reads_nested_response_and_saves_requested_context() { "POST", "/v1/databases?organization=team%20alpha&cluster=prod%2Feu", response.clone(), - json!({"token": "fixture-session", "default_organization": "old-org", "database": "old-db"}), + json!({"token": "rt_fixture", "default_organization": "old-org", "database": "old-db"}), ); assert_eq!(result, response); assert_eq!(body, json!({"name": "analytics"})); - assert_eq!(saved["token"], "fixture-session"); + assert_eq!(saved["token"], "rt_fixture"); assert_eq!(saved["default_organization"], "team alpha"); assert_eq!(saved["cluster"], "prod/eu"); assert_eq!(saved["database"], "analytics"); @@ -48,7 +48,7 @@ fn database_create_reads_nested_response_and_saves_requested_context() { #[test] fn database_create_failure_does_not_overwrite_config() { - let config = json!({"token": "fixture-session", "default_organization": "team", "cluster": "prod", "database": "old-db"}); + let config = json!({"token": "rt_fixture", "default_organization": "team", "cluster": "prod", "database": "old-db"}); let (output, saved, _) = common::run_cli( &[( "POST /v1/databases?organization=team&cluster=prod".to_string(), diff --git a/tests/browser_login.rs b/tests/browser_login.rs index dbba85f..60e12ea 100644 --- a/tests/browser_login.rs +++ b/tests/browser_login.rs @@ -45,15 +45,15 @@ fn organizations(names: &[&str]) -> Response { ( "GET /v1/organizations".into(), "200 OK", - json!({"organizations": names.iter().map(|name| json!({"name": name, "role": "owner"})).collect::>()}), + json!({"organizations": names.iter().map(|name| json!({"organization_id": "00000000-0000-4000-8000-000000000001", "name": name, "role": "owner"})).collect::>()}), ) } fn clusters(names: &[&str]) -> Response { ( - "GET /v1/clusters?organization=team".into(), + "GET /v1/organizations/00000000-0000-4000-8000-000000000001/clusters".into(), "200 OK", - json!({"clusters": names.iter().map(|name| json!({"name": name})).collect::>()}), + json!({"clusters": names.iter().map(|name| json!({"id": name, "name": name})).collect::>()}), ) } diff --git a/tests/format.rs b/tests/format.rs index d5cb54c..1ca77ee 100644 --- a/tests/format.rs +++ b/tests/format.rs @@ -5,7 +5,7 @@ use serde_json::{json, Value}; const PATH: &str = "POST /v1/query-format?organization=team&cluster=prod"; fn config() -> Value { - json!({"token": "fixture-token", "default_organization": "team", "cluster": "prod"}) + json!({"token": "rt_fixture", "default_organization": "team", "cluster": "prod"}) } #[test] diff --git a/tests/sql_input.rs b/tests/sql_input.rs index 0dbe881..7a660d7 100644 --- a/tests/sql_input.rs +++ b/tests/sql_input.rs @@ -3,7 +3,7 @@ mod common; use serde_json::{json, Value}; fn config() -> Value { - json!({"token": "fixture-token", "database": "analytics", "default_organization": "team", "cluster": "prod"}) + json!({"token": "rt_fixture", "database": "analytics", "default_organization": "team", "cluster": "prod"}) } fn response(command: &str) -> (String, &'static str, Value) { From e37db943bafc4ac92d0e6d227881391a7575ea78 Mon Sep 17 00:00:00 2001 From: Rafa Moreno Date: Sat, 17 Oct 2026 20:59:07 +0200 Subject: [PATCH 2/3] Bump CLI to 0.7.0 for organization-scoped clusters --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fb37b9d..ef825c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1502,7 +1502,7 @@ dependencies = [ [[package]] name = "rawtree-cli" -version = "0.6.15" +version = "0.7.0" dependencies = [ "anyhow", "axoupdater", diff --git a/Cargo.toml b/Cargo.toml index db300d4..f56d1b5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rawtree-cli" -version = "0.6.15" +version = "0.7.0" edition = "2021" repository = "https://github.com/rawtreedb/rawtree-cli" From 003e91cfb800b892cf84c99c6493c7efe7f058d2 Mon Sep 17 00:00:00 2001 From: Rafa Moreno Date: Sat, 17 Oct 2026 21:22:55 +0200 Subject: [PATCH 3/3] Restore legacy cluster paths and retain deployment routing --- src/client/routing.rs | 168 ++++++++++++++++----------------- src/client/routing_tests.rs | 59 +++++------- src/commands/auth/selection.rs | 2 +- src/commands/cluster.rs | 77 +++++++++------ src/commands/insert.rs | 1 + tests/browser_login.rs | 6 +- 6 files changed, 160 insertions(+), 153 deletions(-) diff --git a/src/client/routing.rs b/src/client/routing.rs index 89e3da4..bef8ed5 100644 --- a/src/client/routing.rs +++ b/src/client/routing.rs @@ -1,5 +1,5 @@ use super::{handle_response, ApiClient}; -use anyhow::{bail, Context, Result}; +use anyhow::{Context, Result}; use reqwest::{blocking::RequestBuilder, Method, Url}; use serde::Deserialize; use std::{ @@ -33,73 +33,58 @@ struct Clusters { #[derive(Default)] pub(super) struct RoutingCache { - organizations: RefCell>>, + organizations: RefCell)>>, clusters: RefCell)>>, } impl ApiClient { pub(super) fn remember_routing_metadata(&self, path: &str, value: &serde_json::Value) { - if path == "/v1/organizations" { + let Ok(url) = Url::parse(&format!("{}{}", self.base_url, path)) else { + return; + }; + if url.path() == "/v1/organizations" { if let Ok(response) = serde_json::from_value::(value.clone()) { - *self.routing.organizations.borrow_mut() = Some(response.organizations); + *self.routing.organizations.borrow_mut() = + Some((Instant::now(), response.organizations)); } - } else if path.starts_with("/v1/organizations/") && path.ends_with("/clusters") { - if let Ok(response) = serde_json::from_value::(value.clone()) { - self.routing - .clusters - .borrow_mut() - .insert(path.to_owned(), (Instant::now(), response.clusters)); + } else if url.path() == "/v1/clusters" { + if let Some((_, organization)) = + url.query_pairs().find(|(key, _)| key == "organization") + { + if let Ok(response) = serde_json::from_value::(value.clone()) { + self.routing.clusters.borrow_mut().insert( + clusters_path(&organization), + (Instant::now(), response.clusters), + ); + } } } } - pub fn clusters_path(&self, organization: Option<&str>) -> Result { - self.clusters_path_before(organization, None) - } - - fn clusters_path_before( - &self, - organization: Option<&str>, - deadline: Option, - ) -> Result { - if organization.is_none() - && !self - .token - .as_deref() - .is_some_and(|token| token.starts_with("rt_")) - { - bail!("No organization selected. Pass --org or select an organization first."); - } - // IDs need no discovery. Names remain supported by CLI flags/configuration. - if let Some(id) = organization.filter(|id| is_uuid(id)) { - return Ok(format!("/v1/organizations/{id}/clusters")); - } - if self.routing.organizations.borrow().is_none() { - let response: Organizations = handle_response( - before_deadline( - self.unrouted_request(Method::GET, "/v1/organizations"), - deadline, - )? - .send()?, - )?; - *self.routing.organizations.borrow_mut() = Some(response.organizations); - } - let organizations = self.routing.organizations.borrow(); - let matched = organizations.as_ref().and_then(|items| { - items - .iter() - .find(|item| organization.is_none_or(|name| item.name == name)) - }); - let Some(matched) = matched else { - bail!("Organization not found. Pass --org with an organization you can access.") - }; - if !is_uuid(&matched.organization_id) { - bail!("Invalid organization ID returned by the API."); + fn organization_name(&self, id: &str, deadline: Option) -> Result { + if let Some((time, organizations)) = self.routing.organizations.borrow().as_ref() { + if time.elapsed() < Duration::from_secs(30) { + if let Some(organization) = + organizations.iter().find(|item| item.organization_id == id) + { + return Ok(organization.name.clone()); + } + } } - Ok(format!( - "/v1/organizations/{}/clusters", - matched.organization_id - )) + let response: Organizations = handle_response( + before_deadline( + self.unrouted_request(Method::GET, "/v1/organizations"), + deadline, + )? + .send()?, + )?; + let name = response + .organizations + .iter() + .find(|item| item.organization_id == id) + .map(|item| item.name.clone()); + *self.routing.organizations.borrow_mut() = Some((Instant::now(), response.organizations)); + name.context("Organization not found.") } pub(super) fn routed_request(&self, method: Method, path: &str) -> Result { @@ -125,7 +110,9 @@ impl ApiClient { deadline: Option, ) -> Result { let pair = self.deployment_for(path, deadline)?; - if method != Method::GET && path.starts_with("/v1/organizations") { + if method != Method::GET + && (path.starts_with("/v1/organizations") || path.starts_with("/v1/clusters")) + { self.routing.clusters.borrow_mut().clear(); *self.routing.organizations.borrow_mut() = None; } @@ -154,7 +141,11 @@ impl ApiClient { let Some((organization, cluster)) = scope(&url) else { return Ok(None); }; - let collection = self.clusters_path_before(Some(&organization), deadline)?; + let organization = match organization { + OrganizationSelector::Id(id) => self.organization_name(&id, deadline)?, + OrganizationSelector::Name(name) => name, + }; + let collection = clusters_path(&organization); let cached = self .routing .clusters @@ -209,43 +200,42 @@ enum ClusterSelector { Name(String), } -fn scope(url: &Url) -> Option<(String, ClusterSelector)> { +enum OrganizationSelector { + Id(String), + Name(String), +} + +fn clusters_path(organization: &str) -> String { + format!( + "/v1/clusters?organization={}", + urlencoding::encode(organization) + ) +} + +fn scope(url: &Url) -> Option<(OrganizationSelector, ClusterSelector)> { let query: HashMap<_, _> = url.query_pairs().into_owned().collect(); - let parts: Vec<_> = url.path_segments()?.collect(); - let nested = parts.len() >= 4 - && parts[0] == "v1" - && parts[1] == "organizations" - && parts[3] == "clusters"; - let organization = if nested { - urlencoding::decode(parts[2]).ok()?.into_owned() + let organization = if let Some(id) = query.get("organization_id") { + OrganizationSelector::Id(id.clone()) } else { - query - .get("organization_id") - .or_else(|| query.get("organization"))? - .clone() + OrganizationSelector::Name(query.get("organization")?.clone()) + }; + let path_cluster = url + .path() + .strip_prefix("/v1/clusters/") + .and_then(|path| path.split('/').next()) + .filter(|part| { + !part.is_empty() && !matches!(*part, "sizes" | "metrics" | "verify-s3-access") + }); + let cluster = if let Some(id) = path_cluster { + ClusterSelector::Id(urlencoding::decode(id).ok()?.into_owned()) + } else if let Some(id) = query.get("cluster_id") { + ClusterSelector::Id(id.clone()) + } else { + ClusterSelector::Name(query.get("cluster")?.clone()) }; - let cluster = - if nested && parts.len() >= 5 && !matches!(parts[4], "metrics" | "verify-s3-access") { - ClusterSelector::Id(urlencoding::decode(parts[4]).ok()?.into_owned()) - } else if let Some(id) = query.get("cluster_id") { - ClusterSelector::Id(id.clone()) - } else { - ClusterSelector::Name(query.get("cluster")?.clone()) - }; Some((organization, cluster)) } -fn is_uuid(value: &str) -> bool { - value.len() == 36 - && value.bytes().enumerate().all(|(index, byte)| { - if matches!(index, 8 | 13 | 18 | 23) { - byte == b'-' - } else { - byte.is_ascii_hexdigit() - } - }) -} - // Match the backend/worker syntax; do not maintain a catalog of AWS regions. fn valid_routing(deployment: &str, region: &str) -> bool { let mut parts = region.rsplit('-'); diff --git a/src/client/routing_tests.rs b/src/client/routing_tests.rs index 3ff6d03..33d4b61 100644 --- a/src/client/routing_tests.rs +++ b/src/client/routing_tests.rs @@ -68,16 +68,10 @@ fn server(responses: Vec<(&'static str, Value)>) -> (String, thread::JoinHandle< } fn metadata(pair: Value) -> Vec<(&'static str, Value)> { - vec![ - ( - "200 OK", - json!({"organizations": [{"organization_id": ORG, "name": "acme"}]}), - ), - ( - "200 OK", - json!({"clusters": [{"id": "cluster-id", "name": "production", "deployment_id": pair["deployment_id"], "cloud_region": pair["cloud_region"]}]}), - ), - ] + vec![( + "200 OK", + json!({"clusters": [{"id": "cluster-id", "name": "production", "deployment_id": pair["deployment_id"], "cloud_region": pair["cloud_region"]}]}), + )] } #[test] @@ -98,21 +92,20 @@ fn session_requests_resolve_scope_and_reuse_metadata_for_queries_and_gzip() { ) .unwrap(); let requests = server.join().unwrap(); - assert_eq!(requests[0].path, "GET /v1/organizations HTTP/1.1"); assert_eq!( - requests[1].path, - format!("GET /v1/organizations/{ORG}/clusters HTTP/1.1") + requests[0].path, + "GET /v1/clusters?organization=acme HTTP/1.1" ); assert!(!requests[0].headers.contains_key("x-rawtree-region")); - for request in &requests[2..] { + for request in &requests[1..] { assert_eq!(request.headers["x-rawtree-deployment-id"], "12abcdef"); assert_eq!(request.headers["x-rawtree-region"], "eu-west-1"); assert_eq!(request.headers["authorization"], "Bearer session.fixture"); assert_eq!(request.headers["x-rawtree-client"], "cli"); } - assert_eq!(requests[3].headers["content-encoding"], "gzip"); + assert_eq!(requests[2].headers["content-encoding"], "gzip"); let mut decoded = String::new(); - flate2::read::GzDecoder::new(requests[3].body.as_slice()) + flate2::read::GzDecoder::new(requests[2].body.as_slice()) .read_to_string(&mut decoded) .unwrap(); assert_eq!(decoded, "{\"value\":1}"); @@ -134,8 +127,8 @@ fn absent_partial_and_invalid_metadata_keep_default_routing() { .get("/v1/databases?organization=acme&cluster=production") .unwrap(); let requests = server.join().unwrap(); - assert!(!requests[2].headers.contains_key("x-rawtree-region")); - assert!(!requests[2].headers.contains_key("x-rawtree-deployment-id")); + assert!(!requests[1].headers.contains_key("x-rawtree-region")); + assert!(!requests[1].headers.contains_key("x-rawtree-deployment-id")); } } @@ -156,20 +149,18 @@ fn api_keys_skip_discovery_and_routing_headers() { } #[test] -fn cluster_id_actions_use_organization_id_and_serving_headers() { +fn cluster_id_actions_preserve_organization_query_and_serving_headers() { let mut responses = metadata(json!({"deployment_id": "12abcdef", "cloud_region": "us-east-1"})); - responses.remove(0); // Organization ID is already known: no name resolution. responses.push(("200 OK", json!({}))); let (base, server) = server(responses); let client = ApiClient::new(base, Some("session.fixture".to_string())); - let collection = client.clusters_path(Some(ORG)).unwrap(); let _: Value = client - .post_empty(&format!("{collection}/cluster-id/stop")) + .post_empty("/v1/clusters/cluster-id/stop?organization=acme") .unwrap(); let requests = server.join().unwrap(); assert_eq!( requests[1].path, - format!("POST {collection}/cluster-id/stop HTTP/1.1") + "POST /v1/clusters/cluster-id/stop?organization=acme HTTP/1.1" ); assert_eq!(requests[1].headers["x-rawtree-deployment-id"], "12abcdef"); } @@ -177,7 +168,7 @@ fn cluster_id_actions_use_organization_id_and_serving_headers() { #[test] fn discovery_failure_does_not_execute_the_operation() { let mut responses = metadata(json!({})); - responses[1] = ( + responses[0] = ( "503 Service Unavailable", json!({"message":"discovery unavailable"}), ); @@ -186,7 +177,7 @@ fn discovery_failure_does_not_execute_the_operation() { assert!(client .get::("/v1/databases?organization=acme&cluster=production") .is_err()); - assert_eq!(server.join().unwrap().len(), 2); + assert_eq!(server.join().unwrap().len(), 1); } #[test] @@ -202,13 +193,11 @@ fn cluster_creation_invalidates_previously_discovered_empty_collection() { ]; let (base, server) = server(responses); let client = ApiClient::new(base, Some("session.fixture".to_string())); - let collection = client.clusters_path(Some(ORG)).unwrap(); + let collection = "/v1/clusters?organization=acme"; let _: Value = client.get(&collection).unwrap(); let _: Value = client.post(&collection, &json!({"name":"new"})).unwrap(); let _: Value = client - .get(&format!( - "/v1/databases?organization_id={ORG}&cluster_id=new-id" - )) + .get("/v1/databases?organization=acme&cluster_id=new-id") .unwrap(); let requests = server.join().unwrap(); assert_eq!(requests[2].path, format!("GET {collection} HTTP/1.1")); @@ -227,7 +216,7 @@ fn explicit_timeout_includes_routing_discovery() { let mut reader = BufReader::new(socket); let mut line = String::new(); reader.read_line(&mut line).unwrap(); - assert_eq!(line.trim(), "GET /v1/organizations HTTP/1.1"); + assert_eq!(line.trim(), "GET /v1/clusters?organization=acme HTTP/1.1"); thread::sleep(Duration::from_millis(400)); }); let client = ApiClient::new(base, Some("session.fixture".to_string())); @@ -253,12 +242,16 @@ fn explicit_id_and_name_selectors_cannot_collide() { ]}), ), ("200 OK", json!({})), + ( + "200 OK", + json!({"organizations": [{"organization_id": ORG, "name": "acme"}]}), + ), ("200 OK", json!({})), ("200 OK", json!({})), ]); let client = ApiClient::new(base, Some("session.fixture".to_string())); let _: Value = client - .get(&format!("/v1/organizations/{ORG}/clusters/target-id")) + .get("/v1/clusters/target-id?organization=acme") .unwrap(); let _: Value = client .get(&format!( @@ -272,6 +265,6 @@ fn explicit_id_and_name_selectors_cannot_collide() { .unwrap(); let requests = server.join().unwrap(); assert_eq!(requests[1].headers["x-rawtree-region"], "eu-west-1"); - assert_eq!(requests[2].headers["x-rawtree-region"], "eu-west-1"); - assert_eq!(requests[3].headers["x-rawtree-region"], "ap-south-1"); + assert_eq!(requests[3].headers["x-rawtree-region"], "eu-west-1"); + assert_eq!(requests[4].headers["x-rawtree-region"], "ap-south-1"); } diff --git a/src/commands/auth/selection.rs b/src/commands/auth/selection.rs index 96724ba..d025342 100644 --- a/src/commands/auth/selection.rs +++ b/src/commands/auth/selection.rs @@ -265,7 +265,7 @@ fn list_clusters_for_organization( client: &ApiClient, organization_name: &str, ) -> Result> { - let path = client.clusters_path(Some(organization_name))?; + let path = org::scoped_path("/v1/clusters", Some(organization_name), None); let resp: ListClustersResponse = client.get(&path)?; Ok(resp.clusters.into_iter().map(|item| item.name).collect()) } diff --git a/src/commands/cluster.rs b/src/commands/cluster.rs index 61dffbe..32798ab 100644 --- a/src/commands/cluster.rs +++ b/src/commands/cluster.rs @@ -135,7 +135,7 @@ pub fn create( options.idle_timeout_minutes, ); - let value: Value = client.post(&client.clusters_path(options.organization)?, &body)?; + let value: Value = client.post(&clusters_collection_path(options.organization), &body)?; let created: ClusterItem = serde_json::from_value(value.clone()).context("invalid cluster response from server")?; @@ -190,7 +190,7 @@ pub fn update( } let value: Value = client.patch( - &cluster_path(client, &cluster.id, None, organization)?, + &cluster_path(&cluster.id, None, organization), &Value::Object(body), )?; let updated: ClusterItem = @@ -363,7 +363,7 @@ pub fn delete( ) -> Result<()> { let (_, resp) = load_dedicated_clusters(client, organization)?; let cluster = resolve_cluster(&resp.clusters, name_or_id)?.clone(); - let path = cluster_path(client, &cluster.id, None, organization)?; + let path = cluster_path(&cluster.id, None, organization); let result: DeleteClusterResponse = client.delete(&path)?; let value = delete_output(&cluster, result.deleted); if result.deleted { @@ -411,7 +411,7 @@ fn load_dedicated_clusters( client: &ApiClient, organization: Option<&str>, ) -> Result<(Value, ListClustersResponse)> { - let path = client.clusters_path(organization)?; + let path = clusters_collection_path(organization); let value: Value = client.get(&path)?; let resp = serde_json::from_value(value.clone()).context("invalid clusters response from server")?; @@ -420,7 +420,7 @@ fn load_dedicated_clusters( fn load_cluster_sizes(client: &ApiClient) -> Result<(Value, ClusterSizesResponse)> { let value: Value = client - .get("/v1/cluster-sizes") + .get("/v1/clusters/sizes") .context("invalid cluster sizes response from server")?; let response: ClusterSizesResponse = serde_json::from_value(value.clone()) .context("invalid cluster sizes response from server")?; @@ -509,12 +509,7 @@ fn request_lifecycle( ) -> Result<()> { let (_, resp) = load_dedicated_clusters(client, organization)?; let cluster = resolve_cluster(&resp.clusters, name_or_id)?; - let path = cluster_path( - client, - &cluster.id, - Some(action.path_segment()), - organization, - )?; + let path = cluster_path(&cluster.id, Some(action.path_segment()), organization); let value: Value = client.post_empty(&path)?; let updated: ClusterItem = serde_json::from_value(value.clone()).context("invalid cluster response from server")?; @@ -534,22 +529,24 @@ fn request_lifecycle( Ok(()) } -fn cluster_path( - client: &ApiClient, - cluster_id: &str, - action: Option<&str>, - organization: Option<&str>, -) -> Result { - let mut path = format!( - "{}/{}", - client.clusters_path(organization)?, - urlencoding::encode(cluster_id) - ); +fn clusters_collection_path(organization: Option<&str>) -> String { + match organization { + Some(name) => format!("/v1/clusters?organization={}", urlencoding::encode(name)), + None => "/v1/clusters".to_string(), + } +} + +fn cluster_path(cluster_id: &str, action: Option<&str>, organization: Option<&str>) -> String { + let mut path = format!("/v1/clusters/{}", urlencoding::encode(cluster_id)); if let Some(action) = action { path.push('/'); path.push_str(action); } - Ok(path) + if let Some(name) = organization { + path.push_str("?organization="); + path.push_str(&urlencoding::encode(name)); + } + path } fn format_phase(phase: &str) -> String { @@ -606,13 +603,39 @@ mod tests { use serde_json::json; use super::{ - cluster_size_json, create_request_body, default_cluster_after_delete, delete_output, - empty_list_message, format_created_at, format_idle_timeout, format_phase, - format_size_per_replica, renamed_default_cluster, resolve_cluster, resolve_cluster_size, - ClusterItem, ClusterResources, ClusterSizeItem, ClusterStatus, ListClustersResponse, + cluster_path, cluster_size_json, clusters_collection_path, create_request_body, + default_cluster_after_delete, delete_output, empty_list_message, format_created_at, + format_idle_timeout, format_phase, format_size_per_replica, renamed_default_cluster, + resolve_cluster, resolve_cluster_size, ClusterItem, ClusterResources, ClusterSizeItem, + ClusterStatus, ListClustersResponse, }; use crate::cli::ClusterSizeArg; + #[test] + fn collection_path_encodes_organization() { + assert_eq!( + clusters_collection_path(Some("team alpha")), + "/v1/clusters?organization=team%20alpha" + ); + assert_eq!(clusters_collection_path(None), "/v1/clusters"); + } + + #[test] + fn lifecycle_paths_encode_cluster_and_organization() { + assert_eq!( + cluster_path("cluster/id", Some("stop"), Some("team alpha")), + "/v1/clusters/cluster%2Fid/stop?organization=team%20alpha" + ); + assert_eq!( + cluster_path("cluster-id", Some("resume"), None), + "/v1/clusters/cluster-id/resume" + ); + assert_eq!( + cluster_path("cluster-id", None, Some("team alpha")), + "/v1/clusters/cluster-id?organization=team%20alpha" + ); + } + #[test] fn cluster_size_catalog_resolves_cpu_and_memory_pairs() { let sizes = vec![ diff --git a/src/commands/insert.rs b/src/commands/insert.rs index 6ff7b08..2f9bcde 100644 --- a/src/commands/insert.rs +++ b/src/commands/insert.rs @@ -5,6 +5,7 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::mpsc; use std::sync::Arc; use std::thread; +use std::time::Duration; use anyhow::{bail, Context, Result}; use indicatif::{ProgressBar, ProgressStyle}; diff --git a/tests/browser_login.rs b/tests/browser_login.rs index 60e12ea..7696ac8 100644 --- a/tests/browser_login.rs +++ b/tests/browser_login.rs @@ -45,13 +45,13 @@ fn organizations(names: &[&str]) -> Response { ( "GET /v1/organizations".into(), "200 OK", - json!({"organizations": names.iter().map(|name| json!({"organization_id": "00000000-0000-4000-8000-000000000001", "name": name, "role": "owner"})).collect::>()}), + json!({"organizations": names.iter().map(|name| json!({"name": name, "role": "owner"})).collect::>()}), ) } fn clusters(names: &[&str]) -> Response { ( - "GET /v1/organizations/00000000-0000-4000-8000-000000000001/clusters".into(), + "GET /v1/clusters?organization=team".into(), "200 OK", json!({"clusters": names.iter().map(|name| json!({"id": name, "name": name})).collect::>()}), ) @@ -61,7 +61,7 @@ fn databases(names: &[&str]) -> Response { ( "GET /v1/databases?organization=team&cluster=production".into(), "200 OK", - json!({"databases": names.iter().map(|name| json!({"name": name})).collect::>()}), + json!({"databases": names.iter().map(|name| json!({"id": name, "name": name})).collect::>()}), ) }