Merge pull request #206 from hotdata-dev/feat/ingest-commands

feat(ingest): `hotdata ingest` — connections and imports from external sources

Author: eddietejeda

README.md

@@ -73,6 +73,7 @@ API key priority (lowest to highest): config file → `HOTDATA_API_KEY` env var
 | `embedding-providers` | `list`, `get`, `create`, `update`, `delete` | Manage embedding providers used by vector indexes |
 | `results` | `list` | Retrieve stored query results |
 | `jobs` | `list` | Manage background jobs |
+| `ingest` | `new-connection`, `show-connection`, `list-connections`, `delete-connection`, `connectors`, `new-import`, `list-imports`, `trigger-import`, `status` | Pull data from external sources (databases, APIs, buckets, Iceberg) into managed databases |
 | `skills` | `install`, `status` | Manage the hotdata agent skill |
 
 ## Global options
@@ -288,6 +289,51 @@ hotdata results list [--workspace-id <id>] [--limit <n>] [--offset <n>] [--forma
 
 - Query results include a `result-id` in the table footer — use it to retrieve past results without re-running queries.
 
+## Ingest
+
+Pull data from external sources into managed databases. Two nouns: **connections** (onboarded, credentialed sources — schema discovered, no data loaded) and **imports** (managed databases materialized from a connection).
+
+```bash
+# Browse the connector catalog (SQL dialects, ~150 API services, buckets, Iceberg)
+hotdata ingest connectors postgres
+
+# Add a connection: validates credentials and discovers the schema (loads no data).
+# Run bare for a guided wizard, or non-interactively with --service + --config
+# (@file or @- keeps secrets out of argv):
+hotdata ingest new-connection --service postgres --config @conn.json --schema public
+# conn.json holds {"connection_string": "postgresql://user:pass@host/db"}
+
+# Files in S3/GCS/Azure buckets — public buckets need no credentials:
+hotdata ingest new-connection --service buckets --bucket-url s3://bucket/prefix --format parquet
+
+# Iceberg via a REST catalog — --table required; --config holds the catalog
+# properties (uri, warehouse, token/credential), passed through to the service:
+hotdata ingest new-connection --service iceberg --config @catalog.json --table ns.orders
+
+hotdata ingest list-connections            # each connection has its own id
+hotdata ingest show-connection <id>        # status + discovered tables/columns
+hotdata ingest delete-connection <id>      # registry + credentials + discovery DB
+                                           # (--keep-database keeps the DB)
+
+# Import data — returns immediately; track with `status`:
+hotdata ingest new-import "SELECT * FROM postgres.orders LIMIT 1000"
+hotdata ingest new-import --source postgres --all      # everything, no LIMIT
+hotdata ingest status <ingest-id> --wait               # attach until done/failed
+
+hotdata ingest list-imports                # the SQL behind each import + its database
+hotdata ingest trigger-import <ingest-id>  # re-run: refresh the same database from source
+```
+
+`status` exits `0` (done), `1` (failed), or `2` (still in flight) for scripting.
+Statuses stream through stages (`preparing`, `extracting`, `normalizing`,
+`loading`); treat any non-terminal value as in-flight. Secret-bearing configs
+can be passed as `--config @file.json` or `--config @-` (stdin) to keep them
+out of `argv`. Enqueue commands (`new-connection`, `new-import`,
+`trigger-import`) need a durable workspace API key (`--api-key` /
+`HOTDATA_API_KEY`, `hd_...`) — the ingest pipeline runs after a short-lived
+session token would expire. Once an import is `done`, its database works with
+all the regular commands (`query`, `databases`, `results`).
+
 ## Jobs
 
 ```sh

skills/hotdata/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: hotdata
-description: Use this skill when the user wants to run core hotdata CLI commands — auth, workspaces, connections, managed databases, tables, basic SQL query, database context (context:DATAMODEL), jobs, and skill install. Activate for "run hotdata", "list workspaces", "list connections", "create a connection", "list databases", "managed database", "load parquet", "list tables", "execute a query", "database context", "context:DATAMODEL", or general Hotdata CLI usage. For full-text/vector search and retrieval indexes use hotdata-search; for OLAP analytics, query history, stored results, and Chain materializations use hotdata-analytics; for geospatial/GIS use hotdata-geospatial.
+description: Use this skill when the user wants to run core hotdata CLI commands — auth, workspaces, connections, managed databases, tables, basic SQL query, database context (context:DATAMODEL), jobs, ingest (pull external data), and skill install. Activate for "run hotdata", "list workspaces", "list connections", "create a connection", "list databases", "managed database", "load parquet", "list tables", "execute a query", "database context", "context:DATAMODEL", "ingest", "import data from", "connect a data source", "connector", "pull data from postgres/mysql/an API/S3 buckets/Iceberg", or general Hotdata CLI usage. For full-text/vector search and retrieval indexes use hotdata-search; for OLAP analytics, query history, stored results, and Chain materializations use hotdata-analytics; for geospatial/GIS use hotdata-geospatial.
 version: 0.12.0
 ---
 
@@ -20,7 +20,7 @@ Install all skills with **`hotdata skills install`**. Load specialized skills on
 
 | Skill | Use for |
 |-------|---------|
-| **`hotdata`** (this file) | Auth, workspaces, connections, databases, tables, basic `query`, context, jobs |
+| **`hotdata`** (this file) | Auth, workspaces, connections, databases, tables, basic `query`, context, jobs, ingest |
 | **`hotdata-search`** | BM25, vector search, `hotdata search`, bm25/vector indexes, embedding providers |
 | **`hotdata-analytics`** | OLAP SQL, aggregations, query/results history, Chain materializations, sorted indexes |
 | **`hotdata-geospatial`** | PostGIS-style `ST_*`, WKB, spatial joins |
@@ -72,7 +72,7 @@ Catalog, skill decision tree, epic flows (onboard, chain, retrieval), and manage
 
 ## Available Commands
 
-Top-level subcommands (each detailed below): **`auth`**, **`query`**, **`workspaces`**, **`connections`**, **`databases`**, **`tables`**, **`skills`**, **`results`**, **`jobs`**, **`indexes`**, **`embedding-providers`**, **`search`**, **`queries`**, **`context`**, **`usage`**, **`completions`**, **`upgrade`**. Search, indexes (bm25/vector), and embedding providers are documented in **`hotdata-search`**; query history, results, Chain, and OLAP patterns in **`hotdata-analytics`**.
+Top-level subcommands (each detailed below): **`auth`**, **`query`**, **`workspaces`**, **`connections`**, **`databases`**, **`tables`**, **`skills`**, **`results`**, **`jobs`**, **`ingest`**, **`indexes`**, **`embedding-providers`**, **`search`**, **`queries`**, **`context`**, **`usage`**, **`completions`**, **`upgrade`**. Search, indexes (bm25/vector), and embedding providers are documented in **`hotdata-search`**; query history, results, Chain, and OLAP patterns in **`hotdata-analytics`**.
 
 Global CLI options: **`--api-key`**, **`-v` / `--version`**, **`-h` / `--help`**, **`--no-input`** (disable interactive prompts; commands that require input will error instead — useful in CI or non-TTY environments). Hidden developer flag: **`--debug`** (verbose HTTP logs).
 
@@ -295,6 +295,56 @@ hotdata jobs <job_id> [--workspace-id <workspace_id>] [--output table|json|yaml]
 - `--status`: `pending`, `running`, `succeeded`, `partially_succeeded`, `failed`.
 - Use `hotdata jobs <job_id>` to inspect a specific job's status, error, and result.
 
+### Ingest external data (`ingest`)
+
+Pull data from external sources (SQL databases, APIs, S3/GCS/Azure buckets, Iceberg catalogs) into managed databases. Two nouns: **connections** (onboarded, credentialed sources — schema discovered, no data loaded; each has its own id) and **imports** (managed databases materialized from a connection). A connection registers under the connector name you added (`postgres`, `buckets`, `iceberg`, a named API service) — that name is the `FROM` target for imports.
+
+Enqueue commands (`new-connection`, `new-import`, `trigger-import`, `delete-connection`) **require a workspace API key** (`HOTDATA_API_KEY` / `--api-key`, `hd_...`) — a login session is not enough, because the ingest pipeline runs after a short-lived session token would expire.
+
+```bash
+hotdata ingest connectors [filter]     # browse the catalog; "active" = already added.
+                                       # SQL dialects + ~150 named API services + generic
+                                       # buckets / iceberg / api (bring-your-own config)
+hotdata ingest new-connection --service postgres --config @conn.json --schema public
+# conn.json holds {"connection_string": "postgresql://user:pass@host/db"}
+# --config accepts @file.json, @- (stdin), or inline JSON — keep secrets out of argv.
+# Validates credentials + discovers the schema; loads NO data. Blocks until done
+# and prints the discovered tables (--no-wait to return immediately).
+
+hotdata ingest new-connection --service buckets --bucket-url s3://bucket/prefix --format parquet
+# Files in S3/GCS/Azure buckets (csv, jsonl, parquet); --glob narrows the match.
+# Public buckets need no credentials; private ones take --config @creds.json
+# ({"aws_access_key_id": …, "aws_secret_access_key": …, "endpoint_url": …}).
+
+hotdata ingest new-connection --service iceberg --config @catalog.json --table ns.orders
+# Iceberg via a REST catalog. --table is REQUIRED (repeatable, namespace.table).
+# catalog.json = pyiceberg catalog properties passed verbatim — {"uri": …,
+# "warehouse": …, "token": …} or "credential" + "scope" for an OAuth exchange.
+
+hotdata ingest list-connections               # ids, statuses; --all includes superseded
+hotdata ingest show-connection <id>           # status + discovered tables/columns
+hotdata ingest delete-connection <id>         # registry + credentials + discovery DB
+                                              # (--keep-database keeps the DB)
+
+# Imports return IMMEDIATELY (non-blocking). Track with `status`:
+hotdata ingest new-import "SELECT * FROM postgres.orders LIMIT 1000"
+hotdata ingest new-import --source postgres --all       # everything, no LIMIT
+hotdata ingest status <ingest-id>             # one-shot; exits 0 done / 1 failed / 2 in flight
+hotdata ingest status <ingest-id> --wait      # attach and poll to done/failed
+
+hotdata ingest list-imports                   # the SQL behind each import + its database
+hotdata ingest trigger-import <ingest-id>     # re-run: refresh the same DB from source
+```
+
+Agent tips:
+- Import SQL is a **restricted grammar**: `SELECT <cols|*> FROM <connection>[.<resource>] [WHERE …] [LIMIT n]` — no joins/GROUP BY. Run real analytics with `hotdata query` against the imported database afterward.
+- Prefer `-o json` + the `status` exit codes for scripting; poll one-shot `status` rather than holding `--wait` when doing other work in between.
+- In-flight statuses stream through stages (`preparing`, `extracting`, `normalizing`, `loading`); treat **any non-terminal value as in-flight** — the set can grow. Terminal = `done` | `failed`.
+- Tables print oldest→newest; `-o json` is newest-first (`[0]` = latest).
+- Once an import is `done`, its database is a regular managed DB: query it with `hotdata query --database <db-id> "SELECT … FROM public.<table>"`.
+- Resources per family: `buckets` connections expose a single resource named `files` (`FROM buckets` or `FROM buckets.files`); iceberg resources are the onboarded tables — address one as `FROM iceberg.NAMESPACE_TABLE` (dots become underscores, same case as onboarded) or by its bare table name.
+- `ingest` connections are independent of `hotdata connections` (the federated-query store) — ingest owns its own credentials.
+
 ### Usage
 ```
 hotdata usage [--since <rfc3339>] [--workspace-id <workspace_id>] [--output table|json|yaml]

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<JobsCommands>,
     },
 
+    /// 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<String>,
+
+        /// 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)

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;

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<String> {
+    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]