diff --git a/src/cli.rs b/src/cli.rs index 896b4a5..9dd1e63 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -4,6 +4,7 @@ use crate::commands::context::ContextCommands; use crate::commands::databases::DatabasesCommands; use crate::commands::embedding_providers::EmbeddingProvidersCommands; use crate::commands::indexes::IndexesCommands; +use crate::commands::ingest::IngestCommands; use crate::commands::jobs::JobsCommands; use crate::commands::queries::QueriesCommands; use crate::commands::query::QueryCommands; @@ -132,6 +133,21 @@ pub enum Commands { command: Option, }, + /// Ingest data from external connectors (databases, REST APIs, files, Iceberg) + /// into a managed database via the hotdata ingest service + Ingest { + /// Workspace ID (defaults to first workspace from login) + #[arg(long, short = 'w', global = true)] + workspace_id: Option, + + /// Output format + #[arg(long = "output", short = 'o', default_value = "table", value_parser = ["table", "json", "yaml"], global = true)] + output: String, + + #[command(subcommand)] + command: IngestCommands, + }, + /// Manage indexes on a table Indexes { /// Workspace ID (defaults to first workspace from login) diff --git a/src/client.rs b/src/client.rs index dc9d125..ea504d9 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,5 +1,6 @@ pub mod credentials; pub mod database_session; +pub mod ingest; pub mod jwt; pub mod raw_http; pub mod sdk; diff --git a/src/client/credentials.rs b/src/client/credentials.rs index a32dbee..e6c3b28 100644 --- a/src/client/credentials.rs +++ b/src/client/credentials.rs @@ -50,6 +50,45 @@ pub fn check_status(profile_config: &config::ProfileConfig) -> AuthStatus { } } +/// The workspace a command with no `--workspace-id` targets for this profile — +/// the single source of truth shared by `main`'s `resolve_workspace` and +/// `auth status`, so the status readout can never disagree with where commands +/// actually run. +/// +/// An api-key credential scoped to exactly one workspace (a database API token) +/// pins that workspace. For a multi-workspace or unrestricted api key we honor +/// the saved default (`workspaces set` moves a workspace to the front of the +/// config list) when the key can reach it, otherwise fall back to the +/// credential's own first authorized workspace — never a workspace the gateway +/// would reject. A CLI session uses the saved default. `None` means no default +/// is known and the caller must pass `--workspace-id`. +pub(crate) fn default_workspace_id(profile_config: &config::ProfileConfig) -> Option { + let saved_default = || { + profile_config + .workspaces + .first() + .map(|w| w.public_id.clone()) + }; + if !matches!( + profile_config.api_key_source, + ApiKeySource::Flag | ApiKeySource::Env + ) { + return saved_default(); + } + let ids = api_key_workspace_ids(profile_config); + if let [only] = ids.as_slice() { + return Some(only.clone()); + } + // Multi/unrestricted key: prefer the saved default when the key authorizes + // it (empty `ids` = unrestricted, reaches everything), else the key's first. + if let Some(first) = saved_default() + && (ids.is_empty() || ids.contains(&first)) + { + return Some(first); + } + ids.into_iter().next() +} + /// Workspace public-ids the active api-key credential (`--api-key` / /// `HOTDATA_API_KEY`) is scoped to, read from its minted JWT's `workspaces` /// claim. A database API token carries exactly one. Empty when there's no api @@ -212,6 +251,93 @@ mod tests { assert_eq!(ids, vec!["workbound".to_string()]); } + // --- default_workspace_id tests --- + + fn ws(id: &str) -> config::WorkspaceEntry { + config::WorkspaceEntry { + public_id: id.into(), + name: id.into(), + } + } + + /// Mint mock returning a JWT whose `workspaces` claim is `ids`. + fn mock_token_with_workspaces(server: &mut mockito::Server, ids: &[&str]) -> mockito::Mock { + let claim = serde_json::json!({ "workspaces": ids }).to_string(); + let jwt = format!("header.{}.sig", URL_SAFE_NO_PAD.encode(claim.as_bytes())); + server + .mock("POST", "/o/token/") + .match_body(mockito::Matcher::UrlEncoded( + "grant_type".into(), + "api_token".into(), + )) + .with_status(200) + .with_header("content-type", "application/json") + .with_body(format!( + r#"{{"access_token":"{jwt}","expires_in":300,"refresh_token":"r"}}"# + )) + .create() + } + + #[test] + fn default_workspace_id_session_uses_saved_default_without_network() { + // Config source (a CLI session): the saved default, no mint call. + let (_tmp, _guard) = with_temp_config_dir(); + let profile = ProfileConfig { + workspaces: vec![ws("work_saved"), ws("work_other")], + ..Default::default() // api_key_source defaults to Config + }; + assert_eq!( + default_workspace_id(&profile), + Some("work_saved".to_string()) + ); + } + + #[test] + fn default_workspace_id_single_workspace_token_pins_its_own() { + // A database token authorizes exactly one workspace — use it even when a + // different workspace sits at the front of the (unrelated) config cache. + let (_tmp, _guard) = with_temp_config_dir(); + let mut server = mockito::Server::new(); + let mint = mock_token_with_workspaces(&mut server, &["work_only"]); + let mut profile = mock_profile(&server.url(), Some("hd_dbtoken")); + profile.api_key_source = ApiKeySource::Env; + profile.workspaces = vec![ws("work_saved")]; + assert_eq!( + default_workspace_id(&profile), + Some("work_only".to_string()) + ); + mint.assert(); + } + + #[test] + fn default_workspace_id_multi_key_honors_saved_default_when_authorized() { + // Multi-workspace key + a saved default the key can reach → the saved + // default wins (so `workspaces set` keeps working). + let (_tmp, _guard) = with_temp_config_dir(); + let mut server = mockito::Server::new(); + let _mint = mock_token_with_workspaces(&mut server, &["work_a", "work_saved", "work_b"]); + let mut profile = mock_profile(&server.url(), Some("hd_org")); + profile.api_key_source = ApiKeySource::Env; + profile.workspaces = vec![ws("work_saved")]; + assert_eq!( + default_workspace_id(&profile), + Some("work_saved".to_string()) + ); + } + + #[test] + fn default_workspace_id_multi_key_falls_back_to_first_authorized() { + // Saved default is NOT one the key authorizes → the credential's first + // authorized workspace, never a workspace the gateway would 403. + let (_tmp, _guard) = with_temp_config_dir(); + let mut server = mockito::Server::new(); + let _mint = mock_token_with_workspaces(&mut server, &["work_a", "work_b"]); + let mut profile = mock_profile(&server.url(), Some("hd_org")); + profile.api_key_source = ApiKeySource::Env; + profile.workspaces = vec![ws("work_unauthorized")]; + assert_eq!(default_workspace_id(&profile), Some("work_a".to_string())); + } + // --- check_status tests --- #[test] diff --git a/src/client/ingest.rs b/src/client/ingest.rs new file mode 100644 index 0000000..20ab99a --- /dev/null +++ b/src/client/ingest.rs @@ -0,0 +1,799 @@ +//! Raw-HTTP client for the ingest API (`/v1/ingest/*`). +//! +//! These routes are not in the generated SDK yet, so — like the token/session +//! mints — they ride the hand-rolled `reqwest::blocking` seam in +//! [`crate::client::raw_http`]. +//! Every request carries a bearer + `X-Workspace-Id`; the gateway validates the +//! pair and derives the ingest destination server-side, so the CLI never sends +//! a destination. +//! +//! Auth is split by endpoint kind. The enqueue routes (`POST /sources`, +//! `POST /queries`) require a durable `hd_...` API key as the bearer — the +//! drain job uses the credential *after* the request returns, so the server +//! 422s any 5-minute JWT. Read routes accept *workspace-scoped* JWTs, but the +//! CLI's login session is a user-scoped JWT and the worker refuses to trust +//! `X-Workspace-Id` on the JWT route — so in practice only `/connectors` +//! (workspace-free) works without a key. So: when an API key is available +//! (`--api-key` / `HOTDATA_API_KEY`) it is sent directly on every call; +//! otherwise the CLI's session JWT ([`jwt::ensure_access_token`]) is used, +//! enqueue calls fail fast with [`IngestError::NeedsApiKey`], and +//! workspace-scoped reads get a 403 with an `--api-key` hint. +//! +//! When the ingest routes land in the public OpenAPI and the SDK regenerates, +//! delete this module and move the commands onto `sdk::Api`. +//! +//! Surface: `create_source` (onboard, backs `new-connection`), `create_query` +//! (SQL front-door, backs `new-import`), `list_sources` / `list_queries` (the +//! registries behind `list-connections` / `list-imports`), `rerun` (backs +//! `trigger-import`), `connectors` (the catalog), `drain` + `job_status` (the +//! async run loop). The result-reading endpoints are intentionally absent — +//! that path is the core `query`/`databases`/`results` commands. +#![allow(dead_code)] // Response structs are read only through serde/printing. + +use crate::client::jwt; +use crate::config; +use crate::util; +use serde::{Deserialize, Serialize}; + +/// A typed error from an ingest call. Mirrors the `ApiError::exit()` ergonomics +/// the SDK-backed commands use, so handlers can `.unwrap_or_else(|e| e.exit())`. +#[derive(Debug)] +pub enum IngestError { + /// Non-2xx response; `body` is the server's (unredacted) response text. + Http { status: u16, body: String }, + /// Transport/connection failure. + Connection(String), + /// 2xx whose body didn't match the expected shape. + Decode(String), + /// Enqueue attempted with only a session JWT — the server would 422 it + /// (the drain job runs after a short-lived JWT expires), so fail before + /// sending credentials that can't work. + NeedsApiKey, +} + +impl IngestError { + pub fn message(&self) -> String { + match self { + IngestError::Http { status, body } => format!("HTTP {status}: {body}"), + IngestError::Connection(e) => format!("connection error: {e}"), + IngestError::Decode(e) => format!("malformed response: {e}"), + IngestError::NeedsApiKey => { + "enqueueing an ingest requires a durable API key — the ingest pipeline \ + runs after a short-lived session token would expire" + .into() + } + } + } + + pub fn exit(&self) -> ! { + use crossterm::style::Stylize; + eprintln!("{}", format!("error: {}", self.message()).red()); + // Cold-start / scale-to-zero hint: the worker sits behind KEDA. + if matches!( + self, + IngestError::Http { + status: 502 | 503, + .. + } + ) { + eprintln!( + "{}", + "the ingest service may be starting up — retry in a few seconds".dark_grey() + ); + } + // A transport failure on an enqueue is usually the worker being + // unavailable — a cold start or a rollout — where the gateway holds the + // connection until its timeout rather than returning a status. "error + // sending request" is opaque; point at the actual cause + retry. + if matches!(self, IngestError::Connection(_)) { + eprintln!( + "{}", + "the request didn't complete — the ingest service may be starting up or \ + redeploying; retry in a moment (a timed-out enqueue is safe to re-run)." + .dark_grey() + ); + } + if matches!(self, IngestError::NeedsApiKey) { + eprintln!( + "{}", + "Pass --api-key or set HOTDATA_API_KEY with a workspace API token (hd_...)." + .dark_grey() + ); + } + // The worker refuses to trust X-Workspace-Id on the JWT route, and a + // CLI login session is a *user*-scoped JWT — so every workspace-scoped + // ingest endpoint 403s on it. Only an API key carries the workspace. + if let IngestError::Http { status: 403, body } = self + && body.contains("workspace-scoped credential") + { + eprintln!( + "{}", + "Your CLI session is user-scoped; ingest needs a workspace credential — \ + pass --api-key or set HOTDATA_API_KEY." + .dark_grey() + ); + } + std::process::exit(1); + } +} + +/// Ingest client bound to a workspace + a resolved bearer token. +pub struct IngestClient { + /// `{api_url}/ingest` — api_url already carries the `/v1` suffix. + base: String, + token: String, + /// Whether `token` is a durable `hd_...` API key (vs a session JWT). + /// Enqueue endpoints require the former; see the module docs. + token_is_api_key: bool, + workspace_id: String, + client: reqwest::blocking::Client, +} + +impl IngestClient { + /// Build a client for `workspace_id`. An explicit API key (`--api-key` / + /// `HOTDATA_API_KEY`) is sent as the bearer directly — the extAuth route + /// accepts it everywhere and the enqueue routes *require* it. Without one, + /// fall back to the CLI's session JWT, which covers the read routes. + pub fn new(workspace_id: &str) -> Self { + let profile = config::load("default").unwrap_or_else(|e| { + eprintln!("{e}"); + std::process::exit(1); + }); + let (token, token_is_api_key) = match profile.api_key.clone() { + Some(key) => (key, true), + None => { + let jwt = jwt::ensure_access_token(&profile, None).unwrap_or_else(|e| { + use crossterm::style::Stylize; + eprintln!("{}", format!("auth error: {e}").red()); + eprintln!("Run 'hotdata auth login' to authenticate."); + std::process::exit(1); + }); + (jwt, false) + } + }; + let base = format!("{}/ingest", (*profile.api_url).trim_end_matches('/')); + IngestClient { + base, + token, + token_is_api_key, + workspace_id: workspace_id.to_string(), + client: crate::client::raw_http::build_http_client(), + } + } + + /// Test-only constructor bypassing config/session resolution. + #[cfg(test)] + pub fn from_parts(base: &str, token: &str, token_is_api_key: bool, workspace_id: &str) -> Self { + IngestClient { + base: format!("{}/ingest", base.trim_end_matches('/')), + token: token.to_string(), + token_is_api_key, + workspace_id: workspace_id.to_string(), + client: crate::client::raw_http::build_http_client(), + } + } + + fn url(&self, path: &str) -> String { + format!("{}{}", self.base, path) + } + + /// A request builder with the bearer + workspace headers already set. + fn authed(&self, method: reqwest::Method, path: &str) -> reqwest::blocking::RequestBuilder { + self.client + .request(method, self.url(path)) + .header("Authorization", format!("Bearer {}", self.token)) + .header("X-Workspace-Id", &self.workspace_id) + } + + /// Send a request, enforce a 2xx, and decode the JSON body into `T`. + /// + /// `body_log` is the *printable* form for `--debug` — callers whose body + /// carries secrets must pass a view through [`redact_secret_fields`], never + /// the wire body itself. + fn send Deserialize<'de>>( + &self, + builder: reqwest::blocking::RequestBuilder, + body_log: Option<&serde_json::Value>, + ) -> Result { + let (status, body) = util::send_debug(&self.client, builder, body_log) + .map_err(|e| IngestError::Connection(e.to_string()))?; + if !status.is_success() { + return Err(IngestError::Http { + status: status.as_u16(), + body, + }); + } + serde_json::from_str(&body).map_err(|e| IngestError::Decode(e.to_string())) + } + + // --- write endpoints ------------------------------------------------- + + /// Guard for the enqueue routes: a session JWT is rejected server-side + /// (422) because the drain job outlives it, so fail fast with a message + /// that says what to do instead. + fn require_api_key(&self) -> Result<(), IngestError> { + if self.token_is_api_key { + Ok(()) + } else { + Err(IngestError::NeedsApiKey) + } + } + + pub fn create_source(&self, req: &IngestRequest) -> Result { + self.require_api_key()?; + let body = serde_json::to_value(req).expect("IngestRequest serializes"); + let body_log = redact_secret_fields(&body); + self.send( + self.authed(reqwest::Method::POST, "/sources").json(&body), + Some(&body_log), + ) + } + + pub fn create_query(&self, req: &QueryToIngest) -> Result { + self.require_api_key()?; + let body = serde_json::to_value(req).expect("QueryToIngest serializes"); + self.send( + self.authed(reqwest::Method::POST, "/queries").json(&body), + Some(&body), + ) + } + + pub fn drain(&self) -> Result { + self.send(self.authed(reqwest::Method::POST, "/jobs/drain"), None) + } + + /// Re-run an ingest: the worker resets it to pending and fires the drain. + /// Replace-mode loads + the stored (encrypted) source and destination + /// credentials mean this refreshes the SAME managed DB from source. No + /// API-key guard: the drain reuses the destination stored at enqueue, so + /// the server accepts any credential its write check does. A 409 means a + /// drain is mid-flight — the worker's message says to retry shortly. + pub fn rerun(&self, ingest_id: &str) -> Result { + self.send( + self.authed(reqwest::Method::POST, &format!("/jobs/{ingest_id}/rerun")), + None, + ) + } + + // --- read endpoints -------------------------------------------------- + + /// The connector catalog. REST entries carry a ready-to-edit `template` + /// (a dlt `rest_api` config with the service's `base_url`, auth shape, and + /// resources pre-filled and `` secrets) — this is what lets + /// the `new` wizard fill in everything but the caller's secrets. + pub fn connectors(&self) -> Result { + self.send(self.authed(reqwest::Method::GET, "/connectors"), None) + } + + pub fn job_status(&self, ingest_id: &str) -> Result { + self.send( + self.authed(reqwest::Method::GET, &format!("/jobs/{ingest_id}")), + None, + ) + } + + /// The onboarded-source registry (`GET /sources`) — one row per connection, + /// newest first, each with its own ingest id (the connection id). The + /// default view is the current set: the latest onboard per connector name + /// plus unnamed onboards; `all` includes superseded onboards too. + pub fn list_sources(&self, all: bool) -> Result { + let path = if all { "/sources?all=true" } else { "/sources" }; + self.send(self.authed(reqwest::Method::GET, path), None) + } + + /// The import registry (`GET /queries`) — every SQL import, newest first, + /// with the SQL that produced it, its source connection, and the managed + /// DB it landed in. + pub fn list_queries(&self) -> Result { + self.send(self.authed(reqwest::Method::GET, "/queries"), None) + } + + /// Tables + columns discovered for a result database — the schema preview a + /// metadata-only onboard lands: `{database_id, tables: {table: [columns]}}`. + pub fn schema(&self, database_id: &str) -> Result { + self.send( + self.authed( + reqwest::Method::GET, + &format!("/databases/{database_id}/schema"), + ), + None, + ) + } +} + +/// `IngestRequest` fields that carry source secrets: SQL passwords / +/// connection strings, REST bearer tokens inside `rest_config`, Iceberg +/// catalog tokens, filesystem access keys. +const SECRET_BODY_FIELDS: &[&str] = &["credentials", "rest_config", "catalog_config"]; + +/// Debug-log view of an enqueue body with the secret-bearing subtrees +/// replaced wholesale. These fields are nested *objects* whose secret keys +/// vary by connector, so dropping the whole subtree beats field-level +/// masking (`util::redact_json_fields` only masks string values). Mirrors +/// the `redacted_form_body` pattern in `jwt.rs`. +fn redact_secret_fields(body: &serde_json::Value) -> serde_json::Value { + let mut v = body.clone(); + if let serde_json::Value::Object(map) = &mut v { + for key in SECRET_BODY_FIELDS { + if let Some(val) = map.get_mut(*key) { + *val = serde_json::Value::String("***".into()); + } + } + } + v +} + +// --- request / response types ------------------------------------------- + +/// Mirrors the worker's `IngestRequest`. Sensitive fields (`credentials`, +/// `rest_config`, `catalog_config`) are forwarded over TLS and Fernet-encrypted +/// at rest by the worker. `None`/empty fields are omitted so the worker applies +/// its own defaults. +#[derive(Serialize, Default)] +pub struct IngestRequest { + pub family: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub connector_type: Option, + #[serde(skip_serializing_if = "serde_json::Value::is_null")] + pub credentials: serde_json::Value, + #[serde(skip_serializing_if = "Option::is_none")] + pub rest_config: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub schema: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub table_names: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub bucket_url: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub file_glob: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub file_format: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub catalog_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub catalog_type: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub catalog_config: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub tables: Vec, + #[serde(skip_serializing_if = "std::ops::Not::not")] + pub validate_only: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub database_id: Option, +} + +/// Mirrors the worker's `QueryToIngestRequest`. +#[derive(Serialize)] +pub struct QueryToIngest { + pub query: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub source_ingest_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub database_id: Option, +} + +/// 202 body from `POST /sources` and `POST /queries`. Common fields are typed; +/// everything else (e.g. `resource`, `columns`, `where`, `limit` on queries) is +/// captured for `--output json`. +#[derive(Debug, Deserialize, Serialize)] +pub struct IngestAck { + pub ingest_id: String, + pub database_id: String, + pub status: String, + #[serde(default)] + pub status_url: Option, + #[serde(flatten)] + pub extra: serde_json::Map, +} + +/// `GET /jobs/{id}` body. +#[derive(Debug, Deserialize, Serialize)] +pub struct JobStatus { + pub ingest_id: String, + pub status: String, + #[serde(default)] + pub detail: Option, + #[serde(default)] + pub connector_type: Option, + #[serde(default)] + pub family: Option, + #[serde(default)] + pub database_id: Option, + #[serde(default)] + pub created_at: Option, + #[serde(default)] + pub updated_at: Option, +} + +/// `GET /sources` body. +#[derive(Deserialize)] +pub struct SourcesResponse { + pub sources: Vec, +} + +/// One connection in the onboarded-source registry. `ingest_id` is the +/// connection's own id — the pin key `new-import --source` accepts. `active` +/// marks the row by-name resolution picks (always true in the default view; +/// superseded onboards surface with `all=true`). +#[derive(Debug, Deserialize, Serialize)] +pub struct SourceRow { + pub ingest_id: String, + #[serde(default)] + pub connector_type: Option, + #[serde(default)] + pub family: Option, + #[serde(default)] + pub database_id: Option, + pub status: String, + #[serde(default)] + pub detail: Option, + #[serde(default)] + pub created_at: Option, + #[serde(default)] + pub updated_at: Option, + #[serde(default)] + pub active: bool, +} + +/// `GET /queries` body. +#[derive(Deserialize)] +pub struct QueriesResponse { + pub queries: Vec, +} + +/// One import in the registry: the SQL that produced it (verbatim for new +/// rows, server-reconstructed for older ones), the connection it drew from, +/// and the managed DB it landed in. +#[derive(Debug, Deserialize, Serialize)] +pub struct QueryRow { + pub ingest_id: String, + pub source_ingest_id: String, + #[serde(default)] + pub connector_type: Option, + #[serde(default)] + pub family: Option, + #[serde(default)] + pub query: Option, + #[serde(default)] + pub database_id: Option, + pub status: String, + #[serde(default)] + pub detail: Option, + #[serde(default)] + pub created_at: Option, + #[serde(default)] + pub updated_at: Option, +} + +#[derive(Deserialize)] +pub struct ConnectorsResponse { + pub connectors: Vec, +} + +/// One catalog entry. `sql` names are dialects, `filesystem`/`iceberg`/`rest` +/// are family templates. REST entries additionally carry `auth` (the method +/// name, e.g. `bearer`, `oauth_client_credentials`, `none`) and a `template` +/// dlt config with `` secrets the wizard prompts for. +#[derive(Clone, Deserialize)] +pub struct ConnectorEntry { + pub name: String, + #[serde(default)] + pub family: String, + #[serde(default)] + pub description: String, + #[serde(default)] + pub auth: Option, + #[serde(default)] + pub template: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + + fn api_key_client(server: &mockito::Server) -> IngestClient { + IngestClient::from_parts(&server.url(), "hd_test", true, "ws-1") + } + + fn jwt_client(server: &mockito::Server) -> IngestClient { + IngestClient::from_parts(&server.url(), "eyJ.fake.jwt", false, "ws-1") + } + + // --- enqueue auth ------------------------------------------------------ + + #[test] + fn create_source_sends_api_key_bearer_and_workspace_header() { + let mut server = mockito::Server::new(); + let m = server + .mock("POST", "/ingest/sources") + .match_header("authorization", "Bearer hd_test") + .match_header("x-workspace-id", "ws-1") + .match_body(mockito::Matcher::Json(serde_json::json!({ + "family": "sql", + "connector_type": "postgres", + "credentials": {"connection_string": "postgresql://u:p@h:5432/d"}, + "validate_only": true, + }))) + .with_status(202) + .with_header("content-type", "application/json") + .with_body( + r#"{"ingest_id":"ing-1","database_id":"db-1","workspace_id":"ws-1", + "status":"pending","status_url":"/ingest/jobs/ing-1"}"#, + ) + .create(); + + let req = IngestRequest { + family: "sql".into(), + connector_type: Some("postgres".into()), + credentials: serde_json::json!({"connection_string": "postgresql://u:p@h:5432/d"}), + validate_only: true, + ..Default::default() + }; + let ack = api_key_client(&server).create_source(&req).unwrap(); + m.assert(); + assert_eq!(ack.ingest_id, "ing-1"); + assert_eq!(ack.database_id, "db-1"); + assert_eq!(ack.status, "pending"); + // Untyped extras (workspace_id, …) survive for --output json. + assert_eq!(ack.extra["workspace_id"], "ws-1"); + } + + #[test] + fn enqueue_with_session_jwt_fails_fast_without_http() { + // Point at a dead port: reaching the network would surface as a + // Connection error instead of NeedsApiKey. + let client = IngestClient::from_parts("http://127.0.0.1:1", "eyJ.fake.jwt", false, "ws-1"); + + let source_err = client.create_source(&IngestRequest::default()).unwrap_err(); + assert!(matches!(source_err, IngestError::NeedsApiKey)); + + let query_err = client + .create_query(&QueryToIngest { + query: "SELECT * FROM x".into(), + source_ingest_id: None, + database_id: None, + }) + .unwrap_err(); + assert!(matches!(query_err, IngestError::NeedsApiKey)); + assert!( + query_err.message().contains("API key"), + "got: {}", + query_err.message() + ); + } + + // --- read endpoints accept a JWT --------------------------------------- + + #[test] + fn job_status_works_with_jwt_and_decodes_body() { + let mut server = mockito::Server::new(); + let m = server + .mock("GET", "/ingest/jobs/ing-1") + .match_header("authorization", "Bearer eyJ.fake.jwt") + .with_status(200) + .with_header("content-type", "application/json") + .with_body( + r#"{"ingest_id":"ing-1","status":"failed","detail":"boom", + "connector_type":"postgres","family":"sql","database_id":"db-1", + "created_at":"2026-07-07T15:30:10+00:00", + "updated_at":"2026-07-07T15:30:45+00:00", + "dlt_workspace_id":"ignored-unknown-field"}"#, + ) + .create(); + + let st = jwt_client(&server).job_status("ing-1").unwrap(); + m.assert(); + assert_eq!(st.status, "failed"); + assert_eq!(st.detail.as_deref(), Some("boom")); + assert_eq!(st.database_id.as_deref(), Some("db-1")); + } + + #[test] + fn http_error_carries_status_and_body() { + let mut server = mockito::Server::new(); + let m = server + .mock("GET", "/ingest/jobs/nope") + .with_status(404) + .with_body(r#"{"detail":"ingest not found"}"#) + .create(); + + let err = api_key_client(&server).job_status("nope").unwrap_err(); + m.assert(); + match err { + IngestError::Http { status, body } => { + assert_eq!(status, 404); + assert!(body.contains("ingest not found"), "got: {body}"); + } + other => panic!("expected Http, got: {}", other.message()), + } + } + + #[test] + fn connectors_decodes_rest_template_and_auth() { + let mut server = mockito::Server::new(); + let m = server + .mock("GET", "/ingest/connectors") + .with_status(200) + .with_header("content-type", "application/json") + .with_body( + r#"{"connectors":[ + {"name":"postgres","family":"sql","description":"PostgreSQL"}, + {"name":"aikido","family":"rest","auth":"oauth_client_credentials", + "description":"Security posture", + "template":{"client":{"base_url":"https://app.aikido.dev/api/public/v1/", + "auth":{"type":"oauth2_client_credentials","client_id":""}}}} + ]}"#, + ) + .create(); + + let resp = api_key_client(&server).connectors().unwrap(); + m.assert(); + assert_eq!(resp.connectors.len(), 2); + let pg = &resp.connectors[0]; + assert_eq!(pg.family, "sql"); + assert!(pg.template.is_none() && pg.auth.is_none()); + let aikido = &resp.connectors[1]; + assert_eq!(aikido.auth.as_deref(), Some("oauth_client_credentials")); + assert_eq!( + aikido.template.as_ref().unwrap()["client"]["base_url"], + "https://app.aikido.dev/api/public/v1/" + ); + } + + // --- registry reads ------------------------------------------------------ + + #[test] + fn list_sources_default_and_all_hit_the_right_paths() { + let mut server = mockito::Server::new(); + let m_default = server + .mock("GET", "/ingest/sources") + .match_header("x-workspace-id", "ws-1") + .with_status(200) + .with_header("content-type", "application/json") + .with_body( + r#"{"sources":[{"ingest_id":"a1","connector_type":"postgres","family":"sql", + "database_id":"db-1","status":"done","detail":null, + "created_at":"2026-07-08T10:00:00+00:00","updated_at":null,"active":true}]}"#, + ) + .create(); + let m_all = server + .mock("GET", "/ingest/sources?all=true") + .with_status(200) + .with_header("content-type", "application/json") + .with_body(r#"{"sources":[]}"#) + .create(); + + let client = api_key_client(&server); + let resp = client.list_sources(false).unwrap(); + assert_eq!(resp.sources.len(), 1); + let s = &resp.sources[0]; + assert_eq!(s.ingest_id, "a1"); + assert_eq!(s.connector_type.as_deref(), Some("postgres")); + assert!(s.active); + + assert!(client.list_sources(true).unwrap().sources.is_empty()); + m_default.assert(); + m_all.assert(); + } + + #[test] + fn list_queries_decodes_sql_and_source_link() { + let mut server = mockito::Server::new(); + let m = server + .mock("GET", "/ingest/queries") + .with_status(200) + .with_header("content-type", "application/json") + .with_body( + r#"{"queries":[{"ingest_id":"q1","source_ingest_id":"a1", + "connector_type":"bitcoin","family":"rest", + "query":"SELECT id, height FROM bitcoin.blocks LIMIT 100", + "database_id":"db-2","status":"done","detail":null, + "created_at":"2026-07-08T10:05:00+00:00","updated_at":null}]}"#, + ) + .create(); + + let resp = api_key_client(&server).list_queries().unwrap(); + m.assert(); + let q = &resp.queries[0]; + assert_eq!(q.source_ingest_id, "a1"); + assert_eq!( + q.query.as_deref(), + Some("SELECT id, height FROM bitcoin.blocks LIMIT 100") + ); + assert_eq!(q.database_id.as_deref(), Some("db-2")); + } + + #[test] + fn rerun_posts_and_decodes_the_pending_ack() { + let mut server = mockito::Server::new(); + let m = server + .mock("POST", "/ingest/jobs/q1/rerun") + .match_header("authorization", "Bearer hd_test") + .match_header("x-workspace-id", "ws-1") + .with_status(202) + .with_header("content-type", "application/json") + .with_body( + r#"{"ingest_id":"q1","database_id":"db-2","connector_type":"bitcoin", + "family":"rest","status":"pending","status_url":"/ingest/jobs/q1"}"#, + ) + .create(); + + let ack = api_key_client(&server).rerun("q1").unwrap(); + m.assert(); + assert_eq!(ack.ingest_id, "q1"); + assert_eq!(ack.database_id, "db-2"); + assert_eq!(ack.status, "pending"); + } + + #[test] + fn rerun_409_surfaces_the_in_flight_detail() { + let mut server = mockito::Server::new(); + let m = server + .mock("POST", "/ingest/jobs/q1/rerun") + .with_status(409) + .with_body(r#"{"detail":"a drain appears to be running (in flight: a1); retry once it settles"}"#) + .create(); + + let err = api_key_client(&server).rerun("q1").unwrap_err(); + m.assert(); + match err { + IngestError::Http { status, body } => { + assert_eq!(status, 409); + assert!(body.contains("retry once it settles"), "got: {body}"); + } + other => panic!("expected Http, got: {}", other.message()), + } + } + + // --- debug-log redaction ------------------------------------------------- + + #[test] + fn redact_secret_fields_masks_all_secret_subtrees_and_keeps_the_rest() { + let body = serde_json::json!({ + "family": "rest", + "connector_type": "svc", + "credentials": {"connection_string": "postgresql://u:s3cret@h/db"}, + "rest_config": {"client": {"auth": {"type": "bearer", "token": "tok-abc"}}}, + "catalog_config": {"catalog_uri": "http://c", "token": "iceberg-tok"}, + "table_names": ["users"], + }); + let logged = redact_secret_fields(&body); + for key in super::SECRET_BODY_FIELDS { + assert_eq!( + logged[*key], "***", + "{key} must be dropped from the debug view" + ); + } + let printed = logged.to_string(); + assert!( + !printed.contains("s3cret") + && !printed.contains("tok-abc") + && !printed.contains("iceberg-tok"), + "no secret may survive into the printable body: {printed}" + ); + // Non-secret fields stay readable, and the wire body is untouched. + assert_eq!(logged["family"], "rest"); + assert_eq!(logged["table_names"][0], "users"); + assert_eq!( + body["credentials"]["connection_string"], + "postgresql://u:s3cret@h/db" + ); + } + + // --- request serialization ---------------------------------------------- + + #[test] + fn ingest_request_omits_unset_fields() { + // The worker applies its own defaults; nulls/empties must not be sent. + let req = IngestRequest { + family: "filesystem".into(), + bucket_url: Some("s3://b/prefix".into()), + ..Default::default() + }; + let v = serde_json::to_value(&req).unwrap(); + assert_eq!( + v, + serde_json::json!({"family": "filesystem", "bucket_url": "s3://b/prefix"}) + ); + } +} diff --git a/src/commands.rs b/src/commands.rs index 6e3183e..c6967a9 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -4,6 +4,7 @@ pub mod context; pub mod databases; pub mod embedding_providers; pub mod indexes; +pub mod ingest; pub mod jobs; pub mod queries; pub mod query; diff --git a/src/commands/auth.rs b/src/commands/auth.rs index 16c77d6..6ea6201 100644 --- a/src/commands/auth.rs +++ b/src/commands/auth.rs @@ -86,27 +86,31 @@ pub fn status(profile: &str) { if let Some(src) = api_key_jwt_source(&profile_config) { print_row("API Key Source", &src.as_str().cyan().to_string()); } - // An api key (--api-key / HOTDATA_API_KEY) is authorized for its own - // workspaces, which may differ from the CLI session's cached - // `config.workspaces`. Show the workspace the *credential* can reach - // (a database API token is bound to exactly one) rather than a stale - // session cache. The CLI-session path keeps its cached list. - let display_workspaces = match profile_config.api_key_source { + // Show the workspace commands will actually target — the exact + // value `resolve_workspace` uses — so the readout can't lie about + // where work lands. Resolve a display name from the credential's + // own authorized workspaces (api key) or the cached list (session), + // falling back to the bare id if the name isn't known. + let default_id = crate::client::credentials::default_workspace_id(&profile_config); + let known = match profile_config.api_key_source { ApiKeySource::Flag | ApiKeySource::Env => { api_key_authorized_workspaces(&profile_config) } ApiKeySource::Config => profile_config.workspaces.clone(), }; - match display_workspaces.first() { - Some(w) => { - print_row( - "Workspace", - &format!( - "{} {}", - w.name.as_str().cyan(), - format!("({})", w.public_id).dark_grey() - ), - ); + match default_id { + Some(id) => { + let name = known + .iter() + .find(|w| w.public_id == id) + .map(|w| w.name.clone()); + let display = match &name { + Some(n) => { + format!("{} {}", n.as_str().cyan(), format!("({id})").dark_grey()) + } + None => id.cyan().to_string(), + }; + print_row("Workspace", &display); print_row( "", &"use 'hotdata workspaces set' to switch workspaces" diff --git a/src/commands/ingest.rs b/src/commands/ingest.rs new file mode 100644 index 0000000..f8475cb --- /dev/null +++ b/src/commands/ingest.rs @@ -0,0 +1,1423 @@ +//! `hotdata ingest` — pull data from external sources into a managed database. +//! +//! Two nouns, named explicitly in every command: **connections** (onboarded, +//! credentialed sources — schema discovered, no data loaded) and **imports** +//! (managed DBs materialized from a connection). Commands: `new-connection`, +//! `list-connections`, `connectors` (the catalog), `new-import` (--all or +//! SQL), `list-imports`, `trigger-import` (re-run: refresh an ingest's DB +//! from source). The pre-rename verbs (`new`/`list`/`import`/`update`) remain +//! as hidden aliases for one release. +//! +//! `new-connection` **adds a connection and discovers its schema — it loads no +//! data** (the server's `validate_only` mode: check credentials, reflect the +//! schema, cap extraction). It runs the guided wizard when no `--service` is +//! given on a terminal, and is flag-driven otherwise (`--service` given, a +//! non-TTY, or `--no-input`) — one command, two front doors. Pulling rows is +//! the separate, explicit `new-import` step, read back through the core +//! `query`/`databases`/`results` commands. +//! +//! The wizard is catalog-driven: the `/ingest/connectors` catalog returns a +//! `template` per REST service with the `base_url`, auth shape, and resources +//! already filled in, so the user only supplies the `` secrets — +//! never a URL the catalog already knows. Enqueueing needs a durable `hd_` API +//! key (`--api-key`/`HOTDATA_API_KEY`); results land in the authenticated +//! workspace — there is no destination flag by design. + +use crate::client::ingest::{ + ConnectorEntry, IngestAck, IngestClient, IngestRequest, QueryToIngest, +}; +use crate::util; +use inquire::{Password, Select, Text}; +use std::time::{Duration, Instant}; + +/// Status-poll cadence (what the hotdlt UI uses). Each poll is a control-store +/// query on the worker, so this is deliberately not sub-second. +const POLL_INTERVAL: Duration = Duration::from_millis(2_500); + +/// A drain fired immediately after enqueue can miss the new row (control-store +/// read lag). If a job sits pending this long, kick the drain again. +const DRAIN_REKICK_AFTER: Duration = Duration::from_secs(30); + +#[derive(clap::Subcommand)] +pub enum IngestCommands { + /// Add a source connection and discover its schema (loads no data) + /// + /// Interactive by default; pass `--service` with config flags to add + /// non-interactively. Pull rows separately with `hotdata ingest + /// new-import`; browse connectors with `hotdata ingest connectors`. + #[command(alias = "new")] + NewConnection { + /// Connector to add (a catalog name: postgres, bitcoin, filesystem, …). + /// Given → non-interactive; omit on a terminal → guided wizard. + #[arg(long)] + service: Option, + + /// Family payload as JSON (inline, @file.json, or @-): SQL credentials, + /// a REST rest_config, filesystem creds, or an Iceberg catalog_config + #[arg(long)] + config: Option, + + /// Restrict discovery to this table (repeatable; sql/iceberg) + #[arg(long = "table")] + tables: Vec, + + /// Schema to discover (sql) + #[arg(long)] + schema: Option, + + /// Bucket URL, e.g. s3://bucket/prefix (filesystem) + #[arg(long = "bucket-url")] + bucket_url: Option, + + /// File format (filesystem) + #[arg(long, value_parser = ["csv", "jsonl", "parquet"])] + format: Option, + + /// File glob, e.g. **/*.parquet (filesystem) + #[arg(long)] + glob: Option, + + /// Catalog type, e.g. rest (iceberg) + #[arg(long = "catalog-type")] + catalog_type: Option, + + #[command(flatten)] + wait: WaitArgs, + + /// Reuse an existing managed database (by id) instead of minting one + #[arg(long = "database-id")] + database_id: Option, + }, + + /// List the connections you've added (each has its own connection id) + #[command(alias = "list")] + ListConnections { + /// Include superseded onboards (older connections a newer onboard of + /// the same connector replaced) + #[arg(long)] + all: bool, + }, + + /// Browse the connector catalog + #[command(alias = "supported-connectors")] + Connectors { + /// Filter to connectors whose name contains this text + name: Option, + /// Output format + #[arg(long = "output", short = 'o', default_value = "table", value_parser = ["table", "json", "yaml"])] + output: String, + }, + + /// Import data from a connection into a managed database (--all or SQL) + #[command(alias = "import")] + NewImport { + /// SELECT FROM [.] [WHERE …] [LIMIT n] + sql: Option, + + /// Import everything from --source (SELECT * with no LIMIT) + #[arg(long, conflicts_with = "sql")] + all: bool, + + /// Connection to import from: a connector name (used in FROM) or a + /// connection id from `list-connections` (pins resolution) + #[arg(long)] + source: Option, + + /// Reuse an existing managed database (by id) instead of minting one + #[arg(long = "database-id")] + database_id: Option, + + #[command(flatten)] + wait: WaitArgs, + }, + + /// List your imports: the SQL behind each and the database it landed in + ListImports, + + /// Re-run an ingest: refresh its database from the source + /// + /// The worker resets the ingest to pending and re-drains it; loads are + /// replace-mode, so the same managed database is refreshed with the + /// stored credentials — nothing is re-entered. + #[command(alias = "update")] + TriggerImport { + /// Ingest id: an import from `list-imports`, or a connection from + /// `list-connections` (re-validates and refreshes its schema) + id: String, + + /// Seconds to wait for completion (default 300) + #[arg(long = "wait-timeout", default_value = "300")] + wait_timeout: u64, + }, +} + +/// Shared wait/no-wait flags for the enqueue commands. +#[derive(clap::Args)] +pub struct WaitArgs { + /// Enqueue only; print the ingest id and return without waiting + #[arg(long = "no-wait")] + no_wait: bool, + + /// Seconds to wait for completion before giving up (default 300) + #[arg(long = "wait-timeout", default_value = "300")] + wait_timeout: u64, +} + +/// Entry point from `main`. Keeps `main.rs` thin — one call per group. +pub fn dispatch(workspace_id: &str, output: &str, command: IngestCommands) { + match command { + IngestCommands::NewConnection { + service, + config, + tables, + schema, + bucket_url, + format, + glob, + catalog_type, + wait, + database_id, + } => add_connection( + workspace_id, + output, + CreateArgs { + service, + config, + tables, + schema, + bucket_url, + format, + glob, + catalog_type, + database_id, + }, + &wait, + ), + IngestCommands::ListConnections { all } => list_connections(workspace_id, output, all), + IngestCommands::Connectors { name, output } => { + catalog_list(workspace_id, name.as_deref(), &output) + } + IngestCommands::NewImport { + sql, + all, + source, + database_id, + wait, + } => new_import(workspace_id, output, sql, all, source, database_id, &wait), + IngestCommands::ListImports => list_imports(workspace_id, output), + IngestCommands::TriggerImport { id, wait_timeout } => { + trigger_import(workspace_id, output, &id, wait_timeout) + } + } +} + +// --- new: add a connection (wizard or flag-driven) ----------------------- + +/// `ingest new-connection`. The guided wizard runs when no connector was named and we're +/// on a terminal; otherwise (a `--service` was given, or a non-TTY / `--no-input` +/// run) it's flag-driven and needs `--service`. +fn add_connection(workspace_id: &str, output: &str, args: CreateArgs, wait: &WaitArgs) { + if args.service.is_none() && util::is_interactive() { + wizard(workspace_id, output, wait); + } else { + add_from_flags(workspace_id, output, args, wait); + } +} + +fn wizard(workspace_id: &str, output: &str, wait: &WaitArgs) { + let client = IngestClient::new(workspace_id); + let entries = fetch_catalog(&client); + let entry = select_connector(&entries); + + let mut req = match entry.family.as_str() { + "sql" => build_sql_interactive(&entry), + "filesystem" => build_filesystem_interactive(), + "iceberg" => build_iceberg_interactive(&entry), + _ => build_rest_interactive(&entry), + }; + // Adding a connection discovers the schema only — never loads data. + req.validate_only = true; + run_source(&client, output, wait.no_wait, wait.wait_timeout, req); +} + +/// Present the catalog as a filterable menu and return the chosen entry. +/// Families are grouped (generic sql/filesystem/iceberg first, then the REST +/// services) and inquire's typeahead narrows the ~150 entries as the user types. +fn select_connector(entries: &[ConnectorEntry]) -> ConnectorEntry { + let sorted = sorted_for_display(entries); + let labels: Vec = sorted + .iter() + .map(|c| { + if c.description.is_empty() { + format!("{} ({})", c.name, c.family) + } else { + format!("{} ({}) — {}", c.name, c.family, c.description) + } + }) + .collect(); + let selected = Select::new("Source:", labels.clone()) + .with_page_size(15) + .prompt() + .unwrap_or_else(|_| std::process::exit(0)); + let idx = labels.iter().position(|l| l == &selected).unwrap(); + sorted[idx].clone() +} + +fn family_rank(family: &str) -> u8 { + match family { + "sql" => 0, + "filesystem" => 1, + "iceberg" => 2, + _ => 3, // rest services + } +} + +/// Sort the catalog for display: generic families (sql, filesystem, iceberg) +/// first, then the REST services, each group alphabetical. Redundant SQL +/// dialect aliases are collapsed at the source (the dlthubworker catalog), not +/// here. +fn sorted_for_display(entries: &[ConnectorEntry]) -> Vec { + let mut sorted = entries.to_vec(); + sorted.sort_by(|a, b| { + family_rank(&a.family) + .cmp(&family_rank(&b.family)) + .then_with(|| a.name.cmp(&b.name)) + }); + sorted +} + +fn build_sql_interactive(entry: &ConnectorEntry) -> IngestRequest { + // `entry.name` is the dialect (postgres, mysql, …) — we know the shape, so + // prompt only the connection fields, defaulting the port from the dialect. + let mut m = serde_json::Map::new(); + m.insert("host".into(), ask_text("Host:").into()); + let default_port = default_port_for(&entry.name); + let mut port_prompt = Text::new("Port:"); + if let Some(dp) = default_port { + port_prompt = port_prompt.with_default(dp); + } + let port = port_prompt + .prompt() + .unwrap_or_else(|_| std::process::exit(0)); + if let Ok(n) = port.trim().parse::() { + m.insert("port".into(), n.into()); + } + m.insert("username".into(), ask_text("User:").into()); + let pw = ask_secret("Password (blank if none):"); + if !pw.is_empty() { + m.insert("password".into(), pw.into()); + } + m.insert("database".into(), ask_text("Database:").into()); + + let schema = optional(None, "Schema (blank for all):"); + let tables = prompt_list("Tables (comma-separated, blank for all):"); + + IngestRequest { + family: "sql".into(), + connector_type: Some(entry.name.clone()), + credentials: serde_json::Value::Object(m), + schema, + table_names: tables, + ..Default::default() + } +} + +fn build_filesystem_interactive() -> IngestRequest { + let bucket_url = ask_text("Bucket URL (e.g. s3://bucket/prefix):"); + let format = select_optional("File format:", &["parquet", "csv", "jsonl"]); + let glob = optional(None, "File glob (e.g. **/*.parquet, blank for all):"); + IngestRequest { + family: "filesystem".into(), + credentials: filesystem_credentials(None), + bucket_url: Some(bucket_url), + file_glob: glob, + file_format: format, + ..Default::default() + } +} + +fn build_iceberg_interactive(entry: &ConnectorEntry) -> IngestRequest { + let catalog_type = optional_default("Catalog type:", "rest"); + let tables = prompt_list("Tables (namespace.table, comma-separated):"); + IngestRequest { + family: "iceberg".into(), + catalog_name: Some(entry.name.clone()), + catalog_type, + catalog_config: Some(iceberg_catalog_config(None)), + tables, + ..Default::default() + } +} + +/// REST family. A cataloged service carries a `template` with everything but +/// the secrets filled in — walk it, prompt only the `` tokens, and +/// send the result. The bare `rest` entry (no template) falls back to building +/// a minimal config interactively. +fn build_rest_interactive(entry: &ConnectorEntry) -> IngestRequest { + match &entry.template { + Some(template) => { + let filled = fill_template(template); + IngestRequest { + family: "rest".into(), + connector_type: Some(entry.name.clone()), + rest_config: Some(filled), + ..Default::default() + } + } + None => { + let name = ask_text("Connection name (names the connection for `ingest new-import`):"); + IngestRequest { + family: "rest".into(), + connector_type: Some(name), + rest_config: Some(rest_config(None)), + ..Default::default() + } + } + } +} + +// --- template placeholder filling ---------------------------------------- + +/// Collect the distinct `` tokens in a template, in first-seen +/// order. A placeholder is a whole string value wrapped in angle brackets +/// (``) — the shape the connector catalog uses. +fn collect_placeholders(v: &serde_json::Value, out: &mut Vec) { + match v { + serde_json::Value::String(s) if is_placeholder(s) => { + if !out.contains(s) { + out.push(s.clone()); + } + } + serde_json::Value::Array(a) => a.iter().for_each(|i| collect_placeholders(i, out)), + serde_json::Value::Object(m) => m.values().for_each(|i| collect_placeholders(i, out)), + _ => {} + } +} + +fn is_placeholder(s: &str) -> bool { + s.len() > 2 && s.starts_with('<') && s.ends_with('>') +} + +/// A placeholder names a secret when its token looks like one — prompt those +/// with hidden input. +fn is_secret_placeholder(token: &str) -> bool { + let t = token.to_ascii_uppercase(); + ["SECRET", "TOKEN", "KEY", "PASSWORD", "PASS"] + .iter() + .any(|needle| t.contains(needle)) +} + +/// Prompt for each placeholder in `template` (secrets hidden) and substitute +/// every occurrence, returning the filled config. +fn fill_template(template: &serde_json::Value) -> serde_json::Value { + let mut tokens = Vec::new(); + collect_placeholders(template, &mut tokens); + let mut filled = template.clone(); + for token in tokens { + let label = format!( + "{}:", + token + .trim_matches(|c| c == '<' || c == '>') + .replace('_', " ") + .to_lowercase() + ); + let value = if is_secret_placeholder(&token) { + ask_secret(&label) + } else { + ask_text(&label) + }; + substitute_placeholder(&mut filled, &token, &value); + } + filled +} + +fn substitute_placeholder(v: &mut serde_json::Value, token: &str, value: &str) { + match v { + serde_json::Value::String(s) if s == token => { + *v = serde_json::Value::String(value.to_string()); + } + serde_json::Value::Array(a) => a + .iter_mut() + .for_each(|i| substitute_placeholder(i, token, value)), + serde_json::Value::Object(m) => m + .values_mut() + .for_each(|i| substitute_placeholder(i, token, value)), + _ => {} + } +} + +// --- new (flag-driven): non-interactive add ------------------------------ + +struct CreateArgs { + service: Option, + config: Option, + tables: Vec, + schema: Option, + bucket_url: Option, + format: Option, + glob: Option, + catalog_type: Option, + database_id: Option, +} + +fn add_from_flags(workspace_id: &str, output: &str, args: CreateArgs, wait: &WaitArgs) { + let Some(service) = args.service.as_deref() else { + eprintln!( + "error: --service is required to add a connection non-interactively. \ + Browse connectors with 'hotdata ingest connectors', or run \ + 'hotdata ingest new-connection' in a terminal for the guided wizard." + ); + std::process::exit(1); + }; + let client = IngestClient::new(workspace_id); + let entries = fetch_catalog(&client); + let entry = entries + .iter() + .find(|c| c.name == service) + .cloned() + .unwrap_or_else(|| { + eprintln!("error: unknown connector '{service}'. Run 'hotdata ingest connectors'."); + std::process::exit(1); + }); + + let config = args.config.as_deref().map(parse_config_arg); + let mut req = match entry.family.as_str() { + "sql" => IngestRequest { + family: "sql".into(), + connector_type: Some(entry.name.clone()), + credentials: config.unwrap_or_else(|| { + fail("sql connectors need --config with credentials (a connection_string or host/user/…)") + }), + schema: args.schema, + table_names: args.tables, + ..Default::default() + }, + "filesystem" => { + let bucket_url = args.bucket_url.unwrap_or_else(|| fail("filesystem needs --bucket-url")); + IngestRequest { + family: "filesystem".into(), + credentials: config.unwrap_or(serde_json::Value::Null), + bucket_url: Some(bucket_url), + file_glob: args.glob, + file_format: args.format, + ..Default::default() + } + } + "iceberg" => IngestRequest { + family: "iceberg".into(), + catalog_name: Some(entry.name.clone()), + catalog_type: args.catalog_type.or_else(|| Some("rest".into())), + catalog_config: Some(config.unwrap_or_else(|| fail("iceberg needs --config with the catalog config"))), + tables: args.tables, + ..Default::default() + }, + _ => { + // REST: an explicit --config wins; otherwise the catalog template, + // which only works untouched for keyless services. + let rest_config = config.or_else(|| entry.template.clone()).unwrap_or_else(|| { + fail("this REST connector needs --config with a rest_config") + }); + let mut leftover = Vec::new(); + collect_placeholders(&rest_config, &mut leftover); + if !leftover.is_empty() { + fail(&format!( + "connector '{service}' needs secrets ({}) — pass a filled --config, or use 'hotdata ingest new-connection'", + leftover.join(", ") + )); + } + IngestRequest { + family: "rest".into(), + connector_type: Some(entry.name.clone()), + rest_config: Some(rest_config), + ..Default::default() + } + } + }; + // Adding a connection discovers the schema only — never loads data. + req.validate_only = true; + req.database_id = args.database_id; + run_source(&client, output, wait.no_wait, wait.wait_timeout, req); +} + +fn catalog_list(workspace_id: &str, filter: Option<&str>, output: &str) { + let client = IngestClient::new(workspace_id); + let mut entries = fetch_catalog(&client); + if let Some(f) = filter { + let f = f.to_lowercase(); + entries.retain(|c| c.name.to_lowercase().contains(&f)); + } + let entries = sorted_for_display(&entries); + // "active" = a connector this workspace has already added (it appears in + // the onboarded-source registry). Best-effort: if the registry read fails + // the catalog still shows, just without active marks. + let active = active_source_names(&client); + let is_active = |name: &str| active.contains(name); + + match output { + "json" | "yaml" => { + let v: Vec<_> = entries + .iter() + .map(|c| { + serde_json::json!({ + "name": c.name, + "family": c.family, + "active": is_active(&c.name), + "description": c.description, + }) + }) + .collect(); + if output == "yaml" { + print!("{}", serde_yaml::to_string(&v).unwrap()); + } else { + println!("{}", serde_json::to_string_pretty(&v).unwrap()); + } + } + _ => { + let rows: Vec> = entries + .iter() + .map(|c| { + let status = if is_active(&c.name) { "active" } else { "" }; + vec![ + c.name.clone(), + c.family.clone(), + status.to_string(), + c.description.clone(), + ] + }) + .collect(); + crate::output::table::print(&["NAME", "FAMILY", "STATUS", "DESCRIPTION"], &rows); + } + } +} + +/// Connector names this workspace has onboarded, from the sources registry. +/// Used to flag which catalog connectors are active. Best-effort — an +/// unavailable registry yields an empty set. +fn active_source_names(client: &IngestClient) -> std::collections::HashSet { + client + .list_sources(false) + .map(|r| { + r.sources + .into_iter() + .filter_map(|s| s.connector_type) + .collect() + }) + .unwrap_or_default() +} + +// --- new-import: SQL front-door ------------------------------------------- + +fn new_import( + workspace_id: &str, + output: &str, + sql: Option, + all: bool, + source: Option, + database_id: Option, + wait: &WaitArgs, +) { + // A 32-char hex --source is a connection id (pins resolution); anything + // else is a connector name (the FROM target). + let (pin, name) = match source { + Some(s) if looks_like_ingest_id(&s) => (Some(s), None), + Some(s) => (None, Some(s)), + None => (None, None), + }; + + let client = IngestClient::new(workspace_id); + // `--all` against a pinned id needs the connector name for FROM — resolve + // it from the sources registry, only when actually needed. + let pinned_name = (all && name.is_none()) + .then(|| { + pin.as_deref() + .and_then(|p| pinned_connector_name(&client, p)) + }) + .flatten(); + let query = build_import_query(sql, all, name.as_deref(), pinned_name.as_deref()) + .unwrap_or_else(|msg| fail(msg)); + let req = QueryToIngest { + query, + source_ingest_id: pin, + database_id, + }; + let spinner = util::spinner("submitting import…"); + // The by-name FROM lookup reads control-store snapshots that lag writes, so + // an import right after onboarding can 404 briefly — retry a few times. + let mut ack = client.create_query(&req); + for _ in 0..3 { + match &ack { + Err(crate::client::ingest::IngestError::Http { status: 404, .. }) => { + spinner.set_message("waiting for the source to register…"); + std::thread::sleep(POLL_INTERVAL); + ack = client.create_query(&req); + } + _ => break, + } + } + let ack = ack.unwrap_or_else(|e| { + spinner.finish_and_clear(); + e.exit() + }); + spinner.finish_and_clear(); + if wait.no_wait { + render_ack(&ack, output); + return; + } + let st = poll_ingest( + &client, + &ack.ingest_id, + wait.wait_timeout, + "importing", + true, + ); + render_done(&st, output); +} + +fn looks_like_ingest_id(s: &str) -> bool { + s.len() == 32 && s.chars().all(|c| c.is_ascii_hexdigit()) +} + +/// The SQL a `new-import` runs: explicit SQL wins; `--all` becomes a full +/// `SELECT *` against the connection's name (`pinned_name` when --source was +/// a connection id). Errors are messages for `fail`. +fn build_import_query( + sql: Option, + all: bool, + name: Option<&str>, + pinned_name: Option<&str>, +) -> Result { + match (sql, all) { + (Some(q), _) => Ok(q), // clap rejects sql + --all together + (None, true) => name + .or(pinned_name) + .map(|n| format!("SELECT * FROM {n}")) + .ok_or("--all needs --source (a connector name or connection id)"), + (None, false) => Err( + "provide SQL (SELECT … FROM …), or --all to import \ + everything from --source", + ), + } +} + +/// Resolve a pinned connection id to its connector name (the FROM target) +/// via the sources registry. +fn pinned_connector_name(client: &IngestClient, ingest_id: &str) -> Option { + client + .list_sources(true) + .ok()? + .sources + .into_iter() + .find(|s| s.ingest_id == ingest_id) + .and_then(|s| s.connector_type) +} + +// --- trigger-import -------------------------------------------------------- + +fn trigger_import(workspace_id: &str, output: &str, id: &str, wait_timeout: u64) { + let client = IngestClient::new(workspace_id); + let spinner = util::spinner("requesting re-run…"); + // The worker resets the ingest to pending and fires the drain (409 if one + // is already mid-flight — its message says to retry shortly). + let ack = client.rerun(id).unwrap_or_else(|e| { + spinner.finish_and_clear(); + e.exit() + }); + spinner.finish_and_clear(); + // The rerun already drained; poll only. + let st = poll_ingest(&client, &ack.ingest_id, wait_timeout, "re-running", false); + render_done(&st, output); +} + +// --- list-connections ------------------------------------------------------ + +fn list_connections(workspace_id: &str, output: &str, all: bool) { + let client = IngestClient::new(workspace_id); + let spinner = util::spinner("loading connections…"); + let resp = client.list_sources(all).unwrap_or_else(|e| { + spinner.finish_and_clear(); + e.exit() + }); + spinner.finish_and_clear(); + + match output { + "json" => println!("{}", serde_json::to_string_pretty(&resp.sources).unwrap()), + "yaml" => print!("{}", serde_yaml::to_string(&resp.sources).unwrap()), + _ => { + use crossterm::style::Stylize; + if resp.sources.is_empty() { + eprintln!( + "{}", + "No connections yet. Add one with 'hotdata ingest new-connection'.".dark_grey() + ); + return; + } + let mut headers = vec!["NAME", "FAMILY", "STATUS", "CREATED", "CONNECTION ID"]; + if all { + headers.push("ACTIVE"); + } + let rows: Vec> = resp + .sources + .iter() + .map(|s| { + let mut row = vec![ + s.connector_type.clone().unwrap_or_else(|| "-".into()), + s.family.clone().unwrap_or_default(), + s.status.clone(), + date_only(s.created_at.as_deref()), + s.ingest_id.clone(), + ]; + if all { + row.push(if s.active { + "yes".into() + } else { + String::new() + }); + } + row + }) + .collect(); + crate::output::table::print(&headers, &rows); + } + } +} + +// --- list-imports ---------------------------------------------------------- + +fn list_imports(workspace_id: &str, output: &str) { + let client = IngestClient::new(workspace_id); + let spinner = util::spinner("loading imports…"); + let resp = client.list_queries().unwrap_or_else(|e| { + spinner.finish_and_clear(); + e.exit() + }); + spinner.finish_and_clear(); + + match output { + "json" => println!("{}", serde_json::to_string_pretty(&resp.queries).unwrap()), + "yaml" => print!("{}", serde_yaml::to_string(&resp.queries).unwrap()), + _ => { + use crossterm::style::Stylize; + if resp.queries.is_empty() { + eprintln!( + "{}", + "No imports yet. Create one with 'hotdata ingest new-import \ + --source --all' (or pass SQL)." + .dark_grey() + ); + return; + } + let rows: Vec> = resp + .queries + .iter() + .map(|q| { + vec![ + q.connector_type.clone().unwrap_or_else(|| "-".into()), + q.query.clone().unwrap_or_default(), + q.status.clone(), + date_only(q.created_at.as_deref()), + q.ingest_id.clone(), + q.database_id.clone().unwrap_or_else(|| "-".into()), + ] + }) + .collect(); + crate::output::table::print( + &[ + "SOURCE", + "SQL", + "STATUS", + "CREATED", + "IMPORT ID", + "DATABASE", + ], + &rows, + ); + } + } +} + +/// Date part of an ISO timestamp for table display +/// (`2026-07-08T10:12:00+00:00` → `2026-07-08`). +fn date_only(ts: Option<&str>) -> String { + ts.and_then(|t| t.split('T').next()) + .unwrap_or("-") + .to_string() +} + +// --- shared run + poll ---------------------------------------------------- + +fn run_source( + client: &IngestClient, + output: &str, + no_wait: bool, + wait_timeout: u64, + req: IngestRequest, +) { + // The first add in a workspace deploys the dlt runtime (~15-30s); later + // ones hash-short-circuit. The HTTP client allows 300s. + let spinner = util::spinner("adding connection… (the first one in a workspace takes ~30s)"); + let ack = client.create_source(&req).unwrap_or_else(|e| { + spinner.finish_and_clear(); + e.exit() + }); + spinner.finish_and_clear(); + if no_wait { + render_ack(&ack, output); + return; + } + let st = poll_ingest( + client, + &ack.ingest_id, + wait_timeout, + "discovering schema", + true, + ); + render_connection_added(client, &st, output); +} + +/// Poll the ingest to a terminal state, returning the final (done) status for +/// the caller to render. `kick_drain` fires the drain up front (enqueue paths; +/// `trigger-import` skips it — the rerun already drained). Every non-terminal +/// status is shown live in the spinner (stage + detail); exits the process on +/// failure (1) or timeout (2). `verb` labels the spinner. +fn poll_ingest( + client: &IngestClient, + ingest_id: &str, + timeout_secs: u64, + verb: &str, + kick_drain: bool, +) -> crate::client::ingest::JobStatus { + if kick_drain { + let _ = client.drain(); + } + let mut last_drain = Instant::now(); + + let spinner = util::spinner(&format!("{verb}…")); + let deadline = Instant::now() + Duration::from_secs(timeout_secs); + loop { + let st = client.job_status(ingest_id).unwrap_or_else(|e| { + spinner.finish_and_clear(); + e.exit() + }); + match st.status.as_str() { + "done" => { + spinner.finish_and_clear(); + return st; + } + "failed" => { + spinner.finish_and_clear(); + use crossterm::style::Stylize; + let detail = st.detail.as_deref().unwrap_or("unknown error"); + eprintln!("{}", format!("ingest failed: {detail}").red()); + if detail.contains("Forbidden") { + eprintln!( + "{}", + "Forbidden at load time usually means a database-scoped API token — \ + ingest needs a regular workspace API token." + .dark_grey() + ); + } + std::process::exit(1); + } + // Any other status is in-progress. Surface it live rather than + // treating it as terminal — the worker reports stage-level states + // (e.g. extracting / normalizing / loading) and free-text detail as + // the pipeline advances, and the CLI shouldn't need to know the set. + stage => { + spinner.set_message(progress_message(verb, stage, st.detail.as_deref())); + // Re-kick the drain if the job never left `pending` — the + // enqueue-time drain can race the request row (harmless double + // drain; loads are replace-mode). + if stage == "pending" { + if last_drain.elapsed() > DRAIN_REKICK_AFTER { + let _ = client.drain(); + last_drain = Instant::now(); + } + } else { + last_drain = Instant::now(); // progress seen — reset the clock + } + } + } + if Instant::now() > deadline { + spinner.finish_and_clear(); + use crossterm::style::Stylize; + eprintln!("{}", "ingest timed out".red()); + eprintln!( + "{}", + format!( + "Check progress with: hotdata ingest list-imports (or list-connections) — \ + or re-kick it: hotdata ingest trigger-import {ingest_id}" + ) + .dark_grey() + ); + std::process::exit(2); + } + std::thread::sleep(POLL_INTERVAL); + } +} + +/// Spinner text for an in-progress poll: the verb plus the worker's current +/// stage, and its free-text detail when present (e.g. row counts / table name). +fn progress_message(verb: &str, stage: &str, detail: Option<&str>) -> String { + match detail.map(str::trim).filter(|d| !d.is_empty()) { + Some(d) => format!("{verb}… {stage} — {d}"), + None => format!("{verb}… {stage}"), + } +} + +// --- rendering ----------------------------------------------------------- + +fn render_ack(ack: &IngestAck, output: &str) { + match output { + "json" => println!("{}", serde_json::to_string_pretty(ack).unwrap()), + "yaml" => print!("{}", serde_yaml::to_string(ack).unwrap()), + _ => { + use crossterm::style::Stylize; + let label = |l: &str| format!("{:<14}", l).dark_grey().to_string(); + println!("{}{}", label("ingest id:"), ack.ingest_id); + println!("{}{}", label("database:"), ack.database_id); + println!("{}{}", label("status:"), ack.status.as_str().yellow()); + println!( + "{}", + format!( + "Track it with: hotdata ingest trigger-import {}", + ack.ingest_id + ) + .dark_grey() + ); + } + } +} + +fn render_done(st: &crate::client::ingest::JobStatus, output: &str) { + match output { + "json" => println!("{}", serde_json::to_string_pretty(st).unwrap()), + "yaml" => print!("{}", serde_yaml::to_string(st).unwrap()), + _ => { + use crossterm::style::Stylize; + let db = st.database_id.as_deref().unwrap_or("-"); + println!("{} → {}", "done".green(), db); + println!( + "{}", + format!("Query it: hotdata query --database {db} \"SELECT * FROM …\"").dark_grey() + ); + } + } +} + +/// Render the result of adding a connection: the discovered schema (tables + +/// columns), fetched from the schema-preview the metadata-only run landed. No +/// data was loaded — the closing hint points at `import` for that. +fn render_connection_added( + client: &IngestClient, + st: &crate::client::ingest::JobStatus, + output: &str, +) { + let db = st.database_id.as_deref().unwrap_or(""); + let schema = (!db.is_empty()).then(|| client.schema(db).ok()).flatten(); + let tables = schema + .as_ref() + .and_then(|s| s.get("tables")) + .and_then(|t| t.as_object()); + + match output { + "json" | "yaml" => { + let mut v = serde_json::to_value(st).unwrap_or_default(); + if let serde_json::Value::Object(ref mut m) = v { + m.insert( + "tables".into(), + schema + .as_ref() + .and_then(|s| s.get("tables").cloned()) + .unwrap_or(serde_json::Value::Null), + ); + } + if output == "yaml" { + print!("{}", serde_yaml::to_string(&v).unwrap()); + } else { + println!("{}", serde_json::to_string_pretty(&v).unwrap()); + } + } + _ => { + use crossterm::style::Stylize; + let source = st.connector_type.as_deref().unwrap_or("source"); + println!("{} {}", "connection added".green(), source.dark_grey()); + if !db.is_empty() { + println!("{}{}", format!("{:<12}", "database:").dark_grey(), db); + } + match tables { + Some(t) if !t.is_empty() => { + println!( + "{}", + format!("discovered {} table(s):", t.len()).dark_grey() + ); + for (name, cols) in t { + let names: Vec<&str> = cols.as_array().map_or(Vec::new(), |a| { + a.iter().filter_map(|c| c.as_str()).collect() + }); + println!( + " {} {}", + name.as_str().cyan(), + names.join(", ").dark_grey() + ); + } + } + _ => println!("{}", "no tables discovered".dark_grey()), + } + println!( + "{}", + format!( + "Import data with: hotdata ingest new-import --source {source} --all (or SQL)" + ) + .dark_grey() + ); + } + } +} + +// --- catalog fetch -------------------------------------------------------- + +fn fetch_catalog(client: &IngestClient) -> Vec { + let spinner = util::spinner("loading connectors…"); + let resp = client.connectors().unwrap_or_else(|e| { + spinner.finish_and_clear(); + e.exit() + }); + spinner.finish_and_clear(); + resp.connectors +} + +// --- input helpers -------------------------------------------------------- +// +// All prompting is TTY-gated by callers (the wizard errors early on a non-TTY). +// `inquire` returns Err on Ctrl-C/ESC — exit cleanly, matching +// `connections/interactive`. + +fn ask_text(label: &str) -> String { + Text::new(label) + .prompt() + .unwrap_or_else(|_| std::process::exit(0)) +} + +fn ask_secret(label: &str) -> String { + Password::new(label) + .without_confirmation() + .prompt() + .unwrap_or_else(|_| std::process::exit(0)) +} + +/// Optional value: the flag, else a prompt (TTY) whose blank answer = none. +fn optional(flag: Option, label: &str) -> Option { + if flag.is_some() { + return flag; + } + if util::is_interactive() { + let v = ask_text(label); + return (!v.trim().is_empty()).then_some(v); + } + None +} + +/// Optional value with a default; non-interactive falls back to the default. +fn optional_default(label: &str, default: &str) -> Option { + if !util::is_interactive() { + return Some(default.to_string()); + } + let v = Text::new(label) + .with_default(default) + .prompt() + .unwrap_or_else(|_| std::process::exit(0)); + (!v.trim().is_empty()).then_some(v) +} + +/// Select one of `options` (TTY only; None otherwise or on ESC). +fn select_optional(label: &str, options: &[&str]) -> Option { + if !util::is_interactive() { + return None; + } + Select::new(label, options.to_vec()) + .prompt() + .ok() + .map(|s| s.to_string()) +} + +/// Prompt for a comma-separated list (TTY only; empty otherwise). +fn prompt_list(label: &str) -> Vec { + if !util::is_interactive() { + return Vec::new(); + } + ask_text(label) + .split(',') + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect() +} + +/// Bare `rest` family: build a minimal dlt rest_api config interactively +/// (base_url + optional bearer token + resource paths). +fn rest_config(config: Option<&str>) -> serde_json::Value { + if let Some(c) = config { + return parse_config_arg(c); + } + let base_url = ask_text("REST base URL:"); + let token = ask_secret("Bearer token (blank if none):"); + let resources: Vec = ask_text("Resource paths (comma-separated):") + .split(',') + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + .map(serde_json::Value::from) + .collect(); + let mut client = serde_json::Map::new(); + client.insert("base_url".into(), base_url.into()); + if !token.is_empty() { + client.insert( + "auth".into(), + serde_json::json!({ "type": "bearer", "token": token }), + ); + } + serde_json::json!({ "client": client, "resources": resources }) +} + +/// Filesystem object-store credentials: prompt for S3-style creds (all optional +/// — blank access key = public bucket, no creds sent). +fn filesystem_credentials(config: Option<&str>) -> serde_json::Value { + if let Some(c) = config { + return parse_config_arg(c); + } + if !util::is_interactive() { + return serde_json::Value::Null; + } + let key = ask_text("Object-store access key id (blank for a public bucket):"); + if key.trim().is_empty() { + return serde_json::Value::Null; + } + let mut m = serde_json::Map::new(); + m.insert("aws_access_key_id".into(), key.into()); + m.insert( + "aws_secret_access_key".into(), + ask_secret("Secret access key:").into(), + ); + if let Some(ep) = optional(None, "Endpoint URL (S3-compatible; blank for AWS):") { + m.insert("endpoint_url".into(), ep.into()); + } + if let Some(region) = optional(None, "Region (blank for default):") { + m.insert("region_name".into(), region.into()); + } + serde_json::Value::Object(m) +} + +/// Iceberg catalog connection config: prompt for catalog URI + warehouse + +/// token + namespace. +fn iceberg_catalog_config(config: Option<&str>) -> serde_json::Value { + if let Some(c) = config { + return parse_config_arg(c); + } + let mut m = serde_json::Map::new(); + m.insert("catalog_uri".into(), ask_text("Catalog URI:").into()); + if let Some(w) = optional(None, "Warehouse (blank if none):") { + m.insert("warehouse".into(), w.into()); + } + let token = ask_secret("Catalog token (blank if none):"); + if !token.is_empty() { + m.insert("token".into(), token.into()); + } + if let Some(ns) = optional(None, "Namespace (blank if none):") { + m.insert("namespace".into(), ns.into()); + } + serde_json::Value::Object(m) +} + +/// Parse a `--config` argument: inline JSON, `@file.json`, or `@-` for stdin. +fn parse_config_arg(arg: &str) -> serde_json::Value { + use std::io::Read; + let raw = if arg == "@-" { + let mut s = String::new(); + std::io::stdin() + .read_to_string(&mut s) + .unwrap_or_else(|e| fail(&format!("--config reading stdin: {e}"))); + s + } else if let Some(path) = arg.strip_prefix('@') { + std::fs::read_to_string(path) + .unwrap_or_else(|e| fail(&format!("--config reading {path}: {e}"))) + } else { + arg.to_string() + }; + serde_json::from_str(&raw).unwrap_or_else(|e| fail(&format!("--config invalid JSON: {e}"))) +} + +fn default_port_for(dialect: &str) -> Option<&'static str> { + match dialect { + "postgres" | "postgresql" | "redshift" => Some("5432"), + "mysql" | "mariadb" => Some("3306"), + "mssql" | "sqlserver" => Some("1433"), + "oracle" => Some("1521"), + _ => None, + } +} + +fn fail(msg: &str) -> ! { + use crossterm::style::Stylize; + eprintln!("{}", format!("error: {msg}").red()); + std::process::exit(1); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn placeholders_collected_in_order_without_dupes() { + let t = serde_json::json!({ + "client": { + "base_url": "https://x/api", + "auth": {"client_id": "", "client_secret": ""} + }, + "resources": ["", "teams"] + }); + let mut got = Vec::new(); + collect_placeholders(&t, &mut got); + assert_eq!(got, vec!["", ""]); + } + + #[test] + fn keyless_template_has_no_placeholders() { + let t = serde_json::json!({ + "client": {"base_url": "https://blockstream.info/api"}, + "resources": [{"name": "blocks", "endpoint": {"path": "blocks"}}] + }); + let mut got = Vec::new(); + collect_placeholders(&t, &mut got); + assert!(got.is_empty()); + } + + #[test] + fn substitute_replaces_every_occurrence_only_for_that_token() { + let mut t = serde_json::json!({ + "a": "", "b": {"c": ""}, "d": "", "e": "literal" + }); + substitute_placeholder(&mut t, "", "filled"); + assert_eq!(t["a"], "filled"); + assert_eq!(t["b"]["c"], "filled"); + assert_eq!(t["d"], ""); // untouched + assert_eq!(t["e"], "literal"); + } + + #[test] + fn secret_placeholders_detected_by_token_name() { + assert!(is_secret_placeholder("")); + assert!(is_secret_placeholder("")); + assert!(is_secret_placeholder("")); + assert!(is_secret_placeholder("")); + assert!(!is_secret_placeholder("")); + assert!(!is_secret_placeholder("")); + } + + #[test] + fn ingest_id_recognized_only_for_32_hex() { + assert!(looks_like_ingest_id("6232a1694a1b4451957c053a56756ff7")); + assert!(!looks_like_ingest_id("bitcoin")); + assert!(!looks_like_ingest_id("6232a169")); // too short + assert!(!looks_like_ingest_id("zzzz1694a1b4451957c053a56756ffff")); // non-hex + } + + #[test] + fn import_query_explicit_sql_wins() { + let q = build_import_query( + Some("SELECT id FROM pg.orders LIMIT 5".into()), + false, + Some("pg"), + None, + ); + assert_eq!(q.unwrap(), "SELECT id FROM pg.orders LIMIT 5"); + } + + #[test] + fn import_query_all_uses_the_connection_name() { + // Named source. + assert_eq!( + build_import_query(None, true, Some("bitcoin"), None).unwrap(), + "SELECT * FROM bitcoin" + ); + // Pinned connection id — name resolved from the registry by the caller. + assert_eq!( + build_import_query(None, true, None, Some("postgres")).unwrap(), + "SELECT * FROM postgres" + ); + } + + #[test] + fn import_query_errors_without_sql_or_all() { + // Neither SQL nor --all: the two modes must be chosen explicitly. + let err = build_import_query(None, false, Some("bitcoin"), None).unwrap_err(); + assert!(err.contains("--all"), "got: {err}"); + // --all with no resolvable source name. + let err = build_import_query(None, true, None, None).unwrap_err(); + assert!(err.contains("--source"), "got: {err}"); + } + + #[test] + fn date_only_trims_iso_timestamps() { + assert_eq!(date_only(Some("2026-07-08T10:12:00+00:00")), "2026-07-08"); + assert_eq!(date_only(Some("2026-07-08")), "2026-07-08"); + assert_eq!(date_only(None), "-"); + } + + #[test] + fn family_rank_orders_generic_before_rest() { + assert!(family_rank("sql") < family_rank("rest")); + assert!(family_rank("filesystem") < family_rank("rest")); + assert!(family_rank("iceberg") < family_rank("rest")); + } + + #[test] + fn parse_config_accepts_inline_json() { + let v = parse_config_arg(r#"{"connection_string": "postgresql://u:p@h/db"}"#); + assert_eq!(v["connection_string"], "postgresql://u:p@h/db"); + } + + fn entry(name: &str, family: &str) -> ConnectorEntry { + ConnectorEntry { + name: name.into(), + family: family.into(), + description: String::new(), + auth: None, + template: None, + } + } + + #[test] + fn sorted_for_display_groups_generic_families_before_rest() { + let entries = vec![ + entry("stripe", "rest"), + entry("postgres", "sql"), + entry("filesystem", "filesystem"), + entry("aikido", "rest"), + entry("iceberg", "iceberg"), + ]; + let names: Vec = sorted_for_display(&entries) + .into_iter() + .map(|c| c.name) + .collect(); + // Generic families (sql, filesystem, iceberg) first, then rest A→Z. + assert_eq!( + names, + vec!["postgres", "filesystem", "iceberg", "aikido", "stripe"] + ); + } + + #[test] + fn progress_message_appends_detail_when_present() { + assert_eq!( + progress_message("importing", "running", None), + "importing… running" + ); + // Empty/whitespace detail is dropped, not shown as a dangling dash. + assert_eq!( + progress_message("importing", "pending", Some(" ")), + "importing… pending" + ); + assert_eq!( + progress_message("importing", "loading", Some("region (5 rows)")), + "importing… loading — region (5 rows)" + ); + } +} diff --git a/src/config.rs b/src/config.rs index 74cc20d..4bd9d43 100644 --- a/src/config.rs +++ b/src/config.rs @@ -253,20 +253,6 @@ pub fn clear_current_database(profile: &str, workspace_id: &str) -> Result<(), S write_config(&config_path, &content) } -pub fn resolve_workspace_id( - provided: Option, - profile_config: &ProfileConfig, -) -> Result { - if let Some(id) = provided { - return Ok(id); - } - profile_config - .workspaces - .first() - .map(|w| w.public_id.clone()) - .ok_or_else(|| "no workspace-id provided and no default workspace found. Run 'hotdata auth login' or specify --workspace-id.".to_string()) -} - /// Global API key override set via --api-key flag. /// Call `set_api_key_flag` once at startup; `load` picks it up automatically. static API_KEY_FLAG: std::sync::OnceLock = std::sync::OnceLock::new(); @@ -466,37 +452,4 @@ mod tests { "api_key must not appear in YAML, got:\n{yaml}" ); } - - #[test] - fn resolve_workspace_id_prefers_provided() { - let profile = ProfileConfig { - workspaces: vec![WorkspaceEntry { - public_id: "ws-1".into(), - name: "WS".into(), - }], - ..Default::default() - }; - let result = resolve_workspace_id(Some("explicit-id".into()), &profile).unwrap(); - assert_eq!(result, "explicit-id"); - } - - #[test] - fn resolve_workspace_id_falls_back_to_first() { - let profile = ProfileConfig { - workspaces: vec![WorkspaceEntry { - public_id: "ws-1".into(), - name: "WS".into(), - }], - ..Default::default() - }; - let result = resolve_workspace_id(None, &profile).unwrap(); - assert_eq!(result, "ws-1"); - } - - #[test] - fn resolve_workspace_id_errors_when_none() { - let profile = ProfileConfig::default(); - let result = resolve_workspace_id(None, &profile); - assert!(result.is_err()); - } } diff --git a/src/main.rs b/src/main.rs index dd55804..8c9a4c3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,6 +15,7 @@ use commands::context::{self, ContextCommands}; use commands::databases::{self, DatabaseTablesCommands, DatabasesCommands}; use commands::embedding_providers::{self, EmbeddingProvidersCommands}; use commands::indexes::{self, IndexesCommands}; +use commands::ingest; use commands::jobs::{self, JobsCommands}; use commands::queries::{self, QueriesCommands}; use commands::query::{self, QueryCommands}; @@ -66,40 +67,30 @@ fn resolve_workspace(provided: Option) -> String { let _ = ACTIVE_WORKSPACE_ID.set(ws.clone()); return ws; } - match config::load("default") { - Ok(profile) => { - // An api-key credential (e.g. a database API token, passed via - // --api-key / HOTDATA_API_KEY) is authorized for its own - // workspace(s), independent of the CLI session's saved default. - // When the caller didn't pin one and the token is scoped to exactly - // one workspace, target it — otherwise the gateway rejects the - // mismatched X-Workspace-Id with "workspace not allowed" before the - // request ever reaches the resource. - if provided.is_none() - && matches!( - profile.api_key_source, - config::ApiKeySource::Flag | config::ApiKeySource::Env - ) - { - let ids = credentials::api_key_workspace_ids(&profile); - if let [only] = ids.as_slice() { - let _ = ACTIVE_WORKSPACE_ID.set(only.clone()); - return only.clone(); - } - } - match config::resolve_workspace_id(provided, &profile) { - Ok(id) => { - let _ = ACTIVE_WORKSPACE_ID.set(id.clone()); - id - } - Err(e) => { - eprintln!("error: {e}"); - std::process::exit(1); - } - } + let profile = config::load("default").unwrap_or_else(|e| { + eprintln!("{e}"); + std::process::exit(1); + }); + // An explicit --workspace-id always wins. + if let Some(id) = provided { + let _ = ACTIVE_WORKSPACE_ID.set(id.clone()); + return id; + } + // Otherwise the profile's default, computed by the same helper `auth + // status` displays — so the reported workspace is the one commands hit. + // For an api-key credential that's its own authorized workspace (a database + // token's sole one, or the saved default when the key can reach it), not a + // possibly-different CLI-session cache. + match credentials::default_workspace_id(&profile) { + Some(id) => { + let _ = ACTIVE_WORKSPACE_ID.set(id.clone()); + id } - Err(e) => { - eprintln!("{e}"); + None => { + eprintln!( + "error: no workspace-id provided and no default workspace found. \ + Run 'hotdata auth login' or specify --workspace-id." + ); std::process::exit(1); } } @@ -587,6 +578,14 @@ fn main() { } } } + Commands::Ingest { + workspace_id, + output, + command, + } => { + let workspace_id = resolve_workspace(workspace_id); + ingest::dispatch(&workspace_id, &output, command); + } Commands::Indexes { workspace_id, command,